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

delegate_task gains a control plane: action='list' / 'steer' / 'stop'
let the parent agent see, redirect, and early-stop its own running
subagents mid-flight — the model-facing counterpart of the TUI's
delegation.pause / subagent.interrupt / subagent.steer RPCs.

- action='list': live children of this conversation's spawn tree
  (ids, goal, status, running_seconds, accepting_steer, live
  transcript path). Ownership is enforced via a _delegate_parent_ref
  weakref chain stamped at child build time, so a conversation can
  only control its own descendants, never a sibling tree.
- action='steer': queues text into a running child via the existing
  steer_subagent() registry path (delivered at the child's next tool
  boundary; missed steers surface as missed_steer in the completion).
- action='stop': interrupt_subagent() — child stops at its next
  iteration boundary, partial result still re-enters as a completion.
- Spawn dispatch response now includes subagent_ids + control hint.
- Control actions run synchronously (never backgrounded) and bypass
  the spawn pause gate and depth limit; they also never consume the
  per-turn subagent spawn cap, and remain usable once the cap is hit
  (that is when stop matters most).
- Small-model robustness (found live with gpt-5.4-mini on Nous
  Portal): tasks=[] alongside goal no longer trips the "Batch mode
  requires at least 2 tasks" gate — treated as single-goal.
- CLI display: control calls render as "steer sa-…" / "list" instead
  of an empty goal.

