openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-06 20:20:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: compress rollout files

Browse Source
This commit is contained in:
jif-oai
2026-03-31 16:17:24 +02:00
parent 4c72e62d0b
commit 8e6c8d9de9
22 changed files with 618 additions and 305 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/Cargo.lock
Generated
+1
View File
@@ -2478,6 +2478,7 @@ dependencies = [
"tokio",
"tracing",
"uuid",
"zstd",
]
[[package]]
codex-rs/app-server/src/codex_message_processor.rs
+10 -10
View File
@@ -2972,7 +2972,6 @@ impl CodexMessageProcessor {
});
};
let required_suffix = format!("{thread_id}.jsonl");
let Some(file_name) = canonical_rollout_path.file_name().map(OsStr::to_owned) else {
return Err(JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
@@ -2980,9 +2979,7 @@ impl CodexMessageProcessor {
data: None,
});
};
if !file_name
.to_string_lossy()
.ends_with(required_suffix.as_str())
if !rollout_file_name_matches_thread_id(file_name.to_string_lossy().as_ref(), thread_id)
{
return Err(JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
@@ -5324,7 +5321,6 @@ impl CodexMessageProcessor {
};
// Verify file name matches thread id.
let required_suffix = format!("{thread_id}.jsonl");
let Some(file_name) = canonical_rollout_path.file_name().map(OsStr::to_owned) else {
return Err(JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
@@ -5335,10 +5331,7 @@ impl CodexMessageProcessor {
data: None,
});
};
if !file_name
.to_string_lossy()
.ends_with(required_suffix.as_str())
{
if !rollout_file_name_matches_thread_id(file_name.to_string_lossy().as_ref(), thread_id) {
return Err(JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: format!(
@@ -8274,7 +8267,7 @@ async fn summary_from_thread_list_item(
fn thread_id_from_rollout_path(path: &Path) -> Option<ThreadId> {
let file_name = path.file_name()?.to_str()?;
let stem = file_name.strip_suffix(".jsonl")?;
let stem = codex_core::strip_rollout_file_suffix(file_name)?;
if stem.len() < 37 {
return None;
}
@@ -8285,6 +8278,13 @@ fn thread_id_from_rollout_path(path: &Path) -> Option<ThreadId> {
ThreadId::from_string(&stem[uuid_start..]).ok()
}
fn rollout_file_name_matches_thread_id(file_name: &str, thread_id: ThreadId) -> bool {
let Some(stem) = codex_core::strip_rollout_file_suffix(file_name) else {
return false;
};
stem.ends_with(&format!("-{thread_id}"))
}
#[allow(clippy::too_many_arguments)]
fn summary_from_state_db_metadata(
conversation_id: ThreadId,
codex-rs/core/src/lib.rs
+4
View File
@@ -172,6 +172,7 @@ pub use rollout::find_conversation_path_by_id_str;
pub use rollout::find_thread_name_by_id;
pub use rollout::find_thread_path_by_id_str;
pub use rollout::find_thread_path_by_name_str;
pub use rollout::is_rollout_path;
pub use rollout::list::Cursor;
pub use rollout::list::ThreadItem;
pub use rollout::list::ThreadSortKey;
@@ -180,8 +181,11 @@ pub use rollout::list::parse_cursor;
pub use rollout::list::read_head_for_summary;
pub use rollout::list::read_session_meta_line;
pub use rollout::policy::EventPersistenceMode;
pub use rollout::read_nonempty_rollout_text;
pub use rollout::read_rollout_text;
pub use rollout::rollout_date_parts;
pub use rollout::session_index::find_thread_names_by_ids;
pub use rollout::strip_rollout_file_suffix;
mod function_tool;
mod state;
mod tasks;
codex-rs/core/src/rollout.rs
+4
View File
@@ -12,7 +12,11 @@ pub use codex_rollout::find_conversation_path_by_id_str;
pub use codex_rollout::find_thread_name_by_id;
pub use codex_rollout::find_thread_path_by_id_str;
pub use codex_rollout::find_thread_path_by_name_str;
pub use codex_rollout::is_rollout_path;
pub use codex_rollout::read_nonempty_rollout_text;
pub use codex_rollout::read_rollout_text;
pub use codex_rollout::rollout_date_parts;
pub use codex_rollout::strip_rollout_file_suffix;
impl codex_rollout::RolloutConfigView for Config {
fn codex_home(&self) -> &std::path::Path {
codex-rs/core/tests/suite/cli_stream.rs
+8 -6
View File
@@ -1,5 +1,7 @@
use assert_cmd::Command as AssertCommand;
use codex_core::auth::CODEX_API_KEY_ENV_VAR;
use codex_core::is_rollout_path;
use codex_core::read_rollout_text;
use codex_git_utils::collect_git_info;
use codex_protocol::protocol::GitInfo;
use codex_utils_cargo_bin::find_resource;
@@ -375,10 +377,10 @@ async fn integration_creates_and_checks_session_file() -> anyhow::Result<()> {
// Find the session file that contains `marker`.
let marker_clone = marker.clone();
let path = fs_wait::wait_for_matching_file(&sessions_dir, Duration::from_secs(10), move |p| {
if p.extension().and_then(|ext| ext.to_str()) != Some("jsonl") {
if !is_rollout_path(p) {
return false;
}
let Ok(content) = std::fs::read_to_string(p) else {
let Ok(content) = read_rollout_text(p) else {
return false;
};
content.contains(&marker_clone)
@@ -422,7 +424,7 @@ async fn integration_creates_and_checks_session_file() -> anyhow::Result<()> {
}
let content =
std::fs::read_to_string(&path).unwrap_or_else(|_| panic!("Failed to read session file"));
read_rollout_text(path.as_path()).unwrap_or_else(|_| panic!("Failed to read session file"));
let mut lines = content.lines();
let meta_line = lines
.next()
@@ -491,10 +493,10 @@ async fn integration_creates_and_checks_session_file() -> anyhow::Result<()> {
let marker2_clone = marker2.clone();
let resumed_path =
fs_wait::wait_for_matching_file(&sessions_dir, Duration::from_secs(10), move |p| {
if p.extension().and_then(|ext| ext.to_str()) != Some("jsonl") {
if !is_rollout_path(p) {
return false;
}
std::fs::read_to_string(p)
read_rollout_text(p)
.map(|content| content.contains(&marker2_clone))
.unwrap_or(false)
})
@@ -506,7 +508,7 @@ async fn integration_creates_and_checks_session_file() -> anyhow::Result<()> {
"resume should create a new session file"
);
let resumed_content = std::fs::read_to_string(&resumed_path)?;
let resumed_content = read_rollout_text(resumed_path.as_path())?;
assert!(
resumed_content.contains(&marker),
"resumed file missing original marker"
codex-rs/core/tests/suite/compact.rs
+3 -2
View File
@@ -5,6 +5,7 @@ use codex_core::built_in_model_providers;
use codex_core::compact::SUMMARIZATION_PROMPT;
use codex_core::compact::SUMMARY_PREFIX;
use codex_core::config::Config;
use codex_core::read_rollout_text;
use codex_features::Feature;
use codex_protocol::items::TurnItem;
use codex_protocol::openai_models::ModelInfo;
@@ -368,7 +369,7 @@ async fn summarize_context_three_requests_and_instructions() {
// Verify rollout contains user-turn TurnContext entries and a Compacted entry.
println!("rollout path: {}", rollout_path.display());
let text = std::fs::read_to_string(&rollout_path).unwrap_or_else(|e| {
let text = read_rollout_text(rollout_path.as_path()).unwrap_or_else(|e| {
panic!(
"failed to read rollout file {}: {e}",
rollout_path.display()
@@ -2069,7 +2070,7 @@ async fn auto_compact_persists_rollout_entries() {
wait_for_event(&codex, |ev| matches!(ev, EventMsg::ShutdownComplete)).await;
let rollout_path = session_configured.rollout_path.expect("rollout path");
let text = std::fs::read_to_string(&rollout_path).unwrap_or_else(|e| {
let text = read_rollout_text(rollout_path.as_path()).unwrap_or_else(|e| {
panic!(
"failed to read rollout file {}: {e}",
rollout_path.display()
codex-rs/core/tests/suite/compact_remote.rs
+2 -1
View File
@@ -6,6 +6,7 @@ use std::path::PathBuf;
use anyhow::Result;
use codex_core::CodexAuth;
use codex_core::compact::SUMMARY_PREFIX;
use codex_core::read_rollout_text;
use codex_protocol::items::TurnItem;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ResponseItem;
@@ -1160,7 +1161,7 @@ async fn remote_compact_persists_replacement_history_in_rollout() -> Result<()>
assert_eq!(responses_mock.requests().len(), 1);
assert_eq!(compact_mock.requests().len(), 1);
let rollout_text = fs::read_to_string(&rollout_path)?;
let rollout_text = read_rollout_text(rollout_path.as_path())?;
let mut saw_compacted_history = false;
for line in rollout_text
.lines()
codex-rs/core/tests/suite/fork_thread.rs
+2 -1
View File
@@ -1,6 +1,7 @@
use codex_core::ForkSnapshot;
use codex_core::NewThread;
use codex_core::parse_turn_item;
use codex_core::read_rollout_text;
use codex_protocol::items::TurnItem;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::Op;
@@ -66,7 +67,7 @@ async fn fork_thread_twice_drops_to_first_message() {
// Helper: read rollout items (excluding SessionMeta) from a JSONL path.
let read_items = |p: &std::path::Path| -> Vec<RolloutItem> {
let text = std::fs::read_to_string(p).expect("read rollout file");
let text = read_rollout_text(p).expect("read rollout file");
let mut items: Vec<RolloutItem> = Vec::new();
for line in text.lines() {
if line.trim().is_empty() {
codex-rs/core/tests/suite/hooks.rs
+3 -2
View File
@@ -3,6 +3,7 @@ use std::path::Path;
use anyhow::Context;
use anyhow::Result;
use codex_core::read_rollout_text;
use codex_features::Feature;
use codex_protocol::items::parse_hook_prompt_fragment;
use codex_protocol::models::ContentItem;
@@ -558,7 +559,7 @@ async fn stop_hook_can_block_multiple_times_in_same_turn() -> Result<()> {
);
let rollout_path = test.codex.rollout_path().expect("rollout path");
let rollout_text = fs::read_to_string(&rollout_path)?;
let rollout_text = read_rollout_text(rollout_path.as_path())?;
let hook_prompt_texts = rollout_hook_prompt_texts(&rollout_text)?;
assert!(
hook_prompt_texts.contains(&FIRST_CONTINUATION_PROMPT.to_string()),
@@ -746,7 +747,7 @@ async fn multiple_blocking_stop_hooks_persist_multiple_hook_prompt_fragments() -
);
let rollout_path = test.codex.rollout_path().expect("rollout path");
let rollout_text = fs::read_to_string(&rollout_path)?;
let rollout_text = read_rollout_text(rollout_path.as_path())?;
assert_eq!(
rollout_hook_prompt_texts(&rollout_text)?,
vec![
codex-rs/core/tests/suite/image_rollout.rs
+7 -17
View File
@@ -1,4 +1,5 @@
use anyhow::Context;
use codex_core::read_nonempty_rollout_text;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::AskForApproval;
@@ -22,7 +23,6 @@ use image::ImageBuffer;
use image::Rgba;
use pretty_assertions::assert_eq;
use std::path::Path;
use std::time::Duration;
fn find_user_message_with_image(text: &str) -> Option<ResponseItem> {
for line in text.lines() {
@@ -58,20 +58,6 @@ fn extract_image_url(item: &ResponseItem) -> Option<String> {
}
}
async fn read_rollout_text(path: &Path) -> anyhow::Result<String> {
for _ in 0..50 {
if path.exists()
&& let Ok(text) = std::fs::read_to_string(path)
&& !text.trim().is_empty()
{
return Ok(text);
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
std::fs::read_to_string(path)
.with_context(|| format!("read rollout file at {}", path.display()))
}
fn write_test_png(path: &Path, color: [u8; 4]) -> anyhow::Result<()> {
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)?;
@@ -138,7 +124,9 @@ async fn copy_paste_local_image_persists_rollout_request_shape() -> anyhow::Resu
wait_for_event(&codex, |event| matches!(event, EventMsg::ShutdownComplete)).await;
let rollout_path = codex.rollout_path().expect("rollout path");
let rollout_text = read_rollout_text(&rollout_path).await?;
let rollout_text = read_nonempty_rollout_text(&rollout_path)
.await
.with_context(|| format!("read rollout file at {}", rollout_path.display()))?;
let actual = find_user_message_with_image(&rollout_text)
.expect("expected user message with input image in rollout");
@@ -222,7 +210,9 @@ async fn drag_drop_image_persists_rollout_request_shape() -> anyhow::Result<()>
wait_for_event(&codex, |event| matches!(event, EventMsg::ShutdownComplete)).await;
let rollout_path = codex.rollout_path().expect("rollout path");
let rollout_text = read_rollout_text(&rollout_path).await?;
let rollout_text = read_nonempty_rollout_text(&rollout_path)
.await
.with_context(|| format!("read rollout file at {}", rollout_path.display()))?;
let actual = find_user_message_with_image(&rollout_text)
.expect("expected user message with input image in rollout");
codex-rs/core/tests/suite/override_updates.rs
+4 -17
View File
@@ -1,5 +1,6 @@
use anyhow::Result;
use codex_core::config::Constrained;
use codex_core::read_nonempty_rollout_text;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::COLLABORATION_MODE_CLOSE_TAG;
use codex_protocol::protocol::COLLABORATION_MODE_OPEN_TAG;
@@ -19,7 +20,6 @@ use core_test_support::test_codex::test_codex;
use core_test_support::wait_for_event;
use pretty_assertions::assert_eq;
use std::path::Path;
use std::time::Duration;
use tempfile::TempDir;
fn collab_mode_with_instructions(instructions: Option<&str>) -> CollaborationMode {
@@ -37,19 +37,6 @@ fn collab_xml(text: &str) -> String {
format!("{COLLABORATION_MODE_OPEN_TAG}{text}{COLLABORATION_MODE_CLOSE_TAG}")
}
async fn read_rollout_text(path: &Path) -> anyhow::Result<String> {
for _ in 0..50 {
if path.exists()
&& let Ok(text) = std::fs::read_to_string(path)
&& !text.trim().is_empty()
{
return Ok(text);
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
Ok(std::fs::read_to_string(path)?)
}
fn rollout_developer_texts(text: &str) -> Vec<String> {
let mut texts = Vec::new();
for line in text.lines() {
@@ -132,7 +119,7 @@ async fn override_turn_context_without_user_turn_does_not_record_permissions_upd
wait_for_event(&test.codex, |ev| matches!(ev, EventMsg::ShutdownComplete)).await;
let rollout_path = test.codex.rollout_path().expect("rollout path");
let rollout_text = read_rollout_text(&rollout_path).await?;
let rollout_text = read_nonempty_rollout_text(&rollout_path).await?;
let developer_texts = rollout_developer_texts(&rollout_text);
let approval_texts: Vec<&String> = developer_texts
.iter()
@@ -174,7 +161,7 @@ async fn override_turn_context_without_user_turn_does_not_record_environment_upd
wait_for_event(&test.codex, |ev| matches!(ev, EventMsg::ShutdownComplete)).await;
let rollout_path = test.codex.rollout_path().expect("rollout path");
let rollout_text = read_rollout_text(&rollout_path).await?;
let rollout_text = read_nonempty_rollout_text(&rollout_path).await?;
let env_texts = rollout_environment_texts(&rollout_text);
assert!(
env_texts.is_empty(),
@@ -213,7 +200,7 @@ async fn override_turn_context_without_user_turn_does_not_record_collaboration_u
wait_for_event(&test.codex, |ev| matches!(ev, EventMsg::ShutdownComplete)).await;
let rollout_path = test.codex.rollout_path().expect("rollout path");
let rollout_text = read_rollout_text(&rollout_path).await?;
let rollout_text = read_nonempty_rollout_text(&rollout_path).await?;
let developer_texts = rollout_developer_texts(&rollout_text);
let collab_text = collab_xml(collab_text);
let collab_count = developer_texts
codex-rs/core/tests/suite/review.rs
+3 -2
View File
@@ -1,6 +1,7 @@
use codex_core::CodexThread;
use codex_core::REVIEW_PROMPT;
use codex_core::config::Config;
use codex_core::read_rollout_text;
use codex_core::review_format::render_review_output_text;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ResponseItem;
@@ -120,7 +121,7 @@ async fn review_op_emits_lifecycle_and_review_output() {
// Also verify that a user message with the header and a formatted finding
// was recorded back in the parent session's rollout.
let path = codex.rollout_path().expect("rollout path");
let text = std::fs::read_to_string(&path).expect("read rollout file");
let text = read_rollout_text(path.as_path()).expect("read rollout file");
let mut saw_header = false;
let mut saw_finding_line = false;
@@ -641,7 +642,7 @@ async fn review_input_isolated_from_parent_history() {
// Also verify that a user interruption note was recorded in the rollout.
let path = codex.rollout_path().expect("rollout path");
let text = std::fs::read_to_string(&path).expect("read rollout file");
let text = read_rollout_text(path.as_path()).expect("read rollout file");
let mut saw_interruption_message = false;
for line in text.lines() {
if line.trim().is_empty() {
codex-rs/core/tests/suite/subagent_notifications.rs
+2 -2
View File
@@ -1,6 +1,7 @@
use anyhow::Result;
use codex_core::ThreadConfigSnapshot;
use codex_core::config::AgentRoleConfig;
use codex_core::read_rollout_text;
use codex_features::Feature;
use codex_protocol::ThreadId;
use codex_protocol::openai_models::ReasoningEffort;
@@ -227,8 +228,7 @@ async fn setup_turn_one_with_custom_spawned_child(
.ok_or_else(|| anyhow::anyhow!("expected parent rollout path"))?;
let deadline = Instant::now() + Duration::from_secs(6);
loop {
let has_notification = tokio::fs::read_to_string(&rollout_path)
.await
let has_notification = read_rollout_text(rollout_path.as_path())
.is_ok_and(|rollout| rollout.contains("<subagent_notification>"));
if has_notification {
break;
codex-rs/rollout/Cargo.toml
+1
View File
@@ -43,6 +43,7 @@ tokio = { workspace = true, features = [
] }
tracing = { workspace = true }
uuid = { workspace = true }
zstd = { workspace = true }
[dev-dependencies]
pretty_assertions = { workspace = true }
codex-rs/rollout/src/file_io.rs
+174
View File
@@ -0,0 +1,174 @@
use std::fs::File;
use std::fs::OpenOptions;
use std::io;
use std::io::BufRead;
use std::io::BufReader;
use std::io::Read;
use std::io::Write;
use std::path::Path;
use std::time::Duration;
use tokio::time::sleep;
pub const ROLLOUT_FILE_SUFFIX: &str = ".jsonl";
pub const COMPRESSED_ROLLOUT_FILE_SUFFIX: &str = ".jsonl.zst";
const DEFAULT_ZSTD_LEVEL: i32 = 0;
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum RolloutFileEncoding {
PlainJsonl,
ZstdJsonl,
}
impl RolloutFileEncoding {
pub(crate) fn for_path(path: &Path) -> Self {
path.file_name()
.and_then(|file_name| file_name.to_str())
.and_then(file_encoding_from_name)
.unwrap_or_else(|| Self::PlainJsonl)
}
pub(crate) fn preferred_suffix(self) -> &'static str {
match self {
Self::PlainJsonl => ROLLOUT_FILE_SUFFIX,
Self::ZstdJsonl => COMPRESSED_ROLLOUT_FILE_SUFFIX,
}
}
}
/// Returns the suffix used for newly created rollout files.
pub fn preferred_rollout_file_suffix() -> &'static str {
RolloutFileEncoding::ZstdJsonl.preferred_suffix()
}
/// Returns true when `name` matches a rollout filename in either supported encoding.
pub fn is_rollout_file_name(name: &str) -> bool {
name.starts_with("rollout-") && strip_rollout_file_suffix(name).is_some()
}
/// Returns true when `path` points to a rollout file in either supported encoding.
pub fn is_rollout_path(path: &Path) -> bool {
path.file_name()
.and_then(|file_name| file_name.to_str())
.is_some_and(is_rollout_file_name)
}
/// Removes the rollout suffix from `name`, supporting both plain and compressed files.
pub fn strip_rollout_file_suffix(name: &str) -> Option<&str> {
name.strip_suffix(COMPRESSED_ROLLOUT_FILE_SUFFIX)
.or_else(|| name.strip_suffix(ROLLOUT_FILE_SUFFIX))
}
pub(crate) fn append_text(path: &Path, text: &str) -> io::Result<()> {
match RolloutFileEncoding::for_path(path) {
RolloutFileEncoding::PlainJsonl => {
let mut file = OpenOptions::new().append(true).create(true).open(path)?;
file.write_all(text.as_bytes())?;
file.flush()
}
RolloutFileEncoding::ZstdJsonl => {
let file = OpenOptions::new().append(true).create(true).open(path)?;
let mut encoder = zstd::stream::write::Encoder::new(file, DEFAULT_ZSTD_LEVEL)?;
encoder.write_all(text.as_bytes())?;
let mut file = encoder.finish()?;
file.flush()
}
}
}
/// Reads the full rollout file contents, transparently handling plain and zstd-compressed files.
pub fn read_rollout_text(path: &Path) -> io::Result<String> {
let mut text = String::new();
match RolloutFileEncoding::for_path(path) {
RolloutFileEncoding::PlainJsonl => {
File::open(path)?.read_to_string(&mut text)?;
}
RolloutFileEncoding::ZstdJsonl => {
let file = File::open(path)?;
let mut decoder = zstd::stream::read::Decoder::new(file)?;
decoder.read_to_string(&mut text)?;
}
}
Ok(text)
}
/// Retries reading `path` until the rollout exists and contains non-empty text, or returns the
/// last read attempt.
pub async fn read_nonempty_rollout_text(path: &Path) -> io::Result<String> {
const MAX_ATTEMPTS: usize = 50;
const RETRY_DELAY: Duration = Duration::from_millis(20);
for _ in 0..MAX_ATTEMPTS {
if path.exists()
&& let Ok(text) = read_rollout_text(path)
&& !text.trim().is_empty()
{
return Ok(text);
}
sleep(RETRY_DELAY).await;
}
read_rollout_text(path)
}
pub(crate) struct RolloutLineReader {
inner: RolloutLineReaderInner,
}
enum RolloutLineReaderInner {
Plain(BufReader<File>),
Zstd(BufReader<zstd::stream::read::Decoder<'static, BufReader<File>>>),
}
impl RolloutLineReader {
pub(crate) fn open(path: &Path) -> io::Result<Self> {
let inner = match RolloutFileEncoding::for_path(path) {
RolloutFileEncoding::PlainJsonl => {
RolloutLineReaderInner::Plain(BufReader::new(File::open(path)?))
}
RolloutFileEncoding::ZstdJsonl => {
let file = BufReader::new(File::open(path)?);
let decoder = zstd::stream::read::Decoder::with_buffer(file)?;
RolloutLineReaderInner::Zstd(BufReader::new(decoder))
}
};
Ok(Self { inner })
}
pub(crate) fn next_line(&mut self) -> io::Result<Option<String>> {
let mut line = String::new();
let bytes_read = match &mut self.inner {
RolloutLineReaderInner::Plain(reader) => reader.read_line(&mut line)?,
RolloutLineReaderInner::Zstd(reader) => reader.read_line(&mut line)?,
};
if bytes_read == 0 {
return Ok(None);
}
trim_line_ending(&mut line);
Ok(Some(line))
}
}
fn file_encoding_from_name(name: &str) -> Option<RolloutFileEncoding> {
if name.ends_with(COMPRESSED_ROLLOUT_FILE_SUFFIX) {
return Some(RolloutFileEncoding::ZstdJsonl);
}
if name.ends_with(ROLLOUT_FILE_SUFFIX) {
return Some(RolloutFileEncoding::PlainJsonl);
}
None
}
fn trim_line_ending(line: &mut String) {
if line.ends_with('\n') {
line.pop();
if line.ends_with('\r') {
line.pop();
}
}
}
#[cfg(test)]
#[path = "file_io_tests.rs"]
mod tests;
codex-rs/rollout/src/file_io_tests.rs
+37
View File
@@ -0,0 +1,37 @@
use super::*;
use crate::file_io::RolloutLineReader;
use crate::file_io::append_text;
use std::io;
use tempfile::TempDir;
#[test]
fn strip_rollout_file_suffix_supports_both_formats() {
assert_eq!(
strip_rollout_file_suffix("rollout-2026-01-01T00-00-00-thread.jsonl"),
Some("rollout-2026-01-01T00-00-00-thread")
);
assert_eq!(
strip_rollout_file_suffix("rollout-2026-01-01T00-00-00-thread.jsonl.zst"),
Some("rollout-2026-01-01T00-00-00-thread")
);
assert_eq!(strip_rollout_file_suffix("rollout.txt"), None);
}
#[test]
fn compressed_appends_are_read_back_as_one_stream() -> io::Result<()> {
let temp_dir = TempDir::new()?;
let path = temp_dir
.path()
.join("rollout-2026-01-01T00-00-00-thread.jsonl.zst");
append_text(path.as_path(), "{\"a\":1}\n")?;
append_text(path.as_path(), "{\"b\":2}\n")?;
assert_eq!(read_rollout_text(path.as_path())?, "{\"a\":1}\n{\"b\":2}\n");
let mut reader = RolloutLineReader::open(path.as_path())?;
assert_eq!(reader.next_line()?, Some("{\"a\":1}".to_string()));
assert_eq!(reader.next_line()?, Some("{\"b\":2}".to_string()));
assert_eq!(reader.next_line()?, None);
Ok(())
}
codex-rs/rollout/src/lib.rs
+5
View File
@@ -5,6 +5,7 @@ use std::sync::LazyLock;
use codex_protocol::protocol::SessionSource;
pub mod config;
pub(crate) mod file_io;
pub mod list;
pub mod metadata;
pub mod policy;
@@ -32,6 +33,10 @@ pub static INTERACTIVE_SESSION_SOURCES: LazyLock<Vec<SessionSource>> = LazyLock:
pub use codex_protocol::protocol::SessionMeta;
pub use config::RolloutConfig;
pub use config::RolloutConfigView;
pub use file_io::is_rollout_path;
pub use file_io::read_nonempty_rollout_text;
pub use file_io::read_rollout_text;
pub use file_io::strip_rollout_file_suffix;
pub use list::find_archived_thread_path_by_id_str;
pub use list::find_thread_path_by_id_str;
#[deprecated(note = "use find_thread_path_by_id_str")]
codex-rs/rollout/src/list.rs
+117 -105
View File
@@ -17,6 +17,9 @@ use uuid::Uuid;
use super::ARCHIVED_SESSIONS_SUBDIR;
use super::SESSIONS_SUBDIR;
use super::file_io::RolloutLineReader;
use super::file_io::is_rollout_file_name;
use super::file_io::strip_rollout_file_suffix;
use crate::protocol::EventMsg;
use crate::state_db;
use codex_file_search as file_search;
@@ -378,7 +381,7 @@ pub async fn get_threads_in_root(
/// Load thread file paths from disk using directory traversal.
///
/// Directory layout: `~/.codex/sessions/YYYY/MM/DD/rollout-YYYY-MM-DDThh-mm-ss-<uuid>.jsonl`
/// Directory layout: `~/.codex/sessions/YYYY/MM/DD/rollout-YYYY-MM-DDThh-mm-ss-<uuid>.jsonl(.zst)`
/// Returned newest (based on sort key) first.
async fn traverse_directories_for_paths(
root: PathBuf,
@@ -823,7 +826,7 @@ async fn collect_flat_rollout_files(
let Some(name_str) = file_name.to_str() else {
continue;
};
if !name_str.starts_with("rollout-") || !name_str.ends_with(".jsonl") {
if !is_rollout_file_name(name_str) {
continue;
}
let Some((ts, id)) = parse_timestamp_uuid_from_filename(name_str) else {
@@ -843,7 +846,7 @@ async fn collect_rollout_day_files(
day_path: &Path,
) -> io::Result<Vec<(OffsetDateTime, Uuid, PathBuf)>> {
let mut day_files = collect_files(day_path, |name_str, path| {
if !name_str.starts_with("rollout-") || !name_str.ends_with(".jsonl") {
if !is_rollout_file_name(name_str) {
return None;
}
@@ -856,8 +859,8 @@ async fn collect_rollout_day_files(
}
pub(crate) fn parse_timestamp_uuid_from_filename(name: &str) -> Option<(OffsetDateTime, Uuid)> {
// Expected: rollout-YYYY-MM-DDThh-mm-ss-<uuid>.jsonl
let core = name.strip_prefix("rollout-")?.strip_suffix(".jsonl")?;
// Expected: rollout-YYYY-MM-DDThh-mm-ss-<uuid>.jsonl(.zst)
let core = strip_rollout_file_suffix(name)?.strip_prefix("rollout-")?;
// Scan from the right for a '-' such that the suffix parses as a UUID.
let (sep_idx, uuid) = core
@@ -913,7 +916,7 @@ async fn collect_flat_files_by_updated_at(
let Some(name_str) = file_name.to_str() else {
continue;
};
if !name_str.starts_with("rollout-") || !name_str.ends_with(".jsonl") {
if !is_rollout_file_name(name_str) {
continue;
}
let Some((_ts, id)) = parse_timestamp_uuid_from_filename(name_str) else {
@@ -1001,127 +1004,136 @@ impl<'a> ProviderMatcher<'a> {
}
async fn read_head_summary(path: &Path, head_limit: usize) -> io::Result<HeadTailSummary> {
use tokio::io::AsyncBufReadExt;
read_rollout_head(path, move |reader| {
let mut summary = HeadTailSummary::default();
let mut lines_scanned = 0usize;
let file = tokio::fs::File::open(path).await?;
let reader = tokio::io::BufReader::new(file);
let mut lines = reader.lines();
let mut summary = HeadTailSummary::default();
let mut lines_scanned = 0usize;
while lines_scanned < head_limit
|| (summary.saw_session_meta
&& !summary.saw_user_event
&& lines_scanned < head_limit + USER_EVENT_SCAN_LIMIT)
{
let Some(line) = reader.next_line()? else {
break;
};
let trimmed = line.trim();
if trimmed.is_empty() {
continue;
}
lines_scanned += 1;
while lines_scanned < head_limit
|| (summary.saw_session_meta
&& !summary.saw_user_event
&& lines_scanned < head_limit + USER_EVENT_SCAN_LIMIT)
{
let line_opt = lines.next_line().await?;
let Some(line) = line_opt else { break };
let trimmed = line.trim();
if trimmed.is_empty() {
continue;
}
lines_scanned += 1;
let Ok(rollout_line) = serde_json::from_str::<RolloutLine>(trimmed) else {
continue;
};
let parsed: Result<RolloutLine, _> = serde_json::from_str(trimmed);
let Ok(rollout_line) = parsed else { continue };
match rollout_line.item {
RolloutItem::SessionMeta(session_meta_line) => {
if !summary.saw_session_meta {
summary.source = Some(session_meta_line.meta.source.clone());
summary.agent_nickname = session_meta_line.meta.agent_nickname.clone();
summary.agent_role = session_meta_line.meta.agent_role.clone();
summary.model_provider = session_meta_line.meta.model_provider.clone();
summary.thread_id = Some(session_meta_line.meta.id);
summary.cwd = Some(session_meta_line.meta.cwd.clone());
summary.git_branch = session_meta_line
.git
.as_ref()
.and_then(|git| git.branch.clone());
summary.git_sha = session_meta_line
.git
.as_ref()
.and_then(|git| git.commit_hash.as_ref().map(|sha| sha.0.clone()));
summary.git_origin_url = session_meta_line
.git
.as_ref()
.and_then(|git| git.repository_url.clone());
summary.cli_version = Some(session_meta_line.meta.cli_version);
summary.created_at = Some(session_meta_line.meta.timestamp.clone());
summary.saw_session_meta = true;
match rollout_line.item {
RolloutItem::SessionMeta(session_meta_line) => {
if !summary.saw_session_meta {
summary.source = Some(session_meta_line.meta.source.clone());
summary.agent_nickname = session_meta_line.meta.agent_nickname.clone();
summary.agent_role = session_meta_line.meta.agent_role.clone();
summary.model_provider = session_meta_line.meta.model_provider.clone();
summary.thread_id = Some(session_meta_line.meta.id);
summary.cwd = Some(session_meta_line.meta.cwd.clone());
summary.git_branch = session_meta_line
.git
.as_ref()
.and_then(|git| git.branch.clone());
summary.git_sha = session_meta_line
.git
.as_ref()
.and_then(|git| git.commit_hash.as_ref().map(|sha| sha.0.clone()));
summary.git_origin_url = session_meta_line
.git
.as_ref()
.and_then(|git| git.repository_url.clone());
summary.cli_version = Some(session_meta_line.meta.cli_version);
summary.created_at = Some(session_meta_line.meta.timestamp.clone());
summary.saw_session_meta = true;
}
}
}
RolloutItem::ResponseItem(_) => {
summary.created_at = summary
.created_at
.clone()
.or_else(|| Some(rollout_line.timestamp.clone()));
}
RolloutItem::TurnContext(_) => {
// Not included in `head`; skip.
}
RolloutItem::Compacted(_) => {
// Not included in `head`; skip.
}
RolloutItem::EventMsg(ev) => {
if let EventMsg::UserMessage(user) = ev {
summary.saw_user_event = true;
if summary.first_user_message.is_none() {
let message = strip_user_message_prefix(user.message.as_str()).to_string();
if !message.is_empty() {
summary.first_user_message = Some(message);
RolloutItem::ResponseItem(_) => {
summary.created_at = summary
.created_at
.clone()
.or_else(|| Some(rollout_line.timestamp.clone()));
}
RolloutItem::TurnContext(_) => {}
RolloutItem::Compacted(_) => {}
RolloutItem::EventMsg(event) => {
if let EventMsg::UserMessage(user) = event {
summary.saw_user_event = true;
if summary.first_user_message.is_none() {
let message =
strip_user_message_prefix(user.message.as_str()).to_string();
if !message.is_empty() {
summary.first_user_message = Some(message);
}
}
}
}
}
if summary.saw_session_meta && summary.saw_user_event {
break;
}
}
if summary.saw_session_meta && summary.saw_user_event {
break;
}
}
Ok(summary)
Ok(summary)
})
.await
}
/// Read up to `HEAD_RECORD_LIMIT` records from the start of the rollout file at `path`.
/// This should be enough to produce a summary including the session meta line.
pub async fn read_head_for_summary(path: &Path) -> io::Result<Vec<serde_json::Value>> {
use tokio::io::AsyncBufReadExt;
read_rollout_head(path, |reader| {
let mut head = Vec::new();
let file = tokio::fs::File::open(path).await?;
let reader = tokio::io::BufReader::new(file);
let mut lines = reader.lines();
let mut head = Vec::new();
while head.len() < HEAD_RECORD_LIMIT {
let Some(line) = lines.next_line().await? else {
break;
};
let trimmed = line.trim();
if trimmed.is_empty() {
continue;
}
if let Ok(rollout_line) = serde_json::from_str::<RolloutLine>(trimmed) {
match rollout_line.item {
RolloutItem::SessionMeta(session_meta_line) => {
if let Ok(value) = serde_json::to_value(session_meta_line) {
head.push(value);
while head.len() < HEAD_RECORD_LIMIT {
let Some(line) = reader.next_line()? else {
break;
};
let trimmed = line.trim();
if trimmed.is_empty() {
continue;
}
if let Ok(rollout_line) = serde_json::from_str::<RolloutLine>(trimmed) {
match rollout_line.item {
RolloutItem::SessionMeta(session_meta_line) => {
if let Ok(value) = serde_json::to_value(session_meta_line) {
head.push(value);
}
}
}
RolloutItem::ResponseItem(item) => {
if let Ok(value) = serde_json::to_value(item) {
head.push(value);
RolloutItem::ResponseItem(item) => {
if let Ok(value) = serde_json::to_value(item) {
head.push(value);
}
}
RolloutItem::Compacted(_)
| RolloutItem::TurnContext(_)
| RolloutItem::EventMsg(_) => {}
}
RolloutItem::Compacted(_)
| RolloutItem::TurnContext(_)
| RolloutItem::EventMsg(_) => {}
}
}
}
Ok(head)
Ok(head)
})
.await
}
async fn read_rollout_head<T, F>(path: &Path, read: F) -> io::Result<T>
where
T: Send + 'static,
F: FnOnce(&mut RolloutLineReader) -> io::Result<T> + Send + 'static,
{
let path = path.to_path_buf();
tokio::task::spawn_blocking(move || {
let mut reader = RolloutLineReader::open(path.as_path())?;
read(&mut reader)
})
.await
.map_err(|err| io::Error::other(format!("failed to read rollout head: {err}")))?
}
fn strip_user_message_prefix(text: &str) -> &str {
codex-rs/rollout/src/metadata.rs
+4 -3
View File
@@ -1,6 +1,8 @@
use crate::ARCHIVED_SESSIONS_SUBDIR;
use crate::SESSIONS_SUBDIR;
use crate::config::RolloutConfigView;
use crate::file_io::is_rollout_file_name;
use crate::file_io::strip_rollout_file_suffix;
use crate::list;
use crate::list::parse_timestamp_uuid_from_filename;
use crate::recorder::RolloutRecorder;
@@ -30,7 +32,6 @@ use tracing::info;
use tracing::warn;
const ROLLOUT_PREFIX: &str = "rollout-";
const ROLLOUT_SUFFIX: &str = ".jsonl";
const BACKFILL_BATCH_SIZE: usize = 200;
#[cfg(not(test))]
const BACKFILL_LEASE_SECONDS: i64 = 900;
@@ -80,7 +81,7 @@ pub fn builder_from_items(
}
let file_name = rollout_path.file_name()?.to_str()?;
if !file_name.starts_with(ROLLOUT_PREFIX) || !file_name.ends_with(ROLLOUT_SUFFIX) {
if !is_rollout_file_name(file_name) {
return None;
}
let (created_ts, uuid) = parse_timestamp_uuid_from_filename(file_name)?;
@@ -430,7 +431,7 @@ async fn collect_rollout_paths(root: &Path) -> std::io::Result<Vec<PathBuf>> {
let Some(name) = file_name.to_str() else {
continue;
};
if name.starts_with(ROLLOUT_PREFIX) && name.ends_with(ROLLOUT_SUFFIX) {
if name.starts_with(ROLLOUT_PREFIX) && strip_rollout_file_suffix(name).is_some() {
paths.push(path);
}
}
codex-rs/rollout/src/recorder.rs
+158 -134
View File
@@ -1,7 +1,6 @@
//! Persist Codex session rollouts (.jsonl) so sessions can be replayed or inspected later.
//! Persist Codex session rollouts (`.jsonl` / `.jsonl.zst`) so sessions can be replayed or
//! inspected later.
use std::fs;
use std::fs::File;
use std::io::Error as IoError;
use std::path::Path;
use std::path::PathBuf;
@@ -17,7 +16,6 @@ use time::OffsetDateTime;
use time::format_description::FormatItem;
use time::format_description::well_known::Rfc3339;
use time::macros::format_description;
use tokio::io::AsyncWriteExt;
use tokio::sync::mpsc;
use tokio::sync::mpsc::Sender;
use tokio::sync::oneshot;
@@ -27,6 +25,9 @@ use tracing::warn;
use super::ARCHIVED_SESSIONS_SUBDIR;
use super::SESSIONS_SUBDIR;
use super::file_io::RolloutLineReader;
use super::file_io::append_text as append_rollout_text;
use super::file_io::preferred_rollout_file_suffix;
use super::list::Cursor;
use super::list::ThreadItem;
use super::list::ThreadListConfig;
@@ -61,7 +62,8 @@ use codex_utils_path as path_utils;
/// Records all [`ResponseItem`]s for a session and flushes them to disk after
/// every update.
///
/// Rollouts are recorded as JSONL and can be inspected with tools such as:
/// Rollouts are recorded as JSONL, optionally compressed with zstd, and can be inspected after
/// decompression with tools such as:
///
/// ```ignore
/// $ jq -C . ~/.codex/sessions/rollout-2025-05-07T17-24-21-5973b6c0-94b8-487b-a530-2aeb6098ae0e.jsonl
@@ -366,14 +368,14 @@ impl RolloutRecorder {
/// For newly created sessions, this precomputes path/metadata and defers
/// file creation/open until an explicit `persist()` call.
///
/// For resumed sessions, this immediately opens the existing rollout file.
/// For resumed sessions, this validates the existing rollout file path immediately.
pub async fn new(
config: &impl RolloutConfigView,
params: RolloutRecorderParams,
state_db_ctx: Option<StateDbHandle>,
state_builder: Option<ThreadMetadataBuilder>,
) -> std::io::Result<Self> {
let (file, deferred_log_file_info, rollout_path, meta, event_persistence_mode) =
let (writer, deferred_log_file_info, rollout_path, meta, event_persistence_mode) =
match params {
RolloutRecorderParams::Create {
conversation_id,
@@ -429,18 +431,16 @@ impl RolloutRecorder {
RolloutRecorderParams::Resume {
path,
event_persistence_mode,
} => (
Some(
tokio::fs::OpenOptions::new()
.append(true)
.open(&path)
.await?,
),
None,
path,
None,
event_persistence_mode,
),
} => {
tokio::fs::metadata(&path).await?;
(
Some(RolloutFileWriter::new(path.clone())),
None,
path,
None,
event_persistence_mode,
)
}
};
// Clone the cwd for the spawned task to collect git info asynchronously
@@ -450,11 +450,9 @@ impl RolloutRecorder {
// future will yield, which is fine – we only need to ensure we do not
// perform *blocking* I/O on the caller's thread.
let (tx, rx) = mpsc::channel::<RolloutCmd>(256);
// Spawn a Tokio task that owns the file handle and performs async
// writes. Using `tokio::fs::File` keeps everything on the async I/O
// driver instead of blocking the runtime.
// Spawn a Tokio task that owns the rollout writer.
tokio::task::spawn(rollout_writer(
file,
writer,
deferred_log_file_info,
rx,
meta,
@@ -532,65 +530,10 @@ impl RolloutRecorder {
path: &Path,
) -> std::io::Result<(Vec<RolloutItem>, Option<ThreadId>, usize)> {
trace!("Resuming rollout from {path:?}");
let text = tokio::fs::read_to_string(path).await?;
if text.trim().is_empty() {
return Err(IoError::other("empty session file"));
}
let mut items: Vec<RolloutItem> = Vec::new();
let mut thread_id: Option<ThreadId> = None;
let mut parse_errors = 0usize;
for line in text.lines() {
if line.trim().is_empty() {
continue;
}
let v: Value = match serde_json::from_str(line) {
Ok(v) => v,
Err(e) => {
warn!("failed to parse line as JSON: {line:?}, error: {e}");
parse_errors = parse_errors.saturating_add(1);
continue;
}
};
// Parse the rollout line structure
match serde_json::from_value::<RolloutLine>(v.clone()) {
Ok(rollout_line) => match rollout_line.item {
RolloutItem::SessionMeta(session_meta_line) => {
// Use the FIRST SessionMeta encountered in the file as the canonical
// thread id and main session information. Keep all items intact.
if thread_id.is_none() {
thread_id = Some(session_meta_line.meta.id);
}
items.push(RolloutItem::SessionMeta(session_meta_line));
}
RolloutItem::ResponseItem(item) => {
items.push(RolloutItem::ResponseItem(item));
}
RolloutItem::Compacted(item) => {
items.push(RolloutItem::Compacted(item));
}
RolloutItem::TurnContext(item) => {
items.push(RolloutItem::TurnContext(item));
}
RolloutItem::EventMsg(_ev) => {
items.push(RolloutItem::EventMsg(_ev));
}
},
Err(e) => {
trace!("failed to parse rollout line: {e}");
parse_errors = parse_errors.saturating_add(1);
}
}
}
tracing::debug!(
"Resumed rollout with {} items, thread ID: {:?}, parse errors: {}",
items.len(),
thread_id,
parse_errors,
);
Ok((items, thread_id, parse_errors))
let path = path.to_path_buf();
tokio::task::spawn_blocking(move || load_rollout_items_sync(path.as_path()))
.await
.map_err(|err| IoError::other(format!("failed to read rollout items: {err}")))?
}
pub async fn get_rollout_history(path: &Path) -> std::io::Result<InitialHistory> {
@@ -627,6 +570,85 @@ impl RolloutRecorder {
}
}
fn load_rollout_items_sync(
path: &Path,
) -> std::io::Result<(Vec<RolloutItem>, Option<ThreadId>, usize)> {
let mut reader = RolloutLineReader::open(path)?;
let mut items = Vec::new();
let mut thread_id = None;
let mut parse_errors = 0usize;
let mut saw_content = false;
loop {
let line = match reader.next_line() {
Ok(Some(line)) => line,
Ok(None) => break,
Err(err) => {
if !saw_content {
return Err(err);
}
warn!(
"stopping rollout resume after read error for {path:?}; preserving parsed items: {err}"
);
parse_errors = parse_errors.saturating_add(1);
break;
}
};
if line.trim().is_empty() {
continue;
}
saw_content = true;
let value: Value = match serde_json::from_str(line.as_str()) {
Ok(value) => value,
Err(err) => {
warn!("failed to parse line as JSON: {line:?}, error: {err}");
parse_errors = parse_errors.saturating_add(1);
continue;
}
};
match serde_json::from_value::<RolloutLine>(value.clone()) {
Ok(rollout_line) => match rollout_line.item {
RolloutItem::SessionMeta(session_meta_line) => {
if thread_id.is_none() {
thread_id = Some(session_meta_line.meta.id);
}
items.push(RolloutItem::SessionMeta(session_meta_line));
}
RolloutItem::ResponseItem(item) => {
items.push(RolloutItem::ResponseItem(item));
}
RolloutItem::Compacted(item) => {
items.push(RolloutItem::Compacted(item));
}
RolloutItem::TurnContext(item) => {
items.push(RolloutItem::TurnContext(item));
}
RolloutItem::EventMsg(event) => {
items.push(RolloutItem::EventMsg(event));
}
},
Err(err) => {
trace!("failed to parse rollout line: {err}");
parse_errors = parse_errors.saturating_add(1);
}
}
}
if !saw_content {
return Err(IoError::other("empty session file"));
}
tracing::debug!(
"Resumed rollout with {} items, thread ID: {:?}, parse errors: {}",
items.len(),
thread_id,
parse_errors,
);
Ok((items, thread_id, parse_errors))
}
fn truncate_fs_page(
mut page: ThreadsPage,
page_size: usize,
@@ -680,7 +702,10 @@ fn precompute_log_file_info(
.format(format)
.map_err(|e| IoError::other(format!("failed to format timestamp: {e}")))?;
let filename = format!("rollout-{date_str}-{conversation_id}.jsonl");
let filename = format!(
"rollout-{date_str}-{conversation_id}{}",
preferred_rollout_file_suffix()
);
let path = dir.join(filename);
@@ -690,24 +715,9 @@ fn precompute_log_file_info(
timestamp,
})
}
fn open_log_file(path: &Path) -> std::io::Result<File> {
let Some(parent) = path.parent() else {
return Err(IoError::other(format!(
"rollout path has no parent: {}",
path.display()
)));
};
fs::create_dir_all(parent)?;
std::fs::OpenOptions::new()
.append(true)
.create(true)
.open(path)
}
#[allow(clippy::too_many_arguments)]
async fn rollout_writer(
file: Option<tokio::fs::File>,
mut writer: Option<RolloutFileWriter>,
mut deferred_log_file_info: Option<LogFileInfo>,
mut rx: mpsc::Receiver<RolloutCmd>,
mut meta: Option<SessionMeta>,
@@ -718,7 +728,6 @@ async fn rollout_writer(
default_provider: String,
generate_memories: bool,
) -> std::io::Result<()> {
let mut writer = file.map(|file| JsonlWriter { file });
let mut buffered_items = Vec::<RolloutItem>::new();
if let Some(builder) = state_builder.as_mut() {
builder.rollout_path = rollout_path.clone();
@@ -773,10 +782,7 @@ async fn rollout_writer(
"deferred rollout recorder missing log file metadata",
));
};
let file = open_log_file(log_file_info.path.as_path())?;
writer = Some(JsonlWriter {
file: tokio::fs::File::from_std(file),
});
writer = Some(RolloutFileWriter::new(log_file_info.path));
if let Some(session_meta) = meta.take() {
write_session_meta(
@@ -817,13 +823,6 @@ async fn rollout_writer(
let _ = ack.send(());
}
RolloutCmd::Flush { ack } => {
// Deferred fresh threads may not have an initialized file yet.
if let Some(writer) = writer.as_mut()
&& let Err(e) = writer.file.flush().await
{
let _ = ack.send(());
return Err(e);
}
let _ = ack.send(());
}
RolloutCmd::Shutdown { ack } => {
@@ -837,7 +836,7 @@ async fn rollout_writer(
#[allow(clippy::too_many_arguments)]
async fn write_session_meta(
mut writer: Option<&mut JsonlWriter>,
mut writer: Option<&mut RolloutFileWriter>,
session_meta: SessionMeta,
cwd: &Path,
rollout_path: &Path,
@@ -876,7 +875,7 @@ async fn write_session_meta(
}
async fn write_and_reconcile_items(
mut writer: Option<&mut JsonlWriter>,
mut writer: Option<&mut RolloutFileWriter>,
items: &[RolloutItem],
rollout_path: &Path,
state_db_ctx: Option<&StateRuntime>,
@@ -884,9 +883,7 @@ async fn write_and_reconcile_items(
default_provider: &str,
) -> std::io::Result<()> {
if let Some(writer) = writer.as_mut() {
for item in items {
writer.write_rollout_item(item).await?;
}
writer.write_rollout_items(items).await?;
}
sync_thread_state_after_write(
state_db_ctx,
@@ -949,8 +946,8 @@ async fn sync_thread_state_after_write(
.await;
}
struct JsonlWriter {
file: tokio::fs::File,
struct RolloutFileWriter {
path: PathBuf,
}
#[derive(serde::Serialize)]
@@ -960,27 +957,54 @@ struct RolloutLineRef<'a> {
item: &'a RolloutItem,
}
impl JsonlWriter {
impl RolloutFileWriter {
fn new(path: PathBuf) -> Self {
Self { path }
}
async fn write_rollout_item(&mut self, rollout_item: &RolloutItem) -> std::io::Result<()> {
self.write_rollout_items(std::slice::from_ref(rollout_item))
.await
}
async fn write_rollout_items(&mut self, rollout_items: &[RolloutItem]) -> std::io::Result<()> {
if rollout_items.is_empty() {
return Ok(());
}
let timestamp_format: &[FormatItem] = format_description!(
"[year]-[month]-[day]T[hour]:[minute]:[second].[subsecond digits:3]Z"
);
let timestamp = OffsetDateTime::now_utc()
.format(timestamp_format)
.map_err(|e| IoError::other(format!("failed to format timestamp: {e}")))?;
let line = RolloutLineRef {
timestamp,
item: rollout_item,
};
self.write_line(&line).await
let mut batch = String::new();
for rollout_item in rollout_items {
let timestamp = OffsetDateTime::now_utc()
.format(timestamp_format)
.map_err(|e| IoError::other(format!("failed to format timestamp: {e}")))?;
let line = RolloutLineRef {
timestamp,
item: rollout_item,
};
let mut json = serde_json::to_string(&line)?;
json.push('\n');
batch.push_str(json.as_str());
}
self.write_text(batch).await
}
async fn write_line(&mut self, item: &impl serde::Serialize) -> std::io::Result<()> {
let mut json = serde_json::to_string(item)?;
json.push('\n');
self.file.write_all(json.as_bytes()).await?;
self.file.flush().await?;
Ok(())
async fn write_text(&mut self, text: String) -> std::io::Result<()> {
let path = self.path.clone();
tokio::task::spawn_blocking(move || {
let Some(parent) = path.parent() else {
return Err(IoError::other(format!(
"rollout path has no parent: {}",
path.display()
)));
};
std::fs::create_dir_all(parent)?;
append_rollout_text(path.as_path(), text.as_str())
})
.await
.map_err(|err| IoError::other(format!("failed to write rollout line: {err}")))?
}
}
codex-rs/rollout/src/recorder_tests.rs
+67 -2
View File
@@ -2,6 +2,8 @@
use super::*;
use crate::config::RolloutConfig;
use crate::file_io::append_text;
use crate::file_io::read_rollout_text;
use chrono::TimeZone;
use codex_protocol::config_types::ReasoningSummary as ReasoningSummaryConfig;
use codex_protocol::protocol::AgentMessageEvent;
@@ -124,7 +126,7 @@ async fn recorder_materializes_only_after_explicit_persist() -> std::io::Result<
recorder.persist().await?;
assert!(rollout_path.exists(), "rollout file should be materialized");
let text = std::fs::read_to_string(&rollout_path)?;
let text = read_rollout_text(rollout_path.as_path())?;
assert!(
text.contains("\"type\":\"session_meta\""),
"expected session metadata in rollout"
@@ -139,7 +141,7 @@ async fn recorder_materializes_only_after_explicit_persist() -> std::io::Result<
buffered_idx < user_idx,
"buffered items should preserve ordering"
);
let text_after_second_persist = std::fs::read_to_string(&rollout_path)?;
let text_after_second_persist = read_rollout_text(rollout_path.as_path())?;
assert_eq!(text_after_second_persist, text);
recorder.shutdown().await?;
@@ -483,3 +485,66 @@ async fn resume_candidate_matches_cwd_reads_latest_turn_context() -> std::io::Re
);
Ok(())
}
#[tokio::test]
async fn load_rollout_items_preserves_parsed_items_when_zstd_tail_is_truncated()
-> std::io::Result<()> {
let temp_dir = TempDir::new()?;
let thread_id = ThreadId::new();
let rollout_path = temp_dir
.path()
.join(format!("rollout-2026-01-01T00-00-00-{thread_id}.jsonl.zst"));
let session_meta = serde_json::json!({
"timestamp": "2026-01-01T00:00:00Z",
"type": "session_meta",
"payload": {
"id": thread_id,
"timestamp": "2026-01-01T00:00:00Z",
"cwd": ".",
"originator": "test_originator",
"cli_version": "test_version",
"source": "cli",
"model_provider": "test-provider",
},
});
let user_event = serde_json::json!({
"timestamp": "2026-01-01T00:00:01Z",
"type": "event_msg",
"payload": {
"type": "user_message",
"message": "hello",
"kind": "plain",
},
});
append_text(&rollout_path, &format!("{session_meta}\n"))?;
append_text(&rollout_path, &format!("{user_event}\n"))?;
let file = File::options().write(true).open(&rollout_path)?;
let truncated_len = file.metadata()?.len().saturating_sub(1);
file.set_len(truncated_len)?;
let (items, parsed_thread_id, parse_errors) =
RolloutRecorder::load_rollout_items(&rollout_path).await?;
assert_eq!(parsed_thread_id, Some(thread_id));
assert_eq!(items.len(), 1);
let Some(RolloutItem::SessionMeta(session_meta_line)) = items.first() else {
panic!("expected recovered session meta");
};
assert_eq!(session_meta_line.meta.id, thread_id);
assert_eq!(session_meta_line.meta.timestamp, "2026-01-01T00:00:00Z");
assert_eq!(session_meta_line.meta.cwd, PathBuf::from("."));
assert_eq!(session_meta_line.meta.originator, "test_originator");
assert_eq!(session_meta_line.meta.cli_version, "test_version");
assert_eq!(session_meta_line.meta.source, SessionSource::Cli);
assert_eq!(
session_meta_line.meta.model_provider.as_deref(),
Some("test-provider")
);
assert!(session_meta_line.git.is_none());
assert_eq!(parse_errors, 1);
Ok(())
}
codex-rs/rollout/src/tests.rs
+2 -1
View File
@@ -19,6 +19,7 @@ use time::macros::format_description;
use uuid::Uuid;
use crate::INTERACTIVE_SESSION_SOURCES;
use crate::file_io::read_rollout_text;
use crate::find_thread_path_by_id_str;
use crate::list::Cursor;
use crate::list::ThreadItem;
@@ -918,7 +919,7 @@ async fn test_get_thread_contents() {
.unwrap();
let path = &page.items[0].path;
let content = tokio::fs::read_to_string(path).await.unwrap();
let content = read_rollout_text(path).unwrap();
// Page equality (single item)
let expected_path = home