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 b259135271..0c59ae17be 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; @@ -29,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)] @@ -39,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, @@ -51,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)] @@ -90,13 +75,14 @@ pub enum TurnStatus { } #[derive(Clone)] -pub struct ThreadInitializedInput { +struct ThreadInitializedInput { pub connection_id: u64, pub thread_id: String, pub model: String, pub ephemeral: bool, pub session_source: SessionSource, pub initialization_mode: InitializationMode, + pub created_at: u64, } #[derive(Clone, Copy, Debug, Serialize)] @@ -148,23 +134,20 @@ pub enum AnalyticsFact { Request { connection_id: u64, request_id: RequestId, - request: ClientRequest, + request: Box, }, - Notification { + Response { connection_id: u64, - notification: ServerNotification, + response: 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 { - // 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), - TurnStarted(Box), - TurnCompleted(TurnCompletedInput), + TurnResolvedConfig(Box), SkillInvoked(SkillInvokedInput), AppMentioned(AppMentionedInput), AppUsed(AppUsedInput), @@ -172,15 +155,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, @@ -216,16 +190,42 @@ 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, + client_name: Option, + client_version: Option, + experimental_api_enabled: Option, +} + +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)] @@ -331,10 +331,12 @@ impl AnalyticsEventsClient { }); } - pub fn track_thread_initialized(&self, input: ThreadInitializedInput) { - self.record_fact(AnalyticsFact::Custom( - CustomAnalyticsFact::ThreadInitialized(input), - )); + 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) { @@ -364,19 +366,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) { @@ -421,6 +414,17 @@ 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), + }); + } + + pub fn track_notification(&self, notification: ServerNotification) { + self.record_fact(AnalyticsFact::Notification(Box::new(notification))); + } } const ANALYTICS_EVENTS_QUEUE_SIZE: usize = 256; @@ -436,7 +440,7 @@ struct TrackEventsRequest { #[serde(untagged)] enum TrackEventRequest { SkillInvocation(SkillInvocationEventRequest), - CodexThreadInitialized(CodexThreadInitializedEvent), + ThreadInitialized(ThreadInitializedEvent), AppMentioned(CodexAppMentionedEventRequest), AppUsed(CodexAppUsedEventRequest), TurnEvent(Box), @@ -466,9 +470,12 @@ struct SkillInvocationEventParams { } #[derive(Serialize)] -struct CodexThreadInitializedEventParams { +struct ThreadInitializedEventParams { 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>, @@ -479,9 +486,9 @@ struct CodexThreadInitializedEventParams { } #[derive(Serialize)] -struct CodexThreadInitializedEvent { +struct ThreadInitializedEvent { event_type: &'static str, - event_params: CodexThreadInitializedEventParams, + event_params: ThreadInitializedEventParams, } #[derive(Serialize)] @@ -592,26 +599,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, - } => {} - AnalyticsFact::Notification { - connection_id: _connection_id, - notification: _notification, - } => {} + 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(notification) => { + self.ingest_notification(*notification, out); + } AnalyticsFact::Custom(input) => match input { - CustomAnalyticsFact::ThreadInitialized(input) => { - self.ingest_thread_initialized(input, out); - } - 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; @@ -632,13 +640,36 @@ 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 { - product_client_id: params.client_info.name, + ConnectionState { + 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), }, ); + 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( @@ -646,41 +677,226 @@ 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), + out.push(TrackEventRequest::ThreadInitialized( + thread_initialized_event_request(connection_state, input), )); } - 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 { + 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( + &mut self, + connection_id: u64, + response: ClientResponse, + out: &mut Vec, + ) { + match response { + ClientResponse::ThreadStart { response, .. } => { + let thread = response.thread; + self.ingest_thread_initialized( + ThreadInitializedInput { + connection_id, + thread_id: thread.id, + model: response.model, + ephemeral: thread.ephemeral, + session_source: thread.source.into(), + initialization_mode: InitializationMode::New, + created_at: u64::try_from(thread.created_at).unwrap_or_default(), + }, + out, + ); + } + ClientResponse::ThreadResume { response, .. } => { + let thread = response.thread; + self.ingest_thread_initialized( + ThreadInitializedInput { + connection_id, + thread_id: thread.id, + model: response.model, + ephemeral: thread.ephemeral, + session_source: thread.source.into(), + initialization_mode: InitializationMode::Resumed, + created_at: u64::try_from(thread.created_at).unwrap_or_default(), + }, + out, + ); + } + ClientResponse::ThreadFork { response, .. } => { + let thread = response.thread; + self.ingest_thread_initialized( + ThreadInitializedInput { + connection_id, + thread_id: thread.id, + model: response.model, + ephemeral: thread.ephemeral, + session_source: thread.source.into(), + initialization_mode: InitializationMode::Forked, + created_at: u64::try_from(thread.created_at).unwrap_or_default(), + }, + out, + ); + } + ClientResponse::TurnStart { + request_id, + response, + } => { + let turn_id = response.turn.id; + 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); + } + _ => {} + } + } + + fn ingest_notification( + &mut self, + notification: ServerNotification, + out: &mut Vec, + ) { + match notification { + ServerNotification::TurnStarted(notification) => { + let turn_state = self.turns.entry(notification.turn.id).or_insert(TurnState { + connection_id: None, + thread_id: None, + num_input_images: None, + resolved_config: None, + started_at_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; }; - 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), + 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( @@ -799,49 +1015,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), } } @@ -875,30 +1111,33 @@ fn personality_mode(personality: Option) -> Option { } } -fn codex_thread_initialized_event_request( - product_client_id: String, +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(product_client_id, input), + event_params: thread_initialized_event_params(connection_state, input), } } -fn codex_thread_initialized_event_params( - product_client_id: String, +fn thread_initialized_event_params( + connection_state: &ConnectionState, input: ThreadInitializedInput, -) -> CodexThreadInitializedEventParams { - CodexThreadInitializedEventParams { +) -> ThreadInitializedEventParams { + ThreadInitializedEventParams { 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), initialization_mode: input.initialization_mode, subagent_source: None, parent_thread_id: None, - created_at: now_unix_timestamp_secs(), + created_at: input.created_at, } } @@ -948,12 +1187,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 cd40044bc9..e5d04ed614 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -6,25 +6,44 @@ 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::TurnSubmissionType; 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::codex_turn_event_params; 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; +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; @@ -47,6 +66,215 @@ 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 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()) @@ -211,17 +439,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: Some(TurnSubmissionType::Default), + model: "gpt-5".to_string(), model_provider: "openai".to_string(), sandbox_policy: SandboxPolicy::new_read_only_policy(), reasoning_effort: Some(ReasoningEffort::High), @@ -232,28 +460,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), ), })); @@ -266,7 +481,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", "model": "gpt-5", "submission_type": "default", "model_provider": "openai", @@ -281,7 +496,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, @@ -297,65 +512,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; @@ -364,9 +546,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; @@ -376,12 +611,126 @@ 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] fn thread_initialized_event_serializes_expected_shape() { - let event = TrackEventRequest::CodexThreadInitialized(codex_thread_initialized_event_request( - "codex-tui".to_string(), + let event = TrackEventRequest::ThreadInitialized(thread_initialized_event_request( + &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(), @@ -389,6 +738,7 @@ fn thread_initialized_event_serializes_expected_shape() { ephemeral: true, session_source: SessionSource::Exec, initialization_mode: InitializationMode::New, + created_at: 1, }, )); @@ -401,17 +751,19 @@ 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", "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] @@ -421,16 +773,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; @@ -456,16 +806,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 35b2a86810..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; @@ -15,10 +14,8 @@ 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::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 cbbd9e919a..89c2ac22de 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; @@ -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), }; @@ -6255,6 +6265,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) => { @@ -6977,7 +6994,7 @@ impl CodexMessageProcessor { outgoing, thread_manager, thread_state_manager, - analytics_events_client: _, + analytics_events_client, thread_watch_manager, fallback_model_provider, codex_home, @@ -7024,6 +7041,7 @@ impl CodexMessageProcessor { conversation_id, conversation.clone(), thread_manager.clone(), + analytics_events_client.clone(), thread_outgoing, thread_state.clone(), thread_watch_manager.clone(), @@ -7895,22 +7913,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/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/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 94ce775d2c..e063400c95 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; @@ -64,6 +66,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))] @@ -240,6 +245,75 @@ 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()?; + 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?; + 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 cd64f65f72..397e72dd11 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -350,8 +350,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::TurnSubmissionType; use codex_analytics::build_track_events_context; use codex_async_utils::OrCancelExt; @@ -5843,10 +5843,12 @@ pub(crate) async fn run_turn( .submission_type .map(turn_submission_type) .unwrap_or(TurnSubmissionType::Default); - 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: Some(submission_type), + 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, @@ -5857,34 +5859,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()); @@ -6168,13 +6144,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 } 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.