use crate::events::AppServerRpcTransport; use crate::events::CodexAppMentionedEventRequest; use crate::events::CodexAppServerClientMetadata; use crate::events::CodexAppUsedEventRequest; use crate::events::CodexCollabAgentToolCallEventParams; use crate::events::CodexCollabAgentToolCallEventRequest; use crate::events::CodexCommandExecutionEventParams; use crate::events::CodexCommandExecutionEventRequest; use crate::events::CodexCompactionEventRequest; use crate::events::CodexDynamicToolCallEventParams; use crate::events::CodexDynamicToolCallEventRequest; use crate::events::CodexFileChangeEventParams; use crate::events::CodexFileChangeEventRequest; use crate::events::CodexHookRunEventRequest; use crate::events::CodexImageGenerationEventParams; use crate::events::CodexImageGenerationEventRequest; use crate::events::CodexMcpToolCallEventParams; use crate::events::CodexMcpToolCallEventRequest; use crate::events::CodexPluginEventRequest; use crate::events::CodexPluginUsedEventRequest; use crate::events::CodexRuntimeMetadata; use crate::events::CodexToolCallReviewEventParams; use crate::events::CodexToolCallReviewEventRequest; use crate::events::CodexToolItemEventBase; use crate::events::CodexTurnEventParams; use crate::events::CodexTurnEventRequest; use crate::events::CodexTurnSteerEventParams; use crate::events::CodexTurnSteerEventRequest; use crate::events::CodexWebSearchEventParams; use crate::events::CodexWebSearchEventRequest; use crate::events::CollabAgentToolKind; use crate::events::CommandExecutionFamily; use crate::events::CommandExecutionSourceKind; use crate::events::GuardianReviewEventParams; use crate::events::GuardianReviewEventPayload; use crate::events::GuardianReviewEventRequest; use crate::events::SkillInvocationEventParams; use crate::events::SkillInvocationEventRequest; use crate::events::ThreadInitializedEvent; use crate::events::ThreadInitializedEventParams; use crate::events::ToolItemFailureKind; use crate::events::ToolItemFinalApprovalOutcome; use crate::events::ToolItemTerminalStatus; use crate::events::ToolReviewReviewer; use crate::events::ToolReviewStatus; use crate::events::ToolReviewToolKind; use crate::events::ToolReviewTrigger; use crate::events::TrackEventRequest; use crate::events::WebSearchActionKind; use crate::events::codex_app_metadata; use crate::events::codex_compaction_event_params; use crate::events::codex_hook_run_metadata; use crate::events::codex_plugin_metadata; use crate::events::codex_plugin_used_metadata; use crate::events::plugin_state_event_type; use crate::events::subagent_parent_thread_id; use crate::events::subagent_source_name; use crate::events::subagent_thread_started_event_request; use crate::facts::AnalyticsFact; use crate::facts::AnalyticsJsonRpcError; use crate::facts::AppMentionedInput; use crate::facts::AppUsedInput; use crate::facts::CodexCompactionEvent; use crate::facts::CustomAnalyticsFact; use crate::facts::HookRunInput; use crate::facts::PluginState; use crate::facts::PluginStateChangedInput; use crate::facts::PluginUsedInput; use crate::facts::SkillInvokedInput; use crate::facts::SubAgentThreadStartedInput; use crate::facts::ThreadInitializationMode; use crate::facts::TurnResolvedConfigFact; use crate::facts::TurnStatus; use crate::facts::TurnSteerRejectionReason; use crate::facts::TurnSteerResult; use crate::facts::TurnTokenUsageFact; use crate::now_unix_seconds; use codex_app_server_protocol::ClientRequest; use codex_app_server_protocol::ClientResponse; use codex_app_server_protocol::CodexErrorInfo; use codex_app_server_protocol::CollabAgentStatus; use codex_app_server_protocol::CollabAgentTool; use codex_app_server_protocol::CollabAgentToolCallStatus; use codex_app_server_protocol::CommandExecutionApprovalDecision; use codex_app_server_protocol::CommandExecutionSource; use codex_app_server_protocol::CommandExecutionStatus; use codex_app_server_protocol::DynamicToolCallOutputContentItem; use codex_app_server_protocol::DynamicToolCallStatus; use codex_app_server_protocol::FileChangeApprovalDecision; use codex_app_server_protocol::GuardianApprovalReviewAction; use codex_app_server_protocol::GuardianApprovalReviewStatus; use codex_app_server_protocol::GuardianCommandSource as AppServerGuardianCommandSource; use codex_app_server_protocol::InitializeParams; use codex_app_server_protocol::McpToolCallStatus; use codex_app_server_protocol::NetworkPolicyRuleAction; use codex_app_server_protocol::PatchApplyStatus; use codex_app_server_protocol::PatchChangeKind; use codex_app_server_protocol::PermissionGrantScope; use codex_app_server_protocol::RequestId; use codex_app_server_protocol::ServerNotification; use codex_app_server_protocol::ServerRequest; use codex_app_server_protocol::ServerResponse; use codex_app_server_protocol::ThreadItem; use codex_app_server_protocol::TurnSteerResponse; use codex_app_server_protocol::UserInput; use codex_app_server_protocol::WebSearchAction; use codex_git_utils::collect_git_info; use codex_git_utils::get_git_repo_root; use codex_login::default_client::originator; use codex_protocol::config_types::ModeKind; use codex_protocol::config_types::Personality; use codex_protocol::config_types::ReasoningSummary; use codex_protocol::protocol::SandboxPolicy; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SkillScope; use codex_protocol::protocol::TokenUsage; use sha1::Digest; use std::collections::HashMap; use std::path::Path; use std::time::SystemTime; use std::time::UNIX_EPOCH; #[derive(Default)] pub(crate) struct AnalyticsReducer { requests: HashMap<(u64, RequestId), RequestState>, turns: HashMap, connections: HashMap, thread_connections: HashMap, thread_metadata: HashMap, tool_items: HashMap, tool_review_requests: HashMap, guardian_review_starts: HashMap, tool_review_summaries: HashMap, } struct ConnectionState { app_server_client: CodexAppServerClientMetadata, runtime: CodexRuntimeMetadata, } struct ToolItemState { connection_id: u64, started_at: u64, } #[derive(Clone)] struct PendingToolReviewState { thread_id: String, turn_id: String, item_id: Option, review_id: String, tool_kind: ToolReviewToolKind, tool_name: String, trigger: ToolReviewTrigger, created_at: u64, duration_started: bool, requested_additional_permissions: bool, requested_network_access: bool, } struct PendingGuardianReviewState { created_at: u64, } #[derive(Clone, Default)] struct ToolReviewSummary { review_count: u64, guardian_review_count: u64, user_review_count: u64, final_approval_outcome: Option, requested_additional_permissions: bool, requested_network_access: bool, } #[derive(Clone)] struct ThreadMetadataState { thread_source: Option<&'static str>, initialization_mode: ThreadInitializationMode, subagent_source: Option, parent_thread_id: Option, } impl Default for ThreadMetadataState { fn default() -> Self { Self { thread_source: None, initialization_mode: ThreadInitializationMode::New, subagent_source: None, parent_thread_id: None, } } } impl ThreadMetadataState { fn from_thread_metadata( session_source: &SessionSource, initialization_mode: ThreadInitializationMode, ) -> Self { let (subagent_source, parent_thread_id) = match session_source { SessionSource::SubAgent(subagent_source) => ( Some(subagent_source_name(subagent_source)), subagent_parent_thread_id(subagent_source), ), SessionSource::Cli | SessionSource::VSCode | SessionSource::Exec | SessionSource::Mcp | SessionSource::Custom(_) | SessionSource::Unknown => (None, None), }; Self { thread_source: session_source.thread_source_name(), initialization_mode, subagent_source, parent_thread_id, } } } enum RequestState { TurnStart(PendingTurnStartState), TurnSteer(PendingTurnSteerState), } struct PendingTurnStartState { thread_id: String, num_input_images: usize, } struct PendingTurnSteerState { thread_id: String, expected_turn_id: String, num_input_images: usize, created_at: u64, } #[derive(Clone)] struct CompletedTurnState { status: Option, turn_error: Option, completed_at: u64, duration_ms: Option, } struct TurnState { connection_id: Option, thread_id: Option, num_input_images: Option, resolved_config: Option, started_at: Option, token_usage: Option, completed: Option, steer_count: usize, } impl AnalyticsReducer { pub(crate) async fn ingest(&mut self, input: AnalyticsFact, out: &mut Vec) { match input { AnalyticsFact::Initialize { connection_id, params, product_client_id, runtime, rpc_transport, } => { self.ingest_initialize( connection_id, params, product_client_id, runtime, rpc_transport, ); } AnalyticsFact::ClientRequest { connection_id, request_id, request, } => { self.ingest_request(connection_id, request_id, *request); } AnalyticsFact::ClientResponse { connection_id, response, } => { self.ingest_response(connection_id, *response, out); } AnalyticsFact::ErrorResponse { connection_id, request_id, error: _, error_type, } => { self.ingest_error_response(connection_id, request_id, error_type, out); } AnalyticsFact::Notification { connection_id, notification, } => { self.ingest_notification(connection_id, *notification, out); } AnalyticsFact::ServerRequest { connection_id, request, } => { self.ingest_server_request(connection_id, *request); } AnalyticsFact::ServerResponse { response } => { self.ingest_server_response(*response, out); } AnalyticsFact::Custom(input) => match input { CustomAnalyticsFact::SubAgentThreadStarted(input) => { self.ingest_subagent_thread_started(input, out); } CustomAnalyticsFact::Compaction(input) => { self.ingest_compaction(*input, out); } CustomAnalyticsFact::GuardianReview(input) => { self.ingest_guardian_review(*input, out); } CustomAnalyticsFact::TurnResolvedConfig(input) => { self.ingest_turn_resolved_config(*input, out); } CustomAnalyticsFact::TurnTokenUsage(input) => { self.ingest_turn_token_usage(*input, out); } CustomAnalyticsFact::SkillInvoked(input) => { self.ingest_skill_invoked(input, out).await; } CustomAnalyticsFact::AppMentioned(input) => { self.ingest_app_mentioned(input, out); } CustomAnalyticsFact::AppUsed(input) => { self.ingest_app_used(input, out); } CustomAnalyticsFact::HookRun(input) => { self.ingest_hook_run(input, out); } CustomAnalyticsFact::PluginUsed(input) => { self.ingest_plugin_used(input, out); } CustomAnalyticsFact::PluginStateChanged(input) => { self.ingest_plugin_state_changed(input, out); } }, } } fn ingest_initialize( &mut self, connection_id: u64, params: InitializeParams, product_client_id: String, runtime: CodexRuntimeMetadata, rpc_transport: AppServerRpcTransport, ) { self.connections.insert( connection_id, ConnectionState { app_server_client: CodexAppServerClientMetadata { product_client_id, client_name: Some(params.client_info.name), client_version: Some(params.client_info.version), rpc_transport, experimental_api_enabled: params .capabilities .map(|capabilities| capabilities.experimental_api), }, runtime, }, ); } fn ingest_subagent_thread_started( &mut self, input: SubAgentThreadStartedInput, out: &mut Vec, ) { out.push(TrackEventRequest::ThreadInitialized( subagent_thread_started_event_request(input), )); } fn ingest_guardian_review( &mut self, input: GuardianReviewEventParams, out: &mut Vec, ) { let Some(connection_id) = self.thread_connections.get(&input.thread_id) else { tracing::warn!( thread_id = %input.thread_id, turn_id = %input.turn_id, review_id = %input.review_id, "dropping guardian analytics event: missing thread connection metadata" ); return; }; let Some(connection_state) = self.connections.get(connection_id) else { tracing::warn!( thread_id = %input.thread_id, turn_id = %input.turn_id, review_id = %input.review_id, connection_id, "dropping guardian analytics event: missing connection metadata" ); return; }; out.push(TrackEventRequest::GuardianReview(Box::new( GuardianReviewEventRequest { event_type: "codex_guardian_review", event_params: GuardianReviewEventPayload { app_server_client: connection_state.app_server_client.clone(), runtime: connection_state.runtime.clone(), guardian_review: input, }, }, ))); } fn ingest_request( &mut self, connection_id: u64, request_id: RequestId, request: ClientRequest, ) { match request { ClientRequest::TurnStart { params, .. } => { self.requests.insert( (connection_id, request_id), RequestState::TurnStart(PendingTurnStartState { thread_id: params.thread_id, num_input_images: num_input_images(¶ms.input), }), ); } ClientRequest::TurnSteer { params, .. } => { self.requests.insert( (connection_id, request_id), RequestState::TurnSteer(PendingTurnSteerState { thread_id: params.thread_id, expected_turn_id: params.expected_turn_id, num_input_images: num_input_images(¶ms.input), created_at: now_unix_seconds(), }), ); } _ => {} } } fn ingest_turn_resolved_config( &mut self, input: TurnResolvedConfigFact, out: &mut Vec, ) { let turn_id = input.turn_id.clone(); let thread_id = input.thread_id.clone(); let num_input_images = input.num_input_images; let turn_state = self.turns.entry(turn_id.clone()).or_insert(TurnState { connection_id: None, thread_id: None, num_input_images: None, resolved_config: None, started_at: None, token_usage: None, completed: None, steer_count: 0, }); turn_state.thread_id = Some(thread_id); turn_state.num_input_images = Some(num_input_images); turn_state.resolved_config = Some(input); self.maybe_emit_turn_event(&turn_id, out); } fn ingest_turn_token_usage( &mut self, input: TurnTokenUsageFact, out: &mut Vec, ) { let turn_id = input.turn_id.clone(); let turn_state = self.turns.entry(turn_id.clone()).or_insert(TurnState { connection_id: None, thread_id: None, num_input_images: None, resolved_config: None, started_at: None, token_usage: None, completed: None, steer_count: 0, }); turn_state.thread_id = Some(input.thread_id); turn_state.token_usage = Some(input.token_usage); self.maybe_emit_turn_event(&turn_id, out); } async fn ingest_skill_invoked( &mut self, input: SkillInvokedInput, out: &mut Vec, ) { let SkillInvokedInput { tracking, invocations, } = input; for invocation in invocations { let skill_scope = match invocation.skill_scope { SkillScope::User => "user", SkillScope::Repo => "repo", SkillScope::System => "system", SkillScope::Admin => "admin", }; let repo_root = get_git_repo_root(invocation.skill_path.as_path()); let repo_url = if let Some(root) = repo_root.as_ref() { collect_git_info(root) .await .and_then(|info| info.repository_url) } else { None }; let skill_id = skill_id_for_local_skill( repo_url.as_deref(), repo_root.as_deref(), invocation.skill_path.as_path(), invocation.skill_name.as_str(), ); out.push(TrackEventRequest::SkillInvocation( SkillInvocationEventRequest { event_type: "skill_invocation", skill_id, skill_name: invocation.skill_name.clone(), event_params: SkillInvocationEventParams { thread_id: Some(tracking.thread_id.clone()), invoke_type: Some(invocation.invocation_type), model_slug: Some(tracking.model_slug.clone()), product_client_id: Some(originator().value), repo_url, skill_scope: Some(skill_scope.to_string()), }, }, )); } } fn ingest_app_mentioned(&mut self, input: AppMentionedInput, out: &mut Vec) { let AppMentionedInput { tracking, mentions } = input; out.extend(mentions.into_iter().map(|mention| { let event_params = codex_app_metadata(&tracking, mention); TrackEventRequest::AppMentioned(CodexAppMentionedEventRequest { event_type: "codex_app_mentioned", event_params, }) })); } fn ingest_app_used(&mut self, input: AppUsedInput, out: &mut Vec) { let AppUsedInput { tracking, app } = input; let event_params = codex_app_metadata(&tracking, app); out.push(TrackEventRequest::AppUsed(CodexAppUsedEventRequest { event_type: "codex_app_used", event_params, })); } fn ingest_hook_run(&mut self, input: HookRunInput, out: &mut Vec) { let HookRunInput { tracking, hook } = input; out.push(TrackEventRequest::HookRun(CodexHookRunEventRequest { event_type: "codex_hook_run", event_params: codex_hook_run_metadata(&tracking, hook), })); } fn ingest_plugin_used(&mut self, input: PluginUsedInput, out: &mut Vec) { let PluginUsedInput { tracking, plugin } = input; out.push(TrackEventRequest::PluginUsed(CodexPluginUsedEventRequest { event_type: "codex_plugin_used", event_params: codex_plugin_used_metadata(&tracking, plugin), })); } fn ingest_plugin_state_changed( &mut self, input: PluginStateChangedInput, out: &mut Vec, ) { let PluginStateChangedInput { plugin, state } = input; let event = CodexPluginEventRequest { event_type: plugin_state_event_type(state), event_params: codex_plugin_metadata(plugin), }; out.push(match state { PluginState::Installed => TrackEventRequest::PluginInstalled(event), PluginState::Uninstalled => TrackEventRequest::PluginUninstalled(event), PluginState::Enabled => TrackEventRequest::PluginEnabled(event), PluginState::Disabled => TrackEventRequest::PluginDisabled(event), }); } fn ingest_response( &mut self, connection_id: u64, response: ClientResponse, out: &mut Vec, ) { match response { ClientResponse::ThreadStart { response, .. } => { self.emit_thread_initialized( connection_id, response.thread, response.model, ThreadInitializationMode::New, out, ); } ClientResponse::ThreadResume { response, .. } => { self.emit_thread_initialized( connection_id, response.thread, response.model, ThreadInitializationMode::Resumed, out, ); } ClientResponse::ThreadFork { response, .. } => { self.emit_thread_initialized( connection_id, response.thread, response.model, ThreadInitializationMode::Forked, 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: None, token_usage: None, completed: None, steer_count: 0, }); 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); } ClientResponse::TurnSteer { request_id, response, } => { self.ingest_turn_steer_response(connection_id, request_id, response, out); } _ => {} } } fn ingest_server_request(&mut self, _connection_id: u64, request: ServerRequest) { match request { ServerRequest::CommandExecutionRequestApproval { request_id, params } => { let requested_network_access = params.network_approval_context.is_some() || params .proposed_network_policy_amendments .as_ref() .is_some_and(|amendments| !amendments.is_empty()) || params .additional_permissions .as_ref() .and_then(|permissions| permissions.network.as_ref()) .and_then(|network| network.enabled) .unwrap_or(false); let requested_additional_permissions = params.additional_permissions.is_some() || params.proposed_execpolicy_amendment.is_some(); let trigger = if params.approval_id.is_some() { ToolReviewTrigger::SubcommandExecve } else if requested_network_access { ToolReviewTrigger::NetworkRetry } else if requested_additional_permissions { ToolReviewTrigger::SandboxRetry } else { ToolReviewTrigger::Initial }; self.tool_review_requests.insert( request_id.clone(), PendingToolReviewState { thread_id: params.thread_id, turn_id: params.turn_id, item_id: Some(params.item_id), review_id: user_review_id(&request_id), tool_kind: ToolReviewToolKind::CommandExecution, tool_name: "shell".to_string(), trigger, created_at: now_unix_secs(), duration_started: true, requested_additional_permissions, requested_network_access, }, ); } ServerRequest::FileChangeRequestApproval { request_id, params } => { let requested_additional_permissions = params.grant_root.is_some(); self.tool_review_requests.insert( request_id.clone(), PendingToolReviewState { thread_id: params.thread_id, turn_id: params.turn_id, item_id: Some(params.item_id), review_id: user_review_id(&request_id), tool_kind: ToolReviewToolKind::FileChange, tool_name: "apply_patch".to_string(), trigger: if requested_additional_permissions { ToolReviewTrigger::SandboxRetry } else { ToolReviewTrigger::Initial }, created_at: now_unix_secs(), duration_started: true, requested_additional_permissions, requested_network_access: false, }, ); } ServerRequest::PermissionsRequestApproval { request_id, params } => { let requested_network_access = params .permissions .network .as_ref() .and_then(|network| network.enabled) .unwrap_or(false); let requested_additional_permissions = requested_network_access || params.permissions.file_system.is_some(); let trigger = if requested_network_access { ToolReviewTrigger::NetworkRetry } else if requested_additional_permissions { ToolReviewTrigger::SandboxRetry } else { ToolReviewTrigger::Initial }; self.tool_review_requests.insert( request_id.clone(), PendingToolReviewState { thread_id: params.thread_id, turn_id: params.turn_id, item_id: Some(params.item_id), review_id: user_review_id(&request_id), tool_kind: ToolReviewToolKind::Permissions, tool_name: "permissions".to_string(), trigger, created_at: now_unix_secs(), duration_started: true, requested_additional_permissions, requested_network_access, }, ); } _ => {} } } fn ingest_server_response( &mut self, response: ServerResponse, out: &mut Vec, ) { match response { ServerResponse::CommandExecutionRequestApproval { request_id, response, } => { let Some(pending_review) = self.tool_review_requests.remove(&request_id) else { return; }; let status = command_execution_review_status(response.decision); self.emit_tool_review_event(pending_review, ToolReviewReviewer::User, status, out); } ServerResponse::FileChangeRequestApproval { request_id, response, } => { let Some(pending_review) = self.tool_review_requests.remove(&request_id) else { return; }; let status = file_change_review_status(response.decision); self.emit_tool_review_event(pending_review, ToolReviewReviewer::User, status, out); } ServerResponse::PermissionsRequestApproval { request_id, response, } => { let Some(pending_review) = self.tool_review_requests.remove(&request_id) else { return; }; let status = match response.scope { PermissionGrantScope::Turn => ToolReviewStatus::Approved, PermissionGrantScope::Session => ToolReviewStatus::ApprovedForSession, }; self.emit_tool_review_event(pending_review, ToolReviewReviewer::User, status, out); } _ => {} } } fn ingest_error_response( &mut self, connection_id: u64, request_id: RequestId, error_type: Option, out: &mut Vec, ) { let Some(request) = self.requests.remove(&(connection_id, request_id)) else { return; }; self.ingest_request_error_response(connection_id, request, error_type, out); } fn ingest_request_error_response( &mut self, connection_id: u64, request: RequestState, error_type: Option, out: &mut Vec, ) { match request { RequestState::TurnStart(_) => {} RequestState::TurnSteer(pending_request) => { self.ingest_turn_steer_error_response( connection_id, pending_request, error_type, out, ); } } } fn ingest_turn_steer_error_response( &mut self, connection_id: u64, pending_request: PendingTurnSteerState, error_type: Option, out: &mut Vec, ) { self.emit_turn_steer_event( connection_id, pending_request, /*accepted_turn_id*/ None, TurnSteerResult::Rejected, rejection_reason_from_error_type(error_type), out, ); } fn ingest_notification( &mut self, connection_id: u64, notification: ServerNotification, out: &mut Vec, ) { match notification { ServerNotification::ItemStarted(notification) => { if let Some(item_id) = tool_item_id(¬ification.item) { self.tool_items .entry(item_id.to_string()) .or_insert_with(|| ToolItemState { connection_id, started_at: now_unix_secs(), }); } } ServerNotification::ItemCompleted(notification) => { let Some(item_id) = tool_item_id(¬ification.item) else { return; }; let Some(started) = self.tool_items.remove(item_id) else { return; }; let Some(connection_state) = self.connections.get(&started.connection_id) else { return; }; let completed_at = now_unix_secs(); if let Some(event) = tool_item_event(ToolItemEventInput { thread_id: ¬ification.thread_id, turn_id: ¬ification.turn_id, item: ¬ification.item, started_at: started.started_at, completed_at, connection_state, thread_metadata: self.thread_metadata.get(¬ification.thread_id), review_summary: self.tool_review_summaries.get(item_id), }) { out.push(event); } self.tool_review_summaries.remove(item_id); } ServerNotification::ItemGuardianApprovalReviewStarted(notification) => { self.guardian_review_starts.insert( notification.review_id, PendingGuardianReviewState { created_at: now_unix_secs(), }, ); } ServerNotification::ItemGuardianApprovalReviewCompleted(notification) => { self.ingest_guardian_tool_review_completed(notification, out); } ServerNotification::TurnStarted(notification) => { let turn_state = self.turns.entry(notification.turn.id).or_insert(TurnState { connection_id: None, thread_id: None, num_input_images: None, resolved_config: None, started_at: None, token_usage: None, completed: None, steer_count: 0, }); turn_state.started_at = notification .turn .started_at .and_then(|started_at| u64::try_from(started_at).ok()); } ServerNotification::TurnCompleted(notification) => { let turn_state = self.turns .entry(notification.turn.id.clone()) .or_insert(TurnState { connection_id: None, thread_id: None, num_input_images: None, resolved_config: None, started_at: None, token_usage: None, completed: None, steer_count: 0, }); turn_state.completed = Some(CompletedTurnState { status: analytics_turn_status(notification.turn.status), turn_error: notification .turn .error .and_then(|error| error.codex_error_info), completed_at: notification .turn .completed_at .and_then(|completed_at| u64::try_from(completed_at).ok()) .unwrap_or_default(), duration_ms: notification .turn .duration_ms .and_then(|duration_ms| u64::try_from(duration_ms).ok()), }); let turn_id = notification.turn.id; self.maybe_emit_turn_event(&turn_id, out); } _ => {} } } fn emit_thread_initialized( &mut self, connection_id: u64, thread: codex_app_server_protocol::Thread, model: String, initialization_mode: ThreadInitializationMode, out: &mut Vec, ) { let thread_source: SessionSource = thread.source.into(); let thread_id = thread.id; let Some(connection_state) = self.connections.get(&connection_id) else { return; }; let thread_metadata = ThreadMetadataState::from_thread_metadata(&thread_source, initialization_mode); self.thread_connections .insert(thread_id.clone(), connection_id); self.thread_metadata .insert(thread_id.clone(), thread_metadata.clone()); out.push(TrackEventRequest::ThreadInitialized( ThreadInitializedEvent { event_type: "codex_thread_initialized", event_params: ThreadInitializedEventParams { thread_id, app_server_client: connection_state.app_server_client.clone(), runtime: connection_state.runtime.clone(), model, ephemeral: thread.ephemeral, thread_source: thread_metadata.thread_source, initialization_mode, subagent_source: thread_metadata.subagent_source, parent_thread_id: thread_metadata.parent_thread_id, created_at: u64::try_from(thread.created_at).unwrap_or_default(), }, }, )); } fn ingest_compaction(&mut self, input: CodexCompactionEvent, out: &mut Vec) { let Some(connection_id) = self.thread_connections.get(&input.thread_id) else { tracing::warn!( thread_id = %input.thread_id, turn_id = %input.turn_id, "dropping compaction analytics event: missing thread connection metadata" ); return; }; let Some(connection_state) = self.connections.get(connection_id) else { tracing::warn!( thread_id = %input.thread_id, turn_id = %input.turn_id, connection_id, "dropping compaction analytics event: missing connection metadata" ); return; }; let Some(thread_metadata) = self.thread_metadata.get(&input.thread_id) else { tracing::warn!( thread_id = %input.thread_id, turn_id = %input.turn_id, "dropping compaction analytics event: missing thread lifecycle metadata" ); return; }; out.push(TrackEventRequest::Compaction(Box::new( CodexCompactionEventRequest { event_type: "codex_compaction_event", event_params: codex_compaction_event_params( input, connection_state.app_server_client.clone(), connection_state.runtime.clone(), thread_metadata.thread_source, thread_metadata.subagent_source.clone(), thread_metadata.parent_thread_id.clone(), ), }, ))); } fn ingest_guardian_tool_review_completed( &mut self, notification: codex_app_server_protocol::ItemGuardianApprovalReviewCompletedNotification, out: &mut Vec, ) { let completed_at = now_unix_secs(); let started = self.guardian_review_starts.remove(¬ification.review_id); let created_at = started .as_ref() .map(|start| start.created_at) .unwrap_or(completed_at); let Some(status) = guardian_review_status(notification.review.status) else { return; }; let (tool_kind, tool_name, trigger) = guardian_review_tool_metadata(¬ification.action); let pending_review = PendingToolReviewState { thread_id: notification.thread_id, turn_id: notification.turn_id, item_id: notification.target_item_id, review_id: notification.review_id, tool_kind, tool_name, trigger, created_at, duration_started: started.is_some(), requested_additional_permissions: matches!( notification.action, GuardianApprovalReviewAction::ApplyPatch { .. } | GuardianApprovalReviewAction::NetworkAccess { .. } ), requested_network_access: matches!( notification.action, GuardianApprovalReviewAction::NetworkAccess { .. } ), }; self.emit_tool_review_event_at( pending_review, ToolReviewReviewer::Guardian, status, completed_at, out, ); } fn ingest_turn_steer_response( &mut self, connection_id: u64, request_id: RequestId, response: TurnSteerResponse, out: &mut Vec, ) { let Some(RequestState::TurnSteer(pending_request)) = self.requests.remove(&(connection_id, request_id)) else { return; }; if let Some(turn_state) = self.turns.get_mut(&response.turn_id) { turn_state.steer_count += 1; } self.emit_turn_steer_event( connection_id, pending_request, Some(response.turn_id), TurnSteerResult::Accepted, /*rejection_reason*/ None, out, ); } fn emit_turn_steer_event( &mut self, connection_id: u64, pending_request: PendingTurnSteerState, accepted_turn_id: Option, result: TurnSteerResult, rejection_reason: Option, out: &mut Vec, ) { let Some(connection_state) = self.connections.get(&connection_id) else { return; }; let Some(thread_metadata) = self.thread_metadata.get(&pending_request.thread_id) else { tracing::warn!( thread_id = %pending_request.thread_id, "dropping turn steer analytics event: missing thread lifecycle metadata" ); return; }; out.push(TrackEventRequest::TurnSteer(CodexTurnSteerEventRequest { event_type: "codex_turn_steer_event", event_params: CodexTurnSteerEventParams { thread_id: pending_request.thread_id, expected_turn_id: Some(pending_request.expected_turn_id), accepted_turn_id, app_server_client: connection_state.app_server_client.clone(), runtime: connection_state.runtime.clone(), thread_source: thread_metadata.thread_source.map(str::to_string), subagent_source: thread_metadata.subagent_source.clone(), parent_thread_id: thread_metadata.parent_thread_id.clone(), num_input_images: pending_request.num_input_images, result, rejection_reason, created_at: pending_request.created_at, }, })); } fn emit_tool_review_event( &mut self, pending_review: PendingToolReviewState, reviewer: ToolReviewReviewer, status: ToolReviewStatus, out: &mut Vec, ) { self.emit_tool_review_event_at(pending_review, reviewer, status, now_unix_secs(), out); } fn emit_tool_review_event_at( &mut self, pending_review: PendingToolReviewState, reviewer: ToolReviewReviewer, status: ToolReviewStatus, completed_at: u64, out: &mut Vec, ) { if let Some(item_id) = pending_review.item_id.as_ref() { self.record_tool_review_summary(item_id, reviewer, status, &pending_review); } let thread_metadata = self .thread_metadata .get(&pending_review.thread_id) .cloned() .unwrap_or_default(); out.push(TrackEventRequest::ToolCallReview( CodexToolCallReviewEventRequest { event_type: "codex_tool_call_review_event", event_params: CodexToolCallReviewEventParams { thread_id: pending_review.thread_id, turn_id: pending_review.turn_id, item_id: pending_review.item_id, review_id: pending_review.review_id, thread_source: thread_metadata.thread_source, subagent_source: thread_metadata.subagent_source, parent_thread_id: thread_metadata.parent_thread_id, tool_kind: pending_review.tool_kind, tool_name: pending_review.tool_name, reviewer, trigger: pending_review.trigger, status, created_at: pending_review.created_at, completed_at: Some(completed_at), duration_ms: pending_review .duration_started .then(|| { completed_duration_ms( /*item_duration_ms*/ None, pending_review.created_at, completed_at, ) }) .flatten(), }, }, )); } fn record_tool_review_summary( &mut self, item_id: &str, reviewer: ToolReviewReviewer, status: ToolReviewStatus, pending_review: &PendingToolReviewState, ) { let summary = self .tool_review_summaries .entry(item_id.to_string()) .or_default(); summary.review_count += 1; match reviewer { ToolReviewReviewer::Guardian => summary.guardian_review_count += 1, ToolReviewReviewer::User => summary.user_review_count += 1, } summary.final_approval_outcome = Some(tool_item_final_approval_outcome(reviewer, status)); summary.requested_additional_permissions |= pending_review.requested_additional_permissions; summary.requested_network_access |= pending_review.requested_network_access; } fn maybe_emit_turn_event(&mut self, turn_id: &str, out: &mut Vec) { let Some(turn_state) = self.turns.get(turn_id) else { return; }; if turn_state.thread_id.is_none() || turn_state.num_input_images.is_none() || turn_state.resolved_config.is_none() || turn_state.completed.is_none() { return; } let connection_metadata = turn_state .connection_id .and_then(|connection_id| self.connections.get(&connection_id)) .map(|connection_state| { ( connection_state.app_server_client.clone(), connection_state.runtime.clone(), ) }); let Some((app_server_client, runtime)) = connection_metadata else { if let Some(connection_id) = turn_state.connection_id { tracing::warn!( turn_id, connection_id, "dropping turn analytics event: missing connection metadata" ); } return; }; let Some(thread_id) = turn_state.thread_id.as_ref() else { return; }; let Some(thread_metadata) = self.thread_metadata.get(thread_id) else { tracing::warn!( thread_id, turn_id, "dropping turn analytics event: missing thread lifecycle metadata" ); return; }; out.push(TrackEventRequest::TurnEvent(Box::new( CodexTurnEventRequest { event_type: "codex_turn_event", event_params: codex_turn_event_params( app_server_client, runtime, turn_id.to_string(), turn_state, thread_metadata, ), }, ))); self.turns.remove(turn_id); } } fn now_unix_secs() -> u64 { SystemTime::now() .duration_since(UNIX_EPOCH) .map(|duration| duration.as_secs()) .unwrap_or_default() } fn tool_item_id(item: &ThreadItem) -> Option<&str> { match item { ThreadItem::CommandExecution { id, .. } | ThreadItem::FileChange { id, .. } | ThreadItem::McpToolCall { id, .. } | ThreadItem::DynamicToolCall { id, .. } | ThreadItem::CollabAgentToolCall { id, .. } | ThreadItem::WebSearch { id, .. } | ThreadItem::ImageGeneration { id, .. } => Some(id), _ => None, } } struct ToolItemEventInput<'a> { thread_id: &'a str, turn_id: &'a str, item: &'a ThreadItem, started_at: u64, completed_at: u64, connection_state: &'a ConnectionState, thread_metadata: Option<&'a ThreadMetadataState>, review_summary: Option<&'a ToolReviewSummary>, } fn tool_item_event(input: ToolItemEventInput<'_>) -> Option { let ToolItemEventInput { thread_id, turn_id, item, started_at, completed_at, connection_state, thread_metadata, review_summary, } = input; match item { ThreadItem::CommandExecution { id, source, status, command_actions, exit_code, duration_ms, .. } => { let (terminal_status, failure_kind) = command_execution_outcome(status)?; let base = tool_item_base( thread_id, turn_id, id.clone(), command_execution_tool_name(*source).to_string(), ToolItemOutcome { terminal_status, failure_kind, duration_ms: completed_duration_ms( option_i64_to_u64(*duration_ms), started_at, completed_at, ), }, ToolItemContext { started_at, completed_at, connection_state, thread_metadata, review_summary, }, ); Some(TrackEventRequest::CommandExecution( CodexCommandExecutionEventRequest { event_type: "codex_command_execution_event", event_params: CodexCommandExecutionEventParams { base, command_execution_source: command_execution_source_kind(*source), command_execution_family: command_execution_family(*source), exit_code: *exit_code, command_action_count: Some(usize_to_u64(command_actions.len())), }, }, )) } ThreadItem::FileChange { id, changes, status, } => { let (terminal_status, failure_kind) = patch_apply_outcome(status)?; let counts = file_change_counts(changes); let base = tool_item_base( thread_id, turn_id, id.clone(), "apply_patch".to_string(), ToolItemOutcome { terminal_status, failure_kind, duration_ms: observed_completed_duration_ms(started_at, completed_at), }, ToolItemContext { started_at, completed_at, connection_state, thread_metadata, review_summary, }, ); Some(TrackEventRequest::FileChange(CodexFileChangeEventRequest { event_type: "codex_file_change_event", event_params: CodexFileChangeEventParams { base, file_change_count: usize_to_u64(changes.len()), file_add_count: counts.add, file_update_count: counts.update, file_delete_count: counts.delete, file_move_count: counts.move_, }, })) } ThreadItem::McpToolCall { id, server, tool, status, error, duration_ms, .. } => { let (terminal_status, failure_kind) = mcp_tool_call_outcome(status)?; let base = tool_item_base( thread_id, turn_id, id.clone(), tool.clone(), ToolItemOutcome { terminal_status, failure_kind, duration_ms: completed_duration_ms( option_i64_to_u64(*duration_ms), started_at, completed_at, ), }, ToolItemContext { started_at, completed_at, connection_state, thread_metadata, review_summary, }, ); Some(TrackEventRequest::McpToolCall( CodexMcpToolCallEventRequest { event_type: "codex_mcp_tool_call_event", event_params: CodexMcpToolCallEventParams { base, mcp_server_name: server.clone(), mcp_tool_name: tool.clone(), mcp_error_present: error.is_some(), mcp_error_code: None, }, }, )) } ThreadItem::DynamicToolCall { id, tool, status, content_items, success, duration_ms, .. } => { let (terminal_status, failure_kind) = dynamic_tool_call_outcome(status)?; let counts = content_items .as_ref() .map(|items| dynamic_content_counts(items)); let base = tool_item_base( thread_id, turn_id, id.clone(), tool.clone(), ToolItemOutcome { terminal_status, failure_kind, duration_ms: completed_duration_ms( option_i64_to_u64(*duration_ms), started_at, completed_at, ), }, ToolItemContext { started_at, completed_at, connection_state, thread_metadata, review_summary, }, ); Some(TrackEventRequest::DynamicToolCall( CodexDynamicToolCallEventRequest { event_type: "codex_dynamic_tool_call_event", event_params: CodexDynamicToolCallEventParams { base, dynamic_tool_name: tool.clone(), success: *success, output_content_item_count: counts.map(|counts| counts.total), output_text_item_count: counts.map(|counts| counts.text), output_image_item_count: counts.map(|counts| counts.image), }, }, )) } ThreadItem::CollabAgentToolCall { id, tool, status, sender_thread_id, receiver_thread_ids, model, reasoning_effort, agents_states, .. } => { let (terminal_status, failure_kind) = collab_tool_call_outcome(status)?; let base = tool_item_base( thread_id, turn_id, id.clone(), collab_agent_tool_name(tool).to_string(), ToolItemOutcome { terminal_status, failure_kind, duration_ms: observed_completed_duration_ms(started_at, completed_at), }, ToolItemContext { started_at, completed_at, connection_state, thread_metadata, review_summary, }, ); Some(TrackEventRequest::CollabAgentToolCall( CodexCollabAgentToolCallEventRequest { event_type: "codex_collab_agent_tool_call_event", event_params: CodexCollabAgentToolCallEventParams { base, collab_agent_tool: collab_agent_tool_kind(tool), sender_thread_id: sender_thread_id.clone(), receiver_thread_count: usize_to_u64(receiver_thread_ids.len()), receiver_thread_ids: Some(receiver_thread_ids.clone()), requested_model: model.clone(), requested_reasoning_effort: reasoning_effort .as_ref() .and_then(serialize_enum_as_string), agent_state_count: Some(usize_to_u64(agents_states.len())), completed_agent_count: Some(usize_to_u64( agents_states .values() .filter(|state| state.status == CollabAgentStatus::Completed) .count(), )), failed_agent_count: Some(usize_to_u64( agents_states .values() .filter(|state| { matches!( state.status, CollabAgentStatus::Errored | CollabAgentStatus::Shutdown | CollabAgentStatus::NotFound ) }) .count(), )), }, }, )) } ThreadItem::WebSearch { id, query, action } => { let base = tool_item_base( thread_id, turn_id, id.clone(), "web_search".to_string(), ToolItemOutcome { terminal_status: ToolItemTerminalStatus::Completed, failure_kind: None, duration_ms: observed_completed_duration_ms(started_at, completed_at), }, ToolItemContext { started_at, completed_at, connection_state, thread_metadata, review_summary, }, ); Some(TrackEventRequest::WebSearch(CodexWebSearchEventRequest { event_type: "codex_web_search_event", event_params: CodexWebSearchEventParams { base, web_search_action: action.as_ref().map(web_search_action_kind), query_present: !query.trim().is_empty(), query_count: web_search_query_count(query, action.as_ref()), }, })) } ThreadItem::ImageGeneration { id, status, revised_prompt, saved_path, .. } => { let (terminal_status, failure_kind) = image_generation_outcome(status.as_str()); let base = tool_item_base( thread_id, turn_id, id.clone(), "image_generation".to_string(), ToolItemOutcome { terminal_status, failure_kind, duration_ms: observed_completed_duration_ms(started_at, completed_at), }, ToolItemContext { started_at, completed_at, connection_state, thread_metadata, review_summary, }, ); Some(TrackEventRequest::ImageGeneration( CodexImageGenerationEventRequest { event_type: "codex_image_generation_event", event_params: CodexImageGenerationEventParams { base, image_generation_status: status.clone(), revised_prompt_present: revised_prompt.is_some(), saved_path_present: saved_path.is_some(), }, }, )) } _ => None, } } struct ToolItemOutcome { terminal_status: ToolItemTerminalStatus, failure_kind: Option, duration_ms: Option, } struct ToolItemContext<'a> { started_at: u64, completed_at: u64, connection_state: &'a ConnectionState, thread_metadata: Option<&'a ThreadMetadataState>, review_summary: Option<&'a ToolReviewSummary>, } fn tool_item_base( thread_id: &str, turn_id: &str, item_id: String, tool_name: String, outcome: ToolItemOutcome, context: ToolItemContext<'_>, ) -> CodexToolItemEventBase { let thread_metadata = context.thread_metadata.cloned().unwrap_or_default(); let review_summary = context.review_summary.cloned().unwrap_or_default(); CodexToolItemEventBase { thread_id: thread_id.to_string(), turn_id: turn_id.to_string(), item_id, app_server_client: context.connection_state.app_server_client.clone(), runtime: context.connection_state.runtime.clone(), thread_source: thread_metadata.thread_source, subagent_source: thread_metadata.subagent_source, parent_thread_id: thread_metadata.parent_thread_id, tool_name, started_at: context.started_at, completed_at: Some(context.completed_at), duration_ms: outcome.duration_ms, execution_started: true, review_count: review_summary.review_count, guardian_review_count: review_summary.guardian_review_count, user_review_count: review_summary.user_review_count, final_approval_outcome: review_summary .final_approval_outcome .unwrap_or(ToolItemFinalApprovalOutcome::NotNeeded), terminal_status: outcome.terminal_status, failure_kind: outcome.failure_kind, requested_additional_permissions: review_summary.requested_additional_permissions, requested_network_access: review_summary.requested_network_access, retry_count: review_summary.review_count.saturating_sub(1), } } fn completed_duration_ms( item_duration_ms: Option, started_at: u64, completed_at: u64, ) -> Option { item_duration_ms.or_else(|| { completed_at .checked_sub(started_at) .map(|duration_secs| duration_secs.saturating_mul(1000)) }) } fn observed_completed_duration_ms(started_at: u64, completed_at: u64) -> Option { completed_duration_ms(/*item_duration_ms*/ None, started_at, completed_at) } fn user_review_id(request_id: &RequestId) -> String { format!("user:{request_id}") } fn command_execution_review_status(decision: CommandExecutionApprovalDecision) -> ToolReviewStatus { match decision { CommandExecutionApprovalDecision::Accept => ToolReviewStatus::Approved, CommandExecutionApprovalDecision::AcceptForSession => ToolReviewStatus::ApprovedForSession, CommandExecutionApprovalDecision::AcceptWithExecpolicyAmendment { .. } => { ToolReviewStatus::ApprovedExecpolicyAmendment } CommandExecutionApprovalDecision::ApplyNetworkPolicyAmendment { network_policy_amendment, } => match network_policy_amendment.action { NetworkPolicyRuleAction::Allow => ToolReviewStatus::NetworkPolicyAllow, NetworkPolicyRuleAction::Deny => ToolReviewStatus::NetworkPolicyDeny, }, CommandExecutionApprovalDecision::Decline => ToolReviewStatus::Denied, CommandExecutionApprovalDecision::Cancel => ToolReviewStatus::Aborted, } } fn file_change_review_status(decision: FileChangeApprovalDecision) -> ToolReviewStatus { match decision { FileChangeApprovalDecision::Accept => ToolReviewStatus::Approved, FileChangeApprovalDecision::AcceptForSession => ToolReviewStatus::ApprovedForSession, FileChangeApprovalDecision::Decline => ToolReviewStatus::Denied, FileChangeApprovalDecision::Cancel => ToolReviewStatus::Aborted, } } fn guardian_review_status(status: GuardianApprovalReviewStatus) -> Option { match status { GuardianApprovalReviewStatus::InProgress => None, GuardianApprovalReviewStatus::Approved => Some(ToolReviewStatus::Approved), GuardianApprovalReviewStatus::Denied => Some(ToolReviewStatus::Denied), GuardianApprovalReviewStatus::TimedOut => Some(ToolReviewStatus::TimedOut), GuardianApprovalReviewStatus::Aborted => Some(ToolReviewStatus::Aborted), } } fn guardian_review_tool_metadata( action: &GuardianApprovalReviewAction, ) -> (ToolReviewToolKind, String, ToolReviewTrigger) { match action { GuardianApprovalReviewAction::Command { source, .. } => ( ToolReviewToolKind::CommandExecution, app_server_guardian_command_tool_name(*source).to_string(), ToolReviewTrigger::Initial, ), GuardianApprovalReviewAction::Execve { source, .. } => ( ToolReviewToolKind::CommandExecution, app_server_guardian_command_tool_name(*source).to_string(), ToolReviewTrigger::SubcommandExecve, ), GuardianApprovalReviewAction::ApplyPatch { .. } => ( ToolReviewToolKind::FileChange, "apply_patch".to_string(), ToolReviewTrigger::SandboxRetry, ), GuardianApprovalReviewAction::NetworkAccess { .. } => ( ToolReviewToolKind::NetworkAccess, "network_access".to_string(), ToolReviewTrigger::NetworkRetry, ), GuardianApprovalReviewAction::McpToolCall { tool_name, .. } => ( ToolReviewToolKind::McpToolCall, tool_name.clone(), ToolReviewTrigger::Initial, ), } } fn app_server_guardian_command_tool_name(source: AppServerGuardianCommandSource) -> &'static str { match source { AppServerGuardianCommandSource::Shell => "shell", AppServerGuardianCommandSource::UnifiedExec => "unified_exec", } } fn tool_item_final_approval_outcome( reviewer: ToolReviewReviewer, status: ToolReviewStatus, ) -> ToolItemFinalApprovalOutcome { match (reviewer, status) { (ToolReviewReviewer::Guardian, ToolReviewStatus::Approved) => { ToolItemFinalApprovalOutcome::GuardianApproved } ( ToolReviewReviewer::Guardian, ToolReviewStatus::Denied | ToolReviewStatus::NetworkPolicyDeny, ) => ToolItemFinalApprovalOutcome::GuardianDenied, (ToolReviewReviewer::Guardian, _) => ToolItemFinalApprovalOutcome::GuardianAborted, ( ToolReviewReviewer::User, ToolReviewStatus::Approved | ToolReviewStatus::ApprovedExecpolicyAmendment | ToolReviewStatus::NetworkPolicyAllow, ) => ToolItemFinalApprovalOutcome::UserApproved, (ToolReviewReviewer::User, ToolReviewStatus::ApprovedForSession) => { ToolItemFinalApprovalOutcome::UserApprovedForSession } ( ToolReviewReviewer::User, ToolReviewStatus::Denied | ToolReviewStatus::NetworkPolicyDeny, ) => ToolItemFinalApprovalOutcome::UserDenied, (ToolReviewReviewer::User, _) => ToolItemFinalApprovalOutcome::UserAborted, } } fn command_execution_source_kind(source: CommandExecutionSource) -> CommandExecutionSourceKind { match source { CommandExecutionSource::Agent => CommandExecutionSourceKind::Agent, CommandExecutionSource::UserShell => CommandExecutionSourceKind::UserShell, CommandExecutionSource::UnifiedExecStartup => { CommandExecutionSourceKind::UnifiedExecStartup } CommandExecutionSource::UnifiedExecInteraction => { CommandExecutionSourceKind::UnifiedExecInteraction } } } fn command_execution_family(source: CommandExecutionSource) -> CommandExecutionFamily { match source { CommandExecutionSource::Agent => CommandExecutionFamily::Shell, CommandExecutionSource::UserShell => CommandExecutionFamily::UserShell, CommandExecutionSource::UnifiedExecStartup | CommandExecutionSource::UnifiedExecInteraction => CommandExecutionFamily::UnifiedExec, } } fn command_execution_tool_name(source: CommandExecutionSource) -> &'static str { match source { CommandExecutionSource::UnifiedExecStartup | CommandExecutionSource::UnifiedExecInteraction => "unified_exec", CommandExecutionSource::UserShell => "user_shell", CommandExecutionSource::Agent => "shell", } } fn command_execution_outcome( status: &CommandExecutionStatus, ) -> Option<(ToolItemTerminalStatus, Option)> { match status { CommandExecutionStatus::InProgress => None, CommandExecutionStatus::Completed => Some((ToolItemTerminalStatus::Completed, None)), CommandExecutionStatus::Failed => Some(( ToolItemTerminalStatus::Failed, Some(ToolItemFailureKind::ToolError), )), CommandExecutionStatus::Declined => Some(( ToolItemTerminalStatus::Rejected, Some(ToolItemFailureKind::ApprovalDenied), )), } } fn patch_apply_outcome( status: &PatchApplyStatus, ) -> Option<(ToolItemTerminalStatus, Option)> { match status { PatchApplyStatus::InProgress => None, PatchApplyStatus::Completed => Some((ToolItemTerminalStatus::Completed, None)), PatchApplyStatus::Failed => Some(( ToolItemTerminalStatus::Failed, Some(ToolItemFailureKind::ToolError), )), PatchApplyStatus::Declined => Some(( ToolItemTerminalStatus::Rejected, Some(ToolItemFailureKind::ApprovalDenied), )), } } fn mcp_tool_call_outcome( status: &McpToolCallStatus, ) -> Option<(ToolItemTerminalStatus, Option)> { match status { McpToolCallStatus::InProgress => None, McpToolCallStatus::Completed => Some((ToolItemTerminalStatus::Completed, None)), McpToolCallStatus::Failed => Some(( ToolItemTerminalStatus::Failed, Some(ToolItemFailureKind::ToolError), )), } } fn dynamic_tool_call_outcome( status: &DynamicToolCallStatus, ) -> Option<(ToolItemTerminalStatus, Option)> { match status { DynamicToolCallStatus::InProgress => None, DynamicToolCallStatus::Completed => Some((ToolItemTerminalStatus::Completed, None)), DynamicToolCallStatus::Failed => Some(( ToolItemTerminalStatus::Failed, Some(ToolItemFailureKind::ToolError), )), } } fn collab_tool_call_outcome( status: &CollabAgentToolCallStatus, ) -> Option<(ToolItemTerminalStatus, Option)> { match status { CollabAgentToolCallStatus::InProgress => None, CollabAgentToolCallStatus::Completed => Some((ToolItemTerminalStatus::Completed, None)), CollabAgentToolCallStatus::Failed => Some(( ToolItemTerminalStatus::Failed, Some(ToolItemFailureKind::ToolError), )), } } fn image_generation_outcome(status: &str) -> (ToolItemTerminalStatus, Option) { match status { "failed" | "error" => ( ToolItemTerminalStatus::Failed, Some(ToolItemFailureKind::ToolError), ), _ => (ToolItemTerminalStatus::Completed, None), } } fn collab_agent_tool_name(tool: &CollabAgentTool) -> &'static str { match tool { CollabAgentTool::SpawnAgent => "spawn_agent", CollabAgentTool::SendInput => "send_input", CollabAgentTool::ResumeAgent => "resume_agent", CollabAgentTool::Wait => "wait_agent", CollabAgentTool::CloseAgent => "close_agent", } } fn collab_agent_tool_kind(tool: &CollabAgentTool) -> CollabAgentToolKind { match tool { CollabAgentTool::SpawnAgent => CollabAgentToolKind::SpawnAgent, CollabAgentTool::SendInput => CollabAgentToolKind::SendInput, CollabAgentTool::ResumeAgent => CollabAgentToolKind::ResumeAgent, CollabAgentTool::Wait => CollabAgentToolKind::Wait, CollabAgentTool::CloseAgent => CollabAgentToolKind::CloseAgent, } } #[derive(Default)] struct FileChangeCounts { add: u64, update: u64, delete: u64, move_: u64, } fn file_change_counts(changes: &[codex_app_server_protocol::FileUpdateChange]) -> FileChangeCounts { let mut counts = FileChangeCounts::default(); for change in changes { match &change.kind { PatchChangeKind::Add => counts.add += 1, PatchChangeKind::Delete => counts.delete += 1, PatchChangeKind::Update { move_path: Some(_) } => counts.move_ += 1, PatchChangeKind::Update { move_path: None } => counts.update += 1, } } counts } #[derive(Clone, Copy)] struct DynamicContentCounts { total: u64, text: u64, image: u64, } fn dynamic_content_counts(items: &[DynamicToolCallOutputContentItem]) -> DynamicContentCounts { let mut text = 0; let mut image = 0; for item in items { match item { DynamicToolCallOutputContentItem::InputText { .. } => text += 1, DynamicToolCallOutputContentItem::InputImage { .. } => image += 1, } } DynamicContentCounts { total: usize_to_u64(items.len()), text, image, } } fn web_search_action_kind(action: &WebSearchAction) -> WebSearchActionKind { match action { WebSearchAction::Search { .. } => WebSearchActionKind::Search, WebSearchAction::OpenPage { .. } => WebSearchActionKind::OpenPage, WebSearchAction::FindInPage { .. } => WebSearchActionKind::FindInPage, WebSearchAction::Other => WebSearchActionKind::Other, } } fn web_search_query_count(query: &str, action: Option<&WebSearchAction>) -> Option { match action { Some(WebSearchAction::Search { query, queries }) => queries .as_ref() .map(|queries| usize_to_u64(queries.len())) .or_else(|| query.as_ref().map(|_| 1)), Some(WebSearchAction::OpenPage { .. }) | Some(WebSearchAction::FindInPage { .. }) | Some(WebSearchAction::Other) => None, None => (!query.trim().is_empty()).then_some(1), } } fn serialize_enum_as_string(value: &T) -> Option { serde_json::to_value(value) .ok() .and_then(|value| value.as_str().map(str::to_string)) } fn usize_to_u64(value: usize) -> u64 { u64::try_from(value).unwrap_or(u64::MAX) } fn option_i64_to_u64(value: Option) -> Option { value.and_then(|value| u64::try_from(value).ok()) } fn codex_turn_event_params( app_server_client: CodexAppServerClientMetadata, runtime: CodexRuntimeMetadata, turn_id: String, turn_state: &TurnState, thread_metadata: &ThreadMetadataState, ) -> CodexTurnEventParams { let (Some(thread_id), Some(num_input_images), Some(resolved_config), Some(completed)) = ( turn_state.thread_id.clone(), turn_state.num_input_images, turn_state.resolved_config.clone(), turn_state.completed.clone(), ) else { unreachable!("turn event params require a fully populated turn state"); }; let started_at = turn_state.started_at; let TurnResolvedConfigFact { turn_id: _resolved_turn_id, thread_id: _resolved_thread_id, num_input_images: _resolved_num_input_images, submission_type, ephemeral, session_source: _session_source, 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; let token_usage = turn_state.token_usage.clone(); CodexTurnEventParams { thread_id, turn_id, app_server_client, runtime, submission_type, ephemeral, thread_source: thread_metadata.thread_source.map(str::to_string), initialization_mode: thread_metadata.initialization_mode, subagent_source: thread_metadata.subagent_source.clone(), parent_thread_id: thread_metadata.parent_thread_id.clone(), 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: 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: Some(turn_state.steer_count), 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: token_usage .as_ref() .map(|token_usage| token_usage.input_tokens), cached_input_tokens: token_usage .as_ref() .map(|token_usage| token_usage.cached_input_tokens), output_tokens: token_usage .as_ref() .map(|token_usage| token_usage.output_tokens), reasoning_output_tokens: token_usage .as_ref() .map(|token_usage| token_usage.reasoning_output_tokens), total_tokens: token_usage .as_ref() .map(|token_usage| token_usage.total_tokens), duration_ms: completed.duration_ms, started_at, completed_at: Some(completed.completed_at), } } fn sandbox_policy_mode(sandbox_policy: &SandboxPolicy) -> &'static str { match sandbox_policy { SandboxPolicy::DangerFullAccess => "full_access", SandboxPolicy::ReadOnly { .. } => "read_only", SandboxPolicy::WorkspaceWrite { .. } => "workspace_write", SandboxPolicy::ExternalSandbox { .. } => "external_sandbox", } } fn collaboration_mode_mode(mode: ModeKind) -> &'static str { match mode { ModeKind::Plan => "plan", ModeKind::Default | ModeKind::PairProgramming | ModeKind::Execute => "default", } } fn reasoning_summary_mode(summary: Option) -> Option { match summary { Some(ReasoningSummary::None) | None => None, Some(summary) => Some(summary.to_string()), } } fn personality_mode(personality: Option) -> Option { match personality { Some(Personality::None) | None => None, Some(personality) => Some(personality.to_string()), } } 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 num_input_images(input: &[UserInput]) -> usize { input .iter() .filter(|item| matches!(item, UserInput::Image { .. } | UserInput::LocalImage { .. })) .count() } fn rejection_reason_from_error_type( error_type: Option, ) -> Option { match error_type? { AnalyticsJsonRpcError::TurnSteer(error) => Some(error.into()), AnalyticsJsonRpcError::Input(error) => Some(error.into()), } } pub(crate) fn skill_id_for_local_skill( repo_url: Option<&str>, repo_root: Option<&Path>, skill_path: &Path, skill_name: &str, ) -> String { let path = normalize_path_for_skill_id(repo_url, repo_root, skill_path); let prefix = if let Some(url) = repo_url { format!("repo_{url}") } else { "personal".to_string() }; let raw_id = format!("{prefix}_{path}_{skill_name}"); let mut hasher = sha1::Sha1::new(); sha1::Digest::update(&mut hasher, raw_id.as_bytes()); format!("{:x}", sha1::Digest::finalize(hasher)) } /// Returns a normalized path for skill ID construction. /// /// - Repo-scoped skills use a path relative to the repo root. /// - User/admin/system skills use an absolute path. pub(crate) fn normalize_path_for_skill_id( repo_url: Option<&str>, repo_root: Option<&Path>, skill_path: &Path, ) -> String { let resolved_path = std::fs::canonicalize(skill_path).unwrap_or_else(|_| skill_path.to_path_buf()); match (repo_url, repo_root) { (Some(_), Some(root)) => { let resolved_root = std::fs::canonicalize(root).unwrap_or_else(|_| root.to_path_buf()); resolved_path .strip_prefix(&resolved_root) .unwrap_or(resolved_path.as_path()) .to_string_lossy() .replace('\\', "/") } _ => resolved_path.to_string_lossy().replace('\\', "/"), } }