openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-21 17:48:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

Bound retained turn analytics and clean up closed threads (#55847)

Browse Source 
## 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
This commit is contained in:
jif
2026-10-21 09:54:36 +00:00
committed by copyberry
parent 2213359bd3
commit 21b1f5f534
4 changed files with 230 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/analytics/src/reducer.rs
+74 -28
View File
@@ -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<String, TurnState>,
turn_order: VecDeque<String>,
connections: HashMap<u64, ConnectionState>,
threads: HashMap<String, ThreadAnalyticsState>,
tool_items_started_at_ms: HashMap<ToolItemKey, (u64, Option<ModelInvocationContext>)>,
@@ -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<TrackEventRequest>) {
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<TrackEventRequest>,
) {
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<TrackEventRequest>,
) {
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(&notification.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(&notification.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(&notification.thread_id, &notification.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;
}
}
codex-rs/analytics/src/tests/suite/approvals_tests.rs
+43
View File
@@ -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());
}
codex-rs/analytics/src/tests/suite/reducer_ordering_tests.rs
+110
View File
@@ -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<_>>(),
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());
}
codex-rs/app-server/src/outgoing_message.rs
+3 -1
View File
@@ -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(&notification);