Measure shell snapshot use and wait time per command (#51347)

## What changed

Add `codex.shell_snapshot.command` and `codex.shell_snapshot.wait_ms` metrics for eligible command preparations in the unified exec runtime and exec server. Tag observations with snapshot version, availability state, and whether replay was selected (`used`) or normal shell startup was used (`fallback`). Exclude background prewarming from command observations; `used` measures replay selection, not successful restoration or command completion.

## Testing

Extend snapshot retry tests to verify command counts and wait observations across fallback and recovery, excluding prewarm captures. Extend credential-broker approval tests to verify one `protected` snapshot observation for login-shell commands.

GitOrigin-RevId: d3d9f370b6bbba6dfc930a44044ce09373841799

codex-rs/exec-server/src/shell_snapshot.rs

@@ -1,5 +1,6 @@
 //! Cache captured shell state in executor memory. Replay through a per-launch
 //! unnamed reader when the capture sandbox permits it; otherwise retain env replay.
+//! Command metrics exclude prewarm and measure lookup/capture wait before replay.
 
 use std::collections::HashMap;
 use std::collections::VecDeque;
@@ -63,6 +64,7 @@ struct CachedShellSnapshot {
 }
 
 struct ShellSnapshot {
+    prewarmed: bool,
     state: String,
     file_source: bool,
     environment: HashMap<String, String>,
@@ -121,6 +123,7 @@ impl ShellSnapshotCache {
             }
         };
 
+        let wait_started_at = std::time::Instant::now();
         let (snapshot, attempt) = {
             let mut entries = self.entries.lock().await;
             let position = entries.iter().position(|entry| {
@@ -166,14 +169,27 @@ impl ShellSnapshotCache {
                 (snapshot, 1)
             }
         };
+        let mut availability = match snapshot.get() {
+            Some(Ok(snapshot)) if snapshot.prewarmed => "prewarm_ready",
+            Some(Ok(_)) => "cache_hit",
+            Some(Err(_)) => "unavailable",
+            None => "capture_pending",
+        };
         let capture = async {
+            let prewarmed = purpose == CapturePurpose::Prewarm;
             let attempt = attempt.to_string();
             let purpose = match purpose {
                 CapturePurpose::Execution => "execution",
                 CapturePurpose::Prewarm => "prewarm",
             };
             let started_at = std::time::Instant::now();
-            let result = capture_snapshot(params, prepared, shell_type).await;
+            let result =
+                capture_snapshot(params, prepared, shell_type)
+                    .await
+                    .map(|mut snapshot| {
+                        snapshot.prewarmed = prewarmed;
+                        snapshot
+                    });
             telemetry.shell_snapshot_captured(
                 started_at.elapsed(),
                 result.as_ref().map(|_| ()).map_err(|(reason, _)| *reason),
@@ -190,6 +206,7 @@ impl ShellSnapshotCache {
             CapturePurpose::Execution => {
                 snapshot
                     .get_or_init(|| async {
+                        availability = "on_demand";
                         capture.await.map_err(|err| {
                             tracing::warn!("failed to capture shell snapshot: {err:?}");
                             Instant::now() + SNAPSHOT_RETRY_BACKOFF
@@ -205,12 +222,17 @@ impl ShellSnapshotCache {
                     .await?
             }
         };
-        let Ok(snapshot) = snapshot else {
-            return Ok(None);
-        };
+        let wait = wait_started_at.elapsed();
         if purpose == CapturePurpose::Prewarm {
             return Ok(None);
         }
+        let Ok(snapshot) = snapshot else {
+            telemetry.shell_snapshot_command(wait, availability, "fallback");
+            return Ok(None);
+        };
+        if availability == "capture_pending" && snapshot.prewarmed {
+            availability = "prewarm_pending";
+        }
 
         // POSIX sh cannot portably close arbitrary descriptors above 9. Keep its
         // existing replay until the launcher supports remapping child descriptors.
@@ -227,6 +249,7 @@ impl ShellSnapshotCache {
                         ?error,
                         "cannot prepare shell snapshot transport; using normal startup"
                     );
+                    telemetry.shell_snapshot_command(wait, availability, "fallback");
                     return Ok(None);
                 }
             }
@@ -293,6 +316,7 @@ impl ShellSnapshotCache {
             params.argv[2]
         );
 
+        telemetry.shell_snapshot_command(wait, availability, "used");
         Ok(reader)
     }
 }
@@ -500,6 +524,7 @@ fn parse_snapshot(
     environment.retain(|name, _| !shell_environment::is_non_inheritable_env_var(name));
 
     Ok(ShellSnapshot {
+        prewarmed: false,
         state,
         file_source: false,
         environment,

codex-rs/exec-server/src/shell_snapshot_tests.rs

@@ -85,6 +85,7 @@ async fn snapshot_failure_retries_are_bounded_and_single_flight(
     )?;
     let telemetry = ExecServerTelemetry::new(metrics.clone());
 
+    let mut expected_commands = BTreeMap::new();
     for attempt in 1..=5 {
         if attempt == recovery_attempt {
             std::fs::write(
@@ -142,6 +143,13 @@ async fn snapshot_failure_retries_are_bounded_and_single_flight(
             (&concurrent.command, &concurrent.env)
         );
 
+        let outcome = if concurrent.command != params.argv {
+            "used"
+        } else {
+            "fallback"
+        };
+        *expected_commands.entry(outcome.to_string()).or_insert(0) +=
+            if prewarming { 1 } else { 2 };
         tokio::time::pause();
         if attempt < recovery_attempt || recovery_attempt > MAX_SNAPSHOT_ATTEMPTS {
             cache
@@ -153,6 +161,7 @@ async fn snapshot_failure_retries_are_bounded_and_single_flight(
                 )
                 .await
                 .expect("capture must stay cached during backoff");
+            *expected_commands.entry("fallback".to_string()).or_insert(0) += 1;
             assert_eq!(
                 (&prepared.command, &prepared.env),
                 (&params.argv, &params.env)
@@ -174,6 +183,8 @@ async fn snapshot_failure_retries_are_bounded_and_single_flight(
     let snapshot = metrics.snapshot()?;
     let mut counters = BTreeMap::new();
     let mut durations = BTreeMap::new();
+    let mut commands = BTreeMap::new();
+    let mut waits = BTreeMap::new();
     for metric in snapshot
         .scope_metrics()
         .flat_map(opentelemetry_sdk::metrics::data::ScopeMetrics::metrics)
@@ -203,9 +214,41 @@ async fn snapshot_failure_retries_are_bounded_and_single_flight(
                     durations.insert(tags, point.count());
                 }
             }
+            "codex.shell_snapshot.command" => {
+                let AggregatedMetrics::U64(MetricData::Sum(sum)) = metric.data() else {
+                    panic!("expected command counter");
+                };
+                for point in sum.data_points() {
+                    let outcome = point
+                        .attributes()
+                        .find(|tag| tag.key.as_str() == "outcome")
+                        .unwrap()
+                        .value
+                        .to_string();
+                    *commands.entry(outcome).or_insert(0) += point.value();
+                }
+            }
+            "codex.shell_snapshot.wait_ms" => {
+                let AggregatedMetrics::F64(MetricData::Histogram(histogram)) = metric.data() else {
+                    panic!("expected wait histogram");
+                };
+                for point in histogram.data_points() {
+                    let outcome = point
+                        .attributes()
+                        .find(|tag| tag.key.as_str() == "outcome")
+                        .unwrap()
+                        .value
+                        .to_string();
+                    *waits.entry(outcome).or_insert(0) += point.count();
+                }
+            }
             _ => {}
         }
     }
+    assert_eq!(
+        (commands, waits),
+        (expected_commands.clone(), expected_commands)
+    );
     let mut expected_counters = BTreeMap::new();
     let mut expected_durations = BTreeMap::new();
     let captures = prewarm_fails_first

codex-rs/exec-server/src/telemetry.rs

@@ -217,6 +217,28 @@ impl ExecServerTelemetry {
         let _ = metrics.counter("codex.shell_snapshot", /*inc*/ 1, &tags);
     }
 
+    /// One observation per eligible execution preparation, excluding prewarm.
+    /// `used` means selected for replay, not that restoration or execution succeeded.
+    #[cfg(unix)]
+    pub(crate) fn shell_snapshot_command(
+        &self,
+        wait: Duration,
+        state: &'static str,
+        outcome: &'static str,
+    ) {
+        let Some(metrics) = self
+            .inner
+            .as_ref()
+            .map(|inner| inner.metrics.clone())
+            .or_else(codex_otel::global)
+        else {
+            return;
+        };
+        let tags = [("version", "v2"), ("state", state), ("outcome", outcome)];
+        let _ = metrics.counter("codex.shell_snapshot.command", /*inc*/ 1, &tags);
+        let _ = metrics.record_duration("codex.shell_snapshot.wait_ms", wait, &tags);
+    }
+
     pub(crate) fn remote_registration_completed(&self, result: &'static str, duration: Duration) {
         self.record_operation(REMOTE_REGISTRATION_METRICS, result, duration);
     }