Add subagent tests and helpers; watchdog fixes

2026-11-05 20:10:33 +00:00 · 2025-11-22 09:26:13 -08:00
parent beb71f4a00
commit d560aff1aa
10 changed files with 1037 additions and 554 deletions
@@ -359,7 +359,8 @@ mod tests {
        assert!(
            shell_path == PathBuf::from("/bin/bash")
                || shell_path == PathBuf::from("/usr/bin/bash")
-                || shell_path == PathBuf::from("/usr/local/bin/bash"),
+                || shell_path == PathBuf::from("/usr/local/bin/bash")
+                || shell_path == PathBuf::from("/opt/homebrew/bin/bash"),
            "shell path: {shell_path:?}",
        );
    }
@@ -48,16 +48,18 @@ use crate::codex::Session;
 use crate::codex::TurnContext;
 use crate::codex_delegate::run_codex_conversation_interactive;
 use crate::error::CodexErr;
+use crate::model_family::derive_default_model_family;
+use crate::model_family::find_family_for_model;
 use crate::protocol::SandboxPolicy;
-use crate::model_family::{derive_default_model_family, find_family_for_model};
-use codex_protocol::config_types::ReasoningEffort;
 use crate::subagents::SubagentMetadata;
 use crate::subagents::SubagentOrigin;
 use crate::subagents::SubagentRegistry;
 use crate::subagents::SubagentStatus;
+use codex_protocol::config_types::ReasoningEffort;

 const LOG_CAPACITY: usize = 200;
 const ROOT_AGENT_ID: AgentId = 0;
+pub(crate) const MIN_WATCHDOG_INTERVAL_SECS: u64 = 30;

 #[derive(Clone)]
 struct RootInboxItem {
@@ -1070,6 +1072,13 @@ impl SubagentManager {
                .await);
        }

+        if interval_s < MIN_WATCHDOG_INTERVAL_SECS {
+            return Err(SubagentManagerError::InvalidWatchdogInterval {
+                requested: interval_s,
+                min: MIN_WATCHDOG_INTERVAL_SECS,
+            });
+        }
+
        let root_session_id = if target_agent_id == ROOT_AGENT_ID {
            self.find_root_session_id(caller_session_id)
                .await
@@ -2258,6 +2267,8 @@ pub enum SubagentManagerError {
        requested: SandboxMode,
        parent: SandboxMode,
    },
+    #[error("watchdog interval_s {requested}s is below the minimum supported interval of {min}s")]
+    InvalidWatchdogInterval { requested: u64, min: u64 },
 }

 #[derive(Debug, Clone)]
@@ -3077,4 +3088,454 @@ mod tests {

        assert_eq!(order, vec![2, 1]);
    }
