From adad23f7434220f3b05eaf759685268931d49af7 Mon Sep 17 00:00:00 2001 From: jif-oai Date: Thu, 12 Feb 2026 12:49:31 +0000 Subject: [PATCH] Ensure list_threads drops stale rollout files (#11572) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Summary - trim `state_db::list_threads_db` results to entries whose rollout files still exist, logging and recording a discrepancy for dropped rows - delete stale metadata rows from the SQLite store so future calls don’t surface invalid paths - add regression coverage in `recorder.rs` to verify stale DB paths are dropped when the file is missing --- codex-rs/core/src/rollout/recorder.rs | 65 +++++++++++++++++++++++++++ codex-rs/core/src/state_db.rs | 22 ++++++++- codex-rs/state/src/runtime.rs | 9 ++++ 3 files changed, 95 insertions(+), 1 deletion(-) diff --git a/codex-rs/core/src/rollout/recorder.rs b/codex-rs/core/src/rollout/recorder.rs index a41f15fcd..4e94d0356 100644 --- a/codex-rs/core/src/rollout/recorder.rs +++ b/codex-rs/core/src/rollout/recorder.rs @@ -1115,6 +1115,71 @@ mod tests { Ok(()) } + #[tokio::test] + async fn list_threads_db_enabled_drops_missing_rollout_paths() -> std::io::Result<()> { + let home = TempDir::new().expect("temp dir"); + let mut config = ConfigBuilder::default() + .codex_home(home.path().to_path_buf()) + .build() + .await?; + config.features.enable(Feature::Sqlite); + + let uuid = Uuid::from_u128(9010); + let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id"); + let stale_path = home.path().join(format!( + "sessions/2099/01/01/rollout-2099-01-01T00-00-00-{uuid}.jsonl" + )); + + let runtime = codex_state::StateRuntime::init( + home.path().to_path_buf(), + config.model_provider_id.clone(), + None, + ) + .await + .expect("state db should initialize"); + runtime + .mark_backfill_complete(None) + .await + .expect("backfill should be complete"); + let created_at = chrono::Utc + .with_ymd_and_hms(2025, 1, 3, 13, 0, 0) + .single() + .expect("valid datetime"); + let mut builder = codex_state::ThreadMetadataBuilder::new( + thread_id, + stale_path, + created_at, + SessionSource::Cli, + ); + builder.model_provider = Some(config.model_provider_id.clone()); + builder.cwd = home.path().to_path_buf(); + let mut metadata = builder.build(config.model_provider_id.as_str()); + metadata.first_user_message = Some("Hello from user".to_string()); + runtime + .upsert_thread(&metadata) + .await + .expect("state db upsert should succeed"); + + let default_provider = config.model_provider_id.clone(); + let page = RolloutRecorder::list_threads( + &config, + 10, + None, + ThreadSortKey::CreatedAt, + &[], + None, + default_provider.as_str(), + ) + .await?; + assert_eq!(page.items.len(), 0); + let stored_path = runtime + .find_rollout_path_by_id(thread_id, Some(false)) + .await + .expect("state db lookup should succeed"); + assert_eq!(stored_path, None); + Ok(()) + } + #[tokio::test] async fn list_threads_db_enabled_repairs_stale_rollout_paths() -> std::io::Result<()> { let home = TempDir::new().expect("temp dir"); diff --git a/codex-rs/core/src/state_db.rs b/codex-rs/core/src/state_db.rs index 82f0532f5..0c5a14a51 100644 --- a/codex-rs/core/src/state_db.rs +++ b/codex-rs/core/src/state_db.rs @@ -260,7 +260,27 @@ pub async fn list_threads_db( ) .await { - Ok(page) => Some(page), + Ok(mut page) => { + let mut valid_items = Vec::with_capacity(page.items.len()); + for item in page.items { + if tokio::fs::try_exists(&item.rollout_path) + .await + .unwrap_or(false) + { + valid_items.push(item); + } else { + warn!( + "state db list_threads returned stale rollout path for thread {}: {}", + item.id, + item.rollout_path.display() + ); + record_discrepancy("list_threads_db", "stale_db_path_dropped"); + let _ = ctx.delete_thread(item.id).await; + } + } + page.items = valid_items; + Some(page) + } Err(err) => { warn!("state db list_threads failed: {err}"); None diff --git a/codex-rs/state/src/runtime.rs b/codex-rs/state/src/runtime.rs index 838af81fd..d698bd545 100644 --- a/codex-rs/state/src/runtime.rs +++ b/codex-rs/state/src/runtime.rs @@ -650,6 +650,15 @@ ON CONFLICT(thread_id, position) DO NOTHING self.upsert_thread(&metadata).await } + /// Delete a thread metadata row by id. + pub async fn delete_thread(&self, thread_id: ThreadId) -> anyhow::Result { + let result = sqlx::query("DELETE FROM threads WHERE id = ?") + .bind(thread_id.to_string()) + .execute(self.pool.as_ref()) + .await?; + Ok(result.rows_affected()) + } + async fn ensure_backfill_state_row(&self) -> anyhow::Result<()> { sqlx::query( r#"