Avoid scanning archived rollouts when archiving threads (#41908)

## Why

Archiving a thread only needs to locate rollout files that have not already
been moved, so reading the entire rollout archive on every request is
unnecessary.

## What changed

- Add `RolloutReferenceIndex::scan_unarchived` to scan only the active
  `sessions` directory.
- Use the active-only index when archiving threads while retaining full scans
  for operations that require archived-history reference counts.
- Cover discovery of compressed and uncompressed active rollouts and exclusion
  of archived and unrelated rollouts.

GitOrigin-RevId: 46367127578f20bfa46a8db5c6f2b7f59d29f7a9

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<Self> {
-        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> {
+        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<Option<Self>> {
         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<PathBuf>,
         deadline: ScanDeadline,
     ) -> io::Result<Option<Self>> {
         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);

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()?;

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}"),