From 3f40c4eabff02ddade0d6683dcaebc6bbad68078 Mon Sep 17 00:00:00 2001 From: Ahmed Ibrahim Date: Thu, 28 May 2026 11:40:44 -0700 Subject: [PATCH] Track timing for every thread initialization --- .../analytics/src/analytics_client_tests.rs | 70 +++++++++++++++---- codex-rs/analytics/src/client.rs | 6 +- codex-rs/analytics/src/events.rs | 5 +- codex-rs/analytics/src/facts.rs | 4 +- codex-rs/analytics/src/lib.rs | 2 +- codex-rs/analytics/src/reducer.rs | 46 ++++-------- .../app-server/tests/suite/v2/analytics.rs | 10 +++ .../app-server/tests/suite/v2/thread_start.rs | 8 --- codex-rs/core/src/lib.rs | 2 +- ...ing.rs => thread_initialization_timing.rs} | 10 +-- codex-rs/core/src/thread_manager.rs | 20 +++--- 11 files changed, 106 insertions(+), 77 deletions(-) rename codex-rs/core/src/{thread_start_timing.rs => thread_initialization_timing.rs} (85%) diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index a13cfc2025..74d57c60fd 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -66,7 +66,7 @@ use crate::facts::SkillInvocation; use crate::facts::SkillInvokedInput; use crate::facts::SubAgentThreadStartedInput; use crate::facts::ThreadInitializationMode; -use crate::facts::ThreadStartTimingFact; +use crate::facts::ThreadInitializationTimingFact; use crate::facts::TrackEventsContext; use crate::facts::TurnResolvedConfigFact; use crate::facts::TurnStatus; @@ -1751,15 +1751,15 @@ async fn app_server_started_fact_emits_event() { } #[tokio::test] -async fn thread_start_timing_fact_enriches_thread_initialized_event() { +async fn thread_initialization_timing_fact_enriches_initialized_event() { let mut reducer = AnalyticsReducer::default(); let mut events = Vec::new(); ingest_initialize(&mut reducer, &mut events).await; reducer .ingest( - AnalyticsFact::Custom(CustomAnalyticsFact::ThreadStartTiming( - ThreadStartTimingFact { + AnalyticsFact::Custom(CustomAnalyticsFact::ThreadInitializationTiming( + ThreadInitializationTimingFact { thread_id: "thread-1".to_string(), duration_ms: 222, prepare_duration_ms: 12, @@ -1784,13 +1784,21 @@ async fn thread_start_timing_fact_enriches_thread_initialized_event() { .await; let payload = serde_json::to_value(&events).expect("serialize events"); - assert_eq!(payload[0]["event_type"], json!("codex_thread_initialized")); - assert_eq!(payload[0]["event_params"]["duration_ms"], json!(222)); - assert_eq!(payload[0]["event_params"]["prepare_duration_ms"], json!(12)); - assert_eq!(payload[0]["event_params"]["spawn_duration_ms"], json!(123)); assert_eq!( - payload[0]["event_params"]["finalize_duration_ms"], - json!(87) + json!({ + "event_type": payload[0]["event_type"], + "duration_ms": payload[0]["event_params"]["duration_ms"], + "prepare_duration_ms": payload[0]["event_params"]["prepare_duration_ms"], + "spawn_duration_ms": payload[0]["event_params"]["spawn_duration_ms"], + "finalize_duration_ms": payload[0]["event_params"]["finalize_duration_ms"], + }), + json!({ + "event_type": "codex_thread_initialized", + "duration_ms": 222, + "prepare_duration_ms": 12, + "spawn_duration_ms": 123, + "finalize_duration_ms": 87, + }) ); } @@ -2567,6 +2575,7 @@ fn subagent_thread_started_review_serializes_expected_shape() { subagent_source: SubAgentSource::Review, created_at: 123, }, + ThreadInitializationTimingParams::default(), )); let payload = serde_json::to_value(&event).expect("serialize review subagent event"); @@ -2617,6 +2626,7 @@ fn subagent_thread_started_thread_spawn_serializes_parent_thread_id() { }, created_at: 124, }, + ThreadInitializationTimingParams::default(), )); let payload = serde_json::to_value(&event).expect("serialize thread spawn subagent event"); @@ -2645,6 +2655,7 @@ fn subagent_thread_started_memory_consolidation_serializes_expected_shape() { subagent_source: SubAgentSource::MemoryConsolidation, created_at: 125, }, + ThreadInitializationTimingParams::default(), )); let payload = @@ -2671,6 +2682,7 @@ fn subagent_thread_started_other_serializes_expected_shape() { subagent_source: SubAgentSource::Other("guardian".to_string()), created_at: 126, }, + ThreadInitializationTimingParams::default(), )); let payload = serde_json::to_value(&event).expect("serialize other subagent event"); @@ -2693,6 +2705,7 @@ fn subagent_thread_started_other_serializes_explicit_parent_thread_id() { subagent_source: SubAgentSource::Other("guardian".to_string()), created_at: 126, }, + ThreadInitializationTimingParams::default(), )); let payload = serde_json::to_value(&event).expect("serialize auto-review subagent event"); @@ -2708,6 +2721,20 @@ async fn subagent_thread_started_publishes_without_initialize() { let mut reducer = AnalyticsReducer::default(); let mut events = Vec::new(); + reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::ThreadInitializationTiming( + ThreadInitializationTimingFact { + thread_id: "thread-review".to_string(), + duration_ms: 222, + prepare_duration_ms: 12, + spawn_duration_ms: 123, + finalize_duration_ms: 87, + }, + )), + &mut events, + ) + .await; reducer .ingest( AnalyticsFact::Custom(CustomAnalyticsFact::SubAgentThreadStarted( @@ -2730,13 +2757,30 @@ async fn subagent_thread_started_publishes_without_initialize() { let payload = serde_json::to_value(&events).expect("serialize events"); assert_eq!(payload.as_array().expect("events array").len(), 1); - assert_eq!(payload[0]["event_type"], "codex_thread_initialized"); assert_eq!( payload[0]["event_params"]["app_server_client"]["product_client_id"], "codex-tui" ); - assert_eq!(payload[0]["event_params"]["thread_source"], "subagent"); - assert_eq!(payload[0]["event_params"]["subagent_source"], "review"); + assert_eq!( + json!({ + "event_type": payload[0]["event_type"], + "thread_source": payload[0]["event_params"]["thread_source"], + "subagent_source": payload[0]["event_params"]["subagent_source"], + "duration_ms": payload[0]["event_params"]["duration_ms"], + "prepare_duration_ms": payload[0]["event_params"]["prepare_duration_ms"], + "spawn_duration_ms": payload[0]["event_params"]["spawn_duration_ms"], + "finalize_duration_ms": payload[0]["event_params"]["finalize_duration_ms"], + }), + json!({ + "event_type": "codex_thread_initialized", + "thread_source": "subagent", + "subagent_source": "review", + "duration_ms": 222, + "prepare_duration_ms": 12, + "spawn_duration_ms": 123, + "finalize_duration_ms": 87, + }) + ); } #[tokio::test] diff --git a/codex-rs/analytics/src/client.rs b/codex-rs/analytics/src/client.rs index 70a799d2bd..b35e55ea1a 100644 --- a/codex-rs/analytics/src/client.rs +++ b/codex-rs/analytics/src/client.rs @@ -18,7 +18,7 @@ use crate::facts::PluginStateChangedInput; use crate::facts::SkillInvocation; use crate::facts::SkillInvokedInput; use crate::facts::SubAgentThreadStartedInput; -use crate::facts::ThreadStartTimingFact; +use crate::facts::ThreadInitializationTimingFact; use crate::facts::TrackEventsContext; use crate::facts::TurnResolvedConfigFact; use crate::facts::TurnTimingBreakdownFact; @@ -182,9 +182,9 @@ impl AnalyticsEventsClient { )); } - pub fn track_thread_start_timing(&self, fact: ThreadStartTimingFact) { + pub fn track_thread_initialization_timing(&self, fact: ThreadInitializationTimingFact) { self.record_fact(AnalyticsFact::Custom( - CustomAnalyticsFact::ThreadStartTiming(fact), + CustomAnalyticsFact::ThreadInitializationTiming(fact), )); } diff --git a/codex-rs/analytics/src/events.rs b/codex-rs/analytics/src/events.rs index a0423b292f..bed8043ff3 100644 --- a/codex-rs/analytics/src/events.rs +++ b/codex-rs/analytics/src/events.rs @@ -158,7 +158,7 @@ pub(crate) struct AppServerStartedEventRequest { pub(crate) event_params: AppServerStartedEventParams, } -#[derive(Default, Serialize)] +#[derive(Clone, Copy, Default, Serialize)] pub(crate) struct ThreadInitializationTimingParams { pub(crate) duration_ms: Option, pub(crate) prepare_duration_ms: Option, @@ -1059,6 +1059,7 @@ pub(crate) fn current_runtime_metadata() -> CodexRuntimeMetadata { pub(crate) fn subagent_thread_started_event_request( input: SubAgentThreadStartedInput, + initialization_timing: ThreadInitializationTimingParams, ) -> ThreadInitializedEvent { let event_params = ThreadInitializedEventParams { thread_id: input.thread_id, @@ -1079,7 +1080,7 @@ pub(crate) fn subagent_thread_started_event_request( parent_thread_id: input .parent_thread_id .or_else(|| subagent_parent_thread_id(&input.subagent_source)), - initialization_timing: ThreadInitializationTimingParams::default(), + initialization_timing, created_at: input.created_at, }; ThreadInitializedEvent { diff --git a/codex-rs/analytics/src/facts.rs b/codex-rs/analytics/src/facts.rs index 3dda899012..73adeaa7f5 100644 --- a/codex-rs/analytics/src/facts.rs +++ b/codex-rs/analytics/src/facts.rs @@ -290,7 +290,7 @@ pub(crate) struct AppServerStartedInput { pub duration_ms: u64, } -pub struct ThreadStartTimingFact { +pub struct ThreadInitializationTimingFact { pub thread_id: String, pub duration_ms: u64, pub prepare_duration_ms: u64, @@ -349,7 +349,7 @@ pub(crate) enum AnalyticsFact { pub(crate) enum CustomAnalyticsFact { AppServerStarted(AppServerStartedInput), SubAgentThreadStarted(SubAgentThreadStartedInput), - ThreadStartTiming(ThreadStartTimingFact), + ThreadInitializationTiming(ThreadInitializationTimingFact), Compaction(Box), GuardianReview(Box), TurnResolvedConfig(Box), diff --git a/codex-rs/analytics/src/lib.rs b/codex-rs/analytics/src/lib.rs index e9921e7417..db2883eb0c 100644 --- a/codex-rs/analytics/src/lib.rs +++ b/codex-rs/analytics/src/lib.rs @@ -37,7 +37,7 @@ pub use facts::InvocationType; pub use facts::SkillInvocation; pub use facts::SubAgentThreadStartedInput; pub use facts::ThreadInitializationMode; -pub use facts::ThreadStartTimingFact; +pub use facts::ThreadInitializationTimingFact; pub use facts::TrackEventsContext; pub use facts::TurnResolvedConfigFact; pub use facts::TurnStatus; diff --git a/codex-rs/analytics/src/reducer.rs b/codex-rs/analytics/src/reducer.rs index 3d11c4a421..cf74a6fe76 100644 --- a/codex-rs/analytics/src/reducer.rs +++ b/codex-rs/analytics/src/reducer.rs @@ -75,7 +75,7 @@ use crate::facts::PluginUsedInput; use crate::facts::SkillInvokedInput; use crate::facts::SubAgentThreadStartedInput; use crate::facts::ThreadInitializationMode; -use crate::facts::ThreadStartTimingFact; +use crate::facts::ThreadInitializationTimingFact; use crate::facts::TurnResolvedConfigFact; use crate::facts::TurnStatus; use crate::facts::TurnSteerRejectionReason; @@ -153,15 +153,7 @@ struct ConnectionState { struct ThreadAnalyticsState { connection_id: Option, metadata: Option, - thread_start_timing: Option, -} - -#[derive(Clone, Copy)] -struct ThreadStartTimingState { - duration_ms: u64, - prepare_duration_ms: u64, - spawn_duration_ms: u64, - finalize_duration_ms: u64, + initialization_timing: Option, } #[derive(Clone, Copy)] @@ -479,8 +471,8 @@ impl AnalyticsReducer { CustomAnalyticsFact::SubAgentThreadStarted(input) => { self.ingest_subagent_thread_started(input, out); } - CustomAnalyticsFact::ThreadStartTiming(input) => { - self.ingest_thread_start_timing(input); + CustomAnalyticsFact::ThreadInitializationTiming(input) => { + self.ingest_thread_initialization_timing(input); } CustomAnalyticsFact::Compaction(input) => { self.ingest_compaction(*input, out); @@ -587,22 +579,23 @@ impl AnalyticsReducer { if thread_state.connection_id.is_none() { thread_state.connection_id = parent_connection_id; } + let initialization_timing = thread_state.initialization_timing.unwrap_or_default(); out.push(TrackEventRequest::ThreadInitialized( - subagent_thread_started_event_request(input), + subagent_thread_started_event_request(input, initialization_timing), )); } - fn ingest_thread_start_timing(&mut self, input: ThreadStartTimingFact) { - let thread_start_timing = ThreadStartTimingState { - duration_ms: input.duration_ms, - prepare_duration_ms: input.prepare_duration_ms, - spawn_duration_ms: input.spawn_duration_ms, - finalize_duration_ms: input.finalize_duration_ms, + fn ingest_thread_initialization_timing(&mut self, input: ThreadInitializationTimingFact) { + let initialization_timing = ThreadInitializationTimingParams { + duration_ms: Some(input.duration_ms), + prepare_duration_ms: Some(input.prepare_duration_ms), + spawn_duration_ms: Some(input.spawn_duration_ms), + finalize_duration_ms: Some(input.finalize_duration_ms), }; self.threads .entry(input.thread_id) .or_default() - .thread_start_timing = Some(thread_start_timing); + .initialization_timing = Some(initialization_timing); } fn ingest_guardian_review( @@ -1346,9 +1339,6 @@ impl AnalyticsReducer { let thread_state = self.threads.entry(thread_id.clone()).or_default(); thread_state.connection_id = Some(connection_id); thread_state.metadata = Some(thread_metadata.clone()); - let thread_start_timing = matches!(initialization_mode, ThreadInitializationMode::New) - .then_some(thread_state.thread_start_timing) - .flatten(); out.push(TrackEventRequest::ThreadInitialized( ThreadInitializedEvent { event_type: "codex_thread_initialized", @@ -1363,15 +1353,7 @@ impl AnalyticsReducer { initialization_mode, subagent_source: thread_metadata.subagent_source.clone(), parent_thread_id: thread_metadata.parent_thread_id, - initialization_timing: match thread_start_timing { - Some(timing) => ThreadInitializationTimingParams { - duration_ms: Some(timing.duration_ms), - prepare_duration_ms: Some(timing.prepare_duration_ms), - spawn_duration_ms: Some(timing.spawn_duration_ms), - finalize_duration_ms: Some(timing.finalize_duration_ms), - }, - None => ThreadInitializationTimingParams::default(), - }, + initialization_timing: thread_state.initialization_timing.unwrap_or_default(), created_at: u64::try_from(thread.created_at).unwrap_or_default(), }, }, diff --git a/codex-rs/app-server/tests/suite/v2/analytics.rs b/codex-rs/app-server/tests/suite/v2/analytics.rs index bb2306e1b9..03fd3da9da 100644 --- a/codex-rs/app-server/tests/suite/v2/analytics.rs +++ b/codex-rs/app-server/tests/suite/v2/analytics.rs @@ -182,5 +182,15 @@ pub(crate) fn assert_basic_thread_initialized_event( event["event_params"]["initialization_mode"], initialization_mode ); + assert_eq!( + [ + "duration_ms", + "prepare_duration_ms", + "spawn_duration_ms", + "finalize_duration_ms", + ] + .map(|field| event["event_params"][field].as_u64().is_some()), + [true, true, true, true] + ); assert!(event["event_params"]["created_at"].as_u64().is_some()); } diff --git a/codex-rs/app-server/tests/suite/v2/thread_start.rs b/codex-rs/app-server/tests/suite/v2/thread_start.rs index 1bb4117557..835ab3764f 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_start.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_start.rs @@ -447,14 +447,6 @@ async fn thread_start_tracks_thread_initialized_analytics() -> Result<()> { "new", "user", ); - let thread_start_timings = [ - "duration_ms", - "prepare_duration_ms", - "spawn_duration_ms", - "finalize_duration_ms", - ] - .map(|field| event["event_params"][field].as_u64().is_some()); - assert_eq!(thread_start_timings, [true, true, true, true]); Ok(()) } diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index bd83a0c6a7..f3c229fe18 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -137,8 +137,8 @@ pub mod spawn; pub(crate) mod state_db_bridge; pub use state_db_bridge::StateDbHandle; pub use state_db_bridge::init_state_db; +mod thread_initialization_timing; mod thread_rollout_truncation; -mod thread_start_timing; mod tools; pub(crate) mod turn_diff_tracker; mod turn_metadata; diff --git a/codex-rs/core/src/thread_start_timing.rs b/codex-rs/core/src/thread_initialization_timing.rs similarity index 85% rename from codex-rs/core/src/thread_start_timing.rs rename to codex-rs/core/src/thread_initialization_timing.rs index 0cf5457d5a..a6dbcd4d9a 100644 --- a/codex-rs/core/src/thread_start_timing.rs +++ b/codex-rs/core/src/thread_initialization_timing.rs @@ -1,17 +1,17 @@ use std::time::Duration; use std::time::Instant; -use codex_analytics::ThreadStartTimingFact; +use codex_analytics::ThreadInitializationTimingFact; #[derive(Debug)] -pub(crate) struct ThreadStartTiming { +pub(crate) struct ThreadInitializationTiming { phase_started_at: Instant, prepare_duration: Option, spawn_duration: Option, finalize_duration: Option, } -impl ThreadStartTiming { +impl ThreadInitializationTiming { pub(crate) fn start() -> Self { Self { phase_started_at: Instant::now(), @@ -33,11 +33,11 @@ impl ThreadStartTiming { self.finalize_duration = Some(self.finish_phase()); } - pub(crate) fn into_fact(self, thread_id: String) -> ThreadStartTimingFact { + pub(crate) fn into_fact(self, thread_id: String) -> ThreadInitializationTimingFact { let prepare_duration = self.prepare_duration.unwrap_or_default(); let spawn_duration = self.spawn_duration.unwrap_or_default(); let finalize_duration = self.finalize_duration.unwrap_or_default(); - ThreadStartTimingFact { + ThreadInitializationTimingFact { thread_id, duration_ms: duration_ms(prepare_duration + spawn_duration + finalize_duration), prepare_duration_ms: duration_ms(prepare_duration), diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index ba52a09128..53f641cf0e 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -15,7 +15,7 @@ use crate::session::INITIAL_SUBMIT_ID; use crate::shell_snapshot::ShellSnapshot; use crate::tasks::InterruptedTurnHistoryMarker; use crate::tasks::interrupted_turn_history_marker; -use crate::thread_start_timing::ThreadStartTiming; +use crate::thread_initialization_timing::ThreadInitializationTiming; use codex_analytics::AnalyticsEventsClient; use codex_app_server_protocol::ThreadHistoryBuilder; use codex_app_server_protocol::TurnStatus; @@ -1220,7 +1220,7 @@ impl ThreadManagerState { user_shell_override: Option, ) -> CodexResult { let is_resumed_thread = matches!(&initial_history, InitialHistory::Resumed(_)); - let mut thread_start_timing = ThreadStartTiming::start(); + let mut initialization_timing = ThreadInitializationTiming::start(); if let InitialHistory::Resumed(resumed) = &initial_history { let mut threads = self.threads.write().await; if let Some(thread) = threads.get(&resumed.conversation_id).cloned() { @@ -1233,10 +1233,10 @@ impl ThreadManagerState { resumed.conversation_id ))); } - thread_start_timing.mark_prepare_completed(); + initialization_timing.mark_prepare_completed(); if let Some(analytics_events_client) = self.analytics_events_client.as_ref() { - analytics_events_client.track_thread_start_timing( - thread_start_timing.into_fact(resumed.conversation_id.to_string()), + analytics_events_client.track_thread_initialization_timing( + initialization_timing.into_fact(resumed.conversation_id.to_string()), ); } return Ok(NewThread { @@ -1253,7 +1253,7 @@ impl ThreadManagerState { let parent_rollout_thread_trace = self .parent_rollout_thread_trace_for_source(&session_source, &initial_history) .await; - thread_start_timing.mark_prepare_completed(); + initialization_timing.mark_prepare_completed(); let tracked_session_source = session_source.clone(); let CodexSpawnOk { codex, thread_id, .. @@ -1286,14 +1286,14 @@ impl ThreadManagerState { attestation_provider: self.attestation_provider.clone(), }) .await?; - thread_start_timing.mark_spawn_completed(); + initialization_timing.mark_spawn_completed(); let new_thread = self .finalize_thread_spawn(codex, thread_id, tracked_session_source) .await?; - thread_start_timing.mark_finalize_completed(); + initialization_timing.mark_finalize_completed(); if let Some(analytics_events_client) = self.analytics_events_client.as_ref() { - analytics_events_client.track_thread_start_timing( - thread_start_timing.into_fact(new_thread.thread_id.to_string()), + analytics_events_client.track_thread_initialization_timing( + initialization_timing.into_fact(new_thread.thread_id.to_string()), ); } if is_resumed_thread {