From 5c705ae8925f0a452804be9b5c2b10aa302f5685 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Thu, 13 Aug 2026 03:33:32 -0700 Subject: [PATCH] feat(delegation): live orchestration of running subagents via delegate_task action param MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 9 + agent/tool_executor.py | 5 +- agent/tool_guardrails.py | 12 +- run_agent.py | 3 + tests/tools/test_delegate_control_actions.py | 338 +++++++++++++++++++ tools/delegate_tool.py | 243 ++++++++++++- 6 files changed, 606 insertions(+), 4 deletions(-) create mode 100644 tests/tools/test_delegate_control_actions.py diff --git a/agent/display.py b/agent/display.py index 63f101ae3d9f..2880cecccb84 100644 --- a/agent/display.py +++ b/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) diff --git a/agent/tool_executor.py b/agent/tool_executor.py index 1438cd08fcd2..c1697a6c3199 100644 --- a/agent/tool_executor.py +++ b/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] diff --git a/agent/tool_guardrails.py b/agent/tool_guardrails.py index 444ce3739596..6e6a9cd8f54f 100644 --- a/agent/tool_guardrails.py +++ b/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) diff --git a/run_agent.py b/run_agent.py index 20b8e73ad41d..12e7da647ebb 100644 --- a/run_agent.py +++ b/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, ) diff --git a/tests/tools/test_delegate_control_actions.py b/tests/tools/test_delegate_control_actions.py new file mode 100644 index 000000000000..484760a2d2a0 --- /dev/null +++ b/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" diff --git a/tools/delegate_tool.py b/tools/delegate_tool.py index a045ae7683ab..8021051db937 100644 --- a/tools/delegate_tool.py +++ b/tools/delegate_tool.py @@ -27,6 +27,7 @@ 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 _child_thinking(text: str) -> None: 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 _batch_progress(): "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 @@ def _build_dynamic_schema_overrides() -> dict: "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 @@ def _strip_model_hidden_task_fields(tasks: Any) -> Any: 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,