feat: find_thread_path_by_id_str_in_subdir from DB (#10532)

codex-rs/core/src/rollout/list.rs

@@ -1074,6 +1074,29 @@ async fn find_thread_path_by_id_str_in_subdir(
         return Ok(None);
     }
 
+    // Prefer DB lookup, then fall back to rollout file search.
+    // TODO(jif): sqlite migration phase 1
+    let archived_only = match subdir {
+        SESSIONS_SUBDIR => Some(false),
+        ARCHIVED_SESSIONS_SUBDIR => Some(true),
+        _ => None,
+    };
+    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 db_path = state_db::find_rollout_path_by_id(
+            Some(state_db_ctx),
+            thread_id,
+            archived_only,
+            "find_path_query",
+        )
+        .await;
+        if db_path.is_some() {
+            return Ok(db_path);
+        }
+    }
+
     let mut root = codex_home.to_path_buf();
     root.push(subdir);
     if !root.exists() {
@@ -1093,33 +1116,11 @@ 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());
-
-    // Checking if DB is at parity.
-    // TODO(jif): sqlite migration phase 1
-    let archived_only = match subdir {
-        SESSIONS_SUBDIR => Some(false),
-        ARCHIVED_SESSIONS_SUBDIR => Some(true),
-        _ => None,
-    };
-    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 db_path = state_db::find_rollout_path_by_id(
-            Some(state_db_ctx),
-            thread_id,
-            archived_only,
-            "find_path_query",
-        )
-        .await;
-        let canonical_path = found.as_deref();
-        if db_path.as_deref() != canonical_path {
-            tracing::warn!(
-                "state db path mismatch for thread {thread_id:?}: canonical={canonical_path:?} db={db_path:?}"
-            );
-            state_db::record_discrepancy("find_thread_path_by_id_str_in_subdir", "path_mismatch");
-        }
+    if found.is_some() {
+        tracing::error!("state db missing rollout path for thread {id_str}");
+        state_db::record_discrepancy("find_thread_path_by_id_str_in_subdir", "path_mismatch");
     }
+
     Ok(found)
 }
 

codex-rs/core/tests/suite/rollout_list_find.rs

@@ -3,6 +3,7 @@ use std::io::Write;
 use std::path::Path;
 use std::path::PathBuf;
 
+use chrono::Utc;
 use codex_core::RolloutRecorder;
 use codex_core::RolloutRecorderParams;
 use codex_core::config::ConfigBuilder;
@@ -12,6 +13,8 @@ use codex_core::find_thread_path_by_name_str;
 use codex_core::protocol::SessionSource;
 use codex_protocol::ThreadId;
 use codex_protocol::models::BaseInstructions;
+use codex_state::StateRuntime;
+use codex_state::ThreadMetadataBuilder;
 use pretty_assertions::assert_eq;
 use tempfile::TempDir;
 use uuid::Uuid;
@@ -52,6 +55,21 @@ fn write_minimal_rollout_with_id(codex_home: &Path, id: Uuid) -> PathBuf {
     write_minimal_rollout_with_id_in_subdir(codex_home, "sessions", id)
 }
 
+async fn upsert_thread_metadata(codex_home: &Path, thread_id: ThreadId, rollout_path: PathBuf) {
+    let runtime = StateRuntime::init(codex_home.to_path_buf(), "test-provider".to_string(), None)
+        .await
+        .unwrap();
+    let mut builder = ThreadMetadataBuilder::new(
+        thread_id,
+        rollout_path,
+        Utc::now(),
+        SessionSource::default(),
+    );
+    builder.cwd = codex_home.to_path_buf();
+    let metadata = builder.build("test-provider");
+    runtime.upsert_thread(&metadata).await.unwrap();
+}
+
 #[tokio::test]
 async fn find_locates_rollout_file_by_id() {
     let home = TempDir::new().unwrap();
@@ -81,6 +99,43 @@ async fn find_handles_gitignore_covering_codex_home_directory() {
     assert_eq!(found, Some(expected));
 }
 
+#[tokio::test]
+async fn find_prefers_sqlite_path_by_id() {
+    let home = TempDir::new().unwrap();
+    let id = Uuid::new_v4();
+    let thread_id = ThreadId::from_string(&id.to_string()).unwrap();
+    let db_path = home
+        .path()
+        .join("sessions/2030/12/30/rollout-2030-12-30T00-00-00-db.jsonl");
+    write_minimal_rollout_with_id(home.path(), id);
+    upsert_thread_metadata(home.path(), thread_id, db_path.clone()).await;
+
+    let found = find_thread_path_by_id_str(home.path(), &id.to_string())
+        .await
+        .unwrap();
+
+    assert_eq!(found, Some(db_path));
+}
+
+#[tokio::test]
+async fn find_falls_back_to_filesystem_when_sqlite_has_no_match() {
+    let home = TempDir::new().unwrap();
+    let id = Uuid::new_v4();
+    let expected = write_minimal_rollout_with_id(home.path(), id);
+    let unrelated_id = Uuid::new_v4();
+    let unrelated_thread_id = ThreadId::from_string(&unrelated_id.to_string()).unwrap();
+    let unrelated_path = home
+        .path()
+        .join("sessions/2030/12/30/rollout-2030-12-30T00-00-00-unrelated.jsonl");
+    upsert_thread_metadata(home.path(), unrelated_thread_id, unrelated_path).await;
+
+    let found = find_thread_path_by_id_str(home.path(), &id.to_string())
+        .await
+        .unwrap();
+
+    assert_eq!(found, Some(expected));
+}
+
 #[tokio::test]
 async fn find_ignores_granular_gitignore_rules() {
     let home = TempDir::new().unwrap();