openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-10 21:00:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add multi-agent spawn latency and failure metrics (#47984)

Browse Source 
## 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
This commit is contained in:
Owen Lin
2026-09-25 02:40:55 +00:00
committed by copyberry
parent 75e0e0aad9
commit 9ef08dcf2b
10 changed files with 425 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/agent/control/spawn.rs
+78 -12
View File
@@ -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<Arc<crate::exec_policy::ExecPolicyManager>>,
}
struct SpawnedThreadResult {
new_thread: crate::thread_manager::NewThread,
fork_context: Option<Duration>,
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<SessionSource>,
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<crate::thread_manager::NewThread> {
) -> CodexResult<SpawnedThreadResult> {
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.
codex-rs/core/src/agent/control/spawn_telemetry.rs
+65
View File
@@ -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<Duration>,
pub(super) fork_context: Option<Duration>,
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;
codex-rs/core/src/agent/control/spawn_telemetry_tests.rs
+95
View File
@@ -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::<BTreeMap<_, _>>();
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::<BTreeMap<_, _>>();
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)),
])
);
}