openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-12 21:20:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

Preserve Command Center task ownership across backend switches (#60929)

Browse Source 
## Why

Command Center retains tasks across backend switches. Opening a retained task must use its owning backend, and restoring its draft must account for history or settings that changed while another backend was displayed.

## What changed

- Track each thread's first observed backend, reject conflicting ownership, and switch to the owning backend when opening a retained task. Require users to open tasks on other backends before running task actions.
- Keep discovery cursors, initialization, and pins per backend. Scope refreshes and usage requests to their owner, and handle background archive and delete notifications without consuming the displayed backend's pagination budget.
- Verify cached blank sessions are still idle and have no turns before reusing them, with a history-read fallback for older servers. Otherwise, resume saved history and preserve current server collaboration settings when restoring drafts.
- Use only interactive task-list pages for cloud discovery, which does not support the additional source filters.

## Testing

Add regression coverage for ownership conflicts, action routing, pins across backends, background removals, usage isolation, cloud discovery, and draft restoration with blank or populated history, including older-server fallbacks.

GitOrigin-RevId: 0d0913ec38dc16a839823008a4a84aa46ad8e6c6
This commit is contained in:
Benjamin Carlsson
2026-11-12 00:02:01 +00:00
committed by copyberry
parent c95b72ed6f
commit 27294165dc
37 changed files with 1451 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/tui/src/app.rs
+3
View File
@@ -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<WorkspaceCommandRunner>,
/// 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<EnvironmentManager>,
app_server_target: AppServerTarget,
app_server_backend_id: Option<crate::backend_id::BackendId>,
reconnect: reconnect::ReconnectState,
pending_right_click_paste: Option<right_click_paste::PendingPaste>,
right_click_paste_environment: right_click_paste::PasteEnvironment,
codex-rs/tui/src/app/agent_center_tests.rs
+19 -5
View File
@@ -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",
codex-rs/tui/src/app/agents_overview.rs
+140 -37
View File
@@ -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<ThreadId, super::agents_overview_details::AgentsOverviewActivity>,
pub(super) initialized: bool,
pub(super) discovery: super::agents_overview_discovery::AgentsOverviewDiscovery,
pub(super) initialized: HashSet<BackendId>,
pub(super) discovery:
HashMap<BackendId, super::agents_overview_discovery::AgentsOverviewDiscovery>,
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<Uuid>,
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<ThreadId>,
pub(super) active_refresh_thread_ids: HashSet<ThreadId>,
pub(super) pinned_thread_ids: Option<Vec<ThreadId>>,
pub(super) pending_pin_change: Option<Uuid>,
pub(super) pins_by_backend: std::collections::BTreeMap<BackendId, Option<Vec<ThreadId>>>,
pub(super) pending_pin_change: Option<(Uuid, BackendId)>,
pub(super) refresh_task: Option<tokio::task::AbortHandle>,
pub(super) refresh_notifications: HashMap<ThreadId, Vec<ServerNotification>>,
pub(super) rendered_full_screen: bool,
@@ -83,6 +86,34 @@ pub(super) struct AgentsOverviewState {
pub(super) dispatched_requests: HashMap<ThreadId, Vec<BufferedServerRequest>>,
}
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<AgentsOverviewThreadRefresh, String>,
) {
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<AppRunControl> {
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())
codex-rs/tui/src/app/agents_overview_actions.rs
+12 -6
View File
@@ -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
codex-rs/tui/src/app/agents_overview_actions_tests.rs
+8 -6
View File
@@ -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);
codex-rs/tui/src/app/agents_overview_details.rs
+5
View File
@@ -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(
codex-rs/tui/src/app/agents_overview_discovery.rs
+6
View File
@@ -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)
codex-rs/tui/src/app/agents_overview_discovery_tests.rs
+81 -6
View File
@@ -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));
codex-rs/tui/src/app/agents_overview_new.rs
+2
View File
@@ -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());
codex-rs/tui/src/app/agents_overview_tests.rs
+244 -27
View File
@@ -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<i64> =
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::<Vec<_>>()
.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::<HashSet<_>>(),
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
})
));
}
codex-rs/tui/src/app/agents_overview_threads.rs
+27 -14
View File
@@ -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();
codex-rs/tui/src/app/agents_overview_usage.rs
+25 -5
View File
@@ -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(_) => {}
codex-rs/tui/src/app/agents_overview_usage_tests.rs
+40 -3
View File
@@ -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(
codex-rs/tui/src/app/agents_overview_view.rs
+6
View File
@@ -108,6 +108,7 @@ fn is_subagent(thread: &Thread) -> bool {
#[derive(Clone)]
pub(super) struct AgentsOverviewRow {
pub(super) backend_id: Option<crate::backend_id::BackendId>,
pub(super) search_id: Option<uuid::Uuid>,
pub(super) details: AgentsOverviewDetails,
pub(super) thread: Thread,
@@ -229,6 +230,7 @@ pub(super) struct AgentsOverviewView {
project_groups: Vec<AgentsOverviewProjectGroup>,
directory_group_keys: Vec<PathBuf>,
pub(super) pinned_thread_ranks: Option<HashMap<ThreadId, usize>>,
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()),
});
}
codex-rs/tui/src/app/app_server_events.rs
+15 -9
View File
@@ -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(&notification);
// Retained blank sessions stay subscribed after their event channels are cleared.
if let ServerNotification::ThreadSettingsUpdated(settings) = &notification
&& 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!(
&notification,
@@ -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;
codex-rs/tui/src/app/backend_events.rs
+42 -5
View File
@@ -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;
}
codex-rs/tui/src/app/backend_registry.rs
+38
View File
@@ -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<ThreadId, BackendId>,
}
impl BackendRegistry {
pub(super) fn owner(&self, thread: ThreadId) -> Option<BackendId> {
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
}
}
codex-rs/tui/src/app/backend_selection.rs
+22 -2
View File
@@ -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;
codex-rs/tui/src/app/event_dispatch.rs
+23 -2
View File
@@ -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) => {
codex-rs/tui/src/app/owned_transcript_tests.rs
+1 -1
View File
@@ -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",
codex-rs/tui/src/app/reconnect.rs
+16 -4
View File
@@ -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
codex-rs/tui/src/app/resume_preparation.rs
+88 -9
View File
@@ -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<codex_app_server_protocol::ModelProviderError>,
@@ -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<AppServerStartedThread> {
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::<TypedRequestError>(),
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<PreparedResume, AppRunControl> {
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,
codex-rs/tui/src/app/session_lifecycle.rs
+10
View File
@@ -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())
codex-rs/tui/src/app/startup.rs
+8
View File
@@ -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;
codex-rs/tui/src/app/test_support.rs
+2
View File
@@ -51,6 +51,7 @@ pub(crate) async fn make_test_app() -> Box<App> {
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<App> {
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,
codex-rs/tui/src/app/tests.rs
+4
View File
@@ -8983,6 +8983,7 @@ async fn make_test_app() -> Box<App> {
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<App> {
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(),
codex-rs/tui/src/app/tests/backend_event_tests.rs
+121
View File
@@ -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);
codex-rs/tui/src/app/tests/backend_ownership_tests.rs
+18
View File
@@ -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,
codex-rs/tui/src/app/tests/navigation_reconnect_tests.rs
+14 -5
View File
@@ -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?;
codex-rs/tui/src/app/tests/new_session_tests.rs
+322
View File
@@ -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::<Vec<_>>(),
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 [
codex-rs/tui/src/app/tests/session_lifecycle_requests.rs
+37 -11
View File
@@ -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::<ClientRequest>(
let mut request = serde_json::from_value::<ClientRequest>(
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<_>>(),
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");
codex-rs/tui/src/app/thread_routing.rs
+11
View File
@@ -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<crate::backend_id::BackendId> {
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
codex-rs/tui/src/app/thread_settings.rs
+19
View File
@@ -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, &notification.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,
codex-rs/tui/src/app_event.rs
+2
View File
@@ -387,6 +387,7 @@ pub(crate) enum AppEvent {
},
/// Create an empty thread from the command center.
NewAgentsOverviewSession {
backend_id: Option<crate::backend_id::BackendId>,
cwd: Option<AbsolutePathBuf>,
},
/// 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<crate::backend_id::BackendId>,
cwd: Option<AbsolutePathBuf>,
},
AgentsOverviewWorktreeCreated(Result<crate::app::PendingWorktree, String>),
codex-rs/tui/src/backend_id.rs
+1 -1
View File
@@ -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 {
codex-rs/tui/src/chatwidget/input_restore.rs
+18
View File
@@ -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<ThreadInputState>,
codex-rs/tui/src/chatwidget/reconnect.rs
+1 -6
View File
@@ -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 {