From 8ace915aced81ed841e34fa069b2e489c324731c Mon Sep 17 00:00:00 2001 From: Chris Dong Date: Wed, 16 Sep 2026 23:22:31 +0000 Subject: [PATCH] Attribute analytics events to realtime voice sessions (#46058) ## What changed Add `voice_session_id` to turn, app-use, skill-invocation, and MCP tool-call analytics events. Track realtime session lifecycle and handoff markers to preserve attribution when a handoff starts a turn after the voice session closes, or steers an active turn. Keep that attribution from carrying over to subsequent text turns. Record handoff markers using only the thread ID, without retaining transcript content in analytics facts. ## Testing Add coverage for attribution after realtime closure, active-turn steering without tagging the next text turn, and handoff markers that exclude transcripts. Extend event serialization and MCP tool-call tests for `voice_session_id`. GitOrigin-RevId: 024b0e7e4c6fbaf8325574de6c24c048bb4ca032 --- .../analytics/src/analytics_client_tests.rs | 263 ++++++++++++++++++ codex-rs/analytics/src/client.rs | 12 + codex-rs/analytics/src/client_tests.rs | 31 +++ codex-rs/analytics/src/events.rs | 4 + codex-rs/analytics/src/facts.rs | 3 + codex-rs/analytics/src/reducer.rs | 67 ++++- 6 files changed, 379 insertions(+), 1 deletion(-) diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index d65451cd80..3e81ed54eb 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -151,6 +151,8 @@ use codex_app_server_protocol::ThreadArchiveParams; use codex_app_server_protocol::ThreadArchiveResponse; use codex_app_server_protocol::ThreadArchivedNotification; use codex_app_server_protocol::ThreadItem; +use codex_app_server_protocol::ThreadRealtimeClosedNotification; +use codex_app_server_protocol::ThreadRealtimeStartedNotification; use codex_app_server_protocol::ThreadResumeResponse; use codex_app_server_protocol::ThreadSource as AppServerThreadSource; use codex_app_server_protocol::ThreadStartResponse; @@ -189,6 +191,7 @@ use codex_protocol::protocol::HookExecutionMode; use codex_protocol::protocol::HookHandlerType; use codex_protocol::protocol::HookRunStatus; use codex_protocol::protocol::HookSource; +use codex_protocol::protocol::RealtimeConversationVersion; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SubAgentSource; use codex_protocol::protocol::ThreadSource; @@ -1200,6 +1203,7 @@ fn app_used_event_serializes_expected_shape() { invocation_type: Some(InvocationType::Implicit), }, ), + voice_session_id: None, elicitation_type: None, }, }); @@ -1218,6 +1222,7 @@ fn app_used_event_serializes_expected_shape() { "product_client_id": TEST_PRODUCT_CLIENT_ID, "invoke_type": "implicit", "model_slug": "gpt-5", + "voice_session_id": null, "elicitation_type": null } }) @@ -4600,6 +4605,7 @@ async fn reducer_ingests_skill_invoked_fact() { "remote_plugin_id": null, "thread_id": "thread-1", "turn_id": "turn-1", + "voice_session_id": null, "invoke_type": "explicit", "model_slug": "gpt-5" } @@ -4607,6 +4613,240 @@ async fn reducer_ingests_skill_invoked_fact() { ); } +#[tokio::test] +async fn voice_handoff_attributes_plugin_events_after_realtime_closes() { + 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*/ false, + /*include_token_usage*/ true, + ) + .await; + reducer + .ingest( + AnalyticsFact::Notification(Box::new(ServerNotification::ThreadRealtimeStarted( + ThreadRealtimeStartedNotification { + thread_id: "thread-2".to_string(), + realtime_session_id: Some("work-voice-123".to_string()), + version: RealtimeConversationVersion::V2, + }, + ))), + &mut events, + ) + .await; + reducer + .ingest( + AnalyticsFact::RealtimeHandoffRequested { + thread_id: "thread-2".to_string(), + }, + &mut events, + ) + .await; + reducer + .ingest( + AnalyticsFact::Notification(Box::new(ServerNotification::ThreadRealtimeClosed( + ThreadRealtimeClosedNotification { + thread_id: "thread-2".to_string(), + reason: None, + }, + ))), + &mut events, + ) + .await; + reducer + .ingest( + AnalyticsFact::Notification(Box::new(sample_turn_started_notification( + "thread-2", "turn-2", + ))), + &mut events, + ) + .await; + reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::AppUsed(AppUsedInput { + tracking: test_tracking_context("thread-2", "turn-2"), + app: AppInvocation { + connector_id: Some("drive".to_string()), + app_name: Some("Drive".to_string()), + invocation_type: Some(InvocationType::Implicit), + }, + elicitation_type: None, + })), + &mut events, + ) + .await; + reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::SkillInvoked(SkillInvokedInput { + tracking: test_tracking_context("thread-2", "turn-2"), + invocations: vec![SkillInvocation { + skill_name: "sample:doc".to_string(), + location: SkillInvocationLocation::Resource { + id: "resource-1".to_string(), + skill_id: Some("sample:doc".to_string()), + scope: None, + }, + plugin_id: Some("sample@test".to_string()), + remote_plugin_id: None, + invocation_type: InvocationType::Explicit, + }], + })), + &mut events, + ) + .await; + reducer + .ingest( + AnalyticsFact::Notification(Box::new(sample_turn_completed_notification( + "thread-2", + "turn-2", + AppServerTurnStatus::Completed, + /*codex_error_info*/ None, + ))), + &mut events, + ) + .await; + + for event_type in ["codex_app_used", "skill_invocation", "codex_turn_event"] { + let event = events + .iter() + .map(|event| serde_json::to_value(event).expect("serialize analytics event")) + .find(|event| event["event_type"] == event_type) + .unwrap_or_else(|| panic!("missing {event_type}")); + assert_eq!(event["event_params"]["voice_session_id"], "work-voice-123"); + } + reducer + .ingest( + AnalyticsFact::Notification(Box::new(sample_turn_started_notification( + "thread-2", "turn-3", + ))), + &mut events, + ) + .await; + reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::AppUsed(AppUsedInput { + tracking: test_tracking_context("thread-2", "turn-3"), + app: AppInvocation { + connector_id: Some("drive".to_string()), + app_name: Some("Drive".to_string()), + invocation_type: Some(InvocationType::Implicit), + }, + elicitation_type: None, + })), + &mut events, + ) + .await; + let text_app = serde_json::to_value(events.last().expect("text app event")) + .expect("serialize text app event"); + assert_eq!(text_app["event_params"]["voice_session_id"], json!(null)); +} + +#[tokio::test] +async fn voice_handoff_steering_active_turn_does_not_tag_next_text_turn() { + let mut reducer = AnalyticsReducer::default(); + let mut events = Vec::new(); + reducer + .ingest( + AnalyticsFact::Notification(Box::new(sample_turn_started_notification( + "thread-2", "turn-2", + ))), + &mut events, + ) + .await; + reducer + .ingest( + AnalyticsFact::Notification(Box::new(ServerNotification::ThreadRealtimeStarted( + ThreadRealtimeStartedNotification { + thread_id: "thread-2".to_string(), + realtime_session_id: Some("work-voice-123".to_string()), + version: RealtimeConversationVersion::V2, + }, + ))), + &mut events, + ) + .await; + reducer + .ingest( + AnalyticsFact::RealtimeHandoffRequested { + thread_id: "thread-2".to_string(), + }, + &mut events, + ) + .await; + reducer + .ingest( + AnalyticsFact::Notification(Box::new(ServerNotification::ThreadRealtimeClosed( + ThreadRealtimeClosedNotification { + thread_id: "thread-2".to_string(), + reason: None, + }, + ))), + &mut events, + ) + .await; + reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::AppUsed(AppUsedInput { + tracking: test_tracking_context("thread-2", "turn-2"), + app: AppInvocation { + connector_id: Some("drive".to_string()), + app_name: Some("Drive".to_string()), + invocation_type: Some(InvocationType::Implicit), + }, + elicitation_type: None, + })), + &mut events, + ) + .await; + let voice_app = serde_json::to_value(events.last().expect("voice app event")) + .expect("serialize voice app event"); + assert_eq!( + voice_app["event_params"]["voice_session_id"], + "work-voice-123" + ); + + reducer + .ingest( + AnalyticsFact::Notification(Box::new(sample_turn_completed_notification( + "thread-2", + "turn-2", + AppServerTurnStatus::Completed, + /*codex_error_info*/ None, + ))), + &mut events, + ) + .await; + reducer + .ingest( + AnalyticsFact::Notification(Box::new(sample_turn_started_notification( + "thread-2", "turn-3", + ))), + &mut events, + ) + .await; + reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::AppUsed(AppUsedInput { + tracking: test_tracking_context("thread-2", "turn-3"), + app: AppInvocation { + connector_id: Some("drive".to_string()), + app_name: Some("Drive".to_string()), + invocation_type: Some(InvocationType::Implicit), + }, + elicitation_type: None, + })), + &mut events, + ) + .await; + let text_app = serde_json::to_value(events.last().expect("text app event")) + .expect("serialize text app event"); + assert_eq!(text_app["event_params"]["voice_session_id"], json!(null)); +} + #[tokio::test] async fn reducer_includes_plugin_ids_for_plugin_skill_invocations() { let mut reducer = AnalyticsReducer::default(); @@ -5050,6 +5290,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(), + voice_session_id: None, root_turn_id: Some("turn-2".to_string()), turn_trigger: Some("user".to_string()), codex_turn_source: Some("composer".to_string()), @@ -5127,6 +5368,7 @@ fn turn_event_serializes_expected_shape() { "thread_id": "thread-2", "session_id": "session-thread-2", "turn_id": "turn-2", + "voice_session_id": null, "root_turn_id": "turn-2", "turn_trigger": "user", "codex_turn_source": "composer", @@ -5645,6 +5887,19 @@ async fn turn_event_counts_completed_tool_items() { ) .await; + reducer + .ingest( + AnalyticsFact::Notification(Box::new(ServerNotification::ThreadRealtimeStarted( + ThreadRealtimeStartedNotification { + thread_id: "thread-2".to_string(), + realtime_session_id: Some("work-voice-456".to_string()), + version: RealtimeConversationVersion::V2, + }, + ))), + &mut out, + ) + .await; + let mcp_tool_call_item = |status, duration_ms| ThreadItem::McpToolCall { id: "mcp-1".to_string(), server: "server".to_string(), @@ -5843,6 +6098,10 @@ async fn turn_event_counts_completed_tool_items() { .expect("MCP tool call event should be emitted"); let payload = serde_json::to_value(mcp_tool_call_event).expect("serialize MCP tool call event"); assert_eq!(payload["event_params"]["plugin_id"], json!("sample@test")); + assert_eq!( + payload["event_params"]["voice_session_id"], + "work-voice-456" + ); assert_eq!( payload["event_params"]["connector_id"], json!("connector-test") @@ -5866,6 +6125,10 @@ async fn turn_event_counts_completed_tool_items() { .expect("turn event should be emitted"); let payload = serde_json::to_value(turn_event).expect("serialize turn event"); assert_eq!(payload["event_params"]["root_turn_id"], "root-ancestor"); + assert_eq!( + payload["event_params"]["voice_session_id"], + "work-voice-456" + ); 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)); diff --git a/codex-rs/analytics/src/client.rs b/codex-rs/analytics/src/client.rs index 410cf7def4..373e4e2bea 100644 --- a/codex-rs/analytics/src/client.rs +++ b/codex-rs/analytics/src/client.rs @@ -739,11 +739,23 @@ impl AnalyticsEventsClient { /// Records analytics-relevant notifications without cloning ignored variants. pub fn track_notification(&self, notification: &ServerNotification) { + if let ServerNotification::ThreadRealtimeItemAdded(handoff) = notification { + if handoff.item.get("type").and_then(serde_json::Value::as_str) + == Some("handoff_request") + { + self.record_fact(AnalyticsFact::RealtimeHandoffRequested { + thread_id: handoff.thread_id.clone(), + }); + } + return; + } if !matches!( notification, ServerNotification::ThreadArchived(_) | ServerNotification::ThreadClosed(_) | ServerNotification::ThreadUnarchived(_) + | ServerNotification::ThreadRealtimeStarted(_) + | ServerNotification::ThreadRealtimeClosed(_) | ServerNotification::TurnStarted(_) | ServerNotification::TurnCompleted(_) | ServerNotification::TurnDiffUpdated(_) diff --git a/codex-rs/analytics/src/client_tests.rs b/codex-rs/analytics/src/client_tests.rs index 5cc40f4f72..5175ec1bfd 100644 --- a/codex-rs/analytics/src/client_tests.rs +++ b/codex-rs/analytics/src/client_tests.rs @@ -75,6 +75,7 @@ use codex_app_server_protocol::ThreadArchiveParams; use codex_app_server_protocol::ThreadArchiveResponse; use codex_app_server_protocol::ThreadArchivedNotification; use codex_app_server_protocol::ThreadForkResponse; +use codex_app_server_protocol::ThreadRealtimeItemAddedNotification; use codex_app_server_protocol::ThreadResumeResponse; use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::ThreadStatus as AppServerThreadStatus; @@ -149,6 +150,7 @@ fn sample_skill_track_event(thread_id: &str, plugin_id: Option<&str>) -> TrackEv remote_plugin_id: None, thread_id: Some(thread_id.to_string()), turn_id: Some("turn-1".to_string()), + voice_session_id: None, invoke_type: Some(InvocationType::Explicit), model_slug: Some("gpt-5.1-codex".to_string()), }, @@ -236,6 +238,7 @@ fn sample_mcp_tool_call_event(thread_id: &str, plugin_id: Option<&str>) -> Track mcp_error_present: false, plugin_id: plugin_id.map(str::to_string), connector_id: None, + voice_session_id: None, elicitation_type: None, }, }) @@ -999,6 +1002,34 @@ fn track_notification_only_enqueues_analytics_relevant_notifications() { assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty))); } +#[test] +fn realtime_handoff_tracks_only_marker_without_transcript() { + let (client, mut receiver) = client_with_receiver(); + client.track_notification(&ServerNotification::ThreadRealtimeItemAdded( + ThreadRealtimeItemAddedNotification { + thread_id: "thread-1".to_string(), + item: serde_json::json!({ + "type": "handoff_request", + "input_transcript": "private speech", + }), + }, + )); + let Ok(AnalyticsEventsQueueMessage::Fact(input)) = receiver.try_recv() else { + panic!("expected realtime handoff marker"); + }; + assert!(matches!( + *input, + AnalyticsFact::RealtimeHandoffRequested { thread_id } if thread_id == "thread-1" + )); + client.track_notification(&ServerNotification::ThreadRealtimeItemAdded( + ThreadRealtimeItemAddedNotification { + thread_id: "thread-1".to_string(), + item: serde_json::json!({"type": "input_transcript"}), + }, + )); + assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty))); +} + #[test] fn track_event_request_batches_only_isolates_accepted_line_fingerprint_events() { let batches = track_event_request_batches(vec![ diff --git a/codex-rs/analytics/src/events.rs b/codex-rs/analytics/src/events.rs index a79a7d1874..7f62c974bf 100644 --- a/codex-rs/analytics/src/events.rs +++ b/codex-rs/analytics/src/events.rs @@ -212,6 +212,7 @@ pub(crate) struct SkillInvocationEventParams { pub(crate) remote_plugin_id: Option, pub(crate) thread_id: Option, pub(crate) turn_id: Option, + pub(crate) voice_session_id: Option, pub(crate) invoke_type: Option, pub(crate) model_slug: Option, } @@ -860,6 +861,7 @@ pub(crate) struct CodexMcpToolCallEventParams { pub(crate) mcp_error_present: bool, pub(crate) plugin_id: Option, pub(crate) connector_id: Option, + pub(crate) voice_session_id: Option, pub(crate) elicitation_type: Option, } @@ -973,6 +975,7 @@ pub(crate) struct CodexAppMentionedEventRequest { pub(crate) struct CodexAppUsedMetadata { #[serde(flatten)] pub(crate) app: CodexAppMetadata, + pub(crate) voice_session_id: Option, pub(crate) elicitation_type: Option, } @@ -1065,6 +1068,7 @@ pub(crate) struct CodexTurnEventParams { pub(crate) thread_id: String, pub(crate) session_id: String, pub(crate) turn_id: String, + pub(crate) voice_session_id: Option, pub(crate) root_turn_id: Option, pub(crate) turn_trigger: Option, pub(crate) codex_turn_source: Option, diff --git a/codex-rs/analytics/src/facts.rs b/codex-rs/analytics/src/facts.rs index 05df3bc44c..c86cad09d7 100644 --- a/codex-rs/analytics/src/facts.rs +++ b/codex-rs/analytics/src/facts.rs @@ -573,6 +573,9 @@ pub(crate) enum AnalyticsFact { completed_at_ms: u64, request_id: RequestId, }, + RealtimeHandoffRequested { + thread_id: String, + }, Notification(Box), // Facts that do not naturally exist on the app-server protocol surface, or // would require non-trivial protocol reshaping on this branch. diff --git a/codex-rs/analytics/src/reducer.rs b/codex-rs/analytics/src/reducer.rs index da3452cad8..7fee1ccc5c 100644 --- a/codex-rs/analytics/src/reducer.rs +++ b/codex-rs/analytics/src/reducer.rs @@ -227,6 +227,9 @@ struct ThreadAnalyticsState { connection_id: Option, metadata: Option, originator: Option, + active_voice_session_id: Option, + active_turn_id: Option, + pending_voice_handoffs: VecDeque>, } impl ThreadAnalyticsState { @@ -428,6 +431,7 @@ struct CompletedTurnState { struct TurnState { connection_id: Option, thread_id: Option, + voice_session_id: Option, num_input_images: Option, image_preparations: Vec, resolved_config: Option, @@ -609,6 +613,16 @@ impl AnalyticsReducer { } => { self.ingest_server_request_aborted(completed_at_ms, request_id, out); } + AnalyticsFact::RealtimeHandoffRequested { thread_id } => { + let thread = self.threads.entry(thread_id).or_default(); + // ThreadRealtimeStarted already attributes an active turn; only a new turn + // needs the handoff queued after the realtime session closes. + if thread.active_turn_id.is_none() { + thread + .pending_voice_handoffs + .push_back(thread.active_voice_session_id.clone()); + } + } AnalyticsFact::Custom(input) => match input { CustomAnalyticsFact::ArtifactOperation(input) => { self.ingest_artifact_operation(input, out); @@ -1374,6 +1388,10 @@ impl AnalyticsReducer { event_params: SkillInvocationEventParams { thread_id: Some(tracking.thread_id.clone()), turn_id: Some(tracking.turn_id.clone()), + voice_session_id: self + .turns + .get(&tracking.turn_id) + .and_then(|turn| turn.voice_session_id.clone()), invoke_type: Some(invocation.invocation_type), model_slug: Some(tracking.model_slug.clone()), product_client_id: Some(tracking.product_client_id.clone()), @@ -1405,6 +1423,10 @@ impl AnalyticsReducer { } = input; let event_params = CodexAppUsedMetadata { app: codex_app_metadata(&tracking, app), + voice_session_id: self + .turns + .get(&tracking.turn_id) + .and_then(|turn| turn.voice_session_id.clone()), elicitation_type, }; out.push(TrackEventRequest::AppUsed(CodexAppUsedEventRequest { @@ -2009,7 +2031,7 @@ impl AnalyticsReducer { else { return; }; - if let Some(event) = tool_item_event(ToolItemEventInput { + if let Some(mut event) = tool_item_event(ToolItemEventInput { thread_id: ¬ification.thread_id, turn_id: ¬ification.turn_id, item: ¬ification.item, @@ -2022,6 +2044,12 @@ impl AnalyticsReducer { review_summary: self.item_review_summaries.get(&key), elicitation_type, }) { + if let TrackEventRequest::McpToolCall(mcp) = &mut event { + mcp.event_params.voice_session_id = self + .turns + .get(¬ification.turn_id) + .and_then(|turn| turn.voice_session_id.clone()); + } let root_turn_id = self .turns .get(¬ification.turn_id) @@ -2072,9 +2100,39 @@ impl AnalyticsReducer { self.code_mode_cells.remove(¬ification.thread_id); self.pending_mcp_tool_elicitations .retain(|(key, _)| key.thread_id != notification.thread_id); + if let Some(thread) = self.threads.get_mut(¬ification.thread_id) { + thread.active_voice_session_id = None; + thread.active_turn_id = None; + thread.pending_voice_handoffs.clear(); + } + } + ServerNotification::ThreadRealtimeStarted(notification) => { + let thread = self.threads.entry(notification.thread_id).or_default(); + thread.active_voice_session_id = + notification.realtime_session_id.filter(|id| !id.is_empty()); + if let Some(turn_id) = &thread.active_turn_id { + self.turns + .entry(turn_id.clone()) + .or_default() + .voice_session_id = thread.active_voice_session_id.clone(); + } + } + ServerNotification::ThreadRealtimeClosed(notification) => { + if let Some(thread) = self.threads.get_mut(¬ification.thread_id) { + thread.active_voice_session_id = None; + } } ServerNotification::TurnStarted(notification) => { + let voice_session_id = { + let thread = self.threads.entry(notification.thread_id).or_default(); + thread.active_turn_id = Some(notification.turn.id.clone()); + thread + .pending_voice_handoffs + .pop_front() + .unwrap_or_else(|| thread.active_voice_session_id.clone()) + }; let turn_state = self.turns.entry(notification.turn.id).or_default(); + turn_state.voice_session_id = voice_session_id; turn_state.started_at = notification .turn .started_at @@ -2086,6 +2144,11 @@ impl AnalyticsReducer { turn_state.latest_diff = Some(notification.diff); } ServerNotification::TurnCompleted(notification) => { + if let Some(thread) = self.threads.get_mut(¬ification.thread_id) + && thread.active_turn_id.as_deref() == Some(notification.turn.id.as_str()) + { + thread.active_turn_id = None; + } self.flush_pending_tool_events(¬ification.thread_id, ¬ification.turn.id, out); let turn_state = self.turns.entry(notification.turn.id.clone()).or_default(); turn_state.completed = Some(CompletedTurnState { @@ -2838,6 +2901,7 @@ fn tool_item_event(input: ToolItemEventInput<'_>) -> Option { connector_id: app_context .as_ref() .map(|app_context| app_context.connector_id.clone()), + voice_session_id: None, elicitation_type, }, }, @@ -3594,6 +3658,7 @@ fn codex_turn_event_params( thread_id, session_id: thread_metadata.session_id.clone(), turn_id, + voice_session_id: turn_state.voice_session_id.clone(), root_turn_id: turn_metadata.root_turn_id(), turn_trigger: turn_metadata.turn_trigger(), codex_turn_source: turn_metadata.codex_turn_source(),