From a8a6071279b6f3112fcc5fc3fee69c48473d7149 Mon Sep 17 00:00:00 2001 From: jif-oai Date: Mon, 1 Jun 2026 15:14:19 +0200 Subject: [PATCH] Read compressed rollouts and materialize before append (#25087) ## Why Local rollout compression needs a cold `.jsonl.zst` representation without letting compressed physical paths leak into append-mode writers. The unsafe case is resume or metadata update code successfully reading a compressed rollout and then appending raw JSONL bytes to the zstd file. This PR folds the former #25088 materialization slice into the read-support PR so the reader changes and append-safety invariant land together. ## What Changed - Teach rollout readers, discovery, listing, search, and ID lookup to understand compressed `.jsonl.zst` rollouts. - Keep `.jsonl` as the logical/stored rollout path while allowing read paths to open either plain or compressed storage. - Materialize compressed rollouts back to plain `.jsonl` before append-mode writes, including resume and direct metadata append paths. - Preserve compressed-file permissions when materializing back to plain JSONL. - Refresh thread-store resolved rollout paths after compatibility metadata writes so reconciliation follows the materialized file. - Avoid treating transient compression temp files as real rollout lookup results. ## Remaining Stack #25089 remains the separate worker PR. It is based directly on this PR and stays behind the disabled `local_thread_store_compression` feature flag. The worker still has a broader coordination question: a resume or metadata update can race with background compression while a plain file is being replaced by `.jsonl.zst`. This PR handles the read and materialize-before-append primitives; it does not make the worker production-ready. ## Validation - `just test -p codex-rollout` - `just test -p codex-thread-store` - `just fix -p codex-rollout` - `just fix -p codex-thread-store` - `just bazel-lock-check` --- codex-rs/Cargo.lock | 1 + codex-rs/rollout/Cargo.toml | 3 +- codex-rs/rollout/src/compression.rs | 362 ++++++++++++++++++ codex-rs/rollout/src/compression_tests.rs | 294 ++++++++++++++ codex-rs/rollout/src/lib.rs | 5 + codex-rs/rollout/src/list.rs | 135 ++++--- codex-rs/rollout/src/metadata.rs | 20 +- codex-rs/rollout/src/recorder.rs | 45 ++- codex-rs/rollout/src/search.rs | 63 ++- codex-rs/rollout/src/state_db.rs | 7 +- codex-rs/thread-store/src/local/helpers.rs | 46 ++- .../thread-store/src/local/live_writer.rs | 4 +- .../thread-store/src/local/read_thread.rs | 31 +- .../src/local/update_thread_metadata.rs | 10 +- codex-rs/tui/src/app.rs | 21 +- codex-rs/tui/src/session_resume.rs | 8 +- 16 files changed, 924 insertions(+), 131 deletions(-) create mode 100644 codex-rs/rollout/src/compression.rs create mode 100644 codex-rs/rollout/src/compression_tests.rs diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 710a57bc1..3d55019cd 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -3591,6 +3591,7 @@ dependencies = [ "tokio", "tracing", "uuid", + "zstd 0.13.3", ] [[package]] diff --git a/codex-rs/rollout/Cargo.toml b/codex-rs/rollout/Cargo.toml index 50e5a8594..a1ccba420 100644 --- a/codex-rs/rollout/Cargo.toml +++ b/codex-rs/rollout/Cargo.toml @@ -27,6 +27,7 @@ codex-utils-string = { workspace = true } regex = { workspace = true } serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } +tempfile = { workspace = true } time = { workspace = true, features = [ "formatting", "local-offset", @@ -44,7 +45,7 @@ tokio = { workspace = true, features = [ ] } tracing = { workspace = true } uuid = { workspace = true } +zstd = { workspace = true } [dev-dependencies] pretty_assertions = { workspace = true } -tempfile = { workspace = true } diff --git a/codex-rs/rollout/src/compression.rs b/codex-rs/rollout/src/compression.rs new file mode 100644 index 000000000..2dd1ca1cf --- /dev/null +++ b/codex-rs/rollout/src/compression.rs @@ -0,0 +1,362 @@ +use std::ffi::OsStr; +use std::fs::File; +use std::fs::Permissions; +use std::io; +use std::io::Read; +use std::io::Write; +use std::path::Path; +use std::path::PathBuf; +use std::sync::atomic::AtomicU64; +use std::sync::atomic::Ordering; +use std::time::Duration; + +#[cfg(unix)] +use std::os::unix::fs::OpenOptionsExt; +#[cfg(unix)] +use std::os::unix::fs::PermissionsExt; + +const COMPRESSED_SUFFIX: &str = ".zst"; +const MAX_NOT_FOUND_RETRIES: usize = 3; +const OPEN_ROLLOUT_LINE_READER_RETRY_DELAY: Duration = Duration::from_millis(50); +const TEMP_SUFFIX: &str = ".tmp"; +static TEMP_COUNTER: AtomicU64 = AtomicU64::new(0); + +/// Returns the modified time for the existing plain or compressed rollout file. +pub(crate) async fn file_modified_time(path: &Path) -> io::Result> { + let Some(path) = path::existing_rollout_path(path).await else { + return Ok(None); + }; + let meta = tokio::fs::metadata(path).await?; + let modified = meta.modified().ok(); + Ok(modified.map(time::OffsetDateTime::from)) +} + +/// Opens a rollout line reader that transparently handles plain `.jsonl` and `.jsonl.zst` files. +/// +/// If the requested path disappears during a representation transition, this briefly retries +/// resolution so callers do not need to know which representation is on disk. +pub async fn open_rollout_line_reader(path: &Path) -> io::Result { + for _ in 0..MAX_NOT_FOUND_RETRIES { + match reader::open_once(path).await { + Ok(reader) => return Ok(reader), + Err(err) if err.kind() == io::ErrorKind::NotFound => { + tokio::time::sleep(OPEN_ROLLOUT_LINE_READER_RETRY_DELAY).await; + } + Err(err) => return Err(err), + } + } + reader::open_once(path).await +} + +/// Returns the compressed `.jsonl.zst` path for a rollout path. +#[cfg(test)] +pub(crate) fn compressed_rollout_path(path: &Path) -> PathBuf { + path::compressed_rollout_path(path) +} + +/// Materializes a compressed rollout back to plain `.jsonl` for async append paths. +pub(crate) async fn materialize_rollout_for_append(path: &Path) -> io::Result { + let path = path.to_path_buf(); + tokio::task::spawn_blocking(move || materialize_rollout_for_append_blocking(path.as_path())) + .await + .map_err(io::Error::other)? +} + +/// Materializes a compressed rollout back to plain `.jsonl` for blocking append paths. +pub(crate) fn materialize_rollout_for_append_blocking(path: &Path) -> io::Result { + let plain_path = plain_rollout_path(path); + if plain_path.exists() { + return Ok(plain_path); + } + let compressed_path = path::compressed_rollout_path(plain_path.as_path()); + if !compressed_path.exists() { + return Ok(plain_path); + } + + let temp_path = temp_path_for(plain_path.as_path(), "decompress"); + if let Some(parent) = plain_path.parent() { + std::fs::create_dir_all(parent)?; + } + let result: io::Result<()> = (|| { + let permissions = std::fs::metadata(compressed_path.as_path())?.permissions(); + { + let input = File::open(compressed_path.as_path())?; + let mut decoder = zstd::stream::read::Decoder::new(input)?; + let mut output = create_file_with_permissions(temp_path.as_path(), &permissions)?; + io::copy(&mut decoder, &mut output)?; + output.flush()?; + output.sync_all()?; + } + match std::fs::hard_link(temp_path.as_path(), plain_path.as_path()) { + Ok(()) => {} + Err(err) if err.kind() == io::ErrorKind::AlreadyExists => {} + Err(_) => persist_temp_file_noclobber(temp_path.as_path(), plain_path.as_path())?, + } + let _ = std::fs::remove_file(temp_path.as_path()); + match std::fs::remove_file(compressed_path.as_path()) { + Ok(()) => {} + Err(err) if err.kind() == io::ErrorKind::NotFound => {} + Err(err) => return Err(err), + } + Ok(()) + })(); + if result.is_err() { + let _ = std::fs::remove_file(temp_path.as_path()); + } + result?; + Ok(plain_path) +} + +fn persist_temp_file_noclobber(temp_path: &Path, destination: &Path) -> io::Result<()> { + let temp_path = tempfile::TempPath::try_from_path(temp_path)?; + match temp_path.persist_noclobber(destination) { + Ok(()) => Ok(()), + Err(err) if err.error.kind() == io::ErrorKind::AlreadyExists => Ok(()), + Err(err) => Err(err.error), + } +} + +/// Returns the plain `.jsonl` path for a plain or compressed rollout path. +pub fn plain_rollout_path(path: &Path) -> PathBuf { + path::plain_rollout_path(path) +} + +/// Parses a rollout file name, returning its plain `.jsonl` name when valid. +pub(crate) fn parse_rollout_file_name(name: &str) -> Option<&str> { + file_name::parse_rollout_file_name(name) +} + +/// A discovered rollout file, represented by exactly one physical path. +/// +/// This keeps directory walkers from reimplementing the plain/compressed +/// precedence rules. The physical path may point at either `.jsonl` or +/// `.jsonl.zst`, while `plain_file_name` is always the canonical `.jsonl` +/// filename used for timestamp and id parsing. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct RolloutFile { + path: PathBuf, + plain_file_name: String, +} + +impl RolloutFile { + /// Creates a logical rollout file from a physical path found during discovery. + /// + /// Returns `None` for non-rollout names and for compressed siblings hidden by + /// an existing plain `.jsonl` file. + pub(crate) fn from_path(path: PathBuf) -> Option { + let file_name = path.file_name().and_then(|name| name.to_str())?; + let plain_file_name = file_name::parse_rollout_file_name(file_name)?.to_string(); + if path::should_skip_compressed_sibling(path.as_path()) { + return None; + } + + Some(Self { + path, + plain_file_name, + }) + } + + /// Returns the physical path that should be opened for reads. + pub(crate) fn path(&self) -> &Path { + self.path.as_path() + } + + /// Returns the canonical `.jsonl` filename for timestamp and id parsing. + pub(crate) fn plain_file_name(&self) -> &str { + self.plain_file_name.as_str() + } + + /// Returns whether the physical path is the compressed representation. + pub(crate) fn is_compressed(&self) -> bool { + path::is_compressed_rollout_path(self.path.as_path()) + } + + /// Consumes the entry and returns the physical path that should be read. + pub(crate) fn into_path(self) -> PathBuf { + self.path + } +} + +/// Line-oriented rollout reader returned by [`open_rollout_line_reader`]. +pub struct RolloutLineReader { + inner: RolloutLineReaderInner, +} + +enum RolloutLineReaderInner { + Plain(tokio::io::Lines>), + Blocking(Option), +} + +impl RolloutLineReader { + /// Reads the next JSONL record from the rollout. + pub async fn next_line(&mut self) -> io::Result> { + match &mut self.inner { + RolloutLineReaderInner::Plain(lines) => lines.next_line().await, + RolloutLineReaderInner::Blocking(slot) => { + let Some(mut reader) = slot.take() else { + return Err(io::Error::other("compressed rollout reader is busy")); + }; + let (line, reader) = + tokio::task::spawn_blocking(move || (reader.next().transpose(), reader)) + .await + .map_err(io::Error::other)?; + *slot = Some(reader); + line + } + } + } +} + +type BlockingLineReader = std::io::Lines>>; + +/// Returns the existing rollout path, preferring the plain `.jsonl` file over +/// its `.jsonl.zst` compressed sibling. +pub async fn existing_rollout_path(path: &Path) -> Option { + path::existing_rollout_path(path).await +} + +mod path { + use std::ffi::OsStr; + use std::path::Path; + use std::path::PathBuf; + + use super::COMPRESSED_SUFFIX; + + pub(super) fn compressed_rollout_path(path: &Path) -> PathBuf { + if is_compressed_rollout_path(path) { + return path.to_path_buf(); + } + let mut file_name = path + .file_name() + .map(OsStr::to_os_string) + .unwrap_or_else(|| OsStr::new("rollout.jsonl").to_os_string()); + file_name.push(COMPRESSED_SUFFIX); + path.with_file_name(file_name) + } + + pub(super) fn plain_rollout_path(path: &Path) -> PathBuf { + let Some(file_name) = path.file_name().and_then(OsStr::to_str) else { + return path.to_path_buf(); + }; + let Some(plain_file_name) = file_name.strip_suffix(COMPRESSED_SUFFIX) else { + return path.to_path_buf(); + }; + path.with_file_name(plain_file_name) + } + + pub(super) fn is_compressed_rollout_path(path: &Path) -> bool { + path.file_name() + .and_then(OsStr::to_str) + .is_some_and(|name| name.ends_with(".jsonl.zst")) + } + + pub(super) fn should_skip_compressed_sibling(path: &Path) -> bool { + is_compressed_rollout_path(path) && plain_rollout_path(path).exists() + } + + pub(super) async fn existing_rollout_path(path: &Path) -> Option { + let plain_path = plain_rollout_path(path); + if tokio::fs::try_exists(plain_path.as_path()) + .await + .unwrap_or(false) + { + return Some(plain_path); + } + let compressed_path = compressed_rollout_path(plain_path.as_path()); + if tokio::fs::try_exists(compressed_path.as_path()) + .await + .unwrap_or(false) + { + return Some(compressed_path); + } + None + } +} + +mod file_name { + use super::COMPRESSED_SUFFIX; + + pub(super) fn parse_rollout_file_name(name: &str) -> Option<&str> { + let name = name.strip_suffix(COMPRESSED_SUFFIX).unwrap_or(name); + if name.starts_with("rollout-") && name.ends_with(".jsonl") { + Some(name) + } else { + None + } + } +} + +mod reader { + use std::fs::File; + use std::io; + use std::io::BufRead; + use std::io::Read; + use std::path::Path; + + use super::RolloutLineReader; + use super::RolloutLineReaderInner; + use super::path; + use tokio::io::AsyncBufReadExt; + + pub(super) async fn open_once(path: &Path) -> io::Result { + let path = path::existing_rollout_path(path) + .await + .unwrap_or_else(|| path.to_path_buf()); + if path::is_compressed_rollout_path(path.as_path()) { + let reader = tokio::task::spawn_blocking(move || { + let input = File::open(path.as_path())?; + let decoder = zstd::stream::read::Decoder::new(input)?; + Ok::<_, io::Error>( + io::BufReader::new(Box::new(decoder) as Box).lines(), + ) + }) + .await + .map_err(io::Error::other)??; + return Ok(RolloutLineReader { + inner: RolloutLineReaderInner::Blocking(Some(reader)), + }); + } + let file = tokio::fs::File::open(path).await?; + Ok(RolloutLineReader { + inner: RolloutLineReaderInner::Plain(tokio::io::BufReader::new(file).lines()), + }) + } +} + +#[cfg(unix)] +fn create_file_with_permissions(path: &Path, permissions: &Permissions) -> io::Result { + let file = std::fs::OpenOptions::new() + .write(true) + .create_new(true) + .mode(permissions.mode() & 0o7777) + .open(path)?; + file.set_permissions(permissions.clone())?; + Ok(file) +} + +#[cfg(not(unix))] +fn create_file_with_permissions(path: &Path, permissions: &Permissions) -> io::Result { + let file = std::fs::OpenOptions::new() + .write(true) + .create_new(true) + .open(path)?; + file.set_permissions(permissions.clone())?; + Ok(file) +} + +fn temp_path_for(path: &Path, operation: &str) -> PathBuf { + let mut file_name = path + .file_name() + .map(OsStr::to_os_string) + .unwrap_or_else(|| OsStr::new("rollout").to_os_string()); + let counter = TEMP_COUNTER.fetch_add(1, Ordering::Relaxed); + file_name.push(format!( + ".{operation}.{}.{counter}{TEMP_SUFFIX}", + std::process::id() + )); + path.with_file_name(file_name) +} + +#[cfg(test)] +#[path = "compression_tests.rs"] +mod tests; diff --git a/codex-rs/rollout/src/compression_tests.rs b/codex-rs/rollout/src/compression_tests.rs new file mode 100644 index 000000000..4777aa732 --- /dev/null +++ b/codex-rs/rollout/src/compression_tests.rs @@ -0,0 +1,294 @@ +use std::fs; +#[cfg(unix)] +use std::os::unix::fs::PermissionsExt; + +use codex_protocol::ThreadId; +use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::InitialHistory; +use codex_protocol::protocol::RolloutItem; +use codex_protocol::protocol::RolloutLine; +use codex_protocol::protocol::SessionMeta; +use codex_protocol::protocol::SessionMetaLine; +use codex_protocol::protocol::SessionSource; +use codex_protocol::protocol::UserMessageEvent; +use pretty_assertions::assert_eq; +use tempfile::TempDir; +use uuid::Uuid; + +use super::*; +use crate::RolloutConfig; +use crate::RolloutRecorder; +use crate::RolloutRecorderParams; +use crate::append_rollout_item_to_path; + +#[tokio::test] +async fn load_rollout_items_reads_compressed_rollout() -> anyhow::Result<()> { + let home = TempDir::new()?; + let uuid = Uuid::from_u128(1); + let thread_id = ThreadId::from_string(&uuid.to_string())?; + let rollout_path = rollout_path(home.path(), "2025-01-03T12-00-00", uuid); + write_rollout(&rollout_path, thread_id, "hello compressed")?; + compress_now(&rollout_path)?; + + let (items, loaded_thread_id, parse_errors) = + RolloutRecorder::load_rollout_items(&rollout_path).await?; + + assert_eq!(loaded_thread_id, Some(thread_id)); + assert_eq!(parse_errors, 0); + assert_eq!(items.len(), 2); + assert!(!rollout_path.exists()); + assert!(compressed_rollout_path(&rollout_path).exists()); + Ok(()) +} + +#[test] +fn rollout_file_from_path_normalizes_compressed_file_names() -> anyhow::Result<()> { + let home = TempDir::new()?; + let uuid = Uuid::from_u128(7); + let rollout_path = rollout_path(home.path(), "2025-01-03T12-00-00", uuid); + let compressed_path = compressed_rollout_path(&rollout_path); + + assert_eq!( + RolloutFile::from_path(compressed_path.clone()), + Some(RolloutFile { + path: compressed_path, + plain_file_name: format!("rollout-2025-01-03T12-00-00-{uuid}.jsonl"), + }) + ); + Ok(()) +} + +#[test] +fn rollout_file_from_path_hides_compressed_sibling_when_plain_exists() -> anyhow::Result<()> { + let home = TempDir::new()?; + let uuid = Uuid::from_u128(8); + let thread_id = ThreadId::from_string(&uuid.to_string())?; + let rollout_path = rollout_path(home.path(), "2025-01-03T12-00-00", uuid); + write_rollout(&rollout_path, thread_id, "plain wins")?; + + assert_eq!( + RolloutFile::from_path(compressed_rollout_path(&rollout_path)), + None + ); + Ok(()) +} + +#[tokio::test] +async fn append_rollout_item_materializes_compressed_rollout() -> anyhow::Result<()> { + let home = TempDir::new()?; + let uuid = Uuid::from_u128(2); + let thread_id = ThreadId::from_string(&uuid.to_string())?; + let rollout_path = rollout_path(home.path(), "2025-01-03T12-00-00", uuid); + write_rollout(&rollout_path, thread_id, "hello before append")?; + compress_now(&rollout_path)?; + + append_rollout_item_to_path( + &rollout_path, + &RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent { + message: "hello after append".to_string(), + ..Default::default() + })), + ) + .await?; + + assert!(rollout_path.exists()); + assert!(!compressed_rollout_path(&rollout_path).exists()); + let (items, loaded_thread_id, parse_errors) = + RolloutRecorder::load_rollout_items(&rollout_path).await?; + assert_eq!(loaded_thread_id, Some(thread_id)); + assert_eq!(parse_errors, 0); + assert_eq!(items.len(), 3); + Ok(()) +} + +#[tokio::test] +async fn resume_materializes_compressed_rollout_path() -> anyhow::Result<()> { + let home = TempDir::new()?; + let config = RolloutConfig { + codex_home: home.path().to_path_buf(), + sqlite_home: home.path().to_path_buf(), + cwd: home.path().to_path_buf(), + model_provider_id: "test-provider".to_string(), + generate_memories: true, + }; + let uuid = Uuid::from_u128(3); + let thread_id = ThreadId::from_string(&uuid.to_string())?; + let rollout_path = rollout_path(home.path(), "2025-01-03T12-00-00", uuid); + write_rollout(&rollout_path, thread_id, "hello before resume")?; + compress_now(&rollout_path)?; + let compressed_path = compressed_rollout_path(&rollout_path); + + let InitialHistory::Resumed(history) = + RolloutRecorder::get_rollout_history(compressed_path.as_path()).await? + else { + panic!("expected compressed rollout to load as resumed history"); + }; + assert_eq!(history.rollout_path, Some(rollout_path.clone())); + + let recorder = RolloutRecorder::new( + &config, + RolloutRecorderParams::resume(compressed_path.clone()), + ) + .await?; + + assert_eq!(recorder.rollout_path(), rollout_path.as_path()); + assert!(rollout_path.exists()); + assert!(!compressed_path.exists()); + recorder + .record_canonical_items(&[RolloutItem::EventMsg(EventMsg::UserMessage( + UserMessageEvent { + message: "hello after resume".to_string(), + ..Default::default() + }, + ))]) + .await?; + recorder.flush().await?; + recorder.shutdown().await?; + + let (items, loaded_thread_id, parse_errors) = + RolloutRecorder::load_rollout_items(&rollout_path).await?; + assert_eq!(loaded_thread_id, Some(thread_id)); + assert_eq!(parse_errors, 0); + assert_eq!(items.len(), 3); + Ok(()) +} + +#[cfg(unix)] +#[tokio::test] +async fn append_materialization_preserves_compressed_rollout_permissions() -> anyhow::Result<()> { + let home = TempDir::new()?; + let uuid = Uuid::from_u128(6); + let thread_id = ThreadId::from_string(&uuid.to_string())?; + let rollout_path = rollout_path(home.path(), "2025-01-03T12-00-00", uuid); + write_rollout(&rollout_path, thread_id, "restricted transcript")?; + compress_now(&rollout_path)?; + let compressed_path = compressed_rollout_path(&rollout_path); + fs::set_permissions(&compressed_path, fs::Permissions::from_mode(0o600))?; + + append_rollout_item_to_path( + &rollout_path, + &RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent { + message: "materialize restricted transcript".to_string(), + ..Default::default() + })), + ) + .await?; + + assert!(rollout_path.exists()); + assert!(!compressed_path.exists()); + assert_eq!( + fs::metadata(&rollout_path)?.permissions().mode() & 0o777, + 0o600 + ); + Ok(()) +} + +#[tokio::test] +async fn find_thread_path_by_id_handles_compressed_rollout_filenames() -> anyhow::Result<()> { + let home = TempDir::new()?; + let uuid = Uuid::from_u128(8); + let thread_id = ThreadId::from_string(&uuid.to_string())?; + let rollout_path = rollout_path(home.path(), "2025-01-03T12-00-00", uuid); + write_rollout(&rollout_path, thread_id, "compressed filename lookup")?; + compress_now(&rollout_path)?; + let compressed_path = compressed_rollout_path(&rollout_path); + + assert_eq!( + crate::find_thread_path_by_id_str( + home.path(), + &uuid.to_string(), + /*state_db_ctx*/ None + ) + .await?, + Some(compressed_path) + ); + assert_eq!( + crate::find_thread_path_by_id_str(home.path(), "not-a-uuid", /*state_db_ctx*/ None).await?, + None + ); + Ok(()) +} + +#[tokio::test] +async fn find_thread_path_by_id_ignores_compression_temp_matches() -> anyhow::Result<()> { + let home = TempDir::new()?; + let uuid = Uuid::from_u128(9); + let thread_id = ThreadId::from_string(&uuid.to_string())?; + let temp_path = rollout_path(home.path(), "2025-01-03T12-00-00", uuid).with_file_name(format!( + "rollout-2025-01-03T12-00-00-{uuid}.jsonl.zst.compress.1.0.tmp" + )); + write_rollout(&temp_path, thread_id, "temporary file should not resolve")?; + + assert_eq!( + crate::find_thread_path_by_id_str( + home.path(), + &uuid.to_string(), + /*state_db_ctx*/ None + ) + .await?, + None + ); + Ok(()) +} + +fn rollout_path(home: &std::path::Path, ts: &str, uuid: Uuid) -> std::path::PathBuf { + home.join("sessions/2025/01/03") + .join(format!("rollout-{ts}-{uuid}.jsonl")) +} + +fn write_rollout(path: &std::path::Path, thread_id: ThreadId, message: &str) -> anyhow::Result<()> { + let parent = path.parent().expect("rollout path should have parent"); + fs::create_dir_all(parent)?; + let session_meta_line = SessionMetaLine { + meta: SessionMeta { + id: thread_id, + forked_from_id: None, + timestamp: "2025-01-03T12:00:00Z".to_string(), + cwd: parent.to_path_buf(), + originator: "test".to_string(), + cli_version: "test".to_string(), + source: SessionSource::Cli, + thread_source: None, + agent_path: None, + agent_nickname: None, + agent_role: None, + model_provider: None, + base_instructions: None, + dynamic_tools: None, + memory_mode: None, + }, + git: None, + }; + let lines = [ + RolloutLine { + timestamp: "2025-01-03T12:00:00Z".to_string(), + item: RolloutItem::SessionMeta(session_meta_line), + }, + RolloutLine { + timestamp: "2025-01-03T12:00:01Z".to_string(), + item: RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent { + message: message.to_string(), + ..Default::default() + })), + }, + ]; + let jsonl = lines + .iter() + .map(serde_json::to_string) + .collect::, _>>()? + .join("\n"); + fs::write(path, format!("{jsonl}\n"))?; + Ok(()) +} + +fn compress_now(path: &std::path::Path) -> anyhow::Result<()> { + let compressed_path = compressed_rollout_path(path); + let input = fs::File::open(path)?; + let output = fs::File::create(compressed_path)?; + let mut encoder = zstd::stream::write::Encoder::new(output, 3)?; + let mut input = std::io::BufReader::new(input); + std::io::copy(&mut input, &mut encoder)?; + encoder.finish()?; + fs::remove_file(path)?; + Ok(()) +} diff --git a/codex-rs/rollout/src/lib.rs b/codex-rs/rollout/src/lib.rs index abbad1da0..e646b7da2 100644 --- a/codex-rs/rollout/src/lib.rs +++ b/codex-rs/rollout/src/lib.rs @@ -4,6 +4,7 @@ use std::sync::LazyLock; use codex_protocol::protocol::SessionSource; +pub(crate) mod compression; pub(crate) mod config; pub(crate) mod list; pub(crate) mod metadata; @@ -32,6 +33,10 @@ pub static INTERACTIVE_SESSION_SOURCES: LazyLock> = LazyLock: }); pub use codex_protocol::protocol::SessionMeta; +pub use compression::RolloutLineReader; +pub use compression::existing_rollout_path; +pub use compression::open_rollout_line_reader; +pub use compression::plain_rollout_path; pub use config::Config; pub use config::RolloutConfig; pub use config::RolloutConfigView; diff --git a/codex-rs/rollout/src/list.rs b/codex-rs/rollout/src/list.rs index 8c44c6703..fb740bb1d 100644 --- a/codex-rs/rollout/src/list.rs +++ b/codex-rs/rollout/src/list.rs @@ -18,6 +18,7 @@ use uuid::Uuid; use super::ARCHIVED_SESSIONS_SUBDIR; use super::SESSIONS_SUBDIR; +use super::compression; use crate::protocol::EventMsg; use crate::state_db; use codex_file_search as file_search; @@ -896,21 +897,18 @@ async fn collect_flat_rollout_files( { continue; } - let file_name = entry.file_name(); - let Some(name_str) = file_name.to_str() else { + let Some(rollout_file) = compression::RolloutFile::from_path(entry.path()) else { continue; }; - if !name_str.starts_with("rollout-") || !name_str.ends_with(".jsonl") { - continue; - } - let Some((ts, id)) = parse_timestamp_uuid_from_filename(name_str) else { + let Some((ts, id)) = parse_timestamp_uuid_from_filename(rollout_file.plain_file_name()) + else { continue; }; *scanned_files += 1; if *scanned_files > MAX_SCAN_FILES { break; } - collected.push((ts, id, entry.path())); + collected.push((ts, id, rollout_file.into_path())); } collected.sort_by_key(|(ts, sid, _path)| (Reverse(*ts), Reverse(*sid))); Ok(collected) @@ -919,12 +917,10 @@ async fn collect_flat_rollout_files( async fn collect_rollout_day_files( day_path: &Path, ) -> io::Result> { - let mut day_files = collect_files(day_path, |name_str, path| { - if !name_str.starts_with("rollout-") || !name_str.ends_with(".jsonl") { - return None; - } - - parse_timestamp_uuid_from_filename(name_str).map(|(ts, id)| (ts, id, path.to_path_buf())) + let mut day_files = collect_files(day_path, |_name_str, path| { + let rollout_file = compression::RolloutFile::from_path(path.to_path_buf())?; + parse_timestamp_uuid_from_filename(rollout_file.plain_file_name()) + .map(|(ts, id)| (ts, id, rollout_file.into_path())) }) .await?; // Stable ordering within the same second: (timestamp desc, uuid desc) @@ -933,7 +929,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-.jsonl + // Expected: rollout-YYYY-MM-DDThh-mm-ss-.jsonl[.zst] + let name = compression::parse_rollout_file_name(name)?; let core = name.strip_prefix("rollout-")?.strip_suffix(".jsonl")?; // Scan from the right for a '-' such that the suffix parses as a UUID. @@ -986,23 +983,22 @@ async fn collect_flat_files_by_updated_at( { continue; } - let file_name = entry.file_name(); - let Some(name_str) = file_name.to_str() else { + let Some(rollout_file) = compression::RolloutFile::from_path(entry.path()) else { continue; }; - if !name_str.starts_with("rollout-") || !name_str.ends_with(".jsonl") { - continue; - } - let Some((_ts, id)) = parse_timestamp_uuid_from_filename(name_str) else { + let Some((_ts, id)) = parse_timestamp_uuid_from_filename(rollout_file.plain_file_name()) + else { continue; }; *scanned_files += 1; if *scanned_files > MAX_SCAN_FILES { break; } - let updated_at = file_modified_time(&entry.path()).await.unwrap_or(None); + let updated_at = file_modified_time(rollout_file.path()) + .await + .unwrap_or(None); candidates.push(ThreadCandidate { - path: entry.path(), + path: rollout_file.into_path(), id, updated_at, }); @@ -1078,11 +1074,7 @@ impl<'a> ProviderMatcher<'a> { } async fn read_head_summary(path: &Path, head_limit: usize) -> io::Result { - use tokio::io::AsyncBufReadExt; - - let file = tokio::fs::File::open(path).await?; - let reader = tokio::io::BufReader::new(file); - let mut lines = reader.lines(); + let mut lines = compression::open_rollout_line_reader(path).await?; let mut summary = HeadTailSummary::default(); let mut lines_scanned = 0usize; @@ -1169,11 +1161,7 @@ async fn read_head_summary(path: &Path, head_limit: usize) -> io::Result io::Result> { - use tokio::io::AsyncBufReadExt; - - let file = tokio::fs::File::open(path).await?; - let reader = tokio::io::BufReader::new(file); - let mut lines = reader.lines(); + let mut lines = compression::open_rollout_line_reader(path).await?; let mut head = Vec::new(); while head.len() < HEAD_RECORD_LIMIT { @@ -1257,13 +1245,9 @@ pub async fn read_session_meta_line(path: &Path) -> io::Result } async fn file_modified_time(path: &Path) -> io::Result> { - let meta = tokio::fs::metadata(path).await?; - let modified = meta.modified().ok(); - let Some(modified) = modified else { - return Ok(None); - }; - let dt = OffsetDateTime::from(modified); - Ok(truncate_to_millis(dt)) + Ok(compression::file_modified_time(path) + .await? + .and_then(truncate_to_millis)) } fn format_rfc3339(dt: OffsetDateTime) -> Option { @@ -1304,16 +1288,18 @@ async fn find_thread_path_by_id_str_in_subdir( .await { Ok(Some(db_path)) => { - if tokio::fs::try_exists(&db_path).await.unwrap_or(false) { - match read_session_meta_line(&db_path).await { + if let Some(existing_db_path) = + compression::existing_rollout_path(db_path.as_path()).await + { + match read_session_meta_line(&existing_db_path).await { Ok(meta_line) if meta_line.meta.id == thread_id => { - return Ok(Some(db_path)); + return Ok(Some(existing_db_path)); } Ok(meta_line) => { tracing::error!( "state db returned rollout path for thread {id_str} but file belongs to thread {}: {}", meta_line.meta.id, - db_path.display() + existing_db_path.display() ); tracing::warn!( "state db discrepancy during find_thread_path_by_id_str_in_subdir: mismatched_db_path" @@ -1327,9 +1313,9 @@ async fn find_thread_path_by_id_str_in_subdir( Err(err) => { tracing::debug!( "state db returned rollout path for thread {id_str} that could not be verified: {}: {err}", - db_path.display() + existing_db_path.display() ); - unverified_db_path = Some(db_path); + unverified_db_path = Some(existing_db_path); } } } else { @@ -1372,10 +1358,23 @@ async fn find_thread_path_by_id_str_in_subdir( ..Default::default() }; - let results = file_search::run(id_str, vec![root], options, /*cancel_flag*/ None) - .map_err(|e| io::Error::other(format!("file search failed: {e}")))?; + let results = file_search::run( + id_str, + vec![root.clone()], + options, + /*cancel_flag*/ None, + ) + .map_err(|e| io::Error::other(format!("file search failed: {e}")))?; - let found = results.matches.into_iter().next().map(|m| m.full_path()); + let found = match results + .matches + .into_iter() + .map(|m| m.full_path()) + .find_map(compression::RolloutFile::from_path) + { + Some(rollout_file) => Some(rollout_file.into_path()), + None => find_rollout_path_by_id_from_filenames(root.as_path(), id_str).await?, + }; if let Some(found_path) = found.as_ref() { tracing::debug!("state db missing rollout path for thread {id_str}"); tracing::warn!( @@ -1400,6 +1399,46 @@ async fn find_thread_path_by_id_str_in_subdir( Ok(found.or(unverified_db_path)) } +async fn find_rollout_path_by_id_from_filenames( + root: &Path, + id_str: &str, +) -> io::Result> { + let Ok(target) = Uuid::parse_str(id_str) else { + return Ok(None); + }; + let mut stack = vec![root.to_path_buf()]; + while let Some(dir) = stack.pop() { + let mut read_dir = match tokio::fs::read_dir(dir.as_path()).await { + Ok(read_dir) => read_dir, + Err(err) if err.kind() == io::ErrorKind::NotFound => continue, + Err(err) => return Err(err), + }; + while let Some(entry) = read_dir.next_entry().await? { + let path = entry.path(); + let file_type = entry.file_type().await?; + if file_type.is_dir() { + stack.push(path); + continue; + } + if !file_type.is_file() { + continue; + } + let Some(rollout_file) = compression::RolloutFile::from_path(path) else { + continue; + }; + let Some((_ts, id)) = + parse_timestamp_uuid_from_filename(rollout_file.plain_file_name()) + else { + continue; + }; + if id == target { + return Ok(Some(rollout_file.into_path())); + } + } + } + Ok(None) +} + /// Locate a recorded thread rollout file by its UUID string using the existing /// paginated listing implementation. Returns `Ok(Some(path))` if found, `Ok(None)` if not present /// or the id is invalid. diff --git a/codex-rs/rollout/src/metadata.rs b/codex-rs/rollout/src/metadata.rs index 00c6e3e63..42410b3b7 100644 --- a/codex-rs/rollout/src/metadata.rs +++ b/codex-rs/rollout/src/metadata.rs @@ -1,5 +1,6 @@ use crate::ARCHIVED_SESSIONS_SUBDIR; use crate::SESSIONS_SUBDIR; +use crate::compression; use crate::list::parse_timestamp_uuid_from_filename; use crate::recorder::RolloutRecorder; use crate::state_db::normalize_cwd_for_state_db; @@ -27,8 +28,6 @@ use std::path::PathBuf; 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; @@ -78,9 +77,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) { - return None; - } + let file_name = compression::parse_rollout_file_name(file_name)?; let (created_ts, uuid) = parse_timestamp_uuid_from_filename(file_name)?; let created_at = DateTime::::from_timestamp(created_ts.unix_timestamp(), 0)?.with_nanosecond(0)?; @@ -364,9 +361,8 @@ fn backfill_watermark_for_path(codex_home: &Path, path: &Path) -> String { } async fn file_modified_time_utc(path: &Path) -> Option> { - let modified = tokio::fs::metadata(path).await.ok()?.modified().ok()?; - let updated_at: DateTime = modified.into(); - Some(updated_at) + let modified = compression::file_modified_time(path).await.ok()??; + DateTime::::from_timestamp(modified.unix_timestamp(), modified.nanosecond()) } fn parse_timestamp_to_utc(ts: &str) -> Option> { @@ -421,12 +417,8 @@ async fn collect_rollout_paths(root: &Path) -> std::io::Result> { if !file_type.is_file() { continue; } - let file_name = entry.file_name(); - let Some(name) = file_name.to_str() else { - continue; - }; - if name.starts_with(ROLLOUT_PREFIX) && name.ends_with(ROLLOUT_SUFFIX) { - paths.push(path); + if let Some(rollout_file) = compression::RolloutFile::from_path(path) { + paths.push(rollout_file.into_path()); } } } diff --git a/codex-rs/rollout/src/recorder.rs b/codex-rs/rollout/src/recorder.rs index d908ac270..94e3efe2e 100644 --- a/codex-rs/rollout/src/recorder.rs +++ b/codex-rs/rollout/src/recorder.rs @@ -30,6 +30,7 @@ use tracing::warn; use super::ARCHIVED_SESSIONS_SUBDIR; use super::SESSIONS_SUBDIR; +use super::compression; use super::list::Cursor; use super::list::SortDirection; use super::list::ThreadItem; @@ -699,17 +700,20 @@ impl RolloutRecorder { (None, Some(log_file_info), path, Some(session_meta)) } - RolloutRecorderParams::Resume { path } => ( - Some( - tokio::fs::OpenOptions::new() - .append(true) - .open(&path) - .await?, - ), - None, - path, - None, - ), + 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, + ) + } }; // Clone the cwd for the spawned task to collect git info asynchronously @@ -820,19 +824,17 @@ impl RolloutRecorder { path: &Path, ) -> std::io::Result<(Vec, Option, 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 = Vec::new(); let mut thread_id: Option = None; let mut parse_errors = 0usize; - for line in text.lines() { + let mut reader = compression::open_rollout_line_reader(path).await?; + let mut saw_non_empty_line = false; + while let Some(line) = reader.next_line().await? { if line.trim().is_empty() { continue; } - let mut v: Value = match serde_json::from_str(line) { + saw_non_empty_line = true; + let mut v: Value = match serde_json::from_str(&line) { Ok(v) => v, Err(e) => { warn!("failed to parse line as JSON: {line:?}, error: {e}"); @@ -875,6 +877,9 @@ impl RolloutRecorder { } } } + if !saw_non_empty_line { + return Err(IoError::other("empty session file")); + } tracing::debug!( "Resumed rollout with {} items, thread ID: {:?}, parse errors: {}", @@ -898,7 +903,7 @@ impl RolloutRecorder { Ok(InitialHistory::Resumed(ResumedHistory { conversation_id, history: items, - rollout_path: Some(path.to_path_buf()), + rollout_path: Some(compression::plain_rollout_path(path)), })) } @@ -1366,6 +1371,7 @@ fn precompute_log_file_info( } fn open_log_file(path: &Path) -> std::io::Result { + let path = compression::materialize_rollout_for_append_blocking(path)?; let Some(parent) = path.parent() else { return Err(IoError::other(format!( "rollout path has no parent: {}", @@ -1625,6 +1631,7 @@ 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 file = tokio::fs::OpenOptions::new() .append(true) .open(rollout_path) diff --git a/codex-rs/rollout/src/search.rs b/codex-rs/rollout/src/search.rs index 911e80552..227091354 100644 --- a/codex-rs/rollout/src/search.rs +++ b/codex-rs/rollout/src/search.rs @@ -11,11 +11,11 @@ use codex_protocol::protocol::RolloutLine; use codex_protocol::protocol::USER_MESSAGE_BEGIN; use regex::Regex; use regex::RegexBuilder; -use tokio::io::AsyncBufReadExt; use tokio::process::Command; use super::ARCHIVED_SESSIONS_SUBDIR; use super::SESSIONS_SUBDIR; +use super::compression; const MATCH_CONTEXT_BEFORE_CHARS: usize = 48; const MATCH_CONTEXT_AFTER_CHARS: usize = 96; @@ -32,7 +32,10 @@ pub async fn search_rollout_paths( SESSIONS_SUBDIR }); let search_term = json_escaped_search_term(search_term)?; - ripgrep_rollout_paths(rg_command, root.as_path(), search_term.as_str()).await + let mut matches = + ripgrep_rollout_paths(rg_command, root.as_path(), search_term.as_str()).await?; + matches.extend(scan_compressed_rollout_paths(root.as_path(), search_term.as_str()).await?); + Ok(matches) } async fn ripgrep_rollout_paths( @@ -106,13 +109,14 @@ async fn scan_rollout_paths(root: &Path, search_term: &str) -> io::Result io::Result io::Result { - let file = tokio::fs::File::open(path).await?; - let mut lines = tokio::io::BufReader::new(file).lines(); + let mut lines = compression::open_rollout_line_reader(path).await?; while let Some(line) = lines.next_line().await? { if search_term.is_match(line.as_str()) { return Ok(true); @@ -135,8 +138,7 @@ pub async fn first_rollout_content_match_snippet( path: &Path, search_term: &str, ) -> io::Result> { - let file = tokio::fs::File::open(path).await?; - let mut lines = tokio::io::BufReader::new(file).lines(); + let mut lines = compression::open_rollout_line_reader(path).await?; let json_search_term = case_insensitive_literal_regex(json_escaped_search_term(search_term)?)?; let search_term = case_insensitive_literal_regex(search_term)?; while let Some(line) = lines.next_line().await? { @@ -149,6 +151,45 @@ pub async fn first_rollout_content_match_snippet( Ok(None) } +async fn scan_compressed_rollout_paths( + root: &Path, + search_term: &str, +) -> io::Result> { + let mut matches = HashSet::new(); + let mut dirs = vec![root.to_path_buf()]; + let search_term = case_insensitive_literal_regex(search_term)?; + + while let Some(dir) = dirs.pop() { + let mut entries = match tokio::fs::read_dir(dir).await { + Ok(entries) => entries, + Err(err) if err.kind() == io::ErrorKind::NotFound => continue, + Err(err) => return Err(err), + }; + while let Some(entry) = entries.next_entry().await? { + let path = entry.path(); + let file_type = entry.file_type().await?; + if file_type.is_dir() { + dirs.push(path); + continue; + } + if !file_type.is_file() { + continue; + } + let Some(rollout_file) = compression::RolloutFile::from_path(path) else { + continue; + }; + if !rollout_file.is_compressed() { + continue; + } + if rollout_contains(rollout_file.path(), &search_term).await? { + matches.insert(rollout_file.into_path()); + } + } + } + + Ok(matches) +} + fn json_escaped_search_term(search_term: &str) -> io::Result { let serialized = serde_json::to_string(search_term).map_err(io::Error::other)?; Ok(serialized[1..serialized.len() - 1].to_string()) diff --git a/codex-rs/rollout/src/state_db.rs b/codex-rs/rollout/src/state_db.rs index 2b6153484..48210b095 100644 --- a/codex-rs/rollout/src/state_db.rs +++ b/codex-rs/rollout/src/state_db.rs @@ -412,10 +412,11 @@ pub async fn list_threads_db( Ok(mut page) => { let mut valid_items = Vec::with_capacity(page.items.len()); for item in page.items { - if tokio::fs::try_exists(&item.rollout_path) - .await - .unwrap_or(false) + if let Some(existing_path) = + crate::compression::existing_rollout_path(item.rollout_path.as_path()).await { + let mut item = item; + item.rollout_path = existing_path; valid_items.push(item); } else { warn!( diff --git a/codex-rs/thread-store/src/local/helpers.rs b/codex-rs/thread-store/src/local/helpers.rs index 8981187fe..ae0650855 100644 --- a/codex-rs/thread-store/src/local/helpers.rs +++ b/codex-rs/thread-store/src/local/helpers.rs @@ -74,10 +74,11 @@ pub(super) fn matching_rollout_file_name( ), }); }; - let required_suffix = format!("{thread_id}.jsonl"); - if file_name - .to_string_lossy() - .ends_with(required_suffix.as_str()) + let required_plain_suffix = format!("{thread_id}.jsonl"); + let required_compressed_suffix = format!("{required_plain_suffix}.zst"); + let file_name_str = file_name.to_string_lossy(); + if file_name_str.ends_with(required_plain_suffix.as_str()) + || file_name_str.ends_with(required_compressed_suffix.as_str()) { Ok(file_name) } else { @@ -117,10 +118,11 @@ pub(super) fn stored_thread_from_rollout_item( .clone() .or_else(|| item.first_user_message.clone()) .unwrap_or_default(); + let rollout_path = codex_rollout::plain_rollout_path(item.path.as_path()); Some(StoredThread { thread_id, - rollout_path: Some(item.path), + rollout_path: Some(rollout_path), forked_from_id: None, parent_thread_id: item.parent_thread_id, preview, @@ -224,6 +226,7 @@ pub(super) fn git_info_from_parts( fn thread_id_from_rollout_path(path: &Path) -> Option { let file_name = path.file_name()?.to_str()?; + let file_name = file_name.strip_suffix(".zst").unwrap_or(file_name); let stem = file_name.strip_suffix(".jsonl")?; if stem.len() < 37 { return None; @@ -234,3 +237,36 @@ fn thread_id_from_rollout_path(path: &Path) -> Option { } ThreadId::from_string(&stem[uuid_start..]).ok() } + +#[cfg(test)] +mod tests { + use codex_rollout::ThreadItem; + use pretty_assertions::assert_eq; + use uuid::Uuid; + + use super::*; + + #[test] + fn stored_thread_from_rollout_item_returns_logical_rollout_path() { + let uuid = Uuid::from_u128(1); + let compressed_path = PathBuf::from(format!( + "/tmp/sessions/2025/01/03/rollout-2025-01-03T12-00-00-{uuid}.jsonl.zst" + )); + let thread = stored_thread_from_rollout_item( + ThreadItem { + path: compressed_path.clone(), + ..Default::default() + }, + /*archived*/ false, + "test-provider", + ) + .expect("stored thread"); + + assert_eq!( + thread.rollout_path, + Some( + compressed_path.with_file_name(format!("rollout-2025-01-03T12-00-00-{uuid}.jsonl")) + ) + ); + } +} diff --git a/codex-rs/thread-store/src/local/live_writer.rs b/codex-rs/thread-store/src/local/live_writer.rs index 9db49af76..0b31ae0ea 100644 --- a/codex-rs/thread-store/src/local/live_writer.rs +++ b/codex-rs/thread-store/src/local/live_writer.rs @@ -156,9 +156,9 @@ async fn sync_materialized_rollout_path( thread_id: ThreadId, ) -> ThreadStoreResult<()> { let rollout_path = rollout_path(store, thread_id).await?; - if !tokio::fs::try_exists(rollout_path.as_path()) + if codex_rollout::existing_rollout_path(rollout_path.as_path()) .await - .unwrap_or(false) + .is_none() { return Ok(()); } diff --git a/codex-rs/thread-store/src/local/read_thread.rs b/codex-rs/thread-store/src/local/read_thread.rs index b2a15688c..e960bbd62 100644 --- a/codex-rs/thread-store/src/local/read_thread.rs +++ b/codex-rs/thread-store/src/local/read_thread.rs @@ -90,7 +90,7 @@ async fn sqlite_rollout_path_can_load_history_for_thread( path: &std::path::Path, thread_id: codex_protocol::ThreadId, ) -> bool { - if !tokio::fs::try_exists(path).await.unwrap_or(false) { + if codex_rollout::existing_rollout_path(path).await.is_none() { return false; } // SQLite metadata can outlive a moved/recreated rollout path. When history is @@ -107,7 +107,7 @@ pub(super) async fn read_thread_by_rollout_path( include_archived: bool, include_history: bool, ) -> ThreadStoreResult { - let path = resolve_requested_rollout_path(store, rollout_path)?; + let path = resolve_requested_rollout_path(store, rollout_path).await?; let mut thread = read_thread_from_rollout_path(store, path).await?; if !include_archived && thread.archived_at.is_some() { return Err(ThreadStoreError::InvalidRequest { @@ -134,7 +134,7 @@ pub(super) async fn read_thread_by_rollout_path( Ok(thread) } -fn resolve_requested_rollout_path( +async fn resolve_requested_rollout_path( store: &LocalThreadStore, rollout_path: std::path::PathBuf, ) -> ThreadStoreResult { @@ -143,7 +143,15 @@ fn resolve_requested_rollout_path( } else { rollout_path }; - std::fs::canonicalize(&path).map_err(|err| ThreadStoreError::InvalidRequest { + let Some(path) = codex_rollout::existing_rollout_path(path.as_path()).await else { + return Err(ThreadStoreError::InvalidRequest { + message: format!( + "failed to resolve rollout path `{}`: file does not exist", + path.display() + ), + }); + }; + std::fs::canonicalize(path.as_path()).map_err(|err| ThreadStoreError::InvalidRequest { message: format!("failed to resolve rollout path `{}`: {err}", path.display()), }) } @@ -172,11 +180,9 @@ async fn resolve_rollout_path( include_archived: bool, ) -> ThreadStoreResult> { if let Ok(path) = live_writer::rollout_path(store, thread_id).await - && tokio::fs::try_exists(path.as_path()).await.map_err(|err| { - ThreadStoreError::InvalidRequest { - message: format!("failed to check rollout path for thread id {thread_id}: {err}"), - } - })? + && codex_rollout::existing_rollout_path(path.as_path()) + .await + .is_some() && (include_archived || !rollout_path_is_archived(store.config.codex_home.as_path(), &path)) { return Ok(Some(path)); @@ -233,6 +239,7 @@ async fn read_thread_from_rollout_path( .ok_or_else(|| ThreadStoreError::Internal { message: format!("failed to read thread id from {}", path.display()), })?; + thread.rollout_path = Some(codex_rollout::plain_rollout_path(path.as_path())); if let Ok(meta_line) = read_session_meta_line(path.as_path()).await { thread.forked_from_id = meta_line.meta.forked_from_id; thread.parent_thread_id = meta_line.meta.parent_thread_id; @@ -287,6 +294,7 @@ async fn stored_thread_from_sqlite_metadata( .await .ok() .map(|meta_line| meta_line.meta); + let rollout_path = codex_rollout::plain_rollout_path(metadata.rollout_path.as_path()); let forked_from_id = session_meta.as_ref().and_then(|meta| meta.forked_from_id); let parent_thread_id = session_meta.as_ref().and_then(|meta| meta.parent_thread_id); let preview = metadata @@ -298,7 +306,7 @@ async fn stored_thread_from_sqlite_metadata( permission_profile_from_metadata_value(&metadata.sandbox_policy, metadata.cwd.as_path()); StoredThread { thread_id: metadata.id, - rollout_path: Some(metadata.rollout_path), + rollout_path: Some(rollout_path), forked_from_id, parent_thread_id, preview, @@ -360,9 +368,10 @@ fn stored_thread_from_meta_line( .and_then(|meta| meta.modified().ok()) .map(DateTime::::from) .unwrap_or(created_at); + let rollout_path = codex_rollout::plain_rollout_path(path.as_path()); StoredThread { thread_id: meta_line.meta.id, - rollout_path: Some(path), + rollout_path: Some(rollout_path), forked_from_id: meta_line.meta.forked_from_id, parent_thread_id: meta_line.meta.parent_thread_id, preview: String::new(), diff --git a/codex-rs/thread-store/src/local/update_thread_metadata.rs b/codex-rs/thread-store/src/local/update_thread_metadata.rs index 3dd0b39a3..e30c9d3b1 100644 --- a/codex-rs/thread-store/src/local/update_thread_metadata.rs +++ b/codex-rs/thread-store/src/local/update_thread_metadata.rs @@ -69,13 +69,14 @@ pub(super) async fn update_thread_metadata( if live_writer::rollout_path(store, thread_id).await.is_ok() { live_writer::persist_thread(store, thread_id).await?; } - let resolved_rollout_path = + let mut resolved_rollout_path = resolve_rollout_path(store, thread_id, params.include_archived).await?; let name = patch.name; let git_info = patch.git_info; if let Some(memory_mode) = patch.memory_mode { apply_thread_memory_mode(resolved_rollout_path.path.as_path(), thread_id, memory_mode) .await?; + refresh_resolved_rollout_path(&mut resolved_rollout_path).await; } let state_db_ctx = store.state_db().await; @@ -143,6 +144,7 @@ pub(super) async fn update_thread_metadata( memory_mode.as_deref(), ) .await?; + refresh_resolved_rollout_path(&mut resolved_rollout_path).await; apply_thread_git_info(store, thread_id, sha, branch, origin_url).await?; } @@ -173,6 +175,12 @@ pub(super) async fn update_thread_metadata( Ok(thread) } +async fn refresh_resolved_rollout_path(resolved: &mut ResolvedRolloutPath) { + if let Some(path) = codex_rollout::existing_rollout_path(resolved.path.as_path()).await { + resolved.path = path; + } +} + async fn apply_metadata_update( store: &LocalThreadStore, thread_id: ThreadId, diff --git a/codex-rs/tui/src/app.rs b/codex-rs/tui/src/app.rs index e31b18044..4a8ec3b90 100644 --- a/codex-rs/tui/src/app.rs +++ b/codex-rs/tui/src/app.rs @@ -1105,16 +1105,15 @@ See the Codex keymap documentation for supported actions and examples." #[cfg(not(debug_assertions))] let pre_loop_exit_reason = if let Some(latest_version) = upgrade_version { - let control = app - .handle_event( - tui, - &mut app_server, - AppEvent::InsertHistoryCell(Box::new(UpdateAvailableHistoryCell::new( - latest_version, - crate::update_action::get_update_action(), - ))), - ) - .await?; + let control = Box::pin(app.handle_event( + tui, + &mut app_server, + AppEvent::InsertHistoryCell(Box::new(UpdateAvailableHistoryCell::new( + latest_version, + crate::update_action::get_update_action(), + ))), + )) + .await?; match control { AppRunControl::Continue => None, AppRunControl::Exit(exit_reason) => Some(exit_reason), @@ -1131,7 +1130,7 @@ See the Codex keymap documentation for supported actions and examples." loop { let control = select! { Some(event) = app_event_rx.recv() => { - match app.handle_event(tui, &mut app_server, event).await { + match Box::pin(app.handle_event(tui, &mut app_server, event)).await { Ok(control) => control, Err(err) => break Err(err), } diff --git a/codex-rs/tui/src/session_resume.rs b/codex-rs/tui/src/session_resume.rs index 9b47599ed..56b258704 100644 --- a/codex-rs/tui/src/session_resume.rs +++ b/codex-rs/tui/src/session_resume.rs @@ -14,11 +14,11 @@ use crate::cwd_prompt::CwdPromptOutcome; use crate::cwd_prompt::CwdSelection; use crate::tui::Tui; use codex_protocol::ThreadId; +use codex_rollout::open_rollout_line_reader; use codex_state::StateRuntime; use codex_utils_path as path_utils; use serde::Deserialize; use serde_json::Value; -use tokio::io::AsyncBufReadExt; #[derive(Default)] struct RolloutResumeState { @@ -142,13 +142,11 @@ pub(crate) fn cwds_differ(current_cwd: &Path, session_cwd: &Path) -> bool { } async fn read_rollout_resume_state(path: &Path) -> io::Result { - let file = tokio::fs::File::open(path).await?; - let reader = tokio::io::BufReader::new(file); - let mut lines = reader.lines(); + let mut reader = open_rollout_line_reader(path).await?; let mut state = RolloutResumeState::default(); let mut saw_record = false; - while let Some(line) = lines.next_line().await? { + while let Some(line) = reader.next_line().await? { let trimmed = line.trim(); if trimmed.is_empty() { continue;