From 41e2f67e56bdcf08886c54820b4e2c17c77c6795 Mon Sep 17 00:00:00 2001 From: Charlie Marsh Date: Mon, 3 Aug 2026 23:03:18 +0000 Subject: [PATCH] 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/src/lib.rs | 20 ++- codex-rs/exec/tests/suite/resume.rs | 190 ++++++++++++++++++++++++++++ 2 files changed, 209 insertions(+), 1 deletion(-) diff --git a/codex-rs/exec/src/lib.rs b/codex-rs/exec/src/lib.rs index 097816cda9..e61e7df26a 100644 --- a/codex-rs/exec/src/lib.rs +++ b/codex-rs/exec/src/lib.rs @@ -72,6 +72,7 @@ use codex_core::config::resolve_profile_v2_config_path; use codex_core::find_thread_meta_by_name_str; use codex_core::format_exec_policy_error_with_source; use codex_core::path_utils; +use codex_core::read_session_meta_line; use codex_feedback::CodexFeedback; use codex_git_utils::get_git_repo_root; use codex_login::AuthConfig; @@ -1466,6 +1467,7 @@ async fn resolve_resume_thread_id( let model_providers = resume_lookup_model_providers(config, args); if args.last { + let mut use_state_db_only = state_db.is_some(); let mut cursor = None; loop { let response: ThreadListResponse = send_request_with_response( @@ -1484,7 +1486,7 @@ async fn resolve_resume_thread_id( parent_thread_id: None, ancestor_thread_id: None, cwd: None, - use_state_db_only: false, + use_state_db_only, search_term: None, }, }, @@ -1493,12 +1495,28 @@ async fn resolve_resume_thread_id( .await .map_err(anyhow::Error::msg)?; for thread in response.data { + if use_state_db_only && let Some(path) = thread.path.as_deref() { + let Ok(session_meta) = read_session_meta_line(path).await else { + continue; + }; + if session_meta.meta.id.to_string() != thread.id { + continue; + } + } let latest_cwd = latest_thread_cwd(&thread).await; if args.all || cwds_match(config.cwd.as_path(), latest_cwd.as_path()) { + // A usable SQLite candidate is authoritative. Scanning is reserved for a + // complete miss so successful `--last` lookups avoid auditing every rollout. return Ok(Some(thread.id)); } } let Some(next_cursor) = response.next_cursor else { + if use_state_db_only { + // Repair from rollouts before giving up on a missing SQLite match. + use_state_db_only = false; + cursor = None; + continue; + } return Ok(None); }; cursor = Some(next_cursor); diff --git a/codex-rs/exec/tests/suite/resume.rs b/codex-rs/exec/tests/suite/resume.rs index fc4720fc18..d6b1d3af1d 100644 --- a/codex-rs/exec/tests/suite/resume.rs +++ b/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(()));