fix(gateway): arm thread shutdown watchdog + loop heartbeat (#66892)

A frozen asyncio loop mid-SIGTERM drain cannot run the drain timeout or
status rewrites, so KeepAlive never sees a dead process. Arm an OS-thread
watchdog at stop() (drain+60s → faulthandler dump + os._exit) and rewrite
state/gateway.heartbeat from a loop task for external liveness checks.

gateway/run.py

@@ -1917,6 +1917,12 @@ from gateway.platforms.base import (
     merge_pending_message_event,
     utf16_len,
 )
+from gateway.shutdown_watchdog import (
+    DEFAULT_HEARTBEAT_INTERVAL_S,
+    arm_shutdown_watchdog,
+    loop_heartbeat_forever,
+    resolve_shutdown_watchdog_delay,
+)
 from gateway.restart import (
     DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT,
     GATEWAY_FATAL_CONFIG_EXIT_CODE,
@@ -2972,6 +2978,12 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
     _session_model_overrides: Dict[str, Dict[str, str]] = {}
     _session_reasoning_overrides: Dict[str, Dict[str, Any]] = {}
     _startup_restore_in_progress: bool = False
+    # Loop-liveness heartbeat / shutdown-watchdog handles (#66892). 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
+    _gateway_started_at: float = 0.0
+    _shutdown_watchdog_done: Optional["threading.Event"] = None
 
     def __init__(self, config: Optional[GatewayConfig] = None):
         global _gateway_runner_ref
@@ -3297,6 +3309,12 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         # Track background tasks to prevent garbage collection mid-execution
         self._background_tasks: set = set()
 
+        # Event-loop liveness heartbeat (#66892): rewritten every 30s while
+        # the loop is dispatching. External supervisors use the file mtime /
+        # updated_at to distinguish "process alive" from "loop frozen".
+        self._gateway_started_at: float = time.time()
+        self._loop_heartbeat_task: Optional[asyncio.Task] = 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
         # idle predicate needs this. Stamped in _handle_message (the single inbound
@@ -7535,7 +7553,27 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
 
         self._running = True
         self._update_runtime_status("running")
-        
+
+        # Loop-liveness heartbeat (#66892): an asyncio task so a frozen loop
+        # stops refreshing ``state/gateway.heartbeat``. Cancelled with the
+        # other background tasks during stop(). Best-effort — a liveness probe
+        # must never be able to abort startup.
+        try:
+            _existing_hb = getattr(self, "_loop_heartbeat_task", None)
+            if _existing_hb is None or _existing_hb.done():
+                self._loop_heartbeat_task = asyncio.create_task(
+                    loop_heartbeat_forever(
+                        interval_s=DEFAULT_HEARTBEAT_INTERVAL_S,
+                        start_time=getattr(self, "_gateway_started_at", 0.0),
+                    )
+                )
+                _bg = getattr(self, "_background_tasks", None)
+                if _bg is not None:
+                    _bg.add(self._loop_heartbeat_task)
+                    self._loop_heartbeat_task.add_done_callback(_bg.discard)
+        except Exception:
+            logger.debug("Failed to start gateway loop heartbeat", exc_info=True)
+
         # Emit gateway:startup hook
         hook_count = len(self.hooks.loaded_hooks)
         if hook_count:
@@ -8499,11 +8537,57 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                 except Exception as _e:
                     logger.debug("cleanup_all_browsers (%s) error: %s", phase, _e)
 
+            # Thread-based shutdown watchdog (#66892): asyncio timeouts cannot
+            # recover a frozen loop. Arm a plain OS thread at the start of
+            # stop(); if teardown never finishes within drain+grace it dumps
+            # faulthandler stacks and os._exit so KeepAlive/systemd can revive.
+            # Skip under pytest so stop()-driving unit tests don't get a
+            # delayed hard-exit in the worker.
+            _watchdog_done = threading.Event()
+            self._shutdown_watchdog_done = _watchdog_done
+            _stop_started_at_box: dict[str, float] = {}
+
+            def _shutdown_watchdog_snapshot() -> dict:
+                started = _stop_started_at_box.get("t")
+                return {
+                    "restart_requested": bool(self._restart_requested),
+                    "draining": bool(self._draining),
+                    "running": bool(self._running),
+                    "active_agents": self._running_agent_count(),
+                    "active_cron_jobs": self._active_cron_job_count(),
+                    "active_api_runs": self._active_api_run_count(),
+                    "restart_drain_timeout": self._restart_drain_timeout,
+                    "watchdog_delay_s": resolve_shutdown_watchdog_delay(
+                        self._restart_drain_timeout
+                    ),
+                    "phase_elapsed_s": (
+                        time.monotonic() - started if started is not None else None
+                    ),
+                }
+
+            if not os.environ.get("PYTEST_CURRENT_TEST"):
+                arm_shutdown_watchdog(
+                    resolve_shutdown_watchdog_delay(self._restart_drain_timeout),
+                    done_event=_watchdog_done,
+                    snapshot_fn=_shutdown_watchdog_snapshot,
+                    exit_code=1,
+                )
+
+            try:
+                await _stop_impl_body(
+                    _kill_tool_subprocesses,
+                    _stop_started_at_box,
+                )
+            finally:
+                _watchdog_done.set()
+
+        async def _stop_impl_body(_kill_tool_subprocesses, _stop_started_at_box) -> None:
             logger.info(
                 "Stopping gateway%s...",
                 " for restart" if self._restart_requested else "",
             )
             _stop_started_at = time.monotonic()
+            _stop_started_at_box["t"] = _stop_started_at
 
             def _phase_elapsed() -> float:
                 return time.monotonic() - _stop_started_at

gateway/shutdown_watchdog.py

@@ -0,0 +1,257 @@
+"""Out-of-loop shutdown backstop + event-loop liveness heartbeat (#66892).
+
+When the asyncio loop freezes mid-drain, every asyncio-based recovery path is
+structurally unable to fire: the drain deadline, status rewrites, and forensics
+all need the same loop that is stuck. launchd/systemd KeepAlive only restarts a
+*dead* process, so a wedged-but-alive gateway sits as a zombie until manual
+SIGKILL.
+
+This module provides:
+
+1. A plain OS-thread shutdown watchdog armed at ``stop()``. If shutdown has not
+   completed within ``restart_drain_timeout + grace``, it dumps all-thread
+   stacks via ``faulthandler`` plus a metadata snapshot, then ``os._exit`` so
+   the service manager can revive the process.
+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).
+"""
+
+from __future__ import annotations
+
+import asyncio
+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, Callable, Dict, Optional
+
+from hermes_constants import get_hermes_home
+from utils import atomic_json_write
+
+logger = logging.getLogger(__name__)
+
+# Extra leash beyond ``agent.restart_drain_timeout`` so a slow-but-progressing
+# drain is not cut short. Matches the issue #66892 suggested hardening.
+DEFAULT_SHUTDOWN_WATCHDOG_GRACE_S = 60.0
+DEFAULT_HEARTBEAT_INTERVAL_S = 30.0
+_HEARTBEAT_RELATIVE = ("state", "gateway.heartbeat")
+_WATCHDOG_DUMP_RELATIVE = ("logs", "gateway-shutdown-watchdog.log")
+
+
+def _process_hermes_home() -> Path:
+    """HERMES_HOME for process-level identity files (ignore profile overrides)."""
+    val = os.environ.get("HERMES_HOME", "").strip()
+    if val:
+        return Path(val)
+    return get_hermes_home()
+
+
+def get_loop_heartbeat_path(home: Optional[Path] = None) -> Path:
+    """Return ``<HERMES_HOME>/state/gateway.heartbeat``."""
+    base = home if home is not None else _process_hermes_home()
+    return base.joinpath(*_HEARTBEAT_RELATIVE)
+
+
+def get_shutdown_watchdog_dump_path(home: Optional[Path] = None) -> Path:
+    """Return the faulthandler / metadata dump path for a fired watchdog."""
+    base = home if home is not None else _process_hermes_home()
+    return base.joinpath(*_WATCHDOG_DUMP_RELATIVE)
+
+
+def write_loop_heartbeat(
+    *,
+    pid: Optional[int] = None,
+    start_time: Optional[float] = None,
+    home: Optional[Path] = None,
+    extra: Optional[Dict[str, Any]] = None,
+) -> Path:
+    """Atomically rewrite the loop-liveness heartbeat file.
+
+    ``start_time`` is the gateway process start (``time.time()`` epoch seconds)
+    so supervisors can detect PID reuse. Best-effort — never raises.
+    """
+    path = get_loop_heartbeat_path(home)
+    payload: Dict[str, Any] = {
+        "pid": int(pid if pid is not None else os.getpid()),
+        "updated_at": datetime.now(timezone.utc).isoformat(),
+        "monotonic": time.monotonic(),
+    }
+    if start_time is not None:
+        payload["start_time"] = float(start_time)
+    if extra:
+        payload.update(extra)
+    try:
+        atomic_json_write(path, payload, indent=None)
+    except Exception:
+        logger.debug("Failed to write gateway loop heartbeat", exc_info=True)
+    return path
+
+
+def resolve_shutdown_watchdog_delay(
+    drain_timeout: float,
+    *,
+    grace_s: float = DEFAULT_SHUTDOWN_WATCHDOG_GRACE_S,
+) -> float:
+    """Return the wall-clock leash for the shutdown watchdog thread."""
+    try:
+        drain = max(float(drain_timeout), 0.0)
+    except (TypeError, ValueError):
+        drain = 0.0
+    try:
+        grace = max(float(grace_s), 0.0)
+    except (TypeError, ValueError):
+        grace = DEFAULT_SHUTDOWN_WATCHDOG_GRACE_S
+    return drain + grace
+
+
+def _write_watchdog_dump(
+    dump_path: Path,
+    *,
+    delay_s: float,
+    snapshot: Optional[Dict[str, Any]],
+) -> None:
+    """Best-effort faulthandler + metadata dump before hard-exit."""
+    try:
+        dump_path.parent.mkdir(parents=True, exist_ok=True)
+    except OSError:
+        return
+
+    header = {
+        "event": "shutdown_watchdog_fired",
+        "pid": os.getpid(),
+        "delay_s": delay_s,
+        "fired_at": datetime.now(timezone.utc).isoformat(),
+        "snapshot": snapshot or {},
+    }
+    try:
+        with open(dump_path, "a", encoding="utf-8") as fh:
+            fh.write(json.dumps(header, default=str) + "\n")
+            fh.write("--- faulthandler dump (all threads) ---\n")
+            fh.flush()
+            try:
+                faulthandler.dump_traceback(file=fh, all_threads=True)
+            except Exception:
+                fh.write("(faulthandler.dump_traceback failed)\n")
+            fh.write("--- end dump ---\n")
+            fh.flush()
+    except Exception:
+        pass
+
+    # Also dump to stderr so journald/launchd capture it even if the file
+    # write failed (wedged disk was one of the #66892 hypotheses).
+    try:
+        sys.stderr.write(
+            f"Gateway shutdown watchdog fired after {delay_s:.0f}s "
+            f"(pid={os.getpid()}); dumping all thread stacks.\n"
+        )
+        sys.stderr.flush()
+        faulthandler.dump_traceback(all_threads=True)
+    except Exception:
+        pass
+
+
+def arm_shutdown_watchdog(
+    delay_s: float,
+    *,
+    done_event: Optional[threading.Event] = None,
+    snapshot_fn: Optional[Callable[[], Dict[str, Any]]] = None,
+    exit_code: int = 1,
+    dump_path: Optional[Path] = None,
+    name: str = "gateway-shutdown-watchdog",
+) -> threading.Event:
+    """Arm a daemon-thread hard-exit backstop for a wedged shutdown path.
+
+    If ``done_event`` is set before ``delay_s`` elapses, the thread exits
+    quietly (normal / progressing shutdown completed). Otherwise it dumps
+    diagnostics and calls ``os._exit(exit_code)``.
+
+    Never raises. Returns the ``done_event`` (creating one when omitted) so
+    the caller can disarm on successful completion.
+    """
+    done = done_event if done_event is not None else threading.Event()
+    try:
+        delay = max(float(delay_s), 0.0)
+    except (TypeError, ValueError):
+        delay = DEFAULT_SHUTDOWN_WATCHDOG_GRACE_S
+
+    if delay <= 0:
+        return done
+
+    def _watchdog() -> None:
+        # Wait with interruptible chunks so a late disarm doesn't need the
+        # full remaining sleep to observe done_event.
+        deadline = time.monotonic() + delay
+        while time.monotonic() < deadline:
+            remaining = deadline - time.monotonic()
+            if done.wait(timeout=min(remaining, 1.0)):
+                return
+        if done.is_set():
+            return
+
+        snapshot: Optional[Dict[str, Any]] = None
+        if snapshot_fn is not None:
+            try:
+                snapshot = snapshot_fn()
+            except Exception as exc:
+                snapshot = {"snapshot_error": repr(exc)}
+
+        target = dump_path if dump_path is not None else get_shutdown_watchdog_dump_path()
+        _write_watchdog_dump(target, delay_s=delay, snapshot=snapshot)
+
+        try:
+            logger.critical(
+                "Shutdown watchdog fired after %.0fs — forcing process exit "
+                "(asyncio drain path appears wedged; see %s)",
+                delay,
+                target,
+            )
+        except Exception:
+            pass
+
+        for stream in (sys.stdout, sys.stderr):
+            try:
+                stream.flush()
+            except Exception:
+                pass
+        os._exit(exit_code)
+
+    try:
+        threading.Thread(target=_watchdog, daemon=True, name=name).start()
+    except Exception:
+        logger.debug("Failed to arm shutdown watchdog", exc_info=True)
+    return done
+
+
+async def loop_heartbeat_forever(
+    *,
+    interval_s: float = DEFAULT_HEARTBEAT_INTERVAL_S,
+    start_time: Optional[float] = None,
+    home: Optional[Path] = None,
+    should_continue: Optional[Callable[[], bool]] = None,
+) -> None:
+    """Rewrite the loop heartbeat file on a cadence until cancelled / gated off.
+
+    Runs as an asyncio task on the gateway loop — if the loop freezes, this
+    task stops and the file mtime/updated_at goes stale for external monitors.
+    """
+    try:
+        interval = max(float(interval_s), 1.0)
+    except (TypeError, ValueError):
+        interval = DEFAULT_HEARTBEAT_INTERVAL_S
+
+    # Immediate first write so monitors see a fresh file as soon as the
+    # gateway is running, not after the first interval.
+    write_loop_heartbeat(start_time=start_time, home=home)
+    while True:
+        if should_continue is not None and not should_continue():
+            return
+        await asyncio.sleep(interval)
+        if should_continue is not None and not should_continue():
+            return
+        write_loop_heartbeat(start_time=start_time, home=home)