fix(kanban): surface missing assignee profiles

hermes_cli/kanban.py

@@ -2717,7 +2717,7 @@ def _coerce_positive_int(value):
             )
     if res.skipped_nonspawnable:
         print(
-            f"Skipped (non-spawnable assignee — terminal lane, OK): "
+            f"Dispatch failed (missing assignee profile; create it or reassign): "
             f"{', '.join(res.skipped_nonspawnable)}"
         )
     return 0

hermes_cli/kanban_db.py

@@ -7951,12 +7951,10 @@ class DispatchResult:
     operator can see when the dispatcher is acting on the fallback rule
     rather than on explicit per-task assignments."""
     skipped_nonspawnable: list[str] = field(default_factory=list)
-    """Ready task ids skipped because their assignee names a control-plane
-    lane (a Claude Code terminal like ``orion-cc``) rather than a Hermes
-    profile. Expected steady-state on multi-lane setups; NOT an
-    operator-actionable failure. Tracked separately so health telemetry
-    can distinguish "real stuck" (nothing spawned but spawnable work
-    available) from "correctly idle" (nothing spawnable in the queue)."""
+    """Ready/review task ids skipped because their assignee is not an
+    installed Hermes profile. Each real (non-dry-run) skip also emits a
+    deduplicated ``dispatch_nonspawnable_assignee`` event so the failure is
+    durable and machine-readable instead of silently leaving work queued."""
     skipped_per_profile_capped: list[tuple[str, str, int]] = field(default_factory=list)
     """Tasks deferred this tick because their assignee is already at
     ``kanban.max_in_progress_per_profile`` (#21582). Each entry is
@@ -9686,6 +9684,55 @@ def _dispatch_once_locked(
     ):
         max_spawn = max_in_progress
 
+    def record_nonspawnable(task_id: str, assignee: str, task_status: str) -> None:
+        """Persist one actionable event for a missing-profile assignment.
+
+        Dispatcher ticks are frequent, so identical unresolved failures are
+        deduplicated. Reassignment (or a later transition back into the same
+        bad assignment) changes the task's event stream and permits a fresh
+        signal without producing one event per tick.
+        """
+        if dry_run:
+            return
+        error = (
+            f"Assignee profile {assignee!r} does not exist; create that Hermes "
+            "profile or reassign the task to an installed profile."
+        )
+        # Ignore unrelated comments/diagnostic events when deduplicating, but
+        # let an explicit reassignment reset the signal so assigning back to
+        # the same missing name is reported again.
+        latest = conn.execute(
+            "SELECT kind, payload FROM task_events WHERE task_id = ? "
+            "AND kind IN ('dispatch_nonspawnable_assignee', 'assigned') "
+            "ORDER BY id DESC LIMIT 1",
+            (task_id,),
+        ).fetchone()
+        if latest is not None and latest["kind"] == "dispatch_nonspawnable_assignee":
+            try:
+                prior = json.loads(latest["payload"] or "{}")
+            except (TypeError, ValueError, json.JSONDecodeError):
+                prior = {}
+            if (
+                prior.get("assignee") == assignee
+                and prior.get("task_status") == task_status
+                and prior.get("error_code") == "missing_assignee_profile"
+            ):
+                return
+        with write_txn(conn):
+            _append_event(
+                conn,
+                task_id,
+                "dispatch_nonspawnable_assignee",
+                {
+                    "outcome": "dispatch_failed",
+                    "error_code": "missing_assignee_profile",
+                    "error": error,
+                    "assignee": assignee,
+                    "task_status": task_status,
+                    "action": "create_profile_or_reassign",
+                },
+            )
+
     # Count tasks already running so max_spawn enforces concurrency rather
     # than a per-tick spawn budget. See the docstring above for the full
     # rationale; the short version is that a 60-second tick interval with a
@@ -9790,7 +9837,7 @@ def _dispatch_once_locked(
             else:
                 result.skipped_unassigned.append(row["id"])
                 continue
-        # Skip ready tasks whose assignee is not a real Hermes profile.
+        # Fail loudly for ready tasks whose assignee is not a real Hermes profile.
         # `_default_spawn` invokes ``hermes -p <assignee>`` which fails
         # with "Profile 'X' does not exist" when the assignee names a
         # control-plane lane (e.g. an interactive Claude Code terminal
@@ -9805,13 +9852,8 @@ def _dispatch_once_locked(
         except Exception:
             profile_exists = None  # type: ignore[assignment]
         if profile_exists is not None and not profile_exists(row_assignee):
-            # Bucket separately from skipped_unassigned: the operator
-            # cannot fix this by assigning a profile (the assignee IS the
-            # intended owner — a terminal lane). Health telemetry uses
-            # this distinction to suppress spurious "stuck" warnings on
-            # multi-lane setups where the ready queue is steadily full
-            # of human-pulled work.
             result.skipped_nonspawnable.append(row["id"])
+            record_nonspawnable(row["id"], row_assignee, "ready")
             continue
         # Per-profile concurrency cap (#21582): even if there's global
         # headroom, refuse to spawn for an assignee that's already at
@@ -9960,6 +10002,7 @@ def _dispatch_once_locked(
             profile_exists = None  # type: ignore[assignment]
         if profile_exists is not None and not profile_exists(row["assignee"]):
             result.skipped_nonspawnable.append(row["id"])
+            record_nonspawnable(row["id"], row["assignee"], "review")
             continue
         if _per_profile_cap is not None:
             current = _per_profile_running.get(row["assignee"], 0)
@@ -11640,6 +11683,66 @@ def known_assignees(conn: sqlite3.Connection) -> list[dict]:
     ]
 
 
+def nonspawnable_assignee_health(conn: sqlite3.Connection) -> dict[str, Any]:
+    """Return actionable health for queued tasks with missing profiles.
+
+    Only ``ready`` and ``review`` tasks are dispatch candidates. Historical,
+    completed, blocked, and archived assignments do not degrade current board
+    health. If profile discovery itself is unavailable, report ``unknown``
+    rather than manufacturing a clean bill of health.
+    """
+    rows = conn.execute(
+        "SELECT id, title, assignee, status FROM tasks "
+        "WHERE status IN ('ready', 'review') AND claim_lock IS NULL "
+        "AND assignee IS NOT NULL AND assignee != '' "
+        "ORDER BY priority DESC, created_at ASC, id ASC"
+    ).fetchall()
+    try:
+        from hermes_cli.profiles import profile_exists
+    except Exception as exc:
+        return {
+            "status": "unknown",
+            "count": 0,
+            "tasks": [],
+            "error_code": "profile_discovery_unavailable",
+            "error": str(exc),
+        }
+
+    existence: dict[str, bool] = {}
+    affected: list[dict[str, Any]] = []
+    for row in rows:
+        assignee = row["assignee"]
+        if assignee not in existence:
+            try:
+                existence[assignee] = bool(profile_exists(assignee))
+            except Exception as exc:
+                return {
+                    "status": "unknown",
+                    "count": 0,
+                    "tasks": [],
+                    "error_code": "profile_discovery_unavailable",
+                    "error": str(exc),
+                }
+        if existence[assignee]:
+            continue
+        affected.append(
+            {
+                "task_id": row["id"],
+                "title": row["title"],
+                "assignee": assignee,
+                "task_status": row["status"],
+                "error_code": "missing_assignee_profile",
+                "action": "create_profile_or_reassign",
+            }
+        )
+
+    return {
+        "status": "degraded" if affected else "ok",
+        "count": len(affected),
+        "tasks": affected,
+    }
+
+
 # ---------------------------------------------------------------------------
 # Runs (attempt history on a task)
 # ---------------------------------------------------------------------------

plugins/kanban/dashboard/plugin_api.py

@@ -505,6 +505,10 @@ def get_board(
             ],
             "tenants": tenants,
             "assignees": assignees,
+            "health": {
+                "nonspawnable_assignees":
+                    kanban_db.nonspawnable_assignee_health(conn),
+            },
             "latest_event_id": int(latest_event_id),
             "now": int(time.time()),
         }
@@ -2234,6 +2238,21 @@ def get_assignees(board: Optional[str] = Query(None)):
         conn.close()
 
 
+@router.get("/health")
+def get_health(board: Optional[str] = Query(None)):
+    """Actionable dispatch health for the selected board."""
+    board = _resolve_board(board)
+    conn = _conn(board=board)
+    try:
+        nonspawnable = kanban_db.nonspawnable_assignee_health(conn)
+        return {
+            "status": nonspawnable["status"],
+            "nonspawnable_assignees": nonspawnable,
+        }
+    finally:
+        conn.close()
+
+
 # ---------------------------------------------------------------------------
 # Worker log (read-only; file written by _default_spawn)
 # ---------------------------------------------------------------------------

tests/hermes_cli/test_kanban_cli_dispatch_passthrough.py

@@ -94,3 +94,28 @@ def test_cli_max_flag_overrides_config_max_spawn(isolated_kanban_home, monkeypat
     )
 
 
+def test_cli_dispatch_calls_missing_profile_a_failure(
+    isolated_kanban_home, monkeypatch, capsys,
+):
+    """Operator output must not describe a missing assignee profile as OK."""
+    from hermes_cli import kanban as kb_cli
+    from hermes_cli import kanban_db
+
+    monkeypatch.setattr("hermes_cli.config.load_config", lambda: {"kanban": {}})
+    monkeypatch.setattr(
+        kanban_db,
+        "dispatch_once",
+        lambda conn, **kwargs: kanban_db.DispatchResult(
+            skipped_nonspawnable=["t_missing"],
+        ),
+    )
+
+    args = argparse.Namespace(dry_run=False, max=None, failure_limit=2, json=False)
+    kb_cli._cmd_dispatch(args)
+    output = capsys.readouterr().out
+
+    assert "Dispatch failed" in output
+    assert "missing assignee profile" in output
+    assert "OK" not in output
+
+

tests/hermes_cli/test_kanban_db.py

@@ -502,6 +502,149 @@ def test_delete_task_removes_task_and_cascades(kanban_home):
 # ---------------------------------------------------------------------------
 
 
+def test_dispatch_missing_profile_emits_durable_failure_event(kanban_home, monkeypatch):
+    """A missing assignee profile must fail loudly without a spawn loop."""
+    from hermes_cli import profiles
+
+    monkeypatch.setattr(profiles, "profile_exists", lambda name: False)
+    with kb.connect() as conn:
+        task_id = kb.create_task(
+            conn, title="for-missing-profile", assignee="ghost",
+        )
+        result = kb.dispatch_once(conn)
+        events = kb.list_events(conn, task_id)
+        # Frequent ticks must not produce an unbounded duplicate event.
+        kb.dispatch_once(conn)
+        repeated_events = kb.list_events(conn, task_id)
+
+    assert result.skipped_nonspawnable == [task_id]
+    assert not result.spawned
+    failure = events[-1]
+    assert failure.kind == "dispatch_nonspawnable_assignee"
+    assert failure.payload is not None
+    assert failure.payload == {
+        "outcome": "dispatch_failed",
+        "error_code": "missing_assignee_profile",
+        "error": (
+            "Assignee profile 'ghost' does not exist; create that Hermes "
+            "profile or reassign the task to an installed profile."
+        ),
+        "assignee": "ghost",
+        "task_status": "ready",
+        "action": "create_profile_or_reassign",
+    }
+    assert len(repeated_events) == len(events)
+
+
+def test_dispatch_missing_review_profile_emits_failure_event(kanban_home, monkeypatch):
+    """The separately-dispatched review queue gets the same durable signal."""
+    from hermes_cli import profiles
+
+    monkeypatch.setattr(profiles, "profile_exists", lambda name: False)
+    with kb.connect() as conn:
+        task_id = kb.create_task(
+            conn, title="review for missing profile", assignee="ghost",
+        )
+        conn.execute("UPDATE tasks SET status = 'review' WHERE id = ?", (task_id,))
+        conn.commit()
+        result = kb.dispatch_once(conn)
+        failure = kb.list_events(conn, task_id)[-1]
+
+    assert result.skipped_nonspawnable == [task_id]
+    assert failure.kind == "dispatch_nonspawnable_assignee"
+    assert failure.payload is not None
+    assert failure.payload["task_status"] == "review"
+    assert failure.payload["error_code"] == "missing_assignee_profile"
+
+
+def test_dispatch_missing_profile_dry_run_does_not_write_event(
+    kanban_home, monkeypatch,
+):
+    """Dry-run remains read-only while reporting the failed candidate."""
+    from hermes_cli import profiles
+
+    monkeypatch.setattr(profiles, "profile_exists", lambda name: False)
+    with kb.connect() as conn:
+        task_id = kb.create_task(conn, title="dry run", assignee="ghost")
+        before = kb.list_events(conn, task_id)
+        result = kb.dispatch_once(conn, dry_run=True)
+        after = kb.list_events(conn, task_id)
+
+    assert result.skipped_nonspawnable == [task_id]
+    assert after == before
+
+
+def test_dispatch_missing_profile_reassignment_resets_event_dedup(
+    kanban_home, monkeypatch,
+):
+    """Assigning back to the missing profile emits a fresh failure signal."""
+    from hermes_cli import profiles
+
+    monkeypatch.setattr(profiles, "profile_exists", lambda name: False)
+    with kb.connect() as conn:
+        task_id = kb.create_task(conn, title="reassign", assignee="ghost")
+        kb.dispatch_once(conn)
+        kb.assign_task(conn, task_id, "other-ghost")
+        kb.assign_task(conn, task_id, "ghost")
+        kb.dispatch_once(conn)
+        failures = [
+            event for event in kb.list_events(conn, task_id)
+            if event.kind == "dispatch_nonspawnable_assignee"
+        ]
+
+    assert len(failures) == 2
+    assert failures[-1].payload is not None
+    assert failures[-1].payload["assignee"] == "ghost"
+
+
+def test_nonspawnable_assignee_health_reports_only_dispatch_candidates(
+    kanban_home, monkeypatch,
+):
+    """Health degrades for ready/review tasks, not historical assignments."""
+    from hermes_cli import profiles
+
+    monkeypatch.setattr(
+        profiles,
+        "profile_exists",
+        lambda name: name == "installed-worker",
+    )
+    with kb.connect() as conn:
+        kb.create_task(conn, title="valid", assignee="installed-worker")
+        ready_id = kb.create_task(conn, title="missing ready", assignee="ghost")
+        review_id = kb.create_task(conn, title="missing review", assignee="ghost")
+        done_id = kb.create_task(conn, title="missing done", assignee="ghost")
+        conn.execute("UPDATE tasks SET status = 'review' WHERE id = ?", (review_id,))
+        conn.execute("UPDATE tasks SET status = 'done' WHERE id = ?", (done_id,))
+        conn.commit()
+        health = kb.nonspawnable_assignee_health(conn)
+
+    assert health["status"] == "degraded"
+    assert health["count"] == 2
+    assert {task["task_id"] for task in health["tasks"]} == {
+        ready_id, review_id,
+    }
+    assert all(
+        task["error_code"] == "missing_assignee_profile"
+        and task["action"] == "create_profile_or_reassign"
+        for task in health["tasks"]
+    )
+
+
+def test_nonspawnable_assignee_health_is_ok_when_queue_is_spawnable(
+    kanban_home, monkeypatch,
+):
+    from hermes_cli import profiles
+
+    monkeypatch.setattr(profiles, "profile_exists", lambda name: True)
+    with kb.connect() as conn:
+        kb.create_task(conn, title="valid", assignee="installed-worker")
+        assert kb.nonspawnable_assignee_health(conn) == {
+            "status": "ok",
+            "count": 0,
+            "tasks": [],
+        }
+
+