From d560aff1aae329d24b4f5b810db213b9fda0c612 Mon Sep 17 00:00:00 2001 From: Friel Date: Sat, 22 Nov 2025 09:26:13 -0800 Subject: [PATCH] Add subagent tests and helpers; watchdog fixes --- codex-rs/core/src/shell.rs | 3 +- codex-rs/core/src/subagents/manager.rs | 465 ++++++++++++- codex-rs/core/src/subagents/mod.rs | 1 + codex-rs/core/src/tools/handlers/subagent.rs | 7 +- codex-rs/core/tests/common/lib.rs | 61 ++ codex-rs/core/tests/responses_headers.rs | 130 ++-- .../core/tests/subagent_logs_snapshots.rs | 621 ++++++++---------- .../core/tests/suite/subagent_exec_events.rs | 42 +- codex-rs/tui/src/bottom_pane/mod.rs | 1 + codex-rs/tui/src/chatwidget/tests.rs | 260 +++++--- 10 files changed, 1037 insertions(+), 554 deletions(-) diff --git a/codex-rs/core/src/shell.rs b/codex-rs/core/src/shell.rs index ac115facb6..325c0dcc9c 100644 --- a/codex-rs/core/src/shell.rs +++ b/codex-rs/core/src/shell.rs @@ -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:?}", ); } diff --git a/codex-rs/core/src/subagents/manager.rs b/codex-rs/core/src/subagents/manager.rs index 76d5f9f885..d177bf3d29 100644 --- a/codex-rs/core/src/subagents/manager.rs +++ b/codex-rs/core/src/subagents/manager.rs @@ -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(®istry), 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::>(); + 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(®istry), 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(®istry), 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(®istry), 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(®istry), 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::>(); + 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(®istry), 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(®istry), 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(®istry), 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()); + } } diff --git a/codex-rs/core/src/subagents/mod.rs b/codex-rs/core/src/subagents/mod.rs index 29b8b2d20a..51f7e541a4 100644 --- a/codex-rs/core/src/subagents/mod.rs +++ b/codex-rs/core/src/subagents/mod.rs @@ -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; diff --git a/codex-rs/core/src/tools/handlers/subagent.rs b/codex-rs/core/src/tools/handlers/subagent.rs index 8f55754ac5..24f26e5b26 100644 --- a/codex-rs/core/src/tools/handlers/subagent.rs +++ b/codex-rs/core/src/tools/handlers/subagent.rs @@ -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, diff --git a/codex-rs/core/tests/common/lib.rs b/codex-rs/core/tests/common/lib.rs index e7b1e71efa..6026094630 100644 --- a/codex-rs/core/tests/common/lib.rs +++ b/codex-rs/core/tests/common/lib.rs @@ -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 = None; + let mut end: Option = 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( codex: &CodexConversation, mut predicate: F, diff --git a/codex-rs/core/tests/responses_headers.rs b/codex-rs/core/tests/responses_headers.rs index 7b6f645f2f..6787bb0a2e 100644 --- a/codex-rs/core/tests/responses_headers.rs +++ b/codex-rs/core/tests/responses_headers.rs @@ -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 { + 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; } diff --git a/codex-rs/core/tests/subagent_logs_snapshots.rs b/codex-rs/core/tests/subagent_logs_snapshots.rs index e111f62ca0..35484a2381 100644 --- a/codex-rs/core/tests/subagent_logs_snapshots.rs +++ b/codex-rs/core/tests/subagent_logs_snapshots.rs @@ -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 { - 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 = - 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 = - 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 = - 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 = 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!( diff --git a/codex-rs/core/tests/suite/subagent_exec_events.rs b/codex-rs/core/tests/suite/subagent_exec_events.rs index 2edcbe212a..fa3f25baae 100644 --- a/codex-rs/core/tests/suite/subagent_exec_events.rs +++ b/codex-rs/core/tests/suite/subagent_exec_events.rs @@ -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 { - if let EventMsg::ExecCommandBegin(ev) = ev { - Some(ev.clone()) - } else { - None - } -} - -fn is_exec_end(ev: &EventMsg) -> Option { - 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 = None; - let mut end: Option = 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); diff --git a/codex-rs/tui/src/bottom_pane/mod.rs b/codex-rs/tui/src/bottom_pane/mod.rs index 606ae18271..d8b894f4fb 100644 --- a/codex-rs/tui/src/bottom_pane/mod.rs +++ b/codex-rs/tui/src/bottom_pane/mod.rs @@ -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); diff --git a/codex-rs/tui/src/chatwidget/tests.rs b/codex-rs/tui/src/chatwidget/tests.rs index 7a271d7fb2..09c3054156 100644 --- a/codex-rs/tui/src/chatwidget/tests.rs +++ b/codex-rs/tui/src/chatwidget/tests.rs @@ -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) -> Self { + self.summary.label = Some(label.into()); + self + } + + fn summary_text(mut self, summary: impl Into) -> Self { + self.summary.summary = Some(summary.into()); + self + } + + fn reasoning_header(mut self, header: impl Into) -> 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, + 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,