+
+    #[tokio::test]
+    async fn drain_root_inbox_sorts_messages_per_sender() {
+        let registry = Arc::new(SubagentRegistry::new());
+        let manager = SubagentManager::new(Arc::clone(&registry), 4, false, true, false);
+        let root_session = ConversationId::new();
+        let child_session = ConversationId::new();
+
+        let meta = registry
+            .register_spawn(
+                child_session,
+                Some(root_session),
+                Some(ROOT_AGENT_ID),
+                3,
+                0,
+                Some("sort-me".to_string()),
+                None,
+            )
+            .await;
+
+        {
+            let mut inbox = manager.root_inbox.write().await;
+            inbox.insert(
+                root_session,
+                vec![
+                    RootInboxItem {
+                        sender_agent_id: meta.agent_id,
+                        timestamp_ms: 20,
+                        payload: RootInboxPayload::Message(InboxMessage {
+                            sender_agent_id: meta.agent_id,
+                            recipient_agent_id: ROOT_AGENT_ID,
+                            interrupt: false,
+                            prompt: Some("later".to_string()),
+                            timestamp_ms: 20,
+                        }),
+                        metadata: Some(meta.clone()),
+                    },
+                    RootInboxItem {
+                        sender_agent_id: meta.agent_id,
+                        timestamp_ms: 5,
+                        payload: RootInboxPayload::Message(InboxMessage {
+                            sender_agent_id: meta.agent_id,
+                            recipient_agent_id: ROOT_AGENT_ID,
+                            interrupt: false,
+                            prompt: Some("earlier".to_string()),
+                            timestamp_ms: 5,
+                        }),
+                        metadata: Some(meta.clone()),
+                    },
+                ],
+            );
+        }
+
+        let items = manager.drain_root_inbox_to_items(&root_session).await;
+        assert_eq!(items.len(), 2);
+
+        let payload: serde_json::Value = match &items[1] {
+            ResponseItem::FunctionCallOutput { output, .. } => {
+                serde_json::from_str(&output.content).expect("payload json")
+            }
+            other => panic!("unexpected item layout: {other:?}"),
+        };
+
+        let prompts = payload["messages"]
+            .as_array()
+            .expect("messages array")
+            .iter()
+            .map(|entry| entry["prompt"].as_str().unwrap().to_string())
+            .collect::<Vec<_>>();
+        assert_eq!(prompts, vec!["earlier".to_string(), "later".to_string()]);
+    }
+
+    #[tokio::test]
+    async fn drain_root_inbox_uses_registry_metadata_when_missing() {
+        let registry = Arc::new(SubagentRegistry::new());
+        let manager = SubagentManager::new(Arc::clone(&registry), 4, false, true, false);
+        let root_session = ConversationId::new();
+        let child_session = ConversationId::new();
+
+        registry
+            .register_spawn(
+                child_session,
+                Some(root_session),
+                Some(ROOT_AGENT_ID),
+                7,
+                0,
+                Some("from-registry".to_string()),
+                Some("registry summary".to_string()),
+            )
+            .await;
+
+        manager
+            .enqueue_root_inbox_message(
+                &root_session,
+                &child_session,
+                InboxMessage {
+                    sender_agent_id: 7,
+                    recipient_agent_id: ROOT_AGENT_ID,
+                    interrupt: false,
+                    prompt: Some("hello from child".to_string()),
+                    timestamp_ms: 5,
+                },
+            )
+            .await;
+
+        let items = manager.drain_root_inbox_to_items(&root_session).await;
+        assert_eq!(items.len(), 2);
+
+        let payload: serde_json::Value = match &items[1] {
+            ResponseItem::FunctionCallOutput { output, .. } => {
+                serde_json::from_str(&output.content).expect("payload json")
+            }
+            other => panic!("unexpected item layout: {other:?}"),
+        };
+
+        assert_eq!(
+            payload["metadata"]["session_id"],
+            serde_json::Value::String(child_session.to_string())
+        );
+        assert_eq!(payload["metadata"]["label"], "from-registry");
+        assert_eq!(payload["metadata"]["summary"], "registry summary");
+
+        let messages = payload["messages"]
+            .as_array()
+            .expect("messages array from payload");
+        assert_eq!(messages.len(), 1);
+        assert_eq!(messages[0]["prompt"], "hello from child");
+    }
+
+    #[tokio::test]
+    async fn drain_root_inbox_falls_back_to_root_metadata() {
+        let registry = Arc::new(SubagentRegistry::new());
+        let manager = SubagentManager::new(Arc::clone(&registry), 2, false, true, false);
+        let root_session = ConversationId::new();
+
+        {
+            let mut inbox = manager.root_inbox.write().await;
+            inbox.insert(
+                root_session,
+                vec![RootInboxItem {
+                    sender_agent_id: ROOT_AGENT_ID,
+                    timestamp_ms: 1,
+                    payload: RootInboxPayload::Message(InboxMessage {
+                        sender_agent_id: ROOT_AGENT_ID,
+                        recipient_agent_id: ROOT_AGENT_ID,
+                        interrupt: false,
+                        prompt: Some("hello root".to_string()),
+                        timestamp_ms: 1,
+                    }),
+                    metadata: None,
+                }],
+            );
+        }
+
+        let items = manager.drain_root_inbox_to_items(&root_session).await;
+        assert_eq!(items.len(), 2);
+
+        let payload: serde_json::Value = match &items[1] {
+            ResponseItem::FunctionCallOutput { output, .. } => {
+                serde_json::from_str(&output.content).expect("payload json")
+            }
+            other => panic!("unexpected output item: {other:?}"),
+        };
+
+        assert_eq!(
+            payload["metadata"]["session_id"],
+            serde_json::Value::String(root_session.to_string())
+        );
+        assert_eq!(payload["metadata"]["label"], "root");
+        assert_eq!(
+            payload["metadata"]["agent_id"].as_u64(),
+            Some(ROOT_AGENT_ID)
+        );
+
+        let messages = payload["messages"]
+            .as_array()
+            .expect("messages array from payload");
+        assert_eq!(messages.len(), 1);
+        assert_eq!(messages[0]["prompt"], "hello root");
+    }
+
+    #[tokio::test]
+    async fn drain_root_inbox_merges_messages_and_completion() {
+        let registry = Arc::new(SubagentRegistry::new());
+        let manager = SubagentManager::new(Arc::clone(&registry), 4, false, true, false);
+        let root_session = ConversationId::new();
+        let first_session = ConversationId::new();
+        let second_session = ConversationId::new();
+
+        let first_meta = registry
+            .register_spawn(
+                first_session,
+                Some(root_session),
+                Some(ROOT_AGENT_ID),
+                1,
+                0,
+                Some("first".to_string()),
+                None,
+            )
+            .await;
+        let second_meta = registry
+            .register_spawn(
+                second_session,
+                Some(root_session),
+                Some(ROOT_AGENT_ID),
+                2,
+                0,
+                Some("second".to_string()),
+                None,
+            )
+            .await;
+
+        {
+            let mut inbox = manager.root_inbox.write().await;
+            inbox.insert(
+                root_session,
+                vec![
+                    RootInboxItem {
+                        sender_agent_id: first_meta.agent_id,
+                        timestamp_ms: 1,
+                        payload: RootInboxPayload::Message(InboxMessage {
+                            sender_agent_id: first_meta.agent_id,
+                            recipient_agent_id: ROOT_AGENT_ID,
+                            interrupt: false,
+                            prompt: Some("first message".to_string()),
+                            timestamp_ms: 1,
+                        }),
+                        metadata: Some(first_meta.clone()),
+                    },
+                    RootInboxItem {
+                        sender_agent_id: first_meta.agent_id,
+                        timestamp_ms: 5,
+                        payload: RootInboxPayload::Completion(SubagentCompletion::Completed {
+                            last_message: Some("done".to_string()),
+                        }),
+                        metadata: Some(first_meta.clone()),
+                    },
+                    RootInboxItem {
+                        sender_agent_id: second_meta.agent_id,
+                        timestamp_ms: 3,
+                        payload: RootInboxPayload::Message(InboxMessage {
+                            sender_agent_id: second_meta.agent_id,
+                            recipient_agent_id: ROOT_AGENT_ID,
+                            interrupt: false,
+                            prompt: Some("second message".to_string()),
+                            timestamp_ms: 3,
+                        }),
+                        metadata: Some(second_meta.clone()),
+                    },
+                ],
+            );
+        }
+
+        let items = manager.drain_root_inbox_to_items(&root_session).await;
+        assert_eq!(items.len(), 4, "expected two call/output pairs");
+
+        let mut payloads = Vec::new();
+        for chunk in items.chunks(2) {
+            if let [
+                ResponseItem::FunctionCall { call_id, .. },
+                ResponseItem::FunctionCallOutput { output, .. },
+            ] = chunk
+            {
+                let parsed: serde_json::Value =
+                    serde_json::from_str(&output.content).expect("payload json");
+                payloads.push((call_id.clone(), parsed));
+            } else {
+                panic!("unexpected chunk layout: {chunk:?}");
+            }
+        }
+
+        assert_eq!(payloads.len(), 2);
+        assert_eq!(payloads[0].0, "await-1");
+        assert_eq!(payloads[1].0, "await-2");
+
+        let first_payload = &payloads[0].1;
+        let messages = first_payload["messages"]
+            .as_array()
+            .expect("messages array");
+        assert_eq!(messages.len(), 1);
+        assert_eq!(messages[0]["prompt"], "first message");
+
+        let completion = first_payload["completion"]
+            .get("Completed")
+            .and_then(|v| v.get("last_message"))
+            .and_then(|v| v.as_str());
+        assert_eq!(completion, Some("done"));
+
+        assert_eq!(first_payload["completion_status"], "completed");
+
+        let second_payload = &payloads[1].1;
+        let second_messages = second_payload["messages"]
+            .as_array()
+            .expect("messages array");
+        assert_eq!(second_messages.len(), 1);
+        assert_eq!(second_messages[0]["prompt"], "second message");
+        assert!(second_payload["completion"].is_null());
+    }
+
+    #[tokio::test]
+    async fn drain_root_inbox_merges_completion_then_message() {
+        let registry = Arc::new(SubagentRegistry::new());
+        let manager = SubagentManager::new(Arc::clone(&registry), 4, false, true, false);
+        let root_session = ConversationId::new();
+        let child_session = ConversationId::new();
+
+        let meta = registry
+            .register_spawn(
+                child_session,
+                Some(root_session),
+                Some(ROOT_AGENT_ID),
+                3,
+                0,
+                Some("has-completion".to_string()),
+                None,
+            )
+            .await;
+
+        {
+            let mut inbox = manager.root_inbox.write().await;
+            inbox.insert(
+                root_session,
+                vec![
+                    RootInboxItem {
+                        sender_agent_id: meta.agent_id,
+                        timestamp_ms: 7,
+                        payload: RootInboxPayload::Completion(SubagentCompletion::Failed {
+                            message: "boom".to_string(),
+                        }),
+                        metadata: Some(meta.clone()),
+                    },
+                    RootInboxItem {
+                        sender_agent_id: meta.agent_id,
+                        timestamp_ms: 8,
+                        payload: RootInboxPayload::Message(InboxMessage {
+                            sender_agent_id: meta.agent_id,
+                            recipient_agent_id: ROOT_AGENT_ID,
+                            interrupt: false,
+                            prompt: Some("after completion".to_string()),
+                            timestamp_ms: 8,
+                        }),
+                        metadata: Some(meta.clone()),
+                    },
+                ],
+            );
+        }
+
+        let items = manager.drain_root_inbox_to_items(&root_session).await;
+        assert_eq!(items.len(), 2, "expected a single call/output pair");
+
+        let payload: serde_json::Value = match &items[1] {
+            ResponseItem::FunctionCallOutput { output, .. } => {
+                serde_json::from_str(&output.content).expect("payload json")
+            }
+            other => panic!("unexpected output item: {other:?}"),
+        };
+
+        assert_eq!(payload["completion_status"], "failed");
+        assert_eq!(payload["completion"]["Failed"]["message"], "boom");
+
+        let prompts = payload["messages"]
+            .as_array()
+            .expect("messages array")
+            .iter()
+            .map(|entry| entry["prompt"].as_str().unwrap().to_string())
+            .collect::<Vec<_>>();
+        assert_eq!(prompts, vec!["after completion".to_string()]);
+    }
+
+    #[tokio::test]
+    async fn watchdog_action_rejects_short_intervals() {
+        let registry = Arc::new(SubagentRegistry::new());
+        let manager = SubagentManager::new(Arc::clone(&registry), 1, false, false, false);
+        let root_session = ConversationId::new();
+
+        let result = manager
+            .watchdog_action(
+                root_session,
+                ROOT_AGENT_ID,
+                ROOT_AGENT_ID,
+                MIN_WATCHDOG_INTERVAL_SECS - 1,
+                "ping".to_string(),
+                false,
+            )
+            .await;
+
+        assert!(matches!(
+            result,
+            Err(SubagentManagerError::InvalidWatchdogInterval { .. })
+        ));
+        assert!(manager.watchdogs.read().await.is_empty());
+    }
+
+    #[tokio::test]
+    async fn watchdog_action_cancel_returns_not_found() {
+        let registry = Arc::new(SubagentRegistry::new());
+        let manager = SubagentManager::new(Arc::clone(&registry), 1, false, false, false);
+        let root_session = ConversationId::new();
+
+        let action = manager
+            .watchdog_action(
+                root_session,
+                ROOT_AGENT_ID,
+                42,
+                MIN_WATCHDOG_INTERVAL_SECS,
+                "ping".to_string(),
+                true,
+            )
+            .await
+            .expect("cancel result");
+
+        assert_eq!(action, WatchdogAction::NotFound);
+        assert!(manager.watchdogs.read().await.is_empty());
+    }
+
+    #[tokio::test]
+    async fn watchdog_action_cancel_stops_existing_watchdog() {
+        let registry = Arc::new(SubagentRegistry::new());
+        let manager = SubagentManager::new(Arc::clone(&registry), 1, false, false, false);
+        let root_session = ConversationId::new();
+
+        let started = manager
+            .watchdog_action(
+                root_session,
+                ROOT_AGENT_ID,
+                ROOT_AGENT_ID,
+                MIN_WATCHDOG_INTERVAL_SECS,
+                "ping".to_string(),
+                false,
+            )
+            .await
+            .expect("start watchdog");
+        assert_eq!(started, WatchdogAction::Started);
+        assert_eq!(manager.watchdogs.read().await.len(), 1);
+
+        let canceled = manager
+            .watchdog_action(
+                root_session,
+                ROOT_AGENT_ID,
+                ROOT_AGENT_ID,
+                MIN_WATCHDOG_INTERVAL_SECS,
+                "ping".to_string(),
+                true,
+            )
+            .await
+            .expect("cancel watchdog");
+
+        assert_eq!(canceled, WatchdogAction::Canceled);
+        assert!(manager.watchdogs.read().await.is_empty());
+    }
 }
