Gateway shutdown, restart, and hang hardening

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #66955 fix(gateway): thread shutdown watchdog + loop heartbeat for frozen SIGTERM drain (#66892)

  merged to main

  files: gateway/run.py, gateway/shutdown_watchdog.py, tests/gateway/test_shutdown_watchdog.py

  Show the diff

  • gateway/run.py

    @@ -1917,6 +1917,12 @@ def _platform_has_bot_credential(platform: "Platform", platform_config: "Platfor
         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 @@ def __init__(self, config: Optional[GatewayConfig] = None):
             # 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 @@ async def start(self) -> bool:
     
             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 @@ def _kill_tool_subprocesses(phase: str) -> None:
                     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,272 @@
    +"""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
    +        # Mirror _exit_after_graceful_shutdown: release PID file + runtime
    +        # lock BEFORE the log drain (locks must never be stranded), then
    +        # drain the async log queue so the logger.critical above actually
    +        # reaches the file before os._exit bypasses atexit. (#66892)
    +        try:
    +            from gateway.status import remove_pid_file, release_gateway_runtime_lock
    +            remove_pid_file()
    +            release_gateway_runtime_lock()
    +        except Exception:
    +            pass
    +        try:
    +            from hermes_logging import drain_log_queue
    +            drain_log_queue(timeout=1.0)
    +        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)

  • tests/gateway/test_shutdown_watchdog.py

    @@ -0,0 +1,150 @@
    +"""Shutdown watchdog + loop heartbeat coverage for #66892.
    +
    +The drain path is asyncio-based; a frozen loop makes every asyncio timeout
    +structurally unable to fire. These tests pin the out-of-loop backstop
    +(thread watchdog) and the loop-liveness heartbeat file contract.
    +"""
    +
    +from __future__ import annotations
    +
    +import asyncio
    +import json
    +import threading
    +import time
    +from unittest.mock import patch
    +
    +import pytest
    +
    +from gateway.shutdown_watchdog import (
    +    DEFAULT_SHUTDOWN_WATCHDOG_GRACE_S,
    +    arm_shutdown_watchdog,
    +    get_loop_heartbeat_path,
    +    get_shutdown_watchdog_dump_path,
    +    loop_heartbeat_forever,
    +    resolve_shutdown_watchdog_delay,
    +    write_loop_heartbeat,
    +)
    +
    +def test_resolve_shutdown_watchdog_delay_adds_grace():
    +    assert resolve_shutdown_watchdog_delay(180) == 180 + DEFAULT_SHUTDOWN_WATCHDOG_GRACE_S
    +    assert resolve_shutdown_watchdog_delay(0) == DEFAULT_SHUTDOWN_WATCHDOG_GRACE_S
    +    assert resolve_shutdown_watchdog_delay("bad") == DEFAULT_SHUTDOWN_WATCHDOG_GRACE_S
    +    assert resolve_shutdown_watchdog_delay(10, grace_s=5) == 15.0
    +
    +
    +def test_write_loop_heartbeat_atomic_json(tmp_path, monkeypatch):
    +    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
    +    path = write_loop_heartbeat(pid=4242, start_time=100.5, home=tmp_path)
    +    assert path == tmp_path / "state" / "gateway.heartbeat"
    +    assert path.is_file()
    +    data = json.loads(path.read_text(encoding="utf-8"))
    +    assert data["pid"] == 4242
    +    assert data["start_time"] == 100.5
    +    assert "updated_at" in data
    +    assert "monotonic" in data
    +    assert get_loop_heartbeat_path(tmp_path) == path
    +
    +
    +def test_arm_shutdown_watchdog_disarm_before_fire(tmp_path):
    +    done = threading.Event()
    +    exited = []
    +
    +    def fake_exit(code):
    +        exited.append(code)
    +        raise _ExitCalled(code)
    +
    +    with patch("gateway.shutdown_watchdog.os._exit", side_effect=fake_exit):
    +        arm_shutdown_watchdog(
    +            0.4,
    +            done_event=done,
    +            dump_path=tmp_path / "dump.log",
    +            exit_code=7,
    +        )
    +        time.sleep(0.1)
    +        done.set()
    +        time.sleep(0.5)
    +
    +    assert exited == []
    +
    +
    +def test_arm_shutdown_watchdog_fires_with_dump_and_exit(tmp_path):
    +    done = threading.Event()
    +    fired = threading.Event()
    +    dump = tmp_path / "logs" / "watchdog.log"
    +    snapshot_calls = []
    +    exit_codes = []
    +
    +    def snapshot():
    +        snapshot_calls.append(1)
    +        return {"active_agents": 1, "draining": True}
    +
    +    def fake_exit(code):
    +        exit_codes.append(code)
    +        fired.set()
    +
    +    with patch("gateway.shutdown_watchdog.os._exit", side_effect=fake_exit):
    +        arm_shutdown_watchdog(
    +            0.15,
    +            done_event=done,
    +            snapshot_fn=snapshot,
    +            dump_path=dump,
    +            exit_code=9,
    +        )
    +        assert fired.wait(timeout=5.0), "watchdog did not fire"
    +
    +    assert exit_codes == [9]
    +    assert snapshot_calls == [1]
    +    assert dump.is_file()
    +    text = dump.read_text(encoding="utf-8")
    +    assert "shutdown_watchdog_fired" in text
    +    assert "faulthandler dump" in text
    +    assert get_shutdown_watchdog_dump_path(tmp_path).name == "gateway-shutdown-watchdog.log"
    +
    +
    [redacted:email]
    +async def test_loop_heartbeat_rewrites_until_cancelled(tmp_path):
    +    path = get_loop_heartbeat_path(tmp_path)
    +    task = asyncio.create_task(
    +        loop_heartbeat_forever(
    +            interval_s=0.05,
    +            start_time=12.0,
    +            home=tmp_path,
    +        )
    +    )
    +    try:
    +        # First write is immediate.
    +        for _ in range(50):
    +            if path.is_file():
    +                break
    +            await asyncio.sleep(0.02)
    +        assert path.is_file()
    +        first = path.read_text(encoding="utf-8")
    +        assert json.loads(first)["start_time"] == 12.0
    +
    +        # Poll until a refresh lands (monotonic / updated_at change).
    +        second = first
    +        for _ in range(100):
    +            await asyncio.sleep(0.03)
    +            second = path.read_text(encoding="utf-8")
    +            if second != first:
    +                break
    +        assert second != first
    +        assert json.loads(second)["start_time"] == 12.0
    +    finally:
    +        task.cancel()
    +        with pytest.raises(asyncio.CancelledError):
    +            await task
    +
    +
    +def test_gateway_runner_exposes_shutdown_watchdog_state():
    +    """Attrs used by stop()/start() exist after normal construction hooks."""
    +    from gateway.run import GatewayRunner
    +
    +    runner = object.__new__(GatewayRunner)
    +    runner._shutdown_watchdog_done = threading.Event()
    +    runner._loop_heartbeat_task = None
    +    runner._gateway_started_at = time.time()
    +    assert not runner._shutdown_watchdog_done.is_set()
    +    runner._shutdown_watchdog_done.set()
    +    assert runner._shutdown_watchdog_done.is_set()
    +    assert runner._loop_heartbeat_task is None