[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
@@ -1442,6 +1454,10 @@ fn turn_event_serializes_expected_shape() {
             runtime: sample_runtime_metadata(),
             submission_type: None,
             ephemeral: false,
+            thread_source: Some("user".to_string()),
+            initialization_mode: ThreadInitializationMode::New,
+            subagent_source: None,
+            parent_thread_id: None,
             model: Some("gpt-5".to_string()),
             model_provider: "openai".to_string(),
             sandbox_policy: Some("read_only"),
@@ -1499,6 +1515,10 @@ fn turn_event_serializes_expected_shape() {
                     "runtime_arch": "aarch64"
                 },
                 "ephemeral": false,
+                "thread_source": "user",
+                "initialization_mode": "new",
+                "subagent_source": null,
+                "parent_thread_id": null,
                 "model": "gpt-5",
                 "model_provider": "openai",
                 "sandbox_policy": "read_only",
@@ -1598,6 +1618,9 @@ 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"]["subagent_source"], json!(null));
+    assert_eq!(payload["event_params"]["parent_thread_id"], json!(null));
     assert!(payload["event_params"].get("product_client_id").is_none());
 }
 
@@ -1626,6 +1649,9 @@ 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"]["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

@@ -1,6 +1,5 @@
 use crate::facts::AppInvocation;
 use crate::facts::CodexCompactionEvent;
-use crate::facts::CodexTurnSteerEvent;
 use crate::facts::InvocationType;
 use crate::facts::PluginState;
 use crate::facts::SubAgentThreadStartedInput;
@@ -341,6 +340,10 @@ pub(crate) struct CodexTurnEventParams {
     pub(crate) app_server_client: CodexAppServerClientMetadata,
     pub(crate) runtime: CodexRuntimeMetadata,
     pub(crate) ephemeral: bool,
+    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) model: Option<String>,
     pub(crate) model_provider: String,
     pub(crate) sandbox_policy: Option<&'static str>,
@@ -390,6 +393,9 @@ 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) 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>,
@@ -524,25 +530,6 @@ pub(crate) fn codex_plugin_used_metadata(
     }
 }
 
-pub(crate) fn codex_turn_steer_event_params(
-    app_server_client: CodexAppServerClientMetadata,
-    runtime: CodexRuntimeMetadata,
-    tracking: &TrackEventsContext,
-    turn_steer: CodexTurnSteerEvent,
-) -> CodexTurnSteerEventParams {
-    CodexTurnSteerEventParams {
-        thread_id: tracking.thread_id.clone(),
-        expected_turn_id: turn_steer.expected_turn_id,
-        accepted_turn_id: turn_steer.accepted_turn_id,
-        app_server_client,
-        runtime,
-        num_input_images: turn_steer.num_input_images,
-        result: turn_steer.result,
-        rejection_reason: turn_steer.rejection_reason,
-        created_at: turn_steer.created_at,
-    }
-}
-
 pub(crate) fn thread_source_name(thread_source: &SessionSource) -> Option<&'static str> {
     match thread_source {
         SessionSource::Cli | SessionSource::VSCode | SessionSource::Exec => Some("user"),

codex-rs/analytics/src/reducer.rs

@@ -8,6 +8,7 @@ use crate::events::CodexPluginUsedEventRequest;
 use crate::events::CodexRuntimeMetadata;
 use crate::events::CodexTurnEventParams;
 use crate::events::CodexTurnEventRequest;
+use crate::events::CodexTurnSteerEventParams;
 use crate::events::CodexTurnSteerEventRequest;
 use crate::events::GuardianReviewEventParams;
 use crate::events::GuardianReviewEventPayload;
@@ -21,7 +22,6 @@ use crate::events::codex_app_metadata;
 use crate::events::codex_compaction_event_params;
 use crate::events::codex_plugin_metadata;
 use crate::events::codex_plugin_used_metadata;
-use crate::events::codex_turn_steer_event_params;
 use crate::events::plugin_state_event_type;
 use crate::events::subagent_parent_thread_id;
 use crate::events::subagent_source_name;
@@ -32,7 +32,6 @@ use crate::facts::AnalyticsJsonRpcError;
 use crate::facts::AppMentionedInput;
 use crate::facts::AppUsedInput;
 use crate::facts::CodexCompactionEvent;
-use crate::facts::CodexTurnSteerEvent;
 use crate::facts::CustomAnalyticsFact;
 use crate::facts::PluginState;
 use crate::facts::PluginStateChangedInput;
@@ -40,7 +39,6 @@ use crate::facts::PluginUsedInput;
 use crate::facts::SkillInvokedInput;
 use crate::facts::SubAgentThreadStartedInput;
 use crate::facts::ThreadInitializationMode;
-use crate::facts::TrackEventsContext;
 use crate::facts::TurnResolvedConfigFact;
 use crate::facts::TurnStatus;
 use crate::facts::TurnSteerRejectionReason;
@@ -86,12 +84,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_metadata(
+        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)),
@@ -106,6 +108,7 @@ impl ThreadMetadataState {
         };
         Self {
             thread_source: thread_source_name(session_source),
+            initialization_mode,
             subagent_source,
             parent_thread_id,
         }
@@ -657,7 +660,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_metadata(&thread_source, initialization_mode);
         self.thread_connections
             .insert(thread_id.clone(), connection_id);
         self.thread_metadata
@@ -759,30 +763,29 @@ impl AnalyticsReducer {
         let Some(connection_state) = self.connections.get(&connection_id) else {
             return;
         };
-        let tracking = TrackEventsContext {
-            model_slug: String::new(),
-            thread_id: pending_request.thread_id,
-            turn_id: accepted_turn_id
-                .as_deref()
-                .unwrap_or(pending_request.expected_turn_id.as_str())
-                .to_string(),
-        };
-        let turn_steer = CodexTurnSteerEvent {
-            expected_turn_id: Some(pending_request.expected_turn_id),
-            accepted_turn_id,
-            num_input_images: pending_request.num_input_images,
-            result,
-            rejection_reason,
-            created_at: pending_request.created_at,
+        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;
         };
         out.push(TrackEventRequest::TurnSteer(CodexTurnSteerEventRequest {
             event_type: "codex_turn_steer_event",
-            event_params: codex_turn_steer_event_params(
-                connection_state.app_server_client.clone(),
-                connection_state.runtime.clone(),
-                &tracking,
-                turn_steer,
-            ),
+            event_params: CodexTurnSteerEventParams {
+                thread_id: pending_request.thread_id,
+                expected_turn_id: Some(pending_request.expected_turn_id),
+                accepted_turn_id,
+                app_server_client: connection_state.app_server_client.clone(),
+                runtime: connection_state.runtime.clone(),
+                thread_source: thread_metadata.thread_source.map(str::to_string),
+                subagent_source: thread_metadata.subagent_source.clone(),
+                parent_thread_id: thread_metadata.parent_thread_id.clone(),
+                num_input_images: pending_request.num_input_images,
+                result,
+                rejection_reason,
+                created_at: pending_request.created_at,
+            },
         }));
     }
 
@@ -807,6 +810,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(
@@ -817,6 +838,7 @@ impl AnalyticsReducer {
                     runtime,
                     turn_id.to_string(),
                     turn_state,
+                    thread_metadata,
                 ),
             },
         )));
@@ -829,6 +851,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(),
@@ -867,6 +890,10 @@ fn codex_turn_event_params(
         runtime,
         submission_type,
         ephemeral,
+        thread_source: thread_metadata.thread_source.map(str::to_string),
+        initialization_mode: thread_metadata.initialization_mode,
+        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());