diff --git a/docs/pr-assets/66892-shutdown-drain-watchdog.png b/docs/pr-assets/66892-shutdown-drain-watchdog.png new file mode 100644 index 000000000000..cb97f34e40be Binary files /dev/null and b/docs/pr-assets/66892-shutdown-drain-watchdog.png differ diff --git a/gateway/run.py b/gateway/run.py index 4cd3a94d17b7..b447d329ee4f 100644 --- a/gateway/run.py +++ b/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 diff --git a/gateway/shutdown_watchdog.py b/gateway/shutdown_watchdog.py new file mode 100644 index 000000000000..090e76c0561c --- /dev/null +++ b/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 ``/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 ``/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)