diff --git a/tests/tools/test_subagent_steer.py b/tests/tools/test_subagent_steer.py new file mode 100644 index 0000000000000..c3e94dec07ec6 --- /dev/null +++ b/tests/tools/test_subagent_steer.py @@ -0,0 +1,198 @@ +"""steer_subagent — redirecting a live delegated child without stopping it. + +Registry-level coverage for the delegation-side mirror of +interrupt_subagent(): text reaches the live child's AIAgent.steer(), and +every failure shape (unknown id, dead record, empty text, a steer that +raises) degrades to False instead of an exception. Also covers the +missed-steer retention race (a child that finishes before the drain) and +the subagent.steer gateway RPC that fronts the helper. +""" + +from tools.delegate_tool import ( + _register_subagent, + _unregister_subagent, + steer_subagent, +) + + +class _StubAgent: + def __init__(self, accept: bool = True, boom: bool = False): + self.accept = accept + self.boom = boom + self.steered: list[str] = [] + + def steer(self, text: str) -> bool: + if self.boom: + raise RuntimeError("steer exploded") + self.steered.append(text) + return self.accept + + +def _with_registered(sid: str, agent) -> None: + _register_subagent( + { + "subagent_id": sid, + "parent_id": "root", + "depth": 1, + "goal": "test goal", + "status": "running", + "agent": agent, + } + ) + + +def test_steer_reaches_the_live_child(): + agent = _StubAgent() + _with_registered("sid-steer-1", agent) + try: + assert steer_subagent("sid-steer-1", "focus on pricing instead") is True + assert agent.steered == ["focus on pricing instead"] + finally: + _unregister_subagent("sid-steer-1") + + +def test_unknown_subagent_is_false_not_an_error(): + assert steer_subagent("sid-not-registered", "hello") is False + + +def test_empty_text_is_refused_without_a_lookup(): + agent = _StubAgent() + _with_registered("sid-steer-2", agent) + try: + assert steer_subagent("sid-steer-2", " ") is False + assert agent.steered == [] + finally: + _unregister_subagent("sid-steer-2") + + +def test_record_without_live_agent_is_false(): + _register_subagent({"subagent_id": "sid-steer-3", "status": "running", "agent": None}) + try: + assert steer_subagent("sid-steer-3", "hello") is False + finally: + _unregister_subagent("sid-steer-3") + + +def test_agent_rejection_propagates_as_false(): + agent = _StubAgent(accept=False) + _with_registered("sid-steer-4", agent) + try: + assert steer_subagent("sid-steer-4", "hello") is False + finally: + _unregister_subagent("sid-steer-4") + + +def test_exception_in_steer_degrades_to_false(): + agent = _StubAgent(boom=True) + _with_registered("sid-steer-5", agent) + try: + assert steer_subagent("sid-steer-5", "hello") is False + finally: + _unregister_subagent("sid-steer-5") + + +class TestMissedSteerRetention: + """The final-answer race: a steer with no boundary left is NAMED, not lost.""" + + def test_pending_steer_lands_in_completion_entry(self): + import json + from unittest.mock import MagicMock, patch + + from tools.delegate_tool import delegate_task + + parent = MagicMock() + parent._delegate_depth = 0 + parent.model = "test-model" + parent.interactive_mode = False + + with patch("run_agent.AIAgent") as MockAgent: + mock_child = MagicMock() + mock_child.model = "test-model" + mock_child.session_prompt_tokens = 0 + mock_child.session_completion_tokens = 0 + mock_child.run_conversation.return_value = { + "final_response": "done", + "completed": True, + "interrupted": False, + "api_calls": 1, + "messages": [], + # The finalizer's undelivered-steer hand-back + # (turn_finalizer.py "pending_steer"). + "pending_steer": "focus on pricing instead", + } + MockAgent.return_value = mock_child + + result = json.loads(delegate_task(goal="race test", parent_agent=parent)) + entry = result["results"][0] + + assert entry["missed_steer"] == "focus on pricing instead" + assert "steer did not land" in entry["summary"] + assert "focus on pricing instead" in entry["summary"] + # The race must not corrupt the outcome of the work itself. + assert entry["status"] == "completed" + + def test_no_pending_steer_leaves_entry_untouched(self): + import json + from unittest.mock import MagicMock, patch + + from tools.delegate_tool import delegate_task + + parent = MagicMock() + parent._delegate_depth = 0 + parent.model = "test-model" + parent.interactive_mode = False + + with patch("run_agent.AIAgent") as MockAgent: + mock_child = MagicMock() + mock_child.model = "test-model" + mock_child.session_prompt_tokens = 0 + mock_child.session_completion_tokens = 0 + mock_child.run_conversation.return_value = { + "final_response": "done", + "completed": True, + "interrupted": False, + "api_calls": 1, + "messages": [], + } + MockAgent.return_value = mock_child + + result = json.loads(delegate_task(goal="clean run", parent_agent=parent)) + entry = result["results"][0] + + assert "missed_steer" not in entry + assert "steer did not land" not in entry["summary"] + + +class TestSubagentSteerRPC: + """subagent.steer gateway RPC — the programmatic caller beside subagent.interrupt.""" + + def _call(self, params: dict) -> dict: + import tui_gateway.server as srv + + return srv._methods["subagent.steer"](1, params) + + def test_missing_subagent_id_is_4000(self): + envelope = self._call({"text": "hello"}) + assert envelope["error"]["code"] == 4000 + + def test_empty_text_is_4002(self): + envelope = self._call({"subagent_id": "sid-rpc-1", "text": " "}) + assert envelope["error"]["code"] == 4002 + + def test_live_child_queues_and_receives_text(self): + agent = _StubAgent() + _with_registered("sid-rpc-2", agent) + try: + envelope = self._call({"subagent_id": "sid-rpc-2", "text": "check the edge cases"}) + assert envelope["result"] == { + "status": "queued", + "subagent_id": "sid-rpc-2", + "text": "check the edge cases", + } + assert agent.steered == ["check the edge cases"] + finally: + _unregister_subagent("sid-rpc-2") + + def test_unknown_child_is_rejected_not_an_error(self): + envelope = self._call({"subagent_id": "sid-rpc-gone", "text": "hello"}) + assert envelope["result"]["status"] == "rejected" diff --git a/tools/delegate_tool.py b/tools/delegate_tool.py index eb4df253e77b9..6d90b6f9b9b00 100644 --- a/tools/delegate_tool.py +++ b/tools/delegate_tool.py @@ -205,6 +205,38 @@ def interrupt_subagent(subagent_id: str) -> bool: return True +def steer_subagent(subagent_id: str, text: str) -> bool: + """Queue steering text into a single running subagent without stopping it. + + The redirection-side mirror of interrupt_subagent(): resolves the live + child in the registry and calls AIAgent.steer(), which appends the text + to the child's last tool result at its next iteration boundary — the + current tool call is never cut. Returns True if a matching subagent + QUEUED the text; False for an unknown id, a record with no live agent, + or empty text. + + Queued is not delivered: a child already past its final tool batch has + no boundary left to drain into. That race is surfaced, not swallowed — + the finalizer returns the undelivered text as ``pending_steer`` and + ``_run_single_child`` retains it in the completion entry as + ``missed_steer`` with a note appended to the summary. + """ + if not text or not text.strip(): + return False + with _active_subagents_lock: + record = _active_subagents.get(subagent_id) + if not record: + return False + agent = record.get("agent") + if agent is None: + return False + try: + return bool(agent.steer(text)) + except Exception as exc: + logger.debug("steer_subagent(%s) failed: %s", subagent_id, exc) + return False + + def list_active_subagents() -> List[Dict[str, Any]]: """Snapshot of the currently running subagent tree. @@ -2438,6 +2470,21 @@ def _run_single_child( if status == "failed": entry["error"] = result.get("error", "Subagent did not produce a response.") + # A steer that queued after the child's final assistant turn had no + # tool batch left to drain into. The finalizer hands the undelivered + # text back (turn_finalizer.py "pending_steer"); retain it here so the + # parent sees the steer was MISSED rather than silently absorbed — + # steer_subagent() returning True means "queued", and this is where a + # queued-but-never-delivered steer gets named. + _missed_steer = result.get("pending_steer") + if isinstance(_missed_steer, str) and _missed_steer.strip(): + entry["missed_steer"] = _missed_steer + _miss_note = ( + "[steer did not land — the subagent finished before it could " + f"be delivered: {_missed_steer}]" + ) + entry["summary"] = f"{summary}\n\n{_miss_note}" if summary else _miss_note + # Cross-agent file-state reminder. If this subagent wrote any # files the parent had already read, surface it so the parent # knows to re-read before editing — the scenario that motivated diff --git a/tui_gateway/methods_session.py b/tui_gateway/methods_session.py index bdbcd1cfafafe..c0b2ac74f4059 100644 --- a/tui_gateway/methods_session.py +++ b/tui_gateway/methods_session.py @@ -2934,6 +2934,37 @@ def _(rid, params: dict) -> dict: return _ok(rid, {"found": ok, "subagent_id": subagent_id}) +@method("subagent.steer") +def _(rid, params: dict) -> dict: + """Queue steering text into a live delegated child without stopping it. + + The redirection-side mirror of subagent.interrupt: resolves the child in + the delegation registry and calls AIAgent.steer(), which appends the text + to the child's last tool result at its next iteration boundary — the + in-flight tool call is never cut. "queued" is not "delivered": a child + already past its final tool batch has no boundary left to drain into, + and that race surfaces as ``missed_steer`` on the parent's completion + entry instead of being silently dropped. + """ + from tools.delegate_tool import steer_subagent + + subagent_id = str(params.get("subagent_id") or "").strip() + if not subagent_id: + return _err(rid, 4000, "subagent_id required") + text = (params.get("text") or "").strip() + if not text: + return _err(rid, 4002, "text is required") + queued = steer_subagent(subagent_id, text) + return _ok( + rid, + { + "status": "queued" if queued else "rejected", + "subagent_id": subagent_id, + "text": text, + }, + ) + + @method("spawn_tree.save") def _(rid, params: dict) -> dict: session_id = str(params.get("session_id") or "").strip() diff --git a/website/docs/developer-guide/programmatic-integration.md b/website/docs/developer-guide/programmatic-integration.md index cc4f8a98bfb5b..251c6bec30e3c 100644 --- a/website/docs/developer-guide/programmatic-integration.md +++ b/website/docs/developer-guide/programmatic-integration.md @@ -50,7 +50,8 @@ clarify.respond sudo.respond secret.respond approval.respond config.set / config.get commands.catalog command.resolve command.dispatch cli.exec reload.mcp reload.env process.stop -delegation.status subagent.interrupt spawn_tree.save / list / load +delegation.status subagent.interrupt subagent.steer +spawn_tree.save / list / load terminal.resize clipboard.paste image.attach ``` diff --git a/website/docs/user-guide/features/delegation.md b/website/docs/user-guide/features/delegation.md index 625ecd7ec1f00..53be6b2fa896a 100644 --- a/website/docs/user-guide/features/delegation.md +++ b/website/docs/user-guide/features/delegation.md @@ -269,6 +269,24 @@ A delegation the stall monitor has flagged shows as children show their quiet time so you can tell "slow" from "stuck" at a glance. +## Steering a Running Subagent + +Interrupting a child throws away its in-flight work; often you just want to redirect it. `steer_subagent(subagent_id, text)` in `tools/delegate_tool.py` is the redirection-side mirror of `interrupt_subagent()`: it queues text into a live child through the same mechanism as [`/steer`](/reference/slash-commands) — the text is appended to the child's last tool result at its next iteration boundary, the in-flight tool call is never cut, and the child sees it as an out-of-band user message. Programmatic hosts reach it through the `subagent.steer` gateway RPC, which sits beside `subagent.interrupt`: + +```json +{"method": "subagent.steer", "params": {"subagent_id": "sa-0-1a2b3c4d", "text": "focus on pricing instead"}} +``` + +Subagent ids come from `delegation.status` (or `list_active_subagents()`) — the same place `subagent.interrupt` gets them. + +**Queued is not delivered.** A `"queued"` response means the text is staged, not that the child has seen it. A child that has already produced its final answer has no tool batch left to drain the steer into. That race is surfaced instead of swallowed: the turn finalizer hands the undelivered text back as `pending_steer`, and the completion entry the parent receives retains it as `missed_steer`, with a note appended to the summary: + +``` +[steer did not land — the subagent finished before it could be delivered: focus on pricing instead] +``` + +So the parent (or the operator driving it) can tell a steered child from one that finished on the old instructions, and re-issue the guidance as a follow-up instead of trusting that it landed. + ## Live Transcripts Every `delegate_task` dispatch also creates one **append-only, human-readable log per task** so you (or the parent agent) can watch a subagent work in real time instead of waiting for the consolidated summary: