diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index fa64b46d7a..75822e4b52 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; @@ -30,6 +32,7 @@ use crate::events::ReviewTrigger; use crate::events::Reviewer; use crate::events::ThreadInitializedEvent; use crate::events::ThreadInitializedEventParams; +use crate::events::ThreadStartTimingEventParams; use crate::events::ToolItemTerminalStatus; use crate::events::TrackEventRequest; use crate::events::codex_app_metadata; @@ -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,6 +65,7 @@ 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; @@ -1329,6 +1334,12 @@ fn thread_initialized_event_serializes_expected_shape() { initialization_mode: ThreadInitializationMode::New, subagent_source: None, parent_thread_id: None, + thread_start_timing: ThreadStartTimingEventParams { + thread_start_duration_ms: Some(321), + thread_start_prepare_duration_ms: Some(111), + thread_start_spawn_duration_ms: Some(123), + thread_start_finalize_duration_ms: Some(87), + }, created_at: 1, }, }); @@ -1361,12 +1372,49 @@ fn thread_initialized_event_serializes_expected_shape() { "initialization_mode": "new", "subagent_source": null, "parent_thread_id": null, + "thread_start_duration_ms": 321, + "thread_start_prepare_duration_ms": 111, + "thread_start_spawn_duration_ms": 123, + "thread_start_finalize_duration_ms": 87, "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(), + remote_control_enabled: true, + 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" + }, + "remote_control_enabled": true, + "startup_duration_ms": 987, + "completed_at": 12 + } + }) + ); +} + #[test] fn command_execution_event_serializes_expected_shape() { let event = TrackEventRequest::CommandExecution(CodexCommandExecutionEventRequest { @@ -1643,6 +1691,114 @@ 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) + ); + assert_eq!( + payload[0]["event_params"]["thread_start_prepare_duration_ms"], + json!(null) + ); + assert_eq!( + payload[0]["event_params"]["thread_start_spawn_duration_ms"], + json!(null) + ); + assert_eq!( + payload[0]["event_params"]["thread_start_finalize_duration_ms"], + json!(null) + ); +} + +#[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(), + remote_control_enabled: true, + 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" + }, + "remote_control_enabled": true, + "startup_duration_ms": 456, + "completed_at": 12 + } + }]) + ); +} + +#[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, + prepare_duration_ms: 12, + spawn_duration_ms: 123, + finalize_duration_ms: 87, + }, + )), + &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) + ); + assert_eq!( + payload[0]["event_params"]["thread_start_prepare_duration_ms"], + json!(12) + ); + assert_eq!( + payload[0]["event_params"]["thread_start_spawn_duration_ms"], + json!(123) + ); + assert_eq!( + payload[0]["event_params"]["thread_start_finalize_duration_ms"], + json!(87) + ); } #[tokio::test] diff --git a/codex-rs/analytics/src/client.rs b/codex-rs/analytics/src/client.rs index fbcfa32dc5..3124ecc256 100644 --- a/codex-rs/analytics/src/client.rs +++ b/codex-rs/analytics/src/client.rs @@ -8,6 +8,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::CustomAnalyticsFact; use crate::facts::HookRunFact; @@ -17,9 +18,11 @@ use crate::facts::PluginStateChangedInput; 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::TurnTokenUsageFact; +use crate::now_unix_seconds; use crate::reducer::AnalyticsReducer; use codex_app_server_protocol::ClientRequest; use codex_app_server_protocol::ClientResponsePayload; @@ -38,12 +41,35 @@ use std::collections::HashSet; use std::sync::Arc; use std::sync::Mutex; use std::time::Duration; +use std::time::Instant; use tokio::sync::mpsc; const ANALYTICS_EVENTS_QUEUE_SIZE: usize = 256; const ANALYTICS_EVENTS_TIMEOUT: Duration = Duration::from_secs(10); const ANALYTICS_EVENT_DEDUPE_MAX_KEYS: usize = 4096; +#[derive(Clone, Copy, Debug)] +pub struct StartedTimer { + started_at: Instant, +} + +impl StartedTimer { + #[must_use] + pub fn start() -> Self { + Self { + started_at: Instant::now(), + } + } + + fn elapsed_ms(self) -> u64 { + self.started_at + .elapsed() + .as_millis() + .try_into() + .unwrap_or(u64::MAX) + } +} + #[derive(Clone)] pub(crate) struct AnalyticsEventsQueue { pub(crate) sender: mpsc::Sender, @@ -163,12 +189,29 @@ impl AnalyticsEventsClient { }); } + pub fn track_app_server_started(&self, timer: StartedTimer, remote_control_enabled: bool) { + self.record_fact(AnalyticsFact::Custom( + CustomAnalyticsFact::AppServerStarted(AppServerStartedInput { + runtime: current_runtime_metadata(), + remote_control_enabled, + startup_duration_ms: timer.elapsed_ms(), + completed_at: now_unix_seconds(), + }), + )); + } + pub fn track_subagent_thread_started(&self, input: SubAgentThreadStartedInput) { self.record_fact(AnalyticsFact::Custom( CustomAnalyticsFact::SubAgentThreadStarted(input), )); } + pub fn track_thread_start_timing(&self, fact: ThreadStartTimingFact) { + self.record_fact(AnalyticsFact::Custom( + CustomAnalyticsFact::ThreadStartTiming(fact), + )); + } + pub fn track_guardian_review( &self, tracking: &GuardianReviewTrackContext, diff --git a/codex-rs/analytics/src/events.rs b/codex-rs/analytics/src/events.rs index 8791e5767a..4b74331115 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,28 @@ pub(crate) struct CodexRuntimeMetadata { pub(crate) runtime_arch: String, } +#[derive(Serialize)] +pub(crate) struct CodexAppServerStartedEventParams { + pub(crate) runtime: CodexRuntimeMetadata, + pub(crate) remote_control_enabled: bool, + 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(Default, Serialize)] +pub(crate) struct ThreadStartTimingEventParams { + pub(crate) thread_start_duration_ms: Option, + pub(crate) thread_start_prepare_duration_ms: Option, + pub(crate) thread_start_spawn_duration_ms: Option, + pub(crate) thread_start_finalize_duration_ms: Option, +} + #[derive(Serialize)] pub(crate) struct ThreadInitializedEventParams { pub(crate) thread_id: String, @@ -156,6 +179,8 @@ pub(crate) struct ThreadInitializedEventParams { pub(crate) initialization_mode: ThreadInitializationMode, pub(crate) subagent_source: Option, pub(crate) parent_thread_id: Option, + #[serde(flatten)] + pub(crate) thread_start_timing: ThreadStartTimingEventParams, pub(crate) created_at: u64, } @@ -1050,6 +1075,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_timing: ThreadStartTimingEventParams::default(), created_at: input.created_at, }; ThreadInitializedEvent { diff --git a/codex-rs/analytics/src/facts.rs b/codex-rs/analytics/src/facts.rs index 9a85276712..4c043ac041 100644 --- a/codex-rs/analytics/src/facts.rs +++ b/codex-rs/analytics/src/facts.rs @@ -275,6 +275,21 @@ pub struct CodexCompactionEvent { pub duration_ms: Option, } +pub(crate) struct AppServerStartedInput { + pub runtime: CodexRuntimeMetadata, + pub remote_control_enabled: bool, + pub startup_duration_ms: u64, + pub completed_at: u64, +} + +pub struct ThreadStartTimingFact { + pub thread_id: String, + pub duration_ms: u64, + pub prepare_duration_ms: u64, + pub spawn_duration_ms: u64, + pub finalize_duration_ms: u64, +} + #[allow(dead_code)] pub(crate) enum AnalyticsFact { Initialize { @@ -324,7 +339,9 @@ pub(crate) enum AnalyticsFact { } pub(crate) enum CustomAnalyticsFact { + AppServerStarted(AppServerStartedInput), SubAgentThreadStarted(SubAgentThreadStartedInput), + ThreadStartTiming(ThreadStartTimingFact), Compaction(Box), GuardianReview(Box), TurnResolvedConfig(Box), diff --git a/codex-rs/analytics/src/lib.rs b/codex-rs/analytics/src/lib.rs index a33ca7b9e3..d4bbcabc11 100644 --- a/codex-rs/analytics/src/lib.rs +++ b/codex-rs/analytics/src/lib.rs @@ -10,6 +10,7 @@ use std::time::UNIX_EPOCH; pub use accepted_lines::accepted_line_fingerprints_from_unified_diff; pub use accepted_lines::fingerprint_hash; pub use client::AnalyticsEventsClient; +pub use client::StartedTimer; pub use events::AppServerRpcTransport; pub use events::GuardianApprovalRequestSource; pub use events::GuardianReviewAnalyticsResult; @@ -37,6 +38,7 @@ pub use facts::InvocationType; pub use facts::SkillInvocation; pub use facts::SubAgentThreadStartedInput; pub use facts::ThreadInitializationMode; +pub use facts::ThreadStartTimingFact; 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 f2f036b8f4..6b69fea18c 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; @@ -45,6 +47,7 @@ use crate::events::SkillInvocationEventParams; use crate::events::SkillInvocationEventRequest; use crate::events::ThreadInitializedEvent; use crate::events::ThreadInitializedEventParams; +use crate::events::ThreadStartTimingEventParams; use crate::events::ToolItemFailureKind; use crate::events::ToolItemTerminalStatus; use crate::events::TrackEventRequest; @@ -61,6 +64,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,6 +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::TurnResolvedConfigFact; use crate::facts::TurnStatus; use crate::facts::TurnSteerRejectionReason; @@ -147,6 +152,15 @@ 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, } #[derive(Clone, Copy)] @@ -449,9 +463,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); } @@ -511,6 +531,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, + remote_control_enabled: input.remote_control_enabled, + startup_duration_ms: input.startup_duration_ms, + completed_at: input.completed_at, + }, + }, + )); + } + fn ingest_subagent_thread_started( &mut self, input: SubAgentThreadStartedInput, @@ -542,6 +580,19 @@ impl AnalyticsReducer { )); } + 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, + }; + self.threads + .entry(input.thread_id) + .or_default() + .thread_start_timing = Some(thread_start_timing); + } + fn ingest_guardian_review( &mut self, input: GuardianReviewEventParams, @@ -1247,13 +1298,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_timing = matches!(initialization_mode, ThreadInitializationMode::New) + .then_some(thread_state.thread_start_timing) + .flatten(); out.push(TrackEventRequest::ThreadInitialized( ThreadInitializedEvent { event_type: "codex_thread_initialized", @@ -1268,6 +1318,15 @@ impl AnalyticsReducer { initialization_mode, subagent_source: thread_metadata.subagent_source.clone(), parent_thread_id: thread_metadata.parent_thread_id, + thread_start_timing: match thread_start_timing { + Some(timing) => ThreadStartTimingEventParams { + thread_start_duration_ms: Some(timing.duration_ms), + thread_start_prepare_duration_ms: Some(timing.prepare_duration_ms), + thread_start_spawn_duration_ms: Some(timing.spawn_duration_ms), + thread_start_finalize_duration_ms: Some(timing.finalize_duration_ms), + }, + None => ThreadStartTimingEventParams::default(), + }, created_at: u64::try_from(thread.created_at).unwrap_or_default(), }, }, diff --git a/codex-rs/app-server/src/lib.rs b/codex-rs/app-server/src/lib.rs index 80305d2d9f..16d79e5741 100644 --- a/codex-rs/app-server/src/lib.rs +++ b/codex-rs/app-server/src/lib.rs @@ -41,6 +41,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::StartedTimer; use codex_app_server_protocol::ConfigLayerSource; use codex_app_server_protocol::ConfigWarningNotification; use codex_app_server_protocol::JSONRPCMessage; @@ -428,6 +429,7 @@ pub async fn run_main_with_transport_options( auth: AppServerWebsocketAuthSettings, runtime_options: AppServerRuntimeOptions, ) -> IoResult<()> { + let app_server_start_timer = StartedTimer::start(); let (transport_event_tx, mut transport_event_rx) = mpsc::channel::(CHANNEL_CAPACITY); let (outgoing_tx, mut outgoing_rx) = mpsc::channel::(CHANNEL_CAPACITY); @@ -698,6 +700,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(); @@ -787,8 +792,7 @@ pub async fn run_main_with_transport_options( let processor_handle = tokio::spawn({ let auth_manager = Arc::clone(&auth_manager); - let analytics_events_client = - analytics_events_client_from_config(Arc::clone(&auth_manager), &config); + let analytics_events_client = analytics_events_client.clone(); let outgoing_message_sender = Arc::new(OutgoingMessageSender::new( outgoing_tx, analytics_events_client.clone(), @@ -809,7 +813,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, })); @@ -1065,6 +1069,8 @@ pub async fn run_main_with_transport_options( info!("processor task exited (channel closed)"); } }); + analytics_events_client + .track_app_server_started(app_server_start_timer, remote_control_enabled); drop(transport_event_tx); diff --git a/codex-rs/app-server/tests/suite/v2/analytics.rs b/codex-rs/app-server/tests/suite/v2/analytics.rs index 8e8f2a7557..bb2306e1b9 100644 --- a/codex-rs/app-server/tests/suite/v2/analytics.rs +++ b/codex-rs/app-server/tests/suite/v2/analytics.rs @@ -21,6 +21,12 @@ use wiremock::matchers::path; const SERVICE_VERSION: &str = "0.0.0-test"; +#[cfg(any(target_os = "macos", windows))] +pub(crate) const ANALYTICS_TEST_TIMEOUT: Duration = Duration::from_secs(25); + +#[cfg(not(any(target_os = "macos", windows)))] +pub(crate) const ANALYTICS_TEST_TIMEOUT: Duration = Duration::from_secs(10); + fn set_metrics_exporter(config: &mut codex_core::config::Config) { config.otel.metrics_exporter = OtelExporterKind::OtlpHttp { endpoint: "http://localhost:4318".to_string(), @@ -98,28 +104,6 @@ pub(crate) async fn mount_analytics_capture(server: &MockServer, codex_home: &Pa Ok(()) } -pub(crate) async fn wait_for_analytics_payload( - server: &MockServer, - read_timeout: Duration, -) -> Result { - let body = timeout(read_timeout, async { - loop { - let Some(requests) = server.received_requests().await else { - tokio::time::sleep(Duration::from_millis(25)).await; - continue; - }; - if let Some(request) = requests.iter().find(|request| { - request.method == "POST" && request.url.path() == "/codex/analytics-events/events" - }) { - break request.body.clone(); - } - tokio::time::sleep(Duration::from_millis(25)).await; - } - }) - .await?; - serde_json::from_slice(&body).map_err(|err| anyhow::anyhow!("invalid analytics payload: {err}")) -} - pub(crate) async fn wait_for_analytics_event( server: &MockServer, read_timeout: Duration, @@ -133,7 +117,10 @@ pub(crate) async fn wait_for_analytics_event( }; for request in &requests { if request.method != "POST" - || request.url.path() != "/codex/analytics-events/events" + || !request + .url + .path() + .ends_with("/codex/analytics-events/events") { continue; } @@ -155,16 +142,6 @@ pub(crate) async fn wait_for_analytics_event( .await? } -pub(crate) fn thread_initialized_event(payload: &Value) -> Result<&Value> { - let events = payload["events"] - .as_array() - .ok_or_else(|| anyhow::anyhow!("analytics payload missing events array"))?; - events - .iter() - .find(|event| event["event_type"] == "codex_thread_initialized") - .ok_or_else(|| anyhow::anyhow!("codex_thread_initialized event should be present")) -} - pub(crate) fn assert_basic_thread_initialized_event( event: &Value, thread_id: &str, diff --git a/codex-rs/app-server/tests/suite/v2/plugin_install.rs b/codex-rs/app-server/tests/suite/v2/plugin_install.rs index 9d17e8747f..bd80ce4a3a 100644 --- a/codex-rs/app-server/tests/suite/v2/plugin_install.rs +++ b/codex-rs/app-server/tests/suite/v2/plugin_install.rs @@ -58,6 +58,8 @@ use wiremock::matchers::method; use wiremock::matchers::path; use wiremock::matchers::query_param; +use super::analytics::wait_for_analytics_event; + // Plugin install tests wait on connector discovery after the install response path // starts, which is noticeably slower on Windows CI. const DEFAULT_TIMEOUT: Duration = Duration::from_secs(60); @@ -727,22 +729,22 @@ async fn plugin_install_tracks_analytics_event() -> Result<()> { let response: PluginInstallResponse = to_response(response)?; assert_eq!(response.apps_needing_auth, Vec::::new()); - let payload = wait_for_plugin_analytics_payload(&analytics_server).await?; + let event = + wait_for_analytics_event(&analytics_server, DEFAULT_TIMEOUT, "codex_plugin_installed") + .await?; assert_eq!( - payload, + event, json!({ - "events": [{ - "event_type": "codex_plugin_installed", - "event_params": { - "plugin_id": "sample-plugin@debug", - "plugin_name": "sample-plugin", - "marketplace_name": "debug", - "has_skills": false, - "mcp_server_count": 0, - "connector_ids": [], - "product_client_id": DEFAULT_CLIENT_NAME, - } - }] + "event_type": "codex_plugin_installed", + "event_params": { + "plugin_id": "sample-plugin@debug", + "plugin_name": "sample-plugin", + "marketplace_name": "debug", + "has_skills": false, + "mcp_server_count": 0, + "connector_ids": [], + "product_client_id": DEFAULT_CLIENT_NAME, + } }) ); Ok(()) @@ -780,22 +782,21 @@ async fn plugin_install_tracks_remote_plugin_analytics_event() -> Result<()> { let response: PluginInstallResponse = to_response(response)?; assert_eq!(response.apps_needing_auth, Vec::::new()); - let payload = wait_for_plugin_analytics_payload(&server).await?; + let event = + wait_for_analytics_event(&server, DEFAULT_TIMEOUT, "codex_plugin_installed").await?; assert_eq!( - payload, + event, json!({ - "events": [{ - "event_type": "codex_plugin_installed", - "event_params": { - "plugin_id": REMOTE_PLUGIN_ID, - "plugin_name": "linear", - "marketplace_name": "openai-curated-remote", - "has_skills": true, - "mcp_server_count": 0, - "connector_ids": [], - "product_client_id": DEFAULT_CLIENT_NAME, - } - }] + "event_type": "codex_plugin_installed", + "event_params": { + "plugin_id": REMOTE_PLUGIN_ID, + "plugin_name": "linear", + "marketplace_name": "openai-curated-remote", + "has_skills": true, + "mcp_server_count": 0, + "connector_ids": [], + "product_client_id": DEFAULT_CLIENT_NAME, + } }) ); Ok(()) @@ -1288,29 +1289,6 @@ async fn mount_backend_analytics_events(server: &MockServer) { .await; } -async fn wait_for_plugin_analytics_payload(server: &MockServer) -> Result { - timeout(DEFAULT_TIMEOUT, async { - loop { - let Some(requests) = server.received_requests().await else { - tokio::time::sleep(Duration::from_millis(25)).await; - continue; - }; - if let Some(request) = requests.iter().find(|request| { - request.method == "POST" - && request - .url - .path() - .ends_with("/codex/analytics-events/events") - }) { - return serde_json::from_slice(&request.body) - .map_err(|err| anyhow::anyhow!("invalid analytics payload: {err}")); - } - tokio::time::sleep(Duration::from_millis(25)).await; - } - }) - .await? -} - fn write_remote_plugin_catalog_config( codex_home: &std::path::Path, base_url: &str, diff --git a/codex-rs/app-server/tests/suite/v2/plugin_share.rs b/codex-rs/app-server/tests/suite/v2/plugin_share.rs index 69f93f0b7e..dd1dba4ef2 100644 --- a/codex-rs/app-server/tests/suite/v2/plugin_share.rs +++ b/codex-rs/app-server/tests/suite/v2/plugin_share.rs @@ -383,7 +383,13 @@ plugin_sharing = false .received_requests() .await .expect("wiremock should record requests") - .is_empty() + .iter() + .all(|request| { + request + .url + .path() + .ends_with("/codex/analytics-events/events") + }) ); Ok(()) } diff --git a/codex-rs/app-server/tests/suite/v2/plugin_uninstall.rs b/codex-rs/app-server/tests/suite/v2/plugin_uninstall.rs index 4feede934d..5256dec571 100644 --- a/codex-rs/app-server/tests/suite/v2/plugin_uninstall.rs +++ b/codex-rs/app-server/tests/suite/v2/plugin_uninstall.rs @@ -24,6 +24,8 @@ use wiremock::matchers::header; use wiremock::matchers::method; use wiremock::matchers::path; +use super::analytics::wait_for_analytics_event; + const DEFAULT_TIMEOUT: Duration = Duration::from_secs(10); const REMOTE_PLUGIN_ID: &str = "plugins~Plugin_linear"; const WORKSPACE_REMOTE_PLUGIN_ID: &str = "plugins_69f27c3e67848191a45cbaa5f2adb39d"; @@ -116,37 +118,25 @@ async fn plugin_uninstall_tracks_analytics_event() -> Result<()> { let response: PluginUninstallResponse = to_response(response)?; assert_eq!(response, PluginUninstallResponse {}); - let payload = timeout(DEFAULT_TIMEOUT, async { - loop { - let Some(requests) = analytics_server.received_requests().await else { - tokio::time::sleep(Duration::from_millis(25)).await; - continue; - }; - if let Some(request) = requests.iter().find(|request| { - request.method == "POST" && request.url.path() == "/codex/analytics-events/events" - }) { - break request.body.clone(); - } - tokio::time::sleep(Duration::from_millis(25)).await; - } - }) + let event = wait_for_analytics_event( + &analytics_server, + DEFAULT_TIMEOUT, + "codex_plugin_uninstalled", + ) .await?; - let payload: serde_json::Value = serde_json::from_slice(&payload).expect("analytics payload"); assert_eq!( - payload, + event, json!({ - "events": [{ - "event_type": "codex_plugin_uninstalled", - "event_params": { - "plugin_id": "sample-plugin@debug", - "plugin_name": "sample-plugin", - "marketplace_name": "debug", - "has_skills": false, - "mcp_server_count": 0, - "connector_ids": [], - "product_client_id": DEFAULT_CLIENT_NAME, - } - }] + "event_type": "codex_plugin_uninstalled", + "event_params": { + "plugin_id": "sample-plugin@debug", + "plugin_name": "sample-plugin", + "marketplace_name": "debug", + "has_skills": false, + "mcp_server_count": 0, + "connector_ids": [], + "product_client_id": DEFAULT_CLIENT_NAME, + } }) ); Ok(()) diff --git a/codex-rs/app-server/tests/suite/v2/remote_control.rs b/codex-rs/app-server/tests/suite/v2/remote_control.rs index d71f05bf58..2422558497 100644 --- a/codex-rs/app-server/tests/suite/v2/remote_control.rs +++ b/codex-rs/app-server/tests/suite/v2/remote_control.rs @@ -17,6 +17,8 @@ use codex_config::types::AuthCredentialsStoreMode; use pretty_assertions::assert_eq; use tempfile::TempDir; use tokio::io::AsyncBufReadExt; +use tokio::io::AsyncReadExt; +use tokio::io::AsyncWriteExt; use tokio::io::BufReader; use tokio::net::TcpListener; use tokio::net::TcpStream; @@ -150,8 +152,12 @@ impl BlockingRemoteControlBackend { let (enroll_request_tx, enroll_request_rx) = oneshot::channel(); let server_task = tokio::spawn(async move { match read_enroll_request(listener).await { - Ok((request_line, _reader)) => { + Ok(PendingHttpRequest { + request_line, + connection, + }) => { let _ = enroll_request_tx.send(Ok(request_line)); + let _connection = connection; std::future::pending::<()>().await; } Err(err) => { @@ -181,20 +187,48 @@ impl Drop for BlockingRemoteControlBackend { } } -async fn read_enroll_request(listener: TcpListener) -> Result<(String, BufReader)> { - let (stream, _) = listener.accept().await?; - let mut reader = BufReader::new(stream); - - let mut request_line = String::new(); - reader.read_line(&mut request_line).await?; - - loop { - let mut line = String::new(); - reader.read_line(&mut line).await?; - if line == "\r\n" { - break; - } - } - - Ok((request_line.trim_end().to_string(), reader)) +struct PendingHttpRequest { + request_line: String, + connection: BufReader, +} + +async fn read_enroll_request(listener: TcpListener) -> Result { + loop { + let (stream, _) = listener.accept().await?; + let mut reader = BufReader::new(stream); + + let mut request_line = String::new(); + reader.read_line(&mut request_line).await?; + + let mut content_length = 0usize; + loop { + let mut line = String::new(); + reader.read_line(&mut line).await?; + if let Some(value) = line.strip_prefix("Content-Length: ") { + content_length = value.trim().parse().unwrap_or_default(); + } + if line == "\r\n" { + break; + } + } + + if content_length > 0 { + let mut body = vec![0; content_length]; + reader.read_exact(&mut body).await?; + } + + let request_line = request_line.trim_end().to_string(); + if request_line.contains("/codex/analytics-events/events ") { + reader + .get_mut() + .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n") + .await?; + continue; + } + + return Ok(PendingHttpRequest { + request_line, + connection: reader, + }); + } } diff --git a/codex-rs/app-server/tests/suite/v2/thread_fork.rs b/codex-rs/app-server/tests/suite/v2/thread_fork.rs index 287afe1016..6ef2662db4 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_fork.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_fork.rs @@ -41,10 +41,10 @@ use wiremock::ResponseTemplate; use wiremock::matchers::method; use wiremock::matchers::path; +use super::analytics::ANALYTICS_TEST_TIMEOUT; use super::analytics::assert_basic_thread_initialized_event; use super::analytics::mount_analytics_capture; -use super::analytics::thread_initialized_event; -use super::analytics::wait_for_analytics_payload; +use super::analytics::wait_for_analytics_event; #[cfg(windows)] const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(25); @@ -403,10 +403,11 @@ async fn thread_fork_tracks_thread_initialized_analytics() -> Result<()> { .await??; let ThreadForkResponse { thread, .. } = to_response::(fork_resp)?; - let payload = wait_for_analytics_payload(&server, DEFAULT_READ_TIMEOUT).await?; - let event = thread_initialized_event(&payload)?; + let event = + wait_for_analytics_event(&server, ANALYTICS_TEST_TIMEOUT, "codex_thread_initialized") + .await?; assert_basic_thread_initialized_event( - event, + &event, &thread.id, &thread.session_id, "mock-model", diff --git a/codex-rs/app-server/tests/suite/v2/thread_resume.rs b/codex-rs/app-server/tests/suite/v2/thread_resume.rs index bb2de2cfbb..1f7a873081 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_resume.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_resume.rs @@ -93,10 +93,10 @@ use wiremock::ResponseTemplate; use wiremock::matchers::method; use wiremock::matchers::path; +use super::analytics::ANALYTICS_TEST_TIMEOUT; use super::analytics::assert_basic_thread_initialized_event; use super::analytics::mount_analytics_capture; -use super::analytics::thread_initialized_event; -use super::analytics::wait_for_analytics_payload; +use super::analytics::wait_for_analytics_event; #[cfg(windows)] const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(25); @@ -420,10 +420,11 @@ async fn thread_resume_tracks_thread_initialized_analytics() -> Result<()> { ); assert_eq!(thread.thread_source, Some(ThreadSource::User)); - let payload = wait_for_analytics_payload(&server, DEFAULT_READ_TIMEOUT).await?; - let event = thread_initialized_event(&payload)?; + let event = + wait_for_analytics_event(&server, ANALYTICS_TEST_TIMEOUT, "codex_thread_initialized") + .await?; assert_basic_thread_initialized_event( - event, + &event, &thread.id, &thread.session_id, "gpt-5.3-codex", 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 6be6d838ba..ddca8dc178 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_start.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_start.rs @@ -44,10 +44,10 @@ use wiremock::ResponseTemplate; use wiremock::matchers::method; use wiremock::matchers::path; +use super::analytics::ANALYTICS_TEST_TIMEOUT; use super::analytics::assert_basic_thread_initialized_event; use super::analytics::mount_analytics_capture; -use super::analytics::thread_initialized_event; -use super::analytics::wait_for_analytics_payload; +use super::analytics::wait_for_analytics_event; const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); const INVALID_REQUEST_ERROR_CODE: i64 = -32600; @@ -436,17 +436,25 @@ async fn thread_start_tracks_thread_initialized_analytics() -> Result<()> { .await??; let ThreadStartResponse { thread, .. } = to_response::(resp)?; - let payload = wait_for_analytics_payload(&server, DEFAULT_READ_TIMEOUT).await?; - assert_eq!(payload["events"].as_array().expect("events array").len(), 1); - let event = thread_initialized_event(&payload)?; + let event = + wait_for_analytics_event(&server, ANALYTICS_TEST_TIMEOUT, "codex_thread_initialized") + .await?; assert_basic_thread_initialized_event( - event, + &event, &thread.id, &thread.session_id, "mock-model", "new", "user", ); + let thread_start_timings = [ + "thread_start_duration_ms", + "thread_start_prepare_duration_ms", + "thread_start_spawn_duration_ms", + "thread_start_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 aa5784d4fe..bd83a0c6a7 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -138,6 +138,7 @@ pub(crate) mod state_db_bridge; pub use state_db_bridge::StateDbHandle; pub use state_db_bridge::init_state_db; 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_manager.rs b/codex-rs/core/src/thread_manager.rs index d209026b62..ba52a09128 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -15,6 +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 codex_analytics::AnalyticsEventsClient; use codex_app_server_protocol::ThreadHistoryBuilder; use codex_app_server_protocol::TurnStatus; @@ -1219,6 +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(); if let InitialHistory::Resumed(resumed) = &initial_history { let mut threads = self.threads.write().await; if let Some(thread) = threads.get(&resumed.conversation_id).cloned() { @@ -1231,6 +1233,12 @@ impl ThreadManagerState { resumed.conversation_id ))); } + thread_start_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()), + ); + } return Ok(NewThread { thread_id: resumed.conversation_id, session_configured: thread.session_configured(), @@ -1245,6 +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(); let tracked_session_source = session_source.clone(); let CodexSpawnOk { codex, thread_id, .. @@ -1277,9 +1286,16 @@ impl ThreadManagerState { attestation_provider: self.attestation_provider.clone(), }) .await?; + thread_start_timing.mark_spawn_completed(); let new_thread = self .finalize_thread_spawn(codex, thread_id, tracked_session_source) .await?; + thread_start_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()), + ); + } if is_resumed_thread { new_thread.thread.emit_thread_resume_lifecycle().await; if let Err(err) = new_thread.thread.apply_goal_resume_runtime_effects().await { diff --git a/codex-rs/core/src/thread_start_timing.rs b/codex-rs/core/src/thread_start_timing.rs new file mode 100644 index 0000000000..0cf5457d5a --- /dev/null +++ b/codex-rs/core/src/thread_start_timing.rs @@ -0,0 +1,58 @@ +use std::time::Duration; +use std::time::Instant; + +use codex_analytics::ThreadStartTimingFact; + +#[derive(Debug)] +pub(crate) struct ThreadStartTiming { + phase_started_at: Instant, + prepare_duration: Option, + spawn_duration: Option, + finalize_duration: Option, +} + +impl ThreadStartTiming { + pub(crate) fn start() -> Self { + Self { + phase_started_at: Instant::now(), + prepare_duration: None, + spawn_duration: None, + finalize_duration: None, + } + } + + pub(crate) fn mark_prepare_completed(&mut self) { + self.prepare_duration = Some(self.finish_phase()); + } + + pub(crate) fn mark_spawn_completed(&mut self) { + self.spawn_duration = Some(self.finish_phase()); + } + + pub(crate) fn mark_finalize_completed(&mut self) { + self.finalize_duration = Some(self.finish_phase()); + } + + pub(crate) fn into_fact(self, thread_id: String) -> ThreadStartTimingFact { + 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 { + thread_id, + duration_ms: duration_ms(prepare_duration + spawn_duration + finalize_duration), + prepare_duration_ms: duration_ms(prepare_duration), + spawn_duration_ms: duration_ms(spawn_duration), + finalize_duration_ms: duration_ms(finalize_duration), + } + } + + fn finish_phase(&mut self) -> Duration { + let duration = self.phase_started_at.elapsed(); + self.phase_started_at = Instant::now(); + duration + } +} + +fn duration_ms(duration: Duration) -> u64 { + duration.as_millis().try_into().unwrap_or(u64::MAX) +}