From 763b10c320aafdd671934c9f94cd348505ee1402 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Sat, 15 Aug 2026 13:36:38 -0700 Subject: [PATCH] fix(gateway): run session-finalize plugin hooks off-loop and bounded 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. --- gateway/run.py | 85 ++++++++-- gateway/slash_commands.py | 8 +- plugins/observability/nemo_relay/__init__.py | 88 +++++++++- .../gateway/test_finalize_session_off_loop.py | 152 ++++++++++++++++++ 4 files changed, 312 insertions(+), 21 deletions(-) create mode 100644 tests/gateway/test_finalize_session_off_loop.py diff --git a/gateway/run.py b/gateway/run.py index 855f287667..e59af63550 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -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", diff --git a/gateway/slash_commands.py b/gateway/slash_commands.py index 523af310e9..49ba11c572 100644 --- a/gateway/slash_commands.py +++ b/gateway/slash_commands.py @@ -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", diff --git a/plugins/observability/nemo_relay/__init__.py b/plugins/observability/nemo_relay/__init__.py index 3d31f4c6b6..3c7e0698da 100644 --- a/plugins/observability/nemo_relay/__init__.py +++ b/plugins/observability/nemo_relay/__init__.py @@ -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") diff --git a/tests/gateway/test_finalize_session_off_loop.py b/tests/gateway/test_finalize_session_off_loop.py new file mode 100644 index 0000000000..cf10d13617 --- /dev/null +++ b/tests/gateway/test_finalize_session_off_loop.py @@ -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"