From 9ef08dcf2bda8ba57c569bc69dd079b4643805f2 Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Fri, 25 Sep 2026 02:24:54 +0000 Subject: [PATCH] Add multi-agent spawn latency and failure metrics (#47984) ## What changed - Record successful spawn phase durations in `codex.multi_agent.spawn.phase.duration_ms`, covering residency reservation and fork context when applicable, child creation, durability waits, input admission, and total latency. - Count spawn failures from the V1 and V2 tool handlers in `codex.multi_agent.spawn.failure`, grouped by bounded error categories. - Use bounded labels for fork mode, multi-agent version, product SKU, and successful-spawn history mode. Emit phase timings only after a successful spawn. ## Testing Add in-memory metrics tests for phase durations and labels, plus a tool-handler test verifying that a thread-limit failure emits one counter increment with the expected labels. GitOrigin-RevId: 04cba9042b9b34acdf57376ace5288c3f741f3b2 --- codex-rs/core/src/agent/control.rs | 1 + codex-rs/core/src/agent/control/spawn.rs | 90 +++++++++++++++--- .../core/src/agent/control/spawn_telemetry.rs | 65 +++++++++++++ .../agent/control/spawn_telemetry_tests.rs | 95 +++++++++++++++++++ .../src/tools/handlers/multi_agents/spawn.rs | 17 +++- .../src/tools/handlers/multi_agents_common.rs | 33 +++++++ .../src/tools/handlers/multi_agents_tests.rs | 75 +++++++++++++++ .../tools/handlers/multi_agents_v2/spawn.rs | 13 ++- codex-rs/otel/src/events/session_telemetry.rs | 51 +++++++++- codex-rs/otel/src/metrics/names.rs | 3 + 10 files changed, 425 insertions(+), 18 deletions(-) create mode 100644 codex-rs/core/src/agent/control/spawn_telemetry.rs create mode 100644 codex-rs/core/src/agent/control/spawn_telemetry_tests.rs diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index 719336828f..705d48c1d5 100644 --- a/codex-rs/core/src/agent/control.rs +++ b/codex-rs/core/src/agent/control.rs @@ -77,6 +77,7 @@ mod sender_context; mod service_tier; mod spawn; mod spawn_guard; +mod spawn_telemetry; mod target; mod user_authorization; mod watch; diff --git a/codex-rs/core/src/agent/control/spawn.rs b/codex-rs/core/src/agent/control/spawn.rs index 7fe302678d..9b642981a3 100644 --- a/codex-rs/core/src/agent/control/spawn.rs +++ b/codex-rs/core/src/agent/control/spawn.rs @@ -1,5 +1,7 @@ use super::residency::is_v2_resident_session_source; use super::spawn_guard::PendingSpawn; +use super::spawn_telemetry::SpawnMeasurements; +use super::spawn_telemetry::record_spawn_success; use super::*; use crate::agent::child_config::build_agent_resume_config; use crate::agent::role::apply_role_to_config; @@ -31,6 +33,8 @@ use codex_thread_store::PersistContext; use codex_utils_path_uri::PathUri; use futures::StreamExt; use futures::stream; +use std::time::Duration; +use std::time::Instant; const AGENT_NAMES: &str = include_str!("../../../assets/agent/agent_names.txt"); @@ -39,6 +43,12 @@ struct SpawnAgentThreadInheritance { exec_policy: Option>, } +struct SpawnedThreadResult { + new_thread: crate::thread_manager::NewThread, + fork_context: Option, + child_create: Duration, +} + /// Initial input delivered after a spawned agent acquires execution capacity. /// /// V2 communication spawns keep the communication and its context paired so centralized @@ -624,6 +634,7 @@ impl LocalAgentControl { session_source: Option, options: SpawnAgentOptions, ) -> CodexResult<(LiveAgent, ThreadConfigSnapshot)> { + let spawn_started_at = Instant::now(); let state = self.runtime.upgrade()?; let multi_agent_version = state .effective_multi_agent_version_for_spawn( @@ -634,6 +645,7 @@ impl LocalAgentControl { &config, ) .await; + let product_sku = config.apps_mcp_product_sku.clone(); if let Some(session_source) = session_source.as_ref() { self.ensure_execution_capacity(multi_agent_version, session_source)?; } @@ -642,13 +654,17 @@ impl LocalAgentControl { && session_source .as_ref() .is_some_and(is_v2_resident_session_source); - let residency_slot = if spawn_uses_v2_residency { - Some( - self.reserve_v2_residency_slot(&state, &config, /*protected_thread_id*/ None) - .await?, + let (residency_slot, residency_reservation) = if spawn_uses_v2_residency { + let residency_reservation_started_at = Instant::now(); + let residency_slot = self + .reserve_v2_residency_slot(&state, &config, /*protected_thread_id*/ None) + .await?; + ( + Some(residency_slot), + Some(residency_reservation_started_at.elapsed()), ) } else { - None + (None, None) }; let reservation_max_threads = if spawn_uses_v2_residency { None @@ -691,7 +707,11 @@ impl LocalAgentControl { let notification_source = session_source.clone(); // The same `LocalAgentControl` is sent to spawn the thread. - let new_thread = match (session_source, options.fork_mode.as_ref(), inheritance) { + let SpawnedThreadResult { + new_thread, + fork_context, + child_create, + } = match (session_source, options.fork_mode.as_ref(), inheritance) { (Some(session_source), Some(_), inheritance) => { Box::pin(self.spawn_forked_thread( &state, @@ -715,7 +735,8 @@ impl LocalAgentControl { } else { None }; - Box::pin(state.spawn_new_thread_with_source( + let child_create_started_at = Instant::now(); + let new_thread = Box::pin(state.spawn_new_thread_with_source( config.clone(), self.clone(), session_source, @@ -728,9 +749,23 @@ impl LocalAgentControl { inheritance.exec_policy, options.environments.clone(), )) - .await? + .await?; + SpawnedThreadResult { + new_thread, + fork_context: None, + child_create: child_create_started_at.elapsed(), + } + } + (None, _, _) => { + let child_create_started_at = Instant::now(); + let new_thread = + Box::pin(state.spawn_new_thread(config.clone(), self.clone())).await?; + SpawnedThreadResult { + new_thread, + fork_context: None, + child_create: child_create_started_at.elapsed(), + } } - (None, _, _) => Box::pin(state.spawn_new_thread(config.clone(), self.clone())).await?, }; agent_metadata.agent_id = Some(new_thread.thread_id); let mut pending_spawn = PendingSpawn::new(Arc::clone(&state), new_thread.thread_id); @@ -781,6 +816,7 @@ impl LocalAgentControl { ) .await; })); + let durability_wait_started_at = Instant::now(); if options.fork_mode.is_some() { tokio::join!( new_thread @@ -792,6 +828,7 @@ impl LocalAgentControl { } else { pending_spawn.wait_for_edge().await; } + let durability_wait = durability_wait_started_at.elapsed(); let start_options = TurnStartOptions { parent_turn_id: options.parent_turn_id, @@ -800,6 +837,7 @@ impl LocalAgentControl { cyber_access_program: options.cyber_access_program, ..Default::default() }; + let input_admission_started_at = Instant::now(); match initial_input { SpawnInitialInput::UserInput(input) => { self.send_input(new_thread.thread_id, input, start_options) @@ -816,6 +854,7 @@ impl LocalAgentControl { .await?; } } + let input_admission = input_admission_started_at.elapsed(); reservation.commit(agent_metadata.clone()); if let Some(residency_slot) = residency_slot { residency_slot.commit(new_thread.thread_id); @@ -846,6 +885,24 @@ impl LocalAgentControl { status: self.get_status(new_thread.thread_id).await, }; let config = new_thread.thread.config_snapshot().await; + let session_telemetry = new_thread + .thread + .session_telemetry() + .with_product_sku(product_sku.as_deref()); + record_spawn_success( + &session_telemetry, + options.fork_mode.as_ref(), + multi_agent_version, + SpawnMeasurements { + history_mode: config.history_mode, + residency_reservation, + fork_context, + child_create, + durability_wait, + input_admission, + total: spawn_started_at.elapsed(), + }, + ); Ok((agent, config)) } @@ -857,7 +914,7 @@ impl LocalAgentControl { options: &SpawnAgentOptions, inheritance: SpawnAgentThreadInheritance, multi_agent_version: MultiAgentVersion, - ) -> CodexResult { + ) -> CodexResult { let SpawnAgentThreadInheritance { environments: inherited_environments, exec_policy: inherited_exec_policy, @@ -881,6 +938,7 @@ impl LocalAgentControl { )); }; + let fork_context_started_at = Instant::now(); let parent_thread_id = *parent_thread_id; let parent_thread = state.get_thread(parent_thread_id).await?; let (subagent_developer_instructions, parent_developer_instructions) = match ( @@ -1157,7 +1215,9 @@ impl LocalAgentControl { let mut thread_extension_init = ExtensionDataInit::new(); thread_extension_init.insert(selected_capability_roots); - state + let fork_context = fork_context_started_at.elapsed(); + let child_create_started_at = Instant::now(); + let new_thread = state .fork_thread_with_source( config.clone(), InitialHistory::Forked(forked_rollout_items), @@ -1172,7 +1232,13 @@ impl LocalAgentControl { options.environments.clone(), thread_extension_init, ) - .await + .await?; + let child_create = child_create_started_at.elapsed(); + Ok(SpawnedThreadResult { + new_thread, + fork_context: Some(fork_context), + child_create, + }) } /// Resume an existing agent thread from a recorded rollout file. diff --git a/codex-rs/core/src/agent/control/spawn_telemetry.rs b/codex-rs/core/src/agent/control/spawn_telemetry.rs new file mode 100644 index 0000000000..7b681ce9c7 --- /dev/null +++ b/codex-rs/core/src/agent/control/spawn_telemetry.rs @@ -0,0 +1,65 @@ +//! Successful multi-agent spawn latency metrics. +//! +//! Runtime values are collapsed to bounded labels before emission. Failed spawns do not emit +//! partial phase timings. + +use crate::agent::types::SpawnAgentForkMode; +use codex_otel::SessionTelemetry; +use codex_protocol::protocol::MultiAgentVersion; +use codex_protocol::protocol::ThreadHistoryMode; +use std::time::Duration; + +pub(super) struct SpawnMeasurements { + pub(super) history_mode: ThreadHistoryMode, + pub(super) residency_reservation: Option, + pub(super) fork_context: Option, + pub(super) child_create: Duration, + pub(super) durability_wait: Duration, + pub(super) input_admission: Duration, + pub(super) total: Duration, +} + +pub(super) fn record_spawn_success( + telemetry: &SessionTelemetry, + fork_mode: Option<&SpawnAgentForkMode>, + multi_agent_version: MultiAgentVersion, + measurements: SpawnMeasurements, +) { + let fork_mode = match fork_mode { + None => "none", + Some(SpawnAgentForkMode::FullHistory) => "all", + Some(SpawnAgentForkMode::LastNTurns(_)) => "last_n", + }; + let history_mode = match measurements.history_mode { + ThreadHistoryMode::Legacy => "legacy", + ThreadHistoryMode::Paginated => "paginated", + }; + let multi_agent_version = match multi_agent_version { + MultiAgentVersion::Disabled => "disabled", + MultiAgentVersion::V1 => "v1", + MultiAgentVersion::V2 => "v2", + }; + let record_phase = |phase, duration| { + telemetry.record_multi_agent_spawn_phase( + phase, + duration, + fork_mode, + history_mode, + multi_agent_version, + ); + }; + if let Some(duration) = measurements.residency_reservation { + record_phase("residency_reservation", duration); + } + if let Some(duration) = measurements.fork_context { + record_phase("fork_context", duration); + } + record_phase("child_create", measurements.child_create); + record_phase("durability_wait", measurements.durability_wait); + record_phase("input_admission", measurements.input_admission); + record_phase("total", measurements.total); +} + +#[cfg(test)] +#[path = "spawn_telemetry_tests.rs"] +mod tests; diff --git a/codex-rs/core/src/agent/control/spawn_telemetry_tests.rs b/codex-rs/core/src/agent/control/spawn_telemetry_tests.rs new file mode 100644 index 0000000000..410ace28d9 --- /dev/null +++ b/codex-rs/core/src/agent/control/spawn_telemetry_tests.rs @@ -0,0 +1,95 @@ +use super::*; +use codex_otel::MULTI_AGENT_SPAWN_PHASE_DURATION_METRIC; +use codex_otel::MetricsClient; +use codex_otel::MetricsConfig; +use codex_protocol::ThreadId; +use codex_protocol::protocol::SessionSource; +use opentelemetry_sdk::metrics::InMemoryMetricExporter; +use opentelemetry_sdk::metrics::data::AggregatedMetrics; +use opentelemetry_sdk::metrics::data::MetricData; +use pretty_assertions::assert_eq; +use std::collections::BTreeMap; + +#[test] +fn records_bounded_spawn_phase_metrics() { + let metrics = MetricsClient::new( + MetricsConfig::in_memory( + "test", + "codex-core", + env!("CARGO_PKG_VERSION"), + InMemoryMetricExporter::default(), + ) + .with_runtime_reader(), + ) + .expect("create in-memory metrics client"); + let telemetry = SessionTelemetry::new( + ThreadId::new(), + "test-model", + "test-model", + /*account_id*/ None, + /*account_email*/ None, + /*auth_mode*/ None, + "test-originator".to_string(), + /*log_user_prompts*/ false, + "test-terminal".to_string(), + SessionSource::Cli, + ) + .with_product_sku(Some("codex")) + .with_metrics_without_metadata_tags(metrics.clone()); + + record_spawn_success( + &telemetry, + Some(&SpawnAgentForkMode::LastNTurns(3)), + MultiAgentVersion::V2, + SpawnMeasurements { + history_mode: ThreadHistoryMode::Paginated, + residency_reservation: Some(Duration::from_millis(11)), + fork_context: Some(Duration::from_millis(13)), + child_create: Duration::from_millis(17), + durability_wait: Duration::from_millis(19), + input_admission: Duration::from_millis(23), + total: Duration::from_millis(29), + }, + ); + + let snapshot = metrics.snapshot().expect("snapshot spawn metrics"); + let metric = snapshot + .scope_metrics() + .flat_map(opentelemetry_sdk::metrics::data::ScopeMetrics::metrics) + .find(|metric| metric.name() == MULTI_AGENT_SPAWN_PHASE_DURATION_METRIC) + .expect("spawn phase metric"); + let AggregatedMetrics::F64(MetricData::Histogram(histogram)) = metric.data() else { + panic!("expected spawn phase histogram"); + }; + let phases = histogram + .data_points() + .map(|point| { + let attributes = point + .attributes() + .map(|attribute| { + ( + attribute.key.as_str().to_string(), + attribute.value.as_str().to_string(), + ) + }) + .collect::>(); + assert_eq!(attributes.len(), 5); + assert_eq!(attributes["fork_mode"], "last_n"); + assert_eq!(attributes["history_mode"], "paginated"); + assert_eq!(attributes["multi_agent_version"], "v2"); + assert_eq!(attributes["product_sku"], "codex"); + (attributes["phase"].clone(), (point.count(), point.sum())) + }) + .collect::>(); + assert_eq!( + phases, + BTreeMap::from([ + ("child_create".to_string(), (1, 17.0)), + ("durability_wait".to_string(), (1, 19.0)), + ("fork_context".to_string(), (1, 13.0)), + ("input_admission".to_string(), (1, 23.0)), + ("residency_reservation".to_string(), (1, 11.0)), + ("total".to_string(), (1, 29.0)), + ]) + ); +} diff --git a/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs b/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs index 0961db39f8..9bfc5571c4 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs @@ -12,6 +12,7 @@ use crate::agent::types::SpawnAgentForkMode; use crate::agent::types::SpawnAgentOptions; use crate::tools::handlers::multi_agents_spec::SpawnAgentToolOptions; use crate::tools::handlers::multi_agents_spec::create_spawn_agent_tool_v1; +use codex_protocol::protocol::MultiAgentVersion; use codex_tools::ToolSpec; #[derive(Default)] @@ -108,6 +109,7 @@ async fn handle_spawn_agent( .await .map_err(FunctionCallError::RespondToModel)?; let config = prepared.config; + let fork_mode = args.fork_context.then_some(SpawnAgentForkMode::FullHistory); let result = session .services .agent_control @@ -123,8 +125,8 @@ async fn handle_spawn_agent( /*task_name*/ None, )?, options: SpawnAgentOptions { - fork_parent_spawn_call_id: args.fork_context.then(|| call_id.clone()), - fork_mode: args.fork_context.then_some(SpawnAgentForkMode::FullHistory), + fork_parent_spawn_call_id: fork_mode.as_ref().map(|_| call_id.clone()), + fork_mode: fork_mode.clone(), parent_thread_id: Some(session.thread_id), parent_turn_id: Some(turn.sub_id.clone()), root_turn_id: turn.turn_metadata_state.root_turn_id(), @@ -135,7 +137,16 @@ async fn handle_spawn_agent( }, }) .await - .map_err(collab_spawn_error); + .map_err(|err| { + record_collab_spawn_failure( + &turn.session_telemetry, + turn.config.apps_mcp_product_sku.as_deref(), + &err, + fork_mode.as_ref(), + MultiAgentVersion::V1, + ); + collab_spawn_error(err) + }); let (new_thread_id, status) = match &result { Ok((spawned_agent, _)) => (Some(spawned_agent.thread_id), spawned_agent.status.clone()), Err(_) => (None, AgentStatus::NotFound), 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 4f8f9072d0..be17be61b3 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_common.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_common.rs @@ -1,14 +1,17 @@ +use crate::agent::types::SpawnAgentForkMode; use crate::config::DEFAULT_MULTI_AGENT_V2_MIN_WAIT_TIMEOUT_MS; use crate::config::HARD_MAX_MULTI_AGENT_V2_TIMEOUT_MS; use crate::function_tool::FunctionCallError; use crate::tools::context::FunctionToolOutput; use crate::tools::context::ToolOutput; use crate::tools::context::ToolPayload; +use codex_otel::SessionTelemetry; use codex_protocol::AgentPath; use codex_protocol::ThreadId; use codex_protocol::error::CodexErr; use codex_protocol::error::CodexErrorDetails; use codex_protocol::models::ResponseInputItem; +use codex_protocol::protocol::MultiAgentVersion; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SubAgentSource; use codex_protocol::user_input::UserInput; @@ -72,6 +75,36 @@ pub(crate) fn collab_spawn_error(err: CodexErr) -> FunctionCallError { } } +pub(crate) fn record_collab_spawn_failure( + telemetry: &SessionTelemetry, + product_sku: Option<&str>, + err: &CodexErr, + fork_mode: Option<&SpawnAgentForkMode>, + multi_agent_version: MultiAgentVersion, +) { + let reason = match err.details() { + CodexErrorDetails::AgentLimitReached { .. } => "limit_reached", + CodexErrorDetails::InvalidRequest(_) => "invalid_request", + CodexErrorDetails::ThreadNotFound(_) => "thread_not_found", + CodexErrorDetails::UnsupportedOperation(_) => "unsupported_operation", + _ => "internal", + }; + let fork_mode = match fork_mode { + None => "none", + Some(SpawnAgentForkMode::FullHistory) => "all", + Some(SpawnAgentForkMode::LastNTurns(_)) => "last_n", + }; + let multi_agent_version = match multi_agent_version { + MultiAgentVersion::Disabled => "disabled", + MultiAgentVersion::V1 => "v1", + MultiAgentVersion::V2 => "v2", + }; + telemetry + .clone() + .with_product_sku(product_sku) + .record_multi_agent_spawn_failure(reason, fork_mode, multi_agent_version); +} + pub(crate) fn collab_agent_error(agent_id: ThreadId, err: CodexErr) -> FunctionCallError { match err.details() { CodexErrorDetails::ThreadNotFound(id) => { 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 98360657ed..9c8268c542 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_tests.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_tests.rs @@ -36,6 +36,9 @@ use codex_login::CodexAuth; use codex_model_provider::create_model_provider; use codex_model_provider_info::built_in_model_providers; use codex_models_manager::manager::StaticModelsManager; +use codex_otel::MULTI_AGENT_SPAWN_FAILURE_METRIC; +use codex_otel::MetricsClient; +use codex_otel::MetricsConfig; use codex_protocol::AgentPath; use codex_protocol::ThreadId; use codex_protocol::config_types::ApprovalsReviewer; @@ -73,9 +76,13 @@ use codex_protocol::protocol::TurnCompleteEvent; use codex_protocol::user_input::UserInput; use codex_state::DirectionalThreadSpawnEdgeStatus; use core_test_support::TempDirExt; +use opentelemetry_sdk::metrics::InMemoryMetricExporter; +use opentelemetry_sdk::metrics::data::AggregatedMetrics; +use opentelemetry_sdk::metrics::data::MetricData; use pretty_assertions::assert_eq; use serde::Deserialize; use serde_json::json; +use std::collections::BTreeMap; use std::collections::HashMap; use std::path::PathBuf; use std::sync::Arc; @@ -294,6 +301,74 @@ async fn spawn_agent_rejects_when_message_and_items_are_both_set() { ); } +#[tokio::test] +async fn spawn_agent_limit_failure_emits_bounded_metric() { + let metrics = MetricsClient::new( + MetricsConfig::in_memory( + "test", + "codex-core", + env!("CARGO_PKG_VERSION"), + InMemoryMetricExporter::default(), + ) + .with_runtime_reader(), + ) + .expect("create in-memory metrics client"); + let (mut session, mut turn) = make_session_and_context().await; + turn.session_telemetry = turn.session_telemetry.clone().with_metrics(metrics.clone()); + let mut config = (*turn.config).clone(); + config.agent_max_threads = Some(0); + config.apps_mcp_product_sku = Some("codex".to_string()); + turn.config = Arc::new(config); + let manager = thread_manager(); + set_agent_control(&mut session, manager.agent_control()); + + let Err(err) = SpawnAgentHandler::default() + .handle(invocation( + Arc::new(session), + Arc::new(turn), + "spawn_agent", + function_payload(json!({"message": "inspect this repo"})), + )) + .await + else { + panic!("spawn_agent should respect the thread limit"); + }; + assert_eq!( + err, + FunctionCallError::RespondToModel( + "collab spawn failed: agent thread limit reached".to_string() + ) + ); + + let snapshot = metrics.snapshot().expect("snapshot spawn metrics"); + let failure_metric = snapshot + .scope_metrics() + .flat_map(opentelemetry_sdk::metrics::data::ScopeMetrics::metrics) + .find(|metric| metric.name() == MULTI_AGENT_SPAWN_FAILURE_METRIC) + .expect("spawn failure metric"); + let AggregatedMetrics::U64(MetricData::Sum(sum)) = failure_metric.data() else { + panic!("expected spawn failure counter"); + }; + let points = sum.data_points().collect::>(); + assert_eq!(points.len(), 1); + assert_eq!(points[0].value(), 1); + assert_eq!( + points[0] + .attributes() + .map(|attribute| ( + attribute.key.as_str().to_string(), + attribute.value.as_str().to_string(), + )) + .collect::>(), + BTreeMap::from([ + ("fork_mode".to_string(), "none".to_string()), + ("multi_agent_version".to_string(), "v1".to_string()), + ("product_sku".to_string(), "codex".to_string()), + ("reason".to_string(), "limit_reached".to_string()), + ]) + ); +} + #[tokio::test] async fn spawn_agent_uses_explorer_role_and_preserves_approval_policy() { #[derive(Debug, Deserialize)] 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 9f4c0c24ff..ff00a422e6 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 @@ -194,7 +194,7 @@ async fn handle_spawn_agent( source: spawn_source, options: SpawnAgentOptions { fork_parent_spawn_call_id: fork_mode.as_ref().map(|_| call_id.clone()), - fork_mode, + fork_mode: fork_mode.clone(), parent_thread_id: Some(session.thread_id), parent_turn_id: Some(turn.sub_id.clone()), root_turn_id: turn.turn_metadata_state.root_turn_id(), @@ -205,7 +205,16 @@ async fn handle_spawn_agent( }, }) .await - .map_err(collab_spawn_error)?; + .map_err(|err| { + record_collab_spawn_failure( + &turn.session_telemetry, + turn.config.apps_mcp_product_sku.as_deref(), + &err, + fork_mode.as_ref(), + MultiAgentVersion::V2, + ); + collab_spawn_error(err) + })?; let new_thread_id = spawned_agent.thread_id; let agent_status = spawned_agent.status; let nickname = agent_snapshot diff --git a/codex-rs/otel/src/events/session_telemetry.rs b/codex-rs/otel/src/events/session_telemetry.rs index 01789e3c3e..33f2c82037 100644 --- a/codex-rs/otel/src/events/session_telemetry.rs +++ b/codex-rs/otel/src/events/session_telemetry.rs @@ -5,6 +5,8 @@ use crate::events::shared::log_event; use crate::events::shared::trace_event; use crate::metrics::API_CALL_COUNT_METRIC; use crate::metrics::API_CALL_DURATION_METRIC; +use crate::metrics::MULTI_AGENT_SPAWN_FAILURE_METRIC; +use crate::metrics::MULTI_AGENT_SPAWN_PHASE_DURATION_METRIC; use crate::metrics::MetricsClient; use crate::metrics::MetricsConfig; use crate::metrics::MetricsError; @@ -152,7 +154,7 @@ impl SessionTelemetry { self } - /// Attributes tool telemetry without turning arbitrary configuration into metric labels. + /// Attributes bounded telemetry without turning arbitrary configuration into metric labels. pub fn with_product_sku(mut self, product_sku: Option<&str>) -> Self { const KNOWN_PRODUCT_SKUS: &[&str] = &["codex"]; @@ -261,6 +263,53 @@ impl SessionTelemetry { } } + /// Records one successful multi-agent spawn phase with bounded spawn dimensions. + pub fn record_multi_agent_spawn_phase( + &self, + phase: &'static str, + duration: Duration, + fork_mode: &'static str, + history_mode: &'static str, + multi_agent_version: &'static str, + ) { + let Some(metrics) = &self.metrics else { + return; + }; + let mut tags = self.multi_agent_spawn_tags(fork_mode, multi_agent_version); + tags.push(("history_mode", history_mode)); + tags.push(("phase", phase)); + let _ = metrics.record_duration(MULTI_AGENT_SPAWN_PHASE_DURATION_METRIC, duration, &tags); + } + + /// Records one failed multi-agent spawn after request validation has completed. + pub fn record_multi_agent_spawn_failure( + &self, + reason: &'static str, + fork_mode: &'static str, + multi_agent_version: &'static str, + ) { + let Some(metrics) = &self.metrics else { + return; + }; + let mut tags = self.multi_agent_spawn_tags(fork_mode, multi_agent_version); + tags.push(("reason", reason)); + let _ = metrics.counter(MULTI_AGENT_SPAWN_FAILURE_METRIC, /*inc*/ 1, &tags); + } + + fn multi_agent_spawn_tags( + &self, + fork_mode: &'static str, + multi_agent_version: &'static str, + ) -> Vec<(&'static str, &'static str)> { + let mut tags = Vec::with_capacity(5); + tags.push(("fork_mode", fork_mode)); + tags.push(("multi_agent_version", multi_agent_version)); + if let Some(product_sku) = self.metadata.product_sku { + tags.push(("product_sku", product_sku)); + } + tags + } + fn record_duration_ms_f64(&self, name: &str, duration_ms: f64, tags: &[(&str, &str)]) { let res: MetricsResult<()> = (|| { let Some(metrics) = &self.metrics else { diff --git a/codex-rs/otel/src/metrics/names.rs b/codex-rs/otel/src/metrics/names.rs index f27d77a27b..fcc15025dd 100644 --- a/codex-rs/otel/src/metrics/names.rs +++ b/codex-rs/otel/src/metrics/names.rs @@ -1,6 +1,9 @@ pub const TOOL_CALL_COUNT_METRIC: &str = "codex.tool.call"; pub const TOOL_CALL_DURATION_METRIC: &str = "codex.tool.call.duration_ms"; pub const TOOL_CALL_UNIFIED_EXEC_METRIC: &str = "codex.tool.unified_exec"; +pub const MULTI_AGENT_SPAWN_FAILURE_METRIC: &str = "codex.multi_agent.spawn.failure"; +pub const MULTI_AGENT_SPAWN_PHASE_DURATION_METRIC: &str = + "codex.multi_agent.spawn.phase.duration_ms"; pub const ARTIFACT_OPERATION_STARTED_METRIC: &str = "codex.artifact.operation.started"; pub const ARTIFACT_OPERATION_EXPECTED_OUTPUT_COUNT_METRIC: &str = "codex.artifact.operation.expected_output_count";