diff --git a/PLAN.md b/PLAN.md index 5434aae..72ff792 100644 --- a/PLAN.md +++ b/PLAN.md @@ -270,6 +270,17 @@ also makes adoption safe: if a backend restart falls between the deltas and the completion, the correction is still meaningful without process-local memory of which item ids streamed. +**Codex subagents share that one app-server process** (2026-09-13). Every +thread in the session tree is multiplexed onto its stdout and identified by +the notification's `threadId`. `collabAgentToolCall` carries the spawn prompt; +`subAgentActivity` carries the child thread id, path and lifecycle. The driver +uses that child thread id as the existing `Subagents` registry key, routes the +child's ordinary items through a separate translator into its own transcript, +and keeps only the root thread's process state in `CodexDriver`. A child turn +ending does not finish the child: `completed` or `interrupted` activity does. +An `agentMessage` marked `delivery: async` is a peer message delivered to a +thread, never that thread's own assistant reply. + ### The llama driver One `llama-server` per session, started through the same `Transport` as any @@ -877,13 +888,12 @@ session, which is what auto-resume needs. model, with no process and no controls of its own.** Full design and wire shape in `SUBAGENTS.md`, kept separate because the app half is being built against it in parallel and it is the shared contract between the two. The -one-paragraph reason: a session's Task-tool helpers already speak the common -event model on the parent's own stdout (each line carrying -`parent_tool_use_id`), so giving each one its own small transcript — same -file format, same paging routes, same SSE stream, reused by addressing rather -than by copying — costs a routing step in the translator and a registry -(`session/subagent.rs`) rather than a second session type with a driver, a -process and a config entry it does not need. +one-paragraph reason: Claude's Task helpers and Codex's collaboration threads +already speak their parent's event stream with a child identifier, so giving +each one its own small transcript — same file format, same paging routes, same +SSE stream, reused by addressing rather than by copying — costs a routing step +in the translator and a registry (`session/subagent.rs`) rather than a second +session type with a driver, a process and a config entry it does not need. ### HTTP surface diff --git a/SUBAGENTS.md b/SUBAGENTS.md index 1f8c19c..91ac4bd 100644 --- a/SUBAGENTS.md +++ b/SUBAGENTS.md @@ -1,9 +1,10 @@ # Subagents -A session's subagents -- the helpers a Claude Code session starts through its -Task tool -- each get a transcript of their own, listed under the session's -card and readable in the same transcript view the session has. Designed -2026-09-05; the decisions Bryan has not yet reviewed are in `DECISIONS.md`. +A session's subagents -- helpers started by Claude Code's Task tool or Codex's +collaboration tools -- each get a transcript of their own, listed under the +session's card and readable in the same transcript view the session has. +Designed 2026-09-05; extended to Codex's multiplexed app-server threads on +2026-09-13. The decisions Bryan has not yet reviewed are in `DECISIONS.md`. ## What a subagent is here @@ -14,24 +15,33 @@ own. Everything it shares with a session -- the transcript file format, the paging routes, the SSE stream, the phone's cache and rendering -- is reused by addressing, not by copying. -The CLI reports a subagent's messages on the parent's own stream-json -output, each carrying `parent_tool_use_id` = the id of the Task `tool_use` -that started it. Before this the translator dropped those lines +Claude reports a subagent's messages on the parent's own stream-json output, +each carrying `parent_tool_use_id` = the id of the Task `tool_use` that started +it. Before this the translator dropped those lines (`subagent_events_are_not_duplicated_into_the_transcript`); now it routes them to that subagent's own translator and transcript. The parent's transcript still shows only the Task call itself. +Codex app-server multiplexes every thread in the session tree onto the root +process's stdout. Its notifications carry `threadId`; `subAgentActivity` +items name the child thread and its lifecycle, and `collabAgentToolCall` +items carry the spawn prompt. The Codex translator routes a non-root +`threadId` exactly as Claude routes a `parent_tool_use_id`. The child thread +id is the subagent id on disk. An asynchronously delivered `agentMessage` is +a `PeerMessage`, not assistant text from the recipient. + ## Storage Under the session directory: ``` -/subagents//meta.json {title, created} -/subagents//transcript.jsonl same SeqEvent lines as the session's +/subagents//meta.json {title, created} +/subagents//transcript.jsonl same SeqEvent lines as the session's ``` -The id is the Task tool_use id (`toolu_…`), which is unique, stable across a -backend restart, and already the key everything on the parent side uses. +The id is Claude's Task tool_use id (`toolu_…`) or Codex's child thread id. +Both are unique, stable across a backend restart, and already the key their +parent-side lifecycle uses. Only ids matching `[A-Za-z0-9_-]+` are ever created or looked up, since the id becomes a path. @@ -119,6 +129,16 @@ transcript is still being written to and its process is the session's to stop. precisely the one nothing in this process has touched -- and it would otherwise read `running` again every time its session was started. +For Codex the same lifecycle is expressed by app-server rather than Claude's +task notices: `subAgentActivity.started` creates the child, +`subAgentActivity.interacted` reopens it, and `completed` or `interrupted` +finishes it. A child's own `turn/completed` is not its end; it remains running +until that activity edge. The root's `turn/completed` reports `waiting` while +the registry contains an open child, and the last activity completion reports +`idle` if the root is between turns. Because the child thread id is also the +on-disk id, an adopted driver can route and finish a child whose spawn record +is already behind the durable stdout offset. + A subagent that was mid-flight when the backend restarted keeps working: the registry reopens the existing transcript on the next child line, and the file continues its sequence -- the same reopening #4 describes, whether @@ -127,9 +147,12 @@ the backend was down nothing recorded that until the next line arrives, so its last status stays `Running`, which the list reports as **unknown** rather than as running (see the wire shape) until then. -Title: the Task call's `description` input, then ` ()` when -one is given; falling back to the tool's name when the child arrives before -(or without) the parent call being seen. +Title: for Claude, the Task call's `description` input, then +` ()` when one is given; falling back to the tool's name when +the child arrives before (or without) the parent call being seen. For Codex, +the first lifecycle record uses the spawned thread's name or the last segment +of `agentPath`, with underscores shown as spaces, then falls back to +`subagent`. ## Server layout @@ -142,6 +165,9 @@ one is given; falling back to the tool's name when the child arrives before - `session/claude/translate.rs` -- routes child lines by parent id, holds one child `Translator` per subagent, remembers pending Task calls' description/prompt/subagent_type. +- `session/codex/translate.rs` -- routes multiplexed app-server notifications + by thread id, remembers collaboration prompts, and translates activity + edges into the same registry lifecycle. - `session/echo.rs` -- `/subagent [n]`: the test rig. Starts *n* (default 1) subagents at once, each named "helper k". Each writes the prompt as its user message, streams a few words of text, runs one `Bash` tool call, then diff --git a/server/src/session/codex.rs b/server/src/session/codex.rs index 5d26279..c26bae2 100644 --- a/server/src/session/codex.rs +++ b/server/src/session/codex.rs @@ -1,9 +1,11 @@ //! Codex CLI sessions over its persistent app-server protocol. //! -//! One app-server owns one conversation. Unlike `codex exec --json`, this -//! surface can interrupt a turn without killing the connection and can steer -//! an active turn through `turn/steer`. Its stdio is a fifo plus logs so both -//! the process and an in-flight turn survive an ai-server restart. +//! One app-server owns one conversation tree. Unlike `codex exec --json`, +//! this surface can interrupt a turn without killing the connection and can +//! steer an active turn through `turn/steer`. Child threads are multiplexed +//! onto the same stdout and routed into subagent transcripts. Its stdio is a +//! fifo plus logs so both the process and in-flight turns survive an +//! ai-server restart. mod translate; @@ -22,6 +24,7 @@ use super::driver::{ AttachmentRef, Driver, Event, EventSink, SessionStatus, Unqueued, store_image, }; use super::process; +use super::subagent::Subagents; use super::transport::{Launch, Streams, Transport}; use crate::config::{ProviderConfig, SessionConfig}; use translate::Translator; @@ -96,6 +99,7 @@ struct Inner { to_child: mpsc::UnboundedSender, transport: Transport, session_dir: PathBuf, + subagents: Arc, reading: AtomicBool, } @@ -110,6 +114,7 @@ impl CodexDriver { transport: Transport, session_dir: &Path, sink: EventSink, + subagents: Arc, ) -> Result { let mut state = read_state(session_dir); let recorded = process::recorded(session_dir); @@ -170,6 +175,7 @@ impl CodexDriver { to_child, transport, session_dir: session_dir.to_path_buf(), + subagents, reading: AtomicBool::new(true), }); @@ -577,7 +583,11 @@ fn spawn_follower(inner: Arc, record: process::Record) { async fn follow(inner: Arc, mut record: process::Record, mut offset: u64) { let stdout = inner.session_dir.join(STDOUT_LOG); let stderr = inner.session_dir.join(STDERR_LOG); - let mut translator = Translator::default(); + let mut translator = Translator::new( + Arc::clone(&inner.subagents), + read_thread(&inner.session_dir), + inner.state.lock().unwrap().active_turn.is_some(), + ); // `offset` is the durable boundary after the last complete record. `read_at` may move beyond // it while app-server is still writing one record. Image-bearing tool results can be several // megabytes long, and rereading their incomplete prefix every 50 ms made arrival over ssh @@ -664,7 +674,8 @@ fn handle_line(inner: &Arc, translator: &mut Translator, line: &Value) { } let method = line.get("method").and_then(Value::as_str); let params = line.get("params").unwrap_or(&Value::Null); - match method { + let parent = translator.is_parent(line); + match method.filter(|_| parent) { Some("thread/started") => { if let Some(thread) = params.pointer("/thread/id").and_then(Value::as_str) { write_thread(&inner.session_dir, thread); @@ -697,9 +708,6 @@ fn handle_line(inner: &Arc, translator: &mut Translator, line: &Value) { if item.get("type").and_then(Value::as_str) == Some("userMessage") { announce_user(inner, item); } - for event in image_events(inner, item) { - let _ = inner.sink.send(event); - } } Some("turn/completed") => { let mut state = inner.state.lock().unwrap(); @@ -711,10 +719,15 @@ fn handle_line(inner: &Arc, translator: &mut Translator, line: &Value) { } _ => {} } - for event in translator.translate(line) { + let images = if method == Some("item/completed") { + image_events(inner, ¶ms["item"]) + } else { + Vec::new() + }; + for event in translator.translate_with_prefix(line, images) { let _ = inner.sink.send(event); } - if method == Some("turn/completed") { + if parent && method == Some("turn/completed") { dispatch_waiting(inner); } } diff --git a/server/src/session/codex/translate.rs b/server/src/session/codex/translate.rs index 82a9ef2..b76cf88 100644 --- a/server/src/session/codex/translate.rs +++ b/server/src/session/codex/translate.rs @@ -4,9 +4,13 @@ //! 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; +use std::sync::Arc; + use serde_json::{Value, json}; use super::super::driver::{Event, SessionStatus, patch_start}; +use super::super::subagent::Subagents; #[derive(Default)] pub(super) struct Translator { @@ -14,6 +18,10 @@ pub(super) struct Translator { completed: bool, limited: bool, pending_usage: Option, + subagents: Option>, + children: HashMap, + prompts: HashMap, + in_turn: bool, } struct Usage { @@ -22,10 +30,87 @@ struct Usage { } impl Translator { + pub(super) fn new(subagents: Arc, thread_id: Option, in_turn: bool) -> Self { + Self { + thread_id, + subagents: Some(subagents), + 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; + } + if line.get("method").and_then(Value::as_str) == Some("thread/started") + && line + .pointer("/params/thread/parentThreadId") + .is_some_and(|parent| !parent.is_null()) + { + return false; + } + 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, + 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 child = thread.filter(|thread| { + 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 translate_line(&mut self, kind: Option<&str>, body: &Value, line: &Value) -> Vec { match kind { Some("thread.started") => { self.thread_id = line @@ -113,7 +198,15 @@ impl Translator { events.extend(self.failure(error)); } events.push(Event::Status { - state: SessionStatus::Idle, + state: if self + .subagents + .as_ref() + .is_some_and(|subagents| subagents.any_open(true)) + { + SessionStatus::Waiting + } else { + SessionStatus::Idle + }, }); events } @@ -122,6 +215,181 @@ impl Translator { } } + /// 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); + } + Vec::new() + } + 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 + .subagents + .as_ref() + .is_some_and(|subagents| !subagents.any_open(true)) + { + 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 @@ -148,6 +416,23 @@ impl Translator { } } +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 start_item(item: &Value) -> Vec { if matches!( item.get("type").and_then(Value::as_str), @@ -185,6 +470,11 @@ fn update_item(item: &Value) -> Vec { 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) @@ -222,6 +512,9 @@ fn complete_item(item: &Value, include_agent_message: bool) -> Vec { 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) @@ -256,6 +549,30 @@ fn tool(item: &Value) -> Option<(String, String, Value)> { .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") { + input["prompt"] = prompt.clone(); + } + if let Some(model) = item.get("model") { + input["model"] = model.clone(); + } + if let Some(effort) = item.get("reasoningEffort") { + input["reasoningEffort"] = effort.clone(); + } + (name.to_string(), input) + } "functionCallOutput" => ( item.get("name") .and_then(Value::as_str) @@ -277,6 +594,29 @@ fn tool(item: &Value) -> Option<(String, String, Value)> { 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 @@ -744,4 +1084,132 @@ mod tests { }] ); } + + #[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 mut translator = Translator::new( + Arc::clone(&subagents), + 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", "model": null, "reasoningEffort": null}) + }] + ); + assert!( + 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"}}}"# + )) + .is_empty() + ); + 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 + }] + ); + + 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 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 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), + Some("parent-thread".to_string()), + true, + ); + + assert!( + 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"}}}"# + )) + .is_empty() + ); + assert_eq!(subagents.list(true)[0].status, SessionStatus::Exited); + } } diff --git a/server/src/session/mod.rs b/server/src/session/mod.rs index 2d8b69e..1dc3fe9 100644 --- a/server/src/session/mod.rs +++ b/server/src/session/mod.rs @@ -2548,6 +2548,7 @@ fn make_driver( Transport::for_machine(machine), dir, sink.clone(), + Arc::clone(subagents), )?), }) } diff --git a/server/src/session/subagent.rs b/server/src/session/subagent.rs index 0fbc005..a17c830 100644 --- a/server/src/session/subagent.rs +++ b/server/src/session/subagent.rs @@ -7,9 +7,10 @@ //! work on a subagent's file unchanged. //! //! Storage is `/subagents//{meta.json,transcript.jsonl}`, -//! where `` is the Task tool_use id that started it -- unique, stable -//! across a backend restart, and already the key the parent side uses. Only -//! ids matching [`is_subagent_id`] are ever turned into a path. +//! where `` is Claude's Task tool_use id or Codex's child thread id -- +//! unique, stable across a backend restart, and already the key the parent +//! side uses. Only ids matching [`is_subagent_id`] are ever turned into a +//! path. use std::collections::HashMap; use std::fs; @@ -112,8 +113,8 @@ impl Subagent { } } -/// Every subagent one session has started, keyed by the Task tool_use id -/// that names it. +/// Every subagent one session has started, keyed by the provider's stable +/// parent-side id for it. /// /// Lives beside a session's driver rather than inside it: a claude driver /// holds an `Arc` to this and routes child lines into it; echo uses it for