From ff079effc857d9a02154c75d610be68c7db7ac0b Mon Sep 17 00:00:00 2001 From: rhan-oai Date: Mon, 20 Apr 2026 13:07:01 -0700 Subject: [PATCH] [codex-analytics] emit tool item events from item lifecycle --- .../analytics/src/analytics_client_tests.rs | 172 ++++- codex-rs/analytics/src/client.rs | 22 +- codex-rs/analytics/src/facts.rs | 5 +- codex-rs/analytics/src/reducer.rs | 709 +++++++++++++++++- codex-rs/app-server/src/outgoing_message.rs | 10 +- 5 files changed, 890 insertions(+), 28 deletions(-) diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index 679e4a6899..4ec8a2c50a 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -62,8 +62,12 @@ use codex_app_server_protocol::ClientInfo; use codex_app_server_protocol::ClientRequest; use codex_app_server_protocol::ClientResponse; use codex_app_server_protocol::CodexErrorInfo; +use codex_app_server_protocol::CommandExecutionSource; +use codex_app_server_protocol::CommandExecutionStatus; use codex_app_server_protocol::InitializeCapabilities; use codex_app_server_protocol::InitializeParams; +use codex_app_server_protocol::ItemCompletedNotification; +use codex_app_server_protocol::ItemStartedNotification; use codex_app_server_protocol::JSONRPCErrorError; use codex_app_server_protocol::NonSteerableTurnKind; use codex_app_server_protocol::RequestId; @@ -71,6 +75,7 @@ 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::ThreadItem; use codex_app_server_protocol::ThreadResumeResponse; use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::ThreadStatus as AppServerThreadStatus; @@ -297,6 +302,13 @@ fn sample_turn_completed_notification( }) } +fn notification_fact(notification: ServerNotification) -> AnalyticsFact { + AnalyticsFact::Notification { + connection_id: 7, + notification: Box::new(notification), + } +} + fn sample_turn_resolved_config(turn_id: &str) -> TurnResolvedConfigFact { TurnResolvedConfigFact { turn_id: turn_id.to_string(), @@ -521,9 +533,7 @@ async fn ingest_turn_prerequisites( if include_started { reducer .ingest( - AnalyticsFact::Notification(Box::new(sample_turn_started_notification( - "thread-2", "turn-2", - ))), + notification_fact(sample_turn_started_notification("thread-2", "turn-2")), out, ) .await; @@ -541,6 +551,50 @@ async fn ingest_turn_prerequisites( } } +fn sample_initialize_fact(connection_id: u64) -> AnalyticsFact { + AnalyticsFact::Initialize { + connection_id, + params: InitializeParams { + client_info: ClientInfo { + name: "codex-tui".to_string(), + title: None, + version: "1.0.0".to_string(), + }, + capabilities: Some(InitializeCapabilities { + experimental_api: false, + opt_out_notification_methods: None, + }), + }, + product_client_id: DEFAULT_ORIGINATOR.to_string(), + runtime: CodexRuntimeMetadata { + codex_rs_version: "0.99.0".to_string(), + runtime_os: "linux".to_string(), + runtime_os_version: "24.04".to_string(), + runtime_arch: "x86_64".to_string(), + }, + rpc_transport: AppServerRpcTransport::Websocket, + } +} + +fn sample_command_execution_item( + status: CommandExecutionStatus, + exit_code: Option, + duration_ms: Option, +) -> ThreadItem { + ThreadItem::CommandExecution { + id: "item-1".to_string(), + command: "echo hi".to_string(), + cwd: test_path_buf("/tmp").abs(), + process_id: Some("pid-1".to_string()), + source: CommandExecutionSource::Agent, + status, + command_actions: Vec::new(), + aggregated_output: None, + exit_code, + duration_ms, + } +} + fn expected_absolute_path(path: &PathBuf) -> String { std::fs::canonicalize(path) .unwrap_or_else(|_| path.to_path_buf()) @@ -1203,6 +1257,86 @@ async fn compaction_event_ingests_custom_fact() { assert_eq!(payload[0]["event_params"]["status"], "failed"); } +#[tokio::test] +async fn item_lifecycle_notifications_publish_command_execution_event() { + let mut reducer = AnalyticsReducer::default(); + let mut events = Vec::new(); + + reducer + .ingest(sample_initialize_fact(/*connection_id*/ 7), &mut events) + .await; + reducer + .ingest( + AnalyticsFact::Notification { + connection_id: 7, + notification: Box::new(ServerNotification::ItemStarted(ItemStartedNotification { + thread_id: "thread-1".to_string(), + turn_id: "turn-1".to_string(), + item: sample_command_execution_item( + CommandExecutionStatus::InProgress, + /*exit_code*/ None, + /*duration_ms*/ None, + ), + })), + }, + &mut events, + ) + .await; + assert!( + events.is_empty(), + "tool item event should emit on completion" + ); + + reducer + .ingest( + AnalyticsFact::Notification { + connection_id: 7, + notification: Box::new(ServerNotification::ItemCompleted( + ItemCompletedNotification { + thread_id: "thread-1".to_string(), + turn_id: "turn-1".to_string(), + item: sample_command_execution_item( + CommandExecutionStatus::Completed, + Some(0), + Some(42), + ), + }, + )), + }, + &mut events, + ) + .await; + + let payload = serde_json::to_value(&events).expect("serialize events"); + assert_eq!(payload.as_array().expect("events array").len(), 1); + assert_eq!(payload[0]["event_type"], "codex_command_execution_event"); + assert_eq!(payload[0]["event_params"]["thread_id"], "thread-1"); + assert_eq!(payload[0]["event_params"]["turn_id"], "turn-1"); + assert_eq!(payload[0]["event_params"]["item_id"], "item-1"); + assert_eq!(payload[0]["event_params"]["tool_name"], "shell"); + assert_eq!( + payload[0]["event_params"]["command_execution_source"], + "agent" + ); + assert_eq!( + payload[0]["event_params"]["command_execution_family"], + "shell" + ); + assert_eq!(payload[0]["event_params"]["terminal_status"], "completed"); + assert_eq!( + payload[0]["event_params"]["failure_kind"], + serde_json::Value::Null + ); + assert_eq!(payload[0]["event_params"]["exit_code"], 0); + assert_eq!(payload[0]["event_params"]["duration_ms"], 42); + assert_eq!(payload[0]["event_params"]["execution_started"], true); + assert_eq!( + payload[0]["event_params"]["app_server_client"]["client_name"], + "codex-tui" + ); + assert_eq!(payload[0]["event_params"]["thread_source"], json!(null)); +} + #[test] fn subagent_thread_started_review_serializes_expected_shape() { let event = TrackEventRequest::ThreadInitialized(subagent_thread_started_event_request( @@ -2070,12 +2204,12 @@ async fn turn_start_error_response_discards_pending_start_request() { .await; reducer .ingest( - AnalyticsFact::Notification(Box::new(sample_turn_completed_notification( + notification_fact(sample_turn_completed_notification( "thread-2", "turn-2", AppServerTurnStatus::Completed, /*codex_error_info*/ None, - ))), + )), &mut out, ) .await; @@ -2099,12 +2233,12 @@ async fn turn_lifecycle_emits_turn_event() { .await; reducer .ingest( - AnalyticsFact::Notification(Box::new(sample_turn_completed_notification( + notification_fact(sample_turn_completed_notification( "thread-2", "turn-2", AppServerTurnStatus::Completed, /*codex_error_info*/ None, - ))), + )), &mut out, ) .await; @@ -2236,12 +2370,12 @@ async fn accepted_steers_increment_turn_steer_count() { reducer .ingest( - AnalyticsFact::Notification(Box::new(sample_turn_completed_notification( + notification_fact(sample_turn_completed_notification( "thread-2", "turn-2", AppServerTurnStatus::Completed, /*codex_error_info*/ None, - ))), + )), &mut out, ) .await; @@ -2270,12 +2404,12 @@ async fn turn_does_not_emit_without_required_prerequisites() { .await; reducer .ingest( - AnalyticsFact::Notification(Box::new(sample_turn_completed_notification( + notification_fact(sample_turn_completed_notification( "thread-2", "turn-2", AppServerTurnStatus::Completed, /*codex_error_info*/ None, - ))), + )), &mut out, ) .await; @@ -2295,12 +2429,12 @@ async fn turn_does_not_emit_without_required_prerequisites() { .await; reducer .ingest( - AnalyticsFact::Notification(Box::new(sample_turn_completed_notification( + notification_fact(sample_turn_completed_notification( "thread-2", "turn-2", AppServerTurnStatus::Completed, /*codex_error_info*/ None, - ))), + )), &mut out, ) .await; @@ -2323,12 +2457,12 @@ async fn turn_lifecycle_emits_failed_turn_event() { .await; reducer .ingest( - AnalyticsFact::Notification(Box::new(sample_turn_completed_notification( + notification_fact(sample_turn_completed_notification( "thread-2", "turn-2", AppServerTurnStatus::Failed, Some(codex_app_server_protocol::CodexErrorInfo::BadRequest), - ))), + )), &mut out, ) .await; @@ -2355,12 +2489,12 @@ async fn turn_lifecycle_emits_interrupted_turn_event_without_error() { .await; reducer .ingest( - AnalyticsFact::Notification(Box::new(sample_turn_completed_notification( + notification_fact(sample_turn_completed_notification( "thread-2", "turn-2", AppServerTurnStatus::Interrupted, /*codex_error_info*/ None, - ))), + )), &mut out, ) .await; @@ -2387,12 +2521,12 @@ async fn turn_completed_without_started_notification_emits_null_started_at() { .await; reducer .ingest( - AnalyticsFact::Notification(Box::new(sample_turn_completed_notification( + notification_fact(sample_turn_completed_notification( "thread-2", "turn-2", AppServerTurnStatus::Completed, /*codex_error_info*/ None, - ))), + )), &mut out, ) .await; diff --git a/codex-rs/analytics/src/client.rs b/codex-rs/analytics/src/client.rs index e99e4cf572..64bf34e19e 100644 --- a/codex-rs/analytics/src/client.rs +++ b/codex-rs/analytics/src/client.rs @@ -301,10 +301,6 @@ impl AnalyticsEventsClient { }); } - pub fn track_notification(&self, notification: ServerNotification) { - self.record_fact(AnalyticsFact::Notification(Box::new(notification))); - } - pub fn track_server_request(&self, connection_id: u64, request: ServerRequest) { self.record_fact(AnalyticsFact::ServerRequest { connection_id, @@ -317,6 +313,24 @@ impl AnalyticsEventsClient { response: Box::new(response), }); } + + pub fn track_notification(&self, notification: ServerNotification) { + self.record_fact(AnalyticsFact::Notification { + connection_id: 0, + notification: Box::new(notification), + }); + } + + pub fn track_connection_notification( + &self, + connection_id: u64, + notification: ServerNotification, + ) { + self.record_fact(AnalyticsFact::Notification { + connection_id, + notification: Box::new(notification), + }); + } } async fn send_track_events( diff --git a/codex-rs/analytics/src/facts.rs b/codex-rs/analytics/src/facts.rs index 42f2d9023e..36846def16 100644 --- a/codex-rs/analytics/src/facts.rs +++ b/codex-rs/analytics/src/facts.rs @@ -295,7 +295,10 @@ pub(crate) enum AnalyticsFact { ServerResponse { response: Box, }, - Notification(Box), + Notification { + connection_id: u64, + 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), diff --git a/codex-rs/analytics/src/reducer.rs b/codex-rs/analytics/src/reducer.rs index 3fd9c5691b..8a3659dd3b 100644 --- a/codex-rs/analytics/src/reducer.rs +++ b/codex-rs/analytics/src/reducer.rs @@ -2,15 +2,33 @@ 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::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; @@ -18,7 +36,11 @@ 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::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; @@ -50,11 +72,23 @@ 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::CommandExecutionSource; +use codex_app_server_protocol::CommandExecutionStatus; +use codex_app_server_protocol::DynamicToolCallOutputContentItem; +use codex_app_server_protocol::DynamicToolCallStatus; use codex_app_server_protocol::InitializeParams; +use codex_app_server_protocol::McpToolCallStatus; +use codex_app_server_protocol::PatchApplyStatus; +use codex_app_server_protocol::PatchChangeKind; use codex_app_server_protocol::RequestId; use codex_app_server_protocol::ServerNotification; +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; @@ -68,6 +102,8 @@ 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 { @@ -76,6 +112,7 @@ pub(crate) struct AnalyticsReducer { connections: HashMap, thread_connections: HashMap, thread_metadata: HashMap, + tool_items: HashMap, } struct ConnectionState { @@ -83,6 +120,11 @@ struct ConnectionState { runtime: CodexRuntimeMetadata, } +struct ToolItemState { + connection_id: u64, + started_at: u64, +} + #[derive(Clone)] struct ThreadMetadataState { thread_source: Option<&'static str>, @@ -91,6 +133,17 @@ struct ThreadMetadataState { 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, @@ -192,8 +245,11 @@ impl AnalyticsReducer { } => { self.ingest_error_response(connection_id, request_id, error_type, out); } - AnalyticsFact::Notification(notification) => { - self.ingest_notification(*notification, out); + AnalyticsFact::Notification { + connection_id, + notification, + } => { + self.ingest_notification(connection_id, *notification, out); } AnalyticsFact::ServerRequest { connection_id: _connection_id, @@ -610,10 +666,44 @@ impl AnalyticsReducer { 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( + ¬ification.thread_id, + ¬ification.turn_id, + ¬ification.item, + started.started_at, + completed_at, + connection_state, + self.thread_metadata.get(¬ification.thread_id), + ) { + out.push(event); + } + } ServerNotification::TurnStarted(notification) => { let turn_state = self.turns.entry(notification.turn.id).or_insert(TurnState { connection_id: None, @@ -866,6 +956,621 @@ impl AnalyticsReducer { } } +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, + } +} + +fn tool_item_event( + thread_id: &str, + turn_id: &str, + item: &ThreadItem, + started_at: u64, + completed_at: u64, + connection_state: &ConnectionState, + thread_metadata: Option<&ThreadMetadataState>, +) -> Option { + 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, + }, + ); + 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, + }, + ); + 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, + }, + ); + 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, + }, + ); + 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, + }, + ); + 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, + }, + ); + 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, + }, + ); + 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>, +} + +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(); + 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: 0, + guardian_review_count: 0, + user_review_count: 0, + final_approval_outcome: ToolItemFinalApprovalOutcome::NotNeeded, + terminal_status: outcome.terminal_status, + failure_kind: outcome.failure_kind, + requested_additional_permissions: false, + requested_network_access: false, + retry_count: 0, + } +} + +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 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, diff --git a/codex-rs/app-server/src/outgoing_message.rs b/codex-rs/app-server/src/outgoing_message.rs index b80614ca1b..c79be4c4f0 100644 --- a/codex-rs/app-server/src/outgoing_message.rs +++ b/codex-rs/app-server/src/outgoing_message.rs @@ -532,7 +532,7 @@ impl OutgoingMessageSender { targeted_connections = connection_ids.len(), "app-server event: {notification}" ); - let outgoing_message = OutgoingMessage::AppServerNotification(notification); + let outgoing_message = OutgoingMessage::AppServerNotification(notification.clone()); if connection_ids.is_empty() { if let Err(err) = self .sender @@ -556,6 +556,9 @@ impl OutgoingMessageSender { .await { warn!("failed to send server notification to client: {err:?}"); + } else { + self.analytics_events_client + .track_connection_notification(connection_id.0, notification.clone()); } } } @@ -566,7 +569,7 @@ impl OutgoingMessageSender { notification: ServerNotification, ) { tracing::trace!("app-server event: {notification}"); - let outgoing_message = OutgoingMessage::AppServerNotification(notification); + let outgoing_message = OutgoingMessage::AppServerNotification(notification.clone()); let (write_complete_tx, write_complete_rx) = oneshot::channel(); if let Err(err) = self .sender @@ -578,6 +581,9 @@ impl OutgoingMessageSender { .await { warn!("failed to send server notification to client: {err:?}"); + } else { + self.analytics_events_client + .track_connection_notification(connection_id.0, notification); } let _ = write_complete_rx.await; }