mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-31 19:16:29 +00:00
fix(gateway): bound the startup-restore inbound gate on a slow boot-resume turn
This commit is contained in:
Kyzcreig
Teknium
parent
0fb46b8185
commit
769dba1758
3 changed files with 303 additions and 5 deletions
111
gateway/run.py
111
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.
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue