fix(hermes): fence direct CLI heartbeat reclaim

dockerfiles/hermes_run_safety_regression.py

@@ -342,6 +342,153 @@ class ExactRunSafetyTests(ExecutionSafetyTestCase):
         self.assertEqual(task.status, "ready")
         self.assertIsNone(task.current_run_id)
 
+    def test_watchdog_reclaim_atomically_rechecks_current_run_liveness(self) -> None:
+        task_id = kanban_db.create_task(self.connection, title="heartbeat race")
+        self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
+        run_id = kanban_db.get_task(self.connection, task_id).current_run_id
+        now = int(kanban_db.time.time())
+
+        self.connection.execute(
+            "UPDATE task_runs SET last_heartbeat_at = ? WHERE id = ?",
+            (now, run_id),
+        )
+        self.assertFalse(
+            kanban_db.reclaim_task(
+                self.connection,
+                task_id,
+                reason="stale scan raced a heartbeat",
+                expected_run_id=run_id,
+                expected_run_liveness_before=now,
+            )
+        )
+        self.assertEqual(kanban_db.get_task(self.connection, task_id).status, "running")
+
+        self.connection.execute(
+            "UPDATE task_runs SET last_heartbeat_at = ? WHERE id = ?",
+            (now - 601, run_id),
+        )
+        self.assertTrue(
+            kanban_db.reclaim_task(
+                self.connection,
+                task_id,
+                reason="heartbeat remained dead",
+                expected_run_id=run_id,
+                expected_run_liveness_before=now - 600,
+            )
+        )
+        task = kanban_db.get_task(self.connection, task_id)
+        self.assertEqual(task.status, "ready")
+        self.assertEqual(task.consecutive_failures, 0)
+        run = kanban_db.get_run(self.connection, run_id)
+        self.assertEqual(run.outcome, "reclaimed")
+
+    def test_watchdog_reclaim_preserves_accumulated_failure_budget(self) -> None:
+        task_id = kanban_db.create_task(self.connection, title="repeated worker death")
+        self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
+        run_id = kanban_db.get_task(self.connection, task_id).current_run_id
+        now = int(kanban_db.time.time())
+        with kanban_db.write_txn(self.connection):
+            self.connection.execute(
+                "UPDATE tasks SET consecutive_failures = 2, "
+                "last_failure_error = ? WHERE id = ?",
+                ("worker repeatedly disappeared", task_id),
+            )
+            self.connection.execute(
+                "UPDATE task_runs SET last_heartbeat_at = ? WHERE id = ?",
+                (now - 601, run_id),
+            )
+
+        self.assertTrue(
+            kanban_db.reclaim_task(
+                self.connection,
+                task_id,
+                reason="heartbeat remained dead",
+                expected_run_id=run_id,
+                expected_run_liveness_before=now - 600,
+                preserve_failure_counter=True,
+            )
+        )
+        task = kanban_db.get_task(self.connection, task_id)
+        self.assertEqual(task.status, "ready")
+        self.assertEqual(task.consecutive_failures, 2)
+        self.assertEqual(task.last_failure_error, "worker repeatedly disappeared")
+
+    def test_ttl_reclaimer_never_competes_for_direct_lane_claims(self) -> None:
+        task_id = kanban_db.create_task(self.connection, title="durable lane result")
+        self.assertIsNotNone(
+            kanban_db.claim_task(
+                self.connection,
+                task_id,
+                ttl_seconds=60,
+                claimer="direct-cli-lane",
+            )
+        )
+        run_id = kanban_db.get_task(self.connection, task_id).current_run_id
+        expired = int(kanban_db.time.time()) - 1
+        with kanban_db.write_txn(self.connection):
+            self.connection.execute(
+                "UPDATE tasks SET claim_expires = ? WHERE id = ?",
+                (expired, task_id),
+            )
+            self.connection.execute(
+                "UPDATE task_runs SET claim_expires = ? WHERE id = ?",
+                (expired, run_id),
+            )
+
+        self.assertEqual(kanban_db.release_stale_claims(self.connection), 0)
+        task = kanban_db.get_task(self.connection, task_id)
+        self.assertEqual(task.status, "running")
+        self.assertEqual(task.current_run_id, run_id)
+        self.assertEqual(task.claim_lock, "direct-cli-lane")
+
+    def test_ttl_reclaimer_still_handles_non_direct_claims(self) -> None:
+        task_id = kanban_db.create_task(self.connection, title="expired native claim")
+        self.assertIsNotNone(
+            kanban_db.claim_task(
+                self.connection,
+                task_id,
+                ttl_seconds=60,
+                claimer="other-dispatcher",
+            )
+        )
+        expired = int(kanban_db.time.time()) - 1
+        with kanban_db.write_txn(self.connection):
+            self.connection.execute(
+                "UPDATE tasks SET claim_expires = ? WHERE id = ?",
+                (expired, task_id),
+            )
+
+        self.assertEqual(kanban_db.release_stale_claims(self.connection), 1)
+        self.assertEqual(kanban_db.get_task(self.connection, task_id).status, "ready")
+
+    def test_operator_reclaim_still_resets_failure_budget(self) -> None:
+        task_id = kanban_db.create_task(self.connection, title="operator recovery")
+        self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
+        with kanban_db.write_txn(self.connection):
+            self.connection.execute(
+                "UPDATE tasks SET consecutive_failures = 2, "
+                "last_failure_error = ? WHERE id = ?",
+                ("old worker failure", task_id),
+            )
+
+        self.assertTrue(
+            kanban_db.reclaim_task(
+                self.connection,
+                task_id,
+                reason="operator inspected and retried",
+            )
+        )
+        task = kanban_db.get_task(self.connection, task_id)
+        self.assertEqual(task.status, "ready")
+        self.assertEqual(task.consecutive_failures, 0)
+        self.assertIsNone(task.last_failure_error)
+
+    def test_patched_module_advertises_complete_direct_reclaim_safety(self) -> None:
+        self.assertGreaterEqual(
+            kanban_db.DIRECT_CLI_RECLAIM_SAFETY_VERSION,
+            1,
+        )
+
     def test_stale_reclaim_before_transaction_preserves_replacement_run(self) -> None:
         task_id = kanban_db.create_task(self.connection, title="replacement before txn")
         self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))

