From 963ce10299a11f3526e8c6a64c12ea472d2feb66 Mon Sep 17 00:00:00 2001 From: rhan-oai Date: Fri, 10 Apr 2026 17:21:06 -0700 Subject: [PATCH] [codex-analytics] denormalize thread metadata onto turn events --- .../analytics/src/analytics_client_tests.rs | 20 ++++++++ codex-rs/analytics/src/events.rs | 12 +++++ codex-rs/analytics/src/reducer.rs | 49 ++++++++++++++++--- .../app-server/tests/suite/v2/turn_start.rs | 11 +++++ 4 files changed, 86 insertions(+), 6 deletions(-) diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index b6f9d8aa3b..a89e684aa2 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -455,6 +455,18 @@ async fn ingest_turn_prerequisites( ) { if include_initialize { ingest_initialize(reducer, out).await; + reducer + .ingest( + AnalyticsFact::Response { + connection_id: 7, + response: Box::new(sample_thread_start_response( + "thread-2", /*ephemeral*/ false, "gpt-5", + )), + }, + out, + ) + .await; + out.clear(); } reducer @@ -1610,6 +1622,10 @@ async fn accepted_turn_steer_emits_expected_event() { payload["event_params"]["runtime"]["codex_rs_version"], json!("0.1.0") ); + assert_eq!(payload["event_params"]["thread_source"], json!("user")); + assert_eq!(payload["event_params"]["initialization_mode"], json!("new")); + assert_eq!(payload["event_params"]["subagent_source"], json!(null)); + assert_eq!(payload["event_params"]["parent_thread_id"], json!(null)); assert!(payload["event_params"].get("product_client_id").is_none()); } @@ -1638,6 +1654,10 @@ async fn rejected_turn_steer_uses_request_connection_metadata() { payload["event_params"]["runtime"]["codex_rs_version"], json!("0.1.0") ); + assert_eq!(payload["event_params"]["thread_source"], json!("user")); + assert_eq!(payload["event_params"]["initialization_mode"], json!("new")); + assert_eq!(payload["event_params"]["subagent_source"], json!(null)); + assert_eq!(payload["event_params"]["parent_thread_id"], json!(null)); assert_eq!(payload["event_params"]["result"], json!("rejected")); assert_eq!( payload["event_params"]["rejection_reason"], diff --git a/codex-rs/analytics/src/events.rs b/codex-rs/analytics/src/events.rs index 4bc3ccde9d..74105a742d 100644 --- a/codex-rs/analytics/src/events.rs +++ b/codex-rs/analytics/src/events.rs @@ -213,6 +213,10 @@ pub(crate) struct CodexTurnSteerEventParams { pub(crate) accepted_turn_id: Option, pub(crate) app_server_client: CodexAppServerClientMetadata, pub(crate) runtime: CodexRuntimeMetadata, + pub(crate) thread_source: Option, + pub(crate) initialization_mode: ThreadInitializationMode, + pub(crate) subagent_source: Option, + pub(crate) parent_thread_id: Option, pub(crate) num_input_images: usize, pub(crate) result: TurnSteerResult, pub(crate) rejection_reason: Option, @@ -351,6 +355,10 @@ pub(crate) fn codex_turn_steer_event_params( app_server_client: CodexAppServerClientMetadata, runtime: CodexRuntimeMetadata, tracking: &TrackEventsContext, + thread_source: Option<&'static str>, + initialization_mode: ThreadInitializationMode, + subagent_source: Option, + parent_thread_id: Option, turn_steer: CodexTurnSteerEvent, ) -> CodexTurnSteerEventParams { CodexTurnSteerEventParams { @@ -359,6 +367,10 @@ pub(crate) fn codex_turn_steer_event_params( accepted_turn_id: turn_steer.accepted_turn_id, app_server_client, runtime, + thread_source: thread_source.map(str::to_string), + initialization_mode, + subagent_source, + parent_thread_id, num_input_images: turn_steer.num_input_images, result: turn_steer.result, rejection_reason: turn_steer.rejection_reason, diff --git a/codex-rs/analytics/src/reducer.rs b/codex-rs/analytics/src/reducer.rs index d49c88e483..00e4b7abd7 100644 --- a/codex-rs/analytics/src/reducer.rs +++ b/codex-rs/analytics/src/reducer.rs @@ -83,12 +83,16 @@ struct ConnectionState { #[derive(Clone)] struct ThreadMetadataState { thread_source: Option<&'static str>, + initialization_mode: ThreadInitializationMode, subagent_source: Option, parent_thread_id: Option, } impl ThreadMetadataState { - fn from_session_source(session_source: &SessionSource) -> Self { + fn from_thread_lifecycle( + session_source: &SessionSource, + initialization_mode: ThreadInitializationMode, + ) -> Self { let (subagent_source, parent_thread_id) = match session_source { SessionSource::SubAgent(subagent_source) => ( Some(subagent_source_name(subagent_source)), @@ -103,6 +107,7 @@ impl ThreadMetadataState { }; Self { thread_source: thread_source_name(session_source), + initialization_mode, subagent_source, parent_thread_id, } @@ -615,7 +620,8 @@ impl AnalyticsReducer { let Some(connection_state) = self.connections.get(&connection_id) else { return; }; - let thread_metadata = ThreadMetadataState::from_session_source(&thread_source); + let thread_metadata = + ThreadMetadataState::from_thread_lifecycle(&thread_source, initialization_mode); self.thread_connections .insert(thread_id.clone(), connection_id); self.thread_metadata @@ -717,6 +723,13 @@ impl AnalyticsReducer { let Some(connection_state) = self.connections.get(&connection_id) else { return; }; + let Some(thread_metadata) = self.thread_metadata.get(&pending_request.thread_id) else { + tracing::warn!( + thread_id = %pending_request.thread_id, + "dropping turn steer analytics event: missing thread lifecycle metadata" + ); + return; + }; let tracking = TrackEventsContext { model_slug: String::new(), thread_id: pending_request.thread_id, @@ -739,6 +752,10 @@ impl AnalyticsReducer { connection_state.app_server_client.clone(), connection_state.runtime.clone(), &tracking, + thread_metadata.thread_source, + thread_metadata.initialization_mode, + thread_metadata.subagent_source.clone(), + thread_metadata.parent_thread_id.clone(), turn_steer, ), })); @@ -765,6 +782,24 @@ impl AnalyticsReducer { ) }); let Some((app_server_client, runtime)) = connection_metadata else { + if let Some(connection_id) = turn_state.connection_id { + tracing::warn!( + turn_id, + connection_id, + "dropping turn analytics event: missing connection metadata" + ); + } + return; + }; + let Some(thread_id) = turn_state.thread_id.as_ref() else { + return; + }; + let Some(thread_metadata) = self.thread_metadata.get(thread_id) else { + tracing::warn!( + thread_id, + turn_id, + "dropping turn analytics event: missing thread lifecycle metadata" + ); return; }; out.push(TrackEventRequest::TurnEvent(Box::new( @@ -775,6 +810,7 @@ impl AnalyticsReducer { runtime, turn_id.to_string(), turn_state, + thread_metadata, ), }, ))); @@ -787,6 +823,7 @@ fn codex_turn_event_params( runtime: CodexRuntimeMetadata, turn_id: String, turn_state: &TurnState, + thread_metadata: &ThreadMetadataState, ) -> CodexTurnEventParams { let (Some(thread_id), Some(num_input_images), Some(resolved_config), Some(completed)) = ( turn_state.thread_id.clone(), @@ -803,7 +840,7 @@ fn codex_turn_event_params( num_input_images: _resolved_num_input_images, submission_type, ephemeral, - session_source, + session_source: _session_source, model, model_provider, sandbox_policy, @@ -825,10 +862,10 @@ fn codex_turn_event_params( runtime, submission_type, ephemeral, - thread_source: thread_source_name(&session_source).map(str::to_string), + thread_source: thread_metadata.thread_source.map(str::to_string), initialization_mode: thread_metadata.initialization_mode, - subagent_source: None, - parent_thread_id: None, + subagent_source: thread_metadata.subagent_source.clone(), + parent_thread_id: thread_metadata.parent_thread_id.clone(), model: Some(model), model_provider, sandbox_policy: Some(sandbox_policy_mode(&sandbox_policy)), 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 ea3ba02b46..694d0a6eef 100644 --- a/codex-rs/app-server/tests/suite/v2/turn_start.rs +++ b/codex-rs/app-server/tests/suite/v2/turn_start.rs @@ -395,6 +395,17 @@ async fn turn_start_tracks_turn_event_analytics() -> Result<()> { assert_eq!(event["event_params"]["model"], "mock-model"); assert_eq!(event["event_params"]["model_provider"], "mock_provider"); assert_eq!(event["event_params"]["sandbox_policy"], "read_only"); + assert_eq!(event["event_params"]["ephemeral"], false); + assert_eq!(event["event_params"]["thread_source"], "user"); + assert_eq!(event["event_params"]["initialization_mode"], "new"); + assert_eq!( + event["event_params"]["subagent_source"], + serde_json::Value::Null + ); + assert_eq!( + event["event_params"]["parent_thread_id"], + serde_json::Value::Null + ); assert_eq!(event["event_params"]["num_input_images"], 1); assert_eq!(event["event_params"]["status"], "completed"); assert!(event["event_params"]["started_at"].as_u64().is_some());