From 7a01eb0c02f7bdecadfab0d7e9ec004fd286cf24 Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Fri, 21 Aug 2026 05:55:46 +0000 Subject: [PATCH] fix(hermes): reclaim heartbeat-dead CLI runs --- dockerfiles/hermes_run_safety_regression.py | 40 ++++++ dockerfiles/patch_hermes_run_safety.py | 19 ++- services/hermes/agent-configmap.yaml | 4 + services/hermes/scripts/cli_lane_config.py | 29 +++- services/hermes/scripts/cli_lane_dispatch.py | 7 +- services/hermes/scripts/cli_lane_execution.py | 29 +++- services/hermes/scripts/kanban_supervisor.py | 126 +++++++++++++++++ testing/tests/test_hermes_cli_dispatch.py | 27 ++++ .../tests/test_hermes_cli_execution_edges.py | 63 +++++++++ .../tests/test_hermes_kanban_supervisor.py | 130 +++++++++++++++++- 10 files changed, 463 insertions(+), 11 deletions(-) diff --git a/dockerfiles/hermes_run_safety_regression.py b/dockerfiles/hermes_run_safety_regression.py index 7f242e98..0cae0148 100644 --- a/dockerfiles/hermes_run_safety_regression.py +++ b/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)) diff --git a/dockerfiles/patch_hermes_run_safety.py b/dockerfiles/patch_hermes_run_safety.py index 09eecb2f..58b34dde 100644 --- a/dockerfiles/patch_hermes_run_safety.py +++ b/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 diff --git a/services/hermes/agent-configmap.yaml b/services/hermes/agent-configmap.yaml index d339b8c2..0adeb455 100644 --- a/services/hermes/agent-configmap.yaml +++ b/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 diff --git a/services/hermes/scripts/cli_lane_config.py b/services/hermes/scripts/cli_lane_config.py index 895aa5cc..e18a7939 100644 --- a/services/hermes/scripts/cli_lane_config.py +++ b/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 diff --git a/services/hermes/scripts/cli_lane_dispatch.py b/services/hermes/scripts/cli_lane_dispatch.py index 2630f8d7..060cb56a 100644 --- a/services/hermes/scripts/cli_lane_dispatch.py +++ b/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 diff --git a/services/hermes/scripts/cli_lane_execution.py b/services/hermes/scripts/cli_lane_execution.py index f89ded06..8f67126f 100644 --- a/services/hermes/scripts/cli_lane_execution.py +++ b/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: diff --git a/services/hermes/scripts/kanban_supervisor.py b/services/hermes/scripts/kanban_supervisor.py index 6d02f30c..d25ebec3 100644 --- a/services/hermes/scripts/kanban_supervisor.py +++ b/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) diff --git a/testing/tests/test_hermes_cli_dispatch.py b/testing/tests/test_hermes_cli_dispatch.py index 4138ab63..b5610c5b 100644 --- a/testing/tests/test_hermes_cli_dispatch.py +++ b/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 diff --git a/testing/tests/test_hermes_cli_execution_edges.py b/testing/tests/test_hermes_cli_execution_edges.py index d1ffbda8..4153c1d9 100644 --- a/testing/tests/test_hermes_cli_execution_edges.py +++ b/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", diff --git a/testing/tests/test_hermes_kanban_supervisor.py b/testing/tests/test_hermes_kanban_supervisor.py index b1ddde09..29989701 100644 --- a/testing/tests/test_hermes_kanban_supervisor.py +++ b/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( -- 2.47.2