diff --git a/hermes_cli/cli_single_query.py b/hermes_cli/cli_single_query.py index 3c6257053d..ed548e1233 100644 --- a/hermes_cli/cli_single_query.py +++ b/hermes_cli/cli_single_query.py @@ -457,6 +457,12 @@ def _run_single_query_mode(cli, query, image, quiet, oneshot, stream_json: bool # full timeout. See #86878. os.environ["HERMES_SINGLE_QUERY_SESSION"] = "1" from hermes_cli.quiet_single_query import exit_single_query + if os.environ.get("HERMES_KANBAN_TASK"): + from tools.kanban_tools import register_current_worker_from_env + if not register_current_worker_from_env(): + # No exit trailer: the task log now belongs to the run that replaced this one. + print("kanban: this worker's run was reclaimed before it started; exiting", file=sys.stderr) + sys.exit(0) if not cli._claim_active_session("cli", stderr=bool(quiet)): exit_single_query(1) try: diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index 06dba3338f..326f1c4d17 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -2479,8 +2479,10 @@ def release_stale_claims( "UPDATE tasks SET status = ?, claim_lock = NULL, " "claim_expires = NULL, worker_pid = NULL, worker_started_at = NULL " "WHERE id = ? AND status = 'running' AND claim_lock IS ? " - "AND claim_expires IS NOT NULL AND claim_expires < ?", - (retry_status, row["id"], row["claim_lock"], now), + "AND claim_expires IS NOT NULL AND claim_expires < ? " + # A worker that registered its own pid since the SELECT keeps its claim. + "AND worker_pid IS ?", + (retry_status, row["id"], row["claim_lock"], now, row["worker_pid"]), ) if cur.rowcount != 1: continue diff --git a/hermes_cli/kanban_db_dispatch.py b/hermes_cli/kanban_db_dispatch.py index 7b71be5569..0fc321320c 100644 --- a/hermes_cli/kanban_db_dispatch.py +++ b/hermes_cli/kanban_db_dispatch.py @@ -1473,6 +1473,31 @@ def _set_worker_pid(conn: sqlite3.Connection, task_id: str, pid: int) -> None: _kb._append_event(conn, task_id, "spawned", {"pid": int(pid), "started_at": started_at}, run_id=run_id) +def adopt_worker_pid(conn: sqlite3.Connection, task_id: str, run_id: int, pid: int) -> bool: + """Worker-side half of ``_set_worker_pid``, run by the worker before its first model call. + + A dispatcher killed between spawning the worker and ``_set_worker_pid`` leaves the run with no + pid: no liveness check can see the worker, so a TTL expiry reclaims the card and spawns a second + worker beside it. The worker fills the missing pid itself (``worker_registered``). False when + ``run_id`` is no longer the card's live run: the card was reclaimed before this worker got here, + and it must exit without working it.""" + started_at = _process_fingerprint(int(pid)) or UNVERIFIED_WORKER_FINGERPRINT + with _kb.write_txn(conn): + row = conn.execute("SELECT status, current_run_id, worker_pid, claim_lock FROM tasks WHERE id = ?", + (task_id,)).fetchone() + if row is None or row["status"] != "running" or row["current_run_id"] != int(run_id): + return False + # Liveness checks are host-local: a pid from another host (or pid namespace) proves nothing here. + if row["worker_pid"] is None and (row["claim_lock"] or "").startswith(_kb._host_prefix()): + conn.execute("UPDATE tasks SET worker_pid = ?, worker_started_at = ? WHERE id = ?", + (int(pid), started_at, task_id)) + conn.execute("UPDATE task_runs SET worker_pid = ?, worker_started_at = ? WHERE id = ?", + (int(pid), started_at, int(run_id))) + _kb._append_event(conn, task_id, "worker_registered", {"pid": int(pid), "started_at": started_at}, + run_id=int(run_id)) + return True + + def _clear_failure_counter(conn: sqlite3.Connection, task_id: str) -> None: """Reset the unified consecutive-failures counter. diff --git a/tests/e2e/core/kanban/test_kanban_dispatcher_restart.py b/tests/e2e/core/kanban/test_kanban_dispatcher_restart.py index 096d9d58b2..c645100c29 100644 --- a/tests/e2e/core/kanban/test_kanban_dispatcher_restart.py +++ b/tests/e2e/core/kanban/test_kanban_dispatcher_restart.py @@ -10,7 +10,8 @@ Invariants, read from kanban.db and the provider's request log: * the ``tasks`` table holds exactly the created ids with their titles and bodies — no row destroyed, replaced or invented (no status-word placeholder id); * every card ends ``done`` with exactly one ``completed`` run and one ``completed`` event; -* each card's ``kanban_complete`` was billed exactly once (no duplicate worker ran a card); +* each card's ``kanban_complete`` was billed exactly once (no duplicate worker ran a card) — also + when the kill lands after the spawn but before the dispatcher recorded the worker's pid; * no card ever had two runs open at the same time. A second scenario gives a live worker a claim TTL shorter than its provider call: the dispatcher @@ -119,7 +120,8 @@ def test_dispatcher_sigkill_mid_tick_never_destroys_or_duplicates_cards(tmp_path assert len(done) == 1 and done[0]["summary"] == f"done {tid}", board.diag(tid) assert len(board.events(tid, "completed")) == 1, board.diag(tid) assert not _overlapping_runs(runs), board.diag(tid) - assert dict(completer.completes) == {t: 1 for t in created}, completer.completes + assert dict(completer.completes) == {t: 1 for t in created}, ( + completer.completes, [board.diag(t) for t, n in completer.completes.items() if n != 1]) # Vacuity guard: the claim-then-kill round really stranded a claim that the restarted # dispatcher had to recover (a kill that always landed between ticks proves nothing). outcomes = Counter(r["outcome"] for t in created for r in board.runs(t)) diff --git a/tests/tools/test_kanban_tools.py b/tests/tools/test_kanban_tools.py index 690edeac68..195fbae4eb 100644 --- a/tests/tools/test_kanban_tools.py +++ b/tests/tools/test_kanban_tools.py @@ -205,7 +205,7 @@ def test_request_review_rejects_unknown_reviewer_without_mutation(monkeypatch, w from tools import kanban_tools as kt (tmp_path / ".hermes" / "profiles" / "verifier").mkdir(parents=True) - (tmp_path / ".hermes" / "profiles" / "verifier" / "config.yaml").write_text("{}\n") # identity marker + (tmp_path / ".hermes" / "profiles" / "verifier" / "config.yaml").write_text("{}\n", encoding="utf-8") # identity marker with kbc.connect() as conn: before = kb.get_task(conn, worker_env) before_events = kb.list_events(conn, worker_env) @@ -226,7 +226,7 @@ def test_request_review_accepts_installed_profile(monkeypatch, worker_env, tmp_p from tools import kanban_tools as kt (tmp_path / ".hermes" / "profiles" / "verifier").mkdir(parents=True) - (tmp_path / ".hermes" / "profiles" / "verifier" / "config.yaml").write_text("{}\n") # identity marker + (tmp_path / ".hermes" / "profiles" / "verifier" / "config.yaml").write_text("{}\n", encoding="utf-8") # identity marker with kbc.connect() as conn: monkeypatch.setenv("HERMES_KANBAN_RUN_ID", str(kb.get_task(conn, worker_env).current_run_id)) @@ -509,6 +509,61 @@ def test_heartbeat_extends_claim_expires(worker_env): ) +def _expire_claim(conn, tid): + conn.execute("UPDATE tasks SET claim_expires = 1 WHERE id = ?", (tid,)) + conn.commit() + + +def test_worker_the_dispatcher_never_recorded_keeps_its_claim_or_never_starts(monkeypatch, worker_env): + """``worker_env`` is a claim whose dispatcher died between spawning the worker and recording its + pid. The worker registers itself, so the expired claim is extended, not handed to a second worker; + a worker that starts only after its run was reclaimed is told not to work the card.""" + import os as _os + from hermes_cli import kanban_db as kb + from hermes_cli import kanban_db_connect as kbc + from tools import kanban_tools as kt + + assert kt.register_current_worker_from_env() is True + with kbc.connect_closing() as conn: + assert kb.get_task(conn, worker_env).worker_pid == _os.getpid() + _expire_claim(conn, worker_env) + assert kb.release_stale_claims(conn) == 0 + assert kb.get_task(conn, worker_env).status == "running" + + late = kb.create_task(conn, title="late orphan", assignee="test-worker") + kb.claim_task(conn, late) + stale_run = kb._current_run_id(conn, late) + _expire_claim(conn, late) + assert kb.release_stale_claims(conn) == 1 + kb.claim_task(conn, late) + monkeypatch.setenv("HERMES_KANBAN_TASK", late) + monkeypatch.setenv("HERMES_KANBAN_RUN_ID", str(stale_run)) + assert kt.register_current_worker_from_env() is False + with kbc.connect_closing() as conn: + assert kb.get_task(conn, late).worker_pid is None + + +def test_reclaim_loses_to_a_worker_registering_mid_sweep(monkeypatch, worker_env): + """The worker registers between the stale-claim SELECT and its UPDATE: the claim stays its own.""" + import os as _os + from hermes_cli import kanban_db as kb + from hermes_cli import kanban_db_connect as kbc + from tools import kanban_tools as kt + + real_terminate = kb._terminate_reclaimed_worker + + def _register_then_terminate(*args, **kwargs): + assert kt.register_current_worker_from_env() is True + return real_terminate(*args, **kwargs) + + monkeypatch.setattr(kb, "_terminate_reclaimed_worker", _register_then_terminate) + with kbc.connect_closing() as conn: + _expire_claim(conn, worker_env) + assert kb.release_stale_claims(conn) == 0 + task = kb.get_task(conn, worker_env) + assert (task.status, task.worker_pid) == ("running", _os.getpid()) + + def test_comment_rejects_caller_supplied_author(worker_env): """Reject an undeclared author override before a worker can forge a comment.""" from tools import kanban_tools as kt diff --git a/tools/kanban_tools.py b/tools/kanban_tools.py index 64db0c316c..e0d4a20bda 100644 --- a/tools/kanban_tools.py +++ b/tools/kanban_tools.py @@ -516,6 +516,24 @@ _auto_heartbeat_last_attempt: float = 0.0 _auto_heartbeat_fence_warned = False +def register_current_worker_from_env() -> bool: + """Record this worker's pid on its run when the dispatcher died before it could + (``adopt_worker_pid``). False only when the board says the run was already reclaimed: + the caller must exit. Anything unreadable (no run id, delegate child, board error) + lets the worker run, as before.""" + tid = os.environ.get("HERMES_KANBAN_TASK") + run_id = _worker_run_id(tid) if tid else None + if run_id is None or _is_delegated_child_context(): + return True + try: + from hermes_cli import kanban_db_dispatch as kbd + with _board(None, quiet_close=True) as (_kb, conn): + return kbd.adopt_worker_pid(conn, tid, run_id, os.getpid()) + except Exception: + logger.debug("kanban worker registration for %s failed", tid, exc_info=True) + return True + + def heartbeat_current_worker_from_env() -> bool: """Claim extension + board heartbeat for the current worker; True iff both writes succeed. ``HERMES_KANBAN_RUN_ID`` pins the run row so a reclaimed stale run is not diff --git a/website/docs/user-guide/features/kanban.md b/website/docs/user-guide/features/kanban.md index 507fa598ce..0363bbd6c8 100644 --- a/website/docs/user-guide/features/kanban.md +++ b/website/docs/user-guide/features/kanban.md @@ -1415,6 +1415,7 @@ Every transition appends a row to `task_events`. Each row carries an optional `r | Kind | Payload | When | |---|---|---| | `spawned` | `{pid}` | Dispatcher successfully started a worker process. | +| `worker_registered` | `{pid, started_at}` | The dispatcher died after starting the worker but before recording its pid, so the worker recorded it itself before its first model call. Liveness checks then see it and an expired claim is extended instead of spawning a second worker. A worker whose run was reclaimed before it got that far exits without working the card. | | `heartbeat` | `{note?}` | Worker called `hermes kanban heartbeat $TASK` to signal liveness during long operations. | | `reclaimed` | `{stale_lock}` | Claim TTL expired without a completion; task goes back to `ready`. An automatic reclaim counts as one non-successful attempt toward the `gave_up` breaker (a claim that never spawned a worker would otherwise loop claim → reclaim → claim forever); an operator `reclaim` resets the counter instead. | | `crashed` | `{pid, claimer, exit_kind?, exit_code?, worker_output?}` | Worker PID no longer alive but TTL hadn't expired yet. `worker_output` is the tail of the worker's own log (its final response or the rendered provider error, chrome stripped, ≤ 400 chars) and is also appended to the task's `last_failure_error`, so the board shows *why* instead of only the exit code. |