fix(omnio): harden conversation browser sessions

gateway/kanban_watchers.py

@@ -493,8 +493,8 @@ def _collect():
                         )
                         if sub.get("thread_id") and not metadata.get("thread_id"):
                             metadata["thread_id"] = sub["thread_id"]
-                        # Adapters with no push channel (the API server —
-                        # ``supports_async_delivery = False``) can NEVER
+                        # Adapters with no push channel (the API server sets
+                        # ``supports_push_delivery = False``) can NEVER
                         # satisfy a text-send: ``send()`` always reports
                         # SendResult(success=False) by design (see
                         # ApiServerAdapter.send()). Treating that as a

gateway/platforms/api_server.py

@@ -1246,15 +1246,11 @@ class APIServerAdapter(BasePlatformAdapter):
     and routes them through hermes-agent's AIAgent.
     """
 
-    # Stateless request/response: every route (the OpenAI-spec
-    # /v1/chat/completions and /v1/responses, and the proprietary /v1/runs SSE
-    # stream) tears down its channel when the turn ends. There is no persistent
-    # outbound channel to push a background completion to a client that already
-    # received its response, and ``send()`` is a no-op stub. So async-delivery
-    # tools (terminal notify_on_complete / watch_patterns, delegate_task
-    # background=True) must NOT promise delivery on this path — see
-    # ``async_delivery_supported()``.
-    supports_async_delivery: bool = False
+    # The HTTP response channel is stateless, but the gateway can wake the raw
+    # session by self-POSTing through /v1/chat/completions. Async jobs are
+    # therefore supported even though direct push via handle_message is not.
+    supports_async_delivery: bool = True
+    supports_push_delivery: bool = False
 
     # Same statelessness applies to the startup auto-resume prompt: no client
     # is waiting to answer "session restored — what next?", so a resumed turn
@@ -6270,12 +6266,9 @@ def _bind_api_server_session(
         """Bind session contextvars for an API-server agent run.
 
         This is the SINGLE structural chokepoint every API-server agent-entry
