diff --git a/codex-rs/rollout/src/lib.rs b/codex-rs/rollout/src/lib.rs index 23be07996c..29aa35f930 100644 --- a/codex-rs/rollout/src/lib.rs +++ b/codex-rs/rollout/src/lib.rs @@ -48,6 +48,7 @@ pub use list::ThreadListConfig; pub use list::ThreadListLayout; pub use list::ThreadSortKey; pub use list::ThreadsPage; +pub use list::find_any_thread_path_by_id_str; pub use list::find_archived_thread_path_by_id_str; pub use list::find_thread_path_by_id_str; #[deprecated(note = "use find_thread_path_by_id_str")] diff --git a/codex-rs/rollout/src/list.rs b/codex-rs/rollout/src/list.rs index d30bf9c7fc..9cd1005ba6 100644 --- a/codex-rs/rollout/src/list.rs +++ b/codex-rs/rollout/src/list.rs @@ -1766,9 +1766,9 @@ async fn find_thread_path_by_id_str_in_subdir( state_db_ctx: Option<&codex_state::StateRuntime>, ) -> io::Result> { // Validate UUID format early. - if Uuid::parse_str(id_str).is_err() { + let Ok(target_uuid) = Uuid::parse_str(id_str) else { return Ok(None); - } + }; // Prefer DB lookup, then fall back to rollout file search. // TODO(jif): sqlite migration phase 1 @@ -1778,19 +1778,23 @@ async fn find_thread_path_by_id_str_in_subdir( _ => None, }; let thread_id = ThreadId::from_string(id_str).ok(); + let mut state_metadata_path = None; let mut unverified_db_path = None; let mut fallback_reason = state_db_ctx.is_none().then_some("db_unavailable"); if let Some(state_db_ctx) = state_db_ctx && let Some(thread_id) = thread_id { match state_db_ctx - .find_rollout_path_by_id(thread_id, archived_only) + .find_rollout_path_and_created_at_by_id(thread_id, archived_only) .await { - Ok(Some(db_path)) => { + Ok(Some((db_path, created_at))) => { if let Some(existing_db_path) = - compression::existing_rollout_path(db_path.as_path()).await + compression::existing_rollout_path_blocking(db_path.as_path()) { + if rollout_file_name_matches_uuid(existing_db_path.as_path(), target_uuid) { + return Ok(Some(existing_db_path)); + } match read_session_meta_line(&existing_db_path).await { Ok(meta_line) if meta_line.meta.id == thread_id => { return Ok(Some(existing_db_path)); @@ -1809,6 +1813,17 @@ async fn find_thread_path_by_id_str_in_subdir( "mismatch", /*telemetry_override*/ None, ); + state_metadata_path = find_rollout_path_by_id_from_state_created_at( + codex_home, + subdir, + target_uuid, + thread_id, + archived_only, + state_db_ctx, + db_path.as_path(), + created_at, + ) + .await; } Err(err) => { tracing::debug!( @@ -1816,6 +1831,7 @@ async fn find_thread_path_by_id_str_in_subdir( existing_db_path.display() ); unverified_db_path = Some(existing_db_path); + fallback_reason = Some("unverified_path"); } } } else { @@ -1831,6 +1847,17 @@ async fn find_thread_path_by_id_str_in_subdir( "stale_path", /*telemetry_override*/ None, ); + state_metadata_path = find_rollout_path_by_id_from_state_created_at( + codex_home, + subdir, + target_uuid, + thread_id, + archived_only, + state_db_ctx, + db_path.as_path(), + created_at, + ) + .await; } } Ok(None) => fallback_reason = Some("missing_row"), @@ -1843,61 +1870,13 @@ async fn find_thread_path_by_id_str_in_subdir( } } - let mut root = codex_home.to_path_buf(); - root.push(subdir); - if !root.exists() { - return Ok(unverified_db_path); - } - let (filename_match, filename_scan_error) = match find_rollout_path_by_id_from_filenames( - root.as_path(), - id_str, - ) - .await - { - Ok(path) => (path, None), - Err(err) => { - tracing::warn!( - "rollout filename lookup failed during find_thread_path_by_id_str_in_subdir: {err}" - ); - (None, Some(err)) - } - }; - - let found = match filename_match { - Some(path) => Some(path), - None => { - // This is safe because we know the values are valid. - #[allow(clippy::unwrap_used)] - let limit = NonZero::new(1).unwrap(); - let options = file_search::FileSearchOptions { - limit, - compute_indices: false, - respect_gitignore: false, - ..Default::default() - }; - - let results = file_search::run( - id_str, - vec![root.clone()], - options, - /*cancel_flag*/ None, - ) - .map_err(|e| io::Error::other(format!("file search failed: {e}")))?; - - let found = results - .matches - .into_iter() - .map(|m| m.full_path()) - .find_map(compression::RolloutFile::from_path) - .map(compression::RolloutFile::into_path); - - if found.is_none() - && let Some(err) = filename_scan_error - { - return Err(err); - } - found + let mut found_path_repaired = false; + let found = match state_metadata_path { + Some((path, repaired)) => { + found_path_repaired = repaired; + Some(path) } + None => find_thread_path_by_id_str_in_subdir_without_db(codex_home, subdir, id_str).await?, }; if let Some(found_path) = found.as_ref() { tracing::debug!("state db missing rollout path for thread {id_str}"); @@ -1911,18 +1890,294 @@ async fn find_thread_path_by_id_str_in_subdir( /*telemetry_override*/ None, ); } - state_db::read_repair_rollout_path( - state_db_ctx, - thread_id, - archived_only, - found_path.as_path(), - ) - .await; + if !found_path_repaired { + state_db::read_repair_rollout_path( + state_db_ctx, + thread_id, + archived_only, + found_path.as_path(), + ) + .await; + } } Ok(found.or(unverified_db_path)) } +async fn find_thread_path_by_id_str_in_any_subdir( + codex_home: &Path, + id_str: &str, + state_db_ctx: Option<&codex_state::StateRuntime>, +) -> io::Result> { + let Ok(target_uuid) = Uuid::parse_str(id_str) else { + return Ok(None); + }; + let thread_id = ThreadId::from_string(id_str).ok(); + let mut preferred_subdir = None; + let mut unverified_db_path = None; + let mut fallback_reason = state_db_ctx.is_none().then_some("db_unavailable"); + if let Some(state_db_ctx) = state_db_ctx + && let Some(thread_id) = thread_id + { + match state_db_ctx + .find_rollout_path_and_created_at_by_id(thread_id, /*archived_only*/ None) + .await + { + Ok(Some((db_path, created_at))) => { + preferred_subdir = rollout_subdir_hint(codex_home, db_path.as_path()); + if let Some(existing_db_path) = + compression::existing_rollout_path_blocking(db_path.as_path()) + { + if rollout_file_name_matches_uuid(existing_db_path.as_path(), target_uuid) { + return Ok(Some(existing_db_path)); + } + match read_session_meta_line(&existing_db_path).await { + Ok(meta_line) if meta_line.meta.id == thread_id => { + return Ok(Some(existing_db_path)); + } + Ok(meta_line) => { + tracing::error!( + "state db returned rollout path for thread {id_str} but file belongs to thread {}: {}", + meta_line.meta.id, + existing_db_path.display() + ); + tracing::warn!( + "state db discrepancy during find_thread_path_by_id_str_in_any_subdir: mismatched_db_path" + ); + codex_state::record_fallback( + "find_thread_path", + "mismatch", + /*telemetry_override*/ None, + ); + fallback_reason = Some("mismatch"); + } + Err(err) => { + tracing::debug!( + "state db returned rollout path for thread {id_str} that could not be verified: {}: {err}", + existing_db_path.display() + ); + unverified_db_path = Some(existing_db_path); + fallback_reason = Some("unverified_path"); + } + } + } else { + tracing::error!( + "state db returned stale rollout path for thread {id_str}: {}", + db_path.display() + ); + tracing::warn!( + "state db discrepancy during find_thread_path_by_id_str_in_any_subdir: stale_db_path" + ); + codex_state::record_fallback( + "find_thread_path", + "stale_path", + /*telemetry_override*/ None, + ); + fallback_reason = Some("stale_path"); + } + if let Some(subdir) = preferred_subdir + && let Some((path, _repaired)) = find_rollout_path_by_id_from_state_created_at( + codex_home, + subdir, + target_uuid, + thread_id, + archived_only_for_subdir(subdir), + state_db_ctx, + db_path.as_path(), + created_at, + ) + .await + { + return Ok(Some(path)); + } + } + Ok(None) => fallback_reason = Some("missing_row"), + Err(err) => { + tracing::warn!( + "state db find_rollout_path_by_id failed during find_any_path_query: {err}" + ); + fallback_reason = Some("db_error"); + } + } + } + + for (subdir, archived_only) in ordered_session_subdirs(preferred_subdir) { + if let Some(path) = + find_thread_path_by_id_str_in_subdir_without_db(codex_home, subdir, id_str).await? + { + tracing::debug!("state db missing rollout path for thread {id_str}"); + tracing::warn!( + "state db discrepancy during find_thread_path_by_id_str_in_any_subdir: falling_back" + ); + if let Some(reason) = fallback_reason { + codex_state::record_fallback( + "find_thread_path", + reason, + /*telemetry_override*/ None, + ); + } + state_db::read_repair_rollout_path( + state_db_ctx, + thread_id, + archived_only, + path.as_path(), + ) + .await; + return Ok(Some(path)); + } + } + + Ok(unverified_db_path) +} + +pub(crate) async fn find_thread_path_by_id_str_in_subdir_without_db( + codex_home: &Path, + subdir: &str, + id_str: &str, +) -> io::Result> { + let mut root = codex_home.to_path_buf(); + root.push(subdir); + if !root.exists() { + return Ok(None); + } + let (filename_match, filename_scan_error) = match find_rollout_path_by_id_from_filenames( + root.as_path(), + id_str, + ) + .await + { + Ok(path) => (path, None), + Err(err) => { + tracing::warn!( + "rollout filename lookup failed during find_thread_path_by_id_str fallback scan: {err}" + ); + (None, Some(err)) + } + }; + + match filename_match { + Some(path) => Ok(Some(path)), + None => { + #[allow(clippy::unwrap_used)] + let limit = NonZero::new(1).unwrap(); + let options = file_search::FileSearchOptions { + limit, + compute_indices: false, + respect_gitignore: false, + ..Default::default() + }; + + let results = file_search::run(id_str, vec![root], options, /*cancel_flag*/ None) + .map_err(|e| io::Error::other(format!("file search failed: {e}")))?; + + let found = results + .matches + .into_iter() + .map(|m| m.full_path()) + .find_map(compression::RolloutFile::from_path) + .map(compression::RolloutFile::into_path); + + if found.is_none() + && let Some(err) = filename_scan_error + { + return Err(err); + } + Ok(found) + } + } +} + +fn archived_only_for_subdir(subdir: &str) -> Option { + match subdir { + SESSIONS_SUBDIR => Some(false), + ARCHIVED_SESSIONS_SUBDIR => Some(true), + _ => None, + } +} + +fn rollout_subdir_hint(codex_home: &Path, path: &Path) -> Option<&'static str> { + if path.starts_with(codex_home.join(ARCHIVED_SESSIONS_SUBDIR)) { + return Some(ARCHIVED_SESSIONS_SUBDIR); + } + if path.starts_with(codex_home.join(SESSIONS_SUBDIR)) { + return Some(SESSIONS_SUBDIR); + } + None +} + +fn ordered_session_subdirs( + preferred_subdir: Option<&'static str>, +) -> [(&'static str, Option); 2] { + match preferred_subdir { + Some(ARCHIVED_SESSIONS_SUBDIR) => [ + (ARCHIVED_SESSIONS_SUBDIR, Some(true)), + (SESSIONS_SUBDIR, Some(false)), + ], + _ => [ + (SESSIONS_SUBDIR, Some(false)), + (ARCHIVED_SESSIONS_SUBDIR, Some(true)), + ], + } +} + +async fn find_rollout_path_by_id_from_state_created_at( + codex_home: &Path, + subdir: &str, + target_uuid: Uuid, + thread_id: ThreadId, + archived_only: Option, + state_db_ctx: &codex_state::StateRuntime, + stale_path: &Path, + created_at: DateTime, +) -> Option<(PathBuf, bool)> { + let created_at = created_at.with_timezone(&chrono::Local); + let file_name = format!( + "rollout-{:04}-{:02}-{:02}T{:02}-{:02}-{:02}-{target_uuid}.jsonl", + created_at.year(), + created_at.month(), + created_at.day(), + created_at.hour(), + created_at.minute(), + created_at.second(), + ); + let candidate = if subdir == ARCHIVED_SESSIONS_SUBDIR { + codex_home.join(subdir).join(file_name) + } else { + codex_home + .join(subdir) + .join(format!("{:04}", created_at.year())) + .join(format!("{:02}", created_at.month())) + .join(format!("{:02}", created_at.day())) + .join(file_name) + }; + let existing_path = compression::existing_rollout_path_blocking(candidate.as_path())?; + if !rollout_file_name_matches_uuid(existing_path.as_path(), target_uuid) { + return None; + } + let repaired = state_db_ctx + .update_thread_rollout_path_if_current( + thread_id, + stale_path, + existing_path.as_path(), + archived_only, + ) + .await + .unwrap_or_else(|err| { + tracing::warn!( + "state db stale path direct repair failed for thread {thread_id}: {err}" + ); + false + }); + Some((existing_path, repaired)) +} + +fn rollout_file_name_matches_uuid(path: &Path, target: Uuid) -> bool { + path.file_name() + .and_then(OsStr::to_str) + .and_then(parse_timestamp_uuid_from_filename) + .is_some_and(|(_ts, id)| id == target) +} + async fn find_rollout_path_by_id_from_filenames( root: &Path, id_str: &str, @@ -1930,33 +2185,41 @@ async fn find_rollout_path_by_id_from_filenames( let Ok(target) = Uuid::parse_str(id_str) else { return Ok(None); }; + let root = root.to_path_buf(); + tokio::task::spawn_blocking(move || { + find_rollout_path_by_id_from_filenames_blocking(root.as_path(), target) + }) + .await + .map_err(io::Error::other)? +} + +fn find_rollout_path_by_id_from_filenames_blocking( + root: &Path, + target: Uuid, +) -> io::Result> { let mut stack = vec![root.to_path_buf()]; while let Some(dir) = stack.pop() { - let mut read_dir = match tokio::fs::read_dir(dir.as_path()).await { + let read_dir = match std::fs::read_dir(dir.as_path()) { Ok(read_dir) => read_dir, Err(err) if err.kind() == io::ErrorKind::NotFound => continue, Err(err) => return Err(err), }; - while let Some(entry) = read_dir.next_entry().await? { - let path = entry.path(); - let file_type = entry.file_type().await?; + for entry in read_dir { + let entry = entry?; + let file_type = entry.file_type()?; if file_type.is_dir() { + let path = entry.path(); stack.push(path); continue; } if !file_type.is_file() { continue; } - let Some(rollout_file) = compression::RolloutFile::from_path(path) else { - continue; - }; - let Some((_ts, id)) = - parse_timestamp_uuid_from_filename(rollout_file.plain_file_name()) - else { + let Some((_ts, id, path)) = rollout_parts_from_blocking_dir_entry(&entry) else { continue; }; if id == target { - return Ok(Some(rollout_file.into_path())); + return Ok(Some(path)); } } } @@ -1984,6 +2247,15 @@ pub async fn find_archived_thread_path_by_id_str( .await } +/// Locate a recorded thread rollout file by UUID across active and archived session roots. +pub async fn find_any_thread_path_by_id_str( + codex_home: &Path, + id_str: &str, + state_db_ctx: Option<&codex_state::StateRuntime>, +) -> io::Result> { + find_thread_path_by_id_str_in_any_subdir(codex_home, id_str, state_db_ctx).await +} + /// Extract the `YYYY/MM/DD` directory components from a rollout filename. pub fn rollout_date_parts(file_name: &OsStr) -> Option<(String, String, String)> { let name = file_name.to_string_lossy(); diff --git a/codex-rs/rollout/src/recorder_tests.rs b/codex-rs/rollout/src/recorder_tests.rs index 68896ad202..08e7962f39 100644 --- a/codex-rs/rollout/src/recorder_tests.rs +++ b/codex-rs/rollout/src/recorder_tests.rs @@ -706,6 +706,73 @@ async fn list_threads_db_enabled_repairs_stale_rollout_paths() -> std::io::Resul Ok(()) } +#[tokio::test] +async fn state_db_only_list_repairs_stale_rollout_path() -> std::io::Result<()> { + let home = TempDir::new().expect("temp dir"); + let config = test_config(home.path()); + + let uuid = Uuid::from_u128(9016); + let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id"); + let real_path = write_session_file(home.path(), "2025-01-03T17-00-00", uuid)?; + 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(), + ) + .await + .expect("state db should initialize"); + runtime + .mark_backfill_complete(/*last_watermark*/ None) + .await + .expect("backfill should be complete"); + let created_at = chrono::Utc + .with_ymd_and_hms(2025, 1, 3, 17, 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()); + metadata.preview = metadata.first_user_message.clone(); + runtime + .upsert_thread(&metadata) + .await + .expect("state db upsert should succeed"); + + let page = RolloutRecorder::list_threads_from_state_db( + Some(runtime.clone()), + &config, + /*page_size*/ 1, + /*cursor*/ None, + ThreadSortKey::CreatedAt, + SortDirection::Desc, + &[], + /*model_providers*/ None, + /*cwd_filters*/ None, + config.model_provider_id.as_str(), + /*search_term*/ None, + ) + .await?; + assert_eq!(page.items.len(), 1); + assert_eq!(page.items[0].path, real_path); + + let repaired_path = runtime + .find_rollout_path_by_id(thread_id, Some(false)) + .await + .expect("state db lookup should succeed"); + assert_eq!(repaired_path, Some(real_path)); + Ok(()) +} + #[tokio::test] async fn list_threads_state_db_only_skips_jsonl_repair_scan() -> std::io::Result<()> { let home = TempDir::new().expect("temp dir"); diff --git a/codex-rs/rollout/src/state_db.rs b/codex-rs/rollout/src/state_db.rs index 0735ddbc08..d339b59c46 100644 --- a/codex-rs/rollout/src/state_db.rs +++ b/codex-rs/rollout/src/state_db.rs @@ -438,6 +438,10 @@ pub async fn list_threads_db( let mut item = item; item.rollout_path = existing_path; valid_items.push(item); + } else if let Some(repaired_item) = + repair_stale_rollout_path_from_metadata(ctx, codex_home, &item, archived).await + { + valid_items.push(repaired_item); } else { warn!( "state db list_threads returned stale rollout path for thread {}: {}", @@ -458,6 +462,44 @@ pub async fn list_threads_db( } } +async fn repair_stale_rollout_path_from_metadata( + ctx: &codex_state::StateRuntime, + codex_home: &Path, + item: &codex_state::ThreadMetadata, + archived: bool, +) -> Option { + let subdir = if archived { + crate::ARCHIVED_SESSIONS_SUBDIR + } else { + crate::SESSIONS_SUBDIR + }; + let candidate = crate::list::find_thread_path_by_id_str_in_subdir_without_db( + codex_home, + subdir, + &item.id.to_string(), + ) + .await + .ok() + .flatten()?; + let existing_path = crate::compression::existing_rollout_path_blocking(candidate.as_path())?; + + warn!( + "state db list_threads repaired stale rollout path for thread {}: {} -> {}", + item.id, + item.rollout_path.display(), + existing_path.display() + ); + let mut item = item.clone(); + item.rollout_path = existing_path; + if let Err(err) = ctx.upsert_thread(&item).await { + warn!( + "state db list_threads stale path repair upsert failed for thread {}: {err}", + item.id + ); + } + Some(item) +} + /// Look up the rollout path for a thread id using SQLite. pub async fn find_rollout_path_by_id( context: Option<&codex_state::StateRuntime>, @@ -581,28 +623,9 @@ pub async fn read_repair_rollout_path( && let Ok(Some(metadata)) = ctx.get_thread(thread_id).await { saw_existing_metadata = true; - let mut repaired = metadata.clone(); - repaired.rollout_path = rollout_path.to_path_buf(); - repaired.cwd = normalize_cwd_for_state_db(&repaired.cwd); - match archived_only { - Some(true) if repaired.archived_at.is_none() => { - repaired.archived_at = Some(repaired.updated_at); - } - Some(false) => { - repaired.archived_at = None; - } - Some(true) | None => {} - } - if repaired == metadata { - return; - } - warn!("state db discrepancy during read_repair_rollout_path: upsert_needed (fast path)"); - if let Err(err) = ctx.upsert_thread(&repaired).await { - warn!( - "state db read-repair upsert failed for {}: {err}", - rollout_path.display() - ); - } else { + if read_repair_rollout_path_from_metadata(Some(ctx), metadata, archived_only, rollout_path) + .await + { return; } } @@ -629,6 +652,44 @@ pub async fn read_repair_rollout_path( .await; } +/// Repair a rollout path when the caller already has the metadata row. +pub async fn read_repair_rollout_path_from_metadata( + context: Option<&codex_state::StateRuntime>, + mut metadata: codex_state::ThreadMetadata, + archived_only: Option, + rollout_path: &Path, +) -> bool { + let Some(ctx) = context else { + return true; + }; + + let original = metadata.clone(); + metadata.rollout_path = rollout_path.to_path_buf(); + metadata.cwd = normalize_cwd_for_state_db(&metadata.cwd); + 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 metadata == original { + return true; + } + + warn!("state db discrepancy during read_repair_rollout_path: upsert_needed (fast path)"); + if let Err(err) = ctx.upsert_thread(&metadata).await { + warn!( + "state db read-repair upsert failed for {}: {err}", + rollout_path.display() + ); + return false; + } + true +} + /// Apply rollout items incrementally to SQLite. #[allow(clippy::too_many_arguments)] pub async fn apply_rollout_items( diff --git a/codex-rs/rollout/src/tests.rs b/codex-rs/rollout/src/tests.rs index 3c1fc8bb57..4a122a637f 100644 --- a/codex-rs/rollout/src/tests.rs +++ b/codex-rs/rollout/src/tests.rs @@ -7,6 +7,7 @@ use std::fs::File; use std::fs::FileTimes; use std::io::Write; use std::path::Path; +use std::path::PathBuf; use chrono::TimeZone; use pretty_assertions::assert_eq; @@ -19,6 +20,7 @@ use time::macros::format_description; use uuid::Uuid; use crate::INTERACTIVE_SESSION_SOURCES; +use crate::find_any_thread_path_by_id_str; use crate::find_thread_path_by_id_str; use crate::list::Cursor; use crate::list::ThreadItem; @@ -230,6 +232,85 @@ async fn find_thread_path_accepts_existing_state_db_path_without_canonical_filen assert_eq!(found, Some(db_rollout_path)); } +#[tokio::test] +async fn find_any_thread_path_prefers_archived_state_db_path() { + let temp = TempDir::new().unwrap(); + let home = temp.path(); + let uuid = Uuid::from_u128(306); + let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id"); + let active_ts = "2025-01-03T13-00-00"; + write_session_file( + home, + active_ts, + uuid, + /*num_records*/ 1, + Some(SessionSource::Cli), + ) + .unwrap(); + + let archived_ts = "2025-01-04T13-00-00"; + let archived_rollout_path = write_archived_session_file( + home, + archived_ts, + uuid, + /*num_records*/ 1, + Some(SessionSource::Cli), + ) + .unwrap(); + let runtime = insert_state_db_thread( + home, + thread_id, + archived_rollout_path.as_path(), + /*archived*/ true, + ) + .await; + + let found = find_any_thread_path_by_id_str(home, &uuid.to_string(), Some(runtime.as_ref())) + .await + .expect("lookup should succeed"); + assert_eq!(found, Some(archived_rollout_path)); +} + +#[tokio::test] +async fn find_any_thread_path_repairs_stale_archived_db_path() { + let temp = TempDir::new().unwrap(); + let home = temp.path(); + let uuid = Uuid::from_u128(307); + let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id"); + let ts = "2025-01-03T12-00-00"; + let archived_rollout_path = write_archived_session_file( + home, + ts, + uuid, + /*num_records*/ 1, + Some(SessionSource::Cli), + ) + .unwrap(); + + let stale_db_path = home.join(format!( + "archived_sessions/rollout-2099-01-01T00-00-00-{uuid}.jsonl" + )); + let runtime = insert_state_db_thread( + home, + thread_id, + stale_db_path.as_path(), + /*archived*/ true, + ) + .await; + + let found = find_any_thread_path_by_id_str(home, &uuid.to_string(), Some(runtime.as_ref())) + .await + .expect("lookup should succeed"); + assert_eq!(found, Some(archived_rollout_path.clone())); + assert_state_db_rollout_path_with_archive( + home, + thread_id, + Some(true), + Some(archived_rollout_path.as_path()), + ) + .await; +} + #[test] fn rollout_date_parts_extracts_directory_components() { let file_name = OsStr::new("rollout-2025-03-01T09-00-00-123.jsonl"); @@ -244,12 +325,21 @@ async fn assert_state_db_rollout_path( home: &Path, thread_id: ThreadId, expected_path: Option<&Path>, +) { + assert_state_db_rollout_path_with_archive(home, thread_id, Some(false), expected_path).await; +} + +async fn assert_state_db_rollout_path_with_archive( + home: &Path, + thread_id: ThreadId, + archived_only: Option, + expected_path: Option<&Path>, ) { let runtime = codex_state::StateRuntime::init(home.to_path_buf(), TEST_PROVIDER.to_string()) .await .expect("state db should initialize"); let path = runtime - .find_rollout_path_by_id(thread_id, Some(false)) + .find_rollout_path_by_id(thread_id, archived_only) .await .expect("state db lookup should succeed"); assert_eq!(path.as_deref(), expected_path); @@ -343,6 +433,58 @@ fn write_session_file_with_provider( Ok((dt, uuid)) } +fn write_archived_session_file( + root: &Path, + ts_str: &str, + uuid: Uuid, + num_records: usize, + source: Option, +) -> std::io::Result { + let format: &[FormatItem] = + format_description!("[year]-[month]-[day]T[hour]-[minute]-[second]"); + let dt = PrimitiveDateTime::parse(ts_str, format) + .unwrap() + .assume_utc(); + let dir = root.join("archived_sessions"); + fs::create_dir_all(&dir)?; + + let filename = format!("rollout-{ts_str}-{uuid}.jsonl"); + let file_path = dir.join(filename); + let mut file = File::create(&file_path)?; + + let mut payload = serde_json::json!({ + "id": uuid, + "timestamp": ts_str, + "cwd": ".", + "originator": "test_originator", + "cli_version": "test_version", + "base_instructions": null, + }); + + if let Some(source) = source { + payload["source"] = serde_json::to_value(source).unwrap(); + } + payload["model_provider"] = serde_json::Value::String(TEST_PROVIDER.to_string()); + + let meta = serde_json::json!({ + "timestamp": ts_str, + "type": "session_meta", + "payload": payload, + }); + writeln!(file, "{meta}")?; + + for i in 0..num_records { + let rec = serde_json::json!({ + "record_type": "response", + "index": i + }); + writeln!(file, "{rec}")?; + } + let times = FileTimes::new().set_modified(dt.into()); + file.set_times(times)?; + Ok(file_path) +} + fn write_goal_started_session_file( root: &Path, ts_str: &str, diff --git a/codex-rs/state/src/runtime/threads.rs b/codex-rs/state/src/runtime/threads.rs index 4a90b9ad4e..9da54489ce 100644 --- a/codex-rs/state/src/runtime/threads.rs +++ b/codex-rs/state/src/runtime/threads.rs @@ -348,6 +348,64 @@ ON CONFLICT(child_thread_id) DO NOTHING .map(PathBuf::from)) } + /// Find a rollout path and creation timestamp by thread id using the underlying database. + pub async fn find_rollout_path_and_created_at_by_id( + &self, + id: ThreadId, + archived_only: Option, + ) -> anyhow::Result)>> { + let mut builder = QueryBuilder::::new( + "SELECT rollout_path, created_at_ms AS created_at FROM threads WHERE id = ", + ); + builder.push_bind(id.to_string()); + match archived_only { + Some(true) => { + builder.push(" AND archived = 1"); + } + Some(false) => { + builder.push(" AND archived = 0"); + } + None => {} + } + let row = builder.build().fetch_optional(self.pool.as_ref()).await?; + row.map(|row| { + let rollout_path: String = row.try_get("rollout_path")?; + let created_at: i64 = row.try_get("created_at")?; + Ok(( + PathBuf::from(rollout_path), + epoch_millis_to_datetime(created_at)?, + )) + }) + .transpose() + } + + /// Repair a stale rollout path in place when the row still points at the expected path. + pub async fn update_thread_rollout_path_if_current( + &self, + id: ThreadId, + expected_rollout_path: &Path, + repaired_rollout_path: &Path, + archived_only: Option, + ) -> anyhow::Result { + let mut builder = QueryBuilder::::new("UPDATE threads SET rollout_path = "); + builder.push_bind(repaired_rollout_path.display().to_string()); + builder.push(" WHERE id = "); + builder.push_bind(id.to_string()); + builder.push(" AND rollout_path = "); + builder.push_bind(expected_rollout_path.display().to_string()); + match archived_only { + Some(true) => { + builder.push(" AND archived = 1"); + } + Some(false) => { + builder.push(" AND archived = 0"); + } + None => {} + } + let result = builder.build().execute(self.pool.as_ref()).await?; + Ok(result.rows_affected() > 0) + } + /// Find the newest thread whose user-facing title exactly matches `title`. #[allow(clippy::too_many_arguments)] pub async fn find_thread_by_exact_title( diff --git a/codex-rs/thread-store/src/local/read_thread.rs b/codex-rs/thread-store/src/local/read_thread.rs index 4778dd69d9..9f98cdd67d 100644 --- a/codex-rs/thread-store/src/local/read_thread.rs +++ b/codex-rs/thread-store/src/local/read_thread.rs @@ -5,7 +5,7 @@ use codex_protocol::protocol::AskForApproval; use codex_protocol::protocol::SessionMetaLine; use codex_protocol::protocol::SessionSource; use codex_rollout::RolloutRecorder; -use codex_rollout::find_archived_thread_path_by_id_str; +use codex_rollout::find_any_thread_path_by_id_str; use codex_rollout::find_thread_name_by_id; use codex_rollout::find_thread_path_by_id_str; use codex_rollout::read_session_meta_line; @@ -38,35 +38,34 @@ pub(super) async fn read_thread( store.config.codex_home.as_path(), metadata.rollout_path.as_path(), ))) - && (!params.include_history - || sqlite_rollout_path_can_load_history_for_thread( - store, - &metadata.rollout_path, - thread_id, - ) - .await) { let metadata_sandbox_policy = metadata.sandbox_policy.clone(); + let rollout_thread = matching_rollout_thread_from_sqlite_path( + store, + metadata.rollout_path.as_path(), + thread_id, + params.include_archived, + ) + .await; let mut thread = stored_thread_from_sqlite_metadata(store, metadata).await; - if !params.include_history - && let Some(rollout_path) = thread.rollout_path.clone() - && let Ok(mut rollout_thread) = read_thread_from_rollout_path(store, rollout_path).await - && rollout_thread.thread_id == thread_id - && (params.include_archived || rollout_thread.archived_at.is_none()) - && !rollout_thread.preview.is_empty() - { - if thread.name.is_some() { - rollout_thread.name = thread.name; + if !params.include_history || rollout_thread.is_some() { + if !params.include_history + && let Some(mut rollout_thread) = rollout_thread + && !rollout_thread.preview.is_empty() + { + if thread.name.is_some() { + rollout_thread.name = thread.name; + } + rollout_thread.git_info = thread.git_info; + rollout_thread.permission_profile = permission_profile_from_metadata_value( + &metadata_sandbox_policy, + rollout_thread.cwd.as_path(), + ); + thread = rollout_thread; } - rollout_thread.git_info = thread.git_info; - rollout_thread.permission_profile = permission_profile_from_metadata_value( - &metadata_sandbox_policy, - rollout_thread.cwd.as_path(), - ); - thread = rollout_thread; + attach_history_if_requested(&mut thread, params.include_history).await?; + return Ok(thread); } - attach_history_if_requested(&mut thread, params.include_history).await?; - return Ok(thread); } let path = resolve_rollout_path(store, thread_id, params.include_archived) @@ -85,20 +84,21 @@ pub(super) async fn read_thread( Ok(thread) } -async fn sqlite_rollout_path_can_load_history_for_thread( +async fn matching_rollout_thread_from_sqlite_path( store: &LocalThreadStore, path: &std::path::Path, thread_id: codex_protocol::ThreadId, -) -> bool { - if codex_rollout::existing_rollout_path(path).await.is_none() { - return false; - } - // SQLite metadata can outlive a moved/recreated rollout path. When history is - // requested, verify the path still resolves to the requested thread before - // trusting it as the source replay. + include_archived: bool, +) -> Option { + codex_rollout::existing_rollout_path(path).await.as_ref()?; + // SQLite metadata can outlive a moved or recreated rollout path. Verify the + // path still resolves to the requested thread before trusting its summary or history. read_thread_from_rollout_path(store, path.to_path_buf()) .await - .is_ok_and(|thread| thread.thread_id == thread_id) + .ok() + .filter(|thread| { + thread.thread_id == thread_id && (include_archived || thread.archived_at.is_none()) + }) } pub(super) async fn read_thread_by_rollout_path( @@ -209,7 +209,7 @@ async fn resolve_rollout_path( let state_db_ctx = store.state_db().await; if include_archived { - match find_thread_path_by_id_str( + find_any_thread_path_by_id_str( store.config.codex_home.as_path(), &thread_id.to_string(), state_db_ctx.as_deref(), @@ -217,18 +217,7 @@ async fn resolve_rollout_path( .await .map_err(|err| ThreadStoreError::InvalidRequest { message: format!("failed to locate thread id {thread_id}: {err}"), - })? { - Some(path) => Ok(Some(path)), - None => find_archived_thread_path_by_id_str( - store.config.codex_home.as_path(), - &thread_id.to_string(), - state_db_ctx.as_deref(), - ) - .await - .map_err(|err| ThreadStoreError::InvalidRequest { - message: format!("failed to locate archived thread id {thread_id}: {err}"), - }), - } + }) } else { find_thread_path_by_id_str( store.config.codex_home.as_path(),