From 769dba1758ad0adadc8e9b511f9e076d68ee0d5d Mon Sep 17 00:00:00 2001 From: Kyzcreig <9063726+Kyzcreig@users.noreply.github.com> Date: Sun, 26 Jul 2026 03:37:54 -0700 Subject: [PATCH] fix(gateway): bound the startup-restore inbound gate on a slow boot-resume turn --- gateway/run.py | 111 ++++++++++- hermes_cli/config.py | 12 ++ tests/gateway/test_restart_resume_pending.py | 185 +++++++++++++++++++ 3 files changed, 303 insertions(+), 5 deletions(-) diff --git a/gateway/run.py b/gateway/run.py index c45b53fdfec..927bd6069fe 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -741,6 +741,14 @@ def _telegramize_command_mentions(text: str, platform: Any) -> str: # ``config.yaml`` ``agent.gateway_auto_continue_freshness``. _AUTO_CONTINUE_FRESHNESS_SECS_DEFAULT = 60 * 60 +# Default bound for how long ``_finish_startup_restore`` waits on boot +# auto-resume turns before releasing the inbound gate (see +# ``_startup_restore_drain_timeout_secs``). 30s is comfortably longer than a +# normal resume turn's first response yet short enough that one pathologically +# long resumed turn can't hold every channel's inbound queued for minutes. +# Override via ``config.yaml`` ``agent.gateway_startup_restore_drain_timeout``. +_STARTUP_RESTORE_DRAIN_TIMEOUT_SECS_DEFAULT = 30.0 + def _coerce_gateway_timestamp(value: Any) -> Optional[float]: """Best-effort conversion of stored gateway timestamps to epoch seconds. @@ -794,6 +802,39 @@ def _auto_continue_freshness_window() -> float: return auto_continue_freshness_window() +def _startup_restore_drain_timeout_secs() -> float: + """Max seconds ``_finish_startup_restore`` waits on boot auto-resume turns + before releasing the inbound gate and draining the queue. + + While startup restore is in progress the gateway QUEUES every inbound + message (``_queue_startup_restore_event``) instead of processing it, so no + channel gets a reply until the gate opens. The gate is opened by + ``_finish_startup_restore``, which waits for the synthetic boot + auto-resume turns to finish. A single long resumed turn therefore held + the gate shut for every channel — inbound piled up unanswered for as long + as that one turn ran. + + This bounds that wait. Duplicate-agent safety does NOT depend on the + wait: ``_schedule_resume_pending_sessions`` claims each session's + ``_running_agents`` slot SYNCHRONOUSLY (before the gate ever runs), so a + message drained while a resume turn is still running queues behind that + slot rather than spawning a second agent. So on timeout we release the + gate and let the slow turn finish in the background. + + Reads ``HERMES_STARTUP_RESTORE_DRAIN_TIMEOUT`` (bridged from + ``config.yaml`` ``agent.gateway_startup_restore_drain_timeout`` at gateway + startup, same pattern as the other ``agent.*`` knobs). Non-positive + disables the bound (restores the historical "wait forever" behaviour). + """ + raw = os.environ.get("HERMES_STARTUP_RESTORE_DRAIN_TIMEOUT") + if raw is None or raw == "": + return float(_STARTUP_RESTORE_DRAIN_TIMEOUT_SECS_DEFAULT) + try: + return float(raw) + except (TypeError, ValueError): + return float(_STARTUP_RESTORE_DRAIN_TIMEOUT_SECS_DEFAULT) + + def _float_env(name: str, default: float) -> float: """Read an env var as float, falling back to ``default`` on typos/empty. @@ -1951,6 +1992,10 @@ if _config_path.exists(): os.environ["HERMES_AUTO_CONTINUE_FRESHNESS"] = str( _agent_cfg["gateway_auto_continue_freshness"] ) + if "gateway_startup_restore_drain_timeout" in _agent_cfg: + os.environ["HERMES_STARTUP_RESTORE_DRAIN_TIMEOUT"] = str( + _agent_cfg["gateway_startup_restore_drain_timeout"] + ) # config-authoritative knobs for the session-search index; same # bridge semantics as the agent settings above. _sessions_cfg = _cfg.get("sessions", {}) @@ -7482,15 +7527,56 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew return drained async def _finish_startup_restore(self) -> None: - """Wait for startup auto-resume, then release and drain inbound queue.""" + """Wait (BOUNDED) for startup auto-resume, then release + drain inbound. + + The wait is bounded by ``_startup_restore_drain_timeout_secs`` so that + a single pathologically long boot-resume turn cannot hold the inbound + gate shut for every channel. On timeout we release the gate and let + the still-running resume turn(s) finish in the background — they are + NOT cancelled. This is safe because duplicate-agent protection does + not depend on the wait: ``_schedule_resume_pending_sessions`` claims + each session's ``_running_agents`` slot SYNCHRONOUSLY before this gate + runs, so any inbound message drained while a resume turn is still in + flight queues behind that slot instead of spawning a second agent. + """ tasks = list(getattr(self, "_startup_restore_tasks", []) or []) if tasks: - results = await asyncio.gather(*tasks, return_exceptions=True) - for result in results: - if isinstance(result, Exception): + timeout = _startup_restore_drain_timeout_secs() + if timeout > 0: + # asyncio.wait (unlike wait_for / gather+timeout) does NOT + # cancel the pending tasks on timeout — the slow resume turn + # keeps running in the background instead of being killed. + done, pending = await asyncio.wait(tasks, timeout=timeout) + if pending: + logger.warning( + "Startup-restore gate released after %.0fs with %d boot " + "auto-resume turn(s) still running; draining inbound " + "queue now (resume slots already claimed, so no " + "duplicate agents). Slow turn(s) continue in the " + "background.", + timeout, + len(pending), + ) + # These tasks outlive the gate. Their normal done-callback + # only discards them from _background_tasks, so a LATER + # failure would be silently swallowed. Attach a logging + # callback so a background resume turn that fails after the + # timeout is still recorded. + for task in pending: + task.add_done_callback(self._log_background_resume_result) + else: + # Non-positive timeout => opt out of the bound (historical + # "wait forever" behaviour). + await asyncio.gather(*tasks, return_exceptions=True) + done = set(tasks) + for task in done: + if task.cancelled(): + continue + exc = task.exception() + if exc is not None: logger.debug( "startup auto-resume task failed", - exc_info=(type(result), result, result.__traceback__), + exc_info=(type(exc), exc, exc.__traceback__), ) self._startup_restore_tasks = [] drained = await self._drain_startup_restore_queue() @@ -7498,6 +7584,21 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew if drained: logger.info("Drained %d inbound message(s) queued during startup restore", drained) + @staticmethod + def _log_background_resume_result(task: "asyncio.Task") -> None: + """Done-callback for a boot-resume turn that outlived the + startup-restore gate. Logs a late failure that would otherwise be + swallowed once the task is discarded from ``_background_tasks``. + Cancellation is expected (shutdown) and is not an error.""" + if task.cancelled(): + return + exc = task.exception() + if exc is not None: + logger.debug( + "background startup auto-resume task failed after gate release", + exc_info=(type(exc), exc, exc.__traceback__), + ) + async def _redeliver_pending_obligations(self) -> int: """Redeliver final responses recorded in the delivery ledger by a previous (now dead) gateway process. diff --git a/hermes_cli/config.py b/hermes_cli/config.py index 176f13012ec..342631126d9 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -1108,6 +1108,18 @@ DEFAULT_CONFIG = { # default is 1800s) plus runtime slack. Set to 0 to disable the # gate and restore pre-fix behaviour (always inject). "gateway_auto_continue_freshness": 3600, + # Max seconds the gateway waits for boot auto-resume turns to finish + # before it releases the startup-restore inbound gate. While startup + # restore is in progress the gateway QUEUES every inbound message + # instead of replying, so no channel gets an answer until this gate + # opens. Without a bound, one pathologically long resumed turn holds + # the gate shut and every channel's inbound piles up unanswered for as + # long as that turn runs. On timeout the gate releases and the slow + # resume turn keeps running in the background; duplicate-agent + # protection is unaffected because the resume slot is claimed + # synchronously before the gate runs. Set to 0 to disable the bound + # (historical "wait forever" behaviour). + "gateway_startup_restore_drain_timeout": 30, # Stale-stream ceiling for local providers (Ollama, oMLX, llama-cpp) in # seconds. When the base stale timeout is at its default (180s) and a # local endpoint is detected, this finite ceiling replaces the former diff --git a/tests/gateway/test_restart_resume_pending.py b/tests/gateway/test_restart_resume_pending.py index 6e05d2fe9a0..4be587093f7 100644 --- a/tests/gateway/test_restart_resume_pending.py +++ b/tests/gateway/test_restart_resume_pending.py @@ -1937,3 +1937,188 @@ async def test_auto_resume_runs_agent_exactly_once_through_full_path(): # No leaked sentinel and no orphaned queued event. assert session_key not in runner._running_agents assert session_key not in getattr(adapter, "_pending_messages", {}) + + +# --------------------------------------------------------------------------- +# Startup-restore inbound gate must be BOUNDED +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_startup_restore_gate_releases_when_resume_turn_outlives_timeout( + monkeypatch, +): + """A single slow boot-resume turn must not hold the inbound gate shut. + + While ``_startup_restore_in_progress`` is set, every inbound message is + QUEUED instead of answered. The gate is opened by + ``_finish_startup_restore``, which waits on the synthetic boot + auto-resume turns. Without a bound, one pathologically long resumed + turn holds the gate — and therefore every channel's inbound queue — + for the entire duration of that turn. + """ + monkeypatch.setenv("HERMES_STARTUP_RESTORE_DRAIN_TIMEOUT", "0.05") + + runner, adapter = make_restart_runner() + runner._startup_restore_in_progress = True + runner._startup_restore_queue = [] + runner._background_tasks = set() + + seen: list[str] = [] + never_finishes = asyncio.Event() + + async def slow_resume_turn() -> None: + await never_finishes.wait() + + async def fake_handle_message(event: MessageEvent) -> None: + seen.append(f"inbound:{event.text}") + + adapter.handle_message = fake_handle_message + + slow_task = asyncio.create_task(slow_resume_turn()) + runner._startup_restore_tasks = [slow_task] + + inbound = MessageEvent( + text="hello", + message_type=MessageType.TEXT, + source=make_restart_source(chat_id="restore-chat"), + ) + assert await runner._handle_message(inbound) is None + assert runner._startup_restore_queue == [inbound] + + # The gate must release on the bound even though the resume turn is + # still running. + await asyncio.wait_for(runner._finish_startup_restore(), timeout=5) + + assert seen == ["inbound:hello"], ( + "startup-restore gate never released: queued inbound was not drained " + "while a slow boot-resume turn was still running" + ) + assert runner._startup_restore_queue == [] + assert runner._startup_restore_in_progress is False + # The slow turn is NOT cancelled — it finishes in the background. + assert not slow_task.done() + + never_finishes.set() + await slow_task + + +@pytest.mark.asyncio +async def test_startup_restore_gate_still_waits_for_a_prompt_resume_turn( + monkeypatch, +): + """The bound must not truncate a normal-speed resume turn. + + Feature preservation: with the default (generous) timeout, a resume turn + that completes promptly is still fully awaited before the gate opens, so + the queued inbound lands behind a finished turn. + """ + monkeypatch.delenv("HERMES_STARTUP_RESTORE_DRAIN_TIMEOUT", raising=False) + + runner, adapter = make_restart_runner() + runner._startup_restore_in_progress = True + runner._startup_restore_queue = [] + runner._background_tasks = set() + + seen: list[str] = [] + resume_done = asyncio.Event() + + async def resume_turn() -> None: + await resume_done.wait() + seen.append("resume-finished") + + async def fake_handle_message(event: MessageEvent) -> None: + seen.append(f"inbound:{event.text}") + + adapter.handle_message = fake_handle_message + + runner._startup_restore_tasks = [asyncio.create_task(resume_turn())] + + inbound = MessageEvent( + text="hello", + message_type=MessageType.TEXT, + source=make_restart_source(chat_id="restore-chat"), + ) + assert await runner._handle_message(inbound) is None + + finish_task = asyncio.create_task(runner._finish_startup_restore()) + for _ in range(5): + await asyncio.sleep(0) + assert seen == [], "gate opened before the resume turn finished" + + resume_done.set() + await finish_task + assert seen == ["resume-finished", "inbound:hello"] + + +@pytest.mark.asyncio +async def test_startup_restore_drain_timeout_zero_restores_unbounded_wait( + monkeypatch, +): + """A non-positive bound opts back into the historical wait-forever gate.""" + monkeypatch.setenv("HERMES_STARTUP_RESTORE_DRAIN_TIMEOUT", "0") + + runner, adapter = make_restart_runner() + runner._startup_restore_in_progress = True + runner._startup_restore_queue = [] + runner._background_tasks = set() + + seen: list[str] = [] + resume_done = asyncio.Event() + + async def resume_turn() -> None: + await resume_done.wait() + + async def fake_handle_message(event: MessageEvent) -> None: + seen.append(f"inbound:{event.text}") + + adapter.handle_message = fake_handle_message + runner._startup_restore_tasks = [asyncio.create_task(resume_turn())] + + inbound = MessageEvent( + text="hello", + message_type=MessageType.TEXT, + source=make_restart_source(chat_id="restore-chat"), + ) + assert await runner._handle_message(inbound) is None + + finish_task = asyncio.create_task(runner._finish_startup_restore()) + await asyncio.sleep(0.15) + assert seen == [], "unbounded gate released early" + + resume_done.set() + await finish_task + assert seen == ["inbound:hello"] + + +def test_startup_restore_drain_timeout_reads_config_bridged_env(monkeypatch): + """The bound is a config.yaml knob bridged to an internal env var.""" + from gateway.run import ( + _STARTUP_RESTORE_DRAIN_TIMEOUT_SECS_DEFAULT, + _startup_restore_drain_timeout_secs, + ) + + monkeypatch.delenv("HERMES_STARTUP_RESTORE_DRAIN_TIMEOUT", raising=False) + assert ( + _startup_restore_drain_timeout_secs() + == _STARTUP_RESTORE_DRAIN_TIMEOUT_SECS_DEFAULT + ) + + monkeypatch.setenv("HERMES_STARTUP_RESTORE_DRAIN_TIMEOUT", "12.5") + assert _startup_restore_drain_timeout_secs() == 12.5 + + # A malformed value must fall back to the default, never raise. + monkeypatch.setenv("HERMES_STARTUP_RESTORE_DRAIN_TIMEOUT", "not-a-number") + assert ( + _startup_restore_drain_timeout_secs() + == _STARTUP_RESTORE_DRAIN_TIMEOUT_SECS_DEFAULT + ) + + +def test_startup_restore_drain_timeout_is_a_documented_config_key(): + """agent.gateway_startup_restore_drain_timeout ships in DEFAULT_CONFIG.""" + from hermes_cli.config import DEFAULT_CONFIG + + assert ( + "gateway_startup_restore_drain_timeout" in DEFAULT_CONFIG["agent"] + ), "the bound must be a config.yaml knob, not an undocumented env var"