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/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(())
+}