[codex-analytics] denormalize thread metadata onto turn events

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"],

codex-rs/analytics/src/events.rs

@@ -213,6 +213,10 @@ pub(crate) struct CodexTurnSteerEventParams {
     pub(crate) accepted_turn_id: Option<String>,
     pub(crate) app_server_client: CodexAppServerClientMetadata,
     pub(crate) runtime: CodexRuntimeMetadata,
+    pub(crate) thread_source: Option<String>,
+    pub(crate) initialization_mode: ThreadInitializationMode,
+    pub(crate) subagent_source: Option<String>,
+    pub(crate) parent_thread_id: Option<String>,
     pub(crate) num_input_images: usize,
     pub(crate) result: TurnSteerResult,
     pub(crate) rejection_reason: Option<TurnSteerRejectionReason>,
@@ -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<String>,
+    parent_thread_id: Option<String>,
     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,

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<String>,
     parent_thread_id: Option<String>,
 }
 
 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)),

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