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

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,

codex-rs/analytics/src/events.rs

@@ -914,6 +914,7 @@ pub(crate) struct CodexTurnEventParams {
     pub(crate) total_tokens: Option<i64>,
     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,

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,

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,

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.

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);

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()),

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<i64>, Option<i64>) {
+    pub(crate) async fn complete_profile_and_duration_ms(
+        &self,
+    ) -> (Option<i64>, Option<i64>, 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<i64> {
@@ -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<Self>) -> 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<Self>) -> 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<Self>) -> 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,

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"))
+    );
+}