diff --git a/contributors/emails/hello@ianks.com b/contributors/emails/hello@ianks.com new file mode 100644 index 000000000000..bd967db2227e --- /dev/null +++ b/contributors/emails/hello@ianks.com @@ -0,0 +1 @@ +ianks diff --git a/gateway/kanban_watchers.py b/gateway/kanban_watchers.py index eb1c68ffd66b..50e44f980304 100644 --- a/gateway/kanban_watchers.py +++ b/gateway/kanban_watchers.py @@ -336,6 +336,14 @@ def _collect(): continue title = (task.title if task else sub["task_id"])[:120] board_tag = f"[{board_slug}] " if board_slug else "" + # Per-subscription failure-counter key. Hoisted out of the + # event loop: the wake self-post path (in the loop's + # ``else`` clause) needs it even when every event in the + # claim was skipped before reaching the send site. + sub_key = ( + sub["task_id"], sub["platform"], + sub["chat_id"], sub.get("thread_id") or "", + ) for ev in d["events"]: kind = ev.kind # Identity prefix: attribute terminal pings to the @@ -408,14 +416,49 @@ def _collect(): metadata: dict[str, Any] = {} if sub.get("thread_id"): metadata["thread_id"] = sub["thread_id"] - sub_key = ( - 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"], + ) + # Do NOT reset the failure counter here: on this + # path the wake self-post below IS the delivery, + # so the counter is resolved (reset or bumped) by + # the self-post outcome, not by skipping the send. + 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, @@ -475,12 +518,108 @@ def _collect(): # dropping the subscription is the terminal action. break else: - # All events delivered; advance cursor. The cursor + # All text pings delivered (or intentionally skipped + # for non-push adapters, whose delivery is the wake + # self-post below). Whether the cursor may advance now + # depends on the adapter class: + # + # * push-capable: the text send WAS the delivery, so + # advance immediately (pre-existing behavior); the + # wake injection below stays best-effort. + # * non-push (api_server): the wake self-post IS the + # delivery. Advancing first would let a failed / + # retry-exhausted self-post (swallowed by the + # best-effort except) permanently lose the event. + # So the self-post runs FIRST and the cursor only + # advances after it succeeds — a failure rewinds the + # claim exactly like a failed send() above, so the + # next tick retries. + task_terminal = task and task.status in {"done", "archived"} + _WAKE_KINDS = ("completed", "gave_up", "crashed", "timed_out", "blocked") + _wake_kinds = {ev.kind for ev in d["events"] if ev.kind in _WAKE_KINDS} + from gateway.wake import adapter_supports_push as _adapter_push_ok + + _is_push_adapter = _adapter_push_ok(adapter) + _session_key = "" + _synth = "" + if _wake_kinds: + _session_key = getattr(task, "session_id", None) or "" + if _wake_kinds and _session_key: + _title = (task.title if task else sub["task_id"])[:120] + _assignee = task.assignee if task else "" + _parts = [] + if "completed" in _wake_kinds: _parts.append(t("gateway.kanban.wake.completed")) + if "gave_up" in _wake_kinds: _parts.append(t("gateway.kanban.wake.gave_up")) + if "crashed" in _wake_kinds: _parts.append(t("gateway.kanban.wake.crashed")) + if "timed_out" in _wake_kinds: _parts.append(t("gateway.kanban.wake.timed_out")) + if "blocked" in _wake_kinds: _parts.append(t("gateway.kanban.wake.blocked")) + _status = t("gateway.kanban.wake.status_joiner").join(_parts) or t("gateway.kanban.wake.status_default") + _synth = t( + "gateway.kanban.wake.message", + task_id=sub["task_id"], + status=_status, + title=_title, + assignee=_assignee, + board=board_slug, + ) + + if not _is_push_adapter and _wake_kinds and _session_key: + # Wake self-post IS the delivery on this path — + # it must succeed BEFORE the cursor advances. + from gateway.wake import deliver_wake + + try: + await deliver_wake( + adapter, + text=_synth, + session_id=_session_key, + ) + 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, + ) + sub_fail_counts.pop(sub_key, None) + except Exception as _wk_err: + fails = sub_fail_counts.get(sub_key, 0) + 1 + sub_fail_counts[sub_key] = fails + logger.warning( + "kanban notifier: wake self-post failed " + "for %s (attempt %d/%d): %s", + sub["task_id"], fails, + MAX_SEND_FAILURES, _wk_err, exc_info=True, + ) + if fails >= MAX_SEND_FAILURES: + logger.warning( + "kanban notifier: dropping subscription " + "%s on %s after %d consecutive wake failures", + sub["task_id"], platform_str, fails, + ) + await asyncio.to_thread(self._kanban_unsub, sub, board_slug) + sub_fail_counts.pop(sub_key, None) + else: + # Rewind the pre-send claim so the next + # tick retries the self-post — the event + # is NOT lost. + await asyncio.to_thread( + self._kanban_rewind, + sub, + d["cursor"], + d.get("old_cursor", 0), + board_slug, + ) + continue + + # Delivery complete (text ping for push adapters, wake + # self-post for non-push): advance cursor. The cursor # is the dedup mechanism — it prevents re-delivery # of the same event on subsequent ticks. await asyncio.to_thread( self._kanban_advance, sub, d["cursor"], board_slug, ) + if not _is_push_adapter: + # Nothing left to deliver on this path (the wake, + # if any, already succeeded above). + sub_fail_counts.pop(sub_key, None) # Unsubscribe only when the task has reached a truly # final status (done / archived). For blocked / # gave_up / crashed / timed_out the subscription is @@ -488,68 +627,50 @@ def _collect(): # dispatcher respawns the task and it cycles into the # same state. See the longer comment on TERMINAL_KINDS # above for the failure mode this prevents. - task_terminal = task and task.status in {"done", "archived"} - _WAKE_KINDS = ("completed", "gave_up", "crashed", "timed_out", "blocked") - _wake_kinds = {ev.kind for ev in d["events"] if ev.kind in _WAKE_KINDS} - if _wake_kinds: + if _is_push_adapter and _wake_kinds and _session_key: try: - _session_key = getattr(task, "session_id", None) or "" - if _session_key: - _title = (task.title if task else sub["task_id"])[:120] - _assignee = task.assignee if task else "" - _parts = [] - if "completed" in _wake_kinds: _parts.append(t("gateway.kanban.wake.completed")) - if "gave_up" in _wake_kinds: _parts.append(t("gateway.kanban.wake.gave_up")) - if "crashed" in _wake_kinds: _parts.append(t("gateway.kanban.wake.crashed")) - if "timed_out" in _wake_kinds: _parts.append(t("gateway.kanban.wake.timed_out")) - if "blocked" in _wake_kinds: _parts.append(t("gateway.kanban.wake.blocked")) - _status = t("gateway.kanban.wake.status_joiner").join(_parts) or t("gateway.kanban.wake.status_default") - _synth = t( - "gateway.kanban.wake.message", - task_id=sub["task_id"], - status=_status, - title=_title, - assignee=_assignee, - board=board_slug, - ) - from gateway.session import SessionSource - from gateway.platforms.base import MessageEvent, MessageType - # KNOWN LIMITATION (tracked follow-up): the - # subscription row does not persist the - # creator's chat_type, and it is not carried - # on the session-context bridge, so we cannot - # faithfully reconstruct the creator's real - # session key here. build_session_key() keys - # DMs (":dm:") on a wholly different - # shape from group/thread, so any hardcoded - # value mis-routes some creators. "group" is - # the least-surprising default for the - # dashboard/group flows this wake primarily - # serves; DM-originated creators are handled - # by the follow-up that stamps + persists - # chat_type end-to-end. handle_message() - # get_or_create_session's the target, so a - # mismatch degrades to "wake lands in a fresh - # group session" — never an exception. - _source = SessionSource( - platform=plat, - chat_id=sub["chat_id"], - chat_type="group", - thread_id=sub.get("thread_id") or None, - user_id=sub.get("user_id"), - profile=sub_profile or None, - ) - _synth_event = MessageEvent( - text=_synth, - message_type=MessageType.TEXT, - 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, - ) + from gateway.session import SessionSource + 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 + # on the session-context bridge, so we cannot + # faithfully reconstruct the creator's real + # session key here. build_session_key() keys + # DMs (":dm:") on a wholly different + # shape from group/thread, so any hardcoded + # value mis-routes some creators. "group" is + # the least-surprising default for the + # dashboard/group flows this wake primarily + # serves; DM-originated creators are handled + # by the follow-up that stamps + persists + # chat_type end-to-end. handle_message() + # get_or_create_session's the target, so a + # mismatch degrades to "wake lands in a fresh + # group session" — never an exception. + _source = SessionSource( + platform=plat, + chat_id=sub["chat_id"], + chat_type="group", + thread_id=sub.get("thread_id") or None, + user_id=sub.get("user_id"), + profile=sub_profile or None, + ) + # deliver_wake preserves the synthetic + # MessageEvent/handle_message path for + # push-capable adapters (the non-push / + # self-post branch is handled BEFORE the + # cursor advance above). + await deliver_wake( + adapter, + text=_synth, + session_id=_session_key, + source=_source, + ) + 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, + ) except Exception as _wk_err: # Best-effort: the notification itself already # delivered and the cursor has advanced, so a diff --git a/gateway/platforms/api_server.py b/gateway/platforms/api_server.py index b783f34db468..589bb3910c97 100644 --- a/gateway/platforms/api_server.py +++ b/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( diff --git a/gateway/run.py b/gateway/run.py index 9afea42c3228..c6ca17f32593 100644 --- a/gateway/run.py +++ b/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() 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_api_server_runs.py b/tests/gateway/test_api_server_runs.py index 147a75ab4b30..ed0240a9ff2b 100644 --- a/tests/gateway/test_api_server_runs.py +++ b/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) diff --git a/tests/gateway/test_background_process_notifications.py b/tests/gateway/test_background_process_notifications.py index fe3a6588b1b9..4cd5f2a83ba2 100644 --- a/tests/gateway/test_background_process_notifications.py +++ b/tests/gateway/test_background_process_notifications.py @@ -556,3 +556,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..4d05ee3ee4ce --- /dev/null +++ b/tests/gateway/test_kanban_notifier_apiserver_wake.py @@ -0,0 +1,219 @@ +"""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") == [] + + +def test_apiserver_failed_self_post_rewinds_cursor(tmp_path, monkeypatch): + """A failed/exhausted wake self-post must NOT advance the cursor: on the + api_server path the self-post IS the delivery, so advancing first would + permanently lose the event behind a best-effort except. The claim is + rewound and the event stays visible for the next tick's retry.""" + monkeypatch.setenv("HERMES_KANBAN_DB", str(tmp_path / "apiserver_fail.db")) + kb.init_db() + tid = _create_completed_subscription( + "api_server", "raw-sid-999", session_id="raw-sid-999", + ) + + async def failing_self_post(adapter, *, text, session_id): + raise RuntimeError("self-post exhausted retries") + + import gateway.wake as wake_mod + + monkeypatch.setattr(wake_mod, "_self_post_chat_completion", failing_self_post) + + adapter = ApiServerLikeAdapter() + runner = _make_runner({Platform.API_SERVER: adapter}) + asyncio.run(_run_one_notifier_tick(monkeypatch, runner)) + + # Event NOT lost: the cursor was rewound, so the completed event is still + # unseen and will be re-claimed (and the self-post retried) next tick. + assert [ev.kind for ev in _unseen_terminal_events(tid, "api_server", "raw-sid-999")] == [ + "completed" + ] + # And the failure was counted toward the drop threshold. + assert list(runner._kanban_sub_fail_counts.values()) == [1] + + +def test_apiserver_self_post_succeeds_after_earlier_failure(tmp_path, monkeypatch): + """The rewound event is retried on the next tick; a successful self-post + then advances the cursor and clears the failure counter.""" + monkeypatch.setenv("HERMES_KANBAN_DB", str(tmp_path / "apiserver_retry.db")) + kb.init_db() + tid = _create_completed_subscription( + "api_server", "raw-sid-777", session_id="raw-sid-777", + ) + + calls = {"n": 0} + + async def flaky_self_post(adapter, *, text, session_id): + calls["n"] += 1 + if calls["n"] == 1: + raise RuntimeError("transient outage") + + import gateway.wake as wake_mod + + monkeypatch.setattr(wake_mod, "_self_post_chat_completion", flaky_self_post) + + adapter = ApiServerLikeAdapter() + runner = _make_runner({Platform.API_SERVER: adapter}) + asyncio.run(_run_one_notifier_tick(monkeypatch, runner)) + assert calls["n"] == 1 + assert len(_unseen_terminal_events(tid, "api_server", "raw-sid-777")) == 1 + + # Second tick: the re-claimed event's self-post succeeds → cursor advances. + runner._running = True + asyncio.run(_run_one_notifier_tick(monkeypatch, runner)) + assert calls["n"] == 2 + assert _unseen_terminal_events(tid, "api_server", "raw-sid-777") == [] + assert runner._kanban_sub_fail_counts == {} 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_async_delegation.py b/tests/tools/test_async_delegation.py index 0c7bd44c1b89..cc6082ec66b0 100644 --- a/tests/tools/test_async_delegation.py +++ b/tests/tools/test_async_delegation.py @@ -397,6 +397,81 @@ def test_recover_marks_abandoned_running_record_unknown(tmp_path, monkeypatch): assert restored.get_nowait()["status"] == "unknown" +def test_origin_session_id_survives_persistence_round_trip(tmp_path, monkeypatch): + """origin_session_id (the api_server wake self-post target) must be + persisted with the durable dispatch record and restored on recovery — + otherwise completions recovered after a process restart are unroutable + to api_server sessions (in-memory record is gone).""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + record = { + "delegation_id": "deleg_wake_target", + "session_key": "owner", + "origin_ui_session_id": "", + "origin_session_id": "raw-api-sid-42", + "parent_session_id": None, + "dispatched_at": 1.0, + } + ad._persist_dispatch(record) + + # Durable record carries the wake target. + durable = ad.get_durable_delegation("deleg_wake_target") + assert durable["origin_session_id"] == "raw-api-sid-42" + + # Simulate the owning process dying, then recovery after restart: the + # regenerated completion event must still carry the wake target. + with ad._DB_LOCK, ad._connect() as conn: + conn.execute( + "UPDATE async_delegations SET owner_pid=?, owner_started_at=NULL WHERE delegation_id=?", + (99999999, "deleg_wake_target"), + ) + restored = queue.Queue() + assert ad.restore_undelivered_completions(restored) == 1 + evt = restored.get_nowait() + assert evt["delegation_id"] == "deleg_wake_target" + assert evt["origin_session_id"] == "raw-api-sid-42" + assert evt["restored"] is True + + +def test_origin_session_id_migration_backfills_legacy_rows(tmp_path, monkeypatch): + """Rows written by a pre-origin_session_id build must survive the ALTER + TABLE migration and read back as an empty wake target.""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + # Create a legacy-schema DB (no origin_session_id column). + import sqlite3 + + db_path = ad._db_path() + db_path.parent.mkdir(parents=True, exist_ok=True) + legacy = sqlite3.connect(str(db_path)) + legacy.execute( + """CREATE TABLE async_delegations ( + delegation_id TEXT PRIMARY KEY, + origin_session TEXT NOT NULL, + origin_ui_session_id TEXT NOT NULL DEFAULT '', + parent_session_id TEXT, + state TEXT NOT NULL, + dispatched_at REAL NOT NULL, + completed_at REAL, + updated_at REAL NOT NULL, + event_json TEXT, + result_json TEXT, + delivery_state TEXT NOT NULL DEFAULT 'pending', + delivery_attempts INTEGER NOT NULL DEFAULT 0, + delivered_at REAL + )""" + ) + legacy.execute( + """INSERT INTO async_delegations + (delegation_id, origin_session, state, dispatched_at, updated_at) + VALUES ('deleg_legacy', 'owner', 'running', 1.0, 1.0)""" + ) + legacy.commit() + legacy.close() + + durable = ad.get_durable_delegation("deleg_legacy") + assert durable is not None + assert durable["origin_session_id"] == "" + + def test_durable_delivery_claim_is_exclusive_and_retryable(tmp_path, monkeypatch): monkeypatch.setenv("HERMES_HOME", str(tmp_path)) record = { 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..e456aed50f86 100644 --- a/tools/async_delegation.py +++ b/tools/async_delegation.py @@ -115,7 +115,8 @@ def _connect() -> sqlite3.Connection: owner_started_at INTEGER, task_json TEXT, delivery_claim TEXT, - delivery_claimed_at REAL + delivery_claimed_at REAL, + origin_session_id TEXT NOT NULL DEFAULT '' )""" ) columns = {row[1] for row in conn.execute("PRAGMA table_info(async_delegations)")} @@ -125,6 +126,11 @@ def _connect() -> sqlite3.Connection: ("task_json", "TEXT"), ("delivery_claim", "TEXT"), ("delivery_claimed_at", "REAL"), + # Raw api_server session id (X-Hermes-Session-Id) of the ORIGINATING + # request — the wake self-post target. Without persisting it, + # completions recovered after a process restart are unroutable on + # api_server (the in-memory record that carried it is gone). + ("origin_session_id", "TEXT"), ): if name not in columns: conn.execute(f"ALTER TABLE async_delegations ADD COLUMN {name} {sql_type}") @@ -149,12 +155,13 @@ def _persist_dispatch(record: Dict[str, Any]) -> None: (delegation_id, origin_session, origin_ui_session_id, parent_session_id, state, dispatched_at, updated_at, delivery_state, delivery_attempts, owner_pid, - owner_started_at, task_json) - VALUES (?, ?, ?, ?, 'running', ?, ?, 'pending', 0, ?, ?, ?)""", + owner_started_at, task_json, origin_session_id) + VALUES (?, ?, ?, ?, 'running', ?, ?, 'pending', 0, ?, ?, ?, ?)""", (record["delegation_id"], record.get("session_key", ""), record.get("origin_ui_session_id", ""), record.get("parent_session_id"), record["dispatched_at"], now, __import__("os").getpid(), - owner_started_at, json.dumps(task_payload)), + owner_started_at, json.dumps(task_payload), + record.get("origin_session_id", "")), ) _prune_durable_records() @@ -235,11 +242,12 @@ def recover_abandoned_delegations() -> int: rows = conn.execute( """SELECT delegation_id, origin_session, origin_ui_session_id, parent_session_id, dispatched_at, owner_pid, - owner_started_at, task_json + owner_started_at, task_json, origin_session_id FROM async_delegations WHERE state IN ('running','finalizing')""" ).fetchall() for row in rows: - delegation_id, session_key, origin_ui, parent_id, dispatched_at, pid, started, task_json = row + (delegation_id, session_key, origin_ui, parent_id, dispatched_at, + pid, started, task_json, origin_session_id) = row live = False if pid: live = _pid_exists(int(pid)) @@ -251,6 +259,9 @@ def recover_abandoned_delegations() -> int: event = { "type": "async_delegation", "delegation_id": delegation_id, "session_key": session_key, "origin_ui_session_id": origin_ui, + # Restore the durable wake target so completions recovered + # after a restart remain routable to api_server sessions. + "origin_session_id": origin_session_id or "", "parent_session_id": parent_id, "goal": task.get("goal", ""), "goals": task.get("goals"), "context": task.get("context"), "toolsets": task.get("toolsets"), "role": task.get("role"), @@ -427,7 +438,8 @@ def get_durable_delegation(delegation_id: str) -> Optional[Dict[str, Any]]: with _DB_LOCK, _connect() as conn: row = conn.execute( """SELECT origin_session, state, dispatched_at, completed_at, - result_json, delivery_state, delivery_attempts + result_json, delivery_state, delivery_attempts, + origin_session_id FROM async_delegations WHERE delegation_id=?""", (delegation_id,), ).fetchone() if row is None: @@ -437,6 +449,7 @@ def get_durable_delegation(delegation_id: str) -> Optional[Dict[str, Any]]: "dispatched_at": row[2], "completed_at": row[3], "result": json.loads(row[4]) if row[4] else None, "delivery_state": row[5], "delivery_attempts": row[6], + "origin_session_id": row[7] or "", } @@ -487,6 +500,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 +539,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 +588,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 +709,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 +749,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 +791,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 +892,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 _execute_and_aggregate() -> dict: _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 _batch_interrupt(): 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