From 155f1ca9e512c867792518e657a2cd79c05698f2 Mon Sep 17 00:00:00 2001 From: jif Date: Mon, 3 Aug 2026 10:12:44 +0000 Subject: [PATCH] Correlate code mode tool analytics with model responses (#36729) ## What changed - Emit dynamic tool-call analytics for code mode `exec` and `wait` calls, including duration and terminal status. - Associate code-mode cells and their nested tool calls with the originating and subsequent model response IDs. - Buffer correlated tool events until the next response is known, while flushing pending events when turns, threads, or the analytics queue close. ## Testing - Add reducer coverage for correlating `exec`, `wait`, and nested tool events across turns. - Add an app-server test that verifies production analytics include the cell and response IDs for a code-mode `exec` call. GitOrigin-RevId: 8b34ab59c1ad68ebbd57f9c8a05f19ab04df48a4 --- .../analytics/src/analytics_client_tests.rs | 114 ++++++ codex-rs/analytics/src/client.rs | 12 +- codex-rs/analytics/src/client_tests.rs | 4 + codex-rs/analytics/src/events.rs | 4 + codex-rs/analytics/src/facts.rs | 45 +++ codex-rs/analytics/src/lib.rs | 2 + codex-rs/analytics/src/reducer.rs | 371 +++++++++++++++++- .../app-server/tests/suite/v2/turn_start.rs | 59 +++ codex-rs/core/src/session/turn.rs | 28 ++ .../src/tools/code_mode/execute_handler.rs | 39 +- codex-rs/core/src/tools/code_mode/mod.rs | 12 +- .../core/src/tools/code_mode/telemetry.rs | 59 +++ .../core/src/tools/code_mode/wait_handler.rs | 83 ++-- 13 files changed, 797 insertions(+), 35 deletions(-) create mode 100644 codex-rs/core/src/tools/code_mode/telemetry.rs diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index 8ba7faafc3..caea115825 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -47,6 +47,8 @@ use crate::facts::AnalyticsJsonRpcError; use crate::facts::AppInvocation; use crate::facts::AppMentionedInput; use crate::facts::AppUsedInput; +use crate::facts::CodeModeToolCallFact; +use crate::facts::CodeModeToolCallStatus; use crate::facts::CodexCompactionEvent; use crate::facts::CodexErrKind; use crate::facts::CompactionImplementation; @@ -1665,6 +1667,10 @@ fn command_execution_event_serializes_expected_shape() { session_id: "session-thread-1".to_string(), turn_id: "turn-1".to_string(), item_id: "item-1".to_string(), + cell_id: None, + parent_call_id: None, + originating_response_id: None, + subsequent_response_id: None, app_server_client: CodexAppServerClientMetadata { product_client_id: "codex_tui".to_string(), client_name: Some("codex-tui".to_string()), @@ -1717,6 +1723,10 @@ fn command_execution_event_serializes_expected_shape() { "session_id": "session-thread-1", "turn_id": "turn-1", "item_id": "item-1", + "cell_id": null, + "parent_call_id": null, + "originating_response_id": null, + "subsequent_response_id": null, "app_server_client": { "product_client_id": "codex_tui", "client_name": "codex-tui", @@ -2634,6 +2644,110 @@ async fn command_execution_approval_response_publishes_user_review_event() { assert_eq!(payload[0]["event_params"]["duration_ms"], 42); } +async fn ingest_code_mode_facts( + reducer: &mut AnalyticsReducer, + events: &mut Vec, + facts: impl IntoIterator, +) { + for fact in facts { + reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::CodeModeToolCall(fact)), + events, + ) + .await; + } +} + +fn sampling_response( + turn_id: &str, + response_id: &str, + tool_call_ids: &[&str], +) -> CodeModeToolCallFact { + CodeModeToolCallFact::SamplingResponseCompleted { + thread_id: "thread-1".into(), + turn_id: turn_id.into(), + response_id: response_id.into(), + tool_call_ids: tool_call_ids.iter().map(|id| (*id).into()).collect(), + } +} + +#[tokio::test] +async fn code_mode_exec_wait_and_child_events_share_cell_and_response_ids() { + let mut reducer = AnalyticsReducer::default(); + let mut events = Vec::new(); + ingest_review_prerequisites(&mut reducer, &mut events).await; + + let completed = + |turn_id: &str, call_id: &str, tool_name: &str| CodeModeToolCallFact::Completed { + thread_id: "thread-1".into(), + turn_id: turn_id.into(), + call_id: call_id.into(), + cell_id: Some("cell-1".into()), + tool_name: tool_name.into(), + started_at_ms: 1_000, + completed_at_ms: 1_010, + status: CodeModeToolCallStatus::Completed, + }; + ingest_code_mode_facts( + &mut reducer, + &mut events, + [ + sampling_response("turn-1", "resp-a", &["exec-1"]), + CodeModeToolCallFact::CellStarted { + thread_id: "thread-1".into(), + turn_id: "turn-1".into(), + call_id: "exec-1".into(), + cell_id: "cell-1".into(), + }, + CodeModeToolCallFact::ChildStarted { + thread_id: "thread-1".into(), + turn_id: "turn-1".into(), + call_id: "child-1".into(), + cell_id: "cell-1".into(), + }, + completed("turn-1", "exec-1", "exec"), + ], + ) + .await; + ingest_completed_command_execution_item(&mut reducer, &mut events, "thread-1", "child-1").await; + assert!(events.is_empty()); + + ingest_code_mode_facts( + &mut reducer, + &mut events, + [ + sampling_response("turn-1", "resp-b", &[]), + sampling_response("turn-2", "resp-c", &["wait-1"]), + completed("turn-2", "wait-1", "wait"), + sampling_response("turn-2", "resp-d", &[]), + ], + ) + .await; + + let actual = events + .iter() + .map(|event| { + let event = serde_json::to_value(event).expect("serialize tool event"); + serde_json::json!({ + "item": event["event_params"]["item_id"], + "cell": event["event_params"]["cell_id"], + "parent": event["event_params"]["parent_call_id"], + "origin": event["event_params"]["originating_response_id"], + "subsequent": event["event_params"]["subsequent_response_id"], + }) + }) + .collect::>(); + assert_eq!( + actual, + vec![ + serde_json::json!({"item":"exec-1","cell":"cell-1","parent":null,"origin":"resp-a","subsequent":"resp-b"}), + serde_json::json!({"item":"child-1","cell":"cell-1","parent":"exec-1","origin":"resp-a","subsequent":"resp-b"}), + serde_json::json!({"item":"wait-1","cell":"cell-1","parent":"exec-1","origin":"resp-c","subsequent":"resp-d"}), + ] + ); +} + #[tokio::test] async fn permissions_reviews_emit_events_without_denormalizing_onto_tool_items() { let mut reducer = AnalyticsReducer::default(); diff --git a/codex-rs/analytics/src/client.rs b/codex-rs/analytics/src/client.rs index c76a7d741e..84377c6419 100644 --- a/codex-rs/analytics/src/client.rs +++ b/codex-rs/analytics/src/client.rs @@ -141,6 +141,9 @@ impl AnalyticsEventsQueue { let input = match input { AnalyticsEventsQueueMessage::Fact(input) => *input, AnalyticsEventsQueueMessage::Flush(done_tx) => { + let mut events = Vec::new(); + reducer.flush(&mut events); + send_track_events(&auth_manager, &destination, events).await; let _ = done_tx.send(()); continue; } @@ -288,6 +291,12 @@ impl AnalyticsEventsClient { )); } + pub fn track_code_mode_tool_call(&self, input: crate::facts::CodeModeToolCallFact) { + self.record_fact(AnalyticsFact::Custom( + CustomAnalyticsFact::CodeModeToolCall(input), + )); + } + pub fn track_guardian_review( &self, tracking: &GuardianReviewTrackContext, @@ -607,7 +616,8 @@ impl AnalyticsEventsClient { pub fn track_notification(&self, notification: &ServerNotification) { if !matches!( notification, - ServerNotification::TurnStarted(_) + ServerNotification::ThreadClosed(_) + | ServerNotification::TurnStarted(_) | ServerNotification::TurnCompleted(_) | ServerNotification::TurnDiffUpdated(_) | ServerNotification::ItemStarted(_) diff --git a/codex-rs/analytics/src/client_tests.rs b/codex-rs/analytics/src/client_tests.rs index 4391f2c661..816a981317 100644 --- a/codex-rs/analytics/src/client_tests.rs +++ b/codex-rs/analytics/src/client_tests.rs @@ -145,6 +145,10 @@ fn sample_mcp_tool_call_event(thread_id: &str, plugin_id: Option<&str>) -> Track session_id: format!("session-{thread_id}"), turn_id: "turn-1".to_string(), item_id: format!("item-{thread_id}"), + cell_id: None, + parent_call_id: None, + originating_response_id: None, + subsequent_response_id: None, app_server_client: CodexAppServerClientMetadata { product_client_id: "codex_desktop".to_string(), client_name: None, diff --git a/codex-rs/analytics/src/events.rs b/codex-rs/analytics/src/events.rs index 58aa003b2c..0e6c2fa50e 100644 --- a/codex-rs/analytics/src/events.rs +++ b/codex-rs/analytics/src/events.rs @@ -539,6 +539,10 @@ pub(crate) struct CodexToolItemEventBase { /// App-server ThreadItem.id. For tool-originated items this generally /// corresponds to the originating core call_id. pub(crate) item_id: String, + pub(crate) cell_id: Option, + pub(crate) parent_call_id: Option, + pub(crate) originating_response_id: Option, + pub(crate) subsequent_response_id: Option, pub(crate) app_server_client: CodexAppServerClientMetadata, pub(crate) runtime: CodexRuntimeMetadata, pub(crate) thread_source: Option, diff --git a/codex-rs/analytics/src/facts.rs b/codex-rs/analytics/src/facts.rs index 73847aceab..0547aa1119 100644 --- a/codex-rs/analytics/src/facts.rs +++ b/codex-rs/analytics/src/facts.rs @@ -45,6 +45,50 @@ pub struct TrackEventsContext { pub product_client_id: String, } +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum CodeModeToolCallFact { + CellStarted { + thread_id: String, + turn_id: String, + call_id: String, + cell_id: String, + }, + ChildStarted { + thread_id: String, + turn_id: String, + call_id: String, + cell_id: String, + }, + CellClosed { + thread_id: String, + turn_id: String, + cell_id: String, + }, + SamplingResponseCompleted { + thread_id: String, + turn_id: String, + response_id: String, + tool_call_ids: Vec, + }, + Completed { + thread_id: String, + turn_id: String, + call_id: String, + cell_id: Option, + tool_name: String, + started_at_ms: u64, + completed_at_ms: u64, + status: CodeModeToolCallStatus, + }, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum CodeModeToolCallStatus { + Completed, + Failed, + Interrupted, +} + pub fn build_track_events_context( model_slug: String, thread_id: String, @@ -445,6 +489,7 @@ pub(crate) enum AnalyticsFact { } pub(crate) enum CustomAnalyticsFact { + CodeModeToolCall(CodeModeToolCallFact), SubAgentThreadStarted(SubAgentThreadStartedInput), Compaction(Box), Goal(Box), diff --git a/codex-rs/analytics/src/lib.rs b/codex-rs/analytics/src/lib.rs index dc3e8ad81f..29ab3bbbc0 100644 --- a/codex-rs/analytics/src/lib.rs +++ b/codex-rs/analytics/src/lib.rs @@ -26,6 +26,8 @@ pub use events::GuardianReviewedAction; pub use facts::AcceptedLineFingerprint; pub use facts::AnalyticsJsonRpcError; pub use facts::AppInvocation; +pub use facts::CodeModeToolCallFact; +pub use facts::CodeModeToolCallStatus; pub use facts::CodexCompactionEvent; pub use facts::CodexErrKind; pub use facts::CodexGoalEvent; diff --git a/codex-rs/analytics/src/reducer.rs b/codex-rs/analytics/src/reducer.rs index 5dd9cb2c7b..03a9e6ea51 100644 --- a/codex-rs/analytics/src/reducer.rs +++ b/codex-rs/analytics/src/reducer.rs @@ -71,6 +71,8 @@ use crate::facts::AnalyticsFact; use crate::facts::AnalyticsJsonRpcError; use crate::facts::AppMentionedInput; use crate::facts::AppUsedInput; +use crate::facts::CodeModeToolCallFact; +use crate::facts::CodeModeToolCallStatus; use crate::facts::CodexCompactionEvent; use crate::facts::CodexGoalEvent; use crate::facts::CustomAnalyticsFact; @@ -145,8 +147,10 @@ use codex_protocol::request_permissions::PermissionGrantScope as CorePermissionG use codex_protocol::request_permissions::RequestPermissionsResponse as CoreRequestPermissionsResponse; use sha1::Digest; use std::collections::HashMap; +use std::collections::VecDeque; use std::path::Path; use std::path::PathBuf; +const MAX_TOOL_RESPONSE_ENTRIES: usize = 256; #[derive(Default)] pub(crate) struct AnalyticsReducer { @@ -155,6 +159,8 @@ pub(crate) struct AnalyticsReducer { connections: HashMap, threads: HashMap, tool_items_started_at_ms: HashMap, + tool_response_states: HashMap<(String, String), ToolResponseState>, + code_mode_cells: HashMap>, pending_reviews: HashMap, item_review_summaries: HashMap, } @@ -384,13 +390,31 @@ struct TurnState { tool_counts: TurnToolCounts, } -#[derive(Hash, Eq, PartialEq)] +#[derive(Clone, Hash, Eq, PartialEq)] struct ToolItemKey { thread_id: String, turn_id: String, item_id: String, } +#[derive(Default)] +struct ToolResponseState { + response_ids_by_call_id: HashMap, + cell_ids_by_child_call_id: HashMap, + pending_tool_events: VecDeque, +} + +struct CodeModeCellState { + parent_call_id: String, + originating_response_id: Option, + closed_in_turn_id: Option, +} + +enum ToolEventEmission { + ImmediateUnlessCorrelated, + AwaitResponse, +} + #[derive(Default)] struct TurnToolCounts { total: usize, @@ -522,6 +546,9 @@ impl AnalyticsReducer { self.ingest_server_request_aborted(completed_at_ms, request_id, out); } AnalyticsFact::Custom(input) => match input { + CustomAnalyticsFact::CodeModeToolCall(input) => { + self.ingest_code_mode_tool_call(input, out); + } CustomAnalyticsFact::SubAgentThreadStarted(input) => { self.ingest_subagent_thread_started(input, out); } @@ -583,6 +610,281 @@ impl AnalyticsReducer { } } + fn ingest_code_mode_tool_call( + &mut self, + input: CodeModeToolCallFact, + out: &mut Vec, + ) { + let thread_id = match &input { + CodeModeToolCallFact::CellStarted { thread_id, .. } + | CodeModeToolCallFact::ChildStarted { thread_id, .. } + | CodeModeToolCallFact::CellClosed { thread_id, .. } + | CodeModeToolCallFact::SamplingResponseCompleted { thread_id, .. } + | CodeModeToolCallFact::Completed { thread_id, .. } => thread_id, + }; + let has_thread_context = self.threads.get(thread_id).is_some_and(|thread| { + thread.metadata.is_some() + && thread + .connection_id + .is_some_and(|connection_id| self.connections.contains_key(&connection_id)) + }); + if !has_thread_context { + return; + } + + match input { + CodeModeToolCallFact::CellStarted { + thread_id, + turn_id, + call_id, + cell_id, + } => { + let cells = self.code_mode_cells.entry(thread_id.clone()).or_default(); + if cells.contains_key(&cell_id) || cells.len() < MAX_TOOL_RESPONSE_ENTRIES { + let originating_response_id = self + .tool_response_states + .get(&(thread_id, turn_id)) + .and_then(|state| state.response_ids_by_call_id.get(&call_id)) + .cloned(); + cells.insert( + cell_id, + CodeModeCellState { + parent_call_id: call_id, + originating_response_id, + closed_in_turn_id: None, + }, + ); + } + } + CodeModeToolCallFact::ChildStarted { + thread_id, + turn_id, + call_id, + cell_id, + } => { + if self + .code_mode_cells + .get(&thread_id) + .is_some_and(|cells| cells.contains_key(&cell_id)) + { + let state = self + .tool_response_states + .entry((thread_id, turn_id)) + .or_default(); + if state.cell_ids_by_child_call_id.contains_key(&call_id) + || state.cell_ids_by_child_call_id.len() < MAX_TOOL_RESPONSE_ENTRIES + { + state.cell_ids_by_child_call_id.insert(call_id, cell_id); + } + } + } + CodeModeToolCallFact::CellClosed { + thread_id, + turn_id, + cell_id, + } => { + if let Some(cell) = self + .code_mode_cells + .get_mut(&thread_id) + .and_then(|cells| cells.get_mut(&cell_id)) + { + cell.closed_in_turn_id = Some(turn_id); + } + } + CodeModeToolCallFact::SamplingResponseCompleted { + thread_id, + turn_id, + response_id, + tool_call_ids, + } => { + self.ingest_sampling_response_completed( + thread_id, + turn_id, + response_id, + tool_call_ids, + out, + ); + } + CodeModeToolCallFact::Completed { + thread_id, + turn_id, + call_id, + cell_id, + tool_name, + started_at_ms, + completed_at_ms, + status, + } => { + let drop_site = AnalyticsDropSite { + event_name: "code mode tool", + thread_id: &thread_id, + turn_id: Some(&turn_id), + review_id: None, + item_id: Some(&call_id), + }; + let Some((connection_state, thread_state, thread_metadata)) = + self.thread_context_or_warn(drop_site) + else { + return; + }; + let terminal_status = match status { + CodeModeToolCallStatus::Completed => ToolItemTerminalStatus::Completed, + CodeModeToolCallStatus::Failed => ToolItemTerminalStatus::Failed, + CodeModeToolCallStatus::Interrupted => ToolItemTerminalStatus::Interrupted, + }; + let success = status == CodeModeToolCallStatus::Completed; + let mut base = tool_item_base( + &thread_id, + &turn_id, + call_id, + tool_name.clone(), + ToolItemOutcome { + terminal_status, + failure_kind: (status == CodeModeToolCallStatus::Failed) + .then_some(ToolItemFailureKind::ToolError), + execution_duration_ms: observed_duration_ms(started_at_ms, completed_at_ms), + }, + ToolItemContext { + started_at_ms, + completed_at_ms, + connection_state, + thread_state, + thread_metadata, + review_summary: None, + }, + ); + base.cell_id = cell_id; + let event = TrackEventRequest::DynamicToolCall(CodexDynamicToolCallEventRequest { + event_type: "codex_dynamic_tool_call_event", + event_params: CodexDynamicToolCallEventParams { + base, + dynamic_tool_name: tool_name, + success: Some(success), + output_content_item_count: None, + output_text_item_count: None, + output_image_item_count: None, + output_audio_item_count: None, + }, + }); + let counts = &mut self.turns.entry(turn_id.clone()).or_default().tool_counts; + counts.total += 1; + counts.dynamic_tool_call += 1; + self.record_tool_event( + &thread_id, + &turn_id, + event, + ToolEventEmission::AwaitResponse, + out, + ); + } + } + } + + fn ingest_sampling_response_completed( + &mut self, + thread_id: String, + turn_id: String, + response_id: String, + tool_call_ids: Vec, + out: &mut Vec, + ) { + let turn_key = (thread_id.clone(), turn_id); + let state = self.tool_response_states.entry(turn_key).or_default(); + for call_id in tool_call_ids { + if state.response_ids_by_call_id.contains_key(&call_id) + || state.response_ids_by_call_id.len() < MAX_TOOL_RESPONSE_ENTRIES + { + state + .response_ids_by_call_id + .insert(call_id, response_id.clone()); + } + } + if let Some(cells) = self.code_mode_cells.get_mut(&thread_id) { + for cell in cells.values_mut() { + if cell.originating_response_id.is_none() + && let Some(originating_response_id) = + state.response_ids_by_call_id.get(&cell.parent_call_id) + { + cell.originating_response_id = Some(originating_response_id.clone()); + } + } + } + + let mut remaining = VecDeque::new(); + while let Some(mut event) = state.pending_tool_events.pop_front() { + enrich_tool_response_event(&mut event, state, self.code_mode_cells.get(&thread_id)); + let is_subsequent_response = tool_event_base_mut(&mut event) + .and_then(|base| base.originating_response_id.as_deref()) + .is_some_and(|originating_response_id| originating_response_id != response_id); + if is_subsequent_response { + if let Some(base) = tool_event_base_mut(&mut event) { + base.subsequent_response_id = Some(response_id.clone()); + } + out.push(event); + } else { + remaining.push_back(event); + } + } + state.pending_tool_events = remaining; + } + + fn record_tool_event( + &mut self, + thread_id: &str, + turn_id: &str, + mut event: TrackEventRequest, + emission: ToolEventEmission, + out: &mut Vec, + ) { + if tool_event_base_mut(&mut event).is_none() { + out.push(event); + return; + } + let state = self + .tool_response_states + .entry((thread_id.to_string(), turn_id.to_string())) + .or_default(); + enrich_tool_response_event(&mut event, state, self.code_mode_cells.get(thread_id)); + let is_correlated = tool_event_base_mut(&mut event) + .is_some_and(|base| base.cell_id.is_some() || base.originating_response_id.is_some()); + if matches!(emission, ToolEventEmission::ImmediateUnlessCorrelated) && !is_correlated { + out.push(event); + return; + } + if state.pending_tool_events.len() == MAX_TOOL_RESPONSE_ENTRIES + && let Some(oldest) = state.pending_tool_events.pop_front() + { + out.push(oldest); + } + state.pending_tool_events.push_back(event); + } + + fn flush_pending_tool_events( + &mut self, + thread_id: &str, + turn_id: &str, + out: &mut Vec, + ) { + let key = (thread_id.to_string(), turn_id.to_string()); + if let Some(state) = self.tool_response_states.remove(&key) { + out.extend(state.pending_tool_events); + } + let cells = self.code_mode_cells.get_mut(thread_id); + let remove_cells = cells.is_some_and(|cells| { + cells.retain(|_, cell| cell.closed_in_turn_id.as_deref() != Some(turn_id)); + cells.is_empty() + }); + if remove_cells { + self.code_mode_cells.remove(thread_id); + } + } + + pub(crate) fn flush(&mut self, out: &mut Vec) { + for state in self.tool_response_states.values_mut() { + out.extend(state.pending_tool_events.drain(..)); + } + } + fn ingest_initialize( &mut self, connection_id: u64, @@ -1347,7 +1649,13 @@ impl AnalyticsReducer { thread_metadata, review_summary: self.item_review_summaries.get(&key), }) { - out.push(event); + self.record_tool_event( + ¬ification.thread_id, + ¬ification.turn_id, + event, + ToolEventEmission::ImmediateUnlessCorrelated, + out, + ); } self.item_review_summaries.remove(&key); } @@ -1357,6 +1665,17 @@ impl AnalyticsReducer { ServerNotification::ItemGuardianApprovalReviewCompleted(notification) => { self.ingest_guardian_review_completed(notification, out); } + ServerNotification::ThreadClosed(notification) => { + self.tool_response_states.retain(|(thread_id, _), state| { + if thread_id == ¬ification.thread_id { + out.extend(state.pending_tool_events.drain(..)); + false + } else { + true + } + }); + self.code_mode_cells.remove(¬ification.thread_id); + } ServerNotification::TurnStarted(notification) => { let turn_state = self.turns.entry(notification.turn.id).or_default(); turn_state.started_at = notification @@ -1370,6 +1689,7 @@ impl AnalyticsReducer { turn_state.latest_diff = Some(notification.diff); } ServerNotification::TurnCompleted(notification) => { + self.flush_pending_tool_events(¬ification.thread_id, ¬ification.turn.id, out); let turn_state = self.turns.entry(notification.turn.id.clone()).or_default(); turn_state.completed = Some(CompletedTurnState { status: analytics_turn_status(notification.turn.status), @@ -1809,6 +2129,49 @@ fn tracked_tool_item_id(item: &ThreadItem) -> Option<&str> { } } +fn tool_event_base_mut(event: &mut TrackEventRequest) -> Option<&mut CodexToolItemEventBase> { + match event { + TrackEventRequest::CommandExecution(event) => Some(&mut event.event_params.base), + TrackEventRequest::FileChange(event) => Some(&mut event.event_params.base), + TrackEventRequest::McpToolCall(event) => Some(&mut event.event_params.base), + TrackEventRequest::DynamicToolCall(event) => Some(&mut event.event_params.base), + TrackEventRequest::CollabAgentToolCall(event) => Some(&mut event.event_params.base), + TrackEventRequest::WebSearch(event) => Some(&mut event.event_params.base), + TrackEventRequest::ImageGeneration(event) => Some(&mut event.event_params.base), + _ => None, + } +} + +fn enrich_tool_response_event( + event: &mut TrackEventRequest, + state: &ToolResponseState, + cells: Option<&HashMap>, +) { + let Some(base) = tool_event_base_mut(event) else { + return; + }; + if base.cell_id.is_none() { + base.cell_id = state.cell_ids_by_child_call_id.get(&base.item_id).cloned(); + } + + let cell = base + .cell_id + .as_ref() + .and_then(|cell_id| cells?.get(cell_id)); + if let Some(cell) = cell { + base.parent_call_id = + (cell.parent_call_id != base.item_id).then(|| cell.parent_call_id.clone()); + } else { + base.cell_id = None; + base.parent_call_id = None; + } + base.originating_response_id = state + .response_ids_by_call_id + .get(&base.item_id) + .cloned() + .or_else(|| cell.and_then(|cell| cell.originating_response_id.clone())); +} + fn item_review_summary_key(pending_review: &PendingReviewState) -> Option { match pending_review.subject_kind { ReviewSubjectKind::CommandExecution @@ -2229,6 +2592,10 @@ fn tool_item_base( session_id: thread_metadata.session_id.clone(), turn_id: turn_id.to_string(), item_id, + cell_id: None, + parent_call_id: None, + originating_response_id: None, + subsequent_response_id: None, app_server_client: context .thread_state .app_server_client(context.connection_state), diff --git a/codex-rs/app-server/tests/suite/v2/turn_start.rs b/codex-rs/app-server/tests/suite/v2/turn_start.rs index a2e4886cc4..38a9e3d05b 100644 --- a/codex-rs/app-server/tests/suite/v2/turn_start.rs +++ b/codex-rs/app-server/tests/suite/v2/turn_start.rs @@ -962,6 +962,65 @@ async fn turn_start_tracks_thread_originator_in_analytics() -> Result<()> { Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn code_mode_exec_emits_correlated_production_analytics() -> Result<()> { + let server = responses::start_mock_server().await; + let _responses = responses::mount_sse_sequence( + &server, + vec![ + responses::sse(vec![ + responses::ev_response_created("resp-1"), + responses::ev_custom_tool_call("exec-1", "exec", "text('analytics');"), + responses::ev_completed("resp-1"), + ]), + responses::sse(vec![responses::ev_completed("resp-2")]), + ], + ) + .await; + let codex_home = TempDir::new()?; + MockResponsesConfig::new(&server.uri()) + .enable_feature(Feature::CodeModeOnly) + .with_root_config(&format!("chatgpt_base_url = \"{}\"", server.uri())) + .write(codex_home.path())?; + mount_analytics_capture(&server, codex_home.path()).await?; + + let mut app_server = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .without_managed_config() + .build_initialized() + .await?; + let params = ThreadStartParams::default(); + let thread = app_server.start_thread(params).await?; + app_server + .start_turn_and_wait_for_completion(TurnStartParams { + thread_id: thread.thread.id, + input: vec![V2UserInput::Text { + text: "run exec".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }) + .await?; + + let event = wait_for_analytics_event( + &server, + DEFAULT_READ_TIMEOUT, + "codex_dynamic_tool_call_event", + ) + .await?; + assert_eq!( + json!({ + "tool": event["event_params"]["tool_name"], + "origin": event["event_params"]["originating_response_id"], + "subsequent": event["event_params"]["subsequent_response_id"], + "hasCell": event["event_params"]["cell_id"].as_str().is_some(), + }), + json!({"tool":"exec","origin":"resp-1","subsequent":"resp-2","hasCell":true}) + ); + + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn turn_profile_tracks_blocking_tool_and_follow_up_sampling() -> Result<()> { let responses = vec![ diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index 3521fbf1aa..70e09205dc 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -2207,6 +2207,8 @@ async fn try_run_sampling_request( )> = None; let mut should_emit_turn_diff = false; let mut should_emit_token_count = false; + const MAX_ANALYTICS_TOOL_CALL_IDS_PER_RESPONSE: usize = 256; + let mut analytics_tool_call_ids = Vec::new(); let reasoning_effort = turn_context.effective_reasoning_effort_for_tracing(); let plan_mode = turn_context.mode == ModeKind::Plan; let mut assistant_message_stream_parsers = AssistantMessageStreamParsers::new(plan_mode); @@ -2262,6 +2264,22 @@ async fn try_run_sampling_request( ResponseEvent::Created => {} ResponseEvent::OutputItemDone(mut item) => { assign_missing_streamed_response_item_id(&mut item, active_item.as_ref()); + if analytics_tool_call_ids.len() < MAX_ANALYTICS_TOOL_CALL_IDS_PER_RESPONSE { + let call_id = match &item { + ResponseItem::FunctionCall { call_id, .. } + | ResponseItem::CustomToolCall { call_id, .. } => Some(call_id.as_str()), + ResponseItem::ToolSearchCall { call_id, .. } + | ResponseItem::LocalShellCall { call_id, .. } => call_id.as_deref(), + ResponseItem::WebSearchCall { id, .. } + | ResponseItem::ImageGenerationCall { id, .. } => { + id.as_ref().map(codex_protocol::ResponseItemId::as_str) + } + _ => None, + }; + if let Some(call_id) = call_id { + analytics_tool_call_ids.push(call_id.to_string()); + } + } if let Some((_, mut consumer)) = active_tool_argument_diff_consumer.take() && let Ok(Some(event)) = consumer.finish() { @@ -2493,6 +2511,16 @@ async fn try_run_sampling_request( token_usage, end_turn, } => { + sess.services + .analytics_events_client + .track_code_mode_tool_call( + codex_analytics::CodeModeToolCallFact::SamplingResponseCompleted { + thread_id: sess.thread_id.to_string(), + turn_id: turn_context.sub_id.clone(), + response_id: response_id.clone(), + tool_call_ids: std::mem::take(&mut analytics_tool_call_ids), + }, + ); flush_assistant_text_segments_all( &sess, &turn_context, diff --git a/codex-rs/core/src/tools/code_mode/execute_handler.rs b/codex-rs/core/src/tools/code_mode/execute_handler.rs index ffc9e43feb..e6ea6ff461 100644 --- a/codex-rs/core/src/tools/code_mode/execute_handler.rs +++ b/codex-rs/core/src/tools/code_mode/execute_handler.rs @@ -12,6 +12,7 @@ use super::ExecContext; use super::PUBLIC_TOOL_NAME; use super::handle_runtime_response; use super::is_exec_tool_name; +use super::telemetry::CodeModeToolCallGuard; pub struct CodeModeExecuteHandler { spec: ToolSpec, @@ -32,6 +33,7 @@ impl CodeModeExecuteHandler { turn: std::sync::Arc, call_id: String, code: String, + telemetry: &mut CodeModeToolCallGuard, ) -> Result { let args = codex_code_mode::parse_exec_source(&code).map_err(FunctionCallError::RespondToModel)?; @@ -53,6 +55,16 @@ impl CodeModeExecuteHandler { .await .map_err(FunctionCallError::RespondToModel)?; let cell_id = started_cell.cell_id.clone(); + telemetry.cell_id = Some(cell_id.to_string()); + exec.session + .services + .analytics_events_client + .track_code_mode_tool_call(codex_analytics::CodeModeToolCallFact::CellStarted { + thread_id: exec.session.thread_id.to_string(), + turn_id: exec.turn.sub_id.clone(), + call_id: call_id.clone(), + cell_id: cell_id.to_string(), + }); if let Some(executed_tool_calls) = exec.session.services.executed_tool_calls.as_ref() { executed_tool_calls.register_cell(&cell_id, &call_id); } @@ -87,6 +99,14 @@ impl CodeModeExecuteHandler { .services .code_mode_service .finish_cell_dispatch(&cell_id); + exec.session + .services + .analytics_events_client + .track_code_mode_tool_call(codex_analytics::CodeModeToolCallFact::CellClosed { + thread_id: exec.session.thread_id.to_string(), + turn_id: exec.turn.sub_id.clone(), + cell_id: cell_id.to_string(), + }); } exec.session.services.elicitations.wait_until_clear().await; handle_runtime_response(&exec, response, args.max_output_tokens, started_at) @@ -123,15 +143,28 @@ impl CodeModeExecuteHandler { .. } = invocation; - match payload { + let mut telemetry = CodeModeToolCallGuard::new( + session.services.analytics_events_client.clone(), + session.thread_id.to_string(), + turn.sub_id.clone(), + call_id.clone(), + PUBLIC_TOOL_NAME, + ); + let result = match payload { ToolPayload::Custom { input } if is_exec_tool_name(&tool_name) => self - .execute(session, turn, call_id, input) + .execute(session, turn, call_id, input, &mut telemetry) .await .map(boxed_tool_output), _ => Err(FunctionCallError::RespondToModel(format!( "{PUBLIC_TOOL_NAME} expects raw JavaScript source text" ))), - } + }; + telemetry.finish( + result + .as_ref() + .is_ok_and(|output| output.success_for_logging()), + ); + result } } diff --git a/codex-rs/core/src/tools/code_mode/mod.rs b/codex-rs/core/src/tools/code_mode/mod.rs index 144334c8ba..5054f68c9c 100644 --- a/codex-rs/core/src/tools/code_mode/mod.rs +++ b/codex-rs/core/src/tools/code_mode/mod.rs @@ -2,6 +2,7 @@ mod delegate; mod execute_handler; pub(crate) mod execute_spec; mod response_adapter; +mod telemetry; mod wait_handler; pub(crate) mod wait_spec; @@ -314,7 +315,7 @@ fn truncate_code_mode_result( } async fn call_nested_tool( - _exec: ExecContext, + exec: ExecContext, tool_runtime: ToolCallRuntime, invocation: CodeModeNestedToolCall, cancellation_token: CancellationToken, @@ -343,6 +344,15 @@ async fn call_nested_tool( payload, encrypted_function_args: None, }; + exec.session + .services + .analytics_events_client + .track_code_mode_tool_call(codex_analytics::CodeModeToolCallFact::ChildStarted { + thread_id: exec.session.thread_id.to_string(), + turn_id: exec.turn.sub_id.clone(), + call_id: call.call_id.clone(), + cell_id: cell_id.to_string(), + }); let result = tool_runtime .handle_tool_call_with_source( call, diff --git a/codex-rs/core/src/tools/code_mode/telemetry.rs b/codex-rs/core/src/tools/code_mode/telemetry.rs new file mode 100644 index 0000000000..9d1b2584db --- /dev/null +++ b/codex-rs/core/src/tools/code_mode/telemetry.rs @@ -0,0 +1,59 @@ +use codex_analytics::AnalyticsEventsClient; +use codex_analytics::CodeModeToolCallFact; +use codex_analytics::CodeModeToolCallStatus; + +pub(super) struct CodeModeToolCallGuard { + analytics: AnalyticsEventsClient, + thread_id: String, + turn_id: String, + call_id: String, + pub(super) cell_id: Option, + tool_name: &'static str, + started_at_ms: u64, + status: CodeModeToolCallStatus, +} + +impl CodeModeToolCallGuard { + pub(super) fn new( + analytics: AnalyticsEventsClient, + thread_id: String, + turn_id: String, + call_id: String, + tool_name: &'static str, + ) -> Self { + Self { + analytics, + thread_id, + turn_id, + call_id, + cell_id: None, + tool_name, + started_at_ms: codex_analytics::now_unix_millis(), + status: CodeModeToolCallStatus::Interrupted, + } + } + + pub(super) fn finish(&mut self, success: bool) { + self.status = if success { + CodeModeToolCallStatus::Completed + } else { + CodeModeToolCallStatus::Failed + }; + } +} + +impl Drop for CodeModeToolCallGuard { + fn drop(&mut self) { + self.analytics + .track_code_mode_tool_call(CodeModeToolCallFact::Completed { + thread_id: self.thread_id.clone(), + turn_id: self.turn_id.clone(), + call_id: self.call_id.clone(), + cell_id: self.cell_id.clone(), + tool_name: self.tool_name.to_string(), + started_at_ms: self.started_at_ms, + completed_at_ms: codex_analytics::now_unix_millis(), + status: self.status, + }); + } +} diff --git a/codex-rs/core/src/tools/code_mode/wait_handler.rs b/codex-rs/core/src/tools/code_mode/wait_handler.rs index d128850a23..d9e656775a 100644 --- a/codex-rs/core/src/tools/code_mode/wait_handler.rs +++ b/codex-rs/core/src/tools/code_mode/wait_handler.rs @@ -16,6 +16,7 @@ use super::DEFAULT_WAIT_YIELD_TIME_MS; use super::ExecContext; use super::WAIT_TOOL_NAME; use super::handle_runtime_response; +use super::telemetry::CodeModeToolCallGuard; use super::wait_spec::create_wait_tool; pub struct CodeModeWaitHandler; @@ -72,19 +73,23 @@ impl CodeModeWaitHandler { .. } = invocation; - match payload { + let mut telemetry = CodeModeToolCallGuard::new( + session.services.analytics_events_client.clone(), + session.thread_id.to_string(), + turn.sub_id.clone(), + call_id.clone(), + WAIT_TOOL_NAME, + ); + let result = match payload { ToolPayload::Function { arguments } if tool_name.namespace.is_none() && tool_name.name.as_str() == WAIT_TOOL_NAME => { - let args: ExecWaitArgs = parse_arguments(&arguments)?; + let args: ExecWaitArgs = parse_arguments(&arguments).inspect_err(|_error| { + telemetry.finish(/*success*/ false); + })?; let exec = ExecContext { session, turn }; let started_at = std::time::Instant::now(); let cell_id = codex_code_mode::CellId::new(args.cell_id); - if let Some(executed_tool_calls) = - exec.session.services.executed_tool_calls.as_ref() - { - executed_tool_calls.register_cell(&cell_id, &call_id); - } let wait_response = if args.terminate { exec.session .services @@ -101,41 +106,63 @@ impl CodeModeWaitHandler { }) .await } - .map_err(FunctionCallError::RespondToModel)?; - if let codex_code_mode::WaitOutcome::LiveCell(response) = &wait_response - && !matches!(response, codex_code_mode::RuntimeResponse::Yielded { .. }) - { - // Only a live-cell wait can close a CodeCell. A missing - // cell is still an ordinary `wait` tool result, but there - // is no runtime object for the reducer to complete. + .map_err(|error| { + telemetry.finish(/*success*/ false); + FunctionCallError::RespondToModel(error) + })?; + if let codex_code_mode::WaitOutcome::LiveCell(response) = &wait_response { let runtime_cell_id = match response { codex_code_mode::RuntimeResponse::Yielded { cell_id, .. } | codex_code_mode::RuntimeResponse::Terminated { cell_id, .. } | codex_code_mode::RuntimeResponse::Result { cell_id, .. } => cell_id, }; - exec.session - .services - .rollout_thread_trace - .code_cell_trace_context( - exec.turn.sub_id.as_str(), - runtime_cell_id.as_str(), - ) - .record_ended(response); - exec.session - .services - .code_mode_service - .finish_cell_dispatch(runtime_cell_id); + telemetry.cell_id = Some(runtime_cell_id.to_string()); + if let Some(executed_tool_calls) = + exec.session.services.executed_tool_calls.as_ref() + { + executed_tool_calls.register_cell(runtime_cell_id, &call_id); + } + if !matches!(response, codex_code_mode::RuntimeResponse::Yielded { .. }) { + exec.session + .services + .rollout_thread_trace + .code_cell_trace_context( + exec.turn.sub_id.as_str(), + runtime_cell_id.as_str(), + ) + .record_ended(response); + exec.session + .services + .code_mode_service + .finish_cell_dispatch(runtime_cell_id); + exec.session + .services + .analytics_events_client + .track_code_mode_tool_call( + codex_analytics::CodeModeToolCallFact::CellClosed { + thread_id: exec.session.thread_id.to_string(), + turn_id: exec.turn.sub_id.clone(), + cell_id: runtime_cell_id.to_string(), + }, + ); + } } exec.session.services.elicitations.wait_until_clear().await; handle_runtime_response(&exec, wait_response.into(), args.max_tokens, started_at) .await - .map(boxed_tool_output) .map_err(FunctionCallError::RespondToModel) + .map(boxed_tool_output) } _ => Err(FunctionCallError::RespondToModel(format!( "{WAIT_TOOL_NAME} expects JSON arguments" ))), - } + }; + telemetry.finish( + result + .as_ref() + .is_ok_and(codex_tools::ToolOutput::success_for_logging), + ); + result } }