From 3b1fd5bb1c08a1f0ee90221732c1b90307cf2de7 Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Thu, 9 Jul 2026 11:56:21 -0700 Subject: [PATCH 1/2] rollout: add ordinals to paginated records --- .../thread_processor_tests.rs | 3 + codex-rs/cli/src/doctor/thread_inventory.rs | 1 + .../core/src/personality_migration_tests.rs | 2 + codex-rs/core/tests/suite/client.rs | 11 + .../core/tests/suite/personality_migration.rs | 3 + codex-rs/core/tests/suite/sqlite_state.rs | 2 + codex-rs/memories/write/src/startup_tests.rs | 1 + codex-rs/protocol/src/protocol.rs | 2 + codex-rs/rollout/src/compression_tests.rs | 2 + codex-rs/rollout/src/lib.rs | 1 + codex-rs/rollout/src/metadata_tests.rs | 5 + codex-rs/rollout/src/ordinal.rs | 110 +++++++++ codex-rs/rollout/src/recorder.rs | 118 ++++++++-- codex-rs/rollout/src/recorder_tests.rs | 211 ++++++++++++++++++ codex-rs/rollout/src/session_index_tests.rs | 1 + codex-rs/rollout/src/state_db_tests.rs | 2 + codex-rs/rollout/src/tests.rs | 3 + 17 files changed, 454 insertions(+), 24 deletions(-) create mode 100644 codex-rs/rollout/src/ordinal.rs diff --git a/codex-rs/app-server/src/request_processors/thread_processor_tests.rs b/codex-rs/app-server/src/request_processors/thread_processor_tests.rs index 8c0d59efed63..38a94ab8121a 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor_tests.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor_tests.rs @@ -1015,6 +1015,7 @@ mod thread_processor_behavior_tests { let line = RolloutLine { timestamp: timestamp.clone(), + ordinal: None, item: RolloutItem::SessionMeta(SessionMetaLine { meta: session_meta.clone(), git: None, @@ -1082,6 +1083,7 @@ mod thread_processor_behavior_tests { let line = RolloutLine { timestamp, + ordinal: None, item: RolloutItem::SessionMeta(SessionMetaLine { meta: session_meta, git: None, @@ -1124,6 +1126,7 @@ mod thread_processor_behavior_tests { let line = RolloutLine { timestamp, + ordinal: None, item: RolloutItem::SessionMeta(SessionMetaLine { meta: session_meta, git: None, diff --git a/codex-rs/cli/src/doctor/thread_inventory.rs b/codex-rs/cli/src/doctor/thread_inventory.rs index c50e6debdc08..94f68a679e92 100644 --- a/codex-rs/cli/src/doctor/thread_inventory.rs +++ b/codex-rs/cli/src/doctor/thread_inventory.rs @@ -819,6 +819,7 @@ mod tests { let parsed_thread_id = ThreadId::from_string(thread_id).expect("thread id"); let rollout_line = RolloutLine { timestamp: timestamp.to_string(), + ordinal: None, item: RolloutItem::SessionMeta(codex_protocol::protocol::SessionMetaLine { meta: codex_protocol::protocol::SessionMeta { session_id: parsed_thread_id.into(), diff --git a/codex-rs/core/src/personality_migration_tests.rs b/codex-rs/core/src/personality_migration_tests.rs index 80f5b3c02054..12369027415a 100644 --- a/codex-rs/core/src/personality_migration_tests.rs +++ b/codex-rs/core/src/personality_migration_tests.rs @@ -69,10 +69,12 @@ async fn write_rollout_with_user_event(dir: &Path, thread_id: ThreadId) -> io::R }; let meta_line = RolloutLine { timestamp: TEST_TIMESTAMP.to_string(), + ordinal: None, item: RolloutItem::SessionMeta(session_meta), }; let user_event = RolloutLine { timestamp: TEST_TIMESTAMP.to_string(), + ordinal: None, item: RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent { client_id: None, message: "hello".to_string(), diff --git a/codex-rs/core/tests/suite/client.rs b/codex-rs/core/tests/suite/client.rs index 4d298998afb0..284361e0369d 100644 --- a/codex-rs/core/tests/suite/client.rs +++ b/codex-rs/core/tests/suite/client.rs @@ -370,6 +370,7 @@ async fn synthetic_call_output_id_is_stable_across_resumes() -> anyhow::Result<( let rollout = vec![ RolloutLine { timestamp: "2024-01-01T00:00:00.000Z".to_string(), + ordinal: None, item: RolloutItem::SessionMeta(SessionMetaLine { meta: SessionMeta { session_id: thread_id.into(), @@ -387,6 +388,7 @@ async fn synthetic_call_output_id_is_stable_across_resumes() -> anyhow::Result<( }, RolloutLine { timestamp: "2024-01-01T00:00:01.000Z".to_string(), + ordinal: None, item: RolloutItem::ResponseItem(ResponseItem::FunctionCall { id: Some("fc_existing".to_string()), name: "do_it".to_string(), @@ -895,6 +897,7 @@ async fn resume_replays_legacy_js_repl_image_rollout_shapes() { let rollout = vec![ RolloutLine { timestamp: "2024-01-01T00:00:00.000Z".to_string(), + ordinal: None, item: RolloutItem::SessionMeta(SessionMetaLine { meta: SessionMeta { session_id: thread_id.into(), @@ -912,10 +915,12 @@ async fn resume_replays_legacy_js_repl_image_rollout_shapes() { }, RolloutLine { timestamp: "2024-01-01T00:00:01.000Z".to_string(), + ordinal: None, item: RolloutItem::ResponseItem(legacy_custom_tool_call), }, RolloutLine { timestamp: "2024-01-01T00:00:02.000Z".to_string(), + ordinal: None, item: RolloutItem::ResponseItem(ResponseItem::CustomToolCallOutput { id: None, call_id: "legacy-js-call".to_string(), @@ -926,6 +931,7 @@ async fn resume_replays_legacy_js_repl_image_rollout_shapes() { }, RolloutLine { timestamp: "2024-01-01T00:00:03.000Z".to_string(), + ordinal: None, item: RolloutItem::ResponseItem(ResponseItem::Message { id: None, role: "user".to_string(), @@ -1031,6 +1037,7 @@ async fn resume_replays_image_tool_outputs_with_detail() { let rollout = vec![ RolloutLine { timestamp: "2024-01-01T00:00:00.000Z".to_string(), + ordinal: None, item: RolloutItem::SessionMeta(SessionMetaLine { meta: SessionMeta { session_id: thread_id.into(), @@ -1048,6 +1055,7 @@ async fn resume_replays_image_tool_outputs_with_detail() { }, RolloutLine { timestamp: "2024-01-01T00:00:01.000Z".to_string(), + ordinal: None, item: RolloutItem::ResponseItem(ResponseItem::FunctionCall { id: None, name: "view_image".to_string(), @@ -1059,6 +1067,7 @@ async fn resume_replays_image_tool_outputs_with_detail() { }, RolloutLine { timestamp: "2024-01-01T00:00:01.500Z".to_string(), + ordinal: None, item: RolloutItem::ResponseItem(ResponseItem::FunctionCallOutput { id: None, call_id: function_call_id.to_string(), @@ -1073,6 +1082,7 @@ async fn resume_replays_image_tool_outputs_with_detail() { }, RolloutLine { timestamp: "2024-01-01T00:00:02.000Z".to_string(), + ordinal: None, item: RolloutItem::ResponseItem(ResponseItem::CustomToolCall { id: None, status: Some("completed".to_string()), @@ -1085,6 +1095,7 @@ async fn resume_replays_image_tool_outputs_with_detail() { }, RolloutLine { timestamp: "2024-01-01T00:00:02.500Z".to_string(), + ordinal: None, item: RolloutItem::ResponseItem(ResponseItem::CustomToolCallOutput { id: None, call_id: custom_call_id.to_string(), diff --git a/codex-rs/core/tests/suite/personality_migration.rs b/codex-rs/core/tests/suite/personality_migration.rs index 0ca3e1b2a9c5..e89c011a4080 100644 --- a/codex-rs/core/tests/suite/personality_migration.rs +++ b/codex-rs/core/tests/suite/personality_migration.rs @@ -85,10 +85,12 @@ async fn write_rollout_with_user_event(dir: &Path, thread_id: ThreadId) -> io::R }; let meta_line = RolloutLine { timestamp: TEST_TIMESTAMP.to_string(), + ordinal: None, item: RolloutItem::SessionMeta(session_meta), }; let user_event = RolloutLine { timestamp: TEST_TIMESTAMP.to_string(), + ordinal: None, item: RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent { client_id: None, message: "hello".to_string(), @@ -139,6 +141,7 @@ async fn write_rollout_with_meta_only(dir: &Path, thread_id: ThreadId) -> io::Re }; let meta_line = RolloutLine { timestamp: TEST_TIMESTAMP.to_string(), + ordinal: None, item: RolloutItem::SessionMeta(session_meta), }; diff --git a/codex-rs/core/tests/suite/sqlite_state.rs b/codex-rs/core/tests/suite/sqlite_state.rs index 726d74ebfdad..8759d83fa3c3 100644 --- a/codex-rs/core/tests/suite/sqlite_state.rs +++ b/codex-rs/core/tests/suite/sqlite_state.rs @@ -384,10 +384,12 @@ async fn backfill_scans_existing_rollouts() -> Result<()> { let lines = [ RolloutLine { timestamp: "2026-01-27T12:00:00Z".to_string(), + ordinal: None, item: RolloutItem::SessionMeta(session_meta_line), }, RolloutLine { timestamp: "2026-01-27T12:00:01Z".to_string(), + ordinal: None, item: RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent { client_id: None, message: "hello from backfill".to_string(), diff --git a/codex-rs/memories/write/src/startup_tests.rs b/codex-rs/memories/write/src/startup_tests.rs index fad1ce14a425..35862eed349d 100644 --- a/codex-rs/memories/write/src/startup_tests.rs +++ b/codex-rs/memories/write/src/startup_tests.rs @@ -678,6 +678,7 @@ async fn seed_stage1_candidate( let rollout_path = codex_home.join(format!("rollout-{thread_id}.jsonl")); let line = RolloutLine { timestamp: updated_at.to_rfc3339(), + ordinal: None, item: RolloutItem::ResponseItem(ResponseItem::Message { id: None, role: "user".to_string(), diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index 57a36c3312c9..bd339f273381 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -3325,6 +3325,8 @@ impl Mul for TruncationPolicy { #[derive(Serialize, Deserialize, Clone, JsonSchema)] pub struct RolloutLine { pub timestamp: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub ordinal: Option, #[serde(flatten)] pub item: RolloutItem, } diff --git a/codex-rs/rollout/src/compression_tests.rs b/codex-rs/rollout/src/compression_tests.rs index 46c90d0120d3..3b8132534072 100644 --- a/codex-rs/rollout/src/compression_tests.rs +++ b/codex-rs/rollout/src/compression_tests.rs @@ -483,10 +483,12 @@ fn write_rollout(path: &std::path::Path, thread_id: ThreadId, message: &str) -> let lines = [ RolloutLine { timestamp: "2025-01-03T12:00:00Z".to_string(), + ordinal: None, item: RolloutItem::SessionMeta(session_meta_line), }, RolloutLine { timestamp: "2025-01-03T12:00:01Z".to_string(), + ordinal: None, item: RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent { message: message.to_string(), ..Default::default() diff --git a/codex-rs/rollout/src/lib.rs b/codex-rs/rollout/src/lib.rs index 43e7a9216f05..7cf140dc2082 100644 --- a/codex-rs/rollout/src/lib.rs +++ b/codex-rs/rollout/src/lib.rs @@ -8,6 +8,7 @@ pub(crate) mod compression; pub(crate) mod config; pub(crate) mod list; pub(crate) mod metadata; +mod ordinal; mod persistence_metrics; pub(crate) mod policy; pub(crate) mod recorder; diff --git a/codex-rs/rollout/src/metadata_tests.rs b/codex-rs/rollout/src/metadata_tests.rs index 2e1ea98f3aca..c3163b6ef1d6 100644 --- a/codex-rs/rollout/src/metadata_tests.rs +++ b/codex-rs/rollout/src/metadata_tests.rs @@ -62,6 +62,7 @@ async fn extract_metadata_from_rollout_uses_session_meta() { }; let rollout_line = RolloutLine { timestamp: "2026-01-27T12:34:56Z".to_string(), + ordinal: Some(0), item: RolloutItem::SessionMeta(session_meta_line.clone()), }; let json = serde_json::to_string(&rollout_line).expect("rollout json"); @@ -93,6 +94,7 @@ async fn extract_metadata_from_rollout_rejects_unknown_history_mode() { .join(format!("rollout-2026-01-27T12-34-56-{uuid}.jsonl")); let mut rollout_line = serde_json::to_value(RolloutLine { timestamp: "2026-01-27T12:34:56Z".to_string(), + ordinal: None, item: RolloutItem::SessionMeta(SessionMetaLine { meta: SessionMeta { session_id: id.into(), @@ -158,6 +160,7 @@ async fn extract_metadata_from_rollout_returns_latest_memory_mode() { let lines = vec![ RolloutLine { timestamp: "2026-01-27T12:34:56Z".to_string(), + ordinal: None, item: RolloutItem::SessionMeta(SessionMetaLine { meta: session_meta, git: None, @@ -165,6 +168,7 @@ async fn extract_metadata_from_rollout_returns_latest_memory_mode() { }, RolloutLine { timestamp: "2026-01-27T12:35:00Z".to_string(), + ordinal: None, item: RolloutItem::SessionMeta(SessionMetaLine { meta: polluted_meta, git: None, @@ -427,6 +431,7 @@ fn write_rollout_in_sessions_with_cwd( }; let rollout_line = RolloutLine { timestamp: event_ts.to_string(), + ordinal: None, item: RolloutItem::SessionMeta(session_meta_line), }; let json = serde_json::to_string(&rollout_line).expect("serialize rollout"); diff --git a/codex-rs/rollout/src/ordinal.rs b/codex-rs/rollout/src/ordinal.rs new file mode 100644 index 000000000000..c87b32c02bb8 --- /dev/null +++ b/codex-rs/rollout/src/ordinal.rs @@ -0,0 +1,110 @@ +use std::fs::File; +use std::io; +use std::io::BufRead; +use std::io::BufReader; +use std::path::Path; + +use codex_protocol::protocol::RolloutItem; +use codex_protocol::protocol::RolloutLine; +use codex_protocol::protocol::ThreadHistoryMode; + +use crate::reverse_jsonl_scanner::ReverseJsonlScanner; +use crate::reverse_jsonl_scanner::ScanOutcome; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum RolloutOrdinalState { + Legacy, + Paginated { next: Option }, +} + +impl RolloutOrdinalState { + pub(crate) fn for_new_rollout(history_mode: ThreadHistoryMode) -> Self { + match history_mode { + ThreadHistoryMode::Legacy => Self::Legacy, + ThreadHistoryMode::Paginated => Self::Paginated { next: Some(0) }, + } + } + + pub(crate) fn current(&self) -> io::Result> { + match self { + Self::Legacy => Ok(None), + Self::Paginated { next } => { + let ordinal = (*next) + .ok_or_else(|| io::Error::other("paginated rollout record ordinal overflow"))?; + Ok(Some(ordinal)) + } + } + } + + pub(crate) fn advance(&mut self) { + if let Self::Paginated { next } = self + && let Some(ordinal) = *next + { + *next = ordinal.checked_add(1); + } + } +} + +pub(crate) async fn ordinal_state_for_rollout(path: &Path) -> io::Result { + let path = path.to_path_buf(); + tokio::task::spawn_blocking(move || ordinal_state_for_rollout_blocking(path.as_path())) + .await + .map_err(io::Error::other)? +} + +fn ordinal_state_for_rollout_blocking(path: &Path) -> io::Result { + let history_mode = read_history_mode(path)?; + if matches!(history_mode, ThreadHistoryMode::Legacy) { + return Ok(RolloutOrdinalState::Legacy); + } + + let mut scanner = ReverseJsonlScanner::new(File::open(path)?)?; + let record = loop { + match scanner.scan_next::()? { + Some(ScanOutcome::Parsed(record)) => break record, + Some(ScanOutcome::Rejected(_)) => continue, + None => { + return Err(io::Error::other(format!( + "rollout at {} contains no valid records", + path.display() + ))); + } + } + }; + let ordinal = record.ordinal.ok_or_else(|| { + io::Error::other(format!( + "final paginated rollout record at {} is missing an ordinal", + path.display() + )) + })?; + Ok(RolloutOrdinalState::Paginated { + next: ordinal.checked_add(1), + }) +} + +fn read_history_mode(path: &Path) -> io::Result { + let reader = BufReader::new(File::open(path)?); + for line in reader.lines() { + let line = line?; + if line.trim().is_empty() { + continue; + } + let record: RolloutLine = serde_json::from_str(line.as_str()).map_err(|error| { + io::Error::other(format!( + "failed to parse first rollout record at {}: {error}", + path.display() + )) + })?; + let RolloutItem::SessionMeta(session_meta) = record.item else { + return Err(io::Error::other(format!( + "rollout at {} does not start with session metadata", + path.display() + ))); + }; + return Ok(session_meta.meta.history_mode); + } + Err(io::Error::other(format!( + "rollout at {} contains no records", + path.display() + ))) +} diff --git a/codex-rs/rollout/src/recorder.rs b/codex-rs/rollout/src/recorder.rs index 87e97997a245..a482973bd1d9 100644 --- a/codex-rs/rollout/src/recorder.rs +++ b/codex-rs/rollout/src/recorder.rs @@ -4,6 +4,10 @@ use std::collections::HashSet; use std::fs; use std::fs::File; use std::io::Error as IoError; +use std::io::Read; +use std::io::Seek; +use std::io::SeekFrom; +use std::io::Write; use std::path::Path; use std::path::PathBuf; use std::sync::Arc; @@ -45,6 +49,8 @@ use super::list::get_threads_in_root; use super::list::parse_cursor; use super::list::parse_timestamp_uuid_from_filename; use super::metadata; +use super::ordinal::RolloutOrdinalState; +use super::ordinal::ordinal_state_for_rollout; use super::session_index::find_thread_names_by_ids; use crate::config::RolloutConfigView; use crate::state_db; @@ -751,7 +757,7 @@ impl RolloutRecorder { config: &impl RolloutConfigView, params: RolloutRecorderParams, ) -> std::io::Result { - let (file, deferred_log_file_info, rollout_path, meta) = match params { + let (file, deferred_log_file_info, rollout_path, meta, ordinal_state) = match params { RolloutRecorderParams::Create { session_id, conversation_id, @@ -767,6 +773,7 @@ impl RolloutRecorder { history_mode, initial_window_id, } => { + let ordinal_state = RolloutOrdinalState::for_new_rollout(history_mode); let log_file_info = precompute_log_file_info(config, conversation_id)?; let path = log_file_info.path.clone(); let thread_id = log_file_info.conversation_id; @@ -808,25 +815,22 @@ impl RolloutRecorder { context_window: initial_window_id.map(SessionContextWindow::new), }; - (None, Some(log_file_info), path, Some(session_meta)) - } - RolloutRecorderParams::Resume { path } => { - let path = compression::materialize_rollout_for_append(path.as_path()).await?; ( - Some( - tokio::fs::OpenOptions::new() - .append(true) - .open(&path) - .await?, - ), None, + Some(log_file_info), path, - None, + Some(session_meta), + ordinal_state, ) } + RolloutRecorderParams::Resume { path } => { + let (path, file) = open_rollout_for_append(path.as_path()).await?; + let ordinal_state = ordinal_state_for_rollout(path.as_path()).await?; + (Some(file), None, path, None, ordinal_state) + } }; - // Clone the cwd for the spawned task to collect git info asynchronously + // Clone the cwd for the spawned task to collect git info asynchronously. let cwd = config.cwd().to_path_buf(); // A reasonably-sized bounded channel. If the buffer fills up the send @@ -847,6 +851,7 @@ impl RolloutRecorder { meta, cwd, rollout_path_for_spawn.clone(), + ordinal_state, ) .await; if let Err(err) = result { @@ -1554,6 +1559,7 @@ struct RolloutWriterState { meta: Option, cwd: PathBuf, rollout_path: PathBuf, + ordinal_state: RolloutOrdinalState, last_logged_error: Option, } @@ -1564,6 +1570,7 @@ impl RolloutWriterState { meta: Option, cwd: PathBuf, rollout_path: PathBuf, + ordinal_state: RolloutOrdinalState, ) -> Self { Self { writer: file.map(|file| JsonlWriter { file }), @@ -1572,6 +1579,7 @@ impl RolloutWriterState { meta, cwd, rollout_path, + ordinal_state, last_logged_error: None, } } @@ -1675,7 +1683,13 @@ impl RolloutWriterState { let Some(session_meta) = self.meta.as_ref().cloned() else { return Ok(()); }; - write_session_meta(self.writer.as_mut(), session_meta, &self.cwd).await?; + write_session_meta( + self.writer.as_mut(), + &mut self.ordinal_state, + session_meta, + &self.cwd, + ) + .await?; self.meta = None; Ok(()) } @@ -1700,9 +1714,18 @@ impl RolloutWriterState { let mut written_count = 0usize; let mut write_result = Ok(()); for item in &self.pending_items { - if let Err(err) = writer.write_rollout_item(item).await { - write_result = Err(err); - break; + match self.ordinal_state.current() { + Ok(ordinal) => match writer.write_rollout_item(item, ordinal).await { + Ok(()) => self.ordinal_state.advance(), + Err(err) => { + write_result = Err(err); + break; + } + }, + Err(err) => { + write_result = Err(err); + break; + } } written_count += 1; } @@ -1722,8 +1745,16 @@ async fn rollout_writer( meta: Option, cwd: PathBuf, rollout_path: PathBuf, + ordinal_state: RolloutOrdinalState, ) -> std::io::Result<()> { - let mut state = RolloutWriterState::new(file, deferred_log_file_info, meta, cwd, rollout_path); + let mut state = RolloutWriterState::new( + file, + deferred_log_file_info, + meta, + cwd, + rollout_path, + ordinal_state, + ); // Process rollout commands while let Some(cmd) = rx.recv().await { @@ -1755,6 +1786,7 @@ async fn rollout_writer( async fn write_session_meta( mut writer: Option<&mut JsonlWriter>, + ordinal_state: &mut RolloutOrdinalState, session_meta: SessionMeta, cwd: &Path, ) -> std::io::Result<()> { @@ -1774,7 +1806,9 @@ async fn write_session_meta( let rollout_item = RolloutItem::SessionMeta(session_meta_line); if let Some(writer) = writer.as_mut() { - writer.write_rollout_item(&rollout_item).await?; + let ordinal = ordinal_state.current()?; + writer.write_rollout_item(&rollout_item, ordinal).await?; + ordinal_state.advance(); } Ok(()) } @@ -1788,13 +1822,42 @@ pub async fn append_rollout_item_to_path( rollout_path: &Path, item: &RolloutItem, ) -> std::io::Result<()> { - let rollout_path = compression::materialize_rollout_for_append(rollout_path).await?; + let (rollout_path, file) = open_rollout_for_append(rollout_path).await?; + let ordinal_state = ordinal_state_for_rollout(rollout_path.as_path()).await?; + let ordinal = ordinal_state.current()?; + let mut writer = JsonlWriter { file }; + writer.write_rollout_item(item, ordinal).await +} + +async fn open_rollout_for_append(path: &Path) -> std::io::Result<(PathBuf, tokio::fs::File)> { + let path = compression::materialize_rollout_for_append(path).await?; + let path_for_repair = path.clone(); + tokio::task::spawn_blocking(move || { + ensure_rollout_is_newline_terminated(path_for_repair.as_path()) + }) + .await + .map_err(IoError::other)??; let file = tokio::fs::OpenOptions::new() .append(true) - .open(rollout_path) + .open(&path) .await?; - let mut writer = JsonlWriter { file }; - writer.write_rollout_item(item).await + Ok((path, file)) +} + +fn ensure_rollout_is_newline_terminated(path: &Path) -> std::io::Result<()> { + let mut file = File::options().read(true).append(true).open(path)?; + if file.metadata()?.len() == 0 { + return Ok(()); + } + + file.seek(SeekFrom::End(-1))?; + let mut final_byte = [0]; + file.read_exact(&mut final_byte)?; + if final_byte[0] != b'\n' { + file.write_all(b"\n")?; + file.flush()?; + } + Ok(()) } struct JsonlWriter { @@ -1804,12 +1867,18 @@ struct JsonlWriter { #[derive(serde::Serialize)] struct RolloutLineRef<'a> { timestamp: String, + #[serde(skip_serializing_if = "Option::is_none")] + ordinal: Option, #[serde(flatten)] item: &'a RolloutItem, } impl JsonlWriter { - async fn write_rollout_item(&mut self, rollout_item: &RolloutItem) -> std::io::Result<()> { + async fn write_rollout_item( + &mut self, + rollout_item: &RolloutItem, + ordinal: Option, + ) -> std::io::Result<()> { let timestamp_format: &[FormatItem] = format_description!( "[year]-[month]-[day]T[hour]:[minute]:[second].[subsecond digits:3]Z" ); @@ -1819,6 +1888,7 @@ impl JsonlWriter { let line = RolloutLineRef { timestamp, + ordinal, item: rollout_item, }; self.write_line(&line).await diff --git a/codex-rs/rollout/src/recorder_tests.rs b/codex-rs/rollout/src/recorder_tests.rs index ba07a4f5c0c2..a5efbaafa147 100644 --- a/codex-rs/rollout/src/recorder_tests.rs +++ b/codex-rs/rollout/src/recorder_tests.rs @@ -38,6 +38,67 @@ fn test_config(codex_home: &Path) -> RolloutConfig { } } +fn paginated_session_meta_item(thread_id: ThreadId, cwd: &Path) -> RolloutItem { + RolloutItem::SessionMeta(SessionMetaLine { + meta: SessionMeta { + session_id: thread_id.into(), + id: thread_id, + timestamp: "2026-07-09T00:00:00Z".to_string(), + cwd: cwd.to_path_buf(), + originator: "test".to_string(), + cli_version: "test".to_string(), + source: SessionSource::Exec, + history_mode: ThreadHistoryMode::Paginated, + ..SessionMeta::default() + }, + git: None, + }) +} + +fn agent_message_item(message: &str) -> RolloutItem { + RolloutItem::EventMsg(EventMsg::AgentMessage(AgentMessageEvent { + message: message.to_string(), + phase: None, + memory_citation: None, + })) +} + +fn write_paginated_rollout( + path: &Path, + thread_id: ThreadId, + subsequent_ordinals: &[u64], +) -> std::io::Result<()> { + let mut records = vec![RolloutLine { + timestamp: "2026-07-09T00:00:00Z".to_string(), + ordinal: Some(0), + item: paginated_session_meta_item(thread_id, path.parent().unwrap_or(path)), + }]; + records.extend( + subsequent_ordinals + .iter() + .enumerate() + .map(|(index, ordinal)| RolloutLine { + timestamp: format!("2026-07-09T00:00:{:02}Z", index + 1), + ordinal: Some(*ordinal), + item: agent_message_item(format!("message-{index}").as_str()), + }), + ); + let jsonl = records + .iter() + .map(serde_json::to_string) + .collect::, _>>()? + .join("\n"); + fs::write(path, format!("{jsonl}\n")) +} + +fn read_rollout_lines(path: &Path) -> std::io::Result> { + fs::read_to_string(path)? + .lines() + .filter(|line| !line.trim().is_empty()) + .map(|line| serde_json::from_str(line).map_err(std::io::Error::other)) + .collect() +} + fn write_session_file(root: &Path, ts: &str, uuid: Uuid) -> std::io::Result { let day_dir = root.join("sessions/2025/01/03"); fs::create_dir_all(&day_dir)?; @@ -113,10 +174,12 @@ async fn state_db_init_backfills_before_returning() -> anyhow::Result<()> { let lines = [ RolloutLine { timestamp: "2026-01-27T12:34:56Z".to_string(), + ordinal: None, item: RolloutItem::SessionMeta(session_meta_line), }, RolloutLine { timestamp: "2026-01-27T12:34:57Z".to_string(), + ordinal: None, item: RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent { client_id: None, message: "hello from startup backfill".to_string(), @@ -476,6 +539,11 @@ async fn recorder_materializes_on_flush_with_pending_items() -> std::io::Result< assert!(rollout_path.exists(), "rollout file should be materialized"); let text = std::fs::read_to_string(&rollout_path)?; + let lines = read_rollout_lines(&rollout_path)?; + assert_eq!( + lines.iter().map(|line| line.ordinal).collect::>(), + vec![Some(0), Some(1), Some(2)] + ); let first_line = text.lines().next().expect("session metadata line"); let session_meta: RolloutLine = serde_json::from_str(first_line)?; let RolloutItem::SessionMeta(session_meta) = session_meta.item else { @@ -507,6 +575,39 @@ async fn recorder_materializes_on_flush_with_pending_items() -> std::io::Result< Ok(()) } +#[tokio::test] +async fn recorder_omits_ordinals_from_legacy_rollouts() -> std::io::Result<()> { + let home = TempDir::new().expect("temp dir"); + let config = test_config(home.path()); + let recorder = RolloutRecorder::new( + &config, + RolloutRecorderParams::new( + ThreadId::new(), + /*forked_from_id*/ None, + /*parent_thread_id*/ None, + SessionSource::Exec, + /*thread_source*/ None, + "test_originator".to_string(), + BaseInstructions::default(), + Vec::new(), + ), + ) + .await?; + recorder + .record_canonical_items(&[agent_message_item("legacy")]) + .await?; + recorder.flush().await?; + + let text = fs::read_to_string(recorder.rollout_path())?; + let values = text + .lines() + .map(serde_json::from_str::) + .collect::, _>>()?; + assert!(values.iter().all(|value| value.get("ordinal").is_none())); + + recorder.shutdown().await +} + #[tokio::test] async fn persist_reports_filesystem_error_and_retries_buffered_items() -> std::io::Result<()> { let home = TempDir::new().expect("temp dir"); @@ -574,6 +675,7 @@ async fn writer_state_retries_write_error_before_reporting_flush_success() -> st /*meta*/ None, home.path().to_path_buf(), rollout_path.clone(), + RolloutOrdinalState::Legacy, ); state.add_items(vec![RolloutItem::EventMsg(EventMsg::AgentMessage( AgentMessageEvent { @@ -592,6 +694,114 @@ async fn writer_state_retries_write_error_before_reporting_flush_success() -> st Ok(()) } +#[tokio::test] +async fn resumed_paginated_rollout_continues_after_ordinal_gap() -> std::io::Result<()> { + let home = TempDir::new().expect("temp dir"); + let config = test_config(home.path()); + let rollout_path = home.path().join("rollout.jsonl"); + write_paginated_rollout(&rollout_path, ThreadId::new(), &[4])?; + + let recorder = + RolloutRecorder::new(&config, RolloutRecorderParams::resume(rollout_path.clone())).await?; + recorder + .record_canonical_items(&[agent_message_item("after-resume")]) + .await?; + recorder.flush().await?; + + let lines = read_rollout_lines(&rollout_path)?; + assert_eq!( + lines.iter().map(|line| line.ordinal).collect::>(), + vec![Some(0), Some(4), Some(5)] + ); + recorder.shutdown().await +} + +#[tokio::test] +async fn resumed_paginated_rollout_repairs_unsafe_tail() -> std::io::Result<()> { + let valid_unterminated = serde_json::to_string(&RolloutLine { + timestamp: "2026-07-09T00:00:05Z".to_string(), + ordinal: Some(5), + item: agent_message_item("valid unterminated"), + })?; + for (name, tail, expected_ordinals) in [ + ( + "valid unterminated", + valid_unterminated, + vec![Some(0), Some(4), Some(5), Some(6)], + ), + ( + "invalid unterminated", + "{\"timestamp\":\"unterminated\"".to_string(), + vec![Some(0), Some(4), Some(5)], + ), + ] { + let home = TempDir::new().expect("temp dir"); + let config = test_config(home.path()); + let rollout_path = home.path().join("rollout.jsonl"); + write_paginated_rollout(&rollout_path, ThreadId::new(), &[4])?; + let mut file = fs::OpenOptions::new().append(true).open(&rollout_path)?; + write!(file, "{tail}")?; + drop(file); + + let recorder = + RolloutRecorder::new(&config, RolloutRecorderParams::resume(rollout_path.clone())) + .await?; + recorder + .record_canonical_items(&[agent_message_item("after-tail-repair")]) + .await?; + recorder.flush().await?; + + let contents = fs::read_to_string(&rollout_path)?; + assert!(contents.ends_with('\n'), "{name} tail should be terminated"); + let ordinals = contents + .lines() + .filter_map(|line| serde_json::from_str::(line).ok()) + .map(|line| line.ordinal) + .collect::>(); + assert_eq!( + ordinals, expected_ordinals, + "unexpected ordinals after repairing {name} tail" + ); + recorder.shutdown().await?; + } + Ok(()) +} + +#[tokio::test] +async fn paginated_ordinal_overflow_fails_without_appending() -> std::io::Result<()> { + let home = TempDir::new().expect("temp dir"); + let config = test_config(home.path()); + let rollout_path = home.path().join("rollout.jsonl"); + write_paginated_rollout(&rollout_path, ThreadId::new(), &[u64::MAX])?; + let before = fs::read(&rollout_path)?; + + let recorder = + RolloutRecorder::new(&config, RolloutRecorderParams::resume(rollout_path.clone())).await?; + recorder + .record_canonical_items(&[agent_message_item("overflow")]) + .await?; + let err = recorder + .flush() + .await + .expect_err("ordinal overflow should fail the append"); + assert!(err.to_string().contains("overflow")); + assert_eq!(fs::read(&rollout_path)?, before); + Ok(()) +} + +#[tokio::test] +async fn append_rollout_item_to_path_assigns_next_paginated_ordinal() -> std::io::Result<()> { + let home = TempDir::new().expect("temp dir"); + let rollout_path = home.path().join("rollout.jsonl"); + write_paginated_rollout(&rollout_path, ThreadId::new(), &[4])?; + + append_rollout_item_to_path(&rollout_path, &agent_message_item("offline")).await?; + + let lines = read_rollout_lines(&rollout_path)?; + assert_eq!(lines.last().and_then(|line| line.ordinal), Some(5)); + Ok(()) +} + #[tokio::test] async fn list_threads_db_disabled_does_not_skip_paginated_items() -> std::io::Result<()> { let home = TempDir::new().expect("temp dir"); @@ -1204,6 +1414,7 @@ async fn resume_candidate_matches_cwd_reads_latest_turn_context() -> std::io::Re let mut file = std::fs::OpenOptions::new().append(true).open(&path)?; let turn_context = RolloutLine { timestamp: "2025-01-03T13:00:01Z".to_string(), + ordinal: None, item: RolloutItem::TurnContext(TurnContextItem { turn_id: Some("turn-1".to_string()), cwd: serde_json::from_value(serde_json::json!(&latest_cwd)) diff --git a/codex-rs/rollout/src/session_index_tests.rs b/codex-rs/rollout/src/session_index_tests.rs index 45aad8279cfb..b7008dfc8bde 100644 --- a/codex-rs/rollout/src/session_index_tests.rs +++ b/codex-rs/rollout/src/session_index_tests.rs @@ -23,6 +23,7 @@ fn write_rollout_with_metadata(path: &Path, thread_id: ThreadId) -> std::io::Res let timestamp = "2024-01-01T00-00-00Z".to_string(); let line = RolloutLine { timestamp: timestamp.clone(), + ordinal: None, item: RolloutItem::SessionMeta(SessionMetaLine { meta: SessionMeta { session_id: thread_id.into(), diff --git a/codex-rs/rollout/src/state_db_tests.rs b/codex-rs/rollout/src/state_db_tests.rs index 18566d887ebc..dd67a9759b23 100644 --- a/codex-rs/rollout/src/state_db_tests.rs +++ b/codex-rs/rollout/src/state_db_tests.rs @@ -156,6 +156,7 @@ fn write_rollout_with_user_message( let lines = [ RolloutLine { timestamp: "2026-06-01T14:26:25Z".to_string(), + ordinal: None, item: RolloutItem::SessionMeta(SessionMetaLine { meta: SessionMeta { session_id: thread_id.into(), @@ -185,6 +186,7 @@ fn write_rollout_with_user_message( }, RolloutLine { timestamp: "2026-06-01T14:26:26Z".to_string(), + ordinal: None, item: RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent { message: message.to_string(), ..Default::default() diff --git a/codex-rs/rollout/src/tests.rs b/codex-rs/rollout/src/tests.rs index 23f72cb9473e..9285faacdbc0 100644 --- a/codex-rs/rollout/src/tests.rs +++ b/codex-rs/rollout/src/tests.rs @@ -1314,6 +1314,7 @@ async fn test_updated_at_uses_file_mtime() -> Result<()> { let conversation_id = ThreadId::from_string(&uuid.to_string())?; let meta_line = RolloutLine { timestamp: ts.to_string(), + ordinal: None, item: RolloutItem::SessionMeta(SessionMetaLine { meta: SessionMeta { session_id: conversation_id.into(), @@ -1345,6 +1346,7 @@ async fn test_updated_at_uses_file_mtime() -> Result<()> { let user_event_line = RolloutLine { timestamp: ts.to_string(), + ordinal: None, item: RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent { client_id: None, message: "hello".into(), @@ -1360,6 +1362,7 @@ async fn test_updated_at_uses_file_mtime() -> Result<()> { for idx in 0..total_messages { let response_line = RolloutLine { timestamp: format!("{ts}-{idx:02}"), + ordinal: None, item: RolloutItem::ResponseItem(ResponseItem::Message { id: None, role: "assistant".into(), From 660395c47d0c02b42adedfaf676b62f0de72c21c Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Thu, 9 Jul 2026 21:55:48 -0700 Subject: [PATCH 2/2] codex: fix CI failure on PR #31859 --- codex-rs/rollout/src/ordinal.rs | 13 ++++++------- codex-rs/rollout/src/recorder_tests.rs | 24 ++++++++++++++++++++++++ 2 files changed, 30 insertions(+), 7 deletions(-) diff --git a/codex-rs/rollout/src/ordinal.rs b/codex-rs/rollout/src/ordinal.rs index c87b32c02bb8..7325b29793b5 100644 --- a/codex-rs/rollout/src/ordinal.rs +++ b/codex-rs/rollout/src/ordinal.rs @@ -53,7 +53,9 @@ pub(crate) async fn ordinal_state_for_rollout(path: &Path) -> io::Result io::Result { - let history_mode = read_history_mode(path)?; + let Some(history_mode) = read_history_mode(path)? else { + return Ok(RolloutOrdinalState::Legacy); + }; if matches!(history_mode, ThreadHistoryMode::Legacy) { return Ok(RolloutOrdinalState::Legacy); } @@ -82,7 +84,7 @@ fn ordinal_state_for_rollout_blocking(path: &Path) -> io::Result io::Result { +fn read_history_mode(path: &Path) -> io::Result> { let reader = BufReader::new(File::open(path)?); for line in reader.lines() { let line = line?; @@ -101,10 +103,7 @@ fn read_history_mode(path: &Path) -> io::Result { path.display() ))); }; - return Ok(session_meta.meta.history_mode); + return Ok(Some(session_meta.meta.history_mode)); } - Err(io::Error::other(format!( - "rollout at {} contains no records", - path.display() - ))) + Ok(None) } diff --git a/codex-rs/rollout/src/recorder_tests.rs b/codex-rs/rollout/src/recorder_tests.rs index a5efbaafa147..bece7643d7ce 100644 --- a/codex-rs/rollout/src/recorder_tests.rs +++ b/codex-rs/rollout/src/recorder_tests.rs @@ -608,6 +608,30 @@ async fn recorder_omits_ordinals_from_legacy_rollouts() -> std::io::Result<()> { recorder.shutdown().await } +#[tokio::test] +async fn resumed_empty_rollout_omits_ordinals() -> std::io::Result<()> { + let home = TempDir::new().expect("temp dir"); + let config = test_config(home.path()); + let rollout_path = home.path().join("rollout.jsonl"); + File::create(&rollout_path)?; + + let recorder = + RolloutRecorder::new(&config, RolloutRecorderParams::resume(rollout_path.clone())).await?; + recorder + .record_canonical_items(&[agent_message_item("legacy")]) + .await?; + recorder.flush().await?; + + let text = fs::read_to_string(rollout_path)?; + let values = text + .lines() + .map(serde_json::from_str::) + .collect::, _>>()?; + assert!(values.iter().all(|value| value.get("ordinal").is_none())); + + recorder.shutdown().await +} + #[tokio::test] async fn persist_reports_filesystem_error_and_retries_buffered_items() -> std::io::Result<()> { let home = TempDir::new().expect("temp dir");