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(),