From 77b4a025d7fa89e0763d5eba3789233f4919a7b0 Mon Sep 17 00:00:00 2001 From: Sergei Kabuldzhanov <55783724+sergeikabuldzhanov@users.noreply.github.com> Date: Fri, 22 May 2026 15:49:12 +0200 Subject: [PATCH] feat: allow forced kanban dispatch --- hermes_cli/kanban.py | 23 ++++++++++++++ hermes_cli/kanban_db.py | 44 +++++++++++++++++++------- tests/hermes_cli/test_kanban_db.py | 50 ++++++++++++++++++++++++++++++ 3 files changed, 106 insertions(+), 11 deletions(-) diff --git a/hermes_cli/kanban.py b/hermes_cli/kanban.py index 4e975bb3e8d78..23b9c41894e56 100644 --- a/hermes_cli/kanban.py +++ b/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: diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index 7a30b70987f65..4c23a28c9d2cd 100644 --- a/hermes_cli/kanban_db.py +++ b/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 diff --git a/tests/hermes_cli/test_kanban_db.py b/tests/hermes_cli/test_kanban_db.py index 435ef41001a9b..41dfc93234925 100644 --- a/tests/hermes_cli/test_kanban_db.py +++ b/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 ):