fix(gateway): give in-flight cron work its own drain floor

`agent.restart_drain_timeout` defaults to 0 and governed every class of
in-flight work at once. That default is deliberate for chat turns: the
gateway announces the restart to the user and pre-marks the session
resume_pending, so interrupting one is cheap and recoverable.

A cron run has neither property. Nobody is waiting on it, it is written
to jobs.json as a permanent failure, and a recurring job simply skips to
its next schedule. Sharing the chat budget meant `_drain_active_agents()`
short-circuited on `timeout <= 0` before entering the wait loop, so the
drain reported `drain took 0.00s, timed_out=True, cron_at_start=1,
cron_now=1` — it detected the job and killed it anyway.

Cron work now drains on its own deadline, `agent.cron_drain_timeout`
(default 30s, 0 opts out). The floor is clamped to the shutdown-watchdog
leash minus a teardown reserve, so the longer wait can never consume the
post-drain cleanup window: being SIGKILLed mid-cleanup would leave the
job wedged at `last_status=running`, strictly worse than the bug. Being
bounded also means a cron-triggered restart cannot deadlock on itself.

The `timeout <= 0` special case is gone — an expired deadline expresses
the legacy "interrupt immediately" behaviour, so `timed_out` is always
computed from real state instead of asserted up front. The drain-timeout
warning now reports the elapsed wait rather than the configured budget,
which is what made "timed out after 0.0s" so confusing in the report.

Chat-only shutdowns are unchanged: `restart_drain_timeout: 0` still
interrupts chat turns immediately.