Live-tested E2E on Nous Portal (fable-5 + gpt-5.4-mini): full
spawn→list→steer→stop cycle, plus a steer-efficacy run where the
child acked the steer mid-essay and switched topics before finishing.

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 _get_cute_tool_message(
         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_tool_calls_sequential(agent, assistant_message, messages: list, effe
             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 @@ class ToolCallGuardrailController:
             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 @@ class AIAgent:
             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,
         )
 

tests/tools/test_delegate_control_actions.py

@@ -0,0 +1,338 @@
+"""delegate_task(action=...) — model-facing live orchestration of subagents.
+
+Covers the control plane added to delegate_task: action='list' /
+'steer' / 'stop' resolve against the module-level _active_subagents
+registry, scoped by the _delegate_parent_ref ownership chain so a
+conversation can only control its own spawn tree. Also pins the two
+integration contracts: control actions are synchronous (never
+backgrounded) and never consume the per-turn subagent spawn cap.
+"""
+
+import json
+import weakref
+
+from tools.delegate_tool import (
+    _handle_control_action,
+    _is_descendant_of,
+    _register_subagent,
+    _unregister_subagent,
+    delegate_task,
+)
+
+
+class _StubChild:
+    """Weakref-able stand-in for a live child AIAgent."""
+
+    def __init__(self, parent=None, accept_steer: bool = True):
+        self.steered: list[str] = []
+        self.accept_steer = accept_steer
+        self._live_transcript_path = "/tmp/live/task-0.log"
+        if parent is not None:
+            self._delegate_parent_ref = weakref.ref(parent)
+
+    def steer(self, text: str) -> bool:
+        if not self.accept_steer:
+            return False
+        self.steered.append(text)
+        return True
+
+
+class _StubParent:
+    pass
+
+
+def _register(sid: str, child, **extra) -> None:
+    record = {
+        "subagent_id": sid,
+        "parent_id": None,
+        "depth": 0,
+        "goal": "test goal",
+        "model": "test-model",
+        "started_at": 1000.0,
+        "status": "running",
+        "tool_count": 0,
+        "agent": child,
+    }
+    record.update(extra)
+    _register_subagent(record)
+
+
+# ---------------------------------------------------------------------------
+# Ownership chain
+# ---------------------------------------------------------------------------
+
+
+def test_direct_child_is_descendant():
+    parent = _StubParent()
+    child = _StubChild(parent)
+    assert _is_descendant_of(child, parent) is True
+
+
+def test_grandchild_is_descendant():
+    parent = _StubParent()
+    mid = _StubChild(parent)
+    grandchild = _StubChild(mid)
+    assert _is_descendant_of(grandchild, parent) is True
+
+
+def test_foreign_agent_is_not_descendant():
+    parent = _StubParent()
+    other_parent = _StubParent()
+    foreign = _StubChild(other_parent)
+    assert _is_descendant_of(foreign, parent) is False
+
+
+def test_missing_ref_is_not_descendant():
+    parent = _StubParent()
+    orphan = _StubChild()  # no parent ref
+    assert _is_descendant_of(orphan, parent) is False
+    assert _is_descendant_of(None, parent) is False
+
+
+def test_dead_parent_ref_is_not_descendant():
+    parent = _StubParent()
+    child = _StubChild(parent)
+    del parent
+    import gc
+
+    gc.collect()
+    assert _is_descendant_of(child, _StubParent()) is False
+
+
+# ---------------------------------------------------------------------------
+# action='list'
+# ---------------------------------------------------------------------------
+
+
+def test_list_shows_only_own_children():
+    parent = _StubParent()
+    mine = _StubChild(parent)
+    foreign = _StubChild(_StubParent())
+    _register("sid-ctl-list-1", mine)
+    _register("sid-ctl-list-2", foreign)
+    try:
+        out = json.loads(_handle_control_action("list", None, None, parent))
+        assert out["count"] == 1
+        entry = out["subagents"][0]
+        assert entry["subagent_id"] == "sid-ctl-list-1"
+        assert entry["goal"] == "test goal"
+        assert entry["accepting_steer"] is True
+        assert entry["live_transcript"] == "/tmp/live/task-0.log"
+        # Internal fields must not leak
+        assert "agent" not in entry
+        assert "owner_transport" not in entry
+    finally:
+        _unregister_subagent("sid-ctl-list-1")
+        _unregister_subagent("sid-ctl-list-2")
+
+
+def test_list_empty_registry_has_note():
+    out = json.loads(_handle_control_action("list", None, None, _StubParent()))
+    assert out["count"] == 0
+    assert "note" in out
+
+
+# ---------------------------------------------------------------------------
+# action='steer'
+# ---------------------------------------------------------------------------
+
+
+def test_steer_reaches_owned_child():
+    parent = _StubParent()
+    child = _StubChild(parent)
+    _register("sid-ctl-steer-1", child)
+    try:
+        out = json.loads(
+            _handle_control_action("steer", "sid-ctl-steer-1", "focus on X", parent)
+        )
+        assert out["status"] == "queued"
+        assert child.steered == ["focus on X"]
+    finally:
+        _unregister_subagent("sid-ctl-steer-1")
+
+
+def test_steer_foreign_child_is_refused():
+    parent = _StubParent()
+    foreign = _StubChild(_StubParent())
+    _register("sid-ctl-steer-2", foreign)
+    try:
+        out = _handle_control_action("steer", "sid-ctl-steer-2", "hijack", parent)
+        assert "No live subagent" in out
+        assert foreign.steered == []
+    finally:
+        _unregister_subagent("sid-ctl-steer-2")
+
+
+def test_steer_requires_message():
+    parent = _StubParent()
+    child = _StubChild(parent)
+    _register("sid-ctl-steer-3", child)
+    try:
+        out = _handle_control_action("steer", "sid-ctl-steer-3", "   ", parent)
+        assert "requires a non-empty 'message'" in out
+    finally:
+        _unregister_subagent("sid-ctl-steer-3")
+
+
+def test_steer_requires_subagent_id():
+    out = _handle_control_action("steer", "", "text", _StubParent())
+    assert "requires subagent_id" in out
+
+
+def test_steer_closed_acceptance_is_refused():
+    parent = _StubParent()
+    child = _StubChild(parent)
+    _register("sid-ctl-steer-4", child, accepting_steer=False)
+    try:
+        out = _handle_control_action("steer", "sid-ctl-steer-4", "late", parent)
+        assert "no longer accepting" in out
+        assert child.steered == []
+    finally:
+        _unregister_subagent("sid-ctl-steer-4")
+
+
+# ---------------------------------------------------------------------------
+# action='stop'
+# ---------------------------------------------------------------------------
+
+
+def test_stop_interrupts_owned_child(monkeypatch):
+    import tools.delegate_tool as dt
+
+    parent = _StubParent()
+    child = _StubChild(parent)
+    _register("sid-ctl-stop-1", child)
+    interrupted = []
+    monkeypatch.setattr(
+        dt, "request_hard_interrupt", lambda agent, reason: interrupted.append(agent) or True
+    )
+    try:
+        out = json.loads(
+            _handle_control_action("stop", "sid-ctl-stop-1", None, parent)
+        )
+        assert out["status"] == "interrupt_requested"
+        assert interrupted == [child]
+    finally:
+        _unregister_subagent("sid-ctl-stop-1")
+
+
+def test_stop_foreign_child_is_refused(monkeypatch):
+    import tools.delegate_tool as dt
+
+    parent = _StubParent()
+    foreign = _StubChild(_StubParent())
+    _register("sid-ctl-stop-2", foreign)
+    interrupted = []
+    monkeypatch.setattr(
+        dt, "request_hard_interrupt", lambda agent, reason: interrupted.append(agent) or True
+    )
+    try:
+        out = _handle_control_action("stop", "sid-ctl-stop-2", None, parent)
+        assert "No live subagent" in out
+        assert interrupted == []
+    finally:
+        _unregister_subagent("sid-ctl-stop-2")
+
+
+def test_stop_unknown_id_mentions_completion_path():
+    out = _handle_control_action("stop", "sid-gone", None, _StubParent())
+    assert "No live subagent" in out
+    assert "completion message" in out
+
+
+# ---------------------------------------------------------------------------
+# delegate_task() entrypoint routing
+# ---------------------------------------------------------------------------
+
+
+def test_delegate_task_routes_control_action_before_spawn_machinery():
+    """action='list' must return synchronously without touching spawn paths
+    (no goal/tasks required, no pause gate, no depth checks)."""
+    parent = _StubParent()
+    out = json.loads(delegate_task(action="list", parent_agent=parent))
+    assert out["action"] == "list"
+
+
+def test_delegate_task_control_action_bypasses_spawn_pause():
+    from tools.delegate_tool import set_spawn_paused
+
+    parent = _StubParent()
+    set_spawn_paused(True)
+    try:
+        out = json.loads(delegate_task(action="list", parent_agent=parent))
+        assert out["action"] == "list"
+    finally:
+        set_spawn_paused(False)
+
+
+def test_delegate_task_unknown_action_is_an_error():
+    out = delegate_task(action="pause", goal="g", parent_agent=_StubParent())
+    assert "Unknown action" in out
+
+
+def test_delegate_task_spawn_action_still_validates_goal():
+    out = delegate_task(action="spawn", parent_agent=_StubParent())
+    assert "Provide either 'goal'" in out
+
+
+def test_delegate_task_requires_parent_agent_for_control():
+    out = delegate_task(action="list", parent_agent=None)
+    assert "requires a parent agent" in out
+
+
+def test_empty_tasks_array_with_goal_is_single_task_not_batch_error():
+    """Small models emit tasks=[] alongside goal; that must not trip the
+    'Batch mode requires at least 2 tasks' gate (observed live with
+    gpt-5.4-mini on Nous Portal)."""
+    out = delegate_task(tasks=[], goal="", parent_agent=_StubParent())
+    # Falls through to the single-goal validation, not the batch gate.
+    assert "Provide either 'goal'" in out
+    assert "at least 2 tasks" not in out
+
+
+# ---------------------------------------------------------------------------
+# Guardrail: control actions never consume the spawn cap
+# ---------------------------------------------------------------------------
+
+
+def test_spawn_count_zero_for_control_actions():
+    from agent.tool_guardrails import _subagent_spawn_count
+
+    assert _subagent_spawn_count({"action": "list"}) == 0
+    assert _subagent_spawn_count({"action": "steer", "subagent_id": "x"}) == 0
+    assert _subagent_spawn_count({"action": "stop", "subagent_id": "x"}) == 0
+    # Spawn shapes unchanged
+    assert _subagent_spawn_count({"goal": "g"}) == 1
+    assert _subagent_spawn_count({"action": "spawn", "goal": "g"}) == 1
+    assert _subagent_spawn_count({"tasks": [{"goal": "a"}, {"goal": "b"}]}) == 2
+
+
+def test_control_action_not_blocked_at_spawn_cap():
+    """Once the cap is hit, steer/stop must STILL work — that's when the
+    user most needs to rein children in."""
+    from agent.tool_guardrails import (
+        LoopCapConfig,
+        ToolCallGuardrailConfig,
+        ToolCallGuardrailController,
+    )
+
+    cfg = ToolCallGuardrailConfig(loop_caps=LoopCapConfig(max_subagents=1))
+    ctl = ToolCallGuardrailController(cfg)
+    # Exhaust the cap with a spawn
+    assert ctl.before_call("delegate_task", {"goal": "a"}).action == "allow"
+    # A second spawn is blocked
+    assert ctl.before_call("delegate_task", {"goal": "b"}).action == "block"
+    # Control actions still pass on a fresh controller after cap exhaustion
+    ctl2 = ToolCallGuardrailController(cfg)
+    assert ctl2.before_call("delegate_task", {"goal": "a"}).action == "allow"
+    assert (
+        ctl2.before_call(
+            "delegate_task", {"action": "stop", "subagent_id": "x"}
+        ).action
+        == "allow"
+    )
+    assert (
+        ctl2.before_call("delegate_task", {"action": "list"}).action == "allow"
+    )
+    # And spawns remain blocked afterwards — the control call didn't reset it
+    assert ctl2.before_call("delegate_task", {"goal": "c"}).action == "block"

tools/delegate_tool.py

@@ -27,6 +27,7 @@ logger = logging.getLogger(__name__)
 import os
 import threading
 import time
+import weakref
 from concurrent.futures import (
     TimeoutError as FuturesTimeoutError,
 )
@@ -323,6 +324,152 @@ def list_active_subagents() -> List[Dict[str, Any]]:
         ]
 
 
