feat: canonical workload assessment layer

docs/superpowers/specs/2026-10-06-canonical-load-assessment.md

@@ -0,0 +1,82 @@
+# Canonical Load Assessment
+
+## Purpose
+
+KERNEL already coordinates execution. This layer adds the missing operator-control surface:
+reduce concurrent work to a bounded view, expose dependency pressure, flag likely duplicate work,
+and measure context cost only when a producer supplies an actual measurement.
+
+It does not create a second scheduler, second project registry, or second memory system.
+
+## Canonical ownership
+
+- **Uniformity**: portfolio truth, cross-system identity, dependency/state projection.
+- **Hermes**: operator routing, delegation, verification, human-facing next action.
+- **KERNEL**: local execution, leases, bounded concurrency, provider/cost gates, cache, append-only execution evidence, load assessment.
+- **Domain systems**: own domain state and domain evidence.
+- **Fridge Brain**: diagnostic reasoning and workshop evidence; it does not become the portfolio scheduler.
+- **RESELL / SUBANGEL / LOGARHYTHM**: remain domain boundaries and only exchange explicit contracts.
+
+## Assessment states
+
+NOW = one actively consuming execution slot.
+
+NEXT = up to three dependency-ready items that can safely become the next work.
+
+WAITING = work whose dependency is not verified/published.
+
+BLOCKED = work that cannot proceed because of an explicit failure, rejection, or block.
+
+PARKED = valid work that is not currently consequential enough to compete for attention.
+
+Overflow remains visible as evidence, not as another queue.
+
+## Duplicate-work detection
+
+KERNEL computes a deterministic fingerprint from project identity plus normalized task title.
+A duplicate candidate is a signal, not an automatic merge. The operator must inspect whether
+the tasks actually represent the same work before consolidation.
+
+The canonical rule is:
+
+discover once -> record once -> reuse the evidence everywhere appropriate.
+
+Do not repeatedly rediscover the same repository state, architecture decision, test result,
+or source document in each domain.
+
+## Context-cost measurement
+
+The assessment surface accepts optional estimated_tokens and context_cost fields.
+It reports only values actually supplied by instrumentation.
+
+It must never infer or fabricate provider billing/token usage.
+
+Next instrumentation target: record context-cost receipts at Hermes/KERNEL boundaries so we can
+compare repeated retrieval, repeated summarization, and repeated decision context.
+
+## Dependency implementation order
+
+1. KERNEL assessment primitive (this slice).
+2. Hermes consumes the assessment instead of maintaining another work queue.
+3. Uniformity supplies canonical project/dependency projections.
+4. Add cross-system evidence receipts and identity references.
+5. Instrument context cost and repeated retrieval.
+6. Use measured duplication to remove redundant work.
+7. Only then add broader automation or additional agent capacity.
+
+## Release gate
+
+The layer is useful when:
+
+- one operator view identifies the current work without reading every project;
+- dependencies explain why work is waiting;
+- duplicate candidates are visible without destructive auto-merges;
+- no new scheduler or memory store is introduced;
+- measured context cost can be traced to a source event;
+- the Golden Loop remains the release gate for physical/workshop functionality.
+
+## Explicit non-goals
+
+No new LLM, no new plugin, no autonomous task deletion, no automatic cross-repository merging,
+no replacement of Uniformity, and no claim that ChatGPT billing/token totals are measured until
+a real provider receipt exists.

scripts/assessment.py

