From b7e39aa31608b6eaba4f317538a8f82985a9e854 Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Mon, 20 Jul 2026 20:28:29 +0000 Subject: [PATCH] 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 | 5 +- codex-rs/thread-store/src/local/mod.rs | 3 + .../thread-store/src/local/rollout_lineage.rs | 142 +++++++++ .../src/local/rollout_lineage_tests.rs | 299 ++++++++++++++++++ 4 files changed, 447 insertions(+), 2 deletions(-) create mode 100644 codex-rs/thread-store/src/local/rollout_lineage.rs create mode 100644 codex-rs/thread-store/src/local/rollout_lineage_tests.rs diff --git a/codex-rs/rollout/src/ordinal.rs b/codex-rs/rollout/src/ordinal.rs index 0c17607149f3..fbb75e7e22b0 100644 --- a/codex-rs/rollout/src/ordinal.rs +++ b/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 { diff --git a/codex-rs/thread-store/src/local/mod.rs b/codex-rs/thread-store/src/local/mod.rs index f0f8adbec6b6..576642267bf5 100644 --- a/codex-rs/thread-store/src/local/mod.rs +++ b/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; diff --git a/codex-rs/thread-store/src/local/rollout_lineage.rs b/codex-rs/thread-store/src/local/rollout_lineage.rs new file mode 100644 index 000000000000..aed97f55f19f --- /dev/null +++ b/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, +} + +/// 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, +} + +impl LocalThreadStore { + pub(super) async fn resolve_rollout_lineage( + &self, + requested_thread_id: ThreadId, + ) -> ThreadStoreResult { + 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 { + 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; diff --git a/codex-rs/thread-store/src/local/rollout_lineage_tests.rs b/codex-rs/thread-store/src/local/rollout_lineage_tests.rs new file mode 100644 index 000000000000..801db259dab3 --- /dev/null +++ b/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, + 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, + 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::(); + 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, + } +}