fix(gateway): detect and escape silent event-loop freezes

- A self-rescheduling 5s call_later floor timer, armed before any
  adapter connects, guarantees the selector always has a finite
  timeout, so the existing async defenses (polling heartbeat, timeout
  guards) regain a chance to run after a zero-pending-timer stall.
- A resident daemon-thread liveness watchdog probes the loop via
  call_soon_threadsafe every 30s; after 3 consecutive 10s-timeout
  misses (~120s of total unresponsiveness) it dumps all thread
  tracebacks and exits with the established
  GATEWAY_SERVICE_RESTART_EXIT_CODE (75) so a supervisor restarts the
  gateway - async-level recovery cannot run on a frozen loop.
- stop() disarms both guards before any teardown await so a busy
  shutdown is never misjudged as a freeze.
  HERMES_GATEWAY_LOOP_WATCHDOG=0 disables; _INTERVAL/_TIMEOUT/_STRIKES
  tune the thresholds.

Fixes #69089

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

gateway/run.py

@@ -2130,9 +2130,11 @@ from gateway.platforms.base import (
 )
 from gateway.shutdown_watchdog import (
     DEFAULT_HEARTBEAT_INTERVAL_S,
+    _arm_loop_floor_timer,
     arm_shutdown_watchdog,
     loop_heartbeat_forever,
     resolve_shutdown_watchdog_delay,
+    start_loop_liveness_watchdog,
 )
 from gateway.restart import (
     DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT,
@@ -3286,10 +3288,12 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
     _session_ephemeral_pin: Dict[str, tuple] = {}
     _session_vc_last: Dict[str, str] = {}
     _startup_restore_in_progress: bool = False
-    # Loop-liveness heartbeat / shutdown-watchdog handles (#66892). Class-level
+    # Loop-liveness heartbeat / watchdog handles (#66892, #69089). Class-level
     # defaults so partial construction in tests doesn't blow up on access; the
     # real values are set in __init__ / start() / stop().
     _loop_heartbeat_task: Optional["asyncio.Task"] = None
+    _loop_floor_timer_handle: Optional[Any] = None
+    _loop_liveness_watchdog: Optional[Any] = None
     _gateway_started_at: float = 0.0
     _shutdown_watchdog_done: Optional["threading.Event"] = None
     _platform_lock_takeover_on_start: bool = False
@@ -3673,6 +3677,8 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         # updated_at to distinguish "process alive" from "loop frozen".
         self._gateway_started_at: float = time.time()
         self._loop_heartbeat_task: Optional[asyncio.Task] = None
+        self._loop_floor_timer_handle = None
+        self._loop_liveness_watchdog = None
 
         # scale-to-zero (Phase 0, F13): gateway-scoped "last inbound seen" clock.
         # There is no such clock today (only a per-agent _last_activity_ts), so the
@@ -7756,6 +7762,39 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
             )
         return True
 
