[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,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());
 }
 
@@ -1626,6 +1650,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

@@ -86,6 +86,13 @@ pub(crate) struct CodexRuntimeMetadata {
     pub(crate) runtime_arch: String,
 }
 
+pub(crate) struct ThreadMetadata {
+    pub(crate) thread_source: Option<&'static str>,
+    pub(crate) initialization_mode: ThreadInitializationMode,
+    pub(crate) subagent_source: Option<String>,
+    pub(crate) parent_thread_id: Option<String>,
+}
+
 #[derive(Serialize)]
 pub(crate) struct ThreadInitializedEventParams {
     pub(crate) thread_id: String,
@@ -341,6 +348,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 +401,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>,
@@ -528,6 +543,7 @@ pub(crate) fn codex_turn_steer_event_params(
     app_server_client: CodexAppServerClientMetadata,
     runtime: CodexRuntimeMetadata,
     tracking: &TrackEventsContext,
+    thread_metadata: ThreadMetadata,
     turn_steer: CodexTurnSteerEvent,
 ) -> CodexTurnSteerEventParams {
     CodexTurnSteerEventParams {
@@ -536,6 +552,10 @@ pub(crate) fn codex_turn_steer_event_params(
         accepted_turn_id: turn_steer.accepted_turn_id,
         app_server_client,
         runtime,
+        thread_source: thread_metadata.thread_source.map(str::to_string),
+        initialization_mode: thread_metadata.initialization_mode,
+        subagent_source: thread_metadata.subagent_source,
+        parent_thread_id: thread_metadata.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

@@ -16,6 +16,7 @@ use crate::events::SkillInvocationEventParams;
 use crate::events::SkillInvocationEventRequest;
 use crate::events::ThreadInitializedEvent;
 use crate::events::ThreadInitializedEventParams;
+use crate::events::ThreadMetadata;
 use crate::events::TrackEventRequest;
 use crate::events::codex_app_metadata;
 use crate::events::codex_compaction_event_params;
@@ -86,12 +87,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 +111,7 @@ impl ThreadMetadataState {
         };
         Self {
             thread_source: thread_source_name(session_source),
+            initialization_mode,
             subagent_source,
             parent_thread_id,
         }
@@ -657,7 +663,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,6 +766,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,
@@ -781,6 +795,12 @@ impl AnalyticsReducer {
                 connection_state.app_server_client.clone(),
                 connection_state.runtime.clone(),
                 &tracking,
+                ThreadMetadata {
+                    thread_source: thread_metadata.thread_source,
+                    initialization_mode: thread_metadata.initialization_mode,
+                    subagent_source: thread_metadata.subagent_source.clone(),
+                    parent_thread_id: thread_metadata.parent_thread_id.clone(),
+                },
                 turn_steer,
             ),
         }));
@@ -807,6 +827,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 +855,7 @@ impl AnalyticsReducer {
                     runtime,
                     turn_id.to_string(),
                     turn_state,
+                    thread_metadata,
                 ),
             },
         )));
@@ -829,6 +868,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 +907,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());