diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index b684450e6bb4..c82d762d5924 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -121,6 +121,16 @@ # effect of normal API traffic. DEFAULT_CLAIM_HEARTBEAT_MAX_STALE_SECONDS = 60 * 60 +# Grace added to a claim when a reclaim is deferred because the previous +# host-local worker is still alive after a termination attempt. Releasing the +# claim in that state would spawn a duplicate alongside the surviving worker — +# the runaway seen when a cgroup memory.high throttle parks a worker in +# uninterruptible (D) state, where a pending SIGKILL cannot be delivered until +# the throttle lifts. Holding the claim a short grace and retrying next tick +# stops the duplication; once no duplicate is spawned the pressure eases, the +# signal lands, and the following tick reclaims cleanly. +RECLAIM_DEFER_GRACE_SECONDS = 120 + def _resolve_claim_ttl_seconds(ttl_seconds: Optional[int] = None) -> int: """Return the effective claim TTL, honoring the kanban env override. @@ -3286,6 +3296,14 @@ def release_stale_claims( termination = _terminate_reclaimed_worker( row["worker_pid"], row["claim_lock"], signal_fn=signal_fn, ) + # Never release a claim while our own worker is still alive: that would + # spawn a duplicate beside it. Hold the claim and retry next tick. + if _worker_survived_termination(termination): + _defer_reclaim_for_live_worker( + conn, row["id"], row["claim_lock"], now, termination, + reason="ttl_expired_worker_alive", + ) + continue with write_txn(conn): cur = conn.execute( "UPDATE tasks SET status = 'ready', claim_lock = NULL, " @@ -5113,7 +5131,13 @@ def _terminate_reclaimed_worker( info["termination_attempted"] = True try: kill(int(pid), signal.SIGTERM) - except (ProcessLookupError, OSError): + except ProcessLookupError: + # Process is already gone — that's a successful termination, not a + # survival. Leaving terminated=False here would make the reclaim guard + # misread a dead worker as still-alive and defer forever. + info["terminated"] = True + return info + except OSError: return info for _ in range(10): @@ -5136,6 +5160,63 @@ def _terminate_reclaimed_worker( return info +def _worker_survived_termination(termination: dict) -> bool: + """True when we tried to kill our own host-local worker and it is still alive. + + Reclaiming in this state would release the claim and let the dispatcher + spawn a second worker while the first is still running — the duplication + loop. Only host-local workers we actually signalled count: a non-local + claim lock or a no-op attempt (no ``os.kill`` available) must fall through + to the normal release path, since we cannot manage that worker anyway. + """ + return bool( + termination.get("termination_attempted") + and termination.get("host_local") + and not termination.get("terminated") + ) + + +def _defer_reclaim_for_live_worker( + conn: sqlite3.Connection, + task_id: str, + claim_lock: Optional[str], + now: int, + termination: dict, + *, + reason: str, +) -> None: + """Hold a claim whose worker survived termination instead of releasing it. + + Extends ``claim_expires`` by ``RECLAIM_DEFER_GRACE_SECONDS`` so the task + stays ``running`` (no duplicate spawn) and records a ``reclaim_deferred`` + event so the hold is visible in ``hermes kanban tail``. The next dispatch + tick retries the kill; this is self-correcting because not spawning a + duplicate is what lets the throttled worker finally die. + """ + grace = now + RECLAIM_DEFER_GRACE_SECONDS + with write_txn(conn): + cur = conn.execute( + "UPDATE tasks SET claim_expires = ? " + "WHERE id = ? AND status = 'running' AND claim_lock IS ?", + (grace, task_id, claim_lock), + ) + if cur.rowcount != 1: + return + run_id = _current_run_id(conn, task_id) + if run_id is not None: + conn.execute( + "UPDATE task_runs SET claim_expires = ? WHERE id = ?", + (grace, run_id), + ) + payload = { + "reason": reason, + "claim_lock": claim_lock, + "claim_expires_now": grace, + } + payload.update(termination) + _append_event(conn, task_id, "reclaim_deferred", payload, run_id=run_id) + + def heartbeat_worker( conn: sqlite3.Connection, task_id: str, @@ -5374,6 +5455,15 @@ def detect_stale_running( pid, lock, signal_fn=signal_fn, ) + # Never release a claim while our own worker is still alive: that would + # spawn a duplicate beside it. Hold the claim and retry next tick. + if _worker_survived_termination(termination): + _defer_reclaim_for_live_worker( + conn, tid, lock, now, termination, + reason="heartbeat_stale_worker_alive", + ) + continue + with write_txn(conn): cur = conn.execute( "UPDATE tasks SET status = 'ready', claim_lock = NULL, " diff --git a/scripts/release.py b/scripts/release.py index a4bf3c797642..772b11541cd9 100755 --- a/scripts/release.py +++ b/scripts/release.py @@ -1584,6 +1584,7 @@ "andrewdmwalker@gmail.com": "capt-marbles", # PR #38440 salvage (resolve xAI OAuth credentials across profiles; #43589) "infinitycrew39@gmail.com": "infinitycrew39", # PR #47945 salvage (scope langfuse trace state by turn/request ids; #48292) "eurekaxun@163.com": "huangxun375-stack", # PR #37251 / #48894 structured OpenViking sync + "218421507+Sahil-SS9@users.noreply.github.com": "Sahil-SS9", # PR #48466/#44919/#44909/#42209 salvage (cron/checkpoint/kanban/skill) } diff --git a/tests/hermes_cli/test_kanban_db.py b/tests/hermes_cli/test_kanban_db.py index 8bb5c1a7b85c..1386b1ebdc47 100644 --- a/tests/hermes_cli/test_kanban_db.py +++ b/tests/hermes_cli/test_kanban_db.py @@ -505,6 +505,171 @@ def test_stale_claim_with_live_pid_uses_env_ttl_override( assert task.claim_expires > int(time.time()) + 3000 +def test_stale_claim_deferred_when_live_worker_survives_termination( + kanban_home, monkeypatch, +): + """A TTL-expired claim whose worker survives the kill must NOT be released. + + Releasing would let the dispatcher spawn a duplicate beside the still-alive + worker — the runaway seen when a cgroup memory.high throttle parks a worker + in uninterruptible (D) state, where a pending SIGKILL cannot land. The claim + is held (extended) and retried next tick instead. + """ + import hermes_cli.kanban_db as _kb + + with kb.connect() as conn: + t = kb.create_task(conn, title="x", assignee="a") + host = _kb._claimer_id().split(":", 1)[0] + kb.claim_task(conn, t, claimer=f"{host}:worker") + kb._set_worker_pid(conn, t, 12345) + + old_expires = int(time.time()) - 60 + # Heartbeat stale by > 1h so the live-pid EXTEND branch is skipped and + # the terminate path (the wedged-worker case) runs. + conn.execute( + "UPDATE tasks SET claim_expires = ?, last_heartbeat_at = ? " + "WHERE id = ?", + (old_expires, int(time.time()) - 7200, t), + ) + monkeypatch.setattr(_kb, "_pid_alive", lambda _pid: True) + monkeypatch.setattr( + _kb, "_terminate_reclaimed_worker", + lambda *a, **k: { + "termination_attempted": True, + "host_local": True, + "terminated": False, + }, + ) + reclaimed = kb.release_stale_claims(conn, signal_fn=lambda _p, _s: None) + assert reclaimed == 0 + + assert kb.get_task(conn, t).status == "running" + worker_pid = conn.execute( + "SELECT worker_pid FROM tasks WHERE id = ?", (t,), + ).fetchone()[0] + assert worker_pid == 12345 # worker not orphaned + claim_expires = conn.execute( + "SELECT claim_expires FROM tasks WHERE id = ?", (t,), + ).fetchone()[0] + assert claim_expires > old_expires # claim held, not released + + kinds = [ + r["kind"] for r in conn.execute( + "SELECT kind FROM task_events WHERE task_id = ?", (t,), + ).fetchall() + ] + assert "reclaim_deferred" in kinds + assert "reclaimed" not in kinds + + +def test_stale_claim_reclaimed_when_termination_succeeds( + kanban_home, monkeypatch, +): + """When the worker is actually killed, the claim is released as before.""" + import hermes_cli.kanban_db as _kb + + with kb.connect() as conn: + t = kb.create_task(conn, title="x", assignee="a") + host = _kb._claimer_id().split(":", 1)[0] + kb.claim_task(conn, t, claimer=f"{host}:worker") + kb._set_worker_pid(conn, t, 12345) + conn.execute( + "UPDATE tasks SET claim_expires = ?, last_heartbeat_at = ? " + "WHERE id = ?", + (int(time.time()) - 60, int(time.time()) - 7200, t), + ) + monkeypatch.setattr(_kb, "_pid_alive", lambda _pid: False) + monkeypatch.setattr( + _kb, "_terminate_reclaimed_worker", + lambda *a, **k: { + "termination_attempted": True, + "host_local": True, + "terminated": True, + }, + ) + reclaimed = kb.release_stale_claims(conn, signal_fn=lambda _p, _s: None) + assert reclaimed == 1 + assert kb.get_task(conn, t).status == "ready" + + +def test_stale_claim_released_when_worker_not_host_local( + kanban_home, monkeypatch, +): + """The defer guard only holds OUR own surviving workers. + + A claim we cannot manage (different host, or no kill attempted) must still + be released, otherwise a foreign-host claim could strand a task forever. + """ + import hermes_cli.kanban_db as _kb + + with kb.connect() as conn: + t = kb.create_task(conn, title="x", assignee="a") + host = _kb._claimer_id().split(":", 1)[0] + kb.claim_task(conn, t, claimer=f"{host}:worker") + kb._set_worker_pid(conn, t, 12345) + conn.execute( + "UPDATE tasks SET claim_expires = ?, last_heartbeat_at = ? " + "WHERE id = ?", + (int(time.time()) - 60, int(time.time()) - 7200, t), + ) + monkeypatch.setattr(_kb, "_pid_alive", lambda _pid: True) + monkeypatch.setattr( + _kb, "_terminate_reclaimed_worker", + lambda *a, **k: { + "termination_attempted": False, + "host_local": False, + "terminated": False, + }, + ) + reclaimed = kb.release_stale_claims(conn, signal_fn=lambda _p, _s: None) + assert reclaimed == 1 + assert kb.get_task(conn, t).status == "ready" + + +def test_detect_stale_defers_when_live_worker_survives(kanban_home, monkeypatch): + """detect_stale_running must also hold the claim when the worker survives.""" + import hermes_cli.kanban_db as _kb + + with kb.connect() as conn: + t = kb.create_task(conn, title="wedged", assignee="worker") + kb.claim_task(conn, t) + kb._set_worker_pid(conn, t, os.getpid()) + + five_hours_ago = int(time.time()) - (5 * 3600) + with kb.write_txn(conn): + conn.execute( + "UPDATE tasks SET started_at = ?, last_heartbeat_at = NULL " + "WHERE id = ?", + (five_hours_ago, t), + ) + conn.execute( + "UPDATE task_runs SET started_at = ? " + "WHERE id = (SELECT current_run_id FROM tasks WHERE id = ?)", + (five_hours_ago, t), + ) + + monkeypatch.setattr(_kb, "_pid_alive", lambda _pid: True) + monkeypatch.setattr( + _kb, "_terminate_reclaimed_worker", + lambda *a, **k: { + "termination_attempted": True, + "host_local": True, + "terminated": False, + }, + ) + stale = kb.detect_stale_running( + conn, stale_timeout_seconds=14400, signal_fn=lambda p, s: None, + ) + assert stale == [] + assert kb.get_task(conn, t).status == "running" + kinds = [ + r["kind"] for r in conn.execute( + "SELECT kind FROM task_events WHERE task_id = ?", (t,), + ).fetchall() + ] + assert "reclaim_deferred" in kinds + + def test_stale_claim_reclaim_event_records_diagnostic_payload( kanban_home, monkeypatch, ):