From 36ecc2047a0f8337623f0f4c7ab87da9acc24b53 Mon Sep 17 00:00:00 2001 From: Roy Han Date: Thu, 26 Mar 2026 21:59:22 -0700 Subject: [PATCH 1/8] add response to turn facts --- codex-rs/analytics/src/analytics_client.rs | 72 +++++++++++--- .../analytics/src/analytics_client_tests.rs | 98 +++++++++++++++---- codex-rs/analytics/src/lib.rs | 1 - .../app-server/src/codex_message_processor.rs | 60 +++++------- codex-rs/core/src/lib.rs | 2 - 5 files changed, 156 insertions(+), 77 deletions(-) diff --git a/codex-rs/analytics/src/analytics_client.rs b/codex-rs/analytics/src/analytics_client.rs index f178aa0418..eadcfe876f 100644 --- a/codex-rs/analytics/src/analytics_client.rs +++ b/codex-rs/analytics/src/analytics_client.rs @@ -1,4 +1,5 @@ use codex_app_server_protocol::ClientRequest; +use codex_app_server_protocol::ClientResponse; use codex_app_server_protocol::InitializeParams; use codex_app_server_protocol::RequestId; use codex_app_server_protocol::ServerNotification; @@ -30,7 +31,7 @@ pub struct TrackEventsContext { } #[derive(Clone)] -pub struct ThreadInitializedInput { +struct ThreadInitializedInput { pub connection_id: u64, pub thread_id: String, pub model: String, @@ -88,11 +89,15 @@ pub enum AnalyticsFact { Request { connection_id: u64, request_id: RequestId, - request: ClientRequest, + request: Box, + }, + Response { + connection_id: u64, + response: Box, }, Notification { connection_id: u64, - notification: ServerNotification, + notification: Box, }, // Facts that do not naturally exist on the app-server protocol surface, or // would require non-trivial protocol reshaping on this branch. @@ -100,9 +105,6 @@ pub enum AnalyticsFact { } pub enum CustomAnalyticsFact { - // This remains custom on this branch because app-server-protocol does not - // yet expose a generic client response enum we can reduce over directly. - ThreadInitialized(ThreadInitializedInput), SkillInvoked(SkillInvokedInput), AppMentioned(AppMentionedInput), AppUsed(AppUsedInput), @@ -255,12 +257,6 @@ impl AnalyticsEventsClient { }); } - pub fn track_thread_initialized(&self, input: ThreadInitializedInput) { - self.record_fact(AnalyticsFact::Custom( - CustomAnalyticsFact::ThreadInitialized(input), - )); - } - pub fn track_app_mentioned(&self, tracking: TrackEventsContext, mentions: Vec) { if mentions.is_empty() { return; @@ -330,6 +326,13 @@ impl AnalyticsEventsClient { } self.queue.try_send(input); } + + pub fn track_response(&self, connection_id: u64, response: ClientResponse) { + self.record_fact(AnalyticsFact::Response { + connection_id, + response: Box::new(response), + }); + } } const ANALYTICS_EVENTS_QUEUE_SIZE: usize = 256; @@ -461,14 +464,17 @@ impl AnalyticsReducer { request_id: _request_id, request: _request, } => {} + AnalyticsFact::Response { + connection_id, + response, + } => { + self.ingest_response(connection_id, *response, out); + } AnalyticsFact::Notification { connection_id: _connection_id, notification: _notification, } => {} AnalyticsFact::Custom(input) => match input { - CustomAnalyticsFact::ThreadInitialized(input) => { - self.ingest_thread_initialized(input, out); - } CustomAnalyticsFact::SkillInvoked(input) => { self.ingest_skill_invoked(input, out).await; } @@ -510,6 +516,42 @@ impl AnalyticsReducer { )); } + fn ingest_response( + &mut self, + connection_id: u64, + response: ClientResponse, + out: &mut Vec, + ) { + let input = match response { + ClientResponse::ThreadStart { response, .. } => ThreadInitializedInput { + connection_id, + thread_id: response.thread.id, + model: response.model, + ephemeral: response.thread.ephemeral, + session_source: response.thread.source.into(), + initialization_mode: InitializationMode::New, + }, + ClientResponse::ThreadResume { response, .. } => ThreadInitializedInput { + connection_id, + thread_id: response.thread.id, + model: response.model, + ephemeral: response.thread.ephemeral, + session_source: response.thread.source.into(), + initialization_mode: InitializationMode::Resumed, + }, + ClientResponse::ThreadFork { response, .. } => ThreadInitializedInput { + connection_id, + thread_id: response.thread.id, + model: response.model, + ephemeral: response.thread.ephemeral, + session_source: response.thread.source.into(), + initialization_mode: InitializationMode::Forked, + }, + _ => return, + }; + self.ingest_thread_initialized(input, out); + } + async fn ingest_skill_invoked( &mut self, input: SkillInvokedInput, diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index b9b918a76a..6156015bef 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -6,7 +6,6 @@ use super::CodexAppMentionedEventRequest; use super::CodexAppUsedEventRequest; use super::CodexPluginEventRequest; use super::CodexPluginUsedEventRequest; -use super::CustomAnalyticsFact; use super::InitializationMode; use super::InvocationType; use super::ThreadInitializedInput; @@ -17,8 +16,18 @@ use super::codex_plugin_metadata; use super::codex_plugin_used_metadata; use super::codex_thread_initialized_event_request; use super::normalize_path_for_skill_id; +use codex_app_server_protocol::ApprovalsReviewer as AppServerApprovalsReviewer; +use codex_app_server_protocol::AskForApproval as AppServerAskForApproval; use codex_app_server_protocol::ClientInfo; +use codex_app_server_protocol::ClientResponse; use codex_app_server_protocol::InitializeParams; +use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::SandboxPolicy as AppServerSandboxPolicy; +use codex_app_server_protocol::SessionSource as AppServerSessionSource; +use codex_app_server_protocol::Thread; +use codex_app_server_protocol::ThreadResumeResponse; +use codex_app_server_protocol::ThreadStartResponse; +use codex_app_server_protocol::ThreadStatus as AppServerThreadStatus; use codex_login::default_client::originator; use codex_plugin::AppConnectorId; use codex_plugin::PluginCapabilitySummary; @@ -33,6 +42,61 @@ use std::sync::Arc; use std::sync::Mutex; use tokio::sync::mpsc; +fn sample_thread(thread_id: &str, ephemeral: bool) -> Thread { + Thread { + id: thread_id.to_string(), + preview: "first prompt".to_string(), + ephemeral, + model_provider: "openai".to_string(), + created_at: 1, + updated_at: 2, + status: AppServerThreadStatus::Idle, + path: None, + cwd: PathBuf::from("/tmp"), + cli_version: "0.0.0".to_string(), + source: AppServerSessionSource::Exec, + agent_nickname: None, + agent_role: None, + git_info: None, + name: None, + turns: Vec::new(), + } +} + +fn sample_thread_start_response(thread_id: &str, ephemeral: bool, model: &str) -> ClientResponse { + ClientResponse::ThreadStart { + request_id: RequestId::Integer(1), + response: ThreadStartResponse { + thread: sample_thread(thread_id, ephemeral), + model: model.to_string(), + model_provider: "openai".to_string(), + service_tier: None, + cwd: PathBuf::from("/tmp"), + approval_policy: AppServerAskForApproval::OnFailure, + approvals_reviewer: AppServerApprovalsReviewer::User, + sandbox: AppServerSandboxPolicy::DangerFullAccess, + reasoning_effort: None, + }, + } +} + +fn sample_thread_resume_response(thread_id: &str, ephemeral: bool, model: &str) -> ClientResponse { + ClientResponse::ThreadResume { + request_id: RequestId::Integer(2), + response: ThreadResumeResponse { + thread: sample_thread(thread_id, ephemeral), + model: model.to_string(), + model_provider: "openai".to_string(), + service_tier: None, + cwd: PathBuf::from("/tmp"), + approval_policy: AppServerAskForApproval::OnFailure, + approvals_reviewer: AppServerApprovalsReviewer::User, + sandbox: AppServerSandboxPolicy::DangerFullAccess, + reasoning_effort: None, + }, + } +} + fn expected_absolute_path(path: &PathBuf) -> String { std::fs::canonicalize(path) .unwrap_or_else(|_| path.to_path_buf()) @@ -238,16 +302,14 @@ async fn initialize_caches_client_and_thread_lifecycle_publishes_once_initialize reducer .ingest( - AnalyticsFact::Custom(CustomAnalyticsFact::ThreadInitialized( - ThreadInitializedInput { - connection_id: 7, - thread_id: "thread-no-client".to_string(), - model: "gpt-5".to_string(), - ephemeral: false, - session_source: SessionSource::Exec, - initialization_mode: InitializationMode::New, - }, - )), + AnalyticsFact::Response { + connection_id: 7, + response: Box::new(sample_thread_start_response( + "thread-no-client", + false, + "gpt-5", + )), + }, &mut events, ) .await; @@ -273,16 +335,10 @@ async fn initialize_caches_client_and_thread_lifecycle_publishes_once_initialize reducer .ingest( - AnalyticsFact::Custom(CustomAnalyticsFact::ThreadInitialized( - ThreadInitializedInput { - connection_id: 7, - thread_id: "thread-1".to_string(), - model: "gpt-5".to_string(), - ephemeral: true, - session_source: SessionSource::Exec, - initialization_mode: InitializationMode::Resumed, - }, - )), + AnalyticsFact::Response { + connection_id: 7, + response: Box::new(sample_thread_resume_response("thread-1", true, "gpt-5")), + }, &mut events, ) .await; diff --git a/codex-rs/analytics/src/lib.rs b/codex-rs/analytics/src/lib.rs index 1efe63904a..4ae9d9b51a 100644 --- a/codex-rs/analytics/src/lib.rs +++ b/codex-rs/analytics/src/lib.rs @@ -14,6 +14,5 @@ pub use analytics_client::PluginStateChangedInput; pub use analytics_client::PluginUsedInput; pub use analytics_client::SkillInvocation; pub use analytics_client::SkillInvokedInput; -pub use analytics_client::ThreadInitializedInput; pub use analytics_client::TrackEventsContext; pub use analytics_client::build_track_events_context; diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index 5918fad0af..31af6070cf 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -32,6 +32,7 @@ use codex_app_server_protocol::CancelLoginAccountParams; use codex_app_server_protocol::CancelLoginAccountResponse; use codex_app_server_protocol::CancelLoginAccountStatus; use codex_app_server_protocol::ClientRequest; +use codex_app_server_protocol::ClientResponse; use codex_app_server_protocol::CodexErrorInfo as AppServerCodexErrorInfo; use codex_app_server_protocol::CollaborationModeListParams; use codex_app_server_protocol::CollaborationModeListResponse; @@ -185,13 +186,11 @@ use codex_core::CodexAuth; use codex_core::CodexThread; use codex_core::Cursor as RolloutCursor; use codex_core::ForkSnapshot; -use codex_core::InitializationMode; use codex_core::NewThread; use codex_core::RolloutRecorder; use codex_core::SessionMeta; use codex_core::SteerInputError; use codex_core::ThreadConfigSnapshot; -use codex_core::ThreadInitializedInput; use codex_core::ThreadManager; use codex_core::ThreadSortKey as CoreThreadSortKey; use codex_core::auth::AuthMode as CoreAuthMode; @@ -2170,12 +2169,13 @@ impl CodexMessageProcessor { }; listener_task_context .analytics_events_client - .track_thread_initialized(thread_initialize_input( - request_id.connection_id, - &thread, - response.model.clone(), - InitializationMode::New, - )); + .track_response( + request_id.connection_id.0, + ClientResponse::ThreadStart { + request_id: request_id.request_id.clone(), + response: response.clone(), + }, + ); listener_task_context .outgoing @@ -3654,13 +3654,13 @@ impl CodexMessageProcessor { sandbox: session_configured.sandbox_policy.into(), reasoning_effort: session_configured.reasoning_effort, }; - self.analytics_events_client - .track_thread_initialized(thread_initialize_input( - request_id.connection_id, - &response.thread, - response.model.clone(), - InitializationMode::Resumed, - )); + self.analytics_events_client.track_response( + request_id.connection_id.0, + ClientResponse::ThreadResume { + request_id: request_id.request_id.clone(), + response: response.clone(), + }, + ); self.outgoing.send_response(request_id, response).await; } @@ -4268,13 +4268,13 @@ impl CodexMessageProcessor { sandbox: session_configured.sandbox_policy.into(), reasoning_effort: session_configured.reasoning_effort, }; - self.analytics_events_client - .track_thread_initialized(thread_initialize_input( - request_id.connection_id, - &thread, - response.model.clone(), - InitializationMode::Forked, - )); + self.analytics_events_client.track_response( + request_id.connection_id.0, + ClientResponse::ThreadFork { + request_id: request_id.request_id.clone(), + response: response.clone(), + }, + ); self.outgoing.send_response(request_id, response).await; @@ -7894,22 +7894,6 @@ fn cloud_requirements_load_error(err: &std::io::Error) -> Option<&CloudRequireme None } -fn thread_initialize_input( - connection_id: ConnectionId, - thread: &Thread, - model: String, - initialization_mode: InitializationMode, -) -> ThreadInitializedInput { - ThreadInitializedInput { - connection_id: connection_id.0, - thread_id: thread.id.clone(), - model, - ephemeral: thread.ephemeral, - session_source: thread.source.clone().into(), - initialization_mode, - } -} - fn config_load_error(err: &std::io::Error) -> JSONRPCErrorError { let data = cloud_requirements_load_error(err).map(|cloud_error| { let mut data = serde_json::json!({ diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index 358ae58f69..d0e28a8d48 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -136,8 +136,6 @@ pub use auth::CodexAuth; pub use codex_analytics::AnalyticsEventsClient; pub use codex_analytics::AnalyticsFact; pub use codex_analytics::CustomAnalyticsFact; -pub use codex_analytics::InitializationMode; -pub use codex_analytics::ThreadInitializedInput; mod default_client_forwarding; /// Default Codex HTTP client headers and reqwest construction. From b8615d471f45bc02bcc00d629e4c0bef9bdc306c Mon Sep 17 00:00:00 2001 From: Roy Han Date: Fri, 27 Mar 2026 10:25:52 -0700 Subject: [PATCH 2/8] shift towards app-server stream consumption --- codex-rs/Cargo.lock | 1 + codex-rs/analytics/src/analytics_client.rs | 496 +++++++++++++----- .../analytics/src/analytics_client_tests.rs | 441 +++++++++++++--- codex-rs/analytics/src/lib.rs | 4 +- codex-rs/app-server/Cargo.toml | 1 + .../app-server/src/bespoke_event_handling.rs | 59 ++- .../app-server/src/codex_message_processor.rs | 20 +- .../app-server/tests/suite/v2/analytics.rs | 35 ++ .../app-server/tests/suite/v2/turn_start.rs | 68 +++ codex-rs/core/src/codex.rs | 45 +- 10 files changed, 910 insertions(+), 260 deletions(-) diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index ce8e81f539..b800a64f6e 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -1400,6 +1400,7 @@ dependencies = [ "base64 0.22.1", "chrono", "clap", + "codex-analytics", "codex-app-server-protocol", "codex-arg0", "codex-backend-client", diff --git a/codex-rs/analytics/src/analytics_client.rs b/codex-rs/analytics/src/analytics_client.rs index f3c3c88de7..1d8f690db8 100644 --- a/codex-rs/analytics/src/analytics_client.rs +++ b/codex-rs/analytics/src/analytics_client.rs @@ -30,6 +30,8 @@ use std::path::PathBuf; use std::sync::Arc; use std::sync::Mutex; use std::time::Duration; +use std::time::SystemTime; +use std::time::UNIX_EPOCH; use tokio::sync::mpsc; #[derive(Clone)] @@ -40,8 +42,10 @@ pub struct TrackEventsContext { } #[derive(Clone)] -pub struct CodexTurnEvent { +pub struct TurnResolvedConfigFact { + pub turn_id: String, pub submission_type: Option, + pub model: String, pub model_provider: String, pub sandbox_policy: SandboxPolicy, pub reasoning_effort: Option, @@ -52,27 +56,7 @@ pub struct CodexTurnEvent { pub sandbox_network_access: bool, pub collaboration_mode: ModeKind, pub personality: Option, - pub num_input_images: usize, pub is_first_turn: bool, - pub status: Option, - pub turn_error: Option, - pub steer_count: Option, - pub total_tool_call_count: Option, - pub shell_command_count: Option, - pub file_change_count: Option, - pub mcp_tool_call_count: Option, - pub dynamic_tool_call_count: Option, - pub subagent_tool_call_count: Option, - pub web_search_count: Option, - pub image_generation_count: Option, - pub input_tokens: Option, - pub cached_input_tokens: Option, - pub output_tokens: Option, - pub reasoning_output_tokens: Option, - pub total_tokens: Option, - pub duration_ms: Option, - pub started_at: Option, - pub completed_at: Option, } #[derive(Clone, Copy, Debug, Serialize)] @@ -155,18 +139,14 @@ pub enum AnalyticsFact { connection_id: u64, response: Box, }, - Notification { - connection_id: u64, - notification: Box, - }, + Notification(Box), // Facts that do not naturally exist on the app-server protocol surface, or // would require non-trivial protocol reshaping on this branch. Custom(CustomAnalyticsFact), } pub enum CustomAnalyticsFact { - TurnStarted(Box), - TurnCompleted(TurnCompletedInput), + TurnResolvedConfig(Box), SkillInvoked(SkillInvokedInput), AppMentioned(AppMentionedInput), AppUsed(AppUsedInput), @@ -174,15 +154,6 @@ pub enum CustomAnalyticsFact { PluginStateChanged(PluginStateChangedInput), } -pub struct TurnStartedInput { - pub tracking: TrackEventsContext, - pub turn_event: CodexTurnEvent, -} - -pub struct TurnCompletedInput { - pub turn_id: String, -} - pub struct SkillInvokedInput { pub tracking: TrackEventsContext, pub invocations: Vec, @@ -218,16 +189,39 @@ pub enum PluginState { #[derive(Default)] pub struct AnalyticsReducer { + requests: HashMap<(u64, RequestId), RequestState>, turns: HashMap, - clients: HashMap, + connections: HashMap, } -struct ClientState { +struct ConnectionState { product_client_id: String, } +enum RequestState { + TurnStart(PendingTurnStartState), +} + +struct PendingTurnStartState { + thread_id: String, + num_input_images: usize, +} + +#[derive(Clone)] +struct CompletedTurnState { + status: Option, + turn_error: Option, + completed_at_secs: u64, + duration_ms: Option, +} + struct TurnState { - started_input: TurnStartedInput, + connection_id: Option, + thread_id: Option, + num_input_images: Option, + resolved_config: Option, + started_at_ms: Option, + completed: Option, } #[derive(Clone)] @@ -333,6 +327,14 @@ impl AnalyticsEventsClient { }); } + pub fn track_request(&self, connection_id: u64, request_id: RequestId, request: ClientRequest) { + self.record_fact(AnalyticsFact::Request { + connection_id, + request_id, + request: Box::new(request), + }); + } + pub fn track_app_mentioned(&self, tracking: TrackEventsContext, mentions: Vec) { if mentions.is_empty() { return; @@ -360,19 +362,10 @@ impl AnalyticsEventsClient { ))); } - pub fn track_turn_started(&self, tracking: TrackEventsContext, turn_event: CodexTurnEvent) { - self.record_fact(AnalyticsFact::Custom(CustomAnalyticsFact::TurnStarted( - Box::new(TurnStartedInput { - tracking, - turn_event, - }), - ))); - } - - pub fn track_turn_completed(&self, turn_id: String) { - self.record_fact(AnalyticsFact::Custom(CustomAnalyticsFact::TurnCompleted( - TurnCompletedInput { turn_id }, - ))); + pub fn track_turn_resolved_config(&self, fact: TurnResolvedConfigFact) { + self.record_fact(AnalyticsFact::Custom( + CustomAnalyticsFact::TurnResolvedConfig(Box::new(fact)), + )); } pub fn track_plugin_installed(&self, plugin: PluginTelemetryMetadata) { @@ -424,6 +417,10 @@ impl AnalyticsEventsClient { response: Box::new(response), }); } + + pub fn track_notification(&self, notification: ServerNotification) { + self.record_fact(AnalyticsFact::Notification(Box::new(notification))); + } } const ANALYTICS_EVENTS_QUEUE_SIZE: usize = 256; @@ -595,29 +592,27 @@ impl AnalyticsReducer { connection_id, params, } => { - self.ingest_initialize(connection_id, params); + self.ingest_initialize(connection_id, params, out); } AnalyticsFact::Request { - connection_id: _connection_id, - request_id: _request_id, - request: _request, - } => {} + connection_id, + request_id, + request, + } => { + self.ingest_request(connection_id, request_id, *request); + } AnalyticsFact::Response { connection_id, response, } => { self.ingest_response(connection_id, *response, out); } - AnalyticsFact::Notification { - connection_id: _connection_id, - notification: _notification, - } => {} + AnalyticsFact::Notification(notification) => { + self.ingest_notification(*notification, out); + } AnalyticsFact::Custom(input) => match input { - CustomAnalyticsFact::TurnStarted(input) => { - self.ingest_turn_started(*input); - } - CustomAnalyticsFact::TurnCompleted(input) => { - self.ingest_turn_completed(input, out); + CustomAnalyticsFact::TurnResolvedConfig(input) => { + self.ingest_turn_resolved_config(*input, out); } CustomAnalyticsFact::SkillInvoked(input) => { self.ingest_skill_invoked(input, out).await; @@ -638,13 +633,31 @@ impl AnalyticsReducer { } } - fn ingest_initialize(&mut self, connection_id: u64, params: InitializeParams) { - self.clients.insert( + fn ingest_initialize( + &mut self, + connection_id: u64, + params: InitializeParams, + out: &mut Vec, + ) { + self.connections.insert( connection_id, - ClientState { + ConnectionState { product_client_id: params.client_info.name, }, ); + let ready_turn_ids = self + .turns + .iter() + .filter_map(|(turn_id, state)| { + (state.connection_id == Some(connection_id) + && state.resolved_config.is_some() + && state.completed.is_some()) + .then_some(turn_id.clone()) + }) + .collect::>(); + for turn_id in ready_turn_ids { + self.maybe_emit_turn_event(&turn_id, out); + } } fn ingest_thread_initialized( @@ -652,7 +665,7 @@ impl AnalyticsReducer { input: ThreadInitializedInput, out: &mut Vec, ) { - let Some(client_state) = self.clients.get(&input.connection_id) else { + let Some(client_state) = self.connections.get(&input.connection_id) else { return; }; out.push(TrackEventRequest::CodexThreadInitialized( @@ -660,33 +673,50 @@ impl AnalyticsReducer { )); } - fn ingest_turn_started(&mut self, input: TurnStartedInput) { - self.turns.insert( - input.tracking.turn_id.clone(), - TurnState { - started_input: input, - }, + fn ingest_request( + &mut self, + connection_id: u64, + request_id: RequestId, + request: ClientRequest, + ) { + let ClientRequest::TurnStart { params, .. } = request else { + return; + }; + self.requests.insert( + (connection_id, request_id), + RequestState::TurnStart(PendingTurnStartState { + thread_id: params.thread_id, + num_input_images: params + .input + .iter() + .filter(|item| { + matches!( + item, + codex_app_server_protocol::UserInput::Image { .. } + | codex_app_server_protocol::UserInput::LocalImage { .. } + ) + }) + .count(), + }), ); } - fn ingest_turn_completed( + fn ingest_turn_resolved_config( &mut self, - input: TurnCompletedInput, + input: TurnResolvedConfigFact, out: &mut Vec, ) { - let Some(TurnState { started_input }) = self.turns.remove(&input.turn_id) else { - return; - }; - let TurnStartedInput { - tracking, - turn_event, - } = started_input; - out.push(TrackEventRequest::TurnEvent(Box::new( - CodexTurnEventRequest { - event_type: "codex_turn_event", - event_params: codex_turn_event_params(&tracking, turn_event), - }, - ))); + 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_ms: None, + completed: None, + }); + turn_state.resolved_config = Some(input); + self.maybe_emit_turn_event(&turn_id, out); } fn ingest_response( @@ -720,11 +750,127 @@ impl AnalyticsReducer { session_source: response.thread.source.into(), initialization_mode: InitializationMode::Forked, }, + ClientResponse::TurnStart { + request_id, + response, + } => { + let turn_id = response.turn.id.clone(); + let Some(RequestState::TurnStart(pending_request)) = + self.requests.remove(&(connection_id, request_id)) + 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_ms: None, + completed: None, + }); + 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); + self.maybe_emit_turn_event(&turn_id, out); + return; + } _ => return, }; self.ingest_thread_initialized(input, out); } + fn ingest_notification( + &mut self, + notification: ServerNotification, + out: &mut Vec, + ) { + match notification { + ServerNotification::TurnStarted(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_ms: None, + completed: None, + }); + turn_state.started_at_ms = Some(now_unix_timestamp_millis()); + } + 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_ms: None, + completed: None, + }); + let completed_at_secs = now_unix_timestamp_secs(); + let completed_at_ms = completed_at_secs.saturating_mul(1000); + turn_state.completed = Some(CompletedTurnState { + status: analytics_turn_status(notification.turn.status), + turn_error: notification.turn.error.and_then(|error| { + error + .codex_error_info + .map(app_server_codex_error_info_to_core) + }), + completed_at_secs, + duration_ms: turn_state + .started_at_ms + .map(|started_at_ms| completed_at_ms.saturating_sub(started_at_ms)), + }); + let turn_id = notification.turn.id; + self.maybe_emit_turn_event(&turn_id, out); + } + _ => {} + } + } + + fn maybe_emit_turn_event(&mut self, turn_id: &str, out: &mut Vec) { + let Some(turn_state) = self.turns.get(turn_id) else { + return; + }; + let Some(connection_id) = turn_state.connection_id else { + return; + }; + let Some(connection_state) = self.connections.get(&connection_id) else { + return; + }; + let Some(thread_id) = turn_state.thread_id.clone() else { + return; + }; + let Some(num_input_images) = turn_state.num_input_images else { + return; + }; + let Some(resolved_config) = turn_state.resolved_config.clone() else { + return; + }; + let Some(completed) = turn_state.completed.clone() else { + return; + }; + out.push(TrackEventRequest::TurnEvent(Box::new( + CodexTurnEventRequest { + event_type: "codex_turn_event", + event_params: codex_turn_event_params( + connection_state.product_client_id.clone(), + thread_id, + turn_id.to_string(), + num_input_images, + resolved_config, + completed, + turn_state.started_at_ms.map(|value| value / 1000), + ), + }, + ))); + self.turns.remove(turn_id); + } + async fn ingest_skill_invoked( &mut self, input: SkillInvokedInput, @@ -841,49 +987,69 @@ fn codex_app_metadata(tracking: &TrackEventsContext, app: AppInvocation) -> Code } fn codex_turn_event_params( - tracking: &TrackEventsContext, - turn_event: CodexTurnEvent, + product_client_id: String, + thread_id: String, + turn_id: String, + num_input_images: usize, + resolved_config: TurnResolvedConfigFact, + completed: CompletedTurnState, + started_at: Option, ) -> CodexTurnEventParams { + let TurnResolvedConfigFact { + turn_id: _resolved_turn_id, + submission_type, + model, + model_provider, + sandbox_policy, + reasoning_effort, + reasoning_summary, + service_tier, + approval_policy, + approvals_reviewer, + sandbox_network_access, + collaboration_mode, + personality, + is_first_turn, + } = resolved_config; CodexTurnEventParams { - thread_id: tracking.thread_id.clone(), - turn_id: tracking.turn_id.clone(), - product_client_id: Some(originator().value), - submission_type: turn_event.submission_type, - model: Some(tracking.model_slug.clone()), - model_provider: turn_event.model_provider, - sandbox_policy: Some(sandbox_policy_mode(&turn_event.sandbox_policy)), - reasoning_effort: turn_event.reasoning_effort.map(|value| value.to_string()), - reasoning_summary: reasoning_summary_mode(turn_event.reasoning_summary), - service_tier: turn_event - .service_tier + thread_id, + turn_id, + product_client_id: Some(product_client_id), + submission_type, + model: Some(model), + model_provider, + sandbox_policy: Some(sandbox_policy_mode(&sandbox_policy)), + reasoning_effort: reasoning_effort.map(|value| value.to_string()), + reasoning_summary: reasoning_summary_mode(reasoning_summary), + service_tier: service_tier .map(|value| value.to_string()) .unwrap_or_else(|| "default".to_string()), - approval_policy: turn_event.approval_policy.to_string(), - approvals_reviewer: turn_event.approvals_reviewer.to_string(), - sandbox_network_access: turn_event.sandbox_network_access, - collaboration_mode: Some(collaboration_mode_mode(turn_event.collaboration_mode)), - personality: personality_mode(turn_event.personality), - num_input_images: turn_event.num_input_images, - is_first_turn: turn_event.is_first_turn, - status: turn_event.status, - turn_error: turn_event.turn_error, - steer_count: turn_event.steer_count, - total_tool_call_count: turn_event.total_tool_call_count, - shell_command_count: turn_event.shell_command_count, - file_change_count: turn_event.file_change_count, - mcp_tool_call_count: turn_event.mcp_tool_call_count, - dynamic_tool_call_count: turn_event.dynamic_tool_call_count, - subagent_tool_call_count: turn_event.subagent_tool_call_count, - web_search_count: turn_event.web_search_count, - image_generation_count: turn_event.image_generation_count, - input_tokens: turn_event.input_tokens, - cached_input_tokens: turn_event.cached_input_tokens, - output_tokens: turn_event.output_tokens, - reasoning_output_tokens: turn_event.reasoning_output_tokens, - total_tokens: turn_event.total_tokens, - duration_ms: turn_event.duration_ms, - started_at: turn_event.started_at, - completed_at: turn_event.completed_at, + approval_policy: approval_policy.to_string(), + approvals_reviewer: approvals_reviewer.to_string(), + sandbox_network_access, + collaboration_mode: Some(collaboration_mode_mode(collaboration_mode)), + personality: personality_mode(personality), + num_input_images, + is_first_turn, + status: completed.status, + turn_error: completed.turn_error, + steer_count: None, + total_tool_call_count: None, + shell_command_count: None, + file_change_count: None, + mcp_tool_call_count: None, + dynamic_tool_call_count: None, + subagent_tool_call_count: None, + web_search_count: None, + image_generation_count: None, + input_tokens: None, + cached_input_tokens: None, + output_tokens: None, + reasoning_output_tokens: None, + total_tokens: None, + duration_ms: completed.duration_ms, + started_at, + completed_at: Some(completed.completed_at_secs), } } @@ -990,12 +1156,78 @@ fn session_source_name(session_source: &SessionSource) -> Option<&'static str> { } fn now_unix_timestamp_secs() -> u64 { - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) + SystemTime::now() + .duration_since(UNIX_EPOCH) .unwrap_or_default() .as_secs() } +fn now_unix_timestamp_millis() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis() as u64 +} + +fn analytics_turn_status(status: codex_app_server_protocol::TurnStatus) -> Option { + match status { + codex_app_server_protocol::TurnStatus::Completed => Some(TurnStatus::Completed), + codex_app_server_protocol::TurnStatus::Failed => Some(TurnStatus::Failed), + codex_app_server_protocol::TurnStatus::Interrupted => Some(TurnStatus::Interrupted), + codex_app_server_protocol::TurnStatus::InProgress => None, + } +} + +fn app_server_codex_error_info_to_core( + info: codex_app_server_protocol::CodexErrorInfo, +) -> CodexErrorInfo { + match info { + codex_app_server_protocol::CodexErrorInfo::ContextWindowExceeded => { + CodexErrorInfo::ContextWindowExceeded + } + codex_app_server_protocol::CodexErrorInfo::UsageLimitExceeded => { + CodexErrorInfo::UsageLimitExceeded + } + codex_app_server_protocol::CodexErrorInfo::ServerOverloaded => { + CodexErrorInfo::ServerOverloaded + } + codex_app_server_protocol::CodexErrorInfo::HttpConnectionFailed { http_status_code } => { + CodexErrorInfo::HttpConnectionFailed { http_status_code } + } + codex_app_server_protocol::CodexErrorInfo::ResponseStreamConnectionFailed { + http_status_code, + } => CodexErrorInfo::ResponseStreamConnectionFailed { http_status_code }, + codex_app_server_protocol::CodexErrorInfo::InternalServerError => { + CodexErrorInfo::InternalServerError + } + codex_app_server_protocol::CodexErrorInfo::Unauthorized => CodexErrorInfo::Unauthorized, + codex_app_server_protocol::CodexErrorInfo::BadRequest => CodexErrorInfo::BadRequest, + codex_app_server_protocol::CodexErrorInfo::ThreadRollbackFailed => { + CodexErrorInfo::ThreadRollbackFailed + } + codex_app_server_protocol::CodexErrorInfo::SandboxError => CodexErrorInfo::SandboxError, + codex_app_server_protocol::CodexErrorInfo::ResponseStreamDisconnected { + http_status_code, + } => CodexErrorInfo::ResponseStreamDisconnected { http_status_code }, + codex_app_server_protocol::CodexErrorInfo::ResponseTooManyFailedAttempts { + http_status_code, + } => CodexErrorInfo::ResponseTooManyFailedAttempts { http_status_code }, + codex_app_server_protocol::CodexErrorInfo::ActiveTurnNotSteerable { turn_kind } => { + CodexErrorInfo::ActiveTurnNotSteerable { + turn_kind: match turn_kind { + codex_app_server_protocol::NonSteerableTurnKind::Review => { + codex_protocol::protocol::NonSteerableTurnKind::Review + } + codex_app_server_protocol::NonSteerableTurnKind::Compact => { + codex_protocol::protocol::NonSteerableTurnKind::Compact + } + }, + } + } + codex_app_server_protocol::CodexErrorInfo::Other => CodexErrorInfo::Other, + } +} + async fn send_track_events( auth_manager: &AuthManager, base_url: &str, diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index 136b739672..3f07af0c02 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -6,16 +6,16 @@ use super::CodexAppMentionedEventRequest; use super::CodexAppUsedEventRequest; use super::CodexPluginEventRequest; use super::CodexPluginUsedEventRequest; -use super::CodexTurnEvent; use super::CodexTurnEventRequest; +use super::CompletedTurnState; use super::CustomAnalyticsFact; use super::InitializationMode; use super::InvocationType; use super::ThreadInitializedInput; use super::TrackEventRequest; use super::TrackEventsContext; -use super::TurnCompletedInput; -use super::TurnStartedInput; +use super::TurnResolvedConfigFact; +use super::TurnStatus; use super::codex_app_metadata; use super::codex_plugin_metadata; use super::codex_plugin_used_metadata; @@ -25,15 +25,24 @@ use super::normalize_path_for_skill_id; use codex_app_server_protocol::ApprovalsReviewer as AppServerApprovalsReviewer; use codex_app_server_protocol::AskForApproval as AppServerAskForApproval; use codex_app_server_protocol::ClientInfo; +use codex_app_server_protocol::ClientRequest; use codex_app_server_protocol::ClientResponse; use codex_app_server_protocol::InitializeParams; use codex_app_server_protocol::RequestId; use codex_app_server_protocol::SandboxPolicy as AppServerSandboxPolicy; +use codex_app_server_protocol::ServerNotification; use codex_app_server_protocol::SessionSource as AppServerSessionSource; use codex_app_server_protocol::Thread; use codex_app_server_protocol::ThreadResumeResponse; use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::ThreadStatus as AppServerThreadStatus; +use codex_app_server_protocol::Turn; +use codex_app_server_protocol::TurnCompletedNotification; +use codex_app_server_protocol::TurnError as AppServerTurnError; +use codex_app_server_protocol::TurnStartParams; +use codex_app_server_protocol::TurnStartedNotification; +use codex_app_server_protocol::TurnStatus as AppServerTurnStatus; +use codex_app_server_protocol::UserInput as AppServerUserInput; use codex_login::default_client::originator; use codex_plugin::AppConnectorId; use codex_plugin::PluginCapabilitySummary; @@ -111,6 +120,160 @@ fn sample_thread_resume_response(thread_id: &str, ephemeral: bool, model: &str) } } +fn sample_turn_start_request(thread_id: &str, request_id: i64) -> ClientRequest { + ClientRequest::TurnStart { + request_id: RequestId::Integer(request_id), + params: TurnStartParams { + thread_id: thread_id.to_string(), + input: vec![ + AppServerUserInput::Text { + text: "hello".to_string(), + text_elements: vec![], + }, + AppServerUserInput::Image { + url: "https://example.com/a.png".to_string(), + }, + ], + ..Default::default() + }, + } +} + +fn sample_turn_start_response(turn_id: &str, request_id: i64) -> ClientResponse { + ClientResponse::TurnStart { + request_id: RequestId::Integer(request_id), + response: codex_app_server_protocol::TurnStartResponse { + turn: Turn { + id: turn_id.to_string(), + items: vec![], + status: AppServerTurnStatus::InProgress, + error: None, + }, + }, + } +} + +fn sample_turn_started_notification(thread_id: &str, turn_id: &str) -> ServerNotification { + ServerNotification::TurnStarted(TurnStartedNotification { + thread_id: thread_id.to_string(), + turn: Turn { + id: turn_id.to_string(), + items: vec![], + status: AppServerTurnStatus::InProgress, + error: None, + }, + }) +} + +fn sample_turn_completed_notification( + thread_id: &str, + turn_id: &str, + status: AppServerTurnStatus, + codex_error_info: Option, +) -> ServerNotification { + ServerNotification::TurnCompleted(TurnCompletedNotification { + thread_id: thread_id.to_string(), + turn: Turn { + id: turn_id.to_string(), + items: vec![], + status, + error: codex_error_info.map(|codex_error_info| AppServerTurnError { + message: "turn failed".to_string(), + codex_error_info: Some(codex_error_info), + additional_details: None, + }), + }, + }) +} + +fn sample_turn_resolved_config(turn_id: &str) -> TurnResolvedConfigFact { + TurnResolvedConfigFact { + turn_id: turn_id.to_string(), + submission_type: None, + model: "gpt-5".to_string(), + model_provider: "openai".to_string(), + sandbox_policy: SandboxPolicy::new_read_only_policy(), + reasoning_effort: None, + reasoning_summary: None, + service_tier: None, + approval_policy: AskForApproval::OnRequest, + approvals_reviewer: ApprovalsReviewer::GuardianSubagent, + sandbox_network_access: true, + collaboration_mode: ModeKind::Plan, + personality: None, + is_first_turn: true, + } +} + +async fn ingest_turn_prerequisites( + reducer: &mut AnalyticsReducer, + out: &mut Vec, + include_initialize: bool, + include_resolved_config: bool, + include_started: bool, +) { + if include_initialize { + reducer + .ingest( + AnalyticsFact::Initialize { + connection_id: 7, + params: InitializeParams { + client_info: ClientInfo { + name: "codex-tui".to_string(), + title: None, + version: "1.0.0".to_string(), + }, + capabilities: None, + }, + }, + out, + ) + .await; + } + + reducer + .ingest( + AnalyticsFact::Request { + connection_id: 7, + request_id: RequestId::Integer(3), + request: Box::new(sample_turn_start_request("thread-2", 3)), + }, + out, + ) + .await; + reducer + .ingest( + AnalyticsFact::Response { + connection_id: 7, + response: Box::new(sample_turn_start_response("turn-2", 3)), + }, + out, + ) + .await; + + if include_resolved_config { + reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::TurnResolvedConfig(Box::new( + sample_turn_resolved_config("turn-2"), + ))), + out, + ) + .await; + } + + if include_started { + reducer + .ingest( + AnalyticsFact::Notification(Box::new(sample_turn_started_notification( + "thread-2", "turn-2", + ))), + out, + ) + .await; + } +} + fn expected_absolute_path(path: &PathBuf) -> String { std::fs::canonicalize(path) .unwrap_or_else(|_| path.to_path_buf()) @@ -275,17 +438,17 @@ fn app_used_dedupe_is_keyed_by_turn_and_connector() { #[test] fn turn_event_serializes_expected_shape() { - let tracking = TrackEventsContext { - model_slug: "gpt-5".to_string(), - thread_id: "thread-2".to_string(), - turn_id: "turn-2".to_string(), - }; let event = TrackEventRequest::TurnEvent(Box::new(CodexTurnEventRequest { event_type: "codex_turn_event", event_params: codex_turn_event_params( - &tracking, - CodexTurnEvent { + "codex-tui".to_string(), + "thread-2".to_string(), + "turn-2".to_string(), + 2, + TurnResolvedConfigFact { + turn_id: "turn-2".to_string(), submission_type: None, + model: "gpt-5".to_string(), model_provider: "openai".to_string(), sandbox_policy: SandboxPolicy::new_read_only_policy(), reasoning_effort: Some(ReasoningEffort::High), @@ -296,28 +459,15 @@ fn turn_event_serializes_expected_shape() { sandbox_network_access: true, collaboration_mode: ModeKind::Plan, personality: Some(Personality::Pragmatic), - num_input_images: 2, is_first_turn: true, - status: None, - turn_error: None, - steer_count: None, - total_tool_call_count: None, - shell_command_count: None, - file_change_count: None, - mcp_tool_call_count: None, - dynamic_tool_call_count: None, - subagent_tool_call_count: None, - web_search_count: None, - image_generation_count: None, - input_tokens: None, - cached_input_tokens: None, - output_tokens: None, - reasoning_output_tokens: None, - total_tokens: None, - duration_ms: None, - started_at: None, - completed_at: None, }, + CompletedTurnState { + status: Some(TurnStatus::Completed), + turn_error: None, + completed_at_secs: 456, + duration_ms: Some(1234), + }, + Some(455), ), })); @@ -330,7 +480,7 @@ fn turn_event_serializes_expected_shape() { "event_params": { "thread_id": "thread-2", "turn_id": "turn-2", - "product_client_id": originator().value, + "product_client_id": "codex-tui", "submission_type": null, "model": "gpt-5", "model_provider": "openai", @@ -345,7 +495,7 @@ fn turn_event_serializes_expected_shape() { "personality": "pragmatic", "num_input_images": 2, "is_first_turn": true, - "status": null, + "status": "completed", "turn_error": null, "steer_count": null, "total_tool_call_count": null, @@ -361,65 +511,32 @@ fn turn_event_serializes_expected_shape() { "output_tokens": null, "reasoning_output_tokens": null, "total_tokens": null, - "duration_ms": null, - "started_at": null, - "completed_at": null + "duration_ms": 1234, + "started_at": 455, + "completed_at": 456 } }) ); } #[tokio::test] -async fn turn_started_then_completed_emits_turn_event() { - let tracking = TrackEventsContext { - model_slug: "gpt-5".to_string(), - thread_id: "thread-2".to_string(), - turn_id: "turn-2".to_string(), - }; +async fn turn_lifecycle_emits_turn_event() { let mut reducer = AnalyticsReducer::default(); let mut out = Vec::new(); reducer .ingest( - AnalyticsFact::Custom(CustomAnalyticsFact::TurnStarted(Box::new( - TurnStartedInput { - tracking: tracking.clone(), - turn_event: CodexTurnEvent { - submission_type: None, - model_provider: "openai".to_string(), - sandbox_policy: SandboxPolicy::new_read_only_policy(), - reasoning_effort: Some(ReasoningEffort::High), - reasoning_summary: Some(ReasoningSummary::Detailed), - service_tier: Some(ServiceTier::Flex), - approval_policy: AskForApproval::OnRequest, - approvals_reviewer: ApprovalsReviewer::GuardianSubagent, - sandbox_network_access: true, - collaboration_mode: ModeKind::Plan, - personality: Some(Personality::Pragmatic), - num_input_images: 2, - is_first_turn: true, - status: None, - turn_error: None, - steer_count: None, - total_tool_call_count: None, - shell_command_count: None, - file_change_count: None, - mcp_tool_call_count: None, - dynamic_tool_call_count: None, - subagent_tool_call_count: None, - web_search_count: None, - image_generation_count: None, - input_tokens: None, - cached_input_tokens: None, - output_tokens: None, - reasoning_output_tokens: None, - total_tokens: None, - duration_ms: None, - started_at: None, - completed_at: None, + AnalyticsFact::Initialize { + connection_id: 7, + params: InitializeParams { + client_info: ClientInfo { + name: "codex-tui".to_string(), + title: None, + version: "1.0.0".to_string(), }, + capabilities: None, }, - ))), + }, &mut out, ) .await; @@ -428,9 +545,62 @@ async fn turn_started_then_completed_emits_turn_event() { reducer .ingest( - AnalyticsFact::Custom(CustomAnalyticsFact::TurnCompleted(TurnCompletedInput { - turn_id: tracking.turn_id.clone(), - })), + AnalyticsFact::Request { + connection_id: 7, + request_id: RequestId::Integer(3), + request: Box::new(sample_turn_start_request("thread-2", 3)), + }, + &mut out, + ) + .await; + reducer + .ingest( + AnalyticsFact::Response { + connection_id: 7, + response: Box::new(sample_turn_start_response("turn-2", 3)), + }, + &mut out, + ) + .await; + reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::TurnResolvedConfig(Box::new( + TurnResolvedConfigFact { + turn_id: "turn-2".to_string(), + submission_type: None, + model: "gpt-5".to_string(), + model_provider: "openai".to_string(), + sandbox_policy: SandboxPolicy::new_read_only_policy(), + reasoning_effort: Some(ReasoningEffort::High), + reasoning_summary: Some(ReasoningSummary::Detailed), + service_tier: Some(ServiceTier::Flex), + approval_policy: AskForApproval::OnRequest, + approvals_reviewer: ApprovalsReviewer::GuardianSubagent, + sandbox_network_access: true, + collaboration_mode: ModeKind::Plan, + personality: Some(Personality::Pragmatic), + is_first_turn: true, + }, + ))), + &mut out, + ) + .await; + reducer + .ingest( + AnalyticsFact::Notification(Box::new(sample_turn_started_notification( + "thread-2", "turn-2", + ))), + &mut out, + ) + .await; + reducer + .ingest( + AnalyticsFact::Notification(Box::new(sample_turn_completed_notification( + "thread-2", + "turn-2", + AppServerTurnStatus::Completed, + None, + ))), &mut out, ) .await; @@ -440,6 +610,115 @@ async fn turn_started_then_completed_emits_turn_event() { assert_eq!(payload["event_type"], json!("codex_turn_event")); assert_eq!(payload["event_params"]["thread_id"], json!("thread-2")); assert_eq!(payload["event_params"]["turn_id"], json!("turn-2")); + assert_eq!( + payload["event_params"]["product_client_id"], + json!("codex-tui") + ); + assert_eq!(payload["event_params"]["num_input_images"], json!(1)); + assert_eq!(payload["event_params"]["status"], json!("completed")); + assert!(payload["event_params"]["started_at"].as_u64().is_some()); + assert!(payload["event_params"]["completed_at"].as_u64().is_some()); +} + +#[tokio::test] +async fn turn_does_not_emit_without_required_prerequisites() { + let mut reducer = AnalyticsReducer::default(); + let mut out = Vec::new(); + + ingest_turn_prerequisites(&mut reducer, &mut out, false, true, false).await; + reducer + .ingest( + AnalyticsFact::Notification(Box::new(sample_turn_completed_notification( + "thread-2", + "turn-2", + AppServerTurnStatus::Completed, + None, + ))), + &mut out, + ) + .await; + assert!(out.is_empty()); + + let mut reducer = AnalyticsReducer::default(); + let mut out = Vec::new(); + + ingest_turn_prerequisites(&mut reducer, &mut out, true, false, false).await; + reducer + .ingest( + AnalyticsFact::Notification(Box::new(sample_turn_completed_notification( + "thread-2", + "turn-2", + AppServerTurnStatus::Completed, + None, + ))), + &mut out, + ) + .await; + assert!(out.is_empty()); +} + +#[tokio::test] +async fn turn_completed_without_started_notification_emits_null_started_at() { + let mut reducer = AnalyticsReducer::default(); + let mut out = Vec::new(); + + ingest_turn_prerequisites(&mut reducer, &mut out, true, true, false).await; + reducer + .ingest( + AnalyticsFact::Notification(Box::new(sample_turn_completed_notification( + "thread-2", + "turn-2", + AppServerTurnStatus::Completed, + None, + ))), + &mut out, + ) + .await; + + let payload = serde_json::to_value(&out[0]).expect("serialize turn event"); + assert_eq!(payload["event_params"]["started_at"], json!(null)); + assert_eq!(payload["event_params"]["duration_ms"], json!(null)); +} + +#[tokio::test] +async fn turn_completed_maps_completion_variants() { + for (status, codex_error_info, expected_status, expected_turn_error) in [ + ( + AppServerTurnStatus::Failed, + Some(codex_app_server_protocol::CodexErrorInfo::BadRequest), + json!("failed"), + Some(json!("bad_request")), + ), + ( + AppServerTurnStatus::Interrupted, + None, + json!("interrupted"), + None, + ), + ] { + let mut reducer = AnalyticsReducer::default(); + let mut out = Vec::new(); + + ingest_turn_prerequisites(&mut reducer, &mut out, true, true, false).await; + reducer + .ingest( + AnalyticsFact::Notification(Box::new(sample_turn_completed_notification( + "thread-2", + "turn-2", + status, + codex_error_info, + ))), + &mut out, + ) + .await; + + let payload = serde_json::to_value(&out[0]).expect("serialize turn event"); + assert_eq!(payload["event_params"]["status"], expected_status); + assert_eq!( + payload["event_params"]["turn_error"], + expected_turn_error.unwrap_or(json!(null)) + ); + } } #[test] diff --git a/codex-rs/analytics/src/lib.rs b/codex-rs/analytics/src/lib.rs index 1c1f5a66b5..29e6df8413 100644 --- a/codex-rs/analytics/src/lib.rs +++ b/codex-rs/analytics/src/lib.rs @@ -6,7 +6,6 @@ pub use analytics_client::AnalyticsReducer; pub use analytics_client::AppInvocation; pub use analytics_client::AppMentionedInput; pub use analytics_client::AppUsedInput; -pub use analytics_client::CodexTurnEvent; pub use analytics_client::CustomAnalyticsFact; pub use analytics_client::InitializationMode; pub use analytics_client::InvocationType; @@ -16,8 +15,7 @@ pub use analytics_client::PluginUsedInput; pub use analytics_client::SkillInvocation; pub use analytics_client::SkillInvokedInput; pub use analytics_client::TrackEventsContext; -pub use analytics_client::TurnCompletedInput; -pub use analytics_client::TurnStartedInput; +pub use analytics_client::TurnResolvedConfigFact; pub use analytics_client::TurnStatus; pub use analytics_client::TurnSubmissionType; pub use analytics_client::build_track_events_context; diff --git a/codex-rs/app-server/Cargo.toml b/codex-rs/app-server/Cargo.toml index ee12e87acb..413cffa5f7 100644 --- a/codex-rs/app-server/Cargo.toml +++ b/codex-rs/app-server/Cargo.toml @@ -30,6 +30,7 @@ axum = { workspace = true, default-features = false, features = [ "ws", ] } codex-arg0 = { workspace = true } +codex-analytics = { workspace = true } codex-cloud-requirements = { workspace = true } codex-core = { workspace = true } codex-exec-server = { workspace = true } diff --git a/codex-rs/app-server/src/bespoke_event_handling.rs b/codex-rs/app-server/src/bespoke_event_handling.rs index 9a850c0b1e..a2b3210f12 100644 --- a/codex-rs/app-server/src/bespoke_event_handling.rs +++ b/codex-rs/app-server/src/bespoke_event_handling.rs @@ -12,6 +12,7 @@ use crate::thread_state::ThreadState; use crate::thread_state::TurnSummary; use crate::thread_status::ThreadWatchActiveGuard; use crate::thread_status::ThreadWatchManager; +use codex_analytics::AnalyticsEventsClient; use codex_app_server_protocol::AccountRateLimitsUpdatedNotification; use codex_app_server_protocol::AdditionalPermissionProfile as V2AdditionalPermissionProfile; use codex_app_server_protocol::AgentMessageDeltaNotification; @@ -257,6 +258,7 @@ pub(crate) async fn apply_bespoke_event_handling( conversation_id: ThreadId, conversation: Arc, thread_manager: Arc, + analytics_events_client: AnalyticsEventsClient, outgoing: ThreadScopedOutgoingMessageSender, thread_state: Arc>, thread_watch_manager: ThreadWatchManager, @@ -289,6 +291,8 @@ pub(crate) async fn apply_bespoke_event_handling( thread_id: conversation_id.to_string(), turn, }; + analytics_events_client + .track_notification(ServerNotification::TurnStarted(notification.clone())); outgoing .send_server_notification(ServerNotification::TurnStarted(notification)) .await; @@ -301,7 +305,14 @@ pub(crate) async fn apply_bespoke_event_handling( thread_watch_manager .note_turn_completed(&conversation_id.to_string(), turn_failed) .await; - handle_turn_complete(conversation_id, event_turn_id, &outgoing, &thread_state).await; + handle_turn_complete( + conversation_id, + event_turn_id, + Some(&analytics_events_client), + &outgoing, + &thread_state, + ) + .await; } EventMsg::SkillsUpdateAvailable => { if let ApiVersion::V2 = api_version { @@ -1946,6 +1957,7 @@ async fn emit_turn_completed_with_status( event_turn_id: String, status: TurnStatus, error: Option, + analytics_events_client: Option<&AnalyticsEventsClient>, outgoing: &ThreadScopedOutgoingMessageSender, ) { let notification = TurnCompletedNotification { @@ -1957,6 +1969,10 @@ async fn emit_turn_completed_with_status( status, }, }; + if let Some(analytics_events_client) = analytics_events_client { + analytics_events_client + .track_notification(ServerNotification::TurnCompleted(notification.clone())); + } outgoing .send_server_notification(ServerNotification::TurnCompleted(notification)) .await; @@ -2100,6 +2116,7 @@ async fn find_and_remove_turn_summary( async fn handle_turn_complete( conversation_id: ThreadId, event_turn_id: String, + analytics_events_client: Option<&AnalyticsEventsClient>, outgoing: &ThreadScopedOutgoingMessageSender, thread_state: &Arc>, ) { @@ -2110,7 +2127,15 @@ async fn handle_turn_complete( None => (TurnStatus::Completed, None), }; - emit_turn_completed_with_status(conversation_id, event_turn_id, status, error, outgoing).await; + emit_turn_completed_with_status( + conversation_id, + event_turn_id, + status, + error, + analytics_events_client, + outgoing, + ) + .await; } async fn handle_turn_interrupted( @@ -2126,6 +2151,7 @@ async fn handle_turn_interrupted( event_turn_id, TurnStatus::Interrupted, /*error*/ None, + None, outgoing, ) .await; @@ -3280,6 +3306,7 @@ mod tests { handle_turn_complete( conversation_id, event_turn_id.clone(), + None, &outgoing, &thread_state, ) @@ -3368,6 +3395,7 @@ mod tests { handle_turn_complete( conversation_id, event_turn_id.clone(), + None, &outgoing, &thread_state, ) @@ -3628,7 +3656,14 @@ mod tests { &thread_state, ) .await; - handle_turn_complete(conversation_a, a_turn1.clone(), &outgoing, &thread_state).await; + handle_turn_complete( + conversation_a, + a_turn1.clone(), + None, + &outgoing, + &thread_state, + ) + .await; // Turn 1 on conversation B let b_turn1 = "b_turn1".to_string(); @@ -3642,11 +3677,25 @@ mod tests { &thread_state, ) .await; - handle_turn_complete(conversation_b, b_turn1.clone(), &outgoing, &thread_state).await; + handle_turn_complete( + conversation_b, + b_turn1.clone(), + None, + &outgoing, + &thread_state, + ) + .await; // Turn 2 on conversation A let a_turn2 = "a_turn2".to_string(); - handle_turn_complete(conversation_a, a_turn2.clone(), &outgoing, &thread_state).await; + handle_turn_complete( + conversation_a, + a_turn2.clone(), + None, + &outgoing, + &thread_state, + ) + .await; // Verify: A turn 1 let msg = recv_broadcast_message(&mut rx).await?; diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index 31af6070cf..f68b30cb30 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -6179,6 +6179,16 @@ impl CodexMessageProcessor { return; } + let analytics_turn_start_params = params.clone(); + self.analytics_events_client.track_request( + request_id.connection_id.0, + request_id.request_id.clone(), + ClientRequest::TurnStart { + request_id: request_id.request_id.clone(), + params: analytics_turn_start_params, + }, + ); + let collaboration_modes_config = CollaborationModesConfig { default_mode_request_user_input: thread.enabled(Feature::DefaultModeRequestUserInput), }; @@ -6254,6 +6264,13 @@ impl CodexMessageProcessor { }; let response = TurnStartResponse { turn }; + self.analytics_events_client.track_response( + request_id.connection_id.0, + ClientResponse::TurnStart { + request_id: request_id.request_id.clone(), + response: response.clone(), + }, + ); self.outgoing.send_response(request_id, response).await; } Err(err) => { @@ -6976,7 +6993,7 @@ impl CodexMessageProcessor { outgoing, thread_manager, thread_state_manager, - analytics_events_client: _, + analytics_events_client, thread_watch_manager, fallback_model_provider, codex_home, @@ -7023,6 +7040,7 @@ impl CodexMessageProcessor { conversation_id, conversation.clone(), thread_manager.clone(), + analytics_events_client.clone(), thread_outgoing, thread_state.clone(), thread_watch_manager.clone(), diff --git a/codex-rs/app-server/tests/suite/v2/analytics.rs b/codex-rs/app-server/tests/suite/v2/analytics.rs index 1c8a500aeb..c95dc6f416 100644 --- a/codex-rs/app-server/tests/suite/v2/analytics.rs +++ b/codex-rs/app-server/tests/suite/v2/analytics.rs @@ -120,6 +120,41 @@ pub(crate) async fn wait_for_analytics_payload( 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, + event_type: &str, +) -> Result { + timeout(read_timeout, async { + loop { + let Some(requests) = server.received_requests().await else { + tokio::time::sleep(Duration::from_millis(25)).await; + continue; + }; + for request in &requests { + if request.method != "POST" + || request.url.path() != "/codex/analytics-events/events" + { + continue; + } + let payload: Value = serde_json::from_slice(&request.body) + .map_err(|err| anyhow::anyhow!("invalid analytics payload: {err}"))?; + let Some(events) = payload["events"].as_array() else { + continue; + }; + if let Some(event) = events + .iter() + .find(|event| event["event_type"] == event_type) + { + return Ok::(event.clone()); + } + } + tokio::time::sleep(Duration::from_millis(25)).await; + } + }) + .await? +} + pub(crate) fn thread_initialized_event(payload: &Value) -> Result<&Value> { let events = payload["events"] .as_array() diff --git a/codex-rs/app-server/tests/suite/v2/turn_start.rs b/codex-rs/app-server/tests/suite/v2/turn_start.rs index 8d7ca02613..0e16df9d85 100644 --- a/codex-rs/app-server/tests/suite/v2/turn_start.rs +++ b/codex-rs/app-server/tests/suite/v2/turn_start.rs @@ -64,6 +64,9 @@ use std::path::Path; use tempfile::TempDir; use tokio::time::timeout; +use super::analytics::enable_analytics_capture; +use super::analytics::wait_for_analytics_event; + #[cfg(windows)] const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(25); #[cfg(not(windows))] @@ -238,6 +241,71 @@ async fn turn_start_emits_user_message_item_with_text_elements() -> Result<()> { Ok(()) } +#[tokio::test] +async fn turn_start_tracks_turn_event_analytics() -> Result<()> { + let responses = vec![create_final_assistant_message_sse_response("Done")?]; + let server = create_mock_responses_server_sequence_unchecked(responses).await; + + let codex_home = TempDir::new()?; + create_config_toml_with_chatgpt_base_url(codex_home.path(), &server.uri(), &server.uri())?; + enable_analytics_capture(&server, codex_home.path()).await?; + + let mut mcp = McpProcess::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let thread_req = mcp + .send_thread_start_request(ThreadStartParams { + model: Some("mock-model".to_string()), + ..Default::default() + }) + .await?; + let thread_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(thread_req)), + ) + .await??; + let ThreadStartResponse { thread, .. } = to_response::(thread_resp)?; + + let turn_req = mcp + .send_turn_start_request(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![V2UserInput::Image { + url: "https://example.com/a.png".to_string(), + }], + ..Default::default() + }) + .await?; + let turn_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(turn_req)), + ) + .await??; + let TurnStartResponse { turn } = to_response::(turn_resp)?; + + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("turn/completed"), + ) + .await??; + + let event = wait_for_analytics_event(&server, DEFAULT_READ_TIMEOUT, "codex_turn_event").await?; + assert_eq!(event["event_params"]["thread_id"], thread.id); + assert_eq!(event["event_params"]["turn_id"], turn.id); + assert_eq!( + event["event_params"]["product_client_id"], + DEFAULT_CLIENT_NAME + ); + assert_eq!(event["event_params"]["model"], "mock-model"); + assert_eq!(event["event_params"]["model_provider"], "mock_provider"); + assert_eq!(event["event_params"]["sandbox_policy"], "read_only"); + assert_eq!(event["event_params"]["num_input_images"], 1); + assert_eq!(event["event_params"]["status"], "completed"); + assert!(event["event_params"]["started_at"].as_u64().is_some()); + assert!(event["event_params"]["completed_at"].as_u64().is_some()); + + Ok(()) +} + #[tokio::test] async fn turn_start_accepts_text_at_limit_with_mention_item() -> Result<()> { let responses = vec![create_final_assistant_message_sse_response("Done")?]; diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 99b2d51d0f..8447449216 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -349,8 +349,8 @@ use crate::util::backoff; use crate::windows_sandbox::WindowsSandboxLevelExt; use codex_analytics::AnalyticsEventsClient; use codex_analytics::AppInvocation; -use codex_analytics::CodexTurnEvent; use codex_analytics::InvocationType; +use codex_analytics::TurnResolvedConfigFact; use codex_analytics::build_track_events_context; use codex_async_utils::OrCancelExt; use codex_git_utils::get_git_repo_root; @@ -5807,10 +5807,12 @@ pub(crate) async fn run_turn( let mut state = sess.state.lock().await; state.take_next_turn_is_first() }; - sess.services.analytics_events_client.track_turn_started( - tracking.clone(), - CodexTurnEvent { + sess.services + .analytics_events_client + .track_turn_resolved_config(TurnResolvedConfigFact { + turn_id: tracking.turn_id.clone(), submission_type: None, + model: turn_context.model_info.slug.clone(), model_provider: turn_context.config.model_provider_id.clone(), sandbox_policy: turn_context.sandbox_policy.get().clone(), reasoning_effort: turn_context.reasoning_effort, @@ -5821,34 +5823,8 @@ pub(crate) async fn run_turn( sandbox_network_access: turn_context.network_sandbox_policy.is_enabled(), collaboration_mode: turn_context.collaboration_mode.mode, personality: turn_context.personality, - num_input_images: input - .iter() - .filter(|item| { - matches!(item, UserInput::Image { .. } | UserInput::LocalImage { .. }) - }) - .count(), is_first_turn, - status: None, - turn_error: None, - steer_count: None, - total_tool_call_count: None, - shell_command_count: None, - file_change_count: None, - mcp_tool_call_count: None, - dynamic_tool_call_count: None, - subagent_tool_call_count: None, - web_search_count: None, - image_generation_count: None, - input_tokens: None, - cached_input_tokens: None, - output_tokens: None, - reasoning_output_tokens: None, - total_tokens: None, - duration_ms: None, - started_at: None, - completed_at: None, - }, - ); + }); } let skills_outcome = Some(turn_context.turn_skills.outcome.as_ref()); @@ -6132,13 +6108,6 @@ pub(crate) async fn run_turn( } } } - - if !input.is_empty() { - sess.services - .analytics_events_client - .track_turn_completed(tracking.turn_id.clone()); - } - last_agent_message } From 4381ebb4f4fb2f5ddde5bc7f1bc1aa637fd5abba Mon Sep 17 00:00:00 2001 From: Roy Han Date: Fri, 27 Mar 2026 10:35:00 -0700 Subject: [PATCH 3/8] lint --- codex-rs/analytics/src/analytics_client.rs | 21 +++++++++------------ 1 file changed, 9 insertions(+), 12 deletions(-) diff --git a/codex-rs/analytics/src/analytics_client.rs b/codex-rs/analytics/src/analytics_client.rs index 1d8f690db8..b78e238b97 100644 --- a/codex-rs/analytics/src/analytics_client.rs +++ b/codex-rs/analytics/src/analytics_client.rs @@ -754,7 +754,7 @@ impl AnalyticsReducer { request_id, response, } => { - let turn_id = response.turn.id.clone(); + let turn_id = response.turn.id; let Some(RequestState::TurnStart(pending_request)) = self.requests.remove(&(connection_id, request_id)) else { @@ -786,17 +786,14 @@ impl AnalyticsReducer { ) { match notification { ServerNotification::TurnStarted(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_ms: None, - completed: None, - }); + 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_ms: None, + completed: None, + }); turn_state.started_at_ms = Some(now_unix_timestamp_millis()); } ServerNotification::TurnCompleted(notification) => { From 2e725aa97c617aabc92d0dfdd00e681237d7b9db Mon Sep 17 00:00:00 2001 From: Roy Han Date: Fri, 27 Mar 2026 10:38:08 -0700 Subject: [PATCH 4/8] connection cleanup --- codex-rs/analytics/src/analytics_client.rs | 10 ++-------- 1 file changed, 2 insertions(+), 8 deletions(-) diff --git a/codex-rs/analytics/src/analytics_client.rs b/codex-rs/analytics/src/analytics_client.rs index eadcfe876f..d9edc200d0 100644 --- a/codex-rs/analytics/src/analytics_client.rs +++ b/codex-rs/analytics/src/analytics_client.rs @@ -95,10 +95,7 @@ pub enum AnalyticsFact { connection_id: u64, response: Box, }, - Notification { - connection_id: u64, - notification: Box, - }, + Notification(Box), // Facts that do not naturally exist on the app-server protocol surface, or // would require non-trivial protocol reshaping on this branch. Custom(CustomAnalyticsFact), @@ -470,10 +467,7 @@ impl AnalyticsReducer { } => { self.ingest_response(connection_id, *response, out); } - AnalyticsFact::Notification { - connection_id: _connection_id, - notification: _notification, - } => {} + AnalyticsFact::Notification(_notification) => {} AnalyticsFact::Custom(input) => match input { CustomAnalyticsFact::SkillInvoked(input) => { self.ingest_skill_invoked(input, out).await; From a1e506f17d9f8d6147fb78b728591f656a129c30 Mon Sep 17 00:00:00 2001 From: Roy Han Date: Fri, 27 Mar 2026 10:43:05 -0700 Subject: [PATCH 5/8] rename client to connection --- codex-rs/analytics/src/analytics_client.rs | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/codex-rs/analytics/src/analytics_client.rs b/codex-rs/analytics/src/analytics_client.rs index d9edc200d0..32b91205a8 100644 --- a/codex-rs/analytics/src/analytics_client.rs +++ b/codex-rs/analytics/src/analytics_client.rs @@ -144,10 +144,10 @@ pub enum PluginState { #[derive(Default)] pub struct AnalyticsReducer { - clients: HashMap, + connections: HashMap, } -struct ClientState { +struct ConnectionState { product_client_id: String, } @@ -489,9 +489,9 @@ impl AnalyticsReducer { } fn ingest_initialize(&mut self, connection_id: u64, params: InitializeParams) { - self.clients.insert( + self.connections.insert( connection_id, - ClientState { + ConnectionState { product_client_id: params.client_info.name, }, ); @@ -502,11 +502,14 @@ impl AnalyticsReducer { input: ThreadInitializedInput, out: &mut Vec, ) { - let Some(client_state) = self.clients.get(&input.connection_id) else { + let Some(connection_state) = self.connections.get(&input.connection_id) else { return; }; out.push(TrackEventRequest::CodexThreadInitialized( - codex_thread_initialized_event_request(client_state.product_client_id.clone(), input), + codex_thread_initialized_event_request( + connection_state.product_client_id.clone(), + input, + ), )); } From 7afe91c1df05add6f86518c6f6443df167579639 Mon Sep 17 00:00:00 2001 From: Roy Han Date: Fri, 27 Mar 2026 11:00:11 -0700 Subject: [PATCH 6/8] add client metadata --- codex-rs/analytics/src/analytics_client.rs | 29 +++++++++++++------ .../analytics/src/analytics_client_tests.rs | 22 ++++++++++++-- 2 files changed, 40 insertions(+), 11 deletions(-) diff --git a/codex-rs/analytics/src/analytics_client.rs b/codex-rs/analytics/src/analytics_client.rs index 32b91205a8..b856118899 100644 --- a/codex-rs/analytics/src/analytics_client.rs +++ b/codex-rs/analytics/src/analytics_client.rs @@ -149,6 +149,9 @@ pub struct AnalyticsReducer { struct ConnectionState { product_client_id: String, + client_name: Option, + client_version: Option, + experimental_api_enabled: Option, } #[derive(Clone)] @@ -377,6 +380,9 @@ struct SkillInvocationEventParams { struct CodexThreadInitializedEventParams { thread_id: String, product_client_id: String, + client_name: Option, + client_version: Option, + experimental_api_enabled: Option, model: String, ephemeral: bool, session_source: Option<&'static str>, @@ -492,7 +498,12 @@ impl AnalyticsReducer { self.connections.insert( connection_id, ConnectionState { - product_client_id: params.client_info.name, + product_client_id: params.client_info.name.clone(), + client_name: Some(params.client_info.name), + client_version: Some(params.client_info.version), + experimental_api_enabled: params + .capabilities + .map(|capabilities| capabilities.experimental_api), }, ); } @@ -506,10 +517,7 @@ impl AnalyticsReducer { return; }; out.push(TrackEventRequest::CodexThreadInitialized( - codex_thread_initialized_event_request( - connection_state.product_client_id.clone(), - input, - ), + codex_thread_initialized_event_request(connection_state, input), )); } @@ -666,22 +674,25 @@ fn codex_app_metadata(tracking: &TrackEventsContext, app: AppInvocation) -> Code } fn codex_thread_initialized_event_request( - product_client_id: String, + connection_state: &ConnectionState, input: ThreadInitializedInput, ) -> CodexThreadInitializedEvent { CodexThreadInitializedEvent { event_type: "codex_thread_initialized", - event_params: codex_thread_initialized_event_params(product_client_id, input), + event_params: codex_thread_initialized_event_params(connection_state, input), } } fn codex_thread_initialized_event_params( - product_client_id: String, + connection_state: &ConnectionState, input: ThreadInitializedInput, ) -> CodexThreadInitializedEventParams { CodexThreadInitializedEventParams { thread_id: input.thread_id, - product_client_id, + product_client_id: connection_state.product_client_id.clone(), + client_name: connection_state.client_name.clone(), + client_version: connection_state.client_version.clone(), + experimental_api_enabled: connection_state.experimental_api_enabled, model: input.model, ephemeral: input.ephemeral, session_source: session_source_name(&input.session_source), diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index 6156015bef..431453ca51 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -20,6 +20,7 @@ use codex_app_server_protocol::ApprovalsReviewer as AppServerApprovalsReviewer; use codex_app_server_protocol::AskForApproval as AppServerAskForApproval; use codex_app_server_protocol::ClientInfo; use codex_app_server_protocol::ClientResponse; +use codex_app_server_protocol::InitializeCapabilities; use codex_app_server_protocol::InitializeParams; use codex_app_server_protocol::RequestId; use codex_app_server_protocol::SandboxPolicy as AppServerSandboxPolicy; @@ -262,7 +263,12 @@ fn app_used_dedupe_is_keyed_by_turn_and_connector() { #[test] fn thread_initialized_event_serializes_expected_shape() { let event = TrackEventRequest::CodexThreadInitialized(codex_thread_initialized_event_request( - "codex-tui".to_string(), + &super::ConnectionState { + product_client_id: "codex-tui".to_string(), + client_name: Some("codex-tui".to_string()), + client_version: Some("1.0.0".to_string()), + experimental_api_enabled: Some(true), + }, ThreadInitializedInput { connection_id: 1, thread_id: "thread-0".to_string(), @@ -282,6 +288,9 @@ fn thread_initialized_event_serializes_expected_shape() { "event_params": { "thread_id": "thread-0", "product_client_id": "codex-tui", + "client_name": "codex-tui", + "client_version": "1.0.0", + "experimental_api_enabled": true, "model": "gpt-5", "ephemeral": true, "session_source": "user", @@ -325,7 +334,10 @@ async fn initialize_caches_client_and_thread_lifecycle_publishes_once_initialize title: None, version: "1.0.0".to_string(), }, - capabilities: None, + capabilities: Some(InitializeCapabilities { + experimental_api: false, + opt_out_notification_methods: None, + }), }, }, &mut events, @@ -347,6 +359,12 @@ async fn initialize_caches_client_and_thread_lifecycle_publishes_once_initialize 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"]["product_client_id"], "codex-tui"); + assert_eq!(payload[0]["event_params"]["client_name"], "codex-tui"); + assert_eq!(payload[0]["event_params"]["client_version"], "1.0.0"); + assert_eq!( + payload[0]["event_params"]["experimental_api_enabled"], + false + ); assert_eq!(payload[0]["event_params"]["initialization_mode"], "resumed"); assert_eq!(payload[0]["event_params"]["session_source"], "user"); assert_eq!(payload[0]["event_params"]["subagent_source"], json!(null)); From ed327248a25fd2175f7b48a63d3f46c758c221bb Mon Sep 17 00:00:00 2001 From: Roy Han Date: Fri, 27 Mar 2026 11:15:51 -0700 Subject: [PATCH 7/8] various fixes --- codex-rs/analytics/src/analytics_client.rs | 84 ++++++++----------- .../analytics/src/analytics_client_tests.rs | 8 +- 2 files changed, 41 insertions(+), 51 deletions(-) diff --git a/codex-rs/analytics/src/analytics_client.rs b/codex-rs/analytics/src/analytics_client.rs index b856118899..275073c92f 100644 --- a/codex-rs/analytics/src/analytics_client.rs +++ b/codex-rs/analytics/src/analytics_client.rs @@ -38,6 +38,7 @@ struct ThreadInitializedInput { pub ephemeral: bool, pub session_source: SessionSource, pub initialization_mode: InitializationMode, + pub created_at: u64, } #[derive(Clone, Copy, Debug, Serialize)] @@ -348,7 +349,7 @@ struct TrackEventsRequest { #[serde(untagged)] enum TrackEventRequest { SkillInvocation(SkillInvocationEventRequest), - CodexThreadInitialized(CodexThreadInitializedEvent), + ThreadInitialized(ThreadInitializedEvent), AppMentioned(CodexAppMentionedEventRequest), AppUsed(CodexAppUsedEventRequest), PluginUsed(CodexPluginUsedEventRequest), @@ -377,7 +378,7 @@ struct SkillInvocationEventParams { } #[derive(Serialize)] -struct CodexThreadInitializedEventParams { +struct ThreadInitializedEventParams { thread_id: String, product_client_id: String, client_name: Option, @@ -393,9 +394,9 @@ struct CodexThreadInitializedEventParams { } #[derive(Serialize)] -struct CodexThreadInitializedEvent { +struct ThreadInitializedEvent { event_type: &'static str, - event_params: CodexThreadInitializedEventParams, + event_params: ThreadInitializedEventParams, } #[derive(Serialize)] @@ -516,8 +517,8 @@ impl AnalyticsReducer { let Some(connection_state) = self.connections.get(&input.connection_id) else { return; }; - out.push(TrackEventRequest::CodexThreadInitialized( - codex_thread_initialized_event_request(connection_state, input), + out.push(TrackEventRequest::ThreadInitialized( + thread_initialized_event_request(connection_state, input), )); } @@ -527,34 +528,30 @@ impl AnalyticsReducer { response: ClientResponse, out: &mut Vec, ) { - let input = match response { - ClientResponse::ThreadStart { response, .. } => ThreadInitializedInput { - connection_id, - thread_id: response.thread.id, - model: response.model, - ephemeral: response.thread.ephemeral, - session_source: response.thread.source.into(), - initialization_mode: InitializationMode::New, - }, - ClientResponse::ThreadResume { response, .. } => ThreadInitializedInput { - connection_id, - thread_id: response.thread.id, - model: response.model, - ephemeral: response.thread.ephemeral, - session_source: response.thread.source.into(), - initialization_mode: InitializationMode::Resumed, - }, - ClientResponse::ThreadFork { response, .. } => ThreadInitializedInput { - connection_id, - thread_id: response.thread.id, - model: response.model, - ephemeral: response.thread.ephemeral, - session_source: response.thread.source.into(), - initialization_mode: InitializationMode::Forked, - }, + let (thread, model, initialization_mode) = match response { + ClientResponse::ThreadStart { response, .. } => { + (response.thread, response.model, InitializationMode::New) + } + ClientResponse::ThreadResume { response, .. } => { + (response.thread, response.model, InitializationMode::Resumed) + } + ClientResponse::ThreadFork { response, .. } => { + (response.thread, response.model, InitializationMode::Forked) + } _ => return, }; - self.ingest_thread_initialized(input, out); + self.ingest_thread_initialized( + ThreadInitializedInput { + connection_id, + thread_id: thread.id, + model, + ephemeral: thread.ephemeral, + session_source: thread.source.into(), + initialization_mode, + created_at: u64::try_from(thread.created_at).unwrap_or_default(), + }, + out, + ); } async fn ingest_skill_invoked( @@ -673,21 +670,21 @@ fn codex_app_metadata(tracking: &TrackEventsContext, app: AppInvocation) -> Code } } -fn codex_thread_initialized_event_request( +fn thread_initialized_event_request( connection_state: &ConnectionState, input: ThreadInitializedInput, -) -> CodexThreadInitializedEvent { - CodexThreadInitializedEvent { +) -> ThreadInitializedEvent { + ThreadInitializedEvent { event_type: "codex_thread_initialized", - event_params: codex_thread_initialized_event_params(connection_state, input), + event_params: thread_initialized_event_params(connection_state, input), } } -fn codex_thread_initialized_event_params( +fn thread_initialized_event_params( connection_state: &ConnectionState, input: ThreadInitializedInput, -) -> CodexThreadInitializedEventParams { - CodexThreadInitializedEventParams { +) -> ThreadInitializedEventParams { + ThreadInitializedEventParams { thread_id: input.thread_id, product_client_id: connection_state.product_client_id.clone(), client_name: connection_state.client_name.clone(), @@ -699,7 +696,7 @@ fn codex_thread_initialized_event_params( initialization_mode: input.initialization_mode, subagent_source: None, parent_thread_id: None, - created_at: now_unix_timestamp_secs(), + created_at: input.created_at, } } @@ -748,13 +745,6 @@ fn session_source_name(session_source: &SessionSource) -> Option<&'static str> { } } -fn now_unix_timestamp_secs() -> u64 { - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_secs() -} - async fn send_track_events( auth_manager: &AuthManager, base_url: &str, diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index 431453ca51..6689ba5e79 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -14,8 +14,8 @@ use super::TrackEventsContext; use super::codex_app_metadata; use super::codex_plugin_metadata; use super::codex_plugin_used_metadata; -use super::codex_thread_initialized_event_request; use super::normalize_path_for_skill_id; +use super::thread_initialized_event_request; use codex_app_server_protocol::ApprovalsReviewer as AppServerApprovalsReviewer; use codex_app_server_protocol::AskForApproval as AppServerAskForApproval; use codex_app_server_protocol::ClientInfo; @@ -262,7 +262,7 @@ fn app_used_dedupe_is_keyed_by_turn_and_connector() { #[test] fn thread_initialized_event_serializes_expected_shape() { - let event = TrackEventRequest::CodexThreadInitialized(codex_thread_initialized_event_request( + let event = TrackEventRequest::ThreadInitialized(thread_initialized_event_request( &super::ConnectionState { product_client_id: "codex-tui".to_string(), client_name: Some("codex-tui".to_string()), @@ -276,6 +276,7 @@ fn thread_initialized_event_serializes_expected_shape() { ephemeral: true, session_source: SessionSource::Exec, initialization_mode: InitializationMode::New, + created_at: 1, }, )); @@ -297,11 +298,10 @@ fn thread_initialized_event_serializes_expected_shape() { "initialization_mode": "new", "subagent_source": null, "parent_thread_id": null, - "created_at": payload["event_params"]["created_at"] + "created_at": 1 } }) ); - assert!(payload["event_params"]["created_at"].as_u64().is_some()); } #[tokio::test] From 1ce0530baf65f00f0192368208b85b06d1e2f3cf Mon Sep 17 00:00:00 2001 From: Roy Han Date: Fri, 27 Mar 2026 14:26:15 -0700 Subject: [PATCH 8/8] fixes --- codex-rs/app-server/tests/common/config.rs | 28 +++++++++++++++++++ codex-rs/app-server/tests/common/lib.rs | 1 + .../app-server/tests/suite/v2/turn_start.rs | 8 +++++- 3 files changed, 36 insertions(+), 1 deletion(-) diff --git a/codex-rs/app-server/tests/common/config.rs b/codex-rs/app-server/tests/common/config.rs index deb16c6322..1ac2572fa2 100644 --- a/codex-rs/app-server/tests/common/config.rs +++ b/codex-rs/app-server/tests/common/config.rs @@ -78,3 +78,31 @@ model_provider = "{model_provider_id}" ), ) } + +pub fn write_mock_responses_config_toml_with_chatgpt_base_url( + codex_home: &Path, + server_uri: &str, + chatgpt_base_url: &str, +) -> std::io::Result<()> { + let config_toml = codex_home.join("config.toml"); + std::fs::write( + config_toml, + format!( + r#" +model = "mock-model" +approval_policy = "never" +sandbox_mode = "read-only" +chatgpt_base_url = "{chatgpt_base_url}" + +model_provider = "mock_provider" + +[model_providers.mock_provider] +name = "Mock provider for test" +base_url = "{server_uri}/v1" +wire_api = "responses" +request_max_retries = 0 +stream_max_retries = 0 +"# + ), + ) +} diff --git a/codex-rs/app-server/tests/common/lib.rs b/codex-rs/app-server/tests/common/lib.rs index 3f89765851..90553760d9 100644 --- a/codex-rs/app-server/tests/common/lib.rs +++ b/codex-rs/app-server/tests/common/lib.rs @@ -14,6 +14,7 @@ pub use auth_fixtures::encode_id_token; pub use auth_fixtures::write_chatgpt_auth; use codex_app_server_protocol::JSONRPCResponse; pub use config::write_mock_responses_config_toml; +pub use config::write_mock_responses_config_toml_with_chatgpt_base_url; pub use core_test_support::format_with_current_shell; pub use core_test_support::format_with_current_shell_display; pub use core_test_support::format_with_current_shell_display_non_login; diff --git a/codex-rs/app-server/tests/suite/v2/turn_start.rs b/codex-rs/app-server/tests/suite/v2/turn_start.rs index 0e16df9d85..6381ca8b92 100644 --- a/codex-rs/app-server/tests/suite/v2/turn_start.rs +++ b/codex-rs/app-server/tests/suite/v2/turn_start.rs @@ -1,4 +1,5 @@ use anyhow::Result; +use app_test_support::DEFAULT_CLIENT_NAME; use app_test_support::McpProcess; use app_test_support::create_apply_patch_sse_response; use app_test_support::create_exec_command_sse_response; @@ -9,6 +10,7 @@ use app_test_support::create_mock_responses_server_sequence_unchecked; use app_test_support::create_shell_command_sse_response; use app_test_support::format_with_current_shell_display; use app_test_support::to_response; +use app_test_support::write_mock_responses_config_toml_with_chatgpt_base_url; use codex_app_server::INPUT_TOO_LARGE_ERROR_CODE; use codex_app_server::INVALID_PARAMS_ERROR_CODE; use codex_app_server_protocol::ByteRange; @@ -247,7 +249,11 @@ async fn turn_start_tracks_turn_event_analytics() -> Result<()> { let server = create_mock_responses_server_sequence_unchecked(responses).await; let codex_home = TempDir::new()?; - create_config_toml_with_chatgpt_base_url(codex_home.path(), &server.uri(), &server.uri())?; + write_mock_responses_config_toml_with_chatgpt_base_url( + codex_home.path(), + &server.uri(), + &server.uri(), + )?; enable_analytics_capture(&server, codex_home.path()).await?; let mut mcp = McpProcess::new(codex_home.path()).await?;