From 8d4d57387a90635d1033eeae70105465ae9f96d0 Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Wed, 12 Aug 2026 20:17:05 +0000 Subject: [PATCH] Resolve paginated thread history by rollout ID (#38244) ## Why `thread/revert` keeps the logical thread ID while selecting a new immutable rollout. Using the thread ID for persisted history can therefore read or update the wrong rollout. ## What changed - Centralize resolution of a thread's current rollout across live writers, SQLite metadata, and filesystem fallbacks. - Key history materialization, lineage traversal, paging, turn lookup, and fork boundaries by immutable rollout ID while retaining the thread ID for the logical thread. - Treat SQLite's selected rollout as authoritative for paginated threads so a filesystem scan cannot select an older rollout after `thread/revert`. ## Testing Add a regression test showing that update-ordinal item paging reads from the selected rollout ID when it differs from the thread ID. GitOrigin-RevId: 1101e548a6e6c6d319996e24cb650e5b03ab7fbb --- .../thread-store/src/local/live_writer.rs | 27 ++- codex-rs/thread-store/src/local/mod.rs | 6 + .../thread-store/src/local/model_context.rs | 21 +- .../thread-store/src/local/paginated_fork.rs | 63 +++-- .../thread-store/src/local/read_thread.rs | 69 +----- .../thread-store/src/local/rollout_lineage.rs | 65 ++++-- .../src/local/rollout_lineage_tests.rs | 10 +- .../src/local/thread_history/read.rs | 14 +- .../src/local/thread_history/read_tests.rs | 47 ++++ .../src/local/thread_history/search.rs | 4 +- .../local/thread_history/segment_paging.rs | 34 ++- .../src/local/thread_history/turn_lookup.rs | 6 +- .../src/local/thread_rollout_resolver.rs | 216 ++++++++++++++++++ .../src/local/update_thread_metadata.rs | 102 +++------ 14 files changed, 471 insertions(+), 213 deletions(-) create mode 100644 codex-rs/thread-store/src/local/thread_rollout_resolver.rs diff --git a/codex-rs/thread-store/src/local/live_writer.rs b/codex-rs/thread-store/src/local/live_writer.rs index a71ec86729..8472027401 100644 --- a/codex-rs/thread-store/src/local/live_writer.rs +++ b/codex-rs/thread-store/src/local/live_writer.rs @@ -33,7 +33,7 @@ pub(super) async fn create_thread( let writer_lock = store.writer_lock_coordinator.acquire(thread_id)?; let recorder = create_thread::create_thread(store, params).await?; store - .insert_live_recorder(thread_id, recorder, history_mode, writer_lock) + .insert_live_recorder(thread_id, recorder, thread_id, history_mode, writer_lock) .await } @@ -100,13 +100,24 @@ pub(super) async fn resume_thread( model_provider_id: params.metadata.model_provider.clone(), generate_memories: matches!(params.metadata.memory_mode, ThreadMemoryMode::Enabled), }; + let rollout_id = super::thread_rollout_resolver::rollout_id_from_path_or_legacy_thread_id( + rollout_path.as_path(), + params.thread_id, + history_mode, + )?; let recorder = RolloutRecorder::new(&config, RolloutRecorderParams::resume(rollout_path)) .await .map_err(|err| ThreadStoreError::Internal { message: format!("failed to resume local thread recorder: {err}"), })?; store - .insert_live_recorder(params.thread_id, recorder, history_mode, writer_lock) + .insert_live_recorder( + params.thread_id, + recorder, + rollout_id, + history_mode, + writer_lock, + ) .await } @@ -146,7 +157,7 @@ pub(super) async fn shutdown_thread( thread_id: ThreadId, ) -> ThreadStoreResult<()> { let _live_writer_guard = store.live_writer_locks.lock(thread_id).await; - let (recorder, history_mode) = live_writer_parts(store, thread_id).await?; + let (recorder, rollout_id, history_mode) = live_writer_parts(store, thread_id).await?; let rollout_path = recorder.rollout_path().to_path_buf(); if matches!(history_mode, ThreadHistoryMode::Legacy) { recorder.shutdown().await.map_err(thread_store_io_error)?; @@ -154,7 +165,7 @@ pub(super) async fn shutdown_thread( recorder.shutdown().await.map_err(thread_store_io_error)?; if let Err(err) = super::thread_history_materialization::materialize_to_sqlite( store, - thread_id, + rollout_id, rollout_path.as_path(), ) .await @@ -269,12 +280,12 @@ enum RolloutWriteOp { async fn live_writer_parts( store: &LocalThreadStore, thread_id: ThreadId, -) -> ThreadStoreResult<(RolloutRecorder, ThreadHistoryMode)> { +) -> ThreadStoreResult<(RolloutRecorder, ThreadId, ThreadHistoryMode)> { let live_recorders = store.live_recorders.lock().await; let entry = live_recorders .get(&thread_id) .ok_or(ThreadStoreError::ThreadNotFound { thread_id })?; - Ok((entry.recorder.clone(), entry.history_mode)) + Ok((entry.recorder.clone(), entry.rollout_id, entry.history_mode)) } async fn write_and_project( @@ -286,7 +297,7 @@ async fn write_and_project( // shutdown/discard/delete removes it. Keep the lookup defensive so late writes fail after // teardown. let _live_writer_guard = store.live_writer_locks.lock(thread_id).await; - let (recorder, history_mode) = live_writer_parts(store, thread_id).await?; + let (recorder, rollout_id, history_mode) = live_writer_parts(store, thread_id).await?; let sync_rollout_path = matches!(&write_op, RolloutWriteOp::Persist | RolloutWriteOp::Flush); let write_op = match write_op { RolloutWriteOp::AppendItems(mut items) => { @@ -308,7 +319,7 @@ async fn write_and_project( durable_write(&recorder, write_op).await?; if let Err(err) = super::thread_history_materialization::materialize_to_sqlite( store, - thread_id, + rollout_id, rollout_path, ) .await diff --git a/codex-rs/thread-store/src/local/mod.rs b/codex-rs/thread-store/src/local/mod.rs index f4f14fe730..bb3b3967ef 100644 --- a/codex-rs/thread-store/src/local/mod.rs +++ b/codex-rs/thread-store/src/local/mod.rs @@ -15,6 +15,7 @@ mod rollout_lineage; mod search_threads; mod thread_history; mod thread_history_materialization; +mod thread_rollout_resolver; mod thread_sections; mod unarchive_thread; mod update_thread_metadata; @@ -112,6 +113,9 @@ pub struct LocalThreadStore { struct LiveRecorderEntry { recorder: RolloutRecorder, + // Rollout projection rows are keyed by immutable rollout ID, not the stable thread ID used + // to find this live writer. + rollout_id: ThreadId, // Local rollout files are materialized lazily, but metadata updates can arrive before the // canonical SessionMeta is durable. Retain the mode captured when live persistence was opened // so missing SQLite rows can still be seeded. @@ -291,6 +295,7 @@ impl LocalThreadStore { &self, thread_id: ThreadId, recorder: RolloutRecorder, + rollout_id: ThreadId, history_mode: ThreadHistoryMode, writer_lock: WriterLockGuard, ) -> ThreadStoreResult<()> { @@ -301,6 +306,7 @@ impl LocalThreadStore { Entry::Vacant(entry) => { entry.insert(LiveRecorderEntry { recorder, + rollout_id, history_mode, writer_lock, }); diff --git a/codex-rs/thread-store/src/local/model_context.rs b/codex-rs/thread-store/src/local/model_context.rs index 45fd1a9e76..ce70a00bf4 100644 --- a/codex-rs/thread-store/src/local/model_context.rs +++ b/codex-rs/thread-store/src/local/model_context.rs @@ -14,6 +14,7 @@ use codex_rollout::ScanOutcome; use super::LocalThreadStore; use super::read_thread; use super::rollout_lineage::RolloutLineage; +use super::thread_rollout_resolver; use crate::LoadThreadHistoryParams; use crate::StoredModelContext; use crate::ThreadStoreError; @@ -27,7 +28,7 @@ mod tests; /// /// Plain 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 head `SessionMeta` followed by that newest suffix. When no +/// 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. /// @@ -36,11 +37,17 @@ pub(super) async fn load_latest_model_context( store: &LocalThreadStore, params: LoadThreadHistoryParams, ) -> ThreadStoreResult { - let path = read_thread::resolve_rollout_path(store, params.thread_id, params.include_archived) - .await? - .ok_or_else(|| ThreadStoreError::InvalidRequest { - message: format!("no rollout found for thread id {}", params.thread_id), - })?; + let resolved = if params.include_archived { + thread_rollout_resolver::resolve_current_including_archived(store, params.thread_id).await? + } else { + thread_rollout_resolver::resolve_current(store, params.thread_id).await? + }; + let path = + resolved + .map(|resolved| resolved.path) + .ok_or_else(|| ThreadStoreError::InvalidRequest { + message: format!("no rollout found for thread id {}", params.thread_id), + })?; let session_meta = codex_rollout::read_session_meta_line(path.as_path()) .await @@ -139,7 +146,7 @@ fn scan_model_context_from_lineage_blocking( let ScanOutcome::Parsed(line) = outcome else { continue; }; - // Each physical segment contributes only its local delta. Its head metadata is + // Each rollout segment contributes only its local delta. Its session metadata is // replaced with the requested thread's canonical SessionMeta after replay. if matches!(&line.item, RolloutItem::SessionMeta(_)) { break; diff --git a/codex-rs/thread-store/src/local/paginated_fork.rs b/codex-rs/thread-store/src/local/paginated_fork.rs index 810443a7a2..0b14091267 100644 --- a/codex-rs/thread-store/src/local/paginated_fork.rs +++ b/codex-rs/thread-store/src/local/paginated_fork.rs @@ -55,10 +55,10 @@ pub(super) async fn prepare( .iter() .take(lineage.segments().len().saturating_sub(1)) { - let _ancestor_writer_guard = store.live_writer_locks.lock(segment.thread_id()).await; + let _ancestor_writer_guard = store.live_writer_locks.lock(segment.rollout_id()).await; super::thread_history_materialization::materialize_to_sqlite( store, - segment.thread_id(), + segment.rollout_id(), segment.rollout_path.as_path(), ) .await?; @@ -67,18 +67,43 @@ pub(super) async fn prepare( let source_writer_guard = store.live_writer_locks.lock(thread_id).await; super::thread_history_materialization::materialize_to_sqlite( store, - thread_id, + source_segment.rollout_id(), source_segment.rollout_path.as_path(), ) .await?; - let latest_projection_state = super::thread_history::projection_state(store, thread_id) - .await? - .ok_or_else(|| ThreadStoreError::Internal { - message: format!("missing projection state for paginated thread {thread_id}"), - })?; - let latest_position = HistoryPosition { + let history_base = history_base_at_boundary(store, thread_id, boundary, &lineage).await?; + drop(source_writer_guard); + let model_context = Arc::new(model_context::load_for_fork(lineage, history_base).await?); + + Ok(PreparedFork::new( thread_id, + history_base, + model_context, + source_reservation, + )) +} + +pub(super) async fn history_base_at_boundary( + store: &LocalThreadStore, + thread_id: codex_protocol::ThreadId, + boundary: ForkBoundary, + lineage: &super::rollout_lineage::RolloutLineage, +) -> ThreadStoreResult> { + let source_segment = lineage + .segments() + .last() + .ok_or_else(|| ThreadStoreError::Internal { + message: "fork lineage has no source segment".to_string(), + })?; + let latest_projection_state = + super::thread_history::projection_state(store, source_segment.rollout_id()) + .await? + .ok_or_else(|| ThreadStoreError::Internal { + message: format!("missing projection state for paginated thread {thread_id}"), + })?; + let latest_position = HistoryPosition { + thread_id: source_segment.rollout_id(), end_ordinal_exclusive: latest_projection_state.next_ordinal, end_byte_offset: latest_projection_state.next_byte_offset, }; @@ -86,7 +111,7 @@ pub(super) async fn prepare( let position = match boundary { ForkBoundary::Latest => latest_position, ForkBoundary::ThroughTurn(turn_id) => { - let row = find_visible_turn(pool, &lineage, turn_id.as_str()).await?; + let row = find_visible_turn(pool, lineage, turn_id.as_str()).await?; if row.status == "inProgress" { return Err(ThreadStoreError::InvalidRequest { message: format!("lastTurnId '{turn_id}' identifies an in-progress turn"), @@ -99,7 +124,7 @@ pub(super) async fn prepare( .rollout_end_byte_offset .ok_or_else(|| missing_turn_position(turn_id.as_str()))?; HistoryPosition { - thread_id: row.physical_thread_id, + thread_id: row.rollout_id, end_ordinal_exclusive: u64::try_from(rollout_end_ordinal) .map_err(|_| invalid_turn_position(turn_id.as_str()))? .checked_add(1) @@ -109,7 +134,7 @@ pub(super) async fn prepare( } } ForkBoundary::BeforeTurn(turn_id) => { - let row = find_source_turn(pool, &lineage, turn_id.as_str()).await?; + let row = find_source_turn(pool, lineage, turn_id.as_str()).await?; if row.rollout_end_ordinal == Some(row.rollout_ordinal) { return Err(ThreadStoreError::InvalidRequest { message: format!("turn {turn_id} does not have a persisted start boundary"), @@ -119,7 +144,7 @@ pub(super) async fn prepare( .rollout_byte_offset .ok_or_else(|| missing_turn_position(turn_id.as_str()))?; HistoryPosition { - thread_id: row.physical_thread_id, + thread_id: row.rollout_id, end_ordinal_exclusive: u64::try_from(row.rollout_ordinal) .map_err(|_| invalid_turn_position(turn_id.as_str()))?, end_byte_offset: u64::try_from(rollout_byte_offset) @@ -130,7 +155,7 @@ pub(super) async fn prepare( let segment_index = lineage .segments() .iter() - .position(|segment| segment.thread_id() == position.thread_id) + .position(|segment| segment.rollout_id() == position.thread_id) .ok_or_else(|| ThreadStoreError::Internal { message: "fork position is outside the source lineage".to_string(), })?; @@ -150,15 +175,7 @@ pub(super) async fn prepare( } else { Some(position) }; - drop(source_writer_guard); - let model_context = Arc::new(model_context::load_for_fork(lineage, history_base).await?); - - Ok(PreparedFork::new( - thread_id, - history_base, - model_context, - source_reservation, - )) + Ok(history_base) } fn missing_turn_position(turn_id: &str) -> ThreadStoreError { diff --git a/codex-rs/thread-store/src/local/read_thread.rs b/codex-rs/thread-store/src/local/read_thread.rs index d99ad6d42f..4b84ebfbaf 100644 --- a/codex-rs/thread-store/src/local/read_thread.rs +++ b/codex-rs/thread-store/src/local/read_thread.rs @@ -6,9 +6,7 @@ use codex_protocol::protocol::SessionMetaLine; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::ThreadHistoryMode; use codex_rollout::RolloutRecorder; -use codex_rollout::find_archived_thread_path_by_id_str; use codex_rollout::find_thread_name_by_id; -use codex_rollout::find_thread_path_by_id_str; use codex_rollout::read_session_meta_line; use codex_rollout::read_thread_item_from_rollout; use codex_state::ThreadMetadata; @@ -21,7 +19,7 @@ use super::helpers::rollout_path_is_archived; use super::helpers::set_thread_name; use super::helpers::sqlite_thread_name; use super::helpers::stored_thread_from_rollout_item; -use super::live_writer; +use super::thread_rollout_resolver; use crate::ReadThreadParams; use crate::StoredThread; use crate::StoredThreadHistory; @@ -79,11 +77,17 @@ pub(super) async fn read_thread( return Ok(thread); } - let path = resolve_rollout_path(store, thread_id, params.include_archived) - .await? - .ok_or_else(|| ThreadStoreError::InvalidRequest { - message: format!("no rollout found for thread id {thread_id}"), - })?; + let resolved = if params.include_archived { + thread_rollout_resolver::resolve_current_including_archived(store, thread_id).await? + } else { + thread_rollout_resolver::resolve_current(store, thread_id).await? + }; + let path = + resolved + .map(|resolved| resolved.path) + .ok_or_else(|| ThreadStoreError::InvalidRequest { + message: format!("no rollout found for thread id {thread_id}"), + })?; let mut thread = read_thread_from_rollout_path(store, path).await?; if !params.include_archived && thread.archived_at.is_some() { @@ -233,55 +237,6 @@ async fn attach_history_if_requested( Ok(()) } -pub(super) async fn resolve_rollout_path( - store: &LocalThreadStore, - thread_id: codex_protocol::ThreadId, - include_archived: bool, -) -> ThreadStoreResult> { - if let Ok(path) = live_writer::rollout_path(store, thread_id).await - && codex_rollout::existing_rollout_path(path.as_path()) - .await - .is_some() - && (include_archived || !rollout_path_is_archived(store.config.codex_home.as_path(), &path)) - { - return Ok(Some(path)); - } - - let state_db_ctx = store.state_db().await; - if include_archived { - match find_thread_path_by_id_str( - store.config.codex_home.as_path(), - &thread_id.to_string(), - state_db_ctx.as_deref(), - ) - .await - .map_err(|err| ThreadStoreError::InvalidRequest { - message: format!("failed to locate thread id {thread_id}: {err}"), - })? { - Some(path) => Ok(Some(path)), - None => find_archived_thread_path_by_id_str( - store.config.codex_home.as_path(), - &thread_id.to_string(), - state_db_ctx.as_deref(), - ) - .await - .map_err(|err| ThreadStoreError::InvalidRequest { - message: format!("failed to locate archived thread id {thread_id}: {err}"), - }), - } - } else { - find_thread_path_by_id_str( - store.config.codex_home.as_path(), - &thread_id.to_string(), - state_db_ctx.as_deref(), - ) - .await - .map_err(|err| ThreadStoreError::InvalidRequest { - message: format!("failed to locate thread id {thread_id}: {err}"), - }) - } -} - async fn read_thread_from_rollout_path( store: &LocalThreadStore, path: std::path::PathBuf, diff --git a/codex-rs/thread-store/src/local/rollout_lineage.rs b/codex-rs/thread-store/src/local/rollout_lineage.rs index 332cd2dbaa..e5d5a83437 100644 --- a/codex-rs/thread-store/src/local/rollout_lineage.rs +++ b/codex-rs/thread-store/src/local/rollout_lineage.rs @@ -7,23 +7,23 @@ use codex_protocol::protocol::HistoryPosition; use codex_protocol::protocol::ThreadHistoryMode; use super::LocalThreadStore; -use super::read_thread; +use super::thread_rollout_resolver; use crate::ThreadStoreError; use crate::ThreadStoreResult; -/// One physical rollout range contributing to a logical paginated history. +/// One immutable rollout range contributing to a paginated thread's history. #[derive(Clone, Debug, PartialEq, Eq)] pub(super) struct RolloutLineageSegment { - pub(super) thread_id: ThreadId, + pub(super) rollout_id: ThreadId, pub(super) rollout_path: PathBuf, pub(super) start_ordinal: u64, pub(super) end: Option, } -/// Ordered physical rollout ranges contributing to one logical forked history. +/// Ordered rollout ranges contributing to one 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. +/// consume its bounded rollout segments without resolving or mutating fork pointers themselves. #[derive(Clone, Debug, PartialEq, Eq)] pub(super) struct RolloutLineage { pub(super) segments: Vec, @@ -59,23 +59,39 @@ impl LocalThreadStore { ) -> ThreadStoreResult { let mut segments = Vec::new(); let mut seen = HashSet::new(); - let mut thread_id = requested_thread_id; + let mut next_rollout_id = None; let mut end = None; loop { - if !seen.insert(thread_id) { - return Err(malformed_lineage(requested_thread_id, "cycle detected")); - } + let coordination_id = next_rollout_id.unwrap_or(requested_thread_id); let _writer_guard = match representation { LineageRepresentation::Existing => None, LineageRepresentation::PlainForReference => { - Some(self.live_writer_locks.lock(thread_id).await) + Some(self.live_writer_locks.lock(coordination_id).await) } }; - let rollout_path = - read_thread::resolve_rollout_path(self, thread_id, /*include_archived*/ true) + let (rollout_id, rollout_path) = match next_rollout_id { + Some(rollout_id) => { + let rollout_path = resolve_rollout_path_by_id(self, rollout_id) + .await? + .ok_or_else(|| malformed_lineage(rollout_id, "missing source rollout"))?; + (rollout_id, rollout_path) + } + None => { + let resolved = thread_rollout_resolver::resolve_current_including_archived( + self, + requested_thread_id, + ) .await? - .ok_or_else(|| malformed_lineage(thread_id, "missing source rollout"))?; + .ok_or_else(|| { + malformed_lineage(requested_thread_id, "missing source rollout") + })?; + (resolved.rollout_id, resolved.path) + } + }; + if !seen.insert(rollout_id) { + return Err(malformed_lineage(requested_thread_id, "cycle detected")); + } let rollout_path = match representation { LineageRepresentation::Existing => rollout_path, LineageRepresentation::PlainForReference => { @@ -102,7 +118,7 @@ impl LocalThreadStore { rollout_path.display() ), })?; - if meta.meta.id != thread_id { + if next_rollout_id.is_none() && meta.meta.id != requested_thread_id { return Err(malformed_lineage( requested_thread_id, "source rollout belongs to another thread", @@ -124,7 +140,7 @@ impl LocalThreadStore { None => 1, }; segments.push(RolloutLineageSegment { - thread_id, + rollout_id, rollout_path, start_ordinal, end, @@ -133,7 +149,7 @@ impl LocalThreadStore { let Some(base) = meta.meta.history_base else { break; }; - thread_id = base.thread_id; + next_rollout_id = Some(base.thread_id); end = Some(base); } @@ -142,6 +158,17 @@ impl LocalThreadStore { } } +async fn resolve_rollout_path_by_id( + store: &LocalThreadStore, + rollout_id: ThreadId, +) -> ThreadStoreResult> { + codex_rollout::find_rollout_path_by_rollout_id(store.config.codex_home.as_path(), rollout_id) + .await + .map_err(|err| ThreadStoreError::Internal { + message: format!("failed to locate rollout {rollout_id}: {err}"), + }) +} + #[derive(Clone, Copy)] enum LineageRepresentation { Existing, @@ -169,7 +196,7 @@ impl RolloutLineage { let segment_index = self .segments .iter() - .position(|segment| segment.thread_id == end.thread_id) + .position(|segment| segment.rollout_id == end.thread_id) .ok_or_else(|| ThreadStoreError::Internal { message: "fork position is outside the source lineage".to_string(), })?; @@ -187,8 +214,8 @@ impl RolloutLineage { } impl RolloutLineageSegment { - pub(super) fn thread_id(&self) -> ThreadId { - self.thread_id + pub(super) fn rollout_id(&self) -> ThreadId { + self.rollout_id } pub(super) fn start_ordinal(&self) -> u64 { diff --git a/codex-rs/thread-store/src/local/rollout_lineage_tests.rs b/codex-rs/thread-store/src/local/rollout_lineage_tests.rs index c120d5a583..2935ac3889 100644 --- a/codex-rs/thread-store/src/local/rollout_lineage_tests.rs +++ b/codex-rs/thread-store/src/local/rollout_lineage_tests.rs @@ -51,19 +51,19 @@ async fn resolves_nested_lineage_with_empty_intermediate_segments() { lineage.segments, vec![ RolloutLineageSegment { - thread_id: root, + rollout_id: root, rollout_path: root_path.clone(), start_ordinal: 1, end: Some(root_end), }, RolloutLineageSegment { - thread_id: middle, + rollout_id: middle, rollout_path: middle_path.clone(), start_ordinal: 5, end: Some(middle_end), }, RolloutLineageSegment { - thread_id: child, + rollout_id: child, rollout_path: child_path, start_ordinal: 6, end: None, @@ -135,13 +135,13 @@ async fn resolves_lineage_at_explicit_history_position() { lineage.segments, vec![ RolloutLineageSegment { - thread_id: root, + rollout_id: root, rollout_path: root_path.clone(), start_ordinal: 1, end: Some(root_end), }, RolloutLineageSegment { - thread_id: child, + rollout_id: child, rollout_path: child_path.clone(), start_ordinal: 5, end: Some(end), diff --git a/codex-rs/thread-store/src/local/thread_history/read.rs b/codex-rs/thread-store/src/local/thread_history/read.rs index 59d817ee94..b2e3e6713b 100644 --- a/codex-rs/thread-store/src/local/thread_history/read.rs +++ b/codex-rs/thread-store/src/local/thread_history/read.rs @@ -45,12 +45,12 @@ pub(super) struct HistoryCursor { } #[derive(Clone, Copy)] -pub(super) struct PhysicalHistoryPosition { +pub(super) struct RolloutHistoryPosition { pub rollout_ordinal: i64, } pub(super) struct StoredTurnRow { - pub position: PhysicalHistoryPosition, + pub position: RolloutHistoryPosition, pub turn_id: String, pub status: StoredTurnStatus, pub error: Option, @@ -88,7 +88,7 @@ struct StoredSummaryItemColumns { } pub(super) struct StoredThreadItemRow { - pub position: PhysicalHistoryPosition, + pub position: RolloutHistoryPosition, pub item: StoredThreadItem, } @@ -212,7 +212,7 @@ async fn load_inherited_summary_items( let Some(segment) = lineage .segments() .iter() - .find(|segment| segment.thread_id() == source.physical_thread_id) + .find(|segment| segment.rollout_id() == source.rollout_id) else { return Ok(Vec::new()); }; @@ -233,7 +233,7 @@ WHERE thread_id = ? ORDER BY rollout_ordinal ASC "#, ) - .bind(source.physical_thread_id.to_string()) + .bind(source.rollout_id.to_string()) .bind(turn.turn_id.as_str()) .bind(start_ordinal) .bind(end_ordinal) @@ -298,7 +298,7 @@ pub(super) fn stored_turn_row(row: sqlx::sqlite::SqliteRow) -> ThreadStoreResult .transpose() .map_err(super::thread_history_error)?; Ok(StoredTurnRow { - position: PhysicalHistoryPosition { + position: RolloutHistoryPosition { rollout_ordinal: row.try_get("rollout_ordinal")?, }, turn_id: row.try_get("turn_id")?, @@ -389,7 +389,7 @@ pub(super) fn stored_thread_item_row( }); } Ok(StoredThreadItemRow { - position: PhysicalHistoryPosition { rollout_ordinal }, + position: RolloutHistoryPosition { rollout_ordinal }, item: stored_thread_item(row)?, }) } diff --git a/codex-rs/thread-store/src/local/thread_history/read_tests.rs b/codex-rs/thread-store/src/local/thread_history/read_tests.rs index aa5adae78f..fdffc43394 100644 --- a/codex-rs/thread-store/src/local/thread_history/read_tests.rs +++ b/codex-rs/thread-store/src/local/thread_history/read_tests.rs @@ -378,6 +378,53 @@ async fn list_items_rejects_update_ordinals_outside_sqlite_integer_range() { } } +#[tokio::test] +async fn list_items_update_ordinals_use_selected_rollout_id() { + let (home, store, thread_id) = store_with_mode(ThreadHistoryMode::Paginated).await; + let rollout_id = ThreadId::new(); + let selected_rollout_path = home.path().join(format!( + "sessions/2026/07/16/rollout-2026-07-16T00-00-00-{thread_id}_{rollout_id}.jsonl" + )); + write_rollout( + selected_rollout_path.as_path(), + thread_id, + /*history_base*/ None, + ); + let state_db = store.state_db().await.expect("state runtime"); + let mut metadata = state_db + .get_thread(thread_id) + .await + .expect("read metadata") + .expect("thread metadata"); + metadata.rollout_path = selected_rollout_path; + state_db + .upsert_thread(&metadata) + .await + .expect("select replacement rollout"); + insert_item( + history_db(&store).await, + rollout_id, + "turn-1", + "item-1", + /*rollout_ordinal*/ 1, + ) + .await; + + let page = store + .list_items(updated_item_params( + thread_id, /*after_updated_at_ordinal*/ 0, + )) + .await + .expect("update-ordered item page"); + + assert_eq!( + page.items, + vec![expected_item( + "turn-1", "item-1", /*rollout_ordinal*/ 1 + )] + ); +} + #[tokio::test] async fn list_history_keeps_legacy_threads_unsupported() { let (_home, store, thread_id) = store_with_mode(ThreadHistoryMode::Legacy).await; diff --git a/codex-rs/thread-store/src/local/thread_history/search.rs b/codex-rs/thread-store/src/local/thread_history/search.rs index dc310152ab..59c49e4820 100644 --- a/codex-rs/thread-store/src/local/thread_history/search.rs +++ b/codex-rs/thread-store/src/local/thread_history/search.rs @@ -152,12 +152,12 @@ FROM ( ORDER BY rollout_ordinal ASC "#, ) - .bind(segment.thread_id().to_string()) + .bind(segment.rollout_id().to_string()) .bind(next_rollout_ordinal) .bind(end_rollout_ordinal) .bind(segment_start_ordinal) .bind(end_rollout_ordinal) - .bind(segment.thread_id().to_string()) + .bind(segment.rollout_id().to_string()) .bind(next_rollout_ordinal) .bind(end_rollout_ordinal) .bind(segment_start_ordinal) diff --git a/codex-rs/thread-store/src/local/thread_history/segment_paging.rs b/codex-rs/thread-store/src/local/thread_history/segment_paging.rs index 9171c1c5a6..45b087b675 100644 --- a/codex-rs/thread-store/src/local/thread_history/segment_paging.rs +++ b/codex-rs/thread-store/src/local/thread_history/segment_paging.rs @@ -7,7 +7,7 @@ use super::super::rollout_lineage::RolloutLineage; use super::super::rollout_lineage::RolloutLineageSegment; use super::read::CursorScope; use super::read::HistoryCursor; -use super::read::PhysicalHistoryPosition; +use super::read::RolloutHistoryPosition; use super::read::StoredSummaryColumns; use super::read::StoredThreadItemRow; use super::read::StoredTurnRow; @@ -92,12 +92,12 @@ FROM thread_turns WHERE thread_id = "# }); - query.push_bind(segment.thread_id().to_string()); + query.push_bind(segment.rollout_id().to_string()); push_segment_range(&mut query, segment)?; for newer_segment in &lineage.segments()[segment_index + 1..] { query .push(" AND NOT EXISTS (SELECT 1 FROM thread_turns AS newer_turn WHERE newer_turn.thread_id = ") - .push_bind(newer_segment.thread_id().to_string()) + .push_bind(newer_segment.rollout_id().to_string()) .push(" AND newer_turn.turn_id = thread_turns.turn_id AND newer_turn.rollout_ordinal >= ") .push_bind(sqlite_integer(newer_segment.start_ordinal())?); if let Some(end_ordinal) = newer_segment.end_ordinal() { @@ -178,7 +178,7 @@ fn push_summary_item_join( item_id_column: &str, ) -> ThreadStoreResult<()> { query - .push_bind(segment.thread_id().to_string()) + .push_bind(segment.rollout_id().to_string()) .push(" AND ") .push(alias) .push(".turn_id = page_turns.turn_id AND ") @@ -204,7 +204,7 @@ pub(super) async fn page_item_rows( lineage: &RolloutLineage, params: &ListItemsParams, ) -> ThreadStoreResult> { - // Update ordinals are local to a physical rollout. Forked lineages need a structured + // Update ordinals are local to a rollout. Forked lineages need a structured // watermark before incremental replay can safely span their segments. if params.after_updated_at_ordinal.is_some() && lineage.segments().len() > 1 { return Err(ThreadStoreError::InvalidRequest { @@ -217,7 +217,18 @@ pub(super) async fn page_item_rows( message: "update-ordinal item sorting requires an update watermark".to_string(), }); }; - return page_updated_item_rows(pool, params, after_updated_at_ordinal).await; + let [segment] = lineage.segments() else { + return Err(ThreadStoreError::Internal { + message: "update-ordinal item paging requires one rollout segment".to_string(), + }); + }; + return page_updated_item_rows( + pool, + segment.rollout_id(), + params, + after_updated_at_ordinal, + ) + .await; } let cursor = parse_cursor( params.cursor.as_deref(), @@ -239,7 +250,7 @@ FROM thread_items WHERE thread_id = "#, ); - query.push_bind(segment.thread_id().to_string()); + query.push_bind(segment.rollout_id().to_string()); push_segment_range(&mut query, segment)?; if let Some(after_updated_at_ordinal) = params.after_updated_at_ordinal { query @@ -272,6 +283,7 @@ WHERE thread_id = async fn page_updated_item_rows( pool: &sqlx::SqlitePool, + rollout_id: ThreadId, params: &ListItemsParams, after_updated_at_ordinal: u64, ) -> ThreadStoreResult> { @@ -288,7 +300,7 @@ WHERE thread_id = "#, ); query - .push_bind(params.thread_id.to_string()) + .push_bind(rollout_id.to_string()) .push(" AND updated_at_ordinal > ") .push_bind(sqlite_integer(after_updated_at_ordinal)?); if let Some(turn_id) = params.turn_id.as_deref() { @@ -462,17 +474,17 @@ fn finish_page( } trait HasPosition { - fn position(&self) -> PhysicalHistoryPosition; + fn position(&self) -> RolloutHistoryPosition; } impl HasPosition for StoredTurnRow { - fn position(&self) -> PhysicalHistoryPosition { + fn position(&self) -> RolloutHistoryPosition { self.position } } impl HasPosition for StoredThreadItemRow { - fn position(&self) -> PhysicalHistoryPosition { + fn position(&self) -> RolloutHistoryPosition { self.position } } diff --git a/codex-rs/thread-store/src/local/thread_history/turn_lookup.rs b/codex-rs/thread-store/src/local/thread_history/turn_lookup.rs index ad4d926055..7fe20d6809 100644 --- a/codex-rs/thread-store/src/local/thread_history/turn_lookup.rs +++ b/codex-rs/thread-store/src/local/thread_history/turn_lookup.rs @@ -8,7 +8,7 @@ use crate::ThreadStoreError; use crate::ThreadStoreResult; pub(in crate::local) struct TurnRow { - pub physical_thread_id: ThreadId, + pub rollout_id: ThreadId, pub rollout_ordinal: i64, pub rollout_byte_offset: Option, pub rollout_end_ordinal: Option, @@ -75,7 +75,7 @@ WHERE thread_id = ? AND (? IS NULL OR rollout_ordinal < ?) "#, ) - .bind(segment.thread_id().to_string()) + .bind(segment.rollout_id().to_string()) .bind(turn_id) .bind(sqlite_integer(segment.start_ordinal(), "rollout ordinal")?) .bind(end_ordinal) @@ -87,7 +87,7 @@ WHERE thread_id = ? }) .map(|row| { row.map(|row| TurnRow { - physical_thread_id: segment.thread_id(), + rollout_id: segment.rollout_id(), rollout_ordinal: row.get("rollout_ordinal"), rollout_byte_offset: row.get("rollout_byte_offset"), rollout_end_ordinal: row.get("rollout_end_ordinal"), diff --git a/codex-rs/thread-store/src/local/thread_rollout_resolver.rs b/codex-rs/thread-store/src/local/thread_rollout_resolver.rs new file mode 100644 index 0000000000..5fad0edd50 --- /dev/null +++ b/codex-rs/thread-store/src/local/thread_rollout_resolver.rs @@ -0,0 +1,216 @@ +//! Resolves a thread ID to the rollout file the thread currently uses. +//! +//! Most threads have one rollout file. `thread/revert` keeps the thread ID stable while switching +//! the thread to a new rollout file, so callers cannot infer the selected path from the thread ID +//! alone. This module centralizes live-writer, SQLite, and filesystem fallback resolution. + +use std::path::PathBuf; + +use codex_protocol::ThreadId; +use codex_protocol::protocol::ThreadHistoryMode; +use codex_rollout::find_archived_thread_path_by_id_str; +use codex_rollout::find_thread_path_by_id_str; + +use super::LocalThreadStore; +use super::helpers::rollout_path_is_archived; +use super::live_writer; +use crate::ThreadStoreError; +use crate::ThreadStoreResult; + +/// One thread resolved to the concrete rollout file it currently uses. +/// +/// For ordinary threads, `thread_id` and `rollout_id` are the same. After `thread/revert`, +/// `thread_id` stays stable while `rollout_id` identifies the new immutable rollout file. +#[derive(Clone, Debug, PartialEq, Eq)] +pub(super) struct ResolvedThreadRollout { + pub(super) thread_id: ThreadId, + pub(super) rollout_id: ThreadId, + pub(super) path: PathBuf, + pub(super) location: RolloutLocation, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(super) enum RolloutLocation { + Unarchived, + Archived, +} + +pub(super) async fn resolve_current( + store: &LocalThreadStore, + thread_id: ThreadId, +) -> ThreadStoreResult> { + resolve(store, thread_id, LookupScope::ExcludeArchived).await +} + +pub(super) async fn resolve_current_including_archived( + store: &LocalThreadStore, + thread_id: ThreadId, +) -> ThreadStoreResult> { + resolve(store, thread_id, LookupScope::IncludeArchived).await +} + +#[derive(Clone, Copy)] +enum LookupScope { + ExcludeArchived, + IncludeArchived, +} + +impl LookupScope { + fn accepts(self, location: RolloutLocation) -> bool { + match self { + Self::ExcludeArchived => location == RolloutLocation::Unarchived, + Self::IncludeArchived => true, + } + } +} + +/// Resolves a thread's selected rollout in this order: +/// +/// 1. The live writer, when one exists. +/// 2. SQLite's selected rollout path. +/// 3. Filesystem fallback for threads SQLite does not identify as paginated. +/// 4. Archived filesystem fallback, when requested. +async fn resolve( + store: &LocalThreadStore, + thread_id: ThreadId, + scope: LookupScope, +) -> ThreadStoreResult> { + if let Ok(path) = live_writer::rollout_path(store, thread_id).await + && codex_rollout::existing_rollout_path(path.as_path()) + .await + .is_some() + && let Some(resolved) = resolve_path_in_scope(store, thread_id, path, scope).await? + { + return Ok(Some(resolved)); + } + + let state_db_ctx = store.state_db().await; + if let Some(state_db_ctx) = state_db_ctx.as_deref() { + match state_db_ctx.get_thread(thread_id).await { + Ok(Some(metadata)) => { + // Once SQLite identifies a thread as paginated, its rollout path is + // authoritative: after `thread/revert`, a scan could find an older immutable + // rollout for the same thread. Filesystem fallback remains available when SQLite + // has no row or identifies the thread as legacy. + if let Some(path) = + codex_rollout::existing_rollout_path(metadata.rollout_path.as_path()).await + { + let belongs_to_thread = + match codex_rollout::read_session_meta_line(path.as_path()).await { + Ok(session_meta) => session_meta.meta.id == thread_id, + Err(_) => true, + }; + if belongs_to_thread { + return resolve_path_in_scope(store, thread_id, path, scope).await; + } + } + if metadata.history_mode == ThreadHistoryMode::Paginated { + return Ok(None); + } + } + Ok(None) => {} + Err(err) => { + return Err(ThreadStoreError::Internal { + message: format!("failed to read thread metadata for {thread_id}: {err}"), + }); + } + } + } + if let Some(path) = find_thread_path_by_id_str( + store.config.codex_home.as_path(), + &thread_id.to_string(), + state_db_ctx.as_deref(), + ) + .await + .map_err(|err| ThreadStoreError::InvalidRequest { + message: format!("failed to locate thread id {thread_id}: {err}"), + })? && let Some(resolved) = resolve_path_in_scope(store, thread_id, path, scope).await? + { + return Ok(Some(resolved)); + } + if !scope.accepts(RolloutLocation::Archived) { + return Ok(None); + } + let path = find_archived_thread_path_by_id_str( + store.config.codex_home.as_path(), + &thread_id.to_string(), + state_db_ctx.as_deref(), + ) + .await + .map_err(|err| ThreadStoreError::InvalidRequest { + message: format!("failed to locate archived thread id {thread_id}: {err}"), + })?; + match path { + Some(path) => resolve_path_in_scope(store, thread_id, path, scope).await, + None => Ok(None), + } +} + +async fn resolve_path_in_scope( + store: &LocalThreadStore, + thread_id: ThreadId, + path: PathBuf, + scope: LookupScope, +) -> ThreadStoreResult> { + let location = location_for_path(store, path.as_path()); + if !scope.accepts(location) { + return Ok(None); + } + resolve_path(thread_id, path, location).await.map(Some) +} + +fn location_for_path(store: &LocalThreadStore, path: &std::path::Path) -> RolloutLocation { + if rollout_path_is_archived(store.config.codex_home.as_path(), path) { + RolloutLocation::Archived + } else { + RolloutLocation::Unarchived + } +} + +async fn resolve_path( + thread_id: ThreadId, + path: PathBuf, + location: RolloutLocation, +) -> ThreadStoreResult { + let rollout_id = match codex_rollout::rollout_id_from_path(path.as_path()) { + Some(rollout_id) => rollout_id, + None => { + let history_mode = codex_rollout::read_session_meta_line(path.as_path()) + .await + .map_err(|err| ThreadStoreError::Internal { + message: format!("failed to read session metadata {}: {err}", path.display()), + })? + .meta + .history_mode; + rollout_id_from_path_or_legacy_thread_id(path.as_path(), thread_id, history_mode)? + } + }; + Ok(ResolvedThreadRollout { + thread_id, + rollout_id, + path, + location, + }) +} + +/// Returns the immutable rollout ID for a path while preserving legacy noncanonical filenames. +pub(super) fn rollout_id_from_path_or_legacy_thread_id( + path: &std::path::Path, + thread_id: ThreadId, + history_mode: ThreadHistoryMode, +) -> ThreadStoreResult { + Ok(match codex_rollout::rollout_id_from_path(path) { + Some(rollout_id) => rollout_id, + None => { + if history_mode == ThreadHistoryMode::Paginated { + return Err(ThreadStoreError::InvalidRequest { + message: format!( + "paginated rollout path `{}` does not have a canonical rollout filename", + path.display() + ), + }); + } + thread_id + } + }) +} diff --git a/codex-rs/thread-store/src/local/update_thread_metadata.rs b/codex-rs/thread-store/src/local/update_thread_metadata.rs index 2d9f4b45a9..8062edeb3d 100644 --- a/codex-rs/thread-store/src/local/update_thread_metadata.rs +++ b/codex-rs/thread-store/src/local/update_thread_metadata.rs @@ -7,12 +7,9 @@ use codex_protocol::protocol::GitInfo; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::ThreadHistoryMode; use codex_protocol::protocol::ThreadMemoryMode; -use codex_rollout::ARCHIVED_SESSIONS_SUBDIR; use codex_rollout::RolloutItem; use codex_rollout::append_rollout_item_to_path; use codex_rollout::append_thread_name; -use codex_rollout::find_archived_thread_path_by_id_str; -use codex_rollout::find_thread_path_by_id_str; use codex_rollout::read_session_meta_line; use codex_state::ThreadMetadataBuilder; use tracing::warn; @@ -21,6 +18,9 @@ use super::LocalThreadStore; use super::helpers::git_info_from_parts; use super::helpers::permission_profile_to_metadata_value; use super::live_writer; +use super::thread_rollout_resolver; +use super::thread_rollout_resolver::ResolvedThreadRollout; +use super::thread_rollout_resolver::RolloutLocation; use crate::GitInfoPatch; use crate::ReadThreadParams; use crate::StoredThread; @@ -30,11 +30,6 @@ use crate::ThreadStoreResult; use crate::UpdateThreadMetadataParams; use crate::local::read_thread; -struct ResolvedRolloutPath { - path: PathBuf, - archived: bool, -} - pub(super) async fn update_thread_metadata( store: &LocalThreadStore, params: UpdateThreadMetadataParams, @@ -127,24 +122,30 @@ pub(super) async fn update_thread_metadata( if live_writer::rollout_path(store, thread_id).await.is_ok() { live_writer::persist_thread(store, thread_id).await?; } - let mut resolved_rollout_path = - resolve_rollout_path(store, thread_id, params.include_archived).await?; + let mut resolved_rollout = if params.include_archived { + thread_rollout_resolver::resolve_current_including_archived(store, thread_id).await? + } else { + thread_rollout_resolver::resolve_current(store, thread_id).await? + } + .ok_or_else(|| ThreadStoreError::InvalidRequest { + message: format!("thread not found: {thread_id}"), + })?; let name = patch.name; let git_info = patch.git_info; if let Some(memory_mode) = patch.memory_mode { - apply_thread_memory_mode(resolved_rollout_path.path.as_path(), thread_id, memory_mode) - .await?; - refresh_resolved_rollout_path(&mut resolved_rollout_path).await; + apply_thread_memory_mode(resolved_rollout.path.as_path(), thread_id, memory_mode).await?; + refresh_resolved_rollout_path(&mut resolved_rollout).await; } let state_db_ctx = store.state_db().await; codex_rollout::state_db::reconcile_rollout( state_db_ctx.as_deref(), - resolved_rollout_path.path.as_path(), + resolved_rollout.path.as_path(), store.config.default_model_provider_id.as_str(), /*builder*/ None, &[], - /*archived_only*/ resolved_rollout_path.archived.then_some(true), + /*archived_only*/ + (resolved_rollout.location == RolloutLocation::Archived).then_some(true), /*new_thread_memory_mode*/ None, ) .await; @@ -202,7 +203,7 @@ pub(super) async fn update_thread_metadata( }; if let Some(((sha, branch, origin_url), memory_mode)) = resolved_git_info.as_ref() { apply_thread_git_info_to_rollout( - resolved_rollout_path.path.as_path(), + resolved_rollout.path.as_path(), thread_id, sha, branch, @@ -210,7 +211,7 @@ pub(super) async fn update_thread_metadata( memory_mode.as_deref(), ) .await?; - refresh_resolved_rollout_path(&mut resolved_rollout_path).await; + refresh_resolved_rollout_path(&mut resolved_rollout).await; apply_thread_git_info(store, thread_id, sha, branch, origin_url).await?; } @@ -228,7 +229,7 @@ pub(super) async fn update_thread_metadata( Err(_) => { read_thread::read_thread_by_rollout_path( store, - resolved_rollout_path.path, + resolved_rollout.path, params.include_archived, /*include_history*/ false, ) @@ -241,7 +242,7 @@ pub(super) async fn update_thread_metadata( Ok(thread) } -async fn refresh_resolved_rollout_path(resolved: &mut ResolvedRolloutPath) { +async fn refresh_resolved_rollout_path(resolved: &mut ResolvedThreadRollout) { if let Some(path) = codex_rollout::existing_rollout_path(resolved.path.as_path()).await { resolved.path = path; } @@ -273,8 +274,16 @@ async fn apply_metadata_update( })?; let advance_recency_at = patch.advance_recency_at; if existing.is_none() && rollout_path.is_none() { - let resolved = resolve_rollout_path(store, thread_id, include_archived).await?; - rollout_path_archived = resolved.archived; + let resolved = if include_archived { + thread_rollout_resolver::resolve_current_including_archived(store, thread_id) + .await? + } else { + thread_rollout_resolver::resolve_current(store, thread_id).await? + } + .ok_or_else(|| ThreadStoreError::InvalidRequest { + message: format!("thread not found: {thread_id}"), + })?; + rollout_path_archived = resolved.location == RolloutLocation::Archived; rollout_path = Some(resolved.path); } let mut metadata = match existing.clone() { @@ -739,57 +748,8 @@ fn memory_mode_as_str(mode: ThreadMemoryMode) -> &'static str { } } -async fn resolve_rollout_path( - store: &LocalThreadStore, - thread_id: ThreadId, - include_archived: bool, -) -> ThreadStoreResult { - if let Ok(path) = live_writer::rollout_path(store, thread_id).await { - let archived = rollout_path_is_archived(store, path.as_path()); - return Ok(ResolvedRolloutPath { path, archived }); - } - - let state_db_ctx = store.state_db().await; - let active_path = find_thread_path_by_id_str( - store.config.codex_home.as_path(), - &thread_id.to_string(), - state_db_ctx.as_deref(), - ) - .await - .map_err(|err| ThreadStoreError::InvalidRequest { - message: format!("failed to locate thread id {thread_id}: {err}"), - })?; - if let Some(path) = active_path { - return Ok(ResolvedRolloutPath { - path, - archived: false, - }); - } - if !include_archived { - return Err(ThreadStoreError::InvalidRequest { - message: format!("thread not found: {thread_id}"), - }); - } - find_archived_thread_path_by_id_str( - store.config.codex_home.as_path(), - &thread_id.to_string(), - state_db_ctx.as_deref(), - ) - .await - .map_err(|err| ThreadStoreError::InvalidRequest { - message: format!("failed to locate archived thread id {thread_id}: {err}"), - })? - .map(|path| ResolvedRolloutPath { - path, - archived: true, - }) - .ok_or_else(|| ThreadStoreError::InvalidRequest { - message: format!("thread not found: {thread_id}"), - }) -} - fn rollout_path_is_archived(store: &LocalThreadStore, path: &Path) -> bool { - path.starts_with(store.config.codex_home.join(ARCHIVED_SESSIONS_SUBDIR)) + super::helpers::rollout_path_is_archived(store.config.codex_home.as_path(), path) } #[cfg(test)]