@@ -6,6 +6,7 @@ pub use manager::AwaitResult;
 pub use manager::ForkRequest;
 pub use manager::InboxMessage;
 pub use manager::LogEntry;
+pub(crate) use manager::MIN_WATCHDOG_INTERVAL_SECS;
 pub use manager::PruneErrorEntry;
 pub use manager::PruneReport;
 pub use manager::PruneRequest;
@@ -23,6 +23,7 @@ use crate::function_tool::FunctionCallError;
 use crate::parse_command::parse_command;
 use crate::subagents::ForkRequest;
 use crate::subagents::LogEntry;
+use crate::subagents::MIN_WATCHDOG_INTERVAL_SECS;
 use crate::subagents::PruneRequest;
 use crate::subagents::SendMessageRequest;
 use crate::subagents::SpawnRequest;
@@ -50,7 +51,6 @@ use codex_protocol::protocol::TaskCompleteEvent;
 use codex_protocol::protocol::TokenCountEvent;

 const MAX_AWAIT_TIMEOUT_SECS: u64 = 30 * 60;
-const MIN_WATCHDOG_INTERVAL_SECS: u64 = 30;
 const DEFAULT_WATCHDOG_INTERVAL_SECS: u64 = 300;

 const ROOT_AGENT_ID: AgentId = 0;
@@ -1708,6 +1708,11 @@ fn map_manager_error(
                "sandbox_mode {requested:?} not permitted; parent session runs in {parent:?}"
            ))
        }
+        SubagentManagerError::InvalidWatchdogInterval { requested, min } => {
+            FunctionCallError::RespondToModel(format!(
+                "watchdog interval_s {requested}s is below the minimum supported interval of {min}s"
+            ))
+        }
        SubagentManagerError::AgentIdMismatch {
            session_id,
            agent_id,
@@ -6,6 +6,9 @@ use codex_core::CodexConversation;
 use codex_core::config::Config;
 use codex_core::config::ConfigOverrides;
 use codex_core::config::ConfigToml;
+use codex_core::protocol::EventMsg;
+use codex_core::protocol::ExecCommandBeginEvent;
+use codex_core::protocol::ExecCommandEndEvent;
 use regex_lite::Regex;

 #[cfg(target_os = "linux")]
@@ -142,6 +145,64 @@ where
    matcher(&ev).unwrap()
 }

