Preserve response IDs for fast collaborator tool events (#42752)

## Why

Collaborator tools can finish before their sampling response arrives, causing
their analytics events to be emitted without response correlation.

## What changed

Hold completed collaborator tool events until the reducer can attach their
originating and subsequent response IDs.

## Testing

Add coverage for both orderings of collaborator completion and sampling
response arrival.

GitOrigin-RevId: 5933424dd90702685cdb4a9693fed96c7885989a
This commit is contained in:
jif
2026-09-04 12:39:40 +00:00
committed by copyberry
parent a1294e57f1
commit 1cd78651e2
2 changed files with 96 additions and 2 deletions
@@ -2937,6 +2937,92 @@ fn sampling_response(
}
}
#[tokio::test]
async fn collaborator_tool_events_keep_response_ids_when_completion_races_sampling() {
for response_first in [false, true] {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
ingest_review_prerequisites(&mut reducer, &mut events).await;
reducer
.ingest(
AnalyticsFact::Notification(Box::new(sample_turn_started_notification(
"thread-1", "turn-1",
))),
&mut events,
)
.await;
let item = ThreadItem::CollabAgentToolCall {
id: "call-1".into(),
tool: CollabAgentTool::SendMessage,
status: CollabAgentToolCallStatus::Failed,
sender_thread_id: "thread-1".into(),
receiver_thread_ids: Vec::new(),
prompt: None,
model: None,
reasoning_effort: None,
agents_states: Default::default(),
};
reducer
.ingest(
AnalyticsFact::Notification(Box::new(ServerNotification::ItemStarted(
ItemStartedNotification {
thread_id: "thread-1".into(),
turn_id: "turn-1".into(),
started_at_ms: 1_000,
item: item.clone(),
},
))),
&mut events,
)
.await;
let response = AnalyticsFact::Custom(CustomAnalyticsFact::CodeModeToolCall(
sampling_response("turn-1", "response-1", &["call-1"]),
));
let completion = AnalyticsFact::Notification(Box::new(ServerNotification::ItemCompleted(
ItemCompletedNotification {
thread_id: "thread-1".into(),
turn_id: "turn-1".into(),
completed_at_ms: 1_010,
item,
},
)));
let facts = if response_first {
[response, completion]
} else {
[completion, response]
};
for fact in facts {
reducer.ingest(fact, &mut events).await;
assert!(events.is_empty(), "emitted before response correlation");
}
ingest_code_mode_facts(
&mut reducer,
&mut events,
[sampling_response("turn-1", "response-2", &[])],
)
.await;
let payload = serde_json::to_value(&events).expect("serialize collaborator event");
assert_eq!(payload.as_array().expect("events array").len(), 1);
let params = &payload[0]["event_params"];
assert_eq!(
json!({
"type": payload[0]["event_type"],
"item": params["item_id"],
"origin": params["originating_response_id"],
"subsequent": params["subsequent_response_id"],
"status": params["terminal_status"],
}),
json!({
"type": "codex_collab_agent_tool_call_event",
"item": "call-1",
"origin": "response-1",
"subsequent": "response-2",
"status": "failed",
}),
);
}
}
#[tokio::test]
async fn code_mode_exec_wait_and_child_events_share_cell_and_response_ids() {
let mut reducer = AnalyticsReducer::default();
@@ -5413,9 +5499,9 @@ async fn turn_event_counts_completed_tool_items() {
"codex_file_change_event",
"codex_mcp_tool_call_event",
"codex_dynamic_tool_call_event",
"codex_collab_agent_tool_call_event",
"codex_web_search_event",
"codex_image_generation_event",
"codex_collab_agent_tool_call_event",
"codex_control_tool_call_event",
]
.map(|event_type| (event_type, "session-thread-2", "turn-2", "root-ancestor"))
+9 -1
View File
@@ -1966,12 +1966,20 @@ impl AnalyticsReducer {
.get(&notification.turn_id)
.and_then(|turn| turn.resolved_config.as_ref())
.and_then(|config| config.turn_metadata.root_turn_id());
// Fast collaborator tools can complete before their sampling response.
// Keep the event until that response can supply its originating ID.
let emission =
if matches!(&notification.item, ThreadItem::CollabAgentToolCall { .. }) {
ToolEventEmission::AwaitResponse
} else {
ToolEventEmission::ImmediateUnlessCorrelated
};
self.record_tool_event(
&notification.thread_id,
&notification.turn_id,
root_turn_id,
event,
ToolEventEmission::ImmediateUnlessCorrelated,
emission,
out,
);
}