From 129ea2aaf5fb426d8ba683ee53f290742f41dd31 Mon Sep 17 00:00:00 2001 From: Michael Bolin Date: Wed, 1 Jul 2026 18:11:09 -0700 Subject: [PATCH] Log multi-agent communication lifecycle (#30872) ## Why [#30867](https://github.com/openai/codex/pull/30867) makes `submit_inter_agent_communication` the common outbound sink for multi-agent v2 communications. This follow-up uses that single point to log every communication lifecycle without requiring new hooks as spawn, messaging, follow-up, or result paths evolve. For each communication, the logs need to identify its type, sender and receiver threads, and content, while correlating the successful send with receipt by the destination mailbox. The logging path must not query externally supplied time providers because those calls can be expensive for app-server clients. ## What changed - Added structured `INFO` events on the OpenTelemetry-exported `codex_otel.agent_communication` target for `spawn`, `message`, `followup`, and `result` communications. - Logged successful sends from `submit_inter_agent_communication` with the communication kind, sender and receiver thread IDs, content, and submission ID. - Logged receives after the communication has been enqueued in the receiver mailbox, using the same submission ID. - Avoided time-provider calls and other asynchronous work in the logging path. - Narrowed ordinary spawn and send-input APIs to `Vec` so `Op::InterAgentCommunication` cannot bypass the context-bearing centralized path. The refactor does not change submission IDs, capacity checks, last-task bookkeeping, mailbox ordering, protocol types, rollout data, or model-visible context. ## Event shape Illustrative JSON representation of the two independently emitted records: ```json [ { "event.name": "codex.agent_communication", "communication_id": "019f20e1-40d1-7890-a123-456789abcdef", "kind": "spawn", "state": "send", "sender_thread_id": "019f20df-fbe1-7890-a123-456789abcdef", "receiver_thread_id": "019f20e1-3f79-7890-a123-456789abcdef", "content": "inspect the repository" }, { "event.name": "codex.agent_communication", "communication_id": "019f20e1-40d1-7890-a123-456789abcdef", "state": "receive" } ] ``` Consumers join the receive record to the send record by `communication_id` for the immutable communication metadata. ## Testing - Extended the existing end-to-end multi-agent v2 spawn test to verify content, both thread IDs, and a correlated send/receive submission ID. - Re-ran focused control and handler coverage for direct messages, follow-up tasks, and completion results. --- codex-rs/core/src/agent/control.rs | 97 ++++++++++++------- codex-rs/core/src/agent/control/execution.rs | 11 ++- codex-rs/core/src/agent/control/spawn.rs | 62 ++++++++++-- codex-rs/core/src/agent/control_tests.rs | 32 +++--- codex-rs/core/src/agent_communication.rs | 79 +++++++++++++++ codex-rs/core/src/lib.rs | 1 + codex-rs/core/src/session/handlers.rs | 1 + codex-rs/core/src/session/mod.rs | 6 +- .../core/src/tools/handlers/agent_jobs.rs | 2 +- .../src/tools/handlers/multi_agents_common.rs | 8 +- .../src/tools/handlers/multi_agents_tests.rs | 18 ++-- .../src/tools/handlers/multi_agents_v2.rs | 1 - .../handlers/multi_agents_v2/message_tool.rs | 11 ++- .../tools/handlers/multi_agents_v2/spawn.rs | 54 +++++------ .../tests/suite/subagent_notifications.rs | 56 ++++++++++- 15 files changed, 332 insertions(+), 107 deletions(-) create mode 100644 codex-rs/core/src/agent_communication.rs diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index 8de6443402..85e54e2fb9 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 { - self.send_input(agent_id, Op::InterAgentCommunication { communication }) + 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.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 42aaa50e02..d8559c1499 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 c4c40577ca..9078434f1f 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 9863bfb7e2..4741c3c101 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 0000000000..e71f50ad96 --- /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 e9b749fea9..df8753a386 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 a483ce5e0f..9733a9f66a 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 d43724e5db..809fa92015 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 360744973c..6405414731 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 8f91ce5b8e..12ec89d651 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 8acb0a8c8d..f313fcf0d9 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 cae3497a13..8e7f73a762 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 7fdb80e8bc..9501b4355a 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 2186a4a6d6..8eeb6ddd1b 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 951fb92f17..7424d3f360 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(()) }