From 8fb892f1c8f0efe19481241da3d20c7b44646605 Mon Sep 17 00:00:00 2001 From: Casey West Date: Fri, 26 Jun 2026 16:33:22 -0400 Subject: [PATCH] =?UTF-8?q?=F0=9F=90=9B=20fix(kanban):=20never=20spawn=20a?= =?UTF-8?q?=20tool-less=20worker;=20bound=20per-tick=20spawn=20burst?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A dispatcher-spawned worker (Salton, knowledge board) came up with ONLY the base kanban_* coordination tools — no web/shell/git/file — despite its profile correctly declaring a full toolset, then self-blocked ("only kanban_* coordination tools"). Root cause: under a stuck->mass-spawn recovery the dispatcher spawned all 20 ready cards in one tick; _resolve_worker_cli_toolsets came up degenerate and _default_spawn silently launched the worker WITHOUT a --toolsets pin (the `if worker_toolsets:` guard), letting it fall back to a kanban-only surface and waste a full LLM cycle. Two layers, smallest-footprint first, fixing the whole class (the single _default_spawn helper covers both the ready and review dispatch call paths): 1. Never spawn tool-less. _default_spawn now REQUIRES a non-empty resolved CLI toolset and raises when resolution is degenerate. dispatch_once's existing spawn-failure handler records the failure with release_claim=True, so the card is reclaimed to `ready` for a clean retry instead of running crippled. _resolve_worker_cli_toolsets always recovers at least the kanban lifecycle surface for a real profile home, so None/empty is a genuine failure, not a legitimately tool-less profile. 2. Bound the per-tick spawn burst. New kanban.max_spawn_per_tick caps how many workers a single tick may launch (ready + review combined), distinct from max_spawn (a live concurrency cap). Prevents a stuck->recovery tick from dumping the whole ready queue at once — the condition under which workers raced into tool-less spawns. Wired through the gateway dispatcher and the CLI dispatch path; unset preserves historical unbounded behavior. Behavior-contract tests: a spawned worker's resolved toolset is pinned and a degenerate resolution reclaims the card (real _default_spawn against a temp HERMES_HOME, not a mock); N>cap ready cards spawn at most cap per tick; and the gateway forwards kanban.max_spawn_per_tick to dispatch_once. --- gateway/kanban_watchers.py | 12 ++ hermes_cli/config.py | 9 ++ hermes_cli/kanban.py | 6 + hermes_cli/kanban_db.py | 49 +++++++- .../test_kanban_core_functionality.py | 88 +++++++++++++++ tests/hermes_cli/test_kanban_db.py | 105 ++++++++++++++++++ .../test_kanban_worker_spawn_toolsets.py | 69 ++++++++++++ 7 files changed, 336 insertions(+), 2 deletions(-) diff --git a/gateway/kanban_watchers.py b/gateway/kanban_watchers.py index a16007074ab8..0771509872de 100644 --- a/gateway/kanban_watchers.py +++ b/gateway/kanban_watchers.py @@ -736,6 +736,17 @@ async def _kanban_dispatcher_watcher(self) -> None: if max_spawn is not None: logger.info(f"kanban dispatcher: max_spawn={max_spawn}") + # Read max_spawn_per_tick — the per-tick burst bound (distinct from + # max_spawn, which is a live concurrency cap). Caps how many workers a + # single tick may launch so a stuck->recovery tick can't dump the whole + # ready queue at once (incident 2026-06-26). Invalid/<1 values are + # normalized to None (= unbounded) inside dispatch_once. + max_spawn_per_tick = kanban_cfg.get("max_spawn_per_tick", None) + if max_spawn_per_tick is not None: + logger.info( + f"kanban dispatcher: max_spawn_per_tick={max_spawn_per_tick}" + ) + # Cap the number of simultaneously running tasks so slow workers # (local LLMs, resource-constrained hosts) don't pile up and time # out. When set, the dispatcher skips spawning when the board @@ -922,6 +933,7 @@ def _tick_once_for_board(slug: str) -> "Optional[object]": conn, board=slug, max_spawn=max_spawn, + max_spawn_per_tick=max_spawn_per_tick, max_in_progress=max_in_progress, failure_limit=failure_limit, stale_timeout_seconds=stale_timeout_seconds, diff --git a/hermes_cli/config.py b/hermes_cli/config.py index d36f7e8a9c97..d690b4c6b811 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -2343,6 +2343,15 @@ def _ensure_hermes_home_managed(home: Path): # otherwise saturate one profile's local model / API quota / # browser pool while leaving other profiles idle. "max_in_progress_per_profile": None, + # Per-tick spawn burst bound (incident 2026-06-26). Distinct from + # max_spawn (a live concurrency cap counting all running workers): + # this caps how many workers a SINGLE dispatcher tick may launch + # (ready + review combined). Prevents a stuck->recovery tick from + # dumping the whole ready queue at once — the burst that raced + # workers into spawning tool-less. Unset (None) means "no per-tick + # bound" (historical behavior). Invalid/<1 values are treated as + # None. Both caps apply together when set. + "max_spawn_per_tick": None, # When true, the kanban dispatcher auto-runs the decomposer on # tasks that land in Triage (every dispatcher tick). When false, # decomposition is manual via `hermes kanban decompose ` or diff --git a/hermes_cli/kanban.py b/hermes_cli/kanban.py index 31c4bf68ae85..d63eb233168d 100644 --- a/hermes_cli/kanban.py +++ b/hermes_cli/kanban.py @@ -2132,16 +2132,22 @@ def _coerce_positive_int(value): max_spawn = cli_max if cli_max is not None else _coerce_positive_int( _kanban_cfg.get("max_spawn") ) + # Per-tick burst bound (distinct from max_spawn live-concurrency cap). + max_spawn_per_tick = _coerce_positive_int( + _kanban_cfg.get("max_spawn_per_tick") + ) except Exception: default_assignee = None max_in_progress_per_profile = None max_in_progress = None max_spawn = getattr(args, "max", None) + max_spawn_per_tick = None with kb.connect_closing() as conn: res = kb.dispatch_once( conn, dry_run=args.dry_run, max_spawn=max_spawn, + max_spawn_per_tick=max_spawn_per_tick, max_in_progress=max_in_progress, failure_limit=getattr(args, "failure_limit", kb.DEFAULT_SPAWN_FAILURE_LIMIT), default_assignee=default_assignee, diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index c0314c6c101e..9822d61a40ce 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -6223,6 +6223,7 @@ def dispatch_once( ttl_seconds: Optional[int] = None, dry_run: bool = False, max_spawn: Optional[int] = None, + max_spawn_per_tick: Optional[int] = None, max_in_progress: Optional[int] = None, failure_limit: int = DEFAULT_SPAWN_FAILURE_LIMIT, stale_timeout_seconds: int = 0, @@ -6254,6 +6255,15 @@ def dispatch_once( a 60-second tick interval could grow concurrency by N every minute on a busy board and accumulate without bound. + ``max_spawn_per_tick`` is a **per-tick burst bound** (distinct from + ``max_spawn``): it caps how many workers a single tick may launch, + counting ready + review spawns together. Defense-in-depth against the + stuck->mass-spawn recovery pattern (incident 2026-06-26: the dispatcher + sat stuck for 16 ticks, then spawned all 20 ready cards in ONE tick). + Dumping the whole ready queue at once is the condition under which workers + raced and came up tool-less. ``None`` (the default) preserves the + historical unbounded per-tick behavior; both caps apply when set. + ``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. @@ -6322,6 +6332,18 @@ def dispatch_once( if max_spawn is None or max_spawn > remaining: max_spawn = remaining spawned = 0 + # Per-tick burst bound (#incident-2026-06-26). When set, no single tick + # launches more than this many workers (ready + review combined). Normalize + # invalid/<1 values to None (= unbounded) so a typo in config can't wedge + # the dispatcher into never spawning. + _per_tick_cap: Optional[int] = None + if max_spawn_per_tick is not None: + try: + _candidate = int(max_spawn_per_tick) + except (TypeError, ValueError): + _candidate = 0 + if _candidate >= 1: + _per_tick_cap = _candidate # Per-profile concurrency cap (#21582): when set, track how many # workers each assignee already has in flight, and refuse to spawn # when this would push that assignee past the cap. Prevents @@ -6359,6 +6381,8 @@ def dispatch_once( # there, with the existing diagnostic. _default_assignee_resolved = True for row in ready_rows: + if _per_tick_cap is not None and spawned >= _per_tick_cap: + break if max_spawn is not None and running_count + spawned >= max_spawn: break row_assignee = row["assignee"] @@ -6544,6 +6568,8 @@ def dispatch_once( "ORDER BY priority DESC, created_at ASC" ).fetchall() for row in review_rows: + if _per_tick_cap is not None and spawned >= _per_tick_cap: + break if max_spawn is not None and running_count + spawned >= max_spawn: break if not row["assignee"]: @@ -7036,9 +7062,28 @@ def _default_spawn( cmd.extend(["--skills", sk]) if task.model_override: cmd.extend(["-m", task.model_override]) + # Resolve the assignee profile's CLI toolset and pin it explicitly so the + # worker never falls back to a stale root/active-profile config. This MUST + # succeed: a worker spawned without its toolset comes up with only the + # base kanban_* coordination tools (no web/shell/file/git), can't do its + # job, and self-blocks — wasting a full LLM cycle. Under a stuck->mass-spawn + # burst (incident 2026-06-26) resolution can come up degenerate; rather + # than silently launching a crippled worker, FAIL the spawn so the caller + # (dispatch_once) records a spawn failure and RECLAIMS the card to ``ready`` + # for a clean retry on the next tick. _resolve_worker_cli_toolsets always + # recovers at least the kanban lifecycle surface for a real profile home, + # so a None/empty result here is a genuine resolution failure, not a + # legitimately tool-less profile. worker_toolsets = _resolve_worker_cli_toolsets(env.get("HERMES_HOME")) - if worker_toolsets: - cmd.extend(["--toolsets", ",".join(worker_toolsets)]) + if not worker_toolsets: + raise RuntimeError( + f"kanban worker spawn aborted for task {task.id} (profile " + f"{profile_arg!r}): could not resolve a non-empty CLI toolset from " + f"HERMES_HOME={env.get('HERMES_HOME')!r}. Refusing to spawn a " + "tool-less worker (only kanban_* coordination tools); the card will " + "be reclaimed for a clean retry." + ) + cmd.extend(["--toolsets", ",".join(worker_toolsets)]) cmd.extend([ "chat", "-q", prompt, diff --git a/tests/hermes_cli/test_kanban_core_functionality.py b/tests/hermes_cli/test_kanban_core_functionality.py index b51e63f3dc63..de3ac350b100 100644 --- a/tests/hermes_cli/test_kanban_core_functionality.py +++ b/tests/hermes_cli/test_kanban_core_functionality.py @@ -4023,6 +4023,94 @@ async def _sleep(_delay): assert calls["tick"] == 3 +def test_gateway_dispatcher_passes_max_spawn_per_tick_from_config( + monkeypatch, tmp_path +): + """The gateway dispatcher must forward kanban.max_spawn_per_tick. + + Config-propagation guard: the per-tick burst bound is dead code unless the + gateway loop reads ``kanban.max_spawn_per_tick`` and passes it to + ``dispatch_once``. Capture the kwargs the watcher hands ``dispatch_once`` + and assert the configured value arrives. + """ + import asyncio + + from gateway.run import GatewayRunner + import hermes_cli.config as _cfg_mod + import hermes_cli.kanban_db as _kb + + runner = object.__new__(GatewayRunner) + runner._running = True + + monkeypatch.setattr( + _cfg_mod, + "load_config", + lambda: { + "kanban": { + "dispatch_in_gateway": True, + "dispatch_interval_seconds": 1, + "auto_decompose": False, + "max_spawn_per_tick": 3, + } + }, + ) + monkeypatch.setattr( + _kb, "list_boards", + lambda include_archived=False: [{"slug": _kb.DEFAULT_BOARD}], + ) + monkeypatch.setattr(_kb, "read_board_metadata", lambda slug: {"slug": slug}) + monkeypatch.setattr(_kb, "reap_worker_zombies", lambda: []) + monkeypatch.setattr( + _kb, "kanban_db_path", lambda board=None: tmp_path / "kanban.db" + ) + + class _DummyConn: + def close(self): + pass + + monkeypatch.setattr(_kb, "connect", lambda *a, **k: _DummyConn()) + monkeypatch.setattr(_kb, "has_spawnable_ready", lambda conn: True) + monkeypatch.setattr(_kb, "has_spawnable_review", lambda conn: False) + + captured = {} + tick_ran = threading.Event() + + def _dispatch_once(*args, **kwargs): + captured.update(kwargs) + tick_ran.set() + return SimpleNamespace( + spawned=[], reclaimed=0, crashed=[], timed_out=[], + promoted=0, auto_blocked=[], + ) + + monkeypatch.setattr(_kb, "dispatch_once", _dispatch_once) + + async def _scenario(): + watcher = asyncio.ensure_future(runner._kanban_dispatcher_watcher()) + try: + await asyncio.sleep(5.0) # startup delay inside the watcher + t0 = time.monotonic() + while not tick_ran.is_set(): + if time.monotonic() - t0 > 3.0: + break + await asyncio.sleep(0.02) + finally: + runner._running = False + watcher.cancel() + try: + await watcher + except (asyncio.CancelledError, Exception): + pass + + asyncio.run(asyncio.wait_for(_scenario(), timeout=30.0)) + + assert tick_ran.is_set(), "dispatcher tick never ran" + assert captured.get("max_spawn_per_tick") == 3, ( + "gateway dispatcher must forward kanban.max_spawn_per_tick to " + f"dispatch_once; got {captured.get('max_spawn_per_tick')!r}" + ) + + def test_gateway_dispatcher_tick_not_starved_by_busy_default_executor( monkeypatch, tmp_path ): diff --git a/tests/hermes_cli/test_kanban_db.py b/tests/hermes_cli/test_kanban_db.py index c996ae018169..862176ea00f1 100644 --- a/tests/hermes_cli/test_kanban_db.py +++ b/tests/hermes_cli/test_kanban_db.py @@ -1775,6 +1775,111 @@ def fake_spawn(task, workspace): assert kb.get_task(conn, ready_b).status == "ready" +def test_dispatch_reclaims_when_worker_would_spawn_toolless( + kanban_home, all_assignees_spawnable, monkeypatch +): + """E2E: a degenerate toolset resolution must reclaim the card, not block it. + + Exercises the REAL ``_default_spawn`` path (not a stub) against the temp + HERMES_HOME. When toolset resolution comes up degenerate (the burst-race + failure mode), ``_default_spawn`` raises, ``dispatch_once`` records a spawn + failure with ``release_claim=True``, and the card returns to ``ready`` for + a clean retry on the next tick — never a running worker with only + ``kanban_*`` tools. + """ + # Force the burst-race failure mode at the resolution seam. + monkeypatch.setattr(kb, "_resolve_worker_cli_toolsets", lambda home: None) + monkeypatch.setattr(kb, "_resolve_hermes_argv", lambda: ["hermes"]) + + import subprocess as _subprocess + + def fail_popen(*args, **kwargs): # pragma: no cover - must not be reached + raise AssertionError("toolless worker must not be spawned") + + monkeypatch.setattr(_subprocess, "Popen", fail_popen) + + with kb.connect() as conn: + t = kb.create_task(conn, title="needs-tools", assignee="alice") + res = kb.dispatch_once(conn) # real _default_spawn + # Card must NOT have launched a worker; it must be back in ready. + assert res.spawned == [] + task = kb.get_task(conn, t) + assert task.status == "ready" + assert task.claim_lock is None + + +def test_dispatch_max_spawn_per_tick_caps_burst( + kanban_home, all_assignees_spawnable +): + """A single tick must not dump the whole ready queue. + + Defense-in-depth against the 2026-06-26 burst (stuck 16 ticks then + spawned=20 in one tick). ``max_spawn_per_tick`` bounds spawns per tick + distinct from ``max_spawn`` (a live concurrency cap). With N ready cards + and a per-tick cap of C, at most C spawn this tick; the rest stay ready. + """ + spawns = [] + + def fake_spawn(task, workspace): + spawns.append(task.id) + + with kb.connect() as conn: + ready_ids = [ + kb.create_task(conn, title=f"r{i}", assignee="alice") + for i in range(5) + ] + res = kb.dispatch_once(conn, spawn_fn=fake_spawn, max_spawn_per_tick=2) + + assert len(res.spawned) == 2 + assert len(spawns) == 2 + still_ready = [ + tid for tid in ready_ids + if kb.get_task(conn, tid).status == "ready" + ] + assert len(still_ready) == 3 + + +def test_dispatch_max_spawn_per_tick_none_is_unbounded( + kanban_home, all_assignees_spawnable +): + """Omitting max_spawn_per_tick preserves the historical unbounded behavior.""" + spawns = [] + + def fake_spawn(task, workspace): + spawns.append(task.id) + + with kb.connect() as conn: + for i in range(4): + kb.create_task(conn, title=f"r{i}", assignee="alice") + res = kb.dispatch_once(conn, spawn_fn=fake_spawn) + assert len(res.spawned) == 4 + + +def test_dispatch_max_spawn_per_tick_counts_review_spawns( + kanban_home, all_assignees_spawnable +): + """Review spawns count against the per-tick cap alongside ready spawns.""" + spawns = [] + + def fake_spawn(task, workspace): + spawns.append(task.id) + + with kb.connect() as conn: + # 2 ready + 1 review task; per-tick cap of 2 means the review task + # cannot also spawn this tick. + kb.create_task(conn, title="r0", assignee="alice") + kb.create_task(conn, title="r1", assignee="alice") + rev = kb.create_task(conn, title="rev", assignee="alice") + kb.claim_task(conn, rev) + conn.execute("UPDATE tasks SET status = 'review' WHERE id = ?", (rev,)) + conn.execute("UPDATE tasks SET claim_lock = NULL WHERE id = ?", (rev,)) + conn.commit() + + res = kb.dispatch_once(conn, spawn_fn=fake_spawn, max_spawn_per_tick=2) + assert len(res.spawned) == 2 + assert kb.get_task(conn, rev).status == "review" + + def test_dispatch_reclaims_stale_before_spawning(kanban_home): with kb.connect() as conn: t = kb.create_task(conn, title="x", assignee="alice") diff --git a/tests/hermes_cli/test_kanban_worker_spawn_toolsets.py b/tests/hermes_cli/test_kanban_worker_spawn_toolsets.py index 7469a7bf0577..ab1bd59c3c04 100644 --- a/tests/hermes_cli/test_kanban_worker_spawn_toolsets.py +++ b/tests/hermes_cli/test_kanban_worker_spawn_toolsets.py @@ -116,3 +116,72 @@ def test_resolve_worker_cli_toolsets_uses_profile_home_not_parent_config(monkeyp assert "web" in resolved assert "kanban" in resolved # recovered worker lifecycle surface assert resolved != ["kanban"] + + +def test_default_spawn_raises_when_toolset_resolution_degenerate(monkeypatch, tmp_path): + """A worker must NEVER spawn with a degenerate (None/empty) toolset. + + Core regression guard for the 2026-06-26 burst incident: under a + stuck->mass-spawn recovery, ``_resolve_worker_cli_toolsets`` came up empty + and the spawn path silently launched workers with only ``kanban_*`` tools, + which then self-blocked ("only kanban_* coordination tools"). The spawn + must instead FAIL loudly so ``dispatch_once`` reclaims the card for a clean + retry rather than burning an LLM cycle on a crippled worker. + """ + root = tmp_path / ".hermes" + profile = root / "profiles" / "salton" + profile.mkdir(parents=True) + profile.joinpath("config.yaml").write_text( + "toolsets:\n - hermes-cli\n - terminal\n - web\n", + encoding="utf-8", + ) + monkeypatch.setenv("HERMES_HOME", str(root)) + + from hermes_cli import kanban_db as kb + + monkeypatch.setattr(kb, "_resolve_hermes_argv", lambda: ["hermes"]) + # Simulate the burst-race failure mode: resolution comes up degenerate. + monkeypatch.setattr(kb, "_resolve_worker_cli_toolsets", lambda home: None) + + def fail_popen(*args, **kwargs): # pragma: no cover - must not be reached + raise AssertionError("Popen must not run when toolset resolution fails") + + monkeypatch.setattr(subprocess, "Popen", fail_popen) + + workspace = tmp_path / "workspace" + workspace.mkdir() + + import pytest + + with pytest.raises(RuntimeError, match="toolset"): + kb._default_spawn(_make_task(kb, assignee="salton"), str(workspace)) + + +def test_default_spawn_raises_when_toolset_resolution_empty_list(monkeypatch, tmp_path): + """An empty resolved toolset is just as degenerate as None — fail the spawn.""" + root = tmp_path / ".hermes" + profile = root / "profiles" / "salton" + profile.mkdir(parents=True) + profile.joinpath("config.yaml").write_text( + "toolsets:\n - hermes-cli\n - terminal\n", + encoding="utf-8", + ) + monkeypatch.setenv("HERMES_HOME", str(root)) + + from hermes_cli import kanban_db as kb + + monkeypatch.setattr(kb, "_resolve_hermes_argv", lambda: ["hermes"]) + monkeypatch.setattr(kb, "_resolve_worker_cli_toolsets", lambda home: []) + + def fail_popen(*args, **kwargs): # pragma: no cover - must not be reached + raise AssertionError("Popen must not run when toolset resolution is empty") + + monkeypatch.setattr(subprocess, "Popen", fail_popen) + + workspace = tmp_path / "workspace" + workspace.mkdir() + + import pytest + + with pytest.raises(RuntimeError, match="toolset"): + kb._default_spawn(_make_task(kb, assignee="salton"), str(workspace))