diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index 808f64ba8a89..c0b25dd8e03e 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -163,6 +163,15 @@ def _resolve_claim_ttl_seconds(ttl_seconds: Optional[int] = None) -> int: # during the launch window. DEFAULT_CRASH_GRACE_SECONDS = 30 +# If a host-local PID probe says a running worker is gone, but the current +# run has emitted a heartbeat within this window, treat the PID result as +# advisory and defer crash reclaim. This protects Docker / PID-namespace and +# host/container source-authority mismatches where the dispatcher cannot see +# the child PID even though the worker is actively heartbeating through the +# shared Kanban DB. The heartbeat-staleness paths remain the authority for +# reclaiming workers that stop making progress. +DEFAULT_CRASH_HEARTBEAT_FRESH_SECONDS = 2 * 60 + # Sentinel exit code a kanban worker uses to signal "I bailed because the # provider rate-limited / exhausted quota, not because the task failed." @@ -194,6 +203,27 @@ def _resolve_crash_grace_seconds() -> int: return DEFAULT_CRASH_GRACE_SECONDS +def _resolve_crash_heartbeat_fresh_seconds() -> int: + """Return the heartbeat freshness window for PID-mismatch crash deferral. + + A positive heartbeat window makes ``detect_crashed_workers`` trust a fresh + ``current_run_id`` heartbeat over a host-local PID miss. This is deliberately + much shorter than the stale-heartbeat reclaim threshold: it covers PID + namespace / launch telemetry disagreement without masking workers that stop + making progress. Set ``HERMES_KANBAN_CRASH_HEARTBEAT_FRESH_SECONDS=0`` in + tests or unusual deployments to restore PID-only crash detection. + """ + raw = os.environ.get("HERMES_KANBAN_CRASH_HEARTBEAT_FRESH_SECONDS", "").strip() + if raw: + try: + parsed = int(raw) + except ValueError: + parsed = -1 + if parsed >= 0: + return parsed + return DEFAULT_CRASH_HEARTBEAT_FRESH_SECONDS + + def _resolve_rate_limit_cooldown_seconds() -> int: """Return the rate-limit requeue cooldown in seconds. @@ -5793,7 +5823,9 @@ def detect_crashed_workers(conn: sqlite3.Connection) -> list[str]: # (task_id, pid, claimer, protocol_violation, error_text) with write_txn(conn): rows = conn.execute( - "SELECT id, worker_pid, claim_lock, started_at FROM tasks " + "SELECT id, worker_pid, claim_lock, started_at, " + " last_heartbeat_at, current_run_id " + "FROM tasks " "WHERE status = 'running' AND worker_pid IS NOT NULL" ).fetchall() host_prefix = f"{_claimer_id().split(':', 1)[0]}:" @@ -5813,6 +5845,20 @@ def detect_crashed_workers(conn: sqlite3.Connection) -> list[str]: if _pid_alive(row["worker_pid"]): continue + heartbeat_window = _resolve_crash_heartbeat_fresh_seconds() + last_hb = row["last_heartbeat_at"] + if ( + heartbeat_window > 0 + and row["current_run_id"] is not None + and last_hb is not None + and time.time() - int(last_hb) <= heartbeat_window + ): + # PID visibility is advisory when the active run is still + # writing fresh heartbeats through the board DB. This is the + # host/container PID-namespace case: reclaiming here would + # create a duplicate live run beside a healthy worker. + continue + pid = int(row["worker_pid"]) kind, code = _classify_worker_exit(pid) rate_limited_exit = False diff --git a/tests/hermes_cli/test_kanban_db.py b/tests/hermes_cli/test_kanban_db.py index 24b0e7b0fade..818cdf99096e 100644 --- a/tests/hermes_cli/test_kanban_db.py +++ b/tests/hermes_cli/test_kanban_db.py @@ -843,6 +843,86 @@ def test_detect_crashed_workers_grace_period_env_override( assert tid in kb.detect_crashed_workers(conn) +def test_detect_crashed_workers_trusts_fresh_current_run_heartbeat_on_pid_miss( + kanban_home, monkeypatch, +): + """Fresh current-run heartbeat beats host-local PID mismatch. + + This covers Docker / PID-namespace disagreement: the dispatcher may not see + the child PID, but the worker is still proving liveness through the shared + Kanban DB. Reclaiming in that state would spawn a duplicate live run. + """ + import hermes_cli.kanban_db as _kb + + monkeypatch.setattr(_kb, "_pid_alive", lambda _pid: False) + monkeypatch.setenv("HERMES_KANBAN_CRASH_GRACE_SECONDS", "0") + monkeypatch.setenv("HERMES_KANBAN_CRASH_HEARTBEAT_FRESH_SECONDS", "120") + + now = 3_000_000.0 + monkeypatch.setattr(_kb.time, "time", lambda: now) + + with kb.connect() as conn: + host = _kb._claimer_id().split(":", 1)[0] + tid = kb.create_task(conn, title="pid namespace mismatch", assignee="a") + claimed = kb.claim_task(conn, tid, claimer=f"{host}:w") + assert claimed is not None and claimed.current_run_id is not None + kb._set_worker_pid(conn, tid, 424242) + assert kb.heartbeat_worker(conn, tid, expected_run_id=claimed.current_run_id) + + # Past launch grace, PID probe says dead, but heartbeat is fresh. + monkeypatch.setattr(_kb.time, "time", lambda: now + 90) + assert kb.detect_crashed_workers(conn) == [] + task = kb.get_task(conn, tid) + assert task is not None + assert task.status == "running" + assert task.current_run_id == claimed.current_run_id + + # Once the heartbeat is no longer fresh, PID death is authoritative. + monkeypatch.setattr(_kb.time, "time", lambda: now + 300) + assert kb.detect_crashed_workers(conn) == [tid] + task = kb.get_task(conn, tid) + assert task is not None + assert task.status == "ready" + + +def test_stale_run_completion_cannot_overwrite_new_current_run(kanban_home): + """A duplicate/stale worker finishing late must not corrupt newer run state.""" + with kb.connect() as conn: + tid = kb.create_task(conn, title="stale duplicate completion", assignee="a") + first = kb.claim_task(conn, tid, claimer="host:first") + assert first is not None and first.current_run_id is not None + first_run = int(first.current_run_id) + + with kb.write_txn(conn): + conn.execute( + "UPDATE tasks SET status='ready', claim_lock=NULL, " + "claim_expires=NULL, worker_pid=NULL, current_run_id=NULL " + "WHERE id=?", + (tid,), + ) + second = kb.claim_task(conn, tid, claimer="host:second") + assert second is not None and second.current_run_id is not None + second_run = int(second.current_run_id) + assert second_run != first_run + + assert kb.complete_task( + conn, tid, result="late stale result", expected_run_id=first_run, + ) is False + task = kb.get_task(conn, tid) + assert task is not None + assert task.status == "running" + assert task.current_run_id == second_run + assert task.result is None + + assert kb.complete_task( + conn, tid, result="current result", expected_run_id=second_run, + ) is True + task = kb.get_task(conn, tid) + assert task is not None + assert task.status == "done" + assert task.result == "current result" + + def test_resolve_crash_grace_seconds_handles_bad_env(monkeypatch): """Bad env values fall back to DEFAULT_CRASH_GRACE_SECONDS.""" import hermes_cli.kanban_db as _kb