[codex-analytics] denormalize thread metadata onto turn events

codex-rs/analytics/src/analytics_client_tests.rs

@@ -456,6 +456,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

codex-rs/analytics/src/reducer.rs

@@ -765,6 +765,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 +793,7 @@ impl AnalyticsReducer {
                     runtime,
                     turn_id.to_string(),
                     turn_state,
+                    thread_metadata,
                 ),
             },
         )));
@@ -787,6 +806,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 +823,7 @@ fn codex_turn_event_params(
         num_input_images: _resolved_num_input_images,
         submission_type,
         ephemeral,
-        session_source,
+        session_source: _session_source,
         initialization_mode,
         model,
         model_provider,
@@ -826,10 +846,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,
-        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());

codex-rs/core/src/codex.rs

@@ -1094,6 +1094,14 @@ impl TurnContext {
     }
 }
 
+fn thread_initialization_mode(initial_history: &InitialHistory) -> ThreadInitializationMode {
+    match initial_history {
+        InitialHistory::New | InitialHistory::Cleared => ThreadInitializationMode::New,
+        InitialHistory::Forked(_) => ThreadInitializationMode::Forked,
+        InitialHistory::Resumed(_) => ThreadInitializationMode::Resumed,
+    }
+}
+
 fn local_time_context() -> (String, String) {
     match iana_time_zone::get_timezone() {
         Ok(timezone) => (Local::now().format("%Y-%m-%d").to_string(), timezone),
@@ -1622,6 +1630,8 @@ impl Session {
             session_configuration.collaboration_mode.model(),
             session_configuration.provider
         );
+        session_configuration.thread_initialization_mode =
+            thread_initialization_mode(&initial_history);
         let forked_from_id = initial_history.forked_from_id();
 
         let (conversation_id, rollout_params) = match &initial_history {

codex-rs/core/src/codex_tests.rs

@@ -15,6 +15,7 @@ use crate::mcp_tool_exposure::build_mcp_tool_exposure;
 use crate::shell::default_user_shell;
 use crate::tools::format_exec_output_str;
 
+use codex_analytics::ThreadInitializationMode;
 use codex_features::Features;
 use codex_login::CodexAuth;
 use codex_mcp::CODEX_APPS_MCP_SERVER_NAME;