fix(gateway): deliver kanban/delegate wake-ups to api_server sessions (#64998 salvage)

contributors/emails/hello@ianks.com

@@ -0,0 +1 @@
+ianks

gateway/platforms/api_server.py

@@ -5101,7 +5101,17 @@ def _run_sync():
                             # environment state.
                             approval_token = set_current_session_key(approval_session_key)
                             session_tokens = self._bind_api_server_session(
+                                # chat_id carries the raw session id (the
+                                # X-Hermes-Session-Id equivalent) exactly like
+                                # the other agent-entry routes bind it via
+                                # _run_agent(). Without it,
+                                # tools.async_delegation reads an empty
+                                # HERMES_SESSION_CHAT_ID on /v1/runs and
+                                # background delegations stay forced-sync
+                                # (no wake target).
+                                chat_id=session_id or "",
                                 session_key=approval_session_key,
+                                session_id=session_id or "",
                             )
                             register_gateway_notify(approval_session_key, _approval_notify)
                             r = agent.run_conversation(

gateway/run.py

@@ -17387,6 +17387,43 @@ async def _inject_watch_notification(
         """
         source = self._build_process_event_source(evt)
         if not source:
+            # API-server-originated sessions bind a RAW session key (the
+            # X-Hermes-Session-Id value — see _bind_api_server_session), not a
+            # structured ``agent:main:...`` key, so _build_process_event_source
+            # cannot derive routing metadata from it and returns None above.
+            # Recover the raw session id and wake the real session via the API
+            # server's own /v1/chat/completions entry point instead of
+            # dropping the event.
+            raw_sid = str(evt.get("origin_session_id") or "").strip()
+            if not raw_sid:
+                _sk = str(evt.get("session_key") or "").strip()
+                if _sk and _parse_session_key(_sk) is None:
+                    raw_sid = _sk
+            if raw_sid:
+                adapter = self.adapters.get(Platform.API_SERVER)
+                from gateway.wake import adapter_supports_push, deliver_wake
+                if adapter is not None and not adapter_supports_push(adapter):
+                    try:
+                        logger.info(
+                            "Watch pattern notification — waking api_server "
+                            "session %s via self-post",
+                            raw_sid,
+                        )
+                        await deliver_wake(adapter, text=synth_text, session_id=raw_sid)
+                        return True
+                    except Exception as e:
+                        logger.warning(
+                            "Watch notification self-post wake failed for "
+                            "session %s: %s",
+                            raw_sid, e,
+                        )
+                        return False
+                logger.warning(
+                    "Dropping watch notification for raw session %s: no "
+                    "api_server adapter to self-post through",
+                    raw_sid,
+                )
+                return None
             logger.warning(
                 "Dropping watch notification with no routing metadata for process %s",
                 evt.get("session_id", "unknown"),
@@ -17400,6 +17437,30 @@ async def _inject_watch_notification(
                 break
         if not adapter:
             return None
+        from gateway.wake import adapter_supports_push as _wake_push_ok
+        if not _wake_push_ok(adapter):
+            # Non-push adapter (api_server) resolved WITH routing metadata:
+            # its chat_id is the raw session id (see _bind_api_server_session,
+            # which binds chat_id = session_id). handle_message would run the
+            # wake under a build_session_key()-derived key that never matches
+            # the raw X-Hermes-Session-Id session — self-post instead.
+            from gateway.wake import deliver_wake
+            raw_sid = str(evt.get("origin_session_id") or "").strip() or str(source.chat_id or "")
+            try:
+                logger.info(
+                    "Watch pattern notification — waking api_server session "
+                    "%s via self-post",
+                    raw_sid,
+                )
+                await deliver_wake(adapter, text=synth_text, session_id=raw_sid)
+                return True
+            except Exception as e:
+                logger.warning(
+                    "Watch notification self-post wake failed for session "
+                    "%s: %s",
+                    raw_sid, e,
+                )
+                return False
         try:
             metadata = {}
             parent_session_id = str(evt.get("parent_session_id") or "").strip()

gateway/wake.py

@@ -0,0 +1,184 @@
+"""Wake an existing agent session from a background completion event.
+
+Two delivery strategies, selected by the target adapter's
+``supports_async_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
+  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
+  (``_bind_api_server_session``), so the wake lands in a parallel, invisible
+  session. Instead we self-POST ``/v1/chat/completions`` on the in-pod API
+  server with the raw session id in the ``X-Hermes-Session-Id`` header — the
+  exact entry point real turns use — so the wake turn resumes the REAL
+  session, with full history, and its result is visible the next time the
+  client polls/reopens the conversation.
+
+Failures RAISE (after bounded retries on transient errors) so callers can
+rewind cursors / retry instead of silently losing the event.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import Any, Optional
+
+logger = logging.getLogger(__name__)
+
+# A wake self-post runs the entire agent turn synchronously (stream=false);
+# generous ceiling so long tool-using turns aren't killed mid-flight.
+WAKE_TURN_TIMEOUT_SECONDS = 600.0
+
+# Backoff delays between retries on transient failures (429 concurrency cap,
+# connection errors). The API server has no per-session lock — concurrent
+# turns on one session are last-writer-wins — but it DOES enforce a global
+# max_concurrent_runs cap via HTTP 429, which is worth waiting out.
+_RETRY_DELAYS_SECONDS = (2.0, 5.0, 10.0)
+
+
+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.
+    """
+    return bool(getattr(adapter, "supports_async_delivery", True))
+
+
+async def deliver_wake(
+    adapter: Any,
+    *,
+    text: str,
+    session_id: str = "",
+    source: Any = None,
+) -> None:
+    """Deliver a wake turn to the session behind ``adapter``.
+
+    ``session_id`` is the RAW session id (the ``X-Hermes-Session-Id`` value /
+    ``state.db`` key) — required for non-push adapters. ``source`` is the
+    ``SessionSource`` used to build the synthetic event — required for
+    push-capable adapters.
+
+    Raises on failure (bad arguments, exhausted retries, HTTP error) so the
+    caller can rewind/retry instead of treating the wake as delivered.
+    """
+    if adapter_supports_push(adapter):
+        if source is None:
+            raise ValueError(
+                "deliver_wake: push-capable adapter requires a SessionSource"
+            )
+        from gateway.platforms.base import MessageEvent, MessageType
+
+        synth_event = MessageEvent(
+            text=text,
+            message_type=MessageType.TEXT,
+            source=source,
+            internal=True,
+        )
+        await adapter.handle_message(synth_event)
+        return
+
+    if not session_id:
+        raise ValueError(
+            "deliver_wake: non-push adapter (supports_async_delivery=False) "
+            "requires the raw session id to self-post the wake turn"
+        )
+    await _self_post_chat_completion(adapter, text=text, session_id=session_id)
+
+
+async def _self_post_chat_completion(
+    adapter: Any, *, text: str, session_id: str
+) -> None:
+    """POST the wake text to the in-pod API server as a normal session turn.
+
+    Uses the adapter's own bind host/port/key (``ApiServerAdapter.__init__``).
+    Session continuation via ``X-Hermes-Session-Id`` is 403-gated on
+    ``API_SERVER_KEY`` being configured, so a missing key is a hard error —
+    raise loudly rather than run the wake in a fresh fingerprint-derived
+    session nobody is looking at.
+    """
+    import aiohttp
+
+    host = str(getattr(adapter, "_host", "") or "127.0.0.1")
+    if host in ("0.0.0.0", "::", "*"):
+        # Wildcard bind address — connect over loopback.
+        host = "127.0.0.1"
+    port = int(getattr(adapter, "_port", 0) or 8642)
+    api_key = str(getattr(adapter, "_api_key", "") or "")
+    if not api_key:
+        raise RuntimeError(
+            "wake self-post requires API_SERVER_KEY: session continuation via "
+            "X-Hermes-Session-Id is rejected (403) on an unauthenticated API "
+            "server, so the wake cannot reach the target session"
+        )
+
+    if ":" in host and not host.startswith("["):
+        host = f"[{host}]"  # bare IPv6 literal
+    url = f"http://{host}:{port}/v1/chat/completions"
+    headers = {
+        "Authorization": f"Bearer {api_key}",
+        "X-Hermes-Session-Id": session_id,
+    }
+    payload = {
+        "model": str(getattr(adapter, "_model_name", "") or "hermes-agent"),
+        "messages": [{"role": "user", "content": text}],
+        "stream": False,
+    }
+
+    last_err: Optional[BaseException] = None
+    attempts = 1 + len(_RETRY_DELAYS_SECONDS)
+    for attempt in range(attempts):
+        if attempt:
+            await asyncio.sleep(_RETRY_DELAYS_SECONDS[attempt - 1])
+        try:
+            timeout = aiohttp.ClientTimeout(total=WAKE_TURN_TIMEOUT_SECONDS)
+            async with aiohttp.ClientSession(timeout=timeout) as http:
+                async with http.post(url, json=payload, headers=headers) as resp:
+                    if resp.status == 429:
+                        # Global concurrency cap (max_concurrent_runs) —
+                        # transient; back off and retry.
+                        last_err = RuntimeError(
+                            f"wake self-post got HTTP 429 (concurrency cap) "
+                            f"for session {session_id}"
+                        )
+                        logger.warning(
+                            "%s; attempt %d/%d", last_err, attempt + 1, attempts
+                        )
+                        continue
+                    if resp.status >= 400:
+                        body = (await resp.text())[:300]
+                        # Non-transient (auth/validation) — fail immediately.
+                        raise RuntimeError(
+                            f"wake self-post failed for session {session_id}: "
+                            f"HTTP {resp.status}: {body}"
+                        )
+                    await resp.read()
+                    logger.info(
+                        "wake self-post delivered for session %s (attempt %d)",
+                        session_id,
+                        attempt + 1,
+                    )
+                    return
+        except (aiohttp.ClientError, asyncio.TimeoutError, OSError) as exc:
+            last_err = exc
+            logger.warning(
+                "wake self-post transient failure for session %s "
+                "(attempt %d/%d): %s",
+                session_id,
+                attempt + 1,
+                attempts,
+                exc,
+            )
+            continue
+    raise RuntimeError(
+        f"wake self-post gave up for session {session_id} after "
+        f"{attempts} attempts: {last_err}"
+    ) from last_err

tests/gateway/test_api_server_runs.py

@@ -145,6 +145,51 @@ async def test_start_returns_202(self, adapter):
                 assert status["status"] in {"queued", "running", "completed"}
                 assert status["object"] == "hermes.run"
 
+    @pytest.mark.asyncio
+    async def test_start_binds_chat_id_for_delegation_wake_target(self, adapter):
+        """/v1/runs must bind the raw session id as the api_server chat_id
+        (like every other agent-entry route does via _run_agent): the async
+        delegation dispatch reads HERMES_SESSION_CHAT_ID to pick its wake
+        self-post target, and an empty binding forces background delegations
+        on this route back to synchronous execution."""
+        app = _create_runs_app(adapter)
+        captured = {}
+
+        async with TestClient(TestServer(app)) as cli:
+            with patch.object(adapter, "_create_agent") as mock_create:
+                mock_agent = MagicMock()
+
+                def _capture_run(user_message=None, conversation_history=None, task_id=None):
+                    from tools.async_delegation import _current_origin_session_id
+
+                    captured["origin_session_id"] = _current_origin_session_id()
+                    return {"final_response": "done"}
+
+                mock_agent.run_conversation.side_effect = _capture_run
+                mock_agent.session_prompt_tokens = 0
+                mock_agent.session_completion_tokens = 0
+                mock_agent.session_total_tokens = 0
+                mock_create.return_value = mock_agent
+
+                resp = await cli.post(
+                    "/v1/runs",
+                    json={"input": "hello", "session_id": "runs-raw-sid"},
+                )
+                assert resp.status == 202
+                data = await resp.json()
+                run_id = data["run_id"]
+
+                for _ in range(40):
+                    status_resp = await cli.get(f"/v1/runs/{run_id}")
+                    status = await status_resp.json()
+                    if status["status"] == "completed":
+                        break
+                    await asyncio.sleep(0.05)
+
+        assert captured.get("origin_session_id") == "runs-raw-sid", (
+            "runs route must bind chat_id so delegation dispatch sees a wake target"
+        )
+
     @pytest.mark.asyncio
     async def test_start_invalid_json_returns_400(self, adapter):
         app = _create_runs_app(adapter)