Resolve paginated rollout lineages (#34407)

## What changed

- Add a shared local thread-store resolver that follows `history_base` links into ordered, bounded rollout segments, including archived ancestors.
- Support resolving a lineage at an explicit `HistoryPosition`.
- Reject malformed lineages with cycles, missing or mismatched source rollouts, non-paginated sources, or invalid cutoff bounds.

## Testing

- Add unit coverage for nested and archived lineages, explicit history positions, cycles, missing sources, and out-of-bounds offsets.

GitOrigin-RevId: a9063ad87e81d9865bd17fc588ea5d8e9ac74c0a

codex-rs/rollout/src/ordinal.rs

@@ -78,8 +78,9 @@ pub(crate) fn ordinal_state_for_rollout(
             path.display()
         ))
     })?;
-    // The child metadata is written before its inherited prefix. Never resume a
-    // partial initialization: later child records would otherwise stay below the boundary.
+    // Child records must start at `subagent_history_start_ordinal`. If initialization died while
+    // copying the inherited parent records, resuming would append child records before that
+    // boundary.
     if let Some(prefix_end) = subagent_history_start_ordinal.and_then(|start| start.checked_sub(1))
         && ordinal < prefix_end
     {

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

@@ -6,6 +6,9 @@ mod list_threads;
 mod live_writer;
 mod model_context;
 mod read_thread;
+// This lands before the reader PRs that consume the shared lineage resolver.
+#[allow(dead_code)]
+mod rollout_lineage;
 mod search_threads;
 mod thread_history;
 mod thread_history_materialization;

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

@@ -0,0 +1,142 @@
+use std::collections::HashSet;
+use std::path::Path;
+use std::path::PathBuf;
+
+use codex_protocol::ThreadId;
+use codex_protocol::protocol::HistoryPosition;
+use codex_protocol::protocol::ThreadHistoryMode;
+
+use super::LocalThreadStore;
+use super::read_thread;
+use crate::ThreadStoreError;
+use crate::ThreadStoreResult;
+
+/// One physical rollout range contributing to a logical paginated history.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub(super) struct RolloutLineageSegment {
+    pub(super) thread_id: ThreadId,
+    pub(super) rollout_path: PathBuf,
+    pub(super) end: Option<HistoryPosition>,
+}
+
+/// Ordered physical rollout ranges contributing to one logical forked history.
+///
+/// This is the only local abstraction that follows SessionMeta.history_base pointers. Readers
+/// consume its bounded physical segments without resolving or mutating fork pointers themselves.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub(super) struct RolloutLineage {
+    pub(super) segments: Vec<RolloutLineageSegment>,
+}
+
+impl LocalThreadStore {
+    pub(super) async fn resolve_rollout_lineage(
+        &self,
+        requested_thread_id: ThreadId,
+    ) -> ThreadStoreResult<RolloutLineage> {
+        let mut segments = Vec::new();
+        let mut seen = HashSet::new();
+        let mut thread_id = requested_thread_id;
+        let mut end = None;
+
+        loop {
+            if !seen.insert(thread_id) {
+                return Err(malformed_lineage(requested_thread_id, "cycle detected"));
+            }
+            let rollout_path =
+                read_thread::resolve_rollout_path(self, thread_id, /*include_archived*/ true)
+                    .await?
+                    .ok_or_else(|| malformed_lineage(thread_id, "missing source rollout"))?;
+            let meta = codex_rollout::read_session_meta_line(rollout_path.as_path())
+                .await
+                .map_err(|err| ThreadStoreError::Internal {
+                    message: format!(
+                        "failed to read lineage metadata {}: {err}",
+                        rollout_path.display()
+                    ),
+                })?;
+            if meta.meta.id != thread_id {
+                return Err(malformed_lineage(
+                    requested_thread_id,
+                    "source rollout belongs to another thread",
+                ));
+            }
+            if meta.meta.history_mode != ThreadHistoryMode::Paginated {
+                return Err(malformed_lineage(
+                    requested_thread_id,
+                    "source rollout is not paginated",
+                ));
+            }
+            if let Some(end) = end {
+                validate_cutoff_bounds(requested_thread_id, rollout_path.as_path(), &end).await?;
+            }
+            segments.push(RolloutLineageSegment {
+                thread_id,
+                rollout_path,
+                end,
+            });
+
+            let Some(base) = meta.meta.history_base else {
+                break;
+            };
+            thread_id = base.thread_id;
+            end = Some(base);
+        }
+
+        segments.reverse();
+        Ok(RolloutLineage { segments })
+    }
+
+    pub(super) async fn resolve_rollout_lineage_at(
+        &self,
+        end: HistoryPosition,
+    ) -> ThreadStoreResult<RolloutLineage> {
+        let mut lineage = self.resolve_rollout_lineage(end.thread_id).await?;
+        let Some(segment) = lineage.segments.last_mut() else {
+            return Err(ThreadStoreError::Internal {
+                message: "rollout lineage has no segments".to_string(),
+            });
+        };
+        validate_cutoff_bounds(end.thread_id, segment.rollout_path.as_path(), &end).await?;
+        segment.end = Some(end);
+        Ok(lineage)
+    }
+}
+
+async fn validate_cutoff_bounds(
+    requested_thread_id: ThreadId,
+    rollout_path: &Path,
+    end: &HistoryPosition,
+) -> ThreadStoreResult<()> {
+    if end.end_ordinal_exclusive == 0 {
+        return Err(malformed_lineage(
+            requested_thread_id,
+            "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 {
+        return Err(malformed_lineage(
+            requested_thread_id,
+            "cutoff byte offset is past the source rollout",
+        ));
+    }
+    Ok(())
+}
+
+fn malformed_lineage(thread_id: ThreadId, detail: &str) -> ThreadStoreError {
+    ThreadStoreError::InvalidRequest {
+        message: format!("invalid paginated history lineage for {thread_id}: {detail}"),
+    }
+}
+
+#[cfg(test)]
+#[path = "rollout_lineage_tests.rs"]
+mod tests;

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

@@ -0,0 +1,299 @@
+use std::fs;
+use std::path::Path;
+
+use codex_protocol::ThreadId;
+use codex_protocol::protocol::HistoryPosition;
+use codex_protocol::protocol::RolloutItem;
+use codex_protocol::protocol::RolloutLine;
+use codex_protocol::protocol::SessionMeta;
+use codex_protocol::protocol::SessionMetaLine;
+use codex_protocol::protocol::ThreadHistoryMode;
+use pretty_assertions::assert_eq;
+use tempfile::TempDir;
+
+use super::super::LocalThreadStore;
+use super::super::test_support::test_config;
+use super::RolloutLineageSegment;
+
+#[tokio::test]
+async fn resolves_nested_lineage_with_empty_intermediate_segments() {
+    let home = TempDir::new().expect("temp dir");
+    let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
+    let root = ThreadId::default();
+    let middle = ThreadId::default();
+    let child = ThreadId::default();
+    let root_path = write_rollout(
+        home.path(),
+        root,
+        /*history_base*/ None,
+        /*next_ordinal*/ 6,
+    );
+    let root_end = history_position(root_path.as_path(), root, /*end_ordinal_exclusive*/ 4);
+    let middle_path = write_rollout(home.path(), middle, Some(root_end), /*next_ordinal*/ 1);
+    let middle_end = history_position(
+        middle_path.as_path(),
+        middle,
+        /*end_ordinal_exclusive*/ 1,
+    );
+    let child_path = write_rollout(
+        home.path(),
+        child,
+        Some(middle_end),
+        /*next_ordinal*/ 3,
+    );
+
+    let lineage = store
+        .resolve_rollout_lineage(child)
+        .await
+        .expect("resolve nested lineage");
+
+    assert_eq!(
+        lineage.segments,
+        vec![
+            RolloutLineageSegment {
+                thread_id: root,
+                rollout_path: root_path.clone(),
+                end: Some(root_end),
+            },
+            RolloutLineageSegment {
+                thread_id: middle,
+                rollout_path: middle_path.clone(),
+                end: Some(middle_end),
+            },
+            RolloutLineageSegment {
+                thread_id: child,
+                rollout_path: child_path,
+                end: None,
+            },
+        ]
+    );
+}
+
+#[tokio::test]
+async fn resolves_archived_ancestors() {
+    let home = TempDir::new().expect("temp dir");
+    let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
+    let root = ThreadId::default();
+    let child = ThreadId::default();
+    let root_path = write_rollout_under(
+        home.path().join("archived_sessions"),
+        root,
+        /*history_base*/ None,
+        /*next_ordinal*/ 3,
+    );
+    write_rollout(
+        home.path(),
+        child,
+        Some(history_position(
+            root_path.as_path(),
+            root,
+            /*end_ordinal_exclusive*/ 3,
+        )),
+        /*next_ordinal*/ 2,
+    );
+
+    let lineage = store
+        .resolve_rollout_lineage(child)
+        .await
+        .expect("resolve archived ancestor");
+
+    assert_eq!(lineage.segments[0].rollout_path, root_path);
+}
+
+#[tokio::test]
+async fn resolves_lineage_at_explicit_history_position() {
+    let home = TempDir::new().expect("temp dir");
+    let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
+    let root = ThreadId::default();
+    let child = ThreadId::default();
+    let root_path = write_rollout(
+        home.path(),
+        root,
+        /*history_base*/ None,
+        /*next_ordinal*/ 6,
+    );
+    let root_end = history_position(root_path.as_path(), root, /*end_ordinal_exclusive*/ 4);
+    let child_path = write_rollout(home.path(), child, Some(root_end), /*next_ordinal*/ 4);
+    let end = history_position(
+        child_path.as_path(),
+        child,
+        /*end_ordinal_exclusive*/ 2,
+    );
+
+    let lineage = store
+        .resolve_rollout_lineage_at(end)
+        .await
+        .expect("resolve explicit position");
+
+    assert_eq!(
+        lineage.segments,
+        vec![
+            RolloutLineageSegment {
+                thread_id: root,
+                rollout_path: root_path.clone(),
+                end: Some(root_end),
+            },
+            RolloutLineageSegment {
+                thread_id: child,
+                rollout_path: child_path.clone(),
+                end: Some(end),
+            },
+        ]
+    );
+}
+
+#[tokio::test]
+async fn rejects_missing_cycles_and_out_of_bounds_offsets() {
+    let home = TempDir::new().expect("temp dir");
+    let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
+    let missing_parent = ThreadId::default();
+    let missing_child = ThreadId::default();
+    write_rollout(
+        home.path(),
+        missing_child,
+        Some(unchecked_history_position(
+            missing_parent,
+            /*end_ordinal_exclusive*/ 1,
+        )),
+        /*next_ordinal*/ 2,
+    );
+    assert_invalid_lineage(&store, missing_child, "missing source rollout").await;
+
+    let cycle_a = ThreadId::default();
+    let cycle_b = ThreadId::default();
+    write_rollout(
+        home.path(),
+        cycle_a,
+        Some(unchecked_history_position(
+            cycle_b, /*end_ordinal_exclusive*/ 1,
+        )),
+        /*next_ordinal*/ 2,
+    );
+    write_rollout(
+        home.path(),
+        cycle_b,
+        Some(unchecked_history_position(
+            cycle_a, /*end_ordinal_exclusive*/ 1,
+        )),
+        /*next_ordinal*/ 2,
+    );
+    assert_invalid_lineage(&store, cycle_a, "cycle detected").await;
+
+    let root = ThreadId::default();
+    let invalid_child = ThreadId::default();
+    let root_path = write_rollout(
+        home.path(),
+        root,
+        /*history_base*/ None,
+        /*next_ordinal*/ 2,
+    );
+    write_rollout(
+        home.path(),
+        invalid_child,
+        Some(HistoryPosition {
+            thread_id: root,
+            end_ordinal_exclusive: 2,
+            end_byte_offset: fs::metadata(root_path).expect("root metadata").len() + 1,
+        }),
+        /*next_ordinal*/ 2,
+    );
+    assert_invalid_lineage(
+        &store,
+        invalid_child,
+        "cutoff byte offset is past the source rollout",
+    )
+    .await;
+}
+
+async fn assert_invalid_lineage(store: &LocalThreadStore, thread_id: ThreadId, detail: &str) {
+    let err = store
+        .resolve_rollout_lineage(thread_id)
+        .await
+        .expect_err("lineage should be invalid");
+    assert!(err.to_string().contains(detail), "{err}");
+}
+
+fn write_rollout(
+    home: &Path,
+    thread_id: ThreadId,
+    history_base: Option<HistoryPosition>,
+    next_ordinal: u64,
+) -> std::path::PathBuf {
+    write_rollout_under(
+        home.join("sessions/2026/07/16"),
+        thread_id,
+        history_base,
+        next_ordinal,
+    )
+}
+
+fn write_rollout_under(
+    directory: std::path::PathBuf,
+    thread_id: ThreadId,
+    history_base: Option<HistoryPosition>,
+    next_ordinal: u64,
+) -> std::path::PathBuf {
+    fs::create_dir_all(directory.as_path()).expect("create rollout directory");
+    let path = directory.join(format!("rollout-2026-07-16T00-00-00-{thread_id}.jsonl"));
+    let mut lines = vec![rollout_line(
+        /*ordinal*/ 0,
+        RolloutItem::SessionMeta(SessionMetaLine {
+            meta: SessionMeta {
+                session_id: thread_id.into(),
+                id: thread_id,
+                history_mode: ThreadHistoryMode::Paginated,
+                history_base,
+                ..SessionMeta::default()
+            },
+            git: None,
+        }),
+    )];
+    for ordinal in 1..next_ordinal {
+        lines.push(rollout_line(
+            ordinal,
+            RolloutItem::EventMsg(codex_protocol::protocol::EventMsg::ShutdownComplete),
+        ));
+    }
+    fs::write(path.as_path(), format!("{}\n", lines.join("\n"))).expect("write rollout");
+    path
+}
+
+fn rollout_line(ordinal: u64, item: RolloutItem) -> String {
+    serde_json::to_string(&RolloutLine {
+        timestamp: "2026-07-16T00:00:00.000Z".to_string(),
+        ordinal: Some(ordinal),
+        item,
+    })
+    .expect("serialize rollout line")
+}
+
+fn history_position(
+    path: &Path,
+    thread_id: ThreadId,
+    end_ordinal_exclusive: u64,
+) -> HistoryPosition {
+    HistoryPosition {
+        thread_id,
+        end_ordinal_exclusive,
+        end_byte_offset: rollout_end_byte_offset(path, end_ordinal_exclusive),
+    }
+}
+
+fn rollout_end_byte_offset(path: &Path, end_ordinal_exclusive: u64) -> u64 {
+    let line_count = usize::try_from(end_ordinal_exclusive).expect("ordinal fits usize");
+    let bytes = fs::read(path).expect("read rollout");
+    let end_byte_offset = bytes
+        .split_inclusive(|byte| *byte == b'\n')
+        .take(line_count)
+        .map(<[u8]>::len)
+        .sum::<usize>();
+    u64::try_from(end_byte_offset).expect("rollout byte offset fits u64")
+}
+
+fn unchecked_history_position(thread_id: ThreadId, end_ordinal_exclusive: u64) -> HistoryPosition {
+    HistoryPosition {
+        thread_id,
+        end_ordinal_exclusive,
+        end_byte_offset: 0,
+    }
+}