openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-27 18:48:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Track CUA tool call starts and deduplicate lifecycle events (#57167)

Browse Source 
## 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
This commit is contained in:
trpcic-openai
2026-10-26 17:57:13 +00:00
committed by copyberry
parent 2d3de37834
commit c21153bd92
4 changed files with 281 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/analytics/src/events.rs
+31
View File
@@ -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<String>,
pub(crate) item_id: String,
pub(crate) cell_id: Option<String>,
pub(crate) parent_call_id: Option<String>,
pub(crate) originating_response_id: Option<String>,
pub(crate) app_server_client: CodexAppServerClientMetadata,
pub(crate) runtime: CodexRuntimeMetadata,
pub(crate) thread_source: Option<ThreadSource>,
pub(crate) subagent_source: Option<String>,
pub(crate) parent_thread_id: Option<String>,
pub(crate) tool_event_type: Option<ToolEventType>,
pub(crate) started_at_ms: u64,
pub(crate) mcp_server_name: String,
pub(crate) mcp_tool_name: String,
pub(crate) plugin_id: Option<String>,
pub(crate) connector_id: Option<String>,
}
#[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")]
codex-rs/analytics/src/product_attribution.rs
+1
View File
@@ -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),
codex-rs/analytics/src/reducer.rs
+155 -27
View File
@@ -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<String>,
steer_count: usize,
tool_counts: TurnToolCounts,
completed_cua_tool_item_ids: HashSet<String>,
resource_skill_invocations: HashSet<String>,
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(&notification.item);
if cua_identity.is_some()
&& self
.turns
.get(&notification.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(
&notification.thread_id,
&notification.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(&notification.thread_id),
);
let root_turn_id = self
.turns
.get(&notification.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(&notification.item).is_some();
if is_cua
&& self
.turns
.get(&notification.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(&notification.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<ToolEventType>,
cell_id: Option<String>,
parent_call_id: Option<String>,
originating_response_id: Option<String>,
}
fn tool_response_attribution(
item_id: &str,
cell_id: Option<String>,
state: Option<&ToolResponseState>,
cells: Option<&HashMap<String, CodeModeCellState>>,
) -> 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<ToolItemKey> {
codex-rs/analytics/src/tests/suite/tools_tests.rs
+94
View File
@@ -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();