ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 10 Packages Projects Releases Wiki Activity

fix(gateway): run session-finalize plugin hooks off-loop and bounded

Browse Source 
Session-finalize hooks ran synchronously on the gateway event loop from
three call sites (shutdown drain, session-expiry watcher, /new reset).
A plugin hook doing heavy blocking work froze the whole loop: adapter
heartbeats stopped, the drain machinery could not run, and systemd
eventually SIGKILLed the process mid-export. Observed live on a
multi-day 4.7G session where the nemo_relay observability plugin
serialized a full-session ATIF trace inside on_session_finalize.

Changes:
- gateway/run.py: new GatewayRunner._finalize_session_off_loop()
  dispatches hermes_cli.lifecycle.finalize_session via the gateway
  executor under asyncio.wait_for (10s budget), mirroring
  _cleanup_agent_resources_off_loop (#53175). Shutdown finalize and
  the session-expiry watcher now use it.
- gateway/slash_commands.py: /new reset path uses the same helper.
- plugins/observability/nemo_relay: ATIF export is now bounded
  (HERMES_NEMO_RELAY_ATIF_EXPORT_TIMEOUT_S, default 30s) and skipped
  entirely for sessions whose Relay scope operations already errored
  (their exporter state is unreliable and the export can be
  pathologically slow).
- tests/gateway/test_finalize_session_off_loop.py: regression tests
  proving the loop stays live under a wedged hook and the budget is
  enforced.
This commit is contained in:
Teknium
2026-08-15 13:36:38 -07:00
parent 0c50bdbdea
commit 763b10c320
4 changed files with 312 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gateway/run.py
+74 -11
View File
@@ -10549,15 +10549,19 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
)
except Exception as _e:
logger.debug("Shutdown transcript flush failed: %s", _e)
try:
from hermes_cli.lifecycle import finalize_session
finalize_session(
session_id=getattr(agent, "session_id", None),
platform="gateway",
reason="shutdown",
)
except Exception:
pass
# Off-loop + bounded: finalize_session fans out to plugin
# on_session_finalize hooks that can do arbitrary synchronous
# work (e.g. an observability plugin serializing a full-session
# trace export). Running it inline on the event loop blocked the
# entire shutdown sequence past systemd's TimeoutStopSec on a
# multi-day 4.7G session — heartbeats froze and the process was
# SIGKILLed mid-export. Same class as the memory-provider hang
# below (#53175).
await self._finalize_session_off_loop(
session_id=getattr(agent, "session_id", None),
platform="gateway",
reason="shutdown",
)
# Off-loop + bounded: a wedged memory provider here used to hang
# the whole shutdown so SIGTERM never completed (#53175).
await self._cleanup_agent_resources_off_loop(
@@ -10636,6 +10640,63 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
tasks.add(task)
task.add_done_callback(tasks.discard)
# Bounded budget for one finalize_session() dispatch (plugin
# on_session_finalize hooks + core Relay conversation close). Generous
# enough for a normal trace-export flush, small enough that a wedged
# plugin can never eat the systemd stop window.
_FINALIZE_TIMEOUT_S = 10.0
async def _finalize_session_off_loop(
self,
*,
session_id: Any,
platform: str,
reason: str,
**extra: Any,
) -> None:
"""Run hermes_cli.lifecycle.finalize_session off the event loop, bounded.
finalize_session() invokes plugin ``on_session_finalize`` hooks
synchronously; a hook doing heavy blocking work (observability trace
export, network flush) on the event loop freezes heartbeats, adapters,
and the shutdown drain itself. Off-loop + ``wait_for`` keeps the loop
live; on timeout the worker thread is left to finish (or leak) on its
own and the caller proceeds — mirroring
``_cleanup_agent_resources_off_loop`` (#53175).
"""
def _call() -> None:
from hermes_cli.lifecycle import finalize_session
finalize_session(
session_id=session_id,
platform=platform,
reason=reason,
**extra,
)
try:
await asyncio.wait_for(
self._run_in_executor_with_context(_call),
timeout=self._FINALIZE_TIMEOUT_S,
)
except asyncio.TimeoutError:
logger.warning(
"Session finalize hooks (%s, reason=%s) exceeded %ss; "
"proceeding without blocking the event loop (the worker "
"thread is left to finish on its own).",
session_id,
reason,
self._FINALIZE_TIMEOUT_S,
)
except Exception as finalize_exc:
logger.debug(
"Session finalize hooks (%s, reason=%s) failed: %s",
session_id,
reason,
finalize_exc,
)
async def _cleanup_agent_resources_off_loop(
self, agent: Any, *, context: str = ""
) -> None:
@@ -13204,10 +13265,12 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
for key, entry in _expired_entries:
try:
try:
from hermes_cli.lifecycle import finalize_session
_parts = key.split(":")
_platform = _parts[2] if len(_parts) > 2 else ""
finalize_session(
# Off-loop + bounded: plugin finalize hooks can
# block arbitrarily (see _finalize_session_off_loop)
# and this watcher runs on the gateway event loop.
await self._finalize_session_off_loop(
session_id=entry.session_id,
platform=_platform,
reason="session_expired",
gateway/slash_commands.py
+5 -3
View File
@@ -246,10 +246,12 @@ class GatewaySlashCommandsMixin:
_old_sid = old_entry.session_id if old_entry else None
# Fire plugin on_session_finalize hook (session boundary)
# Fire plugin on_session_finalize hook (session boundary).
# Off-loop + bounded: finalize hooks can block arbitrarily
# (observability trace exports) and this handler runs on the
# gateway event loop (see GatewayRunner._finalize_session_off_loop).
try:
from hermes_cli.lifecycle import finalize_session
finalize_session(
await self._finalize_session_off_loop(
session_id=_old_sid,
platform=source.platform.value if source.platform else "",
reason="new_session",
plugins/observability/nemo_relay/__init__.py
+81 -7
View File
@@ -34,6 +34,10 @@ class _SessionState:
atif_subscriber_name: str = ""
is_embedded_subagent: bool = False
parent_session_id: str = ""
# Flipped when a Relay scope operation for this session raises — an
# errored session's exporter state is unreliable and its export can be
# pathologically slow, so close_session skips the ATIF export for it.
scope_errored: bool = False
@dataclass
@@ -58,6 +62,11 @@ class _Settings:
atif_agent_name: str = "Hermes Agent"
atif_agent_version: str = "unknown"
atif_model_name: str = "unknown"
# Wall-clock budget for one session's ATIF export (serialize + write).
# A multi-day session's trace can take minutes to serialize; when the
# export runs inside a session-finalize hook that budget is somebody
# else's shutdown window. <= 0 disables the bound.
atif_export_timeout_s: float = 30.0
class _ProcessPluginConfiguration:
@@ -398,24 +407,76 @@ class _Runtime:
) -> Any:
if state.relay_session is None:
raise RuntimeError("Hermes core Relay session is unavailable")
return self.host.run_in_session(
state.relay_session,
callback,
*args,
**kwargs,
)
try:
return self.host.run_in_session(
state.relay_session,
callback,
*args,
**kwargs,
)
except Exception:
# A failed scope operation leaves the session's Relay/exporter
# state unreliable; remember it so close_session can skip the
# (potentially very slow) ATIF export for this session.
state.scope_errored = True
raise
def export_atif(self, state: _SessionState) -> None:
if not self.settings.atif_enabled or state.atif_exporter is None:
return
if state.is_embedded_subagent and self.settings.atif_subagent_export_mode != "all":
return
if state.scope_errored:
logger.warning(
"Skipping ATIF export for session %s: a prior Relay scope "
"operation failed, exporter state is unreliable",
state.session_id,
)
return
output_dir = self.settings.atif_output_directory
if not output_dir:
return
Path(output_dir).mkdir(parents=True, exist_ok=True)
filename = self.settings.atif_filename_template.format(session_id=state.session_id)
Path(output_dir, filename).write_text(state.atif_exporter.export_json(), encoding="utf-8")
def _export() -> None:
Path(output_dir, filename).write_text(
state.atif_exporter.export_json(), encoding="utf-8"
)
timeout = self.settings.atif_export_timeout_s
if timeout is None or timeout <= 0:
_export()
return
# Bounded: export_json() on a long session can take minutes, and
# close_session runs inside session-finalize/shutdown paths where
# that time is somebody else's stop window. On timeout the worker
# thread finishes (or leaks) on its own; the partial file, if any,
# is overwritten by the next successful export.
error: dict[str, BaseException] = {}
def _runner() -> None:
try:
_export()
except BaseException as exc: # re-raised below when it beat the clock
error["exc"] = exc
thread = threading.Thread(
target=_runner,
name=f"hermes-nemo-relay-atif-export-{state.session_id}",
daemon=True,
)
thread.start()
thread.join(timeout=timeout)
if thread.is_alive():
raise TimeoutError(
f"ATIF export for session {state.session_id} exceeded "
f"{timeout:.0f}s; abandoning the wait (export thread left "
"to finish on its own)"
)
if "exc" in error:
raise error["exc"]
def close_session(
self,
@@ -693,6 +754,9 @@ def _load_settings() -> _Settings:
atif_agent_name=_env("HERMES_NEMO_RELAY_ATIF_AGENT_NAME") or "Hermes Agent",
atif_agent_version=_env("HERMES_NEMO_RELAY_ATIF_AGENT_VERSION") or "unknown",
atif_model_name=_env("HERMES_NEMO_RELAY_ATIF_MODEL_NAME") or "unknown",
atif_export_timeout_s=_env_float(
"HERMES_NEMO_RELAY_ATIF_EXPORT_TIMEOUT_S", 30.0
),
)
@@ -893,6 +957,16 @@ def _env_bool(name: str) -> bool:
return _env(name).lower() in {"1", "true", "yes", "on"}
def _env_float(name: str, default: float) -> float:
raw = _env(name)
if not raw:
return default
try:
return float(raw)
except ValueError:
return default
def _session_id(kwargs: dict[str, Any]) -> str:
return str(kwargs.get("session_id") or kwargs.get("parent_session_id") or "default")
tests/gateway/test_finalize_session_off_loop.py
+152
View File
@@ -0,0 +1,152 @@
"""Session-finalize plugin hooks must not block the gateway event loop.
A plugin ``on_session_finalize`` hook doing heavy synchronous work (e.g. an
observability plugin serializing a multi-day session's trace export) used to
run inline on the event loop from three call sites:
* ``GatewayRunner._finalize_shutdown_agents`` (shutdown drain)
* the session-expiry watcher
* the ``/new`` session-reset handler
On a wedged/slow hook the whole loop froze — adapter heartbeats stopped and
systemd SIGKILLed the process mid-shutdown. All three sites now dispatch
through ``GatewayRunner._finalize_session_off_loop``, which runs
``hermes_cli.lifecycle.finalize_session`` in the gateway executor under a
bounded ``asyncio.wait_for``.
"""
import asyncio
import threading
import time
from gateway.run import GatewayRunner
def _make_runner():
runner = object.__new__(GatewayRunner)
return runner
def test_finalize_off_loop_invokes_lifecycle(monkeypatch):
"""The helper reaches the real lifecycle entry point with the kwargs."""
calls = []
def _fake_finalize(**kwargs):
calls.append(kwargs)
return []
import hermes_cli.lifecycle as lifecycle
monkeypatch.setattr(lifecycle, "finalize_session", _fake_finalize)
runner = _make_runner()
asyncio.run(
runner._finalize_session_off_loop(
session_id="s-123",
platform="gateway",
reason="shutdown",
old_session_id="s-123",
)
)
assert len(calls) == 1
assert calls[0]["session_id"] == "s-123"
assert calls[0]["reason"] == "shutdown"
assert calls[0]["old_session_id"] == "s-123"
def test_finalize_off_loop_keeps_loop_alive_and_bounds_wedged_hook(monkeypatch):
"""A hook that blocks past the budget cannot freeze the event loop.
The loop must keep servicing other callbacks while the hook runs, and
the await must return once the budget expires even though the hook
thread is still blocked.
"""
release = threading.Event()
def _wedged_finalize(**kwargs):
# Simulates a multi-minute trace export.
release.wait(timeout=30)
import hermes_cli.lifecycle as lifecycle
monkeypatch.setattr(lifecycle, "finalize_session", _wedged_finalize)
runner = _make_runner()
monkeypatch.setattr(GatewayRunner, "_FINALIZE_TIMEOUT_S", 0.5, raising=False)
loop_ticks = []
async def _ticker():
while True:
loop_ticks.append(time.monotonic())
await asyncio.sleep(0.05)
async def _scenario():
ticker = asyncio.ensure_future(_ticker())
started = time.monotonic()
try:
await runner._finalize_session_off_loop(
session_id="s-wedged", platform="gateway", reason="shutdown"
)
finally:
elapsed = time.monotonic() - started
ticker.cancel()
return elapsed
elapsed = asyncio.run(_scenario())
release.set()
# Returned promptly at the budget, not after the 30s hook.
assert elapsed < 5.0
# The loop stayed live while the hook was blocked off-loop.
assert len(loop_ticks) >= 3
def test_finalize_off_loop_swallows_hook_exceptions(monkeypatch):
"""A raising hook is contained — callers proceed with shutdown."""
def _raising_finalize(**kwargs):
raise RuntimeError("exporter blew up")
import hermes_cli.lifecycle as lifecycle
monkeypatch.setattr(lifecycle, "finalize_session", _raising_finalize)
runner = _make_runner()
# Must not raise.
asyncio.run(
runner._finalize_session_off_loop(
session_id="s-err", platform="gateway", reason="shutdown"
)
)
def test_shutdown_finalize_path_uses_off_loop_dispatch(monkeypatch):
"""_finalize_shutdown_agents routes finalize through the bounded helper."""
seen = []
async def _fake_off_loop(self, **kwargs):
seen.append(kwargs)
monkeypatch.setattr(
GatewayRunner, "_finalize_session_off_loop", _fake_off_loop
)
async def _fake_cleanup(self, agent, *, context=""):
return None
monkeypatch.setattr(
GatewayRunner, "_cleanup_agent_resources_off_loop", _fake_cleanup
)
class _Agent:
session_id = "s-shutdown"
_session_messages = None
runner = _make_runner()
asyncio.run(runner._finalize_shutdown_agents({"k": _Agent()}))
assert len(seen) == 1
assert seen[0]["session_id"] == "s-shutdown"
assert seen[0]["reason"] == "shutdown"