diff --git a/gateway/run.py b/gateway/run.py index 18aa5ef175fc..ca6556538fdf 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -5744,6 +5744,63 @@ def _add(path: str) -> None: path, exc, ) + def _write_dispatcher_heartbeat( + self, + *, + heartbeat_path: Path, + interval: float, + cycles_since_start: int, + cycle_started_at: float, + any_spawned: bool, + spawned_total: int, + ready_pending: bool, + bad_ticks: int, + cycle_error: str | None, + ) -> None: + """Write a one-shot heartbeat snapshot to disk for external observers. + + Called at the end of every dispatcher cycle (success OR failure). + External tools (`hermes gateway status`, monitoring scripts) read + this to detect silent stalls — the gateway PID may stay alive but + the dispatch loop can stop cycling (hermes-agent#6, #7). + + Write is atomic (temp + rename) and never raises — the caller wraps + in try/except so heartbeat-write failures cannot kill the dispatcher. + + Schema (stable; tools may grow tolerant of new fields): + schema_version: int (currently 1) + last_cycle_ts: float (unix seconds, end of cycle) + last_cycle_iso: str (UTC ISO 8601) + cycle_started_at: float (unix seconds, start of cycle) + cycle_duration_seconds: float + interval_seconds: float (configured cadence) + cycles_since_start: int (monotonic counter) + any_spawned_this_cycle: bool + spawned_total_this_cycle: int (count across all boards) + ready_pending: bool (whether the ready queue had work this cycle) + consecutive_bad_ticks: int + gateway_pid: int + cycle_error: str | null (exception text if the cycle errored) + """ + from datetime import timezone as _tz + now = time.time() + payload = { + "schema_version": 1, + "last_cycle_ts": now, + "last_cycle_iso": datetime.fromtimestamp(now, tz=_tz.utc).isoformat(timespec="seconds").replace("+00:00", "Z"), + "cycle_started_at": cycle_started_at, + "cycle_duration_seconds": max(0.0, now - cycle_started_at), + "interval_seconds": float(interval), + "cycles_since_start": cycles_since_start, + "any_spawned_this_cycle": any_spawned, + "spawned_total_this_cycle": spawned_total, + "ready_pending": ready_pending, + "consecutive_bad_ticks": bad_ticks, + "gateway_pid": os.getpid(), + "cycle_error": cycle_error, + } + atomic_json_write(heartbeat_path, payload) + async def _kanban_dispatcher_watcher(self) -> None: """Embedded kanban dispatcher — one tick every `dispatch_interval_seconds`. @@ -5918,6 +5975,13 @@ async def _kanban_dispatcher_watcher(self) -> None: HEALTH_WINDOW = 6 bad_ticks = 0 last_warn_at = 0 + # Dispatcher heartbeat — written to disk every cycle so external + # observers can detect silent stalls (the gateway PID stays alive + # but the dispatch loop has stopped cycling). See `hermes gateway + # status`. Path is HERMES_HOME/state/dispatcher_health.json. + cycles_since_start = 0 + last_cycle_started_at = 0.0 + _heartbeat_path = _hermes_home / "state" / "dispatcher_health.json" # Avoid hot-looping corrupt-looking board DBs, but do not suppress # same-fingerprint retries forever: transient WAL/open races can # surface as "database disk image is malformed" for one tick. @@ -6187,6 +6251,12 @@ def _auto_decompose_tick() -> int: "kanban dispatcher: embedded in gateway (interval=%.1fs)", interval ) while self._running: + cycles_since_start += 1 + last_cycle_started_at = time.time() + cycle_error: str | None = None + any_spawned = False + ready_pending = False + spawned_total = 0 try: # Reap zombie children before per-board work so a board DB # failure cannot block cleanup of unrelated workers. @@ -6204,10 +6274,10 @@ def _auto_decompose_tick() -> int: if auto_decompose_enabled: await asyncio.to_thread(_auto_decompose_tick) results = await asyncio.to_thread(_tick_once) - any_spawned = False for slug, res in (results or []): if res is not None and getattr(res, "spawned", None): any_spawned = True + spawned_total += len(res.spawned) # Quiet by default — only log when something actually # happened, so an idle gateway stays silent. logger.info( @@ -6240,9 +6310,42 @@ def _auto_decompose_tick() -> int: last_warn_at = now except asyncio.CancelledError: logger.debug("kanban dispatcher: cancelled") + # Write a final heartbeat noting the cancellation, then re-raise. + self._write_dispatcher_heartbeat( + heartbeat_path=_heartbeat_path, + interval=interval, + cycles_since_start=cycles_since_start, + cycle_started_at=last_cycle_started_at, + any_spawned=any_spawned, + spawned_total=spawned_total, + ready_pending=ready_pending, + bad_ticks=bad_ticks, + cycle_error="cancelled", + ) raise - except Exception: + except Exception as e: logger.exception("kanban dispatcher: unexpected watcher error") + cycle_error = f"{type(e).__name__}: {e}" + + # Write heartbeat every cycle — success OR exception. + # External observers (hermes gateway status / monitoring) read this + # to detect silent stalls where the gateway PID is alive but the + # dispatcher has stopped cycling. + try: + self._write_dispatcher_heartbeat( + heartbeat_path=_heartbeat_path, + interval=interval, + cycles_since_start=cycles_since_start, + cycle_started_at=last_cycle_started_at, + any_spawned=any_spawned, + spawned_total=spawned_total, + ready_pending=ready_pending, + bad_ticks=bad_ticks, + cycle_error=cycle_error, + ) + except Exception: + # Heartbeat-write failure must NEVER kill the dispatcher. + logger.exception("kanban dispatcher: heartbeat write failed (non-fatal)") # Sleep in 1s slices so shutdown is snappy — otherwise a stop() # waits up to `interval` seconds for the current sleep to finish. diff --git a/tests/gateway/test_dispatcher_heartbeat.py b/tests/gateway/test_dispatcher_heartbeat.py new file mode 100644 index 000000000000..14eb087f0726 --- /dev/null +++ b/tests/gateway/test_dispatcher_heartbeat.py @@ -0,0 +1,186 @@ +"""Tests for the kanban dispatcher heartbeat written by gateway.run.GatewayRunner. + +The heartbeat is a stable contract for external observers (monitoring scripts, +future `hermes gateway status` integration). These tests pin the schema so +silent contract breaks fail loudly. +""" +from __future__ import annotations + +import json +import os +from pathlib import Path +from unittest.mock import MagicMock + +import pytest + + +@pytest.fixture +def runner(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + """Construct a minimal GatewayRunner with a tmp HERMES_HOME.""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + # Importing run.py is heavy (loads adapters); import here so the fixture + # picks up the env override. + from gateway import run as gateway_run + + # Reload module so the module-level _hermes_home picks up our tmp path. + import importlib + + importlib.reload(gateway_run) + + instance = gateway_run.GatewayRunner.__new__(gateway_run.GatewayRunner) + return instance, gateway_run, tmp_path + + +def test_heartbeat_writes_expected_schema_v1(runner) -> None: + instance, _gateway_run, tmp_path = runner + hb_path = tmp_path / "state" / "dispatcher_health.json" + + instance._write_dispatcher_heartbeat( + heartbeat_path=hb_path, + interval=60.0, + cycles_since_start=5, + cycle_started_at=1_780_000_000.0, + any_spawned=True, + spawned_total=3, + ready_pending=False, + bad_ticks=0, + cycle_error=None, + ) + + assert hb_path.exists(), "heartbeat file should exist after write" + data = json.loads(hb_path.read_text()) + + # Schema-v1 required keys — any missing key = silent contract break. + required = { + "schema_version", + "last_cycle_ts", + "last_cycle_iso", + "cycle_started_at", + "cycle_duration_seconds", + "interval_seconds", + "cycles_since_start", + "any_spawned_this_cycle", + "spawned_total_this_cycle", + "ready_pending", + "consecutive_bad_ticks", + "gateway_pid", + "cycle_error", + } + assert required <= set(data.keys()), f"missing keys: {required - set(data.keys())}" + + # Type checks (catches refactors that change shape). + assert data["schema_version"] == 1 + assert isinstance(data["last_cycle_ts"], (int, float)) + assert data["last_cycle_iso"].endswith("Z"), "iso should be UTC with Z suffix" + assert data["interval_seconds"] == 60.0 + assert data["cycles_since_start"] == 5 + assert data["any_spawned_this_cycle"] is True + assert data["spawned_total_this_cycle"] == 3 + assert data["ready_pending"] is False + assert data["consecutive_bad_ticks"] == 0 + assert data["gateway_pid"] == os.getpid() + assert data["cycle_error"] is None + + +def test_heartbeat_records_cycle_error(runner) -> None: + instance, _gateway_run, tmp_path = runner + hb_path = tmp_path / "state" / "dispatcher_health.json" + + instance._write_dispatcher_heartbeat( + heartbeat_path=hb_path, + interval=60.0, + cycles_since_start=1, + cycle_started_at=1_780_000_000.0, + any_spawned=False, + spawned_total=0, + ready_pending=True, + bad_ticks=2, + cycle_error="RuntimeError: simulated provider auth crash", + ) + + data = json.loads(hb_path.read_text()) + assert data["cycle_error"] == "RuntimeError: simulated provider auth crash" + assert data["consecutive_bad_ticks"] == 2 + assert data["any_spawned_this_cycle"] is False + assert data["ready_pending"] is True + + +def test_heartbeat_overwrites_previous_atomic(runner) -> None: + """Second write must replace the first, not append.""" + instance, _gateway_run, tmp_path = runner + hb_path = tmp_path / "state" / "dispatcher_health.json" + + instance._write_dispatcher_heartbeat( + heartbeat_path=hb_path, + interval=60.0, + cycles_since_start=1, + cycle_started_at=1_780_000_000.0, + any_spawned=False, + spawned_total=0, + ready_pending=False, + bad_ticks=0, + cycle_error=None, + ) + instance._write_dispatcher_heartbeat( + heartbeat_path=hb_path, + interval=60.0, + cycles_since_start=2, + cycle_started_at=1_780_000_060.0, + any_spawned=True, + spawned_total=1, + ready_pending=False, + bad_ticks=0, + cycle_error=None, + ) + + data = json.loads(hb_path.read_text()) + assert data["cycles_since_start"] == 2, "second write should replace first" + assert data["any_spawned_this_cycle"] is True + + +def test_heartbeat_counter_starts_at_one_first_cycle(runner) -> None: + """Contract pinned: the dispatcher's watcher increments cycles_since_start + BEFORE the first cycle body runs, so the first heartbeat shows 1, not 0. + Monitors should expect the smallest valid value to be 1; 0 means "never wrote." + """ + instance, _gateway_run, tmp_path = runner + hb_path = tmp_path / "state" / "dispatcher_health.json" + + # Simulate what the watcher does at the very first iteration. + instance._write_dispatcher_heartbeat( + heartbeat_path=hb_path, + interval=60.0, + cycles_since_start=1, # ← contract: first cycle is 1 + cycle_started_at=1_780_000_000.0, + any_spawned=False, + spawned_total=0, + ready_pending=False, + bad_ticks=0, + cycle_error=None, + ) + + data = json.loads(hb_path.read_text()) + assert data["cycles_since_start"] == 1, ( + "first heartbeat must report cycles_since_start=1 — monitors treat 0 as 'never wrote'" + ) + + +def test_heartbeat_creates_parent_state_dir_if_missing(runner) -> None: + """state/ subdir may not exist on first gateway run — atomic_json_write should mkdir.""" + instance, _gateway_run, tmp_path = runner + hb_path = tmp_path / "state" / "deep" / "nested" / "dispatcher_health.json" + assert not hb_path.parent.exists() + + instance._write_dispatcher_heartbeat( + heartbeat_path=hb_path, + interval=60.0, + cycles_since_start=1, + cycle_started_at=1_780_000_000.0, + any_spawned=False, + spawned_total=0, + ready_pending=False, + bad_ticks=0, + cycle_error=None, + ) + + assert hb_path.exists()