@@ -0,0 +1,127 @@
+"""Canonical workload assessment for KERNEL.
+
+This layer does not execute work. It reduces concurrent work into one bounded
+operator view, identifies dependency waits and duplicate candidates, and
+records measurable context-cost fields when producers provide them.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import re
+from collections import defaultdict
+from typing import Any
+
+
+ACTIVE_STATUSES = {"active", "staged"}
+DONE_STATUSES = {"verified", "published"}
+WAITING_STATUSES = {"planned", "ready"}
+BLOCKED_STATUSES = {"blocked", "failed", "rejected"}
+
+
+def _normalise_title(title: str) -> str:
+    text = re.sub(r"[^a-z0-9]+", " ", str(title).lower())
+    return re.sub(r"\s+", " ", text).strip()
+
+
+def _fingerprint(task: dict[str, Any]) -> str:
+    project = str(task.get("project", ""))
+    title = _normalise_title(task.get("title", task.get("id", "")))
+    return hashlib.sha256(f"{project}|{title}".encode("utf-8")).hexdigest()[:16]
+
+
+def assess(tasks: list[dict[str, Any]], *, now_limit: int = 1, next_limit: int = 3) -> dict[str, Any]:
+    """Return a bounded operator assessment without changing task state.
+
+    Priority is deliberately deterministic:
+    1. active work already consuming execution capacity;
+    2. ready work with no unmet dependencies;
+    3. blocked/failed/rejected work;
+    4. dependency-waiting work;
+    5. planned low-risk work is parked rather than promoted.
+    """
+
+    by_id = {str(task.get("id")): task for task in tasks}
+    duplicate_groups: dict[str, list[str]] = defaultdict(list)
+    for task in tasks:
+        duplicate_groups[_fingerprint(task)].append(str(task.get("id")))
+
+    duplicates = [
+        {"fingerprint": key, "task_ids": ids}
+        for key, ids in duplicate_groups.items()
+        if len(ids) > 1
+    ]
+
+    now: list[dict[str, Any]] = []
+    next_items: list[dict[str, Any]] = []
+    waiting: list[dict[str, Any]] = []
+    blocked: list[dict[str, Any]] = []
+    parked: list[dict[str, Any]] = []
+
+    for task in tasks:
+        status = task.get("status", "planned")
+        dependencies = [str(dep) for dep in task.get("depends_on", [])]
+        unmet = [dep for dep in dependencies if by_id.get(dep, {}).get("status") not in DONE_STATUSES]
+
+        item = {
+            "task_id": task.get("id"),
+            "project": task.get("project"),
+            "title": task.get("title", task.get("id")),
+            "status": status,
+            "unmet_dependencies": unmet,
+            "risk": task.get("risk", "low"),
+            "estimated_tokens": task.get("estimated_tokens"),
+            "context_cost": task.get("context_cost"),
+        }
+
+        if status in ACTIVE_STATUSES:
+            now.append(item)
+        elif status in BLOCKED_STATUSES:
+            blocked.append(item)
+        elif status in WAITING_STATUSES and not unmet:
+            next_items.append(item)
+        elif status in WAITING_STATUSES:
+            waiting.append(item)
+        else:
+            parked.append(item)
+
+    now.sort(key=lambda x: str(x["task_id"]))
+    next_items.sort(key=lambda x: (x["risk"] != "high", str(x["task_id"])))
+    waiting.sort(key=lambda x: (len(x["unmet_dependencies"]), str(x["task_id"])))
+    blocked.sort(key=lambda x: str(x["task_id"]))
+    parked.sort(key=lambda x: str(x["task_id"]))
+
+    # Only the bounded top slice is actionable. Everything else remains visible
+    # as waiting/parked evidence rather than becoming another queue to micromanage.
+    bounded_now = now[: max(1, now_limit)]
+    bounded_next = next_items[: max(1, next_limit)]
+
+    token_values = [
+        task.get("estimated_tokens")
+        for task in tasks
+        if isinstance(task.get("estimated_tokens"), (int, float))
+    ]
+    context_values = [
+        task.get("context_cost")
+        for task in tasks
+        if isinstance(task.get("context_cost"), (int, float))
+    ]
+
+    return {
+        "now": bounded_now,
+        "next": bounded_next,
+        "waiting": waiting,
+        "blocked": blocked,
+        "parked": parked,
+        "overflow": {
+            "active": max(0, len(now) - len(bounded_now)),
+            "ready": max(0, len(next_items) - len(bounded_next)),
+        },
+        "duplicates": duplicates,
+        "measurement": {
+            "tasks": len(tasks),
+            "estimated_tokens_total": sum(token_values) if token_values else None,
+            "context_cost_total": sum(context_values) if context_values else None,
+            "measured_task_count": len(token_values),
+        },
+    }

scripts/kernel.py

@@ -154,6 +154,11 @@ def main(argv=None):
         providers = [adapter.health() for adapter in load_adapters(config)] if config.exists() else []
         print(json.dumps({"providers": providers}, indent=2)); return 0
     if args.command == "metrics": print(json.dumps(metrics(args.board), indent=2)); return 0
+    if args.command == "assess":
+        from assessment import assess
+        report = assess(data.get("tasks", []))
+        event(args.board, "board.assessed", {"measurement": report["measurement"], "duplicates": len(report["duplicates"])})
+        print(json.dumps(report, indent=2)); return 0
     return 0
 if __name__ == "__main__": sys.exit(main())