From c21153bd920a514809e58d9bd10326267ab2ca36 Mon Sep 17 00:00:00 2001 From: trpcic-openai Date: Fri, 23 Oct 2026 18:41:34 +0000 Subject: [PATCH] Track CUA tool call starts and deduplicate lifecycle events (#57167) ## What changed Emit `codex_cua_tool_call_started` for `js` MCP calls on Node REPL-backed connectors. Include start time, tool identity, plugin and connector IDs, and available thread and response attribution without recording tool arguments. Suppress duplicate starts and replayed completions for CUA calls while retaining the existing MCP completion event. A completion received without a start does not prevent a later valid lifecycle from being recorded. ## Testing Add a lifecycle regression test covering missing starts, duplicate notifications, replay after completion, attribution fields, timestamps, and exclusion of tool arguments from serialized events. GitOrigin-RevId: 68554ad76c636bb3c2be38bda56f5ea6da8d9830 --- codex-rs/analytics/src/events.rs | 31 +++ codex-rs/analytics/src/product_attribution.rs | 1 + codex-rs/analytics/src/reducer.rs | 182 +++++++++++++++--- .../analytics/src/tests/suite/tools_tests.rs | 94 +++++++++ 4 files changed, 281 insertions(+), 27 deletions(-) diff --git a/codex-rs/analytics/src/events.rs b/codex-rs/analytics/src/events.rs index cb0aacff8d..79dbf28536 100644 --- a/codex-rs/analytics/src/events.rs +++ b/codex-rs/analytics/src/events.rs @@ -85,6 +85,7 @@ pub(crate) enum TrackEventRequest { CommandExecution(CodexCommandExecutionEventRequest), PluginMeasurement(CodexPluginMeasurementEventRequest), FileChange(CodexFileChangeEventRequest), + CuaToolCallStarted(CodexCuaToolCallStartedEventRequest), McpToolCall(CodexMcpToolCallEventRequest), DynamicToolCall(CodexDynamicToolCallEventRequest), ControlToolCall(CodexControlToolCallEventRequest), @@ -167,6 +168,7 @@ impl TrackEventRequest { match self { Self::PluginUsed(event) => event.event_params.plugin.plugin_id.is_some(), Self::SkillInvocation(event) => event.event_params.plugin_id.is_some(), + Self::CuaToolCallStarted(event) => event.event_params.plugin_id.is_some(), Self::McpToolCall(event) => event.event_params.plugin_id.is_some(), Self::ArtifactOperation(event) => !event.event_params.plugin_id.is_empty(), Self::PluginMeasurement(event) => !event.event_params.plugin_id.is_empty(), @@ -714,6 +716,35 @@ pub(crate) struct CodexToolItemEventBase { pub(crate) requested_network_access: bool, } +#[derive(Serialize)] +pub(crate) struct CodexCuaToolCallStartedEventParams { + pub(crate) thread_id: String, + pub(crate) session_id: String, + pub(crate) turn_id: String, + pub(crate) root_turn_id: Option, + pub(crate) item_id: String, + pub(crate) cell_id: Option, + pub(crate) parent_call_id: Option, + pub(crate) originating_response_id: Option, + pub(crate) app_server_client: CodexAppServerClientMetadata, + pub(crate) runtime: CodexRuntimeMetadata, + pub(crate) thread_source: Option, + pub(crate) subagent_source: Option, + pub(crate) parent_thread_id: Option, + pub(crate) tool_event_type: Option, + pub(crate) started_at_ms: u64, + pub(crate) mcp_server_name: String, + pub(crate) mcp_tool_name: String, + pub(crate) plugin_id: Option, + pub(crate) connector_id: Option, +} + +#[derive(Serialize)] +pub(crate) struct CodexCuaToolCallStartedEventRequest { + pub(crate) event_type: &'static str, + pub(crate) event_params: CodexCuaToolCallStartedEventParams, +} + #[allow(dead_code)] #[derive(Clone, Copy, Debug, Serialize)] #[serde(rename_all = "snake_case")] diff --git a/codex-rs/analytics/src/product_attribution.rs b/codex-rs/analytics/src/product_attribution.rs index 4fa872102e..da5958f35d 100644 --- a/codex-rs/analytics/src/product_attribution.rs +++ b/codex-rs/analytics/src/product_attribution.rs @@ -88,6 +88,7 @@ fn event_thread_id(event: &TrackEventRequest) -> Option<&str> { TrackEventRequest::CommandExecution(event) => Some(&event.event_params.base.thread_id), TrackEventRequest::PluginMeasurement(event) => Some(&event.event_params.thread_id), TrackEventRequest::FileChange(event) => Some(&event.event_params.base.thread_id), + TrackEventRequest::CuaToolCallStarted(event) => Some(&event.event_params.thread_id), TrackEventRequest::McpToolCall(event) => Some(&event.event_params.base.thread_id), TrackEventRequest::DynamicToolCall(event) => Some(&event.event_params.base.thread_id), TrackEventRequest::ControlToolCall(event) => Some(&event.event_params.base.thread_id), diff --git a/codex-rs/analytics/src/reducer.rs b/codex-rs/analytics/src/reducer.rs index 0bebe6771d..cef0a71216 100644 --- a/codex-rs/analytics/src/reducer.rs +++ b/codex-rs/analytics/src/reducer.rs @@ -14,6 +14,8 @@ use crate::events::CodexCommandExecutionEventRequest; use crate::events::CodexCompactionEventRequest; use crate::events::CodexControlToolCallEventParams; use crate::events::CodexControlToolCallEventRequest; +use crate::events::CodexCuaToolCallStartedEventParams; +use crate::events::CodexCuaToolCallStartedEventRequest; use crate::events::CodexDynamicToolCallEventParams; use crate::events::CodexDynamicToolCallEventRequest; use crate::events::CodexFileChangeEventParams; @@ -165,6 +167,7 @@ use codex_protocol::config_types::Personality; use codex_protocol::config_types::ReasoningSummary; use codex_protocol::items::ModelInvocationContext; use codex_protocol::items::is_safe_plugin_relative_path; +use codex_protocol::mcp::is_node_repl_backed_connector; use codex_protocol::models::PermissionProfile; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SkillScope; @@ -342,6 +345,16 @@ impl<'a> AnalyticsDropSite<'a> { } } + fn cua_tool_call_started(thread_id: &'a str, turn_id: &'a str, item_id: &'a str) -> Self { + Self { + event_name: "CUA tool call start", + thread_id, + turn_id: Some(turn_id), + review_id: None, + item_id: Some(item_id), + } + } + fn turn_steer(thread_id: &'a str) -> Self { Self { event_name: "turn steer", @@ -478,6 +491,7 @@ struct TurnState { latest_diff: Option, steer_count: usize, tool_counts: TurnToolCounts, + completed_cua_tool_item_ids: HashSet, resource_skill_invocations: HashSet, turn_event_emitted: bool, } @@ -2060,14 +2074,78 @@ impl AnalyticsReducer { self.turn_state(notification.turn_id.clone()) .thread_id .get_or_insert(notification.thread_id.clone()); + let key = ToolItemKey { + thread_id: notification.thread_id.clone(), + turn_id: notification.turn_id.clone(), + item_id: item_id.to_string(), + }; + let cua_identity = cua_mcp_tool_identity(¬ification.item); + if cua_identity.is_some() + && self + .turns + .get(¬ification.turn_id) + .is_some_and(|turn| turn.completed_cua_tool_item_ids.contains(item_id)) + { + return; + } + let first_notification = !self.tool_items_started_at_ms.contains_key(&key); self.tool_items_started_at_ms - .entry(ToolItemKey { - thread_id: notification.thread_id, - turn_id: notification.turn_id, - item_id: item_id.to_string(), - }) + .entry(key) .and_modify(|(timestamp, _)| *timestamp = started_at_ms) .or_insert((started_at_ms, model_context)); + if !first_notification { + return; + } + let Some((server, tool, plugin_id, connector_id)) = cua_identity else { + return; + }; + let Some((connection_state, thread_state, thread_metadata)) = self + .thread_context_or_warn(AnalyticsDropSite::cua_tool_call_started( + ¬ification.thread_id, + ¬ification.turn_id, + item_id, + )) + else { + return; + }; + let attribution = tool_response_attribution( + item_id, + /*cell_id*/ None, + self.tool_response_states + .get(&(notification.thread_id.clone(), notification.turn_id.clone())), + self.code_mode_cells.get(¬ification.thread_id), + ); + let root_turn_id = self + .turns + .get(¬ification.turn_id) + .and_then(|turn| turn.resolved_config.as_ref()) + .and_then(|config| config.turn_metadata.root_turn_id()); + out.push(TrackEventRequest::CuaToolCallStarted( + CodexCuaToolCallStartedEventRequest { + event_type: "codex_cua_tool_call_started", + event_params: CodexCuaToolCallStartedEventParams { + thread_id: notification.thread_id, + session_id: thread_metadata.session_id.clone(), + turn_id: notification.turn_id, + root_turn_id, + item_id: item_id.to_string(), + cell_id: attribution.cell_id, + parent_call_id: attribution.parent_call_id, + originating_response_id: attribution.originating_response_id, + app_server_client: thread_state.app_server_client(connection_state), + runtime: connection_state.runtime.clone(), + thread_source: thread_metadata.thread_source.clone(), + subagent_source: thread_metadata.subagent_source.clone(), + parent_thread_id: thread_metadata.parent_thread_id.clone(), + tool_event_type: attribution.tool_event_type, + started_at_ms, + mcp_server_name: server.to_string(), + mcp_tool_name: tool.to_string(), + plugin_id: plugin_id.map(str::to_string), + connector_id: connector_id.map(str::to_string), + }, + }, + )); } ServerNotification::ItemCompleted(notification) => { if matches!(notification.item, ThreadItem::SubAgentActivity { .. }) { @@ -2090,6 +2168,15 @@ impl AnalyticsReducer { turn_id: notification.turn_id.clone(), item_id: item_id.to_string(), }; + let is_cua = cua_mcp_tool_identity(¬ification.item).is_some(); + if is_cua + && self + .turns + .get(¬ification.turn_id) + .is_some_and(|turn| turn.completed_cua_tool_item_ids.contains(item_id)) + { + return; + } let elicitation_type = if matches!(notification.item, ThreadItem::McpToolCall { .. }) { // Producers enqueue known classifications before completion on @@ -2172,6 +2259,9 @@ impl AnalyticsReducer { emission, out, ); + if is_cua && let Some(turn) = self.turns.get_mut(¬ification.turn_id) { + turn.completed_cua_tool_item_ids.insert(item_id.to_string()); + } } self.item_review_summaries.remove(&key); if self @@ -2837,6 +2927,28 @@ pub(crate) fn tracked_tool_item_id(item: &ThreadItem) -> Option<&str> { } } +fn cua_mcp_tool_identity(item: &ThreadItem) -> Option<(&str, &str, Option<&str>, Option<&str>)> { + let ThreadItem::McpToolCall { + server, + tool, + plugin_id, + app_context, + .. + } = item + else { + return None; + }; + let connector_id = app_context + .as_ref() + .map(|app_context| app_context.connector_id.as_str()); + (tool == "js" && is_node_repl_backed_connector(server, connector_id)).then_some(( + server, + tool, + plugin_id.as_deref(), + connector_id, + )) +} + fn tool_event_base_mut(event: &mut TrackEventRequest) -> Option<&mut CodexToolItemEventBase> { match event { TrackEventRequest::CommandExecution(event) => Some(&mut event.event_params.base), @@ -2859,36 +2971,52 @@ fn enrich_tool_response_event( let Some(base) = tool_event_base_mut(event) else { return; }; + let attribution = + tool_response_attribution(&base.item_id, base.cell_id.clone(), Some(state), cells); + base.tool_event_type = attribution.tool_event_type; + base.cell_id = attribution.cell_id; + base.parent_call_id = attribution.parent_call_id; + base.originating_response_id = attribution.originating_response_id; +} + +#[derive(Default)] +struct ToolResponseAttribution { + tool_event_type: Option, + cell_id: Option, + parent_call_id: Option, + originating_response_id: Option, +} + +fn tool_response_attribution( + item_id: &str, + cell_id: Option, + state: Option<&ToolResponseState>, + cells: Option<&HashMap>, +) -> ToolResponseAttribution { // A cell association can also describe a separately sampled wait call. Classify // only from evidence about this exact call ID, not its parent or response lineage. - base.tool_event_type = match ( - state.response_ids_by_call_id.contains_key(&base.item_id), - state.cell_ids_by_child_call_id.contains_key(&base.item_id), + let tool_event_type = match ( + state.is_some_and(|state| state.response_ids_by_call_id.contains_key(item_id)), + state.is_some_and(|state| state.cell_ids_by_child_call_id.contains_key(item_id)), ) { (true, false) => Some(ToolEventType::ModelToolCall), (false, true) => Some(ToolEventType::InnerToolCall), (false, false) | (true, true) => None, }; - 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() + let cell_id = cell_id + .or_else(|| state.and_then(|state| state.cell_ids_by_child_call_id.get(item_id).cloned())); + let cell = cell_id.as_ref().and_then(|cell_id| cells?.get(cell_id)); + let parent_call_id = + cell.and_then(|cell| (cell.parent_call_id != item_id).then(|| cell.parent_call_id.clone())); + let originating_response_id = state + .and_then(|state| state.response_ids_by_call_id.get(item_id).cloned()) .or_else(|| cell.and_then(|cell| cell.originating_response_id.clone())); + ToolResponseAttribution { + tool_event_type, + cell_id: cell.is_some().then_some(cell_id).flatten(), + parent_call_id, + originating_response_id, + } } fn item_review_summary_key(pending_review: &PendingReviewState) -> Option { diff --git a/codex-rs/analytics/src/tests/suite/tools_tests.rs b/codex-rs/analytics/src/tests/suite/tools_tests.rs index 6272b8458f..41e9497696 100644 --- a/codex-rs/analytics/src/tests/suite/tools_tests.rs +++ b/codex-rs/analytics/src/tests/suite/tools_tests.rs @@ -218,6 +218,100 @@ fn command_execution_event_serializes_expected_shape() { assert_eq!(payload, expected); } +#[tokio::test] +async fn cua_tool_call_lifecycle_requires_a_start_and_deduplicates_replay() { + let mut reducer = AnalyticsReducer::default(); + let mut events = Vec::new(); + ingest_turn_prerequisites( + &mut reducer, + &mut events, + /*include_initialize*/ true, + /*include_resolved_config*/ true, + /*include_started*/ true, + /*include_token_usage*/ false, + ) + .await; + events.clear(); + + let item = |status| ThreadItem::McpToolCall { + id: "cua-call".into(), + server: "codex_apps".into(), + tool: "js".into(), + status, + arguments: json!({"script": "synthetic-private-input"}), + app_context: Some(McpToolCallAppContext { + connector_id: "connector_openai_browser".into(), + link_id: None, + resource_uri: None, + app_name: None, + action_name: None, + }), + mcp_app_resource_uri: None, + mcp_app_ui: None, + plugin_id: Some("browser@openai-curated".into()), + read_only_hint: None, + result: None, + error: None, + duration_ms: None, + }; + let started = ServerNotification::ItemStarted(ItemStartedNotification { + thread_id: "thread-2".into(), + turn_id: "turn-2".into(), + started_at_ms: 1_000, + item: item(McpToolCallStatus::InProgress), + }); + let completed = ServerNotification::ItemCompleted(ItemCompletedNotification { + thread_id: "thread-2".into(), + turn_id: "turn-2".into(), + completed_at_ms: 1_100, + item: item(McpToolCallStatus::Completed), + }); + for notification in [ + completed.clone(), // Missing start must not poison later deduplication. + started.clone(), + started.clone(), + completed.clone(), + started, // Replay after a valid terminal. + completed, + ] { + reducer + .ingest( + AnalyticsFact::Notification(Box::new(notification)), + &mut events, + ) + .await; + } + + let [ + TrackEventRequest::CuaToolCallStarted(start), + TrackEventRequest::McpToolCall(terminal), + ] = events.as_slice() + else { + panic!("expected exactly one CUA start and one terminal"); + }; + assert_eq!( + ( + start.event_type, + start.event_params.connector_id.as_deref(), + start.event_params.plugin_id.as_deref(), + start.event_params.started_at_ms, + terminal.event_params.base.started_at_ms, + ), + ( + "codex_cua_tool_call_started", + Some("connector_openai_browser"), + Some("browser@openai-curated"), + 1_000, + 1_000, + ) + ); + assert!( + !serde_json::to_string(&events) + .expect("serialize CUA lifecycle events") + .contains("synthetic-private-input") + ); +} + #[tokio::test] async fn item_lifecycle_notifications_publish_command_execution_event() { let mut reducer = AnalyticsReducer::default();