From 21b1f5f5344db8adb6eb1be6a2e61f6085fdc8f8 Mon Sep 17 00:00:00 2001 From: jif Date: Wed, 21 Oct 2026 09:46:18 +0000 Subject: [PATCH] Bound retained turn analytics and clean up closed threads (#55847) ## Why Dropped lifecycle facts can leave incomplete turns retaining analytics metadata indefinitely. ## What changed - Limit retained turns to 256, evicting the oldest turn and its associated tool and review state when capacity is reached. - Forward `ThreadClosed` notifications to analytics and discard the closed thread's turn and review state, including reviews received without turn lifecycle facts. ## Testing Add regression tests covering metadata release after capacity eviction, continued reporting of new complete turns, cleanup isolated to the closed thread, and ignored approval responses after thread closure. GitOrigin-RevId: eed0a1fef013381654345d535f03de42b99e8607 --- codex-rs/analytics/src/reducer.rs | 102 +++++++++++----- .../src/tests/suite/approvals_tests.rs | 43 +++++++ .../src/tests/suite/reducer_ordering_tests.rs | 110 ++++++++++++++++++ codex-rs/app-server/src/outgoing_message.rs | 4 +- 4 files changed, 230 insertions(+), 29 deletions(-) diff --git a/codex-rs/analytics/src/reducer.rs b/codex-rs/analytics/src/reducer.rs index 2fb9d7da28..0bebe6771d 100644 --- a/codex-rs/analytics/src/reducer.rs +++ b/codex-rs/analytics/src/reducer.rs @@ -204,10 +204,15 @@ pub(crate) fn valid_plugin_measurement_row(row: &PluginMeasurementRow) -> bool { }) } +// Match the fact queue capacity: analytics must not retain an unlimited number +// of incomplete turns when lifecycle facts are dropped. +const MAX_PENDING_TURNS: usize = 256; + #[derive(Default)] pub(crate) struct AnalyticsReducer { requests: HashMap<(u64, RequestId), RequestState>, turns: HashMap, + turn_order: VecDeque, connections: HashMap, threads: HashMap, tool_items_started_at_ms: HashMap)>, @@ -552,6 +557,35 @@ impl TurnToolCounts { } impl AnalyticsReducer { + fn turn_state(&mut self, turn_id: String) -> &mut TurnState { + if !self.turns.contains_key(&turn_id) { + // Remove stale ordering entries before evicting the oldest retained turn. + // At capacity this may discard incomplete or pending background-tool analytics. + self.turn_order.retain(|id| self.turns.contains_key(id)); + if self.turns.len() >= MAX_PENDING_TURNS + && let Some(oldest) = self.turn_order.pop_front() + { + self.discard_turn(&oldest); + } + self.turn_order.push_back(turn_id.clone()); + } + self.turns.entry(turn_id).or_default() + } + + fn discard_turn(&mut self, turn_id: &str) { + self.turns.remove(turn_id); + self.tool_items_started_at_ms + .retain(|key, _| key.turn_id != turn_id); + self.tool_response_states + .retain(|(_, pending_turn_id), _| pending_turn_id != turn_id); + self.pending_reviews + .retain(|_, review| review.turn_id != turn_id); + self.item_review_summaries + .retain(|key, _| key.turn_id != turn_id); + self.pending_mcp_tool_elicitations + .retain(|(key, _)| key.turn_id != turn_id); + } + pub(crate) async fn ingest(&mut self, input: AnalyticsFact, out: &mut Vec) { match input { AnalyticsFact::Initialize { @@ -1002,7 +1036,7 @@ impl AnalyticsReducer { output_audio_item_count: None, }, }); - let counts = &mut self.turns.entry(turn_id.clone()).or_default().tool_counts; + let counts = &mut self.turn_state(turn_id.clone()).tool_counts; counts.total += 1; counts.dynamic_tool_call += 1; self.record_tool_event( @@ -1081,11 +1115,7 @@ impl AnalyticsReducer { success: status == ControlToolCallStatus::Completed, }, }); - self.turns - .entry(turn_id.clone()) - .or_default() - .tool_counts - .total += 1; + self.turn_state(turn_id.clone()).tool_counts.total += 1; self.record_tool_event( &thread_id, &turn_id, @@ -1334,7 +1364,7 @@ impl AnalyticsReducer { let turn_id = input.turn_id.clone(); let thread_id = input.thread_id.clone(); let num_input_images = input.num_input_images; - let turn_state = self.turns.entry(turn_id.clone()).or_default(); + let turn_state = self.turn_state(turn_id.clone()); turn_state.thread_id = Some(thread_id); turn_state.num_input_images = Some(num_input_images); // Keep the first received plugin inventory, including unknown or empty, @@ -1353,7 +1383,7 @@ impl AnalyticsReducer { out: &mut Vec, ) { let turn_id = input.turn_id.clone(); - let turn_state = self.turns.entry(turn_id.clone()).or_default(); + let turn_state = self.turn_state(turn_id.clone()); turn_state.thread_id = Some(input.thread_id); turn_state.token_usage = Some(input.token_usage); self.maybe_emit_turn_event(&turn_id, out).await; @@ -1365,7 +1395,7 @@ impl AnalyticsReducer { out: &mut Vec, ) { let TurnProfileFact { turn_id, profile } = input; - let turn_state = self.turns.entry(turn_id.clone()).or_default(); + let turn_state = self.turn_state(turn_id.clone()); turn_state.profile = Some(profile); self.maybe_emit_turn_event(&turn_id, out).await; } @@ -1376,13 +1406,13 @@ impl AnalyticsReducer { thread_id, error, } = input; - let turn_state = self.turns.entry(turn_id).or_default(); + let turn_state = self.turn_state(turn_id); turn_state.thread_id.get_or_insert(thread_id); turn_state.codex_error = Some(error); } fn ingest_image_preparation(&mut self, input: ImagePreparationFact) { - let turn_state = self.turns.entry(input.turn_id).or_default(); + let turn_state = self.turn_state(input.turn_id); turn_state.image_preparations.push(input.metadata); } @@ -1427,7 +1457,7 @@ impl AnalyticsReducer { scope, } => { if matches!(invocation.invocation_type, InvocationType::Implicit) { - let turn_state = self.turns.entry(tracking.turn_id.clone()).or_default(); + let turn_state = self.turn_state(tracking.turn_id.clone()); if !turn_state.resource_skill_invocations.insert(id.clone()) { continue; } @@ -1663,7 +1693,7 @@ impl AnalyticsReducer { else { return; }; - let turn_state = self.turns.entry(turn_id.clone()).or_default(); + let turn_state = self.turn_state(turn_id.clone()); turn_state.connection_id = Some(connection_id); turn_state.thread_id = Some(pending_request.thread_id); turn_state.num_input_images = Some(pending_request.num_input_images); @@ -1682,7 +1712,7 @@ impl AnalyticsReducer { return; }; let turn_id = pending_request.turn_id; - let turn_state = self.turns.entry(turn_id.clone()).or_default(); + let turn_state = self.turn_state(turn_id.clone()); let earliest_requested_at_ms = turn_state .explicit_client_interrupt_requested_at_ms .get_or_insert(pending_request.requested_at_ms); @@ -1728,6 +1758,7 @@ impl AnalyticsReducer { let Some(started_at_ms) = option_i64_to_u64(Some(params.started_at_ms)) else { return; }; + self.turn_state(params.turn_id.clone()).thread_id = Some(params.thread_id.clone()); self.pending_reviews.insert( request_id.clone(), PendingReviewState { @@ -1761,6 +1792,7 @@ impl AnalyticsReducer { let Some(started_at_ms) = option_i64_to_u64(Some(params.started_at_ms)) else { return; }; + self.turn_state(params.turn_id.clone()).thread_id = Some(params.thread_id.clone()); self.pending_reviews.insert( request_id.clone(), PendingReviewState { @@ -1800,6 +1832,7 @@ impl AnalyticsReducer { let Some(started_at_ms) = option_i64_to_u64(Some(params.started_at_ms)) else { return; }; + self.turn_state(params.turn_id.clone()).thread_id = Some(params.thread_id.clone()); self.pending_reviews.insert( request_id.clone(), PendingReviewState { @@ -2024,6 +2057,9 @@ impl AnalyticsReducer { ThreadItem::CommandExecution { model_context, .. } => model_context.clone(), _ => None, }; + self.turn_state(notification.turn_id.clone()) + .thread_id + .get_or_insert(notification.thread_id.clone()); self.tool_items_started_at_ms .entry(ToolItemKey { thread_id: notification.thread_id, @@ -2162,6 +2198,17 @@ impl AnalyticsReducer { true } }); + let closed_turns: Vec<_> = self + .turns + .iter() + .filter(|(_, turn)| turn.thread_id.as_deref() == Some(¬ification.thread_id)) + .map(|(turn_id, _)| turn_id.clone()) + .collect(); + for turn_id in closed_turns { + self.discard_turn(&turn_id); + } + self.pending_reviews + .retain(|_, review| review.thread_id != notification.thread_id); self.code_mode_cells.remove(¬ification.thread_id); self.pending_mcp_tool_elicitations .retain(|(key, _)| key.thread_id != notification.thread_id); @@ -2184,14 +2231,18 @@ impl AnalyticsReducer { } ServerNotification::TurnStarted(notification) => { let voice_session_id = { - let thread = self.threads.entry(notification.thread_id).or_default(); + let thread = self + .threads + .entry(notification.thread_id.clone()) + .or_default(); thread.active_turn_id = Some(notification.turn.id.clone()); thread .pending_voice_handoffs .pop_front() .unwrap_or_else(|| thread.active_voice_session_id()) }; - let turn_state = self.turns.entry(notification.turn.id).or_default(); + let turn_state = self.turn_state(notification.turn.id); + turn_state.thread_id = Some(notification.thread_id); turn_state.voice_session_id = voice_session_id; turn_state.started_at = notification .turn @@ -2199,7 +2250,7 @@ impl AnalyticsReducer { .and_then(|started_at| u64::try_from(started_at).ok()); } ServerNotification::TurnDiffUpdated(notification) => { - let turn_state = self.turns.entry(notification.turn_id.clone()).or_default(); + let turn_state = self.turn_state(notification.turn_id.clone()); turn_state.thread_id = Some(notification.thread_id); turn_state.latest_diff = Some(notification.diff); } @@ -2210,7 +2261,8 @@ impl AnalyticsReducer { thread.active_turn_id = None; } self.flush_pending_tool_events(¬ification.thread_id, ¬ification.turn.id, out); - let turn_state = self.turns.entry(notification.turn.id.clone()).or_default(); + let turn_state = self.turn_state(notification.turn.id.clone()); + turn_state.thread_id = Some(notification.thread_id); turn_state.completed = Some(CompletedTurnState { status: analytics_turn_status(notification.turn.status), turn_error: notification @@ -2382,11 +2434,8 @@ impl AnalyticsReducer { return; }; session.session_id.clone_from(&realtime_session_id); - if let Some(turn_id) = &thread.active_turn_id { - self.turns - .entry(turn_id.clone()) - .or_default() - .voice_session_id = realtime_session_id; + if let Some(turn_id) = thread.active_turn_id.clone() { + self.turn_state(turn_id).voice_session_id = realtime_session_id; } } @@ -2404,11 +2453,8 @@ impl AnalyticsReducer { started_at, closed_at: None, }); - if let Some(turn_id) = &thread.active_turn_id { - self.turns - .entry(turn_id.clone()) - .or_default() - .voice_session_id = realtime_session_id; + if let Some(turn_id) = thread.active_turn_id.clone() { + self.turn_state(turn_id).voice_session_id = realtime_session_id; } } diff --git a/codex-rs/analytics/src/tests/suite/approvals_tests.rs b/codex-rs/analytics/src/tests/suite/approvals_tests.rs index 8ac34d2a1b..03caca5d5c 100644 --- a/codex-rs/analytics/src/tests/suite/approvals_tests.rs +++ b/codex-rs/analytics/src/tests/suite/approvals_tests.rs @@ -56,6 +56,7 @@ use codex_app_server_protocol::RequestPermissionProfile; use codex_app_server_protocol::ServerNotification; use codex_app_server_protocol::ServerRequest; use codex_app_server_protocol::ServerResponse; +use codex_app_server_protocol::ThreadClosedNotification; use codex_login::default_client::DEFAULT_ORIGINATOR; use codex_protocol::approvals::NetworkApprovalProtocol; use codex_protocol::models::AdditionalPermissionProfile; @@ -1051,3 +1052,45 @@ fn permission_metadata_preserves_absent_and_empty_requests() { ); } } + +#[tokio::test] +async fn closing_thread_discards_review_without_turn_lifecycle_facts() { + let mut reducer = AnalyticsReducer::default(); + let mut events = Vec::new(); + ingest_review_prerequisites(&mut reducer, &mut events).await; + reducer + .ingest( + AnalyticsFact::ServerRequest { + connection_id: 1, + request: Box::new(sample_command_approval_request( + /*request_id*/ 1, /*approval_id*/ None, + )), + }, + &mut events, + ) + .await; + reducer + .ingest( + AnalyticsFact::Notification(Box::new(ServerNotification::ThreadClosed( + ThreadClosedNotification { + thread_id: "thread-1".to_string(), + }, + ))), + &mut events, + ) + .await; + events.clear(); + reducer + .ingest( + AnalyticsFact::ServerResponse { + completed_at_ms: 2_000, + response: Box::new(sample_command_approval_response( + /*request_id*/ 1, + CommandExecutionApprovalDecision::Accept, + )), + }, + &mut events, + ) + .await; + assert!(events.is_empty()); +} diff --git a/codex-rs/analytics/src/tests/suite/reducer_ordering_tests.rs b/codex-rs/analytics/src/tests/suite/reducer_ordering_tests.rs index 3ac775b61f..1275543ce4 100644 --- a/codex-rs/analytics/src/tests/suite/reducer_ordering_tests.rs +++ b/codex-rs/analytics/src/tests/suite/reducer_ordering_tests.rs @@ -43,6 +43,7 @@ use codex_app_server_protocol::RequestId; use codex_app_server_protocol::ServerNotification; use codex_app_server_protocol::ThreadArchiveParams; use codex_app_server_protocol::ThreadArchiveResponse; +use codex_app_server_protocol::ThreadClosedNotification; use codex_app_server_protocol::ThreadRealtimeClosedNotification; use codex_app_server_protocol::ThreadRealtimeStartedNotification; use codex_app_server_protocol::TurnStatus as AppServerTurnStatus; @@ -692,3 +693,112 @@ async fn voice_handoff_steering_active_turn_does_not_tag_next_text_turn() { .expect("serialize text app event"); assert_eq!(text_app["event_params"]["voice_session_id"], json!(null)); } + +#[tokio::test] +async fn incomplete_turns_release_metadata_when_capacity_is_exceeded() { + let mut reducer = AnalyticsReducer::default(); + let mut events = Vec::new(); + let mut retained = Vec::new(); + ingest_review_prerequisites(&mut reducer, &mut events).await; + events.clear(); + + // Simulate a full queue dropping completion/profile facts for many turns. + for index in 0..1_024 { + let config = sample_turn_resolved_config("thread-1", &format!("lost-{index}")); + retained.push(std::sync::Arc::downgrade(&config.turn_metadata)); + if index % 2 == 0 { + // Tool completions can arrive before the surviving config fact. + reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::ControlToolCall( + ControlToolCallFact { + thread_id: config.thread_id.clone(), + turn_id: config.turn_id.clone(), + turn_metadata: config.turn_metadata.clone(), + call_id: format!("tool-{index}"), + cell_id: None, + tool_name: "view_image".to_string(), + started_at_ms: 100, + completed_at_ms: 200, + status: ControlToolCallStatus::Completed, + }, + )), + &mut events, + ) + .await; + } + reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::TurnResolvedConfig(Box::new(config))), + &mut events, + ) + .await; + } + assert!(retained.first().unwrap().upgrade().is_none()); + assert!(retained.last().unwrap().upgrade().is_some()); + assert!(events.is_empty()); + + // Dropped history must not prevent a new, complete turn from being reported. + ingest_turn_prerequisites( + &mut reducer, + &mut events, + /*include_initialize*/ true, + /*include_resolved_config*/ true, + /*include_started*/ true, + /*include_token_usage*/ true, + ) + .await; + events.clear(); + reducer + .ingest( + AnalyticsFact::Notification(Box::new(sample_turn_completed_notification( + "thread-2", + "turn-2", + AppServerTurnStatus::Completed, + /*codex_error_info*/ None, + ))), + &mut events, + ) + .await; + assert_eq!( + events + .iter() + .filter_map(|event| match event { + TrackEventRequest::TurnEvent(event) => Some(event.event_params.turn_id.as_str()), + _ => None, + }) + .collect::>(), + vec!["turn-2"] + ); +} + +#[tokio::test] +async fn closing_thread_releases_incomplete_turns_without_affecting_other_threads() { + let mut reducer = AnalyticsReducer::default(); + let mut events = Vec::new(); + let closed = sample_turn_resolved_config("closed-thread", "closed-turn"); + let live = sample_turn_resolved_config("live-thread", "live-turn"); + let closed_metadata = std::sync::Arc::downgrade(&closed.turn_metadata); + let live_metadata = std::sync::Arc::downgrade(&live.turn_metadata); + for config in [closed, live] { + reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::TurnResolvedConfig(Box::new(config))), + &mut events, + ) + .await; + } + reducer + .ingest( + AnalyticsFact::Notification(Box::new(ServerNotification::ThreadClosed( + ThreadClosedNotification { + thread_id: "closed-thread".to_string(), + }, + ))), + &mut events, + ) + .await; + assert!(closed_metadata.upgrade().is_none()); + assert!(live_metadata.upgrade().is_some()); + assert!(events.is_empty()); +} diff --git a/codex-rs/app-server/src/outgoing_message.rs b/codex-rs/app-server/src/outgoing_message.rs index 92e5a88391..f893d3d20e 100644 --- a/codex-rs/app-server/src/outgoing_message.rs +++ b/codex-rs/app-server/src/outgoing_message.rs @@ -758,7 +758,9 @@ impl OutgoingMessageSender { pub(crate) async fn send_server_notification(&self, notification: ServerNotification) { if matches!( notification, - ServerNotification::ThreadArchived(_) | ServerNotification::ThreadUnarchived(_) + ServerNotification::ThreadArchived(_) + | ServerNotification::ThreadClosed(_) + | ServerNotification::ThreadUnarchived(_) ) { self.analytics_events_client .track_notification(¬ification);