github/codex
4
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-21 04:50:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

rollout: add ordinals to paginated records

Browse Source
This commit is contained in:
Owen Lin
2026-07-09 21:41:02 -07:00
parent 239e27c004
commit 3b1fd5bb1c
17 changed files with 456 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/request_processors/thread_processor_tests.rs
+3
View File
@@ -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,
codex-rs/cli/src/doctor/thread_inventory.rs
+1
View File
@@ -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(),
codex-rs/core/src/personality_migration_tests.rs
+2
View File
@@ -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(),
codex-rs/core/tests/suite/client.rs
+11
View File
@@ -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(),
codex-rs/core/tests/suite/personality_migration.rs
+3
View File
@@ -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),
};
codex-rs/core/tests/suite/sqlite_state.rs
+2
View File
@@ -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(),
codex-rs/memories/write/src/startup_tests.rs
+1
View File
@@ -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(),
codex-rs/protocol/src/protocol.rs
+2
View File
@@ -3325,6 +3325,8 @@ impl Mul<f64> for TruncationPolicy {
#[derive(Serialize, Deserialize, Clone, JsonSchema)]
pub struct RolloutLine {
pub timestamp: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub ordinal: Option<u64>,
#[serde(flatten)]
pub item: RolloutItem,
}
codex-rs/rollout/src/compression_tests.rs
+2
View File
@@ -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()
codex-rs/rollout/src/lib.rs
+1
View File
@@ -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;
codex-rs/rollout/src/metadata_tests.rs
+5
View File
@@ -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");
codex-rs/rollout/src/ordinal.rs
+110
View File
@@ -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<u64> },
}
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<Option<u64>> {
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<RolloutOrdinalState> {
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<RolloutOrdinalState> {
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::<RolloutLine>()? {
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<ThreadHistoryMode> {
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()
)))
}
codex-rs/rollout/src/recorder.rs
+96 -26
View File
@@ -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<Self> {
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))
(
None,
Some(log_file_info),
path,
Some(session_meta),
ordinal_state,
)
}
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,
path,
None,
)
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<SessionMeta>,
cwd: PathBuf,
rollout_path: PathBuf,
ordinal_state: RolloutOrdinalState,
last_logged_error: Option<String>,
}
@@ -1564,6 +1570,7 @@ impl RolloutWriterState {
meta: Option<SessionMeta>,
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<SessionMeta>,
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<u64>,
#[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<u64>,
) -> 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
codex-rs/rollout/src/recorder_tests.rs
+211
View File
@@ -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::<Result<Vec<_>, _>>()?
.join("\n");
fs::write(path, format!("{jsonl}\n"))
}
fn read_rollout_lines(path: &Path) -> std::io::Result<Vec<RolloutLine>> {
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<PathBuf> {
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<_>>(),
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::<serde_json::Value>)
.collect::<Result<Vec<_>, _>>()?;
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<_>>(),
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::<RolloutLine>(line).ok())
.map(|line| line.ordinal)
.collect::<Vec<_>>();
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))
codex-rs/rollout/src/session_index_tests.rs
+1
View File
@@ -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(),
codex-rs/rollout/src/state_db_tests.rs
+2
View File
@@ -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()
codex-rs/rollout/src/tests.rs
+3
View File
@@ -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(),