From 0c016b4355235fba0370210dc2fdf8bb7ebed1c8 Mon Sep 17 00:00:00 2001 From: Bartok9 Date: Sun, 3 May 2026 18:29:47 -0400 Subject: [PATCH] fix(gateway): set session contextvars in api_server _run_agent so notify_on_complete works MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When the agent runs through the API server path (WebUI, Dashboard, any OpenAI-compatible client), `terminal(background=True, notify_on_complete=True)` was silently a no-op. The terminal tool's watcher-registration block in `tools/terminal_tool.py` is gated on `HERMES_SESSION_PLATFORM` being non-empty (read via `gateway.session_context.get_session_env`), but `APIServerAdapter._run_agent` never seeded those contextvars before scheduling the agent into the executor — and even if it had, plain `loop.run_in_executor` doesn't propagate contextvars into the worker thread. Result: the watcher gate evaluated False, the entire registration block was skipped, no error, no warning. Works fine on Telegram/Discord/Slack because `GatewayRunner` calls `_set_session_env` at `gateway/run.py:5722` before each turn. This change wraps `_run_agent` with `set_session_vars` / `clear_session_vars` and threads the captured `copy_context()` into `loop.run_in_executor` so the contextvars are visible to `terminal_tool` running in the executor thread. Mirrors the `GatewayRunner._set_session_env` pattern. Watcher *registration* on the API server path now succeeds; watcher *delivery* (routing the completion notification back to an HTTP client without a persistent message channel) is a separate, larger problem and is intentionally out of scope — maintainers' call on whether to wire that through dependency injection from GatewayRunner or via a polling/SSE endpoint. Two regression tests added in `TestAgentExecution`: - `test_run_agent_sets_session_contextvars_for_executor` reads `HERMES_SESSION_PLATFORM`, `HERMES_SESSION_CHAT_ID`, and `HERMES_SESSION_KEY` from inside the agent's `run_conversation` callback (which executes in the executor thread, where `terminal_tool`'s lookup actually happens) and asserts they're populated. - `test_run_agent_clears_session_contextvars_after_run` asserts the vars are cleared after the run so a subsequent unrelated request on the same asyncio task can't observe stale routing data. Targeted run: `pytest tests/gateway/test_api_server.py -o addopts=''` → 126 passed, no regressions. Existing `test_run_agent_uses_session_id_as_task_id` still passes unchanged. Fixes #10760 Co-authored-by: Cursor --- gateway/platforms/api_server.py | 74 ++++++++++++++++++++---------- tests/gateway/test_api_server.py | 79 ++++++++++++++++++++++++++++++++ 2 files changed, 128 insertions(+), 25 deletions(-) diff --git a/gateway/platforms/api_server.py b/gateway/platforms/api_server.py index dc608874594fb..35021671f8c11 100644 --- a/gateway/platforms/api_server.py +++ b/gateway/platforms/api_server.py @@ -34,6 +34,7 @@ import sqlite3 import time import uuid +from contextvars import copy_context from typing import Any, Dict, List, Optional try: @@ -2337,34 +2338,57 @@ async def _run_agent( at ``agent_ref[0]`` before ``run_conversation`` begins. This allows callers (e.g. the SSE writer) to call ``agent.interrupt()`` from another thread to stop in-progress LLM calls. + + Session context vars (``HERMES_SESSION_PLATFORM`` etc.) are set for + the duration of the run and propagated into the executor thread via + ``copy_context``. Without this, ``terminal_tool``'s watcher + registration block (which gates on ``HERMES_SESSION_PLATFORM`` being + non-empty) is silently skipped on the API server path, making + ``notify_on_complete=True`` a no-op for WebUI / Dashboard / any + OpenAI-compatible client. Mirrors ``GatewayRunner._set_session_env`` + in ``gateway/run.py``. Issue #10760. """ - loop = asyncio.get_running_loop() + from gateway.session_context import set_session_vars, clear_session_vars + + # API server is stateless w.r.t. chat/user identity (no platform-side + # account model), so chat_id and session_key both carry session_id — + # that's the only stable per-conversation handle we have here. + _session_tokens = set_session_vars( + platform=Platform.API_SERVER.value, + chat_id=session_id or "", + session_key=session_id or "", + ) + try: + loop = asyncio.get_running_loop() + ctx = copy_context() - def _run(): - agent = self._create_agent( - ephemeral_system_prompt=ephemeral_system_prompt, - session_id=session_id, - stream_delta_callback=stream_delta_callback, - tool_progress_callback=tool_progress_callback, - tool_start_callback=tool_start_callback, - tool_complete_callback=tool_complete_callback, - ) - if agent_ref is not None: - agent_ref[0] = agent - effective_task_id = session_id or str(uuid.uuid4()) - result = agent.run_conversation( - user_message=user_message, - conversation_history=conversation_history, - task_id=effective_task_id, - ) - usage = { - "input_tokens": getattr(agent, "session_prompt_tokens", 0) or 0, - "output_tokens": getattr(agent, "session_completion_tokens", 0) or 0, - "total_tokens": getattr(agent, "session_total_tokens", 0) or 0, - } - return result, usage + def _run(): + agent = self._create_agent( + ephemeral_system_prompt=ephemeral_system_prompt, + session_id=session_id, + stream_delta_callback=stream_delta_callback, + tool_progress_callback=tool_progress_callback, + tool_start_callback=tool_start_callback, + tool_complete_callback=tool_complete_callback, + ) + if agent_ref is not None: + agent_ref[0] = agent + effective_task_id = session_id or str(uuid.uuid4()) + result = agent.run_conversation( + user_message=user_message, + conversation_history=conversation_history, + task_id=effective_task_id, + ) + usage = { + "input_tokens": getattr(agent, "session_prompt_tokens", 0) or 0, + "output_tokens": getattr(agent, "session_completion_tokens", 0) or 0, + "total_tokens": getattr(agent, "session_total_tokens", 0) or 0, + } + return result, usage - return await loop.run_in_executor(None, _run) + return await loop.run_in_executor(None, ctx.run, _run) + finally: + clear_session_vars(_session_tokens) # ------------------------------------------------------------------ # /v1/runs — structured event streaming diff --git a/tests/gateway/test_api_server.py b/tests/gateway/test_api_server.py index 74a30541dc701..02efc630b5a14 100644 --- a/tests/gateway/test_api_server.py +++ b/tests/gateway/test_api_server.py @@ -361,6 +361,85 @@ async def test_run_agent_uses_session_id_as_task_id(self, adapter): task_id="session-123", ) + @pytest.mark.asyncio + async def test_run_agent_sets_session_contextvars_for_executor(self, adapter): + """Regression for #10760. + + ``terminal_tool``'s ``notify_on_complete`` watcher registration is gated + on ``HERMES_SESSION_PLATFORM`` being non-empty (read via + ``gateway.session_context.get_session_env``). On the API server path + the agent runs inside ``loop.run_in_executor``; if the adapter doesn't + seed the session contextvars *and* propagate them via ``copy_context``, + the watcher block is silently skipped and ``notify_on_complete`` becomes + a no-op. + + This test asserts the contextvars are visible inside the executor + thread where the agent actually runs (i.e. where ``terminal_tool`` would + read them). + """ + from gateway.session_context import get_session_env + + observed: Dict[str, str] = {} + + def _capture_env_inside_run_conversation(*args, **kwargs): + # This runs in the executor thread, mirroring where terminal_tool's + # ``_gw_platform = _gse("HERMES_SESSION_PLATFORM", "")`` lookup + # actually happens during background-process registration. + observed["platform"] = get_session_env("HERMES_SESSION_PLATFORM", "") + observed["chat_id"] = get_session_env("HERMES_SESSION_CHAT_ID", "") + observed["session_key"] = get_session_env("HERMES_SESSION_KEY", "") + return {"final_response": "ok"} + + mock_agent = MagicMock() + mock_agent.run_conversation.side_effect = _capture_env_inside_run_conversation + mock_agent.session_prompt_tokens = 0 + mock_agent.session_completion_tokens = 0 + mock_agent.session_total_tokens = 0 + + with patch.object(adapter, "_create_agent", return_value=mock_agent): + await adapter._run_agent( + user_message="hello", + conversation_history=[], + session_id="session-watcher-42", + ) + + assert observed["platform"] == "api_server", ( + "terminal_tool's notify_on_complete watcher block is gated on " + "HERMES_SESSION_PLATFORM; it must be set to 'api_server' inside " + "the executor thread or the watcher is silently skipped." + ) + assert observed["chat_id"] == "session-watcher-42" + assert observed["session_key"] == "session-watcher-42" + + @pytest.mark.asyncio + async def test_run_agent_clears_session_contextvars_after_run(self, adapter): + """Regression for #10760. + + Session contextvars must not leak past ``_run_agent`` — otherwise a + subsequent unrelated agent run on the same asyncio task could observe + stale routing data from a previous request. + """ + from gateway.session_context import get_session_env + + mock_agent = MagicMock() + mock_agent.run_conversation.return_value = {"final_response": "ok"} + mock_agent.session_prompt_tokens = 0 + mock_agent.session_completion_tokens = 0 + mock_agent.session_total_tokens = 0 + + with patch.object(adapter, "_create_agent", return_value=mock_agent): + await adapter._run_agent( + user_message="hello", + conversation_history=[], + session_id="session-leak-check", + ) + + # After the run, the contextvars should be cleared (set to "") so a + # subsequent unguarded read returns "" rather than the previous run's + # session_id. This matches GatewayRunner._clear_session_env semantics. + assert get_session_env("HERMES_SESSION_CHAT_ID", "default-fallback") == "" + assert get_session_env("HERMES_SESSION_KEY", "default-fallback") == "" + # --------------------------------------------------------------------------- # /health endpoint