+def _is_descendant_of(child_agent: Any, parent_agent: Any, max_hops: int = 8) -> bool:
+    """True when *child_agent* sits below *parent_agent* in the spawn tree.
+
+    Walks the ``_delegate_parent_ref`` weakref chain stamped at build time.
+    Identity comparison only — a parent may steer/stop its own children and
+    grandchildren, never a sibling tree owned by another conversation.
+    """
+    if child_agent is None or parent_agent is None:
+        return False
+    cur = child_agent
+    for _ in range(max_hops):
+        ref = getattr(cur, "_delegate_parent_ref", None)
+        ancestor = ref() if callable(ref) else None
+        if ancestor is None:
+            return False
+        if ancestor is parent_agent:
+            return True
+        cur = ancestor
+    return False
+
+
+# Model-facing control actions accepted by delegate_task(action=...).
+# "spawn" (or omitted) keeps the historical spawn semantics.
+_CONTROL_ACTIONS = frozenset({"list", "steer", "stop"})
+
+
+def _handle_control_action(
+    action: str,
+    subagent_id: Optional[str],
+    message: Optional[str],
+    parent_agent: Any,
+) -> str:
+    """Synchronous control plane for delegate_task: list/steer/stop.
+
+    Runs in-turn (never backgrounded) and only over subagents descended from
+    *parent_agent* — the same registry the TUI overlay drives, but scoped so
+    a conversation can only control its own spawn tree.
+    """
+    if action == "list":
+        with _active_subagents_lock:
+            records = list(_active_subagents.values())
+        entries = []
+        for r in records:
+            agent = r.get("agent")
+            if not _is_descendant_of(agent, parent_agent):
+                continue
+            started = r.get("started_at")
+            entries.append(
+                {
+                    "subagent_id": r.get("subagent_id"),
+                    "parent_id": r.get("parent_id"),
+                    "goal": r.get("goal"),
+                    "model": r.get("model"),
+                    "status": r.get("status"),
+                    "running_seconds": (
+                        round(time.time() - started, 1)
+                        if isinstance(started, (int, float))
+                        else None
+                    ),
+                    "accepting_steer": bool(r.get("accepting_steer", False)),
+                    "live_transcript": getattr(agent, "_live_transcript_path", None),
+                }
+            )
+        payload: Dict[str, Any] = {
+            "action": "list",
+            "count": len(entries),
+            "subagents": entries,
+        }
+        if not entries:
+            payload["note"] = (
+                "No live subagents right now. Children that already finished "
+                "have delivered (or will deliver) their results as normal "
+                "completion messages — there is nothing to steer or stop."
+            )
+        return json.dumps(payload, ensure_ascii=False)
+
+    # steer / stop need a resolvable, owned target.
+    sid = (subagent_id or "").strip()
+    if not sid:
+        return tool_error(
+            f"action='{action}' requires subagent_id (from the spawn dispatch "
+            "response or action='list')."
+        )
+    with _active_subagents_lock:
+        record = _active_subagents.get(sid)
+        target_agent = record.get("agent") if record else None
+    if record is None or not _is_descendant_of(target_agent, parent_agent):
+        return tool_error(
+            f"No live subagent '{sid}' in this conversation's spawn tree. It "
+            "may have already finished (its result arrives as a normal "
+            "completion message). Use action='list' to see live children."
+        )
+
+    if action == "stop":
+        if interrupt_subagent(sid):
+            return json.dumps(
+                {
+                    "action": "stop",
+                    "subagent_id": sid,
+                    "status": "interrupt_requested",
+                    "note": (
+                        "The subagent stops at its next iteration boundary "
+                        "(in-flight tool calls are asked to cancel). Its "
+                        "partial result still re-enters the conversation as a "
+                        "completion message — do not wait or poll."
+                    ),
+                },
+                ensure_ascii=False,
+            )
+        return tool_error(
+            f"Could not interrupt '{sid}' — it likely finished in the last "
+            "moment. Its result arrives as a normal completion message."
+        )
+
+    if action == "steer":
+        text = (message or "").strip()
+        if not text:
+            return tool_error(
+                "action='steer' requires a non-empty 'message' describing the "
+                "course correction."
+            )
+        if steer_subagent(sid, text):
+            return json.dumps(
+                {
+                    "action": "steer",
+                    "subagent_id": sid,
+                    "status": "queued",
+                    "note": (
+                        "Steering text queued. The subagent sees it appended "
+                        "to its next tool result — the current tool call is "
+                        "never cut. If the child finishes before a delivery "
+                        "boundary remains, the text is reported back as "
+                        "missed_steer in its completion entry."
+                    ),
+                },
+                ensure_ascii=False,
+            )
+        return tool_error(
+            f"Subagent '{sid}' is no longer accepting steering (finishing or "
+            "already finished). Its result arrives as a normal completion "
+            "message; re-delegate a follow-up task if more work is needed."
+        )
+
+    return tool_error(f"Unknown action '{action}'. Use spawn, list, steer, or stop.")
+
+
 def _extract_output_tail(
     result: Dict[str, Any],
     *,
@@ -1684,6 +1831,16 @@ def _build_child_agent(
     child._parent_subagent_id = parent_subagent_id
     child._subagent_goal = goal
     child._parent_turn_id = getattr(parent_agent, "_current_turn_id", "") or ""
+    # Ownership chain for the model-facing control plane (action=list/steer/
+    # stop): a parent may only control agents whose weakref chain reaches it.
+    # Weakref so a finished parent can be collected while a detached child
+    # record briefly lingers in the registry.
+    try:
+        child._delegate_parent_ref = weakref.ref(parent_agent)
+    except TypeError:
+        # Test doubles (MagicMock et al.) may not be weakref-able; control
+        # actions then simply don't resolve ownership for this child.
+        child._delegate_parent_ref = None
     # Stable sidebar marker: delegate subagent sessions must stay out of
     # session pickers even when a parent delete orphans them (parent_session_id
     # → NULL). Mirrors /branch's ``_branched_from`` pattern — see
@@ -3215,15 +3372,25 @@ def delegate_task(
     role: Optional[str] = None,
     background: Optional[bool] = None,
     output_schema: Optional[Dict[str, Any]] = None,
+    action: Optional[str] = None,
+    subagent_id: Optional[str] = None,
+    message: Optional[str] = None,
     parent_agent=None,
 ) -> str:
     """
-    Spawn one or more child agents to handle delegated tasks.
+    Spawn one or more child agents to handle delegated tasks, or control
+    already-running ones.
 
-    Supports two modes:
+    Spawn modes (action='spawn' or omitted):
       - Single: provide goal (+ optional context and role)
       - Batch:  provide tasks array [{goal, context, role}, ...]
 
+    Control modes (synchronous, never backgrounded):
+      - action='list'  -> live children of this conversation's spawn tree
+      - action='steer' -> queue course-correction text into a running child
+                          (subagent_id + message)
+      - action='stop'  -> interrupt a running child early (subagent_id)
+
     The 'role' parameter controls whether a child can further delegate:
     'leaf' (default) cannot; 'orchestrator' retains the delegation
     toolset and can spawn its own workers, bounded by
@@ -3234,6 +3401,19 @@ def delegate_task(
     if parent_agent is None:
         return tool_error("delegate_task requires a parent agent context.")
 
+    # ── Control plane: list/steer/stop run synchronously and return here.
+    # They never spawn, so they bypass the pause gate, depth limit, and the
+    # async dispatch machinery entirely.
+    normalized_action = (action or "").strip().lower()
+    if normalized_action in _CONTROL_ACTIONS:
+        return _handle_control_action(
+            normalized_action, subagent_id, message, parent_agent
+        )
+    if normalized_action and normalized_action != "spawn":
+        return tool_error(
+            f"Unknown action '{action}'. Use spawn (default), list, steer, or stop."
+        )
+
     # Operator-controlled kill switch — lets the TUI freeze new fan-out
     # when a runaway tree is detected, without interrupting already-running
     # children.  Cleared via the matching `delegation.pause` RPC.
@@ -3302,6 +3482,13 @@ def delegate_task(
     if recovered_tasks is not None:
         tasks = recovered_tasks
 
+    # Small models frequently emit an empty tasks array ([]) alongside a
+    # single goal. Treat that as "no batch" instead of letting the batch
+    # quality gate below reject the goal-derived single task ("Batch mode
+    # requires at least 2 tasks") — the intent is unambiguous.
+    if isinstance(tasks, list) and not tasks:
+        tasks = None
+
     if tasks and isinstance(tasks, list):
         if len(tasks) > max_children:
             return tool_error(
@@ -3858,6 +4045,18 @@ def delegate_task(
                 "goals": _goals,
                 "note": note,
             }
+            _sids = [
+                getattr(_c, "_subagent_id", None) for _c in _child_agents
+            ]
+            if any(isinstance(s, str) and s for s in _sids):
+                payload["subagent_ids"] = _sids
+                payload["control_hint"] = (
+                    "While a child runs you can orchestrate it live with this "
+                    "same tool: delegate_task(action='list') to see live "
+                    "children, action='steer' with subagent_id + message to "
+                    "redirect one, action='stop' with subagent_id to end one "
+                    "early."
+                )
             if live_paths:
                 payload["live_transcripts"] = list(live_paths)
                 payload["live_transcripts_hint"] = (
@@ -4168,6 +4367,11 @@ def _build_top_level_description() -> str:
         "transcript paths, and the completed result (one consolidated message "
         "for a batch) re-enters the conversation on its own. Do NOT wait or "
         "poll; continue other work.\n\n"
+        "LIVE ORCHESTRATION: while children run, this tool also controls "
+        "them — action='list' (live children + ids), action='steer' "
+        "(subagent_id + message, redirect without stopping), action='stop' "
+        "(subagent_id, end early; partial result still returns). Steer when "
+        "a live transcript shows a child drifting.\n\n"
         "USE FOR: reasoning-heavy subtasks, work that would flood your context "
         "with intermediate data, or independent parallel workstreams.\n"
         "DO NOT USE FOR (use these instead):\n"
@@ -4365,6 +4569,38 @@ DELEGATE_TASK_SCHEMA = {
                     "backward compatibility."
                 ),
             },
+            "action": {
+                "type": "string",
+                "enum": ["spawn", "list", "steer", "stop"],
+                "description": (
+                    "Default 'spawn' (omit for normal delegation). Live "
+                    "orchestration of running subagents: 'list' shows this "
+                    "conversation's live children (ids, goals, status, "
+                    "transcript paths); 'steer' queues course-correction text "
+                    "into one child (requires subagent_id + message) without "
+                    "stopping it; 'stop' ends one child early (requires "
+                    "subagent_id) — its partial result still returns as a "
+                    "completion message. Control actions return immediately; "
+                    "goal/tasks are ignored when action is not 'spawn'."
+                ),
+            },
+            "subagent_id": {
+                "type": "string",
+                "description": (
+                    "Target for action='steer'/'stop'. Ids are returned in the "
+                    "spawn dispatch response (subagent_ids) and by "
+                    "action='list'."
+                ),
+            },
+            "message": {
+                "type": "string",
+                "description": (
+                    "For action='steer': the course correction. Be directive "
+                    "and specific — the child sees it appended to its next "
+                    "tool result mid-run (e.g. \"Stop exploring X; focus on Y "
+                    "and return early results\")."
+                ),
+            },
         },
         "required": [],
     },
@@ -4426,6 +4662,9 @@ registry.register(
         role=args.get("role"),
         background=_model_background_value(args, kw.get("parent_agent")),
         output_schema=args.get("output_schema"),
+        action=args.get("action"),
+        subagent_id=args.get("subagent_id"),
+        message=args.get("message"),
         parent_agent=kw.get("parent_agent"),
     ),
     check_fn=check_delegate_requirements,