From 3b381a0bc2b88d6cf62b79779025b3f4e0d6f47b Mon Sep 17 00:00:00 2001 From: Shannon Sands Date: Wed, 19 Aug 2026 14:48:37 +1000 Subject: [PATCH 1/5] fix(gateway): startup-liveness watchdog for pre-event-loop deadlocks (OOF-298) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A hosted gateway (hermes-doubleam-2568) deadlocked at startup with every thread parked in futex_wait_queue before the asyncio loop came alive: zero log lines, /health unreachable — but s6 saw a live PID so it never respawned the process, and a stale gateway_state.json from the previous life told every status surface "draining" for ~30 hours. Every existing liveness backstop assumes startup succeeded: the loop-liveness watchdog is armed inside the running loop's startup path, the shutdown watchdog arms at stop(), and the heartbeat file is written by an asyncio task. None can fire when the process wedges before the loop exists. New gateway/startup_watchdog.py: a plain daemon OS thread armed at process entry (both gateway.run.main() and the `hermes gateway run` CLI wrapper), disarmed the moment GatewayRunner confirms a live event loop — the point where the existing loop-liveness watchdog takes over. If startup neither reaches that milestone nor exits within the deadline (default 300s; slowest legitimate pre-loop work is the 120s-bounded MCP discovery wait), the watchdog: * dumps all-thread stacks via faulthandler, * appends a JSON record to logs/gateway-startup-watchdog.log, * records the exit in the NS-608 lifecycle ledger (reason=startup_liveness_watchdog) so the next boot classifies it instead of reporting an unclean SIGKILL/OOM death, * os._exit(75) so s6/systemd respawn the process. Config is env-only (HERMES_STARTUP_WATCHDOG=0 to disable, HERMES_STARTUP_WATCHDOG_TIMEOUT_S to tune, floor-clamped to 30s): the watchdog must be armed before config.yaml is loaded — a wedge during config parsing is exactly in scope — so it cannot depend on config for its own enablement. Everything is best-effort; a watchdog failure never affects the startup it observes. Arm sites are placed after the PID-file/--replace conflict guards so a --replace loser exiting early never arms a watchdog. Disarm happens even when the loop guards are config-disabled (gateway.loop_watchdog: false) — the startup watchdog only covers the pre-loop window, never adapter connects or steady-state, so WhatsApp pairing / npm cold installs are unaffected. Tests: tests/gateway/test_startup_watchdog.py (29 tests — config resolution, arm/disarm idempotency, fire path with captured exit, lifecycle-ledger marking, dump record, disable knob). Fixes OOF-298. --- gateway/run.py | 24 ++- gateway/startup_watchdog.py | 274 +++++++++++++++++++++++++ hermes_cli/gateway.py | 13 ++ tests/gateway/test_startup_watchdog.py | 206 +++++++++++++++++++ 4 files changed, 516 insertions(+), 1 deletion(-) create mode 100644 gateway/startup_watchdog.py create mode 100644 tests/gateway/test_startup_watchdog.py diff --git a/gateway/run.py b/gateway/run.py index 84604b303b7b..1611c09d9f74 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -12442,6 +12442,17 @@ async def start(self) -> bool: self._gateway_loop = None if self._gateway_loop is not None: self._start_loop_liveness_guards(self._gateway_loop) + # The event loop is confirmed live: the startup-liveness watchdog's + # job is done and the loop-liveness watchdog (armed just above) + # takes over from here (OOF-298). Disarm even when the loop guards + # are config-disabled — the startup watchdog only covers the + # pre-loop window, never adapter connects or steady-state. + try: + from gateway.startup_watchdog import disarm_startup_watchdog + + disarm_startup_watchdog() + except Exception: + logger.debug("Startup watchdog disarm failed", exc_info=True) logger.info("Session storage: %s", self.config.sessions_dir) # Sanity-check that systemd's TimeoutStopSec covers our drain @@ -30947,7 +30958,7 @@ def main(): os.environ.setdefault("AI_AGENT", "hermes-agent") os.environ.setdefault("HERMES_AGENT", "true") - # Positive process identity: ledger registration + Windows job-object +# Positive process identity: ledger registration + Windows job-object # self-attach, so update-time reapers can identify this gateway (and its # child tree dies with it on Windows). Best-effort — never blocks startup. try: @@ -30961,6 +30972,17 @@ def main(): except Exception: pass + # Startup-liveness watchdog (OOF-298): armed before ANY other startup + # work — config load, imports, DB opens — so a deadlock anywhere in the + # pre-event-loop window still gets the process respawned by the service + # supervisor instead of wedging as a live-PID zombie. Disarmed by + # GatewayRunner once the event loop is confirmed live. + try: + from gateway.startup_watchdog import arm_startup_watchdog + arm_startup_watchdog() + except Exception: + pass + # Force UTF-8 stdio on Windows — gateway logs and startup banner would # otherwise UnicodeEncodeError on cp1252 consoles. No-op on POSIX. try: diff --git a/gateway/startup_watchdog.py b/gateway/startup_watchdog.py new file mode 100644 index 000000000000..eed5fa04eb19 --- /dev/null +++ b/gateway/startup_watchdog.py @@ -0,0 +1,274 @@ +"""Startup-liveness watchdog — respawn a gateway that wedges before its loop runs (OOF-298). + +The existing liveness backstops all assume startup succeeded: + +* the loop-liveness watchdog (:mod:`gateway.shutdown_watchdog`) is armed by + ``GatewayRunner._start_loop_liveness_guards`` — *inside* the running event + loop's startup path; +* the shutdown watchdog is armed at ``stop()``; +* the loop heartbeat file is written by an asyncio task. + +None of them can fire if the process deadlocks **before the event loop comes +alive**. That failure mode is real: OOF-298 documents a hosted gateway whose +process sat for ~30 hours with every thread parked in ``futex_wait_queue``, +zero log lines written, ``/health`` unreachable — while s6 saw a live PID and +therefore never respawned it, and a stale ``gateway_state.json`` from the +*previous* life told every status surface the gateway was "draining". + +This module closes that gap with the simplest thing that works: a plain +daemon OS thread armed at process entry, disarmed the moment the event loop +is confirmed live (the point where the existing loop-liveness watchdog takes +over). If startup neither reaches that milestone nor exits within the +deadline, the watchdog dumps all-thread stacks via ``faulthandler``, records +the exit in the lifecycle ledger (NS-608) so the next boot classifies it +correctly, and ``os._exit``\\ s with the service-restart code so s6/systemd +revive the process instead of babysitting a zombie. + +Deadline rationale: a healthy startup reaches the disarm point in seconds. +The slowest legitimate pre-loop work is MCP tool discovery (bounded 120s +internal wait), so the 300s default leaves comfortable headroom. Platform +adapter connects — which can genuinely take minutes (WhatsApp pairing, npm +cold installs) — happen *after* the disarm point and are never covered by +this watchdog. + +Config surface is deliberately env-only (``HERMES_STARTUP_WATCHDOG=0`` to +disable, ``HERMES_STARTUP_WATCHDOG_TIMEOUT_S`` to tune): the watchdog must be +armed before config.yaml is loaded — a wedge during config parsing is exactly +in scope — so it cannot depend on config for its own enablement. + +Everything here is best-effort: a watchdog failure must never affect the +startup it is observing. +""" + +from __future__ import annotations + +import faulthandler +import json +import logging +import os +import threading +import time +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict, Optional + +from gateway.restart import GATEWAY_SERVICE_RESTART_EXIT_CODE + +logger = logging.getLogger(__name__) + +DEFAULT_STARTUP_WATCHDOG_TIMEOUT_S = 300.0 +_MIN_TIMEOUT_S = 30.0 + +ENV_STARTUP_WATCHDOG = "HERMES_STARTUP_WATCHDOG" +ENV_STARTUP_WATCHDOG_TIMEOUT_S = "HERMES_STARTUP_WATCHDOG_TIMEOUT_S" + +_DUMP_RELATIVE = ("logs", "gateway-startup-watchdog.log") + +_FALSEY = frozenset({"0", "false", "no", "off"}) + +# Module-level singleton: the arm sites (gateway.run.main and the +# hermes_cli.gateway CLI wrapper) and the disarm site +# (GatewayRunner._start_loop_liveness_guards) have no shared object to hand a +# handle through, and only one gateway startup ever runs per process. +_handle_lock = threading.Lock() +_handle: Optional["StartupWatchdogHandle"] = None + + +def _process_hermes_home() -> Path: + """HERMES_HOME for process-level diagnostic files (ignore task overrides).""" + val = os.environ.get("HERMES_HOME", "").strip() + if val: + return Path(val) + from hermes_constants import get_hermes_home + + return get_hermes_home() + + +def get_startup_watchdog_dump_path(home: Optional[Path] = None) -> Path: + """Return ``/logs/gateway-startup-watchdog.log``.""" + base = home if home is not None else _process_hermes_home() + return base.joinpath(*_DUMP_RELATIVE) + + +def startup_watchdog_disabled() -> bool: + """True when ``HERMES_STARTUP_WATCHDOG`` opts out explicitly.""" + raw = os.environ.get(ENV_STARTUP_WATCHDOG, "").strip().lower() + return raw in _FALSEY + + +def resolve_startup_watchdog_timeout() -> float: + """Deadline in seconds; env override, floor-clamped, default on garbage.""" + raw = os.environ.get(ENV_STARTUP_WATCHDOG_TIMEOUT_S, "").strip() + if not raw: + return DEFAULT_STARTUP_WATCHDOG_TIMEOUT_S + try: + value = float(raw) + except ValueError: + logger.warning( + "Ignoring non-numeric %s=%r; using default %.0fs", + ENV_STARTUP_WATCHDOG_TIMEOUT_S, + raw, + DEFAULT_STARTUP_WATCHDOG_TIMEOUT_S, + ) + return DEFAULT_STARTUP_WATCHDOG_TIMEOUT_S + if value <= 0: + return DEFAULT_STARTUP_WATCHDOG_TIMEOUT_S + return max(value, _MIN_TIMEOUT_S) + + +def _write_dump_record(record: Dict[str, Any]) -> None: + """Append a one-line JSON metadata record beside the faulthandler dump.""" + try: + path = get_startup_watchdog_dump_path() + path.parent.mkdir(parents=True, exist_ok=True) + with open(path, "a", encoding="utf-8") as fh: + fh.write(json.dumps(record, default=str) + "\n") + except Exception: + logger.debug("Failed to write startup watchdog dump record", exc_info=True) + + +class StartupWatchdogHandle: + """Disarm/inspect handle for the armed startup watchdog thread.""" + + def __init__(self, timeout_s: float, exit_code: int): + self.timeout_s = timeout_s + self.exit_code = exit_code + self.armed_at = time.monotonic() + self._disarmed = threading.Event() + self._thread: Optional[threading.Thread] = None + + def disarm(self) -> None: + """Startup reached a live event loop — stand down. Idempotent.""" + self._disarmed.set() + + @property + def disarmed(self) -> bool: + return self._disarmed.is_set() + + def is_alive(self) -> bool: + return self._thread is not None and self._thread.is_alive() + + def join(self, timeout: Optional[float] = None) -> None: + if self._thread is not None: + self._thread.join(timeout=timeout) + + # ── internals ──────────────────────────────────────────────────────── + + def _fire(self) -> None: + elapsed = time.monotonic() - self.armed_at + try: + logger.critical( + "Gateway startup did not reach a live event loop within %.0fs " + "(elapsed %.0fs); dumping all thread stacks and exiting with " + "code %d so the service supervisor can restart it (OOF-298).", + self.timeout_s, + elapsed, + self.exit_code, + ) + except Exception: + pass + _write_dump_record( + { + "ts": datetime.now(timezone.utc).isoformat(), + "tag": "startup_watchdog.fired", + "pid": os.getpid(), + "timeout_s": self.timeout_s, + "elapsed_s": round(elapsed, 3), + "exit_code": self.exit_code, + } + ) + try: + faulthandler.dump_traceback(all_threads=True) + except Exception: + logger.debug("Startup watchdog faulthandler dump failed", exc_info=True) + # Record the exit in the lifecycle sentinel so the next boot reports + # "startup watchdog hard-exit" instead of misclassifying this as an + # unclean SIGKILL/OOM death (NS-608). + try: + from gateway.lifecycle_ledger import mark_exited + + mark_exited(self.exit_code, reason="startup_liveness_watchdog") + except Exception: + pass + self._exit(self.exit_code) + + @staticmethod + def _exit(code: int) -> None: + """Seam for tests; production is a bare ``os._exit``.""" + os._exit(code) + + def _run(self) -> None: + if self._disarmed.wait(timeout=self.timeout_s): + return + if self._disarmed.is_set(): + return + self._fire() + + def _start(self) -> bool: + thread = threading.Thread( + target=self._run, + daemon=True, + name="gateway-startup-watchdog", + ) + try: + thread.start() + except Exception: + logger.debug("Failed to start gateway startup watchdog", exc_info=True) + return False + self._thread = thread + return True + + +def arm_startup_watchdog( + timeout_s: Optional[float] = None, + *, + exit_code: int = GATEWAY_SERVICE_RESTART_EXIT_CODE, +) -> Optional[StartupWatchdogHandle]: + """Arm the process-wide startup watchdog. Idempotent; never raises. + + Returns the (possibly pre-existing) handle, or ``None`` when disabled via + ``HERMES_STARTUP_WATCHDOG=0`` or when the thread could not be started. + """ + global _handle + try: + if startup_watchdog_disabled(): + return None + with _handle_lock: + if _handle is not None and _handle.is_alive(): + return _handle + resolved = ( + float(timeout_s) + if timeout_s is not None and float(timeout_s) > 0 + else resolve_startup_watchdog_timeout() + ) + handle = StartupWatchdogHandle(resolved, exit_code) + if not handle._start(): + return None + _handle = handle + return handle + except Exception: + logger.debug("Failed to arm gateway startup watchdog", exc_info=True) + return None + + +def disarm_startup_watchdog() -> None: + """Disarm the process-wide startup watchdog, if armed. Never raises.""" + global _handle + try: + with _handle_lock: + handle = _handle + _handle = None + if handle is not None: + handle.disarm() + except Exception: + logger.debug("Failed to disarm gateway startup watchdog", exc_info=True) + + +def _reset_for_tests() -> None: + """Drop the module singleton (test isolation only).""" + global _handle + with _handle_lock: + handle = _handle + _handle = None + if handle is not None: + handle.disarm() diff --git a/hermes_cli/gateway.py b/hermes_cli/gateway.py index dced8278a5b8..af60150dbc6f 100644 --- a/hermes_cli/gateway.py +++ b/hermes_cli/gateway.py @@ -5739,6 +5739,19 @@ def run_gateway(verbose: int = 0, quiet: bool = False, replace: bool = False, fo _guard_existing_gateway_process_conflict(replace=replace) sys.path.insert(0, str(PROJECT_ROOT)) + # Startup-liveness watchdog (OOF-298): armed before config load, imports, + # and DB opens so a deadlock anywhere in the pre-event-loop window still + # gets the process respawned by the service supervisor instead of wedging + # as a live-PID zombie that s6/systemd will never restart. Disarmed by + # GatewayRunner once the event loop is confirmed live. Placed after the + # process-conflict guards: a --replace loser exiting above must not have + # armed a watchdog first. + try: + from gateway.startup_watchdog import arm_startup_watchdog + arm_startup_watchdog() + except Exception: + pass + # Detached Windows gateway runs must ignore console-control broadcasts # from sibling CLI processes, but foreground `hermes gateway run` still # needs to obey the banner's "Press Ctrl+C to stop" contract. diff --git a/tests/gateway/test_startup_watchdog.py b/tests/gateway/test_startup_watchdog.py new file mode 100644 index 000000000000..192c37061f9f --- /dev/null +++ b/tests/gateway/test_startup_watchdog.py @@ -0,0 +1,206 @@ +"""Startup-liveness watchdog tests (OOF-298). + +The watchdog covers the pre-event-loop window: armed at process entry, +disarmed once the gateway's asyncio loop is confirmed live. If neither +happens within the deadline it must dump diagnostics, record a lifecycle +exit, and hard-exit with the service-restart code so the supervisor +respawns the process instead of babysitting a live-PID zombie. +""" + +from __future__ import annotations + +import json +import threading +import time +from pathlib import Path + +import pytest + +import gateway.startup_watchdog as sw +from gateway.restart import GATEWAY_SERVICE_RESTART_EXIT_CODE +from gateway.startup_watchdog import ( + StartupWatchdogHandle, + arm_startup_watchdog, + disarm_startup_watchdog, + get_startup_watchdog_dump_path, + resolve_startup_watchdog_timeout, + startup_watchdog_disabled, +) + + +@pytest.fixture(autouse=True) +def _isolate(tmp_path, monkeypatch): + """Every test gets a fresh singleton and its own HERMES_HOME.""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + monkeypatch.delenv(sw.ENV_STARTUP_WATCHDOG, raising=False) + monkeypatch.delenv(sw.ENV_STARTUP_WATCHDOG_TIMEOUT_S, raising=False) + sw._reset_for_tests() + yield + sw._reset_for_tests() + + +class _ExitCapture: + """Replaces StartupWatchdogHandle._exit so _fire() cannot kill pytest.""" + + def __init__(self): + self.codes: list[int] = [] + self.fired = threading.Event() + + def __call__(self, code: int) -> None: + self.codes.append(code) + self.fired.set() + + +@pytest.fixture +def exit_capture(monkeypatch): + capture = _ExitCapture() + monkeypatch.setattr(StartupWatchdogHandle, "_exit", staticmethod(capture)) + return capture + + +class TestConfigResolution: + def test_default_timeout(self): + assert resolve_startup_watchdog_timeout() == sw.DEFAULT_STARTUP_WATCHDOG_TIMEOUT_S + + def test_env_override(self, monkeypatch): + monkeypatch.setenv(sw.ENV_STARTUP_WATCHDOG_TIMEOUT_S, "120") + assert resolve_startup_watchdog_timeout() == 120.0 + + def test_env_override_clamped_to_floor(self, monkeypatch): + monkeypatch.setenv(sw.ENV_STARTUP_WATCHDOG_TIMEOUT_S, "5") + assert resolve_startup_watchdog_timeout() == sw._MIN_TIMEOUT_S + + def test_garbage_env_falls_back_to_default(self, monkeypatch): + monkeypatch.setenv(sw.ENV_STARTUP_WATCHDOG_TIMEOUT_S, "soon") + assert resolve_startup_watchdog_timeout() == sw.DEFAULT_STARTUP_WATCHDOG_TIMEOUT_S + + def test_nonpositive_env_falls_back_to_default(self, monkeypatch): + monkeypatch.setenv(sw.ENV_STARTUP_WATCHDOG_TIMEOUT_S, "-1") + assert resolve_startup_watchdog_timeout() == sw.DEFAULT_STARTUP_WATCHDOG_TIMEOUT_S + + @pytest.mark.parametrize("raw", ["0", "false", "no", "off", "FALSE", "Off"]) + def test_disabled_values(self, monkeypatch, raw): + monkeypatch.setenv(sw.ENV_STARTUP_WATCHDOG, raw) + assert startup_watchdog_disabled() is True + + @pytest.mark.parametrize("raw", ["", "1", "true", "yes"]) + def test_enabled_values(self, monkeypatch, raw): + monkeypatch.setenv(sw.ENV_STARTUP_WATCHDOG, raw) + assert startup_watchdog_disabled() is False + + +class TestArmDisarm: + def test_arm_returns_live_handle(self): + handle = arm_startup_watchdog(timeout_s=60) + assert handle is not None + assert handle.is_alive() + assert not handle.disarmed + disarm_startup_watchdog() + handle.join(timeout=2) + assert not handle.is_alive() + + def test_arm_is_idempotent(self): + first = arm_startup_watchdog(timeout_s=60) + second = arm_startup_watchdog(timeout_s=60) + assert first is second + disarm_startup_watchdog() + + def test_disarm_prevents_fire(self, exit_capture): + handle = arm_startup_watchdog(timeout_s=0.2) + assert handle is not None + disarm_startup_watchdog() + handle.join(timeout=2) + assert not exit_capture.fired.is_set() + assert exit_capture.codes == [] + + def test_disarm_without_arm_is_safe(self): + disarm_startup_watchdog() # must not raise + + def test_disarm_is_idempotent(self): + arm_startup_watchdog(timeout_s=60) + disarm_startup_watchdog() + disarm_startup_watchdog() # must not raise + + def test_disabled_via_env(self, monkeypatch): + monkeypatch.setenv(sw.ENV_STARTUP_WATCHDOG, "0") + assert arm_startup_watchdog(timeout_s=60) is None + + def test_rearm_after_disarm_starts_fresh_thread(self): + first = arm_startup_watchdog(timeout_s=60) + disarm_startup_watchdog() + first.join(timeout=2) + second = arm_startup_watchdog(timeout_s=60) + assert second is not None + assert second is not first + assert second.is_alive() + disarm_startup_watchdog() + + +class TestFire: + def test_fires_after_deadline_with_restart_code(self, exit_capture, tmp_path): + handle = arm_startup_watchdog(timeout_s=0.1) + assert handle is not None + assert exit_capture.fired.wait(timeout=5) + assert exit_capture.codes == [GATEWAY_SERVICE_RESTART_EXIT_CODE] + + def test_fire_writes_dump_record(self, exit_capture, tmp_path): + arm_startup_watchdog(timeout_s=0.1) + assert exit_capture.fired.wait(timeout=5) + dump_path = get_startup_watchdog_dump_path(tmp_path) + # The record write happens before _exit; poll briefly for the file. + deadline = time.monotonic() + 2 + while not dump_path.exists() and time.monotonic() < deadline: + time.sleep(0.02) + assert dump_path.exists() + record = json.loads(dump_path.read_text(encoding="utf-8").splitlines()[0]) + assert record["tag"] == "startup_watchdog.fired" + assert record["exit_code"] == GATEWAY_SERVICE_RESTART_EXIT_CODE + assert record["timeout_s"] == pytest.approx(0.1) + + def test_fire_marks_lifecycle_exit(self, exit_capture, tmp_path, monkeypatch): + marked = {} + + def _fake_mark_exited(code, reason=None): + marked["code"] = code + marked["reason"] = reason + + import gateway.lifecycle_ledger as ledger + + monkeypatch.setattr(ledger, "mark_exited", _fake_mark_exited) + arm_startup_watchdog(timeout_s=0.1) + assert exit_capture.fired.wait(timeout=5) + # mark_exited runs just before _exit on the same thread; once fired + # is set the _exit stub has returned, so mark_exited already ran. + assert marked == { + "code": GATEWAY_SERVICE_RESTART_EXIT_CODE, + "reason": "startup_liveness_watchdog", + } + + def test_custom_exit_code(self, exit_capture): + arm_startup_watchdog(timeout_s=0.1, exit_code=42) + assert exit_capture.fired.wait(timeout=5) + assert exit_capture.codes == [42] + + +class TestFireTimeoutClamp: + def test_explicit_timeout_below_floor_still_used_directly(self, exit_capture): + """arm_startup_watchdog(timeout_s=...) is a trusted caller/test seam — + it bypasses the env floor clamp so tests stay fast. Only env-provided + values are clamped (they come from operators).""" + handle = arm_startup_watchdog(timeout_s=0.1) + assert handle is not None + assert handle.timeout_s == pytest.approx(0.1) + + +class TestDumpPath: + def test_dump_path_under_home(self, tmp_path): + assert get_startup_watchdog_dump_path(tmp_path) == ( + tmp_path / "logs" / "gateway-startup-watchdog.log" + ) + + def test_dump_write_failure_is_swallowed(self, monkeypatch): + # Point the dump at an unwritable location; must not raise. + monkeypatch.setattr( + sw, "get_startup_watchdog_dump_path", lambda home=None: Path("/dev/null/nope") + ) + sw._write_dump_record({"tag": "x"}) From 500c5eabdd99ec980ec89591ef7df3cf0b50bfa2 Mon Sep 17 00:00:00 2001 From: Shannon Sands Date: Wed, 19 Aug 2026 16:50:19 +1000 Subject: [PATCH 2/5] fix(gateway): address startup-watchdog review findings (OOF-298, PR #89750) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Independent review of the initial startup-liveness watchdog surfaced two P1s and three P2s. All are addressed here. P1 — legitimate slow startups (large state.db schema migrations inside SessionDB.__init__, which run synchronously before the loop starts) could exceed the fixed 300s deadline and restart-loop. The watchdog now checks process CPU time (time.process_time(), process-wide) when the deadline expires: continuous CPU consumption means a live migration, so the deadline is extended (with a warning log per extension). The OOF-298 deadlock class parks every thread in futex waits and accrues ~zero CPU, so it still fires on schedule. Documented limitation: a spinning busy-wait deadlock reads as progress and won't fire — the observed incident class is parked threads. P1 — import-time deadlocks were outside coverage. The implementation moved to a stdlib-only top-level module (hermes_startup_watchdog), and hermes_cli/main.py arms it via an argv fast-path ("gateway" + "run" in argv) BEFORE the heavy module-level import graph. gateway/startup_watchdog remains as a re-export shim so the intuitive import path keeps working for the disarm site, tests, and REPL use. Import-lightness is a correctness property, tested via AST inspection: at fire time the wedged main thread may hold the import lock, so the fire path performs no imports on its own thread — the lifecycle-ledger write runs on a bounded-join helper thread and os._exit happens regardless. P2 — disarm/fire race: the handle now has an explicit state machine (armed → disarmed | firing) guarded by a lock; whichever transition takes the lock first wins, so a disarm landing after deadline expiry but before the fire transition is honored. Regression test forces the exact interleaving by blocking inside the CPU probe. P2 — uncovered entry points: cli.py --gateway and scripts/hermes-gateway run_gateway() now arm the watchdog before importing the gateway graph. hermes_cli/gateway.py run_gateway() keeps an idempotent backstop arm for programmatic callers. P2 — respawn-storm backoff interaction: the storm breaker's intentional backoff sleep (up to minutes, ~zero CPU — indistinguishable from a parked deadlock) now calls kick_startup_watchdog(extra_s=backoff) so the deadline is pushed past the sleep instead of firing mid-backoff. Also: the faulthandler stack dump is now additionally written to logs/gateway-startup-watchdog.log (stderr may be absent on detached/ windowless runs); the disarm site in gateway/run.py moved inside the loop-confirmed branch (if the loop is NOT live, the milestone was not reached and the watchdog must stay armed); hermes_startup_watchdog added to pyproject py-modules so sealed venvs ship it; SERVICE_RESTART_EXIT_CODE is duplicated in the stdlib-only module with a parity test against gateway.restart. Tests: 38 in tests/gateway/test_startup_watchdog.py (contracts incl. stdlib-only AST check and shim re-export identity, config resolution, arm/disarm/kick, CPU-progress extension vs no-progress fire, probe-failure fails toward firing, disarm-vs-fire race, dump record + file stacks, lifecycle ledger, custom exit code). --- cli.py | 7 + gateway/run.py | 37 +- gateway/startup_watchdog.py | 306 ++-------------- hermes_cli/gateway.py | 23 +- hermes_cli/main.py | 18 + hermes_startup_watchdog.py | 471 +++++++++++++++++++++++++ pyproject.toml | 1 + scripts/hermes-gateway | 7 + tests/gateway/test_startup_watchdog.py | 208 +++++++++-- 9 files changed, 756 insertions(+), 322 deletions(-) create mode 100644 hermes_startup_watchdog.py diff --git a/cli.py b/cli.py index b53906386ac3..21d6175c77b9 100644 --- a/cli.py +++ b/cli.py @@ -20919,6 +20919,13 @@ def main( # Handle gateway mode (messaging + cron) if gateway: import asyncio + # Startup-liveness watchdog (OOF-298): this legacy entry point must + # be covered too — arm before importing the gateway graph. + try: + from hermes_startup_watchdog import arm_startup_watchdog + arm_startup_watchdog() + except Exception: + pass from gateway.run import start_gateway print("Starting Hermes Gateway (messaging platforms)...") asyncio.run(start_gateway()) diff --git a/gateway/run.py b/gateway/run.py index 1611c09d9f74..d32c938409de 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -12442,17 +12442,20 @@ async def start(self) -> bool: self._gateway_loop = None if self._gateway_loop is not None: self._start_loop_liveness_guards(self._gateway_loop) - # The event loop is confirmed live: the startup-liveness watchdog's - # job is done and the loop-liveness watchdog (armed just above) - # takes over from here (OOF-298). Disarm even when the loop guards - # are config-disabled — the startup watchdog only covers the - # pre-loop window, never adapter connects or steady-state. - try: - from gateway.startup_watchdog import disarm_startup_watchdog + # The event loop is confirmed live: the startup-liveness + # watchdog's job is done and the loop-liveness watchdog (armed + # just above) takes over from here (OOF-298). Disarm even when + # the loop guards are config-disabled — the startup watchdog + # only covers the pre-loop window, never adapter connects or + # steady-state. Deliberately inside the loop-confirmed branch: + # if the loop somehow isn't live, startup has NOT reached the + # milestone and the watchdog must stay armed. + try: + from gateway.startup_watchdog import disarm_startup_watchdog - disarm_startup_watchdog() - except Exception: - logger.debug("Startup watchdog disarm failed", exc_info=True) + disarm_startup_watchdog() + except Exception: + logger.debug("Startup watchdog disarm failed", exc_info=True) logger.info("Session storage: %s", self.config.sessions_dir) # Sanity-check that systemd's TimeoutStopSec covers our drain @@ -30958,7 +30961,7 @@ def main(): os.environ.setdefault("AI_AGENT", "hermes-agent") os.environ.setdefault("HERMES_AGENT", "true") -# Positive process identity: ledger registration + Windows job-object + # Positive process identity: ledger registration + Windows job-object # self-attach, so update-time reapers can identify this gateway (and its # child tree dies with it on Windows). Best-effort — never blocks startup. try: @@ -30972,11 +30975,13 @@ def main(): except Exception: pass - # Startup-liveness watchdog (OOF-298): armed before ANY other startup - # work — config load, imports, DB opens — so a deadlock anywhere in the - # pre-event-loop window still gets the process respawned by the service - # supervisor instead of wedging as a live-PID zombie. Disarmed by - # GatewayRunner once the event loop is confirmed live. + # Startup-liveness watchdog (OOF-298): armed before config load, DB + # opens, and the rest of pre-loop startup so a deadlock in that window + # still gets the process respawned by the service supervisor instead of + # wedging as a live-PID zombie. (Import-time coverage for the standard + # ``hermes gateway run`` path is provided even earlier, by the argv + # fast-path in hermes_cli.main.) Disarmed by GatewayRunner once the + # event loop is confirmed live. try: from gateway.startup_watchdog import arm_startup_watchdog arm_startup_watchdog() diff --git a/gateway/startup_watchdog.py b/gateway/startup_watchdog.py index eed5fa04eb19..3b114c7b272e 100644 --- a/gateway/startup_watchdog.py +++ b/gateway/startup_watchdog.py @@ -1,274 +1,40 @@ -"""Startup-liveness watchdog — respawn a gateway that wedges before its loop runs (OOF-298). +"""Compatibility shim — the real implementation is ``hermes_startup_watchdog``. -The existing liveness backstops all assume startup succeeded: +The startup-liveness watchdog (OOF-298) must be armable *before* the +``gateway`` package is imported: ``gateway/__init__`` eagerly pulls in the +config/session/delivery graph, and an import-time deadlock is squarely inside +the watchdog's coverage mandate. The implementation therefore lives at the +repository top level as a stdlib-only module. -* the loop-liveness watchdog (:mod:`gateway.shutdown_watchdog`) is armed by - ``GatewayRunner._start_loop_liveness_guards`` — *inside* the running event - loop's startup path; -* the shutdown watchdog is armed at ``stop()``; -* the loop heartbeat file is written by an asyncio task. - -None of them can fire if the process deadlocks **before the event loop comes -alive**. That failure mode is real: OOF-298 documents a hosted gateway whose -process sat for ~30 hours with every thread parked in ``futex_wait_queue``, -zero log lines written, ``/health`` unreachable — while s6 saw a live PID and -therefore never respawned it, and a stale ``gateway_state.json`` from the -*previous* life told every status surface the gateway was "draining". - -This module closes that gap with the simplest thing that works: a plain -daemon OS thread armed at process entry, disarmed the moment the event loop -is confirmed live (the point where the existing loop-liveness watchdog takes -over). If startup neither reaches that milestone nor exits within the -deadline, the watchdog dumps all-thread stacks via ``faulthandler``, records -the exit in the lifecycle ledger (NS-608) so the next boot classifies it -correctly, and ``os._exit``\\ s with the service-restart code so s6/systemd -revive the process instead of babysitting a zombie. - -Deadline rationale: a healthy startup reaches the disarm point in seconds. -The slowest legitimate pre-loop work is MCP tool discovery (bounded 120s -internal wait), so the 300s default leaves comfortable headroom. Platform -adapter connects — which can genuinely take minutes (WhatsApp pairing, npm -cold installs) — happen *after* the disarm point and are never covered by -this watchdog. - -Config surface is deliberately env-only (``HERMES_STARTUP_WATCHDOG=0`` to -disable, ``HERMES_STARTUP_WATCHDOG_TIMEOUT_S`` to tune): the watchdog must be -armed before config.yaml is loaded — a wedge during config parsing is exactly -in scope — so it cannot depend on config for its own enablement. - -Everything here is best-effort: a watchdog failure must never affect the -startup it is observing. +This shim keeps the intuitive ``gateway.startup_watchdog`` import path +working for code that runs after the package is loaded (the disarm site in +``gateway.run``, tests, operators poking at a REPL). """ -from __future__ import annotations - -import faulthandler -import json -import logging -import os -import threading -import time -from datetime import datetime, timezone -from pathlib import Path -from typing import Any, Dict, Optional - -from gateway.restart import GATEWAY_SERVICE_RESTART_EXIT_CODE - -logger = logging.getLogger(__name__) - -DEFAULT_STARTUP_WATCHDOG_TIMEOUT_S = 300.0 -_MIN_TIMEOUT_S = 30.0 - -ENV_STARTUP_WATCHDOG = "HERMES_STARTUP_WATCHDOG" -ENV_STARTUP_WATCHDOG_TIMEOUT_S = "HERMES_STARTUP_WATCHDOG_TIMEOUT_S" - -_DUMP_RELATIVE = ("logs", "gateway-startup-watchdog.log") - -_FALSEY = frozenset({"0", "false", "no", "off"}) - -# Module-level singleton: the arm sites (gateway.run.main and the -# hermes_cli.gateway CLI wrapper) and the disarm site -# (GatewayRunner._start_loop_liveness_guards) have no shared object to hand a -# handle through, and only one gateway startup ever runs per process. -_handle_lock = threading.Lock() -_handle: Optional["StartupWatchdogHandle"] = None - - -def _process_hermes_home() -> Path: - """HERMES_HOME for process-level diagnostic files (ignore task overrides).""" - val = os.environ.get("HERMES_HOME", "").strip() - if val: - return Path(val) - from hermes_constants import get_hermes_home - - return get_hermes_home() - - -def get_startup_watchdog_dump_path(home: Optional[Path] = None) -> Path: - """Return ``/logs/gateway-startup-watchdog.log``.""" - base = home if home is not None else _process_hermes_home() - return base.joinpath(*_DUMP_RELATIVE) - - -def startup_watchdog_disabled() -> bool: - """True when ``HERMES_STARTUP_WATCHDOG`` opts out explicitly.""" - raw = os.environ.get(ENV_STARTUP_WATCHDOG, "").strip().lower() - return raw in _FALSEY - - -def resolve_startup_watchdog_timeout() -> float: - """Deadline in seconds; env override, floor-clamped, default on garbage.""" - raw = os.environ.get(ENV_STARTUP_WATCHDOG_TIMEOUT_S, "").strip() - if not raw: - return DEFAULT_STARTUP_WATCHDOG_TIMEOUT_S - try: - value = float(raw) - except ValueError: - logger.warning( - "Ignoring non-numeric %s=%r; using default %.0fs", - ENV_STARTUP_WATCHDOG_TIMEOUT_S, - raw, - DEFAULT_STARTUP_WATCHDOG_TIMEOUT_S, - ) - return DEFAULT_STARTUP_WATCHDOG_TIMEOUT_S - if value <= 0: - return DEFAULT_STARTUP_WATCHDOG_TIMEOUT_S - return max(value, _MIN_TIMEOUT_S) - - -def _write_dump_record(record: Dict[str, Any]) -> None: - """Append a one-line JSON metadata record beside the faulthandler dump.""" - try: - path = get_startup_watchdog_dump_path() - path.parent.mkdir(parents=True, exist_ok=True) - with open(path, "a", encoding="utf-8") as fh: - fh.write(json.dumps(record, default=str) + "\n") - except Exception: - logger.debug("Failed to write startup watchdog dump record", exc_info=True) - - -class StartupWatchdogHandle: - """Disarm/inspect handle for the armed startup watchdog thread.""" - - def __init__(self, timeout_s: float, exit_code: int): - self.timeout_s = timeout_s - self.exit_code = exit_code - self.armed_at = time.monotonic() - self._disarmed = threading.Event() - self._thread: Optional[threading.Thread] = None - - def disarm(self) -> None: - """Startup reached a live event loop — stand down. Idempotent.""" - self._disarmed.set() - - @property - def disarmed(self) -> bool: - return self._disarmed.is_set() - - def is_alive(self) -> bool: - return self._thread is not None and self._thread.is_alive() - - def join(self, timeout: Optional[float] = None) -> None: - if self._thread is not None: - self._thread.join(timeout=timeout) - - # ── internals ──────────────────────────────────────────────────────── - - def _fire(self) -> None: - elapsed = time.monotonic() - self.armed_at - try: - logger.critical( - "Gateway startup did not reach a live event loop within %.0fs " - "(elapsed %.0fs); dumping all thread stacks and exiting with " - "code %d so the service supervisor can restart it (OOF-298).", - self.timeout_s, - elapsed, - self.exit_code, - ) - except Exception: - pass - _write_dump_record( - { - "ts": datetime.now(timezone.utc).isoformat(), - "tag": "startup_watchdog.fired", - "pid": os.getpid(), - "timeout_s": self.timeout_s, - "elapsed_s": round(elapsed, 3), - "exit_code": self.exit_code, - } - ) - try: - faulthandler.dump_traceback(all_threads=True) - except Exception: - logger.debug("Startup watchdog faulthandler dump failed", exc_info=True) - # Record the exit in the lifecycle sentinel so the next boot reports - # "startup watchdog hard-exit" instead of misclassifying this as an - # unclean SIGKILL/OOM death (NS-608). - try: - from gateway.lifecycle_ledger import mark_exited - - mark_exited(self.exit_code, reason="startup_liveness_watchdog") - except Exception: - pass - self._exit(self.exit_code) - - @staticmethod - def _exit(code: int) -> None: - """Seam for tests; production is a bare ``os._exit``.""" - os._exit(code) - - def _run(self) -> None: - if self._disarmed.wait(timeout=self.timeout_s): - return - if self._disarmed.is_set(): - return - self._fire() - - def _start(self) -> bool: - thread = threading.Thread( - target=self._run, - daemon=True, - name="gateway-startup-watchdog", - ) - try: - thread.start() - except Exception: - logger.debug("Failed to start gateway startup watchdog", exc_info=True) - return False - self._thread = thread - return True - - -def arm_startup_watchdog( - timeout_s: Optional[float] = None, - *, - exit_code: int = GATEWAY_SERVICE_RESTART_EXIT_CODE, -) -> Optional[StartupWatchdogHandle]: - """Arm the process-wide startup watchdog. Idempotent; never raises. - - Returns the (possibly pre-existing) handle, or ``None`` when disabled via - ``HERMES_STARTUP_WATCHDOG=0`` or when the thread could not be started. - """ - global _handle - try: - if startup_watchdog_disabled(): - return None - with _handle_lock: - if _handle is not None and _handle.is_alive(): - return _handle - resolved = ( - float(timeout_s) - if timeout_s is not None and float(timeout_s) > 0 - else resolve_startup_watchdog_timeout() - ) - handle = StartupWatchdogHandle(resolved, exit_code) - if not handle._start(): - return None - _handle = handle - return handle - except Exception: - logger.debug("Failed to arm gateway startup watchdog", exc_info=True) - return None - - -def disarm_startup_watchdog() -> None: - """Disarm the process-wide startup watchdog, if armed. Never raises.""" - global _handle - try: - with _handle_lock: - handle = _handle - _handle = None - if handle is not None: - handle.disarm() - except Exception: - logger.debug("Failed to disarm gateway startup watchdog", exc_info=True) - - -def _reset_for_tests() -> None: - """Drop the module singleton (test isolation only).""" - global _handle - with _handle_lock: - handle = _handle - _handle = None - if handle is not None: - handle.disarm() +from hermes_startup_watchdog import ( # noqa: F401 + DEFAULT_STARTUP_WATCHDOG_TIMEOUT_S, + ENV_STARTUP_WATCHDOG, + ENV_STARTUP_WATCHDOG_TIMEOUT_S, + SERVICE_RESTART_EXIT_CODE, + StartupWatchdogHandle, + arm_startup_watchdog, + disarm_startup_watchdog, + get_startup_watchdog_dump_path, + kick_startup_watchdog, + resolve_startup_watchdog_timeout, + startup_watchdog_disabled, +) + +__all__ = [ + "DEFAULT_STARTUP_WATCHDOG_TIMEOUT_S", + "ENV_STARTUP_WATCHDOG", + "ENV_STARTUP_WATCHDOG_TIMEOUT_S", + "SERVICE_RESTART_EXIT_CODE", + "StartupWatchdogHandle", + "arm_startup_watchdog", + "disarm_startup_watchdog", + "get_startup_watchdog_dump_path", + "kick_startup_watchdog", + "resolve_startup_watchdog_timeout", + "startup_watchdog_disabled", +] diff --git a/hermes_cli/gateway.py b/hermes_cli/gateway.py index af60150dbc6f..27cd5ae2018c 100644 --- a/hermes_cli/gateway.py +++ b/hermes_cli/gateway.py @@ -5739,15 +5739,15 @@ def run_gateway(verbose: int = 0, quiet: bool = False, replace: bool = False, fo _guard_existing_gateway_process_conflict(replace=replace) sys.path.insert(0, str(PROJECT_ROOT)) - # Startup-liveness watchdog (OOF-298): armed before config load, imports, - # and DB opens so a deadlock anywhere in the pre-event-loop window still - # gets the process respawned by the service supervisor instead of wedging - # as a live-PID zombie that s6/systemd will never restart. Disarmed by - # GatewayRunner once the event loop is confirmed live. Placed after the + # Startup-liveness watchdog (OOF-298), idempotent backstop: normal + # ``hermes gateway run`` invocations already armed in hermes_cli.main's + # argv fast-path (before the heavy import graph), but programmatic + # callers can enter run_gateway() directly. Placed after the # process-conflict guards: a --replace loser exiting above must not have - # armed a watchdog first. + # armed a watchdog first. Disarmed by GatewayRunner once the event loop + # is confirmed live. try: - from gateway.startup_watchdog import arm_startup_watchdog + from hermes_startup_watchdog import arm_startup_watchdog arm_startup_watchdog() except Exception: pass @@ -5934,6 +5934,15 @@ def _atexit_hook() -> None: _storm.window_s, _storm.backoff_s, ) + # The backoff sleep is intentional idle time — tell the startup + # watchdog (OOF-298) so it isn't mistaken for a parked deadlock + # and hard-exited mid-backoff (which would defeat the breaker). + try: + from gateway.startup_watchdog import kick_startup_watchdog + + kick_startup_watchdog(extra_s=_storm.backoff_s) + except Exception: + pass _time.sleep(_storm.backoff_s) except Exception as _be: logger.debug("respawn-storm breaker check failed (non-fatal): %s", _be) diff --git a/hermes_cli/main.py b/hermes_cli/main.py index 9f5e1ec93421..57e9e9513eb0 100644 --- a/hermes_cli/main.py +++ b/hermes_cli/main.py @@ -102,6 +102,24 @@ except Exception: pass +# Startup-liveness watchdog (OOF-298): for gateway runs, arm BEFORE the heavy +# module-level import graph below — an import-time deadlock (native-extension +# init, contended import lock) is exactly the "wedged before the event loop, +# no logs, live PID" class this watchdog exists for. ``hermes_startup_watchdog`` +# is stdlib-only, so importing it here cannot itself wedge on application +# code. argv sniffing is deliberately crude: over-arming is harmless (any +# non-gateway command either exits well inside the deadline or... should be +# covered anyway if it wedges), while under-arming recreates OOF-298. +# GatewayRunner disarms once the event loop is confirmed live. +if "gateway" in sys.argv[1:] and "run" in sys.argv[1:]: + try: + from hermes_startup_watchdog import arm_startup_watchdog as _arm_sw + + _arm_sw() + del _arm_sw + except Exception: + pass + def _exit_after_oneshot(rc: object) -> None: """Exit one-shot mode without letting late native finalizers change rc. diff --git a/hermes_startup_watchdog.py b/hermes_startup_watchdog.py new file mode 100644 index 000000000000..1ce307e9a658 --- /dev/null +++ b/hermes_startup_watchdog.py @@ -0,0 +1,471 @@ +"""Startup-liveness watchdog — respawn a gateway that wedges before its loop runs (OOF-298). + +The existing liveness backstops all assume startup succeeded: + +* the loop-liveness watchdog (:mod:`gateway.shutdown_watchdog`) is armed by + ``GatewayRunner._start_loop_liveness_guards`` — *inside* the running event + loop's startup path; +* the shutdown watchdog is armed at ``stop()``; +* the loop heartbeat file is written by an asyncio task. + +None of them can fire if the process deadlocks **before the event loop comes +alive**. That failure mode is real: OOF-298 documents a hosted gateway whose +process sat for ~30 hours with every thread parked in ``futex_wait_queue``, +zero log lines written, ``/health`` unreachable — while s6 saw a live PID and +therefore never respawned it, and a stale ``gateway_state.json`` from the +*previous* life told every status surface the gateway was "draining". + +This module closes that gap with a plain daemon OS thread armed at process +entry, disarmed the moment the event loop is confirmed live (the point where +the existing loop-liveness watchdog takes over). If startup neither reaches +that milestone nor exits within the deadline, the watchdog dumps all-thread +stacks via ``faulthandler``, records the exit in the lifecycle ledger +(NS-608) so the next boot classifies it correctly, and ``os._exit``\\ s with +the service-restart code so s6/systemd revive the process instead of +babysitting a zombie. + +Slow-but-alive startups are NOT killed. Before firing, the watchdog checks +whether the process consumed meaningful CPU time during the expired window +(``time.process_time()`` is process-wide). Long-but-legitimate synchronous +startup work — most importantly large ``state.db`` schema migrations, which +run inside ``SessionDB.__init__`` well before the loop starts and can +genuinely exceed any fixed deadline on multi-GB databases — burns CPU +continuously, so the deadline is extended (with a log line each time) for as +long as progress continues. The OOF-298 deadlock class parks every thread in +futex waits and accrues ~zero CPU, so it still fires on schedule. Known +limitation, documented deliberately: a *spinning* (busy-wait) startup +deadlock reads as CPU progress and will not fire; the observed incident class +is parked-thread deadlocks, which this catches. + +Waits that are idle-by-design get explicit handling instead: + +* the respawn-storm breaker's intentional backoff sleep (up to 300s) calls + :func:`kick_startup_watchdog` with the sleep budget before sleeping; +* MCP tool discovery's internal wait is bounded at 120s, comfortably inside + the 300s default deadline. + +IMPORT-LIGHTNESS IS A CORRECTNESS PROPERTY of this module, not a style +preference. It lives at the repository top level (not inside the ``gateway`` +package) and imports **only stdlib** because: + +1. ``gateway/__init__`` eagerly imports the config/session/delivery graph — + hundreds of modules, DB-adjacent code included. Arming must happen + *before* that graph is imported, or an import-time deadlock (a plausible + shape of "wedged before the loop, no logs") sits outside the watchdog's + coverage. +2. At fire time the main thread may be wedged **holding the import lock**; + any import attempted on the watchdog thread could then block forever. + The fire path therefore performs no imports at all on its own thread — + the lifecycle-ledger write (which does import) runs on a short-lived + helper thread joined with a timeout, and ``os._exit`` happens regardless. + +Config surface is deliberately env-only (``HERMES_STARTUP_WATCHDOG=0`` to +disable, ``HERMES_STARTUP_WATCHDOG_TIMEOUT_S`` to tune): the watchdog must be +armed before config.yaml is loaded — a wedge during config parsing is exactly +in scope — so it cannot depend on config for its own enablement. + +Everything here is best-effort: a watchdog failure must never affect the +startup it is observing. +""" + +from __future__ import annotations + +import faulthandler +import json +import logging +import os +import sys +import threading +import time +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict, Optional + +logger = logging.getLogger(__name__) + +DEFAULT_STARTUP_WATCHDOG_TIMEOUT_S = 300.0 +_MIN_TIMEOUT_S = 30.0 + +# Mirrors gateway.restart.GATEWAY_SERVICE_RESTART_EXIT_CODE. Duplicated here +# (with a parity test in tests/gateway/test_startup_watchdog.py) because this +# module must not import the gateway package — see module docstring. +SERVICE_RESTART_EXIT_CODE = 75 + +ENV_STARTUP_WATCHDOG = "HERMES_STARTUP_WATCHDOG" +ENV_STARTUP_WATCHDOG_TIMEOUT_S = "HERMES_STARTUP_WATCHDOG_TIMEOUT_S" + +_DUMP_RELATIVE = ("logs", "gateway-startup-watchdog.log") + +_FALSEY = frozenset({"0", "false", "no", "off"}) + +# The waiter re-reads its deadline at most this often, so kick_/deadline +# extensions take effect promptly without busy-waiting. +_POLL_SLICE_S = 5.0 + +# Minimum process CPU-time delta (seconds) within one expired deadline window +# for startup to count as "making progress" and earn an extension. A parked +# futex deadlock accrues microseconds; a schema migration accrues orders of +# magnitude more than this per window even on slow disks. +_CPU_PROGRESS_MIN_S = 1.0 + +# How long the fire path waits for the lifecycle-ledger helper thread before +# exiting anyway (the import lock may be held by the wedged main thread). +_LEDGER_JOIN_TIMEOUT_S = 5.0 + +# Handle lifecycle states. Transitions are guarded by the handle's state +# lock so a disarm and a fire can never both "win" (P2 race, PR #89750 +# review): armed -> disarmed (startup reached a live loop) or +# armed -> firing (deadline expired with no CPU progress) — never both. +_ARMED = "armed" +_DISARMED = "disarmed" +_FIRING = "firing" + +# Module-level singleton: the arm sites (hermes_cli.main / hermes_cli.gateway +# / gateway.run.main / cli.py --gateway / scripts/hermes-gateway) and the +# disarm site (GatewayRunner, once the loop is live) have no shared object to +# hand a handle through, and only one gateway startup ever runs per process. +_handle_lock = threading.Lock() +_handle: Optional["StartupWatchdogHandle"] = None + + +def _process_hermes_home() -> Path: + """HERMES_HOME for process-level diagnostic files. + + Stdlib-only replica of ``hermes_constants``' platform default — this + module must not import application code (see module docstring). Hosted + images always set ``HERMES_HOME`` explicitly. + """ + val = os.environ.get("HERMES_HOME", "").strip() + if val: + return Path(val) + if sys.platform == "win32": + local_appdata = os.environ.get("LOCALAPPDATA", "").strip() + base = Path(local_appdata) if local_appdata else Path.home() / "AppData" / "Local" + return base / "hermes" + return Path.home() / ".hermes" + + +def get_startup_watchdog_dump_path(home: Optional[Path] = None) -> Path: + """Return ``/logs/gateway-startup-watchdog.log``.""" + base = home if home is not None else _process_hermes_home() + return base.joinpath(*_DUMP_RELATIVE) + + +def startup_watchdog_disabled() -> bool: + """True when ``HERMES_STARTUP_WATCHDOG`` opts out explicitly.""" + raw = os.environ.get(ENV_STARTUP_WATCHDOG, "").strip().lower() + return raw in _FALSEY + + +def resolve_startup_watchdog_timeout() -> float: + """Deadline in seconds; env override, floor-clamped, default on garbage.""" + raw = os.environ.get(ENV_STARTUP_WATCHDOG_TIMEOUT_S, "").strip() + if not raw: + return DEFAULT_STARTUP_WATCHDOG_TIMEOUT_S + try: + value = float(raw) + except ValueError: + logger.warning( + "Ignoring non-numeric %s=%r; using default %.0fs", + ENV_STARTUP_WATCHDOG_TIMEOUT_S, + raw, + DEFAULT_STARTUP_WATCHDOG_TIMEOUT_S, + ) + return DEFAULT_STARTUP_WATCHDOG_TIMEOUT_S + if value <= 0: + return DEFAULT_STARTUP_WATCHDOG_TIMEOUT_S + return max(value, _MIN_TIMEOUT_S) + + +def _write_dump_record(record: Dict[str, Any]) -> None: + """Append a one-line JSON metadata record beside the faulthandler dump.""" + try: + path = get_startup_watchdog_dump_path() + path.parent.mkdir(parents=True, exist_ok=True) + with open(path, "a", encoding="utf-8") as fh: + fh.write(json.dumps(record, default=str) + "\n") + except Exception: + logger.debug("Failed to write startup watchdog dump record", exc_info=True) + + +def _mark_lifecycle_exit(exit_code: int) -> None: + """Record the watchdog exit in the NS-608 lifecycle sentinel. + + Runs on a dedicated helper thread (see ``_fire``): the ``import`` below + can block indefinitely on the interpreter import lock if the wedged main + thread holds it, and the fire path must reach ``os._exit`` regardless. + """ + try: + from gateway.lifecycle_ledger import mark_exited + + mark_exited(exit_code, reason="startup_liveness_watchdog") + except Exception: + pass + + +class StartupWatchdogHandle: + """Disarm/inspect handle for the armed startup watchdog thread.""" + + def __init__(self, timeout_s: float, exit_code: int): + self.timeout_s = timeout_s + self.exit_code = exit_code + self.armed_at = time.monotonic() + self._state = _ARMED + self._state_lock = threading.Lock() + self._deadline = self.armed_at + timeout_s + self._disarmed_event = threading.Event() + self._thread: Optional[threading.Thread] = None + self._extensions = 0 + + def disarm(self) -> None: + """Startup reached a live event loop — stand down. Idempotent. + + Atomic with respect to firing: whichever of disarm/fire takes the + state lock first wins, so a disarm that lands before the fire + sequence begins is always honored (never lost to a deadline that + expired concurrently). + """ + with self._state_lock: + if self._state == _ARMED: + self._state = _DISARMED + self._disarmed_event.set() + + def kick(self, extra_s: float = 0.0) -> None: + """Push the deadline out to ``now + timeout + extra_s``. + + For call sites that are about to block intentionally with ~zero CPU + activity (the respawn-storm breaker's backoff sleep), which would + otherwise be indistinguishable from a parked deadlock. + """ + try: + extra = max(0.0, float(extra_s)) + except (TypeError, ValueError): + extra = 0.0 + with self._state_lock: + self._deadline = time.monotonic() + self.timeout_s + extra + + @property + def disarmed(self) -> bool: + return self._state == _DISARMED + + def is_alive(self) -> bool: + return self._thread is not None and self._thread.is_alive() + + def join(self, timeout: Optional[float] = None) -> None: + if self._thread is not None: + self._thread.join(timeout=timeout) + + # ── internals ──────────────────────────────────────────────────────── + + @staticmethod + def _process_cpu_seconds() -> Optional[float]: + """Process-wide CPU time (user+system, all threads); None on failure.""" + try: + return time.process_time() + except Exception: + return None + + def _fire(self) -> None: + elapsed = time.monotonic() - self.armed_at + try: + logger.critical( + "Gateway startup did not reach a live event loop within %.0fs " + "(elapsed %.0fs, %d extension(s)) and shows no CPU progress; " + "dumping all thread stacks and exiting with code %d so the " + "service supervisor can restart it (OOF-298).", + self.timeout_s, + elapsed, + self._extensions, + self.exit_code, + ) + except Exception: + pass + _write_dump_record( + { + "ts": datetime.now(timezone.utc).isoformat(), + "tag": "startup_watchdog.fired", + "pid": os.getpid(), + "timeout_s": self.timeout_s, + "elapsed_s": round(elapsed, 3), + "extensions": self._extensions, + "exit_code": self.exit_code, + } + ) + try: + faulthandler.dump_traceback(all_threads=True) + except Exception: + logger.debug("Startup watchdog faulthandler dump failed", exc_info=True) + # Also dump stacks into the log file: on detached/windowless runs + # (pythonw, some service managers) stderr may be absent, and the + # whole point of firing is to leave forensics behind. + try: + path = get_startup_watchdog_dump_path() + path.parent.mkdir(parents=True, exist_ok=True) + with open(path, "a", encoding="utf-8") as fh: + faulthandler.dump_traceback(file=fh, all_threads=True) + except Exception: + logger.debug( + "Startup watchdog file-based faulthandler dump failed", exc_info=True + ) + # Lifecycle-ledger write on a helper thread: it imports application + # code, and the wedged main thread may hold the import lock. Bounded + # join, then exit regardless (NS-608 classification is best-effort; + # the respawn is not). + try: + ledger_thread = threading.Thread( + target=_mark_lifecycle_exit, + args=(self.exit_code,), + daemon=True, + name="gateway-startup-watchdog-ledger", + ) + ledger_thread.start() + ledger_thread.join(timeout=_LEDGER_JOIN_TIMEOUT_S) + except Exception: + pass + self._exit(self.exit_code) + + @staticmethod + def _exit(code: int) -> None: + """Seam for tests; production is a bare ``os._exit``.""" + os._exit(code) + + def _run(self) -> None: + last_cpu = self._process_cpu_seconds() + while True: + with self._state_lock: + if self._state != _ARMED: + return + deadline = self._deadline + remaining = deadline - time.monotonic() + if remaining > 0: + if self._disarmed_event.wait(timeout=min(remaining, _POLL_SLICE_S)): + return + continue + # Deadline expired. A slow-but-alive startup (large state.db + # schema migration inside SessionDB.__init__) burns CPU + # continuously; a parked futex deadlock accrues ~none. Extend + # for the former, fire only for the latter. + cpu = self._process_cpu_seconds() + if ( + cpu is not None + and last_cpu is not None + and (cpu - last_cpu) >= _CPU_PROGRESS_MIN_S + ): + window_delta = cpu - last_cpu + last_cpu = cpu + self._extensions += 1 + with self._state_lock: + if self._state != _ARMED: + return + self._deadline = time.monotonic() + self.timeout_s + try: + logger.warning( + "Gateway startup exceeded %.0fs but is consuming CPU " + "(%.1fs this window) — likely a long schema migration; " + "extending the startup watchdog deadline (extension #%d).", + self.timeout_s, + window_delta, + self._extensions, + ) + except Exception: + pass + continue + # No progress: claim the fire transition atomically so a disarm + # racing this exact moment can still win if it gets there first. + with self._state_lock: + if self._state != _ARMED: + return + self._state = _FIRING + self._fire() + return + + def _start(self) -> bool: + thread = threading.Thread( + target=self._run, + daemon=True, + name="gateway-startup-watchdog", + ) + try: + thread.start() + except Exception: + logger.debug("Failed to start gateway startup watchdog", exc_info=True) + return False + self._thread = thread + return True + + +def arm_startup_watchdog( + timeout_s: Optional[float] = None, + *, + exit_code: int = SERVICE_RESTART_EXIT_CODE, +) -> Optional[StartupWatchdogHandle]: + """Arm the process-wide startup watchdog. Idempotent; never raises. + + Returns the (possibly pre-existing) handle, or ``None`` when disabled via + ``HERMES_STARTUP_WATCHDOG=0`` or when the thread could not be started. + """ + global _handle + try: + if startup_watchdog_disabled(): + return None + with _handle_lock: + if _handle is not None and _handle.is_alive(): + return _handle + resolved = ( + float(timeout_s) + if timeout_s is not None and float(timeout_s) > 0 + else resolve_startup_watchdog_timeout() + ) + handle = StartupWatchdogHandle(resolved, exit_code) + if not handle._start(): + return None + _handle = handle + return handle + except Exception: + logger.debug("Failed to arm gateway startup watchdog", exc_info=True) + return None + + +def disarm_startup_watchdog() -> None: + """Disarm the process-wide startup watchdog, if armed. Never raises. + + The handle's ``disarm()`` is called while still holding the singleton + lock — it is non-blocking, and holding the lock closes the window where + a concurrent re-arm could swap in a new handle that the disarm then + misses. + """ + global _handle + try: + with _handle_lock: + handle = _handle + _handle = None + if handle is not None: + handle.disarm() + except Exception: + logger.debug("Failed to disarm gateway startup watchdog", exc_info=True) + + +def kick_startup_watchdog(extra_s: float = 0.0) -> None: + """Extend the armed watchdog's deadline. No-op when not armed; never raises. + + Call before intentionally blocking with ~zero CPU activity (e.g. the + respawn-storm breaker's backoff sleep) so the idle wait is not mistaken + for a parked deadlock. + """ + try: + with _handle_lock: + handle = _handle + if handle is not None: + handle.kick(extra_s) + except Exception: + logger.debug("Failed to kick gateway startup watchdog", exc_info=True) + + +def _reset_for_tests() -> None: + """Drop the module singleton (test isolation only).""" + global _handle + with _handle_lock: + handle = _handle + _handle = None + if handle is not None: + handle.disarm() diff --git a/pyproject.toml b/pyproject.toml index 863115484515..4cfa95995b7d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -435,6 +435,7 @@ py-modules = [ "hermes_state_portability", "hermes_state_schema", "hermes_state_search", + "hermes_startup_watchdog", "hermes_time", "hermes_logging", "utils", diff --git a/scripts/hermes-gateway b/scripts/hermes-gateway index b0d45810e3a6..2d39cfbe9d43 100755 --- a/scripts/hermes-gateway +++ b/scripts/hermes-gateway @@ -294,6 +294,13 @@ def is_windows() -> bool: def run_gateway(): """Run the gateway in foreground.""" + # Startup-liveness watchdog (OOF-298): arm before importing the gateway + # graph so an import-time or pre-loop deadlock still gets respawned. + try: + from hermes_startup_watchdog import arm_startup_watchdog + arm_startup_watchdog() + except Exception: + pass from gateway.run import start_gateway print("Starting Hermes Gateway...") print("Press Ctrl+C to stop.") diff --git a/tests/gateway/test_startup_watchdog.py b/tests/gateway/test_startup_watchdog.py index 192c37061f9f..d4fb6671c431 100644 --- a/tests/gateway/test_startup_watchdog.py +++ b/tests/gateway/test_startup_watchdog.py @@ -1,10 +1,14 @@ """Startup-liveness watchdog tests (OOF-298). -The watchdog covers the pre-event-loop window: armed at process entry, -disarmed once the gateway's asyncio loop is confirmed live. If neither -happens within the deadline it must dump diagnostics, record a lifecycle -exit, and hard-exit with the service-restart code so the supervisor -respawns the process instead of babysitting a live-PID zombie. +The watchdog covers the pre-event-loop window: armed at process entry +(before the gateway package imports — the implementation is the stdlib-only +top-level module ``hermes_startup_watchdog``; ``gateway.startup_watchdog`` +is a re-export shim), disarmed once the gateway's asyncio loop is confirmed +live. If neither happens within the deadline — and the process shows no CPU +progress, so slow-but-alive schema migrations are exempt — it must dump +diagnostics, record a lifecycle exit, and hard-exit with the service-restart +code so the supervisor respawns the process instead of babysitting a +live-PID zombie. """ from __future__ import annotations @@ -16,13 +20,14 @@ import pytest -import gateway.startup_watchdog as sw -from gateway.restart import GATEWAY_SERVICE_RESTART_EXIT_CODE -from gateway.startup_watchdog import ( +import hermes_startup_watchdog as sw +from hermes_startup_watchdog import ( + SERVICE_RESTART_EXIT_CODE, StartupWatchdogHandle, arm_startup_watchdog, disarm_startup_watchdog, get_startup_watchdog_dump_path, + kick_startup_watchdog, resolve_startup_watchdog_timeout, startup_watchdog_disabled, ) @@ -39,6 +44,18 @@ def _isolate(tmp_path, monkeypatch): sw._reset_for_tests() +@pytest.fixture(autouse=True) +def _no_cpu_progress(monkeypatch): + """Freeze process CPU time so fire tests never see 'progress'. + + Individual tests that exercise the CPU-progress extension override this + with their own sequence. + """ + monkeypatch.setattr( + StartupWatchdogHandle, "_process_cpu_seconds", staticmethod(lambda: 0.0) + ) + + class _ExitCapture: """Replaces StartupWatchdogHandle._exit so _fire() cannot kill pytest.""" @@ -58,6 +75,56 @@ def exit_capture(monkeypatch): return capture +class TestContracts: + def test_restart_code_parity_with_gateway_restart(self): + """The stdlib-only module duplicates the exit-code constant; keep it + in lockstep with the canonical gateway.restart definition.""" + from gateway.restart import GATEWAY_SERVICE_RESTART_EXIT_CODE + + assert SERVICE_RESTART_EXIT_CODE == GATEWAY_SERVICE_RESTART_EXIT_CODE + + def test_gateway_shim_reexports_same_objects(self): + import gateway.startup_watchdog as shim + + assert shim.arm_startup_watchdog is arm_startup_watchdog + assert shim.disarm_startup_watchdog is disarm_startup_watchdog + assert shim.kick_startup_watchdog is kick_startup_watchdog + + def test_implementation_module_is_stdlib_only(self): + """Import-lightness is a correctness property (arm-before-imports, + no import-lock dependence at fire time): the implementation module + must not import the gateway/agent/hermes_cli graphs at module level.""" + import ast + import inspect + + source = inspect.getsource(sw) + tree = ast.parse(source) + forbidden_roots = { + "gateway", + "agent", + "hermes_cli", + "hermes_state", + "hermes_constants", + "tools", + "plugins", + } + offenders = [] + for node in ast.walk(tree): + # Only module-level and unconditional imports matter; function- + # bodied imports (the ledger helper) are deliberate and guarded. + if isinstance(node, ast.Import): + names = [alias.name for alias in node.names] + elif isinstance(node, ast.ImportFrom): + names = [node.module or ""] + else: + continue + for name in names: + root = name.split(".")[0] + if root in forbidden_roots and node.col_offset == 0: + offenders.append(name) + assert offenders == [] + + class TestConfigResolution: def test_default_timeout(self): assert resolve_startup_watchdog_timeout() == sw.DEFAULT_STARTUP_WATCHDOG_TIMEOUT_S @@ -135,29 +202,123 @@ def test_rearm_after_disarm_starts_fresh_thread(self): assert second.is_alive() disarm_startup_watchdog() + def test_disarm_after_deadline_expiry_wins_over_fire(self, exit_capture, monkeypatch): + """The P2 race from review: deadline expires, but disarm lands before + the fire transition claims the state — the disarm must win. We force + the interleaving by blocking the watchdog thread inside the CPU probe + (which runs after deadline expiry, before the fire transition). The + probe is also called once at thread start for the baseline, so only + the second call blocks.""" + in_probe = threading.Event() + release_probe = threading.Event() + calls = {"n": 0} + + def _blocking_probe(): + calls["n"] += 1 + if calls["n"] >= 2: + in_probe.set() + release_probe.wait(timeout=10) + return 0.0 + + monkeypatch.setattr( + StartupWatchdogHandle, + "_process_cpu_seconds", + staticmethod(_blocking_probe), + ) + handle = arm_startup_watchdog(timeout_s=0.1) + assert handle is not None + # Wait until the deadline has expired and the thread is inside the + # probe (post-expiry, pre-fire-transition). + assert in_probe.wait(timeout=5) + disarm_startup_watchdog() + release_probe.set() + handle.join(timeout=5) + assert not exit_capture.fired.is_set() + assert exit_capture.codes == [] + + +class TestKick: + def test_kick_extends_deadline(self, exit_capture): + handle = arm_startup_watchdog(timeout_s=0.3) + assert handle is not None + # Kick far enough out that the original 0.3s deadline can't fire + # while we watch. + kick_startup_watchdog(extra_s=60) + time.sleep(0.6) + assert not exit_capture.fired.is_set() + disarm_startup_watchdog() + + def test_kick_without_arm_is_safe(self): + kick_startup_watchdog(extra_s=30) # must not raise + + def test_kick_with_garbage_extra_is_safe(self): + arm_startup_watchdog(timeout_s=60) + kick_startup_watchdog(extra_s="nonsense") # type: ignore[arg-type] + disarm_startup_watchdog() + + +class TestCpuProgressExtension: + def test_cpu_progress_extends_instead_of_firing(self, exit_capture, monkeypatch): + """A long schema migration burns CPU: the watchdog must extend, not + fire (the P1 false-fire/restart-loop case from review).""" + # Each probe call reports +10s CPU — always 'progress'. + counter = {"cpu": 0.0} + + def _busy_probe(): + counter["cpu"] += 10.0 + return counter["cpu"] + + monkeypatch.setattr( + StartupWatchdogHandle, "_process_cpu_seconds", staticmethod(_busy_probe) + ) + handle = arm_startup_watchdog(timeout_s=0.1) + assert handle is not None + time.sleep(0.6) + assert not exit_capture.fired.is_set() + assert handle._extensions >= 1 + disarm_startup_watchdog() + + def test_no_cpu_progress_fires(self, exit_capture): + # autouse fixture pins CPU time at 0.0 — no progress. + arm_startup_watchdog(timeout_s=0.1) + assert exit_capture.fired.wait(timeout=5) + assert exit_capture.codes == [SERVICE_RESTART_EXIT_CODE] + + def test_probe_failure_fails_toward_firing(self, exit_capture, monkeypatch): + """If CPU time can't be read the watchdog must still fire on a real + deadlock rather than extending forever.""" + monkeypatch.setattr( + StartupWatchdogHandle, "_process_cpu_seconds", staticmethod(lambda: None) + ) + arm_startup_watchdog(timeout_s=0.1) + assert exit_capture.fired.wait(timeout=5) + class TestFire: - def test_fires_after_deadline_with_restart_code(self, exit_capture, tmp_path): + def test_fires_after_deadline_with_restart_code(self, exit_capture): handle = arm_startup_watchdog(timeout_s=0.1) assert handle is not None assert exit_capture.fired.wait(timeout=5) - assert exit_capture.codes == [GATEWAY_SERVICE_RESTART_EXIT_CODE] + assert exit_capture.codes == [SERVICE_RESTART_EXIT_CODE] - def test_fire_writes_dump_record(self, exit_capture, tmp_path): + def test_fire_writes_dump_record_and_stacks(self, exit_capture, tmp_path): arm_startup_watchdog(timeout_s=0.1) assert exit_capture.fired.wait(timeout=5) dump_path = get_startup_watchdog_dump_path(tmp_path) - # The record write happens before _exit; poll briefly for the file. deadline = time.monotonic() + 2 while not dump_path.exists() and time.monotonic() < deadline: time.sleep(0.02) assert dump_path.exists() - record = json.loads(dump_path.read_text(encoding="utf-8").splitlines()[0]) + content = dump_path.read_text(encoding="utf-8") + record = json.loads(content.splitlines()[0]) assert record["tag"] == "startup_watchdog.fired" - assert record["exit_code"] == GATEWAY_SERVICE_RESTART_EXIT_CODE + assert record["exit_code"] == SERVICE_RESTART_EXIT_CODE assert record["timeout_s"] == pytest.approx(0.1) + # File-based faulthandler dump follows the JSON record (stderr may be + # absent on detached runs). + assert "Thread" in content or "Current thread" in content - def test_fire_marks_lifecycle_exit(self, exit_capture, tmp_path, monkeypatch): + def test_fire_marks_lifecycle_exit(self, exit_capture, monkeypatch): marked = {} def _fake_mark_exited(code, reason=None): @@ -169,10 +330,10 @@ def _fake_mark_exited(code, reason=None): monkeypatch.setattr(ledger, "mark_exited", _fake_mark_exited) arm_startup_watchdog(timeout_s=0.1) assert exit_capture.fired.wait(timeout=5) - # mark_exited runs just before _exit on the same thread; once fired - # is set the _exit stub has returned, so mark_exited already ran. + # The ledger write runs on a helper thread joined (with timeout) + # before _exit; once fired is set the join already happened. assert marked == { - "code": GATEWAY_SERVICE_RESTART_EXIT_CODE, + "code": SERVICE_RESTART_EXIT_CODE, "reason": "startup_liveness_watchdog", } @@ -182,16 +343,6 @@ def test_custom_exit_code(self, exit_capture): assert exit_capture.codes == [42] -class TestFireTimeoutClamp: - def test_explicit_timeout_below_floor_still_used_directly(self, exit_capture): - """arm_startup_watchdog(timeout_s=...) is a trusted caller/test seam — - it bypasses the env floor clamp so tests stay fast. Only env-provided - values are clamped (they come from operators).""" - handle = arm_startup_watchdog(timeout_s=0.1) - assert handle is not None - assert handle.timeout_s == pytest.approx(0.1) - - class TestDumpPath: def test_dump_path_under_home(self, tmp_path): assert get_startup_watchdog_dump_path(tmp_path) == ( @@ -199,7 +350,6 @@ def test_dump_path_under_home(self, tmp_path): ) def test_dump_write_failure_is_swallowed(self, monkeypatch): - # Point the dump at an unwritable location; must not raise. monkeypatch.setattr( sw, "get_startup_watchdog_dump_path", lambda home=None: Path("/dev/null/nope") ) From 5b793fb75cf4cd7f42108fcd8a8622d8be961320 Mon Sep 17 00:00:00 2001 From: Shannon Sands Date: Thu, 20 Aug 2026 12:09:12 +1000 Subject: [PATCH 3/5] fix(startup-watchdog): bounded hard-exit escort + phase-owned progress leases MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses the two class-level review blockers on PR #89750: 1. Bounded hard-exit seam (escort thread). The forensic fire path (logger.critical, dump record, faulthandler, lifecycle ledger) can itself wedge — the parked main thread may hold the logging handler lock, or the disk may be full/hung. _fire() now starts an exit-escort daemon thread BEFORE any forensics; it is free of log handlers, filesystem access, module loads and application locks, and hard-exits with the restart code after _FIRE_EXIT_BOUND_S unless the normal fire path signals completion. Adversarial tests hold the logging handler lock / hang the dump write at fire time and assert the exit seam is still reached. 2. Phase-owned progress leases (report_startup_progress). Process CPU time proves process activity, not startup progress: an unrelated busy thread could extend forever while startup sits parked (false negative), and I/O-bound repair/backup accrues ~zero CPU and would be killed (false positive). Long synchronous startup phases now declare authoritative, clamped (_MAX_LEASE_S), renewable progress leases: state.db _init_schema + the version-gated data-migration chain (hermes_state_schema) and repair_state_db_schema (hermes_state) are wired. CPU progress remains only as a bounded fallback, capped at _MAX_CPU_EXTENSIONS, with leases outranking the cap. Adversarial tests cover both directions (lease saves zero-CPU legitimate work; capped CPU noise no longer hides a parked deadlock). Fire-path dump record now includes lease_count/last_lease_phase for forensics. gateway/startup_watchdog.py shim re-exports report_startup_progress. OOF-298 --- gateway/startup_watchdog.py | 2 + hermes_startup_watchdog.py | 211 +++++++++++++++++++++--- hermes_state.py | 6 + hermes_state_schema.py | 11 ++ tests/gateway/test_startup_watchdog.py | 218 ++++++++++++++++++++++++- 5 files changed, 423 insertions(+), 25 deletions(-) diff --git a/gateway/startup_watchdog.py b/gateway/startup_watchdog.py index 3b114c7b272e..a7fe9abc24ee 100644 --- a/gateway/startup_watchdog.py +++ b/gateway/startup_watchdog.py @@ -21,6 +21,7 @@ disarm_startup_watchdog, get_startup_watchdog_dump_path, kick_startup_watchdog, + report_startup_progress, resolve_startup_watchdog_timeout, startup_watchdog_disabled, ) @@ -35,6 +36,7 @@ "disarm_startup_watchdog", "get_startup_watchdog_dump_path", "kick_startup_watchdog", + "report_startup_progress", "resolve_startup_watchdog_timeout", "startup_watchdog_disabled", ] diff --git a/hermes_startup_watchdog.py b/hermes_startup_watchdog.py index 1ce307e9a658..52a3bd0556bc 100644 --- a/hermes_startup_watchdog.py +++ b/hermes_startup_watchdog.py @@ -24,18 +24,29 @@ the service-restart code so s6/systemd revive the process instead of babysitting a zombie. -Slow-but-alive startups are NOT killed. Before firing, the watchdog checks -whether the process consumed meaningful CPU time during the expired window -(``time.process_time()`` is process-wide). Long-but-legitimate synchronous -startup work — most importantly large ``state.db`` schema migrations, which -run inside ``SessionDB.__init__`` well before the loop starts and can -genuinely exceed any fixed deadline on multi-GB databases — burns CPU -continuously, so the deadline is extended (with a log line each time) for as -long as progress continues. The OOF-298 deadlock class parks every thread in -futex waits and accrues ~zero CPU, so it still fires on schedule. Known -limitation, documented deliberately: a *spinning* (busy-wait) startup -deadlock reads as CPU progress and will not fire; the observed incident class -is parked-thread deadlocks, which this catches. +Slow-but-alive startups are NOT killed. Two mechanisms, in order of +authority: + +1. **Phase-owned progress leases** (:func:`report_startup_progress`): a + startup phase that is about to do legitimately long synchronous work + (large ``state.db`` schema migrations, corruption repair/backup — both + run inside ``SessionDB.__init__`` well before the loop starts, and both + can be I/O-bound with near-zero CPU) declares a lease for its honest + worst case. The lease is the authoritative signal: it proves the + *startup path itself* is alive, not merely that the process is warm. +2. **CPU progress, as a bounded fallback only**: if the deadline expires + but the process consumed meaningful CPU during the window + (``time.process_time()`` is process-wide), the deadline is extended — + at most ``_MAX_CPU_EXTENSIONS`` times. Process-wide CPU proves activity, + not startup progress (an unrelated daemon thread burning CPU must not + hide a parked startup thread forever), hence the cap. Phases that hold + a current lease are never subject to the cap. + +The OOF-298 deadlock class parks every thread in futex waits, accrues ~zero +CPU, and owns no lease — it fires on schedule. Known limitation, documented +deliberately: a *spinning* (busy-wait) startup deadlock reads as CPU +progress and gets the capped extensions before firing; the observed +incident class is parked-thread deadlocks, which fire immediately. Waits that are idle-by-design get explicit handling instead: @@ -103,15 +114,34 @@ _POLL_SLICE_S = 5.0 # Minimum process CPU-time delta (seconds) within one expired deadline window -# for startup to count as "making progress" and earn an extension. A parked -# futex deadlock accrues microseconds; a schema migration accrues orders of -# magnitude more than this per window even on slow disks. +# for startup to count as "making progress" and earn a fallback extension. A +# parked futex deadlock accrues microseconds; a schema migration accrues +# orders of magnitude more than this per window even on slow disks. _CPU_PROGRESS_MIN_S = 1.0 +# Hard cap on CPU-fallback extensions. CPU is process-wide evidence and can +# be produced by threads unrelated to startup, so it may only stretch the +# runway to (1 + cap) x timeout; anything longer must hold an explicit +# phase lease (report_startup_progress). 3 x 300s default = 20min total. +_MAX_CPU_EXTENSIONS = 3 + +# Per-call clamp on progress leases (report_startup_progress). A phase that +# genuinely needs longer renews its lease — the renewal is itself the +# liveness evidence. 15 minutes covers the observed worst-case single +# migration step on multi-GB state.db files with generous margin. +_MAX_LEASE_S = 900.0 + # How long the fire path waits for the lifecycle-ledger helper thread before # exiting anyway (the import lock may be held by the wedged main thread). _LEDGER_JOIN_TIMEOUT_S = 5.0 +# Upper bound on the ENTIRE forensic fire path (logging, dump record, +# faulthandler, ledger). A sibling escort thread — which touches no logging, +# no filesystem, and no application locks — hard-exits the process if the +# forensics wedge (e.g. the wedged main thread holds the logging handler +# lock, or the disk is full/hung). Must exceed _LEDGER_JOIN_TIMEOUT_S. +_FIRE_EXIT_BOUND_S = 10.0 + # Handle lifecycle states. Transitions are guarded by the handle's state # lock so a disarm and a fire can never both "win" (P2 race, PR #89750 # review): armed -> disarmed (startup reached a live loop) or @@ -216,6 +246,14 @@ def __init__(self, timeout_s: float, exit_code: int): self._disarmed_event = threading.Event() self._thread: Optional[threading.Thread] = None self._extensions = 0 + # Phase-owned progress lease (see lease()). monotonic deadline the + # current startup phase has claimed for legitimately long sync work. + self._lease_until = 0.0 + self._lease_phase: Optional[str] = None + self._lease_count = 0 + # Set by _fire() once forensics complete; the exit escort thread + # uses it to stand down when the normal exit path won the race. + self._fire_done = threading.Event() def disarm(self) -> None: """Startup reached a live event loop — stand down. Idempotent. @@ -244,6 +282,32 @@ def kick(self, extra_s: float = 0.0) -> None: with self._state_lock: self._deadline = time.monotonic() + self.timeout_s + extra + def lease(self, expected_s: float, phase: str = "") -> None: + """Claim a progress lease: this startup phase is alive and expects + up to ``expected_s`` more seconds of legitimate synchronous work. + + This is the authoritative "still making progress" signal — unlike + process-wide CPU time it is owned by the startup path itself, so it + works for I/O-bound phases (corruption repair, backups) that accrue + almost no CPU, and it cannot be counterfeited by unrelated threads. + + Leases are clamped to ``_MAX_LEASE_S`` per call so a single buggy + caller cannot silence the watchdog indefinitely; genuinely long + phases renew periodically (renewal proves continued liveness). + Never raises.""" + try: + expected = float(expected_s) + except (TypeError, ValueError): + return + if expected <= 0: + return + expected = min(expected, _MAX_LEASE_S) + with self._state_lock: + self._lease_until = max(self._lease_until, time.monotonic() + expected) + if phase: + self._lease_phase = str(phase) + self._lease_count += 1 + @property def disarmed(self) -> bool: return self._state == _DISARMED @@ -266,13 +330,33 @@ def _process_cpu_seconds() -> Optional[float]: return None def _fire(self) -> None: + """Forensics, then exit — with the exit itself independently bounded. + + Everything in here that produces forensics (logging, the JSON dump + record, faulthandler, the lifecycle ledger) can in principle block: + the wedged main thread may hold the logging handler lock, the disk + may be full or hung. None of that may stop the respawn. An escort + thread is started FIRST; it touches no logging, no filesystem and + no application locks — it sleeps, checks whether the normal exit + happened, and otherwise calls the exit seam itself. ``os._exit`` + is async-signal-safe and lock-free by design.""" + try: + escort = threading.Thread( + target=self._exit_escort, + daemon=True, + name="gateway-startup-watchdog-exit-escort", + ) + escort.start() + except Exception: + pass elapsed = time.monotonic() - self.armed_at try: logger.critical( "Gateway startup did not reach a live event loop within %.0fs " - "(elapsed %.0fs, %d extension(s)) and shows no CPU progress; " - "dumping all thread stacks and exiting with code %d so the " - "service supervisor can restart it (OOF-298).", + "(elapsed %.0fs, %d extension(s)), holds no progress lease " + "and shows no CPU progress; dumping all thread stacks and " + "exiting with code %d so the service supervisor can restart " + "it (OOF-298).", self.timeout_s, elapsed, self._extensions, @@ -288,6 +372,8 @@ def _fire(self) -> None: "timeout_s": self.timeout_s, "elapsed_s": round(elapsed, 3), "extensions": self._extensions, + "lease_count": self._lease_count, + "last_lease_phase": self._lease_phase, "exit_code": self.exit_code, } ) @@ -322,8 +408,25 @@ def _fire(self) -> None: ledger_thread.join(timeout=_LEDGER_JOIN_TIMEOUT_S) except Exception: pass + self._fire_done.set() self._exit(self.exit_code) + def _exit_escort(self) -> None: + """Hard-exit if the forensic fire path wedges (bounded-exit seam). + + Deliberately free of log handlers, filesystem access, module loads + and any lock shared with application code: its only dependencies + are a monotonic sleep, an Event check, and the exit seam.""" + self._sleep(_FIRE_EXIT_BOUND_S) + if self._fire_done.is_set(): + return + self._exit(self.exit_code) + + @staticmethod + def _sleep(seconds: float) -> None: + """Seam for tests; production is a bare ``time.sleep``.""" + time.sleep(seconds) + @staticmethod def _exit(code: int) -> None: """Seam for tests; production is a bare ``os._exit``.""" @@ -341,15 +444,48 @@ def _run(self) -> None: if self._disarmed_event.wait(timeout=min(remaining, _POLL_SLICE_S)): return continue - # Deadline expired. A slow-but-alive startup (large state.db - # schema migration inside SessionDB.__init__) burns CPU - # continuously; a parked futex deadlock accrues ~none. Extend - # for the former, fire only for the latter. + # Deadline expired. Order of authority: + # + # 1. Phase lease (report_startup_progress): the startup path + # itself declared long legitimate work — honor it outright. + # Works for I/O-bound phases with ~zero CPU (corruption + # repair, backups) and cannot be faked by unrelated threads. + # 2. CPU progress, bounded: process-wide CPU proves the process + # is doing *something*, not that startup is progressing (an + # unrelated daemon thread could burn CPU while the startup + # thread sits parked forever). Extend at most + # _MAX_CPU_EXTENSIONS times, then fire regardless. + now = time.monotonic() + with self._state_lock: + lease_until = self._lease_until + lease_phase = self._lease_phase + if lease_until > now: + with self._state_lock: + if self._state != _ARMED: + return + self._deadline = max( + lease_until, now + min(_POLL_SLICE_S, self.timeout_s) + ) + try: + logger.warning( + "Gateway startup exceeded %.0fs but phase %r holds a " + "progress lease for another %.0fs — honoring it.", + self.timeout_s, + lease_phase or "unknown", + lease_until - now, + ) + except Exception: + pass + # Leased work may be I/O-bound; reset the CPU baseline so a + # post-lease window is judged on its own activity. + last_cpu = self._process_cpu_seconds() + continue cpu = self._process_cpu_seconds() if ( cpu is not None and last_cpu is not None and (cpu - last_cpu) >= _CPU_PROGRESS_MIN_S + and self._extensions < _MAX_CPU_EXTENSIONS ): window_delta = cpu - last_cpu last_cpu = cpu @@ -361,11 +497,14 @@ def _run(self) -> None: try: logger.warning( "Gateway startup exceeded %.0fs but is consuming CPU " - "(%.1fs this window) — likely a long schema migration; " - "extending the startup watchdog deadline (extension #%d).", + "(%.1fs this window); extending the startup watchdog " + "deadline (CPU-fallback extension %d of %d — phases " + "doing long legitimate work should call " + "report_startup_progress instead).", self.timeout_s, window_delta, self._extensions, + _MAX_CPU_EXTENSIONS, ) except Exception: pass @@ -461,6 +600,30 @@ def kick_startup_watchdog(extra_s: float = 0.0) -> None: logger.debug("Failed to kick gateway startup watchdog", exc_info=True) +def report_startup_progress(expected_s: float, phase: str = "") -> None: + """Declare a phase-owned progress lease on the armed startup watchdog. + + Call from startup phases about to perform legitimately long synchronous + work — most importantly ``state.db`` schema migrations and corruption + repair/backup inside ``SessionDB.__init__`` — passing an honest worst + case for the work about to be done, and renew periodically for + multi-step phases. Unlike CPU-time inference, a lease is owned by the + startup path itself: it works for I/O-bound work that accrues ~zero CPU + and cannot be counterfeited by unrelated busy threads. + + Per-call lease duration is clamped to ``_MAX_LEASE_S``; renewals prove + continued liveness. No-op when the watchdog is not armed; never raises — + safe to call unconditionally from application code. + """ + try: + with _handle_lock: + handle = _handle + if handle is not None: + handle.lease(expected_s, phase) + except Exception: + logger.debug("Failed to report startup progress", exc_info=True) + + def _reset_for_tests() -> None: """Drop the module singleton (test isolation only).""" global _handle diff --git a/hermes_state.py b/hermes_state.py index 9fdc820035a8..57bf037b53dc 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -43,6 +43,7 @@ describe_skill_invocation, ) from hermes_constants import get_hermes_home +from hermes_startup_watchdog import report_startup_progress from hermes_cli.sqlite_runtime import ( is_sqlite_wal_reset_vulnerable as _is_sqlite_wal_reset_vulnerable, ) @@ -2302,6 +2303,11 @@ def repair_state_db_schema(db_path: Path, *, backup: bool = True) -> Dict[str, A "error": None, } + # Startup-watchdog progress lease: repair (raw backup copy + surgery + + # VACUUM) is I/O-bound — near-zero CPU on a multi-GB file — which the + # watchdog's CPU fallback would misread as a parked deadlock (OOF-298). + report_startup_progress(900.0, phase="state_db_repair") + db_path = Path(db_path) if not db_path.exists(): report["error"] = f"{db_path} does not exist" diff --git a/hermes_state_schema.py b/hermes_state_schema.py index 3fa101a77284..9788e477b378 100644 --- a/hermes_state_schema.py +++ b/hermes_state_schema.py @@ -14,6 +14,7 @@ from typing import Dict, Optional from hermes_constants import get_hermes_home +from hermes_startup_watchdog import report_startup_progress from hermes_state_common import ( DEFERRED_INDEX_SQL, FTS_CJK_STALE_KEY, @@ -823,6 +824,13 @@ def _init_schema(self): The schema_version table is retained for future data migrations (transforming existing rows) which cannot be handled declaratively. """ + # Declare a startup-watchdog progress lease before potentially long + # synchronous work: on multi-GB state.db files the reconciliation + + # version-gated data migrations below are legitimately slow and can + # be I/O-bound (near-zero CPU), which the watchdog's CPU fallback + # would misread as a parked deadlock (OOF-298 / PR #89750). + report_startup_progress(600.0, phase="state_db_init_schema") + cursor = self._conn.cursor() cursor.executescript(SCHEMA_SQL) @@ -909,6 +917,9 @@ def _init_schema(self): ) else: current_version = row["version"] if isinstance(row, sqlite3.Row) else row[0] + # Renew the progress lease: the version-gated chain below can + # rewrite whole tables (PK rebuilds, backfills) on large DBs. + report_startup_progress(600.0, phase="state_db_data_migrations") # Data migrations that can't be expressed declaratively (row # backfills, index changes tied to a specific version step) stay # in a version-gated chain. Column additions are handled by diff --git a/tests/gateway/test_startup_watchdog.py b/tests/gateway/test_startup_watchdog.py index d4fb6671c431..2d09adc1529b 100644 --- a/tests/gateway/test_startup_watchdog.py +++ b/tests/gateway/test_startup_watchdog.py @@ -28,6 +28,7 @@ disarm_startup_watchdog, get_startup_watchdog_dump_path, kick_startup_watchdog, + report_startup_progress, resolve_startup_watchdog_timeout, startup_watchdog_disabled, ) @@ -260,7 +261,10 @@ def test_kick_with_garbage_extra_is_safe(self): class TestCpuProgressExtension: def test_cpu_progress_extends_instead_of_firing(self, exit_capture, monkeypatch): """A long schema migration burns CPU: the watchdog must extend, not - fire (the P1 false-fire/restart-loop case from review).""" + fire (the P1 false-fire/restart-loop case from review). Cap raised + here to observe pure extension behavior; the cap itself is covered + by test_cpu_extensions_are_capped.""" + monkeypatch.setattr(sw, "_MAX_CPU_EXTENSIONS", 10_000) # Each probe call reports +10s CPU — always 'progress'. counter = {"cpu": 0.0} @@ -278,6 +282,28 @@ def _busy_probe(): assert handle._extensions >= 1 disarm_startup_watchdog() + def test_cpu_extensions_are_capped(self, exit_capture, monkeypatch): + """Adversarial: an unrelated daemon thread burning CPU while the + startup thread sits parked must NOT hide the deadlock forever. + Process-wide CPU is bounded evidence — after _MAX_CPU_EXTENSIONS + the watchdog fires anyway (review blocker #2, false-negative arm).""" + counter = {"cpu": 0.0} + + def _busy_probe(): + counter["cpu"] += 10.0 + return counter["cpu"] + + monkeypatch.setattr( + StartupWatchdogHandle, "_process_cpu_seconds", staticmethod(_busy_probe) + ) + handle = arm_startup_watchdog(timeout_s=0.1) + assert handle is not None + # Perpetual CPU "progress" earns exactly _MAX_CPU_EXTENSIONS + # extensions, then fires. + assert exit_capture.fired.wait(timeout=10) + assert handle._extensions == sw._MAX_CPU_EXTENSIONS + assert exit_capture.codes == [SERVICE_RESTART_EXIT_CODE] + def test_no_cpu_progress_fires(self, exit_capture): # autouse fixture pins CPU time at 0.0 — no progress. arm_startup_watchdog(timeout_s=0.1) @@ -294,6 +320,104 @@ def test_probe_failure_fails_toward_firing(self, exit_capture, monkeypatch): assert exit_capture.fired.wait(timeout=5) +class TestProgressLease: + """Phase-owned progress leases (review blocker #2): the authoritative + 'startup is alive' signal, owned by the startup path itself — works for + I/O-bound phases with ~zero CPU, can't be counterfeited by unrelated + busy threads, and is clamped per call so it can't silence the watchdog + forever without renewal.""" + + def test_lease_prevents_firing_with_zero_cpu(self, exit_capture): + """Adversarial (false-positive arm): an I/O-bound repair/backup + phase accrues ~no CPU. Without a lease it would be killed; with one + it must survive past the deadline.""" + handle = arm_startup_watchdog(timeout_s=0.1) + assert handle is not None + report_startup_progress(60.0, phase="state_db_repair") + time.sleep(0.6) + assert not exit_capture.fired.is_set() + disarm_startup_watchdog() + + def test_expired_lease_no_longer_protects(self, exit_capture): + """A lease is a bounded claim, not a permanent mute: once it expires + (and no renewal arrives, no CPU progress) the watchdog fires.""" + handle = arm_startup_watchdog(timeout_s=0.1) + assert handle is not None + report_startup_progress(0.2, phase="short_phase") + assert exit_capture.fired.wait(timeout=10) + assert exit_capture.codes == [SERVICE_RESTART_EXIT_CODE] + + def test_lease_duration_is_clamped(self): + handle = arm_startup_watchdog(timeout_s=60) + assert handle is not None + report_startup_progress(10**9, phase="greedy") + with handle._state_lock: + remaining = handle._lease_until - time.monotonic() + assert remaining <= sw._MAX_LEASE_S + 1 + disarm_startup_watchdog() + + def test_lease_outranks_cpu_extension_cap(self, exit_capture, monkeypatch): + """A current lease is honored even when the CPU-fallback cap is + exhausted — the lease is the stronger, owned signal. Cap pinned to + 0 so CPU progress alone can never extend; only the lease can.""" + monkeypatch.setattr(sw, "_MAX_CPU_EXTENSIONS", 0) + counter = {"cpu": 0.0} + + def _busy_probe(): + counter["cpu"] += 10.0 + return counter["cpu"] + + monkeypatch.setattr( + StartupWatchdogHandle, "_process_cpu_seconds", staticmethod(_busy_probe) + ) + handle = arm_startup_watchdog(timeout_s=0.1) + assert handle is not None + report_startup_progress(60.0, phase="post_cap_migration") + time.sleep(0.6) + assert not exit_capture.fired.is_set() + disarm_startup_watchdog() + + def test_lease_without_arm_is_safe(self): + report_startup_progress(30.0, phase="x") # must not raise + + def test_lease_with_garbage_is_safe(self): + arm_startup_watchdog(timeout_s=60) + report_startup_progress("nonsense") # type: ignore[arg-type] + report_startup_progress(-5) + disarm_startup_watchdog() + + def test_lease_visible_in_dump_record(self, exit_capture, tmp_path): + arm_startup_watchdog(timeout_s=0.1) + report_startup_progress(0.15, phase="brief_phase") + assert exit_capture.fired.wait(timeout=10) + dump_path = get_startup_watchdog_dump_path(tmp_path) + deadline = time.monotonic() + 2 + while not dump_path.exists() and time.monotonic() < deadline: + time.sleep(0.02) + record = json.loads(dump_path.read_text(encoding="utf-8").splitlines()[0]) + assert record["lease_count"] >= 1 + assert record["last_lease_phase"] == "brief_phase" + + def test_schema_init_declares_lease(self): + """hermes_state_schema._init_schema must hold a progress lease so + multi-GB migrations aren't misread as deadlocks (wiring contract).""" + import inspect + + import hermes_state_schema + + src = inspect.getsource(hermes_state_schema.SessionSchemaMixin._init_schema) + assert "report_startup_progress" in src + + def test_repair_declares_lease(self): + """repair_state_db_schema (I/O-bound, ~zero CPU) must hold a lease.""" + import inspect + + import hermes_state + + src = inspect.getsource(hermes_state.repair_state_db_schema) + assert "report_startup_progress" in src + + class TestFire: def test_fires_after_deadline_with_restart_code(self, exit_capture): handle = arm_startup_watchdog(timeout_s=0.1) @@ -343,6 +467,98 @@ def test_custom_exit_code(self, exit_capture): assert exit_capture.codes == [42] +class TestBoundedExit: + """The fire path's forensics (logging, dump record, faulthandler, + ledger) may themselves wedge — the wedged main thread can hold the + logging handler lock, the disk can be full or hung. An escort thread + free of logging/filesystem/application locks must still reach the exit + seam within _FIRE_EXIT_BOUND_S (review blocker #1).""" + + @pytest.fixture + def fast_escort(self, monkeypatch): + """Shrink the escort bound so tests don't wait 10s.""" + monkeypatch.setattr( + StartupWatchdogHandle, "_sleep", staticmethod(lambda s: time.sleep(0.2)) + ) + + def test_exits_even_when_logging_lock_is_held( + self, exit_capture, fast_escort, monkeypatch + ): + """Adversarial: acquire the lock of every handler reachable from + this module's logger before the deadline expires. logger.critical + in _fire blocks forever — the escort must exit anyway.""" + import logging + + # Ensure there is at least one handler whose lock we can hold. + blocker_handler = logging.StreamHandler() + root = logging.getLogger() + root.addHandler(blocker_handler) + held = [h for h in root.handlers if h.lock is not None] + assert held, "expected at least one lockable logging handler" + for h in held: + h.lock.acquire() + handle = None + try: + handle = arm_startup_watchdog(timeout_s=0.1) + assert handle is not None + # Normal forensic path is stuck on logger.critical; only the + # escort can set fired. + assert exit_capture.fired.wait(timeout=10) + assert SERVICE_RESTART_EXIT_CODE in exit_capture.codes + finally: + for h in held: + h.lock.release() + root.removeHandler(blocker_handler) + # Let the unblocked fire thread finish while _exit is still the + # capture — otherwise it could reach the REAL os._exit after + # monkeypatch teardown and kill the test run. + if handle is not None: + handle.join(timeout=10) + + def test_exits_even_when_dump_write_hangs( + self, exit_capture, fast_escort, monkeypatch + ): + """Adversarial: filesystem forensics hang (full/hung disk). The + escort must exit anyway.""" + forever = threading.Event() + + def _hang(record): + forever.wait(timeout=30) # bounded only so the test can't leak + + monkeypatch.setattr(sw, "_write_dump_record", _hang) + handle = arm_startup_watchdog(timeout_s=0.1) + assert handle is not None + assert exit_capture.fired.wait(timeout=10) + assert SERVICE_RESTART_EXIT_CODE in exit_capture.codes + # Unblock and drain the fire thread before monkeypatch teardown + # (same real-os._exit hazard as above). + forever.set() + handle.join(timeout=10) + + def test_escort_stands_down_when_fire_completes(self, exit_capture, monkeypatch): + """When forensics complete normally the escort must NOT double-exit: + it observes _fire_done and returns.""" + monkeypatch.setattr( + StartupWatchdogHandle, "_sleep", staticmethod(lambda s: time.sleep(0.5)) + ) + handle = arm_startup_watchdog(timeout_s=0.1) + assert handle is not None + assert exit_capture.fired.wait(timeout=5) + # Give the escort time to wake and observe _fire_done. + time.sleep(0.8) + assert exit_capture.codes == [SERVICE_RESTART_EXIT_CODE] + + def test_escort_uses_no_logging_or_filesystem(self): + """Structural guarantee: the escort body must not touch logging, + the filesystem, or imports — only sleep, an Event check, and the + exit seam.""" + import inspect + + src = inspect.getsource(StartupWatchdogHandle._exit_escort) + for banned in ("logger.", "logging", "open(", "Path(", "import ", "mkdir"): + assert banned not in src, f"escort must not use {banned!r}" + + class TestDumpPath: def test_dump_path_under_home(self, tmp_path): assert get_startup_watchdog_dump_path(tmp_path) == ( From 0b1eec1a806d5d6064e5e1eeb76a3f7bb8b34a62 Mon Sep 17 00:00:00 2001 From: Shannon Sands Date: Sat, 22 Aug 2026 02:57:43 +1000 Subject: [PATCH 4/5] ci: retrigger full matrix (flake: test_focus_regain_redraw_is_rate_limited on fresh runner, unrelated to this PR) From 8805432ed60e75053b863ac12a457a1d5b993e01 Mon Sep 17 00:00:00 2001 From: Shannon Sands Date: Sat, 22 Aug 2026 05:03:10 +1000 Subject: [PATCH 5/5] =?UTF-8?q?ci:=20retrigger=20full=20matrix=20(flake:?= =?UTF-8?q?=20e2e=20test=5Fplaintext=5Frestart=5Fgateway=5Fin=5Fgroup=5Fst?= =?UTF-8?q?ays=5Fplain=5Ftext[telegram]=20mock=20assertion=20failure,=20un?= =?UTF-8?q?related=20to=20this=20PR=20=E2=80=94=20empty=20diff=20on=20test?= =?UTF-8?q?=5Fplatform=5Fcommands.py)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit