diff --git a/gateway/session_context.py b/gateway/session_context.py index 2214c2a568839..af7bd1fa3a865 100644 --- a/gateway/session_context.py +++ b/gateway/session_context.py @@ -96,12 +96,11 @@ def session_context_engaged() -> bool: # 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). +# True — long-lived CLI sessions (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 — finite runtimes that can end before a detached completion returns: +# stateless API-server requests and dispatcher-spawned Kanban workers. # # Tools that promise async delivery (terminal notify_on_complete / # watch_patterns, delegate_task background=True) read this via @@ -330,16 +329,26 @@ def get_session_env(name: str, default: str = "") -> str: 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``. + Returns ``False`` for finite runtimes that can end before a detached result + is delivered. That includes dispatcher-spawned Kanban workers (identified + by ``HERMES_KANBAN_TASK``) and sessions explicitly bound by stateless + adapters such as the API server. Long-lived CLI and gateway sessions — and + contextvar-unaware paths — otherwise 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. """ + import os + + # A Kanban worker is a one-shot subprocess. Its parent session and process + # disappear after the quiet turn returns, so a completion queued later has + # no durable consumer even though an ordinary CLI session can drain that + # queue. Force tools onto their existing synchronous/polling fallbacks. + if os.environ.get("HERMES_KANBAN_TASK"): + return False + value = _SESSION_ASYNC_DELIVERY.get() if value is _UNSET: return True diff --git a/tests/gateway/test_async_delivery_capability.py b/tests/gateway/test_async_delivery_capability.py index 084d4dbdf32cb..8e5ec8b97f494 100644 --- a/tests/gateway/test_async_delivery_capability.py +++ b/tests/gateway/test_async_delivery_capability.py @@ -75,6 +75,14 @@ def test_omitted_arg_defaults_supported(self): finally: clear_session_vars(tokens) + def test_dispatcher_spawned_kanban_worker_is_unsupported(self, monkeypatch): + """A one-shot Kanban worker cannot receive a detached completion + after its process exits, even when its CLI session otherwise defaults + to supporting async delivery.""" + monkeypatch.setenv("HERMES_KANBAN_TASK", "t_review") + + assert async_delivery_supported() is False + def test_clear_resets_to_default_supported(self): """A cleared context must fall back to default-supported, NOT be mistaken for an opted-out stateless adapter.""" diff --git a/tests/tools/test_async_delegation.py b/tests/tools/test_async_delegation.py index 7714d3c8c08a3..89cd0a440c935 100644 --- a/tests/tools/test_async_delegation.py +++ b/tests/tools/test_async_delegation.py @@ -291,6 +291,74 @@ def slow_child(task_index, goal, child=None, parent_agent=None, **kw): assert "the real task" in text +def test_delegate_task_background_waits_inside_kanban_worker(monkeypatch): + """A dispatcher-spawned Kanban worker is a finite process, so a required + delegated result must return in-turn instead of becoming an orphaned + background completion after the parent exits.""" + import json + from unittest.mock import MagicMock + import tools.delegate_tool as dt + + monkeypatch.setenv("HERMES_KANBAN_TASK", "t_review") + + parent = MagicMock() + parent._delegate_depth = 0 + parent.session_id = "kanban-worker-session" + parent._interrupt_requested = False + parent._active_children = [] + parent._active_children_lock = None + fake_child = MagicMock() + fake_child._delegate_role = "leaf" + + started = threading.Event() + release = threading.Event() + + def delayed_child(task_index, goal, child=None, parent_agent=None, **kw): + started.set() + release.wait(timeout=5) + return { + "task_index": task_index, + "status": "completed", + "summary": "review approved", + "api_calls": 1, + "duration_seconds": 0.1, + "model": "m", + "exit_reason": "completed", + } + + creds = { + "model": "m", "provider": None, "base_url": None, "api_key": None, + "api_mode": None, "command": None, "args": None, + } + monkeypatch.setattr(dt, "_build_child_agent", lambda **kw: fake_child) + monkeypatch.setattr(dt, "_run_single_child", delayed_child) + monkeypatch.setattr(dt, "_resolve_delegation_credentials", lambda *a, **k: creds) + + captured = {} + + def call_delegate(): + captured["output"] = dt.delegate_task( + goal="independent review", + background=True, + parent_agent=parent, + ) + + caller = threading.Thread(target=call_delegate) + caller.start() + assert started.wait(timeout=2) + assert caller.is_alive(), "Kanban delegate_task returned before its child finished" + assert ad.active_count() == 0 + + release.set() + caller.join(timeout=5) + assert not caller.is_alive() + + parsed = json.loads(captured["output"]) + assert parsed["results"][0]["summary"] == "review approved" + assert "SYNCHRONOUSLY" in parsed["note"] + assert process_registry.completion_queue.empty() + + def test_delegate_task_background_uses_live_tui_agent_session_id(monkeypatch): """TUI async delegation must route to the live/compressed agent id. @@ -676,4 +744,3 @@ def test_gateway_cli_origin_event_left_unrouted(): runner._enrich_async_delegation_routing(evt) assert "platform" not in evt - diff --git a/tools/delegate_tool.py b/tools/delegate_tool.py index 24c2cd3325d5b..1907c78a7f400 100644 --- a/tools/delegate_tool.py +++ b/tools/delegate_tool.py @@ -2796,13 +2796,11 @@ def _execute_and_aggregate() -> dict: from tools.async_delegation import dispatch_async_delegation_batch from tools.approval import get_current_session_key - # Stateless request/response sessions (the API server / WebUI path) - # cannot route a detached subagent result back to the agent after the - # turn ends — there is no persistent channel and the adapter's send() - # is a no-op, so a background dispatch would silently never re-enter the - # conversation (issue #10760). Fall back to SYNCHRONOUS execution: the - # work still runs and its result returns in this same response, which is - # strictly better than a handle that never resolves. Mirrors the + # Finite sessions cannot route a detached subagent result back to the + # agent after their turn/process ends. This includes stateless HTTP + # requests (#10760) and one-shot Kanban workers (#63169). Fall back to + # SYNCHRONOUS execution so the result returns in this same turn instead + # of handing out a handle with no durable consumer. Mirrors the # pool-at-capacity inline fallback below. try: from gateway.session_context import async_delivery_supported @@ -2812,15 +2810,15 @@ def _execute_and_aggregate() -> dict: if not _async_ok: logger.info( "delegate_task: async delivery unsupported on this session " - "(stateless HTTP API); running the batch synchronously instead." + "runtime; running the batch synchronously instead." ) _sync_result = _execute_and_aggregate() if isinstance(_sync_result, dict): _sync_result["note"] = ( - "background=true is not available on this endpoint (stateless " - "HTTP API — no channel to deliver a detached subagent result " - "after the turn ends), so the subagent(s) ran SYNCHRONOUSLY and " - "the result is included above." + "background=true is not available in this session because it " + "cannot deliver a detached subagent result after the turn ends, " + "so the subagent(s) ran SYNCHRONOUSLY and the result is included " + "above." ) return json.dumps(_sync_result, ensure_ascii=False) diff --git a/tools/terminal_tool.py b/tools/terminal_tool.py index 44ef03af78878..a76a2146f0893 100644 --- a/tools/terminal_tool.py +++ b/tools/terminal_tool.py @@ -2516,20 +2516,18 @@ def terminal_tool( get_session_env as _gse, ) - # Stateless request/response sessions (the API server / - # WebUI path) cannot route a completion back to the agent - # after the turn ends — there is no persistent channel and - # send() is a no-op. Registering a watcher there silently - # no-ops (issue #10760). Refuse the promise instead: drop - # the flags and tell the agent to poll. + # Finite sessions (stateless HTTP requests and one-shot + # Kanban workers) cannot route a completion back to the + # agent after the turn/process ends. Refuse the promise: + # drop the flags and tell the agent to poll. if not _async_ok(): notify_on_complete = False watch_patterns = None result_data["notify_on_complete"] = False result_data["notify_unsupported"] = ( - "notify_on_complete / watch_patterns are not available on " - "this endpoint (stateless HTTP API — no channel to deliver " - "an async completion after the turn ends). The process is " + "notify_on_complete / watch_patterns are not available in " + "this session because it cannot deliver an async completion " + "after the turn ends. The process is " "running in the background; retrieve its result with " "process(action='poll') or process(action='wait')." )