Skip unrelated indexed rollouts during selected thread migration (#57483)

## Why

Migrating selected threads reads rollout headers even for unrelated threads. Use the state database to skip files whose indexed owner and canonical filename agree that they belong to an unselected thread.

## What changed

- Read thread IDs and rollout paths without creating or migrating the state database, and fall back to header inspection if the index cannot be read immediately.
- Keep inspecting selected, unindexed, misnamed, and conflicting paths so rollout headers still determine ownership.
- Recognize stable thread IDs in reverted rollout filenames and match plain and compressed paths when filtering.

## Testing

Add a regression test covering pruning of unrelated indexed rollouts, including archived and compressed files, while retaining header inspection for selected and ambiguously named files.

GitOrigin-RevId: d76827174c00f5bc9e38ad9b69dadeee91f4415f

codex-rs/rollout/src/lib.rs

@@ -139,6 +139,7 @@ pub use maintenance::try_acquire_rollout_maintenance_lock;
 pub use metadata::builder_from_items;
 pub use metadata::forked_from_ordinal_exclusive;
 pub use metadata::rollout_id_from_path;
+pub use metadata::thread_id_from_rollout_path;
 pub use model_context::ModelContextScan;
 pub use model_context::ModelContextScanProgress;
 pub use persistence_metrics::RolloutPersistenceBatchMeasurement;

codex-rs/rollout/src/metadata.rs

@@ -10,6 +10,7 @@ use chrono::NaiveDateTime;
 use chrono::Timelike;
 use chrono::Utc;
 use codex_protocol::RolloutId;
+use codex_protocol::ThreadId;
 use codex_protocol::protocol::AskForApproval;
 use codex_protocol::protocol::SandboxPolicy;
 use codex_protocol::protocol::SessionMeta;
@@ -103,6 +104,11 @@ pub fn builder_from_items(
     ))
 }
 
+/// Returns the stable thread ID from a canonical filename, including a reverted rollout's name.
+pub fn thread_id_from_rollout_path(path: &Path) -> Option<ThreadId> {
+    Some(RolloutFileName::parse(path.file_name()?.to_str()?)?.thread_id())
+}
+
 /// Returns the rollout ID encoded in a canonical rollout filename.
 ///
 /// Normal rollouts use `rollout-<timestamp>-<thread-id>.jsonl`, where the thread ID and rollout ID

codex-rs/state/src/audit.rs

@@ -1,8 +1,10 @@
-//! Read-only state database queries for diagnostics.
+//! Read-only queries against an existing state database.
 
 use anyhow::Result;
+use codex_protocol::ThreadId;
 use sqlx::Row;
 use std::path::PathBuf;
+use std::time::Duration;
 
 use crate::SqliteConfig;
 
@@ -46,3 +48,18 @@ FROM threads
         })
         .collect()
 }
+
+/// Reads only the thread IDs and selected paths, without creating or migrating the state DB.
+pub async fn read_thread_rollout_paths(sqlite: &SqliteConfig) -> Result<Vec<(ThreadId, PathBuf)>> {
+    // Migration can scan the rollout headers immediately if the index is locked.
+    let pool = sqlite
+        .open_read_only_pool(&sqlite.state_db_path(), Some(Duration::ZERO))
+        .await?;
+    let rows = sqlx::query_as::<_, (String, String)>("SELECT id, rollout_path FROM threads")
+        .fetch_all(&pool)
+        .await?;
+    pool.close().await;
+    rows.into_iter()
+        .map(|(id, path)| Ok((ThreadId::from_string(&id)?, PathBuf::from(path))))
+        .collect()
+}

codex-rs/state/src/lib.rs

@@ -47,6 +47,7 @@ pub use sqlite::SqliteConfig;
 pub use sqlite::SqlitePoolWriteExt;
 
 pub use audit::ThreadStateAuditRow;
+pub use audit::read_thread_rollout_paths;
 pub use audit::read_thread_state_audit_rows;
 pub use delegated_preview::delegated_output_preview;
 pub use extract::GUARDIAN_THREAD_PREVIEW;

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

@@ -10,6 +10,7 @@
 //! Compression and decompression workers retain these locks if their async caller is cancelled.
 
 use std::collections::HashMap;
+use std::collections::HashSet;
 use std::io;
 use std::path::Path;
 use std::path::PathBuf;
@@ -347,6 +348,28 @@ impl LocalThreadStore {
             }
             RolloutMigrationPaths::Known(paths) => paths,
         };
