fix(gateway): defer in-band restart until active turns finish (#77184)
request_restart was calling stop() immediately, so the requesting turn stayed in the drain wait set and got force-killed at restart_drain_timeout. Wait for active work to reach zero first, then stop against an idle gateway.
This commit is contained in:
parent
f105db2136
commit
db3f7e4eb9
|
|
@ -24,6 +24,15 @@ DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT = float(
|
|||
DEFAULT_CONFIG["agent"]["restart_drain_timeout"]
|
||||
)
|
||||
|
||||
# In-band restart (``/restart``, SIGUSR1, self-restart from a child CLI)
|
||||
# waits for active turns to finish *before* ``stop()`` begins. Distinct
|
||||
# from ``restart_drain_timeout``, which is the force-interrupt budget
|
||||
# once ``stop()`` is running (and must stay short under systemd
|
||||
# TimeoutStopSec). See #77184.
|
||||
DEFAULT_GATEWAY_RESTART_AFTER_TURN_TIMEOUT = float(
|
||||
DEFAULT_CONFIG["agent"]["restart_after_turn_timeout"]
|
||||
)
|
||||
|
||||
|
||||
def is_gateway_supervisor_process(
|
||||
environ: Mapping[str, str] | None = None,
|
||||
|
|
@ -64,3 +73,48 @@ def parse_restart_drain_timeout(raw: object) -> float:
|
|||
except (TypeError, ValueError):
|
||||
return DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT
|
||||
return max(0.0, value)
|
||||
|
||||
|
||||
def parse_restart_after_turn_timeout(raw: object) -> float:
|
||||
"""Parse the after-turn wait cap for in-band restart, falling back to default.
|
||||
|
||||
``0`` is a deliberate disable (legacy immediate drain) and must not fall
|
||||
through to the default — unlike empty/missing input.
|
||||
"""
|
||||
if raw is None:
|
||||
return DEFAULT_GATEWAY_RESTART_AFTER_TURN_TIMEOUT
|
||||
if isinstance(raw, str) and not raw.strip():
|
||||
return DEFAULT_GATEWAY_RESTART_AFTER_TURN_TIMEOUT
|
||||
try:
|
||||
value = float(raw)
|
||||
except (TypeError, ValueError):
|
||||
return DEFAULT_GATEWAY_RESTART_AFTER_TURN_TIMEOUT
|
||||
return max(0.0, value)
|
||||
|
||||
|
||||
def resolve_restart_exit_wait_budget(
|
||||
drain_timeout: float,
|
||||
after_turn_timeout: float,
|
||||
*,
|
||||
headroom: float = 15.0,
|
||||
) -> float:
|
||||
"""Seconds a CLI should wait for the gateway PID to exit after SIGUSR1.
|
||||
|
||||
In-band restart may defer ``stop()`` until active turns finish
|
||||
(``after_turn_timeout``) and then spend up to ``drain_timeout`` inside
|
||||
``stop()``. Callers that fall back to a hard kill on wait expiry must
|
||||
cover both phases or they reintroduce #77184.
|
||||
"""
|
||||
try:
|
||||
drain = max(float(drain_timeout), 0.0)
|
||||
except (TypeError, ValueError):
|
||||
drain = 0.0
|
||||
try:
|
||||
after_turn = max(float(after_turn_timeout), 0.0)
|
||||
except (TypeError, ValueError):
|
||||
after_turn = 0.0
|
||||
try:
|
||||
margin = max(float(headroom), 0.0)
|
||||
except (TypeError, ValueError):
|
||||
margin = 0.0
|
||||
return drain + after_turn + margin
|
||||
|
|
|
|||
108
gateway/run.py
108
gateway/run.py
|
|
@ -2278,9 +2278,11 @@ from gateway.shutdown_watchdog import (
|
|||
start_loop_liveness_watchdog,
|
||||
)
|
||||
from gateway.restart import (
|
||||
DEFAULT_GATEWAY_RESTART_AFTER_TURN_TIMEOUT,
|
||||
DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT,
|
||||
GATEWAY_FATAL_CONFIG_EXIT_CODE,
|
||||
GATEWAY_SERVICE_RESTART_EXIT_CODE,
|
||||
parse_restart_after_turn_timeout,
|
||||
parse_restart_drain_timeout,
|
||||
)
|
||||
|
||||
|
|
@ -5623,6 +5625,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
_busy_input_mode: str = "interrupt"
|
||||
_busy_text_mode: str = "interrupt"
|
||||
_restart_drain_timeout: float = DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT
|
||||
_restart_after_turn_timeout: float = DEFAULT_GATEWAY_RESTART_AFTER_TURN_TIMEOUT
|
||||
_exit_code: Optional[int] = None
|
||||
_draining: bool = False
|
||||
_external_drain_active: bool = False
|
||||
|
|
@ -5759,6 +5762,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
self._busy_input_mode = self._load_busy_input_mode()
|
||||
self._busy_text_mode = self._load_busy_text_mode()
|
||||
self._restart_drain_timeout = self._load_restart_drain_timeout()
|
||||
self._restart_after_turn_timeout = self._load_restart_after_turn_timeout()
|
||||
self._provider_routing = self._load_provider_routing()
|
||||
self._fallback_model = self._load_fallback_model()
|
||||
|
||||
|
|
@ -8182,6 +8186,29 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
)
|
||||
return value
|
||||
|
||||
@staticmethod
|
||||
def _load_restart_after_turn_timeout() -> float:
|
||||
"""Load in-band restart wait-for-idle timeout in seconds (#77184)."""
|
||||
env_raw = os.getenv("HERMES_RESTART_AFTER_TURN_TIMEOUT")
|
||||
if env_raw is not None and str(env_raw).strip() != "":
|
||||
raw: object = env_raw
|
||||
else:
|
||||
cfg = _load_gateway_runtime_config()
|
||||
raw = cfg_get(cfg, "agent", "restart_after_turn_timeout", default=None)
|
||||
value = parse_restart_after_turn_timeout(raw)
|
||||
# Warn only when the user supplied a non-empty value that failed to
|
||||
# parse (parser falls back to the default). ``0`` is valid.
|
||||
if raw is not None and str(raw).strip() != "":
|
||||
try:
|
||||
float(raw)
|
||||
except (TypeError, ValueError):
|
||||
logger.warning(
|
||||
"Invalid restart_after_turn_timeout '%s', using default %.0fs",
|
||||
raw,
|
||||
DEFAULT_GATEWAY_RESTART_AFTER_TURN_TIMEOUT,
|
||||
)
|
||||
return value
|
||||
|
||||
@staticmethod
|
||||
def _load_background_notifications_mode() -> str:
|
||||
"""Load background process notification mode from config or env var.
|
||||
|
|
@ -9903,6 +9930,78 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
except Exception as e:
|
||||
logger.debug("Failed to launch systemd planned-restart helper: %s", e)
|
||||
|
||||
async def _await_active_work_before_restart(self) -> bool:
|
||||
"""Wait for in-flight work to finish before entering ``stop()``.
|
||||
|
||||
In-band restart used to call ``stop()`` immediately, which folded the
|
||||
requesting turn into the drain wait set and force-interrupted it at
|
||||
``restart_drain_timeout`` (#77184). Instead we refuse new turns and
|
||||
wait here for active agents/cron/api work to reach zero, then let
|
||||
``stop()`` run against an idle gateway (drain is instant).
|
||||
|
||||
Returns True when work drained to zero, False when the safety cap
|
||||
elapsed with work still active (caller proceeds to ``stop()``, which
|
||||
may then interrupt remaining runs under ``restart_drain_timeout``).
|
||||
"""
|
||||
active = self._active_work_count()
|
||||
if active <= 0:
|
||||
return True
|
||||
|
||||
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,
|
||||
)
|
||||
try:
|
||||
self._update_runtime_status("draining")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
loop = asyncio.get_running_loop()
|
||||
deadline = loop.time() + timeout
|
||||
last_status_at = 0.0
|
||||
while self._active_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) "
|
||||
"(%.0fs remaining before force drain)",
|
||||
self._active_work_count(),
|
||||
deadline - now,
|
||||
)
|
||||
try:
|
||||
self._update_runtime_status("draining")
|
||||
except Exception:
|
||||
pass
|
||||
last_status_at = now
|
||||
await asyncio.sleep(0.1)
|
||||
|
||||
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
|
||||
|
|
@ -9910,8 +10009,17 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
self._restart_detached = detached
|
||||
self._restart_via_service = via_service
|
||||
self._restart_task_started = True
|
||||
# Refuse new turns immediately while in-flight work finishes.
|
||||
# Keep ``_running`` True so adapters stay connected and the active
|
||||
# turn can still deliver its final response (#77184).
|
||||
self._draining = True
|
||||
|
||||
async def _run_restart() -> None:
|
||||
await self._await_active_work_before_restart()
|
||||
# Launch the detached helper only AFTER the after-turn wait.
|
||||
# Its deadline is drain_timeout+5 and covers stop() teardown —
|
||||
# launching earlier would fire `hermes gateway restart` while
|
||||
# the requesting turn was still running.
|
||||
if detached:
|
||||
try:
|
||||
await self._launch_detached_restart_command()
|
||||
|
|
|
|||
|
|
@ -35,21 +35,23 @@ DEFAULT_CONFIG = {
|
|||
# tools or receiving API responses. Only fires when the agent has
|
||||
# been completely idle for this duration. 0 = unlimited.
|
||||
"gateway_timeout": 1800,
|
||||
# Graceful drain timeout for gateway stop/restart (seconds).
|
||||
# The gateway stops accepting new work, waits for running agents
|
||||
# to finish, then interrupts any remaining runs after the timeout.
|
||||
# 0 = no drain, interrupt immediately (the default).
|
||||
# Force-interrupt budget once gateway stop()/drain has begun
|
||||
# (seconds). Applies to SIGTERM/external stop and to the final
|
||||
# phase of in-band restart after any after-turn wait. 0 = interrupt
|
||||
# immediately (the default).
|
||||
#
|
||||
# Contract: if you restart the gateway, in-flight work stops. We do
|
||||
# not hold the restart open for a grace window — a drain timeout
|
||||
# large enough to "save" a long agent turn would have to outlast an
|
||||
# unbounded task (some runs take days), which is impossible, and a
|
||||
# drain timeout shorter than systemd's TimeoutStopSec invites a
|
||||
# SIGKILL-mid-cleanup race that leaves a stale lock and crash-loops
|
||||
# the service. 0 sidesteps both: interrupt now, clean up, exit fast.
|
||||
# Set a positive value in config.yaml only if you explicitly want a
|
||||
# grace window on /restart (and keep it well under TimeoutStopSec).
|
||||
# Keep this short and under systemd TimeoutStopSec — a long value
|
||||
# here invites SIGKILL-mid-cleanup. For in-band restart
|
||||
# (/restart, SIGUSR1), prefer restart_after_turn_timeout below so
|
||||
# active turns finish *before* stop() begins (#77184).
|
||||
"restart_drain_timeout": 0,
|
||||
# In-band restart wait for active turns to finish before stop()
|
||||
# (seconds). /restart and SIGUSR1 refuse new work, then wait up to
|
||||
# this cap for in-flight agents/cron/api runs to complete naturally
|
||||
# so the requesting turn is not amputated by restart_drain_timeout.
|
||||
# 0 = legacy behaviour (enter stop()/drain immediately). Default
|
||||
# 6h is a safety valve for wedged agents, not a target latency.
|
||||
"restart_after_turn_timeout": 21600,
|
||||
# Upper bound (seconds) a submitted prompt waits for the deferred
|
||||
# agent build (MCP discovery, model metadata, skills scan) before
|
||||
# failing with a visible error (#63078). The gateway's wait is
|
||||
|
|
|
|||
|
|
@ -32,12 +32,15 @@ PROJECT_ROOT = Path(__file__).parent.parent.resolve()
|
|||
from gateway.config import coerce_systemd_watchdog_seconds, load_gateway_config
|
||||
from gateway.status import terminate_pid
|
||||
from gateway.restart import (
|
||||
DEFAULT_GATEWAY_RESTART_AFTER_TURN_TIMEOUT,
|
||||
DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT,
|
||||
EXTERNAL_GATEWAY_SUPERVISOR_ENV,
|
||||
GATEWAY_FATAL_CONFIG_EXIT_CODE,
|
||||
GATEWAY_SERVICE_RESTART_EXIT_CODE,
|
||||
is_gateway_supervisor_process,
|
||||
parse_restart_after_turn_timeout,
|
||||
parse_restart_drain_timeout,
|
||||
resolve_restart_exit_wait_budget,
|
||||
)
|
||||
from hermes_cli.config import (
|
||||
get_env_value,
|
||||
|
|
@ -252,10 +255,12 @@ def _request_gateway_self_restart(pid: int) -> bool:
|
|||
def _graceful_restart_via_sigusr1(pid: int, drain_timeout: float) -> bool:
|
||||
"""Send SIGUSR1 to a gateway PID and wait for it to exit gracefully.
|
||||
|
||||
SIGUSR1 is wired in gateway/run.py to ``request_restart(via_service=True)``
|
||||
which drains in-flight agent runs (up to ``agent.restart_drain_timeout``
|
||||
seconds), then exits. Both systemd (``Restart=always``) and launchd
|
||||
(unconditional ``<key>KeepAlive</key><true/>``) restart on any exit.
|
||||
SIGUSR1 is wired in gateway/run.py to ``request_restart(via_service=True)``,
|
||||
which refuses new turns, waits for in-flight work up to
|
||||
``agent.restart_after_turn_timeout``, then runs ``stop()`` (force-interrupt
|
||||
budget ``agent.restart_drain_timeout``) and exits. Both systemd
|
||||
(``Restart=always``) and launchd (unconditional KeepAlive) restart on
|
||||
any exit.
|
||||
|
||||
This is the drain-aware alternative to ``systemctl restart`` / ``SIGTERM``,
|
||||
which SIGKILL in-flight agents after a short timeout.
|
||||
|
|
@ -264,9 +269,9 @@ def _graceful_restart_via_sigusr1(pid: int, drain_timeout: float) -> bool:
|
|||
pid: Gateway process PID (systemd MainPID, launchd PID, or bare
|
||||
process PID).
|
||||
drain_timeout: Seconds to wait for the process to exit after sending
|
||||
SIGUSR1. Should be slightly larger than the gateway's
|
||||
``agent.restart_drain_timeout`` to allow the drain loop to
|
||||
finish cleanly.
|
||||
SIGUSR1. Must cover the after-turn wait plus the stop()/drain
|
||||
phase (#77184); callers should pass
|
||||
``resolve_restart_exit_wait_budget(...)``.
|
||||
|
||||
Returns:
|
||||
True if the PID was signalled and exited within the timeout.
|
||||
|
|
@ -3276,6 +3281,26 @@ def _get_restart_drain_timeout() -> float:
|
|||
return parse_restart_drain_timeout(raw)
|
||||
|
||||
|
||||
def _get_restart_after_turn_timeout() -> float:
|
||||
"""Return the in-band restart wait-for-idle timeout in seconds (#77184)."""
|
||||
env_raw = os.getenv("HERMES_RESTART_AFTER_TURN_TIMEOUT")
|
||||
if env_raw is not None and str(env_raw).strip() != "":
|
||||
return parse_restart_after_turn_timeout(env_raw)
|
||||
cfg = read_raw_config()
|
||||
agent_cfg = cfg.get("agent", {}) if isinstance(cfg, dict) else {}
|
||||
if isinstance(agent_cfg, dict) and "restart_after_turn_timeout" in agent_cfg:
|
||||
return parse_restart_after_turn_timeout(agent_cfg.get("restart_after_turn_timeout"))
|
||||
return parse_restart_after_turn_timeout(None)
|
||||
|
||||
|
||||
def _get_restart_exit_wait_budget() -> float:
|
||||
"""CLI wait for gateway exit after SIGUSR1 / self-restart (#77184)."""
|
||||
return resolve_restart_exit_wait_budget(
|
||||
_get_restart_drain_timeout(),
|
||||
_get_restart_after_turn_timeout(),
|
||||
)
|
||||
|
||||
|
||||
def systemd_install(
|
||||
force: bool = False,
|
||||
system: bool = False,
|
||||
|
|
@ -3456,10 +3481,12 @@ def systemd_restart(system: bool = False):
|
|||
if pid is not None:
|
||||
scope_label = _service_scope_label(system).capitalize()
|
||||
svc = get_service_name()
|
||||
drain_timeout = _get_restart_drain_timeout()
|
||||
|
||||
print(f"⏳ {scope_label} service restarting gracefully (PID {pid})...")
|
||||
if _graceful_restart_via_sigusr1(pid, drain_timeout + 5):
|
||||
wait_budget = _get_restart_exit_wait_budget()
|
||||
print(
|
||||
f"⏳ {scope_label} service restarting gracefully (PID {pid}) — "
|
||||
f"waiting up to {wait_budget:.0f}s for in-flight turns + drain..."
|
||||
)
|
||||
if _graceful_restart_via_sigusr1(pid, wait_budget):
|
||||
# The gateway exits with code 75 for a planned service restart.
|
||||
# RestartSec can otherwise delay the relaunch even though the
|
||||
# operator asked for an immediate restart, so kick the unit once
|
||||
|
|
@ -3482,7 +3509,7 @@ def systemd_restart(system: bool = False):
|
|||
return
|
||||
|
||||
print(
|
||||
f"⚠ Graceful restart did not complete within {int(drain_timeout + 5)}s; "
|
||||
f"⚠ Graceful restart did not complete within {int(wait_budget)}s; "
|
||||
"forcing a service restart..."
|
||||
)
|
||||
_run_systemctl(
|
||||
|
|
|
|||
|
|
@ -292,7 +292,7 @@ TIPS = [
|
|||
"Delegation has a heartbeat thread — child activity propagates to the parent, preventing gateway timeouts.",
|
||||
"When a provider returns HTTP 402 (payment required), the auxiliary client auto-falls back to the next one.",
|
||||
"agent.tool_use_enforcement steers models that describe actions instead of calling tools — auto for GPT/Codex.",
|
||||
"agent.restart_drain_timeout (default 60s) lets running agents finish before a gateway restart takes effect.",
|
||||
"agent.restart_after_turn_timeout lets in-flight turns finish before /restart enters stop(); restart_drain_timeout is only the force-interrupt budget once stop() begins.",
|
||||
"agent.api_max_retries (default 3) controls how many times the agent retries a failed API call before surfacing the error — lower it for fast fallback.",
|
||||
"The gateway caches AIAgent instances per session — destroying this cache breaks Anthropic prompt caching.",
|
||||
"Any website can expose skills via /.well-known/skills/index.json — the skills hub discovers them automatically.",
|
||||
|
|
|
|||
|
|
@ -4863,37 +4863,20 @@ def _cmd_update_impl(args, gateway_mode: bool):
|
|||
_manage_cmd_cache[scope_] = cmd
|
||||
return cmd
|
||||
|
||||
# Drain budget for graceful SIGUSR1 restarts. The gateway drains
|
||||
# for up to ``agent.restart_drain_timeout`` (default 60s) before
|
||||
# exiting with code 75; we wait slightly longer so the drain
|
||||
# completes before we fall back to a hard restart. On older
|
||||
# systemd units without SIGUSR1 wiring this wait just times out
|
||||
# and we fall back to ``systemctl restart`` (the old behaviour).
|
||||
# Wait budget for graceful SIGUSR1 restarts. In-band restart
|
||||
# may defer stop() until active turns finish
|
||||
# (``restart_after_turn_timeout``, #77184) and then spend up to
|
||||
# ``restart_drain_timeout`` inside stop(). Cover both phases so
|
||||
# we don't fall back to a hard kill while the gateway is still
|
||||
# patiently waiting for the requesting turn. On older systemd
|
||||
# units without SIGUSR1 wiring this wait just times out and we
|
||||
# fall back to ``systemctl restart`` (the old behaviour).
|
||||
try:
|
||||
from hermes_constants import (
|
||||
DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT as _DEFAULT_DRAIN,
|
||||
)
|
||||
except Exception:
|
||||
_DEFAULT_DRAIN = 60.0
|
||||
_cfg_drain = None
|
||||
try:
|
||||
from hermes_cli.config import load_config
|
||||
from hermes_cli.gateway import _get_restart_exit_wait_budget
|
||||
|
||||
_cfg_agent = load_config().get("agent") or {}
|
||||
_cfg_drain = _cfg_agent.get("restart_drain_timeout")
|
||||
_drain_budget = max(float(_get_restart_exit_wait_budget()), 45.0)
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
_drain_budget = (
|
||||
float(_cfg_drain)
|
||||
if _cfg_drain is not None
|
||||
else float(_DEFAULT_DRAIN)
|
||||
)
|
||||
except (TypeError, ValueError):
|
||||
_drain_budget = float(_DEFAULT_DRAIN)
|
||||
# Add a 15s margin so the drain loop + final exit finish before
|
||||
# we escalate to ``systemctl restart`` / SIGTERM.
|
||||
_drain_budget = max(_drain_budget, 30.0) + 15.0
|
||||
_drain_budget = 45.0
|
||||
|
||||
restarted_services = []
|
||||
failed_or_stale_units = []
|
||||
|
|
|
|||
Loading…
Reference in New Issue