Add a bounded dashboard attention projection

docs/task-control-plane.md

@@ -74,8 +74,20 @@ The default Task detail endpoints on the local JSON API are:
 ```text
 GET http://127.0.0.1:8765/v1/receipt?task=task_<opaque-id>
 GET http://127.0.0.1:8765/v1/tasks
+GET http://127.0.0.1:8765/v1/attention?limit=5
 ```
 
+`/v1/attention` computes complete, exclusive counts by each Task's leading
+review reason (failed check, failed step, or blocker) over all visible Tasks,
+then returns only the bounded operational queue. Current failed checks and
+recorded failed steps lead unresolved blockers—even when a Task has both a
+failure and a blocker—and recency orders Tasks within each class. This is review
+ordering, not a claim about business priority. Each row includes its recorded
+reason and next step when available; agentacct does not invent a recovery action
+for a failed check. The complete classification and ordering are cached with the
+parent Task projection, so repeated dashboard polls rebuild them only when that
+projection changes.
+
 ## Owned execution boundary
 
 The local control plane may launch and govern only executions agentacct creates.

src/agentacct/api.py

@@ -101,6 +101,8 @@
 from .task_projection import build_task_projection
 from .receipt import (
     RECEIPT_SCHEMA_VERSION,
+    V1_ATTENTION_SCHEMA_VERSION,
+    build_attention_reason,
     build_receipt,
     build_receipt_summary,
     latest_store_activity,
@@ -2834,6 +2836,7 @@ def v1_version(request: Request) -> dict[str, Any]:
             "session_detail_schema": V1_SESSION_DETAIL_SCHEMA_VERSION,
             "plan_schema": V1_PLAN_SCHEMA_VERSION,
             "receipt_schema": RECEIPT_SCHEMA_VERSION,
+            "attention_schema": V1_ATTENTION_SCHEMA_VERSION,
             "ingestion_schema": V1_INGESTION_SCHEMA_VERSION,
             "pid": os.getpid(),
             "store_dir": str(store_dir),
@@ -2943,6 +2946,8 @@ def v1_plan(
     # rebuilds in parallel and they all blow the client's request timeout.
     v1_receipt_projection_cache: dict[str, tuple[int, float, dict[str, Any]]] = {}
     v1_receipt_projection_lock = threading.Lock()
+    v1_attention_projection_cache: dict[str, Any] = {}
+    v1_attention_projection_lock = threading.Lock()
 
     def _v1_task_projection() -> dict[str, Any]:
         # Keyed on the v1 event log. Machine checks ARE v1 events, so they
@@ -2991,6 +2996,100 @@ def _visible_tasks(projection: Mapping[str, Any]) -> list[dict[str, Any]]:
             if isinstance(task, Mapping) and str(task.get("public_task_id") or "")
         ]
 
+    def _v1_attention_candidates(
+        projection: dict[str, Any],
+    ) -> tuple[
+        list[tuple[int, float, str, Mapping[str, Any], dict[str, Any]]],
+        float | None,
+        dict[str, int],
+    ]:
+        cached = v1_attention_projection_cache.get("value")
+        if cached is not None and cached[0] is projection:
+            return cached[2]
+
+        tasks = _visible_tasks(projection)
+        attention_inputs = [
+            {
+                "task_id": task.get("public_task_id"),
+                "last_activity_at": task.get("last_activity_at"),
+                "work_items": task.get("work_items"),
+                "current_check_events": task.get("current_check_events"),
+                "task_evidence_events": task.get("task_evidence_events"),
+                "finding_episodes": task.get("finding_episodes"),
+            }
+            for task in tasks
+        ]
+        attention_fingerprint = hashlib.sha256(
+            json.dumps(
+                attention_inputs,
+                sort_keys=True,
+                separators=(",", ":"),
+                default=str,
+            ).encode("utf-8")
+        ).hexdigest()
+        with v1_attention_projection_lock:
+            cached = v1_attention_projection_cache.get("value")
+            if cached is not None and cached[0] is projection:
+                return cached[2]
+
+            if cached is not None and cached[1] == attention_fingerprint:
+                cached_candidates, _, cached_counts = cached[2]
+                tasks_by_id = {
+                    str(task.get("public_task_id")): task
+                    for task in tasks
+                }
+                candidates = [
+                    (order_class, recency, task_id, tasks_by_id[task_id], reason)
+                    for order_class, recency, task_id, _, reason in cached_candidates
+                    if task_id in tasks_by_id
+                ]
+                result = (candidates, latest_store_activity(tasks), cached_counts)
+                v1_attention_projection_cache["value"] = (
+                    projection,
+                    attention_fingerprint,
+                    result,
+                )
+                return result
+
+            latest = latest_store_activity(tasks)
+            candidates: list[
+                tuple[int, float, str, Mapping[str, Any], dict[str, Any]]
+            ] = []
+            counts = {"failed_check": 0, "failed_step": 0, "blocker": 0}
+            for task in tasks:
+                classified = build_attention_reason(
+                    task,
+                    latest_store_activity_at=latest,
+                )
+                if classified is None:
+                    continue
+                order_class, reason = classified
+                task_id = str(task.get("public_task_id"))
+                counts[str(reason["kind"])] += 1
+                candidates.append(
+                    (
+                        order_class,
+                        -float(task.get("last_activity_at") or 0.0),
+                        task_id,
+                        task,
+                        reason,
+                    )
+                )
+            candidates.sort(key=lambda candidate: candidate[:3])
+            result = (candidates, latest, counts)
+            # The parent projection is rebuilt on a short TTL even when its
+            # attention inputs are unchanged. Preserve the reduced/sorted index
+            # across those rebuilds by hashing exactly the task state that can
+            # alter classification, reason copy, or ordering. Selected Receipt
+            # rows are still rebuilt from the current projection below, so
+            # cost/project/evidence presentation stays fresh independently.
+            v1_attention_projection_cache["value"] = (
+                projection,
+                attention_fingerprint,
+                result,
+            )
+            return result
+
     @app.get("/v1/tasks")
     def v1_tasks(
         request: Request,
@@ -3028,6 +3127,46 @@ def v1_tasks(
             "truncated": offset + limit < total,
         }
 
+    @app.get("/v1/attention")
+    def v1_attention(
+        request: Request,
+        limit: int = Query(5, ge=1, le=50),
+    ) -> dict[str, Any]:
+        """A complete but bounded review queue over every visible Task.
+
+        This is deliberately separate from ``/v1/tasks``: deriving attention
+        after paginating recent work makes both the count and queue incomplete.
+        The full cached projection is classified once, then only the selected
+        rows pay for compact Receipt summaries. Findings and recorded failed
+        steps lead blockers; recency and the opaque Task id are stable
+        tie-breakers within those operational classes.
+        """
+
+        _require_v1_token(request)
+        projection = _v1_task_projection()
+        candidates, latest, counts = _v1_attention_candidates(projection)
+        selected = candidates[:limit]
+        items = []
+        for _, _, task_id, task, reason in selected:
+            row = build_receipt_summary(
+                task,
+                public_task_id=task_id,
+                title=_task_title(task),
+                latest_store_activity_at=latest,
+            )
+            row["attention"] = reason
+            items.append(row)
+
+        total = len(candidates)
+        return {
+            "schema": V1_ATTENTION_SCHEMA_VERSION,
+            "items": items,
+            "total": total,
+            "counts": counts,
+            "limit": limit,
+            "truncated": total > limit,
+        }
+
     @app.get("/v1/receipt")
     def v1_receipt(
         request: Request,
@@ -3223,6 +3362,7 @@ def index() -> dict[str, Any]:
                 "/v1/session?client=&session_id= (bearer token from the store's local-api.json)",
                 "/v1/plan (bearer token from the store's local-api.json)",
                 "/v1/tasks (bearer token from the store's local-api.json)",
+                "/v1/attention?limit= (bearer token from the store's local-api.json)",
                 "/v1/receipt?task= (bearer token from the store's local-api.json)",
                 "/v1/ingestion (bearer token from the store's local-api.json)",
             ],

src/agentacct/receipt.py

@@ -58,6 +58,7 @@
 
 # frozen: stored/emitted receipts carry this schema string; surfaces pin it.
 RECEIPT_SCHEMA_VERSION = "agentacct.receipt.v1"
+V1_ATTENTION_SCHEMA_VERSION = "agentacct.v1-attention.v1"
 
 # The user-facing provenance vocabulary — ONE list every surface shares. These
 # answer "where did this fact come from", not "how strong is it" (that is the
@@ -1161,10 +1162,129 @@ def build_receipt_summary(
         # The primary root's {client, client_session_id} — the one id a list row
         # carries, so the Work surface can deep-link a session to its Task.
         "primary_root": _mapping(task.get("primary_root")) or None,
+        # Friendly project label from the same boundary reducer as the full
+        # Receipt. This is additive for /v1/tasks and lets a bounded attention
+        # row retain enough context without fetching one full Receipt per row.
+        "project": _boundary(task)["project"],
         "last_activity_at": _number(task.get("last_activity_at")) or None,
     }
 
 
+def build_attention_reason(
+    task: Mapping[str, Any],
+    *,
+    latest_store_activity_at: float | None = None,
+) -> tuple[int, dict[str, Any]] | None:
+    """Return the operational attention class and its truthful leading reason.
+
+    The integer is an internal ordering class, not a business-priority score:
+    standing machine findings and recorded failed steps lead unresolved
+    blockers, matching the existing dashboard grouping. ``None`` means the
+    Task does not currently need review. Human-resolved and superseded findings
+    are excluded by the canonical decision reducer rather than by check totals,
+    so historical failures cannot leak back into the queue.
+    """
+
+    from .finding_disposition import finding_target_digest
+
+    canonical = reduce_task_outcome(task, latest_store_activity_at=latest_store_activity_at)
+    episodes = task.get("finding_episodes") if isinstance(task.get("finding_episodes"), list) else []
+    disposition_by_digest = {
+        str(episode.get("target_digest")): _text(episode.get("disposition_state")) or "open"
+        for episode in episodes
+        if isinstance(episode, Mapping) and episode.get("target_digest")
+    }
+    standing_failures = [
+        check
+        for check in canonical.get("latest_checks", [])
+        if isinstance(check, Mapping)
+        and _text(check.get("result")).lower() in {"failed", "error"}
+        and _text(check.get("supersession_state")).lower() != "superseded"
+    ]
+    failure_states = [
+        (
+            check,
+            disposition_by_digest.get(
+                str(finding_target_digest(check) or ""),
+                "open",
+            ),
+        )
+        for check in standing_failures
+    ]
+    # Preserve the dashboard's established mixed-state rule: a current failed
+    # check leads even when the same Task also carries a recorded blocker. Open
+    # findings lead reviewed-but-still-failing ones, and human-resolved findings
+    # never become the preview merely because they are newer.
+    attention_failures = [row for row in failure_states if row[1] != "resolved"]
+    if attention_failures:
+        preferred_state = (
+            "open"
+            if any(state == "open" for _, state in attention_failures)
+            else "reviewed"
+        )
+        preferred_failures = [
+            check
+            for check, state in attention_failures
+            if state == preferred_state
+        ]
+        finding = max(
+            preferred_failures or [check for check, _ in attention_failures],
+            key=lambda check: (
+                _number(
+                    check.get("created_at")
+                    or check.get("occurred_at")
+                    or check.get("time")
+                ),
+                _int_or_none(check.get("arrival_sequence")) or 0,
+            ),
+        )
+        observed_at = _number(
+            finding.get("created_at")
+            or finding.get("occurred_at")
+            or finding.get("time")
+        )
+        return (
+            0,
+            {
+                "kind": "failed_check",
+                "summary": _text(
+                    finding.get("summary")
+                    or finding.get("name")
+                    or finding.get("evidence_type")
+                )
+                or _DECISION_STATEMENTS["finding"],
+                # A failed check carries no safe inferred remedy. A later PR
+                # may surface an agent-recorded recovery step when one exists.
+                "next_step": None,
+                "observed_at": observed_at or None,
+                "source": _check_source(finding),
+            },
+        )
+
+    decision = _decision_status(
+        task,
+        latest_store_activity_at=latest_store_activity_at,
+        canonical=canonical,
+    )
+    key = _text(decision.get("key"))
+    if key not in {"failed", "blocked"}:
+        return None
+
+    blocker = _mapping(decision.get("blocker"))
+    return (
+        0 if key == "failed" else 1,
+        {
+            "kind": "failed_step" if key == "failed" else "blocker",
+            "summary": _text(
+                blocker.get("text") or blocker.get("step_title") or decision.get("statement")
+            ),
+            "next_step": _text(blocker.get("next_step")) or None,
+            "observed_at": _number(blocker.get("updated_at")) or None,
+            "source": _asserted_by_source(_text(decision.get("asserted_by")), []),
+        },
+    )
+
+
 def latest_store_activity(tasks: list[Mapping[str, Any]]) -> float | None:
     """The newest event time across every Task — the deterministic 'now' the
     outcome reducer compares against (never the wall clock). See
@@ -1178,11 +1298,13 @@ def latest_store_activity(tasks: list[Mapping[str, Any]]) -> float | None:
 
 
 __all__ = [
+    "V1_ATTENTION_SCHEMA_VERSION",
     "RECEIPT_SCHEMA_VERSION",
     "PROVENANCE_LEGEND",
     "EVIDENCE_TIER_LABEL",
     "build_receipt",
     "build_receipt_summary",
+    "build_attention_reason",
     "evidence_coverage_headline",
     "evidence_coverage_ledger",
     "latest_store_activity",