diff --git a/codex-rs/deny.toml b/codex-rs/deny.toml index a5391c4453..714e19b960 100644 --- a/codex-rs/deny.toml +++ b/codex-rs/deny.toml @@ -77,8 +77,7 @@ ignore = [ { id = "RUSTSEC-2025-0057", reason = "fxhash is unmaintained; pulled in via starlark_map under starlark v0.14.2 used by execpolicy/cli/core and bm25 used by core; remove when upstream dependencies drop fxhash" }, { id = "RUSTSEC-2024-0436", reason = "paste is unmaintained; pulled in via starlark/v8 used by execpolicy/code-mode; no fixed release yet" }, { id = "RUSTSEC-2023-0089", reason = "atomic-polyfill is unmaintained; pulled in via postcard/heapless/pagable under starlark v0.14.2 used by execpolicy/cli/core; no fixed starlark release yet" }, - # TODO(fcoury): remove this exception when syntect drops yaml-rust and bincode, or updates to versions that have fixed the vulnerabilities. - { id = "RUSTSEC-2024-0320", reason = "yaml-rust is unmaintained; pulled in via syntect v5.3.0 used by codex-tui for syntax highlighting; no fixed release yet" }, + # TODO(fcoury): remove this exception when syntect drops bincode, or updates to a version that has fixed the vulnerability. { id = "RUSTSEC-2025-0141", reason = "bincode is unmaintained; pulled in via syntect v5.3.0 used by codex-tui for syntax highlighting; no fixed release yet" }, { id = "RUSTSEC-2026-0118", reason = "hickory-proto v0.25.2 is pulled in via rama-dns/rama-tcp used by codex-network-proxy; DNSSEC features are not enabled; remove when rama updates to hickory 0.26.1 or hickory-net" }, { id = "RUSTSEC-2026-0119", reason = "hickory-proto v0.25.2 is pulled in via rama-dns/rama-tcp used by codex-network-proxy; no fixed rama release is available yet; remove when rama updates to hickory 0.26.1 or hickory-net" }, diff --git a/codex-rs/tui/src/app.rs b/codex-rs/tui/src/app.rs index 43ebfd2454..2bd7c36370 100644 --- a/codex-rs/tui/src/app.rs +++ b/codex-rs/tui/src/app.rs @@ -216,6 +216,7 @@ pub(crate) mod app_server_requests; mod app_server_thread_ownership; mod backend_banner_fallback; mod backend_events; +mod backend_selection; pub(crate) mod backend_sessions; mod background_requests; mod composer_hints; diff --git a/codex-rs/tui/src/app/agents_overview.rs b/codex-rs/tui/src/app/agents_overview.rs index 2b3fc447f8..8559f1272f 100644 --- a/codex-rs/tui/src/app/agents_overview.rs +++ b/codex-rs/tui/src/app/agents_overview.rs @@ -872,23 +872,8 @@ impl App { let pending_settings = restored_blank_session .then(|| input_state.pending_thread_settings.take()) .flatten(); - let preserve_in_flight_turn = !read_only - && self - .active_turn_id_for_thread(root_thread_id) - .await - .is_some(); - self.chat_widget.restore_thread_input_state( - Some(input_state), - ThreadInputStateRestoreMode { - preserve_in_flight_turn, - }, - ); - if let Some(settings) = pending_settings { - self.chat_widget.on_thread_settings_updated(settings); - } - if !preserve_in_flight_turn { - self.chat_widget.maybe_send_next_queued_input(); - } + self.restore_session_draft(root_thread_id, input_state, pending_settings) + .await; } if !read_only && !is_new_session && !self.chat_widget.fork_in_progress { self.maybe_prompt_resume_paused_goal_after_resume(app_server, root_thread_id) @@ -898,6 +883,28 @@ impl App { Ok(AppRunControl::Continue) } + pub(super) async fn restore_session_draft( + &mut self, + thread_id: ThreadId, + input: ThreadInputState, + pending_settings: Option, + ) { + let preserve_in_flight_turn = !self.chat_widget.is_external_writer_view() + && self.active_turn_id_for_thread(thread_id).await.is_some(); + self.chat_widget.restore_thread_input_state( + Some(input), + ThreadInputStateRestoreMode { + preserve_in_flight_turn, + }, + ); + if let Some(settings) = pending_settings { + self.chat_widget.on_thread_settings_updated(settings); + } + if !preserve_in_flight_turn { + self.chat_widget.maybe_send_next_queued_input(); + } + } + pub(super) async fn replay_agents_overview_requests( &mut self, app_server: &AppServerSession, diff --git a/codex-rs/tui/src/app/backend_events.rs b/codex-rs/tui/src/app/backend_events.rs index 3f5d599373..f933611f19 100644 --- a/codex-rs/tui/src/app/backend_events.rs +++ b/codex-rs/tui/src/app/backend_events.rs @@ -2,11 +2,12 @@ //! Connection-wide lifecycle and account events may only update the displayed conversation's UI. use super::App; +use super::app_server_event_targets::ServerNotificationThreadTarget; +use super::app_server_event_targets::server_notification_thread_target; use crate::app_server_session::AppServerSession; use crate::app_server_session::AppServerSessionEvent; use crate::backend_id::BackendId; use codex_app_server_client::AppServerEvent; -use codex_app_server_protocol::ServerNotification; impl App { pub(super) async fn handle_polled_backend_event( @@ -15,20 +16,16 @@ impl App { displayed_owner: BackendId, event: AppServerSessionEvent, ) { - if event.backend_id != displayed_owner - && match &event.event { - AppServerEvent::Disconnected { .. } | AppServerEvent::Lagged { .. } => true, - AppServerEvent::ServerNotification(notification) => matches!( - notification.as_ref(), - ServerNotification::AccountUpdated(_) - | ServerNotification::AccountRateLimitsUpdated(_) - ), - AppServerEvent::ServerRequest(_) => false, + if displayed_owner != event.backend_id { + // A parked voice session still needs its own notifications. + if event.backend_id == app_server.backend_id() + && let AppServerEvent::ServerNotification(notification) = &event.event + && let ServerNotificationThreadTarget::Thread(thread_id) = + server_notification_thread_target(notification) + && self.voice_owner_thread_id() == Some(thread_id) + { + self.deliver_background_voice_notification(thread_id, notification); } - { - tracing::debug!( - "ignoring a background backend event that targets displayed conversation state" - ); return; } self.handle_backend_event(app_server, event).await; diff --git a/codex-rs/tui/src/app/backend_selection.rs b/codex-rs/tui/src/app/backend_selection.rs new file mode 100644 index 0000000000..5e777ba283 --- /dev/null +++ b/codex-rs/tui/src/app/backend_selection.rs @@ -0,0 +1,137 @@ +//! Prepare saved conversations through their owner without mutating the UI on failure. +use super::backend_sessions::BackendSessions; +use super::resume_preparation::ResumeSource; +use super::*; +use codex_app_server_protocol::ThreadItemPayload; + +impl App { + pub(super) fn resume_backend_session<'a>( + &'a mut self, + tui: &'a mut tui::Tui, + backends: &'a mut BackendSessions, + target: SessionTarget, + ) -> futures::future::LocalBoxFuture<'a, Result> { + Box::pin(async move { + let displayed_session = backends.displayed; + let owner = target.backend_id; + let outgoing = backends.displayed_owner(); + if owner == outgoing { + return self + .apply_resume_picker_selection( + tui, + backends.session_at_mut(displayed_session), + SessionSelection::Resume(target), + ) + .await; + } + if self.windows_sandbox_blocks_thread_switch() + || self.reject_pending_permission_root_switch() + { + return Ok(AppRunControl::Continue); + } + if let Some(thread_id) = self.current_displayed_thread_id() + && self.active_turn_id_for_thread(thread_id).await.is_some() + { + self.add_session_picker_error( + "Wait for the current turn to finish before switching backends.".into(), + ); + return Ok(AppRunControl::Continue); + } + let Some(settings) = backends.settings.get(&owner).cloned() else { + self.add_session_picker_error( + "This backend has no connection settings. Open the session picker again." + .into(), + ); + return Ok(AppRunControl::Continue); + }; + let Some(destination_slot) = backends.listening_slot(owner) else { + self.add_session_picker_error( + "This session's backend is unavailable. Open the session picker again.".into(), + ); + return Ok(AppRunControl::Continue); + }; + let destination = backends.session_at_mut(destination_slot); + let bootstrap = match destination.bootstrap(&settings.config).await { + Ok(bootstrap) => bootstrap, + Err(error) => { + let message = format!("Failed to prepare session backend: {error}"); + self.add_session_picker_error(message); + return Ok(AppRunControl::Continue); + } + }; + let prepared = match self + .prepare_target_session(tui, destination, &target, ResumeSource::Backend(&settings)) + .await + { + Ok(prepared) => prepared, + Err(control) => return Ok(control), + }; + self.capture_session_inputs().await; + for channel in self.thread_event_channels.values_mut() { + channel.backend_id.get_or_insert(outgoing); + } + backends.settings.insert(outgoing, self.backend_settings()); + self.detach_current_thread_for_navigation( + backends.session_at_mut(displayed_session), + Some(target.thread_id), + ) + .await; + let destination = backends.session_at_mut(destination_slot); + self.store_active_thread_receiver().await; + self.clear_active_thread().await; + self.agents_overview.pending_usage = None; + self.agents_overview.usage_disabled = false; + self.app_event_tx = AppEventSender::new(self.app_event_tx.app_event_tx.clone()); + self.app_server_target = settings.target; + self.reconnect = Default::default(); + if let Ok(mut state) = self.agents_overview.view_state.lock() { + state.connection_notice = None; + } + self.rate_limit_hard_stop_generation = + self.rate_limit_hard_stop_generation.wrapping_add(1); + self.rate_limit_refresh_state.invalidate_recovery(); + self.account_email_request_id = None; + self.file_search.on_user_query(String::new()); + self.bind_backend_workspace(destination, prepared.config.cwd.to_path_buf()); + self.model_catalog = Arc::new( + ModelCatalog::new(bootstrap.available_models) + .with_collaboration_modes(bootstrap.collaboration_modes), + ); + self.feedback_audience = bootstrap.feedback_audience; + self.chat_widget.requires_openai_auth = bootstrap.requires_openai_auth; + self.chat_widget.invalidate_security_setup(); + self.chat_widget.update_account_state( + bootstrap.status_account_display, + bootstrap.plan_type, + bootstrap.has_chatgpt_account, + matches!(bootstrap.auth_mode, Some(TelemetryAuthMode::Chatgpt)), + ); + let thread_id = target.thread_id; + if let Some(input) = self.agents_overview.input_states.get_mut(&thread_id) { + input.provider_error = None; + for item in prepared.thread.turns.iter().flat_map(Turn::item_payloads) { + if let ThreadItemPayload::UserMessage { + client_id: Some(id), + .. + } = item + { + input.pending_steers.retain(|steer| steer.client_id != *id); + } + } + } + let control = self + .activate_prepared_session(tui, destination, prepared) + .await?; + backends.displayed = destination_slot; + if let Some(mut input) = self.agents_overview.input_states.remove(&thread_id) { + let blank = self.agents_overview.blank_sessions.contains_key(&thread_id); + let settings = blank + .then(|| input.pending_thread_settings.take()) + .flatten(); + self.restore_session_draft(thread_id, input, settings).await; + } + tui.frame_requester().schedule_frame(); + Ok(control) + }) + } +} diff --git a/codex-rs/tui/src/app/backend_sessions.rs b/codex-rs/tui/src/app/backend_sessions.rs index 4cfecb12c9..5112df4530 100644 --- a/codex-rs/tui/src/app/backend_sessions.rs +++ b/codex-rs/tui/src/app/backend_sessions.rs @@ -1,9 +1,14 @@ //! Owns backend sessions independently of the displayed conversation's UI state. //! Polling is fair and cancellation-safe; closing one stream never closes another. +use super::resume_preparation::BackendSettings; use crate::app_server_session::AppServerSession; use crate::app_server_session::AppServerSessionEvent; use crate::backend_id::BackendId; +use codex_app_server_client::AppServerEvent; +use codex_app_server_protocol::ServerNotification; +use std::collections::HashMap; +use std::collections::VecDeque; use std::future::Future; use std::task::Poll; @@ -42,6 +47,7 @@ impl InitialBackendSessions { struct SessionEntry { session: AppServerSession, listening: bool, + pending_requests: VecDeque, } /// A session's stable position in an append-only backend collection. @@ -50,6 +56,8 @@ pub(super) struct SessionSlot(usize); pub(super) struct BackendSessions { entries: Vec, + pub(super) displayed: SessionSlot, + pub(super) settings: HashMap, next_source: usize, } @@ -65,6 +73,8 @@ impl BackendSessions { pub(super) fn new(session: AppServerSession) -> Self { let mut sessions = Self { entries: Vec::new(), + displayed: SessionSlot(0), + settings: HashMap::new(), next_source: 0, }; sessions.insert(session); @@ -76,6 +86,7 @@ impl BackendSessions { self.entries.push(SessionEntry { session, listening: true, + pending_requests: VecDeque::new(), }); } @@ -94,7 +105,12 @@ impl BackendSessions { &self.entries[slot.0].session } - pub(super) fn picker_views(&self, displayed_owner: BackendId) -> Vec { + pub(super) fn displayed_owner(&self) -> BackendId { + self.session_at(self.displayed).backend_id() + } + + pub(super) fn picker_views(&self) -> Vec { + let displayed_owner = self.displayed_owner(); self.entries .iter() .filter(|entry| entry.listening || entry.session.backend_id() == displayed_owner) @@ -106,10 +122,13 @@ impl BackendSessions { &mut self.entries[slot.0].session } + pub(super) fn listening_slot(&self, owner: BackendId) -> Option { + self.slot(owner) + .filter(|slot| self.entries[slot.0].listening) + } + pub(super) fn is_listening(&self, owner: BackendId) -> bool { - self.entries - .iter() - .any(|entry| entry.session.backend_id() == owner && entry.listening) + self.listening_slot(owner).is_some() } pub(super) fn resume_events(&mut self, owner: BackendId) { @@ -119,6 +138,7 @@ impl BackendSessions { .find(|entry| entry.session.backend_id() == owner) { entry.listening = true; + entry.pending_requests.clear(); } } @@ -126,6 +146,7 @@ impl BackendSessions { &mut self, paused_backend: Option, ) -> (BackendId, Option) { + let request_owner = self.displayed_owner(); std::future::poll_fn(|cx| { for offset in 0..self.entries.len() { let index = (self.next_source + offset) % self.entries.len(); @@ -134,14 +155,45 @@ impl BackendSessions { if !entry.listening || paused_backend == Some(owner) { continue; } - // Both client transports receive from cancellation-safe queues. Dropping a - // pending receive leaves its event available for the next main-loop poll. - let event = std::pin::pin!(entry.session.next_event()).poll(cx); - if let Poll::Ready(event) = event { - entry.listening = event.is_some(); - self.next_source = (index + 1) % self.entries.len(); - return Poll::Ready((owner, event)); + let event = if owner == request_owner + && let Some(event) = entry.pending_requests.pop_front() + { + Some(event) + } else { + // Both transports have cancellation-safe event queues. + let Poll::Ready(event) = std::pin::pin!(entry.session.next_event()).poll(cx) + else { + continue; + }; + event + }; + let event = match event { + Some(event) + if owner != request_owner + && matches!(&event.event, AppServerEvent::ServerRequest(_)) => + { + entry.pending_requests.push_back(event); + cx.waker().wake_by_ref(); + continue; + } + event => event, + }; + if let Some(AppServerSessionEvent { event: AppServerEvent::ServerNotification(notification), .. }) = &event + && let ServerNotification::ServerRequestResolved(resolved) = notification.as_ref() + { + entry.pending_requests.retain(|pending| !matches!( + &pending.event, AppServerEvent::ServerRequest(request) if request.id() == &resolved.request_id + )); } + if event + .as_ref() + .is_none_or(|event| matches!(event.event, AppServerEvent::Disconnected { .. })) + { + entry.pending_requests.clear(); + } + entry.listening = event.is_some(); + self.next_source = (index + 1) % self.entries.len(); + return Poll::Ready((owner, event)); } Poll::Pending }) diff --git a/codex-rs/tui/src/app/event_dispatch.rs b/codex-rs/tui/src/app/event_dispatch.rs index a9c64f516b..adb3f9eece 100644 --- a/codex-rs/tui/src/app/event_dispatch.rs +++ b/codex-rs/tui/src/app/event_dispatch.rs @@ -362,7 +362,7 @@ impl App { } } } - AppEvent::StartupThreadStarted { result } => { + AppEvent::StartupThreadStarted { result, .. } => { self.handle_startup_thread_started(app_server, result) .await?; } @@ -597,6 +597,9 @@ impl App { Err(err) => return Err(err), } } + AppEvent::ResumeBackendSession(_) => { + unreachable!("backend selection is handled by the session manager") + } AppEvent::ArchiveCurrentThread => { return self.archive_current_thread(tui, app_server).await; } diff --git a/codex-rs/tui/src/app/reconnect.rs b/codex-rs/tui/src/app/reconnect.rs index 91dde41810..e48819ac1f 100644 --- a/codex-rs/tui/src/app/reconnect.rs +++ b/codex-rs/tui/src/app/reconnect.rs @@ -24,7 +24,7 @@ pub(super) struct ReconnectState { } pub(super) struct Reconnected { - session: AppServerSession, + pub(super) session: AppServerSession, bootstrap: AppServerBootstrap, thread: Option, } @@ -305,6 +305,32 @@ impl App { true } + pub(super) fn bind_backend_workspace( + &mut self, + app_server: &AppServerSession, + search_dir: PathBuf, + ) { + self.chat_widget.snapshot_local_images = self.app_server_target.uses_remote_workspace(); + self.chat_widget.set_local_worktree_operations( + !crate::uses_remote_workspace_or_environment( + &self.app_server_target, + self.environment_manager.as_ref(), + ), + ); + self.chat_widget.windows_sandbox_local_server = + !self.app_server_target.uses_remote_workspace() + && app_server.app_server_platform_os() == Some("windows"); + self.chat_widget.remote_connection = + crate::status::remote_connection::remote_connection_status_value( + &self.app_server_target, + app_server.server_version(), + ); + self.workspace_command_runner = Some(Arc::new(AppServerWorkspaceCommandRunner::new( + app_server.request_handle(), + ))); + self.file_search = FileSearchManager::new(search_dir, self.app_event_tx.clone()); + } + pub(super) async fn finish_reconnect( &mut self, tui: &mut tui::Tui, @@ -370,16 +396,6 @@ impl App { self.chat_widget.windows_sandbox_elevated_setup_complete = false; } } - self.chat_widget.snapshot_local_images = self.app_server_target.uses_remote_workspace(); - self.chat_widget.set_local_worktree_operations( - !crate::uses_remote_workspace_or_environment( - &self.app_server_target, - self.environment_manager.as_ref(), - ), - ); - self.chat_widget.windows_sandbox_local_server = - !self.app_server_target.uses_remote_workspace() - && app_server.app_server_platform_os() == Some("windows"); self.chat_widget.windows_sandbox_host = WindowsSandboxHost::Unknown; self.chat_widget.requires_openai_auth = bootstrap.requires_openai_auth; let managed_daemon = self @@ -387,19 +403,10 @@ impl App { .remote_connection .as_ref() .is_some_and(|connection| connection.is_managed_daemon); - self.chat_widget.remote_connection = - crate::status::remote_connection::remote_connection_status_value( - &self.app_server_target, - app_server.server_version(), - ); + self.bind_backend_workspace(app_server, self.config.cwd.to_path_buf()); if let Some(connection) = &mut self.chat_widget.remote_connection { connection.is_managed_daemon = connection.is_local_daemon && managed_daemon; } - self.workspace_command_runner = Some(Arc::new(AppServerWorkspaceCommandRunner::new( - app_server.request_handle(), - ))); - self.file_search = - FileSearchManager::new(self.config.cwd.to_path_buf(), self.app_event_tx.clone()); self.model_catalog = Arc::new( ModelCatalog::new(bootstrap.available_models) .with_collaboration_modes(bootstrap.collaboration_modes), @@ -444,6 +451,7 @@ impl App { // so their late requests cannot leak into recovery. Background threads attach on selection. for channel in self.thread_event_channels.values_mut() { let mut replacement = ThreadEventChannel::new(THREAD_EVENT_CHANNEL_CAPACITY); + replacement.backend_id = channel.backend_id; if channel.attachment() == ThreadEventAttachment::ExternalWriter { replacement.mark_external_writer(); } else { diff --git a/codex-rs/tui/src/app/resume_preparation.rs b/codex-rs/tui/src/app/resume_preparation.rs index 77ea9390c1..55b8791adc 100644 --- a/codex-rs/tui/src/app/resume_preparation.rs +++ b/codex-rs/tui/src/app/resume_preparation.rs @@ -21,7 +21,6 @@ pub(super) struct BackendSettings { #[derive(Clone, Copy)] pub(super) enum ResumeSource<'a> { Active, - #[cfg_attr(not(test), allow(dead_code, reason = "Used by backend handoff."))] Backend(&'a BackendSettings), } @@ -36,7 +35,6 @@ pub(super) struct PreparedResume { } impl App { - #[cfg_attr(not(test), allow(dead_code, reason = "Used by backend handoff."))] pub(super) fn backend_settings(&self) -> BackendSettings { BackendSettings { config: self.config.clone(), diff --git a/codex-rs/tui/src/app/session_lifecycle.rs b/codex-rs/tui/src/app/session_lifecycle.rs index 8f8416f545..a9ac5e5cdb 100644 --- a/codex-rs/tui/src/app/session_lifecycle.rs +++ b/codex-rs/tui/src/app/session_lifecycle.rs @@ -1478,11 +1478,7 @@ impl App { .await; } } - Err(err) => { - self.add_session_picker_error(format!( - "Failed to attach to resumed app-server thread: {err}" - )); - } + Err(err) => return Err(err), } Ok(AppRunControl::Continue) diff --git a/codex-rs/tui/src/app/session_picker.rs b/codex-rs/tui/src/app/session_picker.rs index be8edb7089..37433d7233 100644 --- a/codex-rs/tui/src/app/session_picker.rs +++ b/codex-rs/tui/src/app/session_picker.rs @@ -4,13 +4,28 @@ use super::*; use crate::chatwidget::ThreadInputStateRestoreMode; impl App { + pub(super) async fn capture_session_inputs(&mut self) { + for (id, channel) in &self.thread_event_channels { + if let Some(input_state) = channel.store.lock().await.input_state.clone() { + self.agents_overview.input_states.insert(*id, input_state); + } + } + if let Some(current_thread_id) = self.current_displayed_thread_id() + && let Some(input_state) = self.chat_widget.capture_thread_input_state() + { + self.agents_overview + .input_states + .insert(current_thread_id, input_state); + } + } + pub(super) async fn open_resume_picker_for_backends( &mut self, tui: &mut tui::Tui, backends: &mut super::backend_sessions::BackendSessions, - displayed_session: super::backend_sessions::SessionSlot, ) -> Result { - let displayed_owner = backends.session_at(displayed_session).backend_id(); + let displayed_session = backends.displayed; + let displayed_owner = backends.displayed_owner(); let selection = crate::resume_picker::run_resume_picker_from_existing_session_with_app_servers( crate::uses_remote_workspace_or_environment( @@ -23,7 +38,7 @@ impl App { /*show_all*/ false, /*include_non_interactive*/ false, crate::resume_picker::ResumePickerAppServers::new( - backends.picker_views(displayed_owner), + backends.picker_views(), displayed_owner, ), self.primary_thread_id @@ -55,25 +70,15 @@ impl App { ) -> Result { match selection { SessionSelection::Resume(target_session) => { - if let Err(error) = target_session.ensure_backend(app_server.backend_id()) { - self.add_session_picker_error(error.to_string()); + if target_session.backend_id != app_server.backend_id() { + self.app_event_tx + .send(AppEvent::ResumeBackendSession(target_session)); return Ok(AppRunControl::Continue); } let thread_id = target_session.thread_id; let switching_threads = self.current_displayed_thread_id() != Some(thread_id); if switching_threads { - for (id, channel) in &self.thread_event_channels { - if let Some(input_state) = channel.store.lock().await.input_state.clone() { - self.agents_overview.input_states.insert(*id, input_state); - } - } - if let Some(current_thread_id) = self.current_displayed_thread_id() - && let Some(input_state) = self.chat_widget.capture_thread_input_state() - { - self.agents_overview - .input_states - .insert(current_thread_id, input_state); - } + self.capture_session_inputs().await; } if let AppRunControl::Exit(reason) = self .resume_target_session(tui, app_server, target_session) diff --git a/codex-rs/tui/src/app/startup.rs b/codex-rs/tui/src/app/startup.rs index 7afd8e4791..ac92bc0176 100644 --- a/codex-rs/tui/src/app/startup.rs +++ b/codex-rs/tui/src/app/startup.rs @@ -24,6 +24,7 @@ fn spawn_startup_thread_start( app_event_tx: AppEventSender, worktree: Option, ) { + let backend_id = app_server.backend_id(); let request_handle = app_server.request_handle(); let thread_params_mode = app_server.thread_params_mode(); let remote_cwd_override = app_server.remote_cwd_override().map(Path::to_path_buf); @@ -45,7 +46,7 @@ fn spawn_startup_thread_start( } Ok(started) }); - app_event_tx.send(AppEvent::StartupThreadStarted { result }); + app_event_tx.send(AppEvent::StartupThreadStarted { backend_id, result }); }); } @@ -1280,7 +1281,6 @@ See the Codex keymap documentation for supported actions and examples." #[cfg(debug_assertions)] let pre_loop_exit_reason: Option = None; - let initial_owner = app_server.backend_id(); let mut backends = super::backend_sessions::BackendSessions::from_initial(app_server, other_backends); @@ -1288,12 +1288,9 @@ See the Codex keymap documentation for supported actions and examples." Ok(exit_reason) } else { loop { - let Some(displayed_session) = backends.slot(initial_owner) else { - break Err(color_eyre::eyre::eyre!( - "the initial backend is not registered" - )); - }; - if !backends.is_listening(initial_owner) + let displayed_session = backends.displayed; + let displayed_owner = backends.displayed_owner(); + if !backends.is_listening(displayed_owner) && !app.pending_startup_thread_start && !app.reconnect.offline { @@ -1311,13 +1308,7 @@ See the Codex keymap documentation for supported actions and examples." } if app.pending_open_resume_picker { app.pending_open_resume_picker = false; - match Box::pin(app.open_resume_picker_for_backends( - tui, - &mut backends, - displayed_session, - )) - .await - { + match Box::pin(app.open_resume_picker_for_backends(tui, &mut backends)).await { Ok(AppRunControl::Continue) => {} Ok(AppRunControl::Exit(reason)) => break Ok(reason), Err(err) if app.recover_transport_error(&err) => {} @@ -1372,9 +1363,9 @@ See the Codex keymap documentation for supported actions and examples." if app.reconnect.offline && !app.reconnect.failed && reconnect.is_none() { let presentation = app.reconnect.presentation; let wait_for_drain = - app.pending_startup_thread_start && backends.is_listening(initial_owner); + app.pending_startup_thread_start && backends.is_listening(displayed_owner); let attempt = reconnect::reconnect( - initial_owner, + displayed_owner, app.app_server_target.clone(), app.config.clone(), app.local_settings.clone(), @@ -1388,7 +1379,7 @@ See the Codex keymap documentation for supported actions and examples." .thread_tool_transport(), presentation, ); - reconnect = Some(Box::pin(async move { + let future = Box::pin(async move { if let reconnect::ReconnectPresentation::Startup(deadline) = presentation && wait_for_drain { @@ -1398,7 +1389,8 @@ See the Codex keymap documentation for supported actions and examples." ); } attempt.await - })); + }); + reconnect = Some((displayed_owner, future)); } // Replay queues history and operations. A buffered closure must not switch // widgets before those app events have been applied. @@ -1406,7 +1398,7 @@ See the Codex keymap documentation for supported actions and examples." let paused_backend = ((app.reconnect.offline && !app.pending_startup_thread_start) || (!matches!(app.app_server_target, AppServerTarget::Embedded) && has_pending_app_events)) - .then_some(initial_owner); + .then_some(displayed_owner); let initial_session_header_pending = waiting_for_initial_session_header && app.primary_session_configured.is_some() && has_pending_app_events; @@ -1450,13 +1442,26 @@ See the Codex keymap documentation for supported actions and examples." AppRunControl::Continue } Some(event) = app_event_rx.recv() => { + if let AppEvent::ResumeBackendSession(target) = event { + match app.resume_backend_session(tui, &mut backends, target).await { + Ok(AppRunControl::Continue) => continue, + Ok(AppRunControl::Exit(reason)) => break Ok(reason), + Err(error) => break Err(error), + } + } let is_initial_session_header = matches!( &event, AppEvent::InsertHistoryCell(cell) if cell.as_any().is::() ); + let event_session = match &event { + AppEvent::StartupThreadStarted { backend_id, .. } => { + backends.slot(*backend_id).unwrap_or(displayed_session) + } + _ => displayed_session, + }; let had_active_modal = app.chat_widget.has_active_modal(); - match Box::pin(app.handle_event(tui, backends.session_at_mut(displayed_session), event)).await { + match Box::pin(app.handle_event(tui, backends.session_at_mut(event_session), event)).await { Ok(AppRunControl::Continue) => { if is_initial_session_header { waiting_for_initial_session_header = false; @@ -1528,14 +1533,14 @@ See the Codex keymap documentation for supported actions and examples." } (owner, app_server_event) = backends.next_event(paused_backend) => { match app_server_event { - Some(event) if owner != initial_owner || !app.reconnect.offline => { + Some(event) if owner != displayed_owner || !app.reconnect.offline => { if let Some(session) = backends.session(owner) { - app.handle_polled_backend_event(session, initial_owner, event).await; + app.handle_polled_backend_event(session, displayed_owner, event).await; } } Some(_) => {} None => { - if owner == initial_owner { + if owner == displayed_owner { if app.reconnect.offline && app.pending_startup_thread_start { reconnect = None; } else if !app.pending_startup_thread_start { @@ -1547,14 +1552,18 @@ See the Codex keymap documentation for supported actions and examples." } AppRunControl::Continue } - result = async { match reconnect.as_mut() { Some(future) => future.await, None => std::future::pending().await } }, if reconnect.is_some() && !has_pending_app_events => { + (reconnect_owner, result) = async { match reconnect.as_mut() { Some((owner, future)) => (*owner, future.await), None => std::future::pending().await } }, if reconnect.is_some() && !has_pending_app_events => { reconnect = None; match result { + Ok(connected) if reconnect_owner != displayed_owner => { + let _ = connected.session.shutdown().await; + } + Err(_) if reconnect_owner != displayed_owner => {} Ok(connected) => { if let Err(err) = app.finish_reconnect(tui, backends.session_at_mut(displayed_session), &mut app_event_rx, connected, CODEX_CLI_VERSION).await { break Err(err); } - backends.resume_events(initial_owner); + backends.resume_events(displayed_owner); waiting_for_initial_session_configured = app.pending_startup_thread_start; if app.pending_startup_thread_start { spawn_startup_thread_start( @@ -1620,7 +1629,7 @@ See the Codex keymap documentation for supported actions and examples." } None => std::future::pending().await, } - }, if backends.is_listening(initial_owner) => { + }, if backends.is_listening(displayed_owner) => { app.refresh_rate_limits(backends.session_at(displayed_session), RateLimitRefreshOrigin::Periodic); AppRunControl::Continue } diff --git a/codex-rs/tui/src/app/tests/backend_ownership_tests.rs b/codex-rs/tui/src/app/tests/backend_ownership_tests.rs index 92f22bfe89..4dcbc4d671 100644 --- a/codex-rs/tui/src/app/tests/backend_ownership_tests.rs +++ b/codex-rs/tui/src/app/tests/backend_ownership_tests.rs @@ -43,6 +43,13 @@ async fn foreign_backend_selection_cannot_resume_same_id_or_unarchive() -> Resul SessionSelection::Resume(target.clone()), ) .await?; + assert!( + std::iter::from_fn(|| events.try_recv().ok()).any(|event| matches!( + event, + AppEvent::ResumeBackendSession(routed) + if routed.backend_id == target.backend_id && routed.thread_id == target.thread_id + )) + ); app.resume_target_session(&mut tui, &mut server, target.clone()) .await?; assert_eq!(app.current_displayed_thread_id(), Some(thread_id)); @@ -62,7 +69,6 @@ async fn foreign_backend_selection_cannot_resume_same_id_or_unarchive() -> Resul .collect(); insta::assert_snapshot!(errors.join("\n"), @r###" ■ This session belongs to a different backend. Open the session picker again. - ■ This session belongs to a different backend. Open the session picker again. "###); for action in [ SessionStartAction::Resume( diff --git a/codex-rs/tui/src/app/tests/backend_sessions_tests.rs b/codex-rs/tui/src/app/tests/backend_sessions_tests.rs index cb723322d5..9dcf00d9a7 100644 --- a/codex-rs/tui/src/app/tests/backend_sessions_tests.rs +++ b/codex-rs/tui/src/app/tests/backend_sessions_tests.rs @@ -2,11 +2,161 @@ use super::*; use crate::app::backend_sessions::BackendSessions; +use crate::app_server_session::ThreadParamsMode; +use crate::backend_id::BackendId; +use codex_app_server_client::AppServerClient; +use codex_app_server_client::AppServerEvent; +use codex_app_server_client::CloudClient; use codex_app_server_protocol::RequestId; use codex_app_server_protocol::ThreadSettingsUpdateResponse; use futures::FutureExt; use pretty_assertions::assert_eq; +fn approval_request(thread_id: ThreadId, request_id: i64) -> ServerRequest { + ServerRequest::FileChangeRequestApproval { + request_id: AppServerRequestId::Integer(request_id), + params: FileChangeRequestApprovalParams { + thread_id: thread_id.to_string(), + turn_id: "turn-1".to_string(), + item_id: "item-1".to_string(), + started_at_ms: 0, + reason: None, + grant_root: None, + }, + } +} + +#[tokio::test] +async fn backend_sessions_parked_request_does_not_hide_disconnect() -> Result<()> { + let [ + (displayed, _displayed_events), + (background, background_events), + ] = std::array::from_fn(|_| { + let (client, _, events) = CloudClient::channel(/*capacity*/ 1); + ( + AppServerSession::new( + BackendId::new(), + AppServerClient::Cloud(client), + ThreadParamsMode::Cloud, + ), + events, + ) + }); + let background_id = background.backend_id(); + let mut backends = BackendSessions::from_initial(displayed, vec![background]); + let initial = backends.displayed; + let thread_id = ThreadId::new(); + + background_events + .send(AppServerEvent::ServerRequest(Box::new(approval_request( + thread_id, /*request_id*/ 1, + )))) + .unwrap(); + assert!( + backends + .next_event(/*paused_backend*/ None) + .now_or_never() + .is_none() + ); + backends.displayed = backends.slot(background_id).unwrap(); + let (owner, parked) = backends.next_event(/*paused_backend*/ None).await; + assert_eq!(owner, background_id); + assert!(matches!( + parked.unwrap().event, + AppServerEvent::ServerRequest(_) + )); + + backends.displayed = initial; + background_events + .send(AppServerEvent::ServerRequest(Box::new(approval_request( + thread_id, /*request_id*/ 2, + )))) + .unwrap(); + background_events + .send(AppServerEvent::ServerRequest(Box::new(approval_request( + thread_id, /*request_id*/ 3, + )))) + .unwrap(); + assert!( + backends + .next_event(/*paused_backend*/ None) + .now_or_never() + .is_none() + ); + assert!( + backends + .next_event(/*paused_backend*/ None) + .now_or_never() + .is_none() + ); + background_events + .send(AppServerEvent::ServerNotification(Box::new( + ServerNotification::ServerRequestResolved( + codex_app_server_protocol::ServerRequestResolvedNotification { + thread_id: thread_id.to_string(), + request_id: AppServerRequestId::Integer(2), + }, + ), + ))) + .unwrap(); + let (owner, resolved) = backends.next_event(/*paused_backend*/ None).await; + assert_eq!(owner, background_id); + assert!(matches!( + resolved.unwrap().event, + AppServerEvent::ServerNotification(notification) + if matches!( + notification.as_ref(), + ServerNotification::ServerRequestResolved(resolved) + if resolved.request_id == AppServerRequestId::Integer(2) + ) + )); + backends.displayed = backends.slot(background_id).unwrap(); + let (owner, unresolved) = backends.next_event(/*paused_backend*/ None).await; + assert_eq!(owner, background_id); + assert!(matches!( + unresolved.unwrap().event, + AppServerEvent::ServerRequest(request) + if request.id() == &AppServerRequestId::Integer(3) + )); + assert!( + backends + .next_event(/*paused_backend*/ None) + .now_or_never() + .is_none() + ); + + backends.displayed = initial; + background_events + .send(AppServerEvent::ServerRequest(Box::new(approval_request( + thread_id, /*request_id*/ 4, + )))) + .unwrap(); + assert!( + backends + .next_event(/*paused_backend*/ None) + .now_or_never() + .is_none() + ); + background_events + .send(AppServerEvent::Disconnected { + message: "stream ended".to_string(), + }) + .unwrap(); + drop(background_events); + let (owner, disconnected) = backends.next_event(/*paused_backend*/ None).await; + assert_eq!(owner, background_id); + assert!(matches!( + disconnected.unwrap().event, + AppServerEvent::Disconnected { .. } + )); + backends.displayed = backends.slot(background_id).unwrap(); + let (owner, closed) = backends.next_event(/*paused_backend*/ None).await; + assert_eq!(owner, background_id); + assert!(closed.is_none()); + assert!(!backends.is_listening(background_id)); + Ok(()) +} + #[tokio::test] async fn backend_sessions_poll_three_sources_and_isolate_closed_or_paused_streams() -> Result<()> { let mut app = make_test_app().await; diff --git a/codex-rs/tui/src/app/tests/disconnect_tests.rs b/codex-rs/tui/src/app/tests/disconnect_tests.rs index 15156ba47e..f54d0eeed3 100644 --- a/codex-rs/tui/src/app/tests/disconnect_tests.rs +++ b/codex-rs/tui/src/app/tests/disconnect_tests.rs @@ -475,10 +475,11 @@ async fn lost_initial_thread_reply_keeps_startup_draft_offline() -> Result<()> { ) .await; } + let backend_id = session.backend_id(); app.handle_event( &mut tui, &mut session, - AppEvent::StartupThreadStarted { result }, + AppEvent::StartupThreadStarted { backend_id, result }, ) .await?; assert!(app.reconnect.offline); diff --git a/codex-rs/tui/src/app/tests/new_session_tests.rs b/codex-rs/tui/src/app/tests/new_session_tests.rs index 4c6c7ef2c5..6ac99972f7 100644 --- a/codex-rs/tui/src/app/tests/new_session_tests.rs +++ b/codex-rs/tui/src/app/tests/new_session_tests.rs @@ -1,6 +1,7 @@ //! Replacement starts use server defaults and preserve the current task on failure. use super::*; +use crate::app::backend_sessions::BackendSessions; use codex_app_server_protocol::SkillMetadata; use codex_app_server_protocol::SkillScope; use codex_app_server_protocol::SkillsListEntry; @@ -460,6 +461,117 @@ async fn replacement_failure_keeps_current_task_and_restores_input() -> Result<( Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn backend_handoff_errors_are_snapshotted() -> Result<()> { + let (mut app, mut events, _ops) = make_test_app_with_channels().await; + let source = crate::start_embedded_app_server_for_picker(&app.config).await?; + let (destination, _, destination_proxy) = start_recording_app_server_with_history( + &app.config, + HistoryCapabilities::ConfigReadFails, + /*blocked_thread_list*/ None, + /*failed_thread_name*/ None, + crate::app_server_session::ThreadParamsMode::Embedded, + LoaderOverrides::without_managed_config_for_tests(), + ) + .await?; + let destination_id = destination.backend_id(); + let target = crate::resume_picker::SessionTarget { + backend_id: destination_id, + path: None, + thread_id: ThreadId::new(), + cwd: None, + history_mode: None, + }; + let mut backends = BackendSessions::from_initial(source, vec![destination]); + let mut tui = crate::tui::test_support::make_test_tui()?; + + app.resume_backend_session(&mut tui, &mut backends, target.clone()) + .await?; + insta::assert_snapshot!(next_history_message(&mut events), @"■ This backend has no connection settings. Open the session picker again."); + + backends + .settings + .insert(destination_id, app.backend_settings()); + app.resume_backend_session(&mut tui, &mut backends, target) + .await?; + insta::assert_snapshot!(next_history_message(&mut events), @"■ Unable to check folder trust: config/read failed while checking remote project trust"); + + backends.shutdown().await?; + destination_proxy.await??; + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn resumed_backend_handoff_detaches_through_the_source_connection() -> Result<()> { + let mut app = make_test_app().await; + let (source, source_requests, source_proxy) = start_recording_app_server( + &app.config, + /*blocked_thread_list*/ None, + /*failed_thread_name*/ None, + ) + .await?; + let (mut destination, destination_requests, destination_proxy) = start_recording_app_server( + &app.config, + /*blocked_thread_list*/ None, + /*failed_thread_name*/ None, + ) + .await?; + let destination_id = destination.backend_id(); + let resumed = destination.start_thread(&app.config).await?; + let resumed_id = resumed.session.thread_id; + let saved_item = serde_json::from_str( + r#"{"type":"message","role":"assistant","content":[{"type":"output_text","text":"saved history"}]}"#, + )?; + destination + .thread_inject_items(resumed.session.thread_id, vec![saved_item]) + .await?; + let current = ThreadId::new(); + app.ensure_thread_channel(ThreadId::new()).backend_id = + Some(crate::backend_id::BackendId::new()); + app.enqueue_primary_thread_session( + test_thread_session(current, app.config.cwd.to_path_buf()), + Vec::new(), + ) + .await?; + let target = crate::resume_picker::SessionTarget { + backend_id: destination_id, + path: None, + thread_id: resumed_id, + cwd: Some(resumed.session.cwd.to_path_buf()), + history_mode: None, + }; + let mut backends = BackendSessions::from_initial(source, vec![destination]); + backends + .settings + .insert(destination_id, app.backend_settings()); + let mut tui = crate::tui::test_support::make_test_tui()?; + let mut stale_input = app.chat_widget.capture_thread_input_state().unwrap(); + stale_input.provider_error = Some( + codex_app_server_protocol::ModelProviderError::ModelProviderNotConfigured { + model_provider: "repaired-provider".into(), + }, + ); + app.agents_overview + .input_states + .insert(resumed_id, stale_input); + + app.resume_backend_session(&mut tui, &mut backends, target) + .await?; + + assert_eq!(backends.displayed_owner(), destination_id); + assert_eq!(app.chat_widget.thread_id(), Some(resumed_id)); + assert!(!app.chat_widget.has_provider_error()); + assert_eq!( + recorded_params(&source_requests, "thread/unsubscribe"), + vec![serde_json::json!({"threadId": current.to_string()})] + ); + assert!(recorded_params(&destination_requests, "thread/unsubscribe").is_empty()); + backends.shutdown().await?; + source_proxy.await??; + destination_proxy.await??; + Ok(()) +} + #[tokio::test] async fn replacement_preserves_remote_launch_paths_and_older_servers() -> Result<()> { for (remote_cwd, capabilities) in [ diff --git a/codex-rs/tui/src/app/tests/realtime_start.rs b/codex-rs/tui/src/app/tests/realtime_start.rs index 295067c3b7..512a43d93f 100644 --- a/codex-rs/tui/src/app/tests/realtime_start.rs +++ b/codex-rs/tui/src/app/tests/realtime_start.rs @@ -102,7 +102,7 @@ async fn rejected_voice_stop_clears_the_owned_session() -> Result<()> { #[tokio::test] async fn background_voice_closes_with_its_owner_not_the_visible_thread() -> Result<()> { - for owner_closed in [false, true] { + for (owner_closed, background_backend) in [(false, false), (true, false), (true, true)] { let (mut app, _events, mut ops) = make_test_app_with_channels().await; let owner = ThreadId::new(); let visible = ThreadId::new(); @@ -117,13 +117,26 @@ async fn background_voice_closes_with_its_owner_not_the_visible_thread() -> Resu )); app.active_thread_id = Some(visible); let closed = if owner_closed { owner } else { visible }; - app.enqueue_thread_notification( - closed, - ServerNotification::ThreadClosed(codex_app_server_protocol::ThreadClosedNotification { - thread_id: closed.to_string(), - }), - ) - .await?; + let notification = thread_closed_notification(closed); + if background_backend { + let source = crate::start_embedded_app_server_for_picker(&app.config).await?; + app.handle_polled_backend_event( + &source, + crate::backend_id::BackendId::new(), + crate::app_server_session::AppServerSessionEvent { + backend_id: source.backend_id(), + event: codex_app_server_client::AppServerEvent::ServerNotification(Box::new( + notification, + )), + }, + ) + .await; + source.shutdown().await?; + } else { + app.enqueue_thread_notification(closed, notification) + .await?; + } + assert_eq!(app.chat_widget.thread_id(), Some(visible)); assert_eq!( app.voice_owner_thread_id(), (!owner_closed).then_some(owner) diff --git a/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs b/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs index 0d6cd632c8..53563c33e9 100644 --- a/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs +++ b/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs @@ -952,18 +952,15 @@ async fn in_app_remote_resume_picker_lists_all_providers_after_provider_switch() let remote_cwd = remote_root.path().join("project"); std::fs::create_dir(&remote_cwd)?; let (app_server, requests, proxy) = start_recording_remote_app_server(&app.config).await?; - let backend_id = app_server.backend_id(); let mut app_server = app_server.with_remote_cwd_override(Some(remote_cwd.clone())); // Provider A's conversations must remain visible after switching to provider B. app_server.model_provider_override = Some("provider-b".to_string()); let mut backends = crate::app::backend_sessions::BackendSessions::new(app_server); - let displayed = backends.slot(backend_id).expect("registered backend"); let mut tui = crate::tui::test_support::make_test_tui()?; tui.set_alt_screen_enabled(/*enabled*/ false); tui.pause_events(); - let mut picker = - Box::pin(app.open_resume_picker_for_backends(&mut tui, &mut backends, displayed)); + let mut picker = Box::pin(app.open_resume_picker_for_backends(&mut tui, &mut backends)); let params = tokio::time::timeout(std::time::Duration::from_secs(/*secs*/ 5), async { loop { if let Some(params) = recorded_params(&requests, "thread/list").pop() { diff --git a/codex-rs/tui/src/app/tests/startup.rs b/codex-rs/tui/src/app/tests/startup.rs index 1b1e4ff4e9..316ae0e850 100644 --- a/codex-rs/tui/src/app/tests/startup.rs +++ b/codex-rs/tui/src/app/tests/startup.rs @@ -1126,11 +1126,13 @@ async fn fresh_startup_thread_drains_buffered_approval_before_draft_handoff() -> let mut tui = crate::tui::test_support::make_test_tui()?; let mut app_server = crate::start_embedded_app_server_for_picker(app.chat_widget.config_ref()).await?; + let backend_id = app_server.backend_id(); let control = Box::pin(app.handle_event( &mut tui, &mut app_server, AppEvent::StartupThreadStarted { + backend_id, result: Ok(AppServerStartedThread { session: test_thread_session(thread_id, test_path_buf("/tmp/project")), turns: Vec::new(), diff --git a/codex-rs/tui/src/app/thread_events.rs b/codex-rs/tui/src/app/thread_events.rs index f810d38745..bddbd7719e 100644 --- a/codex-rs/tui/src/app/thread_events.rs +++ b/codex-rs/tui/src/app/thread_events.rs @@ -597,6 +597,8 @@ fn file_change_item_changes( #[derive(Debug)] pub(super) struct ThreadEventChannel { + /// Set on backend handoff so a retained channel is detached through its owning session. + pub(super) backend_id: Option, pub(super) sender: mpsc::Sender, pub(super) receiver: Option>, pub(super) store: Arc>, @@ -610,6 +612,7 @@ impl ThreadEventChannel { sender, receiver: Some(receiver), store: Arc::new(Mutex::new(ThreadEventStore::new(capacity))), + backend_id: None, attachment: ThreadEventAttachment::Live, } } @@ -639,6 +642,7 @@ impl ThreadEventChannel { store: Arc::new(Mutex::new(ThreadEventStore::new_with_session( capacity, session, turns, ))), + backend_id: None, attachment: ThreadEventAttachment::Live, } } diff --git a/codex-rs/tui/src/app/voice_owner.rs b/codex-rs/tui/src/app/voice_owner.rs index 67ca6107ec..3b0af9f66f 100644 --- a/codex-rs/tui/src/app/voice_owner.rs +++ b/codex-rs/tui/src/app/voice_owner.rs @@ -125,7 +125,16 @@ impl App { app_server: &mut AppServerSession, destination: Option, ) { - let thread_ids: Vec<_> = self.thread_event_channels.keys().copied().collect(); + let thread_ids: Vec<_> = self + .thread_event_channels + .iter() + .filter(|(_, channel)| { + channel + .backend_id + .is_none_or(|owner| owner == app_server.backend_id()) + }) + .map(|(id, _)| *id) + .collect(); for thread_id in thread_ids { if self.voice_owner_thread_id() == Some(thread_id) || destination == Some(thread_id) diff --git a/codex-rs/tui/src/app_event.rs b/codex-rs/tui/src/app_event.rs index 883c2516e3..9c848858c6 100644 --- a/codex-rs/tui/src/app_event.rs +++ b/codex-rs/tui/src/app_event.rs @@ -601,6 +601,7 @@ pub(crate) enum AppEvent { /// Result of the fresh startup thread that is attached after the input UI is live. StartupThreadStarted { + backend_id: crate::backend_id::BackendId, result: color_eyre::Result, }, @@ -655,6 +656,9 @@ pub(crate) enum AppEvent { /// Resume a thread by UUID or thread name inside the running TUI session. ResumeSessionByIdOrName(String), + /// Attach a conversation through the registered backend that owns it. + ResumeBackendSession(crate::resume_picker::SessionTarget), + /// Archive the current active main thread and return to the command center. ArchiveCurrentThread,