From ebf2b6c9feb5b37bf9df78a8c0142407a367b1d1 Mon Sep 17 00:00:00 2001 From: joaomarcos Date: Sun, 9 Aug 2026 00:39:43 -0300 Subject: [PATCH 1/2] fix(gateway): give in-flight cron work its own drain floor MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `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). --- cli-config.yaml.example | 13 ++ gateway/restart.py | 76 +++++++++ gateway/run.py | 98 ++++++++++-- hermes_cli/config_defaults.py | 9 ++ tests/gateway/restart_test_helpers.py | 2 + tests/gateway/test_cron_active_work_drain.py | 1 + tests/gateway/test_cron_drain_floor.py | 158 +++++++++++++++++++ 7 files changed, 342 insertions(+), 15 deletions(-) create mode 100644 tests/gateway/test_cron_drain_floor.py diff --git a/cli-config.yaml.example b/cli-config.yaml.example index 08343f15dd77d..d6f9cf7fc3dfa 100644 --- a/cli-config.yaml.example +++ b/cli-config.yaml.example @@ -870,6 +870,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 diff --git a/gateway/restart.py b/gateway/restart.py index 5a2cfbc1b73a1..c1529e320a5f4 100644 --- a/gateway/restart.py +++ b/gateway/restart.py @@ -33,6 +33,24 @@ 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, diff --git a/gateway/run.py b/gateway/run.py index c3aae82cf53af..be548942a4a28 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -2215,6 +2215,8 @@ def _platform_has_bot_credential(platform: "Platform", platform_config: "Platfor ) 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"] @@ -2433,12 +2435,15 @@ def _platform_has_bot_credential(platform: "Platform", platform_config: "Platfor 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, ) @@ -5841,6 +5846,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 @@ -5978,6 +5984,7 @@ def __init__(self, config: Optional[GatewayConfig] = None): 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._cron_drain_timeout = self._load_cron_drain_timeout() self._provider_routing = self._load_provider_routing() self._fallback_model = self._load_fallback_model() @@ -8513,6 +8520,29 @@ def _load_restart_after_turn_timeout() -> float: ) 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. @@ -9341,7 +9371,9 @@ async def _handle_active_session_busy_message(self, event: MessageEvent, session 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() @@ -9378,18 +9410,32 @@ def _maybe_update_status(force: bool = False) -> None: 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 = ( @@ -13058,15 +13104,37 @@ def _phase_elapsed() -> float: _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(), @@ -13095,7 +13163,7 @@ def _phase_elapsed() -> float: "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(), diff --git a/hermes_cli/config_defaults.py b/hermes_cli/config_defaults.py index 8b5ef69c9022e..14ded653ab39e 100644 --- a/hermes_cli/config_defaults.py +++ b/hermes_cli/config_defaults.py @@ -73,6 +73,15 @@ # (/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 diff --git a/tests/gateway/restart_test_helpers.py b/tests/gateway/restart_test_helpers.py index 7de82a65bbfcd..589cdca1d82e4 100644 --- a/tests/gateway/restart_test_helpers.py +++ b/tests/gateway/restart_test_helpers.py @@ -5,6 +5,7 @@ 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 = {} diff --git a/tests/gateway/test_cron_active_work_drain.py b/tests/gateway/test_cron_active_work_drain.py index 07616a7880a88..a2f6ac27382c1 100644 --- a/tests/gateway/test_cron_active_work_drain.py +++ b/tests/gateway/test_cron_active_work_drain.py @@ -85,6 +85,7 @@ async def test_in_flight_cron_job_marked_interrupted_on_forced_kill(self, monkey 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") diff --git a/tests/gateway/test_cron_drain_floor.py b/tests/gateway/test_cron_drain_floor.py new file mode 100644 index 0000000000000..f4065ac02cf38 --- /dev/null +++ b/tests/gateway/test_cron_drain_floor.py @@ -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 From 82f28b9ad6081d2e0ca66da8759e8e06998232ee Mon Sep 17 00:00:00 2001 From: joaomarcos Date: Sun, 9 Aug 2026 01:59:58 -0300 Subject: [PATCH 2/2] fix(gateway): keep the cron drain floor compatible with shutdown test doubles MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CI slice 5/12 caught two ways the new cron budget broke `_stop_impl_body` for callers that are not real GatewayRunner instances: - `_FakeGateway` in test_shutdown_cache_cleanup.py borrows `_stop_impl` without subclassing, so it never picked up the class-level `_cron_drain_timeout` default and raised AttributeError. Read it through the getattr-guard convention the same function already uses for its liveness-guard machinery. - The same double overrides `_drain_active_agents(self, timeout)`, so passing the cron budget raised "takes 2 positional arguments but 3 were given". The double now mirrors the real optional parameter. It is the only override in the tree; test_startup_restart_race.py uses AsyncMock, which accepts any signature. Verified against a stashed clean tree: the 22 gateway test files that still fail locally fail identically with and without this branch (80 = 80, empty set difference both ways) — they are pre-existing Windows-only failures (setsid, POSIX modes) unrelated to this change. --- gateway/run.py | 10 ++++++++-- tests/gateway/test_shutdown_cache_cleanup.py | 2 +- 2 files changed, 9 insertions(+), 3 deletions(-) diff --git a/gateway/run.py b/gateway/run.py index be548942a4a28..b505ed09356ca 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -13107,9 +13107,15 @@ def _phase_elapsed() -> float: # 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). + # getattr-guard: shutdown-path tests drive _stop_impl_body from + # bare doubles that aren't GatewayRunner instances, so they don't + # pick up the class-level default. + _cron_drain_cfg = getattr( + self, "_cron_drain_timeout", DEFAULT_GATEWAY_CRON_DRAIN_TIMEOUT + ) _cron_timeout = resolve_cron_drain_budget( timeout, - self._cron_drain_timeout, + _cron_drain_cfg, watchdog_delay=resolve_shutdown_watchdog_delay(timeout), elapsed=_phase_elapsed(), ) @@ -13120,7 +13126,7 @@ def _phase_elapsed() -> float: "restart_drain_timeout=%.0fs)", _cron_at_start, _cron_timeout, - self._cron_drain_timeout, + _cron_drain_cfg, timeout, ) _drain_started_at = time.monotonic() diff --git a/tests/gateway/test_shutdown_cache_cleanup.py b/tests/gateway/test_shutdown_cache_cleanup.py index ffd4b1b3b6e98..be5089eaa49cf 100644 --- a/tests/gateway/test_shutdown_cache_cleanup.py +++ b/tests/gateway/test_shutdown_cache_cleanup.py @@ -84,7 +84,7 @@ async def _notify_active_sessions_of_shutdown(self): async def _cancel_secondary_profile_reconnect_tasks(self): pass - async def _drain_active_agents(self, timeout): + async def _drain_active_agents(self, timeout, cron_timeout=None): return {}, False async def _finalize_shutdown_agents(self, agents):