fix(api-server): stop silently promising async delivery on stateless HTTP path

gateway/platforms/api_server.py

@@ -749,6 +749,16 @@ class APIServerAdapter(BasePlatformAdapter):
     and routes them through hermes-agent's AIAgent.
     """
 
+    # Stateless request/response: every route (the OpenAI-spec
+    # /v1/chat/completions and /v1/responses, and the proprietary /v1/runs SSE
+    # stream) tears down its channel when the turn ends. There is no persistent
+    # outbound channel to push a background completion to a client that already
+    # received its response, and ``send()`` is a no-op stub. So async-delivery
+    # tools (terminal notify_on_complete / watch_patterns, delegate_task
+    # background=True) must NOT promise delivery on this path — see
+    # ``async_delivery_supported()``.
+    supports_async_delivery: bool = False
+
     def __init__(self, config: PlatformConfig):
         super().__init__(config, Platform.API_SERVER)
         extra = config.extra or {}
@@ -3655,6 +3665,38 @@ def _concurrency_limited_response(self) -> Optional["web.Response"]:
             )
         return None
 
+    @staticmethod
+    def _bind_api_server_session(
+        *,
+        chat_id: str = "",
+        session_key: str = "",
+        session_id: str = "",
+    ) -> list:
+        """Bind session contextvars for an API-server agent run.
+
+        This is the SINGLE structural chokepoint every API-server agent-entry
+        path must use to seed session context — it hardwires
+        ``platform="api_server"`` and ``async_delivery=False`` so a new route
+        physically cannot reintroduce the silent-no-op bug (#10760) by
+        forgetting to mark the channel as non-delivering. There is no
+        ``async_delivery`` parameter to get wrong; the stateless HTTP path can
+        never wake the agent after the turn ends, on ANY route.
+
+        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
+        CLI or a gateway platform, re-binds fresh and is NOT blocked).
+        """
+        from gateway.session_context import set_session_vars
+
+        return set_session_vars(
+            platform="api_server",
+            chat_id=chat_id,
+            session_key=session_key,
+            session_id=session_id,
+            async_delivery=False,
+        )
+
     async def _run_agent(
         self,
         user_message: str,
@@ -3682,10 +3724,9 @@ async def _run_agent(
         loop = asyncio.get_running_loop()
 
         def _run():
-            from gateway.session_context import clear_session_vars, set_session_vars
+            from gateway.session_context import clear_session_vars
 
-            tokens = set_session_vars(
-                platform="api_server",
+            tokens = self._bind_api_server_session(
                 chat_id=session_id or "",
                 session_key=gateway_session_key or session_id or "",
                 session_id=session_id or "",
@@ -3940,7 +3981,7 @@ def _approval_notify(approval_data: Dict[str, Any]) -> None:
                         pass
 
                 def _run_sync():
-                    from gateway.session_context import clear_session_vars, set_session_vars
+                    from gateway.session_context import clear_session_vars
                     from tools.approval import (
                         register_gateway_notify,
                         reset_current_session_key,
@@ -3956,8 +3997,7 @@ def _run_sync():
                         # contextvars so concurrent runs do not share process
                         # environment state.
                         approval_token = set_current_session_key(approval_session_key)
-                        session_tokens = set_session_vars(
-                            platform="api_server",
+                        session_tokens = self._bind_api_server_session(
                             session_key=approval_session_key,
                         )
                         register_gateway_notify(approval_session_key, _approval_notify)

gateway/platforms/base.py

@@ -1925,6 +1925,22 @@ class BasePlatformAdapter(ABC):
     # preview (see gateway/run.py progress_callback).
     supports_code_blocks: bool = False
 
+    # Whether this adapter can deliver an ASYNC notification back to the agent
+    # AFTER a turn ends — i.e. wake a fresh turn to surface a background
+    # process completion (terminal notify_on_complete / watch_patterns) or a
+    # detached subagent result (delegate_task background=True).
+    #
+    # True for adapters that hold a persistent outbound channel (Telegram,
+    # Discord, Slack, ... — they have a real ``send()`` and the gateway runs
+    # the watcher/drain loops). False for stateless request/response adapters
+    # (the API server): every route closes its channel when the turn ends, so
+    # there is nowhere to push a later completion. The gateway propagates this
+    # into the ``HERMES_SESSION_ASYNC_DELIVERY`` contextvar at session-bind
+    # time; tools read it via ``async_delivery_supported()`` and refuse to make
+    # a delivery promise they can't keep. A new stateless adapter only needs to
+    # set this to False to stay correct-by-default.
+    supports_async_delivery: bool = True
+
     # The command prefix users can always TYPE on this platform to reach
     # Hermes commands.  Default "/" (most platforms deliver "/approve" etc.
     # as plain message text).  Platforms where typing a leading "/" is

gateway/run.py

@@ -12683,6 +12683,16 @@ def _set_session_env(self, context: SessionContext) -> list:
         in a ``finally`` block.
         """
         from gateway.session_context import set_session_vars
