diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index ea5fe80ee7..5050d2e4d5 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -68,8 +68,12 @@ use codex_app_server_protocol::ClientInfo; use codex_app_server_protocol::ClientRequest; use codex_app_server_protocol::ClientResponsePayload; use codex_app_server_protocol::CodexErrorInfo; +use codex_app_server_protocol::CommandExecutionSource as AppServerCommandExecutionSource; +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; @@ -79,6 +83,7 @@ use codex_app_server_protocol::SessionSource as AppServerSessionSource; use codex_app_server_protocol::Thread; use codex_app_server_protocol::ThreadArchiveParams; use codex_app_server_protocol::ThreadArchiveResponse; +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; @@ -588,6 +593,72 @@ async fn ingest_turn_prerequisites( } } +async fn ingest_tool_review_prerequisites( + reducer: &mut AnalyticsReducer, + events: &mut Vec, +) { + reducer + .ingest(sample_initialize_fact(/*connection_id*/ 7), events) + .await; + reducer + .ingest( + AnalyticsFact::ClientResponse { + connection_id: 7, + request_id: RequestId::Integer(1), + response: Box::new(sample_thread_start_response( + "thread-1", /*ephemeral*/ false, "gpt-5", + )), + }, + events, + ) + .await; + events.clear(); +} + +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: AppServerCommandExecutionSource::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()) @@ -1387,6 +1458,87 @@ async fn guardian_review_event_ingests_custom_fact_with_optional_target_item() { assert_eq!(payload[0]["event_params"]["review_timeout_ms"], 90_000); } +#[tokio::test] +async fn item_lifecycle_notifications_publish_command_execution_event() { + let mut reducer = AnalyticsReducer::default(); + let mut events = Vec::new(); + + ingest_tool_review_prerequisites(&mut reducer, &mut events).await; + reducer + .ingest( + AnalyticsFact::Notification(Box::new(ServerNotification::ItemStarted( + ItemStartedNotification { + thread_id: "thread-1".to_string(), + turn_id: "turn-1".to_string(), + started_at_ms: 1_000, + 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(Box::new(ServerNotification::ItemCompleted( + ItemCompletedNotification { + thread_id: "thread-1".to_string(), + turn_id: "turn-1".to_string(), + completed_at_ms: 1_042, + 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"]["final_approval_outcome"], + "unknown" + ); + 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"]["started_at_ms"], 1_000); + assert_eq!(payload[0]["event_params"]["completed_at_ms"], 1_042); + assert_eq!(payload[0]["event_params"]["duration_ms"], 42); + assert_eq!( + payload[0]["event_params"]["app_server_client"]["client_name"], + "codex-tui" + ); + assert_eq!(payload[0]["event_params"]["thread_source"], "user"); +} + #[test] fn subagent_thread_started_review_serializes_expected_shape() { let event = TrackEventRequest::ThreadInitialized(subagent_thread_started_event_request( @@ -1670,6 +1822,79 @@ async fn subagent_thread_started_inherits_parent_connection_for_new_thread() { ); } +#[tokio::test] +async fn subagent_tool_items_inherit_parent_connection_metadata() { + let mut reducer = AnalyticsReducer::default(); + let mut events = Vec::new(); + + ingest_tool_review_prerequisites(&mut reducer, &mut events).await; + reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::SubAgentThreadStarted( + SubAgentThreadStartedInput { + thread_id: "thread-subagent".to_string(), + parent_thread_id: Some("thread-1".to_string()), + product_client_id: "codex-tui".to_string(), + client_name: "codex-tui".to_string(), + client_version: "1.0.0".to_string(), + model: "gpt-5".to_string(), + ephemeral: false, + subagent_source: SubAgentSource::Review, + created_at: 128, + }, + )), + &mut events, + ) + .await; + events.clear(); + + reducer + .ingest( + AnalyticsFact::Notification(Box::new(ServerNotification::ItemStarted( + ItemStartedNotification { + thread_id: "thread-subagent".to_string(), + turn_id: "turn-subagent".to_string(), + started_at_ms: 1_000, + item: sample_command_execution_item( + CommandExecutionStatus::InProgress, + /*exit_code*/ None, + /*duration_ms*/ None, + ), + }, + ))), + &mut events, + ) + .await; + reducer + .ingest( + AnalyticsFact::Notification(Box::new(ServerNotification::ItemCompleted( + ItemCompletedNotification { + thread_id: "thread-subagent".to_string(), + turn_id: "turn-subagent".to_string(), + completed_at_ms: 1_042, + 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_source"], "subagent"); + assert_eq!(payload[0]["event_params"]["subagent_source"], "review"); + assert_eq!(payload[0]["event_params"]["parent_thread_id"], "thread-1"); + assert_eq!( + payload[0]["event_params"]["app_server_client"]["client_name"], + "codex-tui" + ); +} + #[test] fn plugin_used_event_serializes_expected_shape() { let tracking = TrackEventsContext { diff --git a/codex-rs/analytics/src/client.rs b/codex-rs/analytics/src/client.rs index d54c53ede9..6d6d446560 100644 --- a/codex-rs/analytics/src/client.rs +++ b/codex-rs/analytics/src/client.rs @@ -333,10 +333,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, @@ -349,6 +345,21 @@ impl AnalyticsEventsClient { response: Box::new(response), }); } + + pub fn track_notification(&self, notification: ServerNotification) { + if !matches!( + notification, + ServerNotification::TurnStarted(_) + | ServerNotification::TurnCompleted(_) + | ServerNotification::ItemStarted(_) + | ServerNotification::ItemCompleted(_) + | ServerNotification::ItemGuardianApprovalReviewStarted(_) + | ServerNotification::ItemGuardianApprovalReviewCompleted(_) + ) { + return; + } + self.record_fact(AnalyticsFact::Notification(Box::new(notification))); + } } async fn send_track_events( diff --git a/codex-rs/analytics/src/lib.rs b/codex-rs/analytics/src/lib.rs index ed0f1036ca..2fb23199cb 100644 --- a/codex-rs/analytics/src/lib.rs +++ b/codex-rs/analytics/src/lib.rs @@ -51,3 +51,27 @@ pub fn now_unix_seconds() -> u64 { .unwrap_or_default() .as_secs() } + +pub fn now_unix_millis() -> u64 { + u64::try_from( + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis(), + ) + .unwrap_or(u64::MAX) +} + +pub(crate) fn serialize_enum_as_string(value: &T) -> Option { + serde_json::to_value(value) + .ok() + .and_then(|value| value.as_str().map(str::to_string)) +} + +pub(crate) fn usize_to_u64(value: usize) -> u64 { + u64::try_from(value).unwrap_or(u64::MAX) +} + +pub(crate) fn option_i64_to_u64(value: Option) -> Option { + value.and_then(|value| u64::try_from(value).ok()) +} diff --git a/codex-rs/analytics/src/reducer.rs b/codex-rs/analytics/src/reducer.rs index b1dc822d43..6edaa2d0fe 100644 --- a/codex-rs/analytics/src/reducer.rs +++ b/codex-rs/analytics/src/reducer.rs @@ -2,15 +2,32 @@ 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::CommandExecutionFamily; +use crate::events::CommandExecutionSource; use crate::events::GuardianReviewEventParams; use crate::events::GuardianReviewEventPayload; use crate::events::GuardianReviewEventRequest; @@ -18,7 +35,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; @@ -47,14 +68,29 @@ use crate::facts::TurnSteerRejectionReason; use crate::facts::TurnSteerResult; use crate::facts::TurnTokenUsageFact; use crate::now_unix_seconds; +use crate::option_i64_to_u64; +use crate::serialize_enum_as_string; +use crate::usize_to_u64; 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 as AppServerCommandExecutionSource; +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; @@ -75,6 +111,7 @@ pub(crate) struct AnalyticsReducer { turns: HashMap, connections: HashMap, threads: HashMap, + tool_items: HashMap<(String, String, String), ToolItemLifecycleState>, } struct ConnectionState { @@ -118,6 +155,19 @@ impl<'a> AnalyticsDropSite<'a> { } } + fn tool_item( + notification: &'a codex_app_server_protocol::ItemCompletedNotification, + item_id: &'a str, + ) -> Self { + Self { + event_name: "tool item", + thread_id: ¬ification.thread_id, + turn_id: Some(¬ification.turn_id), + review_id: None, + item_id: Some(item_id), + } + } + fn turn_steer(thread_id: &'a str) -> Self { Self { event_name: "turn steer", @@ -216,6 +266,10 @@ struct TurnState { steer_count: usize, } +struct ToolItemLifecycleState { + started_at_ms: u64, +} + impl AnalyticsReducer { pub(crate) async fn ingest(&mut self, input: AnalyticsFact, out: &mut Vec) { match input { @@ -686,6 +740,61 @@ impl AnalyticsReducer { out: &mut Vec, ) { match notification { + ServerNotification::ItemStarted(notification) => { + let Some(item_id) = tool_item_id(¬ification.item) else { + return; + }; + let Some(started_at_ms) = option_i64_to_u64(Some(notification.started_at_ms)) else { + return; + }; + self.tool_items.insert( + ( + notification.thread_id, + notification.turn_id, + item_id.to_string(), + ), + ToolItemLifecycleState { started_at_ms }, + ); + } + ServerNotification::ItemCompleted(notification) => { + let Some(item_id) = tool_item_id(¬ification.item) else { + return; + }; + let key = ( + notification.thread_id.clone(), + notification.turn_id.clone(), + item_id.to_string(), + ); + let Some(tool_item_state) = self.tool_items.remove(&key) else { + tracing::warn!( + thread_id = %notification.thread_id, + turn_id = %notification.turn_id, + item_id, + "dropping tool item analytics event: missing item started notification" + ); + return; + }; + let Some(completed_at_ms) = option_i64_to_u64(Some(notification.completed_at_ms)) + else { + return; + }; + let Some((connection_state, thread_metadata)) = self + .thread_context_or_warn(AnalyticsDropSite::tool_item(¬ification, item_id)) + else { + return; + }; + if let Some(event) = tool_item_event( + ¬ification.thread_id, + ¬ification.turn_id, + ¬ification.item, + tool_item_state.started_at_ms, + completed_at_ms, + connection_state, + thread_metadata, + ) { + out.push(event); + } + } ServerNotification::TurnStarted(notification) => { let turn_state = self.turns.entry(notification.turn.id).or_insert(TurnState { connection_id: None, @@ -772,8 +881,8 @@ impl AnalyticsReducer { 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, + subagent_source: thread_metadata.subagent_source.clone(), + parent_thread_id: thread_metadata.parent_thread_id.clone(), created_at: u64::try_from(thread.created_at).unwrap_or_default(), }, }, @@ -976,6 +1085,587 @@ fn warn_missing_analytics_context( ); } +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_ms: u64, + completed_at_ms: u64, + connection_state: &ConnectionState, + thread_metadata: &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: item_or_observed_duration_ms( + option_i64_to_u64(*duration_ms), + started_at_ms, + completed_at_ms, + ), + }, + ToolItemContext { + started_at_ms, + completed_at_ms, + connection_state, + thread_metadata, + }, + ); + Some(TrackEventRequest::CommandExecution( + CodexCommandExecutionEventRequest { + event_type: "codex_command_execution_event", + event_params: CodexCommandExecutionEventParams { + base, + command_execution_source: (*source).into(), + 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_duration_ms(started_at_ms, completed_at_ms), + }, + ToolItemContext { + started_at_ms, + completed_at_ms, + 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: item_or_observed_duration_ms( + option_i64_to_u64(*duration_ms), + started_at_ms, + completed_at_ms, + ), + }, + ToolItemContext { + started_at_ms, + completed_at_ms, + 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(), + }, + }, + )) + } + 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: item_or_observed_duration_ms( + option_i64_to_u64(*duration_ms), + started_at_ms, + completed_at_ms, + ), + }, + ToolItemContext { + started_at_ms, + completed_at_ms, + 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_duration_ms(started_at_ms, completed_at_ms), + }, + ToolItemContext { + started_at_ms, + completed_at_ms, + connection_state, + thread_metadata, + }, + ); + Some(TrackEventRequest::CollabAgentToolCall( + CodexCollabAgentToolCallEventRequest { + event_type: "codex_collab_agent_tool_call_event", + event_params: CodexCollabAgentToolCallEventParams { + base, + 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_duration_ms(started_at_ms, completed_at_ms), + }, + ToolItemContext { + started_at_ms, + completed_at_ms, + 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_duration_ms(started_at_ms, completed_at_ms), + }, + ToolItemContext { + started_at_ms, + completed_at_ms, + 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_ms: u64, + completed_at_ms: u64, + connection_state: &'a ConnectionState, + thread_metadata: &'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; + 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.clone(), + parent_thread_id: thread_metadata.parent_thread_id.clone(), + tool_name, + started_at_ms: context.started_at_ms, + completed_at_ms: context.completed_at_ms, + duration_ms: outcome.duration_ms, + review_count: 0, + guardian_review_count: 0, + user_review_count: 0, + final_approval_outcome: ToolItemFinalApprovalOutcome::Unknown, + terminal_status: outcome.terminal_status, + failure_kind: outcome.failure_kind, + requested_additional_permissions: false, + requested_network_access: false, + retry_count: 0, + } +} + +fn observed_duration_ms(started_at_ms: u64, completed_at_ms: u64) -> Option { + completed_at_ms.checked_sub(started_at_ms) +} + +fn item_or_observed_duration_ms( + item_duration_ms: Option, + started_at_ms: u64, + completed_at_ms: u64, +) -> Option { + item_duration_ms.or_else(|| observed_duration_ms(started_at_ms, completed_at_ms)) +} + +impl From for CommandExecutionSource { + fn from(source: AppServerCommandExecutionSource) -> Self { + match source { + AppServerCommandExecutionSource::Agent => Self::Agent, + AppServerCommandExecutionSource::UserShell => Self::UserShell, + AppServerCommandExecutionSource::UnifiedExecStartup => Self::UnifiedExecStartup, + AppServerCommandExecutionSource::UnifiedExecInteraction => Self::UnifiedExecInteraction, + } + } +} + +fn command_execution_family(source: AppServerCommandExecutionSource) -> CommandExecutionFamily { + match source { + AppServerCommandExecutionSource::Agent => CommandExecutionFamily::Shell, + AppServerCommandExecutionSource::UserShell => CommandExecutionFamily::UserShell, + AppServerCommandExecutionSource::UnifiedExecStartup + | AppServerCommandExecutionSource::UnifiedExecInteraction => { + CommandExecutionFamily::UnifiedExec + } + } +} + +fn command_execution_tool_name(source: AppServerCommandExecutionSource) -> &'static str { + match source { + AppServerCommandExecutionSource::UnifiedExecStartup + | AppServerCommandExecutionSource::UnifiedExecInteraction => "unified_exec", + AppServerCommandExecutionSource::UserShell => "user_shell", + AppServerCommandExecutionSource::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", + } +} + +#[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 codex_turn_event_params( app_server_client: CodexAppServerClientMetadata, runtime: CodexRuntimeMetadata, diff --git a/codex-rs/app-server/src/bespoke_event_handling.rs b/codex-rs/app-server/src/bespoke_event_handling.rs index 89b5d26579..5499281d8e 100644 --- a/codex-rs/app-server/src/bespoke_event_handling.rs +++ b/codex-rs/app-server/src/bespoke_event_handling.rs @@ -12,7 +12,6 @@ use crate::thread_state::TurnSummary; use crate::thread_state::resolve_server_request_on_thread_listener; use crate::thread_status::ThreadWatchActiveGuard; use crate::thread_status::ThreadWatchManager; -use codex_analytics::AnalyticsEventsClient; use codex_app_server_protocol::AccountRateLimitsUpdatedNotification; use codex_app_server_protocol::AdditionalPermissionProfile as V2AdditionalPermissionProfile; use codex_app_server_protocol::CodexErrorInfo as V2CodexErrorInfo; @@ -141,7 +140,6 @@ pub(crate) async fn apply_bespoke_event_handling( conversation_id: ThreadId, conversation: Arc, thread_manager: Arc, - analytics_events_client: Option, outgoing: ThreadScopedOutgoingMessageSender, thread_state: Arc>, thread_watch_manager: ThreadWatchManager, @@ -176,10 +174,6 @@ pub(crate) async fn apply_bespoke_event_handling( thread_id: conversation_id.to_string(), turn, }; - if let Some(analytics_events_client) = analytics_events_client.as_ref() { - analytics_events_client - .track_notification(ServerNotification::TurnStarted(notification.clone())); - } outgoing .send_server_notification(ServerNotification::TurnStarted(notification)) .await; @@ -196,7 +190,6 @@ pub(crate) async fn apply_bespoke_event_handling( conversation_id, event_turn_id, turn_complete_event, - analytics_events_client.as_ref(), &outgoing, &thread_state, ) @@ -238,10 +231,6 @@ pub(crate) async fn apply_bespoke_event_handling( thread_id: Some(conversation_id.to_string()), message: warning_event.message, }; - if let Some(analytics_events_client) = analytics_events_client.as_ref() { - analytics_events_client - .track_notification(ServerNotification::Warning(notification.clone())); - } outgoing .send_server_notification(ServerNotification::Warning(notification)) .await; @@ -251,10 +240,6 @@ pub(crate) async fn apply_bespoke_event_handling( thread_id: conversation_id.to_string(), message: warning_event.message, }; - if let Some(analytics_events_client) = analytics_events_client.as_ref() { - analytics_events_client - .track_notification(ServerNotification::GuardianWarning(notification.clone())); - } outgoing .send_server_notification(ServerNotification::GuardianWarning(notification)) .await; @@ -1139,7 +1124,6 @@ pub(crate) async fn apply_bespoke_event_handling( conversation_id, event_turn_id, turn_aborted_event, - analytics_events_client.as_ref(), &outgoing, &thread_state, ) @@ -1328,7 +1312,6 @@ async fn emit_turn_completed_with_status( conversation_id: ThreadId, event_turn_id: String, turn_completion_metadata: TurnCompletionMetadata, - analytics_events_client: Option<&AnalyticsEventsClient>, outgoing: &ThreadScopedOutgoingMessageSender, ) { let notification = TurnCompletedNotification { @@ -1343,10 +1326,6 @@ async fn emit_turn_completed_with_status( duration_ms: turn_completion_metadata.duration_ms, }, }; - if let Some(analytics_events_client) = analytics_events_client { - analytics_events_client - .track_notification(ServerNotification::TurnCompleted(notification.clone())); - } outgoing .send_server_notification(ServerNotification::TurnCompleted(notification)) .await; @@ -1510,7 +1489,6 @@ async fn handle_turn_complete( conversation_id: ThreadId, event_turn_id: String, turn_complete_event: TurnCompleteEvent, - analytics_events_client: Option<&AnalyticsEventsClient>, outgoing: &ThreadScopedOutgoingMessageSender, thread_state: &Arc>, ) { @@ -1531,7 +1509,6 @@ async fn handle_turn_complete( completed_at: turn_complete_event.completed_at, duration_ms: turn_complete_event.duration_ms, }, - analytics_events_client, outgoing, ) .await; @@ -1541,7 +1518,6 @@ async fn handle_turn_interrupted( conversation_id: ThreadId, event_turn_id: String, turn_aborted_event: TurnAbortedEvent, - analytics_events_client: Option<&AnalyticsEventsClient>, outgoing: &ThreadScopedOutgoingMessageSender, thread_state: &Arc>, ) { @@ -1557,7 +1533,6 @@ async fn handle_turn_interrupted( completed_at: turn_aborted_event.completed_at, duration_ms: turn_aborted_event.duration_ms, }, - analytics_events_client, outgoing, ) .await; @@ -2109,7 +2084,6 @@ mod tests { use codex_app_server_protocol::GuardianApprovalReviewStatus; use codex_app_server_protocol::JSONRPCErrorError; use codex_app_server_protocol::TurnPlanStepStatus; - use codex_login::AuthManager; use codex_login::CodexAuth; use codex_protocol::items::HookPromptFragment; use codex_protocol::items::build_hook_prompt_message; @@ -2240,7 +2214,6 @@ mod tests { outgoing: ThreadScopedOutgoingMessageSender, thread_state: Arc>, thread_watch_manager: ThreadWatchManager, - analytics_events_client: AnalyticsEventsClient, codex_home: PathBuf, } @@ -2255,7 +2228,6 @@ mod tests { self.conversation_id, self.conversation.clone(), self.thread_manager.clone(), - Some(self.analytics_events_client.clone()), self.outgoing.clone(), self.thread_state.clone(), self.thread_watch_manager.clone(), @@ -2589,13 +2561,6 @@ mod tests { outgoing: outgoing.clone(), thread_state: thread_state.clone(), thread_watch_manager: thread_watch_manager.clone(), - analytics_events_client: AnalyticsEventsClient::new( - AuthManager::from_auth_for_testing( - CodexAuth::create_dummy_chatgpt_auth_for_testing(), - ), - "http://localhost".to_string(), - Some(false), - ), codex_home: codex_home.path().to_path_buf(), }; @@ -3175,7 +3140,6 @@ mod tests { conversation_id, event_turn_id.clone(), turn_complete_event(&event_turn_id), - /*analytics_events_client*/ None, &outgoing, &thread_state, ) @@ -3227,7 +3191,6 @@ mod tests { conversation_id, event_turn_id.clone(), turn_aborted_event(&event_turn_id), - /*analytics_events_client*/ None, &outgoing, &thread_state, ) @@ -3278,7 +3241,6 @@ mod tests { conversation_id, event_turn_id.clone(), turn_complete_event(&event_turn_id), - /*analytics_events_client*/ None, &outgoing, &thread_state, ) @@ -3513,7 +3475,6 @@ mod tests { conversation_a, a_turn1.clone(), turn_complete_event(&a_turn1), - /*analytics_events_client*/ None, &outgoing, &thread_state, ) @@ -3535,7 +3496,6 @@ mod tests { conversation_b, b_turn1.clone(), turn_complete_event(&b_turn1), - /*analytics_events_client*/ None, &outgoing, &thread_state, ) @@ -3547,7 +3507,6 @@ mod tests { conversation_a, a_turn2.clone(), turn_complete_event(&a_turn2), - /*analytics_events_client*/ None, &outgoing, &thread_state, ) diff --git a/codex-rs/app-server/src/outgoing_message.rs b/codex-rs/app-server/src/outgoing_message.rs index f7a90538c2..2b12fb9b1f 100644 --- a/codex-rs/app-server/src/outgoing_message.rs +++ b/codex-rs/app-server/src/outgoing_message.rs @@ -141,6 +141,9 @@ impl ThreadScopedOutgoingMessageSender { } pub(crate) async fn send_server_notification(&self, notification: ServerNotification) { + self.outgoing + .analytics_events_client + .track_notification(notification.clone()); if self.connection_ids.is_empty() { return; } @@ -524,7 +527,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 @@ -558,7 +561,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 diff --git a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs index 9dabdea44f..022118cb80 100644 --- a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs +++ b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs @@ -249,7 +249,7 @@ pub(super) async fn ensure_listener_task_running( thread_manager, thread_state_manager, pending_thread_unloads, - analytics_events_client, + analytics_events_client: _, thread_watch_manager, thread_list_state_permit, fallback_model_provider, @@ -325,7 +325,6 @@ pub(super) async fn ensure_listener_task_running( conversation_id, conversation.clone(), thread_manager.clone(), - Some(analytics_events_client.clone()), thread_outgoing, thread_state.clone(), thread_watch_manager.clone(),