Prefer the state database for exec resume --last (#36809)

## Why

Successful `codex exec resume --last` lookups should not need to audit every rollout file.

## What changed

- Query the state database first when it is available and treat the first usable matching entry as authoritative.
- Verify that an indexed rollout's session ID matches the indexed thread ID before resuming it.
- Fall back to scanning rollouts after a complete database miss, allowing the existing backfill path to repair missing entries.

## Testing

Added integration coverage for missing database entries, usable indexed candidates, and mismatched indexed rollout paths.

GitOrigin-RevId: 9959dacbc908e97f1073118142f276b470aa8e06

codex-rs/exec/tests/suite/resume.rs

@@ -1,5 +1,8 @@
 #![allow(clippy::unwrap_used)]
 use anyhow::Context;
+use codex_core::config::ConfigBuilder;
+use codex_core::init_state_db;
+use codex_protocol::ThreadId;
 use core_test_support::responses;
 use core_test_support::skip_if_no_network;
 use core_test_support::test_codex_exec::test_codex_exec;
@@ -202,6 +205,193 @@ async fn exec_resume_last_appends_to_existing_file() -> anyhow::Result<()> {
     Ok(())
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn exec_resume_last_repairs_rollout_missing_from_state_db() -> anyhow::Result<()> {
+    skip_if_no_network!(Ok(()));
+
+    let test = test_codex_exec();
+    let server = MockServer::start().await;
+    let _response_mock = mount_exec_responses(&server, /*count*/ 2).await;
+    let repo_root = exec_repo_root()?;
+
+    let marker = format!("resume-last-repair-{}", Uuid::new_v4());
+    test.cmd_with_server(&server)
+        .arg("--skip-git-repo-check")
+        .arg("-C")
+        .arg(&repo_root)
+        .arg(format!("echo {marker}"))
+        .assert()
+        .success();
+
+    let sessions_dir = test.home_path().join("sessions");
+    let path = find_session_file_containing_marker(&sessions_dir, &marker)
+        .expect("no session file found after first run");
+    let thread_id = ThreadId::from_string(&extract_conversation_id(&path))?;
+    let config = ConfigBuilder::default()
+        .codex_home(test.home_path().to_path_buf())
+        .build()
+        .await?;
+    let state_db = init_state_db(&config)
+        .await
+        .expect("state DB should initialize");
+    assert_eq!(state_db.delete_thread(thread_id).await?, 1);
+    state_db
+        .mark_backfill_complete(/*last_watermark*/ None)
+        .await?;
+
+    let resumed_marker = format!("resume-last-repaired-{}", Uuid::new_v4());
+    test.cmd_with_server(&server)
+        .arg("--skip-git-repo-check")
+        .arg("-C")
+        .arg(&repo_root)
+        .arg("resume")
+        .arg("--last")
+        .arg(format!("echo {resumed_marker}"))
+        .assert()
+        .success();
+
+    let resumed_path = find_session_file_containing_marker(&sessions_dir, &resumed_marker)
+        .expect("no resumed session file after SQLite repair");
+    assert_eq!(resumed_path, path);
+    assert!(state_db.get_thread(thread_id).await?.is_some());
+    Ok(())
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn exec_resume_last_trusts_usable_state_db_candidate() -> anyhow::Result<()> {
+    skip_if_no_network!(Ok(()));
+
+    let test = test_codex_exec();
+    let server = MockServer::start().await;
+    let _response_mock = mount_exec_responses(&server, /*count*/ 3).await;
+    let repo_root = exec_repo_root()?;
+    let sessions_dir = test.home_path().join("sessions");
+
+    let older_marker = format!("resume-last-indexed-{}", Uuid::new_v4());
+    test.cmd_with_server(&server)
+        .arg("--skip-git-repo-check")
+        .arg("-C")
+        .arg(&repo_root)
+        .arg(format!("echo {older_marker}"))
+        .assert()
+        .success();
+    let older_path = find_session_file_containing_marker(&sessions_dir, &older_marker)
+        .expect("no indexed session file after first run");
+
+    let newer_marker = format!("resume-last-unindexed-{}", Uuid::new_v4());
+    test.cmd_with_server(&server)
+        .arg("--skip-git-repo-check")
+        .arg("-C")
+        .arg(&repo_root)
+        .arg(format!("echo {newer_marker}"))
+        .assert()
+        .success();
+    let newer_path = find_session_file_containing_marker(&sessions_dir, &newer_marker)
+        .expect("no unindexed session file after second run");
+    let newer_thread_id = ThreadId::from_string(&extract_conversation_id(&newer_path))?;
+
+    let config = ConfigBuilder::default()
+        .codex_home(test.home_path().to_path_buf())
+        .build()
+        .await?;
+    let state_db = init_state_db(&config)
+        .await
+        .expect("state DB should initialize");
+    assert_eq!(state_db.delete_thread(newer_thread_id).await?, 1);
+    state_db
+        .mark_backfill_complete(/*last_watermark*/ None)
+        .await?;
+
+    let resumed_marker = format!("resume-last-authoritative-{}", Uuid::new_v4());
+    test.cmd_with_server(&server)
+        .arg("--skip-git-repo-check")
+        .arg("-C")
+        .arg(&repo_root)
+        .arg("resume")
+        .arg("--last")
+        .arg(format!("echo {resumed_marker}"))
+        .assert()
+        .success();
+
+    let resumed_path = find_session_file_containing_marker(&sessions_dir, &resumed_marker)
+        .expect("no resumed session file after SQLite lookup");
+    assert_eq!(
+        (
+            resumed_path,
+            state_db
+                .get_thread(newer_thread_id)
+                .await?
+                .map(|metadata| metadata.id),
+        ),
+        (older_path, None),
+    );
+    Ok(())
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn exec_resume_last_skips_mismatched_state_db_candidate() -> anyhow::Result<()> {
+    skip_if_no_network!(Ok(()));
+
+    let test = test_codex_exec();
+    let server = MockServer::start().await;
+    let _response_mock = mount_exec_responses(&server, /*count*/ 3).await;
+    let repo_root = exec_repo_root()?;
+    let sessions_dir = test.home_path().join("sessions");
+
+    let older_marker = format!("resume-last-valid-{}", Uuid::new_v4());
+    test.cmd_with_server(&server)
+        .arg("--skip-git-repo-check")
+        .arg("-C")
+        .arg(&repo_root)
+        .arg(format!("echo {older_marker}"))
+        .assert()
+        .success();
+    let older_path = find_session_file_containing_marker(&sessions_dir, &older_marker)
+        .expect("no valid session file after first run");
+
+    let newer_marker = format!("resume-last-mismatched-{}", Uuid::new_v4());
+    test.cmd_with_server(&server)
+        .arg("--skip-git-repo-check")
+        .arg("-C")
+        .arg(&repo_root)
+        .arg(format!("echo {newer_marker}"))
+        .assert()
+        .success();
+    let newer_path = find_session_file_containing_marker(&sessions_dir, &newer_marker)
+        .expect("no mismatched session file after second run");
+    let newer_thread_id = ThreadId::from_string(&extract_conversation_id(&newer_path))?;
+
+    let config = ConfigBuilder::default()
+        .codex_home(test.home_path().to_path_buf())
+        .build()
+        .await?;
+    let state_db = init_state_db(&config)
+        .await
+        .expect("state DB should initialize");
+    let mut mismatched = state_db
+        .get_thread(newer_thread_id)
+        .await?
+        .expect("newer thread should be indexed");
+    mismatched.rollout_path = older_path.clone();
+    state_db.upsert_thread(&mismatched).await?;
+
+    let resumed_marker = format!("resume-last-valid-resumed-{}", Uuid::new_v4());
+    test.cmd_with_server(&server)
+        .arg("--skip-git-repo-check")
+        .arg("-C")
+        .arg(&repo_root)
+        .arg("resume")
+        .arg("--last")
+        .arg(format!("echo {resumed_marker}"))
+        .assert()
+        .success();
+
+    let resumed_path = find_session_file_containing_marker(&sessions_dir, &resumed_marker)
+        .expect("no resumed session file after skipping mismatched SQLite row");
+    assert_eq!(resumed_path, older_path);
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn exec_resume_last_accepts_prompt_after_flag_in_json_mode() -> anyhow::Result<()> {
     skip_if_no_network!(Ok(()));