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 }