From b2a1da06644a65eee412cd6d527251ac67eb79ad Mon Sep 17 00:00:00 2001 From: Ahmed Ibrahim Date: Fri, 22 May 2026 12:21:01 -0700 Subject: [PATCH] Track app-server, thread, and turn timing in ChatGPT telemetry --- .../analytics/src/analytics_client_tests.rs | 302 +++++++++++++++++- codex-rs/analytics/src/client.rs | 40 ++- codex-rs/analytics/src/events.rs | 22 ++ codex-rs/analytics/src/facts.rs | 24 ++ codex-rs/analytics/src/lib.rs | 1 + codex-rs/analytics/src/reducer.rs | 201 +++++++----- codex-rs/app-server/src/lib.rs | 23 +- codex-rs/app-server/src/message_processor.rs | 1 + .../request_processors/thread_processor.rs | 14 + codex-rs/core/src/session/turn.rs | 36 ++- codex-rs/core/src/tasks/mod.rs | 12 + codex-rs/core/src/turn_timing.rs | 54 ++++ codex-rs/core/src/turn_timing_tests.rs | 22 ++ 13 files changed, 656 insertions(+), 96 deletions(-) diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index c744eb3f6e..5b1751614b 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -4,6 +4,8 @@ use crate::events::CodexAcceptedLineFingerprintsEventParams; use crate::events::CodexAcceptedLineFingerprintsEventRequest; use crate::events::CodexAppMentionedEventRequest; use crate::events::CodexAppServerClientMetadata; +use crate::events::CodexAppServerStartedEventParams; +use crate::events::CodexAppServerStartedEventRequest; use crate::events::CodexAppUsedEventRequest; use crate::events::CodexCommandExecutionEventParams; use crate::events::CodexCommandExecutionEventRequest; @@ -15,6 +17,7 @@ use crate::events::CodexReviewEventParams; use crate::events::CodexReviewEventRequest; use crate::events::CodexRuntimeMetadata; use crate::events::CodexToolItemEventBase; +use crate::events::CodexTurnEventParams; use crate::events::CodexTurnEventRequest; use crate::events::FinalApprovalOutcome; use crate::events::GuardianApprovalRequestSource; @@ -41,6 +44,7 @@ use crate::facts::AnalyticsFact; use crate::facts::AnalyticsJsonRpcError; use crate::facts::AppInvocation; use crate::facts::AppMentionedInput; +use crate::facts::AppServerStartedInput; use crate::facts::AppUsedInput; use crate::facts::CodexCompactionEvent; use crate::facts::CompactionImplementation; @@ -61,10 +65,12 @@ use crate::facts::SkillInvocation; use crate::facts::SkillInvokedInput; use crate::facts::SubAgentThreadStartedInput; use crate::facts::ThreadInitializationMode; +use crate::facts::ThreadStartTimingFact; use crate::facts::TrackEventsContext; use crate::facts::TurnResolvedConfigFact; use crate::facts::TurnStatus; use crate::facts::TurnSteerRequestError; +use crate::facts::TurnTimingBreakdownFact; use crate::facts::TurnTokenUsageFact; use crate::reducer::AnalyticsReducer; use crate::reducer::normalize_path_for_skill_id; @@ -331,6 +337,16 @@ fn sample_turn_token_usage_fact(thread_id: &str, turn_id: &str) -> TurnTokenUsag } } +fn sample_turn_timing_breakdown_fact(thread_id: &str, turn_id: &str) -> TurnTimingBreakdownFact { + TurnTimingBreakdownFact { + turn_id: turn_id.to_string(), + thread_id: thread_id.to_string(), + request_start_delay_ms: Some(120), + sampling_duration_ms: 900, + blocking_tool_critical_path_duration_ms: 140, + } +} + fn sample_turn_completed_notification( thread_id: &str, turn_id: &str, @@ -1317,6 +1333,7 @@ fn thread_initialized_event_serializes_expected_shape() { initialization_mode: ThreadInitializationMode::New, subagent_source: None, parent_thread_id: None, + thread_start_duration_ms: Some(321), created_at: 1, }, }); @@ -1348,12 +1365,46 @@ fn thread_initialized_event_serializes_expected_shape() { "initialization_mode": "new", "subagent_source": null, "parent_thread_id": null, + "thread_start_duration_ms": 321, "created_at": 1 } }) ); } +#[test] +fn app_server_started_event_serializes_expected_shape() { + let event = TrackEventRequest::AppServerStarted(CodexAppServerStartedEventRequest { + event_type: "codex_app_server_started", + event_params: CodexAppServerStartedEventParams { + runtime: sample_runtime_metadata(), + rpc_transport: AppServerRpcTransport::Websocket, + startup_duration_ms: 987, + completed_at: 12, + }, + }); + + let payload = serde_json::to_value(&event).expect("serialize app-server started event"); + + assert_eq!( + payload, + json!({ + "event_type": "codex_app_server_started", + "event_params": { + "runtime": { + "codex_rs_version": "0.1.0", + "runtime_os": "macos", + "runtime_os_version": "15.3.1", + "runtime_arch": "aarch64" + }, + "rpc_transport": "websocket", + "startup_duration_ms": 987, + "completed_at": 12 + } + }) + ); +} + #[test] fn command_execution_event_serializes_expected_shape() { let event = TrackEventRequest::CommandExecution(CodexCommandExecutionEventRequest { @@ -1629,6 +1680,87 @@ async fn initialize_caches_client_and_thread_lifecycle_publishes_once_initialize payload[0]["event_params"]["runtime"]["runtime_arch"], "x86_64" ); + assert_eq!( + payload[0]["event_params"]["thread_start_duration_ms"], + json!(null) + ); +} + +#[tokio::test] +async fn thread_start_timing_fact_enriches_thread_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 { + thread_id: "thread-1".to_string(), + duration_ms: 222, + }, + )), + &mut events, + ) + .await; + reducer + .ingest( + AnalyticsFact::ClientResponse { + connection_id: 7, + request_id: RequestId::Integer(2), + response: Box::new(sample_thread_start_response( + "thread-1", /*ephemeral*/ true, "gpt-5", + )), + }, + &mut events, + ) + .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"]["thread_start_duration_ms"], + json!(222) + ); +} + +#[tokio::test] +async fn app_server_started_fact_emits_event() { + let mut reducer = AnalyticsReducer::default(); + let mut events = Vec::new(); + + reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::AppServerStarted( + AppServerStartedInput { + runtime: sample_runtime_metadata(), + rpc_transport: AppServerRpcTransport::Stdio, + startup_duration_ms: 456, + completed_at: 12, + }, + )), + &mut events, + ) + .await; + + let payload = serde_json::to_value(&events).expect("serialize events"); + assert_eq!( + payload, + json!([{ + "event_type": "codex_app_server_started", + "event_params": { + "runtime": { + "codex_rs_version": "0.1.0", + "runtime_os": "macos", + "runtime_os_version": "15.3.1", + "runtime_arch": "aarch64" + }, + "rpc_transport": "stdio", + "startup_duration_ms": 456, + "completed_at": 12 + } + }]) + ); } #[tokio::test] @@ -3180,7 +3312,7 @@ async fn reducer_ingests_plugin_state_changed_fact() { fn turn_event_serializes_expected_shape() { let event = TrackEventRequest::TurnEvent(Box::new(CodexTurnEventRequest { event_type: "codex_turn_event", - event_params: crate::events::CodexTurnEventParams { + event_params: CodexTurnEventParams { thread_id: "thread-2".to_string(), turn_id: "turn-2".to_string(), app_server_client: sample_app_server_client_metadata(), @@ -3221,6 +3353,11 @@ fn turn_event_serializes_expected_shape() { reasoning_output_tokens: None, total_tokens: None, duration_ms: Some(1234), + request_start_delay_ms: Some(120), + sampling_duration_ms: Some(900), + blocking_tool_critical_path_duration_ms: Some(140), + approval_wait_duration_ms: Some(0), + finalize_duration_ms: Some(74), started_at: Some(455), completed_at: Some(456), }, @@ -3282,6 +3419,11 @@ fn turn_event_serializes_expected_shape() { "reasoning_output_tokens": null, "total_tokens": null, "duration_ms": 1234, + "request_start_delay_ms": 120, + "sampling_duration_ms": 900, + "blocking_tool_critical_path_duration_ms": 140, + "approval_wait_duration_ms": 0, + "finalize_duration_ms": 74, "started_at": 455, "completed_at": 456 } @@ -3597,6 +3739,164 @@ async fn turn_lifecycle_emits_turn_event() { json!(13) ); assert_eq!(payload["event_params"]["total_tokens"], json!(321)); + assert_eq!( + payload["event_params"]["request_start_delay_ms"], + json!(null) + ); + assert_eq!(payload["event_params"]["sampling_duration_ms"], json!(null)); + assert_eq!( + payload["event_params"]["blocking_tool_critical_path_duration_ms"], + json!(null) + ); + assert_eq!( + payload["event_params"]["approval_wait_duration_ms"], + json!(null) + ); + assert_eq!(payload["event_params"]["finalize_duration_ms"], json!(null)); +} + +#[tokio::test] +async fn turn_timing_breakdown_fact_enriches_turn_event() { + let mut reducer = AnalyticsReducer::default(); + let mut out = Vec::new(); + + ingest_turn_prerequisites( + &mut reducer, + &mut out, + /*include_initialize*/ true, + /*include_resolved_config*/ true, + /*include_started*/ true, + /*include_token_usage*/ false, + ) + .await; + reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::TurnTimingBreakdown(Box::new( + sample_turn_timing_breakdown_fact("thread-2", "turn-2"), + ))), + &mut out, + ) + .await; + reducer + .ingest( + AnalyticsFact::Notification(Box::new(sample_turn_completed_notification( + "thread-2", + "turn-2", + AppServerTurnStatus::Completed, + /*codex_error_info*/ None, + ))), + &mut out, + ) + .await; + + let payload = serde_json::to_value(&out[0]).expect("serialize turn event"); + assert_eq!( + payload["event_params"]["request_start_delay_ms"], + json!(120) + ); + assert_eq!(payload["event_params"]["sampling_duration_ms"], json!(900)); + assert_eq!( + payload["event_params"]["blocking_tool_critical_path_duration_ms"], + json!(140) + ); + assert_eq!( + payload["event_params"]["approval_wait_duration_ms"], + json!(0) + ); + assert_eq!(payload["event_params"]["finalize_duration_ms"], json!(74)); +} + +#[tokio::test] +async fn review_durations_roll_up_into_turn_approval_wait() { + let mut reducer = AnalyticsReducer::default(); + let mut out = Vec::new(); + + ingest_turn_prerequisites( + &mut reducer, + &mut out, + /*include_initialize*/ true, + /*include_resolved_config*/ true, + /*include_started*/ true, + /*include_token_usage*/ false, + ) + .await; + reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::TurnTimingBreakdown(Box::new( + TurnTimingBreakdownFact { + turn_id: "turn-2".to_string(), + thread_id: "thread-2".to_string(), + request_start_delay_ms: Some(100), + sampling_duration_ms: 400, + blocking_tool_critical_path_duration_ms: 700, + }, + ))), + &mut out, + ) + .await; + reducer + .ingest( + AnalyticsFact::ServerRequest { + connection_id: 7, + request: Box::new(ServerRequest::CommandExecutionRequestApproval { + request_id: RequestId::Integer(99), + params: CommandExecutionRequestApprovalParams { + thread_id: "thread-2".to_string(), + turn_id: "turn-2".to_string(), + item_id: "item-1".to_string(), + started_at_ms: 1_000, + approval_id: None, + reason: None, + network_approval_context: None, + command: Some("echo hi".to_string()), + cwd: None, + command_actions: None, + additional_permissions: None, + proposed_execpolicy_amendment: None, + proposed_network_policy_amendments: None, + available_decisions: None, + }, + }), + }, + &mut out, + ) + .await; + reducer + .ingest( + AnalyticsFact::ServerResponse { + completed_at_ms: 1_600, + response: Box::new(ServerResponse::CommandExecutionRequestApproval { + request_id: RequestId::Integer(99), + response: CommandExecutionRequestApprovalResponse { + decision: CommandExecutionApprovalDecision::AllowOnce, + }, + }), + }, + &mut out, + ) + .await; + reducer + .ingest( + AnalyticsFact::Notification(Box::new(sample_turn_completed_notification( + "thread-2", + "turn-2", + AppServerTurnStatus::Completed, + /*codex_error_info*/ None, + ))), + &mut out, + ) + .await; + + let turn_event = out + .iter() + .find(|event| matches!(event, TrackEventRequest::TurnEvent(_))) + .expect("turn event should be emitted"); + let payload = serde_json::to_value(turn_event).expect("serialize turn event"); + assert_eq!( + payload["event_params"]["approval_wait_duration_ms"], + json!(600) + ); + assert_eq!(payload["event_params"]["finalize_duration_ms"], json!(34)); } #[tokio::test] diff --git a/codex-rs/analytics/src/client.rs b/codex-rs/analytics/src/client.rs index fbcfa32dc5..9611510837 100644 --- a/codex-rs/analytics/src/client.rs +++ b/codex-rs/analytics/src/client.rs @@ -8,17 +8,22 @@ use crate::facts::AnalyticsFact; use crate::facts::AnalyticsJsonRpcError; use crate::facts::AppInvocation; use crate::facts::AppMentionedInput; +use crate::facts::AppServerStartedInput; use crate::facts::AppUsedInput; +use crate::facts::CodexCompactionEvent; use crate::facts::CustomAnalyticsFact; use crate::facts::HookRunFact; use crate::facts::HookRunInput; use crate::facts::PluginState; use crate::facts::PluginStateChangedInput; +use crate::facts::PluginUsedInput; use crate::facts::SkillInvocation; use crate::facts::SkillInvokedInput; use crate::facts::SubAgentThreadStartedInput; +use crate::facts::ThreadStartTimingFact; use crate::facts::TrackEventsContext; use crate::facts::TurnResolvedConfigFact; +use crate::facts::TurnTimingBreakdownFact; use crate::facts::TurnTokenUsageFact; use crate::reducer::AnalyticsReducer; use codex_app_server_protocol::ClientRequest; @@ -163,12 +168,37 @@ impl AnalyticsEventsClient { }); } + pub fn track_app_server_started( + &self, + rpc_transport: AppServerRpcTransport, + startup_duration_ms: u64, + completed_at: u64, + ) { + self.record_fact(AnalyticsFact::Custom( + CustomAnalyticsFact::AppServerStarted(AppServerStartedInput { + runtime: current_runtime_metadata(), + rpc_transport, + startup_duration_ms, + completed_at, + }), + )); + } + pub fn track_subagent_thread_started(&self, input: SubAgentThreadStartedInput) { self.record_fact(AnalyticsFact::Custom( CustomAnalyticsFact::SubAgentThreadStarted(input), )); } + pub fn track_thread_start_timing(&self, thread_id: String, duration_ms: u64) { + self.record_fact(AnalyticsFact::Custom( + CustomAnalyticsFact::ThreadStartTiming(ThreadStartTimingFact { + thread_id, + duration_ms, + }), + )); + } + pub fn track_guardian_review( &self, tracking: &GuardianReviewTrackContext, @@ -234,11 +264,11 @@ impl AnalyticsEventsClient { return; } self.record_fact(AnalyticsFact::Custom(CustomAnalyticsFact::PluginUsed( - crate::facts::PluginUsedInput { tracking, plugin }, + PluginUsedInput { tracking, plugin }, ))); } - pub fn track_compaction(&self, event: crate::facts::CodexCompactionEvent) { + pub fn track_compaction(&self, event: CodexCompactionEvent) { self.record_fact(AnalyticsFact::Custom(CustomAnalyticsFact::Compaction( Box::new(event), ))); @@ -256,6 +286,12 @@ impl AnalyticsEventsClient { ))); } + pub fn track_turn_timing(&self, fact: TurnTimingBreakdownFact) { + self.record_fact(AnalyticsFact::Custom( + CustomAnalyticsFact::TurnTimingBreakdown(Box::new(fact)), + )); + } + pub fn track_plugin_installed(&self, plugin: PluginTelemetryMetadata) { self.record_fact(AnalyticsFact::Custom( CustomAnalyticsFact::PluginStateChanged(PluginStateChangedInput { diff --git a/codex-rs/analytics/src/events.rs b/codex-rs/analytics/src/events.rs index f7c64e0e74..b3b161e38e 100644 --- a/codex-rs/analytics/src/events.rs +++ b/codex-rs/analytics/src/events.rs @@ -56,6 +56,7 @@ pub(crate) struct TrackEventsRequest { #[serde(untagged)] pub(crate) enum TrackEventRequest { SkillInvocation(SkillInvocationEventRequest), + AppServerStarted(CodexAppServerStartedEventRequest), ThreadInitialized(ThreadInitializedEvent), GuardianReview(Box), AppMentioned(CodexAppMentionedEventRequest), @@ -144,6 +145,20 @@ pub(crate) struct CodexRuntimeMetadata { pub(crate) runtime_arch: String, } +#[derive(Serialize)] +pub(crate) struct CodexAppServerStartedEventParams { + pub(crate) runtime: CodexRuntimeMetadata, + pub(crate) rpc_transport: AppServerRpcTransport, + pub(crate) startup_duration_ms: u64, + pub(crate) completed_at: u64, +} + +#[derive(Serialize)] +pub(crate) struct CodexAppServerStartedEventRequest { + pub(crate) event_type: &'static str, + pub(crate) event_params: CodexAppServerStartedEventParams, +} + #[derive(Serialize)] pub(crate) struct ThreadInitializedEventParams { pub(crate) thread_id: String, @@ -155,6 +170,7 @@ pub(crate) struct ThreadInitializedEventParams { pub(crate) initialization_mode: ThreadInitializationMode, pub(crate) subagent_source: Option, pub(crate) parent_thread_id: Option, + pub(crate) thread_start_duration_ms: Option, pub(crate) created_at: u64, } @@ -808,6 +824,11 @@ pub(crate) struct CodexTurnEventParams { pub(crate) reasoning_output_tokens: Option, pub(crate) total_tokens: Option, pub(crate) duration_ms: Option, + pub(crate) request_start_delay_ms: Option, + pub(crate) sampling_duration_ms: Option, + pub(crate) blocking_tool_critical_path_duration_ms: Option, + pub(crate) approval_wait_duration_ms: Option, + pub(crate) finalize_duration_ms: Option, pub(crate) started_at: Option, pub(crate) completed_at: Option, } @@ -1041,6 +1062,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)), + thread_start_duration_ms: None, created_at: input.created_at, }; ThreadInitializedEvent { diff --git a/codex-rs/analytics/src/facts.rs b/codex-rs/analytics/src/facts.rs index d7e2c069d6..f84c92e554 100644 --- a/codex-rs/analytics/src/facts.rs +++ b/codex-rs/analytics/src/facts.rs @@ -99,6 +99,15 @@ pub struct TurnTokenUsageFact { pub token_usage: TokenUsage, } +#[derive(Clone)] +pub struct TurnTimingBreakdownFact { + pub turn_id: String, + pub thread_id: String, + pub request_start_delay_ms: Option, + pub sampling_duration_ms: u64, + pub blocking_tool_critical_path_duration_ms: u64, +} + #[derive(Clone, Copy, Debug, Serialize)] #[serde(rename_all = "snake_case")] pub enum TurnStatus { @@ -273,6 +282,18 @@ pub struct CodexCompactionEvent { pub duration_ms: Option, } +pub(crate) struct AppServerStartedInput { + pub runtime: CodexRuntimeMetadata, + pub rpc_transport: AppServerRpcTransport, + pub startup_duration_ms: u64, + pub completed_at: u64, +} + +pub(crate) struct ThreadStartTimingFact { + pub thread_id: String, + pub duration_ms: u64, +} + #[allow(dead_code)] pub(crate) enum AnalyticsFact { Initialize { @@ -322,11 +343,14 @@ pub(crate) enum AnalyticsFact { } pub(crate) enum CustomAnalyticsFact { + AppServerStarted(AppServerStartedInput), SubAgentThreadStarted(SubAgentThreadStartedInput), + ThreadStartTiming(ThreadStartTimingFact), Compaction(Box), GuardianReview(Box), TurnResolvedConfig(Box), TurnTokenUsage(Box), + TurnTimingBreakdown(Box), SkillInvoked(SkillInvokedInput), AppMentioned(AppMentionedInput), AppUsed(AppUsedInput), diff --git a/codex-rs/analytics/src/lib.rs b/codex-rs/analytics/src/lib.rs index a33ca7b9e3..afa3edf9ad 100644 --- a/codex-rs/analytics/src/lib.rs +++ b/codex-rs/analytics/src/lib.rs @@ -43,6 +43,7 @@ pub use facts::TurnStatus; pub use facts::TurnSteerRejectionReason; pub use facts::TurnSteerRequestError; pub use facts::TurnSteerResult; +pub use facts::TurnTimingBreakdownFact; pub use facts::TurnTokenUsageFact; pub use facts::build_track_events_context; diff --git a/codex-rs/analytics/src/reducer.rs b/codex-rs/analytics/src/reducer.rs index d072720d10..3bc7e9367f 100644 --- a/codex-rs/analytics/src/reducer.rs +++ b/codex-rs/analytics/src/reducer.rs @@ -5,6 +5,8 @@ use crate::accepted_lines::accepted_line_repo_hash_for_cwd; use crate::events::AppServerRpcTransport; use crate::events::CodexAppMentionedEventRequest; use crate::events::CodexAppServerClientMetadata; +use crate::events::CodexAppServerStartedEventParams; +use crate::events::CodexAppServerStartedEventRequest; use crate::events::CodexAppUsedEventRequest; use crate::events::CodexCollabAgentToolCallEventParams; use crate::events::CodexCollabAgentToolCallEventRequest; @@ -61,6 +63,7 @@ use crate::events::subagent_thread_started_event_request; use crate::facts::AnalyticsFact; use crate::facts::AnalyticsJsonRpcError; use crate::facts::AppMentionedInput; +use crate::facts::AppServerStartedInput; use crate::facts::AppUsedInput; use crate::facts::CodexCompactionEvent; use crate::facts::CustomAnalyticsFact; @@ -71,10 +74,12 @@ use crate::facts::PluginUsedInput; use crate::facts::SkillInvokedInput; use crate::facts::SubAgentThreadStartedInput; use crate::facts::ThreadInitializationMode; +use crate::facts::ThreadStartTimingFact; use crate::facts::TurnResolvedConfigFact; use crate::facts::TurnStatus; use crate::facts::TurnSteerRejectionReason; use crate::facts::TurnSteerResult; +use crate::facts::TurnTimingBreakdownFact; use crate::facts::TurnTokenUsageFact; use crate::now_unix_seconds; use crate::option_i64_to_u64; @@ -147,6 +152,7 @@ struct ConnectionState { struct ThreadAnalyticsState { connection_id: Option, metadata: Option, + thread_start_duration_ms: Option, } #[derive(Clone, Copy)] @@ -314,6 +320,15 @@ struct CompletedTurnState { duration_ms: Option, } +#[derive(Clone, Default)] +struct TurnTimingState { + request_start_delay_ms: Option, + sampling_duration_ms: Option, + blocking_tool_critical_path_duration_ms: Option, + approval_wait_duration_ms: u64, +} + +#[derive(Default)] struct TurnState { connection_id: Option, thread_id: Option, @@ -325,6 +340,7 @@ struct TurnState { latest_diff: Option, steer_count: usize, tool_counts: TurnToolCounts, + timing: TurnTimingState, } #[derive(Hash, Eq, PartialEq)] @@ -446,9 +462,15 @@ impl AnalyticsReducer { self.ingest_server_request_aborted(completed_at_ms, request_id, out); } AnalyticsFact::Custom(input) => match input { + CustomAnalyticsFact::AppServerStarted(input) => { + self.ingest_app_server_started(input, out); + } CustomAnalyticsFact::SubAgentThreadStarted(input) => { self.ingest_subagent_thread_started(input, out); } + CustomAnalyticsFact::ThreadStartTiming(input) => { + self.ingest_thread_start_timing(input); + } CustomAnalyticsFact::Compaction(input) => { self.ingest_compaction(*input, out); } @@ -461,6 +483,9 @@ impl AnalyticsReducer { CustomAnalyticsFact::TurnTokenUsage(input) => { self.ingest_turn_token_usage(*input, out).await; } + CustomAnalyticsFact::TurnTimingBreakdown(input) => { + self.ingest_turn_timing_breakdown(*input, out).await; + } CustomAnalyticsFact::SkillInvoked(input) => { self.ingest_skill_invoked(input, out).await; } @@ -508,6 +533,24 @@ impl AnalyticsReducer { ); } + fn ingest_app_server_started( + &mut self, + input: AppServerStartedInput, + out: &mut Vec, + ) { + out.push(TrackEventRequest::AppServerStarted( + CodexAppServerStartedEventRequest { + event_type: "codex_app_server_started", + event_params: CodexAppServerStartedEventParams { + runtime: input.runtime, + rpc_transport: input.rpc_transport, + startup_duration_ms: input.startup_duration_ms, + completed_at: input.completed_at, + }, + }, + )); + } + fn ingest_subagent_thread_started( &mut self, input: SubAgentThreadStartedInput, @@ -538,6 +581,13 @@ impl AnalyticsReducer { )); } + fn ingest_thread_start_timing(&mut self, input: ThreadStartTimingFact) { + self.threads + .entry(input.thread_id) + .or_default() + .thread_start_duration_ms = Some(input.duration_ms); + } + fn ingest_guardian_review( &mut self, input: GuardianReviewEventParams, @@ -599,18 +649,7 @@ impl AnalyticsReducer { let turn_id = input.turn_id.clone(); let thread_id = input.thread_id.clone(); let num_input_images = input.num_input_images; - let turn_state = self.turns.entry(turn_id.clone()).or_insert(TurnState { - connection_id: None, - thread_id: None, - num_input_images: None, - resolved_config: None, - started_at: None, - token_usage: None, - completed: None, - latest_diff: None, - steer_count: 0, - tool_counts: TurnToolCounts::default(), - }); + let turn_state = self.turns.entry(turn_id.clone()).or_default(); turn_state.thread_id = Some(thread_id); turn_state.num_input_images = Some(num_input_images); turn_state.resolved_config = Some(input); @@ -623,23 +662,27 @@ impl AnalyticsReducer { out: &mut Vec, ) { let turn_id = input.turn_id.clone(); - let turn_state = self.turns.entry(turn_id.clone()).or_insert(TurnState { - connection_id: None, - thread_id: None, - num_input_images: None, - resolved_config: None, - started_at: None, - token_usage: None, - completed: None, - latest_diff: None, - steer_count: 0, - tool_counts: TurnToolCounts::default(), - }); + let turn_state = self.turns.entry(turn_id.clone()).or_default(); turn_state.thread_id = Some(input.thread_id); turn_state.token_usage = Some(input.token_usage); self.maybe_emit_turn_event(&turn_id, out).await; } + async fn ingest_turn_timing_breakdown( + &mut self, + input: TurnTimingBreakdownFact, + out: &mut Vec, + ) { + let turn_id = input.turn_id.clone(); + let turn_state = self.turns.entry(turn_id.clone()).or_default(); + turn_state.thread_id = Some(input.thread_id); + turn_state.timing.request_start_delay_ms = input.request_start_delay_ms; + turn_state.timing.sampling_duration_ms = Some(input.sampling_duration_ms); + turn_state.timing.blocking_tool_critical_path_duration_ms = + Some(input.blocking_tool_critical_path_duration_ms); + self.maybe_emit_turn_event(&turn_id, out).await; + } + async fn ingest_skill_invoked( &mut self, input: SkillInvokedInput, @@ -788,18 +831,7 @@ impl AnalyticsReducer { else { return; }; - let turn_state = self.turns.entry(turn_id.clone()).or_insert(TurnState { - connection_id: None, - thread_id: None, - num_input_images: None, - resolved_config: None, - started_at: None, - token_usage: None, - completed: None, - latest_diff: None, - steer_count: 0, - tool_counts: TurnToolCounts::default(), - }); + let turn_state = self.turns.entry(turn_id.clone()).or_default(); turn_state.connection_id = Some(connection_id); turn_state.thread_id = Some(pending_request.thread_id); turn_state.num_input_images = Some(pending_request.num_input_images); @@ -1147,58 +1179,19 @@ impl AnalyticsReducer { self.ingest_guardian_review_completed(notification, out); } ServerNotification::TurnStarted(notification) => { - let turn_state = self.turns.entry(notification.turn.id).or_insert(TurnState { - connection_id: None, - thread_id: None, - num_input_images: None, - resolved_config: None, - started_at: None, - token_usage: None, - completed: None, - latest_diff: None, - steer_count: 0, - tool_counts: TurnToolCounts::default(), - }); + let turn_state = self.turns.entry(notification.turn.id).or_default(); turn_state.started_at = notification .turn .started_at .and_then(|started_at| u64::try_from(started_at).ok()); } ServerNotification::TurnDiffUpdated(notification) => { - let turn_state = - self.turns - .entry(notification.turn_id.clone()) - .or_insert(TurnState { - connection_id: None, - thread_id: None, - num_input_images: None, - resolved_config: None, - started_at: None, - token_usage: None, - completed: None, - latest_diff: None, - steer_count: 0, - tool_counts: TurnToolCounts::default(), - }); + let turn_state = self.turns.entry(notification.turn_id.clone()).or_default(); turn_state.thread_id = Some(notification.thread_id); turn_state.latest_diff = Some(notification.diff); } ServerNotification::TurnCompleted(notification) => { - let turn_state = - self.turns - .entry(notification.turn.id.clone()) - .or_insert(TurnState { - connection_id: None, - thread_id: None, - num_input_images: None, - resolved_config: None, - started_at: None, - token_usage: None, - completed: None, - latest_diff: None, - steer_count: 0, - tool_counts: TurnToolCounts::default(), - }); + let turn_state = self.turns.entry(notification.turn.id.clone()).or_default(); turn_state.completed = Some(CompletedTurnState { status: analytics_turn_status(notification.turn.status), turn_error: notification @@ -1240,13 +1233,12 @@ impl AnalyticsReducer { thread.thread_source.map(Into::into), initialization_mode, ); - self.threads.insert( - thread_id.clone(), - ThreadAnalyticsState { - connection_id: Some(connection_id), - metadata: Some(thread_metadata.clone()), - }, - ); + 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_duration_ms = matches!(initialization_mode, ThreadInitializationMode::New) + .then_some(thread_state.thread_start_duration_ms) + .flatten(); out.push(TrackEventRequest::ThreadInitialized( ThreadInitializedEvent { event_type: "codex_thread_initialized", @@ -1260,6 +1252,7 @@ impl AnalyticsReducer { initialization_mode, subagent_source: thread_metadata.subagent_source.clone(), parent_thread_id: thread_metadata.parent_thread_id, + thread_start_duration_ms, created_at: u64::try_from(thread.created_at).unwrap_or_default(), }, }, @@ -1403,6 +1396,7 @@ impl AnalyticsReducer { completed_at_ms: u64, out: &mut Vec, ) { + let duration_ms = observed_duration_ms(pending_review.started_at_ms, completed_at_ms); if let Some(item_key) = item_review_summary_key(&pending_review) { self.record_item_review_summary( item_key, @@ -1412,6 +1406,16 @@ impl AnalyticsReducer { &pending_review, ); } + if let Some(duration_ms) = duration_ms { + let turn_state = self + .turns + .entry(pending_review.turn_id.clone()) + .or_default(); + turn_state.timing.approval_wait_duration_ms = turn_state + .timing + .approval_wait_duration_ms + .saturating_add(duration_ms); + } let Some((connection_state, thread_metadata)) = self.thread_context_or_warn(AnalyticsDropSite::review(&pending_review)) else { @@ -1437,7 +1441,7 @@ impl AnalyticsReducer { resolution, started_at_ms: pending_review.started_at_ms, completed_at_ms, - duration_ms: observed_duration_ms(pending_review.started_at_ms, completed_at_ms), + duration_ms, }, })); } @@ -2445,6 +2449,26 @@ fn codex_turn_event_params( is_first_turn, } = resolved_config; let token_usage = turn_state.token_usage.clone(); + let timing = &turn_state.timing; + let has_turn_timing = timing.request_start_delay_ms.is_some() + || timing.sampling_duration_ms.is_some() + || timing.blocking_tool_critical_path_duration_ms.is_some(); + let approval_wait_duration_ms = has_turn_timing.then_some(timing.approval_wait_duration_ms); + let finalize_duration_ms = completed.duration_ms.and_then(|duration_ms| { + has_turn_timing.then_some( + duration_ms.saturating_sub( + timing + .request_start_delay_ms + .unwrap_or_default() + .saturating_add(timing.sampling_duration_ms.unwrap_or_default()) + .saturating_add( + timing + .blocking_tool_critical_path_duration_ms + .unwrap_or_default(), + ), + ), + ) + }); CodexTurnEventParams { thread_id, turn_id, @@ -2501,6 +2525,11 @@ fn codex_turn_event_params( .as_ref() .map(|token_usage| token_usage.total_tokens), duration_ms: completed.duration_ms, + request_start_delay_ms: timing.request_start_delay_ms, + sampling_duration_ms: timing.sampling_duration_ms, + blocking_tool_critical_path_duration_ms: timing.blocking_tool_critical_path_duration_ms, + approval_wait_duration_ms, + finalize_duration_ms, started_at, completed_at: Some(completed.completed_at), } diff --git a/codex-rs/app-server/src/lib.rs b/codex-rs/app-server/src/lib.rs index ef4106aa14..764e4751c2 100644 --- a/codex-rs/app-server/src/lib.rs +++ b/codex-rs/app-server/src/lib.rs @@ -17,6 +17,7 @@ use std::io::Result as IoResult; use std::sync::Arc; use std::sync::RwLock; use std::sync::atomic::AtomicBool; +use std::time::Instant; use crate::analytics_utils::analytics_events_client_from_config; use crate::config_manager::ConfigManager; @@ -41,6 +42,7 @@ use crate::transport::start_remote_control; use crate::transport::start_stdio_connection; use crate::transport::start_websocket_acceptor; use codex_analytics::AppServerRpcTransport; +use codex_analytics::now_unix_seconds; use codex_app_server_protocol::ConfigLayerSource; use codex_app_server_protocol::ConfigWarningNotification; use codex_app_server_protocol::JSONRPCMessage; @@ -428,6 +430,7 @@ pub async fn run_main_with_transport_options( auth: AppServerWebsocketAuthSettings, runtime_options: AppServerRuntimeOptions, ) -> IoResult<()> { + let app_server_start_started_at = Instant::now(); let (transport_event_tx, mut transport_event_rx) = mpsc::channel::(CHANNEL_CAPACITY); let (outgoing_tx, mut outgoing_rx) = mpsc::channel::(CHANNEL_CAPACITY); @@ -698,6 +701,9 @@ pub async fn run_main_with_transport_options( let auth_manager = AuthManager::shared_from_config(&config, /*enable_codex_api_key_env*/ false).await; + let analytics_events_client = + analytics_events_client_from_config(Arc::clone(&auth_manager), &config); + let analytics_transport = analytics_rpc_transport(&transport); let remote_control_requested = runtime_options.remote_control_enabled; let remote_control_enabled = remote_control_requested && state_db.is_some(); @@ -786,10 +792,8 @@ pub async fn run_main_with_transport_options( }); let processor_handle = tokio::spawn({ - let auth_manager = - AuthManager::shared_from_config(&config, /*enable_codex_api_key_env*/ false).await; - let analytics_events_client = - analytics_events_client_from_config(Arc::clone(&auth_manager), &config); + let auth_manager = Arc::clone(&auth_manager); + let analytics_events_client = analytics_events_client.clone(); let outgoing_message_sender = Arc::new(OutgoingMessageSender::new( outgoing_tx, analytics_events_client.clone(), @@ -810,7 +814,7 @@ pub async fn run_main_with_transport_options( session_source, auth_manager, installation_id, - rpc_transport: analytics_rpc_transport(&transport), + rpc_transport: analytics_transport, remote_control_handle: Some(remote_control_handle.clone()), plugin_startup_tasks: runtime_options.plugin_startup_tasks, })); @@ -1066,6 +1070,15 @@ pub async fn run_main_with_transport_options( info!("processor task exited (channel closed)"); } }); + analytics_events_client.track_app_server_started( + analytics_transport, + app_server_start_started_at + .elapsed() + .as_millis() + .try_into() + .unwrap_or(u64::MAX), + now_unix_seconds(), + ); drop(transport_event_tx); diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index e37c9afd22..d37cfca372 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -410,6 +410,7 @@ impl MessageProcessor { auth_manager.clone(), Arc::clone(&thread_manager), outgoing.clone(), + analytics_events_client.clone(), arg0_paths.clone(), Arc::clone(&config), config_manager.clone(), diff --git a/codex-rs/app-server/src/request_processors/thread_processor.rs b/codex-rs/app-server/src/request_processors/thread_processor.rs index 6b6e1729a5..5ed7407942 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -327,6 +327,7 @@ pub(crate) struct ThreadRequestProcessor { pub(super) auth_manager: Arc, pub(super) thread_manager: Arc, pub(super) outgoing: Arc, + pub(super) analytics_events_client: AnalyticsEventsClient, pub(super) arg0_paths: Arg0DispatchPaths, pub(super) config: Arc, pub(super) config_manager: ConfigManager, @@ -347,6 +348,7 @@ impl ThreadRequestProcessor { auth_manager: Arc, thread_manager: Arc, outgoing: Arc, + analytics_events_client: AnalyticsEventsClient, arg0_paths: Arg0DispatchPaths, config: Arc, config_manager: ConfigManager, @@ -363,6 +365,7 @@ impl ThreadRequestProcessor { auth_manager, thread_manager, outgoing, + analytics_events_client, arg0_paths, config, config_manager, @@ -880,11 +883,13 @@ impl ThreadRequestProcessor { }; let request_trace = request_context.request_trace(); let config_manager = self.config_manager.clone(); + let analytics_events_client = self.analytics_events_client.clone(); let outgoing = Arc::clone(&listener_task_context.outgoing); let error_request_id = request_id.clone(); let thread_start_task = async move { if let Err(error) = Self::thread_start_task( listener_task_context, + analytics_events_client, config_manager, request_id, app_server_client_name, @@ -969,6 +974,7 @@ impl ThreadRequestProcessor { #[allow(clippy::too_many_arguments)] async fn thread_start_task( listener_task_context: ListenerTaskContext, + analytics_events_client: AnalyticsEventsClient, config_manager: ConfigManager, request_id: ConnectionRequestId, app_server_client_name: Option, @@ -1202,6 +1208,14 @@ impl ThreadRequestProcessor { active_permission_profile, reasoning_effort: config_snapshot.reasoning_effort, }; + analytics_events_client.track_thread_start_timing( + thread.id.clone(), + thread_start_started_at + .elapsed() + .as_millis() + .try_into() + .unwrap_or(u64::MAX), + ); let notif = thread_started_notification(thread); listener_task_context .outgoing diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index 4ac511f4fa..24a8abfcbc 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -2,6 +2,7 @@ use std::collections::HashMap; use std::collections::HashSet; use std::sync::Arc; use std::sync::atomic::Ordering; +use std::time::Instant; use crate::SkillInjections; use crate::build_skill_injections; @@ -1733,7 +1734,12 @@ async fn try_run_sampling_request( turn_context.model_info.slug.as_str(), turn_context.provider.info().name.as_str(), ); - let mut stream = client_session + turn_context + .turn_timing_state + .mark_model_request_started() + .await; + let sampling_started_at = Instant::now(); + let stream_result = client_session .stream( prompt, &turn_context.model_info, @@ -1746,7 +1752,24 @@ 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)) => { + turn_context + .turn_timing_state + .record_sampling_duration(sampling_started_at.elapsed()) + .await; + return Err(err); + } + Err(codex_async_utils::CancelErr::Cancelled) => { + turn_context + .turn_timing_state + .record_sampling_duration(sampling_started_at.elapsed()) + .await; + return Err(CodexErr::TurnAborted); + } + }; let mut in_flight: FuturesOrdered>> = FuturesOrdered::new(); let mut needs_follow_up = false; @@ -2151,6 +2174,10 @@ async fn try_run_sampling_request( &mut assistant_message_stream_parsers, ) .await; + turn_context + .turn_timing_state + .record_sampling_duration(sampling_started_at.elapsed()) + .await; if sess .features @@ -2161,7 +2188,12 @@ async fn try_run_sampling_request( client_session.send_response_processed(response_id).await; } + let blocking_tool_started_at = Instant::now(); drain_in_flight(&mut in_flight, sess.clone(), turn_context.clone()).await?; + turn_context + .turn_timing_state + .record_blocking_tool_critical_path_duration(blocking_tool_started_at.elapsed()) + .await; if should_emit_token_count { // A tool call such as request_user_input can intentionally pause the turn. Emit token diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index 164e65f07c..3c42b4bb80 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -33,6 +33,7 @@ use crate::session::turn_context::TurnContext; use crate::state::ActiveTurn; use crate::state::RunningTask; use crate::state::TaskKind; +use codex_analytics::TurnTimingBreakdownFact; use codex_analytics::TurnTokenUsageFact; use codex_login::AuthManager; use codex_models_manager::manager::SharedModelsManager; @@ -767,10 +768,21 @@ impl Session { .turn_timing_state .completed_at_and_duration_ms() .await; + let turn_timing_breakdown = turn_context.turn_timing_state.timing_breakdown().await; let time_to_first_token_ms = turn_context .turn_timing_state .time_to_first_token_ms() .await; + self.services + .analytics_events_client + .track_turn_timing(TurnTimingBreakdownFact { + turn_id: turn_context.sub_id.clone(), + thread_id: self.conversation_id.to_string(), + request_start_delay_ms: turn_timing_breakdown.request_start_delay_ms, + sampling_duration_ms: turn_timing_breakdown.sampling_duration_ms, + blocking_tool_critical_path_duration_ms: turn_timing_breakdown + .blocking_tool_critical_path_duration_ms, + }); if should_clear_active_turn { self.emit_turn_stop_lifecycle(turn_context.extension_data.as_ref()) .await; diff --git a/codex-rs/core/src/turn_timing.rs b/codex-rs/core/src/turn_timing.rs index 570e4a6e32..86de21e556 100644 --- a/codex-rs/core/src/turn_timing.rs +++ b/codex-rs/core/src/turn_timing.rs @@ -36,6 +36,13 @@ pub(crate) async fn record_turn_ttfm_metric(turn_context: &TurnContext, item: &T .record_duration(TURN_TTFM_DURATION_METRIC, duration, &[]); } +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub(crate) struct TurnTimingBreakdown { + pub(crate) request_start_delay_ms: Option, + pub(crate) sampling_duration_ms: u64, + pub(crate) blocking_tool_critical_path_duration_ms: u64, +} + #[derive(Debug, Default)] pub(crate) struct TurnTimingState { state: Mutex, @@ -47,6 +54,9 @@ struct TurnTimingStateInner { started_at_unix_secs: Option, first_token_at: Option, first_message_at: Option, + first_request_started_at: Option, + sampling_duration: Duration, + blocking_tool_critical_path_duration: Duration, } impl TurnTimingState { @@ -57,6 +67,9 @@ impl TurnTimingState { state.started_at_unix_secs = Some(started_at_unix_ms / 1000); state.first_token_at = None; state.first_message_at = None; + state.first_request_started_at = None; + state.sampling_duration = Duration::default(); + state.blocking_tool_critical_path_duration = Duration::default(); started_at_unix_ms } @@ -98,6 +111,36 @@ impl TurnTimingState { let mut state = self.state.lock().await; state.record_turn_ttfm() } + + pub(crate) async fn mark_model_request_started(&self) { + let mut state = self.state.lock().await; + if state.first_request_started_at.is_none() { + state.first_request_started_at = Some(Instant::now()); + } + } + + pub(crate) async fn record_sampling_duration(&self, duration: Duration) { + let mut state = self.state.lock().await; + state.sampling_duration = state.sampling_duration.saturating_add(duration); + } + + pub(crate) async fn record_blocking_tool_critical_path_duration(&self, duration: Duration) { + let mut state = self.state.lock().await; + state.blocking_tool_critical_path_duration = state + .blocking_tool_critical_path_duration + .saturating_add(duration); + } + + pub(crate) async fn timing_breakdown(&self) -> TurnTimingBreakdown { + let state = self.state.lock().await; + TurnTimingBreakdown { + request_start_delay_ms: state.request_start_delay_ms(), + sampling_duration_ms: duration_to_u64_millis(state.sampling_duration), + blocking_tool_critical_path_duration_ms: duration_to_u64_millis( + state.blocking_tool_critical_path_duration, + ), + } + } } fn now_unix_timestamp_secs() -> i64 { @@ -112,6 +155,13 @@ pub(crate) fn now_unix_timestamp_ms() -> i64 { } impl TurnTimingStateInner { + fn request_start_delay_ms(&self) -> Option { + let duration = self + .first_request_started_at? + .duration_since(self.started_at?); + Some(duration_to_u64_millis(duration)) + } + fn time_to_first_token(&self) -> Option { Some(self.first_token_at?.duration_since(self.started_at?)) } @@ -136,6 +186,10 @@ impl TurnTimingStateInner { } } +fn duration_to_u64_millis(duration: Duration) -> u64 { + u64::try_from(duration.as_millis()).unwrap_or(u64::MAX) +} + fn response_event_records_turn_ttft(event: &ResponseEvent) -> bool { match event { ResponseEvent::OutputItemDone(item) | ResponseEvent::OutputItemAdded(item) => { diff --git a/codex-rs/core/src/turn_timing_tests.rs b/codex-rs/core/src/turn_timing_tests.rs index a6675aea8e..93382a9701 100644 --- a/codex-rs/core/src/turn_timing_tests.rs +++ b/codex-rs/core/src/turn_timing_tests.rs @@ -99,6 +99,28 @@ async fn turn_timing_state_records_turn_started_epoch_millis() { ); } +#[tokio::test] +async fn turn_timing_state_tracks_request_start_and_duration_breakdown() { + let state = TurnTimingState::default(); + state.mark_turn_started(Instant::now()).await; + state.mark_model_request_started().await; + state + .record_sampling_duration(std::time::Duration::from_millis(120)) + .await; + state + .record_sampling_duration(std::time::Duration::from_millis(30)) + .await; + state + .record_blocking_tool_critical_path_duration(std::time::Duration::from_millis(45)) + .await; + + let breakdown = state.timing_breakdown().await; + + assert_eq!(breakdown.sampling_duration_ms, 150); + assert_eq!(breakdown.blocking_tool_critical_path_duration_ms, 45); + assert!(breakdown.request_start_delay_ms.is_some()); +} + #[test] fn response_item_records_turn_ttft_for_first_output_signals() { assert!(response_item_records_turn_ttft(