-        path must use to seed session context — it hardwires
-        ``platform="api_server"`` and ``async_delivery=False`` so a new route
-        physically cannot reintroduce the silent-no-op bug (#10760) by
-        forgetting to mark the channel as non-delivering. There is no
-        ``async_delivery`` parameter to get wrong; the stateless HTTP path can
-        never wake the agent after the turn ends, on ANY route.
+        path must use to seed session context. API sessions support asynchronous
+        completion via the gateway's authenticated self-post wake path, while
+        remaining non-push adapters.
 
         Returns reset tokens; pass them to ``clear_session_vars`` in a
         ``finally`` block (the binding is request-scoped and must not outlive
@@ -6289,7 +6282,7 @@ def _bind_api_server_session(
             chat_id=chat_id,
             session_key=session_key,
             session_id=session_id,
-            async_delivery=False,
+            async_delivery=True,
         )
 
     async def _run_agent(

gateway/run.py

@@ -17660,7 +17660,8 @@ def _set_session_env(self, context: SessionContext) -> list:
         # (terminal notify_on_complete / watch_patterns, delegate_task
         # background=True) know whether this channel can wake a later turn.
         # Default True keeps CLI / unknown paths working; stateless adapters
-        # (api_server) declare supports_async_delivery=False. Use getattr so
+        # that cannot arrange any later wake declare supports_async_delivery=False.
+        # The API server supports self-post wakes while remaining non-push. Use getattr so
         # bare runners built via object.__new__ (tests) without self.adapters
         # don't blow up — they simply default to supported.
         _adapters = getattr(self, "adapters", None) or {}

gateway/session_context.py

@@ -100,16 +100,18 @@ def session_context_engaged() -> bool:
 # True  — long-lived CLI sessions (in-process completion_queue drain) and the
 #         real gateway platforms (Telegram/Discord/Slack/...), which hold a
 #         persistent outbound channel and run the watcher/drain loops.
-# False — finite runtimes that can end before a detached completion returns:
-#         stateless API-server requests and dispatcher-spawned Kanban workers.
+# False — finite runtimes that cannot arrange any later wake, such as
+#         dispatcher-spawned one-shot Kanban workers. The API server is
+#         stateless but supports wake delivery through an authenticated
+#         self-post to the raw session, so it binds True.
 #
 # Tools that promise async delivery (terminal notify_on_complete /
 # watch_patterns, delegate_task background=True) read this via
 # ``async_delivery_supported()`` and refuse to hand out a promise the channel
 # can't keep — turning a silent no-op into an explicit contract.
 #
 # Default _UNSET => treated as supported, so CLI (which never sets a platform)
-# and any contextvar-unaware path keep working. Stateless adapters opt OUT by
+# and any contextvar-unaware path keep working. Non-delivering adapters opt OUT by
 # setting ``supports_async_delivery = False`` on the adapter class; the gateway
 # propagates that into this contextvar at session-bind time.
 _SESSION_ASYNC_DELIVERY: ContextVar = ContextVar("HERMES_SESSION_ASYNC_DELIVERY", default=_UNSET)

gateway/wake.py

@@ -1,14 +1,14 @@
 """Wake an existing agent session from a background completion event.
 
 Two delivery strategies, selected by the target adapter's
-``supports_async_delivery`` capability flag:
+``supports_push_delivery`` capability flag:
 
 * Push-capable adapters (telegram, discord, plugin platforms, ...): inject a
   synthetic ``MessageEvent(internal=True)`` through ``adapter.handle_message``
   — the pre-existing wake path, preserved exactly.
 
 * Stateless request/response adapters (the API server,
-  ``supports_async_delivery = False``): ``handle_message`` would run the wake
+  ``supports_push_delivery = False``): ``handle_message`` would run the wake
   turn under a ``build_session_key()``-derived key
   (``agent:main:api_server:group:<sid>``) that NEVER matches the raw
   ``X-Hermes-Session-Id`` key real gateway/HQ turns run under
@@ -45,11 +45,12 @@
 def adapter_supports_push(adapter: Any) -> bool:
     """Whether this adapter can push a message to the user after a turn ends.
 
-    Mirrors ``gateway.session_context.async_delivery_supported`` but reads the
-    capability off the adapter class (``supports_async_delivery``) instead of
-    the request-scoped contextvar — background watchers run outside any bound
-    session context. Adapters that don't declare the flag are push-capable.
+    ``supports_async_delivery`` answers whether a later wake is possible;
+    ``supports_push_delivery`` answers how it is delivered. Older adapters
+    that only declare the former retain their previous behavior.
     """
+    if hasattr(adapter, "supports_push_delivery"):
+        return bool(getattr(adapter, "supports_push_delivery"))
     return bool(getattr(adapter, "supports_async_delivery", True))
 
 
@@ -88,7 +89,7 @@ async def deliver_wake(
 
     if not session_id:
         raise ValueError(
-            "deliver_wake: non-push adapter (supports_async_delivery=False) "
+            "deliver_wake: non-push adapter "
             "requires the raw session id to self-post the wake turn"
         )
     await _self_post_chat_completion(adapter, text=text, session_id=session_id)

tests/gateway/test_async_delivery_capability.py

@@ -1,15 +1,13 @@
 """Tests for the async-delivery capability gate (issue #10760).
 
-Stateless request/response adapters (the API server / WebUI path) cannot route
-a background completion back to the agent after a turn ends — there is no
-persistent channel and ``APIServerAdapter.send()`` is a no-op stub. So tools
-that promise async delivery (``terminal`` notify_on_complete / watch_patterns,
-``delegate_task`` background=True) must refuse the promise on that path instead
-of silently registering a watcher that never fires.
+The API server has no persistent push channel, but it can route a background
+completion back to the raw session through its authenticated self-post wake
+path. Async capability and push capability are therefore separate contracts.
 
 This is wired through:
   - ``BasePlatformAdapter.supports_async_delivery`` (default True)
-  - ``APIServerAdapter.supports_async_delivery = False``
+  - ``APIServerAdapter.supports_async_delivery = True``
+  - ``APIServerAdapter.supports_push_delivery = False``
   - ``gateway.session_context._SESSION_ASYNC_DELIVERY`` contextvar +
     ``async_delivery_supported()`` helper, bound per-session.
 
@@ -208,37 +206,36 @@ def test_base_default_true(self):
 
         assert BasePlatformAdapter.supports_async_delivery is True
 
-    def test_api_server_false(self):
+    def test_api_server_supports_async_self_post_but_not_push(self):
         from gateway.platforms.api_server import APIServerAdapter
 
-        assert APIServerAdapter.supports_async_delivery is False
+        assert APIServerAdapter.supports_async_delivery is True
+        assert APIServerAdapter.supports_push_delivery is False
 
-    def test_api_server_bind_chokepoint_hardwires_no_delivery(self):
+    def test_api_server_bind_chokepoint_enables_self_post_delivery(self):
         """Every API-server agent-entry path binds through
-        _bind_api_server_session, which hardwires async_delivery=False — a new
-        route physically cannot reintroduce the silent no-op (#10760)."""
+        _bind_api_server_session, which hardwires async_delivery=True so a new
+        route cannot accidentally disable the self-post wake contract."""
         from gateway.platforms.api_server import APIServerAdapter
         from gateway.session_context import clear_session_vars, get_session_env
 
         tokens = APIServerAdapter._bind_api_server_session(
             chat_id="c1", session_key="sk1", session_id="sid1"
         )
         try:
-            assert async_delivery_supported() is False
+            assert async_delivery_supported() is True
             assert get_session_env("HERMES_SESSION_PLATFORM") == "api_server"
         finally:
             clear_session_vars(tokens)
 
     def test_api_server_binding_does_not_outlive_turn(self):
-        """The no-delivery decision is request-scoped, NOT stuck to the session.
-        After clear, a session resumed on a delivering interface re-binds fresh
-        and is NOT blocked."""
+        """The delivery decision remains request-scoped and clears cleanly."""
         from gateway.platforms.api_server import APIServerAdapter
         from gateway.session_context import clear_session_vars
 
-        # Turn 1: same session over the API server -> blocked.
+        # Turn 1: same session over the API server -> self-post capable.
         tokens = APIServerAdapter._bind_api_server_session(session_key="shared-key")
-        assert async_delivery_supported() is False
+        assert async_delivery_supported() is True
         clear_session_vars(tokens)
 
         # Turn 2: SAME session_key resumed on a delivering interface (CLI/gateway)
@@ -311,6 +308,26 @@ def test_gateway_registers_watcher(self):
         assert len(process_registry.pending_watchers) == 1
         assert process_registry.pending_watchers[0]["platform"] == "telegram"
 
+    def test_api_server_registers_self_post_watcher(self):
+        from tools.process_registry import process_registry
+
+        tokens = set_session_vars(
+            platform="api_server",
+            chat_id="raw-session",
+            session_key="raw-session",
+            session_id="raw-session",
+            async_delivery=True,
+        )
+        try:
+            d = self._run_bg("sleep 30 && echo DONE")
+        finally:
+            clear_session_vars(tokens)
+
+        assert d.get("notify_on_complete") is True
+        assert not d.get("notify_unsupported")
+        assert len(process_registry.pending_watchers) == 1
+        assert process_registry.pending_watchers[0]["platform"] == "api_server"
+
     def test_cli_stays_supported(self):
         """CLI delivers via the in-process completion_queue: notify stays on,
         no false 'unsupported' note, and no pending_watcher (empty platform)."""

tests/gateway/test_wake_delivery.py

@@ -2,7 +2,7 @@
 
 Two strategies:
 * push-capable adapters keep the synthetic MessageEvent / handle_message path;
-* the stateless API server (supports_async_delivery=False) self-POSTs
+* the stateless API server (supports_push_delivery=False) self-POSTs
   /v1/chat/completions with the RAW session id in X-Hermes-Session-Id, so the
   wake turn resumes the REAL session instead of a parallel invisible one
   keyed by build_session_key().
@@ -53,6 +53,14 @@ def test_adapter_supports_push_default_true():
     assert adapter_supports_push(ApiServerLikeAdapter()) is False
 
 
+def test_explicit_push_capability_is_independent_from_async_capability():
+    adapter = ApiServerLikeAdapter()
+    adapter.supports_async_delivery = True
+    adapter.supports_push_delivery = False
+
+    assert adapter_supports_push(adapter) is False
+
+
 def test_deliver_wake_push_adapter_uses_handle_message():
     adapter = PushAdapter()
     asyncio.run(deliver_wake(adapter, text="wake up", source=_source()))

tests/tools/test_browser_cdp_override.py

@@ -125,7 +125,7 @@ def test_normalizes_provider_returned_http_cdp_url_when_creating_session(self, m
         monkeypatch.setattr(browser_tool, "_session_last_activity", {})
         monkeypatch.setattr(browser_tool, "_start_browser_cleanup_thread", lambda: None)
         monkeypatch.setattr(browser_tool, "_update_session_activity", lambda task_id: None)
-        monkeypatch.setattr(browser_tool, "_get_cdp_override", lambda: "")
+        monkeypatch.setattr(browser_tool, "_get_cdp_override", lambda *_: "")
         monkeypatch.setattr(browser_tool, "_get_cloud_provider", lambda: provider)
 
         with patch("tools.browser_tool.requests.get", return_value=response) as mock_get:
@@ -140,6 +140,52 @@ def test_normalizes_provider_returned_http_cdp_url_when_creating_session(self, m
 
 
 class TestGetCdpOverride:
+    def test_template_routes_each_conversation_to_its_own_endpoint(self, monkeypatch):
+        import tools.browser_tool as browser_tool
+
+        monkeypatch.setenv(
+            "BROWSER_CDP_URL_TEMPLATE",
+            "ws://toolbox.test/sessions/{session_id}",
+        )
+        monkeypatch.delenv("BROWSER_CDP_URL", raising=False)
+        monkeypatch.setattr(
+            "gateway.session_context.get_session_env",
+            lambda _name: "",
+        )
+
+        assert browser_tool._get_cdp_override_raw("conversation-a") == (
+            "ws://toolbox.test/sessions/conversation-a"
+        )
+        assert browser_tool._get_cdp_override_raw("conversation-b") == (
+            "ws://toolbox.test/sessions/conversation-b"
+        )
+
+    def test_template_url_encodes_session_id(self, monkeypatch):
+        import tools.browser_tool as browser_tool
+
+        monkeypatch.setenv(
+            "BROWSER_CDP_URL_TEMPLATE",
+            "ws://toolbox.test/sessions/{session_id}",
+        )
+        monkeypatch.setattr(
+            "gateway.session_context.get_session_env",
+            lambda _name: "",
+        )
+
+        assert browser_tool._get_cdp_override_raw("child/task 1") == (
+            "ws://toolbox.test/sessions/child%2Ftask%201"
+        )
+
+    def test_template_without_session_placeholder_is_rejected(self, monkeypatch):
+        import tools.browser_tool as browser_tool
+
+        monkeypatch.setenv(
+            "BROWSER_CDP_URL_TEMPLATE",
+            "ws://toolbox.test/shared-browser",
+        )
+
+        assert browser_tool._get_cdp_override_raw("conversation-a") == ""
+
     def test_prefers_env_var_over_config(self, monkeypatch):
         import tools.browser_tool as browser_tool
 
@@ -202,6 +248,15 @@ def test_camofox_yields_to_config_cdp_override(self, monkeypatch):
         with patch("hermes_cli.config.read_raw_config", return_value={}):
             assert bc.is_camofox_mode() is False
 
+        # A conversation-scoped gateway also suppresses camofox.
+        monkeypatch.delenv("BROWSER_CDP_URL", raising=False)
+        monkeypatch.setenv(
+            "BROWSER_CDP_URL_TEMPLATE",
+            "ws://toolbox.test/sessions/{session_id}",
+        )
+        with patch("hermes_cli.config.read_raw_config", return_value={}):
+            assert bc.is_camofox_mode() is False
+
 class TestCreateCdpSession:
     """_create_cdp_session() must sanitize the CDP URL before logging.