Add diagnostics for rollout compression metadata failures (#47026)

## What changed

- Tag metadata scan failures with bounded reasons for invalid history modes, responses before metadata, missing metadata, or I/O errors.
- Emit at most five metadata warnings per compression run with the rollout ID, error category, file size, and modification age, excluding paths and error messages. Include the total metadata failure count in the completion log.
- Preserve underlying metadata errors while redacting their details from `MetadataReadError` debug output.

GitOrigin-RevId: 13f8ec265405b8f8bb23a02ee37c1aabdce806eb

codex-rs/rollout/src/compression.rs

@@ -330,6 +330,7 @@ mod worker {
     const WORKER_MAX_RUNTIME: Duration = Duration::from_secs(5 * 60 * 60);
     const RUN_MARKER_FILE_NAME: &str = "rollout-compression.lock";
     const MAX_CONCURRENT_COMPRESSION_JOBS: usize = 2;
+    const MAX_METADATA_WARNINGS_PER_RUN: usize = 5;
 
     #[derive(Default)]
     struct CompressionStats {
@@ -338,6 +339,7 @@ mod worker {
         skipped: usize,
         failed: usize,
         scan_errors: bool,
+        metadata_read_failures: usize,
         cleanup_errors: bool,
         time_budget_exhausted: bool,
     }
@@ -487,8 +489,12 @@ mod worker {
             }
         };
         info!(
+            metadata_read_failures = stats.metadata_read_failures,
             "rollout compression worker finished: scanned={}, compressed={}, skipped={}, failed={}",
-            stats.scanned, stats.compressed, stats.skipped, stats.failed
+            stats.scanned,
+            stats.compressed,
+            stats.skipped,
+            stats.failed
         );
         // Keep the existing completed outcome: it means the pass returned, not
         // that every directory was scanned or every file was compressed.
@@ -614,17 +620,53 @@ mod worker {
                     continue;
                 }
                 let path = rollout_file.into_path();
-                if crate::rollout_id_from_path(path.as_path()).is_none() {
+                let Some(rollout_id) = crate::rollout_id_from_path(path.as_path()) else {
                     stats.scan_errors = true;
                     stats.skipped = stats.skipped.saturating_add(1);
                     metrics::file(trigger, "skipped_unreadable_meta");
                     continue;
-                }
+                };
                 let thread_id = match crate::read_session_meta_line(path.as_path()).await {
                     Ok(metadata) => metadata.meta.id,
                     Err(err) => {
                         stats.scan_errors = true;
-                        FailureMetric::Scan(trigger).record("read_metadata", &err);
+                        let reason = err
+                            .get_ref()
+                            .and_then(|error| {
+                                error.downcast_ref::<crate::list::MetadataReadError>()
+                            })
+                            .map_or("io", |error| error.reason);
+                        let error_kind = super::error_metrics::error_kind(&err);
+                        metrics::counter(
+                            "codex.rollout_compression.scan",
+                            &[
+                                ("outcome", "failed"),
+                                ("stage", "read_metadata"),
+                                ("error_kind", error_kind),
+                                ("trigger", trigger.tag()),
+                                ("reason", reason),
+                            ],
+                        );
+                        stats.metadata_read_failures =
+                            stats.metadata_read_failures.saturating_add(1);
+                        if stats.metadata_read_failures <= MAX_METADATA_WARNINGS_PER_RUN {
+                            let metadata = tokio::fs::metadata(&path).await.ok();
+                            let file_size_bytes = metadata.as_ref().map(std::fs::Metadata::len);
+                            let mtime_age_seconds = metadata
+                                .and_then(|metadata| metadata.modified().ok())
+                                .and_then(|modified| modified.elapsed().ok())
+                                .map(|age| age.as_secs());
+                            // Only bounded labels and parsed IDs, never paths or error messages.
+                            warn!(
+                                trigger = trigger.tag(),
+                                %rollout_id,
+                                reason,
+                                error_kind,
+                                file_size_bytes,
+                                mtime_age_seconds,
+                                "skipping rollout compression because session metadata could not be read"
+                            );
+                        }
                         stats.skipped = stats.skipped.saturating_add(1);
                         metrics::file(trigger, "skipped_unreadable_meta");
                         continue;

codex-rs/rollout/src/list.rs

@@ -3,6 +3,7 @@
 use codex_utils_path as path_utils;
 use std::cmp::Reverse;
 use std::ffi::OsStr;
+use std::fmt;
 use std::io;
 use std::num::NonZero;
 use std::ops::ControlFlow;
@@ -1300,6 +1301,33 @@ fn event_msg_preview(event: &EventMsg) -> Option<String> {
     }
 }
 
+/// A bounded diagnostic reason that preserves the original I/O error and redacts debug output.
+pub(crate) struct MetadataReadError {
+    pub(crate) reason: &'static str,
+    error: io::Error,
+}
+
+impl fmt::Debug for MetadataReadError {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter
+            .debug_struct("MetadataReadError")
+            .field("reason", &self.reason)
+            .finish_non_exhaustive()
+    }
+}
+
+impl fmt::Display for MetadataReadError {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        self.error.fmt(formatter)
+    }
+}
+
+impl std::error::Error for MetadataReadError {
+    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+        Some(&self.error)
+    }
+}
+
 /// Read the SessionMetaLine from the head of a rollout file for reuse by
 /// callers that need the session metadata (e.g. to derive a cwd for config).
 pub async fn read_session_meta_line(path: &Path) -> io::Result<SessionMetaLine> {
@@ -1311,17 +1339,28 @@ pub async fn read_session_meta_line(path: &Path) -> io::Result<SessionMetaLine>
         }
         let Ok(rollout_line) = crate::parse_rollout_line(trimmed) else {
             if let Ok(value) = serde_json::from_str::<serde_json::Value>(trimmed) {
-                crate::recorder::reject_unknown_thread_history_mode(&value)?;
+                crate::recorder::reject_unknown_thread_history_mode(&value).map_err(|error| {
+                    io::Error::new(
+                        error.kind(),
+                        MetadataReadError {
+                            reason: "invalid_history_mode",
+                            error,
+                        },
+                    )
+                })?;
             }
             continue;
         };
         match rollout_line.item {
             RolloutItem::SessionMeta(session_meta_line) => return Ok(session_meta_line),
             RolloutItem::ResponseItem(_) | RolloutItem::InterAgentCommunication(_) => {
-                return Err(io::Error::other(format!(
-                    "rollout at {} does not start with session metadata",
-                    path.display()
-                )));
+                return Err(io::Error::other(MetadataReadError {
+                    reason: "response_before_metadata",
+                    error: io::Error::other(format!(
+                        "rollout at {} does not start with session metadata",
+                        path.display()
+                    )),
+                }));
             }
             RolloutItem::InterAgentCommunicationMetadata { .. }
             | RolloutItem::Compacted(_)
@@ -1334,10 +1373,10 @@ pub async fn read_session_meta_line(path: &Path) -> io::Result<SessionMetaLine>
             | RolloutItem::EventMsg(_) => {}
         }
     }
-    Err(io::Error::other(format!(
-        "rollout at {} is empty",
-        path.display()
-    )))
+    Err(io::Error::other(MetadataReadError {
+        reason: "missing_metadata",
+        error: io::Error::other(format!("rollout at {} is empty", path.display())),
+    }))
 }
 
 async fn file_modified_time(path: &Path) -> io::Result<Option<OffsetDateTime>> {