preview(omnio): background-subagent stack integration (#63 + #67 + #68) — do not merge

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)

gateway/run.py

@@ -18158,15 +18158,41 @@ 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(
                             "Watch pattern notification — waking api_server "
                             "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 "<none>",
+                            e.status_code, raw_sid, e,
+                        )
                         return True
                     except Exception as e:
                         logger.warning(
@@ -18201,15 +18227,34 @@ 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(
                     "Watch pattern notification — waking api_server 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:
+                # 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 "<none>",
+                    e.status_code, raw_sid, e,
+                )
                 return True
             except Exception as e:
                 logger.warning(

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