//! The stream-json dialect: CLI lines in, common [`Event`]s out. //! //! Split from the driver beside it because the two change for unrelated //! reasons. This half moves when the CLI's wire format does, which is what the //! tests at the bottom pin by replaying recorded lines; the driver half moves //! when spawning, resuming or shutting down changes. //! //! The one side effect here is saving images a tool result carries into the //! session directory; everything else is pure, which is what makes the mapping //! testable without a process. use std::collections::{HashMap, HashSet}; use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex}; use serde_json::{Value, json}; use super::super::driver::{Event, QuestionOption, SessionStatus, context_tokens, patch_start}; use super::super::subagent::Subagents; /// Whether this line is the CLI opening a fresh model call. /// /// `message_start` begins one assistant message, and the CLI sends the previous /// call's tool results back before opening the next -- so this is the first /// moment at which anything written since the last one can have been read. /// Nothing earlier will do: the deltas and `tool_use` of a message *already in /// flight* keep arriving after a steer is written, and none of them saw it. /// /// Only present because the driver passes `--include-partial-messages`, which /// is why the caller keeps a fallback that does not depend on it. pub(super) fn starts_a_model_call(message: &Value) -> bool { message.get("type").and_then(Value::as_str) == Some("stream_event") && message["event"].get("type").and_then(Value::as_str) == Some("message_start") } /// What answering a question produced. pub(super) enum AnswerOutcome { /// Send this control_response line to the CLI. Respond(Value), /// Part of a multi-question request; more answers still needed. Pending, Unknown, } /// A setting a control request asked for, held until the CLI says whether it /// took. The CLI answers `set_model` with a bare success -- no value -- so the /// only way to report what was accepted is to remember what was asked. /// `set_permission_mode` does echo its mode back. pub(super) enum Setting { Model(String), PermissionMode(String), } /// A `can_use_tool` request we've surfaced to the phone and not yet answered. /// For plain permissions there is one implicit question (Allow/Deny); for /// AskUserQuestion, one per entry in `questions`. struct PendingRequest { request_id: String, input: Value, /// Question text per sub-question, in order -- the keys the answers map /// uses. Empty for a plain permission request. questions: Vec, answers: HashMap, } /// Translation state: stream-json lines in, common events out. pub(super) struct Translator { pub(super) session_id: Option, pending: HashMap, /// Settings asked for and not yet answered, by request id. Its path out is /// the response: every entry is removed when one arrives, whether it /// succeeded or failed. asked: HashMap, /// Whether this side asked the turn to stop. /// /// The CLI reports an interrupted turn the same way it reports one that /// broke -- a `result` with `is_error` set -- so the line cannot tell them /// apart, and somebody who pressed Stop was shown "the turn ended with an /// error". What separates them is that *we* asked. /// /// Its path out is that result, so a genuine failure in a later turn is /// still reported. interrupting: bool, /// The input side of the newest assistant message, waiting for the `result` /// that ends the turn to carry it out. /// /// Read from the assistant message rather than the result's own usage, /// which is the whole turn added up: measured on 2026-08-30 against 2.1.237, /// a two-message turn reported `cache_read_input_tokens` of 40,211, being /// 14,259 and 25,952 -- the same conversation counted twice. The model held /// 26,131. A turn with ten tool calls would overstate it tenfold. /// /// Its path out is that result, so a turn whose messages carried no usage /// reports none rather than repeating the previous turn's. context: Option, session_dir: PathBuf, /// This session's subagents, shared with every child translator below -- /// see `SUBAGENTS.md`. One registry per session, so a subagent started /// through this translator or any of its children lands in the same /// place a route reads it back from. subagents: Arc, /// One translator per subagent id, holding *its* streaming and /// tool-tracking state -- separate from the parent's because tool ids /// are unique but a `stream_event`'s content-block index is not, and /// parallel subagents interleave their deltas on one stdout. children: HashMap>>, /// 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 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, /// The backgrounded tasks this translator has seen start and not seen /// finish, by `tool_use_id`. /// /// Half of the answer to "does this session still have work outstanding", /// which is the difference between `Idle` and [`SessionStatus::Waiting`]. /// The other half is `Subagents::any_open`, and both are needed: this one /// covers a backgrounded *command*, which has no subagent behind it at /// all, and the registry covers a subagent launched before this /// translator existed, which is every one of them after a backend /// restart adopts a running session. open_tasks: HashSet, /// File-edit calls whose successful boilerplate result should not be drawn below their diff. /// Each leaves here with its `tool_result`; a failure keeps its text because that is the part a /// reader needs to act on. patches: HashSet, /// Whether a turn is open, judged from this translator's own output: the /// events that [`super::proves_a_turn`] accepts open one, and the status /// that ends a turn closes it. /// /// The same rule the driver uses to announce `Running`, read from the same /// side of the translation, because two rules for "is a turn running" /// would disagree the first time either moved. What it decides here is /// narrow: whether a task reporting back means the session has gone idle, /// or is merely one of several things happening inside a turn. in_turn: bool, } impl Translator { pub(super) fn new(session_dir: PathBuf, subagents: Arc) -> Self { Self { session_id: None, pending: HashMap::new(), asked: HashMap::new(), interrupting: false, context: None, session_dir, subagents, children: HashMap::new(), rate_limited: false, tasks: HashMap::new(), open_tasks: HashSet::new(), patches: HashSet::new(), in_turn: false, } } /// Remembers what a control request was for, so its answer can say so. /// Called before the request goes out: the reader thread is already running /// and a fast CLI can answer before this side gets back to it. pub(super) fn expect_setting(&mut self, request_id: String, setting: Setting) { self.asked.insert(request_id, setting); } /// Says that the turn about to end was stopped on purpose. Called before /// the request goes out, for the reason [`Translator::expect_setting`] /// gives. pub(super) fn expect_interrupt(&mut self) { self.interrupting = true; } pub(super) fn translate(&mut self, message: &Value) -> Vec { // Events from subagents (Task tool internals) carry a // parent_tool_use_id; the transcript shows the Task tool's own // start/end instead of every nested step. Routed into that // subagent's own transcript rather than dropped -- see // `SUBAGENTS.md`. if let Some(parent_id) = message.get("parent_tool_use_id").and_then(Value::as_str) { return self.translate_child(parent_id, message); } self.dispatch(message) } /// A line belonging to a subagent rather than to this translator's own /// session. Always returns nothing to the *caller*: everything it /// produces goes into the subagent's own transcript instead. fn translate_child(&mut self, id: &str, message: &Value) -> Vec { match self.subagents.get(id) { Some(subagent) if !subagent.is_open() => { // Not stale: the Task tool runs in the background by // default, so a finished subagent can still be sent another // message later (SendMessage) and start working again. A // line arriving after `finish` means exactly that, not a // conversation that is over -- see `SUBAGENTS.md`. self.subagents.reopen(id); } Some(_) => {} None => { // Nobody has heard of this id yet: the Task call itself // either has not been seen or never will be. Started here // with the best title available -- the tool name of this // first line -- since SUBAGENTS.md's real title only // arrives with the Task call. self.subagents.start(id, &fallback_title(message), None); } } let child = self .children .entry(id.to_string()) .or_insert_with(|| { Arc::new(Mutex::new(Translator::new( self.session_dir.clone(), Arc::clone(&self.subagents), ))) }) .clone(); let events = child.lock().unwrap().dispatch(message); // A limit the account hit while this subagent was working. It belongs // in the subagent's transcript, which is where it happened -- and it // also has to reach the session, which is the only thing auto-resume // can schedule against: a background subagent can run on long after // its parent's own turn ended, so "the main agent was idle when the // limit hit" is the ordinary case rather than an edge of one, and // swallowing it here left that session waiting for a person for ever. let mut hoisted = Vec::new(); for event in events { // The subagent's own vocabulary is Running/Exited/Unknown, never // Idle or Waiting -- a background Task is either working or it has // ended, never merely "between turns" the way a session is. // Dropped here rather than never produced, so a `result` line's // own end-of-turn status (dispatch's ordinary one, for a subagent // dialect that ever sends one) is caught the same way a // `message_delta` would be. if closes_a_turn(&event) { continue; } if matches!(event, Event::LimitReached { .. }) { hoisted.push(event.clone()); } self.subagents.record(id, event); } // What actually ends a subagent's turn: not the parent's // `tool_result`, which for a background Task arrives at launch // ("Async agent launched...") long before the work is done -- see // `SUBAGENTS.md`. if ends_a_turn(message) { self.subagents.finish(id); } hoisted } /// One line of this translator's own session, with [`Translator::in_turn`] /// kept up to date from what came out of it. /// /// Here rather than in each arm because it has to hold for every line /// there is: the set of events that prove a turn is running is /// [`super::proves_a_turn`]'s, and no arm should have to remember it. fn dispatch(&mut self, message: &Value) -> Vec { let events = self.translate_line(message); if events.iter().any(closes_a_turn) { self.in_turn = false; } else if events.iter().any(super::proves_a_turn) { self.in_turn = true; } events } fn translate_line(&mut self, message: &Value) -> Vec { match message.get("type").and_then(Value::as_str) { Some("system") => self.translate_system(message), // The CLI's own announcement that `/clear` took effect, sent just // before the fresh `init` carrying the new session_id. Measured // against 2.1.237: this used to watch for the id being *replaced*, // 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), Some("control_request") => self.translate_control_request(message), Some("control_response") => { let response = &message["response"]; // Answered either way, so the request stops being pending // either way -- a rejected setting that stayed here would be // applied by the next request that reused its id. let asked = response .get("request_id") .and_then(Value::as_str) .and_then(|id| self.asked.remove(id)); if response.get("subtype").and_then(Value::as_str) == Some("error") { let error = response .get("error") .and_then(Value::as_str) .unwrap_or("unknown"); return vec![Event::Error { message: format!("claude rejected a request: {error}"), }]; } // Success, so the setting this request asked for is now the // session's, and this is the only place that says so: the // response carries no value of its own for a model. match asked { Some(Setting::Model(model)) => vec![Event::Settings { model: Some(model), permission_mode: None, }], Some(Setting::PermissionMode(mode)) => vec![Event::Settings { model: None, // The CLI echoes this one, and its answer wins: `auto` // and `manual` are names it accepts on the way in and // reports back under another name, so repeating the // request would show a mode the session is not in. permission_mode: Some( response["response"]["mode"] .as_str() .map(str::to_string) .unwrap_or(mode), ), }], None => Vec::new(), } } Some("result") => { let usage = &message["usage"]; let tokens = usage .get("input_tokens") .and_then(Value::as_u64) .unwrap_or(0) + usage .get("output_tokens") .and_then(Value::as_u64) .unwrap_or(0); let mut events = Vec::new(); // A turn another agent started, which is only knowable here. // // Measured against 2.1.237 (2026-08-31) by sending a real // cross-session message to a real stream-json session: the CLI // emits no `user` record for it and nothing in the // partial-message stream mentions it. The whole of it arrives as // an `origin` object on the turn's `result`, in the same shape // the session file records -- so this is `import::peer_message` // reading a different record. // // The cost is the position: the note lands after the reply it // caused, because at no earlier point does the CLI say why the // turn started. Taken deliberately over a second reader tailing // the CLI's own session file, which is two sources of truth for // one conversation and a poll per live session. // // Only peer-caused turns carry it: four ordinary results over a // real session's stdout had no `origin` between them. if let Some(peer) = crate::session::import::peer_message(message) { events.push(peer); } // Whichever way this result went, the interrupt it may have // been answering is now spent. let asked_to_stop = std::mem::take(&mut self.interrupting); if !asked_to_stop && message .get("is_error") .and_then(Value::as_bool) .unwrap_or(false) { let said = message.get("result").and_then(Value::as_str); events.push(match said.and_then(usage_limit) { Some(resets_at) => Event::LimitReached { resets_at }, None => Event::Error { message: said.unwrap_or("the turn ended with an error").to_string(), }, }); } let context = self.context.take(); if tokens > 0 { events.push(Event::UsageDelta { tokens, context }); } // Idle means "waiting for a person", and a session with a // backgrounded subagent or command still running is not doing // that -- it is waiting for itself, and will speak again with // nobody having typed anything. Reported as what it is, so // that nothing tells the reader the work has finished. events.push(Event::Status { state: if self.work_outstanding() { SessionStatus::Waiting } else { SessionStatus::Idle }, }); events } _ => Vec::new(), } } /// The CLI's own notices: which session this is, and what it is doing that /// is not a turn. /// /// Compaction is the whole of that second kind, and it is announced rather /// than inferred. Measured against 2.1.237 (2026-08-29) by driving a session /// through `/compact`, one produces in order: /// /// - `{"subtype":"status","status":"compacting"}` -- the start; /// - `{"subtype":"status","status":null,"compact_result":"success"}`, or /// `"failed"` with a `compact_error` -- the end; /// - a fresh `init` carrying the same `session_id`; /// - `{"subtype":"compact_boundary","compact_metadata":{…}}` with the token /// counts, and only when it succeeded; /// - the turn's ordinary `result`, which returns it to idle. /// /// The keys are snake_case here and camelCase in the CLI's own transcript /// file, which records the same events -- so reading the shape off that /// file, the obvious place to look, gets every field name wrong and /// silently yields a compaction with no numbers in it. fn translate_system(&mut self, message: &Value) -> Vec { match message.get("subtype").and_then(Value::as_str) { Some("init") => { if let Some(id) = message.get("session_id").and_then(Value::as_str) { self.session_id = Some(id.to_string()); } // The CLI's own account of what it is set to, and the only one // that resolves an alias: a session launched with // `--model haiku` reports `claude-haiku-4-5-20251001` here. It // arrives again after a compaction, which is free. vec![Event::Settings { model: message .get("model") .and_then(Value::as_str) .map(str::to_string), permission_mode: message .get("permissionMode") .and_then(Value::as_str) .map(str::to_string), }] } 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 { pre_tokens: meta.get("pre_tokens").and_then(Value::as_u64), post_tokens: meta.get("post_tokens").and_then(Value::as_u64), trigger: meta .get("trigger") .and_then(Value::as_str) .map(str::to_string), }] } _ => Vec::new(), } } /// 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 { 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 { 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()); } // The tool call this line is about, from the line itself or from // whichever earlier line did carry it. let about = tool_use_id .map(str::to_string) .or_else(|| task_id.and_then(|task| self.tasks.get(task).cloned())); match message.get("subtype").and_then(Value::as_str) { Some("task_notification") => self.task_ended( about, message.get("status").and_then(Value::as_str), text_field(message, "summary"), ), 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") { return Vec::new(); } self.task_ended(about, status, None) } Some("task_started") => { // What makes the session `Waiting` when its turn ends. 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. if let Some(about) = about { self.open_tasks.insert(about); } Vec::new() } // `task_progress`: the mapping above is the whole of what it is // for. _ => Vec::new(), } } /// Whether the session has work of its own still running: the difference /// between `Idle` and [`SessionStatus::Waiting`]. /// /// Two sources because neither covers the other. `open_tasks` holds what /// this translator watched start, which is the only thing that knows /// about a backgrounded *command* -- it has no subagent. The registry /// holds what is on disk, which is the only thing that knows about a /// subagent that started before this translator did. /// /// `session_running` is true by construction: this is only ever asked /// while translating a line the session's process just wrote. fn work_outstanding(&self) -> bool { !self.open_tasks.is_empty() || self.subagents.any_open(true) } /// A task reporting back, from whichever of the two lines got here first. /// /// Handled once. The two shapes can both arrive for one task, and what /// says which of them is the first is that it finds the task still open. /// /// **Nothing about it reaches the parent's transcript**, deliberately. /// The summary is the subagent's own closing words and goes into the /// subagent's own transcript, which is the only place it belongs; a row /// per finished task in the session's transcript is a screen of dividers /// about work the reader was not asking after, and the session did not /// receive a message it could act on. What *does* reach the parent is a /// message a subagent genuinely sends it, which arrives by the peer /// path. The only thing produced here is the status: a session with /// nothing outstanding any more has stopped being /// [`SessionStatus::Waiting`]. fn task_ended( &mut self, about: Option, status: Option<&str>, summary: Option, ) -> Vec { if !ended(status) { return Vec::new(); } let Some(about) = about else { return Vec::new(); }; // Reported once. The two lifecycle shapes can both arrive for one // task, and whichever gets here first is the one that finds it open. // // The registry is asked as well as this translator's own set, and // that is what makes an adopted session work: a subagent launched // before a backend restart has no entry here, because its // `task_started` is behind the offset its session's stdout is read // from. `finish` below closes it either way, so a second line for the // same task still finds nothing. if !self.open_tasks.remove(&about) && !self.subagents.is_open(&about) { return Vec::new(); } let mut events = Vec::new(); // Where the report goes, and the two cases are not the same place. // // A subagent has a transcript of its own, and its closing words are // that transcript's last line -- see `SUBAGENTS.md`. A backgrounded // *command* has none: its own tool card is the only record of it // anywhere, and until this arrives that card is still showing the // launch result, which says the command is running. It stopped being // true at this line, so the card is brought up to date rather than // left making a claim nothing will ever correct. if self.subagents.get(&about).is_some() { if let Some(summary) = summary { self.subagents .record(&about, Event::AssistantText { delta: summary }); } } else { events.push(Event::ToolUpdate { id: about.clone(), // A task that stopped without a word still has to say so: the // states with no summary are exactly the ones that went // wrong, and they are the ones a stale "running in // background" reads worst on. output: summary .unwrap_or_else(|| format!("this background command {}", status_word(status))), }); } self.subagents.finish(&about); // The last outstanding task, with the session's own turn already // over: it has stopped being `Waiting` and nothing else will say so. // Inside a turn there is nothing to announce -- the turn's own // `result` will decide between the two statuses when it lands. if !self.work_outstanding() && !self.in_turn { events.push(Event::Status { state: SessionStatus::Idle, }); } events } /// 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 /// not recognise is left alone rather than mapped onto the nearest one. fn translate_status(&self, message: &Value) -> Vec { // A mode change the CLI has made, announced a moment after it answers // the request. Measured on 2.1.237: // `{"subtype":"status","status":null,"permissionMode":"plan"}`, which // is a leaving edge carrying no compaction result -- so it is checked // before the compaction reading below. if let Some(mode) = message.get("permissionMode").and_then(Value::as_str) { return vec![Event::Settings { model: None, permission_mode: Some(mode.to_string()), }]; } if let Some(status) = message.get("status").and_then(Value::as_str) { return match status { "compacting" => vec![Event::Status { state: SessionStatus::Compacting, }], _ => Vec::new(), }; } let Some(result) = message.get("compact_result").and_then(Value::as_str) else { return Vec::new(); }; let mut events = Vec::new(); if result != "success" { // The CLI's own sentence, because it is specific enough to act on: // "Not enough messages to compact." is a complete answer. events.push(Event::Error { message: match message.get("compact_error").and_then(Value::as_str) { Some(why) => format!("compaction failed: {why}"), None => format!("compaction {result}"), }, }); } events.push(Event::Status { state: SessionStatus::Running, }); events } /// Raw API streaming: only text deltas become events. Consolidated blocks /// arriving later re-carry the same text, so those are skipped in /// `translate_assistant` -- one source per fact. fn translate_stream_event(&mut self, event: &Value) -> Vec { if event.get("type").and_then(Value::as_str) == Some("content_block_delta") && let Some(delta) = event["delta"].get("text") && event["delta"].get("type").and_then(Value::as_str) == Some("text_delta") && let Some(text) = delta.as_str() { return vec![Event::AssistantText { delta: text.to_string(), }]; } Vec::new() } fn translate_assistant(&mut self, message: &Value) -> Vec { if let Some(usage) = message.get("usage") { let field = |name: &str| usage.get(name).and_then(Value::as_u64).unwrap_or(0); self.context = Some(context_tokens( field("input_tokens"), field("cache_creation_input_tokens"), field("cache_read_input_tokens"), )); } let Some(content) = message.get("content").and_then(Value::as_array) else { return Vec::new(); }; content .iter() .filter(|block| block.get("type").and_then(Value::as_str) == Some("tool_use")) .map(|block| { let id = block .get("id") .and_then(Value::as_str) .unwrap_or_default() .to_string(); let tool = block .get("name") .and_then(Value::as_str) .unwrap_or_default() .to_string(); let input = block.get("input").cloned().unwrap_or(Value::Null); // A subagent this call is about to start -- see // `SUBAGENTS.md`'s lifecycle #1. The parent's own transcript // still shows only the Task call itself, below. if tool == "Task" || tool == "Agent" { self.start_subagent_from_task(&id, &input); } if tool == "Edit" { self.patches.insert(id.clone()); patch_start(id, replacement_diff(&input)) } else { Event::ToolStart { id, tool, input } } }) .collect() } /// Starts the subagent a Task call names, with the title and prompt /// SUBAGENTS.md describes: the call's `description`, then /// `()` when one is given, falling back to the tool's own /// name when there is no description to build one from. fn start_subagent_from_task(&self, id: &str, input: &Value) { let description = text_field(input, "description"); let subagent_type = text_field(input, "subagent_type"); let prompt = input.get("prompt").and_then(Value::as_str); let title = match (description, subagent_type) { (Some(description), Some(subagent_type)) => { format!("{description} ({subagent_type})") } (Some(description), None) => description, (None, _) => "Task".to_string(), }; self.subagents.start(id, &title, prompt); } fn translate_control_request(&mut self, message: &Value) -> Vec { let request = &message["request"]; if request.get("subtype").and_then(Value::as_str) != Some("can_use_tool") { return Vec::new(); } let request_id = message .get("request_id") .and_then(Value::as_str) .unwrap_or_default() .to_string(); let tool_name = request .get("tool_name") .and_then(Value::as_str) .unwrap_or("a tool"); let input = request.get("input").cloned().unwrap_or(Value::Null); // Measured, not matched: the request names the call it is about, so the // phone never has to guess which tool row a permission belongs to. let about = request .get("tool_use_id") .and_then(Value::as_str) .map(String::from); let mut events = Vec::new(); let mut questions = Vec::new(); if tool_name == "AskUserQuestion" { for (i, question) in input .get("questions") .and_then(Value::as_array) .into_iter() .flatten() .enumerate() { let text = question .get("question") .and_then(Value::as_str) .unwrap_or("(question)") .to_string(); // Everything the reader decides on, carried in the event. The // alternative -- and what this was -- is the phone reaching into // the tool call's input for the parts the event dropped, which // puts this dialect's schema where no other dialect can reach it. let options = question .get("options") .and_then(Value::as_array) .into_iter() .flatten() .filter_map(|option| { Some(QuestionOption { label: option.get("label").and_then(Value::as_str)?.to_string(), description: text_field(option, "description"), preview: text_field(option, "preview"), }) }) .collect(); events.push(Event::Question { id: format!("{request_id}#{i}"), prompt: text.clone(), header: text_field(question, "header"), options, multi_select: question .get("multiSelect") .and_then(Value::as_bool) .unwrap_or(false), // The call that is asking, so all of this draws as one // thing. It used to be `None` on the grounds that a question // the model asked is not permission for a call -- true, and // beside the point: the reader was shown the AskUserQuestion // call *and* its questions as two separate cards. about: about.clone(), }); questions.push(text); } } else { let summary = serde_json::to_string_pretty(&input).unwrap_or_default(); let summary: String = summary.chars().take(600).collect(); events.push(Event::Question { id: request_id.clone(), prompt: format!("Allow {tool_name}?\n{summary}"), // No header: the question is about the call it names, and the // phone draws it on that call's own row. header: None, options: vec![ QuestionOption::plain("Allow"), QuestionOption::plain("Deny"), ], multi_select: false, about: about.clone(), }); } self.pending.insert( request_id.clone(), PendingRequest { request_id, input, questions, answers: HashMap::new(), }, ); events.push(Event::Status { state: SessionStatus::AwaitingInput, }); events } /// Applies one answer from the phone. Question ids are the control request /// id, suffixed `#i` for AskUserQuestion sub-questions. pub(super) fn answer(&mut self, question_id: &str, answers: &[String]) -> AnswerOutcome { // Where this dialect's shape is put on: the CLI's `answers` map is // string-valued whatever the question, so several choices become one // line here rather than everything upstream pretending a question can // only ever have one answer. let answer = answers.join(", "); let answer = answer.as_str(); let (request_id, sub) = match question_id.split_once('#') { Some((request_id, index)) => (request_id, index.parse::().ok()), None => (question_id, None), }; let Some(pending) = self.pending.get_mut(request_id) else { return AnswerOutcome::Unknown; }; let response = if let Some(index) = sub { let Some(question) = pending.questions.get(index) else { return AnswerOutcome::Unknown; }; pending.answers.insert(question.clone(), answer.to_string()); if pending.answers.len() < pending.questions.len() { return AnswerOutcome::Pending; } let mut updated = pending.input.clone(); updated["answers"] = serde_json::to_value(&pending.answers).expect("string map"); json!({"behavior": "allow", "updatedInput": updated}) } else if answer.eq_ignore_ascii_case("deny") { json!({"behavior": "deny", "message": "The user denied this from the phone."}) } else { json!({"behavior": "allow", "updatedInput": pending.input}) }; let request_id = pending.request_id.clone(); self.pending.remove(&request_id); AnswerOutcome::Respond(json!({ "type": "control_response", "response": {"subtype": "success", "request_id": request_id, "response": response}, })) } /// `user` messages: tool results become ToolEnd, with any image parts saved /// into the session dir and referenced by an Image event. Replayed and /// synthetic user text is skipped -- the manager already recorded the /// user's side. fn translate_user(&mut self, message: &Value) -> Vec { // Only tool results are here. The CLI never echoes a person's own // message back on stdout -- measured, because the obvious way to learn // that a queued message had been taken was to watch for it coming back // -- so the driver reports that itself, at the line it writes. let Some(content) = message["message"].get("content").and_then(Value::as_array) else { return Vec::new(); }; let mut events = Vec::new(); for block in content { if block.get("type").and_then(Value::as_str) != Some("tool_result") { continue; } let mut texts = Vec::new(); // Held until the call's id is in hand a few lines below: an image is // drawn under the call that produced it, so it has to carry that id. let mut images = Vec::new(); match block.get("content") { Some(Value::String(text)) => texts.push(text.clone()), Some(Value::Array(parts)) => { for part in parts { match part.get("type").and_then(Value::as_str) { Some("text") => { if let Some(text) = part.get("text").and_then(Value::as_str) { texts.push(text.to_string()); } } Some("image") => { if let Some(name) = save_image(&self.session_dir, part) { images.push(name); } } _ => {} } } } _ => {} } let about = block .get("tool_use_id") .and_then(Value::as_str) .unwrap_or_default() .to_string(); let patch = self.patches.remove(&about); if patch && block.get("is_error").and_then(Value::as_bool) != Some(true) { texts.clear(); } for image in images { events.push(Event::Image { image, about: Some(about.clone()), }); } events.push(Event::ToolEnd { id: about.clone(), output: texts.join("\n"), }); // Deliberately does *not* finish a subagent `about` might name: // the Task tool runs in the background by default, so this // `tool_result` -- "Async agent launched..." -- arrives at // launch, long before the subagent's own work is done. What // ends it is its own turn ending, handled in `translate_child`. } events } } fn replacement_diff(input: &Value) -> String { let path = input .get("file_path") .and_then(Value::as_str) .unwrap_or("file"); let old = input .get("old_string") .and_then(Value::as_str) .unwrap_or_default(); let new = input .get("new_string") .and_then(Value::as_str) .unwrap_or_default(); format!( "--- {path}\n+++ {path}\n@@\n{}{}", prefixed_lines('-', old), prefixed_lines('+', new) ) } fn prefixed_lines(prefix: char, text: &str) -> String { text.split_inclusive('\n') .map(|line| { if line.ends_with('\n') { format!("{prefix}{line}") } else { format!("{prefix}{line}\n") } }) .collect() } /// The title to start a subagent under when its own first line arrives /// before (or without) its Task call ever being seen: the tool name of that /// first line, which is the only thing known about it yet. `"subagent"` for /// a line this cannot even find a tool name in, such as one that opens with /// something other than a tool call. fn fallback_title(message: &Value) -> String { message["message"]["content"] .as_array() .into_iter() .flatten() .find(|block| block.get("type").and_then(Value::as_str) == Some("tool_use")) .and_then(|block| block.get("name")) .and_then(Value::as_str) .unwrap_or("subagent") .to_string() } /// Whether this line is a subagent's *own* turn ending -- the only thing /// that does, per `SUBAGENTS.md`: not the parent's `tool_result`, which for /// a background Task arrives at launch rather than at completion. /// /// Checked on the raw line rather than on what `dispatch` returns, so this /// never has to touch the shared `translate_stream_event`/`dispatch` code a /// top-level session's own turn-ending also goes through -- a subagent's /// idea of "ended" must not change when a real session's does. /// /// `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. /// /// **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 this event is the end of a turn: the two statuses a turn can /// finish in, and no others. /// /// A sibling of [`super::proves_a_turn`] and deliberately shaped like it -- /// see [`Translator::in_turn`]. `Exited` is not here: a process that has gone /// ends the session rather than the turn, and nothing after it can start one. fn closes_a_turn(event: &Event) -> bool { matches!( event, Event::Status { state: SessionStatus::Idle | SessionStatus::Waiting } ) } /// A task's ending in a word, for a card that has to say what became of it. /// /// `ended` has already said this is an ending, so the fallback is not "still /// going" -- it is the honest answer for an ending this build has no word for, /// which is that nobody here knows which it was. fn status_word(status: Option<&str>) -> &str { status.unwrap_or("ended, and this build cannot say how") } /// 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") => { let event = &message["event"]; event.get("type").and_then(Value::as_str) == Some("message_delta") && event["delta"].get("stop_reason").and_then(Value::as_str) == Some("end_turn") } Some("result") => true, _ => false, } } /// Whether a failed turn failed because the account is out of quota, and when /// the CLI said the limit lifts. /// /// The wording is the CLI's: a turn stopped by the limit ends with `is_error` /// and a result of `Claude AI usage limit reached|1788546972`, the reset being /// epoch seconds after a pipe. Matched on the sentence rather than on a code /// because the CLI sends none, so this is deliberately loose about everything /// but the four words. /// /// The two `None`s mean different things and both are real. The outer one is /// "some other failure". The inner one is "the limit is reached and the CLI did /// not say until when" -- which is not a reason to invent a time: `crate::resume` /// asks the usage endpoint before sending anything, and that answer is the one /// that decides. /// /// Milliseconds are accepted as well as seconds and told apart by magnitude, /// since a wrong guess would schedule a resume tens of thousands of years out /// and look exactly like auto-resume being broken. fn usage_limit(result: &str) -> Option> { if !result.to_ascii_lowercase().contains("usage limit reached") { return None; } let stamp = result .rsplit('|') .next() .and_then(|tail| tail.trim().parse::().ok()) .filter(|stamp| *stamp > 0.0) .map(|stamp| if stamp > 1e11 { stamp / 1000.0 } else { stamp }); Some(stamp) } /// A string field that is there and not empty, or `None`. The CLI omits these /// rather than sending them empty, but a caller that sends `""` means the same /// thing and should not produce a description that draws as a blank line. fn text_field(value: &Value, name: &str) -> Option { value .get(name) .and_then(Value::as_str) .filter(|text| !text.trim().is_empty()) .map(str::to_string) } /// Decodes one base64 image block into `files/` and returns its ref. /// /// A free function rather than a method because the import replay needs exactly /// this too: a session's history carries the same image blocks as its live /// output. Two copies would be two naming schemes for one directory. pub(in crate::session) fn save_image(session_dir: &Path, part: &Value) -> Option { let source = part.get("source")?; let data = source.get("data")?.as_str()?; use base64::Engine; let bytes = base64::engine::general_purpose::STANDARD .decode(data) .ok()?; // Screenshots are the overwhelming case and they are PNG; an unrecognized type is more likely // a dialect change than a JPEG. `store_image` owns that fallback. let media_type = source .get("media_type") .and_then(Value::as_str) .unwrap_or("image/png"); super::super::driver::store_image(session_dir, media_type, &bytes) } #[cfg(test)] mod tests { use super::*; /// What a phone sends back: everything chosen, even when that is one. fn chose(answer: &str) -> Vec { vec![answer.to_string()] } fn labels(options: &[QuestionOption]) -> Vec<&str> { options.iter().map(|option| option.label.as_str()).collect() } fn translate_lines(translator: &mut Translator, lines: &[&str]) -> Vec { lines .iter() .flat_map(|line| translator.translate(&serde_json::from_str(line).expect("json"))) .collect() } /// A fresh, empty subagent registry over the same temp dir a test's /// translator writes into -- every test here is about the parent's own /// events, so what a registry does with a subagent is `subagent.rs`'s /// tests to make, not these. fn test_subagents(dir: &tempfile::TempDir) -> Arc { Arc::new(Subagents::new(dir.path().to_path_buf())) } #[test] fn captures_the_resume_token_and_the_settings_from_init() { let dir = tempfile::tempdir().expect("tempdir"); let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir)); let events = translate_lines( &mut translator, &[ r#"{"type":"system","subtype":"init","cwd":"/x","session_id":"5ecf21da-d53f","tools":[],"model":"claude-haiku-4-5-20251001","permissionMode":"acceptEdits"}"#, ], ); assert_eq!(translator.session_id.as_deref(), Some("5ecf21da-d53f")); // The resolved model, which is the point: a session launched with // `--model haiku` is reported by its full name here. assert_eq!( events, vec![Event::Settings { model: Some("claude-haiku-4-5-20251001".to_string()), permission_mode: Some("acceptEdits".to_string()), }] ); } #[test] fn a_setting_is_reported_when_the_cli_accepts_it_and_not_before() { let dir = tempfile::tempdir().expect("tempdir"); let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir)); // What `set_model` does: remember, send, and say nothing yet. translator.expect_setting("req-a".to_string(), Setting::Model("sonnet".to_string())); translator.expect_setting( "req-b".to_string(), Setting::PermissionMode("plan".to_string()), ); // Success carries no model of its own -- measured on 2.1.237 -- so what // was asked for is the only answer available. let events = translate_lines( &mut translator, &[ r#"{"type":"control_response","response":{"subtype":"success","request_id":"req-a"}}"#, ], ); assert_eq!( events, vec![Event::Settings { model: Some("sonnet".to_string()), permission_mode: None, }] ); // A mode the CLI answers with a value of its own is taken from that // value: `auto` on the way in is `default` coming back. translator.expect_setting( "req-c".to_string(), Setting::PermissionMode("auto".to_string()), ); let events = translate_lines( &mut translator, &[ r#"{"type":"control_response","response":{"subtype":"success","request_id":"req-c","response":{"mode":"default"}}}"#, ], ); assert_eq!( events, vec![Event::Settings { model: None, permission_mode: Some("default".to_string()), }] ); // A refusal changes nothing, and says why rather than claiming a // setting that was rejected. let events = translate_lines( &mut translator, &[ r#"{"type":"control_response","response":{"subtype":"error","request_id":"req-b","error":"unknown mode"}}"#, ], ); assert_eq!( events, vec![Event::Error { message: "claude rejected a request: unknown mode".to_string() }] ); // And neither request is still waiting: a second answer to either id // reports nothing at all. let events = translate_lines( &mut translator, &[ r#"{"type":"control_response","response":{"subtype":"success","request_id":"req-a"}}"#, r#"{"type":"control_response","response":{"subtype":"success","request_id":"req-b"}}"#, ], ); assert!(events.is_empty(), "{events:?}"); } #[test] fn a_mode_the_cli_announces_is_taken_from_the_announcement() { // The line it sends just after answering `set_permission_mode`, which is // also how a mode changed from the terminal arrives. let dir = tempfile::tempdir().expect("tempdir"); let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir)); let events = translate_lines( &mut translator, &[ r#"{"type":"system","subtype":"status","status":null,"permissionMode":"plan","session_id":"s"}"#, ], ); assert_eq!( events, vec![Event::Settings { model: None, permission_mode: Some("plan".to_string()), }] ); } #[test] fn streams_text_deltas_and_skips_the_consolidated_copy() { // Real lines (trimmed) from the 2.1.237 probe. let dir = tempfile::tempdir().expect("tempdir"); let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir)); let events = translate_lines( &mut translator, &[ r#"{"type":"stream_event","event":{"type":"content_block_delta","index":1,"delta":{"type":"text_delta","text":"Done."}},"session_id":"s","parent_tool_use_id":null}"#, r#"{"type":"assistant","message":{"role":"assistant","content":[{"type":"text","text":"Done."}]},"parent_tool_use_id":null,"session_id":"s"}"#, r#"{"type":"stream_event","event":{"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":"hmm"}},"session_id":"s","parent_tool_use_id":null}"#, ], ); assert_eq!( events, vec![Event::AssistantText { delta: "Done.".to_string() }] ); } #[test] fn tool_use_and_result_become_tool_events() { let dir = tempfile::tempdir().expect("tempdir"); let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir)); let events = translate_lines( &mut translator, &[ r#"{"type":"assistant","message":{"role":"assistant","content":[{"type":"tool_use","id":"toolu_01","name":"Bash","input":{"command":"echo probe-ok"}}]},"parent_tool_use_id":null}"#, r#"{"type":"user","message":{"role":"user","content":[{"tool_use_id":"toolu_01","type":"tool_result","content":"probe-ok","is_error":false}]},"parent_tool_use_id":null}"#, ], ); assert_eq!( events, vec![ Event::ToolStart { id: "toolu_01".to_string(), tool: "Bash".to_string(), input: serde_json::json!({"command": "echo probe-ok"}), }, Event::ToolEnd { id: "toolu_01".to_string(), output: "probe-ok".to_string() }, ] ); } #[test] fn an_edit_becomes_a_patch_and_drops_only_its_success_boilerplate() { let dir = tempfile::tempdir().expect("tempdir"); let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir)); let events = translate_lines( &mut translator, &[ r#"{"type":"assistant","message":{"role":"assistant","content":[{"type":"tool_use","id":"edit-1","name":"Edit","input":{"file_path":"src/main.rs","old_string":"old","new_string":"new"}}]},"parent_tool_use_id":null}"#, r#"{"type":"user","message":{"role":"user","content":[{"tool_use_id":"edit-1","type":"tool_result","content":"The file src/main.rs has been updated successfully.","is_error":false}]},"parent_tool_use_id":null}"#, ], ); assert_eq!( events, vec![ patch_start( "edit-1".to_string(), "--- src/main.rs\n+++ src/main.rs\n@@\n-old\n+new\n".to_string() ), Event::ToolEnd { id: "edit-1".to_string(), output: String::new() } ] ); let failed = translate_lines( &mut translator, &[ r#"{"type":"assistant","message":{"role":"assistant","content":[{"type":"tool_use","id":"edit-2","name":"Edit","input":{"file_path":"src/main.rs","old_string":"missing","new_string":"new"}}]},"parent_tool_use_id":null}"#, r#"{"type":"user","message":{"role":"user","content":[{"tool_use_id":"edit-2","type":"tool_result","content":"old_string was not found","is_error":true}]},"parent_tool_use_id":null}"#, ], ); assert!(matches!( &failed[1], Event::ToolEnd { output, .. } if output == "old_string was not found" )); } #[test] fn subagent_events_are_not_duplicated_into_the_transcript() { let dir = tempfile::tempdir().expect("tempdir"); let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir)); let events = translate_lines( &mut translator, &[ r#"{"type":"assistant","message":{"role":"assistant","content":[{"type":"tool_use","id":"toolu_02","name":"Bash","input":{}}]},"parent_tool_use_id":"toolu_parent"}"#, ], ); assert!(events.is_empty()); } /// A child line does not just vanish from the parent -- it lands in its /// own subagent's transcript, with that transcript's own sequence /// numbers, starting at 1 like any other. #[test] fn a_child_line_lands_in_its_own_subagents_transcript() { 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_c1","name":"Bash","input":{"command":"echo hi"}}]},"parent_tool_use_id":"toolu_parent"}"#, ], ); let subagent = subagents.get("toolu_parent").expect("subagent started"); let lines = crate::session::transcript::read_after(&subagent.transcript_path(), 0) .expect("read subagent transcript"); assert_eq!(lines[0].seq, 1); assert_eq!( lines[0].event, Event::Status { state: SessionStatus::Running } ); assert!( lines.iter().any( |entry| matches!(&entry.event, Event::ToolStart { tool, .. } if tool == "Bash") ) ); } /// The title and prompt shown for a subagent come from the Task call /// that started it, not from anything guessed at its first line. #[test] fn the_subagent_takes_its_title_and_prompt_from_the_task_call() { 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_task","name":"Task","input":{"description":"Investigate the bug","prompt":"Find why X fails","subagent_type":"general-purpose"}}]},"parent_tool_use_id":null}"#, ], ); let rows = subagents.list(true); assert_eq!(rows.len(), 1); assert_eq!(rows[0].title, "Investigate the bug (general-purpose)"); let subagent = subagents.get(&rows[0].id).expect("subagent"); let lines = crate::session::transcript::read_after(&subagent.transcript_path(), 0) .expect("read subagent transcript"); assert!(lines.iter().any( |entry| matches!(&entry.event, Event::UserMessage { text, .. } if text == "Find why X fails") )); } /// The parent's `tool_result` for the Task id is what ends the /// subagent -- SUBAGENTS.md's lifecycle #3 -- and nothing else does. #[test] fn the_parents_tool_result_does_not_finish_the_subagent() { // The Task tool runs in the background by default: this // `tool_result` is "Async agent launched...", arriving the moment // the subagent *starts*, while it goes on working for however long // its own turn takes. Finishing it here was the bug -- a running // background agent read as "finished" with its transcript truncated // at launch. 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_task2","name":"Task","input":{"description":"helper"}}]},"parent_tool_use_id":null}"#, ], ); let subagent = subagents.get("toolu_task2").expect("subagent started"); assert!(subagent.is_open()); translate_lines( &mut translator, &[ r#"{"type":"user","message":{"role":"user","content":[{"type":"tool_result","tool_use_id":"toolu_task2","content":"Async agent launched","is_error":false}]},"parent_tool_use_id":null}"#, ], ); assert!(subagent.is_open()); } /// The end of a turn is not the end of the work when the session /// backgrounded something, and `Idle` says it is. Everything that reads a /// status hangs off this: the phone's word for the row, whether a /// "finished" notification goes out, and what auto-resume is looking at. #[test] fn a_turn_that_ends_with_a_task_still_running_is_waiting_rather_than_idle() { let dir = tempfile::tempdir().expect("tempdir"); let subagents = test_subagents(&dir); let mut translator = Translator::new(dir.path().to_path_buf(), Arc::clone(&subagents)); let result = r#"{"type":"result","subtype":"success","is_error":false,"usage":{}}"#; translate_lines( &mut translator, &[ r#"{"type":"assistant","message":{"content":[{"type":"tool_use","id":"toolu_bg","name":"Task","input":{"description":"the Dev Updater agent"}}]},"parent_tool_use_id":null}"#, r#"{"type":"system","subtype":"task_started","task_id":"t1","tool_use_id":"toolu_bg","is_backgrounded":true}"#, ], ); assert_eq!( translate_lines(&mut translator, &[result]).last(), Some(&Event::Status { state: SessionStatus::Waiting }) ); // The task reports back: the message the session received, and only // now the session is genuinely waiting for a person. assert_eq!( translate_lines( &mut translator, &[ r#"{"type":"system","subtype":"task_notification","task_id":"t1","tool_use_id":"toolu_bg","status":"completed","summary":"pushed as c41c36f"}"#, ], ), vec![Event::Status { state: SessionStatus::Idle }] ); // Reported once. The two lifecycle shapes can both arrive for one // task, and a second row for it would be a message that never came. assert!( translate_lines( &mut translator, &[ r#"{"type":"system","subtype":"task_updated","task_id":"t1","patch":{"status":"failed"}}"#, ], ) .is_empty() ); // And with nothing outstanding, the next turn ends idle as before. assert_eq!( translate_lines(&mut translator, &[result]).last(), Some(&Event::Status { state: SessionStatus::Idle }) ); } /// A backgrounded command is a task with no subagent behind it, so there is /// no second transcript for its report to live in -- its own tool card is /// the only record of it anywhere, and until the notification arrives that /// card is still showing the launch result saying it is running. #[test] fn a_backgrounded_command_reports_into_the_card_that_launched_it() { 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":"system","subtype":"task_started","task_id":"bg1","tool_use_id":"toolu_sh","is_backgrounded":true}"#, ], ); assert_eq!( translate_lines( &mut translator, &[ r#"{"type":"system","subtype":"task_notification","task_id":"bg1","tool_use_id":"toolu_sh","status":"completed","summary":"Background command \"run the tests\" completed (exit code 0)"}"#, ], ), vec![ Event::ToolUpdate { id: "toolu_sh".into(), output: r#"Background command "run the tests" completed (exit code 0)"#.into(), }, Event::Status { state: SessionStatus::Idle }, ] ); assert!( subagents.get("toolu_sh").is_none(), "a command is not a subagent and must not become one" ); } /// The states with no summary are exactly the ones that went wrong, and a /// card left saying "running in background" is the worst thing to leave on /// screen for them. #[test] fn a_backgrounded_command_that_failed_says_so_rather_than_saying_nothing() { let dir = tempfile::tempdir().expect("tempdir"); let subagents = test_subagents(&dir); let mut translator = Translator::new(dir.path().to_path_buf(), subagents); translate_lines( &mut translator, &[ r#"{"type":"system","subtype":"task_started","task_id":"bg2","tool_use_id":"toolu_sh2","is_backgrounded":true}"#, ], ); let events = translate_lines( &mut translator, &[ r#"{"type":"system","subtype":"task_updated","task_id":"bg2","patch":{"status":"failed"}}"#, ], ); assert_eq!( events.first(), Some(&Event::ToolUpdate { id: "toolu_sh2".into(), output: "this background command failed".into(), }) ); } /// The case a backend restart produces, which is every subagent a session /// has when the server is updated under it. Adoption picks the session's /// stdout back up from a recorded offset, so the `task_started` lines for /// anything already running are behind it and this translator never sees /// them: it starts empty, and asking only itself would report the session /// idle with a subagent plainly still working. #[test] fn a_subagent_that_started_before_this_translator_still_counts_as_outstanding() { let dir = tempfile::tempdir().expect("tempdir"); let subagents = test_subagents(&dir); // Started by somebody else, exactly as a previous run of the server // would have left it on disk. subagents.start("toolu_old", "the Dev Updater agent", None); let mut translator = Translator::new(dir.path().to_path_buf(), Arc::clone(&subagents)); let result = r#"{"type":"result","subtype":"success","is_error":false,"usage":{}}"#; assert_eq!( translate_lines(&mut translator, &[result]).last(), Some(&Event::Status { state: SessionStatus::Waiting }), "the registry knows about it even though this translator does not" ); // And its ending is reported, though nothing here saw it begin. assert_eq!( translate_lines( &mut translator, &[ r#"{"type":"system","subtype":"task_notification","task_id":"old","tool_use_id":"toolu_old","status":"completed","summary":"pushed"}"#, ], ), vec![Event::Status { state: SessionStatus::Idle }] ); // Once: `finish` closed it, so the second shape finds nothing. assert!( translate_lines( &mut translator, &[ r#"{"type":"system","subtype":"task_updated","task_id":"old","patch":{"status":"failed"}}"#, ], ) .is_empty() ); } /// A task ending *inside* a turn says nothing about the session's status: /// the turn is still running, and its own `result` decides. Without the /// `in_turn` guard this reported the session idle in the middle of one, /// which releases the message queue and tells every phone the work is /// over. #[test] fn a_task_ending_during_a_turn_does_not_report_the_session_idle() { let dir = tempfile::tempdir().expect("tempdir"); let subagents = test_subagents(&dir); let mut translator = Translator::new(dir.path().to_path_buf(), Arc::clone(&subagents)); let events = translate_lines( &mut translator, &[ r#"{"type":"assistant","message":{"content":[{"type":"tool_use","id":"toolu_fg","name":"Task","input":{"description":"a helper"}}]},"parent_tool_use_id":null}"#, r#"{"type":"system","subtype":"task_started","task_id":"t2","tool_use_id":"toolu_fg","is_backgrounded":false}"#, r#"{"type":"stream_event","event":{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"still going"}},"parent_tool_use_id":null}"#, r#"{"type":"system","subtype":"task_notification","task_id":"t2","tool_use_id":"toolu_fg","status":"completed","summary":"done"}"#, ], ); assert!( !events.iter().any(closes_a_turn), "the turn has not ended: {events:?}" ); // The helper's summary went where it belongs and nowhere else. let subagent = subagents.get("toolu_fg").expect("subagent started"); let lines = crate::session::transcript::read_after(&subagent.transcript_path(), 0).expect("read"); assert!( lines.iter().any(|entry| matches!( &entry.event, Event::AssistantText { delta } if delta == "done" )), "{lines:?}" ); } /// A background subagent can still be working long after its parent's own /// turn ended, so the account running out while one is mid-flight is the /// ordinary shape of the problem rather than an edge of it. The limit /// belongs in the subagent's transcript *and* has to reach the session, /// which is the only thing `crate::resume` can schedule against. #[test] fn a_limit_a_subagent_hits_reaches_the_session_as_well_as_the_subagent() { 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_lim","name":"Task","input":{"description":"a helper"}}]},"parent_tool_use_id":null}"#, ], ); let hit = translate_lines( &mut translator, &[ r#"{"type":"result","subtype":"error","is_error":true,"result":"Claude usage limit reached|1788726600","usage":{},"parent_tool_use_id":"toolu_lim"}"#, ], ); assert_eq!( hit, vec![Event::LimitReached { resets_at: Some(1788726600.0) }], "the session has to hear about it, or nothing resumes" ); let subagent = subagents.get("toolu_lim").expect("subagent started"); let lines = crate::session::transcript::read_after(&subagent.transcript_path(), 0).expect("read"); assert!( lines .iter() .any(|entry| matches!(entry.event, Event::LimitReached { .. })), "and so does the transcript it happened in: {lines:?}" ); } /// 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 /// state. #[test] fn the_subagents_own_end_turn_finishes_it() { 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_task3","name":"Task","input":{"description":"helper"}}]},"parent_tool_use_id":null}"#, r#"{"type":"stream_event","event":{"type":"message_delta","delta":{"stop_reason":"end_turn"}},"parent_tool_use_id":"toolu_task3"}"#, ], ); let subagent = subagents.get("toolu_task3").expect("subagent started"); assert!(!subagent.is_open()); let lines = crate::session::transcript::read_after(&subagent.transcript_path(), 0) .expect("read subagent transcript"); assert!( !lines .iter() .any(|entry| matches!(&entry.event, Event::Status { state } if *state == SessionStatus::Idle)), "a subagent's transcript must never carry Idle: {lines:?}" ); assert_eq!( lines.last().unwrap().event, Event::Status { state: SessionStatus::Exited } ); } /// `stop_reason: "tool_use"` is the model about to call a tool, with /// more of the turn still coming -- not an end. #[test] fn a_stop_reason_of_tool_use_does_not_finish_the_subagent() { 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_task4","name":"Task","input":{"description":"helper"}}]},"parent_tool_use_id":null}"#, r#"{"type":"stream_event","event":{"type":"message_delta","delta":{"stop_reason":"tool_use"}},"parent_tool_use_id":"toolu_task4"}"#, ], ); assert!( subagents .get("toolu_task4") .expect("subagent started") .is_open() ); } /// A background Task can be sent another message long after its first /// turn ended -- a further child line for it reopens rather than being /// dropped, and the same transcript and child translator carry on. #[test] fn a_line_after_finish_reopens_the_subagent_rather_than_being_dropped() { 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_task5","name":"Task","input":{"description":"helper"}}]},"parent_tool_use_id":null}"#, r#"{"type":"stream_event","event":{"type":"message_delta","delta":{"stop_reason":"end_turn"}},"parent_tool_use_id":"toolu_task5"}"#, ], ); let subagent = subagents.get("toolu_task5").expect("subagent started"); assert!(!subagent.is_open()); translate_lines( &mut translator, &[ r#"{"type":"assistant","message":{"content":[{"type":"tool_use","id":"toolu_more","name":"Bash","input":{}}]},"parent_tool_use_id":"toolu_task5"}"#, ], ); assert!(subagent.is_open()); let lines = crate::session::transcript::read_after(&subagent.transcript_path(), 0) .expect("read subagent transcript"); // Running, [prompt], Exited, Running (reopened), then the new line's // own ToolStart -- the same transcript throughout, not a new one. assert!( lines.iter().any( |entry| matches!(&entry.event, Event::ToolStart { tool, .. } if tool == "Bash") ) ); assert_eq!( lines .iter() .filter(|entry| matches!( &entry.event, Event::Status { state: SessionStatus::Running } )) .count(), 2, "expected one Running at creation and one at the reopen: {lines:?}" ); } /// Two subagents running at once keep two separate transcripts: tool ids /// are unique but a `stream_event`'s content-block index is not, so /// sharing translation state between them would cross their streams. #[test] fn two_parallel_subagents_keep_separate_transcripts() { 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_a","name":"Bash","input":{}}]},"parent_tool_use_id":"toolu_task_a"}"#, r#"{"type":"assistant","message":{"content":[{"type":"tool_use","id":"toolu_b","name":"Read","input":{}}]},"parent_tool_use_id":"toolu_task_b"}"#, ], ); let a = subagents.get("toolu_task_a").expect("subagent a"); let b = subagents.get("toolu_task_b").expect("subagent b"); let a_events = crate::session::transcript::read_after(&a.transcript_path(), 0) .expect("read a's transcript"); let b_events = crate::session::transcript::read_after(&b.transcript_path(), 0) .expect("read b's transcript"); assert!( a_events.iter().any( |entry| matches!(&entry.event, Event::ToolStart { tool, .. } if tool == "Bash") ) ); assert!( b_events.iter().any( |entry| matches!(&entry.event, Event::ToolStart { tool, .. } if tool == "Read") ) ); assert!( !a_events.iter().any( |entry| matches!(&entry.event, Event::ToolStart { tool, .. } if tool == "Read") ) ); assert!( !b_events.iter().any( |entry| matches!(&entry.event, Event::ToolStart { tool, .. } if tool == "Bash") ) ); } #[test] fn a_permission_request_becomes_an_allow_deny_question() { let dir = tempfile::tempdir().expect("tempdir"); let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir)); let events = translate_lines( &mut translator, &[ r#"{"type":"control_request","request_id":"req-1","request":{"subtype":"can_use_tool","tool_name":"Bash","input":{"command":"rm -rf /tmp/x"},"tool_use_id":"toolu_03"}}"#, ], ); let Event::Question { id, prompt, options, about, .. } = &events[0] else { panic!("expected a question, got {events:?}"); }; assert_eq!(id, "req-1"); // The call being asked about, so the phone draws the ask on that tool's // row instead of as a second card repeating its input. assert_eq!(about.as_deref(), Some("toolu_03")); assert!(prompt.contains("Bash") && prompt.contains("rm -rf /tmp/x")); assert_eq!(labels(options), ["Allow", "Deny"]); assert_eq!( events[1], Event::Status { state: SessionStatus::AwaitingInput } ); // Allowing echoes the input back; the request is then gone. let AnswerOutcome::Respond(response) = translator.answer("req-1", &chose("Allow")) else { panic!("expected a control response"); }; assert_eq!(response["response"]["request_id"], "req-1"); assert_eq!(response["response"]["response"]["behavior"], "allow"); assert_eq!( response["response"]["response"]["updatedInput"]["command"], "rm -rf /tmp/x" ); assert!(matches!( translator.answer("req-1", &chose("Allow")), AnswerOutcome::Unknown )); } #[test] fn denying_a_permission_sends_deny() { let dir = tempfile::tempdir().expect("tempdir"); let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir)); translate_lines( &mut translator, &[ r#"{"type":"control_request","request_id":"req-2","request":{"subtype":"can_use_tool","tool_name":"Write","input":{"file_path":"/etc/passwd"}}}"#, ], ); let AnswerOutcome::Respond(response) = translator.answer("req-2", &chose("Deny")) else { panic!("expected a control response"); }; assert_eq!(response["response"]["response"]["behavior"], "deny"); } #[test] fn ask_user_question_rides_the_same_flow_with_answers_keyed_by_question() { // The real 2.1.237 shape, verified live: answers go back inside // updatedInput, keyed by the question text. let dir = tempfile::tempdir().expect("tempdir"); let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir)); let events = translate_lines( &mut translator, &[ r#"{"type":"control_request","request_id":"req-3","request":{"subtype":"can_use_tool","tool_name":"AskUserQuestion","input":{"questions":[{"question":"Which color?","header":"Color","options":[{"label":"Red"},{"label":"Blue"}],"multiSelect":false},{"question":"Which size?","header":"Size","options":[{"label":"S"},{"label":"L"}],"multiSelect":false}]},"tool_use_id":"toolu_04","requires_user_interaction":true}}"#, ], ); let questions: Vec<_> = events .iter() .filter_map(|event| match event { Event::Question { id, prompt, options, multi_select, .. } => Some((id.clone(), prompt.clone(), options.clone(), *multi_select)), _ => None, }) .collect(); assert_eq!(questions.len(), 2); assert_eq!(questions[0].0, "req-3#0"); // Both belong to the call that asked, so a phone draws them on it. assert!(events.iter().all(|event| match event { Event::Question { about, .. } => about.as_deref() == Some("toolu_04"), _ => true, })); assert_eq!(questions[0].1, "Which color?"); assert_eq!(labels(&questions[0].2), ["Red", "Blue"]); // First answer alone isn't enough; the response goes out when the // last sub-question is answered, with all answers aboard. assert!(matches!( translator.answer("req-3#0", &chose("Blue")), AnswerOutcome::Pending )); let AnswerOutcome::Respond(response) = translator.answer("req-3#1", &chose("L")) else { panic!("expected a control response"); }; let updated = &response["response"]["response"]["updatedInput"]; assert_eq!(updated["answers"]["Which color?"], "Blue"); assert_eq!(updated["answers"]["Which size?"], "L"); assert_eq!(updated["questions"][0]["question"], "Which color?"); } #[test] fn a_question_carries_what_it_takes_to_answer_it() { // Descriptions and previews are what the reader decides on, and a // multi-select is how many answers the question takes. All of it travels // in the event: a phone that had to read this dialect's tool input to // find them would be the only place that knew how. let dir = tempfile::tempdir().expect("tempdir"); let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir)); let events = translate_lines( &mut translator, &[ r#"{"type":"control_request","request_id":"req-9","request":{"subtype":"can_use_tool","tool_name":"AskUserQuestion","input":{"questions":[{"question":"Which to collapse?","header":"Collapsed","multiSelect":true,"options":[{"label":"Tool calls","description":"A run becomes one card."},{"label":"Peer messages","description":"From other agents.","preview":"from: dev-updater\\npull before you touch it"}]}]},"tool_use_id":"toolu_09"}}"#, ], ); let Event::Question { header, options, multi_select, .. } = &events[0] else { panic!("expected a question, got {events:?}"); }; assert_eq!(header.as_deref(), Some("Collapsed")); assert!(multi_select); assert_eq!( options[0].description.as_deref(), Some("A run becomes one card.") ); assert!(options[0].preview.is_none()); assert!( options[1] .preview .as_deref() .unwrap() .contains("dev-updater") ); // Two choices, one answer: the joining is this dialect's shape, done // where it is spoken. The CLI's answers map holds strings. let AnswerOutcome::Respond(response) = translator.answer( "req-9#0", &["Tool calls".to_string(), "Peer messages".to_string()], ) else { panic!("expected a control response"); }; assert_eq!( response["response"]["response"]["updatedInput"]["answers"]["Which to collapse?"], "Tool calls, Peer messages" ); } #[test] fn images_in_tool_results_are_saved_and_referenced() { let dir = tempfile::tempdir().expect("tempdir"); let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir)); // A 1x1 PNG, the smallest real payload worth round-tripping. let png = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg=="; let line = format!( r#"{{"type":"user","message":{{"role":"user","content":[{{"type":"tool_result","tool_use_id":"toolu_05","content":[{{"type":"text","text":"took a screenshot"}},{{"type":"image","source":{{"type":"base64","media_type":"image/png","data":"{png}"}}}}]}}]}},"parent_tool_use_id":null}}"# ); let events = translator.translate(&serde_json::from_str(&line).expect("json")); let Event::Image { image, about } = &events[0] else { panic!("expected an image event, got {events:?}"); }; assert!(image.ends_with(".png")); // Named as belonging to the call that produced it, so a phone draws it // under that row rather than beside it. assert_eq!(about.as_deref(), Some("toolu_05")); let saved = dir.path().join("files").join(image); assert!(saved.is_file(), "image not saved at {}", saved.display()); assert_eq!( events[1], Event::ToolEnd { id: "toolu_05".to_string(), output: "took a screenshot".to_string(), } ); } #[test] fn a_turn_result_reports_usage_and_returns_to_idle() { let dir = tempfile::tempdir().expect("tempdir"); let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir)); let events = translate_lines( &mut translator, &[ r#"{"type":"result","subtype":"success","is_error":false,"num_turns":2,"stop_reason":"end_turn","session_id":"s","total_cost_usd":0.0149,"usage":{"input_tokens":18,"output_tokens":164}}"#, ], ); assert_eq!( events, vec![ Event::UsageDelta { tokens: 182, context: None }, Event::Status { state: SessionStatus::Idle }, ] ); } /// A turn another agent started says so, on the record that carries it. /// /// The line is the real shape, taken from a real cross-session message sent /// to a real stream-json session on 2.1.237 (2026-08-31) -- including the /// `from` socket path, which is deliberately *not* what a reader is shown: /// the sending session's `name` is what they recognise it by. The `body` is /// the message as written; the content the model is given wraps the same /// text in a preamble written for the model rather than for a person. /// /// The note comes before the usage and the idle, so it sits as close to the /// turn it explains as the wire allows. #[test] fn a_turn_started_by_another_agent_records_who_and_what() { let dir = tempfile::tempdir().expect("tempdir"); let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir)); let events = translate_lines( &mut translator, &[ r#"{"type":"result","subtype":"success","is_error":false,"session_id":"s","usage":{"input_tokens":2,"output_tokens":5},"origin":{"kind":"peer","from":"uds:/run/user/1000/cc-socks/137108.sock","verifiedPeerPid":137108,"msg_id":"1e729740","name":"ai-app-2-fb","fromMode":"prompting","body":"Reply with just the word ACK."}}"#, ], ); assert_eq!( events, vec![ Event::PeerMessage { from: "ai-app-2-fb".to_string(), text: "Reply with just the word ACK.".to_string(), // Stamped by the pump, which is the only place that knows // what seq the turn started at. turn_start: None, }, Event::UsageDelta { tokens: 7, context: None }, Event::Status { state: SessionStatus::Idle }, ] ); } /// And an ordinary turn does not, which is the half that decides whether /// the check above is a check or a rubber stamp. Measured over a real /// session's stdout: four results, no `origin` between them. #[test] fn an_ordinary_turn_carries_no_peer_note() { let dir = tempfile::tempdir().expect("tempdir"); let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir)); let events = translate_lines( &mut translator, &[ r#"{"type":"result","subtype":"success","is_error":false,"session_id":"s","usage":{"input_tokens":2,"output_tokens":5}}"#, ], ); assert!( !events .iter() .any(|event| matches!(event, Event::PeerMessage { .. })), "a turn nobody else started must not be attributed to anyone: {events:?}" ); } /// The context is the last assistant message's, not the result's. /// /// Real figures from a two-message haiku turn on 2.1.237, captured /// 2026-08-30. The result adds the turn up -- its `cache_read_input_tokens` /// of 40,211 is 14,259 and 25,952, the same conversation counted twice -- so /// reading the context off it would report a size the model never held, by /// more the more tool calls a turn makes. #[test] fn the_context_is_what_the_last_message_held_not_the_turn_added_up() { let dir = tempfile::tempdir().expect("tempdir"); let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir)); let events = translate_lines( &mut translator, &[ r#"{"type":"assistant","message":{"content":[],"usage":{"input_tokens":9,"cache_creation_input_tokens":11693,"cache_read_input_tokens":14259,"output_tokens":3}}}"#, r#"{"type":"assistant","message":{"content":[],"usage":{"input_tokens":8,"cache_creation_input_tokens":171,"cache_read_input_tokens":25952,"output_tokens":2}}}"#, r#"{"type":"result","subtype":"success","is_error":false,"session_id":"s","usage":{"input_tokens":17,"cache_creation_input_tokens":11864,"cache_read_input_tokens":40211,"output_tokens":156}}"#, ], ); assert_eq!( events.first(), Some(&Event::UsageDelta { tokens: 173, context: Some(26_131), }) ); // Taken by that result, so a following turn whose messages carry no // usage reports none rather than repeating this one's. let events = translate_lines( &mut translator, &[ r#"{"type":"result","subtype":"success","is_error":false,"session_id":"s","usage":{"input_tokens":4,"output_tokens":9}}"#, ], ); assert_eq!( events.first(), Some(&Event::UsageDelta { tokens: 13, context: None, }) ); } #[test] fn a_compaction_reports_its_start_and_what_it_recovered() { // Real lines (trimmed) from a 2.1.237 session driven through `/compact`. // Note the snake_case keys -- the CLI's transcript file writes the same // records in camelCase. let dir = tempfile::tempdir().expect("tempdir"); let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir)); let events = translate_lines( &mut translator, &[ r#"{"type":"system","subtype":"status","status":"compacting","session_id":"s","uuid":"u1"}"#, r#"{"type":"system","subtype":"status","status":null,"compact_result":"success","session_id":"s","uuid":"u2"}"#, r#"{"type":"system","subtype":"compact_boundary","session_id":"s","uuid":"u3","compact_metadata":{"trigger":"manual","pre_tokens":28719,"post_tokens":1125,"duration_ms":17130}}"#, ], ); assert_eq!( events, vec![ Event::Status { state: SessionStatus::Compacting }, Event::Status { state: SessionStatus::Running }, Event::Compacted { pre_tokens: Some(28719), post_tokens: Some(1125), trigger: Some("manual".to_string()), }, ] ); } #[test] fn a_failed_compaction_says_why_and_leaves_the_turn_running() { let dir = tempfile::tempdir().expect("tempdir"); let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir)); let events = translate_lines( &mut translator, &[ r#"{"type":"system","subtype":"status","status":"compacting","session_id":"s","uuid":"u1"}"#, r#"{"type":"system","subtype":"status","status":null,"compact_result":"failed","compact_error":"Not enough messages to compact.","session_id":"s","uuid":"u2"}"#, ], ); assert_eq!( events, vec![ Event::Status { state: SessionStatus::Compacting }, Event::Error { message: "compaction failed: Not enough messages to compact.".to_string() }, Event::Status { state: SessionStatus::Running }, ] ); } #[test] fn a_boundary_without_counts_says_so_rather_than_inventing_them() { let dir = tempfile::tempdir().expect("tempdir"); let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir)); let events = translate_lines( &mut translator, &[ r#"{"type":"system","subtype":"compact_boundary","session_id":"s","compact_metadata":{"trigger":"auto"}}"#, ], ); assert_eq!( events, vec![Event::Compacted { pre_tokens: None, post_tokens: None, trigger: Some("auto".to_string()), }] ); } #[test] fn an_error_result_surfaces_the_message() { let dir = tempfile::tempdir().expect("tempdir"); let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir)); let events = translate_lines( &mut translator, &[ r#"{"type":"result","subtype":"error_during_execution","is_error":true,"result":"something broke","usage":{}}"#, ], ); assert_eq!( events[0], Event::Error { message: "something broke".to_string() } ); assert_eq!( *events.last().unwrap(), Event::Status { state: SessionStatus::Idle } ); } /// Running out of quota is a state, not a failure of the work. /// /// The naive reading -- an error result like any other -- is what shipped /// before this: the transcript said "Claude AI usage limit reached|…" in /// red, which is neither readable nor actionable, and nothing above the /// driver could tell it apart from a broken tool call. #[test] fn a_turn_stopped_by_the_usage_limit_says_so_and_carries_the_reset() { let dir = tempfile::tempdir().expect("tempdir"); let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir)); let events = translate_lines( &mut translator, &[ r#"{"type":"result","subtype":"error_during_execution","is_error":true,"result":"Claude AI usage limit reached|1788546972","usage":{}}"#, ], ); assert_eq!( events[0], Event::LimitReached { resets_at: Some(1_788_546_972.0) } ); } #[test] fn a_limit_the_cli_gave_no_reset_for_is_reported_without_one() { let dir = tempfile::tempdir().expect("tempdir"); let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir)); let events = translate_lines( &mut translator, &[ r#"{"type":"result","subtype":"error_during_execution","is_error":true,"result":"Claude AI usage limit reached","usage":{}}"#, ], ); // Not a time this side invented: the meter is asked before anything is // sent, and a made-up reset would only decide when to ask. assert_eq!(events[0], Event::LimitReached { resets_at: None }); } #[test] fn a_reset_in_milliseconds_is_not_read_as_the_year_58000() { assert_eq!( usage_limit("Claude AI usage limit reached|1788546972000"), Some(Some(1_788_546_972.0)) ); // And anything that is not the limit stays an ordinary failure. assert_eq!(usage_limit("something broke"), None); } /// Pressing Stop is not a failure, and the CLI cannot tell you which it was. /// /// An interrupted turn arrives as exactly the same shape a broken one does, /// so somebody who pressed the button was shown "the turn ended with an /// error". What separates the two is that this side asked. The second half /// of this test is the one that matters, because the naive fix -- never /// reporting an error result -- passes the first half and silences every /// genuine failure afterwards. #[test] fn a_turn_stopped_on_purpose_is_not_an_error() { let dir = tempfile::tempdir().expect("tempdir"); let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir)); let stopped_result = r#"{"type":"result","subtype":"error_during_execution","is_error":true,"result":"Interrupted by user","usage":{}}"#; translator.expect_interrupt(); let events = translate_lines(&mut translator, &[stopped_result]); assert!( !events .iter() .any(|event| matches!(event, Event::Error { .. })), "a stop the driver asked for was reported as a failure: {events:?}" ); assert_eq!( *events.last().unwrap(), Event::Status { state: SessionStatus::Idle }, "an interrupted turn still has to end the turn" ); // The interrupt is spent, so the next failure is a failure again. let later = translate_lines(&mut translator, &[stopped_result]); assert!( later .iter() .any(|event| matches!(event, Event::Error { .. })), "a later failure was swallowed by an interrupt that had already been answered" ); } #[test] fn replayed_and_synthetic_user_text_is_skipped() { let dir = tempfile::tempdir().expect("tempdir"); let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir)); let events = translate_lines( &mut translator, &[ r#"{"type":"user","message":{"role":"user","content":[{"type":"text","text":"hi"}]},"isReplay":true,"parent_tool_use_id":null}"#, r#"{"type":"user","message":{"role":"user","content":[{"type":"text","text":"[continue]"}]},"isSynthetic":true,"parent_tool_use_id":null}"#, ], ); assert!(events.is_empty()); } }