diff --git a/codex-rs/rollout/src/rollout_reference_index.rs b/codex-rs/rollout/src/rollout_reference_index.rs index 1e94bd03df..7a16082eb6 100644 --- a/codex-rs/rollout/src/rollout_reference_index.rs +++ b/codex-rs/rollout/src/rollout_reference_index.rs @@ -36,7 +36,14 @@ struct IndexedRollout { impl RolloutReferenceIndex { /// Scans active and archived local rollout metadata without a deadline. pub async fn scan(codex_home: &Path) -> io::Result { - let Some(index) = Self::scan_with_deadline(codex_home, ScanDeadline::Unlimited).await? + let Some(index) = Self::scan_with_deadline( + vec![ + codex_home.join(ARCHIVED_SESSIONS_SUBDIR), + codex_home.join(SESSIONS_SUBDIR), + ], + ScanDeadline::Unlimited, + ) + .await? else { return Err(io::Error::other( "unlimited rollout reference scan exceeded a deadline", @@ -45,6 +52,19 @@ impl RolloutReferenceIndex { Ok(index) } + /// Scans only unarchived rollouts to locate files that still need to be archived. + /// + /// Reference counts exclude archived history and must not be used to decide whether a + /// rollout can be deleted or compressed. + pub async fn scan_unarchived(codex_home: &Path) -> io::Result { + Self::scan_with_deadline( + vec![codex_home.join(SESSIONS_SUBDIR)], + ScanDeadline::Unlimited, + ) + .await? + .ok_or_else(|| io::Error::other("unlimited rollout reference scan exceeded a deadline")) + } + /// Scans active and archived local rollout metadata until the worker deadline expires. /// /// Returns None instead of a partial index when the deadline expires. @@ -54,7 +74,10 @@ impl RolloutReferenceIndex { max_runtime: Duration, ) -> io::Result> { Self::scan_with_deadline( - codex_home, + vec![ + codex_home.join(ARCHIVED_SESSIONS_SUBDIR), + codex_home.join(SESSIONS_SUBDIR), + ], ScanDeadline::Until { started_at, max_runtime, @@ -90,14 +113,10 @@ impl RolloutReferenceIndex { } async fn scan_with_deadline( - codex_home: &Path, + mut stack: Vec, deadline: ScanDeadline, ) -> io::Result> { let mut rollouts_by_id = HashMap::new(); - let mut stack = vec![ - codex_home.join(ARCHIVED_SESSIONS_SUBDIR), - codex_home.join(SESSIONS_SUBDIR), - ]; while let Some(directory) = stack.pop() { if deadline.expired() { return Ok(None); diff --git a/codex-rs/rollout/src/rollout_reference_index_tests.rs b/codex-rs/rollout/src/rollout_reference_index_tests.rs index 1c4a57cd85..b82fdfbea5 100644 --- a/codex-rs/rollout/src/rollout_reference_index_tests.rs +++ b/codex-rs/rollout/src/rollout_reference_index_tests.rs @@ -108,6 +108,51 @@ async fn indexes_multiple_rollouts_for_the_same_thread() -> anyhow::Result<()> { Ok(()) } +#[tokio::test] +async fn unarchived_scan_finds_all_active_rollouts_owned_by_a_thread() -> anyhow::Result<()> { + let home = TempDir::new()?; + let owner_id = thread_id(Uuid::from_u128(30))?; + let original_uuid = Uuid::from_u128(31); + let replacement_uuid = Uuid::from_u128(32); + let original_path = active_rollout_path(home.path(), original_uuid); + let replacement_path = active_rollout_path(home.path(), replacement_uuid); + write_rollout(original_path.clone(), owner_id, /*history_base*/ None)?; + compress_now(&original_path)?; + write_rollout( + replacement_path.clone(), + owner_id, + /*history_base*/ None, + )?; + write_rollout( + archived_rollout_path(home.path(), Uuid::from_u128(33)), + owner_id, + /*history_base*/ None, + )?; + write_rollout( + active_rollout_path(home.path(), Uuid::from_u128(34)), + thread_id(Uuid::from_u128(34))?, + /*history_base*/ None, + )?; + + let index = RolloutReferenceIndex::scan_unarchived(home.path()).await?; + let mut owned: Vec<_> = index + .rollouts_for_thread(owner_id) + .map(|(id, path)| (id, path.to_path_buf())) + .collect(); + owned.sort_by_key(|(_, path)| path.clone()); + assert_eq!( + owned, + vec![ + ( + thread_id(original_uuid)?, + original_path.with_extension("jsonl.zst") + ), + (thread_id(replacement_uuid)?, replacement_path), + ] + ); + Ok(()) +} + #[tokio::test] async fn self_history_base_does_not_count_as_reference() -> anyhow::Result<()> { let home = TempDir::new()?; diff --git a/codex-rs/thread-store/src/local/archive_thread.rs b/codex-rs/thread-store/src/local/archive_thread.rs index b8c0914812..c1ac4a7946 100644 --- a/codex-rs/thread-store/src/local/archive_thread.rs +++ b/codex-rs/thread-store/src/local/archive_thread.rs @@ -39,7 +39,8 @@ pub(super) async fn archive_threads( } } let _writer_guards = store.acquire_writer_locks(&lock_thread_ids).await?; - let reference_index = RolloutReferenceIndex::scan(store.config.codex_home.as_path()) + // Already-archived rollouts need no move. Avoid reading the entire archive on every request. + let reference_index = RolloutReferenceIndex::scan_unarchived(store.config.codex_home.as_path()) .await .map_err(|err| ThreadStoreError::Internal { message: format!("failed to scan thread rollout files: {err}"),