dockerfiles/patch_hermes_run_safety.py

@@ -11,6 +11,24 @@ source_root = Path(os.environ.get("HERMES_SOURCE_ROOT", "/opt/hermes"))
 db_path = source_root / "hermes_cli/kanban_db.py"
 db = db_path.read_text(encoding="utf-8")
 
+reclaim_safety_marker_before = '''VALID_STATUSES = {"triage", "todo", "scheduled", "ready", "running", "blocked", "review", "done", "archived"}
+VALID_INITIAL_STATUSES = {"running", "blocked"}
+'''
+reclaim_safety_marker_after = '''VALID_STATUSES = {"triage", "todo", "scheduled", "ready", "running", "blocked", "review", "done", "archived"}
+VALID_INITIAL_STATUSES = {"running", "blocked"}
+
+# ConfigMap direct-lane reclaim stays dormant until this complete image patch
+# (atomic liveness fence, failure preservation, and TTL-reclaimer isolation)
+# is loaded. Bump only when the whole safety contract remains present.
+DIRECT_CLI_RECLAIM_SAFETY_VERSION = 1
+'''
+db = replace_once(
+    db,
+    reclaim_safety_marker_before,
+    reclaim_safety_marker_after,
+    "direct CLI reclaim safety capability marker",
+)
+
 complete_signature_before = '''def complete_task(
     conn: sqlite3.Connection,
     task_id: str,
@@ -317,6 +335,8 @@ reclaim_after = '''def reclaim_task(
     reason: Optional[str] = None,
     signal_fn=None,
     expected_run_id: Optional[int] = None,
+    expected_run_liveness_before: Optional[int] = None,
+    preserve_failure_counter: bool = False,
 ) -> bool:
     """Operator-driven reclaim: release the claim and reset to ``ready``.
 
@@ -330,19 +350,37 @@ reclaim_after = '''def reclaim_task(
     ``BEGIN IMMEDIATE`` and included in the guarded update. A replacement run
     is therefore neither signalled nor reclaimed by stale recovery evidence.
 
+    ``expected_run_liveness_before`` adds an atomic heartbeat-age guard for
+    watchdog callers. The current run's own heartbeat (or its start time before
+    the first heartbeat) must still predate that cutoff inside the same write
+    transaction. A heartbeat racing the watchdog therefore wins safely.
+
+    ``preserve_failure_counter`` distinguishes automatic recovery from an
+    operator intervention. Watchdogs preserve accumulated failures so repeated
+    worker death cannot regain a fresh retry budget on every reclaim.
+
     Returns True if a reclaim happened, False if the task isn't in a
     reclaimable state (not running, or doesn't exist), or its run changed.
     """
     with write_txn(conn):
         row = conn.execute(
-            "SELECT status, claim_lock, worker_pid, current_run_id "
+            "SELECT t.status, t.claim_lock, t.worker_pid, t.current_run_id, "
-            "FROM tasks WHERE id = ?",
+            "COALESCE(r.last_heartbeat_at, r.started_at) AS run_liveness_at "
+            "FROM tasks AS t LEFT JOIN task_runs AS r "
+            "ON r.id = t.current_run_id WHERE t.id = ?",
             (task_id,),
         ).fetchone()
         if not row:
             return False
         if expected_run_id is not None and row["current_run_id"] != expected_run_id:
             return False
+        if expected_run_liveness_before is not None:
+            liveness_at = row["run_liveness_at"]
+            if (
+                liveness_at is None
+                or int(liveness_at) >= int(expected_run_liveness_before)
+            ):
+                return False
         if row["status"] != "running" and row["claim_lock"] is None:
             # Nothing to reclaim — already ready / blocked / done.
             return False
@@ -372,13 +410,15 @@ reclaim_after = '''def reclaim_task(
             conn, task_id,
             outcome="reclaimed", status="reclaimed",
             error=(
-                f"manual_reclaim: {reason}" if reason
+                f"{'watchdog' if preserve_failure_counter else 'manual'}_reclaim: {reason}"
-                else f"manual_reclaim lock={prev_lock}"
+                if reason else
+                f"{'watchdog' if preserve_failure_counter else 'manual'}_reclaim "
+                f"lock={prev_lock}"
             ),
             metadata=termination,
         )
         payload = {
-            "manual": True,
+            "manual": not preserve_failure_counter,
             "reason": reason,
             "prev_lock": prev_lock,
         }
@@ -388,11 +428,12 @@ reclaim_after = '''def reclaim_task(
             payload,
             run_id=run_id,
         )
