Kanban worker heartbeat and reclaim lifecycle fixes

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #21183 fix(kanban): heartbeat + reclaim + zombie + retry-cap fixes (#21147, #21141, #21169, #20881)

  merged to main

  files: gateway/run.py, hermes_cli/config.py, hermes_cli/kanban.py, hermes_cli/kanban_db.py, scripts/release.py, tests/hermes_cli/test_kanban_core_functionality.py, tests/hermes_cli/test_kanban_db.py, tests/tools/test_kanban_tools.py, tools/kanban_tools.py

• #23442 fix(kanban): extend stale claim instead of killing live worker (salvage #23071)

  merged to main

  files: hermes_cli/kanban_db.py, scripts/release.py, tests/hermes_cli/test_kanban_db.py

  Show the diff

  • hermes_cli/kanban_db.py

    @@ -1997,16 +1997,69 @@ def release_stale_claims(
     ) -> int:
         """Reset any ``running`` task whose claim has expired.
     
    -    Returns the number of stale claims reclaimed.  Safe to call often.
    +    A stale-by-TTL claim whose host-local worker PID is still alive is
    +    *extended* (with a ``claim_extended`` event) instead of being
    +    reclaimed. Reclaiming a live worker mid-flight produces the spawn-
    +    then-immediately-reclaim loop seen on slow models that spend longer
    +    than ``DEFAULT_CLAIM_TTL_SECONDS`` inside a single tool-free LLM
    +    call (#23025): no tool calls means no ``kanban_heartbeat``, even
    +    though the subprocess is healthy. ``enforce_max_runtime`` and
    +    ``detect_crashed_workers`` remain the upper bounds for genuinely
    +    wedged or dead workers.
    +
    +    Returns the number of stale claims actually reclaimed (live-pid
    +    extensions don't count). Safe to call often.
         """
         now = int(time.time())
         reclaimed = 0
    +    host_prefix = f"{_claimer_id().split(':', 1)[0]}:"
         stale = conn.execute(
    -        "SELECT id, claim_lock, worker_pid FROM tasks "
    -        "WHERE status = 'running' AND claim_expires IS NOT NULL AND claim_expires < ?",
    +        "SELECT id, claim_lock, worker_pid, claim_expires, last_heartbeat_at "
    +        "FROM tasks "
    +        "WHERE status = 'running' AND claim_expires IS NOT NULL "
    +        "  AND claim_expires < ?",
             (now,),
         ).fetchall()
         for row in stale:
    +        lock = row["claim_lock"] or ""
    +        host_local = lock.startswith(host_prefix)
    +        if host_local and row["worker_pid"] and _pid_alive(row["worker_pid"]):
    +            new_expires = now + int(DEFAULT_CLAIM_TTL_SECONDS)
    +            with write_txn(conn):
    +                cur = conn.execute(
    +                    "UPDATE tasks SET claim_expires = ? "
    +                    "WHERE id = ? AND status = 'running' "
    +                    "  AND claim_lock IS ? "
    +                    "  AND claim_expires IS NOT NULL "
    +                    "  AND claim_expires < ?",
    +                    (new_expires, row["id"], row["claim_lock"], now),
    +                )
    +                if cur.rowcount != 1:
    +                    continue
    +                run_id = _current_run_id(conn, row["id"])
    +                if run_id is not None:
    +                    conn.execute(
    +                        "UPDATE task_runs SET claim_expires = ? WHERE id = ?",
    +                        (new_expires, run_id),
    +                    )
    +                _append_event(
    +                    conn, row["id"], "claim_extended",
    +                    {
    +                        "reason": "pid_alive",
    +                        "worker_pid": int(row["worker_pid"]),
    +                        "claim_lock": row["claim_lock"],
    +                        "claim_expires_was": int(row["claim_expires"]),
    +                        "claim_expires_now": new_expires,
    +                        "last_heartbeat_at": (
    +                            int(row["last_heartbeat_at"])
    +                            if row["last_heartbeat_at"] is not None
    +                            else None
    +                        ),
    +                    },
    +                    run_id=run_id,
    +                )
    +            continue
    +
             termination = _terminate_reclaimed_worker(
                 row["worker_pid"], row["claim_lock"], signal_fn=signal_fn,
             )
    @@ -2026,7 +2079,20 @@ def release_stale_claims(
                     error=f"stale_lock={row['claim_lock']}",
                     metadata=termination,
                 )
    -            payload = {"stale_lock": row["claim_lock"]}
    +            payload = {
    +                "stale_lock": row["claim_lock"],
    +                "worker_pid": (
    +                    int(row["worker_pid"])
    +                    if row["worker_pid"] is not None else None
    +                ),
    +                "claim_expires": int(row["claim_expires"]),
    +                "last_heartbeat_at": (
    +                    int(row["last_heartbeat_at"])
    +                    if row["last_heartbeat_at"] is not None else None
    +                ),
    +                "now": now,
    +                "host_local": host_local,
    +            }
                 payload.update(termination)
                 _append_event(
                     conn, row["id"], "reclaimed",

  • scripts/release.py

    @@ -722,6 +722,7 @@
         "[redacted:email]": "yehuosi",
         "[redacted:email]": "yehuosi",
         "[redacted:email]": "jelrod27",
    +    "[redacted:email]": "konsisumer",
         "[redacted:email]": "sicnuyudidi",
         "[redacted:email]": "haimu0x",
         "[redacted:email]": "NIDNASSER-Abdelmajid",

  • tests/hermes_cli/test_kanban_db.py

    @@ -177,12 +177,9 @@ def test_stale_claim_reclaimed(kanban_home, monkeypatch):
             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):
    +        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.
    @@ -190,13 +187,96 @@ def _signal(pid, sig):
                 "UPDATE tasks SET claim_expires = ? WHERE id = ?",
                 (int(time.time()) - 3600, t),
             )
    -        monkeypatch.setattr(_kb, "_pid_alive", lambda _pid: state["alive"])
    +        # Worker PID has died — exactly the case ``release_stale_claims``
    +        # should still reclaim (post-#23025: live PIDs are now extended).
    +        monkeypatch.setattr(_kb, "_pid_alive", lambda _pid: False)
             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_stale_claim_with_live_pid_extends_instead_of_reclaiming(
    +    kanban_home, monkeypatch,
    +):
    +    """A stale-by-TTL claim whose worker PID is still alive should be
    +    extended, not reclaimed (#23025). Slow models can spend longer than
    +    ``DEFAULT_CLAIM_TTL_SECONDS`` inside a single tool-free LLM call;
    +    killing those healthy workers produces a respawn loop with zero
    +    progress."""
    +    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
    +        conn.execute(
    +            "UPDATE tasks SET claim_expires = ? WHERE id = ?",
    +            (old_expires, t),
    +        )
    +
    +        monkeypatch.setattr(_kb, "_pid_alive", lambda _pid: True)
    +        killed: list[int] = []
    +        reclaimed = kb.release_stale_claims(
    +            conn, signal_fn=lambda _p, sig: killed.append(sig),
    +        )
    +        assert reclaimed == 0
    +        task = kb.get_task(conn, t)
    +        assert task.status == "running"
    +        assert task.claim_expires is not None
    +        assert task.claim_expires > old_expires
    +        assert killed == []  # live worker not killed
    +
    +        kinds = [
    +            r["kind"] for r in conn.execute(
    +                "SELECT kind FROM task_events WHERE task_id = ?", (t,),
    +            ).fetchall()
    +        ]
    +        assert "claim_extended" in kinds
    +        assert "reclaimed" not in kinds
    +
    +
    +def test_stale_claim_reclaim_event_records_diagnostic_payload(
    +    kanban_home, monkeypatch,
    +):
    +    """``reclaimed`` events should carry claim_expires, last_heartbeat_at,
    +    and worker_pid so operators can diagnose why a claim went stale
    +    (#23025: previous payload only had ``stale_lock`` which gives no
    +    timing context)."""
    +    import json
    +    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()) - 3600
    +        hb_at = int(time.time()) - 1800
    +        conn.execute(
    +            "UPDATE tasks SET claim_expires = ?, last_heartbeat_at = ? "
    +            "WHERE id = ?",
    +            (old_expires, hb_at, t),
    +        )
    +
    +        monkeypatch.setattr(_kb, "_pid_alive", lambda _pid: False)
    +        kb.release_stale_claims(conn, signal_fn=lambda _p, _s: None)
    +        row = conn.execute(
    +            "SELECT payload FROM task_events "
    +            "WHERE task_id = ? AND kind = 'reclaimed'",
    +            (t,),
    +        ).fetchone()
    +        assert row is not None
    +        payload = json.loads(row["payload"])
    +        assert payload["claim_expires"] == old_expires
    +        assert payload["last_heartbeat_at"] == hb_at
    +        assert payload["worker_pid"] == 12345
    +        assert payload["host_local"] is True
    +
    +
     def test_max_runtime_uses_current_run_start_after_retry(kanban_home):
         """A retry should get a fresh max-runtime window.
     

• #28392 fix(kanban): make claim TTL configurable (#25635)

  merged to main

  files: hermes_cli/kanban_db.py, tests/hermes_cli/test_kanban_db.py

• #34418 fix(kanban): bridge worker runtime activity to board heartbeat (#31752)

  merged to main

  files: run_agent.py, scripts/release.py, tools/kanban_tools.py

  Show the diff

  • run_agent.py

    @@ -2144,9 +2144,26 @@ def _apply_pending_steer_to_tool_results(self, messages: list, num_tool_msgs: in
             return apply_pending_steer_to_tool_results(self, messages, num_tool_msgs)
     
         def _touch_activity(self, desc: str) -> None:
    -        """Update the last-activity timestamp and description (thread-safe)."""
    +        """Update the last-activity timestamp and description (thread-safe).
    +
    +        Also bridges to the kanban board's heartbeat fields when this
    +        process is a dispatcher-spawned worker (HERMES_KANBAN_TASK set),
    +        so the dispatcher watchdog doesn't reclaim an actively-running
    +        worker as stale (#31752). Bridge is rate-limited (60s) and
    +        best-effort — it never raises into the agent loop.
    +        """
             self._last_activity_ts = time.time()
             self._last_activity_desc = desc
    +        if os.environ.get("HERMES_KANBAN_TASK"):
    +            try:
    +                from tools.kanban_tools import heartbeat_current_worker_from_env
    +                heartbeat_current_worker_from_env()
    +            except Exception:
    +                # Never let the bridge break the agent loop.  The function
    +                # already swallows exceptions internally; this outer guard
    +                # covers import-time failures (kanban_tools unavailable,
    +                # etc.) on niche deployment surfaces.
    +                pass
     
         def _capture_rate_limits(self, http_response: Any) -> None:
             """Parse x-ratelimit-* headers from an HTTP response and cache the state.

  • scripts/release.py

    @@ -61,6 +61,7 @@
         "[redacted:email]": "victorGPT",
         "[redacted:email]": "julio-cloudvisor",
         "[redacted:email]": "kweiner",
    +    "[redacted:email]": "faisfamilytravel",
         # teknium (multiple emails)
         "[redacted:email]": "teknium1",
         "[redacted:email]": "kenyonxu",

  • tools/kanban_tools.py

    @@ -176,6 +176,90 @@ def _connect(board: Optional[str] = None):
         return kb, kb.connect(board=board)
     
     
    +# ---------------------------------------------------------------------------
    +# Runtime-activity → board-heartbeat bridge (#31752)
    +# ---------------------------------------------------------------------------
    +# When the agent ticks ``_touch_activity`` during normal work (between
    +# tool calls, mid-stream chunks, etc.), we want the kanban board's
    +# ``last_heartbeat_at`` columns to reflect that liveness so the dispatcher
    +# watchdog (which reads ``tasks.last_heartbeat_at``, not the agent's
    +# in-process timestamp) doesn't reclaim an actively-running worker as
    +# stale. The model is not required to call the explicit ``kanban_heartbeat``
    +# tool for this to work — that tool stays available for workers that want
    +# to attach a note or pre-emptively extend a claim across a known-long op.
    +#
    +# Constraints:
    +#   - Best-effort: never raise. The agent loop must not care if the bridge
    +#     fails (board missing, DB locked, etc.).
    +#   - Rate-limited to one DB write per 60s per-process; runtime activity
    +#     can tick on every chunk/tool result and we don't need that resolution.
    +#   - No-op outside dispatcher-spawned worker context (no ``HERMES_KANBAN_TASK``).
    +#   - No durable note on these auto-heartbeats; that's reserved for the
    +#     explicit tool which carries a model-supplied note.
    +
    +_AUTO_HEARTBEAT_MIN_INTERVAL_SECONDS = 60.0
    +_auto_heartbeat_last_attempt: float = 0.0
    +
    +
    +def heartbeat_current_worker_from_env() -> bool:
    +    """Best-effort: extend the kanban claim + bump board heartbeat for the
    +    current dispatcher-spawned worker, using identity from env vars.
    +
    +    Returns True if a write was attempted (whether or not it succeeded);
    +    False if the call was skipped (not a kanban worker, rate-limited, or
    +    swallowed exception). The boolean is informational — callers should
    +    not branch on it.
    +
    +    Identity comes from:
    +      * ``HERMES_KANBAN_TASK`` — task id (required; absence means no-op)
    +      * ``HERMES_KANBAN_RUN_ID`` — pins the run row so we don't heartbeat
    +        a stale run that may have already been reclaimed
    +      * ``HERMES_KANBAN_CLAIM_LOCK`` — claim lock for ``heartbeat_claim``;
    +        falls back to the default ``_claimer_id()`` for locally-driven
    +        workers that never went through the dispatcher path
    +
    +    Rate-limited via the module-level ``_auto_heartbeat_last_attempt``
    +    timestamp (monotonic clock); not thread-safe in the strict sense, but
    +    the worst case is one extra DB write per race, which is harmless.
    +    """
    +    global _auto_heartbeat_last_attempt
    +    tid = os.environ.get("HERMES_KANBAN_TASK")
    +    if not tid:
    +        return False
    +    import time as _time
    +    now = _time.monotonic()
    +    if (now - _auto_heartbeat_last_attempt) < _AUTO_HEARTBEAT_MIN_INTERVAL_SECONDS:
    +        return False
    +    _auto_heartbeat_last_attempt = now
    +    try:
    +        kb, conn = _connect()
    +        try:
    +            claim_lock = os.environ.get("HERMES_KANBAN_CLAIM_LOCK")
    +            try:
    +                kb.heartbeat_claim(conn, tid, claimer=claim_lock)
    +            except Exception:
    +                logger.debug("auto-heartbeat: heartbeat_claim failed", exc_info=True)
    +            run_id_raw = os.environ.get("HERMES_KANBAN_RUN_ID")
    +            run_id: Optional[int]
    +            try:
    +                run_id = int(run_id_raw) if run_id_raw else None
    +            except (TypeError, ValueError):
    +                run_id = None
    +            try:
    +                kb.heartbeat_worker(conn, tid, note=None, expected_run_id=run_id)
    +            except Exception:
    +                logger.debug("auto-heartbeat: heartbeat_worker failed", exc_info=True)
    +        finally:
    +            try:
    +                conn.close()
    +            except Exception:
    +                pass
    +        return True
    +    except Exception:
    +        logger.debug("auto-heartbeat: bridge failed", exc_info=True)
    +        return False
    +
    +
     def _ok(**fields: Any) -> str:
         return json.dumps({"ok": True, **fields})