diff --git a/codex-rs/analytics/src/client.rs b/codex-rs/analytics/src/client.rs index c6a2f195f7..15e6ddc866 100644 --- a/codex-rs/analytics/src/client.rs +++ b/codex-rs/analytics/src/client.rs @@ -582,6 +582,13 @@ impl AnalyticsEventsClient { ))); } + pub fn track_realtime_session_updated(&self, thread_id: String, realtime_session_id: String) { + self.record_fact(AnalyticsFact::RealtimeSessionUpdated { + thread_id, + realtime_session_id, + }); + } + pub fn track_guardian_v2_event(&self, event: GuardianV2Event) { self.record_fact(AnalyticsFact::Custom(CustomAnalyticsFact::GuardianV2( Box::new(event), @@ -811,6 +818,24 @@ impl AnalyticsEventsClient { /// Records analytics-relevant notifications without cloning ignored variants. pub fn track_notification(&self, notification: &ServerNotification) { + match notification { + ServerNotification::ThreadRealtimeStarted(started) => { + self.record_fact(AnalyticsFact::RealtimeSessionStarted { + thread_id: started.thread_id.clone(), + realtime_session_id: started.realtime_session_id.clone(), + started_at: crate::now_unix_seconds(), + }); + return; + } + ServerNotification::ThreadRealtimeClosed(closed) => { + self.record_fact(AnalyticsFact::RealtimeSessionClosed { + thread_id: closed.thread_id.clone(), + closed_at: crate::now_unix_seconds(), + }); + return; + } + _ => {} + } if let ServerNotification::ThreadRealtimeItemAdded(handoff) = notification { if handoff.item.get("type").and_then(serde_json::Value::as_str) == Some("handoff_request") @@ -826,8 +851,6 @@ impl AnalyticsEventsClient { 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 55f53efe24..13c942612b 100644 --- a/codex-rs/analytics/src/client_tests.rs +++ b/codex-rs/analytics/src/client_tests.rs @@ -76,7 +76,9 @@ 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::ThreadRealtimeClosedNotification; use codex_app_server_protocol::ThreadRealtimeItemAddedNotification; +use codex_app_server_protocol::ThreadRealtimeStartedNotification; use codex_app_server_protocol::ThreadResumeResponse; use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::ThreadStatus as AppServerThreadStatus; @@ -93,6 +95,7 @@ use codex_http_client::HttpClientFactory; use codex_http_client::OutboundProxyPolicy; #[cfg(debug_assertions)] use codex_login::AuthManager; +use codex_protocol::protocol::RealtimeConversationVersion; use codex_utils_absolute_path::test_support::PathBufExt; use codex_utils_absolute_path::test_support::test_path_buf; use pretty_assertions::assert_eq; @@ -1029,6 +1032,54 @@ fn track_notification_only_enqueues_analytics_relevant_notifications() { assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty))); } +#[test] +fn track_notification_records_realtime_session_boundaries() { + let (client, mut receiver) = client_with_receiver(); + + client.track_notification(&ServerNotification::ThreadRealtimeStarted( + ThreadRealtimeStartedNotification { + thread_id: "thread-1".to_string(), + realtime_session_id: Some("voice-session-1".to_string()), + version: RealtimeConversationVersion::V2, + }, + )); + + let Ok(AnalyticsEventsQueueMessage::Fact(input)) = receiver.try_recv() else { + panic!("expected realtime session started fact"); + }; + let AnalyticsFact::RealtimeSessionStarted { + thread_id, + realtime_session_id, + started_at, + } = *input + else { + panic!("expected realtime session started fact"); + }; + assert_eq!(thread_id, "thread-1"); + assert_eq!(realtime_session_id.as_deref(), Some("voice-session-1")); + assert!(started_at > 0); + + client.track_notification(&ServerNotification::ThreadRealtimeClosed( + ThreadRealtimeClosedNotification { + thread_id: "thread-1".to_string(), + reason: Some("completed".to_string()), + }, + )); + + let Ok(AnalyticsEventsQueueMessage::Fact(input)) = receiver.try_recv() else { + panic!("expected realtime session closed fact"); + }; + let AnalyticsFact::RealtimeSessionClosed { + thread_id, + closed_at, + } = *input + else { + panic!("expected realtime session closed fact"); + }; + assert_eq!(thread_id, "thread-1"); + assert!(closed_at >= started_at); +} + #[test] fn realtime_handoff_tracks_only_marker_without_transcript() { let (client, mut receiver) = client_with_receiver(); diff --git a/codex-rs/analytics/src/events.rs b/codex-rs/analytics/src/events.rs index 340860f985..5306749b5c 100644 --- a/codex-rs/analytics/src/events.rs +++ b/codex-rs/analytics/src/events.rs @@ -1027,6 +1027,7 @@ pub(crate) struct CodexCompactionEventParams { pub(crate) phase: CompactionPhase, pub(crate) strategy: CompactionStrategy, pub(crate) status: CompactionStatus, + pub(crate) voice_session_id: Option, pub(crate) codex_error_kind: Option, pub(crate) codex_error_http_status_code: Option, pub(crate) usage_limit_window_minutes: Option, @@ -1396,6 +1397,7 @@ pub(crate) fn codex_compaction_event_params( phase: input.phase, strategy: input.strategy, status: input.status, + voice_session_id: None, codex_error_kind: input.codex_error_kind, codex_error_http_status_code: input.codex_error_http_status_code, usage_limit_window_minutes: input.usage_limit_window_minutes, diff --git a/codex-rs/analytics/src/facts.rs b/codex-rs/analytics/src/facts.rs index f640774190..9189babd9f 100644 --- a/codex-rs/analytics/src/facts.rs +++ b/codex-rs/analytics/src/facts.rs @@ -593,6 +593,19 @@ pub(crate) enum AnalyticsFact { RealtimeHandoffRequested { thread_id: String, }, + RealtimeSessionStarted { + thread_id: String, + realtime_session_id: Option, + started_at: u64, + }, + RealtimeSessionUpdated { + thread_id: String, + realtime_session_id: String, + }, + RealtimeSessionClosed { + thread_id: String, + closed_at: u64, + }, 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 9645787827..2fb9d7da28 100644 --- a/codex-rs/analytics/src/reducer.rs +++ b/codex-rs/analytics/src/reducer.rs @@ -180,6 +180,7 @@ use std::collections::VecDeque; use std::path::Path; use std::path::PathBuf; const MAX_TOOL_RESPONSE_ENTRIES: usize = 256; +const MAX_RECENT_VOICE_SESSION_WINDOWS: usize = 16; pub(crate) const MAX_PLUGIN_MEASUREMENTS_PER_BATCH: usize = 100; const MAX_PLUGIN_MEASUREMENT_DIMENSIONS: usize = 8; @@ -228,11 +229,19 @@ struct ThreadAnalyticsState { connection_id: Option, metadata: Option, originator: Option, - active_voice_session_id: Option, + active_voice_session: Option, + recent_voice_sessions: VecDeque, active_turn_id: Option, pending_voice_handoffs: VecDeque>, } +#[derive(Clone)] +struct VoiceSessionWindow { + session_id: Option, + started_at: u64, + closed_at: Option, +} + impl ThreadAnalyticsState { fn app_server_client( &self, @@ -244,6 +253,25 @@ impl ThreadAnalyticsState { } app_server_client } + + fn active_voice_session_id(&self) -> Option { + self.active_voice_session + .as_ref() + .and_then(|session| session.session_id.clone()) + } + + fn voice_session_id_at(&self, timestamp: u64) -> Option { + self.active_voice_session + .iter() + .chain(self.recent_voice_sessions.iter().rev()) + .find(|session| { + session.started_at <= timestamp + && session + .closed_at + .is_none_or(|closed_at| timestamp <= closed_at) + }) + .and_then(|session| session.session_id.clone()) + } } #[derive(Clone, Copy)] @@ -621,9 +649,28 @@ impl AnalyticsReducer { if thread.active_turn_id.is_none() { thread .pending_voice_handoffs - .push_back(thread.active_voice_session_id.clone()); + .push_back(thread.active_voice_session_id()); } } + AnalyticsFact::RealtimeSessionStarted { + thread_id, + realtime_session_id, + started_at, + } => { + self.start_voice_session(thread_id, realtime_session_id, started_at); + } + AnalyticsFact::RealtimeSessionUpdated { + thread_id, + realtime_session_id, + } => { + self.set_active_voice_session_id(thread_id, Some(realtime_session_id)); + } + AnalyticsFact::RealtimeSessionClosed { + thread_id, + closed_at, + } => { + self.close_voice_session(&thread_id, closed_at); + } AnalyticsFact::Custom(input) => match input { CustomAnalyticsFact::ArtifactOperation(input) => { self.ingest_artifact_operation(input, out); @@ -2119,26 +2166,21 @@ impl AnalyticsReducer { 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_voice_session = None; + thread.recent_voice_sessions.clear(); 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(); - } + self.start_voice_session( + notification.thread_id, + notification.realtime_session_id, + now_unix_seconds(), + ); } ServerNotification::ThreadRealtimeClosed(notification) => { - if let Some(thread) = self.threads.get_mut(¬ification.thread_id) { - thread.active_voice_session_id = None; - } + self.close_voice_session(¬ification.thread_id, now_unix_seconds()); } ServerNotification::TurnStarted(notification) => { let voice_session_id = { @@ -2147,7 +2189,7 @@ impl AnalyticsReducer { thread .pending_voice_handoffs .pop_front() - .unwrap_or_else(|| thread.active_voice_session_id.clone()) + .unwrap_or_else(|| thread.active_voice_session_id()) }; let turn_state = self.turns.entry(notification.turn.id).or_default(); turn_state.voice_session_id = voice_session_id; @@ -2304,22 +2346,87 @@ impl AnalyticsReducer { else { return; }; + let voice_session_id = thread_state + .voice_session_id_at(input.started_at) + .or_else(|| { + self.turns + .get(&input.turn_id) + .and_then(|turn| turn.voice_session_id.clone()) + }); + let mut event_params = codex_compaction_event_params( + input, + thread_metadata.session_id.clone(), + thread_state.app_server_client(connection_state), + connection_state.runtime.clone(), + thread_metadata.thread_source.clone(), + thread_metadata.subagent_source.clone(), + thread_metadata.parent_thread_id.clone(), + ); + event_params.voice_session_id = voice_session_id; out.push(TrackEventRequest::Compaction(Box::new( CodexCompactionEventRequest { event_type: "codex_compaction_event", - event_params: codex_compaction_event_params( - input, - thread_metadata.session_id.clone(), - thread_state.app_server_client(connection_state), - connection_state.runtime.clone(), - thread_metadata.thread_source.clone(), - thread_metadata.subagent_source.clone(), - thread_metadata.parent_thread_id.clone(), - ), + event_params, }, ))); } + fn set_active_voice_session_id( + &mut self, + thread_id: String, + realtime_session_id: Option, + ) { + let thread = self.threads.entry(thread_id).or_default(); + let realtime_session_id = realtime_session_id.filter(|id| !id.is_empty()); + let Some(session) = thread.active_voice_session.as_mut() else { + return; + }; + session.session_id.clone_from(&realtime_session_id); + if let Some(turn_id) = &thread.active_turn_id { + self.turns + .entry(turn_id.clone()) + .or_default() + .voice_session_id = realtime_session_id; + } + } + + fn start_voice_session( + &mut self, + thread_id: String, + realtime_session_id: Option, + started_at: u64, + ) { + self.close_voice_session(&thread_id, started_at); + let realtime_session_id = realtime_session_id.filter(|id| !id.is_empty()); + let thread = self.threads.entry(thread_id).or_default(); + thread.active_voice_session = Some(VoiceSessionWindow { + session_id: realtime_session_id.clone(), + started_at, + closed_at: None, + }); + if let Some(turn_id) = &thread.active_turn_id { + self.turns + .entry(turn_id.clone()) + .or_default() + .voice_session_id = realtime_session_id; + } + } + + fn close_voice_session(&mut self, thread_id: &str, closed_at: u64) { + let Some(thread) = self.threads.get_mut(thread_id) else { + return; + }; + if let Some(mut session) = thread.active_voice_session.take() { + session.closed_at = Some(closed_at); + if session.session_id.is_some() { + if thread.recent_voice_sessions.len() >= MAX_RECENT_VOICE_SESSION_WINDOWS { + thread.recent_voice_sessions.pop_front(); + } + thread.recent_voice_sessions.push_back(session); + } + } + } + fn ingest_goal(&mut self, input: CodexGoalEvent, out: &mut Vec) { let Some((connection_state, thread_state, thread_metadata)) = self.thread_context_or_warn(AnalyticsDropSite::goal(&input)) diff --git a/codex-rs/analytics/src/tests/suite/turns/compaction_tests.rs b/codex-rs/analytics/src/tests/suite/turns/compaction_tests.rs index fa4c642152..2c0400af4b 100644 --- a/codex-rs/analytics/src/tests/suite/turns/compaction_tests.rs +++ b/codex-rs/analytics/src/tests/suite/turns/compaction_tests.rs @@ -17,6 +17,7 @@ use crate::reducer::AnalyticsReducer; use crate::tests::support::sample_app_server_client_metadata; use crate::tests::support::sample_runtime_metadata; use crate::tests::support::sample_thread_resume_response_with_source; +use crate::tests::support::sample_turn_started_notification; use codex_app_server_protocol::ClientInfo; use codex_app_server_protocol::InitializeCapabilities; use codex_app_server_protocol::InitializeParams; @@ -31,38 +32,40 @@ use serde_json::json; #[test] fn compaction_event_serializes_expected_shape() { + let mut event_params = crate::events::codex_compaction_event_params( + CodexCompactionEvent { + thread_id: "thread-1".to_string(), + turn_id: "turn-1".to_string(), + trigger: CompactionTrigger::Auto, + reason: CompactionReason::ContextLimit, + implementation: CompactionImplementation::ResponsesCompactionV2, + phase: CompactionPhase::MidTurn, + strategy: CompactionStrategy::Memento, + status: CompactionStatus::Completed, + codex_error_kind: None, + codex_error_http_status_code: None, + usage_limit_window_minutes: None, + active_context_tokens_before: 120_000, + active_context_tokens_after: 18_000, + retained_image_count: None, + compaction_summary_tokens: None, + cached_input_tokens: None, + cache_write_input_tokens: Some(456), + started_at: 100, + completed_at: 106, + duration_ms: Some(6543), + }, + "session-thread-1".to_string(), + sample_app_server_client_metadata(), + sample_runtime_metadata(), + Some(ThreadSource::User), + /*subagent_source*/ None, + /*parent_thread_id*/ None, + ); + event_params.voice_session_id = Some("voice-session-1".to_string()); let event = TrackEventRequest::Compaction(Box::new(CodexCompactionEventRequest { event_type: "codex_compaction_event", - event_params: crate::events::codex_compaction_event_params( - CodexCompactionEvent { - thread_id: "thread-1".to_string(), - turn_id: "turn-1".to_string(), - trigger: CompactionTrigger::Auto, - reason: CompactionReason::ContextLimit, - implementation: CompactionImplementation::ResponsesCompactionV2, - phase: CompactionPhase::MidTurn, - strategy: CompactionStrategy::Memento, - status: CompactionStatus::Completed, - codex_error_kind: None, - codex_error_http_status_code: None, - usage_limit_window_minutes: None, - active_context_tokens_before: 120_000, - active_context_tokens_after: 18_000, - retained_image_count: None, - compaction_summary_tokens: None, - cached_input_tokens: None, - cache_write_input_tokens: Some(456), - started_at: 100, - completed_at: 106, - duration_ms: Some(6543), - }, - "session-thread-1".to_string(), - sample_app_server_client_metadata(), - sample_runtime_metadata(), - Some(ThreadSource::User), - /*subagent_source*/ None, - /*parent_thread_id*/ None, - ), + event_params, })); let payload = serde_json::to_value(&event).expect("serialize compaction event"); @@ -97,6 +100,7 @@ fn compaction_event_serializes_expected_shape() { "phase": "mid_turn", "strategy": "memento", "status": "completed", + "voice_session_id": "voice-session-1", "codex_error_kind": null, "codex_error_http_status_code": null, "usage_limit_window_minutes": null, @@ -123,7 +127,7 @@ fn compaction_implementation_serializes_remote_v2() { } #[tokio::test] -async fn compaction_event_ingests_custom_fact() { +async fn compaction_events_keep_authoritative_voice_session_across_consecutive_calls() { let mut reducer = AnalyticsReducer::default(); let mut events = Vec::new(); let parent_thread_id = @@ -182,12 +186,79 @@ async fn compaction_event_ingests_custom_fact() { .await; events.clear(); + reducer + .ingest( + AnalyticsFact::RealtimeSessionStarted { + thread_id: "thread-1".to_string(), + realtime_session_id: None, + started_at: 100, + }, + &mut events, + ) + .await; + reducer + .ingest( + AnalyticsFact::RealtimeSessionUpdated { + thread_id: "thread-1".to_string(), + realtime_session_id: "voice-session-1".to_string(), + }, + &mut events, + ) + .await; + reducer + .ingest( + AnalyticsFact::Notification(Box::new(sample_turn_started_notification( + "thread-1", + "turn-compact-1", + ))), + &mut events, + ) + .await; + reducer + .ingest( + AnalyticsFact::RealtimeSessionClosed { + thread_id: "thread-1".to_string(), + closed_at: 150, + }, + &mut events, + ) + .await; + + reducer + .ingest( + AnalyticsFact::RealtimeSessionStarted { + thread_id: "thread-1".to_string(), + realtime_session_id: None, + started_at: 200, + }, + &mut events, + ) + .await; + reducer + .ingest( + AnalyticsFact::RealtimeSessionUpdated { + thread_id: "thread-1".to_string(), + realtime_session_id: "voice-session-2".to_string(), + }, + &mut events, + ) + .await; + reducer + .ingest( + AnalyticsFact::Notification(Box::new(sample_turn_started_notification( + "thread-1", + "turn-compact-2", + ))), + &mut events, + ) + .await; + reducer .ingest( AnalyticsFact::Custom(CustomAnalyticsFact::Compaction(Box::new( CodexCompactionEvent { thread_id: "thread-1".to_string(), - turn_id: "turn-compact".to_string(), + turn_id: "turn-compact-1".to_string(), trigger: CompactionTrigger::Manual, reason: CompactionReason::UserRequested, implementation: CompactionImplementation::Responses, @@ -203,8 +274,8 @@ async fn compaction_event_ingests_custom_fact() { compaction_summary_tokens: None, cached_input_tokens: None, cache_write_input_tokens: None, - started_at: 100, - completed_at: 101, + started_at: 110, + completed_at: 210, duration_ms: Some(1200), }, ))), @@ -212,12 +283,55 @@ async fn compaction_event_ingests_custom_fact() { ) .await; + reducer + .ingest( + AnalyticsFact::RealtimeSessionClosed { + thread_id: "thread-1".to_string(), + closed_at: 250, + }, + &mut events, + ) + .await; + reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::Compaction(Box::new( + CodexCompactionEvent { + thread_id: "thread-1".to_string(), + turn_id: "turn-compact-2".to_string(), + trigger: CompactionTrigger::Auto, + reason: CompactionReason::ContextLimit, + implementation: CompactionImplementation::ResponsesCompactionV2, + phase: CompactionPhase::MidTurn, + strategy: CompactionStrategy::Memento, + status: CompactionStatus::Completed, + codex_error_kind: None, + codex_error_http_status_code: None, + usage_limit_window_minutes: None, + active_context_tokens_before: 100_000, + active_context_tokens_after: 20_000, + retained_image_count: None, + compaction_summary_tokens: None, + cached_input_tokens: None, + cache_write_input_tokens: None, + started_at: 210, + completed_at: 251, + duration_ms: Some(800), + }, + ))), + &mut events, + ) + .await; + let payload = serde_json::to_value(&events).expect("serialize events"); - assert_eq!(payload.as_array().expect("events array").len(), 1); + assert_eq!(payload.as_array().expect("events array").len(), 2); assert_eq!(payload[0]["event_type"], "codex_compaction_event"); assert_eq!(payload[0]["event_params"]["session_id"], "session-thread-1"); assert_eq!(payload[0]["event_params"]["thread_id"], "thread-1"); - assert_eq!(payload[0]["event_params"]["turn_id"], "turn-compact"); + assert_eq!(payload[0]["event_params"]["turn_id"], "turn-compact-1"); + assert_eq!( + payload[0]["event_params"]["voice_session_id"], + "voice-session-1" + ); assert_eq!( payload[0]["event_params"]["codex_error_kind"], json!("context_window_exceeded") @@ -257,4 +371,10 @@ async fn compaction_event_ingests_custom_fact() { assert_eq!(payload[0]["event_params"]["phase"], "standalone_turn"); assert_eq!(payload[0]["event_params"]["strategy"], "memento"); assert_eq!(payload[0]["event_params"]["status"], "failed"); + assert_eq!(payload[1]["event_params"]["turn_id"], "turn-compact-2"); + assert_eq!( + payload[1]["event_params"]["voice_session_id"], + "voice-session-2" + ); + assert_eq!(payload[1]["event_params"]["status"], "completed"); } diff --git a/codex-rs/app-server/src/bespoke_event_handling.rs b/codex-rs/app-server/src/bespoke_event_handling.rs index f632857401..5e5430826e 100644 --- a/codex-rs/app-server/src/bespoke_event_handling.rs +++ b/codex-rs/app-server/src/bespoke_event_handling.rs @@ -498,7 +498,10 @@ pub(crate) async fn apply_bespoke_event_handling( )) .await; } - RealtimeEvent::SessionUpdated { .. } => {} + RealtimeEvent::SessionUpdated { + realtime_session_id, + .. + } => outgoing.track_realtime_session_updated(realtime_session_id), RealtimeEvent::InputAudioSpeechStarted(event) => { let notification = ThreadRealtimeItemAddedNotification { thread_id: conversation_id.to_string(), diff --git a/codex-rs/app-server/src/outgoing_message.rs b/codex-rs/app-server/src/outgoing_message.rs index 961eccc2aa..9ed210ebd3 100644 --- a/codex-rs/app-server/src/outgoing_message.rs +++ b/codex-rs/app-server/src/outgoing_message.rs @@ -196,6 +196,12 @@ impl ThreadScopedOutgoingMessageSender { ); } + pub(crate) fn track_realtime_session_updated(&self, realtime_session_id: String) { + self.outgoing + .analytics_events_client + .track_realtime_session_updated(self.thread_id.to_string(), realtime_session_id); + } + pub(crate) async fn send_server_notification(&self, notification: ServerNotification) { self.outgoing .analytics_events_client