diff --git a/gateway/platforms/api_server.py b/gateway/platforms/api_server.py index 7bcf4bae4bad..97640727429b 100644 --- a/gateway/platforms/api_server.py +++ b/gateway/platforms/api_server.py @@ -1952,6 +1952,7 @@ def _http_route_table(self) -> List[tuple]: ("POST", "/api/sessions/{session_id}/chat", self._handle_session_chat), ("POST", "/api/sessions/{session_id}/chat/stream", self._handle_session_chat_stream), ("POST", "/api/sessions/{session_id}/model", self._handle_session_model_lock), + ("GET", "/api/sessions/{session_id}/delegations", self._handle_session_delegations), ("POST", "/v1/chat/completions", self._handle_chat_completions), ("POST", "/v1/responses", self._handle_responses), ("GET", "/v1/responses/{response_id}", self._handle_get_response), @@ -6385,6 +6386,8 @@ def _bind_api_server_session( chat_id: str = "", session_key: str = "", session_id: str = "", + origin_turn_id: str = "", + delegation_sync_only: bool = False, ) -> list: """Bind session contextvars for an API-server agent run. @@ -6393,6 +6396,21 @@ def _bind_api_server_session( completion via the gateway's authenticated self-post wake path, while remaining non-push adapters. + ``origin_turn_id`` mirrors ``chat_id``: the Omnio ``turn_id`` from the + request body (``/v1/runs``), bound here so a background delegation + dispatched from this run can thread it through to its completion + event (see ``tools.async_delegation._current_origin_session_id`` and + its turn-id sibling). Empty on non-Omnio deployments. + + ``delegation_sync_only`` mirrors ``origin_turn_id``: the Omnio + ``delegation_sync_only`` flag from the request body (``/v1/runs``), + set by the proxy for headless surfaces (crons, trigger.dev runs) that + have no channel to ever receive a background delegation's wake. Bound + here so ``delegate_task(background=True)`` can force its synchronous + fallback for this run regardless of an otherwise-available wake + session id (see ``tools.async_delegation._current_delegation_sync_only`` + and ``tools/delegate_tool.py``). + Returns reset tokens; pass them to ``clear_session_vars`` in a ``finally`` block (the binding is request-scoped and must not outlive the turn — a session resumed later on a delivering interface, e.g. the @@ -6406,6 +6424,8 @@ def _bind_api_server_session( session_key=session_key, session_id=session_id, async_delivery=True, + origin_turn_id=origin_turn_id, + delegation_sync_only=delegation_sync_only, ) async def _run_agent( @@ -7079,6 +7099,11 @@ async def _handle_runs(self, request: "web.Request") -> "web.Response": previous_response_id = body.get("previous_response_id") explicit_session_id = body.get("session_id") turn_id = body.get("turn_id") + # Omnio proxy flag: headless surfaces (crons, trigger.dev runs) have + # no channel to ever receive a background delegation's wake, so they + # force delegate_task(background=True) onto its synchronous fallback + # for this run — see _bind_api_server_session and tools/delegate_tool.py. + delegation_sync_only = bool(body.get("delegation_sync_only")) if explicit_session_id is not None: if not isinstance(explicit_session_id, str) or not explicit_session_id.strip(): @@ -7687,6 +7712,8 @@ def _run_sync(): chat_id=session_id or "", session_key=approval_session_key, session_id=session_id or "", + origin_turn_id=str(turn_id) if turn_id else "", + delegation_sync_only=delegation_sync_only, ) register_gateway_notify(approval_session_key, _approval_notify) # Mark this run's session as an interactive surface so @@ -8076,6 +8103,33 @@ async def _handle_get_run(self, request: "web.Request") -> "web.Response": response_status.setdefault("run_id", run_id) return web.json_response(response_status) + async def _handle_session_delegations(self, request: "web.Request") -> "web.Response": + """GET /api/sessions/{session_id}/delegations — this session's async delegations. + + Filters the process-wide async-delegation registry down to records whose + ``origin_session_id`` matches the requested session, so an external UI can + rebuild "what is this conversation still waiting on" from the process that + owns the children instead of from its own bookkeeping. Records carry the + registry's live-status fields (``children_activity``, per-child + ``finished``) — see ``list_async_delegations``. + """ + auth_err = self._check_auth(request) + if auth_err: + return auth_err + + session_id = request.match_info["session_id"] + try: + from tools.async_delegation import list_async_delegations + + records = list_async_delegations() + except Exception as exc: + logger.exception("[api_server] delegation listing failed: %s", exc) + return web.json_response(_openai_error(str(exc)), status=500) + data = [ + r for r in records if r.get("origin_session_id") == session_id + ] + return web.json_response({"data": data}) + async def _handle_run_events(self, request: "web.Request") -> "web.StreamResponse": """Replay sequence_number > ``after``, then follow the live run.""" auth_err = self._check_auth(request) diff --git a/gateway/run.py b/gateway/run.py index 1b37fc96e10e..3a2ccbaf8b07 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -18158,7 +18158,11 @@ async def _inject_watch_notification( raw_sid = _sk if raw_sid: adapter = self.adapters.get(Platform.API_SERVER) - from gateway.wake import adapter_supports_push, deliver_wake + from gateway.wake import ( + WakeHookPermanentError, + adapter_supports_push, + deliver_wake, + ) if adapter is not None and not adapter_supports_push(adapter): try: logger.info( @@ -18166,7 +18170,29 @@ async def _inject_watch_notification( "session %s via self-post", raw_sid, ) - await deliver_wake(adapter, text=synth_text, session_id=raw_sid) + await deliver_wake( + adapter, + text=synth_text, + session_id=raw_sid, + delegation_id=str(evt.get("delegation_id") or ""), + origin_turn_id=str(evt.get("origin_turn_id") or ""), + subagent_ids=list(evt.get("subagent_ids") or []), + ) + return True + except WakeHookPermanentError as e: + # Unwinnable (e.g. 404 — the proxy no longer + # recognises this turn, most likely the conversation + # was deleted). Retrying can never succeed, so treat + # this as CONSUMED rather than returning False — + # returning False would requeue the completion event + # and redeliver the same 404 forever. + logger.warning( + "wake_dropped origin_turn_id=%s delegation_id=%s " + "status=%s session=%s: %s", + e.origin_turn_id, + evt.get("delegation_id") or "", + e.status_code, raw_sid, e, + ) return True except Exception as e: logger.warning( @@ -18201,7 +18227,7 @@ async def _inject_watch_notification( # 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 + from gateway.wake import WakeHookPermanentError, deliver_wake raw_sid = str(evt.get("origin_session_id") or "").strip() or str(source.chat_id or "") try: logger.info( @@ -18209,7 +18235,26 @@ async def _inject_watch_notification( "%s via self-post", raw_sid, ) - await deliver_wake(adapter, text=synth_text, session_id=raw_sid) + await deliver_wake( + adapter, + text=synth_text, + session_id=raw_sid, + delegation_id=str(evt.get("delegation_id") or ""), + origin_turn_id=str(evt.get("origin_turn_id") or ""), + subagent_ids=list(evt.get("subagent_ids") or []), + ) + return True + except WakeHookPermanentError as e: + # See the twin branch above (no-routing-metadata case) for + # why this is CONSUMED rather than requeued: a 404 (deleted + # conversation) can never succeed on retry. + logger.warning( + "wake_dropped origin_turn_id=%s delegation_id=%s " + "status=%s session=%s: %s", + e.origin_turn_id, + evt.get("delegation_id") or "", + e.status_code, raw_sid, e, + ) return True except Exception as e: logger.warning( diff --git a/gateway/session_context.py b/gateway/session_context.py index 71d822658701..f057cfd503f8 100644 --- a/gateway/session_context.py +++ b/gateway/session_context.py @@ -94,6 +94,28 @@ def session_context_engaged() -> bool: _SESSION_PROFILE: ContextVar = ContextVar("HERMES_SESSION_PROFILE", default=_UNSET) +# Omnio product-turn id (the ``turn_id`` on ``POST /v1/runs``), bound +# alongside ``HERMES_SESSION_CHAT_ID`` for the same request. Empty on any +# non-Omnio deployment. Lets a background delegation dispatched from this +# request thread its ORIGINATING turn id through to its completion event, so +# the async wake can be redirected (via ``OMNIO_WAKE_HOOK``) into a real +# product turn instead of the raw self-post — see gateway/wake.py. +_SESSION_ORIGIN_TURN_ID: ContextVar = ContextVar("HERMES_ORIGIN_TURN_ID", default=_UNSET) + +# Whether the ORIGINATING api_server request forced background delegations +# to run SYNCHRONOUSLY for this run — the Omnio proxy's ``delegation_sync_only`` +# on ``POST /v1/runs``, bound alongside ``HERMES_ORIGIN_TURN_ID`` for the same +# request. Headless Omnio surfaces (crons, trigger.dev runs) have no channel +# to ever receive a background delegation's wake, so they set this to force +# ``delegate_task(background=True)`` onto its synchronous fallback even when a +# raw session id is bound and would otherwise qualify for the self-post wake +# re-enable (see tools/delegate_tool.py). Stored as "1"/"" (not a real bool) +# because get_session_env() only returns strings, matching every other bridged +# var. Empty on any non-Omnio deployment or when the caller omits the flag. +_SESSION_DELEGATION_SYNC_ONLY: ContextVar = ContextVar( + "HERMES_DELEGATION_SYNC_ONLY", default=_UNSET +) + # Whether the current session's delivery channel can route an ASYNC completion # back to the agent AFTER the current turn ends (i.e. wake a fresh turn). # @@ -136,6 +158,8 @@ def session_context_engaged() -> bool: "HERMES_UI_SESSION_ID": _SESSION_UI_SESSION_ID, "HERMES_SESSION_MESSAGE_ID": _SESSION_MESSAGE_ID, "HERMES_SESSION_PROFILE": _SESSION_PROFILE, + "HERMES_ORIGIN_TURN_ID": _SESSION_ORIGIN_TURN_ID, + "HERMES_DELEGATION_SYNC_ONLY": _SESSION_DELEGATION_SYNC_ONLY, "HERMES_CRON_AUTO_DELIVER_PLATFORM": _CRON_AUTO_DELIVER_PLATFORM, "HERMES_CRON_AUTO_DELIVER_CHAT_ID": _CRON_AUTO_DELIVER_CHAT_ID, "HERMES_CRON_AUTO_DELIVER_THREAD_ID": _CRON_AUTO_DELIVER_THREAD_ID, @@ -173,6 +197,8 @@ def set_session_vars( cwd: str = "", async_delivery: bool = True, ui_session_id: str = "", + origin_turn_id: str = "", + delegation_sync_only: bool = False, ) -> list: """Set all session context variables and return reset tokens. @@ -188,6 +214,16 @@ def set_session_vars( background completion back to the agent after the turn ends (see ``_SESSION_ASYNC_DELIVERY`` / ``async_delivery_supported``). Stateless request/response adapters (the API server) pass ``False``. + + ``origin_turn_id`` is the Omnio product turn id (``turn_id`` on + ``POST /v1/runs``), when the caller has one. Empty on any non-Omnio + entry point. + + ``delegation_sync_only`` is the Omnio proxy's ``delegation_sync_only`` on + ``POST /v1/runs`` — set for headless surfaces (crons, trigger.dev runs) + that have no channel to ever receive a background delegation's wake, so + ``delegate_task(background=True)`` must be forced onto its synchronous + fallback for this run. """ # Mark the session-context machinery engaged for this process. The # subprocess-env bridge uses this to switch from "os.environ fallback" to @@ -209,6 +245,8 @@ def set_session_vars( _SESSION_MESSAGE_ID.set(message_id), _SESSION_PROFILE.set(profile), _SESSION_ASYNC_DELIVERY.set(bool(async_delivery)), + _SESSION_ORIGIN_TURN_ID.set(origin_turn_id), + _SESSION_DELEGATION_SYNC_ONLY.set("1" if delegation_sync_only else ""), ] try: from agent.runtime_cwd import set_session_cwd @@ -244,6 +282,8 @@ def clear_session_vars(tokens: list) -> None: _SESSION_UI_SESSION_ID, _SESSION_MESSAGE_ID, _SESSION_PROFILE, + _SESSION_ORIGIN_TURN_ID, + _SESSION_DELEGATION_SYNC_ONLY, ): var.set("") # Reset async-delivery capability to the "never set" sentinel rather than a diff --git a/gateway/wake.py b/gateway/wake.py index 2518b79601ce..88acfaa4becf 100644 --- a/gateway/wake.py +++ b/gateway/wake.py @@ -8,25 +8,42 @@ — the pre-existing wake path, preserved exactly. * Stateless request/response adapters (the API server, - ``supports_push_delivery = False``): ``handle_message`` would run the wake - turn under a ``build_session_key()``-derived key + ``supports_push_delivery = False``): by default 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. ``handle_message`` cannot be used here: it 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. + (``_bind_api_server_session``), so the wake would land in a parallel, + invisible session. + + OPT-IN redirect (Omnio): when env ``OMNIO_WAKE_HOOK`` is set AND the + completion event carries an ``origin_turn_id``, the wake is instead + delivered by POSTing that URL — the Omnio proxy runs the wake as a real + product turn (persisted for the product UI) instead of a same-pod + self-post the proxy never sees. Falls back to the self-post path when the + hook is set but there is no turn id to attribute the wake to (a wake + without turn identity cannot become a product turn). With the env unset, + behavior is byte-identical to the self-post-only path. Failures RAISE (after bounded retries on transient errors) so callers can -rewind cursors / retry instead of silently losing the event. +rewind cursors / retry instead of silently losing the event — WITH ONE +EXCEPTION: a permanent (non-transient) 4xx from the wake hook — e.g. 404 +when the Omnio proxy no longer recognises ``origin_turn_id`` (conversation +deleted) — raises :class:`WakeHookPermanentError` instead of a plain +``RuntimeError``, so callers that rewind/retry on any exception can +distinguish "try again" from "this will never succeed, drop it" and avoid +redelivering the same unwinnable wake forever. """ from __future__ import annotations import asyncio import logging +import os from typing import Any, Optional logger = logging.getLogger(__name__) @@ -41,6 +58,33 @@ # max_concurrent_runs cap via HTTP 429, which is worth waiting out. _RETRY_DELAYS_SECONDS = (2.0, 5.0, 10.0) +# Opt-in redirect target: when set, non-push wakes are delivered by POSTing +# this URL instead of self-posting /v1/chat/completions. Unset (the default) +# preserves today's self-post-only behavior exactly. +OMNIO_WAKE_HOOK_ENV = "OMNIO_WAKE_HOOK" +# Shared service-token header, same coordinate the Omnio turn-finalize hook +# uses (gateway/platforms/api_server.py:_request_turn_finalize_annotations). +_OMNIO_INTERNAL_TOKEN_ENV = "OMNIO_INTERNAL_TOKEN" +_WAKE_HOOK_TIMEOUT_SECONDS = 30.0 + + +class WakeHookPermanentError(RuntimeError): + """A wake-hook POST failed with a permanent (non-retryable) status. + + Raised for any 4xx from ``OMNIO_WAKE_HOOK`` other than 409/429 (both + treated as transient — see ``_post_wake_hook``) — most notably 404, + which the Omnio proxy returns when ``origin_turn_id`` no longer + resolves to a live conversation (e.g. deleted). Distinct from the plain + ``RuntimeError`` retries give up with, so a caller that would otherwise + rewind/requeue on ANY exception can instead drop the event: retrying a + 404 can never succeed and would redeliver the same wake forever. + """ + + def __init__(self, message: str, *, status_code: int, origin_turn_id: str): + super().__init__(message) + self.status_code = status_code + self.origin_turn_id = origin_turn_id + def adapter_supports_push(adapter: Any) -> bool: """Whether this adapter can push a message to the user after a turn ends. @@ -60,6 +104,9 @@ async def deliver_wake( text: str, session_id: str = "", source: Any = None, + delegation_id: str = "", + origin_turn_id: str = "", + subagent_ids: Optional[list] = None, ) -> None: """Deliver a wake turn to the session behind ``adapter``. @@ -68,6 +115,17 @@ async def deliver_wake( ``SessionSource`` used to build the synthetic event — required for push-capable adapters. + ``delegation_id`` / ``origin_turn_id`` are the completion event's async- + delegation id and Omnio product turn id. ``subagent_ids`` is the list of + the completed children's streamed ``subagentId`` values (see + ``tools.delegate_tool._run_single_child`` / ``entry["subagent_id"]``) — + the identity the Omnio proxy actually persisted per-child rows under, as + opposed to ``delegation_id`` which only identifies the batch. All three + are optional (callers that don't carry them, e.g. the kanban notifier, + simply get the self-post path) and are only consulted on the non-push + branch to decide between the ``OMNIO_WAKE_HOOK`` redirect and the + self-post fallback / to fill the hook payload. + Raises on failure (bad arguments, exhausted retries, HTTP error) so the caller can rewind/retry instead of treating the wake as delivered. """ @@ -92,9 +150,130 @@ async def deliver_wake( "deliver_wake: non-push adapter " "requires the raw session id to self-post the wake turn" ) + + hook_url = os.environ.get(OMNIO_WAKE_HOOK_ENV, "").strip() + if hook_url: + if origin_turn_id: + await _post_wake_hook( + hook_url, + text=text, + session_id=session_id, + delegation_id=delegation_id, + origin_turn_id=origin_turn_id, + subagent_ids=list(subagent_ids) if subagent_ids else [], + ) + return + # A wake without a turn id cannot become a product turn on the Omnio + # side — fall back to the self-post rather than dropping the wake. + logger.warning( + "OMNIO_WAKE_HOOK is set but this wake has no origin_turn_id " + "(session %s, delegation %s); falling back to self-post", + session_id, delegation_id or "", + ) + await _self_post_chat_completion(adapter, text=text, session_id=session_id) +async def _post_wake_hook( + hook_url: str, + *, + text: str, + session_id: str, + delegation_id: str, + origin_turn_id: str, + subagent_ids: Optional[list] = None, +) -> None: + """POST a wake completion to the Omnio wake hook instead of self-posting. + + The hook runs on the Omnio proxy side and turns the wake into a real + product turn (so it is persisted for the product UI, unlike the same-pod + self-post the proxy never observes). Retry/error semantics mirror + ``_self_post_chat_completion``: 429/5xx/connection errors are transient + and retried with the same backoff ladder before giving up; any other 4xx + is a permanent (config/validation) failure and raises immediately. + """ + import aiohttp + + service_token = os.environ.get(_OMNIO_INTERNAL_TOKEN_ENV, "") + if not service_token: + raise RuntimeError( + "OMNIO_WAKE_HOOK is set but OMNIO_INTERNAL_TOKEN is missing: the " + "wake hook requires the shared service token to authenticate — " + "refusing to POST an unauthenticated wake" + ) + + headers = { + "Content-Type": "application/json", + "X-Omnio-Service-Token": service_token, + } + payload = { + "origin_turn_id": origin_turn_id, + "delegation_id": delegation_id, + # The children's streamed subagentId values — the proxy matches + # persisted per-child rows on THESE, not on delegation_id (a batch + # id may coincide with a single child's subagent_id, but the proxy + # must not assume that). + "subagent_ids": list(subagent_ids) if subagent_ids else [], + "session_id": session_id, + "text": text, + } + + 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_HOOK_TIMEOUT_SECONDS) + async with aiohttp.ClientSession(timeout=timeout) as http: + async with http.post(hook_url, json=payload, headers=headers) as resp: + if resp.status in (409, 429) or resp.status >= 500: + # Transient — concurrency/lock contention (409), + # concurrency cap (429), or a server-side hiccup. + last_err = RuntimeError( + f"wake hook POST got HTTP {resp.status} for " + f"turn {origin_turn_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/gone) — fail + # immediately with a typed error so callers can tell + # "unwinnable, drop it" (WakeHookPermanentError) from + # "exhausted transient retries" (plain RuntimeError + # below) apart. 404 is the expected shape here: the + # proxy returns it when origin_turn_id no longer + # resolves to a live conversation (e.g. deleted). + raise WakeHookPermanentError( + f"wake hook POST failed for turn {origin_turn_id}: " + f"HTTP {resp.status}: {body}", + status_code=resp.status, + origin_turn_id=origin_turn_id, + ) + await resp.read() + logger.info( + "wake hook delivered for turn %s (delegation %s, " + "attempt %d)", + origin_turn_id, delegation_id or "", attempt + 1, + ) + return + except (aiohttp.ClientError, asyncio.TimeoutError, OSError) as exc: + last_err = exc + logger.warning( + "wake hook POST transient failure for turn %s " + "(attempt %d/%d): %s", + origin_turn_id, attempt + 1, attempts, exc, + ) + continue + raise RuntimeError( + f"wake hook POST gave up for turn {origin_turn_id} after " + f"{attempts} attempts: {last_err}" + ) from last_err + + async def _self_post_chat_completion( adapter: Any, *, text: str, session_id: str ) -> None: diff --git a/tests/gateway/test_api_server_omnio_turn_event_log.py b/tests/gateway/test_api_server_omnio_turn_event_log.py index 84200aa54687..0b2e38322660 100644 --- a/tests/gateway/test_api_server_omnio_turn_event_log.py +++ b/tests/gateway/test_api_server_omnio_turn_event_log.py @@ -2863,3 +2863,61 @@ def run(**_kwargs: Any) -> Dict[str, Any]: assert captured["surface"] is True assert captured["answer"] == "Option B" + + +@pytest.mark.asyncio +async def test_runs_thread_delegation_sync_only_into_session_context() -> None: + """The Omnio proxy's ``delegation_sync_only`` on ``POST /v1/runs`` must + reach the running agent via ``HERMES_DELEGATION_SYNC_ONLY`` — bound by + ``_bind_api_server_session`` exactly like ``turn_id`` -> ``HERMES_ORIGIN_TURN_ID`` + — so ``tools.async_delegation._current_delegation_sync_only()`` (read by + ``delegate_task``) sees it while the run is live.""" + from gateway.session_context import get_session_env + + adapter = _make_adapter() + captured: Dict[str, Any] = {} + + def run(**_kwargs: Any) -> Dict[str, Any]: + captured["delegation_sync_only"] = get_session_env( + "HERMES_DELEGATION_SYNC_ONLY" + ) + return {"final_response": "done", "messages": []} + + with patch.object(adapter, "_create_agent", return_value=_agent(run)): + started, _events = await _run_without_http_server( + adapter, + { + "input": "run headless", + "session_id": "session-sync-only", + "delegation_sync_only": True, + }, + ) + + assert started.status == 202 + assert captured["delegation_sync_only"] == "1" + + +@pytest.mark.asyncio +async def test_runs_default_delegation_sync_only_false_when_omitted() -> None: + """Callers that never pass ``delegation_sync_only`` (every non-Omnio and + most Omnio deployments) must not force delegate_task's synchronous + fallback.""" + from gateway.session_context import get_session_env + + adapter = _make_adapter() + captured: Dict[str, Any] = {} + + def run(**_kwargs: Any) -> Dict[str, Any]: + captured["delegation_sync_only"] = get_session_env( + "HERMES_DELEGATION_SYNC_ONLY" + ) + return {"final_response": "done", "messages": []} + + with patch.object(adapter, "_create_agent", return_value=_agent(run)): + started, _events = await _run_without_http_server( + adapter, + {"input": "run normally", "session_id": "session-normal"}, + ) + + assert started.status == 202 + assert captured["delegation_sync_only"] == "" diff --git a/tests/gateway/test_api_server_session_delegations.py b/tests/gateway/test_api_server_session_delegations.py new file mode 100644 index 000000000000..7c1ed99faa5f --- /dev/null +++ b/tests/gateway/test_api_server_session_delegations.py @@ -0,0 +1,148 @@ +"""Contract tests for ``GET /api/sessions/{session_id}/delegations``. + +The endpoint filters the process-wide async-delegation registry by +``origin_session_id`` so an external UI can rebuild a session's outstanding +background work from the process that owns the children. These tests pin the +contract the Omnio proxy consumes: a ``data`` list, session-scoped filtering, +live-status fields passed through, and auth. +""" + +from __future__ import annotations + +from unittest.mock import patch + +import pytest +from aiohttp import web +from aiohttp.test_utils import TestClient, TestServer + +from gateway.config import PlatformConfig +from gateway.platforms.api_server import APIServerAdapter + + +def _make_adapter(*, api_key: str = "") -> APIServerAdapter: + return APIServerAdapter(PlatformConfig(enabled=True, extra={"key": api_key})) + + +def _make_app(adapter: APIServerAdapter) -> web.Application: + app = web.Application() + app.router.add_get( + "/api/sessions/{session_id}/delegations", + adapter._handle_session_delegations, + ) + return app + + +async def _get(client: TestClient, session_id: str, headers=None): + return await client.get( + f"/api/sessions/{session_id}/delegations", headers=headers or {} + ) + + +_RECORDS = [ + { + "delegation_id": "d_1", + "origin_session_id": "sess-a", + "origin_turn_id": "turn-1", + "goal": "research pricing", + "status": "running", + "children_activity": [ + { + "api_calls": 3, + "current_tool": "web_read", + "seconds_since_activity": 1.2, + "subagent_id": "sub_1", + "finished": False, + } + ], + }, + { + "delegation_id": "d_2", + "origin_session_id": "sess-b", + "origin_turn_id": "turn-9", + "goal": "unrelated", + "status": "running", + }, + { + "delegation_id": "d_3", + # Records predating the origin_session_id column surface it as "". + "origin_session_id": "", + "origin_turn_id": "", + "goal": "orphaned", + "status": "completed", + }, +] + + +@pytest.mark.asyncio +async def test_filters_to_requested_session(): + adapter = _make_adapter() + async with TestClient(TestServer(_make_app(adapter))) as client: + with patch( + "tools.async_delegation.list_async_delegations", return_value=_RECORDS + ): + resp = await _get(client, "sess-a") + assert resp.status == 200 + body = await resp.json() + assert [d["delegation_id"] for d in body["data"]] == ["d_1"] + # Live-status fields ride through untouched — the proxy reads them as-is. + child = body["data"][0]["children_activity"][0] + assert child["subagent_id"] == "sub_1" + assert child["finished"] is False + assert body["data"][0]["origin_turn_id"] == "turn-1" + + +@pytest.mark.asyncio +async def test_unknown_session_returns_empty_data(): + adapter = _make_adapter() + async with TestClient(TestServer(_make_app(adapter))) as client: + with patch( + "tools.async_delegation.list_async_delegations", return_value=_RECORDS + ): + resp = await _get(client, "sess-nope") + assert resp.status == 200 + assert (await resp.json())["data"] == [] + + +@pytest.mark.asyncio +async def test_sessionless_records_never_match(): + """An empty origin_session_id must not leak into any session's listing.""" + adapter = _make_adapter() + async with TestClient(TestServer(_make_app(adapter))) as client: + with patch( + "tools.async_delegation.list_async_delegations", return_value=_RECORDS + ): + resp = await _get(client, "") + # aiohttp may 404 an empty path segment; what matters is that a + # sessionless record (origin_session_id="") can never 200 as data. + assert resp.status in (200, 404) + if resp.status == 200: + assert (await resp.json())["data"] == [] + + +@pytest.mark.asyncio +async def test_registry_failure_returns_500(): + adapter = _make_adapter() + async with TestClient(TestServer(_make_app(adapter))) as client: + with patch( + "tools.async_delegation.list_async_delegations", + side_effect=RuntimeError("boom"), + ): + resp = await _get(client, "sess-a") + assert resp.status == 500 + + +@pytest.mark.asyncio +async def test_auth_required_when_key_configured(): + adapter = _make_adapter(api_key="omnio-test-key") + async with TestClient(TestServer(_make_app(adapter))) as client: + with patch( + "tools.async_delegation.list_async_delegations", return_value=_RECORDS + ): + denied = await _get(client, "sess-a") + allowed = await _get( + client, + "sess-a", + headers={"Authorization": "Bearer omnio-test-key"}, + ) + assert denied.status == 401 + assert allowed.status == 200 diff --git a/tests/gateway/test_async_delivery_capability.py b/tests/gateway/test_async_delivery_capability.py index 188bcdd12c2f..16b02779616b 100644 --- a/tests/gateway/test_async_delivery_capability.py +++ b/tests/gateway/test_async_delivery_capability.py @@ -338,3 +338,135 @@ def test_cli_stays_supported(self): assert not d.get("notify_unsupported") # No platform bound -> no gateway watcher, but completion_queue still fires. assert len(process_registry.pending_watchers) == 0 + + +# --------------------------------------------------------------------------- +# HERMES_ORIGIN_TURN_ID — bound alongside chat_id, empty on non-Omnio paths +# --------------------------------------------------------------------------- + + +class TestOriginTurnIdBinding: + """The Omnio ``turn_id`` bound by ``_bind_api_server_session`` follows + the exact same lifecycle as ``chat_id``: set by ``set_session_vars``, + readable via ``get_session_env``, "" (not the os.environ fallback) once + ``clear_session_vars`` runs, and back to "never bound" after + ``reset_session_vars``.""" + + def test_set_and_read(self): + tokens = set_session_vars( + platform="api_server", + chat_id="sess1", + session_id="sess1", + origin_turn_id="turn-abc", + ) + try: + assert get_session_env("HERMES_ORIGIN_TURN_ID") == "turn-abc" + finally: + clear_session_vars(tokens) + + def test_defaults_empty_when_not_passed(self): + """Non-Omnio callers never pass origin_turn_id — must read "" rather + than raising or falling back to a stale value.""" + tokens = set_session_vars(platform="telegram", chat_id="123") + try: + assert get_session_env("HERMES_ORIGIN_TURN_ID") == "" + finally: + clear_session_vars(tokens) + + def test_cleared_to_empty_not_environ_fallback(self, monkeypatch): + monkeypatch.setenv("HERMES_ORIGIN_TURN_ID", "leaked-from-os-environ") + tokens = set_session_vars( + platform="api_server", chat_id="sess1", origin_turn_id="turn-xyz", + ) + clear_session_vars(tokens) + # Explicitly cleared ("") — must NOT fall back to os.environ, exactly + # like every other _VAR_MAP-mapped session var. + assert get_session_env("HERMES_ORIGIN_TURN_ID") == "" + + def test_reset_restores_environ_fallback(self, monkeypatch): + monkeypatch.setenv("HERMES_ORIGIN_TURN_ID", "cli-env-value") + reset_session_vars() + assert get_session_env("HERMES_ORIGIN_TURN_ID") == "cli-env-value" + + +# --------------------------------------------------------------------------- +# HERMES_DELEGATION_SYNC_ONLY — bound alongside chat_id, empty on non-Omnio +# paths. Mirrors TestOriginTurnIdBinding exactly (same lifecycle, same +# _bind_api_server_session chokepoint). +# --------------------------------------------------------------------------- + + +class TestDelegationSyncOnlyBinding: + """The Omnio ``delegation_sync_only`` flag bound by + ``_bind_api_server_session`` follows the exact same lifecycle as + ``chat_id``/``origin_turn_id``: set by ``set_session_vars``, readable via + ``get_session_env``, "" (not the os.environ fallback) once + ``clear_session_vars`` runs, and back to "never bound" after + ``reset_session_vars``.""" + + def test_set_and_read(self): + tokens = set_session_vars( + platform="api_server", + chat_id="sess1", + session_id="sess1", + delegation_sync_only=True, + ) + try: + assert get_session_env("HERMES_DELEGATION_SYNC_ONLY") == "1" + finally: + clear_session_vars(tokens) + + def test_defaults_empty_when_not_passed(self): + """Non-Omnio callers (and Omnio callers that omit the flag) never + pass delegation_sync_only — must read "" rather than raising or + falling back to a stale value.""" + tokens = set_session_vars(platform="telegram", chat_id="123") + try: + assert get_session_env("HERMES_DELEGATION_SYNC_ONLY") == "" + finally: + clear_session_vars(tokens) + + def test_cleared_to_empty_not_environ_fallback(self, monkeypatch): + monkeypatch.setenv( + "HERMES_DELEGATION_SYNC_ONLY", "leaked-from-os-environ" + ) + tokens = set_session_vars( + platform="api_server", chat_id="sess1", delegation_sync_only=True, + ) + clear_session_vars(tokens) + # Explicitly cleared ("") — must NOT fall back to os.environ, exactly + # like every other _VAR_MAP-mapped session var. + assert get_session_env("HERMES_DELEGATION_SYNC_ONLY") == "" + + def test_reset_restores_environ_fallback(self, monkeypatch): + monkeypatch.setenv("HERMES_DELEGATION_SYNC_ONLY", "1") + reset_session_vars() + assert get_session_env("HERMES_DELEGATION_SYNC_ONLY") == "1" + + def test_bind_chokepoint_threads_the_flag(self): + """``APIServerAdapter._bind_api_server_session`` is the SINGLE + chokepoint every API-server agent-entry path uses — verify it + threads delegation_sync_only exactly like origin_turn_id.""" + from gateway.platforms.api_server import APIServerAdapter + + tokens = APIServerAdapter._bind_api_server_session( + chat_id="c1", + session_key="sk1", + session_id="sid1", + delegation_sync_only=True, + ) + try: + assert get_session_env("HERMES_DELEGATION_SYNC_ONLY") == "1" + finally: + clear_session_vars(tokens) + + def test_bind_chokepoint_defaults_false(self): + from gateway.platforms.api_server import APIServerAdapter + + tokens = APIServerAdapter._bind_api_server_session( + chat_id="c1", session_key="sk1", session_id="sid1", + ) + try: + assert get_session_env("HERMES_DELEGATION_SYNC_ONLY") == "" + finally: + clear_session_vars(tokens) diff --git a/tests/gateway/test_completion_delivery.py b/tests/gateway/test_completion_delivery.py index 4face3841b27..0091573fb81c 100644 --- a/tests/gateway/test_completion_delivery.py +++ b/tests/gateway/test_completion_delivery.py @@ -131,6 +131,48 @@ def test_unroutable_async_event_is_not_requeued_forever( assert isolated.empty() +def test_wake_hook_404_drops_event_without_requeue_loop( + monkeypatch, isolated_registry, caplog, +): + """A permanent 4xx (404 — the Omnio proxy no longer recognises + origin_turn_id, e.g. the conversation was deleted) from OMNIO_WAKE_HOOK + must be treated as CONSUMED, not requeued: retrying a 404 can never + succeed, and returning False here would redeliver the same unwinnable + wake forever (see the twin api_server self-post branches in + gateway.run._inject_watch_notification).""" + import gateway.wake as wake_module + + async def _raise_404(*_a, **_k): + raise wake_module.WakeHookPermanentError( + "wake hook POST failed for turn turn-deleted: HTTP 404: gone", + status_code=404, + origin_turn_id="turn-deleted", + ) + + monkeypatch.setattr(wake_module, "deliver_wake", _raise_404) + + # A non-push api_server-shaped adapter with an unparseable session_key, + # so _inject_watch_notification takes the no-routing-metadata self-post + # branch (mirrors the raw-session-id api_server shape). + adapter = SimpleNamespace(supports_async_delivery=False) + runner = _runner(adapter) + runner.adapters = {Platform.API_SERVER: adapter} + + event = _async_event("deleg_gone_turn") + event["session_key"] = "raw-sid-gone" + event["origin_session_id"] = "raw-sid-gone" + event["origin_turn_id"] = "turn-deleted" + + with caplog.at_level("WARNING"): + delivered = asyncio.run( + runner._deliver_completion_notification("wake text", dict(event)) + ) + + assert delivered is True # consumed, not requeued + assert any("wake_dropped" in r.message for r in caplog.records) + assert any("404" in r.message for r in caplog.records) + + def test_concurrent_claims_share_the_same_narrow_delivery_seam(): """Concurrent consumers in one runner cannot both enter the adapter.""" entered = asyncio.Event() diff --git a/tests/gateway/test_wake_delivery.py b/tests/gateway/test_wake_delivery.py index 9f248a5fd395..6ddc6fa5a15b 100644 --- a/tests/gateway/test_wake_delivery.py +++ b/tests/gateway/test_wake_delivery.py @@ -91,10 +91,15 @@ def test_deliver_wake_non_push_requires_api_key(): async def _serve(handler): """Spin an in-process aiohttp server on an ephemeral loopback port.""" + return await _serve_at(handler, "/v1/chat/completions") + + +async def _serve_at(handler, path): + """Spin an in-process aiohttp server exposing ``handler`` at ``path``.""" from aiohttp import web app = web.Application() - app.router.add_post("/v1/chat/completions", handler) + app.router.add_post(path, handler) runner = web.AppRunner(app) await runner.setup() site = web.TCPSite(runner, "127.0.0.1", 0) @@ -194,3 +199,246 @@ def test_deliver_wake_raises_after_exhausted_retries(monkeypatch): 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")) + + +# --------------------------------------------------------------------------- +# OMNIO_WAKE_HOOK redirect (opt-in, default OFF) +# --------------------------------------------------------------------------- + + +def test_deliver_wake_unset_env_is_byte_identical_to_self_post(monkeypatch): + """With OMNIO_WAKE_HOOK unset, passing the new optional kwargs must not + change the self-post wire contract at all — default-off is sacred.""" + from aiohttp import web + + monkeypatch.delenv("OMNIO_WAKE_HOOK", raising=False) + seen = {} + + async def handler(request): + seen["session_id"] = request.headers.get("X-Hermes-Session-Id") + seen["body"] = await request.json() + return web.json_response({"choices": []}) + + async def run(): + runner, port = await _serve(handler) + try: + adapter = ApiServerLikeAdapter(port=port, key="sekrit") + await deliver_wake( + adapter, + text="task done", + session_id="raw-sid-99", + delegation_id="deleg_1", + origin_turn_id="turn_1", + subagent_ids=["sa-0-abc"], + ) + finally: + await runner.cleanup() + + asyncio.run(run()) + assert seen["session_id"] == "raw-sid-99" + assert seen["body"] == { + "model": "hermes", + "messages": [{"role": "user", "content": "task done"}], + "stream": False, + } + + +def test_deliver_wake_redirects_to_hook_when_env_set(monkeypatch): + """OMNIO_WAKE_HOOK + a non-empty origin_turn_id redirects the wake to + the hook URL instead of self-posting — asserts the exact payload shape + and the service-token header.""" + from aiohttp import web + + seen = {} + + async def handler(request): + seen["headers"] = dict(request.headers) + seen["body"] = await request.json() + return web.json_response({"ok": True}) + + async def run(): + runner, port = await _serve_at(handler, "/omnio/wake") + try: + monkeypatch.setenv("OMNIO_WAKE_HOOK", f"http://127.0.0.1:{port}/omnio/wake") + monkeypatch.setenv("OMNIO_INTERNAL_TOKEN", "svc-tok-1") + # host/port point at a port nothing listens on — if the redirect + # didn't happen, the self-post would fail loudly, not silently + # produce this test's expected payload. + adapter = ApiServerLikeAdapter(host="127.0.0.1", port=1, key="k") + await deliver_wake( + adapter, + text="subagent finished", + session_id="raw-sid-7", + delegation_id="deleg_42", + origin_turn_id="turn_777", + subagent_ids=["sa-0-aaa", "sa-1-bbb"], + ) + finally: + await runner.cleanup() + + asyncio.run(run()) + assert seen["headers"]["X-Omnio-Service-Token"] == "svc-tok-1" + assert seen["headers"]["Content-Type"] == "application/json" + assert seen["body"] == { + "origin_turn_id": "turn_777", + "delegation_id": "deleg_42", + "subagent_ids": ["sa-0-aaa", "sa-1-bbb"], + "session_id": "raw-sid-7", + "text": "subagent finished", + } + + +def test_deliver_wake_falls_back_to_self_post_without_origin_turn_id(monkeypatch): + """OMNIO_WAKE_HOOK set but no origin_turn_id (non-Omnio-attributable + wake) must fall back to the self-post — a wake with no turn identity + can never become a product turn.""" + from aiohttp import web + + seen = {} + + async def hook_handler(request): + seen["hook_hit"] = True + return web.json_response({"ok": True}) + + async def self_post_handler(request): + seen["self_post_hit"] = True + seen["session_id"] = request.headers.get("X-Hermes-Session-Id") + return web.json_response({"choices": []}) + + async def run(): + hook_runner, hook_port = await _serve_at(hook_handler, "/omnio/wake") + self_runner, self_port = await _serve(self_post_handler) + try: + monkeypatch.setenv( + "OMNIO_WAKE_HOOK", f"http://127.0.0.1:{hook_port}/omnio/wake" + ) + monkeypatch.setenv("OMNIO_INTERNAL_TOKEN", "svc-tok-1") + adapter = ApiServerLikeAdapter(port=self_port, key="sekrit") + await deliver_wake( + adapter, text="x", session_id="raw-sid-3", delegation_id="deleg_x", + ) + finally: + await hook_runner.cleanup() + await self_runner.cleanup() + + asyncio.run(run()) + assert "hook_hit" not in seen + assert seen.get("self_post_hit") is True + assert seen["session_id"] == "raw-sid-3" + + +def test_deliver_wake_hook_requires_service_token(monkeypatch): + """A hook set with no OMNIO_INTERNAL_TOKEN is a misconfiguration — fail + loudly rather than POST an unauthenticated wake.""" + monkeypatch.setenv("OMNIO_WAKE_HOOK", "http://127.0.0.1:1/omnio/wake") + monkeypatch.delenv("OMNIO_INTERNAL_TOKEN", raising=False) + adapter = ApiServerLikeAdapter(host="127.0.0.1", port=1, key="k") + with pytest.raises(RuntimeError, match="OMNIO_INTERNAL_TOKEN"): + asyncio.run( + deliver_wake( + adapter, text="x", session_id="sid", origin_turn_id="turn_1", + ) + ) + + +def test_deliver_wake_hook_retries_409_then_succeeds(monkeypatch): + """409 (lock/concurrency contention on the proxy side) is transient — + retried with the same backoff ladder as the self-post path.""" + 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": "locked"}, status=409) + return web.json_response({"ok": True}) + + async def run(): + runner, port = await _serve_at(handler, "/omnio/wake") + try: + monkeypatch.setenv("OMNIO_WAKE_HOOK", f"http://127.0.0.1:{port}/omnio/wake") + monkeypatch.setenv("OMNIO_INTERNAL_TOKEN", "tok") + adapter = ApiServerLikeAdapter(host="127.0.0.1", port=1, key="k") + await deliver_wake( + adapter, text="x", session_id="sid", origin_turn_id="turn_1", + ) + finally: + await runner.cleanup() + + asyncio.run(run()) + assert calls["n"] == 2 + + +def test_deliver_wake_hook_404_is_permanent_no_retry_loop(monkeypatch): + """404 (the proxy no longer recognises origin_turn_id — e.g. the + conversation was deleted) is PERMANENT: a single attempt, a typed + WakeHookPermanentError, and no retry loop — retrying can never + succeed.""" + from aiohttp import web + + import gateway.wake as wake_mod + from gateway.wake import WakeHookPermanentError + + monkeypatch.setattr(wake_mod, "_RETRY_DELAYS_SECONDS", (0.01, 0.01, 0.01)) + calls = {"n": 0} + + async def handler(request): + calls["n"] += 1 + return web.json_response({"error": "unknown turn"}, status=404) + + async def run(): + runner, port = await _serve_at(handler, "/omnio/wake") + try: + monkeypatch.setenv("OMNIO_WAKE_HOOK", f"http://127.0.0.1:{port}/omnio/wake") + monkeypatch.setenv("OMNIO_INTERNAL_TOKEN", "tok") + adapter = ApiServerLikeAdapter(host="127.0.0.1", port=1, key="k") + with pytest.raises(WakeHookPermanentError) as excinfo: + await deliver_wake( + adapter, + text="x", + session_id="sid", + origin_turn_id="turn-deleted", + ) + assert excinfo.value.status_code == 404 + assert excinfo.value.origin_turn_id == "turn-deleted" + finally: + await runner.cleanup() + + asyncio.run(run()) + assert calls["n"] == 1 # exactly one attempt — no retry loop + + +def test_deliver_wake_hook_permanent_400_raises_immediately(monkeypatch): + """Any other permanent 4xx also raises WakeHookPermanentError, not the + generic RuntimeError the self-post path uses for the same class of + error — callers need to tell "unwinnable" apart from "exhausted + retries".""" + from aiohttp import web + + from gateway.wake import WakeHookPermanentError + + calls = {"n": 0} + + async def handler(request): + calls["n"] += 1 + return web.json_response({"error": "bad request"}, status=400) + + async def run(): + runner, port = await _serve_at(handler, "/omnio/wake") + try: + monkeypatch.setenv("OMNIO_WAKE_HOOK", f"http://127.0.0.1:{port}/omnio/wake") + monkeypatch.setenv("OMNIO_INTERNAL_TOKEN", "tok") + adapter = ApiServerLikeAdapter(host="127.0.0.1", port=1, key="k") + with pytest.raises(WakeHookPermanentError): + await deliver_wake( + adapter, text="x", session_id="sid", origin_turn_id="turn_1", + ) + finally: + await runner.cleanup() + + asyncio.run(run()) + assert calls["n"] == 1 diff --git a/tests/tools/test_async_delegation.py b/tests/tools/test_async_delegation.py index 1d7d92b40e8b..a97b49d4ef93 100644 --- a/tests/tools/test_async_delegation.py +++ b/tests/tools/test_async_delegation.py @@ -12,6 +12,7 @@ import sys import threading import time +from types import SimpleNamespace import pytest @@ -441,6 +442,37 @@ def test_list_async_delegations_exposes_live_activity(monkeypatch): gate.set() +def test_list_async_delegations_stamps_per_child_goal(monkeypatch): + """Each children_activity entry must carry its OWN goal from the batch's + goals list (dispatch order), not leave consumers to fall back to the + combined batch-level goal string.""" + monkeypatch.setattr(ad, "_STALE_CHECK_INTERVAL", 0.03) + gate = threading.Event() + ts = time.time() + + res = ad.dispatch_async_delegation_batch( + goals=["research competitor A", "research competitor B"], + context=None, toolsets=None, role="leaf", model="m", session_key="", + max_async_children=2, + runner=lambda: {} if gate.wait(timeout=10) else {}, + progress_fn=lambda: ( + ((1, "web_search", ts, "sub-a", False), (2, "web_read", ts, "sub-b", False)), + True, + ), + ) + try: + assert res["status"] == "dispatched" + item = next( + d for d in ad.list_async_delegations() + if d["delegation_id"] == res["delegation_id"] + ) + first, second = item["children_activity"] + assert first["goal"] == "research competitor A" + assert second["goal"] == "research competitor B" + finally: + gate.set() + + def test_stalled_batch_is_interrupted_then_finalized(monkeypatch): _fast_stale_monitor(monkeypatch) gate = threading.Event() @@ -1198,11 +1230,14 @@ def fake_dispatch(**kwargs): # The dispatch wires a live progress sampler over the child agents so the # async registry's stale monitor can watch the detached batch. The token # includes last_activity_ts so streamed chunks count as liveness (each - # chunk ticks _touch_activity), not just completed API calls. + # chunk ticks _touch_activity), not just completed API calls, plus the + # child's stable subagent_id (its streamed "subagentId") so an opt-in + # progress report can attribute an update to the exact child, and a + # per-child finished flag so that report can announce ONE child ending. progress_fn = captured["progress_fn"] assert callable(progress_fn) token, in_tool = progress_fn() - assert token == ((4, "terminal", 1234.5),) + assert token == ((4, "terminal", 1234.5, "s1", False),) assert in_tool is True @@ -1454,3 +1489,343 @@ def test_gateway_cli_origin_event_left_unrouted(): runner._enrich_async_delegation_routing(evt) assert "platform" not in evt + +# --------------------------------------------------------------------------- +# origin_turn_id / subagent_id threading through the completion event +# --------------------------------------------------------------------------- + + +def test_completion_event_carries_origin_turn_id_and_subagent_ids(): + """origin_turn_id and the batch's per-child subagent_ids must reach the + completion event, mirroring origin_session_id's existing contract.""" + res = ad.dispatch_async_delegation_batch( + goals=["one task"], context=None, toolsets=None, role="leaf", + model="m", session_key="", origin_session_id="raw-sid", + origin_turn_id="turn-batch-1", + runner=lambda: { + "results": [ + {"task_index": 0, "status": "completed", "summary": "ok", + "subagent_id": "sa-0-abc"}, + ], + "total_duration_seconds": 0.1, + }, + ) + assert res["status"] == "dispatched" + + evt = _drain_for(res["delegation_id"], timeout=5.0) + assert evt is not None + assert evt["origin_turn_id"] == "turn-batch-1" + assert evt["subagent_ids"] == ["sa-0-abc"] + + +def test_completion_event_origin_turn_id_defaults_empty(): + """Non-Omnio callers that never pass origin_turn_id get "" — never a + missing key, so downstream .get() callers keep working unconditionally.""" + res = ad.dispatch_async_delegation_batch( + goals=["one task"], context=None, toolsets=None, role="leaf", + model="m", session_key="", + runner=lambda: {"results": [{"task_index": 0, "status": "completed", "summary": "ok"}]}, + ) + evt = _drain_for(res["delegation_id"], timeout=5.0) + assert evt is not None + assert evt["origin_turn_id"] == "" + assert evt["subagent_ids"] == [] + + +# --------------------------------------------------------------------------- +# Opt-in progress reporting (OMNIO_SUBAGENT_PROGRESS_HOOK) — default OFF +# --------------------------------------------------------------------------- + + +def _fast_progress_monitor(monkeypatch, interval=0.05): + monkeypatch.setattr(ad, "_PROGRESS_REPORT_INTERVAL", interval) + + +def test_progress_reporter_off_by_default_no_thread_started(monkeypatch): + """Without OMNIO_SUBAGENT_PROGRESS_HOOK, dispatching with a progress_fn + must never spin up the progress-monitor thread — default-off is + sacred.""" + monkeypatch.delenv("OMNIO_SUBAGENT_PROGRESS_HOOK", raising=False) + _fast_progress_monitor(monkeypatch) + gate = threading.Event() + + res = ad.dispatch_async_delegation( + goal="off by default", context=None, toolsets=None, role="leaf", + model="m", session_key="", origin_turn_id="turn-x", + runner=lambda: (gate.wait(timeout=5), {"status": "completed", "summary": "done"})[1], + progress_fn=lambda: (((1, "terminal", time.time(), "sa-0"),), True), + ) + assert res["status"] == "dispatched" + time.sleep(0.2) + assert ad._progress_monitor_thread is None + + gate.set() + assert _drain_for(res["delegation_id"], timeout=5.0) is not None + + +def test_progress_reporter_posts_per_child_payload(monkeypatch): + """When the hook env is set, a running delegation with an + origin_turn_id reports progress carrying the emitting child's + subagent_id — the same identity it streams as subagentId — plus a + human label/detail, to the configured hook URL with the service-token + header.""" + monkeypatch.setenv("OMNIO_SUBAGENT_PROGRESS_HOOK", "http://example.invalid/progress") + monkeypatch.setenv("OMNIO_INTERNAL_TOKEN", "svc-tok") + _fast_progress_monitor(monkeypatch) + gate = threading.Event() + posted = [] + post_seen = threading.Event() + + def fake_post(url, json=None, headers=None, timeout=None): + posted.append({"url": url, "json": json, "headers": headers}) + post_seen.set() + return SimpleNamespace(status_code=200) + + monkeypatch.setattr("requests.post", fake_post) + + res = ad.dispatch_async_delegation( + goal="progress child", context=None, toolsets=None, role="leaf", + model="m", session_key="", origin_turn_id="turn-progress-1", + runner=lambda: (gate.wait(timeout=5), {"status": "completed", "summary": "done"})[1], + progress_fn=lambda: (((2, "terminal", time.time(), "sa-0-xyz"),), True), + ) + assert res["status"] == "dispatched" + + assert post_seen.wait(timeout=3.0) + gate.set() + evt = _drain_for(res["delegation_id"], timeout=5.0) + assert evt is not None + + assert posted, "progress hook was never POSTed to" + call = posted[0] + assert call["url"] == "http://example.invalid/progress" + assert call["headers"]["X-Omnio-Service-Token"] == "svc-tok" + body = call["json"] + assert body["origin_turn_id"] == "turn-progress-1" + assert body["delegation_id"] == res["delegation_id"] + assert body["subagent_id"] == "sa-0-xyz" + assert body["status"] == "running" + assert body["progress"]["label"] == "progress child" + assert body["progress"]["detail"] == "terminal" + + +def test_progress_reporter_hook_failure_never_raises_into_delegation(monkeypatch): + """A hook returning 500 (or raising) is strictly best-effort: it must + never affect the delegation's own completion event or crash the worker + thread.""" + monkeypatch.setenv("OMNIO_SUBAGENT_PROGRESS_HOOK", "http://example.invalid/progress") + monkeypatch.setenv("OMNIO_INTERNAL_TOKEN", "svc-tok") + _fast_progress_monitor(monkeypatch) + gate = threading.Event() + post_seen = threading.Event() + + def failing_post(url, json=None, headers=None, timeout=None): + post_seen.set() + return SimpleNamespace(status_code=500) + + monkeypatch.setattr("requests.post", failing_post) + + res = ad.dispatch_async_delegation( + goal="progress child failing hook", context=None, toolsets=None, + role="leaf", model="m", session_key="", origin_turn_id="turn-progress-2", + runner=lambda: (gate.wait(timeout=5), {"status": "completed", "summary": "done"})[1], + progress_fn=lambda: (((1, "terminal", time.time(), "sa-1"),), True), + ) + assert res["status"] == "dispatched" + assert post_seen.wait(timeout=3.0) + + gate.set() + evt = _drain_for(res["delegation_id"], timeout=5.0) + assert evt is not None + assert evt["status"] == "completed" # unaffected by the failing hook + + +def test_progress_reporter_skips_delegations_without_origin_turn_id(monkeypatch): + """A delegation with no origin_turn_id (non-Omnio deployment) is never + reported — there is nothing on the product side to attribute it to.""" + monkeypatch.setenv("OMNIO_SUBAGENT_PROGRESS_HOOK", "http://example.invalid/progress") + monkeypatch.setenv("OMNIO_INTERNAL_TOKEN", "svc-tok") + _fast_progress_monitor(monkeypatch) + gate = threading.Event() + posted = [] + + def fake_post(url, json=None, headers=None, timeout=None): + posted.append(json) + return SimpleNamespace(status_code=200) + + monkeypatch.setattr("requests.post", fake_post) + + res = ad.dispatch_async_delegation( + goal="no turn id", context=None, toolsets=None, role="leaf", + model="m", session_key="", # no origin_turn_id + runner=lambda: (gate.wait(timeout=5), {"status": "completed", "summary": "done"})[1], + progress_fn=lambda: (((1, "terminal", time.time(), "sa-2"),), True), + ) + assert res["status"] == "dispatched" + time.sleep(0.25) # several sweeps at the shrunk interval + gate.set() + assert _drain_for(res["delegation_id"], timeout=5.0) is not None + assert posted == [] + + + +# --------------------------------------------------------------------------- +# Per-child completion ticks — one terminal update per finished child +# --------------------------------------------------------------------------- + + +def test_terminal_tick_emitted_for_finished_child_while_siblings_run(monkeypatch): + """A batch's only completion signal is the delegation-level event, which + fires once ALL children are done. A child that finishes on its own must + still get a terminal update carrying its own subagent_id, while its + still-running siblings keep reporting 'running'.""" + monkeypatch.setenv("OMNIO_SUBAGENT_PROGRESS_HOOK", "http://example.invalid/progress") + monkeypatch.setenv("OMNIO_INTERNAL_TOKEN", "svc-tok") + _fast_progress_monitor(monkeypatch) + gate = threading.Event() + posted = [] + post_seen = threading.Event() + + def fake_post(url, json=None, headers=None, timeout=None): + posted.append(json) + post_seen.set() + return SimpleNamespace(status_code=200) + + monkeypatch.setattr("requests.post", fake_post) + + def progress_fn(): + # Child A has returned (finished=True); child B is still in a tool. + return ( + ( + (3, None, time.time(), "sa-a", True), + (1, "terminal", time.time(), "sa-b", False), + ), + True, + ) + + res = ad.dispatch_async_delegation_batch( + goals=["goal a", "goal b"], context=None, toolsets=None, role="leaf", + model="m", session_key="", origin_turn_id="turn-child-done", + runner=lambda: (gate.wait(timeout=5), {"results": []})[1], + progress_fn=progress_fn, + ) + assert res["status"] == "dispatched" + assert post_seen.wait(timeout=3.0) + time.sleep(0.2) # let more sweeps run + gate.set() + assert _drain_for(res["delegation_id"], timeout=5.0) is not None + + done = [p for p in posted if p["status"] == "completed"] + assert len(done) == 1, f"expected exactly one terminal tick, got {done}" + assert done[0]["subagent_id"] == "sa-a" + assert done[0]["origin_turn_id"] == "turn-child-done" + assert done[0]["delegation_id"] == res["delegation_id"] + assert done[0]["progress"]["label"] == "goal a" + assert done[0]["progress"]["detail"] is None + + # The sibling is unaffected — still reported as running, every sweep. + sibling = [p for p in posted if p["subagent_id"] == "sa-b"] + assert sibling, "sibling stopped being reported" + assert all(p["status"] == "running" for p in sibling) + + +def test_terminal_tick_flushed_for_the_last_child(monkeypatch): + """The last child's flip lands together with the batch's own completion, + inside a sweep interval — the finalize-time flush is what keeps it from + being dropped. Sweeps are pinned long enough that only the flush can + have posted.""" + monkeypatch.setenv("OMNIO_SUBAGENT_PROGRESS_HOOK", "http://example.invalid/progress") + monkeypatch.setenv("OMNIO_INTERNAL_TOKEN", "svc-tok") + _fast_progress_monitor(monkeypatch, interval=30.0) + posted = [] + + monkeypatch.setattr( + "requests.post", + lambda url, json=None, headers=None, timeout=None: ( + posted.append(json), SimpleNamespace(status_code=200))[1], + ) + + res = ad.dispatch_async_delegation_batch( + goals=["goal a"], context=None, toolsets=None, role="leaf", + model="m", session_key="", origin_turn_id="turn-last-child", + runner=lambda: {"results": []}, + progress_fn=lambda: (((2, None, time.time(), "sa-last", True),), False), + ) + assert res["status"] == "dispatched" + assert _drain_for(res["delegation_id"], timeout=5.0) is not None + + done = [p for p in posted if p["status"] == "completed"] + assert len(done) == 1 + assert done[0]["subagent_id"] == "sa-last" + + +def test_terminal_tick_not_re_emitted_on_later_sweeps(monkeypatch): + """A finished child stays finished in every later sample. Without a claim + set the sweep would re-announce it on each pass, and the product would + show the same subagent completing over and over.""" + monkeypatch.setenv("OMNIO_SUBAGENT_PROGRESS_HOOK", "http://example.invalid/progress") + monkeypatch.setenv("OMNIO_INTERNAL_TOKEN", "svc-tok") + _fast_progress_monitor(monkeypatch, interval=0.02) + gate = threading.Event() + posted = [] + post_seen = threading.Event() + + def fake_post(url, json=None, headers=None, timeout=None): + posted.append(json) + post_seen.set() + return SimpleNamespace(status_code=200) + + monkeypatch.setattr("requests.post", fake_post) + + def progress_fn(): + return ( + ( + (3, None, time.time(), "sa-done", True), + (1, "terminal", time.time(), "sa-busy", False), + ), + True, + ) + + res = ad.dispatch_async_delegation_batch( + goals=["goal a", "goal b"], context=None, toolsets=None, role="leaf", + model="m", session_key="", origin_turn_id="turn-no-repeat", + runner=lambda: (gate.wait(timeout=5), {"results": []})[1], + progress_fn=progress_fn, + ) + assert res["status"] == "dispatched" + assert post_seen.wait(timeout=3.0) + time.sleep(0.4) # ~20 sweeps at the shrunk interval + gate.set() + assert _drain_for(res["delegation_id"], timeout=5.0) is not None + + done = [p for p in posted if p["subagent_id"] == "sa-done"] + assert len(done) == 1, f"terminal tick re-emitted {len(done)} times" + assert done[0]["status"] == "completed" + + +def test_terminal_tick_hook_failure_is_swallowed(monkeypatch): + """A raising hook on the terminal tick must never reach the delegation: + the batch still finalizes and delivers its completion event.""" + monkeypatch.setenv("OMNIO_SUBAGENT_PROGRESS_HOOK", "http://example.invalid/progress") + monkeypatch.setenv("OMNIO_INTERNAL_TOKEN", "svc-tok") + _fast_progress_monitor(monkeypatch, interval=30.0) + calls = [] + + def raising_post(url, json=None, headers=None, timeout=None): + calls.append(json) + raise RuntimeError("hook exploded") + + monkeypatch.setattr("requests.post", raising_post) + + res = ad.dispatch_async_delegation_batch( + goals=["goal a"], context=None, toolsets=None, role="leaf", + model="m", session_key="", origin_turn_id="turn-hook-boom", + runner=lambda: {"results": [{"task_index": 0, "status": "completed", "summary": "ok"}]}, + progress_fn=lambda: (((1, None, time.time(), "sa-boom", True),), False), + ) + assert res["status"] == "dispatched" + + evt = _drain_for(res["delegation_id"], timeout=5.0) + assert evt is not None + assert evt["status"] == "completed" + assert calls and calls[0]["status"] == "completed" diff --git a/tests/tools/test_delegate.py b/tests/tools/test_delegate.py index 48d5a629d871..45777ac6842f 100644 --- a/tests/tools/test_delegate.py +++ b/tests/tools/test_delegate.py @@ -15,7 +15,7 @@ import time import types import unittest -from unittest.mock import MagicMock, patch +from unittest.mock import MagicMock, call, patch from tools.delegate_tool import ( DELEGATE_BLOCKED_TOOLS, @@ -2103,6 +2103,89 @@ def test_run_single_child_releases_lease_after_failure(self): child._credential_pool.release_lease.assert_called_once_with("cred-a") +class TestBackgroundSubagentStartDedupe(unittest.TestCase): + """A background dispatch emits subagent.start SYNCHRONOUSLY at dispatch + time (delegate_task's `if background:` block, while the parent's turn + is still live) and stamps the child with + ``_subagent_start_emitted_at_dispatch`` so _run_single_child's own + in-thread emission — which normally fires once the child's worker + thread starts, typically well after the parent's run has already gone + terminal — is skipped instead of producing a duplicate persisted row.""" + + def test_run_single_child_skips_start_when_already_emitted_at_dispatch(self): + from tools.delegate_tool import _run_single_child + + child = MagicMock() + child._subagent_start_emitted_at_dispatch = True + child.run_conversation.return_value = { + "final_response": "done", "completed": True, + "interrupted": False, "api_calls": 1, "messages": [], + } + + _run_single_child( + task_index=0, goal="already started at dispatch", + child=child, parent_agent=_make_mock_parent(), + ) + + started_calls = [ + c for c in child.tool_progress_callback.call_args_list + if c.args and c.args[0] == "subagent.start" + ] + self.assertEqual(started_calls, []) + + def test_run_single_child_emits_start_when_not_flagged(self): + """Sync-path (or any caller that never set the dispatch-time flag) + keeps emitting subagent.start from the child thread exactly as + before — this must stay unchanged.""" + from tools.delegate_tool import _run_single_child + + child = MagicMock() + child._subagent_start_emitted_at_dispatch = False + child.run_conversation.return_value = { + "final_response": "done", "completed": True, + "interrupted": False, "api_calls": 1, "messages": [], + } + + _run_single_child( + task_index=0, goal="normal sync child", + child=child, parent_agent=_make_mock_parent(), + ) + + started_calls = [ + c for c in child.tool_progress_callback.call_args_list + if c.args and c.args[0] == "subagent.start" + ] + self.assertEqual( + started_calls, + [call("subagent.start", preview="normal sync child")], + ) + + def test_run_single_child_still_emits_complete_when_start_deduped(self): + """Suppressing the dispatch-time-duplicated subagent.start must not + touch subagent.complete — that stays exactly as-is (fires post-run, + dropped silently if the parent run already went terminal, which is + expected/fine per the coordinator's finding).""" + from tools.delegate_tool import _run_single_child + + child = MagicMock() + child._subagent_start_emitted_at_dispatch = True + child.run_conversation.return_value = { + "final_response": "done", "completed": True, + "interrupted": False, "api_calls": 1, "messages": [], + } + + _run_single_child( + task_index=0, goal="dispatch-time started", + child=child, parent_agent=_make_mock_parent(), + ) + + complete_calls = [ + c for c in child.tool_progress_callback.call_args_list + if c.args and c.args[0] == "subagent.complete" + ] + self.assertEqual(len(complete_calls), 1) + + class TestDelegateHeartbeat(unittest.TestCase): """Heartbeat propagates child activity to parent during delegation. diff --git a/tests/tools/test_delegate_apiserver_background.py b/tests/tools/test_delegate_apiserver_background.py index f0a07d6ddbdd..0423ee0d996c 100644 --- a/tests/tools/test_delegate_apiserver_background.py +++ b/tests/tools/test_delegate_apiserver_background.py @@ -104,9 +104,61 @@ def clobbering_build_child(**kw): 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) + # Test-only handle so callers can inspect the fixed child instance + # clobbering_build_child always returns (e.g. its tool_progress_callback + # mock) without a second, side-effecting call into _build_child_agent. + dt._test_fake_child = fake_child return dt +def test_background_dispatch_emits_subagent_start_synchronously(monkeypatch): + """A background dispatch must emit subagent.start for each child + SYNCHRONOUSLY, before delegate_task returns "dispatched" — while the + parent's turn/run is still live, which is the only point at which the + persisted Turn Event Log row can actually land (once the parent's run + goes terminal, TurnEventLogStore.append_payload silently drops further + writes for that run_id).""" + dt = _patch_delegate(monkeypatch) + # delegate_task tees tool_progress_callback through + # tools.delegation_live_log.wrap_progress_callback for the live + # transcript feature, which replaces fake_child.tool_progress_callback + # with a real wrapping function — irrelevant to this test (it forwards + # every call to the inner callback unchanged) but it would hide the + # MagicMock this test needs to assert against. Make it a passthrough. + import tools.delegation_live_log as dll + + monkeypatch.setattr(dll, "wrap_progress_callback", lambda inner, writer: inner) + + set_session_vars( + platform="api_server", + chat_id="raw-sid-start", + session_key="raw-sid-start", + session_id="raw-sid-start", + 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" + + # By the time delegate_task returned, the child's own progress callback + # must already have received subagent.start — exactly once. + fake_child = dt._test_fake_child + fake_child.tool_progress_callback.assert_called_once_with( + "subagent.start", preview="bg on api_server", + ) + assert fake_child._subagent_start_emitted_at_dispatch is True + + # Drain the async worker's completion event so it can't leak into a + # later test's queue (the fixture only drains what's already queued at + # teardown time, not what a still-running daemon-thread worker later + # pushes). + _drain_one() + + 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 @@ -140,6 +192,42 @@ def test_apiserver_session_with_id_dispatches_background(monkeypatch): assert evt["origin_session_id"] == "raw-sid-7" +def test_healthy_push_path_still_stamps_origin_session_id(monkeypatch): + """async_delivery=True (the normal api_server push path) must stamp the + origin session id on the dispatch record too: the session-delegations + listing matches records by origin_session_id, so a record dispatched on + the healthy path with an empty stamp is invisible to its own session.""" + dt = _patch_delegate(monkeypatch) + monkeypatch.setenv("HERMES_SESSION_ID", "raw-sid-8") + set_session_vars( + platform="api_server", + chat_id="raw-sid-8", + session_key="raw-sid-8", + session_id="raw-sid-8", + async_delivery=True, + ) + + out = dt.delegate_task( + goal="bg on healthy api_server", context="ctx", + background=True, parent_agent=_fake_parent(), + ) + parsed = json.loads(out) + assert parsed["status"] == "dispatched", parsed + assert parsed["mode"] == "background" + + from tools.async_delegation import list_async_delegations + + record = next( + r for r in list_async_delegations() + if r.get("delegation_id") == parsed["delegation_id"] + ) + assert record["origin_session_id"] == "raw-sid-8" + + evt = _drain_one() + assert evt is not None + assert evt["origin_session_id"] == "raw-sid-8" + + # --------------------------------------------------------------------------- # _current_origin_session_id — the clobber-proof origin capture helper # --------------------------------------------------------------------------- @@ -168,6 +256,161 @@ def test_origin_helper_empty_on_push_platforms(monkeypatch): assert _current_origin_session_id() == "" +def test_apiserver_session_carries_origin_turn_id_through_child_clobber(monkeypatch): + """The Omnio turn_id bound alongside chat_id (see + ApiServerAdapter._bind_api_server_session) must reach the completion + event as origin_turn_id, surviving the same child-session clobber that + origin_session_id survives — captured before child construction, from + the request-scoped binding, not from anything the child touches.""" + dt = _patch_delegate(monkeypatch) + monkeypatch.setenv("HERMES_SESSION_ID", "raw-sid-8") + set_session_vars( + platform="api_server", + chat_id="raw-sid-8", + session_key="raw-sid-8", + session_id="raw-sid-8", + async_delivery=False, + origin_turn_id="turn-abc-123", + ) + + out = dt.delegate_task( + goal="bg on api_server with turn id", context="ctx", + background=True, parent_agent=_fake_parent(), + ) + parsed = json.loads(out) + assert parsed["status"] == "dispatched", parsed + + evt = _drain_one() + assert evt is not None + assert evt["origin_session_id"] == "raw-sid-8" + assert evt["origin_turn_id"] == "turn-abc-123" + + +# --------------------------------------------------------------------------- +# _current_origin_turn_id — the clobber-proof origin turn-id capture helper +# --------------------------------------------------------------------------- + + +def test_origin_turn_id_helper_survives_child_session_clobber(monkeypatch): + """Same guarantee as _current_origin_session_id, for the turn id.""" + from gateway.session_context import set_current_session_id + from tools.async_delegation import _current_origin_turn_id + + set_session_vars(platform="api_server", chat_id="raw-origin-2", origin_turn_id="turn-xyz") + assert _current_origin_turn_id() == "turn-xyz" + + set_current_session_id("20260715_child3") # the clobber + assert _current_origin_turn_id() == "turn-xyz" + + +def test_origin_turn_id_helper_empty_on_push_platforms(monkeypatch): + from tools.async_delegation import _current_origin_turn_id + + set_session_vars(platform="telegram", chat_id="123456789", origin_turn_id="turn-should-not-apply") + assert _current_origin_turn_id() == "" + + +def test_origin_turn_id_helper_empty_when_not_bound(monkeypatch): + """Non-Omnio deployments never set the turn id — must degrade to "".""" + from tools.async_delegation import _current_origin_turn_id + + set_session_vars(platform="api_server", chat_id="raw-origin-3") + assert _current_origin_turn_id() == "" + + +# --------------------------------------------------------------------------- +# delegation_sync_only — defeats the wake-sid re-enable above, unconditionally +# --------------------------------------------------------------------------- + + +def test_current_delegation_sync_only_survives_child_session_clobber(monkeypatch): + """Same clobber-proof guarantee as _current_origin_session_id/_turn_id, + for the delegation_sync_only flag.""" + from gateway.session_context import set_current_session_id + from tools.async_delegation import _current_delegation_sync_only + + set_session_vars( + platform="api_server", chat_id="raw-origin-4", delegation_sync_only=True, + ) + assert _current_delegation_sync_only() is True + + set_current_session_id("20260715_child4") # the clobber + assert _current_delegation_sync_only() is True + + +def test_current_delegation_sync_only_false_on_push_platforms(monkeypatch): + from tools.async_delegation import _current_delegation_sync_only + + set_session_vars( + platform="telegram", chat_id="123456789", delegation_sync_only=True, + ) + assert _current_delegation_sync_only() is False + + +def test_current_delegation_sync_only_false_when_not_bound(monkeypatch): + from tools.async_delegation import _current_delegation_sync_only + + set_session_vars(platform="api_server", chat_id="raw-origin-5") + assert _current_delegation_sync_only() is False + + +def test_delegation_sync_only_forces_sync_even_with_wake_sid_available(monkeypatch): + """The exact inverse of test_apiserver_session_with_id_dispatches_background: + async_delivery=False + a raw session id bound (which alone would trigger + the wake-sid re-enable and dispatch in the background) must instead stay + SYNCHRONOUS once delegation_sync_only is set — the flag defeats that + re-enable path unconditionally, because a headless surface (cron, + trigger.dev run) has no channel to ever consume the wake.""" + dt = _patch_delegate(monkeypatch) + monkeypatch.setenv("HERMES_SESSION_ID", "raw-sid-9") + set_session_vars( + platform="api_server", + chat_id="raw-sid-9", + session_key="raw-sid-9", + session_id="raw-sid-9", + async_delivery=False, + delegation_sync_only=True, + ) + + out = dt.delegate_task( + goal="bg on headless api_server", 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 "results" in parsed + assert process_registry.completion_queue.empty() + + +def test_delegation_sync_only_absent_preserves_existing_behavior(monkeypatch): + """Sanity check: without the flag, the exact same session state as above + dispatches in the background (unchanged from + test_apiserver_session_with_id_dispatches_background) — the new + parameter must not alter behavior when omitted/false.""" + dt = _patch_delegate(monkeypatch) + monkeypatch.setenv("HERMES_SESSION_ID", "raw-sid-10") + set_session_vars( + platform="api_server", + chat_id="raw-sid-10", + session_key="raw-sid-10", + session_id="raw-sid-10", + async_delivery=False, + delegation_sync_only=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" + + _drain_one() + + 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).""" diff --git a/tools/async_delegation.py b/tools/async_delegation.py index dbe0b2f1074f..2a5301a75c10 100644 --- a/tools/async_delegation.py +++ b/tools/async_delegation.py @@ -38,6 +38,7 @@ import json import logging +import os import sqlite3 import threading import time @@ -115,6 +116,27 @@ _monitor_thread: Optional[threading.Thread] = None _monitor_stop = threading.Event() +# --------------------------------------------------------------------------- +# Opt-in progress reporting (Omnio) — off by default +# --------------------------------------------------------------------------- +# When env OMNIO_SUBAGENT_PROGRESS_HOOK is set, running delegations report +# best-effort progress (goal + current child activity) to that URL so the +# Omnio product UI can show "subagent working on X" instead of going silent +# until completion. A dedicated monitor thread (same singleton-thread shape +# as the stale monitor above) samples every dispatch's progress_fn on a +# tighter cadence than the stale sweep, since the stale monitor's job is +# "notice a wedge after minutes," not "report fresh status every few +# seconds." Strictly best-effort: a hook failure is logged and dropped, never +# raised into the delegation path, and the thread never keeps the process +# alive (daemon=True, self-stops when nothing is running). +_OMNIO_SUBAGENT_PROGRESS_HOOK_ENV = "OMNIO_SUBAGENT_PROGRESS_HOOK" +_PROGRESS_REPORT_INTERVAL = 10.0 # seconds between monitor sweeps +_PROGRESS_HOOK_TIMEOUT_SECONDS = 5.0 + +_progress_monitor_lock = threading.Lock() +_progress_monitor_thread: Optional[threading.Thread] = None +_progress_monitor_stop = threading.Event() + def _db_path(): return get_hermes_home() / "state.db" @@ -591,6 +613,51 @@ def _current_origin_session_id() -> str: return "" +def _current_origin_turn_id() -> str: + """Omnio product turn id of the ORIGINATING api_server request, or ``""``. + + Mirrors ``_current_origin_session_id`` exactly, and for the same reason: + it must be read BEFORE any child agent is constructed, because + ``agent_init`` clobbers the session-id ContextVar (not this one, but + dispatch-time code that reads session context in one place should read + all of it together to avoid a future divergent read site). The binding + itself (``HERMES_ORIGIN_TURN_ID``) is set alongside ``HERMES_SESSION_CHAT_ID`` + by ``ApiServerAdapter._bind_api_server_session`` and is request-scoped, + so it is empty on any non-Omnio deployment (no ``turn_id`` on the run) or + any non-api_server platform. + """ + try: + from gateway.session_context import get_session_env + + if get_session_env("HERMES_SESSION_PLATFORM", "") != "api_server": + return "" + return get_session_env("HERMES_ORIGIN_TURN_ID", "") or "" + except Exception: + return "" + + +def _current_delegation_sync_only() -> bool: + """Whether the ORIGINATING api_server request forced this run's background + delegations to run SYNCHRONOUSLY, or ``False``. + + Mirrors ``_current_origin_turn_id`` exactly, and for the same reason: it + must be read BEFORE any child agent is constructed, at the same capture + point as the other origin reads. The binding itself + (``HERMES_DELEGATION_SYNC_ONLY``) is set alongside ``HERMES_SESSION_CHAT_ID`` + by ``ApiServerAdapter._bind_api_server_session`` and is request-scoped, so + it is ``False`` on any non-Omnio deployment (no ``delegation_sync_only`` + on the run) or any non-api_server platform. + """ + try: + from gateway.session_context import get_session_env + + if get_session_env("HERMES_SESSION_PLATFORM", "") != "api_server": + return False + return get_session_env("HERMES_DELEGATION_SYNC_ONLY", "") == "1" + except Exception: + return False + + def dispatch_async_delegation( *, goal: str, @@ -603,6 +670,7 @@ def dispatch_async_delegation( runner: Callable[[], Dict[str, Any]], origin_ui_session_id: str = "", origin_session_id: str = "", + origin_turn_id: str = "", interrupt_fn: Optional[Callable[[], None]] = None, max_async_children: int = _DEFAULT_MAX_ASYNC_CHILDREN, progress_fn: Optional[Callable[[], tuple]] = None, @@ -624,6 +692,13 @@ def dispatch_async_delegation( the delegation. Carried on the completion event so the gateway can pin routing to the spawning session instead of recovering the latest ``ended_at IS NULL`` row for the peer tuple (#57498). + origin_turn_id + The Omnio product turn id (``turn_id`` on ``POST /v1/runs``) that + spawned this delegation, captured the same way as + ``origin_session_id`` (before child construction). Empty on any + non-Omnio deployment. Carried on the completion event so + ``gateway.wake`` can redirect the wake into a real product turn via + ``OMNIO_WAKE_HOOK`` instead of self-posting. runner Zero-arg callable that builds + runs the child and returns the same result dict ``_run_single_child`` produces. Runs on the worker thread. @@ -661,6 +736,7 @@ def dispatch_async_delegation( "session_key": session_key, "origin_ui_session_id": origin_ui_session_id, "origin_session_id": origin_session_id, + "origin_turn_id": origin_turn_id, "parent_session_id": parent_session_id, "status": "running", "dispatched_at": dispatched_at, @@ -729,6 +805,8 @@ def _worker() -> None: } if progress_fn is not None: _ensure_stale_monitor() + if os.environ.get(_OMNIO_SUBAGENT_PROGRESS_HOOK_ENV, "").strip(): + _ensure_progress_monitor() logger.info( "Dispatched async delegation %s (session_key=%s): %s", @@ -739,6 +817,15 @@ def _worker() -> None: def _finalize(delegation_id: str, result: Dict[str, Any], status: str) -> None: """Mark a record complete and push the completion event onto the queue.""" + # Before _begin_finalization drops progress_fn: last chance to announce + # children whose terminal tick the sweep never got to. + try: + _flush_child_completions(delegation_id) + except Exception as exc: + logger.debug( + "Child-completion flush failed for delegation %s: %s", + delegation_id, exc, + ) claimed = _begin_finalization(delegation_id) if claimed is None: return @@ -808,6 +895,13 @@ def _push_completion_event( "session_key": record.get("session_key", ""), "origin_ui_session_id": record.get("origin_ui_session_id", ""), "origin_session_id": record.get("origin_session_id", ""), + "origin_turn_id": record.get("origin_turn_id", ""), + # Single-dispatch mirror of the batch path's subagent_ids (see + # _push_batch_completion_event) — a one-element list when the + # runner's result carries the child's streamed subagent_id. + "subagent_ids": ( + [result["subagent_id"]] if result.get("subagent_id") else [] + ), "parent_session_id": record.get("parent_session_id"), "goal": record.get("goal", ""), "context": record.get("context"), @@ -858,6 +952,7 @@ def dispatch_async_delegation_batch( runner: Callable[[], Dict[str, Any]], origin_ui_session_id: str = "", origin_session_id: str = "", + origin_turn_id: str = "", interrupt_fn: Optional[Callable[[], None]] = None, max_async_children: int = _DEFAULT_MAX_ASYNC_CHILDREN, delegation_id: Optional[str] = None, @@ -901,6 +996,7 @@ def dispatch_async_delegation_batch( "session_key": session_key, "origin_ui_session_id": origin_ui_session_id, "origin_session_id": origin_session_id, + "origin_turn_id": origin_turn_id, "parent_session_id": parent_session_id, "status": "running", "dispatched_at": dispatched_at, @@ -970,6 +1066,8 @@ def _worker() -> None: } if progress_fn is not None: _ensure_stale_monitor() + if os.environ.get(_OMNIO_SUBAGENT_PROGRESS_HOOK_ENV, "").strip(): + _ensure_progress_monitor() logger.info( "Dispatched async delegation batch %s (%d task(s), session_key=%s)", @@ -982,6 +1080,15 @@ def _finalize_batch( delegation_id: str, combined: Dict[str, Any], status: str ) -> None: """Mark a batch record complete and push ONE combined completion event.""" + # Before _begin_finalization drops progress_fn: last chance to announce + # children whose terminal tick the sweep never got to. + try: + _flush_child_completions(delegation_id) + except Exception as exc: + logger.debug( + "Child-completion flush failed for delegation %s: %s", + delegation_id, exc, + ) claimed = _begin_finalization(delegation_id) if claimed is None: return @@ -1007,12 +1114,26 @@ def _push_batch_completion_event( dispatched_at = event_record.get("dispatched_at") or time.time() completed_at = event_record.get("completed_at") or time.time() + _results = combined.get("results") or [] + # The exact subagent_id values the children streamed as "subagentId" on + # their subagent.start/complete events (tools.delegate_tool._run_single_child + # -> entry["subagent_id"]) — echoed here so the wake payload can carry + # them and the Omnio proxy can match by the same streamed identity + # instead of delegation_id (which is a batch-level id, not a per-child + # one; a single-task batch's subagent_id can equal delegation_id by + # coincidence but must never be assumed to). + _subagent_ids = [ + r.get("subagent_id") for r in _results + if isinstance(r, dict) and r.get("subagent_id") + ] evt = { "type": "async_delegation", "delegation_id": event_record.get("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", ""), + "origin_turn_id": event_record.get("origin_turn_id", ""), + "subagent_ids": _subagent_ids, "parent_session_id": event_record.get("parent_session_id"), "goal": event_record.get("goal", ""), "goals": event_record.get("goals"), @@ -1024,7 +1145,7 @@ def _push_batch_completion_event( "is_batch": True, # The full per-task results list — the formatter renders a # consolidated multi-task block from this. - "results": combined.get("results") or [], + "results": _results, # Per-task live transcript log paths (cache/delegation/live/...). # They persist after completion and double as the full-fidelity # operational record of each child's run. @@ -1241,8 +1362,11 @@ def _children_activity_from_token(token: Any, now: float) -> Optional[List]: """Parse a progress token into per-child activity dicts (best-effort). delegate_tool's ``_batch_progress`` emits one ``(api_call_count, - current_tool, last_activity_ts)`` tuple per child. Foreign token shapes - (custom dispatchers) degrade to ``None`` entries rather than raising — + current_tool, last_activity_ts, subagent_id, finished)`` tuple per child + (``subagent_id`` is the same stable id the child streams as "subagentId" + on its subagent.start/complete events, ``finished`` flips once that child + returns — see ``tools.delegate_tool._batch_progress``). Shorter tuples and + foreign token shapes (custom dispatchers) degrade rather than raising — the token contract is intentionally opaque to the registry. """ try: @@ -1260,12 +1384,240 @@ def _children_activity_from_token(token: Any, now: float) -> Optional[List]: entry["seconds_since_activity"] = round( max(0.0, now - float(part[2])), 1 ) + if len(part) >= 4 and isinstance(part[3], str) and part[3]: + entry["subagent_id"] = part[3] + if len(part) >= 5: + entry["finished"] = bool(part[4]) out.append(entry) else: out.append(None) return out +def _ensure_progress_monitor() -> None: + """Start (once) the module-level progress-reporting monitor thread. + + Mirrors ``_ensure_stale_monitor``'s singleton-thread shape: one daemon + thread serves every progress-enabled dispatch, exits on its own when no + monitorable records remain, and is restarted by the next dispatch that + carries a ``progress_fn`` while ``OMNIO_SUBAGENT_PROGRESS_HOOK`` is set. + """ + global _progress_monitor_thread + with _progress_monitor_lock: + if _progress_monitor_thread is not None and _progress_monitor_thread.is_alive(): + return + _progress_monitor_stop.clear() + _progress_monitor_thread = threading.Thread( + target=_progress_monitor_loop, + name="async-delegate-progress-monitor", + daemon=True, + ) + _progress_monitor_thread.start() + + +def _progress_report_payload( + record: Dict[str, Any], now: float +) -> Optional[tuple[List[Dict[str, Any]], Any]]: + """Build the progress-hook payload(s) for one running record, or ``None``. + + Best-effort: samples ``progress_fn`` fresh (outside ``_records_lock`` — + see the same rationale in ``list_async_delegations``) and reduces it to + one payload PER CHILD (a batch of N tasks has N children, each streaming + its own ``subagentId`` — the proxy matches progress updates against the + per-child row that identity created, not against the batch-level + ``delegation_id``). Returns ``None`` when there is no origin turn to + attribute the update to, or the sampler is unavailable/errors — the + caller simply skips reporting that sweep. + + Returns ``(payloads, token)`` where ``token`` is the RAW sampled token + (not the per-child breakdown), used by the caller to decide whether + anything meaningfully changed since the last report. + """ + origin_turn_id = record.get("origin_turn_id") or "" + if not origin_turn_id: + return None + progress_fn = record.get("progress_fn") + if progress_fn is None: + return None + try: + token, in_tool = progress_fn() + except Exception: + return None + delegation_id = record.get("delegation_id", "") + goals = record.get("goals") + goal = record.get("goal", "") + activity = _children_activity_from_token(token, now) or [] + payloads: List[Dict[str, Any]] = [] + for idx, a in enumerate(activity): + a = a or {} + label = ( + goals[idx] if isinstance(goals, list) and idx < len(goals) else goal + ) + finished = bool(a.get("finished")) + payloads.append({ + "origin_turn_id": origin_turn_id, + "delegation_id": delegation_id, + "subagent_id": a.get("subagent_id") or "", + "status": "completed" if finished else "running", + "progress": { + "label": str(label)[:200], + "detail": None if finished else a.get("current_tool"), + }, + }) + if not payloads: + # No per-child breakdown available (foreign/opaque token shape, + # e.g. a custom dispatcher) — fall back to one delegation-level + # update with an empty subagent_id rather than reporting nothing. + payloads.append({ + "origin_turn_id": origin_turn_id, + "delegation_id": delegation_id, + "subagent_id": "", + "status": "running", + "progress": { + "label": str(goal)[:200], + "detail": "working" if in_tool else None, + }, + }) + return payloads, token + + +def _claim_child_completions( + delegation_id: str, payloads: List[Dict[str, Any]] +) -> List[Dict[str, Any]]: + """Filter out terminal ticks for children already announced as completed. + + A finished child stays finished in every later sample, so without a claim + set the sweep would re-announce it on each pass. The set lives on the live + record (private key — excluded from ``list_async_delegations`` snapshots) + so the terminal tick fires exactly once per child, independent of whether + its siblings are still running. + """ + kept: List[Dict[str, Any]] = [] + with _records_lock: + record = _records.get(delegation_id) + claimed = record.get("_progress_completed") if record is not None else None + if not isinstance(claimed, set): + claimed = set() + if record is not None: + record["_progress_completed"] = claimed + for payload in payloads: + if payload.get("status") != "completed": + kept.append(payload) + continue + subagent_id = payload.get("subagent_id") or "" + if subagent_id in claimed: + continue + claimed.add(subagent_id) + kept.append(payload) + return kept + + +def _flush_child_completions(delegation_id: str) -> None: + """Announce any child that finished without a terminal tick going out. + + The sweep catches a child the moment it observes the flip, but the LAST + child's flip lands together with the batch's own completion — inside a + sweep interval. Sampling once more here, while the record is still running + and the child agents are still alive, closes that gap. + """ + hook_url = os.environ.get(_OMNIO_SUBAGENT_PROGRESS_HOOK_ENV, "").strip() + if not hook_url: + return + with _records_lock: + record = _records.get(delegation_id) + if record is None: + return + built = _progress_report_payload(record, time.time()) + if built is None: + return + payloads, _token = built + terminal = [p for p in payloads if p.get("status") == "completed"] + for payload in _claim_child_completions(delegation_id, terminal): + _post_progress_update(hook_url, payload) + + +def _post_progress_update(hook_url: str, payload: Dict[str, Any]) -> None: + """POST one progress update. Strictly best-effort — never raises. + + Runs on the progress-monitor thread (a plain daemon thread, not an + asyncio task), so this uses ``requests`` synchronously rather than + aiohttp, matching the sync worker context. Any failure (network, + timeout, non-2xx) is logged and dropped — a lost progress update must + never affect the delegation or the agent loop. + """ + try: + import requests + + headers = { + "Content-Type": "application/json", + "X-Omnio-Service-Token": os.environ.get("OMNIO_INTERNAL_TOKEN", ""), + } + resp = requests.post( + hook_url, + json=payload, + headers=headers, + timeout=_PROGRESS_HOOK_TIMEOUT_SECONDS, + ) + if resp.status_code >= 400: + logger.debug( + "Subagent progress hook returned HTTP %d for delegation %s", + resp.status_code, payload.get("delegation_id"), + ) + except Exception as exc: + logger.debug( + "Subagent progress hook POST failed for delegation %s: %s", + payload.get("delegation_id"), exc, + ) + + +def _progress_monitor_loop() -> None: + """Sweep running delegations and report best-effort progress. + + Per sweep, for every running record with a ``progress_fn`` and a bound + ``origin_turn_id``: sample progress, and POST an update when the sampled + token changed since the last report OR the last report is older than + ``_PROGRESS_REPORT_INTERVAL`` — i.e. report on meaningful state change, + but never faster than the sweep interval either way. Stops itself once + no running record is reportable, same as the stale monitor. + """ + while not _progress_monitor_stop.wait(_PROGRESS_REPORT_INTERVAL): + hook_url = os.environ.get(_OMNIO_SUBAGENT_PROGRESS_HOOK_ENV, "").strip() + if not hook_url: + return # opted out mid-run (env cleared) — nothing left to do + now = time.time() + with _records_lock: + records_snapshot = [ + r for r in _records.values() + if r.get("status") == "running" and r.get("progress_fn") is not None + ] + if not records_snapshot: + return + # Sample + decide OUTSIDE the lock (progress_fn reads child-agent + # attributes and must never run under _records_lock — mirrors + # list_async_delegations' rationale). + to_report: List[Dict[str, Any]] = [] + for record in records_snapshot: + built = _progress_report_payload(record, now) + if built is None: + continue + payloads, token = built + delegation_id = record.get("delegation_id") + last_token = record.get("_progress_report_token") + last_ts = record.get("_progress_report_ts") or 0 + due = token != last_token or (now - last_ts) >= _PROGRESS_REPORT_INTERVAL + if not due: + continue + with _records_lock: + live = _records.get(delegation_id) + if live is None or live.get("status") != "running": + continue + live["_progress_report_token"] = token + live["_progress_report_ts"] = now + to_report.extend(_claim_child_completions(delegation_id, payloads)) + for payload in to_report: + _post_progress_update(hook_url, payload) + + def list_async_delegations() -> List[Dict[str, Any]]: """Snapshot of async delegations (running + recently completed). @@ -1323,6 +1675,15 @@ def list_async_delegations() -> List[Dict[str, Any]]: continue activity = _children_activity_from_token(token, now) if activity is not None: + # Stamp each child's own goal onto its activity entry. The token + # tuples arrive in dispatch order — the same order ``goals`` was + # recorded in — so consumers of this snapshot can name each child + # individually instead of falling back to the batch-level goal. + goals = item.get("goals") + if isinstance(goals, list): + for idx, entry in enumerate(activity): + if entry is not None and idx < len(goals): + entry["goal"] = goals[idx] item["children_activity"] = activity item["in_tool"] = bool(in_tool) return items @@ -1415,7 +1776,7 @@ def interrupt_for_session( def _reset_for_tests() -> None: """Test-only: clear all state and tear down the executor + monitor.""" - global _executor, _executor_max_workers, _monitor_thread + global _executor, _executor_max_workers, _monitor_thread, _progress_monitor_thread with _executor_lock: if _executor is not None: _executor.shutdown(wait=False) @@ -1427,5 +1788,11 @@ def _reset_for_tests() -> None: _monitor_thread = None if thread is not None and thread.is_alive(): thread.join(timeout=2) + _progress_monitor_stop.set() + with _progress_monitor_lock: + progress_thread = _progress_monitor_thread + _progress_monitor_thread = None + if progress_thread is not None and progress_thread.is_alive(): + progress_thread.join(timeout=2) with _records_lock: _records.clear() diff --git a/tools/delegate_tool.py b/tools/delegate_tool.py index f8604ec460d2..4f172450b225 100644 --- a/tools/delegate_tool.py +++ b/tools/delegate_tool.py @@ -2105,7 +2105,20 @@ def _heartbeat_loop(): try: _heartbeat_thread.start() - if child_progress_cb: + # Background dispatches already emitted this SYNCHRONOUSLY at dispatch + # time (see the `if background:` block in delegate_task), while the + # parent's turn/run was still live — that's the only way the event + # both streams to an attached client and persists into the Turn Event + # Log (a background child's own thread runs on a daemon executor + # after the parent's turn/run has typically already gone terminal; + # TurnEventLogStore.append_payload silently drops writes to a + # terminal run — see gateway/turn_event_log.py). Emitting again here + # would double the persisted subagent.start row and break proxy-side + # matching by subagentId, so skip it — never rely on downstream + # dedupe for this. + if child_progress_cb and not getattr( + child, "_subagent_start_emitted_at_dispatch", False + ): try: child_progress_cb("subagent.start", preview=goal) except Exception as e: @@ -2305,6 +2318,7 @@ def _run_with_thread_capture(): ), "_child_role": getattr(child, "_delegate_role", None), "diagnostic_path": diagnostic_path, + "subagent_id": _subagent_id, } finally: # Shut down executor without waiting — if the child thread @@ -2401,6 +2415,13 @@ def _run_with_thread_capture(): "duration_seconds": duration, "model": _model if isinstance(_model, str) else None, "exit_reason": exit_reason, + # The stable id this child streamed as "subagentId" on every + # subagent.start/complete event (api_server.py's _callback, + # ~line 6820). Carried on the per-task result so a batch + # completion event can echo the exact set of child ids the + # Omnio proxy already persisted rows for under, rather than + # re-deriving/guessing them. + "subagent_id": _subagent_id, "tokens": { "input": ( _input_tokens if isinstance(_input_tokens, (int, float)) else 0 @@ -2549,6 +2570,7 @@ def _run_with_thread_capture(): "api_calls": 0, "duration_seconds": duration, "_child_role": getattr(child, "_delegate_role", None), + "subagent_id": _subagent_id, } finally: @@ -2561,6 +2583,16 @@ def _run_with_thread_capture(): if _heartbeat_thread.ident is not None: _heartbeat_thread.join(timeout=5) + # Mark the child terminal for the batch progress sampler, whatever the + # outcome (completed, error, interrupted). This is the only per-child + # completion signal a detached batch has: the parent's progress + # callback is dead by then, and the async registry's completion event + # fires once, for the batch as a whole. + try: + child._subagent_finished = True + except Exception: + pass + # Drop the TUI-facing registry entry. Safe to call even if the # child was never registered (e.g. ID missing on test doubles). if _subagent_id: @@ -2926,9 +2958,23 @@ def delegate_task( # 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 + from tools.async_delegation import ( + _current_delegation_sync_only, + _current_origin_session_id, + _current_origin_turn_id, + ) _origin_wake_sid = _current_origin_session_id() + # Same rationale, same capture point: the Omnio product turn_id bound + # alongside HERMES_SESSION_CHAT_ID (ApiServerAdapter._bind_api_server_session) + # is request-scoped and would be unreadable once a child agent binds its + # own session context. Empty on non-Omnio deployments. + _origin_turn_id = _current_origin_turn_id() + # Same rationale, same capture point: the Omnio proxy's per-run + # delegation_sync_only flag (bound alongside HERMES_SESSION_CHAT_ID) must + # be read before it becomes unreadable once a child agent binds its own + # session context. False on non-Omnio deployments. + _sync_only = _current_delegation_sync_only() # Build all child agents on the main thread (thread-safe construction). # _build_child_preserving_parent_tools saves/restores the parent's @@ -3171,18 +3217,36 @@ def _execute_and_aggregate() -> dict: except Exception: _async_ok = True - _wake_sid = "" - if not _async_ok: + # Stamped on the dispatch record unconditionally: the api_server's + # session-delegations listing matches records by origin_session_id, + # so a record dispatched on the healthy push path must carry it too, + # not only the self-post wake fallback below. + _wake_sid = _origin_wake_sid + if _sync_only: + # The Omnio proxy set delegation_sync_only for this run: this is + # a headless surface (cron, trigger.dev run) with NO channel to + # ever consume a background wake, even though the API server + # always binds a raw session id and would otherwise qualify for + # the self-post wake re-enable below. Force the synchronous + # fallback unconditionally — this defeats that re-enable path + # entirely rather than merely skipping it, since a caller could + # set the flag even when async_delivery_supported() is True. + logger.info( + "delegate_task: delegation_sync_only is set for this run — " + "forcing synchronous execution regardless of wake-session " + "availability." + ) + _async_ok = False + elif 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 + # session id to wake. _wake_sid holds the origin captured before + # child construction — reading HERMES_SESSION_ID here would + # return the subagent's internal id. if _wake_sid: logger.info( "delegate_task: async delivery unsupported on this " @@ -3247,6 +3311,42 @@ def _execute_and_aggregate() -> dict: _parent_session_id = getattr(parent_agent, "session_id", None) _child_agents = [c for (_, _, c) in children] + # Emit subagent.start SYNCHRONOUSLY, now, for every child — before + # this call returns "dispatched". The parent's run is necessarily + # still live at this point (we are inside its tool call), so this + # is the ONLY point at which the event both streams to an attached + # client and persists into the Turn Event Log. The child's own + # thread (tools.async_delegation's daemon executor) typically only + # starts running well after the parent's turn/run has already gone + # terminal, at which point the SAME emission would call + # TurnEventEmitter.omnio_event -> TurnEventLogStore.append_payload, + # which silently drops the write (log.terminal — see + # gateway/turn_event_log.py) — so without this, background + # children's start row never lands. Uses each child's own + # tool_progress_callback (the identity-aware closure + # _build_child_preserving_parent_tools already wired to it), the + # exact same call the child thread would otherwise make + # (`child_progress_cb("subagent.start", preview=goal)` in + # _run_single_child) — subagent_id/task_index/task_count/goal/ + # parent_id/depth/model are all baked into that closure already. + # The flag suppresses the child thread's own later emission (see + # the guard in _run_single_child) so the persisted row is never + # duplicated. + for _i, _t, _c in children: + _start_cb = getattr(_c, "tool_progress_callback", None) + if _start_cb: + try: + _start_cb("subagent.start", preview=_t.get("goal", "")) + except Exception as e: + logger.debug( + "Dispatch-time subagent.start emission failed for " + "task %d: %s", _i, e, + ) + # Set regardless of whether the emission above succeeded — a + # best-effort emission failure must not turn into a doubled + # start row from the child thread retrying it later. + _c._subagent_start_emitted_at_dispatch = True + # Detach every child from the parent's interrupt-propagation list — the # batch's lifecycle is owned by the async registry now, not the parent # turn. _build_child_agent attached them (correct for sync runs). @@ -3277,18 +3377,27 @@ def _batch_interrupt(): def _batch_progress(): # Progress token for the async registry's stale monitor: the - # combined (api_call_count, current_tool, last_activity_ts) of - # every child. last_activity_ts is ticked by _touch_activity on - # every streamed chunk ("receiving stream response"), every tool - # transition, and every API-call start/completion — so a child - # streaming a long response is alive even though api_call_count - # only advances when the call completes (same liveness signal as - # the compaction inactivity budget, PR #71508). A fully frozen - # token past the stale threshold means the detached batch is - # wedged (e.g. stuck inside the first model API call — #60203). - # in_tool=True while ANY child is inside a tool so legitimately - # slow tools get the higher staleness ceiling, mirroring the - # sync-path heartbeat monitor. + # combined (api_call_count, current_tool, last_activity_ts, + # subagent_id) of every child. last_activity_ts is ticked by + # _touch_activity on every streamed chunk ("receiving stream + # response"), every tool transition, and every API-call + # start/completion — so a child streaming a long response is + # alive even though api_call_count only advances when the call + # completes (same liveness signal as the compaction inactivity + # budget, PR #71508). A fully frozen token past the stale + # threshold means the detached batch is wedged (e.g. stuck + # inside the first model API call — #60203). in_tool=True while + # ANY child is inside a tool so legitimately slow tools get the + # higher staleness ceiling, mirroring the sync-path heartbeat + # monitor. subagent_id is the same stable id the child streams + # as "subagentId" on its subagent.start/complete events + # (_run_single_child sets it as child._subagent_id) — carried + # here so the Omnio progress hook can attribute a running + # update to the exact child the proxy already has a row for. + # The trailing `finished` flag (set by _run_single_child's finally) + # is what lets that hook announce ONE child finishing while its + # siblings keep running — the batch's own completion event only + # fires once every child is done. parts = [] in_tool = False for _c in _child_agents: @@ -3300,6 +3409,8 @@ def _batch_progress(): _summary.get("api_call_count", 0), _tool, _summary.get("last_activity_ts"), + getattr(_c, "_subagent_id", None), + getattr(_c, "_subagent_finished", False) is True, ) ) in_tool = in_tool or bool(_tool) @@ -3319,6 +3430,7 @@ def _batch_progress(): session_key=_session_key, origin_ui_session_id=_origin_ui_session_id, origin_session_id=_wake_sid, + origin_turn_id=_origin_turn_id, parent_session_id=_parent_session_id, runner=_batch_runner, interrupt_fn=_batch_interrupt, diff --git a/tools/environments/base.py b/tools/environments/base.py index 20d7d77de112..e3fc5ca71c70 100644 --- a/tools/environments/base.py +++ b/tools/environments/base.py @@ -405,7 +405,8 @@ def _cwd_marker(session_id: str) -> str: # as the Python-side contract for the exclusion set; the dump path unsets by # name/prefix instead of grepping declare lines (see below / issue #71296). _SNAPSHOT_EXCLUDED_ENV_REGEX = ( - "^declare -x (HERMES_SESSION_|HERMES_UI_SESSION_ID|HERMES_CRON_AUTO_DELIVER_)" + "^declare -x (HERMES_SESSION_|HERMES_UI_SESSION_ID|HERMES_CRON_AUTO_DELIVER_" + "|HERMES_ORIGIN_TURN_ID|HERMES_DELEGATION_SYNC_ONLY)" ) @@ -435,7 +436,8 @@ def _export_dump_excluding_session_vars(tmp_path: str) -> str: return ( "{ ( " "unset ${!HERMES_SESSION_*} ${!HERMES_CRON_AUTO_DELIVER_*} " - "HERMES_UI_SESSION_ID 2>/dev/null; " + "HERMES_UI_SESSION_ID HERMES_ORIGIN_TURN_ID HERMES_DELEGATION_SYNC_ONLY " + "2>/dev/null; " "export -p; " ") || true; } " f"> {tmp_path}"