Avoid rollout reads for configured TUI sessions (#39731)

## Why

A newly started thread may not have materialized its rollout before the TUI
receives `ThreadStarted`. Trying to infer session state from that path can wait
through rollout reader retries even when a lifecycle response already provided
the authoritative session.

## What changed

- Preserve session state already stored for a known thread instead of inferring
  it again from `ThreadStarted`.
- Continue updating agent-picker metadata from the notification.
- Restrict fallback session inference to newly observed `ThreadStarted`
  notifications that do not already have session state.

## Testing

Added a startup test that verifies a known thread routes `ThreadStarted`
immediately, retains its configured session, buffers the notification, and
updates agent metadata when the rollout does not exist yet.

GitOrigin-RevId: 7ed98fb46df17566c4a61ac677f60fc8d94f7321

codex-rs/tui/src/app/tests/startup.rs

@@ -803,6 +803,85 @@ async fn queued_startup_app_event_owns_protected_view_before_draft_restore() ->
     Ok(())
 }
 
+#[tokio::test]
+async fn known_thread_started_preserves_session_without_reading_unmaterialized_rollout() {
+    use futures::FutureExt as _;
+
+    let mut app = make_test_app().await;
+    let temp_dir = tempfile::tempdir().expect("temp dir");
+    let thread_id = ThreadId::new();
+    let session = test_thread_session(thread_id, temp_dir.path().to_path_buf());
+    app.primary_session_configured = Some(session.clone());
+    app.thread_event_channels.insert(
+        thread_id,
+        ThreadEventChannel::new_with_session(
+            THREAD_EVENT_CHANNEL_CAPACITY,
+            session.clone(),
+            Vec::new(),
+        ),
+    );
+    let notification = ThreadStartedNotification {
+        thread: Thread {
+            id: thread_id.to_string(),
+            extra: None,
+            session_id: thread_id.to_string(),
+            forked_from_id: None,
+            parent_thread_id: None,
+            preview: String::new(),
+            ephemeral: false,
+            section: None,
+            section_entered_at: None,
+            project_id: None,
+            history_mode: Default::default(),
+            model_provider: "notification-provider".to_string(),
+            created_at: 1,
+            updated_at: 2,
+            recency_at: Some(2),
+            status: codex_app_server_protocol::ThreadStatus::Idle,
+            path: Some(temp_dir.path().join("not-yet-materialized.jsonl")),
+            cwd: session.cwd.clone(),
+            cli_version: "0.0.0".to_string(),
+            source: codex_app_server_protocol::SessionSource::Unknown,
+            can_accept_direct_input: None,
+            thread_source: None,
+            agent_nickname: Some("Robie".to_string()),
+            agent_role: Some("explorer".to_string()),
+            git_info: None,
+            name: Some("notification title".to_string()),
+            turns: Vec::new(),
+        },
+    };
+
+    tokio::task::unconstrained(app.enqueue_thread_notification(
+        thread_id,
+        ServerNotification::ThreadStarted(notification.clone()),
+    ))
+    .now_or_never()
+    .expect("known sessions must not wait for rollout reads")
+    .expect("thread notification should be routed");
+
+    let store = app.thread_event_channels[&thread_id].store.lock().await;
+    assert_eq!(store.session, Some(session));
+    let Some(ThreadBufferedEvent::Notification(buffered)) = store.buffer.back() else {
+        panic!("thread started notification should remain buffered");
+    };
+    let ServerNotification::ThreadStarted(buffered) = buffered.as_ref() else {
+        panic!("buffered notification should be thread started");
+    };
+    assert_eq!(buffered, &notification);
+    drop(store);
+    assert_eq!(
+        app.agent_navigation.get(&thread_id),
+        Some(&AgentPickerThreadEntry {
+            agent_nickname: Some("Robie".to_string()),
+            agent_role: Some("explorer".to_string()),
+            agent_path: None,
+            is_running: false,
+            is_closed: false,
+        })
+    );
+}
+
 #[tokio::test]
 async fn startup_thread_started_submits_queued_startup_input() {
     let (mut app, _app_event_rx, mut op_rx) = make_test_app_with_channels().await;

codex-rs/tui/src/app/thread_routing.rs

@@ -8,6 +8,7 @@ use super::session_lifecycle::ThreadAttachPresentation;
 use super::*;
 use crate::chatwidget::ThreadInputStateRestoreMode;
 use crate::session_resume::read_session_model;
+use codex_app_server_protocol::ThreadStartedNotification;
 use codex_app_server_protocol::TurnInterruptParams;
 use codex_app_server_protocol::TurnInterruptResponse;
 use codex_app_server_protocol::WarningNotification;
@@ -988,9 +989,32 @@ impl App {
             self.apply_thread_settings_to_cached_session(thread_id, &notification.thread_settings)
                 .await;
         }
-        let inferred_session = self
-            .infer_session_for_thread_notification(thread_id, &notification)
-            .await;
+        let inferred_session = if let ServerNotification::ThreadStarted(started) = &notification
+            && self.primary_session_configured.is_some()
+        {
+            self.upsert_agent_picker_thread(
+                thread_id,
+                started.thread.agent_nickname.clone(),
+                started.thread.agent_role.clone(),
+                /*is_closed*/ false,
+            );
+
+            // Lifecycle responses already contain authoritative session state. Their rollout may
+            // not exist until the first turn, so inferring it again can wait through reader retries.
+            let already_has_session = match self.thread_event_channels.get(&thread_id) {
+                Some(channel) => channel.store.lock().await.session.is_some(),
+                None => false,
+            };
+
+            if already_has_session {
+                None
+            } else {
+                self.infer_session_for_started_thread(thread_id, started)
+                    .await
+            }
+        } else {
+            None
+        };
         let is_turn_started = matches!(notification, ServerNotification::TurnStarted(_));
         let notification_status_change = SideParentStatusChange::for_notification(&notification);
         let (sender, store) = {
@@ -1104,14 +1128,11 @@ impl App {
         }
     }
 
-    pub(super) async fn infer_session_for_thread_notification(
-        &mut self,
+    async fn infer_session_for_started_thread(
+        &self,
         thread_id: ThreadId,
-        notification: &ServerNotification,
+        notification: &ThreadStartedNotification,
     ) -> Option<ThreadSessionState> {
-        let ServerNotification::ThreadStarted(notification) = notification else {
-            return None;
-        };
         let mut session = self.primary_session_configured.clone()?;
         session.thread_id = thread_id;
         session.thread_name = notification.thread.name.clone();
@@ -1128,12 +1149,6 @@ impl App {
         }
         session.message_history = None;
         session.rollout_path = rollout_path;
-        self.upsert_agent_picker_thread(
-            thread_id,
-            notification.thread.agent_nickname.clone(),
-            notification.thread.agent_role.clone(),
-            /*is_closed*/ false,
-        );
         Some(session)
     }