fix(gateway): make the restart-loop breaker see slow crash cycles (#81642)

The auto-resume restart-loop breaker (#30719, defense-3) pruned its boot
log against an absolute `window_seconds` (default 60s). That prune is
period-sensitive: a crash cycle slower than the window drops its own
history on every boot, so the counter never leaves 1 and the breaker can
never trip, no matter how long the loop runs.

The cycle reported in #81642 is ~150s — a wedged event loop, the liveness
watchdog hard-exiting at ~90s, a supervisor respawn, and auto-resume
replaying the same session that wedges it again. Structurally invisible to
a 60s window: `gateway/restart_loop.json` kept a single timestamp across 15
kills in one morning. Because every cycle leaves a gateway that cannot
process SIGTERM, `hermes update` has no drainable gateway to stop, which is
the reported hang.

Chain boots on the inter-boot GAP instead of an absolute window: two boots
belong to the same loop when they are no more than `max_gap_seconds` apart
(default 300s, floored by `window_seconds` so widening the window never
makes the breaker less sensitive). The verdict becomes period-agnostic —
the original ~10s respawn loop still trips in 3 boots, and so does a 150s
one — while a boot after real quiet resets the chain, so occasional
operator restarts still never accumulate. The persisted chain is capped at
50 entries.

- gateway/restart_loop_guard.py: gap-chained pruning (`_chain_ending_at`),
  `DEFAULT_MAX_GAP_SECONDS`, `max_gap_seconds` kwarg on the three entry
  points, clock-step tolerance, bounded state file
- gateway/run.py: `_restart_loop_guard_config` reads and returns
  `max_gap_seconds`; the auto-resume call site passes it through
- hermes_cli/config_defaults.py: `gateway.restart_loop_guard.max_gap_seconds`

Tests: 7 new cases in TestRestartLoopGuard covering the slow cycle, chain
persistence, quiet-period reset, the #30719 fast loop, the config knob, the
window floor, and the disabled breaker. Verified RED before the fix (the
slow-cycle case asserted `[1300.0] == [1000.0, 1150.0, 1300.0]`, exactly
the single-timestamp state file from the report) and GREEN after.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

gateway/restart_loop_guard.py

@@ -13,9 +13,13 @@ logic — SIGTERM every ~10 seconds until manually broken.
 
 This module is the last-resort circuit breaker: it records a timestamp
 each time the gateway boots with restart-interrupted sessions pending,
-keeps a rolling window of recent boots persisted across processes (each
+keeps the current chain of such boots persisted across processes (each
 boot is a fresh process, so in-memory state is useless), and reports the
-loop as "tripped" once too many such boots happen inside a short window.
+loop as "tripped" once too many of them chain together.  Boots chain
+while consecutive gaps stay within ``max_gap_seconds``, so the breaker
+sees slow crash cycles (a wedged event loop killed by the liveness
+watchdog every ~150s, #81642) exactly as well as the fast ~10s respawn
+loop it was originally written for.
 When tripped, the caller SKIPS auto-resume for that boot — the gateway
 still starts and serves real inbound messages, it just stops replaying
 the session that keeps killing it, which breaks the cycle and puts a
@@ -43,6 +47,23 @@ logger = logging.getLogger("gateway.run")
 DEFAULT_MAX_RESTARTS = 3
 DEFAULT_WINDOW_SECONDS = 60
 
+# Longest gap between two consecutive restart-interrupted boots that still
+# counts them as the SAME loop (#81642).  A fixed ``window_seconds`` prune can
+# only see crash cycles faster than the window: a loop whose period exceeds it
+# drops its own history on every boot, so the counter never leaves 1 and the
+# breaker never trips no matter how long the loop runs.  The reported cycle was
+# ~150s (wedged event loop -> ~90s liveness watchdog hard-exit -> respawn ->
+# auto-resume replays the same session), i.e. structurally invisible to the 60s
+# window.  Chaining on the inter-boot GAP instead makes the breaker period-
+# agnostic: any repeating cycle trips once ``max_restarts`` links accumulate,
+# and a single boot followed by real quiet resets the chain.
+DEFAULT_MAX_GAP_SECONDS = 300
+
+# Cap the persisted chain so a long-running loop cannot grow the state file
+# without bound.  Only the newest ``max_restarts`` entries can change a
+# verdict; the rest are kept for forensics.
+_MAX_STORED_BOOTS = 50
+
 
 def _state_path():
     return get_hermes_home() / "gateway" / "restart_loop.json"
@@ -67,22 +88,58 @@ def _save_boots(boots: List[float]) -> None:
         pass
 
 
+def _chain_gap(window_seconds: int, max_gap_seconds: int) -> float:
+    """Effective inter-boot gap that still links two boots into one loop.
+
+    Floored by ``window_seconds`` so an operator who widens the window never
+    ends up with a breaker that is *less* sensitive than they asked for.
+    """
+    return float(max(1, window_seconds, max_gap_seconds))
+
+
+def _chain_ending_at(boots: List[float], ts: float, gap: float) -> List[float]:
+    """Return the unbroken chain of boots leading up to ``ts``.
+
+    Walks backwards from ``ts`` and keeps boots while each successive gap stays
+    within ``gap``.  The first gap that exceeds it ends the chain: everything
+    older belongs to a previous, already-resolved episode.  A chain broken at
+    the head (nothing recent enough) yields an empty list, which is how a
+    healthy gateway forgets an old loop.
+    """
+    chain: List[float] = []
+    prev = ts
+    for t in sorted(boots, reverse=True):
+        if t > ts:
+            # Clock moved backwards (NTP step, restored state file). Treat the
+            # future entry as adjacent rather than dropping the whole chain.
+            chain.append(t)
+            continue
+        if prev - t > gap:
+            break
+        chain.append(t)
+        prev = t
+    chain.reverse()
+    return chain
+
+
 def record_restart_interrupted_boot(
     window_seconds: int = DEFAULT_WINDOW_SECONDS,
     *,
     now: Optional[float] = None,
+    max_gap_seconds: int = DEFAULT_MAX_GAP_SECONDS,
 ) -> List[float]:
     """Record that the gateway just booted with restart-interrupted sessions.
 
-    Prunes boots older than ``window_seconds`` and appends the current time.
-    Returns the pruned+appended list (most recent last).  Best-effort — a
-    persistence failure returns the in-memory list without raising.
+    Drops boots that belong to an earlier, already-broken chain (any gap wider
+    than ``max_gap_seconds``) and appends the current time.  Returns the
+    pruned+appended list (most recent last).  Best-effort — a persistence
+    failure returns the in-memory list without raising.
     """
     ts = time.time() if now is None else now
-    cutoff = ts - max(1, window_seconds)
-    boots = [t for t in _load_boots() if t >= cutoff]
+    gap = _chain_gap(window_seconds, max_gap_seconds)
+    boots = _chain_ending_at(_load_boots(), ts, gap)
     boots.append(ts)
-    _save_boots(boots)
+    _save_boots(boots[-_MAX_STORED_BOOTS:])
     return boots
 
 
@@ -91,21 +148,24 @@ def is_restart_loop_tripped(
     window_seconds: int = DEFAULT_WINDOW_SECONDS,
     *,
     now: Optional[float] = None,
+    max_gap_seconds: int = DEFAULT_MAX_GAP_SECONDS,
 ) -> bool:
     """Return True if the gateway has restarted ``>= max_restarts`` times with
-    restart-interrupted sessions inside the last ``window_seconds``.
+    restart-interrupted sessions in one unbroken chain ending at ``now``.
 
     Reads the persisted boot log written by
-    ``record_restart_interrupted_boot`` and counts boots within the window.
+    ``record_restart_interrupted_boot`` and counts the boots that still chain
+    together (consecutive gaps within ``max_gap_seconds``), so the verdict does
+    not depend on how fast the crash cycle happens to be.
     Fails OPEN (returns False) on any error — a broken breaker must never
     wedge a healthy gateway.
     """
     if max_restarts <= 0:
         return False
     ts = time.time() if now is None else now
-    cutoff = ts - max(1, window_seconds)
+    gap = _chain_gap(window_seconds, max_gap_seconds)
     try:
-        recent = [t for t in _load_boots() if t >= cutoff]
+        recent = _chain_ending_at(_load_boots(), ts, gap)
     except Exception:  # pragma: no cover — _load_boots already guards
         return False
     return len(recent) >= max_restarts
@@ -124,26 +184,30 @@ def check_and_record(
     window_seconds: int = DEFAULT_WINDOW_SECONDS,
     *,
     now: Optional[float] = None,
+    max_gap_seconds: int = DEFAULT_MAX_GAP_SECONDS,
 ) -> bool:
     """Record this restart-interrupted boot and report whether the loop is now
     tripped.
 
     This is the single entry point the gateway calls: it appends the current
-    boot, then checks whether the (now-updated) window has reached the
+    boot, then checks whether the (now-updated) chain has reached the
     threshold.  Returns True when auto-resume should be SKIPPED to break the
     loop.
     """
-    boots = record_restart_interrupted_boot(window_seconds, now=now)
+    boots = record_restart_interrupted_boot(
+        window_seconds, now=now, max_gap_seconds=max_gap_seconds
+    )
     tripped = len(boots) >= max_restarts if max_restarts > 0 else False
     if tripped:
         logger.warning(
-            "Restart-loop breaker TRIPPED: %d restart-interrupted gateway "
-            "boots within %ds (threshold %d). Skipping auto-resume to break "
-            "a suspected SIGTERM-respawn loop (#30719). Restart-interrupted "
-            "sessions stay resume-pending and will continue on the next real "
-            "user message. If this is a false positive, delete %s.",
+            "Restart-loop breaker TRIPPED: %d chained restart-interrupted "
+            "gateway boots (no gap wider than %ds; threshold %d). Skipping "
+            "auto-resume to break a suspected SIGTERM-respawn loop (#30719, "
+            "#81642). Restart-interrupted sessions stay resume-pending and "
+            "will continue on the next real user message. If this is a false "
+            "positive, delete %s.",
             len(boots),
-            window_seconds,
+            int(_chain_gap(window_seconds, max_gap_seconds)),
             max_restarts,
             _state_path(),
         )

gateway/run.py

@@ -7645,15 +7645,20 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         return parse_idle_timeout_seconds(raw)
 
     def _restart_loop_guard_config(self) -> tuple:
-        """Return ``(max_restarts, window_seconds)`` for the auto-resume
-        restart-loop breaker (#30719, defense-3), read from
+        """Return ``(max_restarts, window_seconds, max_gap_seconds)`` for the
+        auto-resume restart-loop breaker (#30719, defense-3), read from
         ``gateway.restart_loop_guard`` in config.yaml with the module defaults
         as fallback. ``max_restarts <= 0`` disables the breaker.
+
+        ``max_gap_seconds`` is the longest spacing between two consecutive
+        restart-interrupted boots that still counts them as the same loop, so
+        a crash cycle slower than ``window_seconds`` stays visible (#81642).
         """
         from gateway import restart_loop_guard as _rlg
 
         max_restarts = _rlg.DEFAULT_MAX_RESTARTS
         window_seconds = _rlg.DEFAULT_WINDOW_SECONDS
+        max_gap_seconds = _rlg.DEFAULT_MAX_GAP_SECONDS
         try:
             user_cfg = _load_gateway_config()
             gw = user_cfg.get("gateway") if isinstance(user_cfg, dict) else None
@@ -7663,9 +7668,14 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                     max_restarts = rlg["max_restarts"]
                 if isinstance(rlg.get("window_seconds"), int) and rlg["window_seconds"] > 0:
                     window_seconds = rlg["window_seconds"]
+                if (
+                    isinstance(rlg.get("max_gap_seconds"), int)
+                    and rlg["max_gap_seconds"] > 0
+                ):
+                    max_gap_seconds = rlg["max_gap_seconds"]
         except Exception:  # noqa: BLE001
             pass
-        return max_restarts, window_seconds
+        return max_restarts, window_seconds, max_gap_seconds
 
     def _scale_to_zero_should_arm(self) -> bool:
         """Whether to start the idle watcher (D1/D11/§3.4(1))."""
@@ -10685,8 +10695,10 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
             try:
                 from gateway import restart_loop_guard as _rlg
 
-                _max_restarts, _window = self._restart_loop_guard_config()
-                if _rlg.check_and_record(_max_restarts, _window):
+                _max_restarts, _window, _max_gap = self._restart_loop_guard_config()
+                if _rlg.check_and_record(
+                    _max_restarts, _window, max_gap_seconds=_max_gap
+                ):
                     return 0
             except Exception as exc:  # noqa: BLE001 — breaker must fail OPEN
                 logger.debug("Restart-loop guard check skipped: %s", exc)

hermes_cli/config_defaults.py

@@ -2579,14 +2579,24 @@ DEFAULT_CONFIG = {
         # keeps triggering another kill (e.g. the agent runs a raw
         # `launchctl kickstart ai.hermes.gateway` that defenses 1-2 don't
         # cover), the result is a tight SIGTERM-respawn loop. This breaker
-        # counts restart-interrupted boots in a rolling window and, once
-        # `max_restarts` boots happen within `window_seconds`, SKIPS
-        # auto-resume for that boot — the gateway still starts and serves
-        # real inbound messages, it just stops replaying the session that
-        # keeps killing it. Set `max_restarts` to 0 to disable the breaker.
+        # chains restart-interrupted boots together and, once `max_restarts`
+        # of them chain up, SKIPS auto-resume for that boot — the gateway
+        # still starts and serves real inbound messages, it just stops
+        # replaying the session that keeps killing it. Set `max_restarts` to
+        # 0 to disable the breaker.
+        # Two boots belong to the same chain when they are no more than
+        # `max_gap_seconds` apart (floored by `window_seconds`). Chaining on
+        # the GAP rather than on a fixed window is what makes the breaker see
+        # SLOW crash cycles: a loop whose period exceeds the window used to
+        # prune its own history on every boot, so the counter never left 1 and
+        # the breaker never tripped — e.g. the ~150s wedged-event-loop cycle in
+        # #81642 (stall -> ~90s liveness-watchdog hard-exit -> respawn ->
+        # auto-resume replays the same session), which also makes
+        # `hermes update` hang because it can never drain the gateway.
         "restart_loop_guard": {
             "max_restarts": 3,
             "window_seconds": 60,
+            "max_gap_seconds": 300,
         },
 
         # Portable respawn-storm circuit breaker (complements

tests/hermes_cli/test_gateway_restart_loop.py

@@ -967,6 +967,63 @@ class TestRestartLoopGuard:
         rlg.clear()
         assert rlg.check_and_record(3, 60, now=1002.0) is False
 
+    def test_trips_on_slow_crash_cycle_wider_than_window(self):
+        """#81642: a ~150s crash cycle is wider than the 60s window, so the
+        old absolute-window prune dropped the previous boot on every boot and
+        the counter never left 1.  Chaining on the inter-boot gap sees it."""
+        import gateway.restart_loop_guard as rlg
+        assert rlg.check_and_record(3, 60, now=1000.0) is False
+        assert rlg.check_and_record(3, 60, now=1150.0) is False
+        assert rlg.check_and_record(3, 60, now=1300.0) is True
+
+    def test_slow_cycle_chain_is_persisted_not_truncated(self):
+        """The state file must keep the whole chain — the reported symptom was
+        a restart_loop.json holding a single timestamp after 15 crashes."""
+        import gateway.restart_loop_guard as rlg
+        rlg.record_restart_interrupted_boot(60, now=1000.0)
+        rlg.record_restart_interrupted_boot(60, now=1150.0)
+        boots = rlg.record_restart_interrupted_boot(60, now=1300.0)
+        assert boots == [1000.0, 1150.0, 1300.0]
+
+    def test_quiet_period_breaks_the_chain(self):
+        """A boot after real quiet starts a fresh chain, so occasional
+        operator restarts never accumulate into a trip."""
+        import gateway.restart_loop_guard as rlg
+        rlg.check_and_record(3, 60, now=1000.0)
+        rlg.check_and_record(3, 60, now=1150.0)
+        # 1h later: unrelated restart, chain reset to a single boot.
+        assert rlg.check_and_record(3, 60, now=4800.0) is False
+        assert rlg.is_restart_loop_tripped(3, 60, now=4801.0) is False
+
+    def test_fast_respawn_loop_still_trips(self):
+        """#30719 regression: the original ~10s loop must keep tripping."""
+        import gateway.restart_loop_guard as rlg
+        assert rlg.check_and_record(3, 60, now=1000.0) is False
+        assert rlg.check_and_record(3, 60, now=1010.0) is False
+        assert rlg.check_and_record(3, 60, now=1020.0) is True
+
+    def test_max_gap_seconds_is_configurable(self):
+        """An operator can narrow the chain gap back down; a cycle slower than
+        the configured gap then stops chaining."""
+        import gateway.restart_loop_guard as rlg
+        assert rlg.check_and_record(3, 60, now=1000.0, max_gap_seconds=100) is False
+        assert rlg.check_and_record(3, 60, now=1150.0, max_gap_seconds=100) is False
+        assert rlg.check_and_record(3, 60, now=1300.0, max_gap_seconds=100) is False
+
+    def test_window_seconds_floors_the_gap(self):
+        """A window wider than the gap default still governs, so raising
+        window_seconds never makes the breaker less sensitive."""
+        import gateway.restart_loop_guard as rlg
+        assert rlg.check_and_record(3, 900, now=1000.0, max_gap_seconds=100) is False
+        assert rlg.check_and_record(3, 900, now=1400.0, max_gap_seconds=100) is False
+        assert rlg.check_and_record(3, 900, now=1800.0, max_gap_seconds=100) is True
+
+    def test_disabled_breaker_never_trips(self):
+        import gateway.restart_loop_guard as rlg
+        for ts in (1000.0, 1150.0, 1300.0, 1450.0):
+            assert rlg.check_and_record(0, 60, now=ts) is False
+        assert rlg.is_restart_loop_tripped(0, 60, now=1451.0) is False
+
 class TestTerminalToolGatewayLifecycleGuardRemote:
     """Remote-backend and two-session cwd regression coverage."""