From 6f163252cb1991910d1be461ec4abcf4d12aa682 Mon Sep 17 00:00:00 2001 From: Mustafa Date: Sun, 12 Jul 2026 14:23:29 +0100 Subject: [PATCH] fix(kanban): keep delegated results in worker turn Dispatcher-spawned Kanban workers are finite one-shot processes, so detached delegation completions can outlive their only consumer. Mark that runtime as unable to deliver async completions and reuse the synchronous delegation fallback, returning required child results before the worker exits.\n\nAlso make unsupported-session notes runtime-generic and cover the delayed-child lifecycle regression.\n\nRefs #63169 --- gateway/session_context.py | 29 +++++--- .../gateway/test_async_delivery_capability.py | 8 +++ tests/tools/test_async_delegation.py | 69 ++++++++++++++++++- tools/delegate_tool.py | 22 +++--- tools/terminal_tool.py | 16 ++--- 5 files changed, 112 insertions(+), 32 deletions(-) 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')." )