diff --git a/codex-rs/tui/src/app/tests/startup.rs b/codex-rs/tui/src/app/tests/startup.rs index b0e14416e8..e1f035f7ba 100644 --- a/codex-rs/tui/src/app/tests/startup.rs +++ b/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, ¬ification); + 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; diff --git a/codex-rs/tui/src/app/thread_routing.rs b/codex-rs/tui/src/app/thread_routing.rs index feac96ff86..28a003a66f 100644 --- a/codex-rs/tui/src/app/thread_routing.rs +++ b/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, ¬ification.thread_settings) .await; } - let inferred_session = self - .infer_session_for_thread_notification(thread_id, ¬ification) - .await; + let inferred_session = if let ServerNotification::ThreadStarted(started) = ¬ification + && 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(¬ification); 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 { - 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) }