+        # Propagate the adapter's async-delivery capability so async tools
+        # (terminal notify_on_complete / watch_patterns, delegate_task
+        # background=True) know whether this channel can wake a later turn.
+        # Default True keeps CLI / unknown paths working; stateless adapters
+        # (api_server) declare supports_async_delivery=False. Use getattr so
+        # bare runners built via object.__new__ (tests) without self.adapters
+        # don't blow up — they simply default to supported.
+        _adapters = getattr(self, "adapters", None) or {}
+        _adapter = _adapters.get(context.source.platform)
+        _async_delivery = getattr(_adapter, "supports_async_delivery", True)
         return set_session_vars(
             platform=context.source.platform.value,
             chat_id=context.source.chat_id,
@@ -12692,6 +12702,7 @@ def _set_session_env(self, context: SessionContext) -> list:
             user_name=str(context.source.user_name) if context.source.user_name else "",
             session_key=context.session_key,
             message_id=str(context.source.message_id) if context.source.message_id else "",
+            async_delivery=_async_delivery,
         )
 
     def _clear_session_env(self, tokens: list) -> None:

gateway/session_context.py

@@ -62,6 +62,27 @@
 # private-chat topic (those lanes route only with thread id + reply anchor).
 _SESSION_MESSAGE_ID: ContextVar = ContextVar("HERMES_SESSION_MESSAGE_ID", 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).
+#
+# True  — CLI (in-process completion_queue drain) and the real gateway
+#         platforms (Telegram/Discord/Slack/...), which hold a persistent
+#         outbound channel and run the watcher/drain loops.
+# False — stateless request/response adapters (the API server: every route,
+#         spec and proprietary, tears down its channel when the turn ends, so
+#         a background completion that finishes later has nowhere to go).
+#
+# Tools that promise async delivery (terminal notify_on_complete /
+# watch_patterns, delegate_task background=True) read this via
+# ``async_delivery_supported()`` and refuse to hand out a promise the channel
+# can't keep — turning a silent no-op into an explicit contract.
+#
+# Default _UNSET => treated as supported, so CLI (which never sets a platform)
+# and any contextvar-unaware path keep working. Stateless adapters opt OUT by
+# setting ``supports_async_delivery = False`` on the adapter class; the gateway
+# propagates that into this contextvar at session-bind time.
+_SESSION_ASYNC_DELIVERY: ContextVar = ContextVar("HERMES_SESSION_ASYNC_DELIVERY", default=_UNSET)
+
 # Cron auto-delivery vars — set per-job in run_job() so concurrent jobs
 # don't clobber each other's delivery targets.
 _CRON_AUTO_DELIVER_PLATFORM: ContextVar = ContextVar("HERMES_CRON_AUTO_DELIVER_PLATFORM", default=_UNSET)
@@ -112,6 +133,7 @@ def set_session_vars(
     session_id: str = "",
     message_id: str = "",
     cwd: str = "",
+    async_delivery: bool = True,
 ) -> list:
     """Set all session context variables and return reset tokens.
 
@@ -122,6 +144,11 @@ def set_session_vars(
     only for API compatibility.
 
     ``cwd`` pins the logical working directory for this context.
+
+    ``async_delivery`` declares whether this session's channel can route a
+    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``.
     """
     tokens = [
         _SESSION_PLATFORM.set(platform),
@@ -134,6 +161,7 @@ def set_session_vars(
         _SESSION_KEY.set(session_key),
         _SESSION_ID.set(session_id),
         _SESSION_MESSAGE_ID.set(message_id),
+        _SESSION_ASYNC_DELIVERY.set(bool(async_delivery)),
     ]
     try:
         from agent.runtime_cwd import set_session_cwd
@@ -168,6 +196,11 @@ def clear_session_vars(tokens: list) -> None:
         _SESSION_MESSAGE_ID,
     ):
         var.set("")
+    # Reset async-delivery capability to the "never set" sentinel rather than a
+    # falsy value: a cleared context should fall back to the default-supported
+    # behavior (CLI / unaware paths), not be mistaken for an opted-out
+    # stateless adapter.
+    _SESSION_ASYNC_DELIVERY.set(_UNSET)
     try:
         from agent.runtime_cwd import clear_session_cwd
 
@@ -200,3 +233,22 @@ def get_session_env(name: str, default: str = "") -> str:
             return value
     # Fall back to os.environ for CLI, cron, and test compatibility
     return os.getenv(name, default)
+
+
+def async_delivery_supported() -> bool:
+    """Whether the current session can deliver a background completion later.
+
+    Returns ``False`` only when the active session was explicitly bound by a
+    stateless adapter (the API server) that cannot route a notification back to
+    the agent after the turn ends. CLI, cron, and the real gateway platforms —
+    and any path that never bound the contextvar — return ``True``.
+
+    Tools that promise async delivery (``terminal`` notify_on_complete /
+    watch_patterns, ``delegate_task`` background=True) consult this before
+    registering a watcher / dispatching a detached child, so they can refuse a
+    promise the channel can't keep instead of silently no-op'ing.
+    """
+    value = _SESSION_ASYNC_DELIVERY.get()
+    if value is _UNSET:
+        return True
+    return bool(value)