openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-05 15:18:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

Track timing for every thread initialization

Browse Source
This commit is contained in:
Ahmed Ibrahim
2026-05-28 11:40:44 -07:00
parent 5f31aa4469
commit 3f40c4eabf
11 changed files with 106 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

70
codex-rs/analytics/src/analytics_client_tests.rs
View File

@@ -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]

6
codex-rs/analytics/src/client.rs
View File

@@ -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),
));
}

5
codex-rs/analytics/src/events.rs
View File

@@ -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<u64>,
pub(crate) prepare_duration_ms: Option<u64>,
@@ -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 {

4
codex-rs/analytics/src/facts.rs
View File

@@ -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<CodexCompactionEvent>),
GuardianReview(Box<GuardianReviewEventParams>),
TurnResolvedConfig(Box<TurnResolvedConfigFact>),

2
codex-rs/analytics/src/lib.rs
View File

@@ -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;

46
codex-rs/analytics/src/reducer.rs
View File

@@ -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<u64>,
metadata: Option<ThreadMetadataState>,
thread_start_timing: Option<ThreadStartTimingState>,
}
#[derive(Clone, Copy)]
struct ThreadStartTimingState {
duration_ms: u64,
prepare_duration_ms: u64,
spawn_duration_ms: u64,
finalize_duration_ms: u64,
initialization_timing: Option<ThreadInitializationTimingParams>,
}
#[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(),
},
},

10
codex-rs/app-server/tests/suite/v2/analytics.rs
View File

@@ -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());
}

8
codex-rs/app-server/tests/suite/v2/thread_start.rs
View File

@@ -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(())
}

2
codex-rs/core/src/lib.rs
View File

@@ -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;

10
codex-rs/core/src/thread_start_timing.rs → codex-rs/core/src/thread_initialization_timing.rs
View File

@@ -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<Duration>,
spawn_duration: Option<Duration>,
finalize_duration: Option<Duration>,
}
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),

20
codex-rs/core/src/thread_manager.rs
View File

@@ -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<crate::shell::Shell>,
) -> CodexResult<NewThread> {
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 {