From 96e8afbfb88198cac64c4bc8fa88d027e98b2af8 Mon Sep 17 00:00:00 2001 From: Kyle Brown Date: Wed, 12 Aug 2026 23:45:31 +0000 Subject: [PATCH] Track plugin metrics for background unified exec commands (#38276) ## Why Unified exec can yield while a command is still running. Plugin measurement collection must remain active until that background command exits, including when its item completion arrives after the turn has completed. ## What changed - Keep the plugin metrics sidecar with the stored process and let either the exit watcher or a poll that observes completion finalize it exactly once. - Retain completed turn analytics state while tool items are pending so late command completion events can be emitted without duplicating the turn event. ## Testing - Verify that a background command completed after its turn emits a command execution event and does not emit a second turn event. GitOrigin-RevId: ecf715b3e047aa29ca9a417d12d955257fed8557 --- .../analytics/src/analytics_client_tests.rs | 80 +++++++++++++++++++ codex-rs/analytics/src/reducer.rs | 26 +++++- .../core/src/unified_exec/async_watcher.rs | 15 ++++ .../src/unified_exec/async_watcher_tests.rs | 1 + codex-rs/core/src/unified_exec/mod.rs | 13 +++ codex-rs/core/src/unified_exec/mod_tests.rs | 3 + .../core/src/unified_exec/process_manager.rs | 27 +++++-- .../src/unified_exec/process_manager_tests.rs | 1 + 8 files changed, 158 insertions(+), 8 deletions(-) diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index f609449de3..b594ad5509 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -5062,6 +5062,86 @@ async fn turn_event_counts_completed_tool_items() { assert_eq!(payload["event_params"]["image_generation_count"], json!(1)); } +#[tokio::test] +async fn completed_background_tool_item_emits_after_turn_event() { + let mut reducer = AnalyticsReducer::default(); + let mut out = Vec::new(); + + ingest_turn_prerequisites( + &mut reducer, + &mut out, + /*include_initialize*/ true, + /*include_resolved_config*/ true, + /*include_started*/ true, + /*include_token_usage*/ false, + ) + .await; + + reducer + .ingest( + AnalyticsFact::Notification(Box::new(ServerNotification::ItemStarted( + ItemStartedNotification { + thread_id: "thread-2".to_string(), + turn_id: "turn-2".to_string(), + started_at_ms: 998, + item: sample_command_execution_item( + CommandExecutionStatus::InProgress, + /*exit_code*/ None, + /*duration_ms*/ None, + ), + }, + ))), + &mut out, + ) + .await; + reducer + .ingest( + AnalyticsFact::Notification(Box::new(sample_turn_completed_notification( + "thread-2", + "turn-2", + AppServerTurnStatus::Completed, + /*codex_error_info*/ None, + ))), + &mut out, + ) + .await; + + assert_eq!( + out.iter() + .filter(|event| matches!(event, TrackEventRequest::TurnEvent(_))) + .count(), + 1 + ); + reducer + .ingest( + AnalyticsFact::Notification(Box::new(ServerNotification::ItemCompleted( + ItemCompletedNotification { + thread_id: "thread-2".to_string(), + turn_id: "turn-2".to_string(), + completed_at_ms: 1_000, + item: sample_command_execution_item( + CommandExecutionStatus::Completed, + Some(0), + Some(1), + ), + }, + ))), + &mut out, + ) + .await; + + assert_eq!( + out.iter() + .filter(|event| matches!(event, TrackEventRequest::TurnEvent(_))) + .count(), + 1 + ); + assert!( + out.iter() + .any(|event| matches!(event, TrackEventRequest::CommandExecution(_))) + ); +} + #[tokio::test] async fn item_completed_without_turn_state_does_not_create_turn_state() { let mut reducer = AnalyticsReducer::default(); diff --git a/codex-rs/analytics/src/reducer.rs b/codex-rs/analytics/src/reducer.rs index 0b05cfc4e9..301ed4e5eb 100644 --- a/codex-rs/analytics/src/reducer.rs +++ b/codex-rs/analytics/src/reducer.rs @@ -424,6 +424,7 @@ struct TurnState { steer_count: usize, tool_counts: TurnToolCounts, resource_skill_invocations: HashSet, + turn_event_emitted: bool, } #[derive(Clone, Hash, Eq, PartialEq)] @@ -1757,6 +1758,14 @@ impl AnalyticsReducer { ); } self.item_review_summaries.remove(&key); + if self + .turns + .get(¬ification.turn_id) + .is_some_and(|turn_state| turn_state.turn_event_emitted) + && !self.has_pending_tool_items_for_turn(¬ification.turn_id) + { + self.turns.remove(¬ification.turn_id); + } } ServerNotification::ItemGuardianApprovalReviewStarted(notification) => { let _ = notification; @@ -2128,6 +2137,9 @@ impl AnalyticsReducer { let Some(turn_state) = self.turns.get(turn_id) else { return; }; + if turn_state.turn_event_emitted { + return; + } if turn_state.thread_id.is_none() || turn_state.num_input_images.is_none() || turn_state.resolved_config.is_none() @@ -2179,7 +2191,19 @@ impl AnalyticsReducer { input.repo_hash = accepted_line_repo_hash_for_cwd(cwd.as_path()).await; out.extend(accepted_line_fingerprint_event_requests(input)); } - self.turns.remove(turn_id); + if self.has_pending_tool_items_for_turn(turn_id) { + if let Some(turn_state) = self.turns.get_mut(turn_id) { + turn_state.turn_event_emitted = true; + } + } else { + self.turns.remove(turn_id); + } + } + + fn has_pending_tool_items_for_turn(&self, turn_id: &str) -> bool { + self.tool_items_started_at_ms + .keys() + .any(|key| key.turn_id == turn_id) } /// Resolve the parent connection lazily when a subagent fact arrives first. diff --git a/codex-rs/core/src/unified_exec/async_watcher.rs b/codex-rs/core/src/unified_exec/async_watcher.rs index c301e73298..650d5dfd75 100644 --- a/codex-rs/core/src/unified_exec/async_watcher.rs +++ b/codex-rs/core/src/unified_exec/async_watcher.rs @@ -8,10 +8,13 @@ use tokio::time::Duration; use tokio::time::Instant; use tokio::time::Sleep; +use super::SharedPluginMetricsSidecar; use super::UnifiedExecContext; use super::process::OutputHandles; use super::process::UnifiedExecProcess; +use super::take_plugin_metrics_sidecar; use crate::exec::MAX_EXEC_OUTPUT_DELTAS_PER_CALL; +use crate::plugins::metrics::finish_and_track_measurements; use crate::session::session::Session; use crate::session::turn_context::TurnContext; use crate::tools::events::ToolEmitter; @@ -168,6 +171,7 @@ pub(crate) fn spawn_exit_watcher( transcript: Arc>, started_at: Instant, network_denial_monitor: Option>, + plugin_metrics_sidecar: Option, ) { let exit_token = process.cancellation_token(); let output_drained = process.output_drained_notify(); @@ -185,7 +189,11 @@ pub(crate) fn spawn_exit_watcher( let _interaction_guard = interaction_lock.lock_owned().await; let duration = Instant::now().saturating_duration_since(started_at); + let plugin_metrics_sidecar = plugin_metrics_sidecar + .as_ref() + .and_then(take_plugin_metrics_sidecar); if let Some(message) = process.failure_message() { + drop(plugin_metrics_sidecar); emit_failed_exec_end_for_unified_exec( session_ref, turn_ref, @@ -202,6 +210,13 @@ pub(crate) fn spawn_exit_watcher( .await; } else { let exit_code = process.exit_code().unwrap_or(-1); + finish_and_track_measurements( + plugin_metrics_sidecar, + exit_code, + &session_ref, + &turn_ref, + &call_id, + ); emit_exec_end_for_unified_exec( session_ref, turn_ref, diff --git a/codex-rs/core/src/unified_exec/async_watcher_tests.rs b/codex-rs/core/src/unified_exec/async_watcher_tests.rs index 2899f8cdfe..7f7ccee1ff 100644 --- a/codex-rs/core/src/unified_exec/async_watcher_tests.rs +++ b/codex-rs/core/src/unified_exec/async_watcher_tests.rs @@ -171,6 +171,7 @@ async fn exit_watcher_waits_for_late_network_denial_before_classifying_end() -> transcript, Instant::now(), Some(network_denial_monitor), + /*plugin_metrics_sidecar*/ None, ); let exited_at = Instant::now(); diff --git a/codex-rs/core/src/unified_exec/mod.rs b/codex-rs/core/src/unified_exec/mod.rs index e3a1363ae7..77b23625c6 100644 --- a/codex-rs/core/src/unified_exec/mod.rs +++ b/codex-rs/core/src/unified_exec/mod.rs @@ -43,6 +43,7 @@ use crate::session::turn_context::TurnContext; use crate::session::turn_context::TurnEnvironment; use crate::shell::ShellType; use crate::tools::network_approval::DeferredNetworkApproval; +use codex_core_plugins::PluginMetricsSidecar; mod async_watcher; mod errors; @@ -167,6 +168,7 @@ impl Default for UnifiedExecProcessManager { struct ProcessEntry { process: Arc, + plugin_metrics_sidecar: Option, call_id: String, process_id: i32, cwd: PathUri, @@ -178,6 +180,17 @@ struct ProcessEntry { last_used: tokio::time::Instant, } +type SharedPluginMetricsSidecar = Arc>>; + +fn take_plugin_metrics_sidecar( + sidecar: &SharedPluginMetricsSidecar, +) -> Option { + sidecar + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .take() +} + pub(crate) fn clamp_yield_time(yield_time_ms: u64) -> u64 { let yield_time_ms = if cfg!(windows) { yield_time_ms.max(WINDOWS_INITIAL_EXEC_YIELD_TIME_FLOOR_MS) diff --git a/codex-rs/core/src/unified_exec/mod_tests.rs b/codex-rs/core/src/unified_exec/mod_tests.rs index fd842888b1..d097e783d9 100644 --- a/codex-rs/core/src/unified_exec/mod_tests.rs +++ b/codex-rs/core/src/unified_exec/mod_tests.rs @@ -134,6 +134,7 @@ async fn exec_command_with_tty( if process_started_alive { let entry = ProcessEntry { process: Arc::clone(&process), + plugin_metrics_sidecar: None, call_id: context.call_id.clone(), process_id, cwd: cwd.clone().into(), @@ -607,6 +608,7 @@ async fn terminating_initial_exec_command_rechecks_initial_response_state() -> a process_id, ProcessEntry { process, + plugin_metrics_sidecar: None, call_id: "call".to_string(), process_id, cwd: cwd.into(), @@ -680,6 +682,7 @@ async fn terminating_during_stdin_poll_returns_exited_response() -> anyhow::Resu process_id, ProcessEntry { process: Arc::clone(&process), + plugin_metrics_sidecar: None, call_id: "call".to_string(), process_id, cwd: cwd.into(), diff --git a/codex-rs/core/src/unified_exec/process_manager.rs b/codex-rs/core/src/unified_exec/process_manager.rs index f59ec0fc3b..13292c649c 100644 --- a/codex-rs/core/src/unified_exec/process_manager.rs +++ b/codex-rs/core/src/unified_exec/process_manager.rs @@ -60,6 +60,7 @@ use crate::unified_exec::head_tail_buffer::HeadTailBuffer; use crate::unified_exec::process::OutputHandles; use crate::unified_exec::process::SpawnLifecycleHandle; use crate::unified_exec::process::UnifiedExecProcess; +use crate::unified_exec::take_plugin_metrics_sidecar; use codex_core_plugins::PLUGIN_METRICS_OUTPUT_ENV_VAR; use codex_core_plugins::PluginCommandAttribution; use codex_core_plugins::PluginMetricsSidecar; @@ -538,13 +539,14 @@ impl UnifiedExecProcessManager { request.tty, deferred_network_approval.clone(), network_denial_monitor, + metrics_sidecar, Arc::clone(&transcript), Arc::clone(&initial_exec_command_active), ) .await; InitialExecCommandGuard { active: Some(initial_exec_command_active), - metrics_sidecar, + metrics_sidecar: None, } } else { InitialExecCommandGuard { @@ -629,10 +631,7 @@ impl UnifiedExecProcessManager { exit_code, process_id, .. - } => { - drop(initial_exec_command_guard.metrics_sidecar.take()); - (Some(process_id), exit_code) - } + } => (Some(process_id), exit_code), ProcessStatus::Exited { exit_code, entry } => { if let Err(message) = finish_deferred_network_approval_after_process_exit_for_session( @@ -652,8 +651,17 @@ impl UnifiedExecProcessManager { output_omitted_bytes, ) })?; - initial_exec_command_guard - .finish_plugin_metrics(context, exit_code.unwrap_or(-1)); + let metrics_sidecar = entry + .plugin_metrics_sidecar + .as_ref() + .and_then(take_plugin_metrics_sidecar); + finish_and_track_measurements( + metrics_sidecar, + exit_code.unwrap_or(-1), + &context.session, + &context.step_context.turn, + &context.call_id, + ); (None, exit_code) } ProcessStatus::Unknown => { @@ -988,11 +996,15 @@ impl UnifiedExecProcessManager { tty: bool, network_approval: Option, network_denial_monitor: Option>, + metrics_sidecar: Option, transcript: Arc>, initial_exec_command_active: Arc, ) { + let plugin_metrics_sidecar = + metrics_sidecar.map(|sidecar| Arc::new(std::sync::Mutex::new(Some(sidecar)))); let entry = ProcessEntry { process: Arc::clone(&process), + plugin_metrics_sidecar: plugin_metrics_sidecar.clone(), call_id: context.call_id.clone(), process_id, cwd: cwd.clone(), @@ -1028,6 +1040,7 @@ impl UnifiedExecProcessManager { transcript, started_at, network_denial_monitor, + plugin_metrics_sidecar, ); } diff --git a/codex-rs/core/src/unified_exec/process_manager_tests.rs b/codex-rs/core/src/unified_exec/process_manager_tests.rs index 77ba884d41..4d3429e038 100644 --- a/codex-rs/core/src/unified_exec/process_manager_tests.rs +++ b/codex-rs/core/src/unified_exec/process_manager_tests.rs @@ -581,6 +581,7 @@ async fn pruning_does_not_evict_live_process_while_exited_process_is_finalizing( } else { Arc::clone(&live_process) }, + plugin_metrics_sidecar: None, call_id: format!("call-{process_id}"), process_id, cwd: cwd.clone(),