openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-31 19:28:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

Enable cross-backend session switching in the TUI (#58457)

Browse Source 
## Why

The session picker can list conversations from multiple backends, but selecting a conversation owned by another backend was rejected.

## What changed

- Resume selected conversations through their owning backend, preparing the destination before replacing the current conversation. Preserve drafts and restore the destination's settings, account state, model catalog, and workspace services.
- Block backend switches during an active turn and retain the current conversation when destination preparation fails.
- Queue background backend requests until that backend is displayed, removing requests that are resolved or disconnected. Isolate background events from the displayed conversation while continuing to deliver notifications to a background voice session.
- Track backend ownership for thread detachment, startup results, and reconnect attempts.
- Remove the `yaml-rust` advisory exception from `codex-rs/deny.toml`.

## Testing

Add regression coverage for backend handoff errors, detaching through the source connection, clearing stale provider errors, deferred request delivery and cleanup, and voice-session closure on a background backend.

GitOrigin-RevId: 1bba84f16831af1d5295c1404f48803ce0608d3d
This commit is contained in:
Benjamin Carlsson
2026-10-30 23:56:12 +00:00
committed by copyberry
parent 7656a23b8d
commit 1da2aefe00
22 changed files with 642 additions and 132 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/deny.toml
+1 -2
View File
@@ -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" },
codex-rs/tui/src/app.rs
+1
View File
@@ -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;
codex-rs/tui/src/app/agents_overview.rs
+24 -17
View File
@@ -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<codex_app_server_protocol::ThreadSettingsUpdatedNotification>,
) {
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,
codex-rs/tui/src/app/backend_events.rs
+11 -14
View File
@@ -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;
codex-rs/tui/src/app/backend_selection.rs
+137
View File
@@ -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<AppRunControl>> {
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)
})
}
}
codex-rs/tui/src/app/backend_sessions.rs
+63 -11
View File
@@ -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<AppServerSessionEvent>,
}
/// 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<SessionEntry>,
pub(super) displayed: SessionSlot,
pub(super) settings: HashMap<BackendId, BackendSettings>,
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<AppServerSession> {
pub(super) fn displayed_owner(&self) -> BackendId {
self.session_at(self.displayed).backend_id()
}
pub(super) fn picker_views(&self) -> Vec<AppServerSession> {
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<SessionSlot> {
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>,
) -> (BackendId, Option<AppServerSessionEvent>) {
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
})
codex-rs/tui/src/app/event_dispatch.rs
+4 -1
View File
@@ -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;
}
codex-rs/tui/src/app/reconnect.rs
+29 -21
View File
@@ -24,7 +24,7 @@ pub(super) struct ReconnectState {
}
pub(super) struct Reconnected {
session: AppServerSession,
pub(super) session: AppServerSession,
bootstrap: AppServerBootstrap,
thread: Option<AppServerStartedThread>,
}
@@ -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 {
codex-rs/tui/src/app/resume_preparation.rs
-2
View File
@@ -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(),
codex-rs/tui/src/app/session_lifecycle.rs
+1 -5
View File
@@ -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)
codex-rs/tui/src/app/session_picker.rs
+22 -17
View File
@@ -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<AppRunControl> {
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<AppRunControl> {
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)
codex-rs/tui/src/app/startup.rs
+36 -27
View File
@@ -24,6 +24,7 @@ fn spawn_startup_thread_start(
app_event_tx: AppEventSender,
worktree: Option<crate::ManagedTuiWorktree>,
) {
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<ExitReason> = 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::<history_cell::SessionInfoCell>()
);
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
}
codex-rs/tui/src/app/tests/backend_ownership_tests.rs
+7 -1
View File
@@ -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(
codex-rs/tui/src/app/tests/backend_sessions_tests.rs
+150
View File
@@ -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;
codex-rs/tui/src/app/tests/disconnect_tests.rs
+2 -1
View File
@@ -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);
codex-rs/tui/src/app/tests/new_session_tests.rs
+112
View File
@@ -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 [
codex-rs/tui/src/app/tests/realtime_start.rs
+21 -8
View File
@@ -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)
codex-rs/tui/src/app/tests/session_lifecycle_requests.rs
+1 -4
View File
@@ -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() {
codex-rs/tui/src/app/tests/startup.rs
+2
View File
@@ -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(),
codex-rs/tui/src/app/thread_events.rs
+4
View File
@@ -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<crate::backend_id::BackendId>,
pub(super) sender: mpsc::Sender<ThreadBufferedEvent>,
pub(super) receiver: Option<mpsc::Receiver<ThreadBufferedEvent>>,
pub(super) store: Arc<Mutex<ThreadEventStore>>,
@@ -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,
}
}
codex-rs/tui/src/app/voice_owner.rs
+10 -1
View File
@@ -125,7 +125,16 @@ impl App {
app_server: &mut AppServerSession,
destination: Option<ThreadId>,
) {
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)
codex-rs/tui/src/app_event.rs
+4
View File
@@ -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<AppServerStartedThread>,
},
@@ -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,