dockerfiles/hermes_run_safety_regression.py

@@ -342,6 +342,46 @@ 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_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

@@ -317,6 +317,7 @@ 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,
 ) -> bool:
     """Operator-driven reclaim: release the claim and reset to ``ready``.
 
@@ -330,19 +331,33 @@ 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.
+
     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 "
-            "FROM tasks WHERE id = ?",
+            "SELECT t.status, t.claim_lock, t.worker_pid, t.current_run_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

services/hermes/agent-configmap.yaml

@@ -99,6 +99,10 @@ data:
       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.
+      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/scripts/cli_lane_config.py

@@ -28,9 +28,16 @@ CLAUDE_SETTINGS = DATA_ROOT / "home/.claude/settings.json"
 RESULT_SCHEMA_PATH = STATE_ROOT / "worker-result.schema.json"
 EFFORTS = ("low", "medium", "high", "xhigh")
 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
+# Compatibility constant for callers that only need the default. Direct-lane
+# claims use ``direct_claim_ttl_seconds`` so deployed config is re-read.
+DEFAULT_CLAIM_TTL = (
+    DEFAULT_DIRECT_HEARTBEAT_TIMEOUT_SECONDS + DIRECT_CLAIM_TTL_GRACE_SECONDS
+)
 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 +225,23 @@ 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() -> int:
+    """Return the bounded no-heartbeat window for a direct CLI run."""
+    configured = int(
+        kanban_setting(
+            "direct_lane_heartbeat_timeout_seconds",
+            DEFAULT_DIRECT_HEARTBEAT_TIMEOUT_SECONDS,
+        )
+    )
+    if configured <= 0:
+        configured = 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, configured)
+
+
+def direct_claim_ttl_seconds() -> int:
+    """Keep claim expiry just beyond the independent heartbeat watchdog."""
+    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,
-                            claimer="direct-cli-lane",
+                            ttl_seconds=direct_claim_ttl_seconds(),
+                            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
-                # provider process alive during a transient volume stall and
-                # retry on the next heartbeat instead of losing its work.
+                # Keep the expensive provider process alive during a transient
+                # volume stall. A bounded independent watchdog remains the
+                # authority if storage/heartbeats stay unavailable.
                 return True
 
         def comment(body: str) -> None:

services/hermes/scripts/kanban_supervisor.py

@@ -40,6 +40,12 @@ 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,
+)
+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 +67,9 @@ class Settings:
     enabled: bool
     interval: int
     limits: policy.Limits
+    direct_lane_heartbeat_timeout_seconds: int = (
+        DEFAULT_DIRECT_HEARTBEAT_TIMEOUT_SECONDS
+    )
 
 
 def _kanban_config() -> dict[str, Any]:
@@ -105,6 +114,10 @@ def load_settings() -> Settings:
                 cfg.get("supervise_repair_assignee"), DEFAULT_REPAIR_ASSIGNEE
             ),
         ),
+        direct_lane_heartbeat_timeout_seconds=_positive_int(
+            cfg.get("direct_lane_heartbeat_timeout_seconds"),
+            DEFAULT_DIRECT_HEARTBEAT_TIMEOUT_SECONDS,
+        ),
     )
 
 
@@ -172,6 +185,110 @@ 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,
+                    )
+                    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 without counting a worker failure."
+                        ),
+                    )
+                    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."""
+    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(
@@ -308,6 +425,15 @@ def run_forever(
     ticks = 0
     while max_ticks is None or ticks < max_ticks:
         settings = load()
+        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,33 @@ 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_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

@@ -111,6 +111,7 @@ 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,
         },
     )
     settings = supervisor.load_settings()
@@ -120,6 +121,7 @@ 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
 
 
 def test_malformed_config_falls_back_to_defaults(tmp_path, monkeypatch):
@@ -402,6 +404,105 @@ 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):
+    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,
+            },
+        )
+    ]
+    assert db.comments[-1][0:2] == ("dead", policy.SUPERVISOR_AUTHOR)
+    assert "requeued without counting a worker failure" in db.comments[-1][2]
+
+
+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
@@ -425,9 +526,23 @@ def test_loop_supervises_when_enabled(monkeypatch):
 def test_loop_is_inert_when_flag_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 == [600, 600, 600]
 
 
 def test_loop_survives_a_failing_tick(monkeypatch, capsys):
@@ -479,6 +594,19 @@ def test_auto_supervise_flag_defaults_false_in_configmap():
     assert payload["kanban"]["auto_supervise"] is False
 
 
+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(