From 49880081fe9d6cf89ec86e3d31c99bbe26a736c4 Mon Sep 17 00:00:00 2001 From: Eddie Chen <173840017+eddie-openai@users.noreply.github.com> Date: Mon, 24 Aug 2026 22:23:46 +0000 Subject: [PATCH] Add root turn IDs to turn and tool analytics (#40486) ## Why Analytics need to associate subagent activity with the top-level turn that caused it, while avoiding a stale association when steering makes that origin ambiguous. ## What changed - Add `root_turn_id` to turn and tool-item analytics events. - Read the root from live turn metadata when each event is built, reporting `null` when the root is unknown or ambiguous. - Propagate the metadata through control-tool and code-mode completion facts so correlated and background tool events retain the correct root. ## Testing - Cover root IDs for top-level turns, subagent turns, correlated code-mode calls, and tool events emitted after turn completion. - Verify that queued and subsequent events report `null` after the live root is invalidated. GitOrigin-RevId: 42fd76541c3c716cec74787d47b50adb2d22e6d2 --- .../analytics/src/analytics_client_tests.rs | 308 ++++++++++++++++-- codex-rs/analytics/src/client_tests.rs | 1 + codex-rs/analytics/src/events.rs | 2 + codex-rs/analytics/src/facts.rs | 17 +- codex-rs/analytics/src/lib.rs | 1 + codex-rs/analytics/src/reducer.rs | 19 +- .../app-server/tests/suite/v2/turn_start.rs | 105 +++++- codex-rs/core/src/session/turn.rs | 1 + .../src/tools/code_mode/execute_handler.rs | 1 + .../core/src/tools/code_mode/telemetry.rs | 6 + .../core/src/tools/code_mode/wait_handler.rs | 1 + .../core/src/tools/control_tool_analytics.rs | 1 + codex-rs/core/src/turn_metadata.rs | 6 + 13 files changed, 439 insertions(+), 30 deletions(-) diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index b535c1a34d..0da520714f 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -61,6 +61,8 @@ use crate::facts::CompactionReason; use crate::facts::CompactionStatus; use crate::facts::CompactionStrategy; use crate::facts::CompactionTrigger; +use crate::facts::ControlToolCallFact; +use crate::facts::ControlToolCallStatus; use crate::facts::CustomAnalyticsFact; use crate::facts::ExternalAgentConfigImportCompletedInput; use crate::facts::ExternalAgentConfigImportFailureInput; @@ -88,6 +90,7 @@ use crate::facts::SkillInvokedInput; use crate::facts::SubAgentThreadStartedInput; use crate::facts::ThreadInitializationMode; use crate::facts::TrackEventsContext; +use crate::facts::TurnAnalyticsMetadata; use crate::facts::TurnCodexErrorFact; use crate::facts::TurnProfile; use crate::facts::TurnProfileFact; @@ -198,6 +201,22 @@ use tokio::sync::mpsc; const TEST_PRODUCT_CLIENT_ID: &str = "codex_work_desktop"; +struct TestTurnMetadata { + root_turn_id: Mutex>, +} + +impl TurnAnalyticsMetadata for TestTurnMetadata { + fn root_turn_id(&self) -> Option { + self.root_turn_id.lock().expect("root turn ID").clone() + } +} + +fn test_turn_metadata(root_turn_id: Option<&str>) -> Arc { + Arc::new(TestTurnMetadata { + root_turn_id: Mutex::new(root_turn_id.map(str::to_string)), + }) +} + fn test_tracking_context(thread_id: &str, turn_id: &str) -> TrackEventsContext { TrackEventsContext { model_slug: "gpt-5".to_string(), @@ -438,6 +457,7 @@ fn sample_turn_resolved_config(thread_id: &str, turn_id: &str) -> TurnResolvedCo TurnResolvedConfigFact { turn_id: turn_id.to_string(), thread_id: thread_id.to_string(), + turn_metadata: test_turn_metadata(/*root_turn_id*/ None), num_input_images: 1, submission_type: None, ephemeral: false, @@ -1692,6 +1712,7 @@ fn command_execution_event_serializes_expected_shape() { thread_id: "thread-1".to_string(), session_id: "session-thread-1".to_string(), turn_id: "turn-1".to_string(), + root_turn_id: Some("root-turn".to_string()), item_id: "item-1".to_string(), cell_id: None, parent_call_id: None, @@ -1748,6 +1769,7 @@ fn command_execution_event_serializes_expected_shape() { "thread_id": "thread-1", "session_id": "session-thread-1", "turn_id": "turn-1", + "root_turn_id": "root-turn", "item_id": "item-1", "cell_id": null, "parent_call_id": null, @@ -2873,18 +2895,31 @@ 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; + reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::TurnResolvedConfig(Box::new( + TurnResolvedConfigFact { + turn_metadata: test_turn_metadata(Some("root-a")), + ..sample_turn_resolved_config("thread-1", "turn-1") + }, + ))), + &mut events, + ) + .await; - let completed = - |turn_id: &str, call_id: &str, tool_name: &str| CodeModeToolCallFact::Completed { + let completed = |turn_id: &str, root_turn_id: &str, call_id: &str, tool_name: &str| { + CodeModeToolCallFact::Completed { thread_id: "thread-1".into(), turn_id: turn_id.into(), + turn_metadata: test_turn_metadata(Some(root_turn_id)), 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, @@ -2902,7 +2937,7 @@ async fn code_mode_exec_wait_and_child_events_share_cell_and_response_ids() { call_id: "child-1".into(), cell_id: "cell-1".into(), }, - completed("turn-1", "exec-1", "exec"), + completed("turn-1", "root-a", "exec-1", "exec"), ], ) .await; @@ -2915,7 +2950,7 @@ async fn code_mode_exec_wait_and_child_events_share_cell_and_response_ids() { [ sampling_response("turn-1", "resp-b", &[]), sampling_response("turn-2", "resp-c", &["wait-1"]), - completed("turn-2", "wait-1", "wait"), + completed("turn-2", "root-b", "wait-1", "wait"), sampling_response("turn-2", "resp-d", &[]), ], ) @@ -2927,6 +2962,7 @@ async fn code_mode_exec_wait_and_child_events_share_cell_and_response_ids() { let event = serde_json::to_value(event).expect("serialize tool event"); serde_json::json!({ "item": event["event_params"]["item_id"], + "root": event["event_params"]["root_turn_id"], "cell": event["event_params"]["cell_id"], "parent": event["event_params"]["parent_call_id"], "origin": event["event_params"]["originating_response_id"], @@ -2937,9 +2973,9 @@ async fn code_mode_exec_wait_and_child_events_share_cell_and_response_ids() { 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"}), + serde_json::json!({"item":"exec-1","root":"root-a","cell":"cell-1","parent":null,"origin":"resp-a","subsequent":"resp-b"}), + serde_json::json!({"item":"child-1","root":"root-a","cell":"cell-1","parent":"exec-1","origin":"resp-a","subsequent":"resp-b"}), + serde_json::json!({"item":"wait-1","root":"root-b","cell":"cell-1","parent":"exec-1","origin":"resp-c","subsequent":"resp-d"}), ] ); } @@ -3636,6 +3672,22 @@ async fn subagent_tool_items_inherit_parent_connection_metadata() { ) .await; ingest_review_prerequisites(&mut reducer, &mut events).await; + for (thread_id, turn_id, root_turn_id) in [ + ("thread-1", "turn-parent", "parent-current-root"), + ("thread-subagent", "turn-subagent", "child-causal-root"), + ] { + reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::TurnResolvedConfig(Box::new( + TurnResolvedConfigFact { + turn_metadata: test_turn_metadata(Some(root_turn_id)), + ..sample_turn_resolved_config(thread_id, turn_id) + }, + ))), + &mut events, + ) + .await; + } reducer .ingest( AnalyticsFact::Notification(Box::new(sample_turn_started_notification( @@ -3687,6 +3739,7 @@ async fn subagent_tool_items_inherit_parent_connection_metadata() { [CodeModeToolCallFact::Completed { thread_id: "thread-subagent".into(), turn_id: "turn-subagent".into(), + turn_metadata: test_turn_metadata(Some("child-causal-root")), call_id: "exec-1".into(), cell_id: None, tool_name: "exec".into(), @@ -3700,6 +3753,21 @@ async fn subagent_tool_items_inherit_parent_connection_metadata() { let payload = serde_json::to_value(&events).expect("serialize events"); assert_eq!(payload.as_array().expect("events array").len(), 2); + assert_eq!( + payload + .as_array() + .expect("events array") + .iter() + .map(|event| json!({ + "turn_id": event["event_params"]["turn_id"], + "root_turn_id": event["event_params"]["root_turn_id"], + })) + .collect::>(), + vec![ + json!({"turn_id": "turn-subagent", "root_turn_id": "child-causal-root"}), + json!({"turn_id": "turn-subagent", "root_turn_id": "child-causal-root"}), + ] + ); assert_eq!(payload[0]["event_type"], "codex_command_execution_event"); assert_eq!(payload[0]["event_params"]["thread_id"], "thread-subagent"); assert_eq!(payload[0]["event_params"]["session_id"], "session-thread-1"); @@ -4520,6 +4588,7 @@ fn turn_event_serializes_expected_shape() { thread_id: "thread-2".to_string(), session_id: "session-thread-2".to_string(), turn_id: "turn-2".to_string(), + root_turn_id: Some("turn-2".to_string()), app_server_client: sample_app_server_client_metadata(), runtime: sample_runtime_metadata(), submission_type: None, @@ -4593,6 +4662,7 @@ fn turn_event_serializes_expected_shape() { "thread_id": "thread-2", "session_id": "session-thread-2", "turn_id": "turn-2", + "root_turn_id": "turn-2", "submission_type": null, "app_server_client": { "product_client_id": "codex_cli_rs", @@ -5005,11 +5075,22 @@ async fn turn_event_counts_completed_tool_items() { &mut reducer, &mut out, /*include_initialize*/ true, - /*include_resolved_config*/ true, + /*include_resolved_config*/ false, /*include_started*/ true, /*include_token_usage*/ false, ) .await; + reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::TurnResolvedConfig(Box::new( + TurnResolvedConfigFact { + turn_metadata: test_turn_metadata(Some("root-ancestor")), + ..sample_turn_resolved_config("thread-2", "turn-2") + }, + ))), + &mut out, + ) + .await; let mcp_tool_call_item = |status, duration_ms| ThreadItem::McpToolCall { id: "mcp-1".to_string(), @@ -5121,6 +5202,24 @@ async fn turn_event_counts_completed_tool_items() { .await; } + reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::ControlToolCall(ControlToolCallFact { + thread_id: "thread-2".to_string(), + turn_id: "turn-2".to_string(), + turn_metadata: test_turn_metadata(Some("root-ancestor")), + call_id: "control-1".to_string(), + cell_id: None, + tool_name: "update_plan".to_string(), + started_at_ms: 998, + completed_at_ms: 1_000, + status: ControlToolCallStatus::Completed, + })), + &mut out, + ) + .await; + reducer.flush(&mut out); + let payload = serde_json::to_value(&out).expect("serialize tool item events"); let emitted_tool_events = payload .as_array() @@ -5132,20 +5231,29 @@ async fn turn_event_counts_completed_tool_items() { event["event_params"]["session_id"] .as_str() .expect("tool item event session ID"), + event["event_params"]["turn_id"] + .as_str() + .expect("tool item event turn ID"), + event["event_params"]["root_turn_id"] + .as_str() + .expect("tool item event root turn ID"), ) }) .collect::>(); assert_eq!( emitted_tool_events, - vec![ - ("codex_command_execution_event", "session-thread-2"), - ("codex_file_change_event", "session-thread-2"), - ("codex_mcp_tool_call_event", "session-thread-2"), - ("codex_dynamic_tool_call_event", "session-thread-2"), - ("codex_collab_agent_tool_call_event", "session-thread-2"), - ("codex_web_search_event", "session-thread-2"), - ("codex_image_generation_event", "session-thread-2"), + [ + "codex_command_execution_event", + "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_control_tool_call_event", ] + .map(|event_type| (event_type, "session-thread-2", "turn-2", "root-ancestor")) + .to_vec() ); let mcp_tool_call_event = out @@ -5176,7 +5284,8 @@ async fn turn_event_counts_completed_tool_items() { .find(|event| matches!(event, TrackEventRequest::TurnEvent(_))) .expect("turn event should be emitted"); let payload = serde_json::to_value(turn_event).expect("serialize turn event"); - assert_eq!(payload["event_params"]["total_tool_call_count"], json!(8)); + assert_eq!(payload["event_params"]["root_turn_id"], "root-ancestor"); + assert_eq!(payload["event_params"]["total_tool_call_count"], json!(9)); assert_eq!(payload["event_params"]["shell_command_count"], json!(1)); assert_eq!(payload["event_params"]["file_change_count"], json!(1)); assert_eq!(payload["event_params"]["mcp_tool_call_count"], json!(1)); @@ -5189,20 +5298,129 @@ async fn turn_event_counts_completed_tool_items() { assert_eq!(payload["event_params"]["image_generation_count"], json!(1)); } +#[tokio::test] +async fn turn_and_tool_events_read_current_trusted_root() { + let mut reducer = AnalyticsReducer::default(); + let mut events = Vec::new(); + ingest_review_prerequisites(&mut reducer, &mut events).await; + let turn_metadata = test_turn_metadata(Some("root-ancestor")); + reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::TurnResolvedConfig(Box::new( + TurnResolvedConfigFact { + turn_metadata: turn_metadata.clone(), + ..sample_turn_resolved_config("thread-1", "turn-1") + }, + ))), + &mut events, + ) + .await; + + ingest_completed_command_execution_item( + &mut reducer, + &mut events, + "thread-1", + "before-conflict", + ) + .await; + let queued_control_call = ControlToolCallFact { + thread_id: "thread-1".to_string(), + turn_id: "turn-1".to_string(), + turn_metadata: turn_metadata.clone(), + call_id: "queued-control".to_string(), + cell_id: None, + tool_name: "view_image".to_string(), + started_at_ms: 998, + completed_at_ms: 1_000, + status: ControlToolCallStatus::Completed, + }; + let queued_code_mode_call = CodeModeToolCallFact::Completed { + thread_id: "thread-1".to_string(), + turn_id: "turn-1".to_string(), + turn_metadata: turn_metadata.clone(), + call_id: "queued-code-mode".to_string(), + cell_id: None, + tool_name: "exec".to_string(), + started_at_ms: 998, + completed_at_ms: 1_000, + status: CodeModeToolCallStatus::Completed, + }; + *turn_metadata.root_turn_id.lock().expect("root turn ID") = None; + ingest_completed_command_execution_item( + &mut reducer, + &mut events, + "thread-1", + "after-conflict", + ) + .await; + for fact in [ + AnalyticsFact::Custom(CustomAnalyticsFact::ControlToolCall(queued_control_call)), + AnalyticsFact::Custom(CustomAnalyticsFact::CodeModeToolCall(queued_code_mode_call)), + AnalyticsFact::Custom(CustomAnalyticsFact::TurnProfile(Box::new( + TurnProfileFact { + turn_id: "turn-1".to_string(), + profile: sample_turn_profile(), + }, + ))), + AnalyticsFact::Notification(Box::new(sample_turn_completed_notification( + "thread-1", + "turn-1", + AppServerTurnStatus::Completed, + /*codex_error_info*/ None, + ))), + ] { + reducer.ingest(fact, &mut events).await; + } + + let payload = serde_json::to_value(&events).expect("serialize events"); + assert_eq!( + payload + .as_array() + .expect("events array") + .iter() + .map(|event| json!({ + "event_type": event["event_type"], + "turn_id": event["event_params"]["turn_id"], + "item_id": event["event_params"].get("item_id"), + "root_turn_id": event["event_params"].get("root_turn_id").expect("root field"), + })) + .collect::>(), + vec![ + json!({"event_type": "codex_command_execution_event", "turn_id": "turn-1", "item_id": "before-conflict", "root_turn_id": "root-ancestor"}), + json!({"event_type": "codex_command_execution_event", "turn_id": "turn-1", "item_id": "after-conflict", "root_turn_id": null}), + json!({"event_type": "codex_control_tool_call_event", "turn_id": "turn-1", "item_id": "queued-control", "root_turn_id": null}), + json!({"event_type": "codex_dynamic_tool_call_event", "turn_id": "turn-1", "item_id": "queued-code-mode", "root_turn_id": null}), + json!({"event_type": "codex_turn_event", "turn_id": "turn-1", "item_id": null, "root_turn_id": null}), + ] + ); +} + #[tokio::test] async fn completed_background_tool_item_emits_after_turn_event() { let mut reducer = AnalyticsReducer::default(); let mut out = Vec::new(); + let turn_metadata = test_turn_metadata(Some("root-background")); ingest_turn_prerequisites( &mut reducer, &mut out, /*include_initialize*/ true, - /*include_resolved_config*/ true, + /*include_resolved_config*/ false, /*include_started*/ true, /*include_token_usage*/ false, ) .await; + reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::TurnResolvedConfig(Box::new( + TurnResolvedConfigFact { + turn_metadata: turn_metadata.clone(), + ..sample_turn_resolved_config("thread-2", "turn-2") + }, + ))), + &mut out, + ) + .await; reducer .ingest( @@ -5267,6 +5485,58 @@ async fn completed_background_tool_item_emits_after_turn_event() { out.iter() .any(|event| matches!(event, TrackEventRequest::CommandExecution(_))) ); + reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::ControlToolCall(ControlToolCallFact { + thread_id: "thread-2".to_string(), + turn_id: "turn-2".to_string(), + turn_metadata: turn_metadata.clone(), + call_id: "late-view-image".to_string(), + cell_id: None, + tool_name: "view_image".to_string(), + started_at_ms: 998, + completed_at_ms: 1_001, + status: ControlToolCallStatus::Completed, + })), + &mut out, + ) + .await; + ingest_code_mode_facts( + &mut reducer, + &mut out, + [CodeModeToolCallFact::Completed { + thread_id: "thread-2".to_string(), + turn_id: "turn-2".to_string(), + turn_metadata, + call_id: "late-code-mode".to_string(), + cell_id: None, + tool_name: "exec".to_string(), + started_at_ms: 998, + completed_at_ms: 1_001, + status: CodeModeToolCallStatus::Interrupted, + }], + ) + .await; + reducer.flush(&mut out); + + assert_eq!( + out.iter() + .map(|event| { + let event = serde_json::to_value(event).expect("serialize event"); + json!({ + "event_type": event["event_type"], + "turn_id": event["event_params"]["turn_id"], + "root_turn_id": event["event_params"]["root_turn_id"], + }) + }) + .collect::>(), + vec![ + json!({"event_type": "codex_turn_event", "turn_id": "turn-2", "root_turn_id": "root-background"}), + json!({"event_type": "codex_command_execution_event", "turn_id": "turn-2", "root_turn_id": "root-background"}), + json!({"event_type": "codex_control_tool_call_event", "turn_id": "turn-2", "root_turn_id": "root-background"}), + json!({"event_type": "codex_dynamic_tool_call_event", "turn_id": "turn-2", "root_turn_id": "root-background"}), + ] + ); } #[tokio::test] diff --git a/codex-rs/analytics/src/client_tests.rs b/codex-rs/analytics/src/client_tests.rs index 89070bd096..6e4ee4d1ec 100644 --- a/codex-rs/analytics/src/client_tests.rs +++ b/codex-rs/analytics/src/client_tests.rs @@ -194,6 +194,7 @@ fn sample_mcp_tool_call_event(thread_id: &str, plugin_id: Option<&str>) -> Track thread_id: thread_id.to_string(), session_id: format!("session-{thread_id}"), turn_id: "turn-1".to_string(), + root_turn_id: None, item_id: format!("item-{thread_id}"), cell_id: None, parent_call_id: None, diff --git a/codex-rs/analytics/src/events.rs b/codex-rs/analytics/src/events.rs index 6b7c159109..dd6b44bbd3 100644 --- a/codex-rs/analytics/src/events.rs +++ b/codex-rs/analytics/src/events.rs @@ -629,6 +629,7 @@ pub(crate) struct CodexToolItemEventBase { pub(crate) thread_id: String, pub(crate) session_id: String, pub(crate) turn_id: String, + pub(crate) root_turn_id: Option, /// App-server ThreadItem.id. For tool-originated items this generally /// corresponds to the originating core call_id. pub(crate) item_id: String, @@ -1004,6 +1005,7 @@ pub(crate) struct CodexTurnEventParams { pub(crate) thread_id: String, pub(crate) session_id: String, pub(crate) turn_id: String, + pub(crate) root_turn_id: Option, // TODO(rhan-oai): Populate once queued/default submission type is plumbed from // the turn/start callsites instead of always being reported as None. pub(crate) submission_type: Option, diff --git a/codex-rs/analytics/src/facts.rs b/codex-rs/analytics/src/facts.rs index e240f6c7b1..4616732383 100644 --- a/codex-rs/analytics/src/facts.rs +++ b/codex-rs/analytics/src/facts.rs @@ -34,6 +34,7 @@ use codex_protocol::request_permissions::RequestPermissionsResponse; use serde::Serialize; use std::collections::BTreeMap; use std::path::PathBuf; +use std::sync::Arc; #[derive(Clone)] pub struct TrackEventsContext { @@ -64,7 +65,7 @@ pub struct ArtifactOperation { pub execution_backend: String, } -#[derive(Clone, Debug, PartialEq, Eq)] +#[derive(Clone)] pub enum CodeModeToolCallFact { CellStarted { thread_id: String, @@ -92,6 +93,7 @@ pub enum CodeModeToolCallFact { Completed { thread_id: String, turn_id: String, + turn_metadata: Arc, call_id: String, cell_id: Option, tool_name: String, @@ -108,10 +110,11 @@ pub enum CodeModeToolCallStatus { Interrupted, } -#[derive(Clone, Debug, PartialEq, Eq)] +#[derive(Clone)] pub struct ControlToolCallFact { pub thread_id: String, pub turn_id: String, + pub turn_metadata: Arc, pub call_id: String, pub cell_id: Option, pub tool_name: String, @@ -182,6 +185,7 @@ pub enum TurnSubmissionType { pub struct TurnResolvedConfigFact { pub turn_id: String, pub thread_id: String, + pub turn_metadata: Arc, pub num_input_images: usize, pub submission_type: Option, pub ephemeral: bool, @@ -202,6 +206,15 @@ pub struct TurnResolvedConfigFact { pub is_first_turn: bool, } +/// A live, read-only view of a turn's trusted analytics provenance. +/// +/// Implementations must return `None` for unknown or ambiguous roots. The reducer +/// reads this when constructing each event because steering can invalidate a root +/// after the turn's configuration has been resolved. +pub trait TurnAnalyticsMetadata: Send + Sync { + fn root_turn_id(&self) -> Option; +} + #[derive(Clone, Copy, Debug, Serialize)] #[serde(rename_all = "snake_case")] pub enum ThreadInitializationMode { diff --git a/codex-rs/analytics/src/lib.rs b/codex-rs/analytics/src/lib.rs index 942bfd2b95..49a87ab845 100644 --- a/codex-rs/analytics/src/lib.rs +++ b/codex-rs/analytics/src/lib.rs @@ -60,6 +60,7 @@ pub use facts::SkillInvocationLocation; pub use facts::SubAgentThreadStartedInput; pub use facts::ThreadInitializationMode; pub use facts::TrackEventsContext; +pub use facts::TurnAnalyticsMetadata; pub use facts::TurnCodexErrorFact; pub use facts::TurnProfile; pub use facts::TurnProfileFact; diff --git a/codex-rs/analytics/src/reducer.rs b/codex-rs/analytics/src/reducer.rs index a25627c43c..54ce3df44a 100644 --- a/codex-rs/analytics/src/reducer.rs +++ b/codex-rs/analytics/src/reducer.rs @@ -778,6 +778,7 @@ impl AnalyticsReducer { CodeModeToolCallFact::Completed { thread_id, turn_id, + turn_metadata, call_id, cell_id, tool_name, @@ -842,6 +843,7 @@ impl AnalyticsReducer { self.record_tool_event( &thread_id, &turn_id, + turn_metadata.root_turn_id(), event, ToolEventEmission::AwaitResponse, out, @@ -858,6 +860,7 @@ impl AnalyticsReducer { let ControlToolCallFact { thread_id, turn_id, + turn_metadata, call_id, cell_id, tool_name, @@ -921,6 +924,7 @@ impl AnalyticsReducer { self.record_tool_event( &thread_id, &turn_id, + turn_metadata.root_turn_id(), event, ToolEventEmission::AwaitResponse, out, @@ -979,14 +983,16 @@ impl AnalyticsReducer { &mut self, thread_id: &str, turn_id: &str, + root_turn_id: Option, mut event: TrackEventRequest, emission: ToolEventEmission, out: &mut Vec, ) { - if tool_event_base_mut(&mut event).is_none() { + let Some(base) = tool_event_base_mut(&mut event) else { out.push(event); return; - } + }; + base.root_turn_id = root_turn_id; let state = self .tool_response_states .entry((thread_id.to_string(), turn_id.to_string())) @@ -1873,9 +1879,15 @@ impl AnalyticsReducer { thread_metadata, review_summary: self.item_review_summaries.get(&key), }) { + 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()); self.record_tool_event( ¬ification.thread_id, ¬ification.turn_id, + root_turn_id, event, ToolEventEmission::ImmediateUnlessCorrelated, out, @@ -2888,6 +2900,7 @@ fn tool_item_base( thread_id: thread_id.to_string(), session_id: thread_metadata.session_id.clone(), turn_id: turn_id.to_string(), + root_turn_id: None, item_id, cell_id: None, parent_call_id: None, @@ -3342,6 +3355,7 @@ fn codex_turn_event_params( let TurnResolvedConfigFact { turn_id: _resolved_turn_id, thread_id: _resolved_thread_id, + turn_metadata, num_input_images: _resolved_num_input_images, submission_type, ephemeral, @@ -3377,6 +3391,7 @@ fn codex_turn_event_params( thread_id, session_id: thread_metadata.session_id.clone(), turn_id, + root_turn_id: turn_metadata.root_turn_id(), app_server_client, runtime, submission_type, 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 d7faa5f4ab..baaab3c776 100644 --- a/codex-rs/app-server/tests/suite/v2/turn_start.rs +++ b/codex-rs/app-server/tests/suite/v2/turn_start.rs @@ -219,10 +219,14 @@ async fn turn_start_with_empty_input_runs_model_request() -> Result<()> { let server = create_mock_responses_server_sequence_unchecked(responses).await; let codex_home = TempDir::new()?; - MockResponsesConfig::new(&server.uri()).write(codex_home.path())?; + MockResponsesConfig::new(&server.uri()) + .with_root_config(&format!("chatgpt_base_url = \"{}\"", server.uri())) + .write(codex_home.path())?; + mount_analytics_capture(&server, codex_home.path()).await?; let mut mcp = TestAppServer::builder() .with_codex_home(codex_home.path()) + .without_managed_config() .build_initialized() .await?; @@ -267,6 +271,15 @@ async fn turn_start_with_empty_input_runs_model_request() -> Result<()> { [ThreadItem::AgentMessage { text, .. }] if text == "Done" )); + let event = wait_for_analytics_event(&server, DEFAULT_READ_TIMEOUT, "codex_turn_event").await?; + assert_eq!( + ( + event["event_params"]["turn_id"].as_str(), + event["event_params"].get("root_turn_id"), + ), + (Some(turn.id.as_str()), Some(&Value::Null)) + ); + let requests = server .received_requests() .await @@ -984,6 +997,7 @@ async fn turn_start_tracks_thread_originator_in_analytics() -> Result<()> { assert_eq!(event["event_params"]["thread_id"], thread.id); assert_eq!(event["event_params"]["session_id"], thread.session_id); assert_eq!(event["event_params"]["turn_id"], turn.id); + assert_eq!(event["event_params"]["root_turn_id"], turn.id); assert_eq!( event["event_params"]["app_server_client"]["product_client_id"], "codex_work_desktop" @@ -1084,7 +1098,7 @@ async fn code_mode_exec_emits_correlated_production_analytics() -> Result<()> { .await?; let params = ThreadStartParams::default(); let thread = app_server.start_thread(params).await?; - app_server + let turn = app_server .start_turn_and_wait_for_completion(TurnStartParams { thread_id: thread.thread.id, input: vec![V2UserInput::Text { @@ -1103,12 +1117,21 @@ async fn code_mode_exec_emits_correlated_production_analytics() -> Result<()> { .await?; assert_eq!( json!({ + "turnId": event["event_params"]["turn_id"], + "rootTurnId": event["event_params"]["root_turn_id"], "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}) + json!({ + "turnId": turn.turn.id, + "rootTurnId": turn.turn.id, + "tool": "exec", + "origin": "resp-1", + "subsequent": "resp-2", + "hasCell": true, + }) ); Ok(()) @@ -1175,7 +1198,7 @@ async fn turn_profile_tracks_blocking_tool_and_follow_up_sampling() -> Result<() }) .await?; - let _: TurnStartResponse = mcp + let TurnStartResponse { turn: first_turn } = mcp .request(|request_id| ClientRequest::TurnStart { request_id, params: TurnStartParams { @@ -1227,6 +1250,8 @@ async fn turn_profile_tracks_blocking_tool_and_follow_up_sampling() -> Result<() let params = &event["event_params"]; assert_eq!( json!({ + "turnId": params["turn_id"], + "rootTurnId": params["root_turn_id"], "toolBlockingIsPositive": params["tool_blocking_ms"] .as_u64() .is_some_and(|duration| duration > 0), @@ -1237,6 +1262,8 @@ async fn turn_profile_tracks_blocking_tool_and_follow_up_sampling() -> Result<() "status": params["status"], }), json!({ + "turnId": first_turn.id, + "rootTurnId": first_turn.id, "toolBlockingIsPositive": true, "samplingRequestCount": 2, "samplingRetryCount": 0, @@ -1272,8 +1299,15 @@ async fn turn_profile_tracks_blocking_tool_and_follow_up_sampling() -> Result<() }) .await?; let success = tool_name != "view_image"; + let turn_id = if call_id == "call1" { + &first_turn.id + } else { + &second_turn.turn.id + }; assert_eq!( json!({ + "turnId": event["event_params"]["turn_id"], + "rootTurnId": event["event_params"]["root_turn_id"], "tool": event["event_params"]["tool_name"], "success": event["event_params"]["success"], "status": event["event_params"]["terminal_status"], @@ -1282,6 +1316,8 @@ async fn turn_profile_tracks_blocking_tool_and_follow_up_sampling() -> Result<() .is_some(), }), json!({ + "turnId": turn_id, + "rootTurnId": turn_id, "tool": tool_name, "success": success, "status": if success { "completed" } else { "failed" }, @@ -1310,10 +1346,11 @@ async fn turn_profile_tracks_blocking_tool_and_follow_up_sampling() -> Result<() .await?; assert_eq!( json!({ + "rootTurnId": second_turn_event["event_params"]["root_turn_id"], "total": second_turn_event["event_params"]["total_tool_call_count"], "dynamic": second_turn_event["event_params"]["dynamic_tool_call_count"], }), - json!({"total": 5, "dynamic": 0}) + json!({"rootTurnId": second_turn.turn.id, "total": 5, "dynamic": 0}) ); Ok(()) @@ -3520,6 +3557,7 @@ async fn turn_start_emits_spawn_agent_item_with_model_metadata_v2() -> Result<() const CHILD_PROMPT: &str = "child: do work"; const PARENT_PROMPT: &str = "spawn a child and continue"; const SPAWN_CALL_ID: &str = "spawn-call-1"; + const CHILD_PLAN_CALL_ID: &str = "child-plan-call"; const REQUESTED_MODEL: &str = "gpt-5.2"; const REQUESTED_REASONING_EFFORT: ReasoningEffort = ReasoningEffort::Low; @@ -3544,18 +3582,32 @@ async fn turn_start_emits_spawn_agent_item_with_model_metadata_v2() -> Result<() ]), ) .await; - let _child_turn = responses::mount_sse_once_match( + let child_turn = responses::mount_sse_once_match( &server, |req: &wiremock::Request| { body_contains(req, CHILD_PROMPT) && !body_contains(req, SPAWN_CALL_ID) }, responses::sse(vec![ responses::ev_response_created("resp-child-1"), - responses::ev_assistant_message("msg-child-1", "child done"), + responses::ev_function_call( + CHILD_PLAN_CALL_ID, + "update_plan", + &json!({"plan": [{"step": "child work", "status": "completed"}]}).to_string(), + ), responses::ev_completed("resp-child-1"), ]), ) .await; + let _child_follow_up = responses::mount_sse_once_match( + &server, + |req: &wiremock::Request| body_contains(req, CHILD_PLAN_CALL_ID), + responses::sse(vec![ + responses::ev_response_created("resp-child-2"), + responses::ev_assistant_message("msg-child-2", "child done"), + responses::ev_completed("resp-child-2"), + ]), + ) + .await; let _parent_follow_up = responses::mount_sse_once_match( &server, |req: &wiremock::Request| body_contains(req, SPAWN_CALL_ID), @@ -3570,7 +3622,9 @@ async fn turn_start_emits_spawn_agent_item_with_model_metadata_v2() -> Result<() let codex_home = TempDir::new()?; MockResponsesConfig::new(&server.uri()) .enable_feature(Feature::Collab) + .with_root_config(&format!("chatgpt_base_url = \"{}\"", server.uri())) .write(codex_home.path())?; + mount_analytics_capture(&server, codex_home.path()).await?; let mut mcp = TestAppServer::builder() .with_codex_home(codex_home.path()) @@ -3689,6 +3743,43 @@ async fn turn_start_emits_spawn_agent_item_with_model_metadata_v2() -> Result<() assert_eq!(turn_completed.thread_id, thread.id); assert_eq!(turn_completed.turn.id, turn.turn.id); + let child_turn_event = + wait_for_matching_analytics_event(&server, DEFAULT_READ_TIMEOUT, |event| { + event["event_type"] == "codex_turn_event" + && event["event_params"]["thread_id"] == receiver_thread_id + }) + .await?; + let child_tool_event = + wait_for_matching_analytics_event(&server, DEFAULT_READ_TIMEOUT, |event| { + event["event_type"] == "codex_control_tool_call_event" + && event["event_params"]["item_id"] == CHILD_PLAN_CALL_ID + }) + .await?; + let child_request = child_turn + .requests() + .into_iter() + .find(|request| request.body_json()["client_metadata"]["thread_id"] == receiver_thread_id) + .context("child should issue a model request")? + .body_json(); + let child_turn_id = child_request["client_metadata"]["turn_id"] + .as_str() + .context("child request should have a turn id")?; + assert_ne!(child_turn_id, turn.turn.id); + for event in [child_turn_event, child_tool_event] { + assert_eq!( + json!({ + "thread_id": event["event_params"]["thread_id"], + "turn_id": event["event_params"]["turn_id"], + "root_turn_id": event["event_params"]["root_turn_id"], + }), + json!({ + "thread_id": receiver_thread_id, + "turn_id": child_turn_id, + "root_turn_id": turn.turn.id, + }) + ); + } + // Reuse this live spawn setup to cover thread/delete's ThreadManager descendant path. let _: ThreadDeleteResponse = mcp .request(|request_id| ClientRequest::ThreadDelete { diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index d3dd0b8098..244c116de5 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -972,6 +972,7 @@ async fn track_turn_resolved_config_analytics( .track_turn_resolved_config(TurnResolvedConfigFact { turn_id: turn_context.sub_id.clone(), thread_id: sess.thread_id.to_string(), + turn_metadata: turn_context.turn_metadata_state.clone(), num_input_images: input .iter() .filter_map(|item| match item { 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 87d260be4e..648fff7ecb 100644 --- a/codex-rs/core/src/tools/code_mode/execute_handler.rs +++ b/codex-rs/core/src/tools/code_mode/execute_handler.rs @@ -168,6 +168,7 @@ impl CodeModeExecuteHandler { session.services.analytics_events_client.clone(), session.thread_id.to_string(), turn.sub_id.clone(), + turn.turn_metadata_state.clone(), call_id.clone(), PUBLIC_TOOL_NAME, ); diff --git a/codex-rs/core/src/tools/code_mode/telemetry.rs b/codex-rs/core/src/tools/code_mode/telemetry.rs index 9d1b2584db..b2f8146a67 100644 --- a/codex-rs/core/src/tools/code_mode/telemetry.rs +++ b/codex-rs/core/src/tools/code_mode/telemetry.rs @@ -1,11 +1,14 @@ use codex_analytics::AnalyticsEventsClient; use codex_analytics::CodeModeToolCallFact; use codex_analytics::CodeModeToolCallStatus; +use codex_analytics::TurnAnalyticsMetadata; +use std::sync::Arc; pub(super) struct CodeModeToolCallGuard { analytics: AnalyticsEventsClient, thread_id: String, turn_id: String, + turn_metadata: Arc, call_id: String, pub(super) cell_id: Option, tool_name: &'static str, @@ -18,6 +21,7 @@ impl CodeModeToolCallGuard { analytics: AnalyticsEventsClient, thread_id: String, turn_id: String, + turn_metadata: Arc, call_id: String, tool_name: &'static str, ) -> Self { @@ -25,6 +29,7 @@ impl CodeModeToolCallGuard { analytics, thread_id, turn_id, + turn_metadata, call_id, cell_id: None, tool_name, @@ -48,6 +53,7 @@ impl Drop for CodeModeToolCallGuard { .track_code_mode_tool_call(CodeModeToolCallFact::Completed { thread_id: self.thread_id.clone(), turn_id: self.turn_id.clone(), + turn_metadata: self.turn_metadata.clone(), call_id: self.call_id.clone(), cell_id: self.cell_id.clone(), tool_name: self.tool_name.to_string(), 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 4f65cef15e..63bd28dea4 100644 --- a/codex-rs/core/src/tools/code_mode/wait_handler.rs +++ b/codex-rs/core/src/tools/code_mode/wait_handler.rs @@ -77,6 +77,7 @@ impl CodeModeWaitHandler { session.services.analytics_events_client.clone(), session.thread_id.to_string(), turn.sub_id.clone(), + turn.turn_metadata_state.clone(), call_id.clone(), WAIT_TOOL_NAME, ); diff --git a/codex-rs/core/src/tools/control_tool_analytics.rs b/codex-rs/core/src/tools/control_tool_analytics.rs index 46719201e6..bfde9b24f5 100644 --- a/codex-rs/core/src/tools/control_tool_analytics.rs +++ b/codex-rs/core/src/tools/control_tool_analytics.rs @@ -36,6 +36,7 @@ impl Drop for ControlToolCallGuard<'_> { .track_control_tool_call(ControlToolCallFact { thread_id: invocation.session.thread_id.to_string(), turn_id: invocation.turn.sub_id.clone(), + turn_metadata: invocation.turn.turn_metadata_state.clone(), call_id: invocation.call_id.clone(), cell_id: match &invocation.source { ToolCallSource::CodeMode { cell_id, .. } => Some(cell_id.clone()), diff --git a/codex-rs/core/src/turn_metadata.rs b/codex-rs/core/src/turn_metadata.rs index 7e92aca521..478afaa53d 100644 --- a/codex-rs/core/src/turn_metadata.rs +++ b/codex-rs/core/src/turn_metadata.rs @@ -130,6 +130,12 @@ pub(crate) struct TurnMetadataState { git_enrichment_complete: watch::Sender, } +impl codex_analytics::TurnAnalyticsMetadata for TurnMetadataState { + fn root_turn_id(&self) -> Option { + TurnMetadataState::root_turn_id(self) + } +} + impl TurnMetadataState { #[allow(clippy::too_many_arguments)] pub(crate) fn new(