diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index 67008d6fb7..8895c9cfd6 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -3425,7 +3425,6 @@ async fn subagent_tool_items_inherit_parent_connection_metadata() { let mut reducer = AnalyticsReducer::default(); let mut events = Vec::new(); - ingest_review_prerequisites(&mut reducer, &mut events).await; reducer .ingest( AnalyticsFact::Custom(CustomAnalyticsFact::SubAgentThreadStarted( @@ -3446,7 +3445,7 @@ async fn subagent_tool_items_inherit_parent_connection_metadata() { &mut events, ) .await; - events.clear(); + ingest_review_prerequisites(&mut reducer, &mut events).await; reducer .ingest( AnalyticsFact::Notification(Box::new(sample_turn_started_notification( @@ -3492,8 +3491,25 @@ async fn subagent_tool_items_inherit_parent_connection_metadata() { ) .await; + ingest_code_mode_facts( + &mut reducer, + &mut events, + [CodeModeToolCallFact::Completed { + thread_id: "thread-subagent".into(), + turn_id: "turn-subagent".into(), + call_id: "exec-1".into(), + cell_id: None, + tool_name: "exec".into(), + started_at_ms: 1_000, + completed_at_ms: 1_042, + status: CodeModeToolCallStatus::Completed, + }], + ) + .await; + reducer.flush(&mut events); + let payload = serde_json::to_value(&events).expect("serialize events"); - assert_eq!(payload.as_array().expect("events array").len(), 1); + assert_eq!(payload.as_array().expect("events array").len(), 2); assert_eq!(payload[0]["event_type"], "codex_command_execution_event"); assert_eq!(payload[0]["event_params"]["thread_id"], "thread-subagent"); assert_eq!(payload[0]["event_params"]["session_id"], "session-thread-1"); @@ -3504,6 +3520,8 @@ async fn subagent_tool_items_inherit_parent_connection_metadata() { payload[0]["event_params"]["app_server_client"]["client_name"], "codex-tui" ); + assert_eq!(payload[1]["event_type"], "codex_dynamic_tool_call_event"); + assert_eq!(payload[1]["event_params"]["parent_thread_id"], "thread-1"); } #[test] diff --git a/codex-rs/analytics/src/reducer.rs b/codex-rs/analytics/src/reducer.rs index f4b22f6c0c..bb55a94ae5 100644 --- a/codex-rs/analytics/src/reducer.rs +++ b/codex-rs/analytics/src/reducer.rs @@ -647,8 +647,8 @@ impl AnalyticsReducer { }; let has_thread_context = self.threads.get(thread_id).is_some_and(|thread| { thread.metadata.is_some() - && thread - .connection_id + && self + .thread_connection_id(thread_id) .is_some_and(|connection_id| self.connections.contains_key(&connection_id)) }); if !has_thread_context { @@ -940,9 +940,8 @@ impl AnalyticsReducer { ) { let parent_thread_id = input.parent_thread_id.clone(); let parent_connection_id = parent_thread_id - .as_ref() - .and_then(|parent_thread_id| self.threads.get(parent_thread_id)) - .and_then(|thread| thread.connection_id); + .as_deref() + .and_then(|parent_thread_id| self.thread_connection_id(parent_thread_id)); let thread_state = self.threads.entry(input.thread_id.clone()).or_default(); thread_state .originator @@ -2070,11 +2069,9 @@ impl AnalyticsReducer { return; }; 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 connection_id = turn_state + .connection_id + .or_else(|| self.thread_connection_id(drop_site.thread_id)); let Some(connection_id) = connection_id else { warn_missing_analytics_context(&drop_site, MissingAnalyticsContext::ThreadConnection); return; @@ -2114,15 +2111,24 @@ impl AnalyticsReducer { self.turns.remove(turn_id); } + /// Resolve the parent connection lazily when a subagent fact arrives first. + /// + /// Parents are spawned before their children, so ancestor links cannot cycle. + fn thread_connection_id(&self, thread_id: &str) -> Option { + let mut thread = self.threads.get(thread_id)?; + while thread.connection_id.is_none() { + let thread_metadata = thread.metadata.as_ref()?; + let parent_thread_id = thread_metadata.parent_thread_id.as_deref()?; + thread = self.threads.get(parent_thread_id)?; + } + thread.connection_id + } + fn thread_connection_or_warn( &self, drop_site: AnalyticsDropSite<'_>, ) -> Option<&ConnectionState> { - let Some(thread_state) = self.threads.get(drop_site.thread_id) else { - warn_missing_analytics_context(&drop_site, MissingAnalyticsContext::ThreadConnection); - return None; - }; - let Some(connection_id) = thread_state.connection_id else { + let Some(connection_id) = self.thread_connection_id(drop_site.thread_id) else { warn_missing_analytics_context(&drop_site, MissingAnalyticsContext::ThreadConnection); return None; };