+    def _start_loop_liveness_guards(self, loop: asyncio.AbstractEventLoop) -> None:
+        """Arm the selector floor and out-of-loop watchdog before adapters."""
+        if getattr(self, "_loop_floor_timer_handle", None) is None:
+            try:
+                self._loop_floor_timer_handle = _arm_loop_floor_timer(loop)
+            except Exception:
+                logger.debug("Failed to arm gateway loop floor timer", exc_info=True)
+
+        watchdog = getattr(self, "_loop_liveness_watchdog", None)
+        if watchdog is None or not watchdog.is_alive():
+            try:
+                self._loop_liveness_watchdog = start_loop_liveness_watchdog(loop)
+            except Exception:
+                logger.debug("Failed to start gateway loop liveness watchdog", exc_info=True)
+
+    def _stop_loop_liveness_guards(self) -> None:
+        """Disarm lifetime liveness guards before shutdown can load the loop."""
+        watchdog = getattr(self, "_loop_liveness_watchdog", None)
+        self._loop_liveness_watchdog = None
+        if watchdog is not None:
+            try:
+                watchdog.stop()
+            except Exception:
+                logger.debug("Failed to stop gateway loop liveness watchdog", exc_info=True)
+
+        floor_timer = getattr(self, "_loop_floor_timer_handle", None)
+        self._loop_floor_timer_handle = None
+        if floor_timer is not None:
+            try:
+                floor_timer.cancel()
+            except Exception:
+                logger.debug("Failed to cancel gateway loop floor timer", exc_info=True)
+
     async def start(self) -> bool:
         """
         Start the gateway and all configured platform adapters.
@@ -7796,6 +7835,8 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
             self._gateway_loop = asyncio.get_running_loop()
         except RuntimeError:
             self._gateway_loop = None
+        if self._gateway_loop is not None:
+            self._start_loop_liveness_guards(self._gateway_loop)
         logger.info("Session storage: %s", self.config.sessions_dir)
 
         # Sanity-check that systemd's TimeoutStopSec covers our drain
@@ -9383,6 +9424,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         service_restart: bool = False,
     ) -> None:
         """Stop the gateway and disconnect all adapters."""
+        self._stop_loop_liveness_guards()
         if restart:
             self._restart_requested = True
             self._restart_detached = detached_restart

gateway/shutdown_watchdog.py

@@ -1,4 +1,4 @@
-"""Out-of-loop shutdown backstop + event-loop liveness heartbeat (#66892).
+"""Out-of-loop shutdown and event-loop liveness backstops (#66892, #69089).
 
 When the asyncio loop freezes mid-drain, every asyncio-based recovery path is
 structurally unable to fire: the drain deadline, status rewrites, and forensics
@@ -15,6 +15,10 @@ This module provides:
 2. An event-loop heartbeat file at ``<HERMES_HOME>/state/gateway.heartbeat`` so
    external supervision can distinguish "process alive" from "loop frozen"
    (``gateway_state.json`` alone can't — it only rewrites on transitions/turns).
+3. A lifetime thread watchdog that can still diagnose and hard-exit when the
+   event loop is too frozen to run its own heartbeat or timeout callbacks.
+4. A self-rescheduling floor timer that keeps the loop selector's timeout
+   finite, giving existing async recovery tasks a chance to resume.
 """
 
 from __future__ import annotations
@@ -31,6 +35,7 @@ from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any, Callable, Dict, Optional
 
+from gateway.restart import GATEWAY_SERVICE_RESTART_EXIT_CODE
 from hermes_constants import get_hermes_home
 from utils import atomic_json_write
 
@@ -40,10 +45,181 @@ logger = logging.getLogger(__name__)
 # drain is not cut short. Matches the issue #66892 suggested hardening.
 DEFAULT_SHUTDOWN_WATCHDOG_GRACE_S = 60.0
 DEFAULT_HEARTBEAT_INTERVAL_S = 30.0
+DEFAULT_LOOP_FLOOR_TIMER_INTERVAL_S = 5.0
+DEFAULT_LOOP_WATCHDOG_INTERVAL_S = 30.0
+DEFAULT_LOOP_WATCHDOG_TIMEOUT_S = 10.0
+DEFAULT_LOOP_WATCHDOG_MAX_STRIKES = 3
 _HEARTBEAT_RELATIVE = ("state", "gateway.heartbeat")
 _WATCHDOG_DUMP_RELATIVE = ("logs", "gateway-shutdown-watchdog.log")
 
 
+class _LoopFloorTimerHandle:
+    """Cancelable owner for the currently scheduled selector floor timer."""
+
+    def __init__(self, loop: asyncio.AbstractEventLoop, interval: float):
+        self._loop = loop
+        self._interval = interval
+        self._cancelled = False
+        self._timer: Optional[asyncio.TimerHandle] = None
+        self._schedule()
+
+    def _schedule(self) -> None:
+        self._timer = self._loop.call_later(self._interval, self._tick)
+
+    def _tick(self) -> None:
+        if not self._cancelled:
+            self._schedule()
+
+    def cancel(self) -> None:
+        self._cancelled = True
+        if self._timer is not None:
+            self._timer.cancel()
+
+
+class _LoopLivenessWatchdogHandle:
+    """Small lifecycle handle for the daemon liveness thread."""
+
+    def __init__(self, stop_event: threading.Event, thread: threading.Thread):
+        self._stop_event = stop_event
+        self._thread = thread
+
+    def stop(self) -> None:
+        self._stop_event.set()
+
+    def join(self, timeout: Optional[float] = None) -> None:
+        self._thread.join(timeout=timeout)
+
+    def is_alive(self) -> bool:
+        return self._thread.is_alive()
+
+
+def _arm_loop_floor_timer(
+    loop: asyncio.AbstractEventLoop,
+    interval: float = DEFAULT_LOOP_FLOOR_TIMER_INTERVAL_S,
+) -> _LoopFloorTimerHandle:
+    """Keep at least one timer pending so selector waits remain bounded."""
+    try:
+        resolved_interval = float(interval)
+        if resolved_interval <= 0:
+            raise ValueError
+    except (TypeError, ValueError):
+        resolved_interval = DEFAULT_LOOP_FLOOR_TIMER_INTERVAL_S
+    return _LoopFloorTimerHandle(loop, resolved_interval)
+
+
+def _positive_float_env(name: str, default: float) -> float:
+    raw = os.environ.get(name, "").strip()
+    try:
+        value = float(raw) if raw else float(default)
+    except (TypeError, ValueError):
+        return float(default)
+    return value if value > 0 else float(default)
+
+
+def _positive_int_env(name: str, default: int) -> int:
+    raw = os.environ.get(name, "").strip()
+    try:
+        value = int(raw) if raw else int(default)
+    except (TypeError, ValueError):
+        return int(default)
+    return value if value > 0 else int(default)
+
+
+def start_loop_liveness_watchdog(
+    loop: asyncio.AbstractEventLoop,
+    *,
+    probe_interval: float = DEFAULT_LOOP_WATCHDOG_INTERVAL_S,
+    probe_timeout: float = DEFAULT_LOOP_WATCHDOG_TIMEOUT_S,
+    max_strikes: int = DEFAULT_LOOP_WATCHDOG_MAX_STRIKES,
+    exit_code: int = GATEWAY_SERVICE_RESTART_EXIT_CODE,
+) -> Optional[_LoopLivenessWatchdogHandle]:
+    """Start an out-of-loop watchdog that hard-exits after missed probes.
+
+    Environment overrides are intentionally read here, the owner module, like
+    other gateway runtime-only safety knobs. Set
+    ``HERMES_GATEWAY_LOOP_WATCHDOG=0`` to disable it.
+    """
+    enabled = os.environ.get("HERMES_GATEWAY_LOOP_WATCHDOG", "1").strip().lower()
+    if enabled in {"0", "false", "no", "off"}:
+        return None
+
+    interval = _positive_float_env(
+        "HERMES_GATEWAY_LOOP_WATCHDOG_INTERVAL", probe_interval
+    )
+    timeout = _positive_float_env("HERMES_GATEWAY_LOOP_WATCHDOG_TIMEOUT", probe_timeout)
+    strikes_limit = _positive_int_env(
+        "HERMES_GATEWAY_LOOP_WATCHDOG_STRIKES", max_strikes
+    )
+    stop_event = threading.Event()
+
+    def _wait_for_probe(probe_event: threading.Event) -> Optional[bool]:
+        deadline = time.monotonic() + timeout
+        while True:
+            if stop_event.is_set():
+                return None
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                return probe_event.is_set()
+            if probe_event.wait(timeout=min(remaining, 0.05)):
+                return True
+
+    def _watchdog() -> None:
+        strikes = 0
+        while not stop_event.wait(timeout=interval):
+            probe_event = threading.Event()
+            try:
+                loop.call_soon_threadsafe(probe_event.set)
+            except RuntimeError:
+                # A normally closed loop cannot be probed and no longer needs
+                # a process-liveness backstop.
+                return
+            except Exception:
+                logger.debug(
+                    "Failed to schedule gateway loop liveness probe", exc_info=True
+                )
+                return
+
+            responded = _wait_for_probe(probe_event)
+            if responded is None:
+                return
+            if responded:
+                strikes = 0
+                continue
+
+            strikes += 1
+            if strikes < strikes_limit:
+                continue
+
+            try:
+                logger.critical(
+                    "Gateway event loop missed %d consecutive liveness probes; "
+                    "dumping all thread stacks and exiting with code %d so the "
+                    "service supervisor can restart it.",
+                    strikes,
+                    exit_code,
+                )
+            except Exception:
+                pass
+            try:
+                faulthandler.dump_traceback(all_threads=True)
+            except Exception:
+                logger.debug("Loop liveness faulthandler dump failed", exc_info=True)
+            os._exit(exit_code)
+            return
+
+    thread = threading.Thread(
+        target=_watchdog,
+        daemon=True,
+        name="gateway-loop-liveness-watchdog",
+    )
+    try:
+        thread.start()
+    except Exception:
+        logger.debug("Failed to start gateway loop liveness watchdog", exc_info=True)
+        return None
+    return _LoopLivenessWatchdogHandle(stop_event, thread)
+
+
 def _process_hermes_home() -> Path:
     """HERMES_HOME for process-level identity files (ignore profile overrides)."""
     val = os.environ.get("HERMES_HOME", "").strip()

tests/gateway/test_loop_liveness_watchdog.py

@@ -0,0 +1,264 @@
+"""Gateway event-loop freeze backstops for issue #69089."""
+
+from __future__ import annotations
+
+import asyncio
+import threading
+import time
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from gateway.shutdown_watchdog import (
+    _arm_loop_floor_timer,
+    start_loop_liveness_watchdog,
+)
+
+
+def _immediate_loop() -> MagicMock:
+    loop = MagicMock(spec=asyncio.AbstractEventLoop)
+    loop.call_soon_threadsafe.side_effect = lambda callback: callback()
+    return loop
+
+
+def test_loop_liveness_watchdog_responsive_probe_does_not_fire():
+    loop = _immediate_loop()
+    exit_codes = []
+
+    with (
+        patch("gateway.shutdown_watchdog.faulthandler.dump_traceback") as dump,
+        patch("gateway.shutdown_watchdog.os._exit", side_effect=exit_codes.append),
+    ):
+        handle = start_loop_liveness_watchdog(
+            loop, probe_interval=0.01, probe_timeout=0.01, max_strikes=2
+        )
+        assert handle is not None
+        deadline = time.monotonic() + 2.0
+        while loop.call_soon_threadsafe.call_count < 3 and time.monotonic() < deadline:
+            time.sleep(0.01)
+        handle.stop()
+        handle.join(timeout=1.0)
+
+    assert loop.call_soon_threadsafe.call_count >= 3
+    assert not handle.is_alive()
+    dump.assert_not_called()
+    assert exit_codes == []
+
+
+def test_loop_liveness_watchdog_exits_after_consecutive_misses():
+    loop = MagicMock(spec=asyncio.AbstractEventLoop)
+    fired = threading.Event()
+    exit_codes = []
+
+    def fake_exit(code: int) -> None:
+        exit_codes.append(code)
+        fired.set()
+
+    with (
+        patch("gateway.shutdown_watchdog.faulthandler.dump_traceback") as dump,
+        patch("gateway.shutdown_watchdog.os._exit", side_effect=fake_exit),
+    ):
+        handle = start_loop_liveness_watchdog(
+            loop, probe_interval=0.01, probe_timeout=0.01, max_strikes=2
+        )
+        assert handle is not None
+        assert fired.wait(timeout=2.0), "loop liveness watchdog did not fire"
+        handle.join(timeout=1.0)
+
+    assert not handle.is_alive()
+    assert loop.call_soon_threadsafe.call_count == 2
+    dump.assert_called_once_with(all_threads=True)
+    assert exit_codes == [75]
+
+
+def test_loop_liveness_watchdog_recovery_resets_strikes():
+    loop = MagicMock(spec=asyncio.AbstractEventLoop)
+    four_probes = threading.Event()
+
+    def alternate_response(callback) -> None:
+        count = loop.call_soon_threadsafe.call_count
+        if count in {2, 4}:
+            callback()
+        if count >= 4:
+            four_probes.set()
+
+    loop.call_soon_threadsafe.side_effect = alternate_response
+    with (
+        patch("gateway.shutdown_watchdog.faulthandler.dump_traceback") as dump,
+        patch("gateway.shutdown_watchdog.os._exit") as hard_exit,
+    ):
+        handle = start_loop_liveness_watchdog(
+            loop, probe_interval=0.01, probe_timeout=0.01, max_strikes=2
+        )
+        assert handle is not None
+        assert four_probes.wait(
+            timeout=2.0
+        ), "watchdog did not complete recovery probes"
+        handle.stop()
+        handle.join(timeout=1.0)
+
+    assert not handle.is_alive()
+    dump.assert_not_called()
+    hard_exit.assert_not_called()
+
+
+def test_loop_liveness_watchdog_stop_exits_thread_and_stops_probes():
+    loop = MagicMock(spec=asyncio.AbstractEventLoop)
+    first_probe = threading.Event()
+    loop.call_soon_threadsafe.side_effect = lambda callback: first_probe.set()
+
+    handle = start_loop_liveness_watchdog(
+        loop, probe_interval=0.01, probe_timeout=0.5, max_strikes=10
+    )
+    assert handle is not None
+    assert first_probe.wait(timeout=2.0)
+    handle.stop()
+    handle.join(timeout=1.0)
+    calls_after_stop = loop.call_soon_threadsafe.call_count
+    time.sleep(0.05)
+
+    assert not handle.is_alive()
+    assert loop.call_soon_threadsafe.call_count == calls_after_stop
+
+
+def test_loop_liveness_watchdog_env_can_disable(monkeypatch):
+    monkeypatch.setenv("HERMES_GATEWAY_LOOP_WATCHDOG", "0")
+    loop = MagicMock(spec=asyncio.AbstractEventLoop)
+
+    handle = start_loop_liveness_watchdog(
+        loop, probe_interval=0.01, probe_timeout=0.01, max_strikes=1
+    )
+
+    assert handle is None
+    loop.call_soon_threadsafe.assert_not_called()
+
+
+def test_loop_liveness_watchdog_env_overrides_probe_settings(monkeypatch):
+    monkeypatch.setenv("HERMES_GATEWAY_LOOP_WATCHDOG_INTERVAL", "0.01")
+    monkeypatch.setenv("HERMES_GATEWAY_LOOP_WATCHDOG_TIMEOUT", "0.01")
+    monkeypatch.setenv("HERMES_GATEWAY_LOOP_WATCHDOG_STRIKES", "1")
+    loop = MagicMock(spec=asyncio.AbstractEventLoop)
+    fired = threading.Event()
+
+    with (
+        patch("gateway.shutdown_watchdog.faulthandler.dump_traceback"),
+        patch(
+            "gateway.shutdown_watchdog.os._exit",
+            side_effect=lambda code: fired.set(),
+        ),
+    ):
+        handle = start_loop_liveness_watchdog(
+            loop, probe_interval=10.0, probe_timeout=10.0, max_strikes=10
+        )
+        assert handle is not None
+        assert fired.wait(timeout=2.0), "env overrides were not applied"
+        handle.join(timeout=1.0)
+
+    assert not handle.is_alive()
+    assert loop.call_soon_threadsafe.call_count == 1
+
+
+@pytest.mark.asyncio
+async def test_loop_liveness_watchdog_detects_real_loop_sync_freeze():
+    loop = asyncio.get_running_loop()
+    fired = threading.Event()
+    exit_codes = []
+
+    def fake_exit(code: int) -> None:
+        exit_codes.append(code)
+        fired.set()
+
+    with (
+        patch("gateway.shutdown_watchdog.faulthandler.dump_traceback") as dump,
+        patch("gateway.shutdown_watchdog.os._exit", side_effect=fake_exit),
+    ):
+        handle = start_loop_liveness_watchdog(
+            loop, probe_interval=0.02, probe_timeout=0.03, max_strikes=2
+        )
+        assert handle is not None
+        await asyncio.sleep(0.03)
+        time.sleep(0.25)
+        assert fired.wait(timeout=1.0), "watchdog did not detect the frozen real loop"
+        handle.stop()
+        handle.join(timeout=1.0)
+
+    dump.assert_called_once_with(all_threads=True)
+    assert exit_codes == [75]
+
+
+@pytest.mark.asyncio
+async def test_loop_liveness_watchdog_leaves_responsive_real_loop_running():
+    loop = asyncio.get_running_loop()
+    with (
+        patch("gateway.shutdown_watchdog.faulthandler.dump_traceback") as dump,
+        patch("gateway.shutdown_watchdog.os._exit") as hard_exit,
+    ):
+        handle = start_loop_liveness_watchdog(
+            loop, probe_interval=0.02, probe_timeout=0.03, max_strikes=2
+        )
+        assert handle is not None
+        await asyncio.sleep(0.25)
+        handle.stop()
+        handle.join(timeout=1.0)
+
+    assert not handle.is_alive()
+    dump.assert_not_called()
+    hard_exit.assert_not_called()
+
+
+def test_loop_floor_timer_reschedules_until_cancelled():
+    loop = MagicMock(spec=asyncio.AbstractEventLoop)
+    scheduled = []
+
+    def fake_call_later(delay, callback):
+        timer = MagicMock(spec=asyncio.TimerHandle)
+        scheduled.append((delay, callback, timer))
+        return timer
+
+    loop.call_later.side_effect = fake_call_later
+    handle = _arm_loop_floor_timer(loop, interval=5.0)
+
+    assert len(scheduled) == 1
+    assert scheduled[0][0] == 5.0
+    scheduled[0][1]()
+    assert len(scheduled) == 2
+    assert scheduled[1][0] == 5.0
+
+    handle.cancel()
+    scheduled[1][2].cancel.assert_called_once_with()
+    scheduled[1][1]()
+    assert len(scheduled) == 2
+
+
+def test_gateway_runner_liveness_guards_start_and_stop():
+    from gateway.run import GatewayRunner
+
+    runner = object.__new__(GatewayRunner)
+    runner._loop_floor_timer_handle = None
+    runner._loop_liveness_watchdog = None
+    loop = MagicMock(spec=asyncio.AbstractEventLoop)
+    floor_timer = MagicMock()
+    watchdog = MagicMock()
+    watchdog.is_alive.return_value = True
+
+    with (
+        patch(
+            "gateway.run._arm_loop_floor_timer", return_value=floor_timer
+        ) as arm_floor,
+        patch(
+            "gateway.run.start_loop_liveness_watchdog", return_value=watchdog
+        ) as start_watchdog,
+    ):
+        runner._start_loop_liveness_guards(loop)
+
+    arm_floor.assert_called_once_with(loop)
+    start_watchdog.assert_called_once_with(loop)
+    assert runner._loop_floor_timer_handle is floor_timer
+    assert runner._loop_liveness_watchdog is watchdog
+
+    runner._stop_loop_liveness_guards()
+
+    watchdog.stop.assert_called_once_with()
+    floor_timer.cancel.assert_called_once_with()
+    assert runner._loop_liveness_watchdog is None
+    assert runner._loop_floor_timer_handle is None