+#[allow(dead_code)]
+pub async fn wait_for_exec_command_begin(
+    codex: &CodexConversation,
+    call_id: &str,
+) -> ExecCommandBeginEvent {
+    wait_for_event_match(codex, |msg| match msg {
+        EventMsg::ExecCommandBegin(ev) if ev.call_id == call_id => Some(ev.clone()),
+        _ => None,
+    })
+    .await
+}
+
+#[allow(dead_code)]
+pub async fn wait_for_exec_command_end(
+    codex: &CodexConversation,
+    call_id: &str,
+) -> ExecCommandEndEvent {
+    wait_for_event_match(codex, |msg| match msg {
+        EventMsg::ExecCommandEnd(ev) if ev.call_id == call_id => Some(ev.clone()),
+        _ => None,
+    })
+    .await
+}
+
+pub async fn wait_for_exec_command_pair(
+    codex: &CodexConversation,
+    call_id: &str,
+) -> (ExecCommandBeginEvent, ExecCommandEndEvent) {
+    use tokio::time::Duration;
+    use tokio::time::Instant;
+    use tokio::time::timeout;
+
+    let deadline = Instant::now() + Duration::from_secs(10);
+    let mut begin: Option<ExecCommandBeginEvent> = None;
+    let mut end: Option<ExecCommandEndEvent> = None;
+
+    while begin.is_none() || end.is_none() {
+        let now = Instant::now();
+        if now >= deadline {
+            panic!("timeout waiting for exec events");
+        }
+        let remaining = deadline.saturating_duration_since(now);
+        let wait_for = remaining.max(Duration::from_secs(1));
+        let event = timeout(wait_for, codex.next_event())
+            .await
+            .expect("timeout waiting for exec events")
+            .expect("stream ended unexpectedly");
+
+        match event.msg {
+            EventMsg::ExecCommandBegin(ev) if ev.call_id == call_id => begin = Some(ev),
+            EventMsg::ExecCommandEnd(ev) if ev.call_id == call_id => end = Some(ev),
+            _ => {}
+        }
+    }
+
+    (begin.expect("missing begin"), end.expect("missing end"))
+}
+
 pub async fn wait_for_event_with_timeout<F>(
    codex: &CodexConversation,
    mut predicate: F,
@@ -11,15 +11,24 @@ use codex_core::WireApi;
 use codex_otel::otel_event_manager::OtelEventManager;
 use codex_protocol::ConversationId;
 use codex_protocol::protocol::SessionSource;
+use codex_protocol::protocol::SubAgentSource;
 use core_test_support::load_default_config_for_test;
 use core_test_support::responses;
 use futures::StreamExt;
 use tempfile::TempDir;
 use wiremock::matchers::header;

-#[tokio::test]
-async fn responses_stream_includes_subagent_header_on_review() {
-    core_test_support::skip_if_no_network!();
+struct ResponsesHeaderTest {
+    _server: wiremock::MockServer,
+    request_recorder: responses::ResponseMock,
+    client: ModelClient,
+}
+
+async fn setup_responses_header_test(
+    source: SubAgentSource,
+    expected_header: &str,
+) -> Option<ResponsesHeaderTest> {
+    core_test_support::skip_if_no_network!(None);

    let server = responses::start_mock_server().await;
    let response_body = responses::sse(vec![
@@ -29,7 +38,7 @@ async fn responses_stream_includes_subagent_header_on_review() {

    let request_recorder = responses::mount_sse_once_match(
        &server,
-        header("x-openai-subagent", "review"),
+        header("x-openai-subagent", expected_header),
        response_body,
    )
    .await;
@@ -79,9 +88,26 @@ async fn responses_stream_includes_subagent_header_on_review() {
        effort,
        summary,
        conversation_id,
-        SessionSource::SubAgent(codex_protocol::protocol::SubAgentSource::Review),
+        SessionSource::SubAgent(source),
    );

+    Some(ResponsesHeaderTest {
+        _server: server,
+        request_recorder,
+        client,
+    })
+}
+
+async fn assert_subagent_header(source: SubAgentSource, expected_header: &str) {
+    let Some(ResponsesHeaderTest {
+        _server,
+        request_recorder,
+        client,
+    }) = setup_responses_header_test(source, expected_header).await
+    else {
+        return;
+    };
+
    let mut prompt = Prompt::default();
    prompt.input = vec![ResponseItem::Message {
        id: None,
@@ -101,96 +127,16 @@ async fn responses_stream_includes_subagent_header_on_review() {
    let request = request_recorder.single_request();
    assert_eq!(
        request.header("x-openai-subagent").as_deref(),
-        Some("review")
+        Some(expected_header),
    );
 }

+#[tokio::test]
+async fn responses_stream_includes_subagent_header_on_review() {
+    assert_subagent_header(SubAgentSource::Review, "review").await;
+}
+
 #[tokio::test]
 async fn responses_stream_includes_subagent_header_on_other() {
-    core_test_support::skip_if_no_network!();
-
-    let server = responses::start_mock_server().await;
-    let response_body = responses::sse(vec![
-        responses::ev_response_created("resp-1"),
-        responses::ev_completed("resp-1"),
-    ]);
-
-    let request_recorder = responses::mount_sse_once_match(
-        &server,
-        header("x-openai-subagent", "my-task"),
-        response_body,
-    )
-    .await;
-
-    let provider = ModelProviderInfo {
-        name: "mock".into(),
-        base_url: Some(format!("{}/v1", server.uri())),
-        env_key: None,
-        env_key_instructions: None,
-        experimental_bearer_token: None,
-        wire_api: WireApi::Responses,
-        query_params: None,
-        http_headers: None,
-        env_http_headers: None,
-        request_max_retries: Some(0),
-        stream_max_retries: Some(0),
-        stream_idle_timeout_ms: Some(5_000),
-        requires_openai_auth: false,
-    };
-
-    let codex_home = TempDir::new().expect("failed to create TempDir");
-    let mut config = load_default_config_for_test(&codex_home);
-    config.model_provider_id = provider.name.clone();
-    config.model_provider = provider.clone();
-    let effort = config.model_reasoning_effort;
-    let summary = config.model_reasoning_summary;
-    let config = Arc::new(config);
-
-    let conversation_id = ConversationId::new();
-
-    let otel_event_manager = OtelEventManager::new(
-        conversation_id,
-        config.model.as_str(),
-        config.model_family.slug.as_str(),
-        None,
-        Some("test@test.com".to_string()),
-        Some(AuthMode::ChatGPT),
-        false,
-        "test".to_string(),
-    );
-
-    let client = ModelClient::new(
-        Arc::clone(&config),
-        None,
-        otel_event_manager,
-        provider,
-        effort,
-        summary,
-        conversation_id,
-        SessionSource::SubAgent(codex_protocol::protocol::SubAgentSource::Other(
-            "my-task".to_string(),
-        )),
-    );
-
-    let mut prompt = Prompt::default();
-    prompt.input = vec![ResponseItem::Message {
-        id: None,
-        role: "user".into(),
-        content: vec![ContentItem::InputText {
-            text: "hello".into(),
-        }],
-    }];
-
-    let mut stream = client.stream(&prompt).await.expect("stream failed");
-    while let Some(event) = stream.next().await {
-        if matches!(event, Ok(ResponseEvent::Completed { .. })) {
-            break;
-        }
-    }
-
-    let request = request_recorder.single_request();
-    assert_eq!(
-        request.header("x-openai-subagent").as_deref(),
-        Some("my-task")
-    );
+    assert_subagent_header(SubAgentSource::Other("my-task".to_string()), "my-task").await;
 }
@@ -5,68 +5,269 @@ use codex_core::subagents::LogEntry;
 use codex_protocol::ConversationId;
 use insta::assert_snapshot;
 use pretty_assertions::assert_eq;
-use serde_json::json;

-fn exec_sleep_logs() -> Vec<LogEntry> {
-    let events_json = json!([
-        {
-            "timestamp_ms": 1762823213424i64,
-            "event": {
-                "id": "0",
-                "msg": {
-                    "type": "reasoning_content_delta",
-                    "thread_id": "019a7073-88e5-7461-93a0-ae092f019246",
-                    "turn_id": "0",
-                    "item_id": "rs_0cb9136244ae700b0169128c2c63ec81a084a7fba2604df9fa",
-                    "delta": "**"
+fn session(id: &str) -> ConversationId {
+    ConversationId::from_string(id).expect("valid session id")
+}
+
+fn logs_from_value(value: serde_json::Value) -> Vec<LogEntry> {
+    serde_json::from_value(value).expect("valid subagent_logs JSON")
+}
+
+mod fixtures {
+    use super::LogEntry;
+    use super::logs_from_value;
+    use serde_json::json;
+
+    pub fn exec_sleep() -> Vec<LogEntry> {
+        logs_from_value(json!([
+            {
+                "timestamp_ms": 1762823213424i64,
+                "event": {
+                    "id": "0",
+                    "msg": {
+                        "type": "reasoning_content_delta",
+                        "thread_id": "019a7073-88e5-7461-93a0-ae092f019246",
+                        "turn_id": "0",
+                        "item_id": "rs_0cb9136244ae700b0169128c2c63ec81a084a7fba2604df9fa",
+                        "delta": "**"
+                    }
                }
-            }
-        },
-        {
-            "timestamp_ms": 1762823213442i64,
-            "event": {
-                "id": "0",
-                "msg": {
-                    "type": "item_completed",
-                    "thread_id": "019a7073-88e5-7461-93a0-ae092f019246",
-                    "turn_id": "0",
-                    "item": {
-                        "Reasoning": {
-                            "id": "rs_0cb9136244ae700b0169128c2c63ec81a084a7fba2604df9fa",
-                            "summary_text": ["**Evaluating safe shell command execution**"],
-                            "raw_content": []
+            },
+            {
+                "timestamp_ms": 1762823213442i64,
+                "event": {
+                    "id": "0",
+                    "msg": {
+                        "type": "item_completed",
+                        "thread_id": "019a7073-88e5-7461-93a0-ae092f019246",
+                        "turn_id": "0",
+                        "item": {
+                            "Reasoning": {
+                                "id": "rs_0cb9136244ae700b0169128c2c63ec81a084a7fba2604df9fa",
+                                "summary_text": ["**Evaluating safe shell command execution**"],
+                                "raw_content": []
+                            }
                        }
                    }
                }
-            }
-        },
-        {
-            "timestamp_ms": 1762823213442i64,
-            "event": {
-                "id": "0",
-                "msg": {
-                    "type": "agent_reasoning",
-                    "text": "**Evaluating safe shell command execution**"
+            },
+            {
+                "timestamp_ms": 1762823213442i64,
+                "event": {
+                    "id": "0",
+                    "msg": {
+                        "type": "agent_reasoning",
+                        "text": "**Evaluating safe shell command execution**"
+                    }
+                }
+            },
+            {
+                "timestamp_ms": 1762823213628i64,
+                "event": {
+                    "id": "0",
+                    "msg": {
+                        "type": "exec_command_begin",
+                        "call_id": "call_hBhXJmeCagENc5VGd12udWE3",
+                        "command": ["bash", "-lc", "sleep 60"],
+                        "cwd": "/Users/friel/code/codex",
+                        "parsed_cmd": [ { "type": "unknown", "cmd": "sleep 60" } ],
+                        "is_user_shell_command": false
+                    }
                }
            }
-        },
-        {
-            "timestamp_ms": 1762823213628i64,
-            "event": {
-                "id": "0",
-                "msg": {
-                    "type": "exec_command_begin",
-                    "call_id": "call_hBhXJmeCagENc5VGd12udWE3",
-                    "command": ["bash", "-lc", "sleep 60"],
-                    "cwd": "/Users/friel/code/codex",
-                    "parsed_cmd": [ { "type": "unknown", "cmd": "sleep 60" } ],
-                    "is_user_shell_command": false
-                }
-            }
-        }
-    ]);
+        ]))
+    }

-    serde_json::from_value(events_json).expect("valid exec_sleep_logs JSON")
+    pub fn baseline() -> Vec<LogEntry> {
+        logs_from_value(json!([
+            {
+                "timestamp_ms": 1762823311742i64,
+                "event": { "id": "0", "msg": { "type": "agent_message", "message": "Hello world" } }
+            },
+            {
+                "timestamp_ms": 1762823311766i64,
+                "event": {
+                    "id": "0",
+                    "msg": {
+                        "type": "token_count",
+                        "info": {
+                            "total_token_usage": {
+                                "input_tokens": 11073,
+                                "cached_input_tokens": 11008,
+                                "output_tokens": 8,
+                                "reasoning_output_tokens": 0,
+                                "total_tokens": 11081
+                            },
+                            "last_token_usage": {
+                                "input_tokens": 11073,
+                                "cached_input_tokens": 11008,
+                                "output_tokens": 8,
+                                "reasoning_output_tokens": 0,
+                                "total_tokens": 11081
+                            },
+                            "model_context_window": 258400
+                        },
+                        "rate_limits": { "primary": null, "secondary": null }
+                    }
+                }
+            },
+            {
+                "timestamp_ms": 1762823311766i64,
+                "event": {
+                    "id": "0",
+                    "msg": {
+                        "type": "raw_response_item",
+                        "item": {
+                            "type": "reasoning",
+                            "summary": [ { "type": "summary_text", "text": "**Identifying sandbox requirements**" } ],
+                            "content": null,
+                            "encrypted_content": "[encrypted]"
+                        }
+                    }
+                }
+            },
+            {
+                "timestamp_ms": 1762823311766i64,
+                "event": {
+                    "id": "0",
+                    "msg": {
+                        "type": "raw_response_item",
+                        "item": {
+                            "type": "message",
+                            "role": "assistant",
+                            "content": [ { "type": "output_text", "text": "Hello world" } ]
+                        }
+                    }
+                }
+            },
+            {
+                "timestamp_ms": 1762823311766i64,
+                "event": {
+                    "id": "0",
+                    "msg": { "type": "task_complete", "last_agent_message": "Hello world" }
+                }
+            }
+        ]))
+    }
+
+    pub fn streaming_deltas() -> Vec<LogEntry> {
+        logs_from_value(json!([
+            {
+                "timestamp_ms": 1762836527094i64,
+                "event": { "id": "0", "msg": { "type": "agent_message_content_delta", "thread_id": "019a713e-6ce6-7f82-b1e7-359628267934", "turn_id": "0", "item_id": "msg_0c5117240874292f016912c020d658819cb71e8bad4676a7c0", "delta": " is" } }
+            },
+            {
+                "timestamp_ms": 1762836527105i64,
+                "event": { "id": "0", "msg": { "type": "agent_message_content_delta", "thread_id": "019a713e-6ce6-7f82-b1e7-359628267934", "turn_id": "0", "item_id": "msg_0c5117240874292f016912c020d658819cb71e8bad4676a7c0", "delta": " composing" } }
+            },
+            {
+                "timestamp_ms": 1762836527121i64,
+                "event": { "id": "0", "msg": { "type": "agent_message_content_delta", "thread_id": "019a713e-6ce6-7f82-b1e7-359628267934", "turn_id": "0", "item_id": "msg_0c5117240874292f016912c020d658819cb71e8bad4676a7c0", "delta": " a" } }
+            },
+            {
+                "timestamp_ms": 1762836527137i64,
+                "event": { "id": "0", "msg": { "type": "agent_message_content_delta", "thread_id": "019a713e-6ce6-7f82-b1e7-359628267934", "turn_id": "0", "item_id": "msg_0c5117240874292f016912c020d658819cb71e8bad4676a7c0", "delta": " longer" } }
+            },
+            {
+                "timestamp_ms": 1762836527148i64,
+                "event": { "id": "0", "msg": { "type": "agent_message_content_delta", "thread_id": "019a713e-6ce6-7f82-b1e7-359628267934", "turn_id": "0", "item_id": "msg_0c5117240874292f016912c020d658819cb71e8bad4676a7c0", "delta": " answer" } }
+            }
+        ]))
+    }
+
+    pub fn reasoning_stream() -> Vec<LogEntry> {
+        logs_from_value(json!([
+            {
+                "timestamp_ms": 1_000i64,
+                "event": {
+                    "id": "0",
+                    "msg": {
+                        "type": "reasoning_content_delta",
+                        "thread_id": "thread-1",
+                        "turn_id": "0",
+                        "item_id": "rs_test",
+                        "delta": " thinking"
+                    }
+                }
+            },
+            {
+                "timestamp_ms": 1_050i64,
+                "event": {
+                    "id": "0",
+                    "msg": {
+                        "type": "reasoning_content_delta",
+                        "thread_id": "thread-1",
+                        "turn_id": "0",
+                        "item_id": "rs_test",
+                        "delta": " about"
+                    }
+                }
+            },
+            {
+                "timestamp_ms": 1_100i64,
+                "event": {
+                    "id": "0",
+                    "msg": {
+                        "type": "reasoning_content_delta",
+                        "thread_id": "thread-1",
+                        "turn_id": "0",
+                        "item_id": "rs_test",
+                        "delta": " streaming state"
+                    }
+                }
+            }
+        ]))
+    }
+
+    pub fn single_message() -> Vec<LogEntry> {
+        logs_from_value(json!([
+            {
+                "timestamp_ms": 1_000i64,
+                "event": {
+                    "id": "0",
+                    "msg": { "type": "agent_message", "message": "only event" }
+                }
+            }
+        ]))
+    }
+}
+
+fn render_full(session_id: &ConversationId, logs: &[LogEntry], total: usize, more: bool) -> String {
+    let earliest_ms = logs.first().map(|e| e.timestamp_ms);
+    let latest_ms = logs.last().map(|e| e.timestamp_ms);
+    render_logs_as_text(
+        *session_id,
+        logs,
+        earliest_ms,
+        latest_ms,
+        logs.len(),
+        total,
+        more,
+    )
+}
+
+fn render_with_lines(
+    session_id: &ConversationId,
+    logs: &[LogEntry],
+    total: usize,
+    more: bool,
+    max_lines: usize,
+    direction: PageDirection,
+) -> String {
+    let earliest_ms = logs.first().map(|e| e.timestamp_ms);
+    let latest_ms = logs.last().map(|e| e.timestamp_ms);
+    render_logs_as_text_with_max_lines(
+        *session_id,
+        logs,
+        earliest_ms,
+        latest_ms,
+        logs.len(),
+        total,
+        more,
+        max_lines,
+        direction,
+    )
 }

 #[test]
@@ -74,51 +275,14 @@ fn subagent_logs_paging_tail_vs_full_exec_sleep() {
    // Demonstrate that a one-line tail view is a suffix of the
    // full transcript, and that a generous max_lines reproduces
    // the full rendering.
-    let logs = exec_sleep_logs();
-    let session = ConversationId::from_string("019a7073-88e5-7461-93a0-adf67192b17b")
-        .expect("valid session id");
-    let earliest_ms = logs.first().map(|e| e.timestamp_ms);
-    let latest_ms = logs.last().map(|e| e.timestamp_ms);
-    let returned = logs.len();
+    let logs = fixtures::exec_sleep();
+    let session = session("019a7073-88e5-7461-93a0-adf67192b17b");
    let total = 21; // from exp4-real-run1
    let more = true;

-    // Full transcript for this window.
-    let full = render_logs_as_text(
-        session,
-        &logs,
-        earliest_ms,
-        latest_ms,
-        returned,
-        total,
-        more,
-    );
-
-    // Tail view: header + last content line only.
-    let tail_one = render_logs_as_text_with_max_lines(
-        session,
-        &logs,
-        earliest_ms,
-        latest_ms,
-        returned,
-        total,
-        more,
-        1,
-        PageDirection::Backward,
-    );
-
-    // Generous max_lines reproduces the full transcript.
-    let tail_many = render_logs_as_text_with_max_lines(
-        session,
-        &logs,
-        earliest_ms,
-        latest_ms,
-        returned,
-        total,
-        more,
-        30,
-        PageDirection::Backward,
-    );
+    let full = render_full(&session, &logs, total, more);
+    let tail_one = render_with_lines(&session, &logs, total, more, 1, PageDirection::Backward);
+    let tail_many = render_with_lines(&session, &logs, total, more, 30, PageDirection::Backward);

    assert_eq!(full, tail_many);

@@ -135,23 +299,16 @@ fn subagent_logs_paging_line_by_line_exec_sleep() {
    // Show what the transcript looks like as we increase the
    // line budget from 1 to 3 (backward paging), to mimic a
    // user scrolling back line-by-line.
-    let logs = exec_sleep_logs();
-    let session = ConversationId::from_string("019a7073-88e5-7461-93a0-adf67192b17b")
-        .expect("valid session id");
-    let earliest_ms = logs.first().map(|e| e.timestamp_ms);
-    let latest_ms = logs.last().map(|e| e.timestamp_ms);
-    let returned = logs.len();
+    let logs = fixtures::exec_sleep();
+    let session = session("019a7073-88e5-7461-93a0-adf67192b17b");
    let total = 21; // from exp4-real-run1
    let more = true;

    let mut pages = Vec::new();
    for max_lines in 1..=3 {
-        let rendered = render_logs_as_text_with_max_lines(
-            session,
+        let rendered = render_with_lines(
+            &session,
            &logs,
-            earliest_ms,
-            latest_ms,
-            returned,
            total,
            more,
            max_lines,
@@ -184,94 +341,12 @@ Session 019a7073-88e5-7461-93a0-adf67192b17b • status=waiting_on_tool • olde
 #[test]
 fn subagent_logs_snapshot_baseline() {
    // Grounded in exp1-real-run1 first subagent_logs response (t=0).
-    let events_json = json!([
-        {
-            "timestamp_ms": 1762823311742i64,
-            "event": { "id": "0", "msg": { "type": "agent_message", "message": "Hello world" } }
-        },
-        {
-            "timestamp_ms": 1762823311766i64,
-            "event": {
-                "id": "0",
-                "msg": {
-                    "type": "token_count",
-                    "info": {
-                        "total_token_usage": {
-                            "input_tokens": 11073,
-                            "cached_input_tokens": 11008,
-                            "output_tokens": 8,
-                            "reasoning_output_tokens": 0,
-                            "total_tokens": 11081
-                        },
-                        "last_token_usage": {
-                            "input_tokens": 11073,
-                            "cached_input_tokens": 11008,
-                            "output_tokens": 8,
-                            "reasoning_output_tokens": 0,
-                            "total_tokens": 11081
-                        },
-                        "model_context_window": 258400
-                    },
-                    "rate_limits": { "primary": null, "secondary": null }
-                }
-            }
-        },
-        {
-            "timestamp_ms": 1762823311766i64,
-            "event": {
-                "id": "0",
-                "msg": {
-                    "type": "raw_response_item",
-                    "item": {
-                        "type": "reasoning",
-                        "summary": [ { "type": "summary_text", "text": "**Identifying sandbox requirements**" } ],
-                        "content": null,
-                        "encrypted_content": "[encrypted]"
-                    }
-                }
-            }
-        },
-        {
-            "timestamp_ms": 1762823311766i64,
-            "event": {
-                "id": "0",
-                "msg": {
-                    "type": "raw_response_item",
-                    "item": {
-                        "type": "message",
-                        "role": "assistant",
-                        "content": [ { "type": "output_text", "text": "Hello world" } ]
-                    }
-                }
-            }
-        },
-        {
-            "timestamp_ms": 1762823311766i64,
-            "event": {
-                "id": "0",
-                "msg": { "type": "task_complete", "last_agent_message": "Hello world" }
-            }
-        }
-    ]);
-
-    let logs: Vec<LogEntry> =
-        serde_json::from_value(events_json).expect("valid baseline logs JSON");
-    let session = ConversationId::from_string("019a7075-0760-79c2-8dd1-985772995ecf")
-        .expect("valid session id");
-    let earliest_ms = logs.first().map(|e| e.timestamp_ms);
-    let latest_ms = logs.last().map(|e| e.timestamp_ms);
-    let returned = logs.len();
-    let total = logs.len();
-    let more = false;
-
-    let rendered = render_logs_as_text(
-        session,
+    let logs = fixtures::baseline();
+    let rendered = render_full(
+        &session("019a7075-0760-79c2-8dd1-985772995ecf"),
        &logs,
-        earliest_ms,
-        latest_ms,
-        returned,
-        total,
-        more,
+        logs.len(),
+        false,
    );

    assert_snapshot!(
@@ -286,23 +361,12 @@ fn subagent_logs_snapshot_baseline() {
 #[test]
 fn subagent_logs_snapshot_exec_sleep_command() {
    // Grounded in exp4-real-run1 first subagent_logs response (t=0).
-    let logs = exec_sleep_logs();
-    let session = ConversationId::from_string("019a7073-88e5-7461-93a0-adf67192b17b")
-        .expect("valid session id");
-    let earliest_ms = logs.first().map(|e| e.timestamp_ms);
-    let latest_ms = logs.last().map(|e| e.timestamp_ms);
-    let returned = logs.len();
-    let total = logs.len();
-    let more = false;
-
-    let rendered = render_logs_as_text(
-        session,
+    let logs = fixtures::exec_sleep();
+    let rendered = render_full(
+        &session("019a7073-88e5-7461-93a0-adf67192b17b"),
        &logs,
-        earliest_ms,
-        latest_ms,
-        returned,
-        total,
-        more,
+        logs.len(),
+        false,
    );

    assert_snapshot!(
@@ -317,46 +381,12 @@ fn subagent_logs_snapshot_exec_sleep_command() {
 #[test]
 fn subagent_logs_snapshot_streaming_deltas() {
    // Grounded in exp5-real-run1 agent_message_content_delta stream (t≈?s).
-    let events_json = json!([
-        {
-            "timestamp_ms": 1762836527094i64,
-            "event": { "id": "0", "msg": { "type": "agent_message_content_delta", "thread_id": "019a713e-6ce6-7f82-b1e7-359628267934", "turn_id": "0", "item_id": "msg_0c5117240874292f016912c020d658819cb71e8bad4676a7c0", "delta": " is" } }
-        },
-        {
-            "timestamp_ms": 1762836527105i64,
-            "event": { "id": "0", "msg": { "type": "agent_message_content_delta", "thread_id": "019a713e-6ce6-7f82-b1e7-359628267934", "turn_id": "0", "item_id": "msg_0c5117240874292f016912c020d658819cb71e8bad4676a7c0", "delta": " composing" } }
-        },
-        {
-            "timestamp_ms": 1762836527121i64,
-            "event": { "id": "0", "msg": { "type": "agent_message_content_delta", "thread_id": "019a713e-6ce6-7f82-b1e7-359628267934", "turn_id": "0", "item_id": "msg_0c5117240874292f016912c020d658819cb71e8bad4676a7c0", "delta": " a" } }
-        },
-        {
-            "timestamp_ms": 1762836527137i64,
-            "event": { "id": "0", "msg": { "type": "agent_message_content_delta", "thread_id": "019a713e-6ce6-7f82-b1e7-359628267934", "turn_id": "0", "item_id": "msg_0c5117240874292f016912c020d658819cb71e8bad4676a7c0", "delta": " longer" } }
-        },
-        {
-            "timestamp_ms": 1762836527148i64,
-            "event": { "id": "0", "msg": { "type": "agent_message_content_delta", "thread_id": "019a713e-6ce6-7f82-b1e7-359628267934", "turn_id": "0", "item_id": "msg_0c5117240874292f016912c020d658819cb71e8bad4676a7c0", "delta": " answer" } }
-        }
-    ]);
-    let logs: Vec<LogEntry> =
-        serde_json::from_value(events_json).expect("valid streaming_deltas JSON");
-    let session = ConversationId::from_string("019a713e-6ce4-73e0-bf9b-e070890e3790")
-        .expect("valid session id");
-    let earliest_ms = logs.first().map(|e| e.timestamp_ms);
-    let latest_ms = logs.last().map(|e| e.timestamp_ms);
-    let returned = logs.len();
-    let total = logs.len();
-    let more = false;
-
-    let rendered = render_logs_as_text(
-        session,
+    let logs = fixtures::streaming_deltas();
+    let rendered = render_full(
+        &session("019a713e-6ce4-73e0-bf9b-e070890e3790"),
        &logs,
-        earliest_ms,
-        latest_ms,
-        returned,
-        total,
-        more,
+        logs.len(),
+        false,
    );

    assert_snapshot!(
@@ -369,65 +399,12 @@ fn subagent_logs_snapshot_streaming_deltas() {
 #[test]
 fn subagent_logs_snapshot_reasoning_stream() {
    // Synthetic example of mid-reasoning without a summary yet.
-    let events_json = json!([
-        {
-            "timestamp_ms": 1_000i64,
-            "event": {
-                "id": "0",
-                "msg": {
-                    "type": "reasoning_content_delta",
-                    "thread_id": "thread-1",
-                    "turn_id": "0",
-                    "item_id": "rs_test",
-                    "delta": " thinking"
-                }
-            }
-        },
-        {
-            "timestamp_ms": 1_050i64,
-            "event": {
-                "id": "0",
-                "msg": {
-                    "type": "reasoning_content_delta",
-                    "thread_id": "thread-1",
-                    "turn_id": "0",
-                    "item_id": "rs_test",
-                    "delta": " about"
-                }
-            }
-        },
-        {
-            "timestamp_ms": 1_100i64,
-            "event": {
-                "id": "0",
-                "msg": {
-                    "type": "reasoning_content_delta",
-                    "thread_id": "thread-1",
-                    "turn_id": "0",
-                    "item_id": "rs_test",
-                    "delta": " streaming state"
-                }
-            }
-        }
-    ]);
-    let logs: Vec<LogEntry> =
-        serde_json::from_value(events_json).expect("valid reasoning_stream JSON");
-    let session = ConversationId::from_string("019a713e-eeee-73e0-bf9b-e070890e3790")
-        .expect("valid session id");
-    let earliest_ms = logs.first().map(|e| e.timestamp_ms);
-    let latest_ms = logs.last().map(|e| e.timestamp_ms);
-    let returned = logs.len();
-    let total = logs.len();
-    let more = false;
-
-    let rendered = render_logs_as_text(
-        session,
+    let logs = fixtures::reasoning_stream();
+    let rendered = render_full(
+        &session("019a713e-eeee-73e0-bf9b-e070890e3790"),
        &logs,
-        earliest_ms,
-        latest_ms,
-        returned,
-        total,
-        more,
+        logs.len(),
+        false,
    );

    assert_snapshot!(
@@ -440,32 +417,12 @@ fn subagent_logs_snapshot_reasoning_stream() {
 #[test]
 fn subagent_logs_snapshot_no_older_history() {
    // Minimal case: single assistant message, no older history, at latest.
-    let events_json = json!([
-        {
-            "timestamp_ms": 1_000i64,
-            "event": {
-                "id": "0",
-                "msg": { "type": "agent_message", "message": "only event" }
-            }
-        }
-    ]);
-    let logs: Vec<LogEntry> = serde_json::from_value(events_json).expect("valid single-event JSON");
-    let session = ConversationId::from_string("019a9999-aaaa-bbbb-cccc-ddddeeeeffff")
-        .expect("valid session id");
-    let earliest_ms = logs.first().map(|e| e.timestamp_ms);
-    let latest_ms = logs.last().map(|e| e.timestamp_ms);
-    let returned = logs.len();
-    let total = logs.len();
-    let more = false;
-
-    let rendered = render_logs_as_text(
-        session,
+    let logs = fixtures::single_message();
+    let rendered = render_full(
+        &session("019a9999-aaaa-bbbb-cccc-ddddeeeeffff"),
        &logs,
-        earliest_ms,
-        latest_ms,
-        returned,
-        total,
-        more,
+        logs.len(),
+        false,
    );

    assert_snapshot!(
@@ -2,9 +2,6 @@
 #![allow(clippy::unwrap_used, clippy::expect_used)]

 use anyhow::Result;
-use codex_core::protocol::EventMsg;
-use codex_core::protocol::ExecCommandBeginEvent;
-use codex_core::protocol::ExecCommandEndEvent;
 use codex_core::protocol::Op;
 use codex_core::protocol::SandboxPolicy;
 use codex_protocol::config_types::ReasoningSummary;
@@ -16,28 +13,10 @@ use core_test_support::responses::ev_response_created;
 use core_test_support::responses::sse;
 use core_test_support::responses::start_mock_server;
 use core_test_support::test_codex::test_codex;
-use core_test_support::wait_for_event_with_timeout;
+use core_test_support::wait_for_exec_command_pair;
 use serde_json::json;
-use tokio::time::Duration;
-
-fn is_exec_begin(ev: &EventMsg) -> Option<ExecCommandBeginEvent> {
-    if let EventMsg::ExecCommandBegin(ev) = ev {
-        Some(ev.clone())
-    } else {
-        None
-    }
-}
-
-fn is_exec_end(ev: &EventMsg) -> Option<ExecCommandEndEvent> {
-    if let EventMsg::ExecCommandEnd(ev) = ev {
-        Some(ev.clone())
-    } else {
-        None
-    }
-}

 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-#[ignore = "relies on streaming timing; kept for manual verification"]
 async fn root_subagent_tool_emits_exec_events() -> Result<()> {
    let server = start_mock_server().await;
    let mut builder = test_codex();
@@ -82,27 +61,12 @@ async fn root_subagent_tool_emits_exec_events() -> Result<()> {
        })
        .await?;

-    let mut begin: Option<ExecCommandBeginEvent> = None;
-    let mut end: Option<ExecCommandEndEvent> = None;
-    for _ in 0..40 {
-        let ev = wait_for_event_with_timeout(&test.codex, |_| true, Duration::from_secs(20)).await;
-        if begin.is_none() {
-            begin = is_exec_begin(&ev);
-        }
-        if end.is_none() {
-            end = is_exec_end(&ev);
-        }
-        if matches!(ev, EventMsg::TaskComplete(_)) && begin.is_some() && end.is_some() {
-            break;
-        }
-    }
+    let (begin, end) = wait_for_exec_command_pair(&test.codex, call_id).await;

-    let begin = begin.expect("exec begin");
    assert_eq!(
        begin.command.first().map(String::as_str),
-        Some("subagent_list")
+        Some("Listed subagents"),
    );
-    let end = end.expect("exec end");
    assert_eq!(end.call_id, begin.call_id);
    assert_eq!(end.exit_code, 0);

@@ -808,6 +808,7 @@ mod tests {
            enhanced_keys_supported: false,
            placeholder_text: "Ask Codex to do anything".to_string(),
            disable_paste_burst: false,
+            animations_enabled: true,
        });

        pane.set_task_running(true);
@@ -1040,6 +1040,143 @@ fn exec_history_cell_shows_working_then_completed() {
    );
 }

+struct SummaryBuilder {
+    summary: SubagentSummary,
+}
+
+impl SummaryBuilder {
+    fn new(agent_id: u64, parent_session_id: ConversationId, session_id: ConversationId) -> Self {
+        Self {
+            summary: SubagentSummary {
+                agent_id,
+                parent_agent_id: Some(0),
+                session_id,
+                parent_session_id: Some(parent_session_id),
+                origin: SubagentLifecycleOrigin::Spawn,
+                status: SubagentLifecycleStatus::Running,
+                label: None,
+                summary: None,
+                reasoning_header: None,
+                started_at_ms: 0,
+                pending_messages: 0,
+                pending_interrupts: 0,
+            },
+        }
+    }
+
+    fn origin(mut self, origin: SubagentLifecycleOrigin) -> Self {
+        self.summary.origin = origin;
+        self
+    }
+
+    fn status(mut self, status: SubagentLifecycleStatus) -> Self {
+        self.summary.status = status;
+        self
+    }
+
+    fn label(mut self, label: impl Into<String>) -> Self {
+        self.summary.label = Some(label.into());
+        self
+    }
+
+    fn summary_text(mut self, summary: impl Into<String>) -> Self {
+        self.summary.summary = Some(summary.into());
+        self
+    }
+
+    fn reasoning_header(mut self, header: impl Into<String>) -> Self {
+        self.summary.reasoning_header = Some(header.into());
+        self
+    }
+
+    fn started_at(mut self, started_at_ms: i64) -> Self {
+        self.summary.started_at_ms = started_at_ms;
+        self
+    }
+
+    fn pending(mut self, pending_messages: usize, pending_interrupts: usize) -> Self {
+        self.summary.pending_messages = pending_messages;
+        self.summary.pending_interrupts = pending_interrupts;
+        self
+    }
+
+    fn build(self) -> SubagentSummary {
+        self.summary
+    }
+}
+
+fn dispatch_created(
+    chat: &mut ChatWidget,
+    id: Option<&str>,
+    summary: SubagentSummary,
+    replayed: bool,
+) {
+    chat.dispatch_event_msg(
+        id.map(str::to_string),
+        EventMsg::SubagentLifecycle(SubagentLifecycleEvent::Created(SubagentCreatedEvent {
+            subagent: summary,
+        })),
+        replayed,
+    );
+}
+
+fn dispatch_reasoning_header(
+    chat: &mut ChatWidget,
+    id: Option<&str>,
+    agent_id: u64,
+    session_id: ConversationId,
+    header: impl Into<String>,
+    replayed: bool,
+) {
+    chat.dispatch_event_msg(
+        id.map(str::to_string),
+        EventMsg::SubagentLifecycle(SubagentLifecycleEvent::ReasoningHeader(
+            SubagentReasoningHeaderEvent {
+                agent_id,
+                session_id,
+                reasoning_header: header.into(),
+            },
+        )),
+        replayed,
+    );
+}
+
+fn dispatch_status(
+    chat: &mut ChatWidget,
+    id: Option<&str>,
+    agent_id: u64,
+    session_id: ConversationId,
+    status: SubagentLifecycleStatus,
+    replayed: bool,
+) {
+    chat.dispatch_event_msg(
+        id.map(str::to_string),
+        EventMsg::SubagentLifecycle(SubagentLifecycleEvent::Status(SubagentStatusEvent {
+            agent_id,
+            session_id,
+            status,
+        })),
+        replayed,
+    );
+}
+
+fn dispatch_deleted(
+    chat: &mut ChatWidget,
+    id: Option<&str>,
+    agent_id: u64,
+    session_id: ConversationId,
+    replayed: bool,
+) {
+    chat.dispatch_event_msg(
+        id.map(str::to_string),
+        EventMsg::SubagentLifecycle(SubagentLifecycleEvent::Deleted(SubagentRemovedEvent {
+            agent_id,
+            session_id,
+        })),
+        replayed,
+    );
+}
+
 #[test]
 fn subagent_lifecycle_updates_bottom_pane() {
    let (mut chat, mut _rx, _ops) = make_chatwidget_manual();
@@ -1048,65 +1185,43 @@ fn subagent_lifecycle_updates_bottom_pane() {
    chat.on_task_started();

    let child_id = ConversationId::new();
-    let summary = SubagentSummary {
-        agent_id: 1,
-        parent_agent_id: Some(0),
-        session_id: child_id,
-        parent_session_id: Some(parent_id),
-        origin: SubagentLifecycleOrigin::Fork,
-        status: SubagentLifecycleStatus::Running,
-        label: Some("Docs sweep".to_string()),
-        summary: Some("Rewrite md files".to_string()),
-        reasoning_header: None,
-        started_at_ms: 42,
-        pending_messages: 0,
-        pending_interrupts: 0,
-    };
+    let summary = SummaryBuilder::new(1, parent_id, child_id)
+        .origin(SubagentLifecycleOrigin::Fork)
+        .status(SubagentLifecycleStatus::Running)
+        .label("Docs sweep")
+        .summary_text("Rewrite md files")
+        .started_at(42)
+        .build();

-    chat.dispatch_event_msg(
-        Some("created".into()),
-        EventMsg::SubagentLifecycle(SubagentLifecycleEvent::Created(SubagentCreatedEvent {
-            subagent: summary,
-        })),
-        false,
-    );
+    dispatch_created(&mut chat, Some("created"), summary, false);

    assert_eq!(chat.bottom_pane.subagent_count(), 1);
    assert_eq!(chat.bottom_pane.subagent_entries().len(), 2);

-    chat.dispatch_event_msg(
-        Some("header".into()),
-        EventMsg::SubagentLifecycle(SubagentLifecycleEvent::ReasoningHeader(
-            SubagentReasoningHeaderEvent {
-                agent_id: 1,
-                session_id: child_id,
-                reasoning_header: "Scanning tests".to_string(),
-            },
-        )),
+    dispatch_reasoning_header(
+        &mut chat,
+        Some("header"),
+        1,
+        child_id,
+        "Scanning tests",
        false,
    );
    let entries = chat.bottom_pane.subagent_entries();
    assert_eq!(entries[1].detail.as_deref(), Some("Scanning tests"));

-    chat.dispatch_event_msg(
-        Some("status".into()),
-        EventMsg::SubagentLifecycle(SubagentLifecycleEvent::Status(SubagentStatusEvent {
-            agent_id: 1,
-            session_id: child_id,
-            status: SubagentLifecycleStatus::Idle,
-        })),
+    dispatch_status(
+        &mut chat,
+        Some("status"),
+        1,
+        child_id,
+        SubagentLifecycleStatus::Idle,
        false,
    );
    assert_eq!(chat.bottom_pane.subagent_count(), 0);
+    let entries = chat.bottom_pane.subagent_entries();
+    assert_eq!(entries[1].detail.as_deref(), Some("Scanning tests"));

-    chat.dispatch_event_msg(
-        Some("deleted".into()),
-        EventMsg::SubagentLifecycle(SubagentLifecycleEvent::Deleted(SubagentRemovedEvent {
-            agent_id: 1,
-            session_id: child_id,
-        })),
-        false,
-    );
+    dispatch_deleted(&mut chat, Some("deleted"), 1, child_id, false);
    assert!(chat.bottom_pane.subagent_entries().is_empty());
 }

@@ -1117,28 +1232,13 @@ fn replayed_subagent_events_are_ignored() {
    chat.conversation_id = Some(parent_id);

    let child_id = ConversationId::new();
-    let summary = SubagentSummary {
-        agent_id: 1,
-        parent_agent_id: Some(0),
-        session_id: child_id,
-        parent_session_id: Some(parent_id),
-        origin: SubagentLifecycleOrigin::Spawn,
-        status: SubagentLifecycleStatus::Running,
-        label: None,
-        summary: None,
-        reasoning_header: None,
-        started_at_ms: 100,
-        pending_messages: 0,
-        pending_interrupts: 0,
-    };
+    let summary = SummaryBuilder::new(1, parent_id, child_id)
+        .origin(SubagentLifecycleOrigin::Spawn)
+        .status(SubagentLifecycleStatus::Running)
+        .started_at(100)
+        .build();

-    chat.dispatch_event_msg(
-        None,
-        EventMsg::SubagentLifecycle(SubagentLifecycleEvent::Created(SubagentCreatedEvent {
-            subagent: summary,
-        })),
-        true,
-    );
+    dispatch_created(&mut chat, None, summary, true);

    assert!(chat.bottom_pane.subagent_entries().is_empty());
    assert_eq!(chat.bottom_pane.subagent_count(), 0);
@@ -1151,28 +1251,14 @@ fn agent_inbox_updates_pending_counts() {
    chat.conversation_id = Some(parent_id);

    let child_id = ConversationId::new();
-    let summary = SubagentSummary {
-        agent_id: 1,
-        parent_agent_id: Some(0),
-        session_id: child_id,
-        parent_session_id: Some(parent_id),
-        origin: SubagentLifecycleOrigin::Spawn,
-        status: SubagentLifecycleStatus::Running,
-        label: Some("Docs sweep".to_string()),
-        summary: None,
-        reasoning_header: None,
-        started_at_ms: 100,
-        pending_messages: 0,
-        pending_interrupts: 0,
-    };
+    let summary = SummaryBuilder::new(1, parent_id, child_id)
+        .origin(SubagentLifecycleOrigin::Spawn)
+        .status(SubagentLifecycleStatus::Running)
+        .label("Docs sweep")
+        .started_at(100)
+        .build();

-    chat.dispatch_event_msg(
-        None,
-        EventMsg::SubagentLifecycle(SubagentLifecycleEvent::Created(SubagentCreatedEvent {
-            subagent: summary,
-        })),
-        false,
-    );
+    dispatch_created(&mut chat, None, summary, false);

    chat.dispatch_event_msg(
        None,