From de9d7494b36dd66ed43b083f82791b2bcf3bd493 Mon Sep 17 00:00:00 2001 From: Andrei Hasna Date: Sat, 8 Aug 2026 22:54:14 +0300 Subject: [PATCH 1/2] test(core): cover spawn payload and profile inheritance Agent: Theophrastus --- .../src/tools/handlers/multi_agents_tests.rs | 140 ++++++++++++++++++ .../tests/suite/subagent_notifications.rs | 89 +++++++++++ 2 files changed, 229 insertions(+) diff --git a/codex-rs/core/src/tools/handlers/multi_agents_tests.rs b/codex-rs/core/src/tools/handlers/multi_agents_tests.rs index f7310142d..1395a520b 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_tests.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_tests.rs @@ -1354,6 +1354,146 @@ async fn multi_agent_v2_spawn_partial_fork_turns_allows_agent_type_override() { assert_eq!(snapshot.reasoning_effort, Some(ReasoningEffort::Minimal)); } +#[tokio::test] +async fn multi_agent_v2_spawn_inherits_live_session_auth_profile_for_every_fork_mode() { + for fork_turns in ["none", "1", "all"] { + let (mut session, mut turn) = make_session_and_context().await; + save_test_auth_profile(&turn, "project-default"); + save_test_auth_profile(&turn, "healthy-selected"); + let mut config = (*turn.config).clone(); + config.selected_auth_profile = Some("project-default".to_string()); + config + .features + .enable(Feature::MultiAgentV2) + .expect("test config should allow feature update"); + set_turn_config(&mut turn, config); + session + .update_settings(crate::session::SessionSettingsUpdate { + auth_profile: Some(Some("healthy-selected".to_string())), + ..Default::default() + }) + .await + .expect("live session auth profile should update"); + assert_eq!( + session.selected_auth_profile().await.as_deref(), + Some("healthy-selected") + ); + + let manager = thread_manager(); + let root = manager + .start_thread((*turn.config).clone()) + .await + .expect("root thread should start"); + session.services.agent_control = manager.agent_control(); + session.thread_id = root.thread_id; + let session = Arc::new(session); + let turn = Arc::new(turn); + let task_name = format!("profile_{fork_turns}"); + + let output = SpawnAgentHandlerV2::default() + .handle(invocation( + session.clone(), + turn.clone(), + "spawn_agent", + function_payload(json!({ + "message": "inspect this repo", + "task_name": task_name, + "fork_turns": fork_turns, + })), + )) + .await + .expect("spawn should inherit the live session profile"); + let (content, _) = expect_text_output(output); + let result: serde_json::Value = + serde_json::from_str(&content).expect("spawn_agent result should be json"); + let child_thread_id = session + .services + .agent_control + .resolve_agent_reference( + session.thread_id, + &turn.session_source, + result["task_name"] + .as_str() + .expect("spawned task name should be present"), + ) + .await + .expect("spawned task name should resolve"); + let snapshot = manager + .get_thread(child_thread_id) + .await + .expect("spawned agent thread should exist") + .config_snapshot() + .await; + + assert_eq!( + snapshot.selected_auth_profile.as_deref(), + Some("healthy-selected"), + "fork_turns={fork_turns} must use the live parent session profile" + ); + } +} + +#[tokio::test] +async fn multi_agent_v2_spawn_does_not_substitute_stale_turn_profile() { + let (mut session, mut turn) = make_session_and_context().await; + save_test_auth_profile(&turn, "project-default"); + let mut config = (*turn.config).clone(); + config.selected_auth_profile = Some("project-default".to_string()); + config + .features + .enable(Feature::MultiAgentV2) + .expect("test config should allow feature update"); + set_turn_config(&mut turn, config); + assert_eq!(session.selected_auth_profile().await, None); + + let manager = thread_manager(); + let root = manager + .start_thread((*turn.config).clone()) + .await + .expect("root thread should start"); + session.services.agent_control = manager.agent_control(); + session.thread_id = root.thread_id; + let session = Arc::new(session); + let turn = Arc::new(turn); + + let output = SpawnAgentHandlerV2::default() + .handle(invocation( + session.clone(), + turn.clone(), + "spawn_agent", + function_payload(json!({ + "message": "inspect this repo", + "task_name": "no_profile_substitution", + "fork_turns": "all", + })), + )) + .await + .expect("spawn should not substitute the stale turn profile"); + let (content, _) = expect_text_output(output); + let result: serde_json::Value = + serde_json::from_str(&content).expect("spawn_agent result should be json"); + let child_thread_id = session + .services + .agent_control + .resolve_agent_reference( + session.thread_id, + &turn.session_source, + result["task_name"] + .as_str() + .expect("spawned task name should be present"), + ) + .await + .expect("spawned task name should resolve"); + let snapshot = manager + .get_thread(child_thread_id) + .await + .expect("spawned agent thread should exist") + .config_snapshot() + .await; + + assert_eq!(snapshot.selected_auth_profile, None); +} + #[tokio::test] async fn spawn_agent_returns_agent_id_without_task_name() { let (mut session, turn) = make_session_and_context().await; diff --git a/codex-rs/core/tests/suite/subagent_notifications.rs b/codex-rs/core/tests/suite/subagent_notifications.rs index d3e3d4168..fd53de56f 100644 --- a/codex-rs/core/tests/suite/subagent_notifications.rs +++ b/codex-rs/core/tests/suite/subagent_notifications.rs @@ -1329,6 +1329,95 @@ async fn encrypted_multi_agent_v2_spawn_sends_agent_message_to_child() -> Result Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn multi_agent_v2_spawn_preserves_large_initial_task_for_every_fork_mode() -> Result<()> { + for fork_turns in ["none", "1", "all"] { + let server = start_mock_server().await; + let task_name = format!("large_payload_{fork_turns}"); + let task_path = format!("/root/{task_name}"); + let initial_task = format!( + "verbatim-start-{fork_turns}-{}-verbatim-end-{fork_turns}", + "x".repeat(8 * 1024) + ); + let spawn_args = serde_json::to_string(&json!({ + "message": initial_task, + "task_name": task_name, + "fork_turns": fork_turns, + }))?; + mount_sse_once_match( + &server, + |req: &wiremock::Request| body_contains(req, TURN_1_PROMPT), + sse(vec![ + ev_response_created("resp-parent-1"), + ev_function_call(SPAWN_CALL_ID, "spawn_agent", &spawn_args), + ev_completed("resp-parent-1"), + ]), + ) + .await; + let child_task_path = task_path.clone(); + let child_request_log = mount_sse_once_match( + &server, + move |req: &wiremock::Request| { + body_contains(req, child_task_path.as_str()) && !body_contains(req, SPAWN_CALL_ID) + }, + sse(vec![ + ev_response_created("resp-child-1"), + ev_completed("resp-child-1"), + ]), + ) + .await; + mount_sse_once_match( + &server, + |req: &wiremock::Request| body_contains(req, SPAWN_CALL_ID), + sse(vec![ + ev_response_created("resp-parent-2"), + ev_assistant_message("msg-parent-2", "done"), + ev_completed("resp-parent-2"), + ]), + ) + .await; + + let mut builder = test_codex().with_model("koffing").with_config(|config| { + config + .features + .enable(Feature::Collab) + .expect("test config should allow feature update"); + config + .features + .enable(Feature::MultiAgentV2) + .expect("test config should allow feature update"); + }); + let test = builder.build(&server).await?; + + test.submit_turn(TURN_1_PROMPT).await?; + + let child_request = wait_for_matching_request( + &child_request_log, + format!("large initial task for fork_turns={fork_turns}").as_str(), + |request| request.body_contains_text(initial_task.as_str()), + ) + .await?; + assert!(child_request.body_contains_text("agent_message")); + assert!(child_request.body_contains_text("encrypted_content")); + assert_eq!( + child_request + .body_json() + .to_string() + .matches(initial_task.as_str()) + .count(), + 1, + "initial task must reach the first child request exactly once for fork_turns={fork_turns}" + ); + assert_eq!( + child_request_log.requests().len(), + 1, + "spawn must emit one first child request for fork_turns={fork_turns}" + ); + } + + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn skills_toggle_skips_instructions_for_parent_and_spawned_child() -> Result<()> { skip_if_no_network!(Ok(())); From a723e537f68a546d7e02134f5150646c68093940 Mon Sep 17 00:00:00 2001 From: Andrei Hasna Date: Sat, 8 Aug 2026 22:59:52 +0300 Subject: [PATCH 2/2] fix(core): preserve spawned-agent initial task Agent: Theophrastus --- codex-rs/core/src/agent/control.rs | 29 ++++++ codex-rs/core/src/agent/control/spawn.rs | 16 +++- codex-rs/core/src/codex_thread.rs | 14 +++ codex-rs/core/src/session/handlers.rs | 29 ++++++ codex-rs/core/src/session/input_queue.rs | 96 ++++++++++++++++++- codex-rs/core/src/session/mod.rs | 1 + codex-rs/core/src/thread_manager.rs | 31 ++++++ .../src/tools/handlers/multi_agents/spawn.rs | 1 + .../src/tools/handlers/multi_agents_tests.rs | 35 +++++-- .../tools/handlers/multi_agents_v2/spawn.rs | 2 + .../tests/suite/subagent_notifications.rs | 9 +- 11 files changed, 247 insertions(+), 16 deletions(-) diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index 1bd6776f2..f10be4988 100644 --- a/codex-rs/core/src/agent/control.rs +++ b/codex-rs/core/src/agent/control.rs @@ -60,6 +60,7 @@ pub(crate) enum SpawnAgentForkMode { pub(crate) struct SpawnAgentOptions { pub(crate) fork_parent_spawn_call_id: Option, pub(crate) fork_mode: Option, + pub(crate) initial_task_message_id: Option, pub(crate) parent_thread_id: Option, pub(crate) environments: Option>, } @@ -175,6 +176,34 @@ impl AgentControl { result } + pub(crate) async fn send_initial_agent_task( + &self, + agent_id: ThreadId, + message_id: String, + communication: InterAgentCommunication, + ) -> CodexResult { + let last_task_message = last_task_message_from_communication(&communication); + let state = self.upgrade()?; + let result = self + .handle_thread_request_result( + agent_id, + &state, + state + .deliver_initial_agent_task(agent_id, message_id, communication) + .await, + ) + .await; + if result.is_ok() { + match last_task_message { + Some(last_task_message) => self + .state + .update_last_task_message(agent_id, last_task_message), + None => self.state.clear_last_task_message(agent_id), + } + } + result + } + /// Returns the wire protocol of `agent_id`'s configured provider, or `None` /// when the thread is not currently resolvable. Callers use this to encode an /// inter-agent message in a form the receiver's wire can actually deliver. diff --git a/codex-rs/core/src/agent/control/spawn.rs b/codex-rs/core/src/agent/control/spawn.rs index bb563de6c..fd31c255c 100644 --- a/codex-rs/core/src/agent/control/spawn.rs +++ b/codex-rs/core/src/agent/control/spawn.rs @@ -346,8 +346,20 @@ impl AgentControl { ) .await; - self.send_input(new_thread.thread_id, initial_operation) - .await?; + match (options.initial_task_message_id, initial_operation) { + (Some(initial_task_message_id), Op::InterAgentCommunication { communication }) => { + self.send_initial_agent_task( + new_thread.thread_id, + initial_task_message_id, + communication, + ) + .await?; + } + (_, initial_operation) => { + self.send_input(new_thread.thread_id, initial_operation) + .await?; + } + } if multi_agent_version != MultiAgentVersion::V2 { let child_reference = agent_metadata .agent_path diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index e12006ff9..f3941798e 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -273,6 +273,20 @@ impl CodexThread { Ok(()) } + /// Queues a spawned agent's initial task with a stable id for exact-once model delivery. + pub async fn enqueue_initial_agent_task_with_id( + &self, + message_id: String, + communication: InterAgentCommunication, + ) -> CodexResult<()> { + if !self.is_running() { + return Err(CodexErr::InternalAgentDied); + } + crate::session::enqueue_initial_agent_task(&self.codex.session, message_id, communication) + .await?; + Ok(()) + } + pub async fn queued_mailbox_messages(&self) -> Vec { self.codex.session.input_queue.list_mailbox_messages().await } diff --git a/codex-rs/core/src/session/handlers.rs b/codex-rs/core/src/session/handlers.rs index 7272fa3cf..289b59ea7 100644 --- a/codex-rs/core/src/session/handlers.rs +++ b/codex-rs/core/src/session/handlers.rs @@ -463,6 +463,35 @@ pub async fn enqueue_inter_agent_communication( Ok(()) } +/// Records a spawned agent's initial task under a stable spawn-call id without synchronously +/// waking pending work. +pub async fn enqueue_initial_agent_task( + sess: &Arc, + message_id: String, + communication: InterAgentCommunication, +) -> CodexResult<()> { + let deferred_delivery = if communication.trigger_turn { + sess.input_queue + .defer_mailbox_delivery_for_active_turn(&sess.active_turn) + .await + } else { + None + }; + if let Err(err) = sess + .input_queue + .enqueue_initial_task_with_id(message_id, communication) + .await + { + if let Some((turn_state, previous_phase)) = deferred_delivery { + sess.input_queue + .restore_mailbox_delivery_phase(turn_state.as_ref(), previous_phase) + .await; + } + return Err(CodexErr::InvalidRequest(err.to_string())); + } + Ok(()) +} + pub async fn run_user_shell_command(sess: &Arc, sub_id: String, command: String) { if let Some((turn_context, cancellation_token)) = sess.active_turn_context_and_cancellation_token().await diff --git a/codex-rs/core/src/session/input_queue.rs b/codex-rs/core/src/session/input_queue.rs index 2b5814230..3315a8c5e 100644 --- a/codex-rs/core/src/session/input_queue.rs +++ b/codex-rs/core/src/session/input_queue.rs @@ -7,6 +7,7 @@ use crate::state::TurnState; use codex_protocol::models::ResponseItem; use codex_protocol::protocol::InterAgentCommunication; use codex_protocol::user_input::UserInput; +use std::collections::HashSet; use std::collections::VecDeque; use std::error::Error; use std::fmt; @@ -91,6 +92,7 @@ impl Error for MailboxQueueFull {} pub(crate) struct InputQueue { mailbox_tx: watch::Sender<()>, mailbox_pending_mails: Mutex>, + initial_task_delivery_ids: Mutex>, } impl InputQueue { @@ -99,6 +101,7 @@ impl InputQueue { Self { mailbox_tx, mailbox_pending_mails: Mutex::new(VecDeque::new()), + initial_task_delivery_ids: Mutex::new(HashSet::new()), } } @@ -134,6 +137,36 @@ impl InputQueue { Ok(()) } + /// Enqueue the spawned agent's initial task exactly once under a stable spawn-call id. + /// + /// The id remains recorded after the task drains so an acknowledgement retry cannot inject the + /// task a second time into a later turn. This path is used only for the one initial task of a + /// newly spawned child; retries reuse the same id, so the retained set normally has one entry + /// and cannot grow with arbitrary mailbox traffic. + pub(crate) async fn enqueue_initial_task_with_id( + &self, + id: String, + communication: InterAgentCommunication, + ) -> Result<(), MailboxQueueFull> { + let mut mails = self.mailbox_pending_mails.lock().await; + let mut initial_task_delivery_ids = self.initial_task_delivery_ids.lock().await; + if initial_task_delivery_ids.contains(&id) { + return Ok(()); + } + if mails.len() >= MAX_MAILBOX_CONTEXT_QUEUE_ITEMS { + return Err(MailboxQueueFull::new(MAX_MAILBOX_CONTEXT_QUEUE_ITEMS)); + } + mails.push_back(QueuedMailboxMessage { + id: id.clone(), + communication, + }); + initial_task_delivery_ids.insert(id); + drop(initial_task_delivery_ids); + drop(mails); + self.mailbox_tx.send_replace(()); + Ok(()) + } + pub(crate) async fn has_pending_mailbox_items(&self) -> bool { !self.mailbox_pending_mails.lock().await.is_empty() } @@ -200,16 +233,22 @@ impl InputQueue { pub(crate) async fn drain_mailbox_input_items(&self) -> Vec { let mut mails = self.mailbox_pending_mails.lock().await; + let initial_task_delivery_ids = self.initial_task_delivery_ids.lock().await; let drain_count = mails.len().min(MAX_MAILBOX_CONTEXT_QUEUE_ITEMS); let items = mails .drain(..drain_count) .map(|mail| { - ResponseItem::from( - MailboxContextFragment::new(mail.communication).into_response_input_item(), - ) + if initial_task_delivery_ids.contains(&mail.id) { + mail.communication.to_model_input_item() + } else { + ResponseItem::from( + MailboxContextFragment::new(mail.communication).into_response_input_item(), + ) + } }) .collect(); let has_more = !mails.is_empty(); + drop(initial_task_delivery_ids); drop(mails); if has_more { self.mailbox_tx.send_replace(()); @@ -494,6 +533,57 @@ mod tests { assert!(!input_queue.has_pending_mailbox_items().await); } + #[tokio::test] + async fn input_queue_delivers_large_initial_task_exactly_once() { + let input_queue = InputQueue::new(); + let initial_task = "initial task ".repeat(2_000); + let communication = InterAgentCommunication::new_encrypted( + AgentPath::root(), + AgentPath::try_from("/root/worker").expect("agent path"), + Vec::new(), + initial_task.clone(), + /*trigger_turn*/ true, + ); + + for _ in 0..2 { + input_queue + .enqueue_initial_task_with_id("spawn-call-1".to_string(), communication.clone()) + .await + .expect("initial task should enqueue idempotently"); + } + + let items = input_queue.drain_mailbox_input_items().await; + assert_eq!(items.len(), 1); + let ResponseItem::AgentMessage { + author, + recipient, + content, + } = &items[0] + else { + panic!("encrypted initial task should retain the native agent message shape"); + }; + assert_eq!(author, "/root"); + assert_eq!(recipient, "/root/worker"); + let [ + codex_protocol::models::AgentMessageInputContent::EncryptedContent { + encrypted_content, + }, + ] = content.as_slice() + else { + panic!("initial task should retain one encrypted content item"); + }; + assert_eq!(encrypted_content, &initial_task); + + input_queue + .enqueue_initial_task_with_id("spawn-call-1".to_string(), communication) + .await + .expect("delivered initial task retry should be an idempotent success"); + assert!( + input_queue.drain_mailbox_input_items().await.is_empty(), + "delivered initial task must not be injected into a later turn" + ); + } + #[tokio::test] async fn input_queue_rejects_mailbox_when_context_queue_is_full() { let input_queue = InputQueue::new(); diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index ca47b23fc..bb97cc818 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -205,6 +205,7 @@ use codex_protocol::exec_output::StreamOutput; mod auth_profile_auto_switch; mod config_lock; mod handlers; +pub(crate) use handlers::enqueue_initial_agent_task; pub(crate) use handlers::enqueue_inter_agent_communication; pub(crate) use handlers::inter_agent_communication as handle_inter_agent_communication; mod inject; diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index 56d441515..b840766e8 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -1190,6 +1190,37 @@ impl ThreadManagerState { Ok(message_id) } + /// Delivers a spawned agent's initial task under the stable parent tool-call id. + /// + /// The input queue retains that id after delivery so a retry acknowledges success without + /// injecting a duplicate task into a later child turn. + pub(crate) async fn deliver_initial_agent_task( + &self, + thread_id: ThreadId, + message_id: String, + communication: InterAgentCommunication, + ) -> CodexResult { + let thread = self.get_thread(thread_id).await?; + let trigger_turn = communication.trigger_turn; + if let Some(ops_log) = &self.ops_log + && let Ok(mut log) = ops_log.lock() + { + log.push(( + thread_id, + Op::InterAgentCommunication { + communication: communication.clone(), + }, + )); + } + thread + .enqueue_initial_agent_task_with_id(message_id.clone(), communication) + .await?; + if trigger_turn { + let _ = thread.submit(Op::WakePendingWork).await; + } + Ok(message_id) + } + /// Remove a thread from the manager by ID, returning it when present. pub(crate) async fn remove_thread(&self, thread_id: &ThreadId) -> Option> { self.threads.write().await.remove(thread_id) diff --git a/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs b/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs index ecd531657..d79ad3ae8 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs @@ -142,6 +142,7 @@ async fn handle_spawn_agent( SpawnAgentOptions { fork_parent_spawn_call_id: args.fork_context.then(|| call_id.clone()), fork_mode: args.fork_context.then_some(SpawnAgentForkMode::FullHistory), + initial_task_message_id: None, parent_thread_id: Some(session.thread_id), environments: Some(turn.environments.to_selections()), }, diff --git a/codex-rs/core/src/tools/handlers/multi_agents_tests.rs b/codex-rs/core/src/tools/handlers/multi_agents_tests.rs index 1395a520b..068204a4f 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_tests.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_tests.rs @@ -1418,18 +1418,28 @@ async fn multi_agent_v2_spawn_inherits_live_session_auth_profile_for_every_fork_ ) .await .expect("spawned task name should resolve"); - let snapshot = manager + let child_thread = manager .get_thread(child_thread_id) .await - .expect("spawned agent thread should exist") - .config_snapshot() - .await; + .expect("spawned agent thread should exist"); + let snapshot = child_thread.config_snapshot().await; assert_eq!( snapshot.selected_auth_profile.as_deref(), Some("healthy-selected"), "fork_turns={fork_turns} must use the live parent session profile" ); + assert_eq!( + child_thread + .codex + .session + .services + .auth_manager + .selected_auth_profile() + .as_deref(), + Some("healthy-selected"), + "fork_turns={fork_turns} must bind the child auth manager to the live parent profile" + ); } } @@ -1484,14 +1494,23 @@ async fn multi_agent_v2_spawn_does_not_substitute_stale_turn_profile() { ) .await .expect("spawned task name should resolve"); - let snapshot = manager + let child_thread = manager .get_thread(child_thread_id) .await - .expect("spawned agent thread should exist") - .config_snapshot() - .await; + .expect("spawned agent thread should exist"); + let snapshot = child_thread.config_snapshot().await; assert_eq!(snapshot.selected_auth_profile, None); + assert_eq!( + child_thread + .codex + .session + .services + .auth_manager + .selected_auth_profile(), + None, + "child auth manager must not invent a profile that the parent session did not select" + ); } #[tokio::test] diff --git a/codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs b/codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs index f2799407e..2c8335d60 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs @@ -81,6 +81,7 @@ async fn handle_spawn_agent( .await; let mut config = build_agent_spawn_config(&session.get_base_instructions().await, turn.as_ref())?; + config.selected_auth_profile = session.selected_auth_profile().await; if let Some(service_tier) = args.service_tier.as_ref() { config.service_tier = Some(service_tier.clone()); } @@ -158,6 +159,7 @@ async fn handle_spawn_agent( SpawnAgentOptions { fork_parent_spawn_call_id: fork_mode.as_ref().map(|_| call_id.clone()), fork_mode, + initial_task_message_id: Some(call_id.clone()), parent_thread_id: Some(session.thread_id), environments: Some(turn.environments.to_selections()), }, diff --git a/codex-rs/core/tests/suite/subagent_notifications.rs b/codex-rs/core/tests/suite/subagent_notifications.rs index fd53de56f..10d34a653 100644 --- a/codex-rs/core/tests/suite/subagent_notifications.rs +++ b/codex-rs/core/tests/suite/subagent_notifications.rs @@ -1358,7 +1358,9 @@ async fn multi_agent_v2_spawn_preserves_large_initial_task_for_every_fork_mode() let child_request_log = mount_sse_once_match( &server, move |req: &wiremock::Request| { - body_contains(req, child_task_path.as_str()) && !body_contains(req, SPAWN_CALL_ID) + body_contains(req, child_task_path.as_str()) + && body_contains(req, "\"type\":\"agent_message\"") + && !body_contains(req, SPAWN_CALL_ID) }, sse(vec![ ev_response_created("resp-child-1"), @@ -1391,13 +1393,14 @@ async fn multi_agent_v2_spawn_preserves_large_initial_task_for_every_fork_mode() test.submit_turn(TURN_1_PROMPT).await?; + let request_description = format!("large initial task for fork_turns={fork_turns}"); let child_request = wait_for_matching_request( &child_request_log, - format!("large initial task for fork_turns={fork_turns}").as_str(), + request_description.as_str(), |request| request.body_contains_text(initial_task.as_str()), ) .await?; - assert!(child_request.body_contains_text("agent_message")); + assert!(child_request.body_contains_text("\"type\":\"agent_message\"")); assert!(child_request.body_contains_text("encrypted_content")); assert_eq!( child_request