Support Codex subagent transcripts

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<String>,
     transport: Transport,
     session_dir: PathBuf,
+    subagents: Arc<Subagents>,
     reading: AtomicBool,
 }
 
@@ -110,6 +114,7 @@ impl CodexDriver {
         transport: Transport,
         session_dir: &Path,
         sink: EventSink,
+        subagents: Arc<Subagents>,
     ) -> Result<Self> {
         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<Inner>, record: process::Record) {
 async fn follow(inner: Arc<Inner>, 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<Inner>, 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<Inner>, 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<Inner>, translator: &mut Translator, line: &Value) {
         }
         _ => {}
     }
-    for event in translator.translate(line) {
+    let images = if method == Some("item/completed") {
+        image_events(inner, &params["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);
     }
 }

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<Usage>,
+    subagents: Option<Arc<Subagents>>,
+    children: HashMap<String, Translator>,
+    prompts: HashMap<String, String>,
+    in_turn: bool,
 }
 
 struct Usage {
@@ -22,10 +30,87 @@ struct Usage {
 }
 
 impl Translator {
+    pub(super) fn new(subagents: Arc<Subagents>, thread_id: Option<String>, 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<Event> {
+        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<Event>,
+    ) -> Vec<Event> {
         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<Event> {
         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<Event> {
+        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<Event>) -> Vec<Event> {
+        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<Event> {
     if matches!(
         item.get("type").and_then(Value::as_str),
@@ -185,6 +470,11 @@ fn update_item(item: &Value) -> Vec<Event> {
 
 fn complete_item(item: &Value, include_agent_message: bool) -> Vec<Event> {
     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<Event> {
 
 fn final_item(item: &Value) -> Vec<Event> {
     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<Event> {
+    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);
+    }
 }

server/src/session/mod.rs

@@ -2548,6 +2548,7 @@ fn make_driver(
             Transport::for_machine(machine),
             dir,
             sink.clone(),
+            Arc::clone(subagents),
         )?),
     })
 }

server/src/session/subagent.rs

@@ -7,9 +7,10 @@
 //! work on a subagent's file unchanged.
 //!
 //! Storage is `<session dir>/subagents/<id>/{meta.json,transcript.jsonl}`,
-//! where `<id>` 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 `<id>` 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