fix: async subagent background=True drops user interrupts to queue (#46864)

gateway/run.py

@@ -3742,6 +3742,13 @@ def _agent_has_active_subagents(running_agent: Any) -> bool:
         untouched. Safe-by-default: returns False on any attribute or
         lock error so a missing/broken parent never blocks the existing
         interrupt path.
+
+        #46864 — ``_executing_tools`` is False when the agent's tool loop
+        has returned (i.e. the parent dispatched background subagents via
+        ``delegate_task(background=True)`` and is no longer actively
+        driving tool calls).  In that case any remaining ``_active_children``
+        are background subagents that should NOT block user interrupts.
+        Only demote when the parent is actively mid-tool-loop (sync children).
         """
         if running_agent is None or running_agent is _AGENT_PENDING_SENTINEL:
             return False
@@ -3755,6 +3762,14 @@ def _agent_has_active_subagents(running_agent: Any) -> bool:
             return False
         if not children:
             return False
+        # #46864 — Background children (delegate_task background=True) remain
+        # in _active_children after the parent's tool loop finishes, but
+        # _executing_tools is False because the parent has returned. Only
+        # sync children (parent blocked mid-tool-loop) should trigger the
+        # demotion. Check _executing_tools to distinguish.
+        executing_tools = getattr(running_agent, "_executing_tools", True)
+        if not executing_tools:
+            return False
         lock = getattr(running_agent, "_active_children_lock", None)
         try:
             if lock is not None:

run_agent.py

@@ -5140,6 +5140,7 @@ def _dispatch_delegate_task(self, function_args: dict) -> str:
             acp_command=function_args.get("acp_command"),
             acp_args=function_args.get("acp_args"),
             role=function_args.get("role"),
+            background=function_args.get("background"),
             parent_agent=self,
         )
 

tests/gateway/test_subagent_protection_30170.py

@@ -202,6 +202,23 @@ def test_accepts_list_tuple_set(self, container: Any) -> None:
         parent._active_children_lock = threading.Lock()
         assert GatewayRunner._agent_has_active_subagents(parent) is True
 
+    def test_returns_false_when_parent_not_executing_tools(self) -> None:
+        """#46864 — Background children (delegate_task background=True) remain
+        in _active_children after the parent's tool loop finishes, but
+        _executing_tools is False because the parent has returned.
+        The helper must return False so the gateway does NOT demote user
+        interrupts to queue for background-only subagents."""
+        parent = _make_parent_with_subagents(children=2)
+        parent._executing_tools = False
+        assert GatewayRunner._agent_has_active_subagents(parent) is False
+
+    def test_returns_false_when_executing_tools_attr_missing(self) -> None:
+        """If _executing_tools is not set (e.g. test stubs), default to True
+        so existing behaviour is preserved."""
+        parent = _make_parent_with_subagents(children=1)
+        # Deliberately don't set _executing_tools — getattr should default to True
+        assert GatewayRunner._agent_has_active_subagents(parent) is True
+
 
 # ──────────────────────────────────────────────────────────────────────
 # _handle_active_session_busy_message — interrupt demotion

tools/delegate_tool.py

@@ -1261,14 +1261,13 @@ def _child_thinking(text: str) -> None:
     if child_pool is not None:
         child._credential_pool = child_pool
 
-    # Register child for interrupt propagation
-    if hasattr(parent_agent, "_active_children"):
-        lock = getattr(parent_agent, "_active_children_lock", None)
-        if lock:
-            with lock:
-                parent_agent._active_children.append(child)
-        else:
-            parent_agent._active_children.append(child)
+    # NOTE: child is NOT registered for interrupt propagation here anymore.
+    # The caller decides whether the child is sync (registered) or background
+    # (async delegated). See the caller in delegate_tool._run() for the
+    # background/sync split — registering here introduces a race window where
+    # _active_children is non-empty before the caller removes it for background
+    # subagents, causing _agent_has_active_subagents() to demote user
+    # interrupts to queue (#46864).
 
     # Announce the spawn immediately — the child may sit in a queue
     # for seconds if max_concurrent_children is saturated, so the TUI
@@ -2316,10 +2315,36 @@ def _async_interrupt(_child=child):
                 dispatch.get("error", "Async delegation could not be scheduled.")
             )
 
+        # Register the child for interrupt propagation (sync path).
+        # _build_child_agent no longer does this — background subagents must
+        # NEVER be added to _active_children (they're lifecycle-managed by the
+        # async-delegation registry). Only sync children need the parent's
+        # interrupt to cascade through them. See #46864.
+        if hasattr(parent_agent, "_active_children"):
+            _ac_lock = getattr(parent_agent, "_active_children_lock", None)
+            if _ac_lock:
+                with _ac_lock:
+                    parent_agent._active_children.append(child)
+            else:
+                parent_agent._active_children.append(child)
+
         result = _run_single_child(0, _t["goal"], child, parent_agent)
         results.append(result)
     else:
         # Batch -- run in parallel with per-task progress lines
+        # Register ALL batch children for interrupt propagation before starting
+        # the thread pool, so interrupt cascades through every child even while
+        # others are still queued in the executor. _run_single_child removes
+        # each child when it finishes (in its finally block).
+        if hasattr(parent_agent, "_active_children"):
+            _ac_lock = getattr(parent_agent, "_active_children_lock", None)
+            for _i, _t, _child in children:
+                if _ac_lock:
+                    with _ac_lock:
+                        parent_agent._active_children.append(_child)
+                else:
+                    parent_agent._active_children.append(_child)
+
         completed_count = 0
         spinner_ref = getattr(parent_agent, "_delegate_spinner", None)