fix(kanban): keep delegated results in worker turn

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

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."""

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
 
-

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)
 

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')."
                         )