ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 9 Packages Projects Releases Wiki Activity

fix(kanban): hold reclaim while the worker is still alive

Browse Source 
release_stale_claims and detect_stale_running call _terminate_reclaimed_worker
and then release the task claim unconditionally, even when the termination did
not actually kill the worker. _terminate_reclaimed_worker already reports this
via its "terminated" flag, but the callers ignore it.

When a worker is parked in uninterruptible (D) state — for example throttled by
a cgroup memory.high limit — a pending SIGTERM/SIGKILL cannot be delivered until
the throttle lifts, so the kill is a no-op. The dispatcher then frees the claim
and spawns a fresh worker beside the still-alive one. Repeated every dispatch
tick this accumulates duplicate workers without bound, deepening the memory
pressure that caused the throttle in the first place — a self-reinforcing
runaway.

Fix: gate both automatic reclaim paths on _worker_survived_termination(). When
we attempted to kill our own host-local worker and it is still alive, defer the
reclaim (_defer_reclaim_for_live_worker extends the claim a short grace and
emits a reclaim_deferred event) instead of releasing. This guarantees at most
one live worker per task and is self-correcting: not spawning a duplicate is
what relieves the pressure so the pending signal lands and the worker dies, and
the next tick reclaims cleanly. Non-host-local claims and the operator-driven
reclaim_task() path keep their existing force-release behaviour.

Related: #41448 (concurrent dispatchers amplify this by doubling reclaim
frequency); #42858 (kill the worker rather than orphan it on archive).

Tests: defer-when-worker-survives, reclaim-when-killed,
release-when-not-host-local, and the detect_stale_running path.
This commit is contained in:
Sahil Saghir
2026-06-12 13:59:01 +01:00
committed by Teknium
parent 13d4b5fe2f
commit b9e521da23
2 changed files with 249 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_cli/kanban_db.py
+84
View File
@@ -121,6 +121,16 @@ DEFAULT_CLAIM_TTL_SECONDS = 15 * 60
# 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, "
@@ -5136,6 +5154,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 +5449,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, "
tests/hermes_cli/test_kanban_db.py
+165
View File
@@ -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,
):