From 4dae1e175ff9cd6b9104204f3cbe70e3e693e851 Mon Sep 17 00:00:00 2001 From: rhan-oai Date: Wed, 29 Apr 2026 17:30:44 -0700 Subject: [PATCH] [codex-analytics] centralize thread analytics state --- .../analytics/src/analytics_client_tests.rs | 366 ++++++++++++++++++ codex-rs/analytics/src/reducer.rs | 324 ++++++++++++---- 2 files changed, 609 insertions(+), 81 deletions(-) diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index 9f88ceeb94..aaa1316d4a 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -1376,6 +1376,372 @@ async fn subagent_thread_started_publishes_without_initialize() { assert_eq!(payload[0]["event_params"]["subagent_source"], "review"); } +#[tokio::test] +async fn subagent_thread_started_preserves_existing_connection_when_parent_lookup_fails() { + let mut reducer = AnalyticsReducer::default(); + let mut events = Vec::new(); + let parent_thread_id = + codex_protocol::ThreadId::from_string("22222222-2222-2222-2222-222222222222") + .expect("valid parent thread id"); + + ingest_initialize(&mut reducer, &mut events).await; + reducer + .ingest( + AnalyticsFact::ClientResponse { + connection_id: 7, + response: Box::new(sample_thread_resume_response_with_source( + "thread-review", + /*ephemeral*/ false, + "gpt-5", + AppServerSessionSource::SubAgent(SubAgentSource::ThreadSpawn { + parent_thread_id, + depth: 1, + agent_path: None, + agent_nickname: None, + agent_role: None, + }), + )), + }, + &mut events, + ) + .await; + events.clear(); + + reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::SubAgentThreadStarted( + SubAgentThreadStartedInput { + thread_id: "thread-review".to_string(), + parent_thread_id: Some("missing-parent".to_string()), + product_client_id: "codex-tui".to_string(), + client_name: "codex-tui".to_string(), + client_version: "1.0.0".to_string(), + model: "gpt-5".to_string(), + ephemeral: false, + subagent_source: SubAgentSource::Other("guardian".to_string()), + created_at: 128, + }, + )), + &mut events, + ) + .await; + events.clear(); + + reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::Compaction(Box::new( + CodexCompactionEvent { + thread_id: "thread-review".to_string(), + turn_id: "turn-compact".to_string(), + trigger: CompactionTrigger::Manual, + reason: CompactionReason::UserRequested, + implementation: CompactionImplementation::Responses, + phase: CompactionPhase::StandaloneTurn, + strategy: CompactionStrategy::Memento, + status: CompactionStatus::Failed, + error: Some("context limit exceeded".to_string()), + active_context_tokens_before: 131_000, + active_context_tokens_after: 131_000, + started_at: 100, + completed_at: 101, + duration_ms: Some(1200), + }, + ))), + &mut events, + ) + .await; + + let payload = serde_json::to_value(&events).expect("serialize events"); + assert_eq!(payload.as_array().expect("events array").len(), 1); + assert_eq!(payload[0]["event_type"], "codex_compaction_event"); + assert_eq!( + payload[0]["event_params"]["app_server_client"]["product_client_id"], + "codex-tui" + ); + assert_eq!( + payload[0]["event_params"]["subagent_source"], + "thread_spawn" + ); +} + +#[tokio::test] +async fn subagent_thread_started_preserves_existing_connection_when_parent_lookup_succeeds() { + let mut reducer = AnalyticsReducer::default(); + let mut events = Vec::new(); + let parent_thread_id = + codex_protocol::ThreadId::from_string("44444444-4444-4444-4444-444444444444") + .expect("valid parent thread id"); + let parent_thread_id_string = parent_thread_id.to_string(); + + reducer + .ingest( + AnalyticsFact::Initialize { + connection_id: 7, + params: InitializeParams { + client_info: ClientInfo { + name: "parent-client".to_string(), + title: None, + version: "1.0.0".to_string(), + }, + capabilities: None, + }, + product_client_id: "parent-client".to_string(), + runtime: sample_runtime_metadata(), + rpc_transport: AppServerRpcTransport::Stdio, + }, + &mut events, + ) + .await; + reducer + .ingest( + AnalyticsFact::Initialize { + connection_id: 8, + params: InitializeParams { + client_info: ClientInfo { + name: "child-client".to_string(), + title: None, + version: "1.0.0".to_string(), + }, + capabilities: None, + }, + product_client_id: "child-client".to_string(), + runtime: sample_runtime_metadata(), + rpc_transport: AppServerRpcTransport::Stdio, + }, + &mut events, + ) + .await; + reducer + .ingest( + AnalyticsFact::ClientResponse { + connection_id: 7, + response: Box::new(sample_thread_start_response( + &parent_thread_id_string, + /*ephemeral*/ false, + "gpt-5", + )), + }, + &mut events, + ) + .await; + reducer + .ingest( + AnalyticsFact::ClientResponse { + connection_id: 8, + response: Box::new(sample_thread_resume_response_with_source( + "thread-review", + /*ephemeral*/ false, + "gpt-5", + AppServerSessionSource::SubAgent(SubAgentSource::ThreadSpawn { + parent_thread_id, + depth: 1, + agent_path: None, + agent_nickname: None, + agent_role: None, + }), + )), + }, + &mut events, + ) + .await; + events.clear(); + + reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::SubAgentThreadStarted( + SubAgentThreadStartedInput { + thread_id: "thread-review".to_string(), + parent_thread_id: Some(parent_thread_id_string), + product_client_id: "child-client".to_string(), + client_name: "child-client".to_string(), + client_version: "1.0.0".to_string(), + model: "gpt-5".to_string(), + ephemeral: false, + subagent_source: SubAgentSource::Other("guardian".to_string()), + created_at: 130, + }, + )), + &mut events, + ) + .await; + events.clear(); + + reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::Compaction(Box::new( + CodexCompactionEvent { + thread_id: "thread-review".to_string(), + turn_id: "turn-compact".to_string(), + trigger: CompactionTrigger::Manual, + reason: CompactionReason::UserRequested, + implementation: CompactionImplementation::Responses, + phase: CompactionPhase::StandaloneTurn, + strategy: CompactionStrategy::Memento, + status: CompactionStatus::Failed, + error: Some("context limit exceeded".to_string()), + active_context_tokens_before: 131_000, + active_context_tokens_after: 131_000, + started_at: 100, + completed_at: 101, + duration_ms: Some(1200), + }, + ))), + &mut events, + ) + .await; + + let payload = serde_json::to_value(&events).expect("serialize events"); + assert_eq!(payload.as_array().expect("events array").len(), 1); + assert_eq!(payload[0]["event_type"], "codex_compaction_event"); + assert_eq!( + payload[0]["event_params"]["app_server_client"]["product_client_id"], + "child-client" + ); +} + +#[tokio::test] +async fn subagent_thread_started_preserves_resumed_thread_metadata() { + let mut reducer = AnalyticsReducer::default(); + let mut events = Vec::new(); + let parent_thread_id = + codex_protocol::ThreadId::from_string("33333333-3333-3333-3333-333333333333") + .expect("valid parent thread id"); + let parent_thread_id_string = parent_thread_id.to_string(); + + ingest_initialize(&mut reducer, &mut events).await; + reducer + .ingest( + AnalyticsFact::ClientResponse { + connection_id: 7, + response: Box::new(sample_thread_start_response( + &parent_thread_id_string, + /*ephemeral*/ false, + "gpt-5", + )), + }, + &mut events, + ) + .await; + reducer + .ingest( + AnalyticsFact::ClientResponse { + connection_id: 7, + response: Box::new(sample_thread_resume_response_with_source( + "thread-review", + /*ephemeral*/ false, + "gpt-5", + AppServerSessionSource::SubAgent(SubAgentSource::ThreadSpawn { + parent_thread_id, + depth: 1, + agent_path: None, + agent_nickname: None, + agent_role: None, + }), + )), + }, + &mut events, + ) + .await; + events.clear(); + + reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::SubAgentThreadStarted( + SubAgentThreadStartedInput { + thread_id: "thread-review".to_string(), + parent_thread_id: Some(parent_thread_id_string.clone()), + product_client_id: "codex-tui".to_string(), + client_name: "codex-tui".to_string(), + client_version: "1.0.0".to_string(), + model: "gpt-5".to_string(), + ephemeral: false, + subagent_source: SubAgentSource::Other("guardian".to_string()), + created_at: 129, + }, + )), + &mut events, + ) + .await; + events.clear(); + + reducer + .ingest( + AnalyticsFact::ClientRequest { + connection_id: 7, + 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: 7, + response: Box::new(sample_turn_start_response( + "turn-review", + /*request_id*/ 3, + )), + }, + &mut events, + ) + .await; + let mut resolved_config = sample_turn_resolved_config("turn-review"); + resolved_config.thread_id = "thread-review".to_string(); + reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::TurnResolvedConfig(Box::new( + resolved_config, + ))), + &mut events, + ) + .await; + reducer + .ingest( + AnalyticsFact::Notification(Box::new(sample_turn_started_notification( + "thread-review", + "turn-review", + ))), + &mut events, + ) + .await; + reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::TurnTokenUsage(Box::new( + sample_turn_token_usage_fact("thread-review", "turn-review"), + ))), + &mut events, + ) + .await; + reducer + .ingest( + AnalyticsFact::Notification(Box::new(sample_turn_completed_notification( + "thread-review", + "turn-review", + AppServerTurnStatus::Completed, + /*codex_error_info*/ None, + ))), + &mut events, + ) + .await; + + let payload = serde_json::to_value(&events).expect("serialize events"); + assert_eq!(payload.as_array().expect("events array").len(), 1); + assert_eq!(payload[0]["event_type"], "codex_turn_event"); + assert_eq!(payload[0]["event_params"]["initialization_mode"], "resumed"); + assert_eq!( + payload[0]["event_params"]["subagent_source"], + "thread_spawn" + ); + assert_eq!( + payload[0]["event_params"]["parent_thread_id"], + parent_thread_id_string + ); +} + #[test] fn plugin_used_event_serializes_expected_shape() { let tracking = TrackEventsContext { diff --git a/codex-rs/analytics/src/reducer.rs b/codex-rs/analytics/src/reducer.rs index 33e89df011..9be1845522 100644 --- a/codex-rs/analytics/src/reducer.rs +++ b/codex-rs/analytics/src/reducer.rs @@ -74,8 +74,7 @@ pub(crate) struct AnalyticsReducer { requests: HashMap<(u64, RequestId), RequestState>, turns: HashMap, connections: HashMap, - thread_connections: HashMap, - thread_metadata: HashMap, + threads: HashMap, } struct ConnectionState { @@ -83,6 +82,84 @@ struct ConnectionState { runtime: CodexRuntimeMetadata, } +#[derive(Default)] +struct ThreadAnalyticsState { + connection_id: Option, + metadata: Option, +} + +struct ResolvedAnalyticsContext<'a> { + connection_state: &'a ConnectionState, + thread_metadata: Option<&'a ThreadMetadataState>, +} + +enum AnalyticsConnectionSource<'a> { + Thread(&'a str), + Known(u64), + MaybeKnown(Option), +} + +enum ThreadMetadataRequirement<'a> { + NotRequired, + Required(&'a str), +} + +struct AnalyticsDropSite<'a> { + event_name: &'static str, + thread_id: &'a str, + turn_id: Option<&'a str>, + review_id: Option<&'a str>, + item_id: Option<&'a str>, +} + +impl<'a> AnalyticsDropSite<'a> { + fn guardian(input: &'a GuardianReviewEventParams) -> Self { + Self { + event_name: "guardian", + thread_id: &input.thread_id, + turn_id: Some(&input.turn_id), + review_id: Some(&input.review_id), + item_id: None, + } + } + + fn compaction(input: &'a CodexCompactionEvent) -> Self { + Self { + event_name: "compaction", + thread_id: &input.thread_id, + turn_id: Some(&input.turn_id), + review_id: None, + item_id: None, + } + } + + fn turn_steer(thread_id: &'a str) -> Self { + Self { + event_name: "turn steer", + thread_id, + turn_id: None, + review_id: None, + item_id: None, + } + } + + fn turn(thread_id: &'a str, turn_id: &'a str) -> Self { + Self { + event_name: "turn", + thread_id, + turn_id: Some(turn_id), + review_id: None, + item_id: None, + } + } +} + +enum MissingAnalyticsContext { + ThreadConnection, + Connection { connection_id: u64 }, + ThreadMetadata, +} + #[derive(Clone)] struct ThreadMetadataState { thread_source: Option<&'static str>, @@ -271,6 +348,26 @@ impl AnalyticsReducer { input: SubAgentThreadStartedInput, out: &mut Vec, ) { + let parent_thread_id = input + .parent_thread_id + .clone() + .or_else(|| subagent_parent_thread_id(&input.subagent_source)); + 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); + let thread_state = self.threads.entry(input.thread_id.clone()).or_default(); + thread_state + .metadata + .get_or_insert_with(|| ThreadMetadataState { + thread_source: Some("subagent"), + initialization_mode: ThreadInitializationMode::New, + subagent_source: Some(subagent_source_name(&input.subagent_source)), + parent_thread_id, + }); + if thread_state.connection_id.is_none() { + thread_state.connection_id = parent_connection_id; + } out.push(TrackEventRequest::ThreadInitialized( subagent_thread_started_event_request(input), )); @@ -281,23 +378,15 @@ impl AnalyticsReducer { input: GuardianReviewEventParams, out: &mut Vec, ) { - let Some(connection_id) = self.thread_connections.get(&input.thread_id) else { - tracing::warn!( - thread_id = %input.thread_id, - turn_id = %input.turn_id, - review_id = %input.review_id, - "dropping guardian analytics event: missing thread connection metadata" - ); - return; - }; - let Some(connection_state) = self.connections.get(connection_id) else { - tracing::warn!( - thread_id = %input.thread_id, - turn_id = %input.turn_id, - review_id = %input.review_id, - connection_id, - "dropping guardian analytics event: missing connection metadata" - ); + let Some(ResolvedAnalyticsContext { + connection_state, + thread_metadata: None, + }) = self.resolve_analytics_context( + AnalyticsDropSite::guardian(&input), + AnalyticsConnectionSource::Thread(&input.thread_id), + ThreadMetadataRequirement::NotRequired, + ) + else { return; }; out.push(TrackEventRequest::GuardianReview(Box::new( @@ -683,10 +772,13 @@ impl AnalyticsReducer { }; let thread_metadata = ThreadMetadataState::from_thread_metadata(&thread_source, initialization_mode); - self.thread_connections - .insert(thread_id.clone(), connection_id); - self.thread_metadata - .insert(thread_id.clone(), thread_metadata.clone()); + self.threads.insert( + thread_id.clone(), + ThreadAnalyticsState { + connection_id: Some(connection_id), + metadata: Some(thread_metadata.clone()), + }, + ); out.push(TrackEventRequest::ThreadInitialized( ThreadInitializedEvent { event_type: "codex_thread_initialized", @@ -707,29 +799,15 @@ impl AnalyticsReducer { } fn ingest_compaction(&mut self, input: CodexCompactionEvent, out: &mut Vec) { - let Some(connection_id) = self.thread_connections.get(&input.thread_id) else { - tracing::warn!( - thread_id = %input.thread_id, - turn_id = %input.turn_id, - "dropping compaction analytics event: missing thread connection metadata" - ); - return; - }; - let Some(connection_state) = self.connections.get(connection_id) else { - tracing::warn!( - thread_id = %input.thread_id, - turn_id = %input.turn_id, - connection_id, - "dropping compaction analytics event: missing connection metadata" - ); - return; - }; - let Some(thread_metadata) = self.thread_metadata.get(&input.thread_id) else { - tracing::warn!( - thread_id = %input.thread_id, - turn_id = %input.turn_id, - "dropping compaction analytics event: missing thread lifecycle metadata" - ); + let Some(ResolvedAnalyticsContext { + connection_state, + thread_metadata: Some(thread_metadata), + }) = self.resolve_analytics_context( + AnalyticsDropSite::compaction(&input), + AnalyticsConnectionSource::Thread(&input.thread_id), + ThreadMetadataRequirement::Required(&input.thread_id), + ) + else { return; }; out.push(TrackEventRequest::Compaction(Box::new( @@ -781,14 +859,15 @@ impl AnalyticsReducer { rejection_reason: Option, out: &mut Vec, ) { - 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" - ); + let Some(ResolvedAnalyticsContext { + connection_state, + thread_metadata: Some(thread_metadata), + }) = self.resolve_analytics_context( + AnalyticsDropSite::turn_steer(&pending_request.thread_id), + AnalyticsConnectionSource::Known(connection_id), + ThreadMetadataRequirement::Required(&pending_request.thread_id), + ) + else { return; }; out.push(TrackEventRequest::TurnSteer(CodexTurnSteerEventRequest { @@ -821,42 +900,26 @@ impl AnalyticsReducer { { return; } - let connection_metadata = turn_state - .connection_id - .and_then(|connection_id| self.connections.get(&connection_id)) - .map(|connection_state| { - ( - connection_state.app_server_client.clone(), - connection_state.runtime.clone(), - ) - }); - 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" - ); + let Some(ResolvedAnalyticsContext { + connection_state, + thread_metadata: Some(thread_metadata), + }) = self.resolve_analytics_context( + AnalyticsDropSite::turn(thread_id, turn_id), + AnalyticsConnectionSource::MaybeKnown(turn_state.connection_id), + ThreadMetadataRequirement::Required(thread_id), + ) + else { return; }; out.push(TrackEventRequest::TurnEvent(Box::new( CodexTurnEventRequest { event_type: "codex_turn_event", event_params: codex_turn_event_params( - app_server_client, - runtime, + connection_state.app_server_client.clone(), + connection_state.runtime.clone(), turn_id.to_string(), turn_state, thread_metadata, @@ -865,6 +928,105 @@ impl AnalyticsReducer { ))); self.turns.remove(turn_id); } + + fn resolve_analytics_context( + &self, + drop_site: AnalyticsDropSite<'_>, + connection_source: AnalyticsConnectionSource<'_>, + thread_metadata_requirement: ThreadMetadataRequirement<'_>, + ) -> Option> { + let connection_id = match connection_source { + AnalyticsConnectionSource::Thread(thread_id) => { + let Some(thread_state) = self.threads.get(thread_id) else { + warn_missing_analytics_context( + &drop_site, + MissingAnalyticsContext::ThreadConnection, + ); + return None; + }; + let Some(connection_id) = thread_state.connection_id else { + warn_missing_analytics_context( + &drop_site, + MissingAnalyticsContext::ThreadConnection, + ); + return None; + }; + connection_id + } + AnalyticsConnectionSource::Known(connection_id) => connection_id, + AnalyticsConnectionSource::MaybeKnown(None) => return None, + AnalyticsConnectionSource::MaybeKnown(Some(connection_id)) => connection_id, + }; + let Some(connection_state) = self.connections.get(&connection_id) else { + warn_missing_analytics_context( + &drop_site, + MissingAnalyticsContext::Connection { connection_id }, + ); + return None; + }; + let thread_metadata = match thread_metadata_requirement { + ThreadMetadataRequirement::NotRequired => None, + ThreadMetadataRequirement::Required(thread_id) => { + let Some(thread_metadata) = self.thread_metadata(thread_id) else { + warn_missing_analytics_context( + &drop_site, + MissingAnalyticsContext::ThreadMetadata, + ); + return None; + }; + Some(thread_metadata) + } + }; + Some(ResolvedAnalyticsContext { + connection_state, + thread_metadata, + }) + } + + fn thread_metadata(&self, thread_id: &str) -> Option<&ThreadMetadataState> { + self.threads + .get(thread_id) + .and_then(|thread| thread.metadata.as_ref()) + } +} + +fn warn_missing_analytics_context( + drop_site: &AnalyticsDropSite<'_>, + missing: MissingAnalyticsContext, +) { + match missing { + MissingAnalyticsContext::ThreadConnection => { + tracing::warn!( + thread_id = %drop_site.thread_id, + turn_id = ?drop_site.turn_id, + review_id = ?drop_site.review_id, + item_id = ?drop_site.item_id, + "dropping {} analytics event: missing thread connection metadata", + drop_site.event_name + ); + } + MissingAnalyticsContext::Connection { connection_id } => { + tracing::warn!( + thread_id = %drop_site.thread_id, + turn_id = ?drop_site.turn_id, + review_id = ?drop_site.review_id, + item_id = ?drop_site.item_id, + connection_id, + "dropping {} analytics event: missing connection metadata", + drop_site.event_name + ); + } + MissingAnalyticsContext::ThreadMetadata => { + tracing::warn!( + thread_id = %drop_site.thread_id, + turn_id = ?drop_site.turn_id, + review_id = ?drop_site.review_id, + item_id = ?drop_site.item_id, + "dropping {} analytics event: missing thread lifecycle metadata", + drop_site.event_name + ); + } + } } fn codex_turn_event_params(