From 01cb97851b1634d33b9bbb8e184111611c39b7b5 Mon Sep 17 00:00:00 2001 From: jif-oai Date: Mon, 1 Jun 2026 18:35:58 +0200 Subject: [PATCH] Compress cold local rollouts (#25089) ## Rollout compression stack This stack splits #24941 into reviewable steps for local rollout compression. The design is intentionally staged: 1. Teach readers, listing, search, and lookup to understand compressed rollouts. 2. Make append and resume paths materialize compressed rollouts back to plain JSONL before writing. 3. Add a disabled-by-default worker that can compress cold archived rollouts behind `local_thread_store_compression`. The key invariant is that writers append to plain `.jsonl`. A `.jsonl.zst` file is a cold/read representation; if a write is needed, the compressed file is materialized back to plain JSONL first. Readers prefer plain `.jsonl` when both forms exist and can fall back to the compressed sibling during transitions. The worker is deliberately the last PR and remains behind an under-development feature flag. It currently scans only `archived_sessions`, not active `sessions`, because active sessions have the highest resume/append race risk. That means this stack does not yet compress most unarchived local history. ## Known race / follow-up The remaining unresolved design question is writer/compressor coordination. Even for archived rollouts, a resume or metadata update can append while the worker is replacing the plain file with `.jsonl.zst`; the current double-stat checks narrow but do not fully eliminate the window where a writer has opened the plain file before unlink. Do not treat the worker PR as production-ready until we either: - prevent append/resume paths from racing archived compression, or - introduce a shared representation/append lock or equivalent coordination. The first two PRs are useful independently: they make compressed rollouts readable and make append paths safely recover back to plain JSONL. The third PR isolates the worker behavior so that coordination issue is reviewable separately. ## Validation Focused local validation for the stack includes: - `just test -p codex-rollout` - `just test -p codex-thread-store` where thread-store paths were touched - `just test -p codex-features` for the feature flag slice - `just bazel-lock-check` after dependency graph changes - scoped `just fix -p ...` passes for changed crates CI is still the source of truth for the full platform matrix. ## This PR in the stack This is PR 3/3, based on #25088. It adds the under-development feature flag and starts the best-effort background worker when enabled. The worker currently compresses only cold archived rollouts, skips active sessions, verifies compressed output, preserves mtime and permissions, keeps a store-level lock heartbeat, and cleans stale temp files. Stack order: 1. #25087: read compressed local rollouts. 2. #25088: materialize compressed rollouts before append. 3. This PR: add the disabled local compression worker. --- codex-rs/core/config.schema.json | 6 + codex-rs/core/src/thread_manager.rs | 17 +- codex-rs/features/src/lib.rs | 8 + codex-rs/rollout/src/compression.rs | 396 ++++++++++++++++++++++ codex-rs/rollout/src/compression_tests.rs | 175 ++++++++++ codex-rs/rollout/src/lib.rs | 1 + 6 files changed, 599 insertions(+), 4 deletions(-) diff --git a/codex-rs/core/config.schema.json b/codex-rs/core/config.schema.json index 77d2c97d6..43313fbb0 100644 --- a/codex-rs/core/config.schema.json +++ b/codex-rs/core/config.schema.json @@ -485,6 +485,9 @@ "js_repl_tools_only": { "type": "boolean" }, + "local_thread_store_compression": { + "type": "boolean" + }, "memories": { "type": "boolean" }, @@ -4602,6 +4605,9 @@ "js_repl_tools_only": { "type": "boolean" }, + "local_thread_store_compression": { + "type": "boolean" + }, "memories": { "type": "boolean" }, diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index 98a3d01a7..9f890ae0e 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -22,6 +22,7 @@ use codex_core_plugins::PluginsManager; use codex_exec_server::EnvironmentManager; use codex_extension_api::ExtensionRegistry; use codex_extension_api::empty_extension_registry; +use codex_features::Feature; use codex_login::AuthManager; use codex_login::CodexAuth; use codex_model_provider::create_model_provider; @@ -231,10 +232,18 @@ pub fn thread_store_from_config( state_db: Option, ) -> Arc { match &config.experimental_thread_store { - ThreadStoreConfig::Local => Arc::new(LocalThreadStore::new( - LocalThreadStoreConfig::from_config(config), - state_db, - )), + ThreadStoreConfig::Local => { + if config + .features + .enabled(Feature::LocalThreadStoreCompression) + { + codex_rollout::spawn_rollout_compression_worker(config.codex_home.to_path_buf()); + } + Arc::new(LocalThreadStore::new( + LocalThreadStoreConfig::from_config(config), + state_db, + )) + } ThreadStoreConfig::InMemory { id } => InMemoryThreadStore::for_id(id), } } diff --git a/codex-rs/features/src/lib.rs b/codex-rs/features/src/lib.rs index 116db0a5b..e2ee60bef 100644 --- a/codex-rs/features/src/lib.rs +++ b/codex-rs/features/src/lib.rs @@ -114,6 +114,8 @@ pub enum Feature { RuntimeMetrics, /// Enable startup memory extraction and file-backed memory consolidation. MemoryTool, + /// Compress cold local thread-store rollout files. + LocalThreadStoreCompression, /// Enable the Chronicle sidecar for passive screen-context memories. Chronicle, /// Append additional AGENTS.md guidance to user instructions. @@ -831,6 +833,12 @@ pub const FEATURES: &[FeatureSpec] = &[ }, default_enabled: false, }, + FeatureSpec { + id: Feature::LocalThreadStoreCompression, + key: "local_thread_store_compression", + stage: Stage::UnderDevelopment, + default_enabled: false, + }, FeatureSpec { id: Feature::Chronicle, key: "chronicle", diff --git a/codex-rs/rollout/src/compression.rs b/codex-rs/rollout/src/compression.rs index 2dd1ca1cf..a2a29ab59 100644 --- a/codex-rs/rollout/src/compression.rs +++ b/codex-rs/rollout/src/compression.rs @@ -21,6 +21,15 @@ const OPEN_ROLLOUT_LINE_READER_RETRY_DELAY: Duration = Duration::from_millis(50) const TEMP_SUFFIX: &str = ".tmp"; static TEMP_COUNTER: AtomicU64 = AtomicU64::new(0); +/// Starts a best-effort background job that compresses cold local rollout files. +/// +/// The worker is fire-and-forget: failures are logged, startup is not blocked, +/// and a process-wide lock under `codex_home` prevents overlapping compression +/// runs from the same local store. +pub fn spawn_rollout_compression_worker(codex_home: PathBuf) { + worker::spawn(codex_home) +} + /// 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 { @@ -209,6 +218,393 @@ impl RolloutLineReader { type BlockingLineReader = std::io::Lines>>; +mod worker { + use std::ffi::OsStr; + use std::fs::File; + use std::fs::FileTimes; + use std::fs::Permissions; + use std::io; + use std::io::Write; + use std::path::Path; + use std::path::PathBuf; + use std::time::Duration; + use std::time::Instant; + use std::time::SystemTime; + + use tracing::debug; + use tracing::info; + use tracing::warn; + + use crate::ARCHIVED_SESSIONS_SUBDIR; + use crate::SESSIONS_SUBDIR; + + use super::RolloutFile; + use super::path; + + const TEMP_SUFFIX: &str = ".tmp"; + const COMPRESSION_LEVEL: i32 = 3; + const MIN_ROLLOUT_AGE: Duration = Duration::from_secs(7 * 24 * 60 * 60); + const GLOBAL_LOCK_STALE_AFTER: Duration = Duration::from_secs(6 * 60 * 60); + const TEMP_FILE_STALE_AFTER: Duration = GLOBAL_LOCK_STALE_AFTER; + const WORKER_MAX_RUNTIME: Duration = Duration::from_secs(5 * 60 * 60); + const LOCK_FILE_NAME: &str = "rollout-compression.lock"; + + #[derive(Default)] + struct CompressionStats { + scanned: usize, + compressed: usize, + skipped: usize, + failed: usize, + } + + struct CompressionLock { + path: PathBuf, + } + + impl CompressionLock { + fn try_acquire(codex_home: &Path) -> io::Result> { + let lock_dir = codex_home.join(".tmp"); + std::fs::create_dir_all(lock_dir.as_path())?; + let path = lock_dir.join(LOCK_FILE_NAME); + match create_lock_file(path.as_path()) { + Ok(()) => return Ok(Some(Self { path })), + Err(err) if err.kind() == io::ErrorKind::AlreadyExists => {} + Err(err) => return Err(err), + } + + let stale = std::fs::metadata(path.as_path()) + .and_then(|metadata| metadata.modified()) + .ok() + .and_then(|modified| SystemTime::now().duration_since(modified).ok()) + .is_some_and(|age| age >= GLOBAL_LOCK_STALE_AFTER); + if !stale { + return Ok(None); + } + match std::fs::remove_file(path.as_path()) { + Ok(()) => {} + Err(err) if err.kind() == io::ErrorKind::NotFound => {} + Err(err) => return Err(err), + } + match create_lock_file(path.as_path()) { + Ok(()) => Ok(Some(Self { path })), + Err(err) if err.kind() == io::ErrorKind::AlreadyExists => Ok(None), + Err(err) => Err(err), + } + } + } + + impl Drop for CompressionLock { + fn drop(&mut self) { + let _ = std::fs::remove_file(self.path.as_path()); + } + } + + pub(super) fn spawn(codex_home: PathBuf) { + let Ok(handle) = tokio::runtime::Handle::try_current() else { + warn!( + "failed to start rollout compression worker for {}: no Tokio runtime", + codex_home.display() + ); + return; + }; + handle.spawn(async move { + if let Err(err) = run(codex_home.clone()).await { + warn!( + "rollout compression worker failed for {}: {err}", + codex_home.display() + ); + } + }); + } + + pub(super) async fn run(codex_home: PathBuf) -> io::Result<()> { + let Some(_lock) = CompressionLock::try_acquire(codex_home.as_path())? else { + debug!( + "rollout compression worker already running for {}", + codex_home.display() + ); + return Ok(()); + }; + + let started_at = Instant::now(); + cleanup_stale_temps(codex_home.as_path()).await?; + let mut stats = CompressionStats::default(); + if started_at.elapsed() < WORKER_MAX_RUNTIME { + let archived_root = codex_home.join(ARCHIVED_SESSIONS_SUBDIR); + compress_rollouts_in_root(archived_root.as_path(), started_at, &mut stats).await?; + } + info!( + "rollout compression worker finished: scanned={}, compressed={}, skipped={}, failed={}", + stats.scanned, stats.compressed, stats.skipped, stats.failed + ); + Ok(()) + } + + fn create_lock_file(path: &Path) -> io::Result<()> { + let mut file = std::fs::OpenOptions::new() + .write(true) + .create_new(true) + .open(path)?; + writeln!( + file, + "pid={} started_at={:?}", + std::process::id(), + SystemTime::now() + )?; + Ok(()) + } + + async fn compress_rollouts_in_root( + root: &Path, + started_at: Instant, + stats: &mut CompressionStats, + ) -> io::Result<()> { + if !tokio::fs::try_exists(root).await.unwrap_or(false) { + return Ok(()); + } + let mut stack = vec![root.to_path_buf()]; + while let Some(dir) = stack.pop() { + if started_at.elapsed() >= WORKER_MAX_RUNTIME { + break; + } + let mut read_dir = match tokio::fs::read_dir(dir.as_path()).await { + Ok(read_dir) => read_dir, + Err(err) => { + warn!( + "failed to read rollout compression directory {}: {err}", + dir.display() + ); + continue; + } + }; + while let Some(entry) = read_dir.next_entry().await? { + if started_at.elapsed() >= WORKER_MAX_RUNTIME { + break; + } + let path = entry.path(); + let file_type = match entry.file_type().await { + Ok(file_type) => file_type, + Err(err) => { + warn!( + "failed to read rollout compression file type {}: {err}", + path.display() + ); + continue; + } + }; + if file_type.is_dir() { + stack.push(path); + continue; + } + if !file_type.is_file() { + continue; + } + let Some(rollout_file) = RolloutFile::from_path(path) else { + continue; + }; + if rollout_file.is_compressed() { + continue; + } + let path = rollout_file.into_path(); + stats.scanned = stats.scanned.saturating_add(1); + match compress_rollout_if_cold(path.as_path()).await { + Ok(true) => stats.compressed = stats.compressed.saturating_add(1), + Ok(false) => stats.skipped = stats.skipped.saturating_add(1), + Err(err) => { + stats.failed = stats.failed.saturating_add(1); + warn!("failed to compress rollout {}: {err}", path.display()); + } + } + } + } + Ok(()) + } + + async fn compress_rollout_if_cold(path: &Path) -> io::Result { + let path = path.to_path_buf(); + tokio::task::spawn_blocking(move || compress_rollout_if_cold_blocking(path.as_path())) + .await + .map_err(io::Error::other)? + } + + fn compress_rollout_if_cold_blocking(path: &Path) -> io::Result { + let before = match cold_file_state(path)? { + Some(state) => state, + None => return Ok(false), + }; + let compressed_path = path::compressed_rollout_path(path); + if compressed_path.exists() { + return Ok(false); + } + + let temp_dir = compressed_path + .parent() + .filter(|parent| !parent.as_os_str().is_empty()) + .unwrap_or_else(|| Path::new(".")); + std::fs::create_dir_all(temp_dir)?; + let mut temp_file = tempfile::Builder::new() + .prefix("rollout-compress-") + .suffix(TEMP_SUFFIX) + .tempfile_in(temp_dir)?; + encode_zstd_to_writer(path, temp_file.as_file_mut())?; + temp_file.as_file_mut().flush()?; + verify_zstd(temp_file.path())?; + if !same_file_state(path, &before)? { + return Ok(false); + } + set_file_metadata(temp_file.as_file(), before.modified, &before.permissions)?; + temp_file.as_file().sync_all()?; + + match temp_file.persist_noclobber(compressed_path.as_path()) { + Ok(_) => {} + Err(err) if err.error.kind() == io::ErrorKind::AlreadyExists => return Ok(false), + Err(err) => return Err(err.error), + } + if !same_file_state(path, &before)? { + let _ = std::fs::remove_file(compressed_path.as_path()); + return Ok(false); + } + std::fs::remove_file(path)?; + Ok(true) + } + + struct FileState { + len: u64, + modified: SystemTime, + permissions: Permissions, + } + + fn cold_file_state(path: &Path) -> io::Result> { + let metadata = match std::fs::metadata(path) { + Ok(metadata) => metadata, + Err(err) if err.kind() == io::ErrorKind::NotFound => return Ok(None), + Err(err) => return Err(err), + }; + if !metadata.is_file() { + return Ok(None); + } + let modified = metadata.modified()?; + let age = SystemTime::now() + .duration_since(modified) + .unwrap_or(Duration::ZERO); + if age < MIN_ROLLOUT_AGE { + return Ok(None); + } + Ok(Some(FileState { + len: metadata.len(), + modified, + permissions: metadata.permissions(), + })) + } + + fn same_file_state(path: &Path, expected: &FileState) -> io::Result { + match std::fs::metadata(path) { + Ok(metadata) => Ok(metadata.len() == expected.len + && metadata.modified()? == expected.modified + && metadata.permissions() == expected.permissions), + Err(err) if err.kind() == io::ErrorKind::NotFound => Ok(false), + Err(err) => Err(err), + } + } + + fn encode_zstd_to_writer(source: &Path, output: impl Write) -> io::Result<()> { + let mut input = File::open(source)?; + let mut encoder = zstd::stream::write::Encoder::new(output, COMPRESSION_LEVEL)?; + io::copy(&mut input, &mut encoder)?; + encoder.finish()?; + Ok(()) + } + + fn verify_zstd(path: &Path) -> io::Result<()> { + let input = File::open(path)?; + let mut decoder = zstd::stream::read::Decoder::new(input)?; + let mut sink = io::sink(); + io::copy(&mut decoder, &mut sink)?; + Ok(()) + } + + fn set_file_metadata( + file: &File, + modified: SystemTime, + permissions: &Permissions, + ) -> io::Result<()> { + file.set_times(FileTimes::new().set_modified(modified))?; + file.set_permissions(permissions.clone()) + } + + async fn cleanup_stale_temps(codex_home: &Path) -> io::Result<()> { + for root in [ + codex_home.join(SESSIONS_SUBDIR), + codex_home.join(ARCHIVED_SESSIONS_SUBDIR), + ] { + cleanup_stale_temps_in_root(root.as_path()).await?; + } + Ok(()) + } + + async fn cleanup_stale_temps_in_root(root: &Path) -> io::Result<()> { + if !tokio::fs::try_exists(root).await.unwrap_or(false) { + return Ok(()); + } + 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) => { + warn!( + "failed to read rollout temp cleanup directory {}: {err}", + dir.display() + ); + continue; + } + }; + while let Some(entry) = read_dir.next_entry().await? { + let path = entry.path(); + let file_type = match entry.file_type().await { + Ok(file_type) => file_type, + Err(err) => { + warn!( + "failed to read rollout temp cleanup file type {}: {err}", + path.display() + ); + continue; + } + }; + if file_type.is_dir() { + stack.push(path); + continue; + } + if file_type.is_file() + && path + .file_name() + .and_then(OsStr::to_str) + .is_some_and(|name| name.ends_with(TEMP_SUFFIX)) + { + let stale = entry + .metadata() + .await + .ok() + .and_then(|metadata| metadata.modified().ok()) + .and_then(|modified| SystemTime::now().duration_since(modified).ok()) + .is_some_and(|age| age >= TEMP_FILE_STALE_AFTER); + if !stale { + continue; + } + match tokio::fs::remove_file(path.as_path()).await { + Ok(()) => {} + Err(err) if err.kind() == io::ErrorKind::NotFound => {} + Err(err) => warn!( + "failed to remove stale rollout temp {}: {err}", + path.display() + ), + } + } + } + } + Ok(()) + } +} + /// 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 { diff --git a/codex-rs/rollout/src/compression_tests.rs b/codex-rs/rollout/src/compression_tests.rs index 7dde0735c..cd2618a2f 100644 --- a/codex-rs/rollout/src/compression_tests.rs +++ b/codex-rs/rollout/src/compression_tests.rs @@ -1,6 +1,9 @@ use std::fs; +use std::fs::FileTimes; #[cfg(unix)] use std::os::unix::fs::PermissionsExt; +use std::time::Duration; +use std::time::SystemTime; use codex_protocol::ThreadId; use codex_protocol::protocol::EventMsg; @@ -101,6 +104,46 @@ async fn append_rollout_item_materializes_compressed_rollout() -> anyhow::Result Ok(()) } +#[tokio::test] +async fn worker_compresses_old_archived_rollouts_only() -> anyhow::Result<()> { + let home = TempDir::new()?; + let active_uuid = Uuid::from_u128(3); + let active_id = ThreadId::from_string(&active_uuid.to_string())?; + let active_path = rollout_path(home.path(), "2025-01-03T12-00-00", active_uuid); + write_rollout(&active_path, active_id, "old active")?; + set_old_mtime(&active_path)?; + + let archived_uuid = Uuid::from_u128(4); + let archived_id = ThreadId::from_string(&archived_uuid.to_string())?; + let archived_path = archived_rollout_path(home.path(), "2025-01-04T12-00-00", archived_uuid); + write_rollout(&archived_path, archived_id, "old archived")?; + set_old_mtime(&archived_path)?; + + let fresh_uuid = Uuid::from_u128(5); + let fresh_id = ThreadId::from_string(&fresh_uuid.to_string())?; + let fresh_path = rollout_path(home.path(), "2025-01-05T12-00-00", fresh_uuid); + write_rollout(&fresh_path, fresh_id, "fresh active")?; + + let stale_temp = active_path.with_file_name("rollout-stale.jsonl.zst.tmp"); + fs::write(&stale_temp, "stale temp")?; + set_old_mtime(&stale_temp)?; + + let fresh_temp = active_path.with_file_name("rollout-fresh.jsonl.zst.tmp"); + fs::write(&fresh_temp, "fresh temp")?; + + worker::run(home.path().to_path_buf()).await?; + + assert!(active_path.exists()); + assert!(!compressed_rollout_path(&active_path).exists()); + assert!(!archived_path.exists()); + assert!(compressed_rollout_path(&archived_path).exists()); + assert!(fresh_path.exists()); + assert!(!compressed_rollout_path(&fresh_path).exists()); + assert!(!stale_temp.exists()); + assert!(fresh_temp.exists()); + Ok(()) +} + #[tokio::test] async fn resume_materializes_compressed_rollout_path() -> anyhow::Result<()> { let home = TempDir::new()?; @@ -153,6 +196,28 @@ async fn resume_materializes_compressed_rollout_path() -> anyhow::Result<()> { Ok(()) } +#[cfg(unix)] +#[tokio::test] +async fn compression_preserves_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 = archived_rollout_path(home.path(), "2025-01-03T12-00-00", uuid); + write_rollout(&rollout_path, thread_id, "restricted transcript")?; + fs::set_permissions(&rollout_path, fs::Permissions::from_mode(0o600))?; + set_old_mtime(&rollout_path)?; + + worker::run(home.path().to_path_buf()).await?; + + let compressed_path = compressed_rollout_path(&rollout_path); + assert!(!rollout_path.exists()); + assert_eq!( + fs::metadata(&compressed_path)?.permissions().mode() & 0o777, + 0o600 + ); + Ok(()) +} + #[cfg(unix)] #[tokio::test] async fn append_materialization_preserves_compressed_rollout_permissions() -> anyhow::Result<()> { @@ -183,6 +248,99 @@ async fn append_materialization_preserves_compressed_rollout_permissions() -> an Ok(()) } +#[test] +fn persist_temp_file_noclobber_installs_completed_temp() -> anyhow::Result<()> { + let home = TempDir::new()?; + let temp_path = home.path().join("rollout.jsonl.tmp"); + let destination = home.path().join("rollout.jsonl"); + fs::write(&temp_path, "completed rollout")?; + + persist_temp_file_noclobber(&temp_path, &destination)?; + + assert!(!temp_path.exists()); + assert_eq!(fs::read_to_string(destination)?, "completed rollout"); + Ok(()) +} + +#[test] +fn persist_temp_file_noclobber_does_not_replace_existing_destination() -> anyhow::Result<()> { + let home = TempDir::new()?; + let temp_path = home.path().join("rollout.jsonl.tmp"); + let destination = home.path().join("rollout.jsonl"); + fs::write(&temp_path, "candidate rollout")?; + fs::write(&destination, "existing rollout")?; + + persist_temp_file_noclobber(&temp_path, &destination)?; + + assert!(!temp_path.exists()); + assert_eq!(fs::read_to_string(destination)?, "existing rollout"); + Ok(()) +} + +#[cfg(unix)] +#[tokio::test] +async fn compression_preserves_read_only_rollout_permissions() -> anyhow::Result<()> { + let home = TempDir::new()?; + let uuid = Uuid::from_u128(7); + let thread_id = ThreadId::from_string(&uuid.to_string())?; + let rollout_path = archived_rollout_path(home.path(), "2025-01-03T12-00-00", uuid); + write_rollout(&rollout_path, thread_id, "read-only transcript")?; + set_old_mtime(&rollout_path)?; + fs::set_permissions(&rollout_path, fs::Permissions::from_mode(0o400))?; + let source_modified = fs::metadata(&rollout_path)?.modified()?; + + worker::run(home.path().to_path_buf()).await?; + + let compressed_path = compressed_rollout_path(&rollout_path); + let compressed_metadata = fs::metadata(&compressed_path)?; + assert!(!rollout_path.exists()); + assert_eq!(compressed_metadata.permissions().mode() & 0o777, 0o400); + assert_eq!(compressed_metadata.modified()?, source_modified); + Ok(()) +} + +#[tokio::test] +async fn worker_skips_existing_compressed_archived_rollouts() -> anyhow::Result<()> { + let home = TempDir::new()?; + let uuid = Uuid::from_u128(10); + let thread_id = ThreadId::from_string(&uuid.to_string())?; + let rollout_path = archived_rollout_path(home.path(), "2025-01-03T12-00-00", uuid); + write_rollout(&rollout_path, thread_id, "already compressed")?; + compress_now(&rollout_path)?; + let compressed_path = compressed_rollout_path(&rollout_path); + set_old_mtime(&compressed_path)?; + + worker::run(home.path().to_path_buf()).await?; + + assert!(!rollout_path.exists()); + assert!(compressed_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(), 2); + Ok(()) +} + +#[tokio::test] +async fn worker_skips_when_fresh_lock_exists() -> anyhow::Result<()> { + let home = TempDir::new()?; + let uuid = Uuid::from_u128(11); + let thread_id = ThreadId::from_string(&uuid.to_string())?; + let rollout_path = archived_rollout_path(home.path(), "2025-01-03T12-00-00", uuid); + write_rollout(&rollout_path, thread_id, "locked worker")?; + set_old_mtime(&rollout_path)?; + let lock_dir = home.path().join(".tmp"); + fs::create_dir_all(lock_dir.as_path())?; + fs::write(lock_dir.join("rollout-compression.lock"), "locked")?; + + worker::run(home.path().to_path_buf()).await?; + + assert!(rollout_path.exists()); + assert!(!compressed_rollout_path(&rollout_path).exists()); + Ok(()) +} + #[tokio::test] async fn find_thread_path_by_id_handles_compressed_rollout_filenames() -> anyhow::Result<()> { let home = TempDir::new()?; @@ -236,6 +394,11 @@ fn rollout_path(home: &std::path::Path, ts: &str, uuid: Uuid) -> std::path::Path .join(format!("rollout-{ts}-{uuid}.jsonl")) } +fn archived_rollout_path(home: &std::path::Path, ts: &str, uuid: Uuid) -> std::path::PathBuf { + home.join("archived_sessions") + .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)?; @@ -293,3 +456,15 @@ fn compress_now(path: &std::path::Path) -> anyhow::Result<()> { fs::remove_file(path)?; Ok(()) } + +fn set_old_mtime(path: &std::path::Path) -> anyhow::Result<()> { + let old = SystemTime::now() + .checked_sub(Duration::from_secs(8 * 24 * 60 * 60)) + .expect("old timestamp should be representable"); + let times = FileTimes::new().set_modified(old); + fs::OpenOptions::new() + .write(true) + .open(path)? + .set_times(times)?; + Ok(()) +} diff --git a/codex-rs/rollout/src/lib.rs b/codex-rs/rollout/src/lib.rs index e646b7da2..078e4bde1 100644 --- a/codex-rs/rollout/src/lib.rs +++ b/codex-rs/rollout/src/lib.rs @@ -37,6 +37,7 @@ 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 compression::spawn_rollout_compression_worker; pub use config::Config; pub use config::RolloutConfig; pub use config::RolloutConfigView;