feat(delegation): live orchestration of running subagents via delegate_task action param

agent/display.py

@@ -477,6 +477,11 @@ def build_tool_preview(tool_name: str, args: dict, max_len: int | None = None) -
 
     # delegate_task: show goal (single) or individual task goals (batch)
     if tool_name == "delegate_task":
+        action = str(args.get("action") or "").strip().lower()
+        if action in ("list", "steer", "stop"):
+            sid = str(args.get("subagent_id") or "").strip()
+            preview = f"{action} {sid}".strip()
+            return _truncate_preview(preview, max_len)
         tasks = args.get("tasks")
         if tasks and isinstance(tasks, list):
             task_count, goals = _delegate_task_goal_parts(tasks, per_goal_len=40)
@@ -1550,6 +1555,10 @@ def _wrap(line: str) -> str:
         code = " ".join(str(args.get("code", "") or "").split())
         return _wrap(f"┊ 🌐 browser   {_trunc(code, 35)}  {dur}")
     if tool_name == "delegate_task":
+        _action = str(args.get("action") or "").strip().lower()
+        if _action in ("list", "steer", "stop"):
+            _sid = str(args.get("subagent_id") or "").strip()
+            return _wrap(f"┊ 🔀 delegate  {_trunc(f'{_action} {_sid}'.strip(), 35)}  {dur}")
         tasks = args.get("tasks")
         if tasks and isinstance(tasks, list):
             task_count, goals = _delegate_task_goal_parts(tasks, per_goal_len=30)

agent/tool_executor.py

@@ -1923,8 +1923,11 @@ def _execute(next_args: dict) -> Any:
             if agent._should_emit_quiet_tool_messages():
                 agent._vprint(f"  {_get_cute_tool_message_impl('setup_mcp', function_args, tool_duration, result=function_result)}")
         elif function_name == "delegate_task":
+            _action_arg = str(function_args.get("action") or "").strip().lower()
             tasks_arg = function_args.get("tasks")
-            if tasks_arg and isinstance(tasks_arg, list):
+            if _action_arg in ("list", "steer", "stop"):
+                spinner_label = f"🔀 subagent {_action_arg}"
+            elif tasks_arg and isinstance(tasks_arg, list):
                 spinner_label = f"🔀 delegating {len(tasks_arg)} tasks · (/agents to monitor)"
             else:
                 goal_preview = (function_args.get("goal") or "")[:30]

agent/tool_guardrails.py

@@ -485,6 +485,11 @@ def _check_loop_cap(
             if not cap:
                 return None
             spawn_count = _subagent_spawn_count(args)
+            if spawn_count == 0:
+                # Control action (list/steer/stop) — spawns nothing. Never
+                # block: once the spawn cap is hit, steering/stopping the
+                # existing children is exactly what should still work.
+                return None
             if self._turn_subagent_count >= cap:
                 decision = ToolGuardrailDecision(
                     action="block",
@@ -616,8 +621,13 @@ def _subagent_spawn_count(args: Mapping[str, Any]) -> int:
     delegate_task runs in one of two modes: a batch (``tasks`` is a non-empty
     list, one child per item) or a single task (``goal``). Count the batch size
     when present, otherwise 1, so the session subagent cap reflects real spawns
-    rather than delegate_task invocations.
+    rather than delegate_task invocations. Control actions (list/steer/stop)
+    spawn nothing and must not consume the cap.
     """
+    if isinstance(args, Mapping):
+        action = str(args.get("action") or "").strip().lower()
+        if action in ("list", "steer", "stop"):
+            return 0
     tasks = args.get("tasks") if isinstance(args, Mapping) else None
     if isinstance(tasks, list) and tasks:
         return len(tasks)

run_agent.py

@@ -7918,6 +7918,9 @@ def _dispatch_delegate_task(self, function_args: dict) -> str:
             max_iterations=function_args.get("max_iterations"),
             role=function_args.get("role"),
             background=(not _is_subagent),
+            action=function_args.get("action"),
+            subagent_id=function_args.get("subagent_id"),
+            message=function_args.get("message"),
             parent_agent=self,
         )