+        if !options.thread_ids.is_empty()
+            && let Ok(rows) = codex_state::read_thread_rollout_paths(&self.config.sqlite).await
+        {
+            let selected = options.thread_ids.iter().copied().collect::<HashSet<_>>();
+            let mut needs_header = HashMap::with_capacity(rows.len());
+            for (owner, path) in rows {
+                let inspect = selected.contains(&owner)
+                    || codex_rollout::thread_id_from_rollout_path(&path) != Some(owner);
+                // Conflicting rows can name the same plain or compressed file.
+                needs_header
+                    .entry(codex_rollout::plain_rollout_path(&path))
+                    .and_modify(|inspect| *inspect = true)
+                    .or_insert(inspect);
+            }
+            paths.retain(|path| {
+                // Unindexed files may have been renamed. Their header still decides ownership.
+                needs_header
+                    .get(&codex_rollout::plain_rollout_path(path))
+                    .copied()
+                    .unwrap_or(true)
+            });
+        }
         if options.mode == RolloutMigrationMode::Apply {
             let pending_thread_ids = pending_migration_thread_ids(&self.config.codex_home).await?;
             paths.sort_by_key(|path| {

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

@@ -3173,6 +3173,76 @@ async fn migration_skips_empty_rollout_files() {
     );
 }
 
+#[tokio::test]
+async fn selected_migration_prunes_unrelated_indexed_rollouts() {
+    let home = TempDir::new().expect("create Codex home");
+    let owner = ThreadId::new();
+    let original = write_rollout(home.path(), owner, SessionSource::Cli, vec![]);
+    let directory = original.parent().expect("rollout directory");
+    let legacy = directory.join("rollout-imported.jsonl");
+    let reverted = directory.join(format!(
+        "rollout-2025-01-03T12-00-00-{owner}_{}.jsonl",
+        ThreadId::new()
+    ));
+    fs::copy(&original, &legacy).expect("copy legacy rollout");
+    fs::rename(&original, &reverted).expect("rename reverted rollout");
+    let reverted = compress_rollout(&reverted);
+    let other = write_rollout(home.path(), ThreadId::new(), SessionSource::Cli, vec![]);
+    fs::copy(&other, &original).expect("put another owner's metadata under the selected filename");
+    let archived_owner = ThreadId::new();
+    let archived_reverted = directory.join(format!(
+        "rollout-2025-01-03T12-00-00-{archived_owner}_{}.jsonl",
+        ThreadId::new()
+    ));
+    let archived = write_rollout(home.path(), archived_owner, SessionSource::Cli, vec![]);
+    fs::rename(archived, &archived_reverted).expect("rename unrelated reverted rollout");
+    let archived_reverted = move_to_archived(home.path(), archived_reverted);
+    let misnamed = directory.join(format!(
+        "rollout-2025-01-03T12-00-00-{}.jsonl",
+        ThreadId::new()
+    ));
+    fs::copy(&legacy, &misnamed).expect("copy under a different canonical thread ID");
+    let config = test_config(home.path());
+    let db = codex_state::StateRuntime::init(config.sqlite.clone(), "test-provider".to_string())
+        .await
+        .expect("initialize SQLite");
+    for path in [&other, &archived_reverted, &reverted, &misnamed] {
+        let mut row = codex_rollout::builder_from_items(&[], path)
+            .expect("read indexed filename")
+            .build("test-provider");
+        if path == &misnamed {
+            row.id = ThreadId::new();
+        }
+        db.upsert_thread(&row).await.expect("index rollout");
+    }
+    db.close().await;
+    // SQLite still points to the plain file after compression.
+    compress_rollout(&archived_reverted);
+    let store = LocalThreadStore::new(config, /*state_db*/ None);
+    let mut inspected = 0;
+    let report = store
+        .migrate_rollouts_with_progress(
+            RolloutMigrationOptions {
+                thread_ids: vec![owner],
+                ..Default::default()
+            },
+            |_| inspected += 1,
+        )
+        .await
+        .expect("inspect selected rollouts");
+    let mut actual = report
+        .outcomes
+        .into_iter()
+        .map(|outcome| outcome.rollout_path)
+        .collect::<Vec<_>>();
+    let mut expected = vec![legacy, reverted, misnamed];
+    actual.sort();
+    expected.sort();
+    assert_eq!(actual, expected);
+    // The unindexed file with another owner's header also needs inspection.
+    assert_eq!(inspected, expected.len() + 1);
+}
+
 #[tokio::test]
 async fn migration_reports_missing_sqlite_metadata() {
     let home = TempDir::new().expect("create Codex home");