diff --git a/codex-rs/core/src/rollout/list.rs b/codex-rs/core/src/rollout/list.rs index 2cac898d4..092bb22c6 100644 --- a/codex-rs/core/src/rollout/list.rs +++ b/codex-rs/core/src/rollout/list.rs @@ -1088,9 +1088,10 @@ async fn find_thread_path_by_id_str_in_subdir( ARCHIVED_SESSIONS_SUBDIR => Some(true), _ => None, }; + let thread_id = ThreadId::from_string(id_str).ok(); let state_db_ctx = state_db::open_if_present(codex_home, "").await; if let Some(state_db_ctx) = state_db_ctx.as_deref() - && let Ok(thread_id) = ThreadId::from_string(id_str) + && let Some(thread_id) = thread_id && let Some(db_path) = state_db::find_rollout_path_by_id( Some(state_db_ctx), thread_id, @@ -1128,9 +1129,16 @@ async fn find_thread_path_by_id_str_in_subdir( .map_err(|e| io::Error::other(format!("file search failed: {e}")))?; let found = results.matches.into_iter().next().map(|m| m.full_path()); - if found.is_some() { + if let Some(found_path) = found.as_ref() { tracing::error!("state db missing rollout path for thread {id_str}"); state_db::record_discrepancy("find_thread_path_by_id_str_in_subdir", "falling_back"); + state_db::read_repair_rollout_path( + state_db_ctx.as_deref(), + thread_id, + archived_only, + found_path.as_path(), + ) + .await; } Ok(found) diff --git a/codex-rs/core/src/rollout/recorder.rs b/codex-rs/core/src/rollout/recorder.rs index 89f8829a2..0817577b2 100644 --- a/codex-rs/core/src/rollout/recorder.rs +++ b/codex-rs/core/src/rollout/recorder.rs @@ -568,6 +568,7 @@ async fn rollout_writer( default_provider.as_str(), state_builder.as_ref(), std::slice::from_ref(&rollout_item), + None, ) .await; } diff --git a/codex-rs/core/src/rollout/tests.rs b/codex-rs/core/src/rollout/tests.rs index 89df2897d..c4d93415b 100644 --- a/codex-rs/core/src/rollout/tests.rs +++ b/codex-rs/core/src/rollout/tests.rs @@ -250,7 +250,31 @@ async fn find_thread_path_falls_back_when_db_path_is_stale() { let found = crate::rollout::find_thread_path_by_id_str(home, &uuid.to_string()) .await .expect("lookup should succeed"); - assert_eq!(found, Some(fs_rollout_path)); + assert_eq!(found, Some(fs_rollout_path.clone())); + assert_state_db_rollout_path(home, thread_id, Some(fs_rollout_path.as_path())).await; +} + +#[tokio::test] +async fn find_thread_path_repairs_missing_db_row_after_filesystem_fallback() { + let temp = TempDir::new().unwrap(); + let home = temp.path(); + let uuid = Uuid::from_u128(303); + let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id"); + let ts = "2025-01-03T13-00-00"; + write_session_file(home, ts, uuid, 1, Some(SessionSource::Cli)).unwrap(); + let fs_rollout_path = home.join(format!("sessions/2025/01/03/rollout-{ts}-{uuid}.jsonl")); + + // Create an empty state DB so lookup takes the DB-first path and then falls back to files. + let _runtime = + codex_state::StateRuntime::init(home.to_path_buf(), TEST_PROVIDER.to_string(), None) + .await + .expect("state db should initialize"); + + let found = crate::rollout::find_thread_path_by_id_str(home, &uuid.to_string()) + .await + .expect("lookup should succeed"); + assert_eq!(found, Some(fs_rollout_path.clone())); + assert_state_db_rollout_path(home, thread_id, Some(fs_rollout_path.as_path())).await; } #[test] @@ -263,6 +287,22 @@ fn rollout_date_parts_extracts_directory_components() { ); } +async fn assert_state_db_rollout_path( + home: &Path, + thread_id: ThreadId, + expected_path: Option<&Path>, +) { + let runtime = + codex_state::StateRuntime::init(home.to_path_buf(), TEST_PROVIDER.to_string(), None) + .await + .expect("state db should initialize"); + let path = runtime + .find_rollout_path_by_id(thread_id, Some(false)) + .await + .expect("state db lookup should succeed"); + assert_eq!(path.as_deref(), expected_path); +} + fn write_session_file( root: &Path, ts_str: &str, diff --git a/codex-rs/core/src/state_db.rs b/codex-rs/core/src/state_db.rs index 4ea1435ea..d67a06531 100644 --- a/codex-rs/core/src/state_db.rs +++ b/codex-rs/core/src/state_db.rs @@ -349,6 +349,7 @@ pub async fn reconcile_rollout( default_provider: &str, builder: Option<&ThreadMetadataBuilder>, items: &[RolloutItem], + archived_only: Option, ) { let Some(ctx) = context else { return; @@ -376,7 +377,17 @@ pub async fn reconcile_rollout( return; } }; - if let Err(err) = ctx.upsert_thread(&outcome.metadata).await { + let mut metadata = outcome.metadata; + match archived_only { + Some(true) if metadata.archived_at.is_none() => { + metadata.archived_at = Some(metadata.updated_at); + } + Some(false) => { + metadata.archived_at = None; + } + Some(true) | None => {} + } + if let Err(err) = ctx.upsert_thread(&metadata).await { warn!( "state db reconcile_rollout upsert failed {}: {err}", rollout_path.display() @@ -399,6 +410,56 @@ pub async fn reconcile_rollout( } } +/// Repair a thread's rollout path after filesystem fallback succeeds. +pub async fn read_repair_rollout_path( + context: Option<&codex_state::StateRuntime>, + thread_id: Option, + archived_only: Option, + rollout_path: &Path, +) { + let Some(ctx) = context else { + return; + }; + + if let Some(thread_id) = thread_id + && let Ok(Some(mut metadata)) = ctx.get_thread(thread_id).await + { + metadata.rollout_path = rollout_path.to_path_buf(); + match archived_only { + Some(true) if metadata.archived_at.is_none() => { + metadata.archived_at = Some(metadata.updated_at); + } + Some(false) => { + metadata.archived_at = None; + } + Some(true) | None => {} + } + if let Err(err) = ctx.upsert_thread(&metadata).await { + warn!( + "state db read-repair upsert failed for {}: {err}", + rollout_path.display() + ); + } else { + return; + } + } + + let default_provider = crate::rollout::list::read_session_meta_line(rollout_path) + .await + .ok() + .and_then(|meta| meta.meta.model_provider) + .unwrap_or_default(); + reconcile_rollout( + Some(ctx), + rollout_path, + default_provider.as_str(), + None, + &[], + archived_only, + ) + .await; +} + /// Apply rollout items incrementally to SQLite. pub async fn apply_rollout_items( context: Option<&codex_state::StateRuntime>,