feat: allow forced kanban dispatch

hermes_cli/kanban.py

@@ -575,6 +575,11 @@ def build_parser(parent_subparsers: argparse._SubParsersAction) -> argparse.Argu
                         help="Don't actually spawn processes; just print what would happen")
     p_disp.add_argument("--max", type=int, default=None,
                         help="Cap number of spawns this pass")
+    p_disp.add_argument("--task", "--task-id", dest="task_ids", action="append",
+                        default=None, metavar="TASK_ID",
+                        help="Only dispatch the named task (repeatable).")
+    p_disp.add_argument("--force", action="store_true",
+                        help="Bypass respawn guards such as active_pr/recent_success for selected ready tasks.")
     p_disp.add_argument("--failure-limit", type=int,
                         default=kb.DEFAULT_SPAWN_FAILURE_LIMIT,
                         help=f"Auto-block a task after this many consecutive non-success attempts "
@@ -2008,6 +2013,8 @@ def _cmd_dispatch(args: argparse.Namespace) -> int:
             dry_run=args.dry_run,
             max_spawn=args.max,
             failure_limit=getattr(args, "failure_limit", kb.DEFAULT_SPAWN_FAILURE_LIMIT),
+            task_ids=getattr(args, "task_ids", None),
+            force=getattr(args, "force", False),
         )
     if getattr(args, "json", False):
         print(json.dumps({
@@ -2021,6 +2028,14 @@ def _cmd_dispatch(args: argparse.Namespace) -> int:
                 {"task_id": tid, "assignee": who, "workspace": ws}
                 for (tid, who, ws) in res.spawned
             ],
+            "respawn_guarded": [
+                {"task_id": tid, "reason": reason}
+                for (tid, reason) in res.respawn_guarded
+            ],
+            "respawn_guard_overridden": [
+                {"task_id": tid, "reason": reason}
+                for (tid, reason) in res.respawn_guard_overridden
+            ],
             "skipped_unassigned": res.skipped_unassigned,
             "skipped_nonspawnable": res.skipped_nonspawnable,
         }, indent=2))
@@ -2043,6 +2058,14 @@ def _cmd_dispatch(args: argparse.Namespace) -> int:
     for tid, who, ws in res.spawned:
         tag = " (dry)" if args.dry_run else ""
         print(f"  - {tid}  ->  {who}  @ {ws or '-'}{tag}")
+    if res.respawn_guarded:
+        print("Respawn guarded:")
+        for tid, reason in res.respawn_guarded:
+            print(f"  - {tid}: {reason}")
+    if res.respawn_guard_overridden:
+        print("Respawn guards overridden:")
+        for tid, reason in res.respawn_guard_overridden:
+            print(f"  - {tid}: {reason}")
     if res.skipped_unassigned:
         print(f"Skipped (unassigned): {', '.join(res.skipped_unassigned)}")
     if res.skipped_nonspawnable:

hermes_cli/kanban_db.py

@@ -3710,6 +3710,8 @@ class DispatchResult:
     Reasons: ``"blocker_auth"`` (quota/auth error — also auto-blocked),
     ``"recent_success"`` (completed run within guard window),
     ``"active_pr"`` (GitHub PR URL in a recent comment)."""
+    respawn_guard_overridden: list[tuple[str, str]] = field(default_factory=list)
+    """Tasks whose respawn guard was explicitly ignored by a force dispatch."""
 
 
 # Bounded registry of recently-reaped worker child exits, populated by the
@@ -4710,6 +4712,8 @@ def dispatch_once(
     failure_limit: int = DEFAULT_SPAWN_FAILURE_LIMIT,
     stale_timeout_seconds: int = 0,
     board: Optional[str] = None,
+    task_ids: Optional[Iterable[str]] = None,
+    force: bool = False,
 ) -> DispatchResult:
     """Run one dispatcher tick.
 
