github/codex
4
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-20 00:18:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

Correlate code mode tool analytics with model responses (#36729)

Browse Source 
## 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
This commit is contained in:
jif
2026-08-03 10:19:04 +00:00
committed by copyberry
parent dae2122214
commit 155f1ca9e5
13 changed files with 797 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/analytics/src/analytics_client_tests.rs
+114
View File
@@ -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<TrackEventRequest>,
facts: impl IntoIterator<Item = CodeModeToolCallFact>,
) {
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::<Vec<_>>();
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();
codex-rs/analytics/src/client.rs
+11 -1
View File
@@ -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(_)
codex-rs/analytics/src/client_tests.rs
+4
View File
@@ -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,
codex-rs/analytics/src/events.rs
+4
View File
@@ -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<String>,
pub(crate) parent_call_id: Option<String>,
pub(crate) originating_response_id: Option<String>,
pub(crate) subsequent_response_id: Option<String>,
pub(crate) app_server_client: CodexAppServerClientMetadata,
pub(crate) runtime: CodexRuntimeMetadata,
pub(crate) thread_source: Option<ThreadSource>,
codex-rs/analytics/src/facts.rs
+45
View File
@@ -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<String>,
},
Completed {
thread_id: String,
turn_id: String,
call_id: String,
cell_id: Option<String>,
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<CodexCompactionEvent>),
Goal(Box<CodexGoalEvent>),
codex-rs/analytics/src/lib.rs
+2
View File
@@ -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;
codex-rs/analytics/src/reducer.rs
+369 -2
View File
@@ -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<u64, ConnectionState>,
threads: HashMap<String, ThreadAnalyticsState>,
tool_items_started_at_ms: HashMap<ToolItemKey, u64>,
tool_response_states: HashMap<(String, String), ToolResponseState>,
code_mode_cells: HashMap<String, HashMap<String, CodeModeCellState>>,
pending_reviews: HashMap<RequestId, PendingReviewState>,
item_review_summaries: HashMap<ToolItemKey, ItemReviewSummary>,
}
@@ -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<String, String>,
cell_ids_by_child_call_id: HashMap<String, String>,
pending_tool_events: VecDeque<TrackEventRequest>,
}
struct CodeModeCellState {
parent_call_id: String,
originating_response_id: Option<String>,
closed_in_turn_id: Option<String>,
}
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<TrackEventRequest>,
) {
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<String>,
out: &mut Vec<TrackEventRequest>,
) {
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<TrackEventRequest>,
) {
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<TrackEventRequest>,
) {
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<TrackEventRequest>) {
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(
&notification.thread_id,
&notification.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 == &notification.thread_id {
out.extend(state.pending_tool_events.drain(..));
false
} else {
true
}
});
self.code_mode_cells.remove(&notification.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(&notification.thread_id, &notification.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<String, CodeModeCellState>>,
) {
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<ToolItemKey> {
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),
codex-rs/app-server/tests/suite/v2/turn_start.rs
+59
View File
@@ -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![
codex-rs/core/src/session/turn.rs
+28
View File
@@ -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,
codex-rs/core/src/tools/code_mode/execute_handler.rs
+36 -3
View File
@@ -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<crate::session::turn_context::TurnContext>,
call_id: String,
code: String,
telemetry: &mut CodeModeToolCallGuard,
) -> Result<FunctionToolOutput, FunctionCallError> {
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
}
}
codex-rs/core/src/tools/code_mode/mod.rs
+11 -1
View File
@@ -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,
codex-rs/core/src/tools/code_mode/telemetry.rs
+59
View File
@@ -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<String>,
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,
});
}
}
codex-rs/core/src/tools/code_mode/wait_handler.rs
+55 -28
View File
@@ -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
}
}