From db3f7e4eb99731ddddc8c091f2e4fd28ff835bc9 Mon Sep 17 00:00:00 2001 From: HexLab98 Date: Mon, 3 Aug 2026 07:58:23 +0700 Subject: [PATCH] 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. --- gateway/restart.py | 54 +++++++++++++++++ gateway/run.py | 108 ++++++++++++++++++++++++++++++++++ hermes_cli/config_defaults.py | 28 +++++---- hermes_cli/gateway.py | 51 ++++++++++++---- hermes_cli/tips.py | 2 +- hermes_cli/update_cmd.py | 39 ++++-------- 6 files changed, 228 insertions(+), 54 deletions(-) diff --git a/gateway/restart.py b/gateway/restart.py index 41a7484329578..5a2cfbc1b73a1 100644 --- a/gateway/restart.py +++ b/gateway/restart.py @@ -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 diff --git a/gateway/run.py b/gateway/run.py index fd1d62feb1028..9e6b76ceb2c9b 100644 --- a/gateway/run.py +++ b/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() diff --git a/hermes_cli/config_defaults.py b/hermes_cli/config_defaults.py index 255da8e0a1cb4..b070819017413 100644 --- a/hermes_cli/config_defaults.py +++ b/hermes_cli/config_defaults.py @@ -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 diff --git a/hermes_cli/gateway.py b/hermes_cli/gateway.py index 681ce09c9ef14..2a03ff1004eac 100644 --- a/hermes_cli/gateway.py +++ b/hermes_cli/gateway.py @@ -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 ``KeepAlive``) 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( diff --git a/hermes_cli/tips.py b/hermes_cli/tips.py index be8cb25b15d04..504f29115fd1a 100644 --- a/hermes_cli/tips.py +++ b/hermes_cli/tips.py @@ -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.", diff --git a/hermes_cli/update_cmd.py b/hermes_cli/update_cmd.py index 95982c0f28eed..240bb0fc9b564 100644 --- a/hermes_cli/update_cmd.py +++ b/hermes_cli/update_cmd.py @@ -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 = []