diff --git a/codex-rs/tui/src/app.rs b/codex-rs/tui/src/app.rs index 53c82be9c7..5f7a28b60f 100644 --- a/codex-rs/tui/src/app.rs +++ b/codex-rs/tui/src/app.rs @@ -218,6 +218,7 @@ pub(crate) mod app_server_requests; mod app_server_thread_ownership; mod backend_banner_fallback; mod backend_events; +mod backend_registry; pub(crate) mod backend_selection; pub(crate) mod backend_sessions; mod background_requests; @@ -559,6 +560,7 @@ pub(crate) struct App { pub(crate) session_telemetry: SessionTelemetry, terminal_size_tags: crate::terminal_size_telemetry::LaunchTags, pub(crate) app_event_tx: AppEventSender, + backend_registry: backend_registry::BackendRegistry, pub(crate) chat_widget: ChatWidget, workspace_command_runner: Option, /// Legacy bootstrap and server-setting inputs; local preferences live in `local_settings`. @@ -628,6 +630,7 @@ pub(crate) struct App { feedback_audience: FeedbackAudience, environment_manager: Arc, app_server_target: AppServerTarget, + app_server_backend_id: Option, reconnect: reconnect::ReconnectState, pending_right_click_paste: Option, right_click_paste_environment: right_click_paste::PasteEnvironment, diff --git a/codex-rs/tui/src/app/agent_center_tests.rs b/codex-rs/tui/src/app/agent_center_tests.rs index c1ffef0757..1f72f2540d 100644 --- a/codex-rs/tui/src/app/agent_center_tests.rs +++ b/codex-rs/tui/src/app/agent_center_tests.rs @@ -91,7 +91,11 @@ async fn live_center_keeps_status_order_across_progress_refreshes() { .unwrap(); assert_eq!(app.agents_overview.visible_thread_ids[selected], ids[0]); } - app.agents_overview.pinned_thread_ids = Some(vec![ids[3], ids[2]]); + app.agents_overview.pins_by_backend = [( + crate::backend_id::BackendId::new(), + Some(vec![ids[3], ids[2]]), + )] + .into(); for parent in [ids[0], ids[3]] { threads.push(overview_thread( ThreadId::new(), @@ -273,10 +277,15 @@ async fn live_center_status_order_tracks_updates_while_filtered_and_pinned() { assert_eq!(rendered_titles(&view), vec!["Bravo", "Charlie", "Alpha"]); view.handle_key_event(KeyEvent::new(KeyCode::BackTab, KeyModifiers::SHIFT)); view.handle_key_event(KeyEvent::new(KeyCode::BackTab, KeyModifiers::SHIFT)); - app.agents_overview.pinned_thread_ids = Some(vec![ids[0], ids[2]]); + app.agents_overview.pins_by_backend = [( + crate::backend_id::BackendId::new(), + Some(vec![ids[0], ids[2]]), + )] + .into(); view = app.agents_overview_view(threads.clone(), Some(ids[1])); assert_eq!(rendered_titles(&view), vec!["Alpha", "Charlie", "Bravo"]); - app.agents_overview.pinned_thread_ids = Some(vec![]); + app.agents_overview.pins_by_backend = + [(crate::backend_id::BackendId::new(), Some(vec![]))].into(); view = app.agents_overview_view(threads, Some(ids[1])); assert_eq!(rendered_titles(&view), vec!["Bravo", "Charlie", "Alpha"]); } @@ -333,7 +342,11 @@ async fn live_center_columns() { ThreadStatus::Idle, )); crate::chatwidget::activate_voice_for_thread(&mut app.chat_widget, child); - app.agents_overview.pinned_thread_ids = Some(vec![current, ThreadId::from_u128(/*value*/ 44)]); + app.agents_overview.pins_by_backend = [( + crate::backend_id::BackendId::new(), + Some(vec![current, ThreadId::from_u128(/*value*/ 44)]), + )] + .into(); app.agents_overview .unread .insert(ThreadId::from_u128(/*value*/ 47)); @@ -738,7 +751,8 @@ toggle_pin = 'z p' ) .unwrap(); app.keymap = RuntimeKeymap::from_config(&config).unwrap(); - app.agents_overview.pinned_thread_ids = Some(Vec::new()); + app.agents_overview.pins_by_backend = + [(crate::backend_id::BackendId::new(), Some(Vec::new()))].into(); let view = app.agents_overview_view(Vec::new(), /*selected_thread_id*/ None); insta::assert_snapshot!( "live_center_custom_footer", diff --git a/codex-rs/tui/src/app/agents_overview.rs b/codex-rs/tui/src/app/agents_overview.rs index 6ef505771a..1081c053fe 100644 --- a/codex-rs/tui/src/app/agents_overview.rs +++ b/codex-rs/tui/src/app/agents_overview.rs @@ -22,6 +22,7 @@ use super::app_server_requests::BufferedServerRequest; use super::session_lifecycle::ThreadAttachPresentation; use super::*; use crate::app_event::AgentsOverviewThreadRefresh; +use crate::backend_id::BackendId; use crate::bottom_pane::BottomPaneView; use crate::bottom_pane::SelectionItem; use crate::bottom_pane::SelectionViewParams; @@ -51,19 +52,21 @@ pub(super) struct AgentsOverviewState { pub(super) pending_usage: Option<(ThreadId, Uuid)>, pub(super) usage_disabled: bool, pub(super) activity: HashMap, - pub(super) initialized: bool, - pub(super) discovery: super::agents_overview_discovery::AgentsOverviewDiscovery, + pub(super) initialized: HashSet, + pub(super) discovery: + HashMap, pub(super) show_more_requested: bool, pub(super) refresh_show_more: bool, /// Vacancies left by lifecycle removals, filled without expanding the visible window. pub(super) refill_count: usize, - pub(super) request_id: Option, + pub(super) active_refresh: Option<(Uuid, BackendId)>, pub(super) refresh_pending: bool, + pub(super) refresh_invalidated: bool, pub(super) pin_refresh_requested: bool, pub(super) refresh_thread_ids: HashSet, pub(super) active_refresh_thread_ids: HashSet, - pub(super) pinned_thread_ids: Option>, - pub(super) pending_pin_change: Option, + pub(super) pins_by_backend: std::collections::BTreeMap>>, + pub(super) pending_pin_change: Option<(Uuid, BackendId)>, pub(super) refresh_task: Option, pub(super) refresh_notifications: HashMap>, pub(super) rendered_full_screen: bool, @@ -83,6 +86,34 @@ pub(super) struct AgentsOverviewState { pub(super) dispatched_requests: HashMap>, } +impl AgentsOverviewState { + pub(super) fn invalidate_discovery(&mut self, backend: BackendId, displayed: bool) { + self.initialized.remove(&backend); + self.discovery.remove(&backend); + if displayed { + self.reset_visible_pagination(); + self.view_state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .has_more = false; + } + self.refresh_invalidated |= self + .active_refresh + .is_some_and(|(_, owner)| owner == backend); + } + + pub(super) fn reset_visible_pagination(&mut self) { + self.show_more_requested = false; + self.refill_count = 0; + } + + pub(super) fn has_more(&self, backend_id: BackendId) -> bool { + self.discovery + .get(&backend_id) + .is_none_or(super::agents_overview_discovery::AgentsOverviewDiscovery::has_more) + } +} + impl Drop for AgentsOverviewState { fn drop(&mut self) { if let Some(task) = self.refresh_task.take() { @@ -128,13 +159,17 @@ impl App { request_id: Uuid, result: Result, ) { - if self.agents_overview.request_id != Some(request_id) { + let Some((_, refresh_backend_id)) = self + .agents_overview + .active_refresh + .take_if(|(id, _)| *id == request_id) + else { return; - } - self.agents_overview.request_id = None; + }; self.agents_overview.refresh_show_more = false; self.agents_overview.active_refresh_thread_ids.clear(); self.agents_overview.refresh_task = None; + let invalidated = std::mem::take(&mut self.agents_overview.refresh_invalidated); let refill_succeeded = result .as_ref() .is_ok_and(|refresh| refresh.recent_seed_complete); @@ -145,37 +180,66 @@ impl App { .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); state.loading = false; - state.refresh_failed = !refill_succeeded; + state.refresh_failed = !invalidated && !refill_succeeded; } match result { - Ok(refresh) => { + _ if invalidated => self.agents_overview.refresh_notifications.clear(), + Ok(mut refresh) => { for id in refresh.cleared_unread { self.agents_overview.unread.remove(&id); } if let Some(pinned_thread_ids) = refresh.pinned_thread_ids { - self.agents_overview.pinned_thread_ids = pinned_thread_ids; - } - self.agents_overview.initialized = refresh.recent_seed_complete; - if let Some(discovery) = refresh.discovery { - if !discovery.has_more() { - self.agents_overview.refill_count = 0; - self.agents_overview.show_more_requested = false; - } - self.agents_overview.initialized = true; + let pins = pinned_thread_ids.map(|ids| { + let mut seen = HashSet::new(); + ids.into_iter() + .filter(|id| { + self.backend_registry.observe(refresh_backend_id, *id) + && seen.insert(*id) + }) + .collect() + }); self.agents_overview - .view_state - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .has_more = discovery.has_more(); - self.agents_overview.discovery = discovery; + .pins_by_backend + .insert(refresh_backend_id, pins); + } + let displayed_refresh = refresh_backend_id == app_server.backend_id(); + if refresh.discovery.is_some() + || (displayed_refresh && refresh.recent_seed_complete) + { + self.agents_overview.initialized.insert(refresh_backend_id); + } else if displayed_refresh { + self.agents_overview.initialized.remove(&refresh_backend_id); + } + if let Some(discovery) = refresh.discovery { + if displayed_refresh { + let has_more = discovery.has_more(); + if !has_more { + self.agents_overview.reset_visible_pagination(); + } + self.agents_overview + .view_state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .has_more = has_more; + } + self.agents_overview + .discovery + .insert(refresh_backend_id, discovery); } - self.agents_overview - .last_messages - .extend(refresh.last_messages); for (thread_id, thread) in refresh.threads { if self.agents_overview.removed_threads.contains(&thread_id) { continue; } + if thread.is_some() + && !self.backend_registry.observe(refresh_backend_id, thread_id) + { + continue; + } + if let Some(message) = refresh.last_messages.remove(&thread_id) { + self.agents_overview + .last_messages + .insert(thread_id, message); + } if let Some((mut thread, unread, latest_prompt)) = thread { self.agents_overview.search_matches.remove(&thread_id); if unread { @@ -190,7 +254,8 @@ impl App { self.agents_overview.usage.remove(&thread_id); continue; } - if !self.agents_overview.threads.contains_key(&thread_id) + if displayed_refresh + && !self.agents_overview.threads.contains_key(&thread_id) && !self.agents_overview.hidden_threads.contains(&thread_id) && thread.parent_thread_id.is_none() && !matches!( @@ -238,10 +303,11 @@ impl App { } } if std::mem::take(&mut self.agents_overview.refresh_pending) + || invalidated || (refill_succeeded && (self.agents_overview.refill_count > 0 || self.agents_overview.show_more_requested) - && self.agents_overview.discovery.has_more()) + && self.agents_overview.has_more(app_server.backend_id())) { self.refresh_changed_agents_overview_threads(app_server); } @@ -402,6 +468,10 @@ impl App { continue; } rows.push(AgentsOverviewRow { + backend_id: self + .backend_registry + .owner(thread_id) + .or(self.app_server_backend_id), search_id: self .agents_overview .search_matches @@ -429,8 +499,11 @@ impl App { self.keymap.clone(), Arc::clone(&self.agents_overview.view_state), ); - view.pinned_thread_ranks = self.agents_overview.pinned_thread_ids.as_ref().map(|ids| { - ids.iter() + let pins = &self.agents_overview.pins_by_backend; + view.pinned_thread_ranks = pins.values().any(Option::is_some).then(|| { + pins.values() + .flatten() + .flatten() .copied() .enumerate() .map(|(rank, id)| (id, rank)) @@ -438,6 +511,11 @@ impl App { }); view.can_return_to_chat = self.pending_startup_thread_start || self.current_displayed_thread_id().is_some(); + view.pin_action_supported = self + .displayed_conversation_backend_id() + .and_then(|owner| pins.get(&owner)) + .map(Option::is_some) + .unwrap_or(true); view.pin_action_pending = self.agents_overview.pending_pin_change.is_some(); view } @@ -448,6 +526,25 @@ impl App { app_server: &mut AppServerSession, thread_id: ThreadId, ) -> Result { + if let Some(owner) = self.backend_registry.owner(thread_id) + && owner != app_server.backend_id() + { + let Some(Some(thread)) = self.agents_overview.threads.get(&thread_id) else { + self.add_agents_overview_error(format!( + "Agent session {thread_id} is unavailable. Refresh Command Center and try again." + )); + return Ok(AppRunControl::Continue); + }; + self.app_event_tx + .send(AppEvent::ResumeBackendSession(SessionTarget { + backend_id: owner, + path: thread.path.clone(), + thread_id, + cwd: Some(thread.cwd.to_path_buf()), + history_mode: Some(thread.history_mode), + })); + return Ok(AppRunControl::Continue); + } Box::pin(self.attach_agents_overview_thread( tui, app_server, thread_id, /*started*/ None, /*startup_draft*/ None, )) @@ -484,6 +581,7 @@ impl App { loading::draw(tui)?; } let mut restored_blank_session = false; + let mut resumed_mode = None; let mut provider_resume_checked = false; if self.primary_thread_id != Some(root_thread_id) { let previous_displayed_thread_id = self.current_displayed_thread_id(); @@ -684,14 +782,11 @@ impl App { }; let (resumed, read_only) = if let Some((_, started)) = started { (started, false) - } else if !unloaded - && let Some(blank) = self.agents_overview.blank_sessions.get(&root_thread_id) + } else if let Some(blank) = self + .live_blank_session(app_server, root_thread_id, &target_thread) + .await { - // An untouched task has no rollout for thread/resume yet. Its live - // subscription and saved settings are sufficient to restore the editor. restored_blank_session = true; - let mut blank = blank.clone(); - blank.session.thread_name.clone_from(&target_thread.name); (blank, false) } else { match app_server @@ -706,6 +801,7 @@ impl App { { Ok(resumed) => { provider_resume_checked = true; + resumed_mode = Some(resumed.session.collaboration_mode.is_some()); (resumed, false) } Err(error) @@ -823,6 +919,9 @@ impl App { self.add_agents_overview_error(format!("Failed to attach to task: {error}")); return Ok(AppRunControl::Continue); } + if !is_new_session && !restored_blank_session { + self.agents_overview.blank_sessions.remove(&root_thread_id); + } // Replacing the widget clears the terminal before the remaining server requests. if startup_draft.is_none() { loading::draw(tui)?; @@ -926,6 +1025,10 @@ impl App { if provider_resume_checked { input_state.provider_error = None; } + if let Some(has_resumed_mode) = resumed_mode { + self.chat_widget + .update_input_collaboration_mode(&mut input_state, has_resumed_mode); + } // A saved draft includes model settings, so apply newer server settings after it. let pending_settings = restored_blank_session .then(|| input_state.pending_thread_settings.take()) diff --git a/codex-rs/tui/src/app/agents_overview_actions.rs b/codex-rs/tui/src/app/agents_overview_actions.rs index 588b5927c6..a15cf413f5 100644 --- a/codex-rs/tui/src/app/agents_overview_actions.rs +++ b/codex-rs/tui/src/app/agents_overview_actions.rs @@ -106,7 +106,7 @@ impl App { return; } let request_id = Uuid::new_v4(); - self.agents_overview.pending_pin_change = Some(request_id); + self.agents_overview.pending_pin_change = Some((request_id, app_server.backend_id())); let request_handle = app_server.request_handle(); let app_event_tx = self.app_event_tx.clone(); tokio::spawn(async move { @@ -139,13 +139,19 @@ impl App { pinned: bool, result: Result<(), String>, ) { - if self.agents_overview.pending_pin_change != Some(request_id) { + let Some((_, owner)) = self + .agents_overview + .pending_pin_change + .filter(|(id, _)| *id == request_id) + else { return; - } + }; self.agents_overview.pending_pin_change = None; match result { Ok(()) => { - if let Some(pinned_thread_ids) = self.agents_overview.pinned_thread_ids.as_mut() { + if let Some(Some(pinned_thread_ids)) = + self.agents_overview.pins_by_backend.get_mut(&owner) + { if pinned { if !pinned_thread_ids.contains(&thread_id) { pinned_thread_ids.push(thread_id); @@ -166,7 +172,7 @@ impl App { if let Some(refresh) = self.agents_overview.refresh_task.take() { refresh.abort(); } - self.agents_overview.request_id = None; + self.agents_overview.active_refresh = None; self.agents_overview.refresh_show_more = false; self.agents_overview.refresh_notifications.clear(); let refresh_thread_ids = @@ -374,7 +380,7 @@ impl App { if let Some(refresh) = self.agents_overview.refresh_task.take() { refresh.abort(); } - self.agents_overview.request_id = None; + self.agents_overview.active_refresh = None; self.agents_overview.refresh_show_more = false; self.agents_overview.active_refresh_thread_ids.clear(); self.agents_overview diff --git a/codex-rs/tui/src/app/agents_overview_actions_tests.rs b/codex-rs/tui/src/app/agents_overview_actions_tests.rs index 8b070605ed..ee0b1cc0a9 100644 --- a/codex-rs/tui/src/app/agents_overview_actions_tests.rs +++ b/codex-rs/tui/src/app/agents_overview_actions_tests.rs @@ -30,7 +30,7 @@ async fn pin_action_updates_shared_section_state() -> Result<()> { AppEvent::ToggleAgentsOverviewPin { thread_id, pinned }, )) .await?; - let pending = app + let (pending, _) = app .agents_overview .pending_pin_change .expect("pin request started"); @@ -774,8 +774,8 @@ async fn hidden_task_stays_hidden_through_activity_and_seed_until_explicit_resum }, )); let request_id = Uuid::new_v4(); - app.agents_overview.initialized = false; - app.agents_overview.request_id = Some(request_id); + app.agents_overview.initialized.clear(); + app.agents_overview.active_refresh = Some((request_id, app_server.backend_id())); app.apply_agents_overview_thread_refresh( &app_server, request_id, @@ -1054,7 +1054,7 @@ async fn lifecycle_removes_background_and_current_tasks_without_losing_the_dashb } // Cached details and approval state are invalidated as a group after success. let request_id = Uuid::new_v4(); - app.agents_overview.request_id = Some(request_id); + app.agents_overview.active_refresh = Some((request_id, app_server.backend_id())); let stale_threads = app.agents_overview.threads.clone(); Box::pin(app.run_agents_overview_action(&mut tui, &mut app_server, background, action)) .await?; @@ -1161,7 +1161,9 @@ async fn lifecycle_removes_background_and_current_tasks_without_losing_the_dashb } crate::chatwidget::activate_voice_for_thread(&mut app.chat_widget, primary); // Canceling pagination must allow automatic refill to finish after removing the last task. - app.agents_overview.initialized = true; + app.agents_overview + .initialized + .insert(app_server.backend_id()); app.agents_overview.view_state.lock().unwrap().loading = true; // An invalidated refresh can already be queued when the lifecycle action // aborts its task. It must not satisfy the wait for the replacement refresh. @@ -1178,7 +1180,7 @@ async fn lifecycle_removes_background_and_current_tasks_without_losing_the_dashb }), }); Box::pin(app.handle_event(&mut tui, &mut app_server, confirmed)).await?; - if app.agents_overview.request_id.is_some() { + if app.agents_overview.active_refresh.is_some() { finish_overview_refresh(&mut app, &app_server, &mut rx).await; } assert!(!app.agents_overview.view_state.lock().unwrap().loading); diff --git a/codex-rs/tui/src/app/agents_overview_details.rs b/codex-rs/tui/src/app/agents_overview_details.rs index 04aa6824b9..fe77300f33 100644 --- a/codex-rs/tui/src/app/agents_overview_details.rs +++ b/codex-rs/tui/src/app/agents_overview_details.rs @@ -232,6 +232,11 @@ impl App { }), lines, usage_lines: root_id + .filter(|id| { + self.backend_registry + .owner(*id) + .is_none_or(|owner| self.app_server_backend_id == Some(owner)) + }) .and_then(|id| self.agents_overview.usage.get(&id)) .map(|usage| { super::agents_overview_usage::usage_lines( diff --git a/codex-rs/tui/src/app/agents_overview_discovery.rs b/codex-rs/tui/src/app/agents_overview_discovery.rs index 41464b3ebb..63e53ec2db 100644 --- a/codex-rs/tui/src/app/agents_overview_discovery.rs +++ b/codex-rs/tui/src/app/agents_overview_discovery.rs @@ -148,6 +148,10 @@ impl AgentsOverviewDiscovery { limit: usize, ) -> (Vec<(Thread, bool)>, bool) { let [interactive, other] = &mut self.sources; + // Cloud does not support filtering its task list to Exec and AppServer threads. + if matches!(handle, AppServerRequestHandle::Cloud(_)) { + other.exhausted = true; + } let (left, right) = tokio::join!( interactive.fill(handle, Vec::new()), other.fill( @@ -326,8 +330,10 @@ impl App { let Some(search) = state.history.as_mut().filter(|s| s.id == id) else { return; }; + let backend_id = self.app_server_backend_id; for (thread, unread) in rows { if let Ok(id) = codex_protocol::ThreadId::from_string(&thread.id) + && self.backend_registry.owner(id).or(backend_id) == backend_id && !overview.threads.get(&id).is_some_and(Option::is_some) && !overview.removed_threads.contains(&id) && !overview.hidden_threads.contains(&id) diff --git a/codex-rs/tui/src/app/agents_overview_discovery_tests.rs b/codex-rs/tui/src/app/agents_overview_discovery_tests.rs index 826c08de0b..cd5ee4f7b2 100644 --- a/codex-rs/tui/src/app/agents_overview_discovery_tests.rs +++ b/codex-rs/tui/src/app/agents_overview_discovery_tests.rs @@ -1,8 +1,11 @@ //! Demand-driven discovery, cursor retry, and visible Show more behavior. use super::*; +use crate::app::agents_overview_discovery::AgentsOverviewDiscovery; use crate::app_event::AgentsOverviewAction; use crate::app_server_session::ThreadParamsMode; +use codex_app_server_client::CloudOperation; +use codex_app_server_protocol::ClientRequest; use codex_app_server_protocol::JSONRPCMessage; use futures::SinkExt; use futures::StreamExt; @@ -14,6 +17,50 @@ use std::sync::atomic::Ordering; use tokio::net::TcpListener; use tokio_tungstenite::tungstenite::Message; +#[tokio::test] +async fn overview_cloud_discovery_uses_only_interactive_pages() { + let (client, mut commands, _events) = CloudClient::channel(/*capacity*/ 1); + let handle = AppServerClient::Cloud(client).request_handle(); + let threads = ["First Cloud task", "Second Cloud task"].map(|name| { + overview_thread( + ThreadId::new(), + /*parent_thread_id*/ None, + name, + ThreadStatus::NotLoaded, + ) + }); + + let serve_pages = async { + for (thread, (cursor, next_cursor)) in threads + .iter() + .zip([(None, Some("next")), (Some("next"), None)]) + { + let command = commands.recv().await.unwrap(); + let CloudOperation::Request(request) = command.operation else { + panic!("expected a Cloud request"); + }; + let ClientRequest::ThreadList { params, .. } = *request else { + panic!("expected thread/list"); + }; + assert_eq!(params.source_kinds, Some(Vec::new())); + assert_eq!(params.cursor.as_deref(), cursor); + command + .response + .send(Ok(Ok(json!({"data": [thread], "nextCursor": next_cursor})))) + .unwrap(); + } + }; + let discover = async { + let mut discovery = AgentsOverviewDiscovery::default(); + for (thread, has_more) in threads.iter().zip([true, false]) { + let page = discovery.next_batch(&handle, /*limit*/ 10).await; + assert_eq!(page, (vec![(thread.clone(), false)], true)); + assert_eq!(discovery.has_more(), has_more); + } + }; + tokio::join!(serve_pages, discover); +} + #[tokio::test] async fn overview_show_more_refills_archived_rows_and_retries_without_losing_rows() -> Result<()> { check_discovery(/*mixed_sources*/ false).await?; @@ -174,8 +221,11 @@ async fn check_discovery(mixed_sources: bool) -> Result<()> { app.refresh_agents_overview_threads(&session); finish_overview_refresh(&mut app, &session, &mut rx).await; assert_eq!(app.agents_overview.threads.len(), 10); - assert_eq!(app.agents_overview.pinned_thread_ids, Some(vec![ids[1]])); - assert!(app.agents_overview.discovery.has_more()); + assert_eq!( + app.agents_overview.pins_by_backend[&session.backend_id()], + Some(vec![ids[1]]) + ); + assert!(app.agents_overview.has_more(session.backend_id())); let browse = app.agents_overview.threads.clone(); let browse_rows = || browse.values().flatten().cloned().collect(); let mut view = app.agents_overview_view(browse_rows(), Some(ids[4])); @@ -232,8 +282,33 @@ async fn check_discovery(mixed_sources: bool) -> Result<()> { } assert_eq!(app.agents_overview.threads.len(), 10); assert!(app.agents_overview.threads.contains_key(&ids[10])); + let other = Box::pin(crate::start_embedded_app_server_for_picker(&app.config)).await?; + if mixed_sources { + let archived = ids[22]; + app.backend_registry.observe(session.backend_id(), archived); + app.track_agents_overview_notification(&ServerNotification::ThreadArchived( + ThreadArchivedNotification { + thread_id: archived.to_string(), + }, + )); + expected.remove(&archived); + } + app.agents_overview.reset_visible_pagination(); + if mixed_sources { + app.refresh_agents_overview_threads(&other); + finish_overview_refresh(&mut app, &other, &mut rx).await; + } + app.agents_overview.reset_visible_pagination(); app.show_more_agents_overview(&session); - finish_overview_refresh(&mut app, &session, &mut rx).await; + // The page may finish while another backend is displayed, or after returning. + app.agents_overview.reset_visible_pagination(); + if mixed_sources { + finish_overview_refresh(&mut app, &session, &mut rx).await; + } else { + finish_overview_refresh(&mut app, &other, &mut rx).await; + app.agents_overview.reset_visible_pagination(); + } + other.shutdown().await?; assert_eq!( app.agents_overview .threads @@ -259,7 +334,7 @@ async fn check_discovery(mixed_sources: bool) -> Result<()> { app.refresh_agents_overview_threads(&session); finish_overview_refresh(&mut app, &session, &mut rx).await; assert_eq!(app.agents_overview.threads, expected); - assert!(!app.agents_overview.discovery.has_more()); + assert!(!app.agents_overview.has_more(session.backend_id())); let old = app.agents_overview.threads[&ids[23]].as_ref().unwrap(); assert_eq!(old.preview, "Task 23"); assert_eq!( @@ -280,12 +355,12 @@ async fn check_discovery(mixed_sources: bool) -> Result<()> { ); view.handle_key_event(KeyCode::Esc.into()); app.agents_overview.threads.clear(); - app.agents_overview.pinned_thread_ids = Some(Vec::new()); + app.agents_overview.pins_by_backend = [(session.backend_id(), Some(Vec::new()))].into(); app.agents_overview.refill_count = 0; app.refresh_agents_overview_threads(&session); finish_overview_refresh(&mut app, &session, &mut rx).await; assert_eq!( - app.agents_overview.pinned_thread_ids, + app.agents_overview.pins_by_backend[&session.backend_id()], Some(vec![reopened_pin_id]) ); assert!(app.agents_overview.threads.contains_key(&reopened_pin_id)); diff --git a/codex-rs/tui/src/app/agents_overview_new.rs b/codex-rs/tui/src/app/agents_overview_new.rs index aa926e2dd8..1e64db5296 100644 --- a/codex-rs/tui/src/app/agents_overview_new.rs +++ b/codex-rs/tui/src/app/agents_overview_new.rs @@ -221,6 +221,8 @@ impl App { return Ok(AppRunControl::Continue); } } + self.backend_registry + .register_started(app_server.backend_id(), &started)?; self.agents_overview .blank_sessions .insert(thread_id, started.clone()); diff --git a/codex-rs/tui/src/app/agents_overview_tests.rs b/codex-rs/tui/src/app/agents_overview_tests.rs index 287430e457..9fc0d68905 100644 --- a/codex-rs/tui/src/app/agents_overview_tests.rs +++ b/codex-rs/tui/src/app/agents_overview_tests.rs @@ -10,6 +10,8 @@ use super::super::agents_overview_view::AgentsOverviewGrouping; use super::*; use crate::app::owned_transcript::tests::attach_thread; use crate::app::tests::make_test_app_with_channels; +use crate::app_server_session::ThreadParamsMode; +use crate::backend_id::BackendId; use crate::chatwidget::tests::helpers::normalize_agent_center_snapshot; use codex_config::types::AgentsOverviewGrouping as PersistedAgentsOverviewGrouping; @@ -141,7 +143,9 @@ use crate::render::renderable::Renderable; use crate::test_support::PathBufExt; use crate::test_support::test_path_buf; use crate::test_support::test_path_display; +use codex_app_server_client::AppServerClient; use codex_app_server_client::AppServerEvent; +use codex_app_server_client::CloudClient; use codex_app_server_protocol::CurrentTimeReadParams; use codex_app_server_protocol::ReasoningSummaryTextDeltaNotification; use codex_app_server_protocol::RequestId; @@ -173,6 +177,219 @@ use ratatui::layout::Rect; static OVERVIEW_TIMESTAMP: std::sync::LazyLock = std::sync::LazyLock::new(|| chrono::Utc::now().timestamp() - 120); +#[tokio::test] +async fn refreshing_one_backends_pins_preserves_cross_backend_rank() { + let mut app = make_test_app().await; + let local = BackendId::new(); + let cloud = BackendId::new(); + let (client, _commands, _events) = CloudClient::channel(/*capacity*/ 1); + let session = AppServerSession::new( + local, + AppServerClient::Cloud(client), + ThreadParamsMode::Cloud, + ); + let local_pin = ThreadId::new(); + let cloud_pin = ThreadId::new(); + let mut previous = None; + for (owner, pin) in [(local, local_pin), (cloud, cloud_pin), (local, local_pin)] { + let request = Uuid::new_v4(); + app.agents_overview.active_refresh = Some((request, owner)); + app.apply_agents_overview_thread_refresh( + &session, + request, + Ok(AgentsOverviewThreadRefresh { + cleared_unread: Vec::new(), + last_messages: HashMap::new(), + threads: HashMap::new(), + recent_seed_complete: true, + discovery: None, + pinned_thread_ids: Some(Some(vec![pin])), + }), + ); + let ranks = app + .agents_overview_view(Vec::new(), /*selected_thread_id*/ None) + .pinned_thread_ranks + .unwrap(); + if owner == cloud { + assert_eq!(ranks.len(), 2); + previous = Some(ranks); + } else if let Some(expected) = previous.as_ref() { + assert_eq!(&ranks, expected); + } + } +} + +#[tokio::test] +async fn overview_keeps_rows_and_routes_actions_to_the_owner() -> Result<()> { + let mut app = make_test_app().await; + let stub = |owner, mode| { + let (client, commands, events) = CloudClient::channel(/*capacity*/ 1); + let session = AppServerSession::new(owner, AppServerClient::Cloud(client), mode); + (session, (commands, events)) + }; + let (mut displayed, _displayed_transport) = stub(BackendId::new(), ThreadParamsMode::Cloud); + let remote_owner = BackendId::new(); + let [local_id, remote_id, colliding_id] = std::array::from_fn(|_| ThreadId::new()); + let [local, remote, colliding] = [ + (local_id, "Local task"), + (remote_id, "Remote task"), + (colliding_id, "Colliding task"), + ] + .map(|(id, name)| { + overview_thread(id, /*parent_thread_id*/ None, name, ThreadStatus::Idle) + }); + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + app.app_event_tx = AppEventSender::new(tx); + app.app_server_backend_id = Some(displayed.backend_id()); + let local_reads = HashMap::from([(local_id, Some((local, false, None))), (remote_id, None)]); + let remote_reads = HashMap::from([ + (remote_id, Some((remote.clone(), false, None))), + (colliding_id, Some((colliding, false, None))), + ]); + let name_updated = |id: ThreadId, thread_name| { + AppServerEvent::ServerNotification(Box::new(ServerNotification::ThreadNameUpdated( + ThreadNameUpdatedNotification { + thread_id: id.to_string(), + thread_name, + }, + ))) + }; + for (owner, reads, pins) in [ + (displayed.backend_id(), local_reads, None), + (remote_owner, remote_reads, Some(Vec::new())), + ] { + let request_id = Uuid::new_v4(); + app.agents_overview.active_refresh = Some((request_id, owner)); + if owner == remote_owner { + let event = name_updated(colliding_id, Some("Name from the displayed backend".into())); + app.handle_app_server_event(&displayed, event).await; + app.open_agents_overview(&displayed); + } + app.apply_agents_overview_thread_refresh( + &displayed, + request_id, + Ok(AgentsOverviewThreadRefresh { + cleared_unread: Vec::new(), + threads: reads, + last_messages: HashMap::new(), + recent_seed_complete: owner == remote_owner, + discovery: (owner != displayed.backend_id()).then(Default::default), + pinned_thread_ids: Some(pins), + }), + ); + if owner == displayed.backend_id() { + assert_eq!(app.backend_registry.owner(remote_id), None); + assert_eq!( + app.agents_overview + .pins_by_backend + .get(&displayed.backend_id()), + Some(&None) + ); + } + } + assert_eq!( + app.agents_overview.active_refresh.map(|(_, owner)| owner), + Some(displayed.backend_id()) + ); + app.agents_overview.refresh_task.take().unwrap().abort(); + app.agents_overview.active_refresh = None; + assert_eq!( + app.agents_overview.pins_by_backend.get(&remote_owner), + Some(&Some(Vec::new())) + ); + assert!(!app.agents_overview.view_state.lock().unwrap().has_more); + assert_eq!(app.agents_overview.threads.len(), 2); + assert_eq!(app.backend_registry.owner(remote_id), Some(remote_owner)); + assert_eq!( + app.backend_registry.owner(colliding_id), + Some(displayed.backend_id()) + ); + assert!(!app.agents_overview.threads.contains_key(&colliding_id)); + + app.handle_app_server_event(&displayed, name_updated(remote_id, /*thread_name*/ None)) + .await; + assert_eq!( + app.agents_overview.threads[&remote_id].as_ref(), + Some(&remote) + ); + + let mut tui = crate::tui::test_support::make_test_tui()?; + assert!(matches!( + app.select_agents_overview_thread(&mut tui, &mut displayed, remote_id) + .await?, + AppRunControl::Continue + )); + assert!(matches!( + rx.try_recv()?, + AppEvent::ResumeBackendSession(target) + if target.backend_id == remote_owner && target.thread_id == remote_id + )); + assert!(app.agents_overview.threads.contains_key(&local_id)); + let mut view = app.agents_overview_view(vec![remote], Some(remote_id)); + assert!(!view.pin_action_supported); + view.handle_key_event(KeyCode::Char('n').into()); + let new_session = std::iter::from_fn(|| rx.try_recv().ok()) + .find(|event| matches!(event, AppEvent::NewAgentsOverviewSession { .. })) + .expect("selected foreign row emitted a new-session action"); + // A refresh replaces the visible row before its action reaches dispatch. + app.agents_overview.visible_thread_ids = vec![local_id]; + app.chat_widget.show_bottom_pane_view(Box::new(view)); + app.handle_event(&mut tui, &mut displayed, new_session) + .await?; + let rendered = render_bottom_popup(&app.chat_widget, /*width*/ 72); + let feedback = [ + "Unable to complete action", + "Open this task before using that action.", + "Return to command center", + ] + .into_iter() + .filter(|line| rendered.contains(line)) + .collect::>() + .join("\n"); + insta::assert_snapshot!(feedback, @" + Unable to complete action + Open this task before using that action. + Return to command center + "); + + // The stale-dispatch check replaced the map, not the one-row Remote view. + app.agents_overview.visible_thread_ids = vec![remote_id]; + + // Reverse the foreground so the backend that supports pinning is displayed. + app.chat_widget.handle_key_event(KeyCode::Enter.into()); + let (mut foreground, _foreground_transport) = stub(remote_owner, ThreadParamsMode::Embedded); + app.app_server_backend_id = Some(remote_owner); + app.repaint_agents_overview(); + let selected = app + .chat_widget + .selected_index_for_present_view(AGENTS_OVERVIEW_VIEW_ID) + .and_then(|index| app.agents_overview.visible_thread_ids.get(index)) + .copied(); + assert_eq!(selected, Some(remote_id)); + app.chat_widget.handle_key_event(KeyCode::Down.into()); + let pin = KeyCode::Char('p'); + app.chat_widget.handle_key_event(pin.into()); + let event = rx.try_recv()?; + assert!( + matches!( + &event, + AppEvent::ToggleAgentsOverviewPin { thread_id, pinned: true } if *thread_id == local_id + ), + "unexpected pin event: {event:?}" + ); + app.handle_event(&mut tui, &mut foreground, event).await?; + assert_eq!(app.agents_overview.pending_pin_change, None); + + app.chat_widget.handle_key_event(KeyCode::Enter.into()); + app.chat_widget.handle_key_event(KeyCode::Down.into()); + app.chat_widget.handle_key_event(pin.into()); + assert!(matches!( + rx.try_recv()?, + AppEvent::ToggleAgentsOverviewPin { thread_id, pinned: true } if thread_id == remote_id + )); + Ok(()) +} + #[tokio::test] async fn overview_right_opens_current_or_highlighted_task() { let mut app = make_test_app().await; @@ -326,7 +543,7 @@ async fn shared_overview_keeps_rows_and_replays_changes_over_stale_reads() -> Re overview_thread(id, /*parent_thread_id*/ None, name, ThreadStatus::Idle) }); let request_id = Uuid::new_v4(); - app.agents_overview.request_id = Some(request_id); + app.agents_overview.active_refresh = Some((request_id, app_server.backend_id())); // The seed is still in flight and the command center has never been opened. for notification in [ ServerNotification::ThreadNameUpdated(ThreadNameUpdatedNotification { @@ -403,7 +620,7 @@ async fn shared_overview_keeps_rows_and_replays_changes_over_stale_reads() -> Re assert_eq!(app.agents_overview.threads, expected); let request_id = Uuid::new_v4(); - app.agents_overview.request_id = Some(request_id); + app.agents_overview.active_refresh = Some((request_id, app_server.backend_id())); app.apply_agents_overview_thread_refresh( &app_server, request_id, @@ -418,7 +635,7 @@ async fn shared_overview_keeps_rows_and_replays_changes_over_stale_reads() -> Re ); assert_eq!(app.agents_overview.threads, expected); let request_id = Uuid::new_v4(); - app.agents_overview.request_id = Some(request_id); + app.agents_overview.active_refresh = Some((request_id, app_server.backend_id())); app.apply_agents_overview_thread_refresh( &app_server, request_id, @@ -438,7 +655,7 @@ async fn shared_overview_keeps_rows_and_replays_changes_over_stale_reads() -> Re AppServerEvent::Lagged { skipped: 1 }, ] { let old_request = Uuid::new_v4(); - app.agents_overview.request_id = Some(old_request); + app.agents_overview.active_refresh = Some((old_request, app_server.backend_id())); app.agents_overview.activity.entry(retained).or_default(); app.agents_overview.last_messages = stale_messages.clone(); app.handle_app_server_event(&app_server, event).await; @@ -450,7 +667,7 @@ async fn shared_overview_keeps_rows_and_replays_changes_over_stale_reads() -> Re "new-reasoning", "**Checking the revised task**", )); - let current_request = app.agents_overview.request_id.unwrap(); + let current_request = app.agents_overview.active_refresh.unwrap().0; for (request_id, last_messages) in [ (old_request, stale_messages.clone()), (current_request, HashMap::new()), @@ -470,7 +687,7 @@ async fn shared_overview_keeps_rows_and_replays_changes_over_stale_reads() -> Re assert!(app.agents_overview.last_messages.is_empty()); } assert!(app.agents_overview.activity.contains_key(&retained)); - assert!(app.agents_overview.request_id.is_none()); + assert!(app.agents_overview.active_refresh.is_none()); } app_server.shutdown().await?; Ok(()) @@ -655,7 +872,7 @@ async fn shared_overview_seeds_once_and_retains_locally_resumed_history() -> Res "agents_overview_recent_sessions", age.replace_all(&rendered, " [age]") ); - assert!(restarted.agents_overview.discovery.has_more()); + assert!(restarted.agents_overview.has_more(app_server.backend_id())); // Opening starts a metadata refresh; Show more queues behind it. restarted.show_more_agents_overview(&app_server); finish_overview_refresh(&mut restarted, &app_server, &mut event_rx).await; @@ -669,7 +886,7 @@ async fn shared_overview_seeds_once_and_retains_locally_resumed_history() -> Res .collect::>(), ids.into_iter().collect() ); - assert!(!restarted.agents_overview.discovery.has_more()); + assert!(!restarted.agents_overview.has_more(app_server.backend_id())); app_server.shutdown().await?; Ok(()) } @@ -685,7 +902,7 @@ async fn finish_overview_refresh( events.recv().await.expect("overview event") { // Aborting a refresh task cannot retract an already queued response. - if app.agents_overview.request_id != Some(request_id) { + if app.agents_overview.active_refresh.map(|(id, _)| id) != Some(request_id) { continue; } assert!(result.is_ok(), "{result:?}"); @@ -717,7 +934,7 @@ async fn check_hidden_thread_does_not_refresh_shared_overview(source: &str) { app.chat_widget.show_bottom_pane_view(Box::new(view)); let request_id = uuid::Uuid::new_v4(); - app.agents_overview.request_id = Some(request_id); + app.agents_overview.active_refresh = Some((request_id, app_server.backend_id())); let parent_thread_id = ThreadId::new(); app.primary_thread_id = Some(parent_thread_id); @@ -747,7 +964,7 @@ async fn check_hidden_thread_does_not_refresh_shared_overview(source: &str) { assert_eq!( ( - app.agents_overview.request_id, + app.agents_overview.active_refresh.map(|(id, _)| id), app.agents_overview.refresh_pending, ), (Some(request_id), false) @@ -778,7 +995,7 @@ async fn check_hidden_thread_does_not_refresh_shared_overview(source: &str) { assert_eq!( ( - app.agents_overview.request_id, + app.agents_overview.active_refresh.map(|(id, _)| id), app.agents_overview.refresh_pending, ), (Some(request_id), true) @@ -890,7 +1107,7 @@ async fn agents_overview_details_show_available_attention_without_expanding_rows assert!(!rendered.contains("Which dependency version")); assert!(rendered.contains("Waiting for approval.")); assert!(app.thread_event_channels.is_empty()); - assert!(app.agents_overview.request_id.is_none()); + assert!(app.agents_overview.active_refresh.is_none()); let view = app.agents_overview_view(threads, Some(unloaded)); app.chat_widget.show_bottom_pane_view(Box::new(view)); @@ -1091,7 +1308,7 @@ async fn agents_overview_reasoning_uses_existing_events_and_expires_with_attachm insta::with_settings!({snapshot_path => "../snapshots"}, { insta::assert_snapshot!("agents_overview_live_activity", render_bottom_popup(&app.chat_widget, /*width*/ 96).replace(&format!("{project} 1"), &normalized_group).replace(&project, "/tmp/project")); }); - assert!(app.agents_overview.request_id.is_none()); + assert!(app.agents_overview.active_refresh.is_none()); // A working child can stream through a server attachment without a local channel. let parent = overview_thread( @@ -2771,7 +2988,7 @@ async fn command_center_refresh_failure_is_inline_and_clears_on_success() -> Res }), ] { let request_id = Uuid::new_v4(); - app.agents_overview.request_id = Some(request_id); + app.agents_overview.active_refresh = Some((request_id, server.backend_id())); app.apply_agents_overview_thread_refresh(&server, request_id, result); assert!( app.agents_overview @@ -2790,7 +3007,7 @@ async fn command_center_refresh_failure_is_inline_and_clears_on_success() -> Res Error loading tasks "); let request_id = Uuid::new_v4(); - app.agents_overview.request_id = Some(request_id); + app.agents_overview.active_refresh = Some((request_id, server.backend_id())); app.apply_agents_overview_thread_refresh( &server, request_id, @@ -3116,27 +3333,24 @@ async fn command_center_new_actions_use_selection_and_leave_metadata_text_alone( "Task", ThreadStatus::Idle, ); + let owner = BackendId::new(); + app.backend_registry.observe(owner, id); target.cwd = test_path_buf("/tmp/checkout/subdir").abs(); - app.agents_overview.pinned_thread_ids = Some(Vec::new()); + app.agents_overview.pins_by_backend = + [(crate::backend_id::BackendId::new(), Some(Vec::new()))].into(); let mut view = app.agents_overview_view(vec![target.clone()], Some(id)); view.handle_key_event(KeyCode::Char('p').into()); - assert!(matches!( - rx.try_recv(), - Ok(AppEvent::ToggleAgentsOverviewPin { - thread_id, - pinned: true, - }) if thread_id == id - )); + let _ = rx.try_recv().expect("first pin shortcut emits an event"); view.handle_key_event(KeyCode::Char('p').into()); assert!(rx.try_recv().is_err()); for grouping_index in 0..3 { view.handle_key_event(KeyCode::Char('n').into()); assert!( - matches!(rx.try_recv(), Ok(AppEvent::NewAgentsOverviewSession { cwd: Some(cwd) }) if cwd == target.cwd) + matches!(rx.try_recv(), Ok(AppEvent::NewAgentsOverviewSession { backend_id: Some(id), cwd: Some(cwd) }) if id == owner && cwd == target.cwd) ); view.handle_key_event(KeyCode::Char('w').into()); assert!( - matches!(rx.try_recv(), Ok(AppEvent::NewAgentsOverviewWorktree { cwd: Some(cwd) }) if cwd == target.cwd) + matches!(rx.try_recv(), Ok(AppEvent::NewAgentsOverviewWorktree { backend_id: Some(id), cwd: Some(cwd) }) if id == owner && cwd == target.cwd) ); view.handle_key_event(KeyCode::Char('g').into()); if grouping_index == 0 { @@ -3161,6 +3375,9 @@ async fn command_center_new_actions_use_selection_and_leave_metadata_text_alone( empty.handle_key_event(KeyCode::Char('n').into()); assert!(matches!( rx.try_recv(), - Ok(AppEvent::NewAgentsOverviewSession { cwd: None }) + Ok(AppEvent::NewAgentsOverviewSession { + backend_id: None, + cwd: None + }) )); } diff --git a/codex-rs/tui/src/app/agents_overview_threads.rs b/codex-rs/tui/src/app/agents_overview_threads.rs index e4456c4bf2..e4be36118d 100644 --- a/codex-rs/tui/src/app/agents_overview_threads.rs +++ b/codex-rs/tui/src/app/agents_overview_threads.rs @@ -74,7 +74,7 @@ impl App { } pub(super) fn show_more_agents_overview(&mut self, app_server: &AppServerSession) { - if !self.agents_overview.discovery.has_more() { + if !self.agents_overview.has_more(app_server.backend_id()) { return; } self.agents_overview.show_more_requested |= self.agents_overview.refill_count == 0; @@ -217,7 +217,7 @@ impl App { { self.agents_overview.refresh_thread_ids.insert(thread_id); } - if self.agents_overview.request_id.is_some() { + if self.agents_overview.active_refresh.is_some() { // Replay the latest notification of each kind after the read, in arrival order. // This also protects sessions whose metadata has not arrived in the initial seed. let pending = self @@ -256,7 +256,6 @@ impl App { .chat_widget .selected_index_for_present_view(AGENTS_OVERVIEW_VIEW_ID) .is_none() - || (!self.agents_overview.initialized && self.agents_overview.request_id.is_none()) { return; } @@ -264,26 +263,34 @@ impl App { } fn start_agents_overview_refresh(&mut self, app_server: &AppServerSession) { + let backend_id = app_server.backend_id(); + let initialized = self.agents_overview.initialized.contains(&backend_id); + let source_has_more = self.agents_overview.has_more(backend_id); let visible = self .chat_widget .selected_index_for_present_view(AGENTS_OVERVIEW_VIEW_ID) .is_some(); - if !visible && self.agents_overview.initialized { + if !visible && initialized { return; } - if self.agents_overview.request_id.is_some() { + if self.agents_overview.active_refresh.is_some() { self.agents_overview.refresh_pending = true; return; } - if self.agents_overview.initialized + if initialized && self.agents_overview.refresh_thread_ids.is_empty() && !self.agents_overview.show_more_requested - && !(self.agents_overview.refill_count > 0 && self.agents_overview.discovery.has_more()) + && !(self.agents_overview.refill_count > 0 && source_has_more) && !self.agents_overview.pin_refresh_requested { return; } + let owns_thread = |thread_id| { + self.backend_registry + .owner(thread_id) + .is_none_or(|owner| owner == backend_id) + }; let mut unread_rows = Vec::new(); if std::mem::take(&mut self.agents_overview.pin_refresh_requested) { unread_rows.extend( @@ -291,18 +298,17 @@ impl App { .visible_thread_ids .iter() .copied() - .filter(|id| self.agents_overview.unread.contains(id)), + .filter(|id| self.agents_overview.unread.contains(id) && owns_thread(*id)), ); } let request_id = Uuid::new_v4(); - self.agents_overview.request_id = Some(request_id); - let initialized = self.agents_overview.initialized; + self.agents_overview.active_refresh = Some((request_id, backend_id)); + self.agents_overview.refresh_invalidated = false; let refill = self.agents_overview.refill_count; let show_more = refill == 0 && std::mem::take(&mut self.agents_overview.show_more_requested); self.agents_overview.refresh_show_more = show_more; - let discover = - !initialized || show_more || (refill > 0 && self.agents_overview.discovery.has_more()); + let discover = !initialized || show_more || (refill > 0 && source_has_more); let limit = if initialized && refill > 0 { refill.min(10) } else { @@ -310,13 +316,20 @@ impl App { }; let mut discovery = discover.then(|| { if initialized { - self.agents_overview.discovery.clone() + self.agents_overview + .discovery + .get(&backend_id) + .cloned() + .unwrap_or_default() } else { - self.agents_overview.removed_threads.clear(); + self.agents_overview + .removed_threads + .retain(|id| !owns_thread(*id)); Default::default() } }); let mut thread_ids = std::mem::take(&mut self.agents_overview.refresh_thread_ids); + thread_ids.retain(|id| owns_thread(*id)); self.agents_overview.active_refresh_thread_ids = thread_ids.clone(); let request_handle = app_server.request_handle(); let app_event_tx = self.app_event_tx.clone(); diff --git a/codex-rs/tui/src/app/agents_overview_usage.rs b/codex-rs/tui/src/app/agents_overview_usage.rs index 47a4868d81..d4712475dc 100644 --- a/codex-rs/tui/src/app/agents_overview_usage.rs +++ b/codex-rs/tui/src/app/agents_overview_usage.rs @@ -35,14 +35,20 @@ pub(super) struct AgentsOverviewUsage { } impl App { + pub(super) fn reset_agents_overview_usage_for_backend_switch(&mut self) { + self.agents_overview.usage.retain(|_, usage| { + usage.estimate = None; + usage.fetched_at = None; + usage.tokens.is_some() + }); + } + pub(super) fn refresh_agents_overview_usage( &mut self, app_server: &AppServerSession, frame_requester: FrameRequester, ) { - if self.reconnect.offline - || self.agents_overview.usage_disabled - || self.agents_overview.pending_usage.is_some() + if self.agents_overview.pending_usage.is_some() || !self.chat_widget.has_codex_backend_auth() || !matches!( self.chat_widget.current_plan_type(), @@ -64,6 +70,17 @@ impl App { else { return; }; + let owner = app_server.backend_id(); + // The plan used to format usage belongs to the displayed backend. + if self + .backend_registry + .owner(thread_id) + .is_some_and(|thread_owner| thread_owner != owner) + || self.reconnect.offline + || self.agents_overview.usage_disabled + { + return; + } if let Some(age) = self .agents_overview .usage @@ -136,9 +153,12 @@ impl App { usage.estimate = Some(estimate); } Ok(ThreadUsageOutcome::Disabled) => { + let owner = self.app_server_backend_id; self.agents_overview.usage_disabled = true; - for usage in self.agents_overview.usage.values_mut() { - usage.estimate = None; + for (id, usage) in &mut self.agents_overview.usage { + if self.backend_registry.owner(*id).or(owner) == owner { + usage.estimate = None; + } } } Ok(ThreadUsageOutcome::Available(_)) | Err(_) => {} diff --git a/codex-rs/tui/src/app/agents_overview_usage_tests.rs b/codex-rs/tui/src/app/agents_overview_usage_tests.rs index 35fb2d47ef..6a601bf62d 100644 --- a/codex-rs/tui/src/app/agents_overview_usage_tests.rs +++ b/codex-rs/tui/src/app/agents_overview_usage_tests.rs @@ -14,6 +14,7 @@ use pretty_assertions::assert_eq; async fn selected_usage_is_cached_and_account_changes_discard_old_results() -> Result<()> { let mut app = make_test_app_with_chat().await; let app_server = crate::start_embedded_app_server_for_picker(&app.config).await?; + app.app_server_backend_id = Some(app_server.backend_id()); app.chat_widget.update_account_state( /*status_account_display*/ None, Some(PlanType::Business), @@ -22,6 +23,7 @@ async fn selected_usage_is_cached_and_account_changes_discard_old_results() -> R ); let selected = ThreadId::from_u128(/*value*/ 1); let other = ThreadId::new(); + app.backend_registry.observe(app_server.backend_id(), other); let threads = [(selected, "Review parser"), (other, "Other task")].map(|(id, name)| { overview_thread(id, /*parent_thread_id*/ None, name, ThreadStatus::Idle) }); @@ -83,7 +85,7 @@ async fn selected_usage_is_cached_and_account_changes_discard_old_results() -> R reasoning_output_tokens: 0, }; app.agents_overview.threads.remove(&selected); - app.agents_overview.request_id = Some(request_id); + app.agents_overview.active_refresh = Some((request_id, app_server.backend_id())); app.track_agents_overview_notification(&ServerNotification::ThreadTokenUsageUpdated( ThreadTokenUsageUpdatedNotification { thread_id: selected.to_string(), @@ -159,13 +161,37 @@ async fn selected_usage_is_cached_and_account_changes_discard_old_results() -> R let other_usage = app.agents_overview.usage.entry(other).or_default(); other_usage.estimate = Some(expected.clone()); other_usage.tokens = Some(tokens.clone()); + let foreign = ThreadId::new(); + app.backend_registry + .observe(crate::backend_id::BackendId::new(), foreign); + let foreign_thread = overview_thread( + foreign, + /*parent_thread_id*/ None, + "Foreign task", + ThreadStatus::Idle, + ); + app.agents_overview + .threads + .insert(foreign, Some(foreign_thread.clone())); + let foreign_usage = app.agents_overview.usage.entry(foreign).or_default(); + foreign_usage.estimate = Some(expected.clone()); + let row = app.agents_overview.threads.remove(&selected).unwrap(); + app.agents_overview + .search_matches + .insert(selected, row.clone().unwrap()); app.agents_overview.pending_usage = Some((thread_id, request_id)); app.finish_agents_overview_usage(thread_id, request_id, Ok(ThreadUsageOutcome::Disabled)); + app.agents_overview.threads.insert(selected, row); assert!( app.agents_overview .usage - .values() - .all(|usage| usage.estimate.is_none()) + .iter() + .filter(|(id, _)| **id != foreign) + .all(|(_, usage)| usage.estimate.is_none()) + ); + assert_eq!( + app.agents_overview.usage[&foreign].estimate, + Some(expected.clone()) ); assert_eq!( app.agents_overview.usage[&other].tokens, @@ -178,6 +204,17 @@ async fn selected_usage_is_cached_and_account_changes_discard_old_results() -> R app.refresh_agents_overview_usage(&app_server, tui.frame_requester()); assert!(app.agents_overview.usage_disabled); assert_eq!(app.agents_overview.pending_usage, None); + let view = app.agents_overview_view(vec![foreign_thread], Some(foreign)); + assert!(view.rows[0].details.usage_lines.is_empty()); + app.agents_overview.usage.get_mut(&other).unwrap().estimate = Some(expected.clone()); + app.reset_agents_overview_usage_for_backend_switch(); + assert_eq!( + app.agents_overview.usage[&other].tokens, + Some(tokens.clone()) + ); + assert_eq!(app.agents_overview.usage[&other].estimate, None); + assert!(!app.agents_overview.usage.contains_key(&foreign)); + app.agents_overview.threads.remove(&foreign); for event in [ AppServerEvent::Lagged { skipped: 1 }, AppServerEvent::ServerNotification(Box::new(ServerNotification::AccountUpdated( diff --git a/codex-rs/tui/src/app/agents_overview_view.rs b/codex-rs/tui/src/app/agents_overview_view.rs index 6a2a7a5d3c..5574639619 100644 --- a/codex-rs/tui/src/app/agents_overview_view.rs +++ b/codex-rs/tui/src/app/agents_overview_view.rs @@ -108,6 +108,7 @@ fn is_subagent(thread: &Thread) -> bool { #[derive(Clone)] pub(super) struct AgentsOverviewRow { + pub(super) backend_id: Option, pub(super) search_id: Option, pub(super) details: AgentsOverviewDetails, pub(super) thread: Thread, @@ -229,6 +230,7 @@ pub(super) struct AgentsOverviewView { project_groups: Vec, directory_group_keys: Vec, pub(super) pinned_thread_ranks: Option>, + pub(super) pin_action_supported: bool, pub(super) pin_action_pending: bool, pub(super) can_return_to_chat: bool, selected: usize, @@ -307,6 +309,7 @@ impl AgentsOverviewView { project_groups, directory_group_keys, pinned_thread_ranks: None, + pin_action_supported: true, pin_action_pending: false, can_return_to_chat: false, selected, @@ -362,6 +365,7 @@ impl AgentsOverviewView { fn can_toggle_selected_pin(&self) -> bool { !self.pin_action_pending + && self.pin_action_supported && self.pinned_thread_ranks.is_some() && self .selected_row() @@ -787,6 +791,7 @@ impl BottomPaneView for AgentsOverviewView { } if self.agents_keymap.new_task.is_pressed(key) { self.app_event_tx.send(AppEvent::NewAgentsOverviewSession { + backend_id: self.selected_row().and_then(|row| row.backend_id), cwd: self.selected_row().map(|row| row.thread.cwd.clone()), }); return; @@ -794,6 +799,7 @@ impl BottomPaneView for AgentsOverviewView { if self.agents_keymap.new_worktree.is_pressed(key) { if self.worktrees_enabled { self.app_event_tx.send(AppEvent::NewAgentsOverviewWorktree { + backend_id: self.selected_row().and_then(|row| row.backend_id), cwd: self.selected_row().map(|row| row.thread.cwd.clone()), }); } diff --git a/codex-rs/tui/src/app/app_server_events.rs b/codex-rs/tui/src/app/app_server_events.rs index 5f3680248f..cadda4127d 100644 --- a/codex-rs/tui/src/app/app_server_events.rs +++ b/codex-rs/tui/src/app/app_server_events.rs @@ -78,11 +78,12 @@ impl App { if let Some(task) = self.agents_overview.refresh_task.take() { task.abort(); } - self.agents_overview.request_id = None; + self.agents_overview.active_refresh = None; self.agents_overview.refresh_show_more = false; self.agents_overview.active_refresh_thread_ids.clear(); self.agents_overview.refresh_pending = false; - self.agents_overview.initialized = false; + self.agents_overview + .invalidate_discovery(app_server_client.backend_id(), /*displayed*/ true); self.agents_overview.refresh_notifications.clear(); self.agents_overview.activity.clear(); self.agents_overview.last_messages.clear(); @@ -100,6 +101,14 @@ impl App { } } AppServerEvent::ServerNotification(notification) => { + if let ServerNotificationThreadTarget::Thread(thread_id) = + server_notification_thread_target(notification.as_ref()) + && !self + .backend_registry + .observe(app_server_client.backend_id(), thread_id) + { + return; + } let request_resolved = matches!( notification.as_ref(), ServerNotification::ServerRequestResolved(_) @@ -258,17 +267,11 @@ impl App { .or_default(); } self.track_agents_overview_notification(¬ification); - // Retained blank sessions stay subscribed after their event channels are cleared. if let ServerNotification::ThreadSettingsUpdated(settings) = ¬ification && let Ok(thread_id) = ThreadId::from_string(&settings.thread_id) - && self.agents_overview.blank_sessions.contains_key(&thread_id) - && !self.thread_event_channels.contains_key(&thread_id) { - self.apply_thread_settings_to_cached_session(thread_id, &settings.thread_settings) + self.update_retained_blank_settings(thread_id, settings) .await; - if let Some(input) = self.agents_overview.input_states.get_mut(&thread_id) { - input.pending_thread_settings = Some(settings.clone()); - } } if matches!( ¬ification, @@ -359,6 +362,9 @@ impl App { return; } ServerNotification::AccountUpdated(notification) => { + self.agents_overview + .invalidate_discovery(app_server_client.backend_id(), /*displayed*/ true); + self.refresh_changed_agents_overview_threads(app_server_client); self.agents_overview.usage.clear(); self.agents_overview.pending_usage = None; self.agents_overview.usage_disabled = false; diff --git a/codex-rs/tui/src/app/backend_events.rs b/codex-rs/tui/src/app/backend_events.rs index f933611f19..4c29561596 100644 --- a/codex-rs/tui/src/app/backend_events.rs +++ b/codex-rs/tui/src/app/backend_events.rs @@ -8,6 +8,7 @@ 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( @@ -17,14 +18,50 @@ impl App { event: AppServerSessionEvent, ) { 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 + if event.backend_id != app_server.backend_id() { + return; + } + let invalidates_discovery = match &event.event { + AppServerEvent::Lagged { .. } | AppServerEvent::Disconnected { .. } => true, + AppServerEvent::ServerNotification(notification) => matches!( + notification.as_ref(), + ServerNotification::AccountUpdated(_) + | ServerNotification::ThreadArchived(_) + | ServerNotification::ThreadDeleted(_) + | ServerNotification::ThreadUnarchived(_) + ), + _ => false, + }; + if invalidates_discovery { + self.agents_overview + .invalidate_discovery(event.backend_id, /*displayed*/ false); + } + if 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); + if matches!( + notification.as_ref(), + ServerNotification::ThreadArchived(_) + | ServerNotification::ThreadDeleted(_) + | ServerNotification::ThreadUnarchived(_) + ) && self.backend_registry.owner(thread_id) == Some(event.backend_id) + { + // A parked removal must not consume the displayed backend's page budget. + let refill_count = self.agents_overview.refill_count; + self.track_agents_overview_notification(notification); + self.agents_overview.refill_count = refill_count; + self.repaint_agents_overview(); + } + if let ServerNotification::ThreadSettingsUpdated(settings) = notification.as_ref() + && self.backend_registry.owner(thread_id) == Some(event.backend_id) + { + self.update_retained_blank_settings(thread_id, settings) + .await; + } + if self.voice_owner_thread_id() == Some(thread_id) { + self.deliver_background_voice_notification(thread_id, notification); + } } return; } diff --git a/codex-rs/tui/src/app/backend_registry.rs b/codex-rs/tui/src/app/backend_registry.rs new file mode 100644 index 0000000000..94631faf93 --- /dev/null +++ b/codex-rs/tui/src/app/backend_registry.rs @@ -0,0 +1,38 @@ +//! Tracks conversation ownership across backends. +//! A thread ID keeps its first observed owner across connection replacement. + +use crate::app_server_session::AppServerStartedThread; +use crate::backend_id::BackendId; +use codex_protocol::ThreadId; +use std::collections::HashMap; + +#[derive(Default)] +pub(super) struct BackendRegistry { + owners: HashMap, +} + +impl BackendRegistry { + pub(super) fn owner(&self, thread: ThreadId) -> Option { + self.owners.get(&thread).copied() + } + + pub(super) fn register_started( + &mut self, + owner: BackendId, + started: &AppServerStartedThread, + ) -> color_eyre::Result<()> { + color_eyre::eyre::ensure!( + self.observe(owner, started.session.thread_id), + "conversation belongs to a different backend" + ); + Ok(()) + } + + pub(super) fn observe(&mut self, owner: BackendId, thread: ThreadId) -> bool { + if *self.owners.entry(thread).or_insert(owner) != owner { + tracing::warn!(%thread, "ignoring conflicting thread ownership"); + return false; + } + true + } +} diff --git a/codex-rs/tui/src/app/backend_selection.rs b/codex-rs/tui/src/app/backend_selection.rs index ef345a41b0..5c46dbce9a 100644 --- a/codex-rs/tui/src/app/backend_selection.rs +++ b/codex-rs/tui/src/app/backend_selection.rs @@ -389,12 +389,18 @@ impl App { self.clear_active_thread().await; self.agents_overview.search_matches.clear(); self.agents_overview.pending_usage = None; + self.reset_agents_overview_usage_for_backend_switch(); self.agents_overview.usage_disabled = false; + self.agents_overview.reset_visible_pagination(); self.app_event_tx = AppEventSender::new(self.app_event_tx.app_event_tx.clone()); + self.app_server_backend_id = Some(destination.backend_id()); self.app_server_target = target; self.reconnect = Default::default(); if let Ok(mut state) = self.agents_overview.view_state.lock() { state.connection_notice = None; + let destination_id = destination.backend_id(); + state.has_more = self.agents_overview.initialized.contains(&destination_id) + && self.agents_overview.has_more(destination_id); state.restart_history_search(); } self.rate_limit_hard_stop_generation = self.rate_limit_hard_stop_generation.wrapping_add(1); @@ -428,6 +434,12 @@ impl App { let displayed_session = backends.displayed; let owner = target.backend_id; let outgoing = backends.displayed_owner(); + if let Some(registered_owner) = self.backend_registry.owner(target.thread_id) + && let Err(error) = target.ensure_backend(registered_owner) + { + self.add_session_picker_error(error.to_string()); + return Ok(AppRunControl::Continue); + } if owner == outgoing { return self .apply_resume_picker_selection( @@ -501,13 +513,21 @@ impl App { } } } + let restored_blank_session = prepared.restored_blank_session; + let has_resumed_mode = prepared.thread.session.collaboration_mode.is_some(); let control = self .activate_prepared_session(tui, destination, prepared) .await?; backends.displayed = destination_slot; + if !restored_blank_session { + self.agents_overview.blank_sessions.remove(&thread_id); + } 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 + if !restored_blank_session { + self.chat_widget + .update_input_collaboration_mode(&mut input, has_resumed_mode); + } + let settings = restored_blank_session .then(|| input.pending_thread_settings.take()) .flatten(); self.restore_session_draft(thread_id, input, settings).await; diff --git a/codex-rs/tui/src/app/event_dispatch.rs b/codex-rs/tui/src/app/event_dispatch.rs index 9b695c3adb..d98f762733 100644 --- a/codex-rs/tui/src/app/event_dispatch.rs +++ b/codex-rs/tui/src/app/event_dispatch.rs @@ -103,6 +103,7 @@ impl App { | AppEvent::ManagedWorktreeCreated(_) | AppEvent::AgentsOverviewWorktreeCreated(_) | AppEvent::AgentsOverviewPinToggled { .. } + | AppEvent::AgentsOverviewUsageLoaded { .. } | AppEvent::AppendMessageHistoryEntry { .. } | AppEvent::BeginInitialHistoryReplayBuffer | AppEvent::BeginThreadSwitchHistoryReplayBuffer @@ -112,6 +113,26 @@ impl App { { return Ok(AppRunControl::Continue); } + let overview_action_thread = match &event { + AppEvent::ForkAgentsOverviewThread { thread_id } + | AppEvent::RenameAgentsOverviewThread { thread_id, .. } + | AppEvent::ToggleAgentsOverviewPin { thread_id, .. } + | AppEvent::RunAgentsOverviewAction { thread_id, .. } + | AppEvent::StopAgentsOverviewThread { thread_id } => Some(*thread_id), + _ => None, + }; + let overview_action_owner = match &event { + AppEvent::NewAgentsOverviewSession { backend_id, .. } + | AppEvent::NewAgentsOverviewWorktree { backend_id, .. } => *backend_id, + _ => overview_action_thread.and_then(|id| self.backend_registry.owner(id)), + }; + if overview_action_owner.is_some_and(|owner| owner != app_server.backend_id()) { + if matches!(&event, AppEvent::ToggleAgentsOverviewPin { .. }) { + self.repaint_agents_overview(); + } + self.add_agents_overview_error("Open this task before using that action.".into()); + return Ok(AppRunControl::Continue); + } if matches!( &event, AppEvent::OpenWindowsSandboxEnablePrompt { .. } @@ -3209,7 +3230,7 @@ impl App { self.refresh_changed_agents_overview_threads(app_server); } } - AppEvent::NewAgentsOverviewSession { cwd } => { + AppEvent::NewAgentsOverviewSession { cwd, .. } => { return Box::pin(self.new_agents_overview_session(tui, app_server, cwd)).await; } AppEvent::AgentsOverviewThreadsLoaded { request_id, result } => { @@ -3257,7 +3278,7 @@ impl App { return control; } AppEvent::ForkAgentsOverviewThreadReady { .. } => unreachable!("normalized above"), - AppEvent::NewAgentsOverviewWorktree { cwd } => { + AppEvent::NewAgentsOverviewWorktree { cwd, .. } => { Box::pin(self.new_agents_overview_worktree(tui, app_server, cwd)).await; } AppEvent::AgentsOverviewWorktreeCreated(result) => { diff --git a/codex-rs/tui/src/app/owned_transcript_tests.rs b/codex-rs/tui/src/app/owned_transcript_tests.rs index 6ce1f30645..e8b1919943 100644 --- a/codex-rs/tui/src/app/owned_transcript_tests.rs +++ b/codex-rs/tui/src/app/owned_transcript_tests.rs @@ -121,7 +121,7 @@ async fn external_writer_escape_returns_to_command_center_without_editing() -> R (false, false, String::new()), ); if offline { - assert!(app.agents_overview.request_id.is_none()); + assert!(app.agents_overview.active_refresh.is_none()); assert!(app.reconnect.presentation == reconnect::ReconnectPresentation::Overview); insta::assert_snapshot!( "external_writer_escape_offline_command_center", diff --git a/codex-rs/tui/src/app/reconnect.rs b/codex-rs/tui/src/app/reconnect.rs index d6e2f6f26a..d9f1f1f5c2 100644 --- a/codex-rs/tui/src/app/reconnect.rs +++ b/codex-rs/tui/src/app/reconnect.rs @@ -240,7 +240,12 @@ impl App { self.reconnect.offline = true; self.account_email_request_id = None; // Cached blank sessions are usable only while this connection owns a subscription. - self.agents_overview.blank_sessions.clear(); + self.agents_overview.blank_sessions.retain(|id, _| { + self.backend_registry + .owner(*id) + .zip(self.app_server_backend_id) + .is_some_and(|(owner, reconnecting)| owner != reconnecting) + }); self.reconnect.failed = false; if self.pending_server_version_notice.take().is_some() { self.reconnect.seen_version_notice = None; @@ -256,7 +261,11 @@ impl App { if let Some(task) = self.agents_overview.refresh_task.take() { task.abort(); } - self.agents_overview.request_id = None; + self.agents_overview.active_refresh = None; + if let Some(owner) = self.app_server_backend_id { + self.agents_overview + .invalidate_discovery(owner, /*displayed*/ true); + } self.agents_overview.refresh_show_more = false; self.agents_overview.active_refresh_thread_ids.clear(); self.agents_overview.refresh_pending = false; @@ -372,6 +381,7 @@ impl App { // prevents their late completions from submitting follow-up operations on the new one. let (tx, rx) = mpsc::unbounded_channel(); self.app_event_tx = AppEventSender::new(tx); + self.app_server_backend_id = Some(session.backend_id()); *app_event_rx = rx; // The old timer targets the discarded channel, so a new preview can be scheduled. self.config_notification_test_pending = None; @@ -445,7 +455,7 @@ impl App { cancellation.cancel(); } self.agents_overview.dispatched_requests.clear(); - self.agents_overview.request_id = None; + self.agents_overview.active_refresh = None; self.agents_overview.refresh_show_more = false; self.agents_overview.active_refresh_thread_ids.clear(); self.agents_overview.refresh_pending = false; @@ -560,7 +570,9 @@ impl App { self.refresh_pending_thread_approvals().await; // Discover tasks whose notifications were missed, without clearing retained rows. // A hidden overview performs this discovery when it is next opened. - self.agents_overview.initialized = false; + self.agents_overview + .initialized + .remove(&app_server.backend_id()); if self.reconnect.presentation == ReconnectPresentation::Overview { let threads = self .agents_overview diff --git a/codex-rs/tui/src/app/resume_preparation.rs b/codex-rs/tui/src/app/resume_preparation.rs index 55b8791adc..2735654b35 100644 --- a/codex-rs/tui/src/app/resume_preparation.rs +++ b/codex-rs/tui/src/app/resume_preparation.rs @@ -5,6 +5,7 @@ //! prepared result retains the connection identity and any provider error for read-only history. use super::*; +use codex_app_server_client::TypedRequestError; /// Settings retained for a backend that is not currently displayed. #[derive(Clone)] @@ -29,6 +30,7 @@ pub(super) struct PreparedResume { pub(super) config: Config, pub(super) local_settings: crate::local_settings::LocalSettings, pub(super) thread: AppServerStartedThread, + pub(super) restored_blank_session: bool, pub(super) read_only: bool, pub(super) history_notice: Option<&'static str>, pub(super) provider_error: Option, @@ -43,6 +45,57 @@ impl App { } } + pub(super) async fn live_blank_session( + &mut self, + app_server: &mut AppServerSession, + thread_id: ThreadId, + thread: &codex_app_server_protocol::Thread, + ) -> Option { + if !matches!(thread.status, codex_app_server_protocol::ThreadStatus::Idle) { + return None; + } + let mut blank = self.agents_overview.blank_sessions.get(&thread_id)?.clone(); + + // A turn can start while this backend is parked, without the TUI seeing it. + let paginated = matches!( + thread.history_mode, + codex_app_server_protocol::ThreadHistoryMode::Paginated + ) && app_server.supports_paginated_history(); + let untouched = if paginated { + match app_server.thread_turns_page(thread_id, /*cursor*/ None, /*limit*/ 1).await { + Ok(page) => Some(page.data.is_empty() && page.next_cursor.is_none()), + Err(error) if Self::can_fallback_from_include_turns_error(&error) => Some(true), + Err(error) if error.chain().any(|cause| matches!( + cause.downcast_ref::(), + Some(TypedRequestError::Server { source, .. }) + if crate::app_server_session::is_history_pagination_unsupported(source) + )) => None, + Err(_) => Some(false), + } + } else { + None + }; + let untouched = match untouched { + Some(untouched) => untouched, + None => match app_server + .thread_read(thread_id, /*include_turns*/ true) + .await + { + Ok(thread) => { + thread.turns.is_empty() + && matches!(thread.status, codex_app_server_protocol::ThreadStatus::Idle) + } + Err(error) => Self::can_fallback_from_include_turns_error(&error), + }, + }; + if !untouched { + return None; + } + // An untouched task may have no rollout to resume yet. + blank.session.thread_name.clone_from(&thread.name); + Some(blank) + } + pub(super) async fn prepare_target_session( &mut self, tui: &mut tui::Tui, @@ -50,6 +103,12 @@ impl App { target_session: &SessionTarget, source: ResumeSource<'_>, ) -> std::result::Result { + if let Some(owner) = self.backend_registry.owner(target_session.thread_id) + && let Err(error) = target_session.ensure_backend(owner) + { + self.add_session_picker_error(error.to_string()); + return Err(AppRunControl::Continue); + } let (mut resume_config, local_settings) = self .resume_config_for_target(tui, app_server, target_session, source) .await?; @@ -74,15 +133,34 @@ impl App { if let Some(history_mode) = target_session.history_mode { app_server.remember_thread_history_mode(target_session.thread_id, history_mode); } - let resumed = app_server - .resume_thread_with_permission_overrides( - &local_settings, - resume_config.clone(), - target_session.thread_id, - model_settings, - permission_overrides, - ) - .await; + let thread_id = target_session.thread_id; + let live_blank = if matches!(source, ResumeSource::Backend(_)) + && self.backend_registry.owner(thread_id) == Some(app_server.backend_id()) + && self.agents_overview.blank_sessions.contains_key(&thread_id) + && let Ok(thread) = app_server + .thread_read(thread_id, /*include_turns*/ false) + .await + { + self.live_blank_session(app_server, thread_id, &thread) + .await + } else { + None + }; + let restored_blank_session = live_blank.is_some(); + let resumed = match live_blank { + Some(thread) => Ok(thread), + None => { + app_server + .resume_thread_with_permission_overrides( + &local_settings, + resume_config.clone(), + thread_id, + model_settings, + permission_overrides, + ) + .await + } + }; let mut history_notice = None; let mut provider_error = None; let (resumed, read_only) = match resumed { @@ -125,6 +203,7 @@ impl App { config: resume_config, local_settings, thread: resumed, + restored_blank_session, read_only, history_notice, provider_error, diff --git a/codex-rs/tui/src/app/session_lifecycle.rs b/codex-rs/tui/src/app/session_lifecycle.rs index ac4d2b0376..22a2f0d515 100644 --- a/codex-rs/tui/src/app/session_lifecycle.rs +++ b/codex-rs/tui/src/app/session_lifecycle.rs @@ -1374,6 +1374,13 @@ impl App { config, local_settings, } = prepared; + if let Err(error) = self + .backend_registry + .register_started(app_server.backend_id(), &started) + { + self.chat_widget.add_error_message(error.to_string()); + return; + } if let Some(thread_id) = self.current_displayed_thread_id() && let Some(blank) = self.agents_overview.blank_sessions.get_mut(&thread_id) { @@ -1741,6 +1748,7 @@ impl App { read_only, history_notice, provider_error, + .. } = prepared; if backend_id != app_server.backend_id() { self.add_session_picker_error( @@ -1749,6 +1757,8 @@ impl App { ); return Ok(AppRunControl::Continue); } + self.backend_registry + .observe(backend_id, resumed.session.thread_id); let resumed_thread_id = resumed.session.thread_id; let retained_input = ((self.chat_widget.is_external_writer_view() || self.chat_widget.has_provider_error()) diff --git a/codex-rs/tui/src/app/startup.rs b/codex-rs/tui/src/app/startup.rs index 4ca036004a..7a10e449db 100644 --- a/codex-rs/tui/src/app/startup.rs +++ b/codex-rs/tui/src/app/startup.rs @@ -1040,6 +1040,7 @@ See the Codex keymap documentation for supported actions and examples." feedback_audience, environment_manager, app_server_target, + app_server_backend_id: Some(app_server.backend_id()), pending_right_click_paste: None, right_click_paste_environment: super::right_click_paste::PasteEnvironment::detect(), reconnect: ReconnectState { @@ -1067,6 +1068,7 @@ See the Codex keymap documentation for supported actions and examples." pending_thread_titles: HashMap::new(), thread_event_listener_tasks: HashMap::new(), agent_navigation: AgentNavigationState::default(), + backend_registry: Default::default(), agents_overview, side_threads: HashMap::new(), saved_openings: Default::default(), @@ -1137,6 +1139,10 @@ See the Codex keymap documentation for supported actions and examples." /*server_version*/ None, ); } + if let Some(thread_id) = app.current_displayed_thread_id() { + app.backend_registry + .observe(app_server.backend_id(), thread_id); + } if start_in_agents_overview { app.open_agents_overview(&app_server); } else { @@ -1625,6 +1631,8 @@ See the Codex keymap documentation for supported actions and examples." } Some(_) => {} None => { + app.agents_overview + .invalidate_discovery(owner, owner == displayed_owner); if owner == displayed_owner { if app.reconnect.offline && app.pending_startup_thread_start { reconnect = None; diff --git a/codex-rs/tui/src/app/test_support.rs b/codex-rs/tui/src/app/test_support.rs index a07888d749..74103922cb 100644 --- a/codex-rs/tui/src/app/test_support.rs +++ b/codex-rs/tui/src/app/test_support.rs @@ -51,6 +51,7 @@ pub(crate) async fn make_test_app() -> Box { session_telemetry, terminal_size_tags: Vec::new(), app_event_tx, + backend_registry: Default::default(), chat_widget, workspace_command_runner: None, backend_preference_context: backend_selection::BackendPreferenceContext::new( @@ -101,6 +102,7 @@ pub(crate) async fn make_test_app() -> Box { feedback_audience: FeedbackAudience::External, environment_manager: Arc::new(EnvironmentManager::default_for_tests()), app_server_target: crate::AppServerTarget::Embedded, + app_server_backend_id: None, pending_right_click_paste: None, right_click_paste_environment: super::right_click_paste::PasteEnvironment { primary: false, diff --git a/codex-rs/tui/src/app/tests.rs b/codex-rs/tui/src/app/tests.rs index cf56ac7a0b..a77607b4e6 100644 --- a/codex-rs/tui/src/app/tests.rs +++ b/codex-rs/tui/src/app/tests.rs @@ -8983,6 +8983,7 @@ async fn make_test_app() -> Box { feedback_audience: FeedbackAudience::External, environment_manager: Arc::new(EnvironmentManager::default_for_tests()), app_server_target: crate::AppServerTarget::Embedded, + app_server_backend_id: None, reconnect: Default::default(), pending_right_click_paste: None, right_click_paste_environment: super::right_click_paste::PasteEnvironment { @@ -9008,6 +9009,7 @@ async fn make_test_app() -> Box { thread_event_listener_tasks: HashMap::new(), agent_navigation: AgentNavigationState::default(), pending_server_profiles: HashMap::new(), + backend_registry: Default::default(), agents_overview: Default::default(), side_threads: HashMap::new(), saved_openings: Default::default(), @@ -9123,6 +9125,7 @@ pub(super) async fn make_test_app_with_channels() -> ( feedback_audience: FeedbackAudience::External, environment_manager: Arc::new(EnvironmentManager::default_for_tests()), app_server_target: crate::AppServerTarget::Embedded, + app_server_backend_id: None, reconnect: Default::default(), pending_right_click_paste: None, right_click_paste_environment: super::right_click_paste::PasteEnvironment { @@ -9148,6 +9151,7 @@ pub(super) async fn make_test_app_with_channels() -> ( thread_event_listener_tasks: HashMap::new(), agent_navigation: AgentNavigationState::default(), pending_server_profiles: HashMap::new(), + backend_registry: Default::default(), agents_overview: Default::default(), side_threads: HashMap::new(), saved_openings: Default::default(), diff --git a/codex-rs/tui/src/app/tests/backend_event_tests.rs b/codex-rs/tui/src/app/tests/backend_event_tests.rs index 6f74cb1c6f..879e9b4f81 100644 --- a/codex-rs/tui/src/app/tests/backend_event_tests.rs +++ b/codex-rs/tui/src/app/tests/backend_event_tests.rs @@ -1,12 +1,114 @@ //! Incoming events cannot use a colliding native thread ID to mutate another backend's state. use super::*; +use crate::app::agents_overview::tests::overview_thread; +use crate::app_event::AgentsOverviewThreadRefresh; use crate::app_server_session::AppServerSessionEvent; use crate::backend_id::BackendId; use codex_app_server_client::AppServerEvent; use codex_app_server_protocol::AccountRateLimitsUpdatedNotification; use codex_app_server_protocol::AccountUpdatedNotification; +use codex_app_server_protocol::ThreadStatus; use pretty_assertions::assert_eq; +use std::collections::HashMap; + +#[tokio::test] +async fn background_archive_and_delete_remove_only_owned_rows() -> Result<()> { + let (mut app, _events, _) = make_test_app_with_channels().await; + let (server, _, proxy) = + session_lifecycle_requests::start_recording_remote_app_server(&app.config).await?; + let owner = server.backend_id(); + let displayed = BackendId::new(); + let collision = ThreadId::new(); + assert!(app.backend_registry.observe(displayed, collision)); + let collision_row = overview_thread( + collision, + /*parent_thread_id*/ None, + "other backend", + ThreadStatus::Idle, + ); + app.agents_overview + .threads + .insert(collision, Some(collision_row.clone())); + + for deleted in [false, true] { + let id = ThreadId::new(); + assert!(app.backend_registry.observe(owner, id)); + let row = overview_thread( + id, + /*parent_thread_id*/ None, + "owned row", + ThreadStatus::Idle, + ); + app.agents_overview.threads.insert(id, Some(row.clone())); + let request_id = uuid::Uuid::new_v4(); + app.agents_overview.active_refresh = Some((request_id, owner)); + for target in [id, collision] { + let notification = if deleted { + ServerNotification::ThreadDeleted(serde_json::from_value( + serde_json::json!({ "threadId": target.to_string() }), + )?) + } else { + ServerNotification::ThreadArchived(serde_json::from_value( + serde_json::json!({ "threadId": target.to_string() }), + )?) + }; + app.handle_polled_backend_event( + &server, + displayed, + AppServerSessionEvent { + backend_id: owner, + event: AppServerEvent::ServerNotification(Box::new(notification)), + }, + ) + .await; + } + assert!(!app.agents_overview.threads.contains_key(&id)); + assert!(app.agents_overview.threads.contains_key(&collision)); + assert!(app.agents_overview.refresh_invalidated); + assert_eq!(app.agents_overview.refill_count, 0); + app.apply_agents_overview_thread_refresh( + &server, + request_id, + Ok(AgentsOverviewThreadRefresh { + cleared_unread: Vec::new(), + threads: HashMap::from([(id, Some((row, false, None)))]), + last_messages: HashMap::new(), + recent_seed_complete: true, + discovery: None, + pinned_thread_ids: None, + }), + ); + assert!(!app.agents_overview.threads.contains_key(&id)); + assert_eq!( + app.agents_overview.threads.get(&collision), + Some(&Some(collision_row.clone())) + ); + if !deleted { + let notification = ServerNotification::ThreadUnarchived(serde_json::from_value( + serde_json::json!({ "threadId": id.to_string() }), + )?); + app.handle_polled_backend_event( + &server, + displayed, + AppServerSessionEvent { + backend_id: owner, + event: AppServerEvent::ServerNotification(Box::new(notification)), + }, + ) + .await; + assert!(!app.agents_overview.removed_threads.contains(&id)); + } + } + app.show_more_agents_overview(&server); + assert!(app.agents_overview.refresh_show_more); + if let Some(task) = app.agents_overview.refresh_task.take() { + task.abort(); + } + server.shutdown().await?; + proxy.await??; + Ok(()) +} #[tokio::test] async fn foreign_backend_events_leave_thread_cache_and_connection_untouched() -> Result<()> { @@ -51,6 +153,7 @@ async fn foreign_backend_events_leave_thread_cache_and_connection_untouched() -> ) .await; } + let owner = server.backend_id(); for event in [ AppServerEvent::Lagged { skipped: 1 }, AppServerEvent::Disconnected { @@ -70,6 +173,18 @@ async fn foreign_backend_events_leave_thread_cache_and_connection_untouched() -> }, ))), ] { + let account_updated = matches!( + &event, + AppServerEvent::ServerNotification(notification) + if matches!(notification.as_ref(), ServerNotification::AccountUpdated(_)) + ); + if account_updated { + app.agents_overview.initialized.extend([owner, foreign]); + app.agents_overview + .discovery + .insert(owner, Default::default()); + app.agents_overview.active_refresh = Some((uuid::Uuid::new_v4(), owner)); + } app.handle_polled_backend_event( &server, foreign, @@ -79,6 +194,12 @@ async fn foreign_backend_events_leave_thread_cache_and_connection_untouched() -> }, ) .await; + if account_updated { + assert!(!app.agents_overview.discovery.contains_key(&owner)); + assert!(!app.agents_overview.initialized.contains(&owner)); + assert!(app.agents_overview.initialized.contains(&foreign)); + assert!(app.agents_overview.refresh_invalidated); + } } assert!(!app.reconnect.offline); assert_eq!(app.rate_limit_hard_stop_generation, hard_stop_generation); 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 4dcbc4d671..2ef1b5777f 100644 --- a/codex-rs/tui/src/app/tests/backend_ownership_tests.rs +++ b/codex-rs/tui/src/app/tests/backend_ownership_tests.rs @@ -102,6 +102,23 @@ async fn foreign_backend_selection_cannot_resume_same_id_or_unarchive() -> Resul requests.lock().unwrap().is_empty(), "a foreign selection must cause zero requests" ); + + let collision = ThreadId::new(); + assert!(app.backend_registry.observe(origin.backend_id(), collision)); + let target = SessionTarget { + backend_id: server.backend_id(), + thread_id: collision, + cwd: None, + ..target + }; + app.resume_target_session(&mut tui, &mut server, target) + .await?; + assert_eq!(app.current_displayed_thread_id(), Some(thread_id)); + assert_eq!( + app.chat_widget.composer_text_with_pending(), + "keep this draft" + ); + assert!(requests.lock().unwrap().is_empty()); origin.shutdown().await?; server.shutdown().await?; proxy.await??; @@ -311,6 +328,7 @@ async fn prepared_resume_cannot_activate_through_another_backend() -> Result<()> blocks_direct_input: false, task_tools_available: true, }, + restored_blank_session: false, read_only: false, history_notice: None, provider_error: None, diff --git a/codex-rs/tui/src/app/tests/navigation_reconnect_tests.rs b/codex-rs/tui/src/app/tests/navigation_reconnect_tests.rs index ca127d7080..afff9d14b7 100644 --- a/codex-rs/tui/src/app/tests/navigation_reconnect_tests.rs +++ b/codex-rs/tui/src/app/tests/navigation_reconnect_tests.rs @@ -132,7 +132,6 @@ async fn reconnect_daemon_command_center_after_socket_replacement_without_a_conv app.agents_overview .last_messages .insert(selected, "Pre-disconnect answer".into()); - app.agents_overview.initialized = overview_initialized; // This task was unarchived elsewhere while disconnected. app.agents_overview.removed_threads.insert(added); let mut view = app.agents_overview_view( @@ -158,7 +157,8 @@ async fn reconnect_daemon_command_center_after_socket_replacement_without_a_conv .to_string() }; let stale_request = Uuid::new_v4(); - app.agents_overview.request_id = Some(stale_request); + let refresh_owner = crate::backend_id::BackendId::new(); + app.agents_overview.active_refresh = Some((stale_request, refresh_owner)); app.agents_overview.refresh_pending = true; let refresh = tokio::spawn(std::future::pending::<()>()); app.agents_overview.refresh_task = Some(refresh.abort_handle()); @@ -251,10 +251,13 @@ async fn reconnect_daemon_command_center_after_socket_replacement_without_a_conv Ok::<_, color_eyre::Report>(methods) }); let mut session = AppServerSession::new( - crate::backend_id::BackendId::new(), + refresh_owner, crate::connect_remote_app_server(endpoint).await?, ThreadParamsMode::Embedded, ); + if overview_initialized { + app.agents_overview.initialized.insert(session.backend_id()); + } assert!( session .thread_loaded_list(ThreadLoadedListParams { @@ -272,13 +275,15 @@ async fn reconnect_daemon_command_center_after_socket_replacement_without_a_conv "Waiting for active turns.".into(), ); } + app.agents_overview.pending_usage = Some((ThreadId::new(), Uuid::new_v4())); app.handle_backend_event(&session, disconnected).await; + assert_eq!(app.agents_overview.pending_usage, None); assert!(app.reconnect.offline); assert!(app.agents_overview.last_messages.is_empty()); assert!(refresh.await.unwrap_err().is_cancelled()); assert_eq!( ( - app.agents_overview.request_id, + app.agents_overview.active_refresh.map(|(id, _)| id), app.agents_overview.refresh_pending ), (None, false) @@ -325,6 +330,7 @@ async fn reconnect_daemon_command_center_after_socket_replacement_without_a_conv app.reconnect.presentation, ) .await?; + let old_event_tx = app.app_event_tx.app_event_tx.clone(); app.finish_reconnect( &mut tui, &mut session, @@ -333,6 +339,9 @@ async fn reconnect_daemon_command_center_after_socket_replacement_without_a_conv CODEX_CLI_VERSION, ) .await?; + assert_eq!(app.app_server_backend_id, Some(session.backend_id())); + assert!(old_event_tx.is_closed()); + assert_eq!(app.agents_overview.pending_usage, None); if interrupted_setup { assert!(app.windows_sandbox.pending_setup.is_none()); assert!(app.windows_sandbox.setup_started_at.is_none()); @@ -361,7 +370,7 @@ async fn reconnect_daemon_command_center_after_socket_replacement_without_a_conv tokio::time::timeout(Duration::from_secs(/*secs*/ 5), async { let mut deferred = Vec::new(); - while app.agents_overview.request_id.is_some() { + while app.agents_overview.active_refresh.is_some() { let event = events.recv().await.expect("inventory refresh event"); if matches!(event, AppEvent::AgentsOverviewThreadsLoaded { .. }) { Box::pin(app.handle_event(&mut tui, &mut session, event)).await?; 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 74bf110ff5..d99a276fa0 100644 --- a/codex-rs/tui/src/app/tests/new_session_tests.rs +++ b/codex-rs/tui/src/app/tests/new_session_tests.rs @@ -2,6 +2,7 @@ use super::*; use crate::app::backend_sessions::BackendSessions; +use crate::app_server_session::AppServerSessionEvent; use codex_app_server_protocol::SkillMetadata; use codex_app_server_protocol::SkillScope; use codex_app_server_protocol::SkillsListEntry; @@ -990,6 +991,327 @@ async fn resumed_backend_handoff_detaches_through_the_source_connection() -> Res Ok(()) } +#[tokio::test] +async fn backend_handoff_restores_a_live_blank_session_without_resuming_it() -> Result<()> { + check_retained_draft_handoff( + /*saved_history*/ false, + HistoryCapabilities::Current, + /*retained_blank*/ true, + ) + .await +} + +#[tokio::test] +async fn backend_handoff_restores_a_live_blank_session_on_an_older_server() -> Result<()> { + let capabilities = HistoryCapabilities::LegacyOnly; + check_retained_draft_handoff( + /*saved_history*/ false, + capabilities, + /*retained_blank*/ true, + ) + .await +} + +#[tokio::test] +async fn backend_handoff_falls_back_when_the_first_turn_page_is_unsupported() -> Result<()> { + check_retained_draft_handoff( + /*saved_history*/ false, + HistoryCapabilities::FirstTurnPageUnsupported, + /*retained_blank*/ true, + ) + .await +} + +#[tokio::test] +async fn backend_handoff_reads_saved_turns_instead_of_reusing_a_stale_blank() -> Result<()> { + check_retained_draft_handoff( + /*saved_history*/ true, + HistoryCapabilities::Current, + /*retained_blank*/ true, + ) + .await +} + +#[tokio::test] +async fn backend_handoff_keeps_resumed_settings_for_populated_task_draft() -> Result<()> { + check_retained_draft_handoff( + /*saved_history*/ true, + HistoryCapabilities::Current, + /*retained_blank*/ false, + ) + .await +} + +#[tokio::test] +async fn same_backend_reopen_recovers_history_for_a_stale_blank() -> Result<()> { + let (mut app, mut events, _ops) = make_test_app_with_channels().await; + let model = core_test_support::responses::start_mock_server().await; + app_test_support::MockResponsesConfig::new(&model.uri()).write(&app.config.codex_home)?; + crate::legacy_core::config::set_project_trust_level( + &app.config.codex_home, + app.config.cwd.as_path(), + codex_protocol::config_types::TrustLevel::Trusted, + ) + .map_err(std::io::Error::other)?; + app.config = app + .rebuild_config_for_cwd(app.config.cwd.to_path_buf()) + .await?; + let mut server = + boxed_future!(crate::start_embedded_app_server_for_picker(&app.config)).await?; + let blank = server.start_thread(&app.config).await?; + let thread_id = blank.session.thread_id; + app.backend_registry + .register_started(server.backend_id(), &blank)?; + app.agents_overview.blank_sessions.insert(thread_id, blank); + + // Returning to the backend through another task leaves this cached blank untouched. + let other = server.start_thread(&app.config).await?; + let other_id = other.session.thread_id; + let mut tui = crate::tui::test_support::make_test_tui()?; + app.enqueue_primary_thread_session(other.session, other.turns) + .await?; + drain_app_events(&mut app, &mut tui, &mut server, &mut events).await?; + assert_eq!(app.current_displayed_thread_id(), Some(other_id)); + + // The helper consumes the turn notifications without delivering them to App. + let prompt = "Sent while the backend was parked"; + side_chat_policy_tests::ask(&mut server, &model, thread_id, prompt).await?; + assert!(app.agents_overview.blank_sessions.contains_key(&thread_id)); + let mut expected_mode = app.chat_widget.effective_collaboration_mode(); + app.chat_widget.set_model("stale-model"); + app.chat_widget.apply_external_edit("saved draft".into()); + let draft = app.chat_widget.capture_thread_input_state().unwrap(); + app.agents_overview.input_states.insert(thread_id, draft); + expected_mode.mode = ModeKind::Plan; + expected_mode.settings.developer_instructions = Some("Updated by the other client".into()); + assert!( + server + .thread_settings_update(ThreadSettingsUpdateParams { + thread_id: thread_id.to_string(), + collaboration_mode: Some(expected_mode.clone()), + ..Default::default() + }) + .await? + ); + let _ = next_thread_settings_updated(&mut server, thread_id).await; + + boxed_future!(app.select_agents_overview_thread(&mut tui, &mut server, thread_id)).await?; + assert_eq!(app.current_displayed_thread_id(), Some(thread_id)); + let history = boxed_future!(active_reconnect::drain_history( + &mut app, + &mut tui, + &mut server, + &mut events + )) + .await?; + assert!(history.contains(prompt), "{history}"); + assert_eq!(app.chat_widget.composer_text_with_pending(), "saved draft"); + assert_eq!( + app.chat_widget.effective_collaboration_mode(), + expected_mode + ); + assert!(!app.agents_overview.blank_sessions.contains_key(&thread_id)); + server.shutdown().await?; + Ok(()) +} + +async fn check_retained_draft_handoff( + saved_history: bool, + history_capabilities: HistoryCapabilities, + retained_blank: bool, +) -> 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 (mut destination, requests, proxy) = start_recording_app_server_with_history( + &app.config, + history_capabilities, + /*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 mut blank = if saved_history { + let timestamp = "2025-01-05T12-00-00"; + let thread_id = create_fake_rollout( + &app.config.codex_home, + timestamp, + "2025-01-05T12:00:00Z", + "Sent while the backend was parked", + Some(&app.config.model_provider_id), + /*git_info*/ None, + ) + .expect("materialize the retained thread"); + let path = app_test_support::rollout_path(&app.config.codex_home, timestamp, &thread_id); + let contents = std::fs::read_to_string(&path)?; + let (metadata, history) = contents.split_once('\n').expect("rollout with history"); + let mut metadata: serde_json::Value = serde_json::from_str(metadata)?; + metadata["payload"]["cwd"] = serde_json::to_value(&app.config.cwd)?; + std::fs::write(path, format!("{metadata}\n{history}"))?; + let thread_id = ThreadId::from_string(&thread_id)?; + destination + .resume_thread( + &app.local_settings, + app.config.clone(), + thread_id, + crate::app_server_session::ResumeModelSettings::PreserveExistingThread, + ) + .await? + } else { + destination.start_thread(&app.config).await? + }; + let thread_id = blank.session.thread_id; + let mut server_mode = blank.session.collaboration_mode.as_deref().cloned(); + let initial_policy = blank.session.approval_policy; + blank.turns.clear(); + let resume_count = recorded_params(&requests, "thread/resume").len(); + let target = crate::resume_picker::SessionTarget { + backend_id: destination_id, + path: None, + thread_id, + cwd: Some(app.config.cwd.to_path_buf()), + history_mode: None, + }; + app.backend_registry + .register_started(destination_id, &blank)?; + if retained_blank { + app.agents_overview.blank_sessions.insert(thread_id, blank); + } + app.chat_widget + .restore_user_message_to_composer("draft for the retained task".into()); + let collaboration_mode = if saved_history { + app.chat_widget.set_model("old-model"); + let mode = server_mode + .as_mut() + .expect("server supplies the current mode"); + mode.mode = ModeKind::Plan; + mode.settings.developer_instructions = Some("Instructions from the parked task".into()); + Some(mode.clone()) + } else { + None + }; + let retained = app.chat_widget.capture_thread_input_state().unwrap(); + app.agents_overview.input_states.insert(thread_id, retained); + app.chat_widget + .restore_user_message_to_composer("draft for the displayed task".into()); + + let parked_policy = (!saved_history && history_capabilities == HistoryCapabilities::Current) + .then_some(if initial_policy == AskForApproval::Never { + AskForApproval::OnRequest + } else { + AskForApproval::Never + }); + if collaboration_mode.is_some() || parked_policy.is_some() { + let settings = ThreadSettingsUpdateParams { + thread_id: thread_id.to_string(), + collaboration_mode, + approval_policy: parked_policy, + ..Default::default() + }; + assert!(destination.thread_settings_update(settings).await?); + let update = next_thread_settings_updated(&mut destination, thread_id).await; + if parked_policy.is_some() { + let event = AppServerEvent::ServerNotification(Box::new( + ServerNotification::ThreadSettingsUpdated(update), + )); + let event = AppServerSessionEvent { + backend_id: destination_id, + event, + }; + app.handle_polled_backend_event(&destination, source.backend_id(), event) + .await; + } + } + let mut backends = BackendSessions::from_initial(source, vec![destination]); + let settings = app.backend_settings(); + backends.settings.insert(destination_id, settings); + if history_capabilities == HistoryCapabilities::LegacyOnly { + app.app_server_target = AppServerTarget::Cloud; + app.app_server_backend_id = Some(backends.displayed_owner()); + assert!(app.begin_reconnect()); + } + + let outgoing = backends.displayed_owner(); + app.agents_overview.initialized.insert(outgoing); + app.agents_overview.show_more_requested = true; + let mut tui = crate::tui::test_support::make_test_tui()?; + let pages_before = recorded_params(&requests, "thread/turns/list").len(); + let reads_before = recorded_params(&requests, "thread/read").len(); + app.resume_backend_session(&mut tui, &mut backends, target) + .await?; + + assert_eq!( + backends.displayed_owner(), + destination_id, + "handoff errors: {:?}; reads: {:?}; pages: {:?}; resumes: {:?}", + std::iter::from_fn(|| events.try_recv().ok()) + .filter_map(|event| match event { + AppEvent::InsertHistoryCell(cell) => + Some(lines_to_single_string(&cell.raw_lines())), + _ => None, + }) + .collect::>(), + recorded_params(&requests, "thread/read"), + recorded_params(&requests, "thread/turns/list"), + recorded_params(&requests, "thread/resume"), + ); + assert!(app.agents_overview.initialized.contains(&outgoing)); + assert!(!app.agents_overview.initialized.contains(&destination_id)); + assert!(!app.agents_overview.show_more_requested); + assert_eq!(app.chat_widget.thread_id(), Some(thread_id)); + if saved_history { + assert_eq!( + app.chat_widget.effective_collaboration_mode(), + server_mode.unwrap() + ); + } + if let Some(parked_policy) = parked_policy { + let policy = &app.chat_widget.config_ref().permissions.approval_policy; + assert_eq!(AskForApproval::from(policy.value()), parked_policy); + } + assert_eq!( + app.chat_widget.composer_text_with_pending(), + "draft for the retained task" + ); + let page_probe_count = recorded_params(&requests, "thread/turns/list")[pages_before..] + .iter() + .filter(|params| params["limit"] == 1) + .count(); + // The saved-history case uses a legacy rollout even on the current server. + assert_eq!( + page_probe_count, + usize::from( + !saved_history + && matches!( + history_capabilities, + HistoryCapabilities::Current | HistoryCapabilities::FirstTurnPageUnsupported + ) + ) + ); + if retained_blank + && (page_probe_count == 0 + || history_capabilities == HistoryCapabilities::FirstTurnPageUnsupported) + { + assert!( + recorded_params(&requests, "thread/read")[reads_before..] + .iter() + .any(|params| params["includeTurns"] == true) + ); + } + assert_eq!( + recorded_params(&requests, "thread/resume").len(), + resume_count + usize::from(saved_history), + ); + assert_eq!( + app.agents_overview.blank_sessions.contains_key(&thread_id), + !saved_history, + ); + backends.shutdown().await?; + 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/session_lifecycle_requests.rs b/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs index c2052a24d3..e437376bb6 100644 --- a/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs +++ b/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs @@ -1228,6 +1228,7 @@ pub(super) enum HistoryCapabilities { Current, LegacyOnly, LegacyOnlyUnsupportedVariant, + FirstTurnPageUnsupported, LegacyDynamicToolsAndHistory, ForkHydrationFails, ReadAfterResumeFails, @@ -1430,6 +1431,12 @@ pub(super) async fn start_recording_app_server_with_realtime_speech( } let request_id = request.id.clone(); let params = request.params.as_ref(); + let legacy_history = matches!( + history_capabilities, + HistoryCapabilities::LegacyOnly + | HistoryCapabilities::LegacyOnlyUnsupportedVariant + | HistoryCapabilities::LegacyDynamicToolsAndHistory + ); let requires_pagination = match request.method.as_str() { "thread/start" => params .and_then(|params| params.get("historyMode")) @@ -1600,12 +1607,9 @@ pub(super) async fn start_recording_app_server_with_realtime_speech( message: "missing field `inputSchema`".to_string(), }, }) - } else if matches!( - history_capabilities, - HistoryCapabilities::LegacyOnly - | HistoryCapabilities::LegacyOnlyUnsupportedVariant - | HistoryCapabilities::LegacyDynamicToolsAndHistory - ) && requires_pagination + } else if (legacy_history && requires_pagination) + || (history_capabilities == HistoryCapabilities::FirstTurnPageUnsupported + && request.method == "thread/turns/list") { let (code, message) = if history_capabilities == HistoryCapabilities::LegacyOnlyUnsupportedVariant @@ -1638,9 +1642,20 @@ pub(super) async fn start_recording_app_server_with_realtime_speech( matches!(inventories, 2 | 4) }; let detach = request.method == "thread/unsubscribe"; - let request = serde_json::from_value::( + let mut request = serde_json::from_value::( serde_json::to_value(request)?, )?; + // Simulate an older store while advertising pagination for the first probe. + if (legacy_history + || history_capabilities + == HistoryCapabilities::FirstTurnPageUnsupported) + && let ClientRequest::ThreadStart { params, .. } = &mut request + { + params.history_mode = Some(ThreadHistoryMode::Legacy); + } + let advertise_paginated = history_capabilities + == HistoryCapabilities::FirstTurnPageUnsupported + && matches!(&request, ClientRequest::ThreadRead { params, .. } if !params.include_turns); if let ClientRequest::ThreadList { params, .. } = &request && let Some((root, started, release)) = blocked_thread_list.take() { @@ -1679,6 +1694,9 @@ pub(super) async fn start_recording_app_server_with_realtime_speech( ClientRequest::ThreadRealtimeListVoices { .. } ); let mut result = embedded.request(request).await?; + if advertise_paginated && let Ok(value) = &mut result { + value["thread"]["historyMode"] = serde_json::json!("paginated"); + } if unknown_voice && let Ok(value) = &mut result { value["config"]["realtime"]["voice"] = serde_json::json!("future_voice"); @@ -2300,7 +2318,11 @@ async fn archive_current_thread_returns_to_agents() -> Result<()> { break; } } - assert!(app.agents_overview.initialized); + assert!( + app.agents_overview + .initialized + .contains(&server.backend_id()) + ); assert_eq!( ( app.active_thread_id, @@ -4348,9 +4370,13 @@ async fn agents_overview_seeds_loaded_threads_when_recent_listing_is_unavailable .collect::>(), vec![started.session.thread_id] ); - assert!(app.agents_overview.initialized); + assert!( + app.agents_overview + .initialized + .contains(&app_server.backend_id()) + ); assert_eq!( - app.agents_overview.discovery.has_more(), + app.agents_overview.has_more(app_server.backend_id()), capabilities == HistoryCapabilities::ThreadListFails ); if attempt == 0 { @@ -4366,7 +4392,7 @@ async fn agents_overview_seeds_loaded_threads_when_recent_listing_is_unavailable )), ) .await; - assert!(app.agents_overview.request_id.is_none()); + assert!(app.agents_overview.active_refresh.is_none()); } } let list_requests = recorded_params(&requests, "thread/list"); diff --git a/codex-rs/tui/src/app/thread_routing.rs b/codex-rs/tui/src/app/thread_routing.rs index f333f98752..0de17efef9 100644 --- a/codex-rs/tui/src/app/thread_routing.rs +++ b/codex-rs/tui/src/app/thread_routing.rs @@ -17,6 +17,17 @@ use codex_app_server_protocol::WarningNotification; const REALTIME_STOP_TIMEOUT: Duration = Duration::from_secs(/*secs*/ 1); impl App { + pub(super) fn displayed_conversation_backend_id(&self) -> Option { + self.current_displayed_thread_id() + .and_then(|thread_id| { + self.thread_event_channels + .get(&thread_id) + .and_then(|channel| channel.backend_id) + .or_else(|| self.backend_registry.owner(thread_id)) + }) + .or(self.app_server_backend_id) + } + pub(super) async fn stop_realtime_conversation(&mut self, app_server: &mut AppServerSession) { let thread_id = self .background_voice diff --git a/codex-rs/tui/src/app/thread_settings.rs b/codex-rs/tui/src/app/thread_settings.rs index c7ae7e638e..24172f2972 100644 --- a/codex-rs/tui/src/app/thread_settings.rs +++ b/codex-rs/tui/src/app/thread_settings.rs @@ -11,6 +11,7 @@ use codex_app_server_protocol::ApprovalsReviewer as AppServerApprovalsReviewer; use codex_app_server_protocol::AskForApproval as AppServerAskForApproval; use codex_app_server_protocol::ThreadSettings; use codex_app_server_protocol::ThreadSettingsUpdateParams; +use codex_app_server_protocol::ThreadSettingsUpdatedNotification; use codex_config::types::ApprovalsReviewer; use codex_protocol::ThreadId; use codex_protocol::config_types::ModeKind; @@ -193,6 +194,24 @@ impl App { self.send_thread_settings_update(app_server, params).await; } + // Retained blank sessions stay subscribed after their event channels are cleared. + pub(super) async fn update_retained_blank_settings( + &mut self, + thread_id: ThreadId, + notification: &ThreadSettingsUpdatedNotification, + ) { + if !self.agents_overview.blank_sessions.contains_key(&thread_id) + || self.thread_event_channels.contains_key(&thread_id) + { + return; + } + self.apply_thread_settings_to_cached_session(thread_id, ¬ification.thread_settings) + .await; + if let Some(input) = self.agents_overview.input_states.get_mut(&thread_id) { + input.pending_thread_settings = Some(notification.clone()); + } + } + pub(super) async fn apply_thread_settings_to_cached_session( &mut self, thread_id: ThreadId, diff --git a/codex-rs/tui/src/app_event.rs b/codex-rs/tui/src/app_event.rs index b1dd1bae17..2c01a4fab9 100644 --- a/codex-rs/tui/src/app_event.rs +++ b/codex-rs/tui/src/app_event.rs @@ -387,6 +387,7 @@ pub(crate) enum AppEvent { }, /// Create an empty thread from the command center. NewAgentsOverviewSession { + backend_id: Option, cwd: Option, }, /// Update the agents overview after a background thread listing finishes. @@ -400,6 +401,7 @@ pub(crate) enum AppEvent { }, /// Create an empty session in a worktree from the selected project's default branch. NewAgentsOverviewWorktree { + backend_id: Option, cwd: Option, }, AgentsOverviewWorktreeCreated(Result), diff --git a/codex-rs/tui/src/backend_id.rs b/codex-rs/tui/src/backend_id.rs index 8b95ba7ae7..38c442179e 100644 --- a/codex-rs/tui/src/backend_id.rs +++ b/codex-rs/tui/src/backend_id.rs @@ -2,7 +2,7 @@ //! Reconnecting and auxiliary picker connections reuse the launch owner; a different //! backend gets a fresh identity. These process-local IDs never appear on the wire. -#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] +#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)] pub(crate) struct BackendId(uuid::Uuid); impl BackendId { diff --git a/codex-rs/tui/src/chatwidget/input_restore.rs b/codex-rs/tui/src/chatwidget/input_restore.rs index dd6d4bd659..47359d7632 100644 --- a/codex-rs/tui/src/chatwidget/input_restore.rs +++ b/codex-rs/tui/src/chatwidget/input_restore.rs @@ -736,6 +736,24 @@ impl ChatWidget { }) } + /// Keep the current session's collaboration settings when restoring a saved draft. + pub(crate) fn update_input_collaboration_mode( + &self, + input: &mut ThreadInputState, + has_resumed_mode: bool, + ) { + input.current_collaboration_mode = self.current_collaboration_mode.clone(); + if has_resumed_mode { + input + .active_collaboration_mask + .clone_from(&self.active_collaboration_mask); + } else if let Some(mask) = input.active_collaboration_mask.as_mut() { + // Keep the local selection for older servers. Replay reapplies a supplied mode. + mask.model = Some(self.current_model().to_string()); + mask.reasoning_effort = Some(self.effective_reasoning_effort()); + } + } + pub(crate) fn restore_thread_input_state( &mut self, input_state: Option, diff --git a/codex-rs/tui/src/chatwidget/reconnect.rs b/codex-rs/tui/src/chatwidget/reconnect.rs index d715c5c5cd..dc2855b78b 100644 --- a/codex-rs/tui/src/chatwidget/reconnect.rs +++ b/codex-rs/tui/src/chatwidget/reconnect.rs @@ -88,12 +88,7 @@ impl ChatWidget { .push_front(UserMessageHistoryRecord::UserMessageText); input.recovered_queue |= !input.reconnect_pending; } - input.current_collaboration_mode = self.current_collaboration_mode.clone(); - // Keep the local selection for older servers. Replay reapplies a supplied mode. - if let Some(mask) = input.active_collaboration_mask.as_mut() { - mask.model = Some(self.current_model().to_string()); - mask.reasoning_effort = Some(self.effective_reasoning_effort()); - } + self.update_input_collaboration_mode(&mut input, /*has_resumed_mode*/ false); self.restore_thread_input_state( Some(input), ThreadInputStateRestoreMode {