End subagents on the CLI's own task lifecycle, and detect a limit two ways
Subagents were showing "running" long after they had finished. Measured against 2.1.237 by running a session that launched one Task agent and reading its stdout: a subagent's lines carry no `stream_event` at all -- they are whole `user`/`assistant` lines with a null `stop_reason` -- and no `result` line is sent for one. So `ends_a_turn`, which watches for a raw `message_delta` saying `end_turn`, could never fire for a subagent, and nothing finished one until its session's process exited. What the CLI does send is a task lifecycle, as top-level `system` lines: `task_started` (with the tool_use id), `task_progress`, `task_updated` (status, naming the task only) and `task_notification` (tool id, status, and the agent's own summary). `translate_task` keeps the task -> tool mapping, records the summary as the subagent's closing text -- the run showed its child lines stop at its last tool_result, so without this a finished subagent reads as stopping mid-tool -- and ends it. A `completed` update is deliberately not the end, since its notification carries the summary; any other terminal status is, because the failure to avoid is a subagent nothing ever finishes. `ends_a_turn` stays as a second detector and must never be the only one again. Verified by replaying the captured stream through the server as a fake CLI: running, prompt, Bash call, output, report, exited. `finish_all` now reads the directory rather than the live map, which is what clears the ones already stuck: a subagent left running by an earlier run of the server is exactly the one this process never touched, so it read "running" again every time its session was started. Auto-resume gets the same treatment on its own single point of failure. The only thing that scheduled a resume was the CLI's error sentence at the end of a failed turn; the CLI also sends `rate_limit_event` lines saying where the account stands, and this server ignored them entirely. Both are read now. Anything that is not an `allowed...` status counts as refused and is logged if unfamiliar -- being wrong that way costs one question to the usage meter, which is still what decides whether anything is sent, and being wrong the other way is the feature silently not existing. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
irisandClaude Opus 5 committed
1 parent
13d2d11c2d
commit
74c07d687a
5 files changed
+371
-22
No files matched your search
@@ -104,6 +104,16 @@ pub(super) struct Translator {
|
||||
/// are unique but a `stream_event`'s content-block index is not, and
|
||||
/// parallel subagents interleave their deltas on one stdout.
|
||||
children: HashMap<String, Arc<Mutex<Translator>>>,
|
||||
/// Whether the last `rate_limit_event` said the account is refused, so
|
||||
/// that only the change into that state is reported -- see
|
||||
/// [`Translator::translate_rate_limit`].
|
||||
rate_limited: bool,
|
||||
/// Which Task call each running task belongs to: the CLI's `task_id`
|
||||
/// against the `tool_use_id` this side names a subagent by.
|
||||
///
|
||||
/// Needed because the line that says a task *ended* comes in two shapes
|
||||
/// and only one of them carries the tool id -- see [`Translator::translate_task`].
|
||||
tasks: HashMap<String, String>,
|
||||
}
|
||||
|
||||
impl Translator {
|
||||
@@ -117,6 +127,8 @@ impl Translator {
|
||||
session_dir,
|
||||
subagents,
|
||||
children: HashMap::new(),
|
||||
rate_limited: false,
|
||||
tasks: HashMap::new(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -215,6 +227,7 @@ impl Translator {
|
||||
// which is the same event seen through a side effect. The
|
||||
// announcement lands before the new init rather than after it.
|
||||
Some("conversation_reset") => vec![Event::Cleared],
|
||||
Some("rate_limit_event") => self.translate_rate_limit(message),
|
||||
Some("stream_event") => self.translate_stream_event(&message["event"]),
|
||||
Some("assistant") => self.translate_assistant(&message["message"]),
|
||||
Some("user") => self.translate_user(message),
|
||||
@@ -364,6 +377,9 @@ impl Translator {
|
||||
}]
|
||||
}
|
||||
Some("status") => self.translate_status(message),
|
||||
Some("task_started" | "task_progress" | "task_updated" | "task_notification") => {
|
||||
self.translate_task(message)
|
||||
}
|
||||
Some("compact_boundary") => {
|
||||
let meta = &message["compact_metadata"];
|
||||
vec![Event::Compacted {
|
||||
@@ -382,6 +398,129 @@ impl Translator {
|
||||
/// A `system/status` line: the CLI entering or leaving a state that is not
|
||||
/// a turn.
|
||||
///
|
||||
/// The CLI saying where the account stands with its rate limits, which it
|
||||
/// sends unasked during a turn.
|
||||
///
|
||||
/// A second detector for the one thing auto-resume depends on, beside the
|
||||
/// failed `result` [`usage_limit`] reads. That one is the CLI's error
|
||||
/// sentence and is the only thing this server watched; if a version ever
|
||||
/// stops the turn without it -- or blocks before starting one -- nothing
|
||||
/// is scheduled and a session switched to auto-resume simply never comes
|
||||
/// back, with nothing on screen or in the log saying why. This line says
|
||||
/// it outright.
|
||||
///
|
||||
/// Measured against 2.1.237 on 2026-09-06: `rate_limit_info` carries
|
||||
/// `status`, `resetsAt` (epoch seconds), `rateLimitType`, and on newer
|
||||
/// lines a `unifiedWindows` map of utilizations. Only `allowed` has been
|
||||
/// observed here, so anything that is *not* an `allowed…` word is taken
|
||||
/// as refused rather than assumed harmless -- the state left out of an
|
||||
/// enumeration is the one that costs, and being wrong that way is one
|
||||
/// extra question to the usage meter, which is what decides whether
|
||||
/// anything is actually sent. An unrecognised word is logged, so the next
|
||||
/// one to appear is a fact rather than a guess.
|
||||
///
|
||||
/// Only the *change* into being refused is reported: these arrive
|
||||
/// repeatedly, and a `LimitReached` per line would be a transcript full
|
||||
/// of them.
|
||||
fn translate_rate_limit(&mut self, message: &Value) -> Vec<Event> {
|
||||
let info = &message["rate_limit_info"];
|
||||
let status = info.get("status").and_then(Value::as_str);
|
||||
let refused = !matches!(status, None | Some("allowed"))
|
||||
&& !status.is_some_and(|status| status.starts_with("allowed"));
|
||||
if refused && !status.is_some_and(is_known_refusal) {
|
||||
tracing::warn!(
|
||||
"unrecognised rate limit status {status:?}, read as out of quota -- see translate_rate_limit"
|
||||
);
|
||||
}
|
||||
let was = std::mem::replace(&mut self.rate_limited, refused);
|
||||
if !refused || was {
|
||||
return Vec::new();
|
||||
}
|
||||
vec![Event::LimitReached {
|
||||
resets_at: info.get("resetsAt").and_then(Value::as_f64),
|
||||
}]
|
||||
}
|
||||
|
||||
/// The CLI's own account of a subagent's life, and **the only thing that
|
||||
/// ends one**.
|
||||
///
|
||||
/// Measured against 2.1.237 on 2026-09-06 by running a session that
|
||||
/// launched one Task agent and reading its stdout. A task produces, in
|
||||
/// order and all as top-level `system` lines with no `parent_tool_use_id`:
|
||||
///
|
||||
/// - `task_started` -- `task_id`, `tool_use_id`, `description`,
|
||||
/// `subagent_type`, `is_backgrounded`, `prompt`;
|
||||
/// - `task_progress` -- `last_tool_name` and usage, repeatedly;
|
||||
/// - `task_updated` -- `{patch: {status, end_time}}`, carrying the
|
||||
/// `task_id` but **not** the tool id;
|
||||
/// - `task_notification` -- `tool_use_id`, `status`, and `summary`: the
|
||||
/// agent's own report, which is also what the parent's `tool_result`
|
||||
/// for the Task call is given.
|
||||
///
|
||||
/// What that run also showed is why this exists: **the child lines carry
|
||||
/// no `stream_event` at all.** A subagent's own output arrives as whole
|
||||
/// `user`/`assistant` lines whose `stop_reason` is `null`, and no `result`
|
||||
/// line is ever sent for one -- so [`ends_a_turn`], which watches for a
|
||||
/// raw `message_delta` saying `end_turn`, cannot fire for a subagent in
|
||||
/// this version, and every subagent stayed `Running` until its session's
|
||||
/// process exited. The task lines are the CLI saying it outright, for a
|
||||
/// backgrounded agent and a synchronous one alike, which is what the
|
||||
/// parent's `tool_result` could not do.
|
||||
///
|
||||
/// The summary is recorded as the subagent's own closing text because it
|
||||
/// is the one thing it says that never reaches its transcript otherwise:
|
||||
/// the run above ended the child lines at its last `tool_result`, so
|
||||
/// without this a finished subagent reads as stopping mid-tool.
|
||||
fn translate_task(&mut self, message: &Value) -> Vec<Event> {
|
||||
let task_id = message.get("task_id").and_then(Value::as_str);
|
||||
let tool_use_id = message.get("tool_use_id").and_then(Value::as_str);
|
||||
if let (Some(task), Some(tool)) = (task_id, tool_use_id) {
|
||||
self.tasks.insert(task.to_string(), tool.to_string());
|
||||
}
|
||||
match message.get("subtype").and_then(Value::as_str) {
|
||||
Some("task_notification") => {
|
||||
let Some(id) = tool_use_id else {
|
||||
return Vec::new();
|
||||
};
|
||||
if !ended(message.get("status").and_then(Value::as_str)) {
|
||||
return Vec::new();
|
||||
}
|
||||
if let Some(summary) = text_field(message, "summary") {
|
||||
self.subagents
|
||||
.record(id, Event::AssistantText { delta: summary });
|
||||
}
|
||||
self.subagents.finish(id);
|
||||
}
|
||||
Some("task_updated") => {
|
||||
let status = message
|
||||
.get("patch")
|
||||
.and_then(|patch| patch.get("status"))
|
||||
.and_then(Value::as_str);
|
||||
// `completed` deliberately does nothing here: the
|
||||
// `task_notification` a completed task is always followed by
|
||||
// is what carries its summary, and ending it on this earlier
|
||||
// line would put that summary after the ending. Every other
|
||||
// way a task can stop is taken at face value -- a task that
|
||||
// failed or was cancelled has not been observed here, and the
|
||||
// failure to avoid is the one this whole function exists for:
|
||||
// a subagent that nothing ever finishes.
|
||||
if status != Some("completed")
|
||||
&& ended(status)
|
||||
&& let Some(id) = task_id.and_then(|task| self.tasks.get(task))
|
||||
{
|
||||
let id = id.clone();
|
||||
self.subagents.finish(&id);
|
||||
}
|
||||
}
|
||||
// `task_started` and `task_progress`: the mapping above is the
|
||||
// whole of what they are for. The subagent itself is created by
|
||||
// the Task `tool_use` in the parent's own message, which arrives
|
||||
// first and carries the title this side shows.
|
||||
_ => {}
|
||||
}
|
||||
Vec::new()
|
||||
}
|
||||
|
||||
/// A null `status` is the leaving edge, and it carries how the thing went.
|
||||
/// The turn it happened inside is still going when it ends -- the `result`
|
||||
/// has not arrived -- so leaving says `Running`. A state this build does
|
||||
@@ -740,8 +879,36 @@ fn fallback_title(message: &Value) -> String {
|
||||
/// `message_delta` is the raw API's own signal, carrying the stop reason:
|
||||
/// `end_turn` is genuinely done, `tool_use` means the model is about to call
|
||||
/// one and there is more coming. A `result` line is the CLI's own shape for
|
||||
/// a top-level turn; a subagent has not been observed to send one, but
|
||||
/// SUBAGENTS.md counts it too in case a future CLI version does.
|
||||
/// a top-level turn.
|
||||
///
|
||||
/// **Neither has been observed on a subagent's lines in 2.1.237** -- see
|
||||
/// [`Translator::translate_task`], which is what actually ends one, and which
|
||||
/// exists because relying on this alone left every subagent running for ever.
|
||||
/// Kept because it costs nothing and a dialect that does send either would be
|
||||
/// saying exactly what it means; it must never be the only detector again.
|
||||
/// The words `rate_limit_info.status` has been seen or documented to use for
|
||||
/// "no". Only used to decide whether to *log* an unfamiliar one: an unknown
|
||||
/// status is treated as a refusal either way, since a limit missed is a
|
||||
/// session that never comes back.
|
||||
fn is_known_refusal(status: &str) -> bool {
|
||||
matches!(status, "rejected" | "blocked" | "exceeded" | "limited")
|
||||
}
|
||||
|
||||
/// Whether a task status word means the task is over.
|
||||
///
|
||||
/// Written as "not one of the words that mean it is still going" rather than
|
||||
/// as a list of endings, because the two are not symmetric here: a status
|
||||
/// this build has never seen is far more likely to be a new way of finishing
|
||||
/// than a new way of continuing, and guessing wrong in that direction leaves
|
||||
/// a subagent reading `running` for ever with nothing able to correct it.
|
||||
/// Nothing at all is *not* an ending: a line that said no status said nothing.
|
||||
fn ended(status: Option<&str>) -> bool {
|
||||
!matches!(
|
||||
status,
|
||||
None | Some("running" | "in_progress" | "pending" | "queued" | "started")
|
||||
)
|
||||
}
|
||||
|
||||
fn ends_a_turn(message: &Value) -> bool {
|
||||
match message.get("type").and_then(Value::as_str) {
|
||||
Some("stream_event") => {
|
||||
@@ -1119,6 +1286,126 @@ mod tests {
|
||||
assert!(subagent.is_open());
|
||||
}
|
||||
|
||||
/// The second limit detector, on the shape the CLI actually sends. The
|
||||
/// `allowed` line is copied from a real 2.1.237 run on 2026-09-06; the
|
||||
/// refused one is the same line with the status changed, which is the
|
||||
/// part that has not been observed and is why an unknown word counts as
|
||||
/// refused.
|
||||
#[test]
|
||||
fn a_rate_limit_event_that_is_not_allowed_reports_the_limit_once() {
|
||||
let dir = tempfile::tempdir().expect("tempdir");
|
||||
let subagents = test_subagents(&dir);
|
||||
let mut translator = Translator::new(dir.path().to_path_buf(), subagents);
|
||||
let allowed = r#"{"type":"rate_limit_event","rate_limit_info":{"status":"allowed","resetsAt":1788726600,"rateLimitType":"five_hour"}}"#;
|
||||
let refused = r#"{"type":"rate_limit_event","rate_limit_info":{"status":"rejected","resetsAt":1788726600,"rateLimitType":"five_hour"}}"#;
|
||||
assert!(translate_lines(&mut translator, &[allowed]).is_empty());
|
||||
assert_eq!(
|
||||
translate_lines(&mut translator, &[refused]),
|
||||
vec![Event::LimitReached {
|
||||
resets_at: Some(1788726600.0)
|
||||
}]
|
||||
);
|
||||
// Repeats say nothing: these arrive throughout a turn, and the
|
||||
// schedule was made by the first one.
|
||||
assert!(translate_lines(&mut translator, &[refused]).is_empty());
|
||||
// Allowed again, then refused again, is a new limit and is reported.
|
||||
assert!(translate_lines(&mut translator, &[allowed]).is_empty());
|
||||
assert_eq!(
|
||||
translate_lines(&mut translator, &[refused]),
|
||||
vec![Event::LimitReached {
|
||||
resets_at: Some(1788726600.0)
|
||||
}]
|
||||
);
|
||||
}
|
||||
|
||||
/// What ends a subagent in the CLI as it actually behaves: its task's
|
||||
/// own lifecycle lines. The shapes here are copied from a real 2.1.237
|
||||
/// run on 2026-09-06 -- see [`Translator::translate_task`] -- including
|
||||
/// the detail that killed the previous rule, that a subagent's own lines
|
||||
/// stop at its last `tool_result` and never say `end_turn`.
|
||||
#[test]
|
||||
fn a_tasks_own_completion_finishes_the_subagent_and_records_its_report() {
|
||||
let dir = tempfile::tempdir().expect("tempdir");
|
||||
let subagents = test_subagents(&dir);
|
||||
let mut translator = Translator::new(dir.path().to_path_buf(), Arc::clone(&subagents));
|
||||
translate_lines(
|
||||
&mut translator,
|
||||
&[
|
||||
r#"{"type":"assistant","message":{"content":[{"type":"tool_use","id":"toolu_task9","name":"Task","input":{"description":"echo something","prompt":"say hello"}}]},"parent_tool_use_id":null}"#,
|
||||
r#"{"type":"system","subtype":"task_started","task_id":"a7c5","tool_use_id":"toolu_task9","is_backgrounded":false}"#,
|
||||
r#"{"type":"assistant","message":{"role":"assistant","content":[{"type":"tool_use","id":"toolu_bash","name":"Bash","input":{"command":"echo hello"}}],"stop_reason":null},"parent_tool_use_id":"toolu_task9"}"#,
|
||||
r#"{"type":"system","subtype":"task_progress","task_id":"a7c5","tool_use_id":"toolu_task9","last_tool_name":"Bash"}"#,
|
||||
],
|
||||
);
|
||||
let subagent = subagents.get("toolu_task9").expect("subagent started");
|
||||
assert!(subagent.is_open(), "still working");
|
||||
|
||||
// The completed update carries no tool id and no summary, and is
|
||||
// deliberately not the end: the notification behind it is.
|
||||
translate_lines(
|
||||
&mut translator,
|
||||
&[
|
||||
r#"{"type":"system","subtype":"task_updated","task_id":"a7c5","patch":{"status":"completed","end_time":1788715207655}}"#,
|
||||
],
|
||||
);
|
||||
assert!(subagent.is_open(), "waiting for the report");
|
||||
|
||||
translate_lines(
|
||||
&mut translator,
|
||||
&[
|
||||
r#"{"type":"system","subtype":"task_notification","task_id":"a7c5","tool_use_id":"toolu_task9","status":"completed","summary":"it said hello"}"#,
|
||||
],
|
||||
);
|
||||
assert!(!subagent.is_open());
|
||||
let lines =
|
||||
crate::session::transcript::read_after(&subagent.transcript_path(), 0).expect("read");
|
||||
// The report, then the ending, in that order.
|
||||
let tail: Vec<&Event> = lines
|
||||
.iter()
|
||||
.rev()
|
||||
.take(2)
|
||||
.map(|entry| &entry.event)
|
||||
.collect();
|
||||
assert!(
|
||||
matches!(
|
||||
tail[0],
|
||||
Event::Status {
|
||||
state: SessionStatus::Exited
|
||||
}
|
||||
),
|
||||
"{tail:?}"
|
||||
);
|
||||
assert!(
|
||||
matches!(tail[1], Event::AssistantText { delta } if delta == "it said hello"),
|
||||
"{tail:?}"
|
||||
);
|
||||
}
|
||||
|
||||
/// A task that stops any other way still ends its subagent, from the one
|
||||
/// line that carries it -- the update, which names the task rather than
|
||||
/// the tool call, so the mapping from `task_started` is what finds it.
|
||||
#[test]
|
||||
fn a_task_that_did_not_complete_is_ended_by_its_update() {
|
||||
let dir = tempfile::tempdir().expect("tempdir");
|
||||
let subagents = test_subagents(&dir);
|
||||
let mut translator = Translator::new(dir.path().to_path_buf(), Arc::clone(&subagents));
|
||||
translate_lines(
|
||||
&mut translator,
|
||||
&[
|
||||
r#"{"type":"assistant","message":{"content":[{"type":"tool_use","id":"toolu_task8","name":"Task","input":{"description":"doomed"}}]},"parent_tool_use_id":null}"#,
|
||||
r#"{"type":"system","subtype":"task_started","task_id":"b1f2","tool_use_id":"toolu_task8","is_backgrounded":true}"#,
|
||||
],
|
||||
);
|
||||
let subagent = subagents.get("toolu_task8").expect("subagent started");
|
||||
translate_lines(
|
||||
&mut translator,
|
||||
&[
|
||||
r#"{"type":"system","subtype":"task_updated","task_id":"b1f2","patch":{"status":"failed"}}"#,
|
||||
],
|
||||
);
|
||||
assert!(!subagent.is_open());
|
||||
}
|
||||
|
||||
/// What actually ends a subagent: the raw API's own `message_delta`
|
||||
/// saying its turn stopped with `end_turn`. Never written into the
|
||||
/// subagent's own transcript as `Idle` -- its vocabulary has no such
|
||||
|
||||
@@ -301,11 +301,26 @@ impl Subagents {
|
||||
}
|
||||
|
||||
/// The parent session's process is gone, so nothing still open here has
|
||||
/// a process behind it either -- see `SUBAGENTS.md`'s lifecycle #4.
|
||||
/// a process behind it either -- see `SUBAGENTS.md`'s lifecycle #5.
|
||||
///
|
||||
/// Read from the directory rather than from `live`, because a subagent
|
||||
/// left `Running` by a *previous* run of this server is exactly the one
|
||||
/// that needs closing and is the one `live` does not have: nothing in
|
||||
/// this process ever touched it, so it would keep reading `running`
|
||||
/// every time its session was started again, with nothing able to
|
||||
/// correct it.
|
||||
pub fn finish_all(&self) {
|
||||
let subagents: Vec<Arc<Subagent>> = self.live.lock().unwrap().values().cloned().collect();
|
||||
for subagent in subagents {
|
||||
if subagent.is_open() {
|
||||
// `list(true)` reports each one's own last status rather than
|
||||
// rewriting a running one as unknown -- what is wanted here is which
|
||||
// are open on disk, and this call is the very thing that decides the
|
||||
// session is not running.
|
||||
for info in self.list(true) {
|
||||
if info.status != SessionStatus::Running {
|
||||
continue;
|
||||
}
|
||||
if let Some(subagent) = self.get(&info.id)
|
||||
&& subagent.is_open()
|
||||
{
|
||||
subagent.append(Event::Status {
|
||||
state: SessionStatus::Exited,
|
||||
});
|
||||
@@ -508,6 +523,25 @@ mod tests {
|
||||
subagents.finish("never-started");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn finish_all_closes_one_left_running_by_an_earlier_run() {
|
||||
// The state a backend restart leaves behind: the subagent is on disk
|
||||
// reading `Running` and nothing in this process has touched it, so a
|
||||
// registry that only knew its own `live` map left it running for
|
||||
// ever -- and the phone showed it as running every time the session
|
||||
// was started again.
|
||||
let dir = tempfile::tempdir().expect("tempdir");
|
||||
{
|
||||
let subagents = Subagents::new(dir.path().to_path_buf());
|
||||
subagents.start("toolu_stale", "helper", None);
|
||||
}
|
||||
let subagents = Subagents::new(dir.path().to_path_buf());
|
||||
subagents.finish_all();
|
||||
let rows = subagents.list(true);
|
||||
assert_eq!(rows.len(), 1);
|
||||
assert_eq!(rows[0].status, SessionStatus::Exited);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn deleting_a_finished_subagent_takes_its_directory_with_it() {
|
||||
let dir = tempfile::tempdir().expect("tempdir");
|
||||
|
||||
Reference in new issue
Block a user