From dddc3c0a79528c4de8d9a20d51dd3bd89b02a3e4 Mon Sep 17 00:00:00 2001 From: Charlie Marsh Date: Tue, 27 Oct 2026 15:11:44 +0000 Subject: [PATCH] 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 | 1 + codex-rs/rollout/src/metadata.rs | 6 ++ codex-rs/state/src/audit.rs | 19 ++++- codex-rs/state/src/lib.rs | 1 + .../src/local/rollout_migration.rs | 23 ++++++ .../src/local/rollout_migration_tests.rs | 70 +++++++++++++++++++ 6 files changed, 119 insertions(+), 1 deletion(-) diff --git a/codex-rs/rollout/src/lib.rs b/codex-rs/rollout/src/lib.rs index 66aac2dd24..88feb507b1 100644 --- a/codex-rs/rollout/src/lib.rs +++ b/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; diff --git a/codex-rs/rollout/src/metadata.rs b/codex-rs/rollout/src/metadata.rs index c2469d0d70..93f967217b 100644 --- a/codex-rs/rollout/src/metadata.rs +++ b/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 { + Some(RolloutFileName::parse(path.file_name()?.to_str()?)?.thread_id()) +} + /// Returns the rollout ID encoded in a canonical rollout filename. /// /// Normal rollouts use `rollout--.jsonl`, where the thread ID and rollout ID diff --git a/codex-rs/state/src/audit.rs b/codex-rs/state/src/audit.rs index 57ef9f0673..8bc0ec3fed 100644 --- a/codex-rs/state/src/audit.rs +++ b/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> { + // 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() +} diff --git a/codex-rs/state/src/lib.rs b/codex-rs/state/src/lib.rs index 5c5ecad54b..4c9960624f 100644 --- a/codex-rs/state/src/lib.rs +++ b/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; diff --git a/codex-rs/thread-store/src/local/rollout_migration.rs b/codex-rs/thread-store/src/local/rollout_migration.rs index 3f54ec1fdb..412c5823fb 100644 --- a/codex-rs/thread-store/src/local/rollout_migration.rs +++ b/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::>(); + 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| { diff --git a/codex-rs/thread-store/src/local/rollout_migration_tests.rs b/codex-rs/thread-store/src/local/rollout_migration_tests.rs index 8179691899..a609981196 100644 --- a/codex-rs/thread-store/src/local/rollout_migration_tests.rs +++ b/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::>(); + 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");