From 129ea2aaf5fb426d8ba683ee53f290742f41dd31 Mon Sep 17 00:00:00 2001 From: Michael Bolin Date: Wed, 1 Jul 2026 18:11:09 -0700 Subject: [PATCH 1/2] 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(()) } From f530914a545588ef8f45245ec0e5807b8740f348 Mon Sep 17 00:00:00 2001 From: Michael Bolin Date: Wed, 1 Jul 2026 21:54:40 -0700 Subject: [PATCH 2/2] telemetry: log structured tool timing events --- .../app-server/tests/common/json_logging.rs | 106 ++++++++-- .../tests/common/test_app_server.rs | 40 +++- codex-rs/app-server/tests/suite/logging.rs | 185 ++++++++++++++++++ codex-rs/core/src/session/turn.rs | 32 ++- codex-rs/core/src/tools/parallel.rs | 82 ++++++++ codex-rs/core/src/tools/registry.rs | 54 ++++- codex-rs/core/src/turn_timing.rs | 80 +++++++- codex-rs/core/tests/suite/otel.rs | 2 +- codex-rs/exec-server/src/local_process.rs | 4 +- codex-rs/exec-server/src/server/processor.rs | 24 ++- codex-rs/exec-server/src/telemetry.rs | 73 ++++++- codex-rs/exec-server/src/telemetry_tests.rs | 170 ++++++++++++++++ codex-rs/otel/src/events/session_telemetry.rs | 38 +++- .../tests/suite/otel_export_routing_policy.rs | 27 +++ 14 files changed, 872 insertions(+), 45 deletions(-) create mode 100644 codex-rs/exec-server/src/telemetry_tests.rs diff --git a/codex-rs/app-server/tests/common/json_logging.rs b/codex-rs/app-server/tests/common/json_logging.rs index c6c78bd0a4..8683d1bdaf 100644 --- a/codex-rs/app-server/tests/common/json_logging.rs +++ b/codex-rs/app-server/tests/common/json_logging.rs @@ -1,11 +1,79 @@ use std::path::Path; use std::process::Command; use std::process::Stdio; +use std::sync::Arc; +use std::sync::Mutex; +use std::time::Duration; use anyhow::Context; use anyhow::Result; use serde_json::Value; use serde_json::json; +use tokio::sync::Notify; + +#[derive(Clone, Default)] +pub(crate) struct JsonLogCapture { + lines: Arc>>, + updated: Arc, +} + +impl JsonLogCapture { + pub(crate) fn record(&self, line: String) { + self.lines + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .push(line); + self.updated.notify_one(); + } + + pub(crate) async fn wait_for_event(&self, event_name: &str) -> Result { + let mut events = self.wait_for_events(event_name, /*count*/ 1).await?; + Ok(events.remove(0)) + } + + pub(crate) async fn wait_for_events( + &self, + event_name: &str, + count: usize, + ) -> Result> { + let result = tokio::time::timeout(Duration::from_secs(10), async { + loop { + let updated = self.updated.notified(); + let events = self + .events()? + .into_iter() + .filter(|event| event["fields"]["event.name"].as_str() == Some(event_name)) + .collect::>(); + if events.len() >= count { + return Ok(events); + } + updated.await; + } + }) + .await; + match result { + Ok(result) => result, + Err(_) => { + let lines = self + .lines + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .join("\n"); + anyhow::bail!( + "timed out waiting for {count} JSON log event(s) named `{event_name}`; captured stderr:\n{lines}" + ) + } + } + } + + pub(crate) fn events(&self) -> Result> { + let lines = self + .lines + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + json_log_events(lines.iter().map(String::as_str)) + } +} pub fn app_server_json_shutdown_event( binary: &str, @@ -31,25 +99,39 @@ pub fn app_server_json_shutdown_event( let stderr = String::from_utf8(output.stderr)?; anyhow::ensure!(output.status.success(), "app-server failed: {stderr}"); - let events = stderr - .lines() - .filter(|line| !line.is_empty()) - .map(serde_json::from_str::) - .collect::>>() - .with_context(|| format!("app-server stderr was not JSONL: {stderr}"))?; + let events = json_log_events(stderr.lines()) + .with_context(|| format!("app-server stderr was not valid JSONL: {stderr}"))?; let event = events .iter() .find(|event| event["fields"]["message"] == "processor task exited") .context("missing INFO shutdown event in app-server JSON logs")?; - let timestamp = event["timestamp"] - .as_str() - .context("shutdown event did not include a timestamp")?; - chrono::DateTime::parse_from_rfc3339(timestamp) - .with_context(|| format!("shutdown event timestamp was not RFC 3339: {timestamp}"))?; - Ok(json!({ "level": event["level"], "fields": event["fields"], "target": event["target"], })) } + +fn json_log_events<'a>(lines: impl IntoIterator) -> Result> { + lines + .into_iter() + .filter(|line| !line.is_empty()) + .map(|line| { + let event = serde_json::from_str::(line) + .with_context(|| format!("log line was not JSON: {line}"))?; + anyhow::ensure!( + event["level"].is_string() + && event["fields"].is_object() + && event["target"].is_string(), + "JSON log event did not include level, fields, and target: {line}" + ); + let timestamp = event["timestamp"] + .as_str() + .with_context(|| format!("JSON log event did not include a timestamp: {line}"))?; + chrono::DateTime::parse_from_rfc3339(timestamp).with_context(|| { + format!("JSON log event timestamp was not RFC 3339: {timestamp}") + })?; + Ok(event) + }) + .collect() +} diff --git a/codex-rs/app-server/tests/common/test_app_server.rs b/codex-rs/app-server/tests/common/test_app_server.rs index fbe43e7689..3f68dddf81 100644 --- a/codex-rs/app-server/tests/common/test_app_server.rs +++ b/codex-rs/app-server/tests/common/test_app_server.rs @@ -123,6 +123,8 @@ use core_test_support::test_codex::TestEnv; use core_test_support::test_codex::test_env; use tokio::process::Command; +use crate::json_logging::JsonLogCapture; + pub struct TestAppServer { next_request_id: AtomicI64, /// Retain this child process until the client is dropped. The Tokio runtime @@ -134,6 +136,7 @@ pub struct TestAppServer { stdout: BufReader, pending_messages: VecDeque, auto_env: Option, + json_logs: JsonLogCapture, } pub const DEFAULT_CLIENT_NAME: &str = "codex-app-server-tests"; @@ -160,6 +163,14 @@ impl TestAppServer { /// URL-based configuration, this helper rejects a `codex_home` containing /// that file. pub async fn new_with_auto_env(codex_home: &Path) -> anyhow::Result { + Self::new_with_auto_env_and_env(codex_home, &[]).await + } + + /// Starts an auto-environment app server with child-process environment overrides. + pub async fn new_with_auto_env_and_env( + codex_home: &Path, + extra_env_overrides: &[(&str, Option<&str>)], + ) -> anyhow::Result { let environments_toml = codex_home.join("environments.toml"); ensure!( !environments_toml @@ -172,7 +183,7 @@ impl TestAppServer { let auto_env = test_env().await?; // Noise registry configuration takes precedence over the URL-based // provider, so clear inherited values to keep the selection hermetic. - let env_overrides = [ + let mut env_overrides = vec![ ( CODEX_EXEC_SERVER_URL_ENV_VAR, auto_env.environment().exec_server_url(), @@ -182,6 +193,7 @@ impl TestAppServer { (CODEX_EXEC_SERVER_NOISE_AUTH_TOKEN_ENV_VAR, None), (CODEX_EXEC_SERVER_NOISE_CHATGPT_ACCOUNT_ID_ENV_VAR, None), ]; + env_overrides.extend_from_slice(extra_env_overrides); let mut app_server = Self::new_with_env(codex_home, &env_overrides).await?; app_server.auto_env = Some(auto_env); Ok(app_server) @@ -202,6 +214,28 @@ impl TestAppServer { }) } + /// Waits for a JSON stderr event whose structured `event.name` field matches. + pub async fn wait_for_json_log_event( + &self, + event_name: &str, + ) -> anyhow::Result { + self.json_logs.wait_for_event(event_name).await + } + + /// Waits for the requested number of JSON stderr events with the same `event.name` field. + pub async fn wait_for_json_log_events( + &self, + event_name: &str, + count: usize, + ) -> anyhow::Result> { + self.json_logs.wait_for_events(event_name, count).await + } + + /// Returns every stderr line parsed and validated as a JSON log event. + pub fn json_log_events(&self) -> anyhow::Result> { + self.json_logs.events() + } + pub async fn new_without_managed_config(codex_home: &Path) -> anyhow::Result { Self::new_with_env(codex_home, &[(DISABLE_MANAGED_CONFIG_ENV_VAR, Some("1"))]).await } @@ -322,10 +356,13 @@ impl TestAppServer { // Forward child's stderr to our stderr so failures are visible even // when stdout/stderr are captured by the test harness. + let json_logs = JsonLogCapture::default(); if let Some(stderr) = process.stderr.take() { + let json_logs = json_logs.clone(); let mut stderr_reader = BufReader::new(stderr).lines(); tokio::spawn(async move { while let Ok(Some(line)) = stderr_reader.next_line().await { + json_logs.record(line.clone()); eprintln!("[mcp stderr] {line}"); } }); @@ -337,6 +374,7 @@ impl TestAppServer { stdout, pending_messages: VecDeque::new(), auto_env: None, + json_logs, }) } diff --git a/codex-rs/app-server/tests/suite/logging.rs b/codex-rs/app-server/tests/suite/logging.rs index ea2c31a766..989ec021b7 100644 --- a/codex-rs/app-server/tests/suite/logging.rs +++ b/codex-rs/app-server/tests/suite/logging.rs @@ -1,8 +1,28 @@ use anyhow::Result; +use app_test_support::TestAppServer; use app_test_support::app_server_json_shutdown_event; +use app_test_support::create_exec_command_sse_response; +use app_test_support::create_final_assistant_message_sse_response; +use app_test_support::create_mock_responses_server_sequence; +use app_test_support::to_response; +use app_test_support::write_mock_responses_config_toml; +use codex_app_server_protocol::JSONRPCResponse; +use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::ThreadStartParams; +use codex_app_server_protocol::ThreadStartResponse; +use codex_app_server_protocol::TurnStartParams; +use codex_app_server_protocol::TurnStartResponse; +use codex_app_server_protocol::UserInput; +use codex_features::Feature; +use core_test_support::skip_if_no_network; use pretty_assertions::assert_eq; use serde_json::json; +use std::collections::BTreeMap; use tempfile::TempDir; +use tokio::time::Duration; +use tokio::time::timeout; + +const READ_TIMEOUT: Duration = Duration::from_secs(10); #[test] fn standalone_app_server_emits_json_info_events() -> Result<()> { @@ -25,3 +45,168 @@ fn standalone_app_server_emits_json_info_events() -> Result<()> { Ok(()) } + +#[tokio::test] +async fn app_server_emits_structured_tool_timing_events() -> Result<()> { + skip_if_no_network!(Ok(())); + + let server = create_mock_responses_server_sequence(vec![ + create_exec_command_sse_response("exec-call-1")?, + create_final_assistant_message_sse_response("done")?, + ]) + .await; + let codex_home = TempDir::new()?; + write_mock_responses_config_toml( + codex_home.path(), + &server.uri(), + &BTreeMap::from([(Feature::UnifiedExec, true)]), + /*auto_compact_limit*/ 100_000, + /*requires_openai_auth*/ None, + "mock_provider", + "compact", + )?; + + let mut app_server = TestAppServer::new_with_auto_env_and_env( + codex_home.path(), + &[ + ("LOG_FORMAT", Some("json")), + ( + "RUST_LOG", + Some( + "warn,codex_core::tools::parallel=info,codex_core::turn_timing=info,codex_otel.log_only=info", + ), + ), + ], + ) + .await?; + timeout(READ_TIMEOUT, app_server.initialize()).await??; + + let thread_start_id = app_server + .send_thread_start_request_with_auto_env(ThreadStartParams { + model: Some("mock-model".to_string()), + ..Default::default() + }) + .await?; + let thread_start_response: JSONRPCResponse = timeout( + READ_TIMEOUT, + app_server.read_stream_until_response_message(RequestId::Integer(thread_start_id)), + ) + .await??; + let ThreadStartResponse { thread, .. } = to_response(thread_start_response)?; + + let turn_start_id = app_server + .send_turn_start_request(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![UserInput::Text { + text: "run a command".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }) + .await?; + let turn_start_response: JSONRPCResponse = timeout( + READ_TIMEOUT, + app_server.read_stream_until_response_message(RequestId::Integer(turn_start_id)), + ) + .await??; + let TurnStartResponse { turn } = to_response(turn_start_response)?; + + timeout( + READ_TIMEOUT, + app_server.read_stream_until_notification_message("turn/completed"), + ) + .await??; + + let tool_call = app_server + .wait_for_json_log_event("codex.tool_call") + .await?; + assert_eq!(tool_call["level"], "INFO"); + assert_eq!(tool_call["target"], "codex_core::tools::parallel"); + assert_eq!(tool_call["fields"]["message"], "tool call completed"); + assert!(tool_call["fields"]["trace_id"].is_string()); + assert_eq!(tool_call["fields"]["conversation.id"], thread.id); + assert_eq!(tool_call["fields"]["turn_id"], turn.id); + assert_eq!(tool_call["fields"]["tool_name"], "exec_command"); + assert_eq!(tool_call["fields"]["call_id"], "exec-call-1"); + assert_eq!(tool_call["fields"]["tool_source"], "direct"); + assert_eq!(tool_call["fields"]["code_mode_cell_id"], ""); + assert_eq!(tool_call["fields"]["code_mode_runtime_tool_call_id"], ""); + assert_eq!(tool_call["fields"]["execution_started"], true); + assert_nonnegative_duration_fields( + &tool_call, + &[ + "dispatch_duration_seconds", + "handler_duration_seconds", + "total_duration_seconds", + ], + ); + let dispatch_duration = duration_field(&tool_call, "dispatch_duration_seconds"); + let handler_duration = duration_field(&tool_call, "handler_duration_seconds"); + let total_duration = duration_field(&tool_call, "total_duration_seconds"); + assert!(total_duration > 0.0); + assert!( + (dispatch_duration + handler_duration - total_duration).abs() < 0.000_001, + "dispatch and handler durations must sum to total duration: {tool_call}" + ); + + let tool_result = app_server + .wait_for_json_log_event("codex.tool_result") + .await?; + assert_eq!(tool_result["level"], "INFO"); + assert_eq!(tool_result["target"], "codex_otel.log_only"); + assert!(tool_result["fields"]["trace_id"].is_string()); + assert_eq!(tool_result["fields"]["conversation.id"], thread.id); + assert_eq!(tool_result["fields"]["turn_id"], turn.id); + assert_eq!(tool_result["fields"]["tool_name"], "exec_command"); + assert_eq!(tool_result["fields"]["call_id"], "exec-call-1"); + assert_eq!(tool_result["fields"]["tool_source"], "direct"); + assert!( + matches!( + tool_result["fields"]["success"].as_str(), + Some("true" | "false") + ), + "success must be a boolean string: {tool_result}" + ); + assert_nonnegative_duration_fields(&tool_result, &["duration_seconds"]); + + let inferences = app_server + .wait_for_json_log_events("codex.inference", /*count*/ 2) + .await?; + assert_eq!( + inferences + .iter() + .map(|event| event["fields"]["inference_index"].as_u64()) + .collect::>(), + vec![Some(1), Some(2)] + ); + for inference in inferences { + assert_eq!(inference["level"], "INFO"); + assert_eq!(inference["target"], "codex_core::turn_timing"); + assert_eq!(inference["fields"]["message"], "inference completed"); + assert!(inference["fields"]["trace_id"].is_string()); + assert_eq!(inference["fields"]["conversation.id"], thread.id); + assert_eq!(inference["fields"]["turn_id"], turn.id); + assert_eq!(inference["fields"]["model"], "mock-model"); + assert_eq!( + inference["fields"]["provider_name"], + "Mock provider for test" + ); + assert_eq!(inference["fields"]["result"], "success"); + assert_nonnegative_duration_fields(&inference, &["duration_seconds"]); + } + + Ok(()) +} + +fn assert_nonnegative_duration_fields(event: &serde_json::Value, fields: &[&str]) { + for field in fields { + let duration = duration_field(event, field); + assert!(duration >= 0.0, "{field} must be nonnegative: {event}"); + } +} + +fn duration_field(event: &serde_json::Value, field: &str) -> f64 { + event["fields"][field] + .as_f64() + .unwrap_or_else(|| panic!("{field} must be a JSON number: {event}")) +} diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index f16b525002..c8880cb462 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -65,6 +65,7 @@ use crate::tools::router::extension_tool_executors; use crate::tools::spec_plan::search_tool_enabled; use crate::tools::spec_plan::tool_suggest_enabled; use crate::turn_diff_tracker::TurnDiffTracker; +use crate::turn_timing::InferenceTimingResult; use crate::turn_timing::record_turn_ttft_metric; use crate::util::error_or_panic; use codex_analytics::AppInvocation; @@ -1903,13 +1904,19 @@ async fn try_run_sampling_request( auth_mode = sess.services.auth_manager.auth_mode(), features = sess.features.enabled_features(), ); + let provider_info = turn_context.provider.info(); let inference_trace = sess.services.rollout_thread_trace.inference_trace_context( turn_context.sub_id.as_str(), turn_context.model_info.slug.as_str(), - turn_context.provider.info().name.as_str(), + provider_info.name.as_str(), ); - let sampling_timing_guard = turn_context.turn_timing_state.begin_sampling(); - let mut stream = client_session + let sampling_timing_guard = turn_context.turn_timing_state.begin_sampling( + &sess.thread_id, + &turn_context.sub_id, + &turn_context.model_info.slug, + &provider_info.name, + ); + let stream_result = client_session .stream( prompt, &turn_context.model_info, @@ -1922,7 +1929,18 @@ async fn try_run_sampling_request( ) .instrument(trace_span!("stream_request")) .or_cancel(&cancellation_token) - .await??; + .await; + let mut stream = match stream_result { + Ok(Ok(stream)) => stream, + Ok(Err(err)) => { + sampling_timing_guard.finish_inference(InferenceTimingResult::Error); + return Err(err); + } + Err(codex_async_utils::CancelErr::Cancelled) => { + sampling_timing_guard.finish_inference(InferenceTimingResult::Cancelled); + return Err(CodexErr::TurnAborted); + } + }; let mut in_flight: FuturesOrdered>> = FuturesOrdered::new(); let mut needs_follow_up = false; @@ -2346,7 +2364,11 @@ async fn try_run_sampling_request( } } }; - drop(sampling_timing_guard); + sampling_timing_guard.finish_inference(match &outcome { + Ok(_) => InferenceTimingResult::Success, + Err(CodexErr::TurnAborted) => InferenceTimingResult::Cancelled, + Err(_) => InferenceTimingResult::Error, + }); flush_assistant_text_segments_all( &sess, diff --git a/codex-rs/core/src/tools/parallel.rs b/codex-rs/core/src/tools/parallel.rs index a2ee89ab83..593f7d0213 100644 --- a/codex-rs/core/src/tools/parallel.rs +++ b/codex-rs/core/src/tools/parallel.rs @@ -1,4 +1,5 @@ use std::sync::Arc; +use std::sync::OnceLock; use std::sync::atomic::AtomicBool; use std::sync::atomic::Ordering; use std::time::Instant; @@ -9,6 +10,7 @@ use tokio_util::either::Either; use tokio_util::sync::CancellationToken; use tokio_util::task::AbortOnDropHandle; use tracing::Instrument; +use tracing::info; use tracing::instrument; use tracing::trace_span; @@ -27,6 +29,16 @@ use crate::tools::router::ToolRouter; use codex_protocol::error::CodexErr; use codex_protocol::models::ResponseInputItem; +struct ToolCallTimingGuard { + started_at: Instant, + execution_started_at: Arc>, + conversation_id: String, + turn_id: String, + call_id: String, + tool_name: codex_tools::ToolName, + source: ToolCallSource, +} + #[derive(Clone)] pub(crate) struct ToolCallRuntime { router: Arc, @@ -96,6 +108,11 @@ impl ToolCallRuntime { let invocation_cancellation_token = cancellation_token.clone(); let wait_for_runtime_cancellation = self.router.tool_waits_for_runtime_cancellation(&call); let started = Instant::now(); + let tool_call_timing_guard = + ToolCallTimingGuard::capture(started, &session.thread_id, &turn.sub_id, &call, &source); + let execution_started_at = tool_call_timing_guard + .as_ref() + .map(|timing| Arc::clone(&timing.execution_started_at)); let abort_session = Arc::clone(&session); let abort_source = source.clone(); let abort_turn = Arc::clone(&turn); @@ -119,6 +136,9 @@ impl ToolCallRuntime { } else { Either::Right(lock.write().await) }; + if let Some(execution_started_at) = execution_started_at { + let _ = execution_started_at.set(Instant::now()); + } router .dispatch_tool_call_with_terminal_outcome( @@ -135,6 +155,7 @@ impl ToolCallRuntime { })); async move { + let _tool_call_timing_guard = tool_call_timing_guard; tokio::select! { res = &mut handle => res.map_err(Self::tool_task_join_error)?, _ = cancellation_token.cancelled() => { @@ -237,6 +258,67 @@ impl ToolCallRuntime { } } +impl ToolCallTimingGuard { + fn capture( + started_at: Instant, + conversation_id: &impl std::fmt::Display, + turn_id: &str, + call: &ToolCall, + source: &ToolCallSource, + ) -> Option { + if !tracing::enabled!(tracing::Level::INFO) { + return None; + } + + Some(Self { + started_at, + execution_started_at: Arc::new(OnceLock::new()), + conversation_id: conversation_id.to_string(), + turn_id: turn_id.to_string(), + call_id: call.call_id.clone(), + tool_name: call.tool_name.clone(), + source: source.clone(), + }) + } +} + +impl Drop for ToolCallTimingGuard { + fn drop(&mut self) { + let completed_at = Instant::now(); + info!( + event.name = "codex.tool_call", + trace_id = %codex_otel::current_span_trace_id().unwrap_or_default(), + conversation.id = %self.conversation_id, + turn_id = %self.turn_id, + tool_name = %self.tool_name, + call_id = %self.call_id, + tool_source = match &self.source { + ToolCallSource::Direct => "direct", + ToolCallSource::CodeMode { .. } => "code_mode", + }, + code_mode_cell_id = match &self.source { + ToolCallSource::Direct => "", + ToolCallSource::CodeMode { cell_id, .. } => cell_id.as_str(), + }, + code_mode_runtime_tool_call_id = match &self.source { + ToolCallSource::Direct => "", + ToolCallSource::CodeMode { runtime_tool_call_id, .. } => runtime_tool_call_id.as_str(), + }, + execution_started = self.execution_started_at.get().is_some(), + dispatch_duration_seconds = self.execution_started_at.get().map_or_else( + || completed_at.duration_since(self.started_at).as_secs_f64(), + |execution_started_at| execution_started_at.duration_since(self.started_at).as_secs_f64(), + ), + handler_duration_seconds = self.execution_started_at.get().map_or( + 0.0, + |execution_started_at| completed_at.duration_since(*execution_started_at).as_secs_f64(), + ), + total_duration_seconds = completed_at.duration_since(self.started_at).as_secs_f64(), + "tool call completed" + ); + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/codex-rs/core/src/tools/registry.rs b/codex-rs/core/src/tools/registry.rs index 2220d4244a..5895208d0e 100644 --- a/codex-rs/core/src/tools/registry.rs +++ b/codex-rs/core/src/tools/registry.rs @@ -14,6 +14,7 @@ use crate::sandbox_tags::permission_profile_policy_tag; use crate::sandbox_tags::permission_profile_sandbox_tag; use crate::session::turn_context::TurnContext; use crate::tools::context::FunctionToolOutput; +use crate::tools::context::ToolCallSource; use crate::tools::context::ToolInvocation; use crate::tools::context::ToolOutput; use crate::tools::context::ToolPayload; @@ -444,6 +445,8 @@ impl ToolRegistry { None => { let message = unsupported_tool_call_message(&invocation.payload, &tool_name); let log_payload = invocation.payload.log_payload(); + let extra_trace_fields = + tool_result_trace_fields(&invocation.turn.sub_id, &invocation.source, "", ""); otel.tool_result_with_tags( tool_name_flat.as_ref(), &call_id_owned, @@ -452,7 +455,7 @@ impl ToolRegistry { /*success*/ false, &message, &base_tool_result_tags, - /*extra_trace_fields*/ &[], + &extra_trace_fields, ); let err = FunctionCallError::RespondToModel(message); dispatch_trace.record_failed(&err); @@ -463,18 +466,25 @@ impl ToolRegistry { let telemetry_tags = tool.telemetry_tags(&invocation).await; let mut tool_result_tags = Vec::with_capacity(base_tool_result_tags.len() + telemetry_tags.len()); - let mut extra_trace_fields = Vec::new(); + let mut mcp_server = ""; + let mut mcp_server_origin = ""; tool_result_tags.extend_from_slice(&base_tool_result_tags); for (key, value) in &telemetry_tags { - if matches!(*key, "mcp_server" | "mcp_server_origin") { - extra_trace_fields.push((*key, value.as_str())); - } else { - tool_result_tags.push((*key, value.as_str())); + match *key { + "mcp_server" => mcp_server = value.as_str(), + "mcp_server_origin" => mcp_server_origin = value.as_str(), + _ => tool_result_tags.push((*key, value.as_str())), } } if !tool.matches_kind(&invocation.payload) { let message = format!("tool {tool_name} invoked with incompatible payload"); let log_payload = invocation.payload.log_payload(); + let extra_trace_fields = tool_result_trace_fields( + &invocation.turn.sub_id, + &invocation.source, + mcp_server, + mcp_server_origin, + ); otel.tool_result_with_tags( tool_name_flat.as_ref(), &call_id_owned, @@ -541,6 +551,12 @@ impl ToolRegistry { let response_cell = tokio::sync::Mutex::new(None); let invocation_for_tool = invocation.clone(); let log_payload = invocation.payload.log_payload(); + let extra_trace_fields = tool_result_trace_fields( + &invocation.turn.sub_id, + &invocation.source, + mcp_server, + mcp_server_origin, + ); let result = otel .log_tool_result_with_tags( @@ -670,6 +686,32 @@ impl ToolRegistry { } } +fn tool_result_trace_fields<'a>( + turn_id: &'a str, + source: &'a ToolCallSource, + mcp_server: &'a str, + mcp_server_origin: &'a str, +) -> [(&'static str, &'a str); 6] { + let (tool_source, code_mode_cell_id, code_mode_runtime_tool_call_id) = match source { + ToolCallSource::Direct => ("direct", "", ""), + ToolCallSource::CodeMode { + cell_id, + runtime_tool_call_id, + } => ("code_mode", cell_id.as_str(), runtime_tool_call_id.as_str()), + }; + [ + ("turn_id", turn_id), + ("tool_source", tool_source), + ("code_mode_cell_id", code_mode_cell_id), + ( + "code_mode_runtime_tool_call_id", + code_mode_runtime_tool_call_id, + ), + ("mcp_server", mcp_server), + ("mcp_server_origin", mcp_server_origin), + ] +} + async fn notify_tool_finish_if_unclaimed( invocation: &ToolInvocation, terminal_outcome_reached: Option<&AtomicBool>, diff --git a/codex-rs/core/src/turn_timing.rs b/codex-rs/core/src/turn_timing.rs index 04f7aeb69b..7a13a63bfc 100644 --- a/codex-rs/core/src/turn_timing.rs +++ b/codex-rs/core/src/turn_timing.rs @@ -10,6 +10,7 @@ use codex_otel::TURN_TTFM_DURATION_METRIC; use codex_protocol::items::TurnItem; use codex_protocol::models::ResponseItem; use tokio::sync::Mutex; +use tracing::info; use crate::ResponseEvent; use crate::session::turn_context::TurnContext; @@ -80,6 +81,25 @@ pub(crate) struct TurnProfileTimingGuard { timing: Arc, phase: TurnProfilePhase, active: bool, + inference_log_context: Option, +} + +#[derive(Clone, Copy)] +pub(crate) enum InferenceTimingResult { + Success, + Cancelled, + Error, +} + +struct InferenceLogContext { + started_at: Instant, + conversation_id: String, + turn_id: String, + model: String, + provider_name: String, + trace_id: String, + inference_index: u32, + result: InferenceTimingResult, } impl TurnTimingState { @@ -118,12 +138,35 @@ impl TurnTimingState { self.profile_state().complete(Instant::now()) } - pub(crate) fn begin_sampling(self: &Arc) -> TurnProfileTimingGuard { - let active = self.profile_state().begin_sampling(Instant::now()); + pub(crate) fn begin_sampling( + self: &Arc, + conversation_id: &impl std::fmt::Display, + turn_id: &str, + model: &str, + provider_name: &str, + ) -> TurnProfileTimingGuard { + let started_at = Instant::now(); + let inference_index = self.profile_state().begin_sampling(started_at); + let active = inference_index.is_some(); + let inference_log_context = if tracing::enabled!(tracing::Level::INFO) { + inference_index.map(|inference_index| InferenceLogContext { + started_at, + conversation_id: conversation_id.to_string(), + turn_id: turn_id.to_string(), + model: model.to_string(), + provider_name: provider_name.to_string(), + trace_id: codex_otel::current_span_trace_id().unwrap_or_default(), + inference_index, + result: InferenceTimingResult::Error, + }) + } else { + None + }; TurnProfileTimingGuard { timing: Arc::clone(self), phase: TurnProfilePhase::Sampling, active, + inference_log_context, } } @@ -137,6 +180,7 @@ impl TurnTimingState { timing: Arc::clone(self), phase: TurnProfilePhase::ToolBlocking, active, + inference_log_context: None, } } @@ -166,6 +210,14 @@ impl TurnTimingState { } } +impl TurnProfileTimingGuard { + pub(crate) fn finish_inference(mut self, result: InferenceTimingResult) { + if let Some(log_context) = &mut self.inference_log_context { + log_context.result = result; + } + } +} + impl Drop for TurnProfileTimingGuard { fn drop(&mut self) { if self.active { @@ -173,6 +225,24 @@ impl Drop for TurnProfileTimingGuard { .profile_state() .end_phase(Instant::now(), self.phase); } + if let Some(log_context) = &self.inference_log_context { + info!( + event.name = "codex.inference", + trace_id = %log_context.trace_id, + conversation.id = %log_context.conversation_id, + turn_id = %log_context.turn_id, + model = %log_context.model, + provider_name = %log_context.provider_name, + inference_index = log_context.inference_index, + result = match log_context.result { + InferenceTimingResult::Success => "success", + InferenceTimingResult::Cancelled => "cancelled", + InferenceTimingResult::Error => "error", + }, + duration_seconds = log_context.started_at.elapsed().as_secs_f64(), + "inference completed" + ); + } } } @@ -200,12 +270,12 @@ impl TurnProfileState { }; } - fn begin_sampling(&mut self, now: Instant) -> bool { + fn begin_sampling(&mut self, now: Instant) -> Option { if self.completed_profile.is_some() || self.started_at.is_none() || self.active_phase.is_some() { - return false; + return None; } self.advance(now); if self.seen_sampling { @@ -214,7 +284,7 @@ impl TurnProfileState { self.seen_sampling = true; self.active_phase = Some(TurnProfilePhase::Sampling); self.sampling_request_count = self.sampling_request_count.saturating_add(1); - true + Some(self.sampling_request_count) } fn record_sampling_retry(&mut self) { diff --git a/codex-rs/core/tests/suite/otel.rs b/codex-rs/core/tests/suite/otel.rs index e3415d945b..97c7f6b5cb 100644 --- a/codex-rs/core/tests/suite/otel.rs +++ b/codex-rs/core/tests/suite/otel.rs @@ -710,7 +710,7 @@ async fn turn_and_completed_response_spans_record_token_usage() { ); } -#[tokio::test] +#[tokio::test(flavor = "current_thread")] async fn handle_responses_span_records_response_kind_and_tool_name() { let buffer: &'static Mutex> = Box::leak(Box::new(Mutex::new(Vec::new()))); let subscriber = tracing_subscriber::fmt() diff --git a/codex-rs/exec-server/src/local_process.rs b/codex-rs/exec-server/src/local_process.rs index 3c488f0ad7..08d2d0319f 100644 --- a/codex-rs/exec-server/src/local_process.rs +++ b/codex-rs/exec-server/src/local_process.rs @@ -334,7 +334,7 @@ impl LocalProcess { output_notify: Arc::clone(&output_notify), open_streams: 2, closed: false, - metrics: Some(self.inner.telemetry.process_started()), + metrics: Some(self.inner.telemetry.process_started(process_id.as_ref())), termination_requested: false, sandbox: prepared.sandbox, sandbox_denied: false, @@ -1306,7 +1306,7 @@ mod tests { output_notify: Arc::clone(&output_notify), open_streams: 2, closed: false, - metrics: Some(backend.inner.telemetry.process_started()), + metrics: Some(backend.inner.telemetry.process_started(process_id.as_ref())), termination_requested: false, sandbox: SandboxType::None, sandbox_denied: false, diff --git a/codex-rs/exec-server/src/server/processor.rs b/codex-rs/exec-server/src/server/processor.rs index 63f6bd45d9..56690c65c5 100644 --- a/codex-rs/exec-server/src/server/processor.rs +++ b/codex-rs/exec-server/src/server/processor.rs @@ -128,6 +128,13 @@ async fn run_connection( JsonRpcConnectionEvent::Message(message) => match message { codex_exec_server_protocol::JSONRPCMessage::Request(request) => { let request_started_at = Instant::now(); + let request_log_context = telemetry.request_log_context( + &request.id, + request + .trace + .as_ref() + .and_then(|trace| trace.traceparent.as_deref()), + ); if let Some((method, route)) = router.request_route(request.method.as_str()) { let request_span = request_span(method, &request); let message = tokio::select! { @@ -135,6 +142,7 @@ async fn run_connection( _ = disconnected_rx.changed() => { request_span.record("result", "disconnected"); telemetry.request_completed( + request_log_context.as_ref(), method, "disconnected", request_started_at.elapsed(), @@ -149,6 +157,7 @@ async fn run_connection( { request_span.record("result", "disconnected"); telemetry.request_completed( + request_log_context.as_ref(), method, "disconnected", request_started_at.elapsed(), @@ -156,7 +165,12 @@ async fn run_connection( break; } request_span.record("result", result); - telemetry.request_completed(method, result, request_started_at.elapsed()); + telemetry.request_completed( + request_log_context.as_ref(), + method, + result, + request_started_at.elapsed(), + ); drop(request_span); } else { let method = "unknown"; @@ -174,6 +188,7 @@ async fn run_connection( { request_span.record("result", "disconnected"); telemetry.request_completed( + request_log_context.as_ref(), method, "disconnected", request_started_at.elapsed(), @@ -181,7 +196,12 @@ async fn run_connection( break; } request_span.record("result", "error"); - telemetry.request_completed(method, "error", request_started_at.elapsed()); + telemetry.request_completed( + request_log_context.as_ref(), + method, + "error", + request_started_at.elapsed(), + ); } } codex_exec_server_protocol::JSONRPCMessage::Notification(notification) => { diff --git a/codex-rs/exec-server/src/telemetry.rs b/codex-rs/exec-server/src/telemetry.rs index 650faacb2c..4a3b1d0dcc 100644 --- a/codex-rs/exec-server/src/telemetry.rs +++ b/codex-rs/exec-server/src/telemetry.rs @@ -4,6 +4,7 @@ use std::time::Duration; use std::time::Instant; use codex_otel::MetricsClient; +use tracing::info; use tracing::warn; const CONNECTIONS_ACTIVE_METRIC: &str = "exec_server_connections_active"; @@ -96,10 +97,21 @@ pub(crate) struct ConnectionMetricGuard { pub(crate) struct ProcessMetricGuard { telemetry: ExecServerTelemetry, + log_context: Option, started_at: Instant, result: &'static str, } +struct ProcessLogContext { + process_id: String, + trace_id: String, +} + +pub(crate) struct RequestLogContext { + request_id: String, + traceparent: Option, +} + impl ExecServerTelemetry { pub fn new(metrics: MetricsClient) -> Self { let active = Arc::new(Mutex::new(ActiveCounts::default())); @@ -129,10 +141,22 @@ impl ExecServerTelemetry { pub(crate) fn request_completed( &self, + log_context: Option<&RequestLogContext>, method: &'static str, result: &'static str, duration: Duration, ) { + if let Some(log_context) = log_context { + info!( + event.name = "codex.exec_server_request", + request_id = %log_context.request_id, + method, + result, + duration_seconds = duration.as_secs_f64(), + traceparent = log_context.traceparent.as_deref().unwrap_or(""), + "exec-server request completed" + ); + } self.with_inner(|inner| { let tags = [("method", method), ("result", result)]; inner.counter(REQUESTS_TOTAL_METRIC, REQUESTS_TOTAL_DESCRIPTION, &tags); @@ -145,6 +169,17 @@ impl ExecServerTelemetry { }); } + pub(crate) fn request_log_context( + &self, + request_id: &impl std::fmt::Display, + traceparent: Option<&str>, + ) -> Option { + self.info_events_enabled().then(|| RequestLogContext { + request_id: request_id.to_string(), + traceparent: traceparent.map(str::to_string), + }) + } + pub(crate) fn remote_registration_completed(&self, result: &'static str, duration: Duration) { self.record_operation(REMOTE_REGISTRATION_METRICS, result, duration); } @@ -163,18 +198,37 @@ impl ExecServerTelemetry { }); } - pub(crate) fn process_started(&self) -> ProcessMetricGuard { + pub(crate) fn process_started(&self, process_id: &str) -> ProcessMetricGuard { self.with_inner(|inner| { inner.adjust_process_count(/*delta*/ 1); }); ProcessMetricGuard { telemetry: self.clone(), + log_context: self.info_events_enabled().then(|| ProcessLogContext { + process_id: process_id.to_string(), + trace_id: codex_otel::current_span_trace_id().unwrap_or_default(), + }), started_at: Instant::now(), result: "unknown", } } - fn process_finished(&self, result: &'static str, duration: Duration) { + fn process_finished( + &self, + log_context: Option<&ProcessLogContext>, + result: &'static str, + duration: Duration, + ) { + if let Some(log_context) = log_context { + info!( + event.name = "codex.exec_server_process", + process_id = %log_context.process_id, + trace_id = %log_context.trace_id, + result, + duration_seconds = duration.as_secs_f64(), + "exec-server process completed" + ); + } self.with_inner(|inner| { inner.adjust_process_count(/*delta*/ -1); inner.counter( @@ -191,6 +245,10 @@ impl ExecServerTelemetry { }); } + pub(crate) fn info_events_enabled(&self) -> bool { + tracing::enabled!(tracing::Level::INFO) + } + fn connection_finished(&self, transport: ConnectionTransport) { self.with_inner(|inner| { inner.adjust_connection_count(transport, /*delta*/ -1); @@ -236,8 +294,11 @@ impl ProcessMetricGuard { impl Drop for ProcessMetricGuard { fn drop(&mut self) { - self.telemetry - .process_finished(self.result, self.started_at.elapsed()); + self.telemetry.process_finished( + self.log_context.as_ref(), + self.result, + self.started_at.elapsed(), + ); } } @@ -338,3 +399,7 @@ fn register_active_gauge( warn!(metric = name, "failed to register exec-server gauge"); } } + +#[cfg(test)] +#[path = "telemetry_tests.rs"] +mod tests; diff --git a/codex-rs/exec-server/src/telemetry_tests.rs b/codex-rs/exec-server/src/telemetry_tests.rs new file mode 100644 index 0000000000..05c43cfcc6 --- /dev/null +++ b/codex-rs/exec-server/src/telemetry_tests.rs @@ -0,0 +1,170 @@ +use std::collections::BTreeMap; +use std::sync::Arc; +use std::sync::Mutex; +use std::time::Duration; + +use pretty_assertions::assert_eq; +use tracing::Event; +use tracing::Level; +use tracing::Subscriber; +use tracing::field::Field; +use tracing::field::Visit; +use tracing_subscriber::Layer; +use tracing_subscriber::layer::Context; +use tracing_subscriber::prelude::*; +use tracing_subscriber::registry::LookupSpan; + +use super::ExecServerTelemetry; + +#[derive(Clone, Debug, PartialEq, Eq)] +struct CapturedEvent { + level: Level, + target: String, + fields: BTreeMap, +} + +#[derive(Clone, Default)] +struct CaptureLayer { + events: Arc>>, +} + +impl CaptureLayer { + fn events(&self) -> Vec { + self.events + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .clone() + } +} + +impl Layer for CaptureLayer +where + S: Subscriber + for<'lookup> LookupSpan<'lookup>, +{ + fn on_event(&self, event: &Event<'_>, _context: Context<'_, S>) { + let mut visitor = FieldVisitor::default(); + event.record(&mut visitor); + self.events + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .push(CapturedEvent { + level: *event.metadata().level(), + target: event.metadata().target().to_string(), + fields: visitor.fields, + }); + } +} + +#[derive(Default)] +struct FieldVisitor { + fields: BTreeMap, +} + +impl Visit for FieldVisitor { + fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) { + self.fields + .insert(field.name().to_string(), format!("{value:?}")); + } + + fn record_str(&mut self, field: &Field, value: &str) { + self.fields + .insert(field.name().to_string(), value.to_string()); + } + + fn record_f64(&mut self, field: &Field, value: f64) { + self.fields + .insert(field.name().to_string(), value.to_string()); + } +} + +#[test] +fn exec_server_timing_events_are_structured_info_logs() { + let capture = CaptureLayer::default(); + let subscriber = tracing_subscriber::registry().with(capture.clone()); + + tracing::subscriber::with_default(subscriber, || { + tracing::callsite::rebuild_interest_cache(); + let telemetry = ExecServerTelemetry::default(); + let request_log_context = telemetry + .request_log_context(&"request-1", Some("00-trace-parent")) + .expect("INFO event should be enabled"); + telemetry.request_completed( + Some(&request_log_context), + "process/start", + "success", + Duration::from_millis(42), + ); + telemetry.process_started("process-1").finish("success"); + }); + + let events = capture.events(); + let request = events + .iter() + .find(|event| { + event + .fields + .get("event.name") + .is_some_and(|name| name == "codex.exec_server_request") + }) + .expect("request timing event"); + assert_eq!(request.level, Level::INFO); + assert_eq!(request.target, "codex_exec_server::telemetry"); + assert_eq!( + request.fields, + BTreeMap::from([ + ("duration_seconds".to_string(), "0.042".to_string()), + ( + "event.name".to_string(), + "codex.exec_server_request".to_string(), + ), + ( + "message".to_string(), + "exec-server request completed".to_string(), + ), + ("method".to_string(), "process/start".to_string()), + ("request_id".to_string(), "request-1".to_string()), + ("result".to_string(), "success".to_string()), + ("traceparent".to_string(), "00-trace-parent".to_string()), + ]) + ); + + let process = events + .iter() + .find(|event| { + event + .fields + .get("event.name") + .is_some_and(|name| name == "codex.exec_server_process") + }) + .expect("process timing event"); + assert_eq!(process.level, Level::INFO); + assert_eq!(process.target, "codex_exec_server::telemetry"); + assert_eq!( + process.fields.get("message").map(String::as_str), + Some("exec-server process completed") + ); + assert_eq!( + process.fields.get("process_id").map(String::as_str), + Some("process-1") + ); + assert_eq!(process.fields.get("trace_id").map(String::as_str), Some("")); + assert_eq!( + process.fields.get("result").map(String::as_str), + Some("success") + ); + assert!(process.fields.contains_key("duration_seconds")); +} + +#[test] +fn disabled_info_events_do_not_capture_process_log_values() { + let subscriber = tracing_subscriber::registry() + .with(tracing_subscriber::filter::filter_fn(|_metadata| false)); + + tracing::subscriber::with_default(subscriber, || { + tracing::callsite::rebuild_interest_cache(); + let telemetry = ExecServerTelemetry::default(); + assert!(!telemetry.info_events_enabled()); + let process = telemetry.process_started("process-1"); + assert!(process.log_context.is_none()); + }); +} diff --git a/codex-rs/otel/src/events/session_telemetry.rs b/codex-rs/otel/src/events/session_telemetry.rs index 8210c0e5b9..0a3e552a66 100644 --- a/codex-rs/otel/src/events/session_telemetry.rs +++ b/codex-rs/otel/src/events/session_telemetry.rs @@ -1062,8 +1062,16 @@ impl SessionTelemetry { log_event!( self, event.name = "codex.tool_result", + trace_id = %crate::current_span_trace_id().unwrap_or_default(), + turn_id = "", tool_name = %tool_name, + call_id = "", + tool_source = "", + code_mode_cell_id = "", + code_mode_runtime_tool_call_id = "", + arguments = "", duration_ms = %Duration::ZERO.as_millis(), + duration_seconds = Duration::ZERO.as_secs_f64(), success = %false, output = %error, mcp_server = "", @@ -1072,8 +1080,15 @@ impl SessionTelemetry { trace_event!( self, event.name = "codex.tool_result", + trace_id = %crate::current_span_trace_id().unwrap_or_default(), + turn_id = "", tool_name = %tool_name, + call_id = "", + tool_source = "", + code_mode_cell_id = "", + code_mode_runtime_tool_call_id = "", duration_ms = %Duration::ZERO.as_millis(), + duration_seconds = Duration::ZERO.as_secs_f64(), success = %false, output_length = error.len() as i64, output_line_count = error.lines().count() as i64, @@ -1101,33 +1116,42 @@ impl SessionTelemetry { tags.extend_from_slice(extra_tags); self.counter(TOOL_CALL_COUNT_METRIC, /*inc*/ 1, &tags); self.record_duration(TOOL_CALL_DURATION_METRIC, duration, &tags); - let mcp_server = trace_field_value(extra_trace_fields, "mcp_server").unwrap_or(""); - let mcp_server_origin = - trace_field_value(extra_trace_fields, "mcp_server_origin").unwrap_or(""); log_event!( self, event.name = "codex.tool_result", + trace_id = %crate::current_span_trace_id().unwrap_or_default(), + turn_id = %trace_field_value(extra_trace_fields, "turn_id").unwrap_or(""), tool_name = %tool_name, call_id = %call_id, + tool_source = %trace_field_value(extra_trace_fields, "tool_source").unwrap_or(""), + code_mode_cell_id = %trace_field_value(extra_trace_fields, "code_mode_cell_id").unwrap_or(""), + code_mode_runtime_tool_call_id = %trace_field_value(extra_trace_fields, "code_mode_runtime_tool_call_id").unwrap_or(""), arguments = %arguments, duration_ms = %duration.as_millis(), + duration_seconds = duration.as_secs_f64(), success = %success_str, output = %output, - mcp_server = %mcp_server, - mcp_server_origin = %mcp_server_origin, + mcp_server = %trace_field_value(extra_trace_fields, "mcp_server").unwrap_or(""), + mcp_server_origin = %trace_field_value(extra_trace_fields, "mcp_server_origin").unwrap_or(""), ); trace_event!( self, event.name = "codex.tool_result", + trace_id = %crate::current_span_trace_id().unwrap_or_default(), + turn_id = %trace_field_value(extra_trace_fields, "turn_id").unwrap_or(""), tool_name = %tool_name, call_id = %call_id, + tool_source = %trace_field_value(extra_trace_fields, "tool_source").unwrap_or(""), + code_mode_cell_id = %trace_field_value(extra_trace_fields, "code_mode_cell_id").unwrap_or(""), + code_mode_runtime_tool_call_id = %trace_field_value(extra_trace_fields, "code_mode_runtime_tool_call_id").unwrap_or(""), duration_ms = %duration.as_millis(), + duration_seconds = duration.as_secs_f64(), success = %success_str, arguments_length = arguments.len() as i64, output_length = output.len() as i64, output_line_count = output.lines().count() as i64, - tool_origin = if mcp_server.is_empty() { "builtin" } else { "mcp" }, - mcp_tool = !mcp_server.is_empty(), + tool_origin = if trace_field_value(extra_trace_fields, "mcp_server").unwrap_or("").is_empty() { "builtin" } else { "mcp" }, + mcp_tool = !trace_field_value(extra_trace_fields, "mcp_server").unwrap_or("").is_empty(), ); } diff --git a/codex-rs/otel/tests/suite/otel_export_routing_policy.rs b/codex-rs/otel/tests/suite/otel_export_routing_policy.rs index 17b6f764db..90f18d54f8 100644 --- a/codex-rs/otel/tests/suite/otel_export_routing_policy.rs +++ b/codex-rs/otel/tests/suite/otel_export_routing_policy.rs @@ -255,6 +255,10 @@ fn otel_export_routing_policy_routes_tool_result_log_and_trace_events() { &[ ("mcp_server", "internal-mcp"), ("mcp_server_origin", "stdio"), + ("turn_id", "turn-1"), + ("tool_source", "code_mode"), + ("code_mode_cell_id", "cell-1"), + ("code_mode_runtime_tool_call_id", "runtime-call-1"), ], ); }); @@ -286,6 +290,29 @@ fn otel_export_routing_policy_routes_tool_result_log_and_trace_events() { tool_log_attrs.get("mcp_server_origin").map(String::as_str), Some("stdio") ); + assert_eq!( + tool_log_attrs.get("turn_id").map(String::as_str), + Some("turn-1") + ); + assert_eq!( + tool_log_attrs.get("tool_source").map(String::as_str), + Some("code_mode") + ); + assert_eq!( + tool_log_attrs.get("code_mode_cell_id").map(String::as_str), + Some("cell-1") + ); + assert_eq!( + tool_log_attrs + .get("code_mode_runtime_tool_call_id") + .map(String::as_str), + Some("runtime-call-1") + ); + assert_eq!( + tool_log_attrs.get("duration_seconds").map(String::as_str), + Some("0.042") + ); + assert!(tool_log_attrs.contains_key("trace_id")); let spans = span_exporter.get_finished_spans().expect("span export"); assert_eq!(spans.len(), 1);