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

Wake-ups for kanban notifications and background delegation completions were
injected via handle_message() using a build_session_key()-derived key, which
can never match the raw X-Hermes-Session-Id key that api_server sessions run
under — so the wake landed in a session nobody was reading. On top of that,
ApiServerAdapter.send() reports failure without raising, and that was treated
as a successful delivery, so the notify cursor advanced past events that were
permanently lost; and background delegation was forced synchronous on
api_server since there was no way to wake the session afterward.

Fix: route wake-ups for non-push adapters through a self-post to
/v1/chat/completions with the original session id, treat non-raising send
failures as failures (rewind instead of advancing the cursor), and re-enable
background delegation whenever a session id is available to wake.

The origin session id is captured from the request-scoped api_server chat_id
binding rather than HERMES_SESSION_ID: constructing a child agent calls
set_current_session_id() with the subagent's internal id, clobbering that
variable right before dispatch would read it and misrouting the wake into
the subagent's own session.

Related: #56580, #64609, #53027, #63169, #56531, #50319, #64113

gateway/kanban_watchers.py

@@ -412,10 +412,46 @@ class GatewayKanbanWatchersMixin:
                             sub["task_id"], sub["platform"],
                             sub["chat_id"], sub.get("thread_id") or "",
                         )
+                        # Adapters with no push channel (the API server —
+                        # ``supports_async_delivery = False``) can NEVER
+                        # satisfy a text-send: ``send()`` always reports
+                        # SendResult(success=False) by design (see
+                        # ApiServerAdapter.send()). Treating that as a
+                        # delivery failure would rewind/drop the subscription
+                        # forever and — because the wake dispatch below lives
+                        # in this loop's ``else`` clause — would also make the
+                        # wake-on-completion path (the actual fix for the
+                        # api_server wrong-session bug) unreachable. So for
+                        # non-push adapters, skip the doomed send attempt
+                        # entirely: there is nothing to text-notify, the
+                        # creator is woken via the self-post below instead.
+                        from gateway.wake import adapter_supports_push
+
+                        if not adapter_supports_push(adapter):
+                            logger.debug(
+                                "kanban notifier: adapter %s has no push "
+                                "channel; skipping text ping for %s, relying "
+                                "on wake self-post instead",
+                                platform_str, sub["task_id"],
+                            )
+                            sub_fail_counts.pop(sub_key, None)
+                            continue
                         try:
-                            await adapter.send(
+                            _send_res = await adapter.send(
                                 sub["chat_id"], msg, metadata=metadata,
                             )
+                            # A SendResult(success=False) without an exception
+                            # (returned by push-capable adapters on a genuine
+                            # transient failure) must count as a FAILED
+                            # delivery — otherwise the cursor advances and the
+                            # event is permanently lost. Adapters returning
+                            # None (or anything non-SendResult shaped) keep
+                            # the legacy "no exception == delivered" contract.
+                            if getattr(_send_res, "success", True) is False:
+                                raise RuntimeError(
+                                    "adapter send() reported failure: "
+                                    f"{getattr(_send_res, 'error', None) or 'unknown error'}"
+                                )
                             logger.debug(
                                 "kanban notifier: delivered %s event for %s to %s/%s on board %s",
                                 kind, sub["task_id"], platform_str, sub["chat_id"], board_slug,
@@ -513,7 +549,7 @@ class GatewayKanbanWatchersMixin:
                                         board=board_slug,
                                     )
                                     from gateway.session import SessionSource
-                                    from gateway.platforms.base import MessageEvent, MessageType
+                                    from gateway.wake import deliver_wake
                                     # KNOWN LIMITATION (tracked follow-up): the
                                     # subscription row does not persist the
                                     # creator's chat_type, and it is not carried
@@ -539,13 +575,22 @@ class GatewayKanbanWatchersMixin:
                                         user_id=sub.get("user_id"),
                                         profile=sub_profile or None,
                                     )
