From c38a60ded2ffbad510f8b512e2902f0128e998e8 Mon Sep 17 00:00:00 2001 From: Charlie Marsh Date: Wed, 5 Aug 2026 20:27:59 +0000 Subject: [PATCH] Harden named session lookup in the TUI (#37157) ## What changed - Share exact-name candidate lookup between resume and archive commands. - Prefer valid SQLite names for local sessions, then recover legacy index names without overriding newer metadata. - Validate rollout identity, collection, source, and model provider before selecting a local session, and choose the most recently modified eligible legacy duplicate. - Preserve server-side lookup behavior for remote workspaces. ## Testing - Cover stale and duplicate SQLite entries, legacy-name recovery, source and provider filtering, paginated histories, and duplicate legacy names. GitOrigin-RevId: c69f86d64ff556bc7b463d13522a9241591eb070 --- codex-rs/rollout/src/lib.rs | 1 + codex-rs/rollout/src/session_index.rs | 56 +- codex-rs/rollout/src/session_index_tests.rs | 85 ++- codex-rs/tui/src/app/event_dispatch.rs | 8 +- codex-rs/tui/src/lib.rs | 111 +--- codex-rs/tui/src/named_session_lookup.rs | 297 ++++++++++ .../tui/src/named_session_lookup_tests.rs | 513 ++++++++++++++++++ codex-rs/tui/src/session_archive_commands.rs | 98 +--- 8 files changed, 976 insertions(+), 193 deletions(-) create mode 100644 codex-rs/tui/src/named_session_lookup.rs create mode 100644 codex-rs/tui/src/named_session_lookup_tests.rs diff --git a/codex-rs/rollout/src/lib.rs b/codex-rs/rollout/src/lib.rs index 71dac72632..89a81e4c19 100644 --- a/codex-rs/rollout/src/lib.rs +++ b/codex-rs/rollout/src/lib.rs @@ -87,6 +87,7 @@ pub use search::search_rollout_matches; pub use search::search_rollout_paths; pub use session_index::append_thread_name; pub use session_index::find_thread_meta_by_name_str; +pub use session_index::find_thread_meta_candidates_by_name_str; pub use session_index::find_thread_name_by_id; pub use session_index::find_thread_names_by_ids; pub use session_index::remove_thread_name_entries; diff --git a/codex-rs/rollout/src/session_index.rs b/codex-rs/rollout/src/session_index.rs index 834dd8a70d..ecfcf44796 100644 --- a/codex-rs/rollout/src/session_index.rs +++ b/codex-rs/rollout/src/session_index.rs @@ -7,11 +7,13 @@ use std::path::Path; use std::path::PathBuf; use std::sync::LazyLock; use std::sync::Mutex; +use std::time::SystemTime; use crate::reverse_jsonl_scanner::ReverseJsonlScanner; use crate::reverse_jsonl_scanner::ScanOutcome; use codex_protocol::ThreadId; use codex_protocol::protocol::SessionMetaLine; +use codex_protocol::protocol::SessionSource; use serde::Deserialize; use serde::Serialize; use tokio::io::AsyncBufReadExt; @@ -150,19 +152,42 @@ pub async fn find_thread_names_by_ids( Ok(names) } -/// Locate a recorded thread rollout and read its session metadata by thread name. -/// Returns the newest indexed name that still has a readable rollout header. +/// Locate the readable rollout with the newest modification time for a recorded thread name. pub async fn find_thread_meta_by_name_str( codex_home: &Path, name: &str, state_db_ctx: Option<&codex_state::StateRuntime>, ) -> std::io::Result> { + Ok(find_thread_meta_candidates_by_name_str( + codex_home, + name, + state_db_ctx, + /*allowed_sources*/ &[], + /*allowed_model_providers*/ &[], + ) + .await? + .into_iter() + .next()) +} + +/// Locate readable rollouts for a recorded thread name, newest modification time first. +/// +/// Empty filter slices disable their respective filters. Provider filtering only rejects explicit +/// mismatches because older rollouts omitted provider metadata. Filtering happens before ranking, +/// so an ineligible newer duplicate cannot hide an older usable session. +pub async fn find_thread_meta_candidates_by_name_str( + codex_home: &Path, + name: &str, + state_db_ctx: Option<&codex_state::StateRuntime>, + allowed_sources: &[SessionSource], + allowed_model_providers: &[String], +) -> std::io::Result> { if name.trim().is_empty() { - return Ok(None); + return Ok(Vec::new()); } let path = session_index_path(codex_home); if !path.exists() { - return Ok(None); + return Ok(Vec::new()); } let (tx, mut rx) = tokio::sync::mpsc::channel(1); let name = name.to_string(); @@ -171,6 +196,7 @@ pub async fn find_thread_meta_by_name_str( let scan = tokio::task::spawn_blocking(move || stream_thread_ids_from_end_by_name(&path, &name, tx)); + let mut candidates = Vec::new(); while let Some(thread_id) = rx.recv().await { // Keep walking until a matching id resolves to a loadable rollout so an unsaved or partial // rename cannot shadow an older persisted session with the same name. @@ -181,15 +207,29 @@ pub async fn find_thread_meta_by_name_str( ) .await? && let Ok(session_meta) = super::list::read_session_meta_line(&path).await + && (allowed_sources.is_empty() || allowed_sources.contains(&session_meta.meta.source)) + && session_meta + .meta + .model_provider + .as_ref() + .is_none_or(|provider| { + allowed_model_providers.is_empty() || allowed_model_providers.contains(provider) + }) { - drop(rx); - scan.await.map_err(std::io::Error::other)??; - return Ok(Some((path, session_meta))); + let modified = tokio::fs::metadata(&path) + .await + .and_then(|metadata| metadata.modified()) + .unwrap_or(SystemTime::UNIX_EPOCH); + candidates.push((modified, path, session_meta)); } } scan.await.map_err(std::io::Error::other)??; + candidates.sort_by(|(left, _, _), (right, _, _)| right.cmp(left)); - Ok(None) + Ok(candidates + .into_iter() + .map(|(_, path, session_meta)| (path, session_meta)) + .collect()) } fn session_index_path(codex_home: &Path) -> PathBuf { diff --git a/codex-rs/rollout/src/session_index_tests.rs b/codex-rs/rollout/src/session_index_tests.rs index dee363f73c..0cc20e33b2 100644 --- a/codex-rs/rollout/src/session_index_tests.rs +++ b/codex-rs/rollout/src/session_index_tests.rs @@ -20,6 +20,15 @@ fn write_index(path: &Path, lines: &[SessionIndexEntry]) -> std::io::Result<()> } fn write_rollout_with_metadata(path: &Path, thread_id: ThreadId) -> std::io::Result<()> { + write_rollout_with_source_and_provider(path, thread_id, SessionSource::Cli, "test-provider") +} + +fn write_rollout_with_source_and_provider( + path: &Path, + thread_id: ThreadId, + source: SessionSource, + model_provider: &str, +) -> std::io::Result<()> { let timestamp = "2024-01-01T00-00-00Z".to_string(); let line = RolloutLine { timestamp: timestamp.clone(), @@ -34,12 +43,12 @@ fn write_rollout_with_metadata(path: &Path, thread_id: ThreadId) -> std::io::Res cwd: ".".into(), originator: "test_originator".into(), cli_version: "test_version".into(), - source: SessionSource::Cli, + source, thread_source: None, agent_path: None, agent_nickname: None, agent_role: None, - model_provider: Some("test-provider".into()), + model_provider: Some(model_provider.to_string()), base_instructions: None, dynamic_tools: None, selected_capability_roots: Vec::new(), @@ -190,6 +199,78 @@ async fn find_thread_meta_by_name_str_ignores_historical_name_after_rename() -> Ok(()) } +#[tokio::test] +async fn find_thread_meta_candidates_filter_metadata_before_ranking() -> std::io::Result<()> { + let temp = TempDir::new()?; + let index_path = session_index_path(temp.path()); + let allowed_id = ThreadId::new(); + let other_id = ThreadId::new(); + let noninteractive_id = ThreadId::new(); + let rollout_dir = temp.path().join("sessions/2024/01/01"); + let allowed_path = rollout_dir.join(format!("rollout-2024-01-01T00-00-00-{allowed_id}.jsonl")); + let other_path = rollout_dir.join(format!("rollout-2024-01-01T00-00-01-{other_id}.jsonl")); + let noninteractive_path = rollout_dir.join(format!( + "rollout-2024-01-01T00-00-02-{noninteractive_id}.jsonl" + )); + std::fs::create_dir_all(&rollout_dir)?; + write_rollout_with_metadata(&allowed_path, allowed_id)?; + write_rollout_with_source_and_provider( + &other_path, + other_id, + SessionSource::Cli, + "other-provider", + )?; + write_rollout_with_source_and_provider( + &noninteractive_path, + noninteractive_id, + SessionSource::Exec, + "test-provider", + )?; + write_index( + &index_path, + &[ + SessionIndexEntry { + id: allowed_id, + thread_name: "same".to_string(), + updated_at: "2024-01-01T00:00:00Z".to_string(), + }, + SessionIndexEntry { + id: other_id, + thread_name: "same".to_string(), + updated_at: "2024-01-02T00:00:00Z".to_string(), + }, + SessionIndexEntry { + id: noninteractive_id, + thread_name: "same".to_string(), + updated_at: "2024-01-03T00:00:00Z".to_string(), + }, + ], + )?; + std::fs::OpenOptions::new() + .append(true) + .open(&allowed_path)? + .set_times(std::fs::FileTimes::new().set_modified(std::time::UNIX_EPOCH))?; + let allowed_model_providers = vec!["test-provider".to_string()]; + + let found = find_thread_meta_candidates_by_name_str( + temp.path(), + "same", + /*state_db_ctx*/ None, + &[SessionSource::Cli], + &allowed_model_providers, + ) + .await?; + + assert_eq!( + found + .into_iter() + .map(|(path, session_meta)| (path, session_meta.meta.id)) + .collect::>(), + vec![(allowed_path, allowed_id)], + ); + Ok(()) +} + #[test] fn find_thread_name_by_id_prefers_latest_entry() -> std::io::Result<()> { let temp = TempDir::new()?; diff --git a/codex-rs/tui/src/app/event_dispatch.rs b/codex-rs/tui/src/app/event_dispatch.rs index 95e8564638..2d55735451 100644 --- a/codex-rs/tui/src/app/event_dispatch.rs +++ b/codex-rs/tui/src/app/event_dispatch.rs @@ -171,7 +171,13 @@ impl App { tui.frame_requester().schedule_frame(); } AppEvent::ResumeSessionByIdOrName(id_or_name) => { - match crate::lookup_session_target_with_app_server(app_server, &id_or_name).await? { + match crate::lookup_session_target_with_app_server( + app_server, + &self.config, + &id_or_name, + ) + .await? + { Some(target_session) => { return self .resume_target_session(tui, app_server, target_session) diff --git a/codex-rs/tui/src/lib.rs b/codex-rs/tui/src/lib.rs index b1062103fa..c79e6add72 100644 --- a/codex-rs/tui/src/lib.rs +++ b/codex-rs/tui/src/lib.rs @@ -149,6 +149,7 @@ mod model_catalog; mod model_migration; mod motion; mod multi_agents; +mod named_session_lookup; mod notifications; #[cfg(any(not(debug_assertions), test))] mod npm_registry; @@ -622,45 +623,9 @@ pub(crate) fn resume_source_kinds(include_non_interactive: bool) -> Vec color_eyre::Result> { - let mut cursor = None; - loop { - let response = app_server - .thread_list(ThreadListParams { - cursor: cursor.clone(), - limit: Some(100), - sort_key: Some(AppServerThreadSortKey::UpdatedAt), - sort_direction: None, - model_providers: None, - source_kinds: Some(vec![ThreadSourceKind::Cli, ThreadSourceKind::VsCode]), - archived: Some(false), - section_id: None, - parent_thread_id: None, - ancestor_thread_id: None, - cwd: None, - use_state_db_only: false, - search_term: Some(name.to_string()), - }) - .await?; - if let Some(thread) = response - .data - .into_iter() - .find(|thread| thread.name.as_deref() == Some(name)) - { - return Ok(session_target_from_app_server_thread(thread)); - } - if response.next_cursor.is_none() { - return Ok(None); - } - cursor = response.next_cursor; - } -} - async fn lookup_session_target_with_app_server( app_server: &mut AppServerSession, + config: &Config, id_or_name: &str, ) -> color_eyre::Result> { if Uuid::parse_str(id_or_name).is_ok() { @@ -691,7 +656,7 @@ async fn lookup_session_target_with_app_server( }; } - lookup_session_target_by_name_with_app_server(app_server, id_or_name).await + named_session_lookup::lookup(app_server, config, id_or_name).await } async fn lookup_latest_session_target_with_app_server( @@ -1516,7 +1481,8 @@ async fn run_ratatui_app( let Some(startup_app_server) = app_server.as_mut() else { unreachable!("app server should be initialized for --fork "); }; - match lookup_session_target_with_app_server(startup_app_server, id_str).await? { + match lookup_session_target_with_app_server(startup_app_server, &config, id_str).await? + { Some(target_session) => resume_picker::SessionSelection::Fork(target_session), None => { shutdown_app_server_if_present(app_server.take()).await; @@ -1573,7 +1539,7 @@ async fn run_ratatui_app( let Some(startup_app_server) = app_server.as_mut() else { unreachable!("app server should be initialized for --resume "); }; - match lookup_session_target_with_app_server(startup_app_server, id_str).await? { + match lookup_session_target_with_app_server(startup_app_server, &config, id_str).await? { Some(target_session) => resume_picker::SessionSelection::Resume(target_session), None => { shutdown_app_server_if_present(app_server.take()).await; @@ -3217,71 +3183,6 @@ mod tests { Ok(()) } - #[tokio::test] - async fn lookup_session_target_by_name_uses_backend_title_search() -> color_eyre::Result<()> { - Box::pin(async { - let temp_dir = TempDir::new()?; - let config = build_config(&temp_dir).await?; - let thread_id = ThreadId::new(); - let rollout_path = temp_dir - .path() - .join("sessions/2025/02/01") - .join(format!("rollout-2025-02-01T10-00-00-{thread_id}.jsonl")); - let rollout_dir = rollout_path.parent().expect("rollout parent"); - std::fs::create_dir_all(rollout_dir)?; - std::fs::write(&rollout_path, "")?; - - let state_runtime = codex_state::StateRuntime::init( - codex_state::SqliteConfig::new_for_testing(config.codex_home.as_path().abs()), - config.model_provider_id.clone(), - ) - .await - .map_err(std::io::Error::other)?; - state_runtime - .mark_backfill_complete(/*last_watermark*/ None) - .await - .map_err(std::io::Error::other)?; - - let session_cwd = temp_dir.path().join("project"); - std::fs::create_dir_all(&session_cwd)?; - let created_at = chrono::DateTime::parse_from_rfc3339("2025-02-01T10:00:00Z") - .expect("timestamp should parse") - .with_timezone(&chrono::Utc); - let mut builder = codex_state::ThreadMetadataBuilder::new( - thread_id, - rollout_path.clone(), - created_at, - serde_json::from_value(serde_json::json!("cli")) - .expect("cli session source should deserialize"), - ); - builder.cwd = session_cwd; - let mut metadata = builder.build(config.model_provider_id.as_str()); - metadata.title = "saved-session".to_string(); - metadata.first_user_message = Some("preview text".to_string()); - state_runtime - .upsert_thread(&metadata) - .await - .map_err(std::io::Error::other)?; - - let mut app_server = AppServerSession::new( - codex_app_server_client::AppServerClient::InProcess( - start_test_embedded_app_server(config).await?, - ), - ThreadParamsMode::Embedded, - ); - let target = - lookup_session_target_by_name_with_app_server(&mut app_server, "saved-session") - .await?; - let target = target.expect("name lookup should find the saved thread"); - assert_eq!(target.path, Some(rollout_path)); - assert_eq!(target.thread_id, thread_id); - - app_server.shutdown().await?; - Ok(()) - }) - .await - } - #[tokio::test] async fn embedded_app_server_start_failure_is_returned() -> color_eyre::Result<()> { let temp_dir = TempDir::new()?; diff --git a/codex-rs/tui/src/named_session_lookup.rs b/codex-rs/tui/src/named_session_lookup.rs new file mode 100644 index 0000000000..e1ab21bc8d --- /dev/null +++ b/codex-rs/tui/src/named_session_lookup.rs @@ -0,0 +1,297 @@ +use std::path::Path; + +use crate::app_server_session::AppServerSession; +use crate::legacy_core::config::Config; +use crate::resume_picker::SessionTarget; +use codex_app_server_protocol::Thread as AppServerThread; +use codex_app_server_protocol::ThreadHistoryMode; +use codex_app_server_protocol::ThreadListParams; +use codex_app_server_protocol::ThreadSortKey; +use codex_protocol::ThreadId; +use codex_protocol::protocol::SessionMeta; +use codex_protocol::protocol::SessionSource; + +// Keep this aligned with `resume_source_kinds(/*include_non_interactive*/ false)`. +const RESUME_SESSION_SOURCES: [SessionSource; 2] = [SessionSource::Cli, SessionSource::VSCode]; + +/// Selects whether a local lookup can repair SQLite metadata from rollout files. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(super) enum SessionNameLookupMode { + StateDbOnly, + ScanAndRepair, +} + +/// Selects the rollout collection that a local candidate must belong to. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(super) enum SessionCollection { + Active, + Archived, +} + +/// An exact-name match and the rollout metadata verified for local candidates. +pub(super) struct NamedSessionCandidate { + pub(super) thread: AppServerThread, + /// Remote candidates never inspect local rollout files and therefore have no session metadata. + pub(super) session_meta: Option, +} + +/// Lazily pages through exact-name matches and skips unusable local rollouts. +pub(super) struct NamedSessionCandidates<'a> { + name: &'a str, + codex_home: &'a Path, + collection: SessionCollection, + mode: SessionNameLookupMode, + search_term: Option<&'a str>, + cursor: Option, + pending: std::vec::IntoIter, + has_more: bool, +} + +/// Resolves an active session by exact name. +/// +/// Local lookups treat a usable SQLite match as authoritative and consult legacy rollout metadata +/// only to recover a missing or unusable row. Remote workspaces retain the backend's scan behavior. +pub(super) async fn lookup( + app_server: &mut AppServerSession, + config: &Config, + name: &str, +) -> color_eyre::Result> { + if app_server.uses_remote_workspace() { + return Ok(lookup_from_app_server( + app_server, + config, + name, + SessionNameLookupMode::ScanAndRepair, + ) + .await? + .and_then(super::session_target_from_app_server_thread)); + } + + if let Some(thread) = + lookup_from_app_server(app_server, config, name, SessionNameLookupMode::StateDbOnly).await? + { + return Ok(super::session_target_from_app_server_thread(thread)); + } + + if let Some(target) = lookup_legacy_index_target(app_server, config, name).await? { + return Ok(Some(target)); + } + + Ok(lookup_from_app_server( + app_server, + config, + name, + SessionNameLookupMode::ScanAndRepair, + ) + .await? + .and_then(super::session_target_from_app_server_thread)) +} + +/// Recovers a local target from legacy metadata without overriding an explicit SQLite name. +/// +/// Rewriting the recovered name into SQLite is best effort and cannot invalidate a usable target. +async fn lookup_legacy_index_target( + app_server: &mut AppServerSession, + config: &Config, + name: &str, +) -> color_eyre::Result> { + // Older rollouts omitted the provider and were historically assigned to the current provider + // during scans, so only reject an explicit mismatch here. + let candidates = codex_rollout::find_thread_meta_candidates_by_name_str( + config.codex_home.as_path(), + name, + /*state_db_ctx*/ None, + &RESUME_SESSION_SOURCES, + std::slice::from_ref(&config.model_provider_id), + ) + .await?; + for (path, session_meta) in candidates { + if !path.starts_with(config.codex_home.join(codex_rollout::SESSIONS_SUBDIR)) { + continue; + } + + // `thread/read` resolves the candidate's current SQLite title. A conflicting title means + // the sidecar is stale, while a missing legacy title is the partial-write case it recovers. + let thread = app_server + .thread_read(session_meta.meta.id, /*include_turns*/ false) + .await?; + if !current_name_is_compatible(&thread, name) { + continue; + } + if let Err(err) = app_server + .thread_set_name(session_meta.meta.id, name.to_string()) + .await + { + tracing::warn!( + thread_id = %session_meta.meta.id, + %err, + "failed to repair recovered thread name" + ); + } + return Ok(Some(SessionTarget { + path: Some(path), + thread_id: session_meta.meta.id, + })); + } + Ok(None) +} + +/// Allows a missing SQLite name only for legacy threads, whose name may exist only in the index. +fn current_name_is_compatible(thread: &AppServerThread, name: &str) -> bool { + match thread.history_mode { + ThreadHistoryMode::Legacy => thread.name.as_deref().is_none_or(|current| current == name), + ThreadHistoryMode::Paginated => thread.name.as_deref() == Some(name), + } +} + +/// Returns an eligible active thread for one SQLite-only or scan-and-repair lookup pass. +/// +/// Local resume candidates also honor source/provider filters and reject stale scanned titles. +async fn lookup_from_app_server( + app_server: &mut AppServerSession, + config: &Config, + name: &str, + mode: SessionNameLookupMode, +) -> color_eyre::Result> { + let mut candidates = NamedSessionCandidates::new( + name, + config.codex_home.as_path(), + SessionCollection::Active, + mode, + Some(name), + ); + while let Some(candidate) = candidates.next(app_server).await? { + if let Some(session_meta) = candidate.session_meta + && (!RESUME_SESSION_SOURCES.contains(&session_meta.source) + || session_meta + .model_provider + .as_deref() + .is_some_and(|provider| provider != config.model_provider_id.as_str())) + { + continue; + } + if mode == SessionNameLookupMode::ScanAndRepair && !app_server.uses_remote_workspace() { + // Scans can surface stale sidecar names, so verify the candidate's current title. + let thread = app_server + .thread_read( + ThreadId::from_string(&candidate.thread.id)?, + /*include_turns*/ false, + ) + .await?; + if !current_name_is_compatible(&thread, name) { + continue; + } + return Ok(Some(thread)); + } + return Ok(Some(candidate.thread)); + } + Ok(None) +} + +impl<'a> NamedSessionCandidates<'a> { + pub(super) fn new( + name: &'a str, + codex_home: &'a Path, + collection: SessionCollection, + mode: SessionNameLookupMode, + search_term: Option<&'a str>, + ) -> Self { + Self { + name, + codex_home, + collection, + mode, + search_term, + cursor: None, + pending: Vec::new().into_iter(), + has_more: true, + } + } + + /// Returns the next exact-name match, advancing through server pages as needed. + /// + /// Local candidates must have a rollout in the requested collection with a matching session + /// header; remote candidates preserve server-side lookup without touching local files. + pub(super) async fn next( + &mut self, + app_server: &mut AppServerSession, + ) -> color_eyre::Result> { + loop { + if let Some(thread) = self.pending.next() { + if thread.name.as_deref() != Some(self.name) { + continue; + } + let session_meta = if app_server.uses_remote_workspace() { + None + } else { + let (Ok(thread_id), Some(path)) = + (ThreadId::from_string(&thread.id), thread.path.as_deref()) + else { + continue; + }; + let Some(path) = codex_rollout::existing_rollout_path(path).await else { + continue; + }; + let expected_root = self.codex_home.join(match self.collection { + SessionCollection::Active => codex_rollout::SESSIONS_SUBDIR, + SessionCollection::Archived => codex_rollout::ARCHIVED_SESSIONS_SUBDIR, + }); + if !path.starts_with(expected_root) { + continue; + } + let Ok(session_meta) = + codex_rollout::read_session_meta_line(path.as_path()).await + else { + continue; + }; + if session_meta.meta.id != thread_id { + continue; + } + Some(session_meta.meta) + }; + return Ok(Some(NamedSessionCandidate { + thread, + session_meta, + })); + } + + if !self.has_more { + return Ok(None); + } + // Only embedded SQLite supports recency sorting; scanners and older daemons use mtime. + let sort_key = if self.mode == SessionNameLookupMode::StateDbOnly + && app_server.uses_embedded_app_server() + { + ThreadSortKey::RecencyAt + } else { + ThreadSortKey::UpdatedAt + }; + let response = app_server + .thread_list(ThreadListParams { + cursor: self.cursor.clone(), + limit: Some(100), + sort_key: Some(sort_key), + sort_direction: None, + model_providers: None, + source_kinds: Some(super::resume_source_kinds( + /*include_non_interactive*/ false, + )), + archived: Some(self.collection == SessionCollection::Archived), + section_id: None, + parent_thread_id: None, + ancestor_thread_id: None, + cwd: None, + use_state_db_only: self.mode == SessionNameLookupMode::StateDbOnly, + search_term: self.search_term.map(str::to_string), + }) + .await?; + self.cursor = response.next_cursor; + self.has_more = self.cursor.is_some(); + self.pending = response.data.into_iter(); + } + } +} + +#[cfg(test)] +#[path = "named_session_lookup_tests.rs"] +mod tests; diff --git a/codex-rs/tui/src/named_session_lookup_tests.rs b/codex-rs/tui/src/named_session_lookup_tests.rs new file mode 100644 index 0000000000..eb6e01ad93 --- /dev/null +++ b/codex-rs/tui/src/named_session_lookup_tests.rs @@ -0,0 +1,513 @@ +use std::path::PathBuf; +use std::sync::Arc; + +use codex_protocol::ThreadId; +use codex_protocol::protocol::SessionMeta; +use codex_protocol::protocol::SessionMetaLine; +use codex_protocol::protocol::SessionSource; +use codex_protocol::protocol::ThreadHistoryMode; +use codex_state::ThreadMetadataBuilder; +use codex_utils_absolute_path::test_support::PathExt; +use pretty_assertions::assert_eq; +use tempfile::TempDir; + +use super::lookup; +use crate::app_server_session::AppServerSession; +use crate::app_server_session::ThreadParamsMode; +use crate::legacy_core::config::Config; +use crate::legacy_core::config::ConfigBuilder; +use crate::tests::start_test_embedded_app_server; + +async fn build_config(temp_dir: &TempDir) -> std::io::Result { + ConfigBuilder::default() + .codex_home(temp_dir.path().to_path_buf()) + .build() + .await +} + +async fn state_runtime(config: &Config) -> std::io::Result> { + let runtime = codex_state::StateRuntime::init( + codex_state::SqliteConfig::new_for_testing(config.codex_home.as_path().abs()), + config.model_provider_id.clone(), + ) + .await + .map_err(std::io::Error::other)?; + runtime + .mark_backfill_complete(/*last_watermark*/ None) + .await + .map_err(std::io::Error::other)?; + Ok(runtime) +} + +async fn lookup_name( + config: &Config, + name: &str, +) -> color_eyre::Result> { + let mut app_server = AppServerSession::new( + codex_app_server_client::AppServerClient::InProcess( + start_test_embedded_app_server(config.clone()).await?, + ), + ThreadParamsMode::Embedded, + ); + let target = lookup(&mut app_server, config, name).await?; + app_server.shutdown().await?; + Ok(target) +} + +async fn upsert_thread( + runtime: &codex_state::StateRuntime, + metadata: codex_state::ThreadMetadata, +) -> std::io::Result<()> { + runtime + .upsert_thread(&metadata) + .await + .map_err(std::io::Error::other) +} + +fn thread_metadata( + config: &Config, + thread_id: ThreadId, + rollout_path: PathBuf, + title: &str, +) -> codex_state::ThreadMetadata { + let created_at = chrono::DateTime::parse_from_rfc3339("2025-02-01T10:00:00Z") + .expect("timestamp should parse") + .with_timezone(&chrono::Utc); + let mut builder = ThreadMetadataBuilder::new( + thread_id, + rollout_path, + created_at, + serde_json::from_value(serde_json::json!("cli")) + .expect("cli session source should deserialize"), + ); + builder.cwd = config.codex_home.join("project").to_path_buf(); + let mut metadata = builder.build(config.model_provider_id.as_str()); + metadata.title = title.to_string(); + metadata.first_user_message = Some("preview text".to_string()); + metadata.preview = metadata.first_user_message.clone(); + metadata +} + +fn write_rollout( + config: &Config, + thread_id: ThreadId, + timestamp: &str, + preview: &str, + source: SessionSource, + history_mode: ThreadHistoryMode, +) -> color_eyre::Result { + let rollout_path = config + .codex_home + .join("sessions/2025/02/01") + .join(format!("rollout-2025-02-01T10-00-00-{thread_id}.jsonl")); + std::fs::create_dir_all(rollout_path.parent().expect("rollout parent"))?; + let session_meta = SessionMetaLine { + meta: SessionMeta { + session_id: thread_id.into(), + id: thread_id, + timestamp: timestamp.to_string(), + cwd: config.codex_home.join("project").to_path_buf(), + originator: "codex".to_string(), + cli_version: "0.0.0".to_string(), + source, + model_provider: Some(config.model_provider_id.clone()), + history_mode, + ..Default::default() + }, + git: None, + }; + let lines = [ + serde_json::json!({ + "timestamp": timestamp, + "type": "session_meta", + "payload": session_meta, + }), + serde_json::json!({ + "timestamp": timestamp, + "type": "response_item", + "payload": { + "type": "message", + "role": "user", + "content": [{"type": "input_text", "text": preview}], + }, + }), + ]; + std::fs::write( + &rollout_path, + lines + .iter() + .map(serde_json::Value::to_string) + .collect::>() + .join("\n") + + "\n", + )?; + Ok(rollout_path.to_path_buf()) +} + +async fn add_scan_sentinel(config: &Config, search_term: &str) -> color_eyre::Result { + let thread_id = ThreadId::new(); + write_rollout( + config, + thread_id, + "2025-02-01T11:00:00Z", + "scan sentinel", + SessionSource::Cli, + ThreadHistoryMode::Legacy, + )?; + codex_rollout::append_thread_name( + config.codex_home.as_path(), + thread_id, + &format!("{search_term}-scan-sentinel"), + ) + .await?; + Ok(thread_id) +} + +#[tokio::test] +async fn uses_sqlite_title_without_scanning_rollouts() -> color_eyre::Result<()> { + let temp_dir = TempDir::new()?; + let config = build_config(&temp_dir).await?; + let runtime = state_runtime(&config).await?; + let thread_id = ThreadId::new(); + let rollout_path = write_rollout( + &config, + thread_id, + "2025-02-01T10:00:00Z", + "preview", + SessionSource::Cli, + ThreadHistoryMode::Legacy, + )?; + upsert_thread( + &runtime, + thread_metadata(&config, thread_id, rollout_path.clone(), "saved-session"), + ) + .await?; + // Scan-and-repair would discover this matching rollout and insert it into SQLite. + let scan_sentinel_id = add_scan_sentinel(&config, "saved-session").await?; + let duplicate_id = ThreadId::new(); + write_rollout( + &config, + duplicate_id, + "2025-02-01T11:00:00Z", + "newer duplicate", + SessionSource::Cli, + ThreadHistoryMode::Legacy, + )?; + codex_rollout::append_thread_name(config.codex_home.as_path(), duplicate_id, "saved-session") + .await?; + + let target = lookup_name(&config, "saved-session").await?; + + assert_eq!( + ( + target.map(|target| (target.path, target.thread_id)), + runtime + .get_thread(scan_sentinel_id) + .await + .map_err(std::io::Error::other)?, + runtime + .get_thread(duplicate_id) + .await + .map_err(std::io::Error::other)?, + ), + (Some((Some(rollout_path), thread_id)), None, None), + ); + Ok(()) +} + +#[tokio::test] +async fn skips_stale_sqlite_candidate_for_valid_duplicate() -> color_eyre::Result<()> { + let temp_dir = TempDir::new()?; + let config = build_config(&temp_dir).await?; + let runtime = state_runtime(&config).await?; + let valid_id = ThreadId::new(); + let valid_path = write_rollout( + &config, + valid_id, + "2025-02-01T10:00:00Z", + "valid preview", + SessionSource::Cli, + ThreadHistoryMode::Legacy, + )?; + upsert_thread( + &runtime, + thread_metadata(&config, valid_id, valid_path.clone(), "saved-session"), + ) + .await?; + + let stale_id = ThreadId::new(); + let stale_path = write_rollout( + &config, + ThreadId::new(), + "2025-02-01T11:00:00Z", + "stale preview", + SessionSource::Cli, + ThreadHistoryMode::Legacy, + )?; + upsert_thread( + &runtime, + thread_metadata(&config, stale_id, stale_path, "saved-session"), + ) + .await?; + + let target = lookup_name(&config, "saved-session").await?; + + assert_eq!( + target.map(|target| (target.path, target.thread_id)), + Some((Some(valid_path), valid_id)), + ); + Ok(()) +} + +#[tokio::test] +async fn validates_sqlite_candidate_filters_against_rollout() -> color_eyre::Result<()> { + let mut targets = Vec::new(); + for (source, model_provider) in [ + (SessionSource::Exec, None), + (SessionSource::Cli, Some("other-provider")), + ] { + let temp_dir = TempDir::new()?; + let config = build_config(&temp_dir).await?; + let runtime = state_runtime(&config).await?; + let thread_id = ThreadId::new(); + let mut rollout_config = config.clone(); + if let Some(model_provider) = model_provider { + rollout_config.model_provider_id = model_provider.to_string(); + } + let rollout_path = write_rollout( + &rollout_config, + thread_id, + "2025-02-01T10:00:00Z", + "preview", + source, + ThreadHistoryMode::Legacy, + )?; + upsert_thread( + &runtime, + thread_metadata(&config, thread_id, rollout_path, "saved-session"), + ) + .await?; + + targets.push( + lookup_name(&config, "saved-session") + .await? + .map(|target| target.thread_id), + ); + } + + assert_eq!(targets, vec![None, None]); + Ok(()) +} + +#[tokio::test] +async fn recovers_legacy_index_name_without_requiring_sidecar_rewrite() -> color_eyre::Result<()> { + let temp_dir = TempDir::new()?; + let config = build_config(&temp_dir).await?; + let runtime = state_runtime(&config).await?; + let stale_id = ThreadId::new(); + upsert_thread( + &runtime, + thread_metadata( + &config, + stale_id, + config + .codex_home + .join("missing-rollout.jsonl") + .to_path_buf(), + "saved-session", + ), + ) + .await?; + + let thread_id = ThreadId::new(); + let rollout_path = write_rollout( + &config, + thread_id, + "2025-02-01T11:00:00Z", + "preview", + SessionSource::Cli, + ThreadHistoryMode::Legacy, + )?; + codex_rollout::append_thread_name(config.codex_home.as_path(), thread_id, "saved-session") + .await?; + // The direct index lookup should not scan and repair other matching rollouts. + let scan_sentinel_id = add_scan_sentinel(&config, "saved-session").await?; + + let index_path = config.codex_home.join("session_index.jsonl"); + let original_permissions = std::fs::metadata(&index_path)?.permissions(); + let mut read_only_permissions = original_permissions.clone(); + read_only_permissions.set_readonly(true); + std::fs::set_permissions(&index_path, read_only_permissions)?; + let target = lookup_name(&config, "saved-session").await; + std::fs::set_permissions(&index_path, original_permissions)?; + let target = target?; + + assert_eq!( + ( + target.map(|target| (target.path, target.thread_id)), + runtime + .get_thread(thread_id) + .await + .map_err(std::io::Error::other)? + .map(|metadata| metadata.title), + runtime + .get_thread(scan_sentinel_id) + .await + .map_err(std::io::Error::other)?, + ), + ( + Some((Some(rollout_path), thread_id)), + Some("saved-session".to_string()), + None, + ), + ); + Ok(()) +} + +#[tokio::test] +async fn skips_stale_legacy_index_candidate() -> color_eyre::Result<()> { + let temp_dir = TempDir::new()?; + let config = build_config(&temp_dir).await?; + let runtime = state_runtime(&config).await?; + let stale_id = ThreadId::new(); + let stale_path = write_rollout( + &config, + stale_id, + "2025-02-01T10:00:00Z", + "preview", + SessionSource::Cli, + ThreadHistoryMode::Legacy, + )?; + upsert_thread( + &runtime, + thread_metadata(&config, stale_id, stale_path.clone(), "current-session"), + ) + .await?; + codex_rollout::append_thread_name(config.codex_home.as_path(), stale_id, "saved-session") + .await?; + + let fallback_id = ThreadId::new(); + let fallback_path = write_rollout( + &config, + fallback_id, + "2025-02-01T11:00:00Z", + "fallback", + SessionSource::Cli, + ThreadHistoryMode::Legacy, + )?; + codex_rollout::append_thread_name(config.codex_home.as_path(), fallback_id, "saved-session") + .await?; + for (path, timestamp) in [ + (&stale_path, "2025-02-04T10:00:00Z"), + (&fallback_path, "2025-02-03T10:00:00Z"), + ] { + let modified = chrono::DateTime::parse_from_rfc3339(timestamp)?.into(); + std::fs::OpenOptions::new() + .append(true) + .open(path)? + .set_times(std::fs::FileTimes::new().set_modified(modified))?; + } + + let target = lookup_name(&config, "saved-session").await?; + + assert_eq!( + ( + target.map(|target| (target.path, target.thread_id)), + runtime + .get_thread(stale_id) + .await + .map_err(std::io::Error::other)? + .map(|metadata| metadata.title), + ), + ( + Some((Some(fallback_path), fallback_id)), + Some("current-session".to_string()), + ), + ); + Ok(()) +} + +#[tokio::test] +async fn keeps_newer_paginated_sqlite_name_when_index_is_stale() -> color_eyre::Result<()> { + let temp_dir = TempDir::new()?; + let config = build_config(&temp_dir).await?; + let runtime = state_runtime(&config).await?; + let thread_id = ThreadId::new(); + let rollout_path = write_rollout( + &config, + thread_id, + "2025-02-01T10:00:00Z", + "preview", + SessionSource::Cli, + ThreadHistoryMode::Paginated, + )?; + let mut metadata = thread_metadata(&config, thread_id, rollout_path, "preview text"); + metadata.history_mode = ThreadHistoryMode::Paginated; + upsert_thread(&runtime, metadata).await?; + runtime + .update_thread_name(thread_id, Some("current-session")) + .await + .map_err(std::io::Error::other)?; + codex_rollout::append_thread_name(config.codex_home.as_path(), thread_id, "old-session") + .await?; + + let target = lookup_name(&config, "old-session").await?; + + assert_eq!( + ( + target.map(|target| target.thread_id), + runtime + .get_thread(thread_id) + .await + .map_err(std::io::Error::other)? + .and_then(|metadata| metadata.name), + ), + (None, Some("current-session".to_string())), + ); + Ok(()) +} + +#[tokio::test] +async fn prefers_most_recent_duplicate_legacy_name() -> color_eyre::Result<()> { + let temp_dir = TempDir::new()?; + let config = build_config(&temp_dir).await?; + let _runtime = state_runtime(&config).await?; + let recent_id = ThreadId::new(); + let recent_path = write_rollout( + &config, + recent_id, + "2025-02-01T10:00:00Z", + "recent", + SessionSource::Cli, + ThreadHistoryMode::Legacy, + )?; + let older_id = ThreadId::new(); + let older_path = write_rollout( + &config, + older_id, + "2025-02-02T10:00:00Z", + "older", + SessionSource::Cli, + ThreadHistoryMode::Legacy, + )?; + codex_rollout::append_thread_name(config.codex_home.as_path(), recent_id, "same-name").await?; + codex_rollout::append_thread_name(config.codex_home.as_path(), older_id, "same-name").await?; + for (path, timestamp) in [ + (&recent_path, "2025-02-04T10:00:00Z"), + (&older_path, "2025-02-03T10:00:00Z"), + ] { + let modified = chrono::DateTime::parse_from_rfc3339(timestamp)?.into(); + std::fs::OpenOptions::new() + .append(true) + .open(path)? + .set_times(std::fs::FileTimes::new().set_modified(modified))?; + } + + let target = lookup_name(&config, "same-name").await?; + + assert_eq!( + target.map(|target| (target.path, target.thread_id)), + Some((Some(recent_path), recent_id)), + ); + Ok(()) +} diff --git a/codex-rs/tui/src/session_archive_commands.rs b/codex-rs/tui/src/session_archive_commands.rs index faa4e78e76..e2c378f917 100644 --- a/codex-rs/tui/src/session_archive_commands.rs +++ b/codex-rs/tui/src/session_archive_commands.rs @@ -17,9 +17,10 @@ use crate::legacy_core::config::bootstrap_auth_config; use crate::legacy_core::config::load_config_toml_with_layer_stack; use crate::legacy_core::config::resolve_oss_provider; use crate::legacy_core::config::resolve_profile_v2_config_path; +use crate::named_session_lookup::NamedSessionCandidates; +use crate::named_session_lookup::SessionCollection; +use crate::named_session_lookup::SessionNameLookupMode; use codex_app_server_protocol::Thread as AppServerThread; -use codex_app_server_protocol::ThreadListParams; -use codex_app_server_protocol::ThreadSortKey; use codex_arg0::Arg0DispatchPaths; use codex_cloud_config::cloud_config_bundle_loader_for_storage; use codex_config::CloudConfigBundleLoader; @@ -178,10 +179,9 @@ async fn lookup_session_by_exact_name( name: &str, archived: bool, ) -> Result> { - let uses_remote_workspace = app_server.uses_remote_workspace(); // Remote workspaces stay on their existing server-side path. Local workspaces trust SQLite // names, then scan and repair only after a miss or an unusable rollout path. - let lookup_modes = if uses_remote_workspace { + let lookup_modes = if app_server.uses_remote_workspace() { &[SessionNameLookupMode::ScanAndRepair][..] } else { &[ @@ -190,87 +190,31 @@ async fn lookup_session_by_exact_name( ][..] }; for &lookup_mode in lookup_modes { - // Only the embedded server can safely use SQLite's recency cursor. Daemons may predate - // that sort key, and filesystem repair must paginate in the scanner's mtime order. - let sort_key = if lookup_mode == SessionNameLookupMode::StateDbOnly - && app_server.uses_embedded_app_server() - { - ThreadSortKey::RecencyAt - } else { - ThreadSortKey::UpdatedAt - }; // Search is the fast path, but legacy stores attach renamed titles after filtering. for search_term in [Some(name), None] { - let mut cursor = None; - loop { - let response = app_server - .thread_list(ThreadListParams { - cursor: cursor.clone(), - limit: Some(100), - sort_key: Some(sort_key), - sort_direction: None, - model_providers: None, - source_kinds: Some(super::resume_source_kinds( - /*include_non_interactive*/ false, - )), - archived: Some(archived), - section_id: None, - parent_thread_id: None, - ancestor_thread_id: None, - cwd: None, - use_state_db_only: lookup_mode == SessionNameLookupMode::StateDbOnly, - search_term: search_term.map(str::to_string), - }) - .await - .wrap_err("failed to list sessions while resolving session name")?; - - for thread in response - .data - .into_iter() - .filter(|thread| thread.name.as_deref() == Some(name)) - { - if !uses_remote_workspace { - // The action still requires a real rollout in the requested collection. - let thread_id = ThreadId::from_string(&thread.id).wrap_err_with(|| { - format!("app server returned invalid session id `{}`", thread.id) - })?; - let expected_root = codex_home.join(if archived { - codex_rollout::ARCHIVED_SESSIONS_SUBDIR - } else { - codex_rollout::SESSIONS_SUBDIR - }); - let valid_rollout = if let Some(path) = thread.path.as_deref() - && let Some(path) = codex_rollout::existing_rollout_path(path).await - && path.starts_with(expected_root) - && let Ok(session_meta) = - codex_rollout::read_session_meta_line(path.as_path()).await - { - session_meta.meta.id == thread_id - } else { - false - }; - if !valid_rollout { - continue; - } - } - return Ok(Some(thread)); - } - let Some(next_cursor) = response.next_cursor else { - break; - }; - cursor = Some(next_cursor); + let mut candidates = NamedSessionCandidates::new( + name, + codex_home, + if archived { + SessionCollection::Archived + } else { + SessionCollection::Active + }, + lookup_mode, + search_term, + ); + if let Some(candidate) = candidates + .next(app_server) + .await + .wrap_err("failed to list sessions while resolving session name")? + { + return Ok(Some(candidate.thread)); } } } Ok(None) } -#[derive(Clone, Copy, Debug, Eq, PartialEq)] -enum SessionNameLookupMode { - StateDbOnly, - ScanAndRepair, -} - fn session_target_from_app_server_thread(thread: AppServerThread) -> Result { let session_id = ThreadId::from_string(&thread.id) .wrap_err_with(|| format!("app server returned invalid session id `{}`", thread.id))?;