Relates to #82161 (complements #82195, which removes the `hermes update`
self-deadlock that triggered the reported instance).
This commit is contained in:
joaomarcos
2026-08-09 00:39:43 -03:00
committed by Teknium
parent 1f6f86119f
commit 45bb486b26
7 changed files with 342 additions and 15 deletions
+13
View File
@@ -913,6 +913,19 @@ agent:
# window on /restart, and keep it well under systemd's TimeoutStopSec.
# restart_drain_timeout: 0
# Cron-only floor under the same drain (seconds). Default 30.
# restart_drain_timeout above is written for chat turns, which are cheap to
# interrupt: the user is told the gateway is restarting and the session
# resumes on their next message. A cron run has no such safety net — it is
# recorded in jobs.json as a permanent failure, nobody is waiting on it, and
# a recurring job simply skips to its next schedule. So in-flight cron work
# gets its own grace window instead of inheriting the 0 above.
# Clamped at runtime to the shutdown-watchdog leash (restart_drain_timeout
# + 60s) minus teardown headroom, so values past ~50s need a matching
# TimeoutStopSec bump to take effect. Set 0 to opt out and drain cron on
# restart_drain_timeout like before.
# cron_drain_timeout: 30
# Upper bound (seconds) a submitted prompt waits for the deferred agent
# build (MCP discovery, model metadata, skills scan) before failing with a
# visible error. The wait is patient — the message is delivered as soon as
+76
View File
@@ -33,6 +33,24 @@ DEFAULT_GATEWAY_RESTART_AFTER_TURN_TIMEOUT = float(
DEFAULT_CONFIG["agent"]["restart_after_turn_timeout"]
)
# Cron-only floor under the ``stop()`` drain. ``restart_drain_timeout``
# defaults to 0 because interrupting a *chat* turn is cheap and recoverable:
# the user is told the gateway is restarting and the session is pre-marked
# resume_pending. An interrupted *cron* run has neither property — nobody is
# waiting on it, it lands in jobs.json as a permanent failure, and a recurring
# job just waits for its next schedule — so a zero-second drain silently
# destroys work. See #82161.
DEFAULT_GATEWAY_CRON_DRAIN_TIMEOUT = float(
DEFAULT_CONFIG["agent"]["cron_drain_timeout"]
)
# Seconds of the shutdown watchdog leash held back for the work that still has
# to happen after the drain returns: interrupt agents, kill tool subprocesses,
# mark in-flight jobs interrupted, disconnect adapters. Waiting for cron past
# that point trades a job that is killed *and recorded* for one that is
# SIGKILLed mid-write and stays wedged at ``last_status=running`` forever.
CRON_DRAIN_CLEANUP_RESERVE_S = 10.0
def is_gateway_supervisor_process(
environ: Mapping[str, str] | None = None,
@@ -92,6 +110,64 @@ def parse_restart_after_turn_timeout(raw: object) -> float:
return max(0.0, value)
def parse_cron_drain_timeout(raw: object) -> float:
"""Parse the cron-only drain floor, falling back to the shared default.
``0`` is a deliberate opt-out — cron work is then interrupted on the same
budget as chat work, the pre-#82161 behaviour — and must not fall through
to the default, unlike empty/missing input.
"""
if raw is None:
return DEFAULT_GATEWAY_CRON_DRAIN_TIMEOUT
if isinstance(raw, str) and not raw.strip():
return DEFAULT_GATEWAY_CRON_DRAIN_TIMEOUT
try:
value = float(raw)
except (TypeError, ValueError):
return DEFAULT_GATEWAY_CRON_DRAIN_TIMEOUT
return max(0.0, value)
def resolve_cron_drain_budget(
drain_timeout: float,
cron_drain_timeout: float,
*,
watchdog_delay: float,
elapsed: float = 0.0,
cleanup_reserve_s: float = CRON_DRAIN_CLEANUP_RESERVE_S,
) -> float:
"""Seconds the shutdown drain may spend waiting on in-flight cron work.
The configured floor is clamped to what this process can actually honour.
The shutdown watchdog hard-exits at ``watchdog_delay`` and the service
manager's ``TimeoutStopSec`` is sized from the same drain timeout, so
waiting past that leash (minus ``cleanup_reserve_s`` for the teardown that
follows the drain) would swap a cleanly-interrupted job for a SIGKILL that
leaves it wedged mid-run — strictly worse than the bug being fixed.
Never returns less than ``drain_timeout``: the cron floor only ever
extends the wait, so an operator who deliberately configured a long
``restart_drain_timeout`` keeps it.
"""
def _seconds(value: object, fallback: float = 0.0) -> float:
try:
return max(float(value), 0.0) # type: ignore[arg-type]
except (TypeError, ValueError):
return fallback
drain = _seconds(drain_timeout)
floor = _seconds(cron_drain_timeout)
if floor <= 0.0:
return drain
ceiling = (
_seconds(watchdog_delay)
- _seconds(elapsed)
- _seconds(cleanup_reserve_s, CRON_DRAIN_CLEANUP_RESERVE_S)
)
return max(drain, min(floor, ceiling))
def resolve_restart_exit_wait_budget(
drain_timeout: float,
after_turn_timeout: float,
+83 -15
View File
@@ -2271,6 +2271,8 @@ if _config_path.exists():
)
if "restart_drain_timeout" in _agent_cfg:
os.environ["HERMES_RESTART_DRAIN_TIMEOUT"] = str(_agent_cfg["restart_drain_timeout"])
if "cron_drain_timeout" in _agent_cfg:
os.environ["HERMES_CRON_DRAIN_TIMEOUT"] = str(_agent_cfg["cron_drain_timeout"])
if "gateway_auto_continue_freshness" in _agent_cfg:
os.environ["HERMES_AUTO_CONTINUE_FRESHNESS"] = str(
_agent_cfg["gateway_auto_continue_freshness"]
@@ -2491,12 +2493,15 @@ from gateway.shutdown_watchdog import (
start_loop_liveness_watchdog,
)
from gateway.restart import (
DEFAULT_GATEWAY_CRON_DRAIN_TIMEOUT,
DEFAULT_GATEWAY_RESTART_AFTER_TURN_TIMEOUT,
DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT,
GATEWAY_FATAL_CONFIG_EXIT_CODE,
GATEWAY_SERVICE_RESTART_EXIT_CODE,
parse_cron_drain_timeout,
parse_restart_after_turn_timeout,
parse_restart_drain_timeout,
resolve_cron_drain_budget,
)
@@ -6355,6 +6360,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
_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
_cron_drain_timeout: float = DEFAULT_GATEWAY_CRON_DRAIN_TIMEOUT
_exit_code: Optional[int] = None
_draining: bool = False
_external_drain_active: bool = False
@@ -6497,6 +6503,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
self._busy_text_modes_by_profile: Dict[str, str] = {}
self._restart_drain_timeout = self._load_restart_drain_timeout()
self._restart_after_turn_timeout = self._load_restart_after_turn_timeout()
self._cron_drain_timeout = self._load_cron_drain_timeout()
self._provider_routing = self._load_provider_routing()
self._fallback_model = self._load_fallback_model()
@@ -9199,6 +9206,29 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
)
return value
@staticmethod
def _load_cron_drain_timeout() -> float:
"""Load the cron-only floor under the stop()/drain wait (#82161)."""
env_raw = os.getenv("HERMES_CRON_DRAIN_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", "cron_drain_timeout", default=None)
value = parse_cron_drain_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 cron_drain_timeout '%s', using default %.0fs",
raw,
DEFAULT_GATEWAY_CRON_DRAIN_TIMEOUT,
)
return value
@staticmethod
def _load_background_notifications_mode() -> str:
"""Load background process notification mode from config or env var.
@@ -10050,7 +10080,9 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
return True
async def _drain_active_agents(self, timeout: float) -> tuple[Dict[str, Any], bool]:
async def _drain_active_agents(
self, timeout: float, cron_timeout: Optional[float] = None
) -> tuple[Dict[str, Any], bool]:
snapshot = self._snapshot_running_agents()
last_active_count = self._running_agent_count()
last_cron_count = self._active_cron_job_count()
@@ -10087,18 +10119,32 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
return snapshot, False
_maybe_update_status(force=True)
if timeout <= 0:
return snapshot, True
deadline = asyncio.get_running_loop().time() + timeout
while (
(
len(self._running_agents)
or self._active_cron_job_count()
or self._active_api_run_count()
)
and asyncio.get_running_loop().time() < deadline
):
# Cron work drains on its own deadline. ``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).
loop = asyncio.get_running_loop()
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()
if (
len(self._running_agents) or self._active_api_run_count()
) and now < deadline:
return True
return bool(self._active_cron_job_count()) and now < cron_deadline
# Both budgets at 0 leave this loop unentered, which is the legacy
# "interrupt immediately" behaviour — expressed as an expired
# deadline rather than a special case, so the timed_out value below
# is always computed from real state instead of asserted up front.
while _still_draining():
_maybe_update_status()
await asyncio.sleep(0.1)
timed_out = (
@@ -13897,15 +13943,37 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
_cron_at_start = self._active_cron_job_count()
_api_at_start = self._active_api_run_count()
# In-flight cron work gets its own floor, clamped to the watchdog
# leash we're already running under so the extra wait can never
# cost us the post-drain cleanup window (#82161).
_cron_timeout = resolve_cron_drain_budget(
timeout,
self._cron_drain_timeout,
watchdog_delay=resolve_shutdown_watchdog_delay(timeout),
elapsed=_phase_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,
self._cron_drain_timeout,
timeout,
)
_drain_started_at = time.monotonic()
active_agents, timed_out = await self._drain_active_agents(timeout)
active_agents, timed_out = await self._drain_active_agents(
timeout, _cron_timeout
)
_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)",
_phase_elapsed(),
time.monotonic() - _drain_started_at,
_drain_elapsed,
timed_out,
len(active_agents),
self._running_agent_count(),
@@ -13934,7 +14002,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
"Gateway drain timed out after %.1fs with %d active agent(s), "
"%d in-flight cron job(s), and %d api_server run(s); "
"interrupting remaining work.",
timeout,
_drain_elapsed,
self._running_agent_count(),
self._active_cron_job_count(),
self._active_api_run_count(),
+9
View File
@@ -78,6 +78,15 @@ DEFAULT_CONFIG = {
# (/restart, SIGUSR1), prefer restart_after_turn_timeout below so
# active turns finish *before* stop() begins (#77184).
"restart_drain_timeout": 0,
# Cron-only floor under the stop()/drain wait (seconds). A chat turn
# interrupted by a restart is announced to the user and resumed on
# their next message; an interrupted cron run is written to jobs.json
# as a permanent failure that nobody is waiting on, so it must not
# inherit restart_drain_timeout's 0 (#82161). Clamped at runtime to
# the shutdown-watchdog leash minus teardown headroom, so raising it
# past ~50s has no effect unless TimeoutStopSec is raised too.
# 0 = opt out (cron drains on restart_drain_timeout, legacy).
"cron_drain_timeout": 30,
# 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
+2
View File
@@ -5,6 +5,7 @@ from unittest.mock import AsyncMock, MagicMock
from gateway.config import GatewayConfig, Platform, PlatformConfig
from gateway.platforms.base import BasePlatformAdapter, SendResult
from gateway.restart import (
DEFAULT_GATEWAY_CRON_DRAIN_TIMEOUT,
DEFAULT_GATEWAY_RESTART_AFTER_TURN_TIMEOUT,
DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT,
)
@@ -77,6 +78,7 @@ def make_restart_runner(
runner._restart_command_source = None
runner._restart_drain_timeout = DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT
runner._restart_after_turn_timeout = DEFAULT_GATEWAY_RESTART_AFTER_TURN_TIMEOUT
runner._cron_drain_timeout = DEFAULT_GATEWAY_CRON_DRAIN_TIMEOUT
runner._stop_task = None
runner._busy_input_mode = "interrupt"
runner._update_prompt_pending = {}
@@ -87,6 +87,7 @@ class TestKillToolSubprocessesMarksCronInterrupted:
runner, adapter = make_restart_runner()
runner._restart_drain_timeout = 0.01 # force the timeout path
runner._cron_drain_timeout = 0.01 # ...past the cron floor too (#82161)
adapter.disconnect = _make_async_noop()
sched._running_job_ids.add("job-1")
+158
View File
@@ -0,0 +1,158 @@
"""Regression tests for #82161.
``restart_drain_timeout`` defaults to ``0``, and the drain applied that single
budget to every class of in-flight work. That default is deliberate for chat
turns — the user is told the gateway is restarting and the session is
pre-marked resume_pending, so interrupting one is cheap and recoverable — but
a cron run has neither property: it is written to jobs.json as a permanent
failure that nobody is waiting on, and a recurring job just skips to its next
schedule.
With the shared budget the drain short-circuited on ``timeout <= 0`` before
the wait loop, producing the reported log line: ``drain took 0.00s,
timed_out=True, cron_at_start=1, cron_now=1`` — it detected the job and killed
it anyway. Cron work now drains on its own floor (``cron_drain_timeout``),
clamped to the shutdown-watchdog leash so the extra wait can never eat the
post-drain cleanup window.
"""
import asyncio
import pytest
from gateway.restart import (
CRON_DRAIN_CLEANUP_RESERVE_S,
DEFAULT_GATEWAY_CRON_DRAIN_TIMEOUT,
parse_cron_drain_timeout,
resolve_cron_drain_budget,
)
from tests.gateway.restart_test_helpers import make_restart_runner
@pytest.fixture(autouse=True)
def _reset_cron_running_set():
import cron.scheduler as sched
sched._running_job_ids.clear()
sched._interrupted_job_ids.clear()
yield
sched._running_job_ids.clear()
sched._interrupted_job_ids.clear()
class TestDrainWaitsForCronOnDefaultConfig:
"""The reported repro: default config, cron-only workload."""
@pytest.mark.asyncio
async def test_zero_drain_timeout_still_waits_for_cron(self):
import cron.scheduler as sched
runner, _adapter = make_restart_runner()
sched._running_job_ids.add("be62d36a9914")
async def finish_job():
await asyncio.sleep(0.12)
sched._running_job_ids.discard("be62d36a9914")
task = asyncio.create_task(finish_job())
# restart_drain_timeout=0 (the shipped default) with a 2s cron floor.
_snapshot, timed_out = await runner._drain_active_agents(0.0, 2.0)
await task
assert timed_out is False, (
"drain returned timed_out=True with a cron job in flight — this is "
"the 0.00s drain from #82161"
)
assert runner._active_cron_job_count() == 0
@pytest.mark.asyncio
async def test_cron_floor_is_bounded_not_indefinite(self):
"""A job that never finishes must still lose, or a cron-triggered
restart (the reporter's `hermes update` job) would deadlock: the job
waits for the gateway to exit while the gateway waits for the job."""
import cron.scheduler as sched
runner, _adapter = make_restart_runner()
sched._running_job_ids.add("never-finishes")
_snapshot, timed_out = await runner._drain_active_agents(0.0, 0.2)
assert timed_out is True
assert runner._active_cron_job_count() == 1
@pytest.mark.asyncio
async def test_chat_only_workload_keeps_the_zero_second_drain(self):
"""The cron floor must not silently become a chat-turn grace window —
`restart_drain_timeout: 0` still means "interrupt chat immediately"."""
runner, _adapter = make_restart_runner()
runner._running_agents = {"sess-1": object()}
loop = asyncio.get_running_loop()
before = loop.time()
_snapshot, timed_out = await runner._drain_active_agents(0.0, 30.0)
elapsed = loop.time() - before
assert timed_out is True
assert elapsed < 1.0, f"chat-only drain waited {elapsed:.2f}s on a 0s budget"
@pytest.mark.asyncio
async def test_cron_timeout_defaults_to_the_shared_budget(self):
"""Callers that pass one argument keep the pre-#82161 semantics."""
import cron.scheduler as sched
runner, _adapter = make_restart_runner()
sched._running_job_ids.add("job-1")
_snapshot, timed_out = await runner._drain_active_agents(0.0)
assert timed_out is True
class TestParseCronDrainTimeout:
def test_missing_and_blank_fall_back_to_default(self):
assert parse_cron_drain_timeout(None) == DEFAULT_GATEWAY_CRON_DRAIN_TIMEOUT
assert parse_cron_drain_timeout("") == DEFAULT_GATEWAY_CRON_DRAIN_TIMEOUT
assert parse_cron_drain_timeout(" ") == DEFAULT_GATEWAY_CRON_DRAIN_TIMEOUT
def test_zero_is_a_deliberate_opt_out_not_a_missing_value(self):
assert parse_cron_drain_timeout(0) == 0.0
assert parse_cron_drain_timeout("0") == 0.0
def test_garbage_falls_back_and_negatives_clamp(self):
assert parse_cron_drain_timeout("soon") == DEFAULT_GATEWAY_CRON_DRAIN_TIMEOUT
assert parse_cron_drain_timeout(-5) == 0.0
class TestResolveCronDrainBudget:
def test_extends_a_zero_drain_up_to_the_configured_floor(self):
assert resolve_cron_drain_budget(
0.0, 30.0, watchdog_delay=60.0, elapsed=1.0
) == 30.0
def test_clamped_to_the_watchdog_leash_minus_cleanup_reserve(self):
# Watchdog hard-exits at 60s; waiting 300s would guarantee a SIGKILL
# mid-cleanup, leaving the job wedged at last_status=running.
budget = resolve_cron_drain_budget(
0.0, 300.0, watchdog_delay=60.0, elapsed=5.0
)
assert budget == pytest.approx(60.0 - 5.0 - CRON_DRAIN_CLEANUP_RESERVE_S)
def test_never_shortens_an_explicitly_configured_drain_timeout(self):
assert resolve_cron_drain_budget(
120.0, 30.0, watchdog_delay=180.0, elapsed=0.0
) == 120.0
def test_no_headroom_left_falls_back_to_the_drain_timeout(self):
assert resolve_cron_drain_budget(
0.0, 30.0, watchdog_delay=60.0, elapsed=59.0
) == 0.0
def test_zero_floor_opts_out_entirely(self):
assert resolve_cron_drain_budget(
0.0, 0.0, watchdog_delay=60.0, elapsed=0.0
) == 0.0
def test_non_numeric_inputs_degrade_instead_of_raising(self):
assert resolve_cron_drain_budget(
None, "30", watchdog_delay=60.0, elapsed=None
) == 30.0