fix(kanban): reap a superseded run's live worker before respawn

hermes_cli/kanban_db_dispatch.py

@@ -343,6 +343,106 @@ def _worker_survived_termination(termination: dict) -> bool:
     )
 
 
+def _spawned_event_pids(conn: sqlite3.Connection, task_id: str) -> list[tuple[int, int]]:
+    """``(run_id, pid)`` pairs from this task's ``spawned`` events, newest first.
+
+    ``task_runs.worker_pid`` is cleared when a run closes, but the spawned event
+    payload is append-only — the only reliable place to find a run's worker PID
+    after the fact. Used by the respawn guard to reap a previous run's still-live
+    process before spawning its replacement (goal-mode review/fix cycles used to
+    stack 4-6 workers on one task, t_e46a5fea).
+    """
+    rows = conn.execute(
+        "SELECT run_id, payload FROM task_events "
+        "WHERE task_id = ? AND kind = 'spawned' "
+        "ORDER BY id DESC LIMIT 20",
+        (task_id,),
+    ).fetchall()
+    out: list[tuple[int, int]] = []
+    for row in rows:
+        payload = _kb._json_dict(_kb._row_get(row, "payload"))
+        pid = _kb._row_get(payload, "pid")
+        run_id = row["run_id"]
+        if pid and run_id:
+            try:
+                out.append((int(run_id), int(pid)))
+            except (TypeError, ValueError):
+                continue
+    return out
+
+
+def _worker_query(task_id: str) -> str:
+    """The single-source query that ``_worker_argv`` passes to ``chat -q``.
+
+    One definition, so the respawn reap recognises a worker by the very string
+    the dispatcher spawns: a hand-written copy could drift and either miss the
+    orphan (no reap) or, worse, match an unrelated process.
+    """
+    return f"work kanban task {task_id}"
+
+
+def _cmdline_belongs_to_kanban_worker(pid: int, task_id: str) -> bool:
+    """True when ``pid``'s command line is a kanban worker scoped to ``task_id``.
+
+    Guards the respawn reap against PID recycling: we only SIGTERM a PID whose
+    live cmdline still carries the task-scoped query ``_worker_query`` produces,
+    never an innocent process that inherited the number. This is a last-line
+    ownership check on a PID this task itself recorded as ``spawned``, not a
+    process classifier.
+    """
+    try:
+        from gateway.status import _read_process_cmdline as read_cmdline
+        cmdline = read_cmdline(pid) or ""
+    except Exception:
+        return False
+    return _worker_query(task_id) in cmdline
+
+
+def _reap_superseded_run_workers(
+    conn: sqlite3.Connection,
+    task_id: str,
+    keep_run_id: Optional[int] = None,
+    *,
+    signal_fn=None,
+) -> int:
+    """Kill host-local workers of this task's ALREADY-CLOSED runs before a respawn.
+
+    A run that reached a terminal outcome (review_requested / changes_requested /
+    completed / blocked / reclaimed) must not leave its worker process behind:
+    if the old process is still alive when the dispatcher spawns the next run of
+    the same task, two workers mutate one card concurrently (the t_e46a5fea bug:
+    runs 30, 32-36 overlapped for ~30 min). Best-effort — returns the number of
+    PIDs signalled. ``keep_run_id`` is the freshly-claimed run about to spawn,
+    whose spawned event does not exist yet.
+    """
+    if not task_id:
+        return 0
+    reaped = 0
+    for run_id, pid in _spawned_event_pids(conn, task_id):
+        if keep_run_id is not None and run_id == int(keep_run_id):
+            continue
+        # Only reap runs that are actually closed; a live run's pid belongs to
+        # the worker currently owning the claim and must not be touched here.
+        row = conn.execute(
+            "SELECT outcome FROM task_runs WHERE id = ? AND task_id = ?",
+            (run_id, task_id),
+        ).fetchone()
+        if row is None or not row["outcome"]:
+            continue
+        if not _kb._pid_alive(pid):
+            continue
+        if not _cmdline_belongs_to_kanban_worker(pid, task_id):
+            continue
+        info = _terminate_reclaimed_worker(pid, _kb._host_prefix() + ":reap", signal_fn=signal_fn)
+        if info.get("termination_attempted") and info.get("terminated"):
+            reaped += 1
+            _kb._append_event(
+                conn, task_id, "reaped_superseded_worker",
+                {"run_id": run_id, "pid": pid, **{k: v for k, v in info.items() if k != "prev_pid"}},
+            )
+    return reaped
+
+
 def _defer_reclaim_for_live_worker(
     conn: sqlite3.Connection,
     task_id: str,
@@ -1572,6 +1672,16 @@ def _count_spawn(name: str) -> None:
         # Force-load sdlc-review; the kanban lifecycle is already in every
         # worker's system prompt via KANBAN_GUIDANCE.
         claimed.skills = list(dict.fromkeys([*(claimed.skills or []), "sdlc-review"]))
+    # Reap this task's superseded (already-closed) runs before spawning the
+    # replacement. A previous run whose worker is still alive must not overlap
+    # the new one — the goal-mode review/fix cycle stacked 4-6 workers on one
+    # card (t_e46a5fea) because request_review/request_changes closed the run
+    # in the DB but the old process kept running. Best-effort; never blocks a
+    # spawn on the reap outcome.
+    try:
+        _reap_superseded_run_workers(conn, claimed.id, keep_run_id=claimed.current_run_id)
+    except Exception:
+        _kb._log.debug("kanban dispatch: superseded-worker reap failed", exc_info=True)
     try:
         pid = _call_spawn_fn(spawn_fn if spawn_fn is not None else _default_spawn, claimed, str(workspace), board)
         if pid:
@@ -2123,7 +2233,7 @@ def _worker_argv(task: Task, profile_arg: str, hermes_home: Optional[str]) -> li
     worker_toolsets = _resolve_worker_cli_toolsets(hermes_home)
     if worker_toolsets:
         cmd.extend(["--toolsets", ",".join(worker_toolsets)])
-    cmd.extend(["chat", "-q", f"work kanban task {task.id}"])
+    cmd.extend(["chat", "-q", _worker_query(task.id)])
     if task.goal_mode:
         # The kanban goal-loop hook only runs in cli.py's fully-quiet branch.
         # Without -Q the worker gets one turn, prints text, exits rc=0, and the

tests/hermes_cli/test_kanban_reap_superseded_workers.py

@@ -0,0 +1,216 @@
+"""Tests: the dispatcher reaps superseded runs' still-live worker processes
+before spawning a replacement (goal-mode review/fix cycles used to stack
+4-6 workers on one task, t_e46a5fea).
+
+A run reaches a terminal outcome (review_requested / changes_requested / ...)
+in the DB while its worker process is still alive. The respawn guard must find
+that PID (via the append-only ``spawned`` event — ``task_runs.worker_pid`` is
+cleared when the run closes) and terminate it, so two workers never mutate one
+card concurrently. The reap is cmdline-scoped to avoid PID recycling.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+import pytest
+
+from hermes_cli import kanban_db as kb
+from hermes_cli import kanban_db_connect as kbc
+from hermes_cli import kanban_db_dispatch as kbd
+
+
+@pytest.fixture
+def kanban_home(tmp_path, monkeypatch):
+    home = tmp_path / ".hermes"
+    home.mkdir()
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    monkeypatch.setenv("HERMES_KANBAN_HOME", str(home))
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+    db_path = kb.kanban_db_path(board="default")
+    kb._INITIALIZED_PATHS.discard(str(db_path.resolve()))
+    kb.init_db()
+    return home
+
+
+@pytest.fixture
+def conn(kanban_home):
+    with kbc.connect() as c:
+        yield c
+
+
+def _spawned_event(conn, task_id, run_id, pid):
+    """Record a spawned event the way _set_worker_pid does (append-only)."""
+    conn.execute(
+        "INSERT INTO task_events (task_id, run_id, kind, payload, created_at) "
+        "VALUES (?, ?, 'spawned', ?, ?)",
+        (task_id, run_id, json.dumps({"pid": pid}), int(time.time())),
+    )
+    conn.commit()
+
+
+def test_reap_skips_open_run_and_unmatched_pid(conn, monkeypatch):
+    """Only CLOSED runs' workers are reaped; an innocent PID (no kanban worker
+    cmdline) is never signalled. PID recycling is the risk that makes the
+    cmdline check load-bearing: a closed run's recorded PID may now belong to
+    something else entirely."""
+    host = kb._claimer_id().split(":", 1)[0]
+    tid = kb.create_task(conn, title="reap", assignee="w")
+
+    # Old run: claimed, spawned, then closed by its terminal handoff.
+    kb.claim_task(conn, tid, claimer=f"{host}:old")
+    old_run = conn.execute(
+        "SELECT id FROM task_runs WHERE task_id=? ORDER BY id DESC LIMIT 1", (tid,)
+    ).fetchone()["id"]
+    assert kb.request_review(conn, tid, summary="impl done", expected_run_id=old_run)
+
+    # Innocent live process with a PID that must NOT be killed: ourselves.
+    _spawned_event(conn, tid, old_run, os.getpid())
+
+    # The review lane claims the task, opening a new (still open) run.
+    claimed = kb.claim_review_task(conn, tid, claimer=f"{host}:new")
+    assert claimed is not None
+    new_run = conn.execute(
+        "SELECT id FROM task_runs WHERE task_id=? ORDER BY id DESC LIMIT 1", (tid,)
+    ).fetchone()["id"]
+    assert new_run != old_run
+    _spawned_event(conn, tid, new_run, os.getpid())
+
+    def _must_not_signal(*_args, **_kwargs):
+        pytest.fail("reap signalled a PID that is not a worker of this task")
+
+    monkeypatch.setattr(kbd, "_terminate_reclaimed_worker", _must_not_signal)
+    # Our own cmdline is not a kanban worker for this task → not reaped.
+    assert kbd._reap_superseded_run_workers(conn, tid, keep_run_id=new_run) == 0
+    assert kb._pid_alive(os.getpid())
+
+
+def test_reap_terminates_superseded_live_worker(conn, monkeypatch):
+    """A closed run whose spawned PID is a live kanban worker for THIS task is
+    SIGTERMed before the replacement spawns."""
+    host = kb._claimer_id().split(":", 1)[0]
+    tid = kb.create_task(conn, title="reap-live", assignee="w")
+
+    kb.claim_task(conn, tid, claimer=f"{host}:old")
+    old_run = conn.execute(
+        "SELECT id FROM task_runs WHERE task_id=? ORDER BY id DESC LIMIT 1", (tid,)
+    ).fetchone()["id"]
+    assert kb.request_review(conn, tid, summary="impl done", expected_run_id=old_run)
+
+    # Simulate a lingering worker: a real sleep child whose cmdline we will
+    # fake-match as a kanban worker for tid.
+    linger = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"])
+    try:
+        _spawned_event(conn, tid, old_run, linger.pid)
+        # cmdline guard sees through the fake process only if we patch the
+        # identity check to confirm this pid belongs to our task.
+        monkeypatch.setattr(
+            kbd, "_cmdline_belongs_to_kanban_worker", lambda pid, task: pid == linger.pid
+        )
+        assert kb._pid_alive(linger.pid)
+        assert kbd._reap_superseded_run_workers(conn, tid) == 1
+        # Give the SIGTERM a moment to land.
+        linger.wait(timeout=10)
+        assert linger.returncode is not None
+    finally:
+        if linger.poll() is None:
+            linger.kill()
+            linger.wait()
+
+
+def test_reap_noop_without_spawned_events(conn):
+    tid = kb.create_task(conn, title="clean", assignee="w")
+    assert kbd._reap_superseded_run_workers(conn, tid) == 0
+
+
+def _sleep_worker_cmd(tid):
+    """argv whose cmdline reads as a kanban worker scoped to ``tid``."""
+    return [sys.executable, "-c", "import time; time.sleep(120)", f"work kanban task {tid}"]
+
+
+def test_full_review_fix_cycle_keeps_single_live_worker(
+    conn, all_assignees_spawnable,
+):
+    """Replay the t_e46a5fea cycle (implement -> review -> fix -> review ->
+    complete) through the real dispatcher. Old workers deliberately do NOT exit
+    after their terminal tool (the bug); the respawn reap must SIGTERM each
+    superseded process before the next spawn, so at every instant at most one
+    worker process is alive for the task."""
+    import time as _time
+
+    tid = kb.create_task(
+        conn, title="cycle", assignee="dev",
+        body="acceptance: step done", goal_mode=True,
+    )
+    live: dict[int, subprocess.Popen] = {}
+
+    def spawn_fn(task, workspace, board=None):
+        p = subprocess.Popen(_sleep_worker_cmd(tid))
+        live[p.pid] = p
+        return p.pid
+
+    def live_pids():
+        return [pid for pid, p in list(live.items()) if p.poll() is None]
+
+    def run_id():
+        row = conn.execute("SELECT current_run_id FROM tasks WHERE id=?", (tid,)).fetchone()
+        return row["current_run_id"]
+
+    try:
+        # Tick 1 — implementation run spawns (no prior runs).
+        res = kbd.dispatch_once(conn, spawn_fn=spawn_fn)
+        assert any(row[0] == tid for row in res.spawned)
+        first_run = run_id()
+        assert len(live_pids()) == 1
+
+        # Implementer's terminal tool: request_review. Process stays alive (bug).
+        kb.request_review(conn, tid, summary="impl done", expected_run_id=first_run)
+        assert len(live_pids()) == 1  # orphan still alive — the bug
+
+        # Tick 2 — review lane spawns; the reap must kill run 1's worker first.
+        res = kbd.dispatch_once(conn, spawn_fn=spawn_fn)
+        assert any(row[0] == tid for row in res.spawned)
+        second_run = run_id()
+        assert second_run != first_run
+        # Old worker reaped before/around spawn; exactly one remains.
+        assert len(live_pids()) == 1, f"expected 1 live worker, got {live_pids()}"
+
+        # Reviewer's terminal tool: request_changes.
+        kb.request_changes(conn, tid, reason="fix needed", expected_run_id=second_run)
+
+        # Tick 3 — fixer lane spawns; reap run 2's worker.
+        res = kbd.dispatch_once(conn, spawn_fn=spawn_fn)
+        assert any(row[0] == tid for row in res.spawned)
+        third_run = run_id()
+        assert third_run != second_run
+        assert len(live_pids()) == 1
+
+        # Fixer requests review again.
+        kb.request_review(conn, tid, summary="fixed", expected_run_id=third_run)
+
+        # Tick 4 — review lane spawns; reap run 3's worker.
+        res = kbd.dispatch_once(conn, spawn_fn=spawn_fn)
+        assert any(row[0] == tid for row in res.spawned)
+        fourth_run = run_id()
+        assert fourth_run != third_run
+        assert len(live_pids()) == 1
+
+        # Final reviewer completes the task.
+        kb.complete_task(
+            conn, tid, summary="approved", created_cards=[],
+            expected_run_id=fourth_run,
+        )
+        # Even the final worker is eventually gone (or at most one remains).
+        assert len(live_pids()) <= 1
+        task = kb.get_task(conn, tid)
+        assert task is not None and task.status == "done"
+    finally:
+        for p in live.values():
+            if p.poll() is None:
+                p.kill()
+                p.wait()