-    # Operator intervention — they've looked at the task, so the
+    # Only an explicit operator reclaim resets the retry budget. Automatic
-    # consecutive-failures counter is now stale. Give the next retry
+    # watchdog recovery must retain both the count and its diagnostic.
-    # a fresh budget. (_clear_failure_counter opens its own write_txn,
+    if not preserve_failure_counter:
-    # so it runs after the enclosing one commits.)
+        # _clear_failure_counter opens its own write_txn, so it runs after the
-    _clear_failure_counter(conn, task_id)
+        # enclosing one commits.
+        _clear_failure_counter(conn, task_id)
     return True
 '''
 db = replace_once(
@@ -402,6 +443,22 @@ db = replace_once(
     "transactional exact-run reclaim",
 )
 
+stale_direct_before = '''        "FROM tasks "
+        "WHERE status = 'running' AND claim_expires IS NOT NULL "
+        "  AND claim_expires < ?",
+'''
+stale_direct_after = '''        "FROM tasks "
+        "WHERE status = 'running' AND claim_expires IS NOT NULL "
+        "  AND claim_lock IS NOT 'direct-cli-lane' "
+        "  AND claim_expires < ?",
+'''
+db = replace_once(
+    db,
+    stale_direct_before,
+    stale_direct_after,
+    "exclude direct CLI claims from generic TTL reclaim",
+)
+
 task_field_before = '''    current_run_id: Optional[int] = None
     workflow_template_id: Optional[str] = None
 '''

services/hermes/agent-configmap.yaml

@@ -95,12 +95,18 @@ data:
       # implement->review->repair->re-review chain by creating Kanban cards
       # only (subscription lanes), never merging/approving. Bounded by the
       # cycle and concurrent-chain ceilings below.
-      auto_supervise: true
+      auto_supervise: false
       supervise_interval_seconds: 30
       supervise_max_cycles: 5
       supervise_max_chains: 20
       supervise_review_assignee: cli-claude-xhigh
       supervise_repair_assignee: cli-auto
+      # A separate supervisor process reclaims direct-CLI runs that stop
+      # heartbeating. The lane renews its short claim on each 20s heartbeat;
+      # ten minutes bounds worker death without reacting to brief storage loss.
+      # The image capability gate keeps this fail-closed during split rollouts.
+      direct_lane_watchdog_enabled: true
+      direct_lane_heartbeat_timeout_seconds: 600
       dispatch_stale_timeout_seconds: 14400
       # Steady-state quota-aware routing for the direct CLI lane: below this
       # remaining-percent a provider stops receiving NEW cli-auto work while

services/hermes/agent-deployment.yaml

@@ -947,8 +947,9 @@ spec:
           # Autonomous cross-card review->repair->re-review driver. Reads/writes
           # only the local Kanban DB under /opt/data; creates Kanban cards that
           # route through the existing subscription lanes. No runtime-access
-          # mount and no provider client: it holds no metered/API-key path. Inert
+          # mount and no provider client: it holds no metered/API-key path. Its
-          # until kanban.auto_supervise is set true in the deployed config.
+          # cross-card and direct-lane writes have separate config switches; the
+          # direct watchdog also requires the matching image safety capability.
           command: [/opt/hermes/.venv/bin/python, /opt/coordinator/kanban_supervisor.py]
           env:
             - {name: HERMES_HOME, value: /opt/data}

services/hermes/scripts/cli_lane_config.py

@@ -31,6 +31,10 @@ EXTERNAL_PREFIX = "cli-"
 DEFAULT_CLAIM_TTL = 7 * 24 * 60 * 60
 DEFAULT_MAX_RUNTIME = 12 * 60 * 60
 HEARTBEAT_SECONDS = 20
+DIRECT_CLAIM_LOCK = "direct-cli-lane"
+DEFAULT_DIRECT_HEARTBEAT_TIMEOUT_SECONDS = 10 * 60
+DIRECT_CLAIM_TTL_GRACE_SECONDS = 60
+DIRECT_CLI_RECLAIM_SAFETY_VERSION = 1
 PROVIDER_HEALTH_MAX_AGE_SECONDS = 5 * 60
 PROVIDER_AUTH_FAILURE_MAX_AGE_SECONDS = 12 * 60 * 60
 QUOTA_MIN_REMAINING_PERCENT_DEFAULT = 15.0
@@ -218,3 +222,40 @@ def kanban_setting(name: str, default: float) -> float:
     if isinstance(value, bool) or not isinstance(value, (int, float)):
         return default
     return float(value)
+
+
+def direct_heartbeat_timeout_seconds(configured: Any | None = None) -> int:
+    """Return the bounded no-heartbeat window for a direct CLI run."""
+    if configured is None:
+        configured = kanban_setting(
+            "direct_lane_heartbeat_timeout_seconds",
+            DEFAULT_DIRECT_HEARTBEAT_TIMEOUT_SECONDS,
+        )
+    if isinstance(configured, bool) or not isinstance(configured, (int, float)):
+        parsed = DEFAULT_DIRECT_HEARTBEAT_TIMEOUT_SECONDS
+    else:
+        parsed = int(configured)
+    if parsed <= 0:
+        parsed = DEFAULT_DIRECT_HEARTBEAT_TIMEOUT_SECONDS
+    # Three missed 20s heartbeats is the smallest useful death window. This
+    # also prevents an accidental tiny config value from causing churn.
+    return max(HEARTBEAT_SECONDS * 3, parsed)
+
+
+def direct_reclaim_safety_ready(kanban_db: Any | None = None) -> bool:
+    """Require the image patch that makes both direct reclaimers safe."""
+    if kanban_db is None:
+        try:
+            from hermes_cli import kanban_db as runtime_kanban_db
+        except ImportError:
+            return False
+        kanban_db = runtime_kanban_db
+    version = getattr(kanban_db, "DIRECT_CLI_RECLAIM_SAFETY_VERSION", 0)
+    return type(version) is int and version >= DIRECT_CLI_RECLAIM_SAFETY_VERSION
+
+
+def direct_claim_ttl_seconds(kanban_db: Any | None = None) -> int:
+    """Use the short TTL only after the complete image safety patch is loaded."""
+    if not direct_reclaim_safety_ready(kanban_db):
+        return DEFAULT_CLAIM_TTL
+    return direct_heartbeat_timeout_seconds() + DIRECT_CLAIM_TTL_GRACE_SECONDS

services/hermes/scripts/cli_lane_dispatch.py

@@ -18,10 +18,11 @@ from cli_lane_capabilities import (
 )
 from cli_lane_config import (
     BOARD_CORRUPTION_ERRORS,
-    DEFAULT_CLAIM_TTL,
+    DIRECT_CLAIM_LOCK,
     EXTERNAL_PREFIX,
     RESULT_SCHEMA,
     RESULT_SCHEMA_PATH,
+    direct_claim_ttl_seconds,
 )
 from cli_lane_execution import execute_claim
 from cli_lane_files import atomic_json
@@ -136,8 +137,8 @@ def claim_ready(
                         result = kanban_db.claim_task(
                             conn,
                             task_id,
-                            ttl_seconds=DEFAULT_CLAIM_TTL,
+                            ttl_seconds=direct_claim_ttl_seconds(),
-                            claimer="direct-cli-lane",
+                            claimer=DIRECT_CLAIM_LOCK,
                         )
                     except Exception:
                         continue

services/hermes/scripts/cli_lane_execution.py

@@ -12,8 +12,10 @@ from pathlib import Path
 import cli_lane_goal
 from cli_lane_board import _board_call, _resolve_workspace, _task_context, _task_value
 from cli_lane_config import (
+    DIRECT_CLAIM_LOCK,
     DEFAULT_MAX_RUNTIME,
     TerminalFinalizationPending,
+    direct_claim_ttl_seconds,
 )
 from cli_lane_failover import _routed_or_blocked, capacity_failover
 from cli_lane_files import (
@@ -84,7 +86,7 @@ def execute_claim(board: str, task_id: str) -> None:
 
         def heartbeat(note: str) -> bool:
             try:
-                return bool(
+                alive = bool(
                     _board_call(
                         kanban_db,
                         board,
@@ -96,10 +98,29 @@ def execute_claim(board: str, task_id: str) -> None:
                         ),
                     )
                 )
+                if not alive:
+                    return False
+                renew = getattr(kanban_db, "heartbeat_claim", None)
+                if not callable(renew):
+                    # Compatibility with a runtime that has not yet acquired
+                    # heartbeat_claim. The watchdog still bounds liveness.
+                    return True
+                return bool(
+                    _board_call(
+                        kanban_db,
+                        board,
+                        lambda fresh: renew(
+                            fresh,
+                            task_id,
+                            ttl_seconds=direct_claim_ttl_seconds(),
+                            claimer=DIRECT_CLAIM_LOCK,
+                        ),
+                    )
+                )
             except (OSError, sqlite3.Error):
-                # The claim TTL is deliberately long. Keep the expensive
+                # Keep the expensive provider process alive during a transient
-                # provider process alive during a transient volume stall and
+                # volume stall. A bounded independent watchdog remains the
-                # retry on the next heartbeat instead of losing its work.
+                # authority if storage/heartbeats stay unavailable.
                 return True
 
         def comment(body: str) -> None:

services/hermes/scripts/kanban_supervisor.py

@@ -23,8 +23,10 @@ module is only the I/O shell: config, board iteration, and turning a
 :class:`supervisor_policy.Decision` into ``create_task``/``block_task``/
 ``add_comment`` calls. It imports no provider client and holds no metered path;
 the sole spawn is a Kanban card that routes through the existing subscription
-lanes. It is inert until ``kanban.auto_supervise`` is set true in the deployed
+lanes. Cross-card automation is inert until ``kanban.auto_supervise`` is true.
-config, re-read every tick like ``kanban.auto_decompose``.
+The direct-lane watchdog has a separate switch and stays fail-closed until the
+loaded image advertises its complete reclaim-safety capability. Both switches
+are re-read every tick like ``kanban.auto_decompose``.
 """
 
 from __future__ import annotations
@@ -40,6 +42,14 @@ from typing import Any, Callable
 import yaml
 
 import supervisor_policy as policy
+from cli_lane_config import (
+    DEFAULT_DIRECT_HEARTBEAT_TIMEOUT_SECONDS,
+    DIRECT_CLAIM_LOCK,
+    EXTERNAL_PREFIX,
+    direct_heartbeat_timeout_seconds,
+    direct_reclaim_safety_ready,
+)
+from cli_lane_recovery import _has_pending_finalization
 
 DATA_ROOT = Path(os.environ.get("HERMES_HOME", "/opt/data"))
 CONFIG_PATH = DATA_ROOT / "config.yaml"
@@ -61,6 +71,10 @@ class Settings:
     enabled: bool
     interval: int
     limits: policy.Limits
+    direct_lane_watchdog_enabled: bool = False
+    direct_lane_heartbeat_timeout_seconds: int = (
+        DEFAULT_DIRECT_HEARTBEAT_TIMEOUT_SECONDS
+    )
 
 
 def _kanban_config() -> dict[str, Any]:
@@ -105,6 +119,15 @@ def load_settings() -> Settings:
                 cfg.get("supervise_repair_assignee"), DEFAULT_REPAIR_ASSIGNEE
             ),
         ),
+        direct_lane_watchdog_enabled=_bool(
+            cfg.get("direct_lane_watchdog_enabled"), False
+        ),
+        direct_lane_heartbeat_timeout_seconds=direct_heartbeat_timeout_seconds(
+            cfg.get(
+                "direct_lane_heartbeat_timeout_seconds",
+                DEFAULT_DIRECT_HEARTBEAT_TIMEOUT_SECONDS,
+            )
+        ),
     )
 
 
@@ -172,6 +195,117 @@ def _comment(kanban_db: Any, conn: Any, task_id: str, body: str) -> None:
         _log(f"could not comment on {task_id}: {error}")
 
 
+def _value(record: Any, name: str, default: Any = None) -> Any:
+    if isinstance(record, dict):
+        return record.get(name, default)
+    return getattr(record, name, default)
+
+
+def _watchdog_board(
+    kanban_db: Any,
+    board: str,
+    heartbeat_timeout_seconds: int,
+    now: int,
+) -> int:
+    """Atomically reclaim heartbeat-dead direct-lane runs on one board."""
+    reclaimed = 0
+    cutoff = now - heartbeat_timeout_seconds
+    with kanban_db.scoped_current_board(board):
+        conn = kanban_db.connect(board=board)
+        try:
+            tasks = list(kanban_db.list_tasks(conn))
+            for task in tasks:
+                task_id = str(_value(task, "id", "") or "")
+                run_id = _value(task, "current_run_id")
+                if (
+                    not task_id
+                    or _value(task, "status") != "running"
+                    or _value(task, "claim_lock") != DIRECT_CLAIM_LOCK
+                    or not str(_value(task, "assignee", "") or "").startswith(
+                        EXTERNAL_PREFIX
+                    )
+                    or type(run_id) is not int
+                ):
+                    continue
+                try:
+                    run = kanban_db.get_run(conn, run_id)
+                    if (
+                        run is None
+                        or _value(run, "id") != run_id
+                        or _value(run, "task_id") != task_id
+                        or _value(run, "status") != "running"
+                        or _value(run, "ended_at") is not None
+                    ):
+                        continue
+                    heartbeat_at = _value(run, "last_heartbeat_at")
+                    liveness_at = (
+                        int(heartbeat_at)
+                        if heartbeat_at is not None
+                        else int(_value(run, "started_at"))
+                    )
+                    if liveness_at >= cutoff:
+                        continue
+                    # Accepted terminal evidence outranks liveness recovery.
+                    # The lane finalizer will either commit it or classify it
+                    # before this watchdog is allowed to release the run.
+                    if _has_pending_finalization(board, task_id, run_id):
+                        continue
+                    age = max(0, now - liveness_at)
+                    reason = (
+                        f"direct CLI heartbeat absent for {age}s "
+                        f"(limit {heartbeat_timeout_seconds}s)"
+                    )
+                    won = kanban_db.reclaim_task(
+                        conn,
+                        task_id,
+                        reason=reason,
+                        expected_run_id=run_id,
+                        expected_run_liveness_before=cutoff,
+                        preserve_failure_counter=True,
+                    )
+                    if not won:
+                        continue
+                    _comment(
+                        kanban_db,
+                        conn,
+                        task_id,
+                        (
+                            f"Watchdog reclaimed heartbeat-dead direct CLI run "
+                            f"{run_id}: {reason}. The claim was released and the "
+                            "task requeued; its failure history was preserved."
+                        ),
+                    )
+                    reclaimed += 1
+                except Exception as error:  # noqa: BLE001 - isolate each run
+                    _log(f"watchdog could not inspect {board}/{task_id}: {error}")
+        finally:
+            conn.close()
+    return reclaimed
+
+
+def watchdog_once(
+    kanban_db: Any,
+    heartbeat_timeout_seconds: int = DEFAULT_DIRECT_HEARTBEAT_TIMEOUT_SECONDS,
+) -> int:
+    """One independent heartbeat-death pass over every active board."""
+    # ConfigMap scripts can roll before the hermes-agent image. Keep every
+    # watchdog write inert until the image advertises the whole atomic reclaim
+    # bundle: liveness fencing, failure-budget preservation, and direct-lane
+    # exclusion from the generic TTL reclaimer.
+    if not direct_reclaim_safety_ready(kanban_db):
+        return 0
+    total = 0
+    now = int(time.time())
+    for board in _iter_boards(kanban_db):
+        try:
+            total += _watchdog_board(
+                kanban_db, board, heartbeat_timeout_seconds, now
+            )
+        except Exception as error:  # noqa: BLE001 - isolate a faulty board
+            _log(f"watchdog temporarily skipping board {board!r}: {error}")
+    return total
+
+
 def _mark_ready_for_human(kanban_db: Any, conn: Any, task_id: str, body: str) -> None:
     """Flag the implementation as human-mergeable. Never merges or clears WIP."""
     setter = getattr(kanban_db, "set_task_metadata", None) or getattr(
@@ -304,10 +438,20 @@ def run_forever(
     load: Callable[[], Settings] = load_settings,
     max_ticks: int | None = None,
 ) -> int:
-    """Poll loop. Inert while ``auto_supervise`` is false; re-reads config a tick."""
+    """Poll loop with independent automation and direct-watchdog switches."""
     ticks = 0
     while max_ticks is None or ticks < max_ticks:
         settings = load()
+        if settings.direct_lane_watchdog_enabled:
+            try:
+                watchdog_once(
+                    kanban_db,
+                    heartbeat_timeout_seconds=(
+                        settings.direct_lane_heartbeat_timeout_seconds
+                    ),
+                )
+            except Exception as error:  # noqa: BLE001 - never stop the safety loop
+                _log(f"watchdog tick failed: {error}")
         if settings.enabled:
             try:
                 supervise_once(kanban_db, settings.limits)

testing/tests/test_hermes_cli_dispatch.py

@@ -43,6 +43,42 @@ def test_unassigned_ready_task_is_persistently_routed_to_auto_lane(monkeypatch):
     assert assigned == [("t_auto", "cli-auto")]
 
 
+def test_claim_ready_uses_bounded_heartbeat_coupled_ttl(monkeypatch):
+    task = SimpleNamespace(id="t_ttl", assignee="cli-auto", status="ready")
+    claim_kwargs = []
+
+    class Connection:
+        def close(self):
+            return None
+
+    def claim_task(_conn, _task_id, **kwargs):
+        claim_kwargs.append(kwargs)
+        return task
+
+    fake_db = SimpleNamespace(
+        list_boards=lambda include_archived=False: [{"slug": "cassandra"}],
+        scoped_current_board=lambda _board: nullcontext(),
+        connect=lambda board: Connection(),
+        recompute_ready=lambda _conn: None,
+        list_tasks=lambda _conn: [task],
+        claim_task=claim_task,
+    )
+    monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
+    monkeypatch.setattr(lanes, "direct_claim_ttl_seconds", lambda: 660)
+
+    assert lanes.claim_ready(set(), 1) == [("cassandra", "t_ttl")]
+    assert claim_kwargs == [{"ttl_seconds": 660, "claimer": "direct-cli-lane"}]
+
+
+def test_direct_claim_ttl_stays_legacy_until_complete_image_patch_is_loaded():
+    legacy_db = SimpleNamespace()
+    safe_db = SimpleNamespace(DIRECT_CLI_RECLAIM_SAFETY_VERSION=1)
+
+    assert lanes.DEFAULT_CLAIM_TTL == 7 * 24 * 60 * 60
+    assert lanes.direct_claim_ttl_seconds(legacy_db) == lanes.DEFAULT_CLAIM_TTL
+    assert lanes.direct_claim_ttl_seconds(safe_db) == 660
+
+
 def test_corrupt_board_is_quarantined_without_stopping_healthy_lanes(monkeypatch, capsys):
     class CorruptBoardError(Exception):
         pass

testing/tests/test_hermes_cli_execution_edges.py

@@ -72,6 +72,69 @@ def test_transient_callback_storage_errors_do_not_kill_provider(
     assert blocks and blocks[0]["kind"] == "capability"
 
 
+def test_heartbeat_renews_short_claim_only_while_exact_run_is_owned(
+    tmp_path: Path,
+    monkeypatch,
+):
+    task = SimpleNamespace(
+        id="t_renew",
+        status="running",
+        current_run_id=24,
+        assignee="cli-auto",
+        max_runtime_seconds=60,
+    )
+    operations = []
+    heartbeat_results = iter([True, False])
+    blocks = []
+
+    def heartbeat_worker(*_args, **kwargs):
+        operations.append(("heartbeat", kwargs["expected_run_id"]))
+        return next(heartbeat_results)
+
+    def heartbeat_claim(*_args, **kwargs):
+        operations.append(("renew", kwargs))
+        return True
+
+    db = SimpleNamespace(
+        scoped_current_board=lambda _board: nullcontext(),
+        connect=lambda board: _Connection(),
+        get_task=lambda *_args: task,
+        worker_log_path=lambda *_args, **_kwargs: tmp_path / "worker.log",
+        _resolve_worktree_workspace=lambda *_args, **_kwargs: (tmp_path, "branch"),
+        set_branch_name=lambda *_args: None,
+        set_workspace_path=lambda *_args: None,
+        build_worker_context=lambda *_args: "renew the lease",
+        heartbeat_worker=heartbeat_worker,
+        heartbeat_claim=heartbeat_claim,
+        add_comment=lambda *_args, **_kwargs: None,
+        block_task=lambda *_args, **kwargs: blocks.append(kwargs),
+    )
+    monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=db))
+    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "lanes")
+    monkeypatch.setattr(lanes, "direct_claim_ttl_seconds", lambda: 660)
+    monkeypatch.setattr(
+        lanes,
+        "select_route",
+        lambda *_args, **_kwargs: lanes.Route(
+            "codex", "gpt", "high", "p", "c", "r", 1, ()
+        ),
+    )
+
+    def provider(*args, **_kwargs):
+        assert args[6]("provider alive") is True
+        assert args[6]("provider lost lease") is False
+        return lanes.ProcessResult(1, "lost lease", None, False)
+
+    monkeypatch.setattr(lanes, "run_provider", provider)
+    lanes.execute_claim("cassandra", "t_renew")
+
+    assert operations == [
+        ("heartbeat", 24),
+        ("renew", {"ttl_seconds": 660, "claimer": "direct-cli-lane"}),
+        ("heartbeat", 24),
+    ]
+
+
 def test_restart_handoff_survives_missing_prior_log(tmp_path: Path, monkeypatch):
     task = SimpleNamespace(
         id="t_restart",

testing/tests/test_hermes_kanban_supervisor.py

@@ -95,6 +95,7 @@ def test_defaults_are_inert_and_conservative_when_config_missing(tmp_path, monke
     monkeypatch.setattr(supervisor, "CONFIG_PATH", tmp_path / "absent.yaml")
     settings = supervisor.load_settings()
     assert settings.enabled is False
+    assert settings.direct_lane_watchdog_enabled is False
     assert settings.interval == supervisor.DEFAULT_INTERVAL_SECONDS
     assert settings.limits.max_cycles == supervisor.DEFAULT_MAX_CYCLES
     assert settings.limits.review_assignee == supervisor.DEFAULT_REVIEW_ASSIGNEE
@@ -111,6 +112,8 @@ def test_config_values_are_read_and_sanitized(tmp_path, monkeypatch):
             "supervise_max_chains": 8,
             "supervise_review_assignee": "  ",  # blank -> default
             "supervise_repair_assignee": "cli-codex-high",
+            "direct_lane_heartbeat_timeout_seconds": 540,
+            "direct_lane_watchdog_enabled": True,
         },
     )
     settings = supervisor.load_settings()
@@ -120,6 +123,18 @@ def test_config_values_are_read_and_sanitized(tmp_path, monkeypatch):
     assert settings.limits.max_chains == 8
     assert settings.limits.review_assignee == supervisor.DEFAULT_REVIEW_ASSIGNEE
     assert settings.limits.repair_assignee == "cli-codex-high"
+    assert settings.direct_lane_heartbeat_timeout_seconds == 540
+    assert settings.direct_lane_watchdog_enabled is True
+
+
+def test_watchdog_timeout_uses_the_shared_three_heartbeat_floor(tmp_path, monkeypatch):
+    _write_config(
+        tmp_path,
+        monkeypatch,
+        {"direct_lane_heartbeat_timeout_seconds": 1},
+    )
+    settings = supervisor.load_settings()
+    assert settings.direct_lane_heartbeat_timeout_seconds == 60
 
 
 def test_malformed_config_falls_back_to_defaults(tmp_path, monkeypatch):
@@ -402,6 +417,127 @@ def test_non_actionable_task_is_a_clean_no_op():
     assert db.created == [] and db.blocked == [] and db.comments == []
 
 
+# --- direct-lane heartbeat watchdog --------------------------------------
+
+
+class WatchdogDb(RecordingDb):
+    DIRECT_CLI_RECLAIM_SAFETY_VERSION = 1
+
+    def __init__(self, tasks, runs):
+        super().__init__(tasks)
+        self._runs = runs
+        self.reclaims = []
+
+    def get_run(self, _conn, run_id):
+        return self._runs.get(run_id)
+
+    def reclaim_task(self, _conn, task_id, **kwargs):
+        self.reclaims.append((task_id, kwargs))
+        return True
+
+
+def _direct_running_task(task_id="dead", run_id=86, **kwargs):
+    values = {
+        "id": task_id,
+        "status": "running",
+        "assignee": "cli-auto",
+        "claim_lock": "direct-cli-lane",
+        "current_run_id": run_id,
+    }
+    values.update(kwargs)
+    return _task(**values)
+
+
+def test_watchdog_reclaims_heartbeat_dead_direct_run_with_atomic_age_guard(
+    monkeypatch,
+):
+    task = _direct_running_task()
+    run = SimpleNamespace(
+        id=86,
+        task_id="dead",
+        status="running",
+        ended_at=None,
+        started_at=100,
+        last_heartbeat_at=399,
+    )
+    db = WatchdogDb([task], {86: run})
+    monkeypatch.setattr(supervisor.time, "time", lambda: 1000)
+    monkeypatch.setattr(
+        supervisor, "_has_pending_finalization", lambda *_args: False
+    )
+
+    assert supervisor.watchdog_once(db, heartbeat_timeout_seconds=600) == 1
+    assert db.reclaims == [
+        (
+            "dead",
+            {
+                "reason": "direct CLI heartbeat absent for 601s (limit 600s)",
+                "expected_run_id": 86,
+                "expected_run_liveness_before": 400,
+                "preserve_failure_counter": True,
+            },
+        )
+    ]
+    assert db.comments[-1][0:2] == ("dead", policy.SUPERVISOR_AUTHOR)
+    assert "failure history was preserved" in db.comments[-1][2]
+
+
+def test_watchdog_fails_closed_against_legacy_image_api(monkeypatch):
+    task = _direct_running_task()
+    run = SimpleNamespace(
+        id=86,
+        task_id="dead",
+        status="running",
+        ended_at=None,
+        started_at=100,
+        last_heartbeat_at=100,
+    )
+    db = WatchdogDb([task], {86: run})
+    db.DIRECT_CLI_RECLAIM_SAFETY_VERSION = 0
+    monkeypatch.setattr(supervisor.time, "time", lambda: 1000)
+    monkeypatch.setattr(supervisor, "_has_pending_finalization", lambda *_args: False)
+
+    assert supervisor.watchdog_once(db, heartbeat_timeout_seconds=600) == 0
+    assert db.reclaims == []
+
+
+def test_watchdog_uses_current_run_liveness_and_skips_live_or_unowned_work(
+    monkeypatch,
+):
+    live = _direct_running_task("live", 1)
+    gateway = _direct_running_task("gateway", 2, claim_lock="other-dispatcher")
+    pending = _direct_running_task("pending", 3)
+    replaced = _direct_running_task("replaced", 4)
+    runs = {
+        1: SimpleNamespace(
+            id=1, task_id="live", status="running", ended_at=None,
+            started_at=100, last_heartbeat_at=950,
+        ),
+        2: SimpleNamespace(
+            id=2, task_id="gateway", status="running", ended_at=None,
+            started_at=100, last_heartbeat_at=100,
+        ),
+        3: SimpleNamespace(
+            id=3, task_id="pending", status="running", ended_at=None,
+            started_at=100, last_heartbeat_at=100,
+        ),
+        4: SimpleNamespace(
+            id=99, task_id="replaced", status="running", ended_at=None,
+            started_at=100, last_heartbeat_at=100,
+        ),
+    }
+    db = WatchdogDb([live, gateway, pending, replaced], runs)
+    monkeypatch.setattr(supervisor.time, "time", lambda: 1000)
+    monkeypatch.setattr(
+        supervisor,
+        "_has_pending_finalization",
+        lambda _board, task_id, _run_id: task_id == "pending",
+    )
+
+    assert supervisor.watchdog_once(db, heartbeat_timeout_seconds=600) == 0
+    assert db.reclaims == []
+
+
 def test_board_connect_failure_is_isolated(capsys):
     db = RecordingDb([], boards=[{"slug": "cassandra"}], raise_on={"connect"})
     assert supervisor.supervise_once(db, policy.Limits()) == 0
@@ -422,12 +558,48 @@ def test_loop_supervises_when_enabled(monkeypatch):
     assert len(calls) == 2 and sleeps == [0, 0]
 
 
-def test_loop_is_inert_when_flag_off(monkeypatch):
+def test_loop_is_inert_when_both_flags_are_off(monkeypatch):
     calls = []
     monkeypatch.setattr(supervisor, "supervise_once", lambda *a: calls.append(a) or 0)
-    settings = supervisor.Settings(enabled=False, interval=0, limits=policy.Limits())
+    watchdog = []
+    monkeypatch.setattr(
+        supervisor,
+        "watchdog_once",
+        lambda _db, heartbeat_timeout_seconds: watchdog.append(
+            heartbeat_timeout_seconds
+        ) or 0,
+    )
+    settings = supervisor.Settings(
+        enabled=False,
+        interval=0,
+        limits=policy.Limits(),
+        direct_lane_heartbeat_timeout_seconds=600,
+    )
     supervisor.run_forever(object(), sleep=lambda _s: None, load=lambda: settings, max_ticks=3)
     assert calls == []
+    assert watchdog == []
+
+
+def test_loop_runs_watchdog_only_when_its_dedicated_flag_is_on(monkeypatch):
+    watchdog = []
+    monkeypatch.setattr(
+        supervisor,
+        "watchdog_once",
+        lambda _db, heartbeat_timeout_seconds: watchdog.append(
+            heartbeat_timeout_seconds
+        ) or 0,
+    )
+    settings = supervisor.Settings(
+        enabled=False,
+        interval=0,
+        limits=policy.Limits(),
+        direct_lane_watchdog_enabled=True,
+        direct_lane_heartbeat_timeout_seconds=600,
+    )
+    supervisor.run_forever(
+        object(), sleep=lambda _s: None, load=lambda: settings, max_ticks=3
+    )
+    assert watchdog == [600, 600, 600]
 
 
 def test_loop_survives_a_failing_tick(monkeypatch, capsys):
@@ -479,6 +651,32 @@ def test_auto_supervise_flag_defaults_false_in_configmap():
     assert payload["kanban"]["auto_supervise"] is False
 
 
+def test_direct_lane_watchdog_has_a_dedicated_deployed_switch():
+    documents = list(
+        yaml.safe_load_all((HERMES / "agent-configmap.yaml").read_text(encoding="utf-8"))
+    )
+    config_doc = next(
+        doc
+        for doc in documents
+        if doc and doc.get("metadata", {}).get("name") == "hermes-agent-config"
+    )
+    payload = yaml.safe_load(config_doc["data"]["config.yaml"])
+    assert payload["kanban"]["direct_lane_watchdog_enabled"] is True
+
+
+def test_direct_lane_watchdog_timeout_is_deployed():
+    documents = list(
+        yaml.safe_load_all((HERMES / "agent-configmap.yaml").read_text(encoding="utf-8"))
+    )
+    config_doc = next(
+        doc
+        for doc in documents
+        if doc and doc.get("metadata", {}).get("name") == "hermes-agent-config"
+    )
+    payload = yaml.safe_load(config_doc["data"]["config.yaml"])
+    assert payload["kanban"]["direct_lane_heartbeat_timeout_seconds"] == 600
+
+
 def test_supervisor_scripts_registered_in_coordinator_configmap():
     kustomization = yaml.safe_load((HERMES / "kustomization.yaml").read_text(encoding="utf-8"))
     coordinator = next(