[codex-analytics] emit internally started turn events

codex-rs/analytics/src/analytics_client_tests.rs

@@ -778,6 +778,33 @@ fn sample_initialize_fact(connection_id: u64) -> AnalyticsFact {
     }
 }
 
+async fn ingest_complete_child_turn(
+    reducer: &mut AnalyticsReducer,
+    events: &mut Vec<TrackEventRequest>,
+    thread_id: &str,
+    turn_id: &str,
+) {
+    for fact in [
+        AnalyticsFact::Custom(CustomAnalyticsFact::TurnResolvedConfig(Box::new(
+            sample_turn_resolved_config(thread_id, turn_id),
+        ))),
+        AnalyticsFact::Custom(CustomAnalyticsFact::TurnProfile(Box::new(
+            TurnProfileFact {
+                turn_id: turn_id.to_string(),
+                profile: sample_turn_profile(),
+            },
+        ))),
+        AnalyticsFact::Notification(Box::new(sample_turn_completed_notification(
+            thread_id,
+            turn_id,
+            AppServerTurnStatus::Completed,
+            /*codex_error_info*/ None,
+        ))),
+    ] {
+        reducer.ingest(fact, events).await;
+    }
+}
+
 fn sample_command_execution_item(
     status: CommandExecutionStatus,
     exit_code: Option<i32>,
@@ -2665,7 +2692,7 @@ async fn subagent_thread_started_publishes_without_initialize() {
 }
 
 #[tokio::test]
-async fn subagent_thread_started_inherits_parent_connection_for_new_thread() {
+async fn subagent_events_use_inherited_connection_unless_turn_connection_is_explicit() {
     let mut reducer = AnalyticsReducer::default();
     let mut events = Vec::new();
     let parent_thread_id =
@@ -2772,6 +2799,68 @@ async fn subagent_thread_started_inherits_parent_connection_for_new_thread() {
         payload[0]["event_params"]["parent_thread_id"],
         "44444444-4444-4444-4444-444444444444"
     );
+
+    events.clear();
+    ingest_complete_child_turn(&mut reducer, &mut events, "thread-review", "turn-inherited").await;
+    let [TrackEventRequest::TurnEvent(event)] = events.as_slice() else {
+        panic!("expected one turn event");
+    };
+    let params = &event.event_params;
+    assert_eq!(params.session_id, "session-root");
+    assert_eq!(params.thread_source, Some(ThreadSource::Subagent));
+    assert_eq!(params.subagent_source.as_deref(), Some("thread_spawn"));
+    assert_eq!(
+        params.parent_thread_id.as_deref(),
+        Some("44444444-4444-4444-4444-444444444444")
+    );
+    assert_eq!(params.app_server_client.product_client_id, "parent-client");
+    assert_eq!(params.runtime.codex_rs_version, "0.1.0");
+
+    reducer
+        .ingest(
+            AnalyticsFact::Custom(CustomAnalyticsFact::TurnTokenUsage(Box::new(
+                sample_turn_token_usage_fact("thread-review", "turn-inherited"),
+            ))),
+            &mut events,
+        )
+        .await;
+    assert_eq!(events.len(), 1);
+
+    events.clear();
+    reducer
+        .ingest(sample_initialize_fact(/*connection_id*/ 8), &mut events)
+        .await;
+    reducer
+        .ingest(
+            AnalyticsFact::ClientRequest {
+                connection_id: 8,
+                request_id: RequestId::Integer(3),
+                request: Box::new(sample_turn_start_request(
+                    "thread-review",
+                    /*request_id*/ 3,
+                )),
+            },
+            &mut events,
+        )
+        .await;
+    reducer
+        .ingest(
+            AnalyticsFact::ClientResponse {
+                connection_id: 8,
+                request_id: RequestId::Integer(3),
+                response: Box::new(sample_turn_start_response("turn-explicit")),
+            },
+            &mut events,
+        )
+        .await;
+    ingest_complete_child_turn(&mut reducer, &mut events, "thread-review", "turn-explicit").await;
+    let [TrackEventRequest::TurnEvent(event)] = events.as_slice() else {
+        panic!("expected one turn event");
+    };
+    assert_eq!(
+        event.event_params.app_server_client.product_client_id,
+        DEFAULT_ORIGINATOR
+    );
 }
 
 #[tokio::test]

codex-rs/analytics/src/reducer.rs

@@ -1497,17 +1497,23 @@ impl AnalyticsReducer {
         let Some(thread_id) = turn_state.thread_id.as_ref() else {
             return;
         };
-        let Some(connection_id) = turn_state.connection_id else {
+        let drop_site = AnalyticsDropSite::turn(thread_id, turn_id);
+        let connection_id = turn_state.connection_id.or_else(|| {
+            self.threads
+                .get(drop_site.thread_id)
+                .and_then(|thread| thread.connection_id)
+        });
+        let Some(connection_id) = connection_id else {
+            warn_missing_analytics_context(&drop_site, MissingAnalyticsContext::ThreadConnection);
             return;
         };
         let Some(connection_state) = self.connections.get(&connection_id) else {
             warn_missing_analytics_context(
-                &AnalyticsDropSite::turn(thread_id, turn_id),
+                &drop_site,
                 MissingAnalyticsContext::Connection { connection_id },
             );
             return;
         };
-        let drop_site = AnalyticsDropSite::turn(thread_id, turn_id);
         let Some(thread_metadata) = self
             .threads
             .get(drop_site.thread_id)