From 24e9b849fad8f506971dfa0313dbdea8abd90112 Mon Sep 17 00:00:00 2001 From: rtaylor-oai Date: Fri, 17 Jul 2026 19:08:31 +0000 Subject: [PATCH] Limit rollout metadata reads to headers (#33892) ## Why Callers that only need session metadata should not scan an entire rollout or fail on an unreadable tail after valid metadata. ## What changed - Read session metadata directly from the rollout line reader, stopping as soon as it is found for both plain and compressed rollouts while preserving pre-header and validation behavior. - Bound the doctor thread inventory scan to the first 64 non-empty header lines. Prefer the metadata thread ID and retain the validated filename fallback for legacy rollouts without metadata. ## Testing - Cover invalid UTF-8 tails, compressed rollouts, pre-header records, metadata and filename ID mismatches, the scan boundary, legacy fallback, and unusable headers. GitOrigin-RevId: 49576bd2877942e64082ce5ed057ecc848c025ec --- codex-rs/cli/src/doctor/thread_inventory.rs | 338 +++++++++++++++++++- codex-rs/rollout/src/compression_tests.rs | 131 ++++++++ codex-rs/rollout/src/list.rs | 44 ++- 3 files changed, 486 insertions(+), 27 deletions(-) diff --git a/codex-rs/cli/src/doctor/thread_inventory.rs b/codex-rs/cli/src/doctor/thread_inventory.rs index 94f68a679e..ecef4e6fdc 100644 --- a/codex-rs/cli/src/doctor/thread_inventory.rs +++ b/codex-rs/cli/src/doctor/thread_inventory.rs @@ -5,9 +5,10 @@ use super::Config; use super::DoctorCheck; use super::DoctorIssue; use codex_protocol::protocol::InternalSessionSource; +use codex_protocol::protocol::RolloutItem; +use codex_protocol::protocol::RolloutLine; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SubAgentSource; -use codex_rollout::RolloutRecorder; use codex_state::ThreadStateAuditRow; use codex_utils_path::normalize_for_path_comparison; use std::collections::BTreeMap; @@ -18,6 +19,7 @@ use std::path::Path; use std::path::PathBuf; const MAX_PARITY_SCAN_FILES: usize = 10_000; +const MAX_ROLLOUT_HEADER_LINES: usize = 64; const SAMPLE_LIMIT: usize = 5; const SUMMARY_LIMIT: usize = 8; const CHECK_ID: &str = "state.rollout_db_parity"; @@ -45,6 +47,12 @@ enum RolloutThreadId { Unusable(String), } +#[derive(serde::Deserialize)] +struct RolloutLineType { + #[serde(rename = "type")] + item_type: String, +} + impl RolloutScan { fn candidate_count(&self) -> usize { self.files.len() + self.malformed_names.len() + self.scan_errors.len() @@ -503,14 +511,54 @@ async fn scan_rollout_root(root: &Path, archived: bool, scan: &mut RolloutScan) } async fn thread_id_from_rollout(path: &Path) -> RolloutThreadId { - let items = match RolloutRecorder::load_rollout_items(path).await { - Ok((items, _, _)) => items, + let mut lines = match codex_rollout::open_rollout_line_reader(path).await { + Ok(lines) => lines, Err(err) => return RolloutThreadId::Unusable(err.to_string()), }; - if items.is_empty() { - return RolloutThreadId::Unusable("no parseable rollout items".to_string()); + let mut has_legacy_item = false; + + for _ in 0..MAX_ROLLOUT_HEADER_LINES { + let line = match lines.next_line().await { + Ok(Some(line)) if line.trim().is_empty() => continue, + Ok(Some(line)) => line, + Ok(None) => break, + Err(err) => return RolloutThreadId::Unusable(err.to_string()), + }; + let item_type = match serde_json::from_str::(line.trim()) { + Ok(line) => line.item_type, + Err(_) => continue, + }; + if item_type == "session_meta" { + return match serde_json::from_str::(line.trim()) { + Ok(line) => match line.item { + RolloutItem::SessionMeta(session_meta) => { + RolloutThreadId::Id(session_meta.meta.id.to_string()) + } + _ => RolloutThreadId::Unusable(format!( + "rollout at {} has invalid session metadata", + path.display() + )), + }, + Err(_) => RolloutThreadId::Unusable(format!( + "rollout at {} has invalid session metadata", + path.display() + )), + }; + } + if !has_legacy_item { + has_legacy_item = serde_json::from_str::(line.trim()).is_ok(); + } } - codex_rollout::builder_from_items(items.as_slice(), path) + + if !has_legacy_item { + return RolloutThreadId::Unusable(format!( + "rollout at {} has no usable header record", + path.display() + )); + } + // Legacy rollouts can omit session metadata, so use the validated filename fallback after + // the bounded prefix without retaining the first item or loading the full history. + codex_rollout::builder_from_items(&[], path) .map(|builder| RolloutThreadId::Id(builder.id.to_string())) .unwrap_or(RolloutThreadId::MalformedName) } @@ -678,8 +726,6 @@ where mod tests { use super::*; use codex_protocol::ThreadId; - use codex_protocol::protocol::RolloutItem; - use codex_protocol::protocol::RolloutLine; use pretty_assertions::assert_eq; use sqlx::sqlite::SqliteConnectOptions; use sqlx::sqlite::SqlitePoolOptions; @@ -779,12 +825,276 @@ mod tests { .as_deref() .is_some_and(|remedy| remedy.starts_with("Restart Codex")) })); - assert!( - check - .details - .iter() - .any(|detail| detail.contains(missing_path.to_string_lossy().as_ref())) - ); + let missing_sample = check + .details + .iter() + .find_map(|detail| detail.strip_prefix("rollout DB missing active sample: ")) + .expect("missing active sample"); + assert_eq!(Path::new(missing_sample), missing_path.as_path()); + } + + #[tokio::test] + async fn thread_inventory_check_uses_metadata_id_when_filename_and_db_disagree() { + let fixture = Fixture::new().await; + let filename_id = "00000000-0000-0000-0000-000000000001"; + let metadata_id = ThreadId::from_string("00000000-0000-0000-0000-000000000002") + .expect("metadata thread id"); + let path = + fixture.write_rollout(/*archived*/ false, "2025-01-02T10-00-00", filename_id); + let contents = std::fs::read_to_string(&path).expect("rollout file"); + let mut rollout_line = + serde_json::from_str::(contents.trim()).expect("rollout line"); + let RolloutItem::SessionMeta(session_meta) = &mut rollout_line.item else { + panic!("expected session metadata"); + }; + session_meta.meta.session_id = metadata_id.into(); + session_meta.meta.id = metadata_id; + session_meta.meta.timestamp = "not-a-timestamp".to_string(); + let contents = serde_json::to_string(&rollout_line).expect("rollout line"); + std::fs::write(&path, format!("{contents}\n")).expect("rollout file"); + fixture + .insert_thread_row(filename_id, path.as_path(), /*archived*/ false) + .await; + + let check = thread_inventory_check_for_roots( + fixture.codex_home.path(), + fixture.sqlite_home.path(), + "test-provider", + ) + .await; + + assert_eq!(check.status, CheckStatus::Warning); + assert_detail(&check, "rollout DB scan errors", "0"); + assert_detail(&check, "rollout DB missing active rows", "1"); + assert_detail(&check, "rollout DB stale rows", "0"); + } + + #[tokio::test] + async fn thread_inventory_check_ignores_invalid_utf8_after_session_metadata() { + let fixture = Fixture::new().await; + let thread_id = "00000000-0000-0000-0000-000000000001"; + let path = fixture.write_rollout(/*archived*/ false, "2025-01-02T10-00-00", thread_id); + let mut contents = std::fs::read(&path).expect("rollout file"); + contents.resize(contents.len() + 128 * 1024, 0xff); + contents.push(b'\n'); + std::fs::write(&path, contents).expect("rollout file"); + fixture + .insert_thread_row(thread_id, path.as_path(), /*archived*/ false) + .await; + + let check = thread_inventory_check_for_roots( + fixture.codex_home.path(), + fixture.sqlite_home.path(), + "test-provider", + ) + .await; + + assert_eq!(check.status, CheckStatus::Ok); + assert_detail(&check, "rollout DB scan errors", "0"); + assert_detail(&check, "rollout DB missing active rows", "0"); + } + + #[tokio::test] + async fn thread_id_from_rollout_falls_back_for_legacy_non_meta_record() { + let home = TempDir::new().expect("temp dir"); + let thread_id = "00000000-0000-0000-0000-000000000001"; + let valid_path = home + .path() + .join(format!("rollout-2025-01-02T10-00-00-{thread_id}.jsonl")); + let malformed_path = home.path().join("rollout-not-a-valid-name.jsonl"); + let legacy_line = serde_json::json!({ + "timestamp": "2025-01-02T10:00:00Z", + "type": "compacted", + "payload": {"message": "legacy history"}, + }); + std::fs::write(&valid_path, format!("not-json\n{legacy_line}\n")).expect("legacy rollout"); + std::fs::write(&malformed_path, format!("{legacy_line}\n")).expect("legacy rollout"); + + assert!(matches!( + thread_id_from_rollout(&valid_path).await, + RolloutThreadId::Id(id) if id == thread_id + )); + assert!(matches!( + thread_id_from_rollout(&malformed_path).await, + RolloutThreadId::MalformedName + )); + } + + #[tokio::test] + async fn thread_id_from_rollout_uses_metadata_after_pre_header_record() { + let fixture = Fixture::new().await; + let filename_id = "00000000-0000-0000-0000-000000000001"; + let metadata_id = ThreadId::from_string("00000000-0000-0000-0000-000000000002") + .expect("metadata thread id"); + let path = + fixture.write_rollout(/*archived*/ false, "2025-01-02T10-00-00", filename_id); + let contents = std::fs::read_to_string(&path).expect("rollout file"); + let mut rollout_line = + serde_json::from_str::(contents.trim()).expect("rollout line"); + let RolloutItem::SessionMeta(session_meta) = &mut rollout_line.item else { + panic!("expected session metadata"); + }; + session_meta.meta.session_id = metadata_id.into(); + session_meta.meta.id = metadata_id; + let response_item = serde_json::json!({ + "timestamp": "2025-01-02T10:00:00Z", + "type": "response_item", + "payload": { + "type": "message", + "role": "assistant", + "content": [{"type": "output_text", "text": "before metadata"}], + }, + }); + let session_meta = serde_json::to_string(&rollout_line).expect("rollout line"); + std::fs::write(&path, format!("{response_item}\n{session_meta}\n")).expect("rollout file"); + + assert!(matches!( + thread_id_from_rollout(&path).await, + RolloutThreadId::Id(id) if id == metadata_id.to_string() + )); + } + + #[tokio::test] + async fn thread_id_from_rollout_uses_metadata_at_header_line_limit() { + let fixture = Fixture::new().await; + let filename_id = "00000000-0000-0000-0000-000000000001"; + let metadata_id = ThreadId::from_string("00000000-0000-0000-0000-000000000002") + .expect("metadata thread id"); + let path = + fixture.write_rollout(/*archived*/ false, "2025-01-02T10-00-00", filename_id); + let contents = std::fs::read_to_string(&path).expect("rollout file"); + let mut rollout_line = + serde_json::from_str::(contents.trim()).expect("rollout line"); + let RolloutItem::SessionMeta(session_meta) = &mut rollout_line.item else { + panic!("expected session metadata"); + }; + session_meta.meta.session_id = metadata_id.into(); + session_meta.meta.id = metadata_id; + let legacy_line = serde_json::json!({ + "timestamp": "2025-01-02T10:00:00Z", + "type": "compacted", + "payload": { + "message": "legacy history", + "replacement_history": [{ + "type": "message", + "role": "assistant", + "content": [{ + "type": "output_text", + "text": "x".repeat(/*n*/ 1024 * 1024), + }], + }], + }, + }); + let mut lines = vec![legacy_line.to_string(); MAX_ROLLOUT_HEADER_LINES - 1]; + lines.push(serde_json::to_string(&rollout_line).expect("rollout line")); + std::fs::write(&path, format!("{}\n", lines.join("\n"))).expect("rollout file"); + + assert!(matches!( + thread_id_from_rollout(&path).await, + RolloutThreadId::Id(id) if id == metadata_id.to_string() + )); + } + + #[tokio::test] + async fn thread_id_from_rollout_stops_after_legacy_header_line_limit() { + let home = TempDir::new().expect("temp dir"); + let thread_id = "00000000-0000-0000-0000-000000000001"; + let path = home + .path() + .join(format!("rollout-2025-01-02T10-00-00-{thread_id}.jsonl")); + let legacy_line = serde_json::json!({ + "timestamp": "2025-01-02T10:00:00Z", + "type": "compacted", + "payload": {"message": "legacy history"}, + }); + let mut contents = format!( + "{}\n", + vec![legacy_line.to_string(); MAX_ROLLOUT_HEADER_LINES].join("\n") + ) + .into_bytes(); + contents.extend([0xff, 0xfe, b'\n']); + std::fs::write(&path, contents).expect("legacy rollout"); + + assert!(matches!( + thread_id_from_rollout(&path).await, + RolloutThreadId::Id(id) if id == thread_id + )); + } + + #[tokio::test] + async fn thread_id_from_rollout_rejects_unusable_headers() { + let home = TempDir::new().expect("temp dir"); + let empty_path = home + .path() + .join("rollout-2025-01-02T10-00-00-00000000-0000-0000-0000-000000000001.jsonl"); + let invalid_path = home + .path() + .join("rollout-2025-01-02T10-00-01-00000000-0000-0000-0000-000000000002.jsonl"); + let invalid_utf8_path = home + .path() + .join("rollout-2025-01-02T10-00-02-00000000-0000-0000-0000-000000000003.jsonl"); + let unknown_history_mode_path = home + .path() + .join("rollout-2025-01-02T10-00-03-00000000-0000-0000-0000-000000000004.jsonl"); + let missing_path = home + .path() + .join("rollout-2025-01-02T10-00-04-00000000-0000-0000-0000-000000000005.jsonl"); + std::fs::write(&empty_path, " \n\t\n").expect("empty rollout"); + std::fs::write(&invalid_path, "not-json\n{also-invalid}\n").expect("invalid rollout"); + std::fs::write(&invalid_utf8_path, [0xff, 0xfe, b'\n']).expect("invalid utf8 rollout"); + let unknown_history_mode = serde_json::json!({ + "timestamp": "2025-01-02T10:00:03Z", + "type": "session_meta", + "payload": { + "session_id": "00000000-0000-0000-0000-000000000004", + "id": "00000000-0000-0000-0000-000000000004", + "timestamp": "2025-01-02T10:00:03Z", + "cwd": ".", + "originator": "test", + "cli_version": "test", + "source": "cli", + "model_provider": "test-provider", + "history_mode": "future", + }, + }); + std::fs::write( + &unknown_history_mode_path, + format!( + "{unknown_history_mode}\n{}\n", + serde_json::json!({ + "timestamp": "2025-01-02T10:00:04Z", + "type": "session_meta", + "payload": { + "session_id": "00000000-0000-0000-0000-000000000005", + "id": "00000000-0000-0000-0000-000000000005", + "timestamp": "2025-01-02T10:00:04Z", + "cwd": ".", + "originator": "test", + "cli_version": "test", + "source": "cli", + "model_provider": "test-provider", + }, + }) + ), + ) + .expect("unknown history mode rollout"); + + for path in [ + &empty_path, + &invalid_path, + &invalid_utf8_path, + &unknown_history_mode_path, + &missing_path, + ] { + assert!( + matches!( + thread_id_from_rollout(path).await, + RolloutThreadId::Unusable(_) + ), + "{} should be unusable", + path.display() + ); + } } struct Fixture { diff --git a/codex-rs/rollout/src/compression_tests.rs b/codex-rs/rollout/src/compression_tests.rs index f5ced62c10..6b2f8d328f 100644 --- a/codex-rs/rollout/src/compression_tests.rs +++ b/codex-rs/rollout/src/compression_tests.rs @@ -23,6 +23,7 @@ use crate::RolloutConfig; use crate::RolloutRecorder; use crate::RolloutRecorderParams; use crate::append_rollout_item_to_path; +use crate::read_session_meta_line; use crate::search_rollout_matches; #[tokio::test] @@ -45,6 +46,136 @@ async fn load_rollout_items_reads_compressed_rollout() -> anyhow::Result<()> { Ok(()) } +#[tokio::test] +async fn read_session_meta_line_stops_before_invalid_utf8_tail() -> anyhow::Result<()> { + let home = TempDir::new()?; + let uuid = Uuid::from_u128(16); + let thread_id = ThreadId::from_string(&uuid.to_string())?; + let rollout_path = rollout_path(home.path(), "2025-01-03T12-00-00", uuid); + write_rollout(&rollout_path, thread_id, "ignored tail")?; + let mut contents = fs::read(&rollout_path)?; + contents.resize(contents.len() + 64 * 1024, 0xff); + fs::write(&rollout_path, contents)?; + + let session_meta = read_session_meta_line(&rollout_path).await?; + + assert_eq!(session_meta.meta.id, thread_id); + Ok(()) +} + +#[tokio::test] +async fn read_session_meta_line_stops_before_invalid_utf8_compressed_tail() -> anyhow::Result<()> { + let home = TempDir::new()?; + let uuid = Uuid::from_u128(17); + let thread_id = ThreadId::from_string(&uuid.to_string())?; + let rollout_path = rollout_path(home.path(), "2025-01-03T12-00-00", uuid); + write_rollout(&rollout_path, thread_id, "ignored compressed tail")?; + let mut contents = fs::read(&rollout_path)?; + contents.resize(contents.len() + 64 * 1024, 0xff); + fs::write(&rollout_path, contents)?; + compress_now(&rollout_path)?; + + let session_meta = read_session_meta_line(&rollout_path).await?; + + assert_eq!(session_meta.meta.id, thread_id); + assert!(!rollout_path.exists()); + assert!(compressed_rollout_path(&rollout_path).exists()); + Ok(()) +} + +#[tokio::test] +async fn read_session_meta_line_preserves_pre_header_and_error_semantics() -> anyhow::Result<()> { + let home = TempDir::new()?; + let uuid = Uuid::from_u128(18); + let thread_id = ThreadId::from_string(&uuid.to_string())?; + let rollout_path = rollout_path(home.path(), "2025-01-03T12-00-00", uuid); + write_rollout(&rollout_path, thread_id, "after metadata")?; + let contents = fs::read_to_string(&rollout_path)?; + let (session_meta, tail) = contents.split_once('\n').expect("session metadata line"); + let mut session_meta = serde_json::from_str::(session_meta)?; + session_meta["payload"] + .as_object_mut() + .expect("session metadata payload") + .remove("session_id"); + let contents = format!("{session_meta}\n{tail}"); + let ignored_event = serde_json::json!({ + "timestamp": "2025-01-03T12:00:00Z", + "type": "event_msg", + "payload": {"type": "user_message", "message": "before metadata", "kind": "plain"} + }); + fs::write( + &rollout_path, + format!("\nnot json\n{ignored_event}\n{contents}"), + )?; + + let session_meta = read_session_meta_line(&rollout_path).await?; + + assert_eq!(session_meta.meta.id, thread_id); + + let response_item = serde_json::json!({ + "timestamp": "2025-01-03T12:00:00Z", + "type": "response_item", + "payload": { + "type": "message", + "role": "assistant", + "content": [{"type": "output_text", "text": "before metadata"}] + } + }); + fs::write(&rollout_path, format!("{response_item}\n{contents}"))?; + + let error = read_session_meta_line(&rollout_path) + .await + .expect_err("response item before metadata should fail"); + + assert_eq!( + error.to_string(), + format!( + "rollout at {} does not start with session metadata", + rollout_path.display() + ) + ); + + let unknown_history_mode = serde_json::json!({ + "timestamp": "2025-01-03T12:00:00Z", + "type": "session_meta", + "payload": { + "session_id": uuid.to_string(), + "id": uuid.to_string(), + "timestamp": "2025-01-03T12:00:00Z", + "cwd": ".", + "originator": "test", + "cli_version": "test", + "source": "cli", + "model_provider": "test-provider", + "history_mode": "future", + }, + }); + fs::write(&rollout_path, format!("{unknown_history_mode}\n{contents}"))?; + + let error = read_session_meta_line(&rollout_path) + .await + .expect_err("unknown history mode before valid metadata should fail"); + + assert!( + error + .to_string() + .starts_with("invalid session metadata history_mode:"), + "{error}" + ); + + fs::write(&rollout_path, "\nnot json\n")?; + + let error = read_session_meta_line(&rollout_path) + .await + .expect_err("rollout without selected records should fail"); + + assert_eq!( + error.to_string(), + format!("rollout at {} is empty", rollout_path.display()) + ); + Ok(()) +} + #[test] fn rollout_file_from_path_normalizes_compressed_file_names() -> anyhow::Result<()> { let home = TempDir::new()?; diff --git a/codex-rs/rollout/src/list.rs b/codex-rs/rollout/src/list.rs index 4729695dcb..137ee6642e 100644 --- a/codex-rs/rollout/src/list.rs +++ b/codex-rs/rollout/src/list.rs @@ -1278,19 +1278,37 @@ fn event_msg_preview(event: &EventMsg) -> Option { /// 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 { - let head = read_head_for_summary(path).await?; - let Some(first) = head.first() else { - return Err(io::Error::other(format!( - "rollout at {} is empty", - path.display() - ))); - }; - serde_json::from_value::(first.clone()).map_err(|_| { - io::Error::other(format!( - "rollout at {} does not start with session metadata", - path.display() - )) - }) + let mut lines = compression::open_rollout_line_reader(path).await?; + while let Some(line) = lines.next_line().await? { + let trimmed = line.trim(); + if trimmed.is_empty() { + continue; + } + let Ok(rollout_line) = serde_json::from_str::(trimmed) else { + if let Ok(value) = serde_json::from_str::(trimmed) { + crate::recorder::reject_unknown_thread_history_mode(&value)?; + } + 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() + ))); + } + RolloutItem::InterAgentCommunicationMetadata { .. } + | RolloutItem::Compacted(_) + | RolloutItem::TurnContext(_) + | RolloutItem::WorldState(_) + | RolloutItem::EventMsg(_) => {} + } + } + Err(io::Error::other(format!( + "rollout at {} is empty", + path.display() + ))) } async fn file_modified_time(path: &Path) -> io::Result> {