//! Codex app-server notifications into the common event model. //! //! The CLI promises JSONL but deliberately leaves room for new item kinds. We //! therefore match only the records that have a useful common equivalent and //! ignore the rest; an added Codex item must not make a live session go deaf. use std::collections::{HashMap, HashSet}; use std::sync::{Arc, Mutex}; use serde_json::{Value, json}; use super::super::driver::{Event, SessionStatus, patch_start, prefixed_lines}; use super::super::subagent::Subagents; #[derive(Default)] pub(super) struct Translator { pub(super) thread_id: Option, completed: bool, limited: bool, subagents: Option>, background_processes: Option>>>, children: HashMap, prompts: HashMap, async_messages: HashSet, in_turn: bool, } impl Translator { pub(super) fn new( subagents: Arc, background_processes: Arc>>, thread_id: Option, in_turn: bool, ) -> Self { Self { thread_id, subagents: Some(subagents), background_processes: Some(background_processes), in_turn, ..Self::default() } } /// Whether a protocol notification belongs to the session's root thread. /// App-server multiplexes every spawned thread onto the same stdout, so /// the process-state bookkeeping in `codex.rs` must not mistake a child's /// turn for the session's own. pub(super) fn is_parent(&self, line: &Value) -> bool { if line.get("id").is_some() && line.get("method").is_none() { return true; } let kind = line.get("method").and_then(Value::as_str); let body = kind.and_then(|_| line.get("params")).unwrap_or(line); if let Some(root) = started_thread_is_root(kind, body) { // A clear or missing-rollout recovery replaces the root id inside this same // app-server. The new root cannot match the id this translator still holds; its null // parent is the authoritative distinction from a newly spawned subagent. return root; } match (self.thread_id.as_deref(), notification_thread(line)) { (Some(parent), Some(thread)) => parent == thread, _ => true, } } #[cfg(test)] pub(super) fn translate(&mut self, line: &Value) -> Vec { self.translate_with_prefix(line, Vec::new()) } /// Translates one multiplexed app-server record. `prefix` contains image /// events extracted by the driver and must precede the item's ToolEnd in /// whichever transcript owns the record. pub(super) fn translate_with_prefix(&mut self, line: &Value, prefix: Vec) -> Vec { let before = self.background_task_count(); let mut events = self.translate_inner(line, prefix); let after = self.background_task_count(); if after != before && let Some(count) = after { events.push(Event::BackgroundTasks { count }); } events } fn translate_inner(&mut self, line: &Value, mut prefix: Vec) -> Vec { let method = line.get("method").and_then(Value::as_str); let body = method.and_then(|_| line.get("params")).unwrap_or(line); let kind = method.or_else(|| line.get("type").and_then(Value::as_str)); let item = body.get("item").unwrap_or(&Value::Null); self.remember_collaboration(item); if item.get("type").and_then(Value::as_str) == Some("subAgentActivity") { return self.translate_activity(item); } self.remember_spawned_thread(kind, body); let thread = notification_thread(line); let root_started = started_thread_is_root(kind, body) == Some(true); let child = thread.filter(|thread| { !root_started && self .thread_id .as_deref() .is_some_and(|parent| parent != *thread) }); if let Some(id) = child { self.ensure_child(id, "subagent", None); if let Some(subagents) = &self.subagents { subagents.reopen(id); } let translator = self.children.entry(id.to_string()).or_default(); prefix.extend(translator.translate_line(kind, body, line)); return self.record_child(id, prefix); } if matches!(kind, Some("turn.started" | "turn/started")) { self.in_turn = true; } prefix.extend(self.translate_line(kind, body, line)); if matches!(kind, Some("turn.completed" | "turn/completed")) { self.in_turn = false; } prefix } fn background_task_count(&self) -> Option { self.subagents.as_ref().map(|subagents| { subagents.open_count() + self .background_processes .as_ref() .map(|processes| processes.lock().unwrap().len()) .unwrap_or(0) }) } fn translate_line(&mut self, kind: Option<&str>, body: &Value, line: &Value) -> Vec { match kind { Some("thread.started") => { self.thread_id = line .get("thread_id") .and_then(Value::as_str) .map(str::to_string); Vec::new() } Some("thread/started") => { self.thread_id = body .pointer("/thread/id") .and_then(Value::as_str) .map(str::to_string); Vec::new() } Some("turn.started") | Some("turn/started") => { self.completed = false; self.limited = false; vec![Event::Status { state: SessionStatus::Running, }] } Some("item.started") | Some("item/started") => { let item = &body["item"]; if let Some(id) = item.get("id").and_then(Value::as_str) && item.get("delivery").and_then(Value::as_str) == Some("async") { self.async_messages.insert(id.to_string()); } start_item(item) } Some("item.updated") => update_item(&line["item"]), // The old `codex exec --json` dialect reports only the completed message. App-server // also reports deltas, but they are provisional: safety buffering can revise their // text before completion. Keep its completed copy as an append-only correction rather // than guessing that the two representations concatenate to the same answer. Some("item.completed") => complete_item(&body["item"], true), Some("item/completed") => { let item = &body["item"]; let events = final_item(item); if let Some(id) = item.get("id").and_then(Value::as_str) { self.async_messages.remove(id); } events } Some("item/agentMessage/delta") => { if body .get("itemId") .and_then(Value::as_str) .is_some_and(|id| self.async_messages.contains(id)) { // The delta notification does not repeat `delivery`. // Appending an asynchronously delivered peer report as // assistant text merges it into the recipient's current // reply; the next completed assistant item then replaces // that whole row. The item/started record did identify it, // so leave its text to the PeerMessage emitted at completion. return Vec::new(); } let Some(delta) = body.get("delta").and_then(Value::as_str) else { return Vec::new(); }; vec![Event::AssistantText { delta: delta.to_string(), }] } Some("item/commandExecution/outputDelta") => { let (Some(id), Some(output)) = ( body.get("itemId").and_then(Value::as_str), body.get("delta").and_then(Value::as_str), ) else { return Vec::new(); }; vec![Event::ToolUpdate { id: id.to_string(), output: output.to_string(), }] } // One of these per *model request*, not per turn: app-server's `last` is the // request that just finished, and a turn is as many requests as it made tool // calls. Reported as each arrives, so `tokens` adds up to what the turn cost -- // held to the end of the turn it was the last request's cost alone, which on a // two-request turn measured 28,878 against the 51,399 actually spent -- and so // the context figure moves while a long turn is still running rather than // standing at what it was before the turn began. Some("thread/tokenUsage/updated") => { let last = &body["tokenUsage"]["last"]; last.get("totalTokens") .and_then(Value::as_u64) .map(|tokens| Event::UsageDelta { tokens, // Cached input is a subset of this figure, not an additional count. context: last.get("inputTokens").and_then(Value::as_u64), tokens_per_second: None, prefill_ms: None, }) .into_iter() .collect() } Some("turn.completed") | Some("turn/completed") => { self.completed = true; let mut events = Vec::new(); // The old `codex exec --json` dialect reports the turn's usage here and // sends no `thread/tokenUsage/updated` at all. if let Some(usage) = line.get("usage") { let input = number(usage, "input_tokens"); let output = number(usage, "output_tokens"); if input.is_some() || output.is_some() { events.push(Event::UsageDelta { tokens: input.unwrap_or(0) + output.unwrap_or(0), context: input, tokens_per_second: None, prefill_ms: None, }); } } if let Some(error) = body.pointer("/turn/error") && !error.is_null() { events.extend(self.failure(error)); } events.push(Event::Status { state: if self.background_task_count().is_some_and(|count| count > 0) { SessionStatus::Waiting } else { SessionStatus::Idle }, }); events } Some("turn.failed") | Some("error") => self.failure(body), _ => Vec::new(), } } /// Remembers prompts before the separate activity item announces which /// child thread a spawn produced. Interactions with an existing child are /// also real user turns in that child's transcript. fn remember_collaboration(&mut self, item: &Value) { if item.get("type").and_then(Value::as_str) != Some("collabAgentToolCall") { return; } let prompt = item .get("prompt") .and_then(Value::as_str) .filter(|prompt| !prompt.is_empty()) .map(str::to_string); if let (Some(id), Some(prompt)) = (item.get("id").and_then(Value::as_str), &prompt) { self.prompts.insert(id.to_string(), prompt.clone()); } let Some(receivers) = item.get("receiverThreadIds").and_then(Value::as_array) else { return; }; for receiver in receivers.iter().filter_map(Value::as_str) { if let Some(prompt) = &prompt { self.prompts.insert(receiver.to_string(), prompt.clone()); } } if item.get("status").and_then(Value::as_str) != Some("completed") || item.get("tool").and_then(Value::as_str) == Some("spawnAgent") { return; } let Some(prompt) = prompt else { return; }; for receiver in receivers.iter().filter_map(Value::as_str) { if self.thread_id.as_deref() == Some(receiver) { continue; } self.ensure_child(receiver, "subagent", None); if let Some(subagents) = &self.subagents { subagents.reopen(receiver); subagents.record( receiver, Event::UserMessage { id: None, text: prompt.clone(), attachments: Vec::new(), }, ); } } } fn remember_spawned_thread(&mut self, kind: Option<&str>, body: &Value) { if kind != Some("thread/started") { return; } let thread = &body["thread"]; if thread.get("parentThreadId").is_none_or(Value::is_null) { return; } let Some(id) = thread.get("id").and_then(Value::as_str) else { return; }; let title = thread .get("name") .and_then(Value::as_str) .or_else(|| { thread .pointer("/source/subAgent/thread_spawn/agent_path") .and_then(Value::as_str) }) .map(subagent_title) .unwrap_or_else(|| "subagent".to_string()); let prompt = self.prompts.get(id).cloned().or_else(|| { thread .get("preview") .and_then(Value::as_str) .map(str::to_string) }); self.ensure_child(id, &title, prompt.as_deref()); if let Some(subagents) = &self.subagents { subagents.reopen(id); } } fn translate_activity(&mut self, item: &Value) -> Vec { let Some(id) = item.get("agentThreadId").and_then(Value::as_str) else { return Vec::new(); }; let title = item .get("agentPath") .and_then(Value::as_str) .map(subagent_title) .unwrap_or_else(|| "subagent".to_string()); match item.get("kind").and_then(Value::as_str) { Some("started") => { let prompt = item .get("id") .and_then(Value::as_str) .and_then(|call| self.prompts.get(call)) .cloned() .or_else(|| self.prompts.get(id).cloned()); self.ensure_child(id, &title, prompt.as_deref()); if let Some(subagents) = &self.subagents { subagents.reopen(id); } item.get("id") .and_then(Value::as_str) .map(|call| { vec![Event::ToolEnd { id: call.to_string(), output: String::new(), }] }) .unwrap_or_default() } Some("interacted") => { self.ensure_child(id, &title, None); if let Some(subagents) = &self.subagents { subagents.reopen(id); } Vec::new() } Some("interrupted" | "completed") => { self.ensure_child(id, &title, None); let was_open = self .subagents .as_ref() .is_some_and(|subagents| subagents.is_open(id)); if let Some(subagents) = &self.subagents { subagents.finish(id); } self.prompts.remove(id); if was_open && !self.in_turn && self.background_task_count() == Some(0) { vec![Event::Status { state: SessionStatus::Idle, }] } else { Vec::new() } } _ => Vec::new(), } } fn ensure_child(&mut self, id: &str, title: &str, prompt: Option<&str>) { let Some(subagents) = &self.subagents else { return; }; if subagents.get(id).is_none() { subagents.start(id, title, prompt); } self.children.entry(id.to_string()).or_default(); } fn record_child(&self, id: &str, events: Vec) -> Vec { let Some(subagents) = &self.subagents else { return events; }; let mut parent = Vec::new(); for event in events { // A child remains running between its own turns and is closed by // SubAgentActivity, not by its turn/completed notification. if matches!( event, Event::Status { state: SessionStatus::Idle | SessionStatus::Waiting | SessionStatus::Exited } ) { continue; } if matches!(event, Event::LimitReached { .. }) { parent.push(event.clone()); } subagents.record(id, event); } parent } #[cfg(test)] pub(super) fn completed(&self) -> bool { self.completed } #[cfg(test)] pub(super) fn limited(&self) -> bool { self.limited } fn failure(&mut self, line: &Value) -> Vec { let detail = error_text(line); if is_limit(&detail) { if self.limited { return Vec::new(); } self.limited = true; vec![Event::LimitReached { resets_at: find_reset(line), }] } else { vec![Event::Error { message: detail }] } } } fn notification_thread(line: &Value) -> Option<&str> { let method = line.get("method").and_then(Value::as_str)?; let body = &line["params"]; if method == "thread/started" { body.pointer("/thread/id").and_then(Value::as_str) } else { body.get("threadId").and_then(Value::as_str) } } fn subagent_title(path: &str) -> String { path.rsplit('/') .find(|part| !part.is_empty()) .unwrap_or(path) .replace('_', " ") } fn started_thread_is_root(kind: Option<&str>, body: &Value) -> Option { (kind == Some("thread/started")).then(|| { body.pointer("/thread/parentThreadId") .is_none_or(Value::is_null) }) } fn start_item(item: &Value) -> Vec { if matches!( item.get("type").and_then(Value::as_str), Some("file_change" | "fileChange") ) { // Unlike command execution, a file change's start notification has no payload. Its // completed copy carries the diff, so that is where both common events are made; recording // this empty shell produced a Patch card containing only `diff:`. return Vec::new(); } let Some((id, tool, input)) = tool(item) else { return Vec::new(); }; vec![Event::ToolStart { id, tool, input }] } fn update_item(item: &Value) -> Vec { let Some(id) = item.get("id").and_then(Value::as_str) else { return Vec::new(); }; let output = item .get("aggregated_output") .or_else(|| item.get("output")) .and_then(value_text); output .filter(|text| !text.is_empty()) .map(|output| { vec![Event::ToolUpdate { id: id.to_string(), output, }] }) .unwrap_or_default() } fn complete_item(item: &Value, include_agent_message: bool) -> Vec { match item.get("type").and_then(Value::as_str) { Some("agent_message" | "agentMessage") if item.get("delivery").and_then(Value::as_str) == Some("async") => { delivered_agent_message(item).into_iter().collect() } Some("agent_message" | "agentMessage") => item .get("text") .and_then(Value::as_str) .filter(|text| !text.is_empty()) .filter(|_| include_agent_message) .map(|delta| { vec![Event::AssistantText { delta: delta.to_string(), }] }) .unwrap_or_default(), Some("reasoning" | "userMessage") => Vec::new(), Some("file_change" | "fileChange") => item .get("id") .and_then(Value::as_str) .map(|id| { vec![ patch_start(id.to_string(), file_change_diff(item)), Event::ToolEnd { id: id.to_string(), output: tool_output(item), }, ] }) .unwrap_or_default(), _ => { let Some((id, _, _)) = tool(item) else { return Vec::new(); }; let output = tool_output(item); vec![Event::ToolEnd { id, output }] } } } fn final_item(item: &Value) -> Vec { match item.get("type").and_then(Value::as_str) { Some("agentMessage") if item.get("delivery").and_then(Value::as_str) == Some("async") => { delivered_agent_message(item).into_iter().collect() } Some("agentMessage") => item .get("text") .and_then(Value::as_str) .map(|text| { vec![Event::AssistantTextFinal { text: text.to_string(), }] }) .unwrap_or_default(), _ => complete_item(item, false), } } fn tool(item: &Value) -> Option<(String, String, Value)> { let id = item.get("id")?.as_str()?.to_string(); let kind = item.get("type")?.as_str()?; let (name, input) = match kind { "command_execution" | "commandExecution" => command_tool(item), "file_change" | "fileChange" => return None, "mcp_tool_call" | "mcpToolCall" => ( item.get("tool") .or_else(|| item.get("name")) .and_then(Value::as_str) .map(|tool| format!("mcp:{tool}")) .unwrap_or_else(|| "mcp".to_string()), item.get("arguments").cloned().unwrap_or(Value::Null), ), "dynamicToolCall" => ( item.get("tool") .and_then(Value::as_str) .unwrap_or("tool") .to_string(), item.get("arguments").cloned().unwrap_or(Value::Null), ), "collabAgentToolCall" => { let name = match item.get("tool").and_then(Value::as_str) { Some("spawnAgent") => "Task", Some("sendInput" | "sendMessage" | "followupTask") => "SendMessage", Some("wait") => "TaskOutput", Some("closeAgent") => "CloseAgent", Some("interruptAgent") => "InterruptAgent", Some("listAgents") => "ListAgents", Some("resumeAgent") => "ResumeAgent", Some(tool) => tool, None => "Agent", }; let mut input = json!({}); if let Some(prompt) = item.get("prompt").and_then(Value::as_str) { input["prompt"] = Value::String(prompt.to_string()); } if let Some(model) = item.get("model").and_then(Value::as_str) { input["model"] = Value::String(model.to_string()); } if let Some(effort) = item.get("reasoningEffort").and_then(Value::as_str) { input["reasoningEffort"] = Value::String(effort.to_string()); } (name.to_string(), input) } "functionCallOutput" => ( item.get("name") .and_then(Value::as_str) .unwrap_or("tool") .to_string(), Value::Null, ), "imageView" => ( "view_image".to_string(), json!({"path": item.get("path").cloned().unwrap_or(Value::Null)}), ), "web_search" | "webSearch" => ( "WebSearch".to_string(), json!({"query": item.get("query").cloned().unwrap_or(Value::Null)}), ), "todo_list" | "todoList" | "plan" => ("update_plan".to_string(), item.clone()), _ => return None, }; Some((id, name, input)) } fn delivered_agent_message(item: &Value) -> Option { let text = item.get("text").and_then(Value::as_str)?; if text.is_empty() { return None; } let from = text .lines() .find_map(|line| line.strip_prefix("Sender: ")) .unwrap_or("another agent") .to_string(); let text = text .split_once("Payload:\n") .map(|(_, payload)| payload.trim()) .filter(|payload| !payload.is_empty()) .unwrap_or(text) .to_string(); Some(Event::PeerMessage { from, text, turn_start: None, }) } /// Codex records the executor's argv, while Claude reports the script handed to its Bash tool. /// Collapse Codex's standard wrapper to the same common shape so the transcript describes the /// command a person wrote, not the implementation used to start it. An unfamiliar executable is /// left intact: hiding that would make a deliberately selected shell look like Bash. fn command_tool(item: &Value) -> (String, Value) { let command = item.get("command").cloned().unwrap_or(Value::Null); if let Some(script) = command.as_str() { let script = rendered_bash_script(script).unwrap_or_else(|| script.to_string()); return ("Bash".to_string(), json!({"command": script})); } if let Some(argv) = command.as_array() && let [program, option, script] = argv.as_slice() && program .as_str() .and_then(|program| program.rsplit('/').next()) == Some("bash") && matches!(option.as_str(), Some("-c" | "-lc")) && let Some(script) = script.as_str() { return ("Bash".to_string(), json!({"command": script})); } let command = command .as_array() .map(|argv| argv.iter().map(shell_word).collect::>().join(" ")) .unwrap_or_else(|| command.to_string()); ("Shell".to_string(), json!({"command": command})) } /// App-server renders the executor argv into one shell-quoted string. Unwrap only an exact Bash /// invocation whose script has matching outer quotes; extra arguments stay visible rather than /// being mistaken for part of the script. fn rendered_bash_script(command: &str) -> Option { let quoted = ["/usr/bin/bash -lc ", "/bin/bash -lc ", "bash -lc "] .iter() .find_map(|prefix| command.strip_prefix(prefix))?; if quoted.len() >= 2 && ((quoted.starts_with('\'') && quoted.ends_with('\'')) || (quoted.starts_with('"') && quoted.ends_with('"'))) { return Some(quoted[1..quoted.len() - 1].to_string()); } (!quoted.is_empty() && !quoted.chars().any(char::is_whitespace)).then(|| quoted.to_string()) } fn shell_word(word: &Value) -> String { let Some(word) = word.as_str() else { return word.to_string(); }; if !word.is_empty() && word .chars() .all(|character| character.is_ascii_alphanumeric() || "/_-.=:,@+".contains(character)) { return word.to_string(); } format!("'{}'", word.replace('\'', "'\\''")) } fn file_change_diff(item: &Value) -> String { match item.get("changes") { // Current app-server protocol: [{path, kind: {type, move_path?}, diff}]. Some(Value::Array(changes)) => changes .iter() .filter_map(|change| { let path = change.get("path")?.as_str()?; let kind = change.pointer("/kind/type").and_then(Value::as_str); let moved = change.pointer("/kind/move_path").and_then(Value::as_str); let body = change .get("diff") .and_then(Value::as_str) .unwrap_or_default(); Some(render_file_diff(path, kind, moved, body)) }) .collect::>() .join("\n"), // Older exec protocol and durable Codex rollouts: {path: {type, unified_diff, ...}}. Some(Value::Object(changes)) => changes .iter() .map(|(path, change)| { let kind = change.get("type").and_then(Value::as_str); render_file_diff( path, kind, change.get("move_path").and_then(Value::as_str), change .get(if matches!(kind, Some("add" | "delete")) { "content" } else { "unified_diff" }) .and_then(Value::as_str) .unwrap_or_default(), ) }) .collect::>() .join("\n"), _ => String::new(), } } fn render_file_diff(path: &str, kind: Option<&str>, moved: Option<&str>, body: &str) -> String { // Some protocol revisions carry complete unified diffs and some carry only their hunks. if !matches!(kind, Some("add" | "delete")) && (body.starts_with("--- ") || body.starts_with("diff --git ")) { return body.to_string(); } let from = if kind == Some("add") { "/dev/null" } else { path }; let to = if kind == Some("delete") { "/dev/null" } else { moved.unwrap_or(path) }; let body = match kind { // Add and delete carry file content, not a unified diff. Prefixing it here makes the // protocol's change kind authoritative: a '-' that belongs to a Markdown bullet can never // be mistaken for diff syntax. Some("add") => prefixed_lines('+', body), Some("delete") => prefixed_lines('-', body), _ => body.to_string(), }; format!("--- {from}\n+++ {to}\n{body}") } fn tool_output(item: &Value) -> String { match item.get("type").and_then(Value::as_str) { Some("file_change" | "fileChange") if item.get("status").and_then(Value::as_str) == Some("completed") => { return String::new(); } Some("collabAgentToolCall") if item.get("status").and_then(Value::as_str) == Some("completed") => { return String::new(); } Some("dynamicToolCall") => { return content_text(item.get("contentItems"), "inputText"); } Some("mcpToolCall") => { if let Some(message) = item.pointer("/error/message").and_then(Value::as_str) { return message.to_string(); } return content_text(item.pointer("/result/content"), "text"); } Some("functionCallOutput") => return content_text(item.get("output"), "input_text"), Some("imageView") => return String::new(), _ => {} } for key in [ "aggregated_output", "aggregatedOutput", "output", "result", "error", ] { if let Some(text) = item.get(key).and_then(value_text) && !text.is_empty() { return text; } } match item.get("status").and_then(Value::as_str) { Some(status) => status.to_string(), None => String::new(), } } /// Text from a structured result, deliberately excluding its image data. The driver saves images /// beside the transcript; serializing a data URL here makes a screenshot a megabytes-long line and /// still cannot draw it. fn content_text(value: Option<&Value>, text_kind: &str) -> String { match value { Some(Value::String(text)) => text.clone(), Some(Value::Array(parts)) => parts .iter() .filter(|part| part.get("type").and_then(Value::as_str) == Some(text_kind)) .filter_map(|part| part.get("text").and_then(Value::as_str)) .collect::>() .join("\n"), _ => String::new(), } } fn value_text(value: &Value) -> Option { value .as_str() .map(str::to_string) .or_else(|| (!value.is_null()).then(|| value.to_string())) } fn number(value: &Value, key: &str) -> Option { value.get(key).and_then(Value::as_u64) } fn error_text(line: &Value) -> String { line.get("message") .and_then(Value::as_str) .or_else(|| line.pointer("/error/message").and_then(Value::as_str)) .or_else(|| line.get("error").and_then(Value::as_str)) .unwrap_or("Codex ended the turn with an unknown error") .to_string() } fn is_limit(detail: &str) -> bool { let lower = detail.to_ascii_lowercase(); lower.contains("usage limit") || lower.contains("rate limit") || lower.contains("quota exceeded") || lower.contains("credits depleted") } fn find_reset(value: &Value) -> Option { for key in ["resets_at", "resetsAt", "reset_at", "resetAt"] { if let Some(at) = value.get(key).and_then(Value::as_f64) { return Some(at); } } value.as_object()?.values().find_map(find_reset) } #[cfg(test)] mod tests { use super::*; fn line(text: &str) -> Value { serde_json::from_str(text).expect("fixture") } #[test] fn translates_the_observed_minimal_stream() { let mut translator = Translator::default(); assert!( translator .translate(&line(r#"{"type":"thread.started","thread_id":"thread-1"}"#)) .is_empty() ); assert_eq!(translator.thread_id.as_deref(), Some("thread-1")); assert_eq!( translator.translate(&line( r#"{"type":"item.completed","item":{"id":"item_0","type":"agent_message","text":"hello"}}"# )), vec![Event::AssistantText { delta: "hello".to_string() }] ); let events = translator.translate(&line( r#"{"type":"turn.completed","usage":{"input_tokens":13,"cached_input_tokens":8,"output_tokens":5}}"#, )); assert_eq!( events[0], Event::UsageDelta { tokens: 18, context: Some(13), tokens_per_second: None, prefill_ms: None, } ); assert!(translator.completed()); } #[test] fn translates_tools_and_limits_without_matching_whole_records() { let mut translator = Translator::default(); let started = translator.translate(&line( r#"{"type":"item.started","item":{"id":"item_1","type":"command_execution","command":["/usr/bin/bash","-lc","pwd"],"status":"in_progress"}}"#, )); assert_eq!( started, vec![Event::ToolStart { id: "item_1".to_string(), tool: "Bash".to_string(), input: json!({"command": "pwd"}) }] ); let ended = translator.translate(&line( r#"{"type":"item.completed","item":{"id":"item_1","type":"command_execution","command":"pwd","aggregated_output":"/tmp\n","exit_code":0,"status":"completed"}}"#, )); assert_eq!( ended, vec![Event::ToolEnd { id: "item_1".to_string(), output: "/tmp\n".to_string() }] ); let limit = translator.translate(&line( r#"{"type":"turn.failed","error":{"message":"usage limit reached","resetsAt":1234}}"#, )); assert_eq!( limit, vec![Event::LimitReached { resets_at: Some(1234.0) }] ); assert!(translator.limited()); } #[test] fn a_replacement_root_is_not_mistaken_for_a_subagent() { let mut translator = Translator { thread_id: Some("old-root".to_string()), ..Translator::default() }; let replacement = line( r#"{"method":"thread/started","params":{"thread":{"id":"new-root","parentThreadId":null}}}"#, ); assert!(translator.is_parent(&replacement)); assert!(translator.translate(&replacement).is_empty()); assert_eq!(translator.thread_id.as_deref(), Some("new-root")); let child = line( r#"{"method":"thread/started","params":{"thread":{"id":"child","parentThreadId":"new-root"}}}"#, ); assert!(!translator.is_parent(&child)); } #[test] fn command_translation_only_hides_the_known_bash_wrapper() { let legacy = tool(&line( r#"{"id":"old","type":"command_execution","command":"pwd"}"#, )); assert_eq!( legacy, Some(( "old".to_string(), "Bash".to_string(), json!({"command": "pwd"}) )) ); let rendered = tool(&json!({ "id": "rendered", "type": "commandExecution", "command": "/usr/bin/bash -lc 'printf \"%s\\n\" hello'" })); assert_eq!( rendered, Some(( "rendered".to_string(), "Bash".to_string(), json!({"command": "printf \"%s\\n\" hello"}) )) ); let double_quoted = tool(&json!({ "id": "double-quoted", "type": "commandExecution", "command": r#"/usr/bin/bash -lc "printf '%s\n' "$HOME" \path""# })); assert_eq!( double_quoted, Some(( "double-quoted".to_string(), "Bash".to_string(), json!({"command": "printf '%s\\n' \"$HOME\" \\path"}) )) ); let fish = tool(&line( r#"{"id":"fish","type":"commandExecution","command":["/usr/bin/fish","-c","pwd"]}"#, )); assert_eq!( fish, Some(( "fish".to_string(), "Shell".to_string(), json!({"command": "/usr/bin/fish -c pwd"}) )) ); } #[test] fn web_search_uses_the_common_tool_shape() { assert_eq!( tool(&line( r#"{"id":"search-1","type":"webSearch","query":"Codex app-server protocol"}"#, )), Some(( "search-1".to_string(), "WebSearch".to_string(), json!({"query": "Codex app-server protocol"}) )) ); } #[test] fn a_file_change_becomes_the_common_patch_shape() { let mut translator = Translator::default(); let started = translator.translate(&line( r#"{"method":"item/started","params":{"item":{"id":"patch-1","type":"fileChange","changes":{},"status":"inProgress"}}}"#, )); assert!(started.is_empty()); assert_eq!( translator.translate(&line( r#"{"method":"item/completed","params":{"item":{"id":"patch-1","type":"fileChange","changes":[{"path":"src/main.rs","kind":{"type":"update","move_path":null},"diff":"@@ -1 +1 @@\n-old\n+new\n"}],"status":"completed","stdout":"Success"}}}"# )), vec![ patch_start( "patch-1".to_string(), "--- src/main.rs\n+++ src/main.rs\n@@ -1 +1 @@\n-old\n+new\n".to_string() ), Event::ToolEnd { id: "patch-1".to_string(), output: String::new() } ] ); } #[test] fn the_legacy_file_change_shape_remains_readable() { let item = line( r#"{"changes":{"old.rs":{"type":"update","unified_diff":"@@ -1 +1 @@\n-a\n+b\n","move_path":"new.rs"}}}"#, ); assert_eq!( file_change_diff(&item), "--- old.rs\n+++ new.rs\n@@ -1 +1 @@\n-a\n+b\n" ); } #[test] fn whole_file_changes_get_markers_from_their_kind_not_their_content() { let item = line( r#"{"changes":[{"path":"added.md","kind":{"type":"add"},"diff":"--- a rule\n- a bullet\nplain\n"},{"path":"deleted.md","kind":{"type":"delete"},"diff":"- another bullet\nplain\n"}]}"#, ); assert_eq!( file_change_diff(&item), "--- /dev/null\n+++ added.md\n+--- a rule\n+- a bullet\n+plain\n\n--- deleted.md\n+++ /dev/null\n-- another bullet\n-plain\n" ); } #[test] fn legacy_whole_file_changes_show_the_content_they_changed() { let item = line( r#"{"changes":{"added.md":{"type":"add","content":"new\n"},"deleted.md":{"type":"delete","content":"old\n"}}}"#, ); assert_eq!( file_change_diff(&item), "--- /dev/null\n+++ added.md\n+new\n\n--- deleted.md\n+++ /dev/null\n-old\n" ); } #[test] fn native_app_server_completion_corrects_provisional_streaming() { let mut translator = Translator::default(); assert_eq!( translator.translate(&line( r#"{"method":"turn/started","params":{"threadId":"thread-1","turn":{"id":"turn-1"}}}"# )), vec![Event::Status { state: SessionStatus::Running }] ); assert_eq!( translator.translate(&line( r#"{"method":"item/agentMessage/delta","params":{"itemId":"message-1","delta":"hello"}}"# )), vec![Event::AssistantText { delta: "hello".to_string() }] ); assert_eq!( translator.translate(&line( r#"{"method":"item/completed","params":{"item":{"id":"message-1","type":"agentMessage","text":"hello, revised"}}}"# )), vec![Event::AssistantTextFinal { text: "hello, revised".to_string() }] ); // One per model request, as it arrives. A turn that made two of them costs both, // and the context figure moves while the turn is still running. assert_eq!( translator.translate(&line( r#"{"method":"thread/tokenUsage/updated","params":{"tokenUsage":{"last":{"inputTokens":39,"cachedInputTokens":30,"outputTokens":3,"reasoningOutputTokens":1,"totalTokens":42},"total":{"inputTokens":100,"cachedInputTokens":80,"outputTokens":9,"reasoningOutputTokens":2,"totalTokens":109},"modelContextWindow":258400}}}"# )), vec![Event::UsageDelta { tokens: 42, context: Some(39), tokens_per_second: None, prefill_ms: None, }] ); assert_eq!( translator.translate(&line( r#"{"method":"thread/tokenUsage/updated","params":{"tokenUsage":{"last":{"inputTokens":61,"cachedInputTokens":39,"outputTokens":4,"reasoningOutputTokens":1,"totalTokens":65},"total":{"inputTokens":161,"cachedInputTokens":119,"outputTokens":13,"reasoningOutputTokens":3,"totalTokens":174},"modelContextWindow":258400}}}"# )), vec![Event::UsageDelta { tokens: 65, context: Some(61), tokens_per_second: None, prefill_ms: None, }] ); assert_eq!( translator.translate(&line( r#"{"method":"turn/completed","params":{"turn":{"id":"turn-1","status":"completed","error":null}}}"# )), vec![Event::Status { state: SessionStatus::Idle }] ); } #[test] fn structured_tool_results_keep_text_but_not_image_data() { let mut translator = Translator::default(); let started = translator.translate(&line( r#"{"method":"item/started","params":{"item":{"id":"tool-1","type":"dynamicToolCall","tool":"view_image","arguments":{"path":"shot.png"},"status":"inProgress"}}}"#, )); assert!(matches!(&started[0], Event::ToolStart { tool, .. } if tool == "view_image")); let ended = translator.translate(&line( r#"{"method":"item/completed","params":{"item":{"id":"tool-1","type":"dynamicToolCall","tool":"view_image","arguments":{},"status":"completed","contentItems":[{"type":"inputText","text":"looked"},{"type":"inputImage","imageUrl":"data:image/png;base64,aGVsbG8="}]}}}"#, )); assert_eq!( ended, vec![Event::ToolEnd { id: "tool-1".to_string(), output: "looked".to_string() }] ); } #[test] fn an_app_server_completion_corrects_streamed_text_after_adoption() { // A newly adopted translator has not seen the deltas already recorded by the previous // backend. The dialect, rather than process-local memory, decides that this is a copy. let mut adopted = Translator::default(); assert_eq!( adopted.translate(&line( r#"{"method":"item/completed","params":{"item":{"id":"message-1","type":"agentMessage","text":"the complete message"}}}"# )), vec![Event::AssistantTextFinal { text: "the complete message".to_string() }] ); } #[test] fn codex_subagents_get_their_own_transcripts_and_hold_the_parent_waiting() { let dir = tempfile::tempdir().expect("tempdir"); let subagents = Arc::new(Subagents::new(dir.path().to_path_buf())); let background_processes = Arc::new(Mutex::new(HashSet::new())); let mut translator = Translator::new( Arc::clone(&subagents), Arc::clone(&background_processes), Some("parent-thread".to_string()), false, ); assert_eq!( translator.translate(&line( r#"{"method":"item/started","params":{"threadId":"parent-thread","turnId":"turn-1","item":{"id":"spawn-1","type":"collabAgentToolCall","tool":"spawnAgent","status":"inProgress","senderThreadId":"parent-thread","receiverThreadIds":[],"agentsStates":{},"prompt":"audit the history","model":null,"reasoningEffort":null}}}"# )), vec![Event::ToolStart { id: "spawn-1".to_string(), tool: "Task".to_string(), input: json!({"prompt": "audit the history"}) }] ); assert_eq!( translator.translate(&line( r#"{"method":"item/completed","params":{"threadId":"parent-thread","turnId":"turn-1","item":{"id":"spawn-1","type":"subAgentActivity","kind":"started","agentThreadId":"child-thread","agentPath":"/root/history_boundaries"}}}"# )), vec![ Event::ToolEnd { id: "spawn-1".to_string(), output: String::new() }, Event::BackgroundTasks { count: 1 } ] ); let rows = subagents.list(true); assert_eq!(rows.len(), 1); assert_eq!(rows[0].id, "child-thread"); assert_eq!(rows[0].title, "history boundaries"); assert!( translator .translate(&line( r#"{"method":"turn/started","params":{"threadId":"child-thread","turn":{"id":"child-turn"}}}"# )) .is_empty() ); assert!( translator .translate(&line( r#"{"method":"item/completed","params":{"threadId":"child-thread","turnId":"child-turn","item":{"id":"child-message","type":"agentMessage","text":"the audit result","phase":"final_answer","delivery":null}}}"# )) .is_empty() ); assert!( translator .translate(&line( r#"{"method":"turn/completed","params":{"threadId":"child-thread","turn":{"id":"child-turn","status":"completed","error":null}}}"# )) .is_empty() ); assert_eq!( translator.translate(&line( r#"{"method":"turn/completed","params":{"threadId":"parent-thread","turn":{"id":"turn-1","status":"completed","error":null}}}"# )), vec![Event::Status { state: SessionStatus::Waiting }] ); assert_eq!( translator.translate(&line( r#"{"method":"item/completed","params":{"threadId":"parent-thread","turnId":"turn-1","item":{"id":"subagent-completed-1","type":"subAgentActivity","kind":"completed","agentThreadId":"child-thread","agentPath":"/root/history_boundaries"}}}"# )), vec![ Event::Status { state: SessionStatus::Idle }, Event::BackgroundTasks { count: 0 } ] ); let child = subagents.get("child-thread").expect("child"); let events = crate::session::transcript::read_after(&child.transcript_path(), 0) .expect("child transcript"); assert!(matches!( &events[1].event, Event::UserMessage { text, .. } if text == "audit the history" )); assert!(events.iter().any(|event| matches!( &event.event, Event::AssistantTextFinal { text } if text == "the audit result" ))); assert_eq!( events.last().map(|event| &event.event), Some(&Event::Status { state: SessionStatus::Exited }) ); } #[test] fn codex_reports_each_change_to_its_live_background_count() { let dir = tempfile::tempdir().expect("tempdir"); let subagents = Arc::new(Subagents::new(dir.path().to_path_buf())); let background_processes = Arc::new(Mutex::new(HashSet::new())); let mut translator = Translator::new( Arc::clone(&subagents), Arc::clone(&background_processes), Some("parent-thread".to_string()), false, ); for (call, child, count) in [("spawn-a", "child-a", 1), ("spawn-b", "child-b", 2)] { let events = translator.translate(&json!({ "method": "item/completed", "params": { "threadId": "parent-thread", "item": { "id": call, "type": "subAgentActivity", "kind": "started", "agentThreadId": child, "agentPath": format!("/root/{child}") } } })); assert_eq!(events.last(), Some(&Event::BackgroundTasks { count })); } background_processes .lock() .unwrap() .insert("command-a".to_string()); for (child, count) in [("child-a", 2), ("child-b", 1)] { let events = translator.translate(&json!({ "method": "item/completed", "params": { "threadId": "parent-thread", "item": { "id": format!("completed-{child}"), "type": "subAgentActivity", "kind": "completed", "agentThreadId": child, "agentPath": format!("/root/{child}") } } })); assert_eq!(events.last(), Some(&Event::BackgroundTasks { count })); assert!(!events.iter().any(|event| { matches!( event, Event::Status { state: SessionStatus::Idle } ) })); } } #[test] fn codex_collaboration_coordination_has_clean_parent_tool_cards() { let mut translator = Translator::default(); for (tool, name) in [ ("wait", "TaskOutput"), ("sendInput", "SendMessage"), ("sendMessage", "SendMessage"), ("followupTask", "SendMessage"), ("closeAgent", "CloseAgent"), ("interruptAgent", "InterruptAgent"), ("listAgents", "ListAgents"), ("resumeAgent", "ResumeAgent"), ] { let started = json!({ "method": "item/started", "params": { "threadId": "parent-thread", "turnId": "turn-1", "item": { "id": format!("{tool}-1"), "type": "collabAgentToolCall", "tool": tool, "status": "inProgress", "senderThreadId": "parent-thread", "receiverThreadIds": [], "agentsStates": {}, "prompt": null, "model": null, "reasoningEffort": null } } }); let completed = json!({ "method": "item/completed", "params": { "threadId": "parent-thread", "turnId": "turn-1", "item": { "id": format!("{tool}-1"), "type": "collabAgentToolCall", "tool": tool, "status": "completed", "senderThreadId": "parent-thread", "receiverThreadIds": [], "agentsStates": {}, "prompt": null, "model": null, "reasoningEffort": null } } }); assert_eq!( translator.translate(&started), vec![Event::ToolStart { id: format!("{tool}-1"), tool: name.to_string(), input: json!({}) }], "{tool} start" ); assert_eq!( translator.translate(&completed), vec![Event::ToolEnd { id: format!("{tool}-1"), output: String::new() }], "{tool} completion" ); } } #[test] fn delivered_agent_messages_are_peer_messages_not_the_parents_reply() { let event = final_item(&json!({ "id": "delivery-1", "type": "agentMessage", "delivery": "async", "text": "Message Type: FINAL_ANSWER\nSender: /root/audit\nPayload:\nFound it." })); assert_eq!( event, vec![Event::PeerMessage { from: "/root/audit".to_string(), text: "Found it.".to_string(), turn_start: None }] ); } #[test] fn delivered_agent_message_deltas_do_not_replace_the_recipients_text() { let mut translator = Translator::default(); assert!( translator .translate(&line( r#"{"method":"item/started","params":{"threadId":"parent-thread","turnId":"turn-1","item":{"id":"delivery-1","type":"agentMessage","text":"","phase":null,"delivery":"async"}}}"# )) .is_empty() ); assert!( translator .translate(&line( r#"{"method":"item/agentMessage/delta","params":{"threadId":"parent-thread","turnId":"turn-1","itemId":"delivery-1","delta":"a child report"}}"# )) .is_empty() ); assert_eq!( translator.translate(&line( r#"{"method":"item/completed","params":{"threadId":"parent-thread","turnId":"turn-1","item":{"id":"delivery-1","type":"agentMessage","text":"a child report","phase":null,"delivery":"async"}}}"# )), vec![Event::PeerMessage { from: "another agent".to_string(), text: "a child report".to_string(), turn_start: None }] ); // Removing the completed id is the path out for this state: a later // item reusing it is not silently suppressed. assert_eq!( translator.translate(&line( r#"{"method":"item/agentMessage/delta","params":{"threadId":"parent-thread","turnId":"turn-2","itemId":"delivery-1","delta":"ordinary text"}}"# )), vec![Event::AssistantText { delta: "ordinary text".to_string() }] ); } #[test] fn an_adopted_turn_does_not_go_idle_when_its_last_child_finishes() { let dir = tempfile::tempdir().expect("tempdir"); let subagents = Arc::new(Subagents::new(dir.path().to_path_buf())); subagents.start("child-thread", "child", Some("work")); let mut translator = Translator::new( Arc::clone(&subagents), Arc::new(Mutex::new(HashSet::new())), Some("parent-thread".to_string()), true, ); assert_eq!( translator.translate(&line( r#"{"method":"item/completed","params":{"threadId":"parent-thread","turnId":"turn-1","item":{"id":"subagent-completed-1","type":"subAgentActivity","kind":"completed","agentThreadId":"child-thread","agentPath":"/root/child"}}}"# )), vec![Event::BackgroundTasks { count: 0 }] ); assert_eq!(subagents.list(true)[0].status, SessionStatus::Exited); } }