hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 28 Packages Projects Releases Wiki Activity
Files
main
hermes-agent/gateway/run_shutdown.py
Hermes Agent ec5c9c738a fix(processes): persist_on_release keeps background jobs alive across lifecycle kill sweeps (#41225) 
Background processes spawned with terminal(background=true) are killed from
three agent-lifecycle sweeps: agent release()'s kill_all, a gateway turn
timeout's kill_started_since, and agent close's owned-process loop. Jobs the
user explicitly wants to outlive the session (overnight batches, watchful
daemons) had no way to opt out.

Add terminal(background=true, persist_on_release=true):
- ProcessSession.persist_on_release, stamped by spawn_local/spawn_via_env,
  carried in crash-recovery checkpoints and exposed via list_sessions()
- kill_all skips persisted sessions only for lifecycle sources
  (_LIFECYCLE_KILL_SOURCES: kill_all, gateway_turn_timeout, agent_close);
  explicit operator stops (process_manage kill, /stop slash + RPC mirror,
  CLI /stop) now pass distinct sources so they still reach persisted jobs
- the agent_close owned-process loop in _close_task_resources skips
  persisted sessions the same way
- gateway shutdown keeps killing persisted jobs (source=gateway_shutdown):
  the host is going away and survivors would become PPID=1 orphans

Co-authored-by: salvaged from #109846 (persist_on_release plumbing) and
extended to the turn-timeout and agent_close paths.
2026-09-25 13:49:31 -05:00

2229 lines
120 KiB
Python
Raw Permalink Blame History

"""Stop/drain/restart, scale-to-zero and active-work accounting methods for GatewayRunner.
Split out of ``gateway/run.py``; bound onto ``GatewayRunner`` via the MRO.
``gateway.run`` internals are imported lazily inside method bodies (import cycle),
so ``patch("gateway.run.X")`` keeps intercepting them at call time.
"""
from __future__ import annotations
import asyncio
import dataclasses
import json
import logging
import os
import shlex
import sys
import threading
import time
from contextlib import contextmanager, nullcontext, suppress
from contextvars import Context
from pathlib import Path
from typing import Any, Callable, Dict, Optional
from gateway.config import Platform
from gateway.restart import (
DEFAULT_GATEWAY_CRON_DRAIN_TIMEOUT, GATEWAY_SERVICE_RESTART_EXIT_CODE,
effective_stop_drain_timeout, effective_stop_watchdog_delay, resolve_cron_drain_budget
)
from gateway.run_common import _UNSET
from gateway.shutdown_watchdog import arm_shutdown_watchdog, resolve_shutdown_watchdog_delay
# Log-record parity with the origin module.
logger = logging.getLogger("gateway.run")
def _exit_with_failure_verdict(runner) -> bool:
"""True (after logging the reason) when the runner asked for a failure exit."""
if not runner.should_exit_with_failure:
return False
if runner.exit_reason:
logger.error("Gateway exiting with failure: %s", runner.exit_reason)
return True
def _resolve_gateway_exit_verdict(runner, signal_initiated_shutdown: bool) -> bool:
"""Resolve the process verdict after either startup abort or normal shutdown."""
if _exit_with_failure_verdict(runner):
return False
if runner.exit_code is not None:
raise SystemExit(runner.exit_code)
if signal_initiated_shutdown and not runner._restart_requested:
logger.info(
"Exiting with code 1 (signal-initiated shutdown without restart "
"request) so the service manager can revive the gateway."
)
return False
# Older restart paths may not set ``runner.exit_code``; retain the service-restart fallback.
if runner._restart_via_service:
logger.info(
"Exiting with code %d (service-restart requested) so the service "
"manager relaunches the gateway.",
GATEWAY_SERVICE_RESTART_EXIT_CODE,
)
raise SystemExit(GATEWAY_SERVICE_RESTART_EXIT_CODE)
return True
# Windows has no bash/setsid chain: a tiny detached Python watcher waits for the gateway PID to
# exit (bounded), then spawns ``hermes gateway restart``.
_WINDOWS_RESTART_WATCHER = """
import os, subprocess, sys, time
from hermes_cli._subprocess_compat import windows_detach_flags_without_breakaway
pid = int(sys.argv[1])
restart_after_s = float(sys.argv[2])
cmd = sys.argv[3:]
deadline = time.monotonic() + restart_after_s
def _alive(p):
# On Windows, os.kill(pid, 0) is NOT a no-op — it maps to
# GenerateConsoleCtrlEvent(0, pid) (bpo-14484). Use the
# Win32 handle-based existence check instead.
if os.name == 'nt':
import ctypes
k32 = ctypes.windll.kernel32
k32.OpenProcess.restype = ctypes.c_void_p
k32.WaitForSingleObject.restype = ctypes.c_uint
k32.GetLastError.restype = ctypes.c_uint
h = k32.OpenProcess(0x1000 | 0x100000, False, int(p))
if not h:
return k32.GetLastError() != 87
try:
return k32.WaitForSingleObject(h, 0) == 0x102
finally:
k32.CloseHandle(h)
try:
os.kill(int(p), 0)
return True
except ProcessLookupError:
return False
except PermissionError:
return True
except OSError:
return False
while time.monotonic() < deadline:
if not _alive(pid):
break
time.sleep(0.2)
subprocess.Popen(
cmd,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
creationflags=windows_detach_flags_without_breakaway(),
)
""".strip()
@contextmanager
def _log_suppressed(level: int, msg: str, *args, exc_info: bool = False):
"""``suppress(Exception)`` that logs the swallowed exception on ``gateway.run``.
Without ``exc_info`` the exception is appended as the last ``%s`` argument (``msg % (*args, exc)``);
with it the traceback is attached instead. Best-effort seams use this everywhere a failure must be
visible in the log but must never propagate.
"""
try:
yield
except Exception as exc:
if exc_info:
logger.log(level, msg, *args, exc_info=(type(exc), exc, exc.__traceback__))
else:
logger.log(level, msg, *args, exc)
def _send_failed(result: Any) -> bool:
"""True when an adapter ``send()`` result explicitly reports failure."""
return result is not None and getattr(result, "success", True) is False
def _send_error(result: Any) -> str:
"""Error text of a failed ``send()`` result (adapters may omit it)."""
return getattr(result, "error", "send returned success=False")
def _notice_target_key(platform_value: str, chat_id, thread_id) -> tuple:
"""Dedup key for one notice destination: thread/topic platforms share a chat but route apart."""
return (platform_value, str(chat_id), str(thread_id) if thread_id else None)
def _effective_watchdog_leash(runner: object) -> float:
"""Thread-watchdog leash for the stop in progress: effective drain + grace, clamped under
launchd's live ``ExitTimeOut`` minus the dump margin. Lives here (not in gateway.restart)
because restart.py cannot import shutdown_watchdog without a cycle."""
return effective_stop_watchdog_delay(runner, resolve_shutdown_watchdog_delay(effective_stop_drain_timeout(runner)))
class GatewayShutdownMixin:
"""Stop/drain/restart, scale-to-zero and active-work accounting methods for GatewayRunner."""
@dataclasses.dataclass
class _StopContext:
"""State threaded through the ``_stop_*`` phases of one ``stop()`` run."""
deferred_count: Callable[[], int]
started_at: Optional[float] = None
active_agents: dict = dataclasses.field(default_factory=dict)
timed_out: bool = False
drain_elapsed: float = 0.0
# API-server runs still live when the adapters were released; the adapter map is empty by the
# time the SessionDB close gate runs, so the count has to be taken before ``adapters.clear()``.
api_live: int = 0
def elapsed(self) -> float:
return time.monotonic() - self.started_at
# Active-work accounting
def _active_work_count(self) -> int:
"""All agent work the gateway must expose and drain as one total."""
return (
self._running_agent_count()
+ self._active_cron_job_count()
+ self._active_api_run_count()
+ self._active_deferred_agent_worker_count()
)
@staticmethod
def _running_cron_job_count() -> int:
# The FULL work aggregate, not _running_agent_count(): cron jobs run on the scheduler's own thread
# pool and API-server runs live on the adapter — both outside _running_agents (the #60432 blind
# spot), so counting agents alone let a suspend land mid-cron-job. Fail-AWAKE accounting: the shared
# shutdown-drain counters (_active_cron_job_count/_active_api_run_count) swallow exceptions to 0,
# which is fine for a drain but unsafe for a suspend predicate — a transient read failure would make
# live work look idle and reopen the mid-job freeze. Here an unreadable source counts as work
# (sentinel 1) so the machine stays awake until the source is readable again.
from cron.scheduler import get_running_job_ids
return len(get_running_job_ids())
def _active_cron_job_count(self) -> int:
"""Cron jobs currently executing — they run outside ``_running_agents``; 0 if cron can't import.
Cron jobs run through a standalone ``AIAgent`` on the scheduler's own thread pool
(``cron/scheduler.py::run_job``), entirely outside ``self._running_agents`` — the dict every OTHER
active-work check on this class (``_running_agent_count``, ``_drain_active_agents``) reads. Without
this, the shutdown drain is structurally blind to in-flight cron work: it can report
``active_at_start=0`` and proceed straight to killing tool subprocesses while a cron job's terminal
command is still running (#60432). Best-effort: returns 0 if the cron module can't be imported (e.g.
a minimal test double for this class).
"""
try:
return self._running_cron_job_count()
except Exception:
return 0
def _api_server_hook(self, name: str, *args: Any) -> int:
"""Call the primary API-server adapter's ``name`` hook, clamped >= 0 (0 when the hook is absent).
Only the primary API server owns the HTTP listener, so only it is a source of this work.
"""
helper = getattr(getattr(self, "adapters", {}).get(Platform.API_SERVER), name, None)
return max(0, int(helper(*args))) if callable(helper) else 0
def _active_api_run_count(self) -> int:
"""API-server work that is outside ``_running_agents``."""
try:
return self._api_server_hook("active_agent_work_count")
except Exception:
return 0
def _active_api_worker_count(self) -> int:
"""API-server executor threads still inside an agent turn (#116535).
Module-level, like the cron registry above: the handler-side adapter count is already
unreachable here (``adapters`` was cleared a phase earlier) and, worse, drops on handler
cancellation while the worker thread lives on. Read live at the close gate instead of
snapshotting.
"""
try:
from gateway.platforms.api_server_runs import api_worker_live_count
return max(0, int(api_worker_live_count()))
except Exception:
return 0
def _interrupt_api_server_runs(self, reason: str) -> int:
"""Interrupt API-server agents not in ``_running_agents`` (same set ``_active_api_run_count`` counts)."""
try:
return self._api_server_hook("interrupt_active_runs", reason)
except Exception as exc:
logger.debug("Failed interrupting api_server runs during shutdown: %s", exc)
return 0
def _mark_api_runs_shutdown_requested(self) -> int:
"""Persist the shutdown boundary on API runs before the drain can await."""
try:
return self._api_server_hook("mark_shutdown_requested")
except Exception as exc:
logger.debug("Failed marking api_server runs as shutdown-requested: %s", exc)
return 0
def _active_deferred_agent_worker_count(self) -> int:
"""Executor workers that outlived their gateway turn (e.g. a timed-out hygiene compression)."""
workers = getattr(self, "_deferred_agent_workers", None)
if not isinstance(workers, dict):
return 0
return sum(1 for future in list(workers) if not future.done())
def _track_deferred_agent_worker(self, future: asyncio.Future, agent: Any) -> None:
"""Expose an executor worker to drain/interrupt until it really exits."""
workers = getattr(self, "_deferred_agent_workers", None)
if workers is None:
workers = self._deferred_agent_workers = {}
workers[future] = agent
def _discard_worker(done_future: asyncio.Future) -> None:
workers.pop(done_future, None)
# Workers that outlive their starting coroutine have no later waiter: consume the
# terminal exception so asyncio emits no unhandled-future warning.
# See #98973.
if not done_future.cancelled():
with suppress(Exception):
done_future.exception()
future.add_done_callback(_discard_worker)
def _interrupt_deferred_agent_workers(self, reason: str) -> int:
"""Request cancellation of detached executor-backed agent work."""
from gateway.run import _INTERRUPT_TOOL_REASON_GATEWAY_SHUTDOWN, request_hard_interrupt
workers = getattr(self, "_deferred_agent_workers", None)
if not isinstance(workers, dict):
return 0
interrupted = 0
seen: set[int] = set()
for future, agent in list(workers.items()):
if future.done() or agent is None or id(agent) in seen:
continue
seen.add(id(agent))
try:
request_hard_interrupt(agent, reason, tool_reason=_INTERRUPT_TOOL_REASON_GATEWAY_SHUTDOWN)
interrupted += 1
except Exception as exc:
logger.debug("Failed interrupting deferred agent worker during shutdown: %s", exc)
return interrupted
# Scale-to-zero idle detection / dormant-quiesce
def _scale_to_zero_has_live_background_work(self) -> bool:
"""Live background work (delegations, processes, pending watchers) that must block a suspend.
PERMANENT supervised watchers (_hermes_supervised_watcher, incl. the scale-to-zero watcher
itself) are excluded, else this would be True forever and the gateway could never go dormant.
"""
if any(
not t.done() and not getattr(t, "_hermes_supervised_watcher", False)
for t in self._background_tasks
):
return True
def _delegations_active() -> bool:
from tools.async_delegation import active_count
return active_count() > 0
def _processes_active() -> bool:
from tools.process_registry import process_registry
return bool(process_registry.has_any_active() or process_registry.pending_watchers)
for label, probe in (("async-delegation", _delegations_active), ("bg-work", _processes_active)):
with _log_suppressed(logging.DEBUG, f"scale-to-zero {label} check failed", exc_info=True):
if probe():
return True
return False
@staticmethod
def _gateway_cfg_section(name: str) -> Optional[dict]:
"""``gateway.<name>`` from the user config when it is a dict, else None (never raises)."""
from gateway.run import _load_gateway_config
try:
user_cfg = _load_gateway_config()
gw = user_cfg.get("gateway") if isinstance(user_cfg, dict) else None
section = gw.get(name) if isinstance(gw, dict) else None
except Exception: # noqa: BLE001
return None
return section if isinstance(section, dict) else None
def _scale_to_zero_idle_timeout_seconds(self) -> float:
from gateway.scale_to_zero import parse_idle_timeout_seconds
stz = self._gateway_cfg_section("scale_to_zero")
return parse_idle_timeout_seconds(stz.get("idle_timeout_minutes") if stz else None)
def _restart_loop_guard_config(self) -> tuple:
"""``(max_restarts, window_seconds, max_gap_seconds)`` for the restart-loop breaker.
``max_restarts <= 0`` disables it; ``max_gap_seconds`` is the longest spacing between
restart-interrupted boots that still counts as one loop.
"""
from gateway import restart_loop_guard as _rlg
rlg = self._gateway_cfg_section("restart_loop_guard") or {}
def _int_or(key: str, default: int, positive: bool) -> int:
value = rlg.get(key)
if isinstance(value, int) and (value > 0 or not positive):
return value
return default
return (
_int_or("max_restarts", _rlg.DEFAULT_MAX_RESTARTS, positive=False),
_int_or("window_seconds", _rlg.DEFAULT_WINDOW_SECONDS, positive=True),
_int_or("max_gap_seconds", _rlg.DEFAULT_MAX_GAP_SECONDS, positive=True),
)
def _scale_to_zero_active_messaging_platforms(self) -> list:
"""Return every enabled or live messaging platform across served profiles.
``self.config`` belongs to the launch profile, while ``_profile_adapters`` holds
live adapters for multiplexed secondary profiles. A direct secondary connection
must block suspension just like a direct primary connection. A secondary adapter
parked in ``_profile_failed_platforms`` (popped from ``_profile_adapters`` while a
retryable fatal reconnects) is still served: suspending mid-reconnect would leave
that reconnect unable to complete, so pending reconnects count as active too.
config.platforms is pre-seeded with disabled placeholders, and the api_server is
force-enabled on every hosted container (counting it silently disarmed the feature).
"""
non_messaging = {Platform.LOCAL, Platform.API_SERVER, Platform.WEBHOOK}
active = []
def add_platform(platform: Platform) -> None:
if platform not in non_messaging and platform not in active:
active.append(platform)
try:
if self.config:
for platform, platform_config in self.config.platforms.items():
if getattr(platform_config, "enabled", False):
add_platform(platform)
for platform in getattr(self, "adapters", {}) or {}:
add_platform(platform)
for profile_adapters in (getattr(self, "_profile_adapters", {}) or {}).values():
for platform in profile_adapters:
add_platform(platform)
for profile_pending in (getattr(self, "_profile_failed_platforms", {}) or {}).values():
for platform in profile_pending or {}:
add_platform(platform)
except Exception: # noqa: BLE001 - unreadable state must keep the gateway awake
logger.debug(
"scale-to-zero: active messaging platforms unreadable — staying awake",
exc_info=True,
)
return ["<unavailable>"]
return active
@staticmethod
def _relay_wake_url_or_none():
from gateway.relay import relay_wake_url
try:
return relay_wake_url()
except Exception: # noqa: BLE001
return None
def _scale_to_zero_should_arm(self) -> bool:
"""Whether to start the idle watcher (D1/D11/§3.4(1))."""
from gateway.scale_to_zero import messaging_is_relay_only_or_absent, scale_to_zero_enabled, should_arm
return should_arm(
enabled=scale_to_zero_enabled(),
relay_only_or_absent=messaging_is_relay_only_or_absent(self._scale_to_zero_active_messaging_platforms()),
wake_url=self._relay_wake_url_or_none(),
)
def _log_scale_to_zero_not_armed_reason(self) -> None:
"""One INFO line on why the idle watcher did NOT arm — only for an OPTED-IN instance."""
from gateway.scale_to_zero import messaging_is_relay_only_or_absent, scale_to_zero_enabled
try:
if not scale_to_zero_enabled():
return # not opted in — normal, stay quiet
active = [getattr(p, "value", p) for p in self._scale_to_zero_active_messaging_platforms()]
logger.info(
"scale-to-zero: NOT armed despite opt-in — relay_only_or_absent=%s (enabled platforms=%s), "
"wake_url=%s. Need relay-only messaging + a registered wake URL.",
messaging_is_relay_only_or_absent(active), active or "none",
"set" if self._relay_wake_url_or_none() else "MISSING",
)
except Exception: # noqa: BLE001 - diagnostics must never block startup
logger.debug("scale-to-zero: not-armed reason logging failed", exc_info=True)
def _scale_to_zero_is_idle(self) -> bool:
from gateway.scale_to_zero import is_idle
# FULL work aggregate with fail-AWAKE reads: the drain counters swallow errors to 0, which a
# suspend predicate would read as idle, so an unreadable source counts as work here.
def _read_or_awake(label: str, fn: Callable[[], Any], busy_sentinel: Any) -> Any:
try:
return fn()
except Exception: # noqa: BLE001 - unreadable source => assume busy
logger.debug("scale-to-zero: %s unreadable — staying awake", label, exc_info=True)
return busy_sentinel
cron_count = _read_or_awake("cron work count", self._running_cron_job_count, 1)
api_count = _read_or_awake("api work count", lambda: self._api_server_hook("active_agent_work_count"), 1)
# An attached dashboard/desktop/TUI client (heartbeat mtime) is inbound activity — folded into
# the inbound clock, not a conjunct, so a lingering marker cannot pin the box.
last_inbound = self._last_inbound_at
from gateway.scale_to_zero import dashboard_client_last_seen
seen = _read_or_awake("dashboard heartbeat", dashboard_client_last_seen, time.time())
if seen is not None and seen > last_inbound:
last_inbound = seen
return is_idle(
active_work_count=self._running_agent_count() + cron_count + api_count,
seconds_since_last_inbound=time.time() - last_inbound,
idle_timeout_seconds=self._scale_to_zero_idle_timeout_seconds(),
has_live_background_work=self._scale_to_zero_has_live_background_work(),
)
def _scale_to_zero_note_real_inbound(self) -> None:
"""Stamp real inbound and flip status back to running after a dormant wake.
Internal completion/replay events deliberately do not call this (they must not keep an idle
gateway awake).
"""
self._last_inbound_at = time.time()
if getattr(self, "_scale_to_zero_cooldown_until", 0.0) > 0:
self._scale_to_zero_status(self._serving_state(), "scale-to-zero: status restore failed")
self._scale_to_zero_cooldown_until = 0.0
def _scale_to_zero_status(self, state: str, fail_msg: str) -> None:
"""Best-effort runtime status write; failures are debug-logged with ``fail_msg``."""
try:
self._update_runtime_status(state)
except Exception: # noqa: BLE001 - status is best-effort
logger.debug(fail_msg, exc_info=True)
def _relay_adapter_for_dormancy(self):
"""Return the connected RELAY adapter, if any (the one go_dormant targets)."""
return self.adapters.get(Platform.RELAY)
async def _scale_to_zero_watcher(self, interval: float = 30.0) -> None:
"""Watch for idle, drive the relay dormant, then self-suspend. On sustained idle: status
`draining` (NOT _running=False), relay go_dormant() (socket close, NOT disconnect()), no
mark_resume_pending (suspend preserves RAM), THEN suspend via the flaps socket — Fly autostop
sees only INBOUND connections and would freeze mid-job. Without a flaps socket NAS brokers
the stop through the stamped GATEWAY_RELAY_SLEEP_URL; with no lever at all the watcher
abstains."""
from gateway.scale_to_zero import messaging_is_relay_only_or_absent
await asyncio.sleep(min(interval, 30.0)) # let startup settle
while self._running:
try:
await asyncio.sleep(interval)
if not self._running:
return
if time.time() < self._scale_to_zero_cooldown_until or not self._scale_to_zero_is_idle():
continue
# The arm gate ran once at boot. A direct adapter that came up since (profile
# reconcile hot-adding a secondary, a re-enabled platform) owns a socket no wake
# URL can revive, so the same gate is re-asked before every dormant sequence.
active = self._scale_to_zero_active_messaging_platforms()
if not messaging_is_relay_only_or_absent(active):
if not self._scale_to_zero_direct_platform_logged:
self._scale_to_zero_direct_platform_logged = True
logger.info(
"scale-to-zero: idle, but directly connected messaging platform(s) %s "
"hold a live socket that a suspended instance cannot wake from — staying "
"awake. Route them through the relay connector or disable them to allow "
"suspend.",
", ".join(str(getattr(p, "value", p)) for p in active),
)
continue
self._scale_to_zero_direct_platform_logged = False
go_dormant = getattr(self._relay_adapter_for_dormancy(), "go_dormant", None)
if not callable(go_dormant):
continue
# Quiesce only when a suspend can follow: otherwise the re-dial after the socket
# close just clears the flip again.
from gateway.scale_to_zero import suspend_available
if not suspend_available():
if not self._scale_to_zero_no_suspend_logged:
self._scale_to_zero_no_suspend_logged = True
logger.info(
"scale-to-zero: idle, but this platform offers no suspend lever (no "
"in-machine API and no brokered sleep URL); staying connected rather "
"than quiescing"
)
continue
logger.info(
"scale-to-zero: gateway idle for >= %.0fs — going dormant "
"(relay buffered, socket closed) then self-suspending",
self._scale_to_zero_idle_timeout_seconds(),
)
self._scale_to_zero_status("draining", "scale-to-zero: status mark failed")
# Both levers: the 1s dormant re-dial can beat either suspend and clear the flip.
# Held BEFORE go_dormant, whose close arms it.
if not self._scale_to_zero_hold_redial(True):
# Without the hold the re-dial can clear the flip before the stop lands, so
# refuse rather than suspend unprotected.
logger.warning(
"scale-to-zero: could not hold the relay re-dial — staying awake rather "
"than suspending unprotected"
)
self._scale_to_zero_abandon_suspend()
continue
dormant_ok = True
try:
result = go_dormant()
if asyncio.iscoroutine(result):
result = await result
# The going_idle ack. Without it inbound is NOT buffered, so suspending would
# freeze a live destination: the whole bug.
if result is not True:
dormant_ok = False
logger.warning(
"scale-to-zero: connector did not ack going_idle — staying awake "
"rather than freezing a live destination"
)
except Exception: # noqa: BLE001 - dormancy is best-effort
dormant_ok = False
logger.debug("scale-to-zero: go_dormant failed", exc_info=True)
# After a wake the drained inbound updates _last_inbound_at; give it a window so we
# don't immediately re-go-dormant on the same idle reading before traffic lands.
self._scale_to_zero_cooldown_until = time.time() + max(interval, 60.0)
# Suspend ONLY after an ACKED quiesce (else inbound black-holes while we sleep), and
# re-check idle — inbound may have landed during the quiesce await.
if not dormant_ok:
self._scale_to_zero_abandon_suspend()
continue
if not self._scale_to_zero_is_idle():
logger.info("scale-to-zero: inbound arrived during quiesce — skipping suspend")
self._scale_to_zero_abandon_suspend()
continue
await self._scale_to_zero_self_suspend()
except asyncio.CancelledError:
raise
except Exception: # noqa: BLE001 - the watcher must never crash the gateway
logger.debug("scale-to-zero watcher iteration error", exc_info=True)
async def _scale_to_zero_self_suspend(self) -> None:
"""Suspend this machine, in-guest where possible and via NAS otherwise (fail-awake).
Called ONLY after a clean, acked go_dormant(), with the re-dial already held.
"""
from gateway.scale_to_zero import (
brokered_sleep_url, request_brokered_suspend, self_suspend_available, suspend_self
)
try:
if self_suspend_available():
accepted = await asyncio.to_thread(suspend_self)
lever = "self-suspend"
if accepted:
# flaps answers seconds BEFORE the kernel freezes, so the fence has to span
# that gap.
await self._scale_to_zero_await_freeze_gap()
self._scale_to_zero_hold_redial(False)
else:
self._scale_to_zero_abandon_suspend()
else:
# No in-guest API (Azure ACA): NAS holds the credential for the stop verb and
# brokers it for us.
url = brokered_sleep_url()
if not url:
# The watcher held on our behalf; nothing is coming to freeze the machine, so
# a held supervisor would just stay offline.
self._scale_to_zero_abandon_suspend()
logger.debug(
"scale-to-zero: no suspend lever available — dormant without platform suspend"
)
return
# The watcher already holds the supervisor across this call.
accepted = await asyncio.to_thread(request_brokered_suspend, url)
lever = "brokered suspend"
if not accepted:
self._scale_to_zero_abandon_suspend()
if not accepted:
logger.warning(
"scale-to-zero: %s not accepted — machine stays awake (fail-awake); will "
"retry on the next idle window", lever,
)
except Exception: # noqa: BLE001 - suspend is best-effort, never crash
logger.debug("scale-to-zero: self-suspend failed", exc_info=True)
self._scale_to_zero_abandon_suspend()
async def _scale_to_zero_await_freeze_gap(self) -> None:
"""Hold the re-dial fence across the flaps-2xx -> kernel-freeze gap.
Sliced on the WALL clock rather than one ``asyncio.sleep`` because a Fly suspend stops
CLOCK_MONOTONIC while CLOCK_REALTIME keeps tracking host time. Measured on a Fly machine
(gru, 2026-09-03) across a 252.219s freeze: ``time.monotonic()`` advanced 0.501s,
``time.time()`` advanced 252.219s. ``asyncio.sleep`` runs on ``loop.time()`` (monotonic),
so a single sleep would resume with its REMAINDER after the wake and delay the drain
re-dial by exactly that much, on every wake of every Fly agent.
On the wall clock the deadline is already past by the time we resume, so the fence costs
nothing after a freeze while still spanning the full gap before one. That decoupling is
what lets FLY_FREEZE_GRACE_S be sized for the slowest (largest-RAM) machine.
"""
from gateway.scale_to_zero import FLY_FREEZE_GRACE_S, FLY_FREEZE_GRACE_TICK_S
deadline = time.time() + FLY_FREEZE_GRACE_S
while time.time() < deadline:
await asyncio.sleep(FLY_FREEZE_GRACE_TICK_S)
def _scale_to_zero_abandon_suspend(self) -> None:
"""Undo a quiesce we are not going to follow with a suspend.
All three together: a released supervisor still advertising `draining` reads as
mid-shutdown until the next real inbound event, and an abort that skips the cooldown
re-runs on every tick.
"""
self._scale_to_zero_hold_redial(False)
# Same guard as _exit_external_drain: a real shutdown drain must win, so never resurrect
# a stopping gateway to `running`.
if not getattr(self, "_draining", False) and self._running:
self._scale_to_zero_status(self._serving_state(), "scale-to-zero: status restore failed")
# An abort before the cooldown is set would otherwise retry every tick.
self._scale_to_zero_cooldown_until = max(
self._scale_to_zero_cooldown_until, time.time() + 60.0
)
def _scale_to_zero_hold_redial(self, held: bool) -> bool:
"""Hold or release the relay's reconnect supervisor. Returns whether the transport actually
took it, so the caller can refuse to suspend without the protection rather than fail open."""
try:
adapter = self._relay_adapter_for_dormancy()
if adapter is None:
return False
method = getattr(adapter, "hold_redial" if held else "release_redial", None)
if not callable(method):
return False
# Trust the adapter's answer rather than the absence of an exception: it deliberately
# never raises, so "did not throw" proves nothing.
return method() is True
except Exception: # noqa: BLE001 - never blocks the suspend it precedes
logger.debug("scale-to-zero: redial hold toggle failed", exc_info=True)
return False
# External drain control: the dashboard writes/removes ``.drain_request.json`` (gateway/drain_control.py);
# the watcher flips between accepting and refusing NEW turns WITHOUT exiting (reversible).
def _enter_external_drain(self) -> None:
"""Begin external drain: refuse NEW turns (in-flight ones are NOT interrupted). Idempotent."""
if self._external_drain_active:
return
self._external_drain_active = True
logger.info(
"External drain ENGAGED (.drain_request.json present) — refusing "
"new turns; %d in-flight turn(s) will finish. Process stays up.", self._active_work_count(),
)
# Persist "draining" so /api/status tracks it; active_agents is read-merged, only state changes.
self._update_runtime_status("draining")
def _exit_external_drain(self) -> None:
"""Cancel external drain: re-accept new turns. Idempotent; never resurrects a stopping gateway."""
if not self._external_drain_active:
return
self._external_drain_active = False
if self._draining or not self._running:
logger.info(
"External drain marker cleared during shutdown — not reverting "
"to running (shutdown takes precedence)."
)
return
logger.info(
"External drain RELEASED (.drain_request.json removed) — "
"re-accepting new turns; gateway_state -> %s.", self._serving_state(),
)
self._update_runtime_status(self._serving_state())
async def _drain_control_watcher(self, interval: float = 1.0) -> None:
"""Poll ``.drain_request.json`` at 1s: present -> enter drain, absent -> exit; a stale epoch = absent."""
from gateway.drain_control import drain_requested
while self._running:
try:
# Off-thread: a synchronous marker read at 1s cadence can stall 30s+ under host I/O
# pressure and take every platform heartbeat down.
if await asyncio.to_thread(drain_requested):
self._enter_external_drain()
# API and cron work live outside messaging's _running_agents map; refresh the
# aggregate while an external caller polls this reversible drain state.
self._persist_active_agents()
else:
self._exit_external_drain()
except asyncio.CancelledError:
raise
except Exception as exc:
logger.debug("Drain-control watcher tick error: %s", exc, exc_info=True)
await asyncio.sleep(interval)
def _update_platform_runtime_status(
self, platform: str, *, platform_state: Optional[str] = None,
error_code: Optional[str] = None, error_message: Optional[str] = None,
needs_attention: Optional[bool] = None, retrying_since: Any = _UNSET,
) -> None:
from gateway.run import _write_runtime_status_quiet
extra: Dict[str, Any] = {}
if needs_attention is not None:
extra["needs_attention"] = needs_attention
if retrying_since is not _UNSET:
extra["retrying_since"] = retrying_since
_write_runtime_status_quiet(
platform=platform, platform_state=platform_state, error_code=error_code,
error_message=error_message, **extra,
)
# Per-platform circuit breaker (pause/resume): reconnect watcher + /platform pause|resume.
def _pause_failed_platform(self, platform, *, reason: str = "") -> None:
"""Pause a queued platform (manual ``/platform pause`` only — the watcher never auto-pauses)."""
info = getattr(self, "_failed_platforms", {}).get(platform)
if info is None or info.get("paused"):
return
info["paused"] = True
info["pause_reason"] = reason or "auto-paused after repeated failures"
# next_retry=inf: a stale code path missing "paused" still never fires.
info["next_retry"] = float("inf")
self._update_platform_runtime_status(
platform.value, platform_state="paused", error_code=None, error_message=info["pause_reason"],
)
logger.warning(
"%s paused after %d consecutive failures (%s) — fix the underlying issue then run `/platform "
"resume %s` to retry, or `hermes gateway restart` to restart the gateway.",
platform.value, info.get("attempts", 0), info["pause_reason"], platform.value,
)
def _resume_paused_platform(self, platform) -> bool:
"""Unpause a platform (reset attempts, retry on the next watcher tick). True iff it was paused."""
info = getattr(self, "_failed_platforms", {}).get(platform)
if info is None or not info.get("paused"):
return False
info["paused"] = False
info.pop("pause_reason", None)
info["attempts"] = 0
info["next_retry"] = time.monotonic() # retry on next watcher tick
self._update_platform_runtime_status(platform.value, platform_state="retrying")
logger.info("%s resumed — retrying on next watcher tick", platform.value)
return True
# Drain / interrupt
def _drain_work_counts(self) -> tuple:
"""``(agents, cron, api, deferred)`` — the four sources the drain waits on."""
return (
self._running_agent_count(), self._active_cron_job_count(),
self._active_api_run_count(), self._active_deferred_agent_worker_count(),
)
async def _drain_active_agents(
self, timeout: float, cron_timeout: Optional[float] = None
) -> tuple[Dict[str, Any], bool]:
snapshot = self._snapshot_running_agents()
loop = asyncio.get_running_loop()
last_counts = self._drain_work_counts()
last_status_at = 0.0
def _maybe_update_status(force: bool = False) -> None:
nonlocal last_counts, last_status_at
now = loop.time()
counts = self._drain_work_counts()
if force or counts != last_counts or (now - last_status_at) >= 1.0:
self._update_runtime_status("draining")
last_counts, last_status_at = counts, now
# Cron/API/deferred work lives outside ``_running_agents``; fold it in or it is killed unwarned.
_cron0, _api0, _deferred0 = last_counts[1:]
_maybe_update_status(force=True)
if not self._running_agents and not (_cron0 or _api0 or _deferred0):
return snapshot, False
# Cron has its own deadline: a chat turn is announced+resumable; a killed cron run is a permanent failure.
# ``timeout`` (``restart_drain_timeout``) defaults to 0 because interrupting a chat turn is
# announced and resumable; a cron run killed mid-flight is recorded in jobs.json as a permanent
# failure nobody is waiting on. Sharing one budget meant the default config could report
# ``timed_out=True`` after 0.00s with a cron job in flight and kill it — the drain never even
# entered this loop (#82161).
started = loop.time()
deadline = started + timeout
cron_deadline = started + (timeout if cron_timeout is None else cron_timeout)
def _still_draining() -> bool:
now = loop.time()
agents, cron, api, deferred = self._drain_work_counts()
return bool(((agents or api or deferred) and now < deadline) or (cron and now < cron_deadline))
# Both budgets at 0 = an expired deadline (loop unentered), so timed_out still comes from real state.
while _still_draining():
_maybe_update_status()
await asyncio.sleep(0.1)
timed_out = any(self._drain_work_counts())
_maybe_update_status(force=True)
return snapshot, timed_out
def _interrupt_running_agents(self, reason: str) -> None:
from gateway.run import _AGENT_PENDING_SENTINEL, _INTERRUPT_TOOL_REASON_GATEWAY_SHUTDOWN, request_hard_interrupt
for session_key, agent in list(self._running_agents.items()):
if agent is _AGENT_PENDING_SENTINEL:
continue
with _log_suppressed(logging.DEBUG, "Failed interrupting agent during shutdown: %s"):
request_hard_interrupt(agent, reason, tool_reason=_INTERRUPT_TOOL_REASON_GATEWAY_SHUTDOWN)
logger.debug("Interrupted running agent for session %s during shutdown", session_key)
# API-server / desk turns are adapter-owned and never enter _running_agents, so the loop above
# cannot see them even though _drain_active_agents() waited for them.
for count, what in (
(self._interrupt_api_server_runs(reason), "api_server run(s)"),
(self._interrupt_deferred_agent_workers(reason), "deferred agent worker(s)"),
):
if count:
logger.debug("Interrupted %d %s during shutdown", count, what)
def _shutdown_interrupt_reason(self) -> str:
from gateway.run import _INTERRUPT_REASON_GATEWAY_RESTART, _INTERRUPT_REASON_GATEWAY_SHUTDOWN
return _INTERRUPT_REASON_GATEWAY_RESTART if self._restart_requested else _INTERRUPT_REASON_GATEWAY_SHUTDOWN
async def _mark_running_sessions_resume_pending(self, log_prefix: str) -> list:
"""Mark every non-pending running session resume_pending; returns the keys marked."""
from gateway.run import _AGENT_PENDING_SENTINEL
reason = "restart_timeout" if self._restart_requested else "shutdown_timeout"
marked: list[str] = []
# Pre-mark sessions as resume_pending BEFORE the drain wait. If the process is killed by the service
# manager during the drain, the durable marker is already written so the next gateway boot can
# recover in-flight sessions (#27856).
for _sk, _agent in list(self._running_agents.items()):
if _agent is _AGENT_PENDING_SENTINEL:
continue
with _log_suppressed(logging.DEBUG, "%s failed for %s: %s", log_prefix, _sk):
await self.async_session_store.mark_resume_pending(_sk, reason)
marked.append(_sk)
return marked
def _restart_notification_allowed(self, platform: Platform) -> bool:
"""False when the platform config sets ``gateway_restart_notification=false``."""
platform_cfg = self.config.platforms.get(platform)
return platform_cfg is None or bool(platform_cfg.gateway_restart_notification)
def _notice_allowed(self, platform: Platform, what: str) -> bool:
"""``_restart_notification_allowed`` with the INFO suppression line for shutdown notices."""
if self._restart_notification_allowed(platform):
return True
logger.info(
"Shutdown notification suppressed for %s: %s has gateway_restart_notification=false", what, platform.value,
)
return False
async def _notify_interrupted_cron_jobs(self, job_ids) -> int:
"""Tell the owner of each just-interrupted cron job that its run died; returns notices sent.
The cron worker can't (its thread reaches ``_deliver_result`` after teardown closed the
transport), so this runs post-interrupt while adapters are still connected. Best-effort.
Its thread reaches ``_deliver_result`` asynchronously, and by then ``_bounded_adapter_teardown`` has
closed the transport — so the notice never leaves the process, and ``_consume_interrupted_flag``
discards the resulting ``delivery_error`` along with it. The run's only trace is a line in jobs.json
nobody reads (#82232).
Must therefore be called from the post-interrupt phase, while adapters are still connected — the
same window ``_notify_active_sessions_of_shutdown`` relies on for chat sessions, which is blind to
cron work because cron runs on the scheduler's own thread pool rather than ``self._running_agents``
(#60432).
"""
if not job_ids:
return 0
try:
from cron.jobs import get_job
from cron.scheduler import _resolve_delivery_targets
except Exception as e:
logger.debug("Cron interrupt notification unavailable: %s", e)
return 0
action = "restarting" if self._restart_requested else "shutting down"
notified: set = set()
for job_id in job_ids:
try:
job = get_job(job_id)
if not job:
continue
# deliver=local / unresolvable-origin jobs resolve to zero targets and stay silent (no home-
# channel fallback). Interrupted notices are failure-category status: honor failure_deliver.
# See #43014.
targets = _resolve_delivery_targets(job, for_failure=True)
except Exception as e:
logger.debug("Cron interrupt targets unresolved for %s: %s", job_id, e)
continue
job_name = job.get("name") or job_id
msg = (
f"⚠️ Scheduled job '{job_name}' was cut short because Hermes is {action}; "
"no result this run. It will run again on schedule, or run it now with "
f"`hermes cron run {job_name}` once Hermes is back."
)
for target in targets or ():
try:
platform = Platform(str(target.get("platform", "")).lower())
except Exception:
continue
adapter = self.adapters.get(platform)
if adapter is None or not self._restart_notification_allowed(platform):
continue
chat_id = str(target.get("chat_id"))
thread_id = target.get("thread_id")
dedup_key = (job_id, *_notice_target_key(platform.value, chat_id, thread_id))
if dedup_key in notified:
continue
with _log_suppressed(logging.DEBUG, "Cron interrupt notice to %s:%s raised: %s", platform.value, chat_id):
metadata = self._thread_metadata_for_target(platform, chat_id, thread_id, adapter=adapter)
async def send_notice():
if await self._send_notice_logged(
adapter, chat_id, msg, platform.value, "Cron interrupt notice to %s:%s failed: %s",
"Cron interrupt notice to %s:%s raised: %s", metadata=metadata,
):
notified.add(dedup_key)
from gateway.warning_notifications import present_notification
await present_notification(send_notice, platform=platform)
if notified:
logger.info("Shutdown: delivered %d interrupted-cron-job notice(s)", len(notified))
return len(notified)
async def _shutdown_notification_target(self, session_key: str):
"""``(source, platform_str, chat_id, thread_id, profile)``: persisted origin > cached source >
parsed key. ``profile`` is the owning profile from the source or the ``agent:<profile>:`` key
namespace (``None`` = default) so the notice leaves through that profile's bot."""
from gateway.run import _parse_session_key
source = None
try:
if getattr(self, "session_store", None) is not None:
await self.async_session_store._ensure_loaded()
entry = self.session_store._entries.get(session_key)
source = getattr(entry, "origin", None) if entry else None
except Exception as e:
logger.debug("Failed to load session origin for shutdown notification %s: %s", session_key, e)
if source is None:
source = self._get_cached_session_source(session_key)
if source is not None:
return source, source.platform.value, str(source.chat_id), source.thread_id, getattr(source, "profile", None)
_parsed = _parse_session_key(session_key)
if not _parsed:
return None
return None, _parsed["platform"], _parsed["chat_id"], _parsed.get("thread_id"), _parsed.get("profile")
async def _send_shutdown_notice(
self, adapter, chat_id: str, msg: str, kind: str, platform_str: str, **send_kwargs
) -> bool:
"""Send one shutdown notice; True when delivered. Failures are debug-logged, never raised."""
where = "home channel " if kind == "home channel" else ""
fail_fmt = f"Failed to send shutdown notification to {where}%s:%s: %s"
if not await self._send_notice_logged(adapter, chat_id, msg, platform_str, fail_fmt, **send_kwargs):
return False
logger.info("Sent shutdown notification to %s %s:%s", kind, platform_str, chat_id)
return True
@staticmethod
async def _send_notice_logged(
adapter, chat_id: str, msg: str, platform_str: str, fail_fmt: str, raise_fmt: Optional[str] = None, **kw
) -> bool:
"""``adapter.send`` whose failure is debug-logged as ``fmt % (platform, chat, error)`` — ``fail_fmt``
for success=False, ``raise_fmt`` (default ``fail_fmt``) for a raise; True only on a delivered send.
Every shutdown notice races live turns, so it always carries the interim marker (#98432)."""
from gateway.run import _interim_metadata
kw["metadata"] = _interim_metadata(kw.get("metadata"))
try:
result = await adapter.send(chat_id, msg, **kw)
except Exception as e:
logger.debug(raise_fmt or fail_fmt, platform_str, chat_id, e)
return False
if _send_failed(result):
logger.debug(fail_fmt, platform_str, chat_id, _send_error(result))
return False
return True
async def _notify_active_sessions_of_shutdown(self) -> None:
"""Send shutdown/restart notifications to active chats and home channels.
Called at the start of stop() while adapters are connected; send failures never block shutdown.
"""
restart_source = self._restart_command_source if self._restart_requested else None
msg = (
"⚠️ Hermes is shutting down — your current task will be interrupted. "
"When it is back online, send any message and I'll try to pick up where we left off."
)
if self._restart_requested:
msg = (
"⚠️ Hermes is restarting — your current task will be interrupted. "
"Send any message after the restart and I'll try to resume where you left off."
)
restart_key = None
if restart_source is not None:
with suppress(Exception):
restart_key = _notice_target_key(
restart_source.platform.value, restart_source.chat_id, restart_source.thread_id
)
notified: set[tuple[str, str, Optional[str]]] = set()
for session_key in self._snapshot_running_agents():
target = await self._shutdown_notification_target(session_key)
if target is None:
continue
source, platform_str, chat_id, thread_id, profile = target
dedup_key = _notice_target_key(platform_str, chat_id, thread_id)
if dedup_key in notified:
continue
try:
platform = Platform(platform_str)
# The session's OWN profile's bot (transport ref → profile map), never a bare
# self.adapters hit: under multiplex that is the default bot, so a secondary session's
# "Gateway shutting down" would land in the user's chat with the wrong bot.
adapter = self._delivery_adapter_for(source) if source is not None else None
if adapter is None:
adapter = self._authorization_adapter(platform, profile)
if not adapter:
continue
if not self._notice_allowed(platform, "active session"):
continue
reply_to_message_id = getattr(source, "message_id", None)
if reply_to_message_id is None and restart_key == dedup_key:
reply_to_message_id = getattr(restart_source, "message_id", None)
metadata = self._thread_metadata_for_target(
platform, chat_id, thread_id, chat_type=getattr(source, "chat_type", None),
reply_to_message_id=reply_to_message_id, adapter=adapter,
)
except Exception as e:
logger.debug("Failed to send shutdown notification to %s:%s: %s", platform_str, chat_id, e)
continue
# Automatic interrupt diagnostic, resolved under the session's own profile scope (same
# shape as the stall watcher). The requester's own chat on an in-chat /restart is the
# requested outcome of that command and is never suppressed.
async def _send_active(adapter=adapter, chat_id=chat_id, platform_str=platform_str,
metadata=metadata, dedup_key=dedup_key):
if await self._send_shutdown_notice(
adapter, chat_id, msg, "active chat", platform_str, metadata=metadata
):
notified.add(dedup_key)
from gateway.warning_notifications import present_notification
from gateway.run import _async_profile_runtime_scope
scope = (_async_profile_runtime_scope(self._resolve_profile_home_for_source(source))
if source is not None else nullcontext())
async with scope:
presented = await present_notification(_send_active, platform=platform, diagnostic=restart_key != dedup_key)
if not presented:
notified.add(dedup_key) # suppressed: latch so the home-channel pass does not re-target it
if self._restart_requested and restart_source is not None:
logger.debug("Skipping home-channel shutdown notifications for in-chat restart")
return
# A quiet drain (routine fleet auto-update) suppresses ONLY the home-channel broadcast; per-session
# pings above stay. Current-epoch marker only; a failing check fails toward the louder behaviour.
with _log_suppressed(logging.DEBUG, "drain_notification_suppressed check failed: %s"):
from gateway.drain_control import drain_notification_suppressed
if drain_notification_suppressed():
logger.info(
"Home-channel shutdown broadcast suppressed by drain marker (suppress_notification=true)"
)
return
# Snapshot adapters: adapter.send() can hit a fatal path (_handle_fatal) that pops the adapter
# from self.adapters -> ``RuntimeError: dictionary changed size during iteration``.
for platform, adapter in list(self.adapters.items()):
home = self.config.get_home_channel(platform)
if not home or not home.chat_id:
continue
if not self._notice_allowed(platform, "home channel"):
continue
dedup_key = _notice_target_key(platform.value, home.chat_id, home.thread_id)
if dedup_key in notified:
continue
try:
metadata = self._thread_metadata_for_target(platform, home.chat_id, home.thread_id, adapter=adapter)
except Exception as e:
logger.debug(
"Failed to send shutdown notification to home channel %s:%s: %s", platform.value, home.chat_id, e,
)
continue
async def _send_home(adapter=adapter, home=home, platform=platform, metadata=metadata):
if await self._send_shutdown_notice(
adapter, str(home.chat_id), msg, "home channel", platform.value, metadata=metadata,
):
notified.add(dedup_key)
from gateway.warning_notifications import present_notification
await present_notification(_send_home, platform=platform)
# Agent finalization / resource cleanup
@staticmethod
def _flush_agent_transcript_at_shutdown(agent: Any) -> None:
"""Persist an in-flight transcript before teardown.
A force-interrupted agent may never reach finalize_turn (the only mid-turn flush), so its
tool rounds would vanish on resume. Idempotent; gracefully finished agents re-flush nothing.
"""
with _log_suppressed(logging.DEBUG, "Shutdown transcript flush failed: %s"):
# Persist any in-flight transcript to the SQLite session store before teardown (#13121). An
# agent forcibly interrupted by the drain-timeout escalation may never reach
# ``turn_finalizer.finalize_turn`` (the only place that flushes the turn to state.db) — e.g. it
# was blocked in a tool call that did not abort within the post-interrupt grace window. Its
# in-flight tool rounds live only in the in-memory ``_session_messages`` (refreshed per tool
# round in ``conversation_loop`` but never written to SQLite mid-turn), so the immediate
# pre-restart turn is silently dropped from ``load_transcript()`` on resume. Flushing here
# closes that gap; the resume_pending / fresh-tool-tail branches in
# ``_handle_message_with_agent`` already expect a transcript whose tail may be a pending tool
# result.
_flush = getattr(agent, "_flush_messages_to_session_db", None)
_session_messages = getattr(agent, "_session_messages", None)
if not (callable(_flush) and isinstance(_session_messages, list) and _session_messages):
return
# Strip empty-response retry scaffolding from the tail first (as ``_persist_session``
# does) so a resumed turn doesn't replay synthetic recovery nudges.
_strip = getattr(agent, "_drop_trailing_empty_response_scaffolding", None)
if callable(_strip):
with suppress(Exception):
_strip(_session_messages)
try:
_flush(_session_messages)
except Exception as _flush_err:
# Transcript could not be persisted (e.g. FTS/SQLite corruption): dump the live history
# to a JSON recovery snapshot rather than lose it. Non-fatal.
logger.warning(
"Shutdown transcript flush failed (%s); preserving %d in-memory message(s) to recovery snapshot",
_flush_err, len(_session_messages),
)
from gateway.shutdown_flush import flush_agent_history_to_file
flush_agent_history_to_file(getattr(agent, "session_id", None), _session_messages)
async def _finalize_shutdown_agents(self, active_agents: Dict[str, Any]) -> None:
for session_key, agent in active_agents.items():
self._flush_agent_transcript_at_shutdown(agent)
# Off-loop + bounded: plugin on_session_finalize hooks can do arbitrary synchronous work
# (e.g. a full-session trace export) — same hang class as the memory provider below.
await self._finalize_session_off_loop(
session_id=getattr(agent, "session_id", None), platform="gateway", reason="shutdown",
session_key=session_key,
)
# Off-loop + bounded: a wedged memory provider here used to hang the whole shutdown so
# SIGTERM never completed.
await self._cleanup_agent_resources_off_loop(agent, context="shutdown finalize", session_key=session_key)
def _should_emit_long_running_notification(
self, session_key: Optional[str], agent: Any, executor_task: Optional[Any],
) -> bool:
"""Emit the heartbeat only while this task still owns the live run (not after ``/new`` rebinds).
Guards against a stale ``running: delegate_task`` heartbeat outliving the run that started it: stop
once the executor finishes, the agent is gone, or the session key has been rebound to a different
live agent (e.g. the user sent ``/new`` and a fresh agent took the slot mid-run, #12029).
"""
if agent is None or (executor_task is not None and executor_task.done()):
return False
# Drain/restart already told the chat the task will be interrupted; a "still working"
# heartbeat after that notice reads as a contradiction (#10990).
if getattr(self, "_draining", False) or getattr(self, "_restart_requested", False):
return False
if session_key:
_hb_state = self._peek_session_state(session_key)
if (_hb_state.turn.agent if _hb_state else None) is not agent:
return False
return True
def _defer_agent_cleanup_until_future_done(self, future: asyncio.Future, agent: Any, *, context: str) -> None:
"""Clean up ``agent`` only after its executor future finishes (it may still use the agent's clients)."""
async def _cleanup_when_done() -> None:
try:
await asyncio.shield(future)
except asyncio.CancelledError:
# Loop shutdown can cancel this waiter while the executor still runs. Never turn
# that cancellation into premature cleanup.
return
except Exception as exc:
logger.debug(
"Deferred agent worker%s finished with an error: %s", f" ({context})" if context else "", exc,
)
await self._cleanup_agent_resources_off_loop(agent, context=context)
self._track_deferred_agent_worker(future, agent)
tasks = getattr(self, "_deferred_agent_cleanup_tasks", None)
if tasks is None:
tasks = self._deferred_agent_cleanup_tasks = set()
self._track_task_in(tasks, asyncio.create_task(_cleanup_when_done()))
async def _finalize_session_off_loop(
self, *, session_id: Any, platform: str, reason: str, session_key: Optional[str] = None, **extra: Any,
) -> None:
"""Run hermes_cli.lifecycle.finalize_session off-loop, bounded; on timeout the worker is left alone.
``session_key`` lets an unscoped caller (shutdown) enter the owning profile's scope: plugin
``on_session_finalize`` observers and the Relay coordinator (``current_profile_key``) resolve
profile state at call time."""
def _call() -> None:
from hermes_cli.lifecycle import finalize_session
finalize_session(session_id=session_id, platform=platform, reason=reason, **extra)
try:
await asyncio.wait_for(
self._run_housekeeping_in_executor(self._run_release_in_profile_scope, _call, (), session_key),
timeout=self._FINALIZE_TIMEOUT_S,
)
except asyncio.TimeoutError:
logger.warning(
"Session finalize hooks (%s, reason=%s) exceeded %ss; proceeding without blocking the event loop "
"(the worker thread is left to finish on its own).", session_id, reason, self._FINALIZE_TIMEOUT_S,
)
except Exception as finalize_exc:
logger.debug("Session finalize hooks (%s, reason=%s) failed: %s", session_id, reason, finalize_exc)
async def _cleanup_agent_resources_off_loop(
self, agent: Any, *, context: str = "", session_key: Optional[str] = None,
) -> None:
"""Run _cleanup_agent_resources in a worker thread, bounded; on timeout the worker is left alone.
The teardown fires the memory-provider lifecycle hooks (``flush_pending`` → ``on_session_end`` →
``shutdown`` → ``close``), which read credentials/home at call time. In-turn callers carry the
profile scope through ``_run_housekeeping_in_executor``; shutdown does not (it runs on the main
loop, outside any adapter handler), so under multiplexing ``on_session_end`` failed closed and the
session tail was never committed (#110622). ``_run_release_in_profile_scope`` enters the OWNING
profile's scope from ``session_key`` when the caller has none, exactly like cache eviction."""
if agent is None:
return
if context.startswith("shutdown") or context == "session expiry":
with suppress(Exception):
agent._end_session_on_close = False
ctx_label = f" ({context})" if context else ""
try:
await asyncio.wait_for(
self._run_housekeeping_in_executor(
self._run_release_in_profile_scope, self._cleanup_agent_resources, (agent,), session_key,
),
timeout=self._CLEANUP_TIMEOUT_S,
)
except asyncio.TimeoutError:
logger.warning(
"Agent resource cleanup%s exceeded %ss; proceeding without blocking the event loop (the worker "
"thread is left to finish on its own). (#53175)", ctx_label, self._CLEANUP_TIMEOUT_S,
)
except Exception as cleanup_exc:
logger.warning("Agent resource cleanup%s failed: %s (#53175)", ctx_label, cleanup_exc)
def _cleanup_agent_resources(self, agent: Any) -> None:
"""Best-effort cleanup for temporary or cached agent instances."""
if agent is None:
return
with suppress(Exception):
if hasattr(agent, "shutdown_memory_provider"):
# Drain queued memory writes BEFORE teardown (shutdown_all() gives the worker only ~5s, so a
# /reset or rotation could drop them). Bounded; a failure never blocks teardown.
# The memory manager persists per-turn sync and end-of-session extraction on a single
# serialized background worker. shutdown_memory_provider() -> shutdown_all() only gives that
# worker a ~5s bounded drain and abandons (cancels) anything still queued past it, so a
# /reset — or any gateway session rotation that reaches this cleanup path — could silently
# drop writes the session had already handed off. The next session then loads stale memory
# (#73297). Give pending work a bounded head start through the manager's own barrier first,
# mirroring the CLI exit path (cli.py). Best-effort: a flush failure must never block
# teardown.
_mm = getattr(agent, "_memory_manager", None)
if _mm is not None and hasattr(_mm, "flush_pending"):
with suppress(Exception):
_mm.flush_pending(timeout=10)
# Pass the real transcript so ``on_session_end`` hooks don't see the empty default.
# ``_session_messages`` may be absent on ``object.__new__`` test stubs, hence getattr.
# ``_session_messages`` is set on ``AIAgent`` (run_agent.py:1518) and refreshed at the end
# of every ``run_conversation`` turn via ``_persist_session``; on an agent built through
# ``object.__new__`` (test stubs) the attribute may be absent, so ``getattr`` with a
# ``None`` default keeps the call signature-compatible with the pre-fix behaviour
# (``shutdown_memory_provider(messages=None)``). See #15165.
session_messages = getattr(agent, "_session_messages", None)
if isinstance(session_messages, list):
agent.shutdown_memory_provider(session_messages)
else:
agent.shutdown_memory_provider()
# Close tool resources (sandboxes, browser daemons, background processes, httpx clients).
with suppress(Exception):
if hasattr(agent, "close"):
agent.close()
# Auxiliary async clients live in a process-global cache created from worker threads; drop
# entries whose event loop is dead so httpx transports don't accumulate across turns.
with suppress(Exception):
from agent.auxiliary_client import cleanup_stale_async_clients
cleanup_stale_async_clients()
# Stuck-loop (restart failure) counters
def _stuck_loop_counts_path(self) -> Path:
from gateway.run import _hermes_home
return _hermes_home / self._STUCK_LOOP_FILE
@staticmethod
def _read_json_counts(path: Path) -> Optional[dict]:
"""Parsed counter dict, or None when the file is missing/unreadable (no exists() pre-check needed)."""
try:
return json.loads(path.read_text(encoding="utf-8-sig"))
except Exception:
return None
def _increment_restart_failure_counts(self, active_session_keys: set) -> None:
"""Increment persisted restart-failure counters for active sessions; drop the rest (loop broken)."""
from utils import atomic_json_write
path = self._stuck_loop_counts_path()
counts = self._read_json_counts(path) or {}
with suppress(Exception):
atomic_json_write(path, {key: counts.get(key, 0) + 1 for key in active_session_keys}, indent=None)
def _suspend_stuck_loop_sessions(self) -> int:
"""Suspend sessions active across too many restarts (startup, AFTER crash-turn recovery)."""
path = self._stuck_loop_counts_path()
if not path.exists():
return 0
counts = self._read_json_counts(path)
if counts is None:
return 0
suspended = 0
for session_key in [k for k, v in counts.items() if v >= self._STUCK_LOOP_THRESHOLD]:
with suppress(Exception):
entry = self.session_store._entries.get(session_key)
if entry and not entry.suspended:
entry.suspended = True
suspended += 1
logger.warning(
"Auto-suspended stuck session %s (active across %d consecutive restarts — likely a stuck loop)",
session_key, counts[session_key],
)
if suspended:
with suppress(Exception):
self.session_store._save()
# Clear the file — counters start fresh after suspension
with suppress(Exception):
path.unlink(missing_ok=True)
return suspended
async def _clear_restart_failure_count(self, session_key: str) -> None:
"""Clear a completed session's restart-failure counter off-loop (atomic_json_write fsyncs)."""
from utils import atomic_json_write
path = self._stuck_loop_counts_path()
if not path.exists():
return
# The whole read/mutate/write is guarded (as on main): a corrupt counters file
# (non-dict JSON) must never raise out of a session-completion path.
try:
counts = self._read_json_counts(path) or {}
if session_key in counts:
del counts[session_key]
if counts:
await asyncio.to_thread(atomic_json_write, path, counts, indent=None)
else:
path.unlink(missing_ok=True)
except Exception:
pass
# Restart orchestration
@staticmethod
def _restart_watcher_env() -> dict:
"""Watcher env minus ``_HERMES_GATEWAY`` (else the CLI's self-restart guard refuses; gateway stays down).
The host multiplexer is respawned with ``host_gateway_child_env`` (default-root
secrets via ``served_profile_child_env``, not ``os.environ.copy()``). A standalone
named-profile gateway keeps that profile's home — only a multiplexer, or a process
already on the default root, is the host.
"""
from gateway.config_loader import drop_bridged_env
from hermes_constants import get_default_hermes_root, get_hermes_home
from tools.environments.local import host_gateway_child_env, served_profile_child_env
home = get_hermes_home()
try:
on_default = home.resolve() == get_default_hermes_root().resolve()
except Exception:
on_default = False
# ``resolve_multiplex_mode`` settles the default-on/unset decision before
# restart. Carry that runtime identity instead of re-reading raw config:
# ``None`` is the normal pre-resolution value for a named launcher.
from agent.secret_scope import is_multiplex_active
settled_multiplex = is_multiplex_active()
multiplex = False
if not on_default and not settled_multiplex:
# Second settled source: the live host gateway's OWN published record
# (its settled served set). Only when NO settled identity exists may the
# raw config re-read stand — it reads the UNSET flag as False, which is
# wrong exactly when this process IS the default-on host (#120305).
try:
from gateway import host_rendezvous as hr
record = hr.read_record(hr.ROLE_GATEWAY)
if record is not None and hr.liveness_is_proven(record) and len(record.profiles) > 1:
multiplex = True
except Exception:
multiplex = False
if not on_default and not settled_multiplex and not multiplex:
try:
from gateway.config import load_gateway_config
multiplex = bool(load_gateway_config().multiplex_profiles)
except Exception:
multiplex = False
if on_default or settled_multiplex or multiplex:
watcher_env = host_gateway_child_env()
else:
watcher_env = served_profile_child_env(
target_home=home, inherit_credentials=True,
)
watcher_env = drop_bridged_env(watcher_env)
watcher_env.pop("_HERMES_GATEWAY", None)
return watcher_env
@staticmethod
def _spawn_windows_restart_watcher(hermes_cmd: list, current_pid: int, restart_after_s: float) -> None:
"""Spawn the detached Windows watcher (``python -c``), retrying once without job breakaway."""
import subprocess
from hermes_cli._subprocess_compat import (
windows_detach_flags_without_breakaway, windows_detach_popen_kwargs
)
watcher_env = GatewayShutdownMixin._restart_watcher_env()
# host_gateway_child_env does not copy the parent dotenv. The watcher
# still has to run inside the venv this process is using, or the
# respawn cannot import hermes.
if not watcher_env.get("VIRTUAL_ENV"):
inherited = os.environ.get("VIRTUAL_ENV")
if inherited:
watcher_env["VIRTUAL_ENV"] = inherited
project_root = Path(__file__).resolve().parent.parent
# Console python under CREATE_NO_WINDOW: nothing flashes. NOT pythonw.exe — a console-less
# watcher makes every console-subsystem descendant allocate a visible conhost (#54220/#56747).
# The watcher runs sys.executable (console python) under the CREATE_NO_WINDOW detach kwargs below:
# it owns one hidden console, inherited by the `hermes gateway restart` child, so nothing flashes.
# See #54220, #56747.
from hermes_cli._launchers import runtime_command
watcher_argv = runtime_command(project_root,
[str(current_pid), str(restart_after_s), *hermes_cmd, "gateway", "restart"],
code=_WINDOWS_RESTART_WATCHER)
watcher_python = watcher_argv[0]
popen_kwargs = dict(stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, env=watcher_env)
# Break away from the parent CLI's job object or be reaped when the CLI exits; a job without
# BREAKAWAY_OK rejects CREATE_BREAKAWAY_FROM_JOB (OSError) — retry once without the bit.
try:
subprocess.Popen(watcher_argv, **popen_kwargs, **windows_detach_popen_kwargs())
except OSError:
try:
subprocess.Popen(
watcher_argv, **popen_kwargs, creationflags=windows_detach_flags_without_breakaway(),
)
except OSError as exc:
# Both spawns failed. Log only the interpreter basename and numeric errno — never
# argv, env, watcher source, or str(exc) (may carry a full path) — and return.
winerror = getattr(exc, "winerror", None)
logger.warning(
"Detached restart watcher was not started after the "
"no-breakaway retry (%s; %s=%r). The gateway will not "
"be respawned by this restart attempt.", os.path.basename(watcher_python),
"winerror" if winerror is not None else "errno",
winerror if winerror is not None else exc.errno,
)
async def _launch_detached_restart_command(self) -> None:
from gateway.run import _resolve_hermes_bin
import shutil
import subprocess
hermes_cmd = _resolve_hermes_bin()
if not hermes_cmd:
logger.error("Could not locate hermes binary for detached /restart")
return
if self._detached_restart_helper_started:
return
self._detached_restart_helper_started = True
current_pid = os.getpid()
restart_after_s = max(float(getattr(self, "_restart_drain_timeout", 0.0) or 0.0) + 5.0, 5.0)
if sys.platform == "win32":
GatewayShutdownMixin._spawn_windows_restart_watcher(hermes_cmd, current_pid, restart_after_s)
return
cmd = " ".join(shlex.quote(part) for part in hermes_cmd)
shell_cmd = (
f"deadline=$(( $(date +%s) + {int(restart_after_s)} )); "
f"while kill -0 {current_pid} 2>/dev/null && [ $(date +%s) -lt $deadline ]; do sleep 0.2; done; "
f"{cmd} gateway restart"
)
setsid_bin = shutil.which("setsid")
argv = [setsid_bin, "bash", "-lc", shell_cmd] if setsid_bin else ["bash", "-lc", shell_cmd]
subprocess.Popen(
argv, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
env=GatewayShutdownMixin._restart_watcher_env(), start_new_session=True,
)
def _wedged_agent_count(self) -> int:
"""Work units the restart wait may skip: chat agents idle past ``agent.gateway_timeout`` and
cron runs older than the scheduler's stale-inflight allowance (#115469).
API work has no activity clock and pending sentinels are brand-new, so neither counts.
"""
return self._wedged_chat_agent_count() + self._wedged_cron_job_count()
def _wedged_cron_job_count(self) -> int:
"""Cron runs past ``cron.scheduler.get_wedged_job_ids``'s allowance; 0 if cron can't import."""
try:
from cron.scheduler import get_wedged_job_ids
return len(get_wedged_job_ids())
except Exception:
return 0
def _wedged_chat_agent_count(self) -> int:
"""Running chat agents with no activity for ``agent.gateway_timeout`` (0 when disabled);
an unreadable activity summary means "not wedged".
"""
from gateway.run import _AGENT_PENDING_SENTINEL, _float_env
timeout = _float_env("HERMES_AGENT_TIMEOUT", 1800)
if timeout <= 0:
return 0
def _idle_seconds(agent: Any) -> Optional[float]:
summary_fn = getattr(agent, "get_activity_summary", None)
if not callable(summary_fn):
return None
try:
summary = summary_fn()
return float(summary.get("seconds_since_activity", 0.0)) if isinstance(summary, dict) else None
except Exception:
return None
return sum(
1
for agent in list((getattr(self, "_running_agents", None) or {}).values())
if agent is not None and agent is not _AGENT_PENDING_SENTINEL
and (idle := _idle_seconds(agent)) is not None and idle >= timeout
)
def _awaitable_work_count(self) -> int:
"""Active work minus wedged turns — what the restart wait waits on."""
return max(0, self._active_work_count() - self._wedged_agent_count())
def _describe_active_work(self) -> list:
"""One dict per in-flight work unit the restart wait is holding for, so an observer
(``hermes update``, ``hermes gateway status``) can name it instead of printing a bare count.
``kind`` ∈ ``chat`` (session turn), ``cron`` (job id + external worker pid when the run was
handed to a restart-safe scope), ``api`` / ``deferred`` (count only — those sources expose
no identity). Best-effort: a source that can't be read is omitted, never raises.
"""
from gateway.run import _AGENT_PENDING_SENTINEL
now = time.time()
units: list = []
for key, state in list(self._sessions_map().items()):
agent = state.turn.agent
if agent is None:
continue
unit: dict = {"kind": "chat", "session": key, "pid": os.getpid()}
if state.turn.started_ts:
unit["elapsed_s"] = round(now - state.turn.started_ts, 1)
if agent is not _AGENT_PENDING_SENTINEL:
unit["model"] = getattr(agent, "model", None)
summary_fn = getattr(agent, "get_activity_summary", None)
if callable(summary_fn):
with suppress(Exception):
summary = summary_fn()
unit["current_tool"] = summary.get("current_tool")
unit["idle_s"] = summary.get("seconds_since_activity")
units.append(unit)
with suppress(Exception):
from cron.scheduler import get_running_job_details, get_wedged_job_ids
wedged = get_wedged_job_ids()
for job in get_running_job_details():
units.append({"kind": "cron", "job_id": job["job_id"], "elapsed_s": job["elapsed_s"],
"pid": job["worker_pid"] or os.getpid(), "external": bool(job["worker_pid"]),
"wedged": job["job_id"] in wedged})
for kind, count in (("api", self._active_api_run_count()), ("deferred", self._active_deferred_agent_worker_count())):
units.extend({"kind": kind, "pid": os.getpid()} for _ in range(count))
return units
async def _await_active_work_before_restart(self) -> bool:
"""Wait for in-flight work before ``stop()`` so the requesting turn isn't force-interrupted.
Wedged turns are excluded (restart is their remedy). True when drained to zero, False when the
cap elapsed or only wedged work remains (caller proceeds to ``stop()``).
"""
active = self._active_work_count()
if active <= 0:
return True
if self._awaitable_work_count() <= 0:
logger.warning(
"Restart requested with %d active work unit(s), all wedged "
"past the inactivity timeout; skipping the after-turn wait "
"and proceeding to stop()/drain which will interrupt them", active,
)
return False
timeout = float(getattr(self, "_restart_after_turn_timeout", 0.0) or 0.0)
if timeout <= 0:
logger.info(
"Restart requested with %d active work unit(s); "
"restart_after_turn_timeout=0 — entering stop()/drain immediately", active,
)
return False
logger.info(
"Restart requested with %d active work unit(s); "
"deferring stop() until they finish (cap=%.0fs) so in-flight "
"turns are not amputated (#77184)", active, timeout,
)
self._scale_to_zero_status("draining", "restart wait: status mark failed")
loop = asyncio.get_running_loop()
deadline = loop.time() + timeout
last_status_at = 0.0
while self._awaitable_work_count() > 0:
now = loop.time()
if now >= deadline:
logger.warning(
"Restart after-turn wait timed out after %.0fs with %d "
"still active; proceeding to stop()/drain which may "
"interrupt remaining work (#77184)", timeout, self._active_work_count(),
)
return False
if (now - last_status_at) >= 30.0:
logger.info(
"Restart deferred: waiting on %d active work unit(s) "
"(%d wedged and excluded; %.0fs remaining before force drain): %s",
self._awaitable_work_count(), self._wedged_agent_count(), deadline - now,
self._describe_active_work(),
)
self._scale_to_zero_status("draining", "restart wait: status mark failed")
last_status_at = now
await asyncio.sleep(0.1)
if self._active_work_count() > 0:
logger.warning(
"Restart deferred wait: %d wedged work unit(s) remain; "
"proceeding to stop()/drain which will interrupt them", self._active_work_count(),
)
return False
logger.info("Restart deferred wait complete — active work drained; proceeding to stop()")
return True
def request_restart(self, *, detached: bool = False, via_service: bool = False) -> bool:
if self._restart_task_started:
return False
self._restart_requested = True
self._restart_detached = detached
self._restart_via_service = via_service
self._restart_task_started = True
# Refuse new turns; keep ``_running`` True so the active turn can still deliver its final response.
self._draining = True
# The restart's after-turn wait is a drain window too: pollers of GET /v1/runs/{id} must see
# the boundary from the moment new turns are refused, not only once stop() begins (#115133).
self._mark_api_runs_shutdown_requested()
async def _run_restart() -> None:
await self._await_active_work_before_restart()
# Detached helper only AFTER the after-turn wait, or its drain_timeout+5 deadline fires mid-turn.
if detached:
with _log_suppressed(logging.ERROR, "Failed to launch detached gateway restart helper: %s"):
await self._launch_detached_restart_command()
await asyncio.sleep(0.05)
await self.stop(restart=True, detached_restart=detached, service_restart=via_service)
# NOT in _background_tasks: _stop_impl cancels those, which would skip _shutdown_event.set() / exit 75.
# _run_restart is a short-lived self-terminating task (calls stop() then returns). Don't add it to
# _background_tasks — _stop_impl cancels all entries in that set, which would cancel _run_restart
# while it's awaiting _stop_task, propagating CancelledError into _stop_impl and preventing
# _shutdown_event.set() / _exit_code = 75. See #12875. We still hold a strong reference in
# self._restart_task: a bare asyncio.create_task() keeps only a weak reference, so the event loop
# may garbage-collect a still-pending task mid-flight. The cancel loop in _stop_impl explicitly
# skips _restart_task for the same reason it skips _stop_task.
# Empty Context: /restart is handled inside the requester's profile scope, and a copied context
# would run the HOST restart as that profile (watcher HERMES_HOME, stop()'s flushes).
self._restart_task = Context().run(lambda: asyncio.create_task(_run_restart()))
return True
def _start_systemd_watchdog(self) -> bool:
"""Start sd_notify only after a configured gateway is truly running."""
if not self._running or self.config.systemd_watchdog_seconds <= 0:
return False
if self._systemd_watchdog is not None:
return True
from gateway.systemd_notify import SystemdWatchdog
watchdog = SystemdWatchdog(config_enabled=True)
if not watchdog.start():
return False
self._systemd_watchdog = watchdog
watchdog.ready("Hermes Gateway running")
return True
async def _stop_systemd_watchdog(self) -> None:
"""Stop heartbeats before any potentially long shutdown drain."""
watchdog = self._systemd_watchdog
if watchdog is None:
return
self._systemd_watchdog = None
await watchdog.stop()
# stop() phases. Invoked as ``GatewayRunner._stop_<phase>(self, ctx)`` so shutdown-path tests
# can drive them from bare doubles that are not GatewayRunner instances.
@staticmethod
def _quiet_step(label: str, fn: Callable[[], Any]) -> Any:
"""Run one best-effort teardown step; a failure is debug-logged as ``"<label>: <exc>"``."""
try:
return fn()
except Exception as _e:
logger.debug("%s: %s", label, _e)
return None
@staticmethod
def _stop_kill_tool_subprocesses(phase: str) -> list:
"""Kill tool subprocesses + terminal envs + browsers; returns cron job IDs marked interrupted.
Called twice: after a drain timeout (reclaim children before systemd SIGKILLs) and as a final
catch-all. Best-effort; one failing subsystem cannot block the rest.
"""
def _step(label: str, fn: Callable[[], Any]) -> Any:
return GatewayShutdownMixin._quiet_step(f"{label} ({phase}) error", fn)
def _count_step(fmt: str, fn: Callable[[], int]) -> None:
n = fn()
if n:
logger.info(fmt, phase, n)
def _kill_processes() -> None:
from tools.process_registry import process_registry
# Host shutdown: kill even persist_on_release jobs or they become
# PPID=1 orphans (#41225/#46778); an explicit source reaches them.
_count_step(
"Shutdown (%s): killed %d tool subprocess(es)",
lambda: process_registry.kill_all(source="gateway_shutdown"))
def _mark_cron_interrupted() -> list:
# kill_all() is global: a cron job mid-dispatch lost its tool subprocess and its agent thread may
# still emit a plausible response from truncated output — mark it interrupted, never success.
# Any cron job still dispatched at this instant just had its tool subprocess killed above
# (kill_all() has no per-job-ID targeting — it's a global sweep). No-op when no cron job is in
# flight. See #60432.
from cron.scheduler import mark_running_jobs_interrupted
_interrupted = mark_running_jobs_interrupted(
f"Gateway shutdown ({phase}) killed the job's tool subprocess before the run finished."
)
if _interrupted:
logger.warning(
"Shutdown (%s): marked %d in-flight cron job(s) interrupted: %s",
phase, len(_interrupted), ", ".join(_interrupted),
)
return _interrupted
def _interrupt_delegations() -> None:
from tools.async_delegation import interrupt_all as _interrupt_async
_count_step(
"Shutdown (%s): interrupted %d background delegation(s)",
lambda: _interrupt_async(reason=f"gateway shutdown ({phase})"),
)
_step("process_registry.kill_all", _kill_processes)
_marked_cron_jobs = _step("mark_running_jobs_interrupted", _mark_cron_interrupted) or []
_step("async interrupt_all", _interrupt_delegations)
def _cleanup_environments() -> None:
from tools.terminal_tool_lifecycle import cleanup_all_environments
cleanup_all_environments()
def _cleanup_browsers() -> None:
from tools.browser_tool_lifecycle import cleanup_all_browsers
cleanup_all_browsers()
_step("cleanup_all_environments", _cleanup_environments)
_step("cleanup_all_browsers", _cleanup_browsers)
return _marked_cron_jobs
@staticmethod
async def _stop_kill_tool_subprocesses_off_loop(phase: str) -> list:
"""Run _stop_kill_tool_subprocesses in a worker thread; returns cron job IDs marked interrupted.
``kill_all`` fans out into per-target ``kill_process`` calls that do blocking work
(registry checkpoint disk I/O, ``subprocess.run`` for systemd scopes, sandbox exec),
so running the sweep inline would monopolize the gateway event loop (#116327).
Offloaded with ``asyncio.to_thread`` — the loop's default executor, deliberately NOT
the gateway-owned ``self._executor``, which ``_stop_quiesce_and_close_session_dbs``
drains right after this phase. Phase order is preserved: callers await this before
cron notices / adapter teardown. If the surrounding stop task is cancelled while the
worker runs, the thread is left to finish on its own; the thread-based shutdown
watchdog remains the hard backstop.
"""
return await asyncio.to_thread(
GatewayShutdownMixin._stop_kill_tool_subprocesses, phase
)
async def _stop_begin_teardown(self, ctx: "GatewayShutdownMixin._StopContext") -> None:
"""Flag teardown, stop room worker/watchdog, notify sessions."""
logger.info("Stopping gateway%s...", " for restart" if self._restart_requested else "")
ctx.started_at = time.monotonic()
self._running = False
self._clear_plugin_message_injector()
self._draining = True
self._mark_api_runs_shutdown_requested()
# getattr-guards: shutdown-path test doubles may lack the room worker / systemd watchdog.
stop_room_worker = getattr(self, "_stop_hosted_room_worker", None)
if callable(stop_room_worker):
try:
if not await stop_room_worker(timeout=5.0):
logger.warning(
"Group Chat worker is still settling durable work; the next gateway start will recover it"
)
except Exception:
logger.warning(
"Group Chat worker could not stop cleanly; the next gateway start will recover durable work",
exc_info=True,
)
stop_watchdog = getattr(self, "_stop_systemd_watchdog", None)
if callable(stop_watchdog):
await stop_watchdog()
await self._cancel_secondary_profile_reconnect_tasks()
# Notify all chats with active agents BEFORE draining — adapters are still connected here.
await self._notify_active_sessions_of_shutdown()
logger.info("Shutdown phase: notify_active_sessions done at +%.2fs", ctx.elapsed())
async def _stop_drain_active_work(self, timeout: float, ctx: "GatewayShutdownMixin._StopContext") -> None:
"""Pre-mark resume_pending, drain agents/cron/API work into ``ctx``."""
from gateway.run import GatewayRunner
# Pre-mark resume_pending BEFORE the drain so a mid-drain SIGKILL still leaves a durable marker.
_pre_drain_keys = await GatewayRunner._mark_running_sessions_resume_pending(
self, "pre-drain mark_resume_pending"
)
_cron_at_start = self._active_cron_job_count()
_api_at_start = self._active_api_run_count()
_deferred_at_start = ctx.deferred_count()
# Cron floor clamped to the watchdog leash; getattr-guard for bare shutdown-path doubles.
_cron_drain_cfg = getattr(self, "_cron_drain_timeout", DEFAULT_GATEWAY_CRON_DRAIN_TIMEOUT)
# Under launchd the real leash is launchd's own exit timeout, not our watchdog
# (drain + grace): a signal-driven stop that lets cron work push past it is SIGKILLed
# before cleanup runs.
# ``timeout`` is already the effective (launchd-capped) drain, so this is the same
# leash the thread watchdog is armed with — dump margin included.
_cron_leash = effective_stop_watchdog_delay(self, resolve_shutdown_watchdog_delay(timeout))
_cron_timeout = resolve_cron_drain_budget(
timeout, _cron_drain_cfg, watchdog_delay=_cron_leash, elapsed=ctx.elapsed(),
)
if _cron_at_start and _cron_timeout > timeout:
logger.info(
"Shutdown drain: %d in-flight cron job(s) — waiting up to "
"%.0fs for them (cron_drain_timeout=%.0fs, restart_drain_timeout=%.0fs)",
_cron_at_start, _cron_timeout, _cron_drain_cfg, timeout,
)
_drain_started_at = time.monotonic()
ctx.active_agents, ctx.timed_out = await self._drain_active_agents(timeout, _cron_timeout)
ctx.drain_elapsed = time.monotonic() - _drain_started_at
logger.info(
"Shutdown phase: drain done at +%.2fs (drain took %.2fs, timed_out=%s, active_at_start=%d, "
"active_now=%d, cron_at_start=%d, cron_now=%d, api_at_start=%d, api_now=%d, "
"deferred_at_start=%d, deferred_now=%d)", ctx.elapsed(), ctx.drain_elapsed,
ctx.timed_out, len(ctx.active_agents), self._running_agent_count(), _cron_at_start,
self._active_cron_job_count(), _api_at_start, self._active_api_run_count(),
_deferred_at_start, ctx.deferred_count(),
)
if ctx.timed_out:
return
# Graceful drain: clear the pre-drain resume_pending markers so sessions that finished
# during the drain window don't carry a stale flag.
for _sk in _pre_drain_keys:
if _sk not in self._running_agents:
try:
await self.async_session_store.clear_resume_pending(_sk)
except Exception as _e:
logger.debug("clear_resume_pending after drain failed for %s: %s", _sk, _e)
async def _stop_interrupt_remaining_work(self, ctx: "GatewayShutdownMixin._StopContext") -> None:
"""Drain timed out: mark resume_pending, interrupt, settle, kill tool subprocesses, notify cron."""
from gateway.run import GatewayRunner
logger.warning(
"Gateway drain timed out after %.1fs with %d active agent(s), "
"%d in-flight cron job(s), %d api_server run(s), and %d deferred agent worker(s); "
"interrupting remaining work.", ctx.drain_elapsed, self._running_agent_count(),
self._active_cron_job_count(), self._active_api_run_count(), ctx.deferred_count(),
)
# Mark resume_pending BEFORE interrupting so the next message auto-resumes (stuck sessions
# still escalate via .restart_failure_counts). CURRENT _running_agents, not the drain snapshot.
await GatewayRunner._mark_running_sessions_resume_pending(self, "mark_resume_pending")
reason = GatewayRunner._shutdown_interrupt_reason(self)
self._interrupt_running_agents(reason)
interrupt_grace_timeout = GatewayRunner._post_interrupt_grace_timeout(self)
loop = asyncio.get_running_loop()
interrupt_deadline = loop.time() + interrupt_grace_timeout
logger.info("Shutdown phase: allowing %.1fs for interrupted agents to unwind", interrupt_grace_timeout)
def _work_live() -> bool:
return bool(self._running_agents or self._active_api_run_count() or ctx.deferred_count())
# Wait on API-server work too, or an API turn's tool subprocesses are killed before it unwinds.
while _work_live() and loop.time() < interrupt_deadline:
self._update_runtime_status("draining")
await asyncio.sleep(0.1)
# Work can materialize AFTER the one-shot interrupt (/v1/runs registers on _create_agent return;
# pending sentinels promote later). Re-signal for a cooperative interrupt, not a bare kill.
if _work_live():
self._interrupt_running_agents(reason)
logger.debug("Re-signaled interrupt for work still live at settle-window exit")
# Kill tool subprocesses NOW: deferring past adapter/DB teardown risks the systemd cgroup SIGKILL.
# Off-loop: the sweep does blocking kills that must not monopolize the event loop (#116327).
_interrupted_cron_jobs = await GatewayRunner._stop_kill_tool_subprocesses_off_loop("post-interrupt")
logger.info("Shutdown phase: post-interrupt tool kill done at +%.2fs", ctx.elapsed())
# Last window with the transport up (the cron worker's own notice arrives after teardown).
with _log_suppressed(logging.DEBUG, "Cron interrupt notification failed: %s"):
# The cron worker whose run we just killed will try to deliver its own "interrupted" notice, but
# it gets there after the adapter teardown below and the message is lost (#82232).
await self._notify_interrupted_cron_jobs(_interrupted_cron_jobs)
logger.info("Shutdown phase: cron interrupt notices done at +%.2fs", ctx.elapsed())
async def _stop_finalize_agents_and_adapters(self, ctx: "GatewayShutdownMixin._StopContext") -> None:
"""Detached restart launch, agent finalization, idle-cache cleanup, adapter teardown."""
if self._restart_requested and self._restart_detached:
with _log_suppressed(logging.ERROR, "Failed to launch detached gateway restart: %s"):
await self._launch_detached_restart_command()
await self._finalize_shutdown_agents(ctx.active_agents)
# Idle cached agents too: their MemoryProviders may never have seen on_session_end().
_cache_lock = getattr(self, "_agent_cache_lock", None)
_cache = getattr(self, "_agent_cache", None)
if _cache_lock is not None and _cache is not None:
with _cache_lock:
_idle_agents = list(_cache.items())
_cache.clear()
for _key, _entry in _idle_agents:
# Bounded + off-loop: a wedged memory provider here once made SIGTERM hang forever.
await self._cleanup_agent_resources_off_loop(
_entry[0] if isinstance(_entry, tuple) else _entry, context="shutdown idle-cache",
session_key=_key,
)
# Settle completion flush tasks while adapters are alive so every watcher gets a retryable result.
cancel_completion_batches = getattr(self, "_cancel_process_completion_batch_tasks", None)
if cancel_completion_batches is not None:
await cancel_completion_batches()
for platform, adapter in list(self.adapters.items()):
await self._bounded_adapter_teardown(adapter, platform)
# Disconnect secondary-profile adapters (multiplex mode).
_profile_adapters = getattr(self, "_profile_adapters", {})
for _prof, _amap in list(_profile_adapters.items()):
for platform, adapter in list(_amap.items()):
await self._bounded_adapter_teardown(adapter, platform, profile=_prof)
_amap.clear()
_profile_adapters.clear()
logger.info("Shutdown phase: all adapters disconnected at +%.2fs", ctx.elapsed())
async def _stop_release_runtime_state(self, ctx: "GatewayShutdownMixin._StopContext") -> None:
"""Cancel background tasks, flush pending messages, clear per-session state, final tool kill."""
from gateway.run import GatewayRunner
for _task in list(self._background_tasks):
# _restart_task awaits _stop_task: cancelling it would tunnel into _stop_impl and skip _shutdown_event.set().
if _task is self._stop_task or _task is self._restart_task:
continue
_task.cancel()
# The restart orchestration task is awaiting _stop_task right now; cancelling it would propagate
# CancelledError into this _stop_impl and skip _shutdown_event.set() / _exit_code = 75 (#12875). It
# self-terminates anyway.
self._background_tasks.clear()
ctx.api_live = self._active_api_run_count()
self.adapters.clear()
for _session_key in list(self._running_agents):
self._release_running_agent_state(_session_key)
# Flush pending messages before clearing: under FTS5 corruption they are the only surviving copy.
with suppress(Exception):
from gateway.shutdown_flush import flush_pending_to_file
flush_pending_to_file(dict(self._pending_messages), reason="shutdown")
# The overflow FIFO tail lives in SessionState.conversation.queued_events — flush it too.
with suppress(Exception):
from gateway.shutdown_flush import flush_overflow_to_file
flush_overflow_to_file(
{_k: list(_v) for _k, _v in dict(getattr(self, "_queued_events", None) or {}).items() if _v},
reason="shutdown",
)
# Live SessionState views: clear() resets one field per session (never a wholesale dict swap).
self._running_agents.clear()
self._running_agents_ts.clear()
self._pending_messages.clear()
self._pending_approvals.clear()
for _attr in ("_active_session_leases", "_busy_ack_ts"): # absent on bare shutdown-path doubles
if hasattr(self, _attr):
getattr(self, _attr).clear()
self._shutdown_event.set()
# Global catch-all subprocess kill (safe to repeat) for the graceful path and late respawns.
# Off-loop: same blocking sweep as the post-interrupt kill (#116327).
await GatewayRunner._stop_kill_tool_subprocesses_off_loop("final-cleanup")
logger.info("Shutdown phase: final-cleanup tool kill done at +%.2fs", ctx.elapsed())
# Reap the auxiliary-client cache: clients bound to dead worker-thread loops leak httpx transports.
def _reap_aux_clients() -> None:
# Reap the process-global auxiliary-client cache once at the very end of teardown. Per-turn
# cleanup runs in _cleanup_agent_resources for each active agent, but clients bound to
# worker-thread loops that died with their ThreadPoolExecutor (notably cron ticks) only get
# swept here. Without this, long-running gateways accumulate async httpx transports until they
# hit EMFILE on macOS's default RLIMIT_NOFILE=256. See #14210.
from agent.auxiliary_client import shutdown_cached_clients
shutdown_cached_clients()
GatewayShutdownMixin._quiet_step("shutdown_cached_clients error", _reap_aux_clients)
def _stop_quiesce_and_close_session_dbs(self, timeout: float, ctx: "GatewayShutdownMixin._StopContext") -> None:
"""Quiesce the executor, then close SessionDB handles only if no worker is still live."""
from gateway.run import GatewayRunner, _EXECUTOR_QUIESCE_TIMEOUT
# Quiesce the thread pool BEFORE closing session DBs: a late executor write after
# SessionDB.close() checkpointed the WAL reopens the handle and splits the WAL generation
# (close-time corruption). Clamped to the remaining watchdog leash minus 1s for the close.
# This used to run *after* the close block below, which left two holes: (a) `_executor_closing` was
# still False during the close, so any coroutine reaching `_run_in_executor_with_context` minted a
# brand-new pool and ran more blocking DB work against handles that had just been closed; (b)
# cancelling `self._background_tasks` above does not stop a `run_in_executor` future that already
# started — the task dies, the worker thread keeps writing. Either way a write lands after
# `SessionDB.close()`, which has already checkpointed the WAL and let SQLite unlink the sidecar. The
# late write silently reopens the handle (#94736) and mints a fresh WAL generation behind that
# checkpoint, so teardown checkpoints the same file a second time from a connection the shutdown log
# never accounts for — the close-time page-write damage in #101093 and the split WAL generation in
# #101064. The wait is bounded and clamped to what is left of the shutdown watchdog leash (minus a
# second for the close itself), so a stuck worker can never cost us the post-close cleanup window
# (#82161).
_exec_quiesce_budget = max(
0.0, min(_EXECUTOR_QUIESCE_TIMEOUT, resolve_shutdown_watchdog_delay(timeout) - ctx.elapsed() - 1.0),
)
_exec_live = GatewayRunner._shutdown_executor(self, drain_timeout=_exec_quiesce_budget)
if _exec_live:
# A live worker may be mid-write (the #101093 corruption sequence): skip the close and let
# SQLite recover from its WAL on next open (at worst a transient "database is locked").
logger.warning(
"Shutdown phase: %d executor worker(s) still running after a %.2fs quiesce — skipping the "
"SessionDB close/checkpoint to avoid racing a live write (#101093); handles are left "
"open for SQLite to recover on next open", _exec_live, _exec_quiesce_budget,
)
return
logger.info("Shutdown phase: executor quiesced at +%.2fs", ctx.elapsed())
# Cron jobs (scheduler pool), API-server runs and deferred hygiene workers (both on the loop's
# default executor) never touch self._executor, so the join above cannot see them. A writer that
# outlived the drain is mid-write for the same #101093 reasons; the drain already spent its
# budget, so no second wait — leave the handles open (#102198). The API count is the snapshot
# taken before the adapters were released; a run whose handler task was cancelled at disconnect
# has already left it, so that term under-counts — the live worker-scoped count below covers
# the cancelled-handler case (#116535).
_cron_live = self._active_cron_job_count()
_api_live = ctx.api_live
_api_worker_live = self._active_api_worker_count()
_deferred_live = ctx.deferred_count()
if _cron_live or _api_live or _api_worker_live or _deferred_live:
logger.warning(
"Shutdown phase: %d cron job(s) / %d API-server run(s) / %d API-server worker(s) / "
"%d deferred worker(s) still running after the executor quiesce — skipping the SessionDB "
"close/checkpoint, leaving state.db open for the live writer (#102198, #116535)",
_cron_live, _api_live, _api_worker_live, _deferred_live,
)
return
_step = GatewayShutdownMixin._quiet_step
# Close SQLite session DBs so --replace's new gateway does not hit 'database is locked'.
# ``_session_db`` is an AsyncSessionDB facade — unwrap; ``session_store`` holds ``_db``.
_self_db = getattr(self, "_session_db", None)
_self_db = getattr(_self_db, "_db", _self_db)
store = getattr(self, "session_store", None)
for _db in (_self_db, getattr(store, "_db", None)):
if _db is not None and hasattr(_db, "close"):
_step("SessionDB close error", _db.close)
# Multiplexed session_store caches one SessionDB per profile; sweep the secondary WAL locks too.
_sweep = getattr(store, "close_all_db_handles", None)
if _sweep is not None:
_step("SessionDB handle sweep error", _sweep)
# Same sweep for the runner's own per-profile session_search handles.
_step("Runner SessionDB handle sweep error", lambda: GatewayRunner.close_all_session_db_handles(self))
def _close_shared() -> None:
# Shared SessionDB instances still held by the process-wide registry (tools, cron, mirror).
# This is the safety net that guarantees no WAL write lock survives past gateway shutdown
# (#90837).
from hermes_state_registry import close_all
closed = close_all()
if closed:
logger.debug("Closed %d shared SessionDB instance(s) at shutdown", closed)
_step("Shared SessionDB close error", _close_shared)
logger.info("Shutdown phase: SessionDB close done at +%.2fs", ctx.elapsed())
async def _stop_persist_exit_state(self, ctx: "GatewayShutdownMixin._StopContext") -> None:
"""PID/lock release, clean-shutdown marker, restart markers, terminal runtime status."""
from gateway.run import _hermes_home, _planned_restart_notification_path, _shutdown_gateway_health_export
from utils import atomic_json_write
from gateway.status import remove_pid_file, release_gateway_runtime_lock
remove_pid_file()
release_gateway_runtime_lock()
# Clean-shutdown marker skips crash-turn recovery next boot; a timed-out drain left
# half-finished sessions, so no marker — the next startup recovers their turn markers.
if not ctx.timed_out:
with suppress(Exception):
(_hermes_home / ".clean_shutdown").touch()
else:
logger.info(
"Skipping .clean_shutdown marker — drain timed out with "
"interrupted agents; next startup will recover their interrupted turns."
)
# Stuck-loop counter: sessions active across 3 consecutive restarts are auto-suspended next boot.
if ctx.active_agents:
self._increment_restart_failure_counts(set(ctx.active_agents.keys()))
if self._restart_requested and self._restart_command_source is None:
with _log_suppressed(logging.DEBUG, "Failed to write planned restart notification marker: %s"):
atomic_json_write(
_planned_restart_notification_path(),
{
"requested_at": time.time(),
"via_service": bool(self._restart_via_service),
"detached": bool(self._restart_detached),
},
indent=None,
)
if self._restart_requested and self._restart_via_service:
# Exit 75 + ``RestartForceExitStatus=75``: systemd replaces us without a racing helper.
self._exit_code = GATEWAY_SERVICE_RESTART_EXIT_CODE
self._exit_reason = self._exit_reason or "Gateway restart requested"
self._draining = False
# Terminal gateway_state: "stopped", or "running" on an UNEXPECTED signal (docker restart,
# OOM) — container_boot.py only auto-starts gateways last seen "running".
if getattr(self, "_signal_initiated_shutdown", False) and not self._restart_requested:
logger.info(
"Gateway stopped by an unexpected signal — persisting "
"gateway_state=running so container_boot auto-starts on the next boot (issue #42675)"
)
self._update_runtime_status("running", self._exit_reason)
else:
self._update_runtime_status("stopped", self._exit_reason)
try:
from gateway.status import flush_runtime_status_async
# Never outlive the launchd exit budget (``_launchd_exit_timeout_s`` is set by the
# supervised-restart path when it exists; a plain 2 s bound otherwise). The cap only
# applies to signal-driven stops — a programmatic stop is not racing the supervisor.
flush_timeout = 2.0
budget = getattr(self, "_launchd_exit_timeout_s", None)
signal_stop = getattr(self, "_stop_requested_by_signal", False)
if signal_stop and isinstance(budget, (int, float)) and budget > 0:
flush_timeout = max(0.0, min(flush_timeout, budget - ctx.elapsed()))
if not await flush_runtime_status_async(timeout=flush_timeout):
logger.warning("Timed out flushing terminal gateway runtime status")
except Exception:
logger.debug("Failed to flush terminal gateway runtime status", exc_info=True)
_shutdown_gateway_health_export(self)
logger.info("Gateway stopped (total teardown %.2fs)", ctx.elapsed())
def _shutdown_watchdog_snapshot(self, ctx: "GatewayShutdownMixin._StopContext") -> dict:
"""State dumped by the thread-based shutdown watchdog when teardown hangs."""
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(),
"active_deferred_agent_workers": ctx.deferred_count(),
"restart_drain_timeout": self._restart_drain_timeout,
"effective_drain_timeout": effective_stop_drain_timeout(self),
"launchd_exit_timeout_s": getattr(self, "_launchd_exit_timeout_s", None),
"watchdog_delay_s": _effective_watchdog_leash(self),
"phase_elapsed_s": ctx.elapsed() if ctx.started_at is not None else None,
}
async def _stop_impl(self) -> None:
"""Run every ``_stop_*`` phase under the thread-based shutdown watchdog."""
from gateway.run import GatewayRunner
# Thread-based watchdog (asyncio timeouts cannot recover a frozen loop): dumps stacks and
# os._exit past drain+grace so the service manager revives us. Skipped under pytest.
# 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. See #66892.
_watchdog_done = threading.Event()
self._shutdown_watchdog_done = _watchdog_done
# Shutdown-path doubles may lack the deferred-worker counter.
ctx = GatewayShutdownMixin._StopContext(
deferred_count=getattr(self, "_active_deferred_agent_worker_count", lambda: 0)
)
if not os.environ.get("PYTEST_CURRENT_TEST"):
arm_shutdown_watchdog(
_effective_watchdog_leash(self), done_event=_watchdog_done,
snapshot_fn=lambda: GatewayRunner._shutdown_watchdog_snapshot(self, ctx), exit_code=1,
)
try:
await GatewayRunner._stop_begin_teardown(self, ctx)
timeout = effective_stop_drain_timeout(self)
if timeout < self._restart_drain_timeout:
logger.warning(
"Shutdown drain capped to %.0fs (configured %.0fs) to fit the live launchd exit "
"timeout of %.0fs — launchd SIGKILLs past it",
timeout, self._restart_drain_timeout, self._launchd_exit_timeout_s,
)
await GatewayRunner._stop_drain_active_work(self, timeout, ctx)
if ctx.timed_out:
await GatewayRunner._stop_interrupt_remaining_work(self, ctx)
await GatewayRunner._stop_finalize_agents_and_adapters(self, ctx)
await GatewayRunner._stop_release_runtime_state(self, ctx)
GatewayRunner._stop_quiesce_and_close_session_dbs(self, timeout, ctx)
await GatewayRunner._stop_persist_exit_state(self, ctx)
finally:
_watchdog_done.set()
async def stop(
self, *, restart: bool = False, detached_restart: bool = False, service_restart: bool = False
) -> None:
"""Stop the gateway and disconnect all adapters."""
from gateway.run import GatewayRunner
# getattr-guard: shutdown-path tests build bare runners via object.__new__ that lack the
# liveness-guard machinery.
_stop_guards = getattr(self, "_stop_loop_liveness_guards", None)
if callable(_stop_guards):
_stop_guards()
if restart:
self._restart_requested = True
self._restart_detached = detached_restart
self._restart_via_service = service_restart
if self._stop_task is not None:
await self._stop_task
return
self._stop_task = asyncio.create_task(GatewayRunner._stop_impl(self))
await self._stop_task
async def wait_for_shutdown(self) -> None:
"""Wait for shutdown signal."""
await self._shutdown_event.wait()
Reference in New Issue View Git Blame Copy Permalink