@@ -4738,6 +4742,9 @@ def dispatch_once(
     ``spawn_fn`` defaults to ``_default_spawn``. Tests pass a stub.
     ``board`` pins workspace/log/db resolution for this tick to a specific
     board. When omitted, the current-board resolution chain is used.
+    ``task_ids`` optionally restricts dispatch to named tasks. ``force``
+    bypasses respawn guards (active PR / recent success / auth blocker) and
+    records a ``respawn_guard_overridden`` event for auditability.
     """
     # Reap zombie children from previously spawned workers.
     # The gateway-embedded dispatcher is the parent of every worker spawned
@@ -4773,6 +4780,8 @@ def dispatch_once(
             pass
 
     result = DispatchResult()
+    task_filter = {str(t) for t in (task_ids or ()) if str(t).strip()}
+
     result.reclaimed = release_stale_claims(conn)
     result.stale = detect_stale_running(
         conn, stale_timeout_seconds=stale_timeout_seconds,
@@ -4809,6 +4818,8 @@ def dispatch_once(
         "WHERE status = 'ready' AND claim_lock IS NULL "
         "ORDER BY priority DESC, created_at ASC"
     ).fetchall()
+    if task_filter:
+        ready_rows = [row for row in ready_rows if row["id"] in task_filter]
     # Honour kanban.max_in_progress: if the board already has enough running
     # tasks, skip spawning this tick so slow workers (local LLMs,
     # resource-constrained hosts) can finish what they have before more tasks
@@ -4863,17 +4874,26 @@ def dispatch_once(
         # blocks via the normal path rather than on first occurrence.
         guard_reason = check_respawn_guard(conn, row["id"])
         if guard_reason is not None:
-            result.respawn_guarded.append((row["id"], guard_reason))
-            # Emit an event so operators can see why the task was
-            # skipped when reading `hermes kanban tail` — without
-            # this the task appears stuck in ready with no diagnosis.
-            if not dry_run:
-                with write_txn(conn):
-                    _append_event(
-                        conn, row["id"], "respawn_guarded",
-                        {"reason": guard_reason},
-                    )
-            continue
+            if force:
+                result.respawn_guard_overridden.append((row["id"], guard_reason))
+                if not dry_run:
+                    with write_txn(conn):
+                        _append_event(
+                            conn, row["id"], "respawn_guard_overridden",
+                            {"reason": guard_reason},
+                        )
+            else:
+                result.respawn_guarded.append((row["id"], guard_reason))
+                # Emit an event so operators can see why the task was
+                # skipped when reading `hermes kanban tail` — without
+                # this the task appears stuck in ready with no diagnosis.
+                if not dry_run:
+                    with write_txn(conn):
+                        _append_event(
+                            conn, row["id"], "respawn_guarded",
+                            {"reason": guard_reason},
+                        )
+                continue
         if dry_run:
             result.spawned.append((row["id"], row["assignee"], ""))
             continue
@@ -4939,6 +4959,8 @@ def dispatch_once(
         "WHERE status = 'review' AND claim_lock IS NULL "
         "ORDER BY priority DESC, created_at ASC"
     ).fetchall()
+    if task_filter:
+        review_rows = [row for row in review_rows if row["id"] in task_filter]
     for row in review_rows:
         if max_spawn is not None and running_count + spawned >= max_spawn:
             break

tests/hermes_cli/test_kanban_db.py

@@ -1373,6 +1373,56 @@ def fake_spawn(task, workspace):
         assert kb.get_task(conn, t).status == "ready"
 
 
+def test_dispatch_force_overrides_active_pr_guard(
+    kanban_home, all_assignees_spawnable
+):
+    """force=True lets operators intentionally re-run a ready task with an active PR."""
+    spawned_ids = []
+
+    def fake_spawn(task, workspace):
+        spawned_ids.append(task.id)
+
+    with kb.connect() as conn:
+        t = kb.create_task(conn, title="has-pr", assignee="alice")
+        kb.add_comment(
+            conn, t, "worker",
+            "Opened https://github.com/totemx-AI/subsidysmart/pull/99",
+        )
+        res = kb.dispatch_once(conn, spawn_fn=fake_spawn, force=True)
+        events = kb.list_events(conn, t)
+
+    assert (t, "active_pr") in res.respawn_guard_overridden
+    assert (t, "active_pr") not in res.respawn_guarded
+    assert t in spawned_ids
+    assert t not in res.auto_blocked
+    overridden = [e for e in events if e.kind == "respawn_guard_overridden"]
+    assert overridden
+    assert overridden[-1].payload.get("reason") == "active_pr"
+    with kb.connect() as conn:
+        assert kb.get_task(conn, t).status == "running"
+
+
+def test_dispatch_task_filter_only_spawns_named_task(
+    kanban_home, all_assignees_spawnable
+):
+    """task_ids narrows a dispatcher pass to an operator-selected task."""
+    spawned_ids = []
+
+    def fake_spawn(task, workspace):
+        spawned_ids.append(task.id)
+
+    with kb.connect() as conn:
+        selected = kb.create_task(conn, title="selected", assignee="alice")
+        other = kb.create_task(conn, title="other", assignee="bob")
+        res = kb.dispatch_once(conn, spawn_fn=fake_spawn, task_ids=[selected])
+
+    assert spawned_ids == [selected]
+    assert [item[0] for item in res.spawned] == [selected]
+    with kb.connect() as conn:
+        assert kb.get_task(conn, selected).status == "running"
+        assert kb.get_task(conn, other).status == "ready"
+
+
 def test_dispatch_respawn_guard_dry_run_no_auto_block(
     kanban_home, all_assignees_spawnable
 ):