Measure Guardian approval blocking time per turn (#55591)

## What changed

Record `codex.guardian.turn.blocking_ms` when a turn completes or aborts, with an `outcome` label. Count overlapping approval waits once, exclude idle gaps, and split pending reviews across the active turns they span. Use each turn's telemetry so samples retain the servicing turn's model attribution.

## Testing

Add unit coverage for overlapping waits, idle gaps, late turn completion, and duplicate stops. Extend integration coverage to verify attribution across turns and model changes for yielded command reviews and MCP elicitation, including aborted turns.

GitOrigin-RevId: a3e6a0c32a06fe58ea93119ecd941076100fb82c

codex-rs/ext/guardian-reviewer/src/blocking.rs

@@ -0,0 +1,149 @@
+//! Splits the union of thread-owned approval waits across active turns.
+//! Turn completion exports one sample; pending reviews survive idle gaps and turn changes.
+//! Saved durations are removed when their turn finishes or its handle drops.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::sync::Mutex;
+use std::sync::PoisonError;
+use std::sync::Weak;
+use std::time::Duration;
+use std::time::Instant;
+
+/// Tracks pending approvals independently of the turn currently servicing them.
+#[derive(Default)]
+pub struct ReviewBlocking {
+    state: Mutex<State>,
+}
+
+#[derive(Default)]
+struct State {
+    interval: Interval,
+    turn: Option<String>,
+    completed: HashMap<String, Duration>,
+}
+
+/// Identifies a turn's sample and removes it when the turn is released.
+pub struct ReviewBlockingTurn {
+    id: String,
+    tracker: Weak<ReviewBlocking>,
+    telemetry: Arc<codex_otel::SessionTelemetry>,
+}
+
+impl State {
+    fn start_turn(&mut self, turn_id: &str, now: Instant) {
+        if let Some(previous) = self.turn.replace(turn_id.to_owned()) {
+            self.completed
+                .insert(previous, self.interval.finish_turn(now));
+        }
+        self.interval.start_turn(now);
+    }
+
+    fn finish_turn(&mut self, turn_id: &str, now: Instant) -> Option<Duration> {
+        if self.turn.as_deref() == Some(turn_id) {
+            self.turn = None;
+            Some(self.interval.finish_turn(now))
+        } else {
+            self.completed.remove(turn_id)
+        }
+    }
+}
+
+#[derive(Default)]
+struct Interval {
+    active: usize,
+    since: Option<Instant>,
+    elapsed: Duration,
+}
+
+impl Interval {
+    fn start_turn(&mut self, now: Instant) {
+        self.elapsed = Duration::ZERO;
+        self.since = (self.active > 0).then_some(now);
+    }
+
+    fn finish_turn(&mut self, now: Instant) -> Duration {
+        if let Some(since) = self.since.take() {
+            self.elapsed += now.saturating_duration_since(since);
+        }
+        std::mem::take(&mut self.elapsed)
+    }
+
+    fn enter(&mut self, now: Instant) {
+        if self.active == 0 {
+            self.since = Some(now);
+        }
+        self.active += 1;
+    }
+    fn leave(&mut self, now: Instant) {
+        self.active = self.active.saturating_sub(1);
+        if self.active == 0
+            && let Some(since) = self.since.take()
+        {
+            self.elapsed += now.saturating_duration_since(since);
+        }
+    }
+}
+
+impl ReviewBlocking {
+    pub fn start_turn(
+        self: &Arc<Self>,
+        turn_id: &str,
+        telemetry: Arc<codex_otel::SessionTelemetry>,
+    ) -> ReviewBlockingTurn {
+        self.state
+            .lock()
+            .unwrap_or_else(PoisonError::into_inner)
+            .start_turn(turn_id, Instant::now());
+        ReviewBlockingTurn {
+            id: turn_id.to_owned(),
+            tracker: Arc::downgrade(self),
+            telemetry,
+        }
+    }
+
+    pub fn finish_turn(&self, turn: &ReviewBlockingTurn, outcome: &'static str) {
+        let elapsed = self
+            .state
+            .lock()
+            .unwrap_or_else(PoisonError::into_inner)
+            .finish_turn(&turn.id, Instant::now());
+        if let Some(elapsed) = elapsed {
+            turn.telemetry.record_duration(
+                "codex.guardian.turn.blocking_ms",
+                elapsed,
+                &[("outcome", outcome)],
+            );
+        }
+    }
+
+    pub(crate) fn enter(self: &Arc<Self>) -> BlockingGuard {
+        let mut state = self.state.lock().unwrap_or_else(PoisonError::into_inner);
+        state.interval.enter(Instant::now());
+        BlockingGuard(Arc::clone(self))
+    }
+}
+
+impl Drop for ReviewBlockingTurn {
+    fn drop(&mut self) {
+        if let Some(tracker) = self.tracker.upgrade() {
+            tracker
+                .state
+                .lock()
+                .unwrap_or_else(PoisonError::into_inner)
+                .finish_turn(&self.id, Instant::now());
+        }
+    }
+}
+
+pub(crate) struct BlockingGuard(Arc<ReviewBlocking>);
+impl Drop for BlockingGuard {
+    fn drop(&mut self) {
+        let mut state = self.0.state.lock().unwrap_or_else(PoisonError::into_inner);
+        state.interval.leave(Instant::now());
+    }
+}
+
+#[cfg(test)]
+#[path = "blocking_tests.rs"]
+mod tests;

codex-rs/ext/guardian-reviewer/src/blocking_tests.rs

@@ -0,0 +1,68 @@
+//! Checks overlapping approvals and turn boundaries with an explicit clock.
+
+use super::*;
+use pretty_assertions::assert_eq;
+
+#[test]
+fn overlapping_reviews_span_turns_without_charging_idle_gaps() {
+    let start = Instant::now();
+    let mut interval = Interval::default();
+    interval.start_turn(start);
+    interval.enter(start);
+    interval.enter(start + Duration::from_millis(10));
+    interval.leave(start + Duration::from_millis(30));
+    assert_eq!(interval.elapsed, Duration::ZERO);
+    assert_eq!(
+        interval.finish_turn(start + Duration::from_millis(40)),
+        Duration::from_millis(40)
+    );
+    interval.start_turn(start + Duration::from_millis(60));
+    interval.leave(start + Duration::from_millis(70));
+    assert_eq!(
+        interval.finish_turn(start + Duration::from_millis(80)),
+        Duration::from_millis(10)
+    );
+    // A review that starts while idle contributes only after the next turn starts.
+    interval.enter(start + Duration::from_millis(100));
+    interval.start_turn(start + Duration::from_millis(200));
+    interval.leave(start + Duration::from_millis(210));
+    assert_eq!(
+        interval.finish_turn(start + Duration::from_millis(220)),
+        Duration::from_millis(10)
+    );
+    interval.start_turn(start + Duration::from_millis(230));
+    interval.enter(start + Duration::from_millis(240));
+    interval.leave(start + Duration::from_millis(250));
+    interval.enter(start + Duration::from_millis(270));
+    interval.leave(start + Duration::from_millis(280));
+    assert_eq!(
+        interval.finish_turn(start + Duration::from_millis(300)),
+        Duration::from_millis(20)
+    );
+    interval.start_turn(start + Duration::from_millis(310));
+    assert_eq!(
+        interval.finish_turn(start + Duration::from_millis(320)),
+        Duration::ZERO
+    );
+}
+
+#[test]
+fn late_turn_completion_preserves_both_samples_and_ignores_duplicate_stops() {
+    let start = Instant::now();
+    let mut state = State::default();
+    state.start_turn("a", start);
+    state.interval.enter(start + Duration::from_millis(10));
+    state.start_turn("b", start + Duration::from_millis(30));
+    assert_eq!(
+        state.finish_turn("a", start + Duration::from_millis(50)),
+        Some(Duration::from_millis(20))
+    );
+    assert_eq!(
+        state.finish_turn("a", start + Duration::from_millis(60)),
+        None
+    );
+    assert_eq!(
+        state.finish_turn("b", start + Duration::from_millis(70)),
+        Some(Duration::from_millis(40))
+    );
+}

codex-rs/ext/guardian-reviewer/src/lib.rs

@@ -3,6 +3,9 @@
 //! The host supplies review attempts and enforces the resulting decision on the bound action.
 
 mod assessment;
+mod blocking;
+pub use blocking::ReviewBlocking;
+pub use blocking::ReviewBlockingTurn;
 mod circuit_breaker;
 mod completion;
 mod conversation;

codex-rs/ext/guardian-reviewer/src/routing.rs

@@ -71,6 +71,10 @@ impl<H: ReviewHost> ReviewRequest<'_, H> {
             .telemetry
             .start_timer("codex.guardian.decision.duration_ms", &[])
             .ok();
+        let _blocking = self
+            .thread_store
+            .get::<crate::ReviewBlocking>()
+            .map(|blocking| blocking.enter());
         let runtime = self.thread_store.get::<crate::ReviewerTasks>();
         let _task = runtime.as_ref().map(|runtime| runtime.tasks.token());
         if runtime

codex-rs/ext/guardian-v2/src/sync_reviewer/mod.rs

@@ -22,6 +22,8 @@ use codex_extension_api::TurnAbortInput;
 use codex_extension_api::TurnLifecycleContributor;
 use codex_extension_api::TurnStartInput;
 use codex_extension_api::TurnStopInput;
+use codex_guardian_reviewer::ReviewBlocking;
+use codex_guardian_reviewer::ReviewBlockingTurn;
 use codex_guardian_reviewer::ReviewDenials;
 use codex_guardian_reviewer::ReviewerPool;
 use codex_guardian_reviewer::ReviewerTasks;
@@ -58,6 +60,7 @@ impl ThreadLifecycleContributor<Config> for GuardianExtension {
             let manager = self.thread_manager.clone();
             let parent_thread_id = input.thread_store.level_id().to_owned();
             let runtime = input.thread_store.get_or_init(ReviewerTasks::default);
+            input.thread_store.get_or_init(ReviewBlocking::default);
             input.thread_store.get_or_init(|| {
                 ReviewerPool::<GuardianReviewSession>::new(
                     Arc::clone(&runtime),
@@ -228,19 +231,39 @@ impl ThreadLifecycleContributor<Config> for GuardianExtension {
 
 impl TurnLifecycleContributor for GuardianExtension {
     fn on_turn_start<'a>(&'a self, input: TurnStartInput<'a>) -> ExtensionFuture<'a, ()> {
-        Box::pin(ReviewDenials::clear_turn(input.thread_store, input.turn_id))
+        Box::pin(async move {
+            if input.thread_store.get::<ReviewerTasks>().is_some()
+                && let Some(telemetry) = input.turn_store.get()
+            {
+                let tracker = input.thread_store.get_or_init(ReviewBlocking::default);
+                input
+                    .turn_store
+                    .get_or_init(|| tracker.start_turn(input.turn_id, telemetry));
+            }
+            ReviewDenials::clear_turn(input.thread_store, input.turn_id).await;
+        })
     }
     fn on_turn_stop<'a>(&'a self, input: TurnStopInput<'a>) -> ExtensionFuture<'a, ()> {
-        Box::pin(ReviewDenials::clear_turn(
-            input.thread_store,
-            input.turn_store.level_id(),
-        ))
+        Box::pin(async move {
+            if let (Some(tracker), Some(turn)) = (
+                input.thread_store.get::<ReviewBlocking>(),
+                input.turn_store.get::<ReviewBlockingTurn>(),
+            ) {
+                tracker.finish_turn(&turn, "completed");
+            }
+            ReviewDenials::clear_turn(input.thread_store, input.turn_store.level_id()).await;
+        })
     }
     fn on_turn_abort<'a>(&'a self, input: TurnAbortInput<'a>) -> ExtensionFuture<'a, ()> {
-        Box::pin(ReviewDenials::clear_turn(
-            input.thread_store,
-            input.turn_store.level_id(),
-        ))
+        Box::pin(async move {
+            if let (Some(tracker), Some(turn)) = (
+                input.thread_store.get::<ReviewBlocking>(),
+                input.turn_store.get::<ReviewBlockingTurn>(),
+            ) {
+                tracker.finish_turn(&turn, "aborted");
+            }
+            ReviewDenials::clear_turn(input.thread_store, input.turn_store.level_id()).await;
+        })
     }
 }