-                                    _synth_event = MessageEvent(
+                                    # deliver_wake preserves the synthetic
+                                    # MessageEvent/handle_message path for
+                                    # push-capable adapters, and self-POSTs
+                                    # /v1/chat/completions with the task's RAW
+                                    # session id for the stateless API server —
+                                    # handle_message there would run the wake
+                                    # under a build_session_key()-derived key
+                                    # that never matches the raw
+                                    # X-Hermes-Session-Id session real turns
+                                    # run under (wrong-session wake bug).
+                                    await deliver_wake(
+                                        adapter,
                                         text=_synth,
-                                        message_type=MessageType.TEXT,
+                                        session_id=_session_key,
                                         source=_source,
-                                        internal=True,
                                     )
-                                    await adapter.handle_message(_synth_event)
                                     logger.info(
                                         "kanban notifier: woke agent for %s on %s/%s profile=%s events=%s",
                                         sub["task_id"], platform_str, sub["chat_id"], sub_profile or "default", _wake_kinds,

gateway/run.py

@@ -17447,6 +17447,43 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         """
         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"),
@@ -17460,6 +17497,30 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                 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_background_process_notifications.py

@@ -621,3 +621,72 @@ def test_parse_session_key_too_short():
 def test_parse_session_key_wrong_prefix():
     assert _parse_session_key("cron:main:telegram:dm:123") is None
     assert _parse_session_key("agent:cron:telegram:dm:123") is None
+
+
+# ---------------------------------------------------------------------------
+# api_server (stateless) wake routing — gateway/wake.py self-post path
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_inject_watch_notification_raw_session_key_self_posts(monkeypatch, tmp_path):
+    """An event whose session_key is a RAW api_server session id (not an
+    agent:main:... structured key) must wake the real session via the
+    /v1/chat/completions self-post instead of being dropped for missing
+    routing metadata."""
+    runner = _build_runner(monkeypatch, tmp_path, "all")
+    api_adapter = SimpleNamespace(
+        supports_async_delivery=False,
+        handle_message=AsyncMock(),
+        _host="127.0.0.1", _port=8642, _api_key="k", _model_name="m",
+    )
+    runner.adapters[Platform.API_SERVER] = api_adapter
+
+    posts = []
+
+    async def fake_self_post(adapter, *, text, session_id):
+        posts.append({"text": text, "session_id": session_id})
+
+    import gateway.wake as wake_mod
+    monkeypatch.setattr(wake_mod, "_self_post_chat_completion", fake_self_post)
+
+    evt = {
+        "session_id": "proc_watch",
+        "session_key": "raw-hq-session-id",  # no agent:main:... structure
+    }
+    result = await runner._inject_watch_notification("[SYSTEM: subagent finished]", evt)
+
+    assert result is True
+    api_adapter.handle_message.assert_not_awaited()
+    assert posts == [
+        {"text": "[SYSTEM: subagent finished]", "session_id": "raw-hq-session-id"}
+    ]
+
+
+@pytest.mark.asyncio
+async def test_inject_watch_notification_origin_session_id_wins(monkeypatch, tmp_path):
+    """origin_session_id (stamped at dispatch time by async_delegation) takes
+    precedence as the wake target."""
+    runner = _build_runner(monkeypatch, tmp_path, "all")
+    api_adapter = SimpleNamespace(
+        supports_async_delivery=False,
+        handle_message=AsyncMock(),
+        _host="127.0.0.1", _port=8642, _api_key="k", _model_name="m",
+    )
+    runner.adapters[Platform.API_SERVER] = api_adapter
+
+    posts = []
+
+    async def fake_self_post(adapter, *, text, session_id):
+        posts.append(session_id)
+
+    import gateway.wake as wake_mod
+    monkeypatch.setattr(wake_mod, "_self_post_chat_completion", fake_self_post)
+
+    evt = {
+        "session_id": "proc_watch",
+        "session_key": "",
+        "origin_session_id": "raw-origin-sid",
+    }
+    result = await runner._inject_watch_notification("[SYSTEM: done]", evt)
+    assert result is True
+    assert posts == ["raw-origin-sid"]

tests/gateway/test_kanban_notifier_apiserver_wake.py

@@ -0,0 +1,154 @@
+"""Kanban notifier behavior on stateless (api_server) subscriptions.
+
+Covers the wrong-session-wake / silent-loss fixes:
+* a SendResult(success=False) return (the API server's send() stub) rewinds
+  the cursor instead of advancing past a never-delivered event;
+* api_server subscriptions wake the creator's REAL session via the
+  /v1/chat/completions self-post (raw task.session_id), never via
+  handle_message (which would run under a build_session_key()-derived key
+  that never matches the raw X-Hermes-Session-Id session real turns use).
+"""
+
+import asyncio
+
+from gateway.config import Platform
+from gateway.platforms.base import SendResult
+from gateway.run import GatewayRunner
+from hermes_cli import kanban_db as kb
+
+
+class SoftFailAdapter:
+    """Push-capable adapter whose send() returns SendResult(success=False)
+    WITHOUT raising — previously treated as delivered (event lost)."""
+
+    def __init__(self):
+        self.attempts = 0
+
+    async def send(self, chat_id, text, metadata=None):
+        self.attempts += 1
+        return SendResult(success=False, error="soft failure")
+
+
+class ApiServerLikeAdapter:
+    supports_async_delivery = False
+
+    def __init__(self):
+        self._host = "127.0.0.1"
+        self._port = 8642
+        self._api_key = "k"
+        self._model_name = "hermes"
+        self.handle_message_calls = []
+        self.send_calls = 0
+
+    async def send(self, chat_id, text, metadata=None):
+        self.send_calls += 1
+        return SendResult(
+            success=False,
+            error="API server uses HTTP request/response, not send()",
+        )
+
+    async def handle_message(self, event):
+        self.handle_message_calls.append(event)
+
+
+async def _run_one_notifier_tick(monkeypatch, runner):
+    real_sleep = asyncio.sleep
+
+    async def fake_sleep(delay):
+        if delay == 5:
+            return None
+        runner._running = False
+        await real_sleep(0)
+
+    monkeypatch.setattr(asyncio, "sleep", fake_sleep)
+    await runner._kanban_notifier_watcher(interval=1)
+
+
+def _make_runner(adapters):
+    runner = GatewayRunner.__new__(GatewayRunner)
+    runner._running = True
+    runner.adapters = adapters
+    runner._kanban_sub_fail_counts = {}
+    return runner
+
+
+def _create_completed_subscription(platform, chat_id, session_id=None):
+    conn = kb.connect()
+    try:
+        tid = kb.create_task(
+            conn, title="notify once", assignee="worker", session_id=session_id,
+        )
+        kb.add_notify_sub(conn, task_id=tid, platform=platform, chat_id=chat_id)
+        kb.complete_task(conn, tid, summary="done once")
+        return tid
+    finally:
+        conn.close()
+
+
+def _unseen_terminal_events(tid, platform, chat_id):
+    conn = kb.connect()
+    try:
+        _, events = kb.unseen_events_for_sub(
+            conn,
+            task_id=tid,
+            platform=platform,
+            chat_id=chat_id,
+            kinds=["completed", "blocked", "gave_up", "crashed", "timed_out"],
+        )
+        return events
+    finally:
+        conn.close()
+
+
+def test_sendresult_failure_rewinds_cursor(tmp_path, monkeypatch):
+    """SendResult(success=False) without an exception must count as a failed
+    delivery — cursor rewound, event retried on the next tick. Previously the
+    cursor advanced and the event was permanently lost."""
+    monkeypatch.setenv("HERMES_KANBAN_DB", str(tmp_path / "softfail.db"))
+    kb.init_db()
+    tid = _create_completed_subscription("telegram", "chat-1")
+
+    adapter = SoftFailAdapter()
+    runner = _make_runner({Platform.TELEGRAM: adapter})
+    asyncio.run(_run_one_notifier_tick(monkeypatch, runner))
+
+    assert adapter.attempts >= 1
+    assert [ev.kind for ev in _unseen_terminal_events(tid, "telegram", "chat-1")] == [
+        "completed"
+    ]
+
+
+def test_apiserver_sub_wakes_real_session_via_self_post(tmp_path, monkeypatch):
+    """An api_server subscription wakes the creator's REAL session by
+    self-posting with the task's raw session_id — never handle_message (which
+    would run the wake under a build_session_key()-derived key that can't
+    match the raw X-Hermes-Session-Id session)."""
+    monkeypatch.setenv("HERMES_KANBAN_DB", str(tmp_path / "apiserver.db"))
+    kb.init_db()
+    tid = _create_completed_subscription(
+        "api_server", "raw-sid-123", session_id="raw-sid-123",
+    )
+
+    posts = []
+
+    async def fake_self_post(adapter, *, text, session_id):
+        posts.append({"text": text, "session_id": session_id})
+
+    import gateway.wake as wake_mod
+
+    monkeypatch.setattr(wake_mod, "_self_post_chat_completion", fake_self_post)
+
+    adapter = ApiServerLikeAdapter()
+    runner = _make_runner({Platform.API_SERVER: adapter})
+    asyncio.run(_run_one_notifier_tick(monkeypatch, runner))
+
+    assert adapter.handle_message_calls == [], (
+        "api_server wake must not go through handle_message (wrong-session bug)"
+    )
+    assert len(posts) == 1
+    assert posts[0]["session_id"] == "raw-sid-123"
+    assert tid in posts[0]["text"]
+    # The wake self-post IS the delivery on this path (no separate text-ping
+    # fallback is attempted for stateless api_server subs) — cursor advances
+    # once the wake succeeds.
+    assert _unseen_terminal_events(tid, "api_server", "raw-sid-123") == []

tests/gateway/test_wake_delivery.py

@@ -0,0 +1,188 @@
+"""Tests for gateway/wake.py — background wake delivery.
+
+Two strategies:
+* push-capable adapters keep the synthetic MessageEvent / handle_message path;
+* the stateless API server (supports_async_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().
+"""
+
+import asyncio
+
+import pytest
+
+from gateway.config import Platform
+from gateway.session import SessionSource
+from gateway.wake import deliver_wake, adapter_supports_push
+
+
+class PushAdapter:
+    """Default adapter shape — no supports_async_delivery attribute."""
+
+    def __init__(self):
+        self.handled = []
+
+    async def handle_message(self, event):
+        self.handled.append(event)
+
+
+class ApiServerLikeAdapter:
+    supports_async_delivery = False
+
+    def __init__(self, host="0.0.0.0", port=0, key="test-key", model="hermes"):
+        self._host = host
+        self._port = port
+        self._api_key = key
+        self._model_name = model
+
+    async def handle_message(self, event):  # pragma: no cover — must NOT be hit
+        raise AssertionError("non-push adapter must not receive handle_message wakes")
+
+
+def _source():
+    return SessionSource(
+        platform=Platform.TELEGRAM,
+        chat_id="chat-1",
+        chat_type="group",
+    )
+
+
+def test_adapter_supports_push_default_true():
+    assert adapter_supports_push(PushAdapter()) is True
+    assert adapter_supports_push(ApiServerLikeAdapter()) is False
+
+
+def test_deliver_wake_push_adapter_uses_handle_message():
+    adapter = PushAdapter()
+    asyncio.run(deliver_wake(adapter, text="wake up", source=_source()))
+    assert len(adapter.handled) == 1
+    evt = adapter.handled[0]
+    assert evt.text == "wake up"
+    assert evt.internal is True
+    assert evt.source.chat_id == "chat-1"
+
+
+def test_deliver_wake_push_adapter_requires_source():
+    with pytest.raises(ValueError):
+        asyncio.run(deliver_wake(PushAdapter(), text="x", session_id="sid"))
+
+
+def test_deliver_wake_non_push_requires_session_id():
+    with pytest.raises(ValueError):
+        asyncio.run(deliver_wake(ApiServerLikeAdapter(), text="x", source=_source()))
+
+
+def test_deliver_wake_non_push_requires_api_key():
+    """Session continuation is 403-gated on API_SERVER_KEY — a missing key
+    must fail loudly instead of running the wake in a fresh session."""
+    adapter = ApiServerLikeAdapter(key="")
+    with pytest.raises(RuntimeError, match="API_SERVER_KEY"):
+        asyncio.run(deliver_wake(adapter, text="x", session_id="raw-sid"))
+
+
+async def _serve(handler):
+    """Spin an in-process aiohttp server on an ephemeral loopback port."""
+    from aiohttp import web
+
+    app = web.Application()
+    app.router.add_post("/v1/chat/completions", handler)
+    runner = web.AppRunner(app)
+    await runner.setup()
+    site = web.TCPSite(runner, "127.0.0.1", 0)
+    await site.start()
+    port = site._server.sockets[0].getsockname()[1]
+    return runner, port
+
+
+def test_deliver_wake_non_push_self_posts_raw_session_id(monkeypatch):
+    """The self-post carries the RAW session id header + bearer auth and a
+    single user message with stream=false — the exact entry point real
+    gateway turns use."""
+    from aiohttp import web
+
+    seen = {}
+
+    async def handler(request):
+        seen["session_id"] = request.headers.get("X-Hermes-Session-Id")
+        seen["auth"] = request.headers.get("Authorization")
+        seen["body"] = await request.json()
+        return web.json_response({"choices": [{"message": {"content": "ok"}}]})
+
+    async def run():
+        runner, port = await _serve(handler)
+        try:
+            adapter = ApiServerLikeAdapter(host="0.0.0.0", port=port, key="sekrit")
+            await deliver_wake(adapter, text="task done — wake", session_id="raw-sid-42")
+        finally:
+            await runner.cleanup()
+
+    asyncio.run(run())
+    assert seen["session_id"] == "raw-sid-42"
+    assert seen["auth"] == "Bearer sekrit"
+    assert seen["body"]["stream"] is False
+    assert seen["body"]["messages"] == [
+        {"role": "user", "content": "task done — wake"}
+    ]
+
+
+def test_deliver_wake_retries_429_then_succeeds(monkeypatch):
+    """HTTP 429 (max_concurrent_runs cap) is transient — retried with backoff."""
+    from aiohttp import web
+
+    import gateway.wake as wake_mod
+
+    monkeypatch.setattr(wake_mod, "_RETRY_DELAYS_SECONDS", (0.01, 0.01, 0.01))
+    calls = {"n": 0}
+
+    async def handler(request):
+        calls["n"] += 1
+        if calls["n"] == 1:
+            return web.json_response({"error": "busy"}, status=429)
+        return web.json_response({"choices": []})
+
+    async def run():
+        runner, port = await _serve(handler)
+        try:
+            adapter = ApiServerLikeAdapter(port=port)
+            await deliver_wake(adapter, text="x", session_id="sid")
+        finally:
+            await runner.cleanup()
+
+    asyncio.run(run())
+    assert calls["n"] == 2
+
+
+def test_deliver_wake_raises_on_permanent_http_error(monkeypatch):
+    """Auth/validation errors (403/400) are permanent — raise immediately so
+    the caller can rewind instead of treating the event as delivered."""
+    from aiohttp import web
+
+    calls = {"n": 0}
+
+    async def handler(request):
+        calls["n"] += 1
+        return web.json_response({"error": "forbidden"}, status=403)
+
+    async def run():
+        runner, port = await _serve(handler)
+        try:
+            adapter = ApiServerLikeAdapter(port=port)
+            with pytest.raises(RuntimeError, match="HTTP 403"):
+                await deliver_wake(adapter, text="x", session_id="sid")
+        finally:
+            await runner.cleanup()
+
+    asyncio.run(run())
+    assert calls["n"] == 1
+
+
+def test_deliver_wake_raises_after_exhausted_retries(monkeypatch):
+    """Connection failures raise after bounded retries — never silent."""
+    import gateway.wake as wake_mod
+
+    monkeypatch.setattr(wake_mod, "_RETRY_DELAYS_SECONDS", (0.01,))
+    # Nothing is listening on this port.
+    adapter = ApiServerLikeAdapter(host="127.0.0.1", port=1, key="k")
+    with pytest.raises(RuntimeError, match="gave up"):
+        asyncio.run(deliver_wake(adapter, text="x", session_id="sid"))

tests/tools/test_delegate_apiserver_background.py

@@ -0,0 +1,190 @@
+"""delegate_task(background=true) on stateless API-server sessions.
+
+Previously async_delivery_supported()=False forced SYNCHRONOUS execution for
+every background dispatch on the API server, blocking the whole turn. Now
+that background completions can wake the originating session via the
+/v1/chat/completions self-post (gateway/wake.py), a session-continuable
+turn (raw session id bound as the api_server chat_id) dispatches async; only
+session-id-less one-shot requests keep the sync fallback.
+
+The wake target must be captured from the request-scoped chat_id binding,
+NOT from HERMES_SESSION_ID: constructing a child agent calls
+set_current_session_id(child.session_id), clobbering the HERMES_SESSION_ID
+ContextVar and os.environ with the subagent's internal id before the
+dispatch code reads it — the fake child build below reproduces that clobber.
+"""
+
+import json
+import time
+from unittest.mock import MagicMock
+
+import pytest
+
+from gateway.session_context import set_session_vars
+from tools.process_registry import process_registry
+
+
+@pytest.fixture(autouse=True)
+def _clean_queue_and_context(monkeypatch):
+    monkeypatch.delenv("HERMES_SESSION_ID", raising=False)
+    while not process_registry.completion_queue.empty():
+        try:
+            process_registry.completion_queue.get_nowait()
+        except Exception:
+            break
+    yield
+    # Restore ContextVars to the pristine "never set" sentinel rather than
+    # clear_session_vars()'s explicit-"" state, which would mask env vars for
+    # unrelated tests running later in the same worker.
+    import gateway.session_context as sc
+
+    for var in sc._VAR_MAP.values():
+        var.set(sc._UNSET)
+    sc._SESSION_ASYNC_DELIVERY.set(sc._UNSET)
+    # set_current_session_id (invoked by the clobber-reproducing fake child
+    # build) writes os.environ directly — scrub it so it can't leak into
+    # other test modules.
+    import os
+
+    os.environ.pop("HERMES_SESSION_ID", None)
+    while not process_registry.completion_queue.empty():
+        try:
+            process_registry.completion_queue.get_nowait()
+        except Exception:
+            break
+
+
+def _drain_one(timeout=5.0):
+    deadline = time.time() + timeout
+    while time.time() < deadline:
+        if not process_registry.completion_queue.empty():
+            return process_registry.completion_queue.get_nowait()
+        time.sleep(0.02)
+    return None
+
+
+def _fake_parent():
+    parent = MagicMock()
+    parent._delegate_depth = 0
+    parent.session_id = "sess"
+    parent._interrupt_requested = False
+    parent._active_children = []
+    parent._active_children_lock = None
+    return parent
+
+
+def _patch_delegate(monkeypatch):
+    import tools.delegate_tool as dt
+
+    fake_child = MagicMock()
+    fake_child._delegate_role = "leaf"
+    fake_child._subagent_id = "s1"
+
+    def fast_child(task_index, goal, child=None, parent_agent=None, **kw):
+        return {
+            "task_index": 0, "status": "completed", "summary": f"done: {goal}",
+            "api_calls": 1, "duration_seconds": 0.1, "model": "m",
+            "exit_reason": "completed",
+        }
+
+    creds = {
+        "model": "m", "provider": None, "base_url": None, "api_key": None,
+        "api_mode": None, "command": None, "args": None,
+    }
+    def clobbering_build_child(**kw):
+        # Reproduce what the real _build_child_agent -> AIAgent -> agent_init
+        # path does: it synchronizes the child's internal session id into the
+        # HERMES_SESSION_ID ContextVar + os.environ, clobbering the spawner's
+        # id ~milliseconds before delegate_tool dispatches the batch.
+        from gateway.session_context import set_current_session_id
+
+        set_current_session_id("20260715_child1")
+        return fake_child
+
+    monkeypatch.setattr(dt, "_build_child_agent", clobbering_build_child)
+    monkeypatch.setattr(dt, "_run_single_child", fast_child)
+    monkeypatch.setattr(dt, "_resolve_delegation_credentials", lambda *a, **k: creds)
+    return dt
+
+
+def test_apiserver_session_with_id_dispatches_background(monkeypatch):
+    """async_delivery=False + a raw session id (HERMES_SESSION_ID) →
+    background dispatch (the completion wakes the session via the
+    api_server self-post), NOT the forced-sync fallback."""
+    dt = _patch_delegate(monkeypatch)
+    monkeypatch.setenv("HERMES_SESSION_ID", "raw-sid-7")
+    set_session_vars(
+        platform="api_server",
+        chat_id="raw-sid-7",
+        session_key="raw-sid-7",
+        session_id="raw-sid-7",
+        async_delivery=False,
+    )
+
+    out = dt.delegate_task(
+        goal="bg on api_server", context="ctx",
+        background=True, parent_agent=_fake_parent(),
+    )
+    parsed = json.loads(out)
+    assert parsed["status"] == "dispatched", parsed
+    assert parsed["mode"] == "background"
+
+    evt = _drain_one()
+    assert evt is not None
+    assert evt["type"] == "async_delegation"
+    # The raw session id is stamped so the gateway drain can self-post the
+    # wake to the REAL session (session_key alone is the raw id here, which
+    # carries no parseable routing metadata). Crucially this is the SPAWNER's
+    # id, not the subagent-internal id the child build clobbered
+    # HERMES_SESSION_ID with (see clobbering_build_child).
+    assert evt["origin_session_id"] == "raw-sid-7"
+
+
+# ---------------------------------------------------------------------------
+# _current_origin_session_id — the clobber-proof origin capture helper
+# ---------------------------------------------------------------------------
+
+
+def test_origin_helper_survives_child_session_clobber(monkeypatch):
+    """set_current_session_id (child agent construction) rewrites the
+    HERMES_SESSION_ID ContextVar + env, but the request-scoped chat_id
+    binding is untouched — the helper must keep returning the spawner's id."""
+    from gateway.session_context import set_current_session_id
+    from tools.async_delegation import _current_origin_session_id
+
+    set_session_vars(platform="api_server", chat_id="raw-origin-1")
+    assert _current_origin_session_id() == "raw-origin-1"
+
+    set_current_session_id("20260715_child2")  # the clobber
+    assert _current_origin_session_id() == "raw-origin-1"
+
+
+def test_origin_helper_empty_on_push_platforms(monkeypatch):
+    """On push platforms chat_id identifies a chat, not a session — the
+    helper must yield empty rather than misroute a wake there."""
+    from tools.async_delegation import _current_origin_session_id
+
+    set_session_vars(platform="telegram", chat_id="123456789")
+    assert _current_origin_session_id() == ""
+
+
+def test_apiserver_session_without_id_stays_synchronous(monkeypatch):
+    """No session id to wake → keep the sync fallback (a detached result
+    would never re-enter any conversation)."""
+    dt = _patch_delegate(monkeypatch)
+    set_session_vars(
+        platform="api_server",
+        chat_id="",
+        session_key="",
+        session_id="",
+        async_delivery=False,
+    )
+
+    out = dt.delegate_task(
+        goal="one-shot", context="ctx",
+        background=True, parent_agent=_fake_parent(),
+    )
+    parsed = json.loads(out)
+    assert parsed.get("status") != "dispatched", parsed
+    assert "SYNCHRONOUSLY" in parsed.get("note", "")
+    assert process_registry.completion_queue.empty()

tools/async_delegation.py

@@ -487,6 +487,34 @@ def _prune_completed_locked() -> None:
         _records.pop(rid, None)
 
 
+def _current_origin_session_id() -> str:
+    """Raw session id of the ORIGINATING api_server request, or ``""``.
+
+    The obvious source — ``HERMES_SESSION_ID`` via ``get_session_env`` — is
+    NOT safe to read at dispatch time: constructing a child agent
+    (``agent/agent_init.py``) calls ``set_current_session_id(child.session_id)``,
+    clobbering that ContextVar *and* ``os.environ`` with the subagent's
+    internal ``{timestamp}_{uuid}`` id moments before the dispatch code reads
+    it, so the completion wake would self-post into the subagent's own
+    (unread) session instead of the spawner's.
+
+    The request-scoped ``HERMES_SESSION_CHAT_ID`` binding survives child
+    construction: ``_bind_api_server_session`` binds ``chat_id`` to the raw
+    ``X-Hermes-Session-Id``, and its only writer is ``set_session_vars`` —
+    ``set_current_session_id`` never touches it. Gate on the platform: on
+    push platforms ``chat_id`` is a chat, not a session, so yield ``""``
+    there.
+    """
+    try:
+        from gateway.session_context import get_session_env
+
+        if get_session_env("HERMES_SESSION_PLATFORM", "") != "api_server":
+            return ""
+        return get_session_env("HERMES_SESSION_CHAT_ID", "") or ""
+    except Exception:
+        return ""
+
+
 def dispatch_async_delegation(
     *,
     goal: str,
@@ -498,6 +526,7 @@ def dispatch_async_delegation(
     parent_session_id: Optional[str] = None,
     runner: Callable[[], Dict[str, Any]],
     origin_ui_session_id: str = "",
+    origin_session_id: str = "",
     interrupt_fn: Optional[Callable[[], None]] = None,
     max_async_children: int = _DEFAULT_MAX_ASYNC_CHILDREN,
 ) -> Dict[str, Any]:
@@ -546,6 +575,7 @@ def dispatch_async_delegation(
         "model": model,
         "session_key": session_key,
         "origin_ui_session_id": origin_ui_session_id,
+        "origin_session_id": origin_session_id,
         "parent_session_id": parent_session_id,
         "status": "running",
         "dispatched_at": dispatched_at,
@@ -666,6 +696,7 @@ def _push_completion_event(
         # session; empty string => CLI (single-session) path.
         "session_key": record.get("session_key", ""),
         "origin_ui_session_id": record.get("origin_ui_session_id", ""),
+        "origin_session_id": record.get("origin_session_id", ""),
         "parent_session_id": record.get("parent_session_id"),
         "goal": record.get("goal", ""),
         "context": record.get("context"),
@@ -705,6 +736,7 @@ def dispatch_async_delegation_batch(
     parent_session_id: Optional[str] = None,
     runner: Callable[[], Dict[str, Any]],
     origin_ui_session_id: str = "",
+    origin_session_id: str = "",
     interrupt_fn: Optional[Callable[[], None]] = None,
     max_async_children: int = _DEFAULT_MAX_ASYNC_CHILDREN,
     delegation_id: Optional[str] = None,
@@ -746,6 +778,7 @@ def dispatch_async_delegation_batch(
         "model": model,
         "session_key": session_key,
         "origin_ui_session_id": origin_ui_session_id,
+        "origin_session_id": origin_session_id,
         "parent_session_id": parent_session_id,
         "status": "running",
         "dispatched_at": dispatched_at,
@@ -846,6 +879,7 @@ def _finalize_batch(
         "delegation_id": delegation_id,
         "session_key": event_record.get("session_key", ""),
         "origin_ui_session_id": event_record.get("origin_ui_session_id", ""),
+        "origin_session_id": event_record.get("origin_session_id", ""),
         "parent_session_id": event_record.get("parent_session_id"),
         "goal": event_record.get("goal", ""),
         "goals": event_record.get("goals"),

tools/delegate_tool.py

@@ -2583,6 +2583,18 @@ def delegate_task(
 
     _parent_tool_names = list(_model_tools._last_resolved_tool_names)
 
+    # Capture the ORIGINATING session's wake target BEFORE any child agent is
+    # constructed: _build_child_agent() -> AIAgent() -> agent_init calls
+    # set_current_session_id(child.session_id), which clobbers the
+    # HERMES_SESSION_ID ContextVar and os.environ with the subagent's internal
+    # id before the background-dispatch code below would read it. The
+    # request-scoped chat_id binding (the raw X-Hermes-Session-Id on
+    # api_server) is untouched by child construction, so read it here and
+    # thread it through the dispatch.
+    from tools.async_delegation import _current_origin_session_id
+
+    _origin_wake_sid = _current_origin_session_id()
+
     # Build all child agents on the main thread (thread-safe construction)
     # Wrapped in try/finally so the global is always restored even if a
     # child build raises (otherwise _last_resolved_tool_names stays corrupted).
@@ -2921,6 +2933,30 @@ def delegate_task(
             _async_ok = async_delivery_supported()
         except Exception:
             _async_ok = True
+
+        _wake_sid = ""
+        if not _async_ok:
+            # The adapter itself cannot push, but if a raw session id is
+            # bound (the API server always binds one — see
+            # ApiServerAdapter._bind_api_server_session), gateway.wake can
+            # still reach the session by self-POSTing /v1/chat/completions
+            # with that id in X-Hermes-Session-Id once the batch completes.
+            # Only fall back to forced-sync execution when there is truly no
+            # session id to wake. Uses the origin captured before child
+            # construction (see _origin_wake_sid above) — reading
+            # HERMES_SESSION_ID here would return the subagent's internal id.
+            _wake_sid = _origin_wake_sid
+            if _wake_sid:
+                logger.info(
+                    "delegate_task: async delivery unsupported on this "
+                    "session, but a session id is bound (%s) — dispatching "
+                    "in the background and waking the session via self-post "
+                    "when it completes instead of forcing synchronous "
+                    "execution.",
+                    _wake_sid,
+                )
+                _async_ok = True
+
         if not _async_ok:
             logger.info(
                 "delegate_task: async delivery unsupported on this session "
@@ -3013,6 +3049,7 @@ def delegate_task(
             model=creds["model"],
             session_key=_session_key,
             origin_ui_session_id=_origin_ui_session_id,
+            origin_session_id=_wake_sid,
             parent_session_id=_parent_session_id,
             runner=_batch_runner,
             interrupt_fn=_batch_interrupt,

tools/kanban_tools.py

@@ -1139,7 +1139,17 @@ def _handle_create(args: dict, **kw) -> str:
     # Stamp the originating session id when the agent loop runs under
     # ACP (which sets HERMES_SESSION_ID before invoking tools). NULL on
     # CLI / dashboard paths and on legacy hosts that don't set the env.
-    session_id = args.get("session_id") or os.environ.get("HERMES_SESSION_ID")
+    # Prefer the request-scoped api_server origin binding: HERMES_SESSION_ID
+    # is clobbered with a subagent's internal id whenever a child agent is
+    # constructed in-process (agent_init calls set_current_session_id), which
+    # would stamp — and later wake — the wrong session.
+    from tools.async_delegation import _current_origin_session_id
+
+    session_id = (
+        args.get("session_id")
+        or _current_origin_session_id()
+        or os.environ.get("HERMES_SESSION_ID")
+    )
     priority = args.get("priority")
     # Resolve workspace. Workspace sharing is always explicit: omitted fields
     # mean a fresh scratch workspace, even when a dispatcher-spawned worker