diff --git a/gateway/kanban_watchers.py b/gateway/kanban_watchers.py index eb1c68ffd66b..3b5692667d96 100644 --- a/gateway/kanban_watchers.py +++ b/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, diff --git a/gateway/run.py b/gateway/run.py index 26cf8859ac42..cee6256cefd6 100644 --- a/gateway/run.py +++ b/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() diff --git a/gateway/wake.py b/gateway/wake.py new file mode 100644 index 000000000000..cee8d45e50c3 --- /dev/null +++ b/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:``) 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 diff --git a/tests/gateway/test_background_process_notifications.py b/tests/gateway/test_background_process_notifications.py index 17ae232b2aae..358b7ff7a795 100644 --- a/tests/gateway/test_background_process_notifications.py +++ b/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"] diff --git a/tests/gateway/test_kanban_notifier_apiserver_wake.py b/tests/gateway/test_kanban_notifier_apiserver_wake.py new file mode 100644 index 000000000000..0fd7f3ed999b --- /dev/null +++ b/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") == [] diff --git a/tests/gateway/test_wake_delivery.py b/tests/gateway/test_wake_delivery.py new file mode 100644 index 000000000000..3c09ef87c18f --- /dev/null +++ b/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")) diff --git a/tests/tools/test_delegate_apiserver_background.py b/tests/tools/test_delegate_apiserver_background.py new file mode 100644 index 000000000000..f0a07d6ddbdd --- /dev/null +++ b/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() diff --git a/tools/async_delegation.py b/tools/async_delegation.py index 01d6b84ab621..1aed7b7f9068 100644 --- a/tools/async_delegation.py +++ b/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"), diff --git a/tools/delegate_tool.py b/tools/delegate_tool.py index ab3224f48163..761785d7aae2 100644 --- a/tools/delegate_tool.py +++ b/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, diff --git a/tools/kanban_tools.py b/tools/kanban_tools.py index b1923a1e2c2b..46991b4a477b 100644 --- a/tools/kanban_tools.py +++ b/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