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

Attribute compaction analytics to the correct voice session (#51963)

Browse Source 
## Why

Compaction analytics lack a voice session ID. A compaction event can arrive after its voice session has closed or another session has started, so attribution needs to account for when compaction began.

## What changed

Add `voice_session_id` to compaction events. Track realtime session starts, ID updates, and closures, retaining up to 16 recent closed session windows. Resolve the session from the compaction start timestamp, falling back to the turn's voice session ID.

Forward realtime session ID updates from the app server to analytics and update the active turn's attribution.

## Testing

Add coverage for realtime session boundary facts, compaction event serialization, and delayed compaction events across consecutive voice sessions.

GitOrigin-RevId: 981da0ea2c38a87079b6bab33daf2c88fd95160f
This commit is contained in:
reia-oai
2026-10-08 05:29:11 +00:00
committed by copyberry
parent e974aad3b1
commit 14c8b7771a
8 changed files with 389 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/analytics/src/client.rs
+25 -2
View File
@@ -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(_)
codex-rs/analytics/src/client_tests.rs
+51
View File
@@ -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();
codex-rs/analytics/src/events.rs
+2
View File
@@ -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<String>,
pub(crate) codex_error_kind: Option<CodexErrKind>,
pub(crate) codex_error_http_status_code: Option<u16>,
pub(crate) usage_limit_window_minutes: Option<u16>,
@@ -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,
codex-rs/analytics/src/facts.rs
+13
View File
@@ -593,6 +593,19 @@ pub(crate) enum AnalyticsFact {
RealtimeHandoffRequested {
thread_id: String,
},
RealtimeSessionStarted {
thread_id: String,
realtime_session_id: Option<String>,
started_at: u64,
},
RealtimeSessionUpdated {
thread_id: String,
realtime_session_id: String,
},
RealtimeSessionClosed {
thread_id: String,
closed_at: u64,
},
Notification(Box<ServerNotification>),
// Facts that do not naturally exist on the app-server protocol surface, or
// would require non-trivial protocol reshaping on this branch.
codex-rs/analytics/src/reducer.rs
+132 -25
View File
@@ -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<u64>,
metadata: Option<ThreadMetadataState>,
originator: Option<String>,
active_voice_session_id: Option<String>,
active_voice_session: Option<VoiceSessionWindow>,
recent_voice_sessions: VecDeque<VoiceSessionWindow>,
active_turn_id: Option<String>,
pending_voice_handoffs: VecDeque<Option<String>>,
}
#[derive(Clone)]
struct VoiceSessionWindow {
session_id: Option<String>,
started_at: u64,
closed_at: Option<u64>,
}
impl ThreadAnalyticsState {
fn app_server_client(
&self,
@@ -244,6 +253,25 @@ impl ThreadAnalyticsState {
}
app_server_client
}
fn active_voice_session_id(&self) -> Option<String> {
self.active_voice_session
.as_ref()
.and_then(|session| session.session_id.clone())
}
fn voice_session_id_at(&self, timestamp: u64) -> Option<String> {
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(&notification.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(&notification.thread_id) {
thread.active_voice_session_id = None;
}
self.close_voice_session(&notification.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<String>,
) {
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<String>,
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<TrackEventRequest>) {
let Some((connection_state, thread_state, thread_metadata)) =
self.thread_context_or_warn(AnalyticsDropSite::goal(&input))
codex-rs/analytics/src/tests/suite/turns/compaction_tests.rs
+156 -36
View File
@@ -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");
}
codex-rs/app-server/src/bespoke_event_handling.rs
+4 -1
View File
@@ -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(),
codex-rs/app-server/src/outgoing_message.rs
+6
View File
@@ -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