diff --git a/gateway/session_context.py b/gateway/session_context.py index a8e7c027fecb..a792726e47e1 100644 --- a/gateway/session_context.py +++ b/gateway/session_context.py @@ -96,12 +96,11 @@ _SESSION_PROFILE: ContextVar = ContextVar("HERMES_SESSION_PROFILE", default=_UNS # 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 @@ -355,18 +354,29 @@ def declare_stateless_channel() -> None: def async_delivery_supported() -> bool: """Whether the current session can deliver a background completion later. - Returns ``False`` when the active session was bound by a stateless channel: - an adapter that cannot route a notification back after the turn ends (the - API server), or a one-shot runner that exits after its final response - (``hermes -z``, cron — see :func:`declare_stateless_channel`). The real - gateway platforms, the interactive CLI, and any path that never bound the - contextvar return ``True``. + Returns ``False`` for finite runtimes that can end before a detached result + is delivered: sessions explicitly bound by a stateless channel — an adapter + that cannot route a notification back after the turn ends (the API server), + or a one-shot runner that exits after its final response (``hermes -z``, + cron — see :func:`declare_stateless_channel`) — and dispatcher-spawned + Kanban workers (identified by ``HERMES_KANBAN_TASK``), which are one-shot + ``chat -q`` subprocesses. The real gateway platforms, the interactive CLI, + and any other 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. """ + 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 4f63f2ed408b..65b9691e3ab4 100644 --- a/tests/gateway/test_async_delivery_capability.py +++ b/tests/gateway/test_async_delivery_capability.py @@ -76,6 +76,14 @@ class TestAsyncDeliverySupported: 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 37aae286394a..0c7bd44c1b89 100644 --- a/tests/tools/test_async_delegation.py +++ b/tests/tools/test_async_delegation.py @@ -487,6 +487,74 @@ def test_delegate_task_background_routes_async_and_does_not_block(monkeypatch): 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. @@ -872,4 +940,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 e64da983f59d..ab3224f48163 100644 --- a/tools/delegate_tool.py +++ b/tools/delegate_tool.py @@ -2910,13 +2910,11 @@ def delegate_task( 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 @@ -2926,16 +2924,16 @@ def delegate_task( 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 in this session — it cannot " "receive a detached subagent result after the turn ends (a " - "one-shot runner such as `hermes -z` or a cron job, or a " - "stateless HTTP endpoint). The subagent(s) ran SYNCHRONOUSLY " - "and the result is included above." + "one-shot runner such as `hermes -z`, a cron job, a Kanban " + "worker, or a stateless HTTP endpoint). 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 57422299d0de..859d30a76497 100644 --- a/tools/terminal_tool.py +++ b/tools/terminal_tool.py @@ -2612,12 +2612,10 @@ 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 @@ -2625,8 +2623,9 @@ def terminal_tool( result_data["notify_unsupported"] = ( "notify_on_complete / watch_patterns are not available in " "this session — it cannot receive an async completion after " - "the turn ends (a one-shot runner such as `hermes -z` or a " - "cron job, or a stateless HTTP endpoint). The process is " + "the turn ends (a one-shot runner such as `hermes -z`, a " + "cron job, a Kanban worker, or a stateless HTTP endpoint). " + "The process is " "running in the background; retrieve its result with " "process(action='poll') or process(action='wait')." )