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
This commit is contained in:
Charlie Marsh authored and copyberry committed 2026-08-05 20:44:03 +00:00
1 parent b6cddbf6d5
commit c38a60ded2
8 files changed
+976 -193

No files matched your search

+1
View File
@@ -87,6 +87,7 @@ pub use search::search_rollout_matches;
pub use search::search_rollout_paths; pub use search::search_rollout_paths;
pub use session_index::append_thread_name; pub use session_index::append_thread_name;
pub use session_index::find_thread_meta_by_name_str; 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_name_by_id;
pub use session_index::find_thread_names_by_ids; pub use session_index::find_thread_names_by_ids;
pub use session_index::remove_thread_name_entries; pub use session_index::remove_thread_name_entries;
+48 -8
View File
@@ -7,11 +7,13 @@ use std::path::Path;
use std::path::PathBuf; use std::path::PathBuf;
use std::sync::LazyLock; use std::sync::LazyLock;
use std::sync::Mutex; use std::sync::Mutex;
use std::time::SystemTime;
use crate::reverse_jsonl_scanner::ReverseJsonlScanner; use crate::reverse_jsonl_scanner::ReverseJsonlScanner;
use crate::reverse_jsonl_scanner::ScanOutcome; use crate::reverse_jsonl_scanner::ScanOutcome;
use codex_protocol::ThreadId; use codex_protocol::ThreadId;
use codex_protocol::protocol::SessionMetaLine; use codex_protocol::protocol::SessionMetaLine;
use codex_protocol::protocol::SessionSource;
use serde::Deserialize; use serde::Deserialize;
use serde::Serialize; use serde::Serialize;
use tokio::io::AsyncBufReadExt; use tokio::io::AsyncBufReadExt;
@@ -150,19 +152,42 @@ pub async fn find_thread_names_by_ids(
Ok(names) Ok(names)
} }
/// Locate a recorded thread rollout and read its session metadata by thread name. /// Locate the readable rollout with the newest modification time for a recorded thread name.
/// Returns the newest indexed name that still has a readable rollout header.
pub async fn find_thread_meta_by_name_str( pub async fn find_thread_meta_by_name_str(
codex_home: &Path, codex_home: &Path,
name: &str, name: &str,
state_db_ctx: Option<&codex_state::StateRuntime>, state_db_ctx: Option<&codex_state::StateRuntime>,
) -> std::io::Result<Option<(PathBuf, SessionMetaLine)>> { ) -> std::io::Result<Option<(PathBuf, SessionMetaLine)>> {
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<Vec<(PathBuf, SessionMetaLine)>> {
if name.trim().is_empty() { if name.trim().is_empty() {
return Ok(None); return Ok(Vec::new());
} }
let path = session_index_path(codex_home); let path = session_index_path(codex_home);
if !path.exists() { if !path.exists() {
return Ok(None); return Ok(Vec::new());
} }
let (tx, mut rx) = tokio::sync::mpsc::channel(1); let (tx, mut rx) = tokio::sync::mpsc::channel(1);
let name = name.to_string(); let name = name.to_string();
@@ -171,6 +196,7 @@ pub async fn find_thread_meta_by_name_str(
let scan = let scan =
tokio::task::spawn_blocking(move || stream_thread_ids_from_end_by_name(&path, &name, tx)); 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 { while let Some(thread_id) = rx.recv().await {
// Keep walking until a matching id resolves to a loadable rollout so an unsaved or partial // 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. // 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? .await?
&& let Ok(session_meta) = super::list::read_session_meta_line(&path).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); let modified = tokio::fs::metadata(&path)
scan.await.map_err(std::io::Error::other)??; .await
return Ok(Some((path, session_meta))); .and_then(|metadata| metadata.modified())
.unwrap_or(SystemTime::UNIX_EPOCH);
candidates.push((modified, path, session_meta));
} }
} }
scan.await.map_err(std::io::Error::other)??; 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 { fn session_index_path(codex_home: &Path) -> PathBuf {
+83 -2
View File
@@ -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<()> { 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 timestamp = "2024-01-01T00-00-00Z".to_string();
let line = RolloutLine { let line = RolloutLine {
timestamp: timestamp.clone(), timestamp: timestamp.clone(),
@@ -34,12 +43,12 @@ fn write_rollout_with_metadata(path: &Path, thread_id: ThreadId) -> std::io::Res
cwd: ".".into(), cwd: ".".into(),
originator: "test_originator".into(), originator: "test_originator".into(),
cli_version: "test_version".into(), cli_version: "test_version".into(),
source: SessionSource::Cli, source,
thread_source: None, thread_source: None,
agent_path: None, agent_path: None,
agent_nickname: None, agent_nickname: None,
agent_role: None, agent_role: None,
model_provider: Some("test-provider".into()), model_provider: Some(model_provider.to_string()),
base_instructions: None, base_instructions: None,
dynamic_tools: None, dynamic_tools: None,
selected_capability_roots: Vec::new(), selected_capability_roots: Vec::new(),
@@ -190,6 +199,78 @@ async fn find_thread_meta_by_name_str_ignores_historical_name_after_rename() ->
Ok(()) 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<_>>(),
vec![(allowed_path, allowed_id)],
);
Ok(())
}
#[test] #[test]
fn find_thread_name_by_id_prefers_latest_entry() -> std::io::Result<()> { fn find_thread_name_by_id_prefers_latest_entry() -> std::io::Result<()> {
let temp = TempDir::new()?; let temp = TempDir::new()?;
+7 -1
View File
@@ -171,7 +171,13 @@ impl App {
tui.frame_requester().schedule_frame(); tui.frame_requester().schedule_frame();
} }
AppEvent::ResumeSessionByIdOrName(id_or_name) => { 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) => { Some(target_session) => {
return self return self
.resume_target_session(tui, app_server, target_session) .resume_target_session(tui, app_server, target_session)
+6 -105
View File
@@ -149,6 +149,7 @@ mod model_catalog;
mod model_migration; mod model_migration;
mod motion; mod motion;
mod multi_agents; mod multi_agents;
mod named_session_lookup;
mod notifications; mod notifications;
#[cfg(any(not(debug_assertions), test))] #[cfg(any(not(debug_assertions), test))]
mod npm_registry; mod npm_registry;
@@ -622,45 +623,9 @@ pub(crate) fn resume_source_kinds(include_non_interactive: bool) -> Vec<ThreadSo
source_kinds source_kinds
} }
async fn lookup_session_target_by_name_with_app_server(
app_server: &mut AppServerSession,
name: &str,
) -> color_eyre::Result<Option<resume_picker::SessionTarget>> {
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( async fn lookup_session_target_with_app_server(
app_server: &mut AppServerSession, app_server: &mut AppServerSession,
config: &Config,
id_or_name: &str, id_or_name: &str,
) -> color_eyre::Result<Option<resume_picker::SessionTarget>> { ) -> color_eyre::Result<Option<resume_picker::SessionTarget>> {
if Uuid::parse_str(id_or_name).is_ok() { 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( 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 { let Some(startup_app_server) = app_server.as_mut() else {
unreachable!("app server should be initialized for --fork <id>"); unreachable!("app server should be initialized for --fork <id>");
}; };
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), Some(target_session) => resume_picker::SessionSelection::Fork(target_session),
None => { None => {
shutdown_app_server_if_present(app_server.take()).await; 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 { let Some(startup_app_server) = app_server.as_mut() else {
unreachable!("app server should be initialized for --resume <id>"); unreachable!("app server should be initialized for --resume <id>");
}; };
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), Some(target_session) => resume_picker::SessionSelection::Resume(target_session),
None => { None => {
shutdown_app_server_if_present(app_server.take()).await; shutdown_app_server_if_present(app_server.take()).await;
@@ -3217,71 +3183,6 @@ mod tests {
Ok(()) 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] #[tokio::test]
async fn embedded_app_server_start_failure_is_returned() -> color_eyre::Result<()> { async fn embedded_app_server_start_failure_is_returned() -> color_eyre::Result<()> {
let temp_dir = TempDir::new()?; let temp_dir = TempDir::new()?;
+297
View File
@@ -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<SessionMeta>,
}
/// 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<String>,
pending: std::vec::IntoIter<AppServerThread>,
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<Option<SessionTarget>> {
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<Option<SessionTarget>> {
// 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<Option<AppServerThread>> {
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<Option<NamedSessionCandidate>> {
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;
@@ -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<Config> {
ConfigBuilder::default()
.codex_home(temp_dir.path().to_path_buf())
.build()
.await
}
async fn state_runtime(config: &Config) -> std::io::Result<Arc<codex_state::StateRuntime>> {
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<Option<crate::resume_picker::SessionTarget>> {
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<PathBuf> {
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::<Vec<_>>()
.join("\n")
+ "\n",
)?;
Ok(rollout_path.to_path_buf())
}
async fn add_scan_sentinel(config: &Config, search_term: &str) -> color_eyre::Result<ThreadId> {
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(())
}
+21 -77
View File
@@ -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::load_config_toml_with_layer_stack;
use crate::legacy_core::config::resolve_oss_provider; use crate::legacy_core::config::resolve_oss_provider;
use crate::legacy_core::config::resolve_profile_v2_config_path; 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::Thread as AppServerThread;
use codex_app_server_protocol::ThreadListParams;
use codex_app_server_protocol::ThreadSortKey;
use codex_arg0::Arg0DispatchPaths; use codex_arg0::Arg0DispatchPaths;
use codex_cloud_config::cloud_config_bundle_loader_for_storage; use codex_cloud_config::cloud_config_bundle_loader_for_storage;
use codex_config::CloudConfigBundleLoader; use codex_config::CloudConfigBundleLoader;
@@ -178,10 +179,9 @@ async fn lookup_session_by_exact_name(
name: &str, name: &str,
archived: bool, archived: bool,
) -> Result<Option<AppServerThread>> { ) -> Result<Option<AppServerThread>> {
let uses_remote_workspace = app_server.uses_remote_workspace();
// Remote workspaces stay on their existing server-side path. Local workspaces trust SQLite // 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. // 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][..] &[SessionNameLookupMode::ScanAndRepair][..]
} else { } else {
&[ &[
@@ -190,87 +190,31 @@ async fn lookup_session_by_exact_name(
][..] ][..]
}; };
for &lookup_mode in lookup_modes { 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. // Search is the fast path, but legacy stores attach renamed titles after filtering.
for search_term in [Some(name), None] { for search_term in [Some(name), None] {
let mut cursor = None; let mut candidates = NamedSessionCandidates::new(
loop { name,
let response = app_server codex_home,
.thread_list(ThreadListParams { if archived {
cursor: cursor.clone(), SessionCollection::Archived
limit: Some(100), } else {
sort_key: Some(sort_key), SessionCollection::Active
sort_direction: None, },
model_providers: None, lookup_mode,
source_kinds: Some(super::resume_source_kinds( search_term,
/*include_non_interactive*/ false, );
)), if let Some(candidate) = candidates
archived: Some(archived), .next(app_server)
section_id: None, .await
parent_thread_id: None, .wrap_err("failed to list sessions while resolving session name")?
ancestor_thread_id: None, {
cwd: None, return Ok(Some(candidate.thread));
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);
} }
} }
} }
Ok(None) Ok(None)
} }
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum SessionNameLookupMode {
StateDbOnly,
ScanAndRepair,
}
fn session_target_from_app_server_thread(thread: AppServerThread) -> Result<ResolvedSessionTarget> { fn session_target_from_app_server_thread(thread: AppServerThread) -> Result<ResolvedSessionTarget> {
let session_id = ThreadId::from_string(&thread.id) let session_id = ThreadId::from_string(&thread.id)
.wrap_err_with(|| format!("app server returned invalid session id `{}`", thread.id))?; .wrap_err_with(|| format!("app server returned invalid session id `{}`", thread.id))?;