From 852db61abe4638242026905ad424a8a11bcce9fc Mon Sep 17 00:00:00 2001 From: Shannon Sands Date: Thu, 20 Aug 2026 12:09:12 +1000 Subject: [PATCH] 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 3b114c7b27..a7fe9abc24 100644 --- a/gateway/startup_watchdog.py +++ b/gateway/startup_watchdog.py @@ -21,6 +21,7 @@ from hermes_startup_watchdog import ( # noqa: F401 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 @@ __all__ = [ "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 1ce307e9a6..52a3bd0556 100644 --- a/hermes_startup_watchdog.py +++ b/hermes_startup_watchdog.py @@ -24,18 +24,29 @@ stacks via ``faulthandler``, records the exit in the lifecycle ledger 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 @@ _FALSEY = frozenset({"0", "false", "no", "off"}) _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 @@ class StartupWatchdogHandle: 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 @@ class StartupWatchdogHandle: 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 @@ class StartupWatchdogHandle: 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 @@ class StartupWatchdogHandle: "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 @@ class StartupWatchdogHandle: 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 @@ class StartupWatchdogHandle: 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 @@ class StartupWatchdogHandle: 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 8040c0f6b4..80896812e3 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -52,6 +52,7 @@ from agent.skill_commands import ( 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, ) @@ -3501,6 +3502,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 5dcf9756df..c6252fe33b 100644 --- a/hermes_state_schema.py +++ b/hermes_state_schema.py @@ -18,6 +18,7 @@ from typing import Dict, Optional, Sequence 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, @@ -971,6 +972,13 @@ class SessionSchemaMixin: 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) @@ -1068,6 +1076,9 @@ class SessionSchemaMixin: 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 d4fb6671c4..2d09adc152 100644 --- a/tests/gateway/test_startup_watchdog.py +++ b/tests/gateway/test_startup_watchdog.py @@ -28,6 +28,7 @@ from hermes_startup_watchdog import ( 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 @@ class TestKick: 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 @@ class TestCpuProgressExtension: 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 @@ class TestCpuProgressExtension: 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 @@ class TestFire: 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) == (