diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index 8de64434029b..85e54e2fb9b7 100644 --- a/codex-rs/core/src/agent/control.rs +++ b/codex-rs/core/src/agent/control.rs @@ -4,6 +4,8 @@ use crate::agent::registry::AgentRegistry; use crate::agent::role::DEFAULT_ROLE_NAME; use crate::agent::role::resolve_role_config; use crate::agent::status::is_final; +use crate::agent_communication::AgentCommunicationContext; +use crate::agent_communication::AgentCommunicationKind; use crate::codex_thread::ThreadConfigSnapshot; use crate::config::Config; use crate::config::RolloutBudgetConfig; @@ -139,12 +141,12 @@ impl AgentControl { pub(crate) async fn send_input( &self, agent_id: ThreadId, - initial_operation: Op, + input: Vec, ) -> CodexResult { let state = self.upgrade()?; - self.ensure_execution_capacity_for_op(agent_id, &initial_operation) + self.ensure_execution_capacity_for_turn_start(agent_id, /*starts_turn*/ true) .await?; - self.send_input_after_capacity_check(agent_id, &state, initial_operation) + self.send_input_after_capacity_check(agent_id, &state, input) .await } @@ -152,20 +154,14 @@ impl AgentControl { &self, agent_id: ThreadId, state: &Arc, - initial_operation: Op, + input: Vec, ) -> CodexResult { - if let Op::InterAgentCommunication { communication } = initial_operation { - return self - .submit_inter_agent_communication(agent_id, state, communication) - .await; - } - - let last_task_message = non_empty_task_message(render_input_preview(&initial_operation)); + let last_task_message = non_empty_task_message(render_input_preview(&input)); let result = self .handle_thread_request_result( agent_id, state, - state.send_op(agent_id, initial_operation).await, + state.send_op(agent_id, input.into()).await, ) .await; if result.is_ok() { @@ -183,8 +179,28 @@ impl AgentControl { &self, agent_id: ThreadId, communication: InterAgentCommunication, + agent_communication_context: AgentCommunicationContext, + ) -> CodexResult { + let state = self.upgrade()?; + self.ensure_execution_capacity_for_turn_start(agent_id, communication.trigger_turn) + .await?; + self.send_inter_agent_communication_after_capacity_check( + agent_id, + &state, + communication, + agent_communication_context, + ) + .await + } + + async fn send_inter_agent_communication_after_capacity_check( + &self, + agent_id: ThreadId, + state: &Arc, + communication: InterAgentCommunication, + context: AgentCommunicationContext, ) -> CodexResult { - self.send_input(agent_id, Op::InterAgentCommunication { communication }) + self.submit_inter_agent_communication(agent_id, state, communication, context) .await } @@ -193,8 +209,11 @@ impl AgentControl { agent_id: ThreadId, state: &Arc, communication: InterAgentCommunication, + context: AgentCommunicationContext, ) -> CodexResult { let last_task_message = last_task_message_from_communication(&communication); + let communication_for_log = + crate::agent_communication::logging_enabled().then(|| communication.clone()); let result = self .handle_thread_request_result( agent_id, @@ -204,6 +223,16 @@ impl AgentControl { .await, ) .await; + if let (Some(communication), Ok(communication_id)) = + (communication_for_log, result.as_ref()) + { + crate::agent_communication::emit_agent_communication_send( + communication_id, + &context, + &communication, + agent_id, + ); + } if result.is_ok() { match last_task_message { Some(last_task_message) => self @@ -494,8 +523,10 @@ impl AgentControl { message, /*trigger_turn*/ false, ); + let context = + AgentCommunicationContext::new(AgentCommunicationKind::Result, child_thread_id); let _ = control - .send_inter_agent_communication(parent_thread_id, communication) + .send_inter_agent_communication(parent_thread_id, communication, context) .await; return; } @@ -720,27 +751,23 @@ fn agent_matches_prefix(agent_path: Option<&AgentPath>, prefix: &AgentPath) -> b }) } -pub(crate) fn render_input_preview(initial_operation: &Op) -> String { - match initial_operation { - Op::UserInput { items, .. } => items - .iter() - .map(|item| match item { - UserInput::Text { text, .. } => text.clone(), - UserInput::Image { .. } => "[image]".to_string(), - UserInput::LocalImage { path, .. } => { - format!("[local_image:{}]", path.display()) - } - UserInput::Skill { name, path, .. } => { - format!("[skill:${name}]({})", path.display()) - } - UserInput::Mention { name, path, .. } => format!("[mention:${name}]({path})"), - _ => "[input]".to_string(), - }) - .collect::>() - .join("\n"), - Op::InterAgentCommunication { communication } => communication.content.clone(), - _ => String::new(), - } +pub(crate) fn render_input_preview(input: &[UserInput]) -> String { + input + .iter() + .map(|item| match item { + UserInput::Text { text, .. } => text.clone(), + UserInput::Image { .. } => "[image]".to_string(), + UserInput::LocalImage { path, .. } => { + format!("[local_image:{}]", path.display()) + } + UserInput::Skill { name, path, .. } => { + format!("[skill:${name}]({})", path.display()) + } + UserInput::Mention { name, path, .. } => format!("[mention:${name}]({path})"), + _ => "[input]".to_string(), + }) + .collect::>() + .join("\n") } fn last_task_message_from_communication(communication: &InterAgentCommunication) -> Option { diff --git a/codex-rs/core/src/agent/control/execution.rs b/codex-rs/core/src/agent/control/execution.rs index 42aaa50e029a..d8559c1499cc 100644 --- a/codex-rs/core/src/agent/control/execution.rs +++ b/codex-rs/core/src/agent/control/execution.rs @@ -32,7 +32,16 @@ impl AgentControl { thread_id: ThreadId, op: &Op, ) -> CodexResult<()> { - if !op_starts_turn(op) { + self.ensure_execution_capacity_for_turn_start(thread_id, op_starts_turn(op)) + .await + } + + pub(super) async fn ensure_execution_capacity_for_turn_start( + &self, + thread_id: ThreadId, + starts_turn: bool, + ) -> CodexResult<()> { + if !starts_turn { return Ok(()); } let state = self.upgrade()?; diff --git a/codex-rs/core/src/agent/control/spawn.rs b/codex-rs/core/src/agent/control/spawn.rs index c4c40577ca4f..9078434f1fe8 100644 --- a/codex-rs/core/src/agent/control/spawn.rs +++ b/codex-rs/core/src/agent/control/spawn.rs @@ -9,6 +9,17 @@ struct SpawnAgentThreadInheritance { exec_policy: Option>, } +/// Initial input delivered after a spawned agent acquires execution capacity. +/// +/// V2 communication spawns keep the communication and its context paired so centralized +/// submission and lifecycle logging cannot receive one without the other. Other spawn sources +/// provide user input directly, making an uncontextualized inter-agent communication +/// unrepresentable. +enum SpawnInitialInput { + UserInput(Vec), + InterAgentCommunication(InterAgentCommunication, AgentCommunicationContext), +} + fn default_agent_nickname_list() -> Vec<&'static str> { AGENT_NAMES .lines() @@ -89,12 +100,12 @@ impl AgentControl { pub(crate) async fn spawn_agent( &self, config: Config, - initial_operation: Op, + initial_input: Vec, session_source: Option, ) -> CodexResult { let spawned_agent = Box::pin(self.spawn_agent_internal( config, - initial_operation, + SpawnInitialInput::UserInput(initial_input), session_source, SpawnAgentOptions::default(), )) @@ -106,12 +117,34 @@ impl AgentControl { pub(crate) async fn spawn_agent_with_metadata( &self, config: Config, - initial_operation: Op, + initial_input: Vec, session_source: Option, options: SpawnAgentOptions, // TODO(jif) drop with new fork. ) -> CodexResult { - Box::pin(self.spawn_agent_internal(config, initial_operation, session_source, options)) - .await + Box::pin(self.spawn_agent_internal( + config, + SpawnInitialInput::UserInput(initial_input), + session_source, + options, + )) + .await + } + + pub(crate) async fn spawn_agent_with_communication( + &self, + config: Config, + communication: InterAgentCommunication, + context: AgentCommunicationContext, + session_source: Option, + options: SpawnAgentOptions, + ) -> CodexResult { + Box::pin(self.spawn_agent_internal( + config, + SpawnInitialInput::InterAgentCommunication(communication, context), + session_source, + options, + )) + .await } pub(crate) async fn ensure_v2_agent_loaded( @@ -197,7 +230,7 @@ impl AgentControl { async fn spawn_agent_internal( &self, config: Config, - initial_operation: Op, + initial_input: SpawnInitialInput, session_source: Option, options: SpawnAgentOptions, ) -> CodexResult { @@ -356,8 +389,21 @@ impl AgentControl { ) .await; - self.send_input_after_capacity_check(new_thread.thread_id, &state, initial_operation) - .await?; + match initial_input { + SpawnInitialInput::UserInput(input) => { + self.send_input_after_capacity_check(new_thread.thread_id, &state, input) + .await?; + } + SpawnInitialInput::InterAgentCommunication(communication, context) => { + self.send_inter_agent_communication_after_capacity_check( + new_thread.thread_id, + &state, + communication, + context, + ) + .await?; + } + } if multi_agent_version != MultiAgentVersion::V2 { let child_reference = agent_metadata .agent_path diff --git a/codex-rs/core/src/agent/control_tests.rs b/codex-rs/core/src/agent/control_tests.rs index 9863bfb7e24c..4741c3c101a3 100644 --- a/codex-rs/core/src/agent/control_tests.rs +++ b/codex-rs/core/src/agent/control_tests.rs @@ -3,6 +3,8 @@ use crate::CodexThread; use crate::StateDbHandle; use crate::ThreadManager; use crate::agent::agent_status_from_event; +use crate::agent_communication::AgentCommunicationContext; +use crate::agent_communication::AgentCommunicationKind; use crate::config::AgentRoleConfig; use crate::config::Config; use crate::config::ConfigBuilder; @@ -65,12 +67,11 @@ async fn test_config() -> (TempDir, Config) { test_config_with_cli_overrides(Vec::new()).await } -fn text_input(text: &str) -> Op { +fn text_input(text: &str) -> Vec { vec![UserInput::Text { text: text.to_string(), text_elements: Vec::new(), }] - .into() } fn assistant_message(text: &str, phase: Option) -> ResponseItem { @@ -311,8 +312,7 @@ async fn send_input_errors_when_manager_dropped() { vec![UserInput::Text { text: "hello".to_string(), text_elements: Vec::new(), - }] - .into(), + }], ) .await .expect_err("send_input should fail without a manager"); @@ -423,8 +423,7 @@ async fn send_input_errors_when_thread_missing() { vec![UserInput::Text { text: "hello".to_string(), text_elements: Vec::new(), - }] - .into(), + }], ) .await .expect_err("send_input should fail for missing thread"); @@ -490,8 +489,7 @@ async fn send_input_submits_user_message() { vec![UserInput::Text { text: "hello from tests".to_string(), text_elements: Vec::new(), - }] - .into(), + }], ) .await .expect("send_input should succeed"); @@ -531,7 +529,11 @@ async fn send_inter_agent_communication_without_turn_queues_message_without_trig let submission_id = harness .control - .send_inter_agent_communication(thread_id, communication.clone()) + .send_inter_agent_communication( + thread_id, + communication.clone(), + AgentCommunicationContext::new(AgentCommunicationKind::Message, ThreadId::new()), + ) .await .expect("send_inter_agent_communication should succeed"); assert!(!submission_id.is_empty()); @@ -656,7 +658,11 @@ async fn ensure_v2_agent_loaded_reloads_registered_unloaded_agent() { ); harness .control - .send_inter_agent_communication(spawned_agent.thread_id, communication.clone()) + .send_inter_agent_communication( + spawned_agent.thread_id, + communication.clone(), + AgentCommunicationContext::new(AgentCommunicationKind::Message, ThreadId::new()), + ) .await .expect("send_inter_agent_communication should succeed after reload"); let expected = ( @@ -803,7 +809,11 @@ async fn encrypted_inter_agent_communication_clears_existing_last_task_message() ); harness .control - .send_inter_agent_communication(spawned_agent.thread_id, communication) + .send_inter_agent_communication( + spawned_agent.thread_id, + communication, + AgentCommunicationContext::new(AgentCommunicationKind::Followup, ThreadId::new()), + ) .await .expect("send_inter_agent_communication should succeed"); diff --git a/codex-rs/core/src/agent_communication.rs b/codex-rs/core/src/agent_communication.rs new file mode 100644 index 000000000000..e71f50ad96a6 --- /dev/null +++ b/codex-rs/core/src/agent_communication.rs @@ -0,0 +1,79 @@ +use codex_protocol::ThreadId; +use codex_protocol::protocol::InterAgentCommunication; + +const AGENT_COMMUNICATION_TARGET: &str = "codex_otel.agent_communication"; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum AgentCommunicationKind { + Spawn, + Message, + Followup, + Result, +} + +impl AgentCommunicationKind { + fn as_str(self) -> &'static str { + match self { + Self::Spawn => "spawn", + Self::Message => "message", + Self::Followup => "followup", + Self::Result => "result", + } + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) struct AgentCommunicationContext { + kind: AgentCommunicationKind, + sender_thread_id: ThreadId, +} + +impl AgentCommunicationContext { + pub(crate) fn new(kind: AgentCommunicationKind, sender_thread_id: ThreadId) -> Self { + Self { + kind, + sender_thread_id, + } + } +} + +pub(crate) fn logging_enabled() -> bool { + tracing::enabled!(target: AGENT_COMMUNICATION_TARGET, tracing::Level::INFO) +} + +pub(crate) fn emit_agent_communication_send( + communication_id: &str, + context: &AgentCommunicationContext, + communication: &InterAgentCommunication, + receiver_thread_id: ThreadId, +) { + tracing::info!( + target: AGENT_COMMUNICATION_TARGET, + { + event.name = "codex.agent_communication", + communication_id, + kind = context.kind.as_str(), + state = "send", + sender_thread_id = %context.sender_thread_id, + receiver_thread_id = %receiver_thread_id, + content = if communication.content.is_empty() { + communication.encrypted_content.as_deref().unwrap_or_default() + } else { + communication.content.as_str() + }, + }, + "agent communication" + ); +} + +pub(crate) fn emit_agent_communication_receive(communication_id: &str) { + tracing::info!( + target: AGENT_COMMUNICATION_TARGET, + { + event.name = "codex.agent_communication", + communication_id, + state = "receive", + }, + "agent communication" + ); +} diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index e9b749fea949..df8753a38641 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -31,6 +31,7 @@ pub use codex_thread::TryStartTurnIfIdleError; pub use codex_thread::TryStartTurnIfIdleRejectionReason; pub use session::turn_context::TurnContext; mod agent; +mod agent_communication; mod attestation; mod codex_delegate; mod command_canonicalization; diff --git a/codex-rs/core/src/session/handlers.rs b/codex-rs/core/src/session/handlers.rs index a483ce5e0f75..9733a9f66aec 100644 --- a/codex-rs/core/src/session/handlers.rs +++ b/codex-rs/core/src/session/handlers.rs @@ -285,6 +285,7 @@ pub async fn inter_agent_communication( sess.input_queue .enqueue_mailbox_communication(communication) .await; + crate::agent_communication::emit_agent_communication_receive(&sub_id); if trigger_turn { sess.maybe_start_turn_for_pending_work_with_sub_id(sub_id) .await; diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index d43724e5db18..809fa92015bf 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -14,6 +14,8 @@ use crate::agent::AgentControl; use crate::agent::AgentStatus; use crate::agent::agent_status_from_event; use crate::agent::status::is_final; +use crate::agent_communication::AgentCommunicationContext; +use crate::agent_communication::AgentCommunicationKind; use crate::attestation::AttestationProvider; use crate::build_available_skills; use crate::compact; @@ -1867,10 +1869,12 @@ impl Session { message, /*trigger_turn*/ false, ); + let context = + AgentCommunicationContext::new(AgentCommunicationKind::Result, self.thread_id); if let Err(err) = self .services .agent_control - .send_inter_agent_communication(parent_thread_id, communication) + .send_inter_agent_communication(parent_thread_id, communication, context) .await { debug!("failed to notify parent thread {parent_thread_id}: {err}"); diff --git a/codex-rs/core/src/tools/handlers/agent_jobs.rs b/codex-rs/core/src/tools/handlers/agent_jobs.rs index 360744973c56..640541473189 100644 --- a/codex-rs/core/src/tools/handlers/agent_jobs.rs +++ b/codex-rs/core/src/tools/handlers/agent_jobs.rs @@ -203,7 +203,7 @@ async fn run_agent_job_loop( .agent_control .spawn_agent_with_metadata( options.spawn_config.clone(), - items.into(), + items, Some(SessionSource::SubAgent(SubAgentSource::Other(format!( "agent_job:{job_id}" )))), diff --git a/codex-rs/core/src/tools/handlers/multi_agents_common.rs b/codex-rs/core/src/tools/handlers/multi_agents_common.rs index 8f91ce5b8ee2..12ec89d651c9 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_common.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_common.rs @@ -18,7 +18,6 @@ use codex_protocol::openai_models::ReasoningEffort; use codex_protocol::openai_models::ReasoningEffortPreset; use codex_protocol::protocol::CollabAgentRef; use codex_protocol::protocol::CollabAgentStatusEntry; -use codex_protocol::protocol::Op; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SubAgentSource; use codex_protocol::user_input::UserInput; @@ -163,7 +162,7 @@ pub(crate) fn thread_spawn_source( pub(crate) fn parse_collab_input( message: Option, items: Option>, -) -> Result { +) -> Result, FunctionCallError> { match (message, items) { (Some(_), Some(_)) => Err(FunctionCallError::RespondToModel( "Provide either message or items, but not both".to_string(), @@ -180,8 +179,7 @@ pub(crate) fn parse_collab_input( Ok(vec![UserInput::Text { text: message, text_elements: Vec::new(), - }] - .into()) + }]) } (None, Some(items)) => { if items.is_empty() { @@ -189,7 +187,7 @@ pub(crate) fn parse_collab_input( "Items can't be empty".to_string(), )); } - Ok(items.into()) + Ok(items) } } } 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 8acb0a8c8d18..f313fcf0d9a7 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_tests.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_tests.rs @@ -1377,8 +1377,7 @@ async fn multi_agent_v2_send_message_accepts_root_target_from_child() { vec![UserInput::Text { text: "inspect this repo".to_string(), text_elements: Vec::new(), - }] - .into(), + }], Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id: root.thread_id, depth: 1, @@ -1454,8 +1453,7 @@ async fn multi_agent_v2_followup_task_rejects_root_target_from_child() { vec![UserInput::Text { text: "inspect this repo".to_string(), text_elements: Vec::new(), - }] - .into(), + }], Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id: root.thread_id, depth: 1, @@ -1626,8 +1624,7 @@ async fn multi_agent_v2_list_agents_filters_by_relative_path_prefix() { vec![UserInput::Text { text: "research".to_string(), text_elements: Vec::new(), - }] - .into(), + }], Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id: root.thread_id, depth: 1, @@ -1647,8 +1644,7 @@ async fn multi_agent_v2_list_agents_filters_by_relative_path_prefix() { vec![UserInput::Text { text: "build".to_string(), text_elements: Vec::new(), - }] - .into(), + }], Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id: root.thread_id, depth: 2, @@ -4097,8 +4093,7 @@ async fn multi_agent_v2_interrupt_agent_rejects_self_target_by_id() { vec![UserInput::Text { text: "inspect this repo".to_string(), text_elements: Vec::new(), - }] - .into(), + }], Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id: root.thread_id, depth: 1, @@ -4165,8 +4160,7 @@ async fn multi_agent_v2_interrupt_agent_rejects_self_target_by_task_name() { vec![UserInput::Text { text: "inspect this repo".to_string(), text_elements: Vec::new(), - }] - .into(), + }], Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id: root.thread_id, depth: 1, diff --git a/codex-rs/core/src/tools/handlers/multi_agents_v2.rs b/codex-rs/core/src/tools/handlers/multi_agents_v2.rs index cae3497a1319..8e7f73a7622f 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_v2.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_v2.rs @@ -19,7 +19,6 @@ use codex_protocol::protocol::CollabWaitingEndEvent; use codex_protocol::protocol::InterAgentCommunication; use codex_protocol::protocol::SubAgentActivityEvent; use codex_protocol::protocol::SubAgentActivityKind; -use codex_protocol::user_input::UserInput; use codex_tools::ToolName; use serde::Deserialize; use serde::Serialize; diff --git a/codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs b/codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs index 7fdb80e8bcb2..9501b4355ac9 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs @@ -4,6 +4,8 @@ //! resulting `InterAgentCommunication` should wake the target immediately. use super::*; +use crate::agent_communication::AgentCommunicationContext; +use crate::agent_communication::AgentCommunicationKind; use crate::tools::context::FunctionToolOutput; use crate::turn_timing::now_unix_timestamp_ms; use codex_protocol::protocol::InterAgentCommunication; @@ -46,7 +48,7 @@ pub(crate) struct FollowupTaskArgs { pub(crate) message: String, } -fn message_content(message: String) -> Result { +pub(super) fn message_content(message: String) -> Result { if message.trim().is_empty() { return Err(FunctionCallError::RespondToModel( "Empty message can't be sent to an agent".to_string(), @@ -101,10 +103,15 @@ pub(crate) async fn handle_message_string_tool( .unwrap_or_else(AgentPath::root); let communication = communication_from_tool_message(author, receiver_agent_path.clone(), message); + let kind = match mode { + MessageDeliveryMode::QueueOnly => AgentCommunicationKind::Message, + MessageDeliveryMode::TriggerTurn => AgentCommunicationKind::Followup, + }; + let context = AgentCommunicationContext::new(kind, session.thread_id); let result = session .services .agent_control - .send_inter_agent_communication(receiver_thread_id, mode.apply(communication)) + .send_inter_agent_communication(receiver_thread_id, mode.apply(communication), context) .await .map_err(|err| collab_agent_error(receiver_thread_id, err)); result?; 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 2186a4a6d605..8eeb6ddd1b78 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 @@ -4,11 +4,13 @@ use crate::agent::control::SpawnAgentOptions; use crate::agent::next_thread_spawn_depth; use crate::agent::role::DEFAULT_ROLE_NAME; use crate::agent::role::apply_role_to_config; +use crate::agent_communication::AgentCommunicationContext; +use crate::agent_communication::AgentCommunicationKind; use crate::tools::handlers::multi_agents_spec::SpawnAgentToolOptions; use crate::tools::handlers::multi_agents_spec::create_spawn_agent_tool_v2; +use crate::tools::handlers::multi_agents_v2::message_tool::message_content; use crate::turn_timing::now_unix_timestamp_ms; use codex_protocol::AgentPath; -use codex_protocol::protocol::Op; use codex_tools::ToolSpec; #[derive(Default)] @@ -55,8 +57,7 @@ async fn handle_spawn_agent( .map(str::trim) .filter(|role| !role.is_empty()); - let message = args.message.clone(); - let initial_operation = parse_collab_input(Some(args.message), /*items*/ None)?; + let message = message_content(args.message)?; let session_source = turn.session_source.clone(); let child_depth = next_thread_spawn_depth(&session_source); let mut config = @@ -104,33 +105,28 @@ async fn handle_spawn_agent( "spawned agent is missing a canonical task name".to_string(), ) })?; + let author = turn + .session_source + .get_agent_path() + .unwrap_or_else(AgentPath::root); + let communication = communication_from_tool_message(author, new_agent_path.clone(), message); + let context = AgentCommunicationContext::new(AgentCommunicationKind::Spawn, session.thread_id); let spawned_agent = Box::pin( - session.services.agent_control.spawn_agent_with_metadata( - config, - match initial_operation { - Op::UserInput { items, .. } - if items - .iter() - .all(|item| matches!(item, UserInput::Text { .. })) => - { - let author = turn - .session_source - .get_agent_path() - .unwrap_or_else(AgentPath::root); - let communication = - communication_from_tool_message(author, new_agent_path.clone(), message); - Op::InterAgentCommunication { communication } - } - initial_operation => initial_operation, - }, - Some(spawn_source), - SpawnAgentOptions { - fork_parent_spawn_call_id: fork_mode.as_ref().map(|_| call_id.clone()), - fork_mode, - parent_thread_id: Some(session.thread_id), - environments: Some(turn.environments.to_selections()), - }, - ), + session + .services + .agent_control + .spawn_agent_with_communication( + config, + communication, + context, + Some(spawn_source), + SpawnAgentOptions { + fork_parent_spawn_call_id: fork_mode.as_ref().map(|_| call_id.clone()), + fork_mode, + parent_thread_id: Some(session.thread_id), + environments: Some(turn.environments.to_selections()), + }, + ), ) .await .map_err(collab_spawn_error)?; diff --git a/codex-rs/core/tests/suite/subagent_notifications.rs b/codex-rs/core/tests/suite/subagent_notifications.rs index 951fb92f17cd..7424d3f360b1 100644 --- a/codex-rs/core/tests/suite/subagent_notifications.rs +++ b/codex-rs/core/tests/suite/subagent_notifications.rs @@ -39,10 +39,13 @@ use serde_json::Value; use serde_json::json; use std::fs; use std::path::Path; +use std::sync::Mutex; use std::time::Duration; use test_case::test_case; use tokio::time::Instant; use tokio::time::sleep; +use tracing::Level; +use tracing_test::internal::MockWriter; use wiremock::MockServer; const SPAWN_CALL_ID: &str = "spawn-call-1"; @@ -96,6 +99,13 @@ fn decoded_body(req: &wiremock::Request) -> Option> { } } +fn log_field<'a>(line: &'a str, name: &str) -> Option<&'a str> { + let prefix = format!("{name}="); + line.split_ascii_whitespace() + .find_map(|field| field.strip_prefix(&prefix)) + .map(|value| value.trim_matches('"')) +} + fn has_subagent_notification(req: &ResponsesRequest) -> bool { req.message_input_texts("user") .iter() @@ -1042,8 +1052,16 @@ async fn spawned_multi_agent_v2_child_inherits_parent_developer_context() -> Res Ok(()) } -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[tokio::test] async fn encrypted_multi_agent_v2_spawn_sends_agent_message_to_child() -> Result<()> { + let output: &'static Mutex> = Box::leak(Box::new(Mutex::new(Vec::new()))); + let subscriber = tracing_subscriber::fmt() + .with_ansi(false) + .with_max_level(Level::INFO) + .with_writer(MockWriter::new(output)) + .finish(); + let _guard = tracing::subscriber::set_default(subscriber); + let server = start_mock_server().await; let encrypted_message = "opaque-encrypted-message"; let spawn_args = serde_json::to_string(&json!({ @@ -1098,6 +1116,7 @@ async fn encrypted_multi_agent_v2_spawn_sends_agent_message_to_child() -> Result .expect("test config should allow feature update"); }); let test = builder.build(&server).await?; + let root_thread_id = test.session_configured.thread_id; test.submit_turn(TURN_1_PROMPT).await?; @@ -1124,6 +1143,41 @@ async fn encrypted_multi_agent_v2_spawn_sends_agent_message_to_child() -> Result })]) ); + let child_thread_id = test + .thread_manager + .list_thread_ids() + .await + .into_iter() + .find(|thread_id| *thread_id != root_thread_id) + .expect("child thread ID"); + let logs = tokio::time::timeout(Duration::from_secs(5), async { + loop { + let logs = String::from_utf8(output.lock().expect("buffer lock").clone()) + .expect("logs should be UTF-8"); + if logs.contains("kind=\"spawn\"") && logs.contains("state=\"receive\"") { + break logs; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("spawn communication logs should be emitted"); + let send = logs + .lines() + .find(|line| line.contains("kind=\"spawn\"") && line.contains("state=\"send\"")) + .expect("spawn send event"); + assert!(send.contains(&format!("sender_thread_id={root_thread_id}"))); + assert!(send.contains(&format!("receiver_thread_id={child_thread_id}"))); + assert!(send.contains(&format!("content=\"{encrypted_message}\""))); + + let communication_id = log_field(send, "communication_id").expect("communication ID"); + logs.lines() + .find(|line| { + line.contains("state=\"receive\"") + && log_field(line, "communication_id") == Some(communication_id) + }) + .expect("correlated receive event"); + Ok(()) }