fix(gateway): set session contextvars in api_server _run_agent so notify_on_complete works

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

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