Fix stale thread-name resume lookups (#16646)

Addresses #15943

Problem: Name-based resume could stop on a newer session_index entry
whose rollout was never persisted, shadowing an older saved thread with
the same name.

Solution: Materialize rollouts before indexing thread names and make
name lookup skip unresolved entries until it finds a persisted rollout.

codex-rs/rollout/src/session_index.rs

@@ -111,11 +111,12 @@ pub async fn find_thread_names_by_ids(
     Ok(names)
 }
 
-/// Find the most recently updated thread id for a thread name, if any.
-pub async fn find_thread_id_by_name(
+/// Locate a recorded thread rollout file by thread name using newest-first ordering.
+/// Returns `Ok(Some(path))` if found, `Ok(None)` if not present.
+pub async fn find_thread_path_by_name_str(
     codex_home: &Path,
     name: &str,
-) -> std::io::Result<Option<ThreadId>> {
+) -> std::io::Result<Option<PathBuf>> {
     if name.trim().is_empty() {
         return Ok(None);
     }
@@ -123,23 +124,28 @@ pub async fn find_thread_id_by_name(
     if !path.exists() {
         return Ok(None);
     }
+    let (tx, mut rx) = tokio::sync::mpsc::channel(1);
     let name = name.to_string();
-    let entry = tokio::task::spawn_blocking(move || scan_index_from_end_by_name(&path, &name))
-        .await
-        .map_err(std::io::Error::other)??;
-    Ok(entry.map(|entry| entry.id))
-}
+    // Stream matching ids newest-first instead of stopping at the first name hit: the newest entry
+    // may point at a thread whose rollout was never materialized.
+    let scan =
+        tokio::task::spawn_blocking(move || stream_thread_ids_from_end_by_name(&path, &name, tx));
 
-/// Locate a recorded thread rollout file by thread name using newest-first ordering.
-/// Returns `Ok(Some(path))` if found, `Ok(None)` if not present.
-pub async fn find_thread_path_by_name_str(
-    codex_home: &Path,
-    name: &str,
-) -> std::io::Result<Option<PathBuf>> {
-    let Some(thread_id) = find_thread_id_by_name(codex_home, name).await? else {
-        return Ok(None);
-    };
-    super::list::find_thread_path_by_id_str(codex_home, &thread_id.to_string()).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
+        // rename cannot shadow an older persisted session with the same name.
+        if let Some(path) =
+            super::list::find_thread_path_by_id_str(codex_home, &thread_id.to_string()).await?
+            && super::list::read_session_meta_line(&path).await.is_ok()
+        {
+            drop(rx);
+            scan.await.map_err(std::io::Error::other)??;
+            return Ok(Some(path));
+        }
+    }
+    scan.await.map_err(std::io::Error::other)??;
+
+    Ok(None)
 }
 
 fn session_index_path(codex_home: &Path) -> PathBuf {
@@ -153,11 +159,22 @@ fn scan_index_from_end_by_id(
     scan_index_from_end(path, |entry| entry.id == *thread_id)
 }
 
-fn scan_index_from_end_by_name(
+fn stream_thread_ids_from_end_by_name(
     path: &Path,
     name: &str,
-) -> std::io::Result<Option<SessionIndexEntry>> {
-    scan_index_from_end(path, |entry| entry.thread_name == name)
+    tx: tokio::sync::mpsc::Sender<ThreadId>,
+) -> std::io::Result<()> {
+    let mut seen = HashSet::new();
+    scan_index_from_end_for_each(path, |entry| {
+        // The first row seen for an id is its latest name. Ignore older rows for that id so a
+        // historical name cannot be treated as the current one after the thread is renamed.
+        if seen.insert(entry.id) && entry.thread_name == name && tx.blocking_send(entry.id).is_err()
+        {
+            return Ok(Some(entry.clone()));
+        }
+        Ok(None)
+    })?;
+    Ok(())
 }
 
 fn scan_index_from_end<F>(
@@ -166,6 +183,21 @@ fn scan_index_from_end<F>(
 ) -> std::io::Result<Option<SessionIndexEntry>>
 where
     F: FnMut(&SessionIndexEntry) -> bool,
+{
+    scan_index_from_end_for_each(path, |entry| {
+        if predicate(entry) {
+            return Ok(Some(entry.clone()));
+        }
+        Ok(None)
+    })
+}
+
+fn scan_index_from_end_for_each<F>(
+    path: &Path,
+    mut visit_entry: F,
+) -> std::io::Result<Option<SessionIndexEntry>>
+where
+    F: FnMut(&SessionIndexEntry) -> std::io::Result<Option<SessionIndexEntry>>,
 {
     let mut file = File::open(path)?;
     let mut remaining = file.metadata()?.len();
@@ -181,7 +213,7 @@ where
 
         for &byte in buf[..read_size].iter().rev() {
             if byte == b'\n' {
-                if let Some(entry) = parse_line_from_rev(&mut line_rev, &mut predicate)? {
+                if let Some(entry) = parse_line_from_rev(&mut line_rev, &mut visit_entry)? {
                     return Ok(Some(entry));
                 }
                 continue;
@@ -190,7 +222,7 @@ where
         }
     }
 
-    if let Some(entry) = parse_line_from_rev(&mut line_rev, &mut predicate)? {
+    if let Some(entry) = parse_line_from_rev(&mut line_rev, &mut visit_entry)? {
         return Ok(Some(entry));
     }
 
@@ -199,10 +231,10 @@ where
 
 fn parse_line_from_rev<F>(
     line_rev: &mut Vec<u8>,
-    predicate: &mut F,
+    visit_entry: &mut F,
 ) -> std::io::Result<Option<SessionIndexEntry>>
 where
-    F: FnMut(&SessionIndexEntry) -> bool,
+    F: FnMut(&SessionIndexEntry) -> std::io::Result<Option<SessionIndexEntry>>,
 {
     if line_rev.is_empty() {
         return Ok(None);
@@ -222,10 +254,7 @@ where
     let Ok(entry) = serde_json::from_str::<SessionIndexEntry>(trimmed) else {
         return Ok(None);
     };
-    if predicate(&entry) {
-        return Ok(Some(entry));
-    }
-    Ok(None)
+    visit_entry(&entry)
 }
 
 #[cfg(test)]