Support compression for shared rollout lineages (#41357)

## Why

Rollout compression skips files that participate in paginated fork lineages because lineage readers rely on byte offsets into the original JSONL. This prevents shared rollout history from being compressed.

## What changed

- Add a seekable rollout reader that preserves logical JSONL offsets for plain and zstd-compressed files without materializing compressed data beside the rollout.
- Use the reader for lineage validation, model-context replay, and SQLite history projection, including when compression races with a read.
- Add the opt-in `local_thread_store_shared_compression` feature. The default mode continues to leave shared lineages uncompressed for compatibility, while the new mode includes referenced sources and fork pointers.
- Record the uncompressed size in new zstd frames so common prefix-bound checks can avoid decoding the full file.

## Testing

Add coverage for seekable compressed reads, prefix validation across zstd frame shapes, compressed lineage replay and projection, concurrent forks, and compression-mode behavior.

GitOrigin-RevId: 83e519f7eabf7448977eff37c55899858cc17a07

codex-rs/thread-store/src/local/model_context.rs

@@ -1,4 +1,3 @@
-use std::fs::File;
 use std::io;
 
 use codex_protocol::protocol::HistoryPosition;
@@ -26,13 +25,14 @@ mod tests;
 
 /// Loads rollout items needed to reconstruct the latest model-visible context.
 ///
-/// Plain paginated JSONL rollouts use a reverse scan. When it finds both a usable replacement-
+/// Paginated JSONL rollouts use a reverse scan. When it finds both a usable replacement-
 /// history checkpoint and the completed user-turn context needed for resume metadata, the returned
 /// replay starts with the canonical `SessionMeta` followed by that newest suffix. When no
 /// bounded cutoff is available, the scan continues to the beginning and returns the complete
 /// replay it already accumulated.
 ///
-/// Legacy and compressed rollout shapes keep the existing full-history path.
+/// Compressed segments are decoded before applying their original JSONL offsets. Legacy rollouts
+/// keep the existing full-history path.
 pub(super) async fn load_latest_model_context(
     store: &LocalThreadStore,
     params: LoadThreadHistoryParams,
@@ -65,12 +65,7 @@ pub(super) async fn load_latest_model_context(
         });
     }
 
