From bd5b55e403e867f957e371840377cd284023f98c Mon Sep 17 00:00:00 2001 From: Ben Romano Date: Wed, 22 Jul 2026 22:08:22 +0000 Subject: [PATCH] Track compaction time in turn profiles (#34835) ## What changed - Add `compaction_ms` to turn profile facts and turn analytics events. - Measure manual and automatic compaction as an exclusive profile phase instead of including it in surrounding idle time. - Complete the profile and total turn duration from the same instant so their elapsed times remain consistent. ## Testing - Cover compaction phase accounting and profile-duration consistency. - Update turn event serialization coverage for `compaction_ms`. GitOrigin-RevId: 50f0e1cf19789a807f9f56c2ef763c620a5d641c --- .../analytics/src/analytics_client_tests.rs | 9 ++- codex-rs/analytics/src/events.rs | 1 + codex-rs/analytics/src/facts.rs | 1 + codex-rs/analytics/src/reducer.rs | 2 + codex-rs/core/src/session/turn.rs | 1 + codex-rs/core/src/tasks/compact.rs | 1 + codex-rs/core/src/tasks/mod.rs | 12 ++-- codex-rs/core/src/turn_timing.rs | 50 ++++++++++++++--- codex-rs/core/src/turn_timing_tests.rs | 55 +++++++++++++++++++ 9 files changed, 114 insertions(+), 18 deletions(-) diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index 164a31daf5a4..563206258c3a 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -441,9 +441,10 @@ fn sample_turn_profile() -> TurnProfile { TurnProfile { before_first_sampling_ms: 100, sampling_ms: 700, + compaction_ms: 40, between_sampling_overhead_ms: 50, tool_blocking_ms: 250, - after_last_sampling_ms: 134, + after_last_sampling_ms: 94, sampling_request_count: 2, sampling_retry_count: 1, } @@ -4065,9 +4066,10 @@ fn turn_event_serializes_expected_shape() { total_tokens: None, before_first_sampling_ms: 100, sampling_ms: 700, + compaction_ms: 40, between_sampling_overhead_ms: 50, tool_blocking_ms: 250, - after_last_sampling_ms: 134, + after_last_sampling_ms: 94, sampling_request_count: 2, sampling_retry_count: 1, duration_ms: Some(1234), @@ -4138,9 +4140,10 @@ fn turn_event_serializes_expected_shape() { "total_tokens": null, "before_first_sampling_ms": 100, "sampling_ms": 700, + "compaction_ms": 40, "between_sampling_overhead_ms": 50, "tool_blocking_ms": 250, - "after_last_sampling_ms": 134, + "after_last_sampling_ms": 94, "sampling_request_count": 2, "sampling_retry_count": 1, "duration_ms": 1234, diff --git a/codex-rs/analytics/src/events.rs b/codex-rs/analytics/src/events.rs index 223d763e8bdd..82b28fcc0bc2 100644 --- a/codex-rs/analytics/src/events.rs +++ b/codex-rs/analytics/src/events.rs @@ -914,6 +914,7 @@ pub(crate) struct CodexTurnEventParams { pub(crate) total_tokens: Option, pub(crate) before_first_sampling_ms: u64, pub(crate) sampling_ms: u64, + pub(crate) compaction_ms: u64, pub(crate) between_sampling_overhead_ms: u64, pub(crate) tool_blocking_ms: u64, pub(crate) after_last_sampling_ms: u64, diff --git a/codex-rs/analytics/src/facts.rs b/codex-rs/analytics/src/facts.rs index d42391eb19e5..9f6a4a9fcc0a 100644 --- a/codex-rs/analytics/src/facts.rs +++ b/codex-rs/analytics/src/facts.rs @@ -108,6 +108,7 @@ pub struct TurnTokenUsageFact { pub struct TurnProfile { pub before_first_sampling_ms: u64, pub sampling_ms: u64, + pub compaction_ms: u64, pub between_sampling_overhead_ms: u64, pub tool_blocking_ms: u64, pub after_last_sampling_ms: u64, diff --git a/codex-rs/analytics/src/reducer.rs b/codex-rs/analytics/src/reducer.rs index b5ef11f64e38..2fddefbf3b31 100644 --- a/codex-rs/analytics/src/reducer.rs +++ b/codex-rs/analytics/src/reducer.rs @@ -2636,6 +2636,7 @@ fn codex_turn_event_params( let TurnProfile { before_first_sampling_ms, sampling_ms, + compaction_ms, between_sampling_overhead_ms, tool_blocking_ms, after_last_sampling_ms, @@ -2708,6 +2709,7 @@ fn codex_turn_event_params( .map(|token_usage| token_usage.total_tokens), before_first_sampling_ms, sampling_ms, + compaction_ms, between_sampling_overhead_ms, tool_blocking_ms, after_last_sampling_ms, diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index 63ccba77235a..883fc5d962e0 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -989,6 +989,7 @@ async fn run_auto_compact( phase: CompactionPhase, ) -> CodexResult<()> { let turn_context = &step_context.turn; + let _profile_guard = turn_context.turn_timing_state.begin_compaction(); if turn_context.config.features.enabled(Feature::TokenBudget) { // Compaction is the reset request, so force a new context window // instead of consuming a pending `new_context` tool request. diff --git a/codex-rs/core/src/tasks/compact.rs b/codex-rs/core/src/tasks/compact.rs index 60cdebb82929..eba04f1d1065 100644 --- a/codex-rs/core/src/tasks/compact.rs +++ b/codex-rs/core/src/tasks/compact.rs @@ -32,6 +32,7 @@ impl SessionTask for CompactTask { _cancellation_token: CancellationToken, ) -> SessionTaskResult { let session = session.clone_session(); + let _profile_guard = ctx.turn_timing_state.begin_compaction(); if ctx.config.features.enabled(Feature::TokenBudget) { crate::compact_token_budget::run_manual_compact_task(session, ctx).await?; return Ok(None); diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index 9dfa559b7d86..1893568e416e 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -758,15 +758,15 @@ impl Session { turn_had_memory_citation, ); let started_at = turn_context.turn_timing_state.started_at_unix_secs().await; - let (completed_at, duration_ms) = turn_context + let (completed_at, duration_ms, profile) = turn_context .turn_timing_state - .completed_at_and_duration_ms() + .complete_profile_and_duration_ms() .await; self.services .analytics_events_client .track_turn_profile(TurnProfileFact { turn_id: turn_context.sub_id.clone(), - profile: turn_context.turn_timing_state.complete_profile(), + profile, }); let event = if let Some(reason) = abort_reason { self.emit_turn_abort_lifecycle(reason.clone(), turn_context.extension_data.as_ref()) @@ -907,16 +907,16 @@ impl Session { .turn_timing_state .started_at_unix_secs() .await; - let (completed_at, duration_ms) = task + let (completed_at, duration_ms, profile) = task .turn_context .turn_timing_state - .completed_at_and_duration_ms() + .complete_profile_and_duration_ms() .await; self.services .analytics_events_client .track_turn_profile(TurnProfileFact { turn_id: task.turn_context.sub_id.clone(), - profile: task.turn_context.turn_timing_state.complete_profile(), + profile, }); let event = EventMsg::TurnAborted(TurnAbortedEvent { turn_id: Some(task.turn_context.sub_id.clone()), diff --git a/codex-rs/core/src/turn_timing.rs b/codex-rs/core/src/turn_timing.rs index d0f4f46cc3fa..0502ad6ee4c3 100644 --- a/codex-rs/core/src/turn_timing.rs +++ b/codex-rs/core/src/turn_timing.rs @@ -61,6 +61,7 @@ struct TurnProfileState { seen_sampling: bool, before_first_sampling: Duration, sampling: Duration, + compaction: Duration, between_sampling_overhead: Duration, tool_blocking: Duration, pending_idle_after_sampling: Duration, @@ -72,6 +73,7 @@ struct TurnProfileState { #[derive(Clone, Copy, Debug, PartialEq, Eq)] enum TurnProfilePhase { Sampling, + Compaction, ToolBlocking, } @@ -98,13 +100,22 @@ impl TurnTimingState { self.state.lock().await.started_at_unix_secs } - pub(crate) async fn completed_at_and_duration_ms(&self) -> (Option, Option) { + pub(crate) async fn complete_profile_and_duration_ms( + &self, + ) -> (Option, Option, TurnProfile) { + let completed_at_instant = Instant::now(); let state = self.state.lock().await; let completed_at = Some(now_unix_timestamp_secs()); - let duration_ms = state - .started_at - .map(|started_at| i64::try_from(started_at.elapsed().as_millis()).unwrap_or(i64::MAX)); - (completed_at, duration_ms) + let duration_ms = state.started_at.map(|started_at| { + i64::try_from( + completed_at_instant + .saturating_duration_since(started_at) + .as_millis(), + ) + .unwrap_or(i64::MAX) + }); + let profile = self.profile_state().complete(completed_at_instant); + (completed_at, duration_ms, profile) } pub(crate) async fn time_to_first_token_ms(&self) -> Option { @@ -114,10 +125,6 @@ impl TurnTimingState { .map(|duration| i64::try_from(duration.as_millis()).unwrap_or(i64::MAX)) } - pub(crate) fn complete_profile(&self) -> TurnProfile { - self.profile_state().complete(Instant::now()) - } - pub(crate) fn begin_sampling(self: &Arc) -> TurnProfileTimingGuard { let active = self.profile_state().begin_sampling(Instant::now()); TurnProfileTimingGuard { @@ -131,6 +138,15 @@ impl TurnTimingState { self.profile_state().record_sampling_retry(); } + pub(crate) fn begin_compaction(self: &Arc) -> TurnProfileTimingGuard { + let active = self.profile_state().begin_compaction(Instant::now()); + TurnProfileTimingGuard { + timing: Arc::clone(self), + phase: TurnProfilePhase::Compaction, + active, + } + } + pub(crate) fn begin_tool_blocking(self: &Arc) -> TurnProfileTimingGuard { let active = self.profile_state().begin_tool_blocking(Instant::now()); TurnProfileTimingGuard { @@ -235,6 +251,18 @@ impl TurnProfileState { true } + fn begin_compaction(&mut self, now: Instant) -> bool { + if self.completed_profile.is_some() + || self.started_at.is_none() + || self.active_phase.is_some() + { + return false; + } + self.advance(now); + self.active_phase = Some(TurnProfilePhase::Compaction); + true + } + fn end_phase(&mut self, now: Instant, phase: TurnProfilePhase) { if self.completed_profile.is_some() || self.active_phase != Some(phase) { return; @@ -250,6 +278,7 @@ impl TurnProfileState { let elapsed = now.saturating_duration_since(previous); match self.active_phase { Some(TurnProfilePhase::Sampling) => self.sampling += elapsed, + Some(TurnProfilePhase::Compaction) => self.compaction += elapsed, Some(TurnProfilePhase::ToolBlocking) => self.tool_blocking += elapsed, None if self.seen_sampling => self.pending_idle_after_sampling += elapsed, None => self.before_first_sampling += elapsed, @@ -272,6 +301,7 @@ impl TurnProfileState { let mut profile = TurnProfile { before_first_sampling_ms: duration_to_u64_ms(self.before_first_sampling), sampling_ms: duration_to_u64_ms(self.sampling), + compaction_ms: duration_to_u64_ms(self.compaction), between_sampling_overhead_ms: duration_to_u64_ms(self.between_sampling_overhead), tool_blocking_ms: duration_to_u64_ms(self.tool_blocking), after_last_sampling_ms: duration_to_u64_ms(after_last_sampling), @@ -285,12 +315,14 @@ impl TurnProfileState { let classified_ms = profile .before_first_sampling_ms .saturating_add(profile.sampling_ms) + .saturating_add(profile.compaction_ms) .saturating_add(profile.between_sampling_overhead_ms) .saturating_add(profile.tool_blocking_ms) .saturating_add(profile.after_last_sampling_ms); let rounding_ms = total_ms.saturating_sub(classified_ms); match final_phase { Some(TurnProfilePhase::Sampling) => profile.sampling_ms += rounding_ms, + Some(TurnProfilePhase::Compaction) => profile.compaction_ms += rounding_ms, Some(TurnProfilePhase::ToolBlocking) => profile.tool_blocking_ms += rounding_ms, None if self.seen_sampling => profile.after_last_sampling_ms += rounding_ms, None => profile.before_first_sampling_ms += rounding_ms, diff --git a/codex-rs/core/src/turn_timing_tests.rs b/codex-rs/core/src/turn_timing_tests.rs index fcb34b474212..e9748547fede 100644 --- a/codex-rs/core/src/turn_timing_tests.rs +++ b/codex-rs/core/src/turn_timing_tests.rs @@ -186,6 +186,7 @@ fn turn_profile_breaks_down_sampling_blocking_and_retry_overhead() { TurnProfile { before_first_sampling_ms: 100, sampling_ms: 700, + compaction_ms: 0, between_sampling_overhead_ms: 100, tool_blocking_ms: 300, after_last_sampling_ms: 100, @@ -194,3 +195,57 @@ fn turn_profile_breaks_down_sampling_blocking_and_retry_overhead() { } ); } + +#[test] +fn turn_profile_counts_compaction_as_an_exclusive_phase() { + let started_at = Instant::now(); + let mut state = TurnProfileState::default(); + state.start(started_at); + + let _ = state.begin_compaction(started_at + Duration::from_millis(100)); + state.end_phase( + started_at + Duration::from_millis(300), + TurnProfilePhase::Compaction, + ); + let _ = state.begin_sampling(started_at + Duration::from_millis(400)); + state.end_phase( + started_at + Duration::from_millis(700), + TurnProfilePhase::Sampling, + ); + let _ = state.begin_compaction(started_at + Duration::from_millis(800)); + + assert_eq!( + state.complete(started_at + Duration::from_millis(1_100)), + TurnProfile { + before_first_sampling_ms: 200, + sampling_ms: 300, + compaction_ms: 500, + between_sampling_overhead_ms: 0, + tool_blocking_ms: 0, + after_last_sampling_ms: 100, + sampling_request_count: 1, + sampling_retry_count: 0, + } + ); +} + +#[tokio::test] +async fn turn_profile_and_duration_share_a_completion_instant() { + let state = TurnTimingState::default(); + state + .mark_turn_started(Instant::now() - Duration::from_millis(100)) + .await; + + let (_, duration_ms, profile) = state.complete_profile_and_duration_ms().await; + let classified_ms = profile.before_first_sampling_ms + + profile.sampling_ms + + profile.compaction_ms + + profile.between_sampling_overhead_ms + + profile.tool_blocking_ms + + profile.after_last_sampling_ms; + + assert_eq!( + duration_ms, + Some(i64::try_from(classified_ms).expect("profile duration should fit i64")) + ); +}