diff --git a/gateway/run.py b/gateway/run.py index 15ce3ab08ce00..42c1f804b8645 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -3645,6 +3645,24 @@ async def _kanban_dispatcher_watcher(self) -> None: if max_spawn is not None: logger.info(f"kanban dispatcher: max_spawn={max_spawn}") + raw_failure_limit = kanban_cfg.get("failure_limit", _kb.DEFAULT_FAILURE_LIMIT) + try: + failure_limit = int(raw_failure_limit) + except (TypeError, ValueError): + logger.warning( + "kanban dispatcher: invalid kanban.failure_limit=%r; using default %d", + raw_failure_limit, + _kb.DEFAULT_FAILURE_LIMIT, + ) + failure_limit = _kb.DEFAULT_FAILURE_LIMIT + if failure_limit < 1: + logger.warning( + "kanban dispatcher: kanban.failure_limit=%r is below 1; using default %d", + raw_failure_limit, + _kb.DEFAULT_FAILURE_LIMIT, + ) + failure_limit = _kb.DEFAULT_FAILURE_LIMIT + # Initial delay so the gateway finishes wiring adapters before the # dispatcher spawns workers (those workers may hit gateway notify # subscriptions etc.). Matches the notifier watcher's delay. @@ -3673,7 +3691,12 @@ def _tick_once_for_board(slug: str) -> "Optional[object]": _kb.init_db(board=slug) # idempotent, handles first-run except Exception: pass - return _kb.dispatch_once(conn, board=slug, max_spawn=max_spawn) + return _kb.dispatch_once( + conn, + board=slug, + max_spawn=max_spawn, + failure_limit=failure_limit, + ) except Exception: logger.exception("kanban dispatcher: tick failed on board %s", slug) return None diff --git a/hermes_cli/config.py b/hermes_cli/config.py index cf2b0b528a6ab..baf73c2ea5508 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -1230,6 +1230,10 @@ def _ensure_hermes_home_managed(home: Path): # Seconds between dispatcher ticks (idle or not). Lower = snappier # pickup of newly-ready tasks; higher = less SQL pressure. "dispatch_interval_seconds": 60, + # Auto-block after this many consecutive non-success attempts for the + # same task/profile (spawn_failed, timed_out, or crashed). Reassignment + # resets the streak for the new profile. + "failure_limit": 2, }, # execute_code settings — controls the tool used for programmatic tool calls. diff --git a/hermes_cli/kanban.py b/hermes_cli/kanban.py index d8bc47a7d7b9a..7301e58b66dfe 100644 --- a/hermes_cli/kanban.py +++ b/hermes_cli/kanban.py @@ -443,8 +443,8 @@ def build_parser(parent_subparsers: argparse._SubParsersAction) -> argparse.Argu help="Cap number of spawns this pass") p_disp.add_argument("--failure-limit", type=int, default=kb.DEFAULT_SPAWN_FAILURE_LIMIT, - help=f"Auto-block a task after this many consecutive spawn failures " - f"(default: {kb.DEFAULT_SPAWN_FAILURE_LIMIT})") + help=f"Auto-block a task after this many consecutive non-success attempts " + f"(spawn_failed, timed_out, or crashed; default: {kb.DEFAULT_SPAWN_FAILURE_LIMIT})") p_disp.add_argument("--json", action="store_true") # --- daemon (deprecated) --- @@ -1657,6 +1657,7 @@ def _cmd_daemon(args: argparse.Namespace) -> int: " kanban:\n" " dispatch_in_gateway: true # default\n" " dispatch_interval_seconds: 60\n" + " failure_limit: 2 # consecutive non-success attempts before auto-block\n" "\n" "Running both the gateway AND this standalone daemon will\n" "race for claims. If you truly need the old standalone\n" diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index 2d2f1b2ecf84f..94968dd87c713 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -1380,7 +1380,7 @@ def assign_task(conn: sqlite3.Connection, task_id: str, profile: Optional[str]) profile = _canonical_assignee(profile) with write_txn(conn): row = conn.execute( - "SELECT status, claim_lock FROM tasks WHERE id = ?", (task_id,) + "SELECT status, claim_lock, assignee FROM tasks WHERE id = ?", (task_id,) ).fetchone() if not row: return False @@ -1389,7 +1389,17 @@ def assign_task(conn: sqlite3.Connection, task_id: str, profile: Optional[str]) f"cannot reassign {task_id}: currently running (claimed). " "Wait for completion or reclaim the stale lock first." ) - conn.execute("UPDATE tasks SET assignee = ? WHERE id = ?", (profile, task_id)) + if row["assignee"] != profile: + # The retry guard is scoped to the task/profile combination. A + # human reassigning the task is an explicit recovery action, so the + # new profile should not inherit the previous profile's streak. + conn.execute( + "UPDATE tasks SET assignee = ?, consecutive_failures = 0, " + "last_failure_error = NULL WHERE id = ?", + (profile, task_id), + ) + else: + conn.execute("UPDATE tasks SET assignee = ? WHERE id = ?", (profile, task_id)) _append_event(conn, task_id, "assigned", {"assignee": profile}) return True @@ -1859,34 +1869,47 @@ def heartbeat_claim( return False -def release_stale_claims(conn: sqlite3.Connection) -> int: +def release_stale_claims( + conn: sqlite3.Connection, + *, + signal_fn=None, +) -> int: """Reset any ``running`` task whose claim has expired. Returns the number of stale claims reclaimed. Safe to call often. """ now = int(time.time()) reclaimed = 0 - with write_txn(conn): - stale = conn.execute( - "SELECT id, claim_lock FROM tasks " - "WHERE status = 'running' AND claim_expires IS NOT NULL AND claim_expires < ?", - (now,), - ).fetchall() - for row in stale: - conn.execute( + stale = conn.execute( + "SELECT id, claim_lock, worker_pid FROM tasks " + "WHERE status = 'running' AND claim_expires IS NOT NULL AND claim_expires < ?", + (now,), + ).fetchall() + for row in stale: + termination = _terminate_reclaimed_worker( + row["worker_pid"], row["claim_lock"], signal_fn=signal_fn, + ) + with write_txn(conn): + cur = conn.execute( "UPDATE tasks SET status = 'ready', claim_lock = NULL, " "claim_expires = NULL, worker_pid = NULL " - "WHERE id = ? AND status = 'running'", - (row["id"],), + "WHERE id = ? AND status = 'running' AND claim_lock IS ? " + "AND claim_expires IS NOT NULL AND claim_expires < ?", + (row["id"], row["claim_lock"], now), ) + if cur.rowcount != 1: + continue run_id = _end_run( conn, row["id"], outcome="reclaimed", status="reclaimed", error=f"stale_lock={row['claim_lock']}", + metadata=termination, ) + payload = {"stale_lock": row["claim_lock"]} + payload.update(termination) _append_event( conn, row["id"], "reclaimed", - {"stale_lock": row["claim_lock"]}, + payload, run_id=run_id, ) reclaimed += 1 @@ -1898,6 +1921,7 @@ def reclaim_task( task_id: str, *, reason: Optional[str] = None, + signal_fn=None, ) -> bool: """Operator-driven reclaim: release the claim and reset to ``ready``. @@ -1910,24 +1934,29 @@ def reclaim_task( Returns True if a reclaim happened, False if the task isn't in a reclaimable state (not running, or doesn't exist). """ + row = conn.execute( + "SELECT status, claim_lock, worker_pid FROM tasks WHERE id = ?", + (task_id,), + ).fetchone() + if not row: + return False + if row["status"] != "running" and row["claim_lock"] is None: + # Nothing to reclaim — already ready / blocked / done. + return False + prev_lock = row["claim_lock"] + termination = _terminate_reclaimed_worker( + row["worker_pid"], prev_lock, signal_fn=signal_fn, + ) with write_txn(conn): - row = conn.execute( - "SELECT status, claim_lock, worker_pid FROM tasks WHERE id = ?", - (task_id,), - ).fetchone() - if not row: - return False - if row["status"] != "running" and row["claim_lock"] is None: - # Nothing to reclaim — already ready / blocked / done. - return False - prev_lock = row["claim_lock"] - prev_pid = row["worker_pid"] - conn.execute( + cur = conn.execute( "UPDATE tasks SET status = 'ready', claim_lock = NULL, " "claim_expires = NULL, worker_pid = NULL " - "WHERE id = ? AND status IN ('running', 'ready', 'blocked')", - (task_id,), + "WHERE id = ? AND status IN ('running', 'ready', 'blocked') " + "AND claim_lock IS ?", + (task_id, prev_lock), ) + if cur.rowcount != 1: + return False run_id = _end_run( conn, task_id, outcome="reclaimed", status="reclaimed", @@ -1935,15 +1964,17 @@ def reclaim_task( f"manual_reclaim: {reason}" if reason else f"manual_reclaim lock={prev_lock}" ), + metadata=termination, ) + payload = { + "manual": True, + "reason": reason, + "prev_lock": prev_lock, + } + payload.update(termination) _append_event( conn, task_id, "reclaimed", - { - "manual": True, - "reason": reason, - "prev_lock": prev_lock, - "prev_pid": prev_pid, - }, + payload, run_id=run_id, ) # Operator intervention — they've looked at the task, so the @@ -2548,11 +2579,11 @@ def set_workspace_path( # Dispatcher (one-shot pass) # --------------------------------------------------------------------------- -# After this many consecutive `spawn_failed` events on a task, the dispatcher -# stops retrying and parks the task in ``blocked`` with a reason so a human -# can investigate. Prevents the dispatcher from thrashing forever on a task -# whose profile doesn't exist, whose workspace is unmountable, etc. -DEFAULT_FAILURE_LIMIT = 5 +# After this many consecutive non-success attempts on a task/profile, the +# dispatcher stops retrying and parks the task in ``blocked`` with a reason so +# a human can investigate. Prevents retry storms when a worker repeatedly times +# out, crashes, or cannot spawn. +DEFAULT_FAILURE_LIMIT = 2 # Legacy alias — callers / tests still reference the old name. DEFAULT_SPAWN_FAILURE_LIMIT = DEFAULT_FAILURE_LIMIT @@ -2652,6 +2683,59 @@ def _pid_alive(pid: Optional[int]) -> bool: return True +def _terminate_reclaimed_worker( + pid: Optional[int], + claim_lock: Optional[str], + *, + signal_fn=None, +) -> dict[str, Any]: + """Best-effort host-local worker termination for reclaim paths.""" + import signal + + info: dict[str, Any] = { + "prev_pid": int(pid) if pid else None, + "host_local": False, + "termination_attempted": False, + "terminated": False, + "sigkill": False, + } + if not pid or pid <= 0 or not claim_lock: + return info + + host_prefix = f"{_claimer_id().split(':', 1)[0]}:" + if not str(claim_lock).startswith(host_prefix): + return info + info["host_local"] = True + + kill = signal_fn if signal_fn is not None else ( + os.kill if hasattr(os, "kill") else None + ) + if kill is None: + return info + + info["termination_attempted"] = True + try: + kill(int(pid), signal.SIGTERM) + except (ProcessLookupError, OSError): + return info + + for _ in range(10): + if not _pid_alive(pid): + info["terminated"] = True + return info + time.sleep(0.5) + + if _pid_alive(pid): + try: + kill(int(pid), signal.SIGKILL) + info["sigkill"] = True + except (ProcessLookupError, OSError): + return info + + info["terminated"] = not _pid_alive(pid) + return info + + def heartbeat_worker( conn: sqlite3.Connection, task_id: str, @@ -3150,6 +3234,25 @@ def dispatch_once( ``board`` pins workspace/log/db resolution for this tick to a specific board. When omitted, the current-board resolution chain is used. """ + # Reap zombie children from previously spawned workers. + # The gateway-embedded dispatcher is the parent of every worker spawned + # via _default_spawn (start_new_session=True only detaches the + # controlling tty, not the parent). Without an explicit waitpid, each + # completed worker becomes a entry that lingers until gateway + # exit. WNOHANG keeps this non-blocking; ChildProcessError means no + # children to reap. Bounded: at most one tick's worth of completions + # can be in at once. + try: + while True: + try: + _pid, _status = os.waitpid(-1, os.WNOHANG) + except ChildProcessError: + break + if _pid == 0: + break + except Exception: + pass + result = DispatchResult() result.reclaimed = release_stale_claims(conn) result.crashed = detect_crashed_workers(conn) diff --git a/scripts/release.py b/scripts/release.py index 8249484e446f5..2c683022847c2 100755 --- a/scripts/release.py +++ b/scripts/release.py @@ -68,6 +68,9 @@ "wysie@users.noreply.github.com": "wysie", "jkausel@gmail.com": "jkausel-ai", "e.silacandmr@gmail.com": "Es1la", + "51599529+stephen0110@users.noreply.github.com": "stephen0110", + "265632032+sonic-netizen@users.noreply.github.com": "sonic-netizen", + "82531659+mwnickerson@users.noreply.github.com": "mwnickerson", "154585401+LeonSGP43@users.noreply.github.com": "LeonSGP43", "zjtan1@gmail.com": "zeejaytan", "asslaenn5@gmail.com": "Aslaaen", @@ -439,6 +442,9 @@ "xowiekk@gmail.com": "Xowiek", "1243352777@qq.com": "zons-zhaozhy", "e.silacandmr@gmail.com": "Es1la", + "51599529+stephen0110@users.noreply.github.com": "stephen0110", + "265632032+sonic-netizen@users.noreply.github.com": "sonic-netizen", + "82531659+mwnickerson@users.noreply.github.com": "mwnickerson", "h3057183414@gmail.com": "CoreyNoDream", "franksong2702@gmail.com": "franksong2702", "673088860@qq.com": "ambition0802", diff --git a/tests/hermes_cli/test_kanban_core_functionality.py b/tests/hermes_cli/test_kanban_core_functionality.py index 1e286d7ce6463..a6d65f6072d0e 100644 --- a/tests/hermes_cli/test_kanban_core_functionality.py +++ b/tests/hermes_cli/test_kanban_core_functionality.py @@ -90,22 +90,20 @@ def _bad_spawn(task, ws): conn = kb.connect() try: tid = kb.create_task(conn, title="x", assignee="worker") - # Three ticks below the default limit (5) → still ready, counter grows. - for i in range(3): - res = kb.dispatch_once(conn, spawn_fn=_bad_spawn, failure_limit=5) - assert tid not in res.auto_blocked + assert kb.DEFAULT_FAILURE_LIMIT == 2 + # One default-limit failure → still ready, counter grows. + res1 = kb.dispatch_once(conn, spawn_fn=_bad_spawn) + assert tid not in res1.auto_blocked task = kb.get_task(conn, tid) assert task.status == "ready" - assert task.consecutive_failures == 3 + assert task.consecutive_failures == 1 - # Two more ticks → fifth failure exceeds the limit. - res1 = kb.dispatch_once(conn, spawn_fn=_bad_spawn, failure_limit=5) - assert tid not in res1.auto_blocked - res2 = kb.dispatch_once(conn, spawn_fn=_bad_spawn, failure_limit=5) + # Second default-limit failure trips the guard. + res2 = kb.dispatch_once(conn, spawn_fn=_bad_spawn) assert tid in res2.auto_blocked task = kb.get_task(conn, tid) assert task.status == "blocked" - assert task.consecutive_failures >= 5 + assert task.consecutive_failures >= 2 assert task.last_failure_error and "no PATH" in task.last_failure_error finally: conn.close() @@ -170,6 +168,27 @@ def test_successful_completion_resets_failure_counter(kanban_home, all_assignees conn.close() +def test_reassign_resets_failure_counter_for_new_profile(kanban_home, all_assignees_spawnable): + """Retry streaks are scoped to a task/profile pair; reassigning is a + human recovery action and gives the new profile a fresh budget.""" + conn = kb.connect() + try: + tid = kb.create_task(conn, title="x", assignee="worker") + with kb.write_txn(conn): + conn.execute( + "UPDATE tasks SET consecutive_failures = 1, " + "last_failure_error = 'timed out' WHERE id = ?", + (tid,), + ) + assert kb.assign_task(conn, tid, "reviewer") is True + task = kb.get_task(conn, tid) + assert task.assignee == "reviewer" + assert task.consecutive_failures == 0 + assert task.last_failure_error is None + finally: + conn.close() + + def test_workspace_resolution_failure_also_counts(kanban_home, all_assignees_spawnable): """`dir:` workspace with no path should fail workspace resolution AND count against the failure budget — not just crash the tick.""" @@ -719,6 +738,48 @@ def _signal_fn(pid, sig): _kb._pid_alive = original_alive +def test_repeated_timeouts_auto_block_at_default_limit(kanban_home): + """Two timed_out outcomes on the same task/profile trip the retry guard.""" + import hermes_cli.kanban_db as _kb + original_alive = _kb._pid_alive + _kb._pid_alive = lambda pid: False + + def _age_active_run(conn, tid): + old_started = int(time.time()) - 30 + with kb.write_txn(conn): + conn.execute( + "UPDATE task_runs SET started_at = ? " + "WHERE id = (SELECT current_run_id FROM tasks WHERE id = ?)", + (old_started, tid), + ) + + try: + conn = kb.connect() + try: + tid = kb.create_task( + conn, title="long job", assignee="worker", + max_runtime_seconds=1, + ) + for expected_failures in (1, 2): + kb.claim_task(conn, tid) + kb._set_worker_pid(conn, tid, os.getpid()) + _age_active_run(conn, tid) + timed_out = kb.enforce_max_runtime(conn, signal_fn=lambda pid, sig: None) + assert tid in timed_out + task = kb.get_task(conn, tid) + assert task.consecutive_failures == expected_failures + task = kb.get_task(conn, tid) + assert task.status == "blocked" + events = kb.list_events(conn, tid) + assert [e.kind for e in events].count("timed_out") == 2 + gave_up = [e for e in events if e.kind == "gave_up"] + assert gave_up and gave_up[-1].payload["trigger_outcome"] == "timed_out" + finally: + conn.close() + finally: + _kb._pid_alive = original_alive + + def test_max_runtime_none_means_no_cap(kanban_home): """A task with max_runtime_seconds=None is never timed out regardless of how long it runs.""" @@ -3283,17 +3344,28 @@ def test_complete_prose_scan_ignores_existing_ids(kanban_home): # Recovery helpers (reclaim + reassign) # --------------------------------------------------------------------------- -def test_reclaim_task_resets_running_to_ready(kanban_home): +def test_reclaim_task_resets_running_to_ready(kanban_home, monkeypatch): """Manual reclaim releases the claim, resets status, and emits a ``reclaimed`` event even when claim_expires has not passed.""" + import signal import time import secrets + import hermes_cli.kanban_db as _kb conn = kb.connect() try: t = kb.create_task(conn, title="stuck", assignee="broken") # Simulate a live claim (not expired). - lock = secrets.token_hex(8) + lock = f"{_kb._claimer_id().split(':', 1)[0]}:{secrets.token_hex(8)}" future = int(time.time()) + 3600 + killed: list[int] = [] + state = {"alive": True} + + def _signal(pid, sig): + killed.append(sig) + if sig == signal.SIGTERM: + state["alive"] = False + + monkeypatch.setattr(_kb, "_pid_alive", lambda _pid: state["alive"]) conn.execute( "UPDATE tasks SET status='running', claim_lock=?, claim_expires=?, " "worker_pid=? WHERE id=?", @@ -3312,7 +3384,7 @@ def test_reclaim_task_resets_running_to_ready(kanban_home): assert kb.release_stale_claims(conn) == 0 # reclaim_task should work immediately. - assert kb.reclaim_task(conn, t, reason="test reason") is True + assert kb.reclaim_task(conn, t, reason="test reason", signal_fn=_signal) is True row = conn.execute( "SELECT status, claim_lock, worker_pid FROM tasks WHERE id=?", @@ -3333,6 +3405,9 @@ def test_reclaim_task_resets_running_to_ready(kanban_home): assert len(reclaim_evs) == 1 assert reclaim_evs[0].get("manual") is True assert reclaim_evs[0].get("reason") == "test reason" + assert reclaim_evs[0].get("termination_attempted") is True + assert reclaim_evs[0].get("terminated") is True + assert killed == [signal.SIGTERM] finally: conn.close() diff --git a/tests/hermes_cli/test_kanban_db.py b/tests/hermes_cli/test_kanban_db.py index 7068e773d1b04..2375d6c4bc441 100644 --- a/tests/hermes_cli/test_kanban_db.py +++ b/tests/hermes_cli/test_kanban_db.py @@ -168,18 +168,33 @@ def test_claim_fails_on_non_ready(kanban_home): assert kb.claim_task(conn, t) is None -def test_stale_claim_reclaimed(kanban_home): +def test_stale_claim_reclaimed(kanban_home, monkeypatch): + import signal + import hermes_cli.kanban_db as _kb + with kb.connect() as conn: t = kb.create_task(conn, title="x", assignee="a") - kb.claim_task(conn, t) + host = _kb._claimer_id().split(":", 1)[0] + kb.claim_task(conn, t, claimer=f"{host}:worker") + killed: list[int] = [] + state = {"alive": True} + + def _signal(pid, sig): + killed.append(sig) + if sig == signal.SIGTERM: + state["alive"] = False + + kb._set_worker_pid(conn, t, 12345) # Rewind claim_expires so it looks stale. conn.execute( "UPDATE tasks SET claim_expires = ? WHERE id = ?", (int(time.time()) - 3600, t), ) - reclaimed = kb.release_stale_claims(conn) + monkeypatch.setattr(_kb, "_pid_alive", lambda _pid: state["alive"]) + reclaimed = kb.release_stale_claims(conn, signal_fn=_signal) assert reclaimed == 1 assert kb.get_task(conn, t).status == "ready" + assert killed == [signal.SIGTERM] def test_max_runtime_uses_current_run_start_after_retry(kanban_home): diff --git a/tests/tools/test_kanban_tools.py b/tests/tools/test_kanban_tools.py index f00a33d544b8e..aa7168da6cb1d 100644 --- a/tests/tools/test_kanban_tools.py +++ b/tests/tools/test_kanban_tools.py @@ -214,6 +214,61 @@ def test_heartbeat_without_note(worker_env): assert d["ok"] is True +def test_heartbeat_extends_claim_expires(worker_env): + """The kanban_heartbeat tool MUST extend claim_expires, not just + update last_heartbeat_at — otherwise long-running workers loop the + heartbeat tool diligently and still get reclaimed by + release_stale_claims at DEFAULT_CLAIM_TTL_SECONDS. + + Regression test for the bug where _handle_heartbeat called + heartbeat_worker but never heartbeat_claim, so claim_expires sat + static while last_heartbeat_at advanced. + """ + import time as _time + from hermes_cli import kanban_db as kb + from tools import kanban_tools as kt + + # Rewind claim_expires into the past so any forward movement is + # unambiguous (avoids time.sleep flakiness). + conn = kb.connect() + try: + conn.execute( + "UPDATE tasks SET claim_expires = ? WHERE id = ?", + (1, worker_env), + ) + conn.commit() + before = conn.execute( + "SELECT claim_expires FROM tasks WHERE id = ?", (worker_env,) + ).fetchone()["claim_expires"] + finally: + conn.close() + assert before == 1 + + out = kt._handle_heartbeat({"note": "still alive"}) + assert json.loads(out).get("ok") is True + + conn = kb.connect() + try: + after = conn.execute( + "SELECT claim_expires FROM tasks WHERE id = ?", (worker_env,) + ).fetchone()["claim_expires"] + finally: + conn.close() + + now = int(_time.time()) + # claim_expires should be roughly now + DEFAULT_CLAIM_TTL_SECONDS. + # We assert a generous floor (now + half the default TTL) to keep the + # test stable against future TTL changes. + assert after > before, ( + f"claim_expires did not advance ({before} -> {after}); workers " + f"would be reclaimed at TTL despite heartbeating" + ) + assert after >= now + (kb.DEFAULT_CLAIM_TTL_SECONDS // 2), ( + f"claim_expires={after} is suspiciously close to now={now}; " + f"expected at least now + {kb.DEFAULT_CLAIM_TTL_SECONDS // 2}" + ) + + def test_comment_happy_path(worker_env): from tools import kanban_tools as kt out = kt._handle_comment({ diff --git a/tools/kanban_tools.py b/tools/kanban_tools.py index 2f40b3f0de1f1..2326895554fe7 100644 --- a/tools/kanban_tools.py +++ b/tools/kanban_tools.py @@ -315,7 +315,15 @@ def _handle_block(args: dict, **kw) -> str: def _handle_heartbeat(args: dict, **kw) -> str: - """Signal that the worker is still alive during a long operation.""" + """Signal that the worker is still alive during a long operation. + + Extends the claim TTL via ``heartbeat_claim`` AND records a heartbeat + event via ``heartbeat_worker``. Without the ``heartbeat_claim`` half, + a diligent worker that loops this tool while a single tool call + blocks the agent for >DEFAULT_CLAIM_TTL_SECONDS still gets reclaimed + by ``release_stale_claims`` — which is exactly the trap that + ``heartbeat_claim``'s docstring warns against. + """ tid = _default_task_id(args.get("task_id")) if not tid: return tool_error( @@ -328,6 +336,14 @@ def _handle_heartbeat(args: dict, **kw) -> str: try: kb, conn = _connect() try: + # Extend the claim TTL first. The dispatcher pins + # HERMES_KANBAN_CLAIM_LOCK in the worker env at spawn time + # (see _default_spawn in kanban_db.py); falling back to the + # default _claimer_id() covers locally-driven workers that + # never went through the dispatcher path. + claim_lock = os.environ.get("HERMES_KANBAN_CLAIM_LOCK") + kb.heartbeat_claim(conn, tid, claimer=claim_lock) + ok = kb.heartbeat_worker( conn, tid,