-    let items = if matches!(session_meta.meta.history_mode, ThreadHistoryMode::Paginated)
-        && !path
-            .file_name()
-            .and_then(|file_name| file_name.to_str())
-            .is_some_and(|file_name| file_name.ends_with(".jsonl.zst"))
-    {
+    let items = if matches!(session_meta.meta.history_mode, ThreadHistoryMode::Paginated) {
         let lineage = store.resolve_rollout_lineage(params.thread_id).await?;
         scan_model_context_from_lineage(lineage, session_meta).await?
     } else {
@@ -137,7 +132,7 @@ fn scan_model_context_from_lineage_blocking(
 ) -> io::Result<Vec<RolloutItem>> {
     let mut scan = ModelContextScan::default();
     'segments: for segment in lineage.segments().iter().rev() {
-        let file = File::open(segment.rollout_path.as_path())?;
+        let file = codex_rollout::open_rollout_seekable_reader(segment.rollout_path.as_path())?;
         let mut scanner = match segment.end.map(|end| end.end_byte_offset) {
             Some(end_byte_offset) => ReverseJsonlScanner::new_at(file, end_byte_offset)?,
             None => ReverseJsonlScanner::new(file)?,

codex-rs/thread-store/src/local/model_context_tests.rs

@@ -386,9 +386,34 @@ async fn replays_nested_archived_lineage_from_frozen_prefix() {
         turn_complete("child-turn"),
     ];
     assert_eq!(
-        serde_json::to_value(context.items).expect("serialize context"),
+        serde_json::to_value(&context.items).expect("serialize context"),
+        serde_json::to_value(&expected).expect("serialize expected context")
+    );
+    // The same frozen lineage must replay from compressed files, without materializing or
+    // accidentally including the archived root's records after the inherited cutoff.
+    for path in [&archived_root, &middle_path, &child_path] {
+        let input = std::fs::File::open(path).expect("open rollout");
+        let output = std::fs::File::create(path.with_extension("jsonl.zst"))
+            .expect("create compressed rollout");
+        zstd::stream::copy_encode(input, output, /*level*/ 3).expect("compress rollout");
+        std::fs::remove_file(path).expect("remove plain rollout");
+    }
+    let compressed_context = store
+        .load_latest_model_context(LoadThreadHistoryParams {
+            thread_id: child_id,
+            include_archived: false,
+        })
+        .await
+        .expect("load compressed lineage model context");
+    assert_eq!(
+        serde_json::to_value(compressed_context.items).expect("serialize compressed context"),
         serde_json::to_value(expected).expect("serialize expected context")
     );
+    assert!(
+        [archived_root, middle_path, child_path]
+            .iter()
+            .all(|path| !path.exists())
+    );
 }
 
 fn write_paginated_rollout<const N: usize>(

codex-rs/thread-store/src/local/revert_thread.rs

@@ -68,19 +68,22 @@ pub(super) async fn revert(
         });
     }
 
-    // HistoryPosition stores byte offsets in plain JSONL, so materialize any compressed
-    // immutable lineage before deriving the retained boundary.
+    // Preserve old-reader compatibility when introducing the first reference to a standalone
+    // source. Already-shared ancestors stay read-only; their offsets address decoded JSONL bytes.
     let mut lineage = store.resolve_rollout_lineage(thread_id).await?;
     for segment in &mut lineage.segments {
-        segment.rollout_path =
-            codex_rollout::materialize_rollout_for_reference(segment.rollout_path.as_path())
-                .await
-                .map_err(|err| ThreadStoreError::Internal {
-                    message: format!(
-                        "failed to materialize rollout {} for revert: {err}",
-                        segment.rollout_path.display()
-                    ),
-                })?;
+        if segment.rollout_id() == current_rollout.rollout_id && source_meta.history_base.is_none()
+        {
+            segment.rollout_path =
+                codex_rollout::materialize_rollout_for_reference(segment.rollout_path.as_path())
+                    .await
+                    .map_err(|err| ThreadStoreError::Internal {
+                        message: format!(
+                            "failed to materialize rollout {} for revert: {err}",
+                            segment.rollout_path.display()
+                        ),
+                    })?;
+        }
         super::thread_history_materialization::materialize_to_sqlite(
             store,
             segment.rollout_id(),

codex-rs/thread-store/src/local/rollout_lineage.rs

@@ -94,21 +94,11 @@ impl LocalThreadStore {
             }
             let rollout_path = match representation {
                 LineageRepresentation::Existing => rollout_path,
-                LineageRepresentation::PlainForReference => {
-                    let rollout_path = super::helpers::scoped_rollout_path(
-                        self.config.codex_home.clone(),
-                        rollout_path.as_path(),
-                        "Codex home",
-                    )?;
-                    codex_rollout::materialize_rollout_for_reference(rollout_path.as_path())
-                        .await
-                        .map_err(|err| ThreadStoreError::Internal {
-                            message: format!(
-                                "failed to materialize referenced rollout {}: {err}",
-                                rollout_path.display()
-                            ),
-                        })?
-                }
+                LineageRepresentation::PlainForReference => super::helpers::scoped_rollout_path(
+                    self.config.codex_home.clone(),
+                    rollout_path.as_path(),
+                    "Codex home",
+                )?,
             };
             let meta = codex_rollout::read_session_meta_line(rollout_path.as_path())
                 .await
@@ -130,6 +120,26 @@ impl LocalThreadStore {
                     "source rollout is not paginated",
                 ));
             }
+            let rollout_path = match representation {
+                LineageRepresentation::Existing => rollout_path,
+                LineageRepresentation::PlainForReference
+                    if next_rollout_id.is_none() && meta.meta.history_base.is_none() =>
+                {
+                    // A newly shared standalone source must remain readable by older binaries.
+                    codex_rollout::materialize_rollout_for_reference(rollout_path.as_path())
+                        .await
+                        .map_err(|err| ThreadStoreError::Internal {
+                            message: format!(
+                                "failed to materialize referenced rollout {}: {err}",
+                                rollout_path.display()
+                            ),
+                        })?
+                }
+                // Already-shared compressed history requires a compatible reader regardless of
+                // new forks. Read it without publishing decoded copies into ancestors' folders;
+                // their owners may concurrently archive or unarchive those immutable files.
+                LineageRepresentation::PlainForReference => rollout_path,
+            };
             if let Some(end) = end {
                 validate_cutoff_bounds(requested_thread_id, rollout_path.as_path(), &end).await?;
             }
@@ -238,16 +248,22 @@ async fn validate_cutoff_bounds(
             "cutoff cannot include source session metadata",
         ));
     }
-    let file_len = tokio::fs::metadata(rollout_path)
-        .await
-        .map_err(|err| ThreadStoreError::Internal {
-            message: format!(
-                "failed to read lineage metadata {}: {err}",
-                rollout_path.display()
-            ),
-        })?
-        .len();
-    if end.end_byte_offset > file_len {
+    let path = rollout_path.to_path_buf();
+    let end_byte_offset = end.end_byte_offset;
+    let contains_prefix = tokio::task::spawn_blocking(move || {
+        codex_rollout::rollout_contains_prefix(&path, end_byte_offset)
+    })
+    .await
+    .map_err(|err| ThreadStoreError::Internal {
+        message: format!("failed to join rollout prefix validation: {err}"),
+    })?
+    .map_err(|err| ThreadStoreError::Internal {
+        message: format!(
+            "failed to read lineage metadata {}: {err}",
+            rollout_path.display()
+        ),
+    })?;
+    if !contains_prefix {
         return Err(malformed_lineage(
             requested_thread_id,
             "cutoff byte offset is past the source rollout",

codex-rs/thread-store/src/local/thread_history_materialization.rs

@@ -29,9 +29,9 @@ pub(super) async fn materialize_to_sqlite(
         .as_ref()
         .map_or(0, |state| state.next_byte_offset);
     if projection_state.is_none()
-        && !tokio::fs::try_exists(rollout_path)
+        && codex_rollout::existing_rollout_path(rollout_path)
             .await
-            .map_err(thread_store_io_error)?
+            .is_none()
     {
         return Ok(());
     }
@@ -76,13 +76,21 @@ async fn read_projection_steps(
     thread_id: ThreadId,
     subagent_history_start_ordinal: Option<u64>,
 ) -> ThreadStoreResult<(Vec<RolloutProjectionStep>, u64)> {
-    let file_end_offset = match tokio::fs::metadata(rollout_path).await {
-        Ok(metadata) => metadata.len(),
+    let path = rollout_path.to_path_buf();
+    let file =
+        tokio::task::spawn_blocking(move || codex_rollout::open_rollout_seekable_reader(&path))
+            .await
+            .map_err(|err| ThreadStoreError::Internal {
+                message: format!("failed to join rollout projection read: {err}"),
+            })?;
+    let mut file = match file {
+        Ok(file) => tokio::fs::File::from_std(file),
         Err(err) if err.kind() == std::io::ErrorKind::NotFound && start_offset == 0 => {
             return Ok((Vec::new(), 0));
         }
         Err(err) => return Err(thread_store_io_error(err)),
     };
+    let file_end_offset = file.metadata().await.map_err(thread_store_io_error)?.len();
     let byte_count =
         file_end_offset
             .checked_sub(start_offset)
@@ -93,9 +101,6 @@ async fn read_projection_steps(
         message: "durable rollout append exceeds addressable memory".to_string(),
     })?;
     let mut bytes = vec![0; byte_count];
-    let mut file = tokio::fs::File::open(rollout_path)
-        .await
-        .map_err(thread_store_io_error)?;
     file.seek(SeekFrom::Start(start_offset))
         .await
         .map_err(thread_store_io_error)?;

codex-rs/thread-store/src/local/thread_history_materialization_tests.rs

@@ -908,6 +908,19 @@ async fn active_turn_stores_only_its_start_position() {
     assert!(prepared.model_context.iter().any(|item| {
         matches!(item, RolloutItem::EventMsg(EventMsg::TurnStarted(event)) if event.turn_id == "turn-1")
     }));
+    // Another store may fork the persisted prefix while this store keeps the source writer open.
+    let other_store = projection_store(home.path()).await;
+    let other_prepared =
+        prepare_paginated_fork(&other_store, thread_id, ForkBoundary::Latest).await;
+    assert_eq!(
+        serde_json::to_value((
+            other_prepared.history_base,
+            other_prepared.model_context.as_ref()
+        ))
+        .expect("serialize other store's fork snapshot"),
+        serde_json::to_value((prepared.history_base, prepared.model_context.as_ref()))
+            .expect("serialize live store's fork snapshot"),
+    );
     assert_eq!(
         prepare_paginated_fork(
             &store,
@@ -943,7 +956,7 @@ async fn paginated_fork_persists_empty_source() {
 }
 
 #[tokio::test]
-async fn paginated_fork_materializes_compressed_source_and_ancestor() {
+async fn paginated_fork_reads_compressed_shared_lineage_without_materializing() {
     let home = TempDir::new().expect("temp dir");
     let store = projection_store(home.path()).await;
     let ancestor_thread_id = ThreadId::default();
@@ -976,6 +989,17 @@ async fn paginated_fork_materializes_compressed_source_and_ancestor() {
         .await
         .expect("shutdown ancestor");
 
+    // A standalone source still becomes plain before its first shared reference, so the default
+    // mode does not introduce compressed lineages that older readers cannot follow.
+    compress_rollout(ancestor_path.as_path());
+    assert_eq!(
+        prepare_paginated_fork(&store, ancestor_thread_id, ForkBoundary::Latest)
+            .await
+            .history_base,
+        Some(ancestor_base)
+    );
+    assert!(ancestor_path.exists());
+
     let source_thread_id = ThreadId::default();
     create_paginated_subagent_thread(
         &store,
@@ -1022,20 +1046,20 @@ async fn paginated_fork_materializes_compressed_source_and_ancestor() {
         prepare_paginated_fork(&store, source_thread_id, ForkBoundary::Latest),
         prepare_paginated_fork(&store, source_thread_id, ForkBoundary::Latest),
     );
-    assert!(ancestor_path.exists());
-    assert!(source_path.exists());
-    assert!(!ancestor_compressed_path.exists());
-    assert!(!source_compressed_path.exists());
+    assert!(!ancestor_path.exists());
+    assert!(!source_path.exists());
+    assert!(ancestor_compressed_path.exists());
+    assert!(source_compressed_path.exists());
     assert_eq!(
-        fs::metadata(&ancestor_path)
+        fs::metadata(&ancestor_compressed_path)
             .and_then(|metadata| metadata.modified())
-            .expect("read materialized ancestor timestamp"),
+            .expect("read compressed ancestor timestamp"),
         ancestor_modified
     );
     assert_eq!(
-        fs::metadata(&source_path)
+        fs::metadata(&source_compressed_path)
             .and_then(|metadata| metadata.modified())
-            .expect("read materialized source timestamp"),
+            .expect("read compressed source timestamp"),
         source_modified
     );
     for prepared in [first, second] {
@@ -1050,10 +1074,48 @@ async fn paginated_fork_materializes_compressed_source_and_ancestor() {
             ));
         }
     }
+
+    // Skipping materialization must not allow a reference that cannot be resolved inside this home.
+    let external_home = TempDir::new().expect("external temp dir");
+    let external_path = external_home.path().join(
+        source_compressed_path
+            .file_name()
+            .expect("source rollout filename"),
+    );
+    fs::rename(&source_compressed_path, &external_path).expect("move shared source outside home");
+    store
+        .resume_thread(ResumeThreadParams {
+            thread_id: source_thread_id,
+            rollout_path: Some(external_path),
+            history: None,
+            include_archived: true,
+            metadata: ThreadPersistenceMetadata {
+                cwd: Some(home.path().to_path_buf()),
+                model_provider: "test-provider".to_string(),
+                memory_mode: ThreadMemoryMode::Enabled,
+            },
+        })
+        .await
+        .expect("resume external shared source");
+    let error = store
+        .prepare_fork(PrepareForkParams {
+            thread_id: source_thread_id,
+            boundary: ForkBoundary::Latest,
+        })
+        .await
+        .expect_err("external shared source cannot be referenced by rollout id");
+    assert!(matches!(
+        error,
+        crate::ThreadStoreError::InvalidRequest { message } if message.contains("must be in Codex home")
+    ));
+    store
+        .shutdown_thread(source_thread_id)
+        .await
+        .expect("shutdown external source");
 }
 
 #[tokio::test]
-async fn cancelled_fork_keeps_source_reserved_until_lineage_materialization_finishes() {
+async fn cancelled_fork_keeps_source_reserved_until_lineage_resolution_finishes() {
     let home = TempDir::new().expect("temp dir");
     let store = projection_store(home.path()).await;
     let ancestor_thread_id = ThreadId::default();
@@ -1096,6 +1158,7 @@ async fn cancelled_fork_keeps_source_reserved_until_lineage_materialization_fini
     compress_rollout(source_path.as_path());
 
     let ancestor_writer_guard = store.live_writer_locks.lock(ancestor_thread_id).await;
+    let source_coordination = store.live_writer_locks.coordination(source_thread_id).await;
     let preparation_store = store.clone();
     let preparation = tokio::spawn(async move {
         preparation_store
@@ -1106,12 +1169,12 @@ async fn cancelled_fork_keeps_source_reserved_until_lineage_materialization_fini
             .await
     });
     tokio::time::timeout(Duration::from_secs(10), async {
-        while !source_path.exists() {
+        while source_coordination.lifecycle.try_write().is_ok() {
             tokio::task::yield_now().await;
         }
     })
     .await
-    .expect("detached lineage task should materialize the source");
+    .expect("fork preparation should reserve the source");
     preparation.abort();
     assert!(
         preparation
@@ -1126,14 +1189,14 @@ async fn cancelled_fork_keeps_source_reserved_until_lineage_materialization_fini
     tokio::select! {
         biased;
         result = &mut delete => {
-            panic!("source deletion completed while lineage materialization was active: {result:?}")
+            panic!("source deletion completed while lineage resolution was active: {result:?}")
         }
         _ = tokio::task::yield_now() => {}
     }
     drop(ancestor_writer_guard);
     delete
         .await
-        .expect("delete source after lineage materialization finishes");
+        .expect("delete source after lineage resolution finishes");
 }
 
 #[tokio::test]