fix(delegation): surface a child's undelivered steer instead of dropping it
The turn finalizer already hands back steer text that queued after the final tool batch — result["pending_steer"], with the comment "hand it back to the caller so it can be delivered as the next user turn instead of being silently lost." Every interactive surface honors that contract (cli.py, gateway/run.py, tui_gateway/server.py all requeue it). The delegation layer doesn't: _run_single_child never reads it, so a steer queued into a delegated child that finishes first vanishes with no trace in the completion entry. There is also no sanctioned sender: the registry has interrupt_subagent() but no redirection-side mirror, and session.steer cannot reach children (lazy watch sessions have agent=None, so it 4010s). Complete the contract for delegated children — both halves: - steer_subagent(subagent_id, text): redirection-side mirror of interrupt_subagent(). Resolves the live child in _active_subagents and queues text via AIAgent.steer(). True means queued, not delivered. - missed_steer retention: when the child's result carries pending_steer, _run_single_child names it on the completion entry (missed_steer field plus a summary note) so the parent can re-issue the guidance instead of trusting it landed. This is what makes adding a sender safe: without it the finish-before-drain race silently loses the text — the exact loss the finalizer contract exists to prevent. - subagent.steer gateway RPC beside subagent.interrupt so programmatic hosts (dashboard, voice layers, ACP bridges) get an in-tree caller; catalogued in programmatic-integration.md. - docs: "Steering a Running Subagent" section in delegation.md covering the queued-vs-delivered semantics. Tests: registry-level steer coverage (delivery, unknown id, empty text, dead record, raising agent), the finish-before-drain race retaining missed_steer, and the RPC contract (4000/4002 validation, queued and rejected envelopes).
This commit is contained in:
SmokeDev
Teknium
parent
6e9cae6ac4
commit
60e1f7517c
|
|
@ -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"
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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
|
||||
```
|
||||
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
Loading…
Reference in New Issue