Refresh the subagent picker in the background (#35693)

## Why

Opening the subagent picker could wait on thread metadata and live event-store locks, delaying terminal input. Its cached entries could also omit descendants that were not observed in the current TUI session.

## What changed

- Render the picker immediately from cached navigation state and refresh root descendants asynchronously with `thread/list`.
- Merge discovered descendants into an open picker while preserving its selection and live status, and coalesce concurrent refresh requests.
- Ignore refresh responses from a previous session and avoid blocking on busy event stores.

## Testing

- Extend session lifecycle coverage for nonblocking picker opens, refresh coalescing, descendant discovery, selection preservation, and status updates.
- Add coverage for rejecting a refresh response after navigation state is cleared.

GitOrigin-RevId: 7eae7f696f8d2d323a47b3fd2800ea1c1cdbf2a0
This commit is contained in:
Eric Traut
2026-07-28 00:38:45 +00:00
committed by copyberry
parent 63682c4e1a
commit f029bb795c
10 changed files with 415 additions and 37 deletions
+1
View File
@@ -201,6 +201,7 @@ use toml::Value as TomlValue;
use uuid::Uuid;
mod agent_message_consolidation;
mod agent_navigation;
mod agent_picker;
mod agent_status_feed;
mod app_server_event_targets;
mod app_server_events;
+39
View File
@@ -28,6 +28,7 @@ use codex_protocol::ThreadId;
use ratatui::text::Span;
use std::collections::HashMap;
use std::collections::HashSet;
use uuid::Uuid;
/// Small state container for multi-agent picker ordering and labeling.
///
@@ -48,6 +49,8 @@ pub(crate) struct AgentNavigationState {
stopped_threads: HashSet<ThreadId>,
/// Spawned child threads whose instructions are owned by their parent agent.
parent_owned_threads: HashSet<ThreadId>,
/// Coalesces root refreshes while rejecting replies from a previous session.
picker_refresh: Option<(ThreadId, Uuid)>,
}
/// Direction of keyboard traversal through the stable picker order.
@@ -60,6 +63,23 @@ pub(crate) enum AgentNavigationDirection {
}
impl AgentNavigationState {
pub(crate) fn begin_picker_refresh(&mut self, thread_id: ThreadId) -> Option<Uuid> {
if self.picker_refresh.is_some() {
return None;
}
let request_id = Uuid::new_v4();
self.picker_refresh = Some((thread_id, request_id));
Some(request_id)
}
pub(crate) fn finish_picker_refresh(&mut self, thread_id: ThreadId, request_id: Uuid) -> bool {
if self.picker_refresh != Some((thread_id, request_id)) {
return false;
}
self.picker_refresh = None;
true
}
/// Returns the cached picker entry for a specific thread id.
///
/// Callers use this when they already know which thread they care about and need the last
@@ -203,6 +223,7 @@ impl AgentNavigationState {
self.order.clear();
self.stopped_threads.clear();
self.parent_owned_threads.clear();
self.picker_refresh = None;
}
/// Removes a tracked thread entirely from picker metadata and traversal order.
@@ -433,6 +454,24 @@ mod tests {
assert!(!state.is_parent_owned(second_agent_id));
}
#[test]
fn picker_refresh_rejects_responses_from_before_clear() {
let mut state = AgentNavigationState::default();
let thread_id = ThreadId::new();
let stale_request = state
.begin_picker_refresh(thread_id)
.expect("first picker refresh");
assert_eq!(state.begin_picker_refresh(thread_id), None);
state.clear();
let current_request = state
.begin_picker_refresh(thread_id)
.expect("refresh after session reset");
assert!(!state.finish_picker_refresh(thread_id, stale_request));
assert!(state.finish_picker_refresh(thread_id, current_request));
}
#[test]
fn adjacent_thread_id_wraps_in_spawn_order() {
let (state, main_thread_id, first_agent_id, second_agent_id) = populated_state();
+146
View File
@@ -0,0 +1,146 @@
//! Root-scoped background refresh for the agent picker.
use super::*;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::SortDirection;
use codex_app_server_protocol::Thread;
use codex_app_server_protocol::ThreadListParams;
use codex_app_server_protocol::ThreadListResponse;
use codex_app_server_protocol::ThreadSourceKind;
use codex_app_server_protocol::ThreadStatus;
use std::collections::HashSet;
pub(super) const AGENT_PICKER_VIEW_ID: &str = "agent-picker";
const AGENT_PICKER_PAGE_SIZE: u32 = 100;
const AGENT_PICKER_MAX_THREADS: usize = 1_000;
impl App {
pub(super) fn refresh_agent_picker_threads(
&mut self,
app_server: &AppServerSession,
root: ThreadId,
) {
let Some(request_id) = self.agent_navigation.begin_picker_refresh(root) else {
return;
};
let request_handle = app_server.request_handle();
let app_event_tx = self.app_event_tx.clone();
tokio::spawn(async move {
let result = async {
let mut threads = Vec::new();
let mut cursor = None;
let mut seen_cursors = HashSet::new();
while threads.len() < AGENT_PICKER_MAX_THREADS
&& seen_cursors.insert(cursor.clone())
{
let page = match request_handle
.request_typed::<ThreadListResponse>(ClientRequest::ThreadList {
request_id: RequestId::String(Uuid::new_v4().to_string()),
params: ThreadListParams {
cursor,
limit: Some(AGENT_PICKER_PAGE_SIZE),
sort_key: None,
sort_direction: Some(SortDirection::Desc),
model_providers: Some(vec![]),
source_kinds: Some(vec![ThreadSourceKind::SubAgentThreadSpawn]),
archived: None,
is_pinned: None,
cwd: None,
use_state_db_only: true,
search_term: None,
parent_thread_id: None,
ancestor_thread_id: Some(root.to_string()),
},
})
.await
{
Ok(page) => page,
Err(err) if threads.is_empty() => return Err(err.to_string()),
Err(err) => {
tracing::warn!(%err, "failed to refresh remaining agent picker descendants");
break;
}
};
threads.extend(
page.data
.into_iter()
.take(AGENT_PICKER_MAX_THREADS - threads.len()),
);
let Some(next_cursor) = page.next_cursor else {
break;
};
cursor = Some(next_cursor);
}
threads.reverse();
Ok(threads)
}
.await;
app_event_tx.send(AppEvent::AgentPickerThreadsLoaded {
primary_thread_id: root,
request_id,
result,
});
});
}
pub(super) fn apply_agent_picker_thread_refresh(
&mut self,
root: ThreadId,
request_id: Uuid,
result: Result<Vec<Thread>, String>,
) {
if !self
.agent_navigation
.finish_picker_refresh(root, request_id)
|| self.primary_thread_id != Some(root)
{
return;
}
let threads = match result {
Ok(threads) => threads,
Err(err) => {
tracing::warn!(%err, "failed to refresh agent picker descendants");
return;
}
};
let selected = self
.chat_widget
.selected_index_for_present_view(AGENT_PICKER_VIEW_ID);
for thread in threads {
let Ok(thread_id) = ThreadId::from_string(&thread.id) else {
continue;
};
let live = self
.thread_event_channels
.get(&thread_id)
.is_some_and(|channel| channel.attachment() == ThreadEventAttachment::Live);
let previous = self.agent_navigation.get(&thread_id);
let is_running = matches!(thread.status, ThreadStatus::Active { .. });
let update_liveness = previous.is_none() || !is_running;
let is_closed = !live && matches!(thread.status, ThreadStatus::NotLoaded);
if !is_closed && previous.is_some_and(|entry| entry.is_closed) {
continue;
}
let agent_path = crate::app_server_session::source_agent_path(&thread.source);
let agent_nickname = thread
.agent_nickname
.or_else(|| previous.and_then(|entry| entry.agent_nickname.clone()));
let agent_role = thread
.agent_role
.or_else(|| previous.and_then(|entry| entry.agent_role.clone()));
if thread.can_accept_direct_input == Some(false) {
self.agent_navigation.mark_parent_owned(thread_id);
}
self.upsert_agent_picker_thread(thread_id, agent_nickname, agent_role, is_closed);
self.agent_navigation.set_agent_path(thread_id, agent_path);
if !live && update_liveness {
self.agent_navigation.set_running(thread_id, is_running);
}
}
let params = self.agent_picker_selection_view_params(selected);
self.chat_widget
.replace_selection_view_if_present(AGENT_PICKER_VIEW_ID, params);
}
}
+7
View File
@@ -2004,6 +2004,13 @@ impl App {
AppEvent::OpenAgentPicker => {
self.open_agent_picker(app_server).await;
}
AppEvent::AgentPickerThreadsLoaded {
primary_thread_id,
request_id,
result,
} => {
self.apply_agent_picker_thread_refresh(primary_thread_id, request_id, result);
}
AppEvent::SelectAgentThread(thread_id) => {
self.select_agent_thread_and_discard_side(tui, app_server, thread_id)
.await?;
+63 -24
View File
@@ -4,6 +4,7 @@
//! resuming/forking saved sessions, replacing ChatWidget instances, and maintaining the agent picker
//! cache used for multi-agent navigation.
use super::agent_picker::AGENT_PICKER_VIEW_ID;
use super::*;
use crate::app_server_session::source_agent_path;
use crate::app_server_session::thread_blocks_direct_input;
@@ -26,7 +27,11 @@ pub(super) struct LoadedSubagentBackfill {
impl App {
pub(super) async fn open_agent_picker(&mut self, app_server: &mut AppServerSession) {
let backfill = self.backfill_loaded_subagent_threads(app_server).await;
let backfill = if self.primary_thread_id.is_none() {
self.backfill_loaded_subagent_threads(app_server).await
} else {
LoadedSubagentBackfill::default()
};
// V2 subagents are identified by canonical paths observed from activity events or loaded
// thread metadata. A buffered active turn is positive liveness evidence; a completed
// snapshot is terminal evidence. An empty store does not clear a successful spawn hint.
@@ -40,22 +45,23 @@ impl App {
if let Some(channel) = self.thread_event_channels.get(&thread_id)
&& channel.attachment() == ThreadEventAttachment::Live
{
let (has_active_turn, has_terminal_snapshot) = {
let store = channel.store.lock().await;
(
store.active_turn_id().is_some(),
store
.turns
.last()
.is_some_and(|turn| !matches!(turn.status, TurnStatus::InProgress)),
)
let Ok(store) = channel.store.try_lock() else {
continue;
};
let has_active_turn = store.active_turn_id().is_some();
let has_terminal_snapshot = store
.turns
.last()
.is_some_and(|turn| !matches!(turn.status, TurnStatus::InProgress));
drop(store);
if has_active_turn {
self.agent_navigation.mark_running(thread_id);
} else if has_terminal_snapshot {
self.agent_navigation.mark_stopped(thread_id);
}
} else if !backfill.refreshed_thread_ids.contains(&thread_id) {
} else if self.primary_thread_id.is_none()
&& !backfill.refreshed_thread_ids.contains(&thread_id)
{
self.refresh_agent_picker_thread_liveness(app_server, thread_id)
.await;
}
@@ -76,10 +82,16 @@ impl App {
let mut entries = Vec::new();
for (thread_id, agent_path) in running_threads {
let preview = if let Some(channel) = self.thread_event_channels.get(&thread_id) {
let store = channel.store.lock().await;
super::agent_status_feed::AgentStatusThreadPreview::from_store(
agent_path, &store,
)
match channel.store.try_lock() {
Ok(store) => {
super::agent_status_feed::AgentStatusThreadPreview::from_store(
agent_path, &store,
)
}
Err(_) => {
super::agent_status_feed::AgentStatusThreadPreview::empty(agent_path)
}
}
} else {
super::agent_status_feed::AgentStatusThreadPreview::empty(agent_path)
};
@@ -105,11 +117,14 @@ impl App {
{
continue;
}
if !self
.refresh_agent_picker_thread_liveness(app_server, thread_id)
.await
{
continue;
if self.primary_thread_id.is_none() {
self.refresh_agent_picker_thread_liveness(app_server, thread_id)
.await;
} else if self.agent_navigation.get(&thread_id).is_none() {
self.upsert_agent_picker_thread(
thread_id, /*agent_nickname*/ None, /*agent_role*/ None,
/*is_closed*/ false,
);
}
}
@@ -117,6 +132,9 @@ impl App {
.agent_navigation
.has_non_primary_thread(self.primary_thread_id);
if !self.config.features.enabled(Feature::Collab) && !has_non_primary_agent_thread {
if let Some(primary_thread_id) = self.primary_thread_id {
self.refresh_agent_picker_threads(app_server, primary_thread_id);
}
self.chat_widget.open_multi_agent_enable_prompt();
return;
}
@@ -127,14 +145,34 @@ impl App {
return;
}
let mut initial_selected_idx = None;
let selected = self
.chat_widget
.selected_index_for_present_view(AGENT_PICKER_VIEW_ID);
let params = self.agent_picker_selection_view_params(selected);
if !self
.chat_widget
.replace_selection_view_if_present(AGENT_PICKER_VIEW_ID, params)
{
let params = self.agent_picker_selection_view_params(selected);
self.chat_widget.show_selection_view(params);
}
if let Some(primary_thread_id) = self.primary_thread_id {
self.refresh_agent_picker_threads(app_server, primary_thread_id);
}
}
pub(super) fn agent_picker_selection_view_params(
&self,
selected: Option<usize>,
) -> SelectionViewParams {
let mut initial_selected_idx = selected;
let items: Vec<SelectionItem> = self
.agent_navigation
.ordered_threads()
.into_iter()
.enumerate()
.map(|(idx, (thread_id, entry))| {
if self.active_thread_id == Some(thread_id) {
if initial_selected_idx.is_none() && self.active_thread_id == Some(thread_id) {
initial_selected_idx = Some(idx);
}
let id = thread_id;
@@ -168,14 +206,15 @@ impl App {
})
.collect();
self.chat_widget.show_selection_view(SelectionViewParams {
SelectionViewParams {
view_id: Some(AGENT_PICKER_VIEW_ID),
title: Some("Subagents".to_string()),
subtitle: Some(AgentNavigationState::picker_subtitle()),
footer_hint: Some(standard_popup_hint_line()),
items,
initial_selected_idx,
..Default::default()
});
}
}
pub(super) fn is_terminal_thread_read_error(err: &color_eyre::Report) -> bool {
@@ -7,6 +7,8 @@ use codex_app_server_protocol::ClientRequest;
use codex_app_server_protocol::JSONRPCError;
use codex_app_server_protocol::JSONRPCMessage;
use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::SortDirection;
use codex_app_server_protocol::ThreadStatus;
use codex_protocol::AgentPath;
use codex_state::SqliteConfig;
use futures::SinkExt;
@@ -14,6 +16,7 @@ use futures::StreamExt;
use pretty_assertions::assert_eq;
use std::sync::Mutex;
use tokio::net::TcpListener;
use tokio::sync::oneshot;
use tokio::task::JoinHandle;
use tokio_tungstenite::accept_async;
use tokio_tungstenite::tungstenite::Message;
@@ -36,6 +39,7 @@ fn take_backfill_counts(requests: &Arc<Mutex<Vec<String>>>) -> (usize, usize) {
/// Starts an embedded app server behind a loopback WebSocket proxy that records JSON-RPC methods.
async fn start_recording_app_server(
config: &Config,
mut blocked_thread_list: Option<(ThreadId, oneshot::Sender<()>, oneshot::Receiver<()>)>,
) -> Result<(
AppServerSession,
Arc<Mutex<Vec<String>>>,
@@ -93,6 +97,14 @@ async fn start_recording_app_server(
let request_id = request.id.clone();
let request =
serde_json::from_value::<ClientRequest>(serde_json::to_value(request)?)?;
if let ClientRequest::ThreadList { params, .. } = &request
&& let Some((root, started, release)) = blocked_thread_list.take()
{
assert_eq!(params.ancestor_thread_id, Some(root.to_string()));
assert_eq!(params.sort_direction, Some(SortDirection::Desc));
let _ = started.send(());
let _ = release.await;
}
let response = match embedded.request(request).await? {
Ok(result) => JSONRPCMessage::Response(JSONRPCResponse {
id: request_id,
@@ -155,7 +167,7 @@ fn fresh_session_applies_requested_name() -> Result<()> {
app.config.codex_home = codex_home.path().to_path_buf().abs();
app.config.sqlite = SqliteConfig::new_for_testing(codex_home.path().abs());
let (mut app_server, requests, proxy) =
start_recording_app_server(&app.config).await?;
start_recording_app_server(&app.config, /*blocked_thread_list*/ None).await?;
let mut tui = crate::tui::test_support::make_test_tui()?;
app.start_fresh_session_with_summary_hint(
@@ -206,7 +218,7 @@ fn session_lifecycle_avoids_redundant_subagent_metadata_reads() -> Result<()> {
.enable_all()
.build()?;
runtime.block_on(async {
let mut app = make_test_app().await;
let (mut app, mut app_event_rx, _op_rx) = make_test_app_with_channels().await;
let codex_home = tempdir()?;
app.config.codex_home = codex_home.path().to_path_buf().abs();
app.config.sqlite =
@@ -250,8 +262,13 @@ fn session_lifecycle_avoids_redundant_subagent_metadata_reads() -> Result<()> {
root_timestamp,
&root_thread_id.to_string(),
);
let (mut app_server, requests, proxy) =
start_recording_app_server(&app.config).await?;
let (started_tx, started_rx) = oneshot::channel();
let (release_tx, release_rx) = oneshot::channel();
let (mut app_server, requests, proxy) = start_recording_app_server(
&app.config,
Some((root_thread_id, started_tx, release_rx)),
)
.await?;
let root = app_server
.resume_thread(
app.config.clone(),
@@ -336,11 +353,107 @@ fn session_lifecycle_avoids_redundant_subagent_metadata_reads() -> Result<()> {
})
);
Box::pin(app.open_agent_picker(&mut app_server)).await;
let child_store = Arc::clone(
&app.thread_event_channels
.entry(child_thread_id)
.or_insert_with(|| ThreadEventChannel::new(/*capacity*/ 1))
.store,
);
let child_store_guard = child_store.lock().await;
futures::FutureExt::now_or_never(app.open_agent_picker(&mut app_server))
.expect("opening the agent picker waited for the app server");
drop(child_store_guard);
insta::assert_snapshot!(
render_bottom_popup(&app.chat_widget, /*width*/ 80)
.replace(&root_thread_id.to_string(), "[root]")
.replace(&child_thread_id.to_string(), "[child]"),
@r###"
Subagents
Select an agent to watch. ⌥ + ← previous, ⌥ + → next.
// The picker refreshes the primary thread once. Discovered children were already
// refreshed by the picker's initial backfill and must not be read a second time.
assert_eq!(take_backfill_counts(&requests), (1, expected_reads + 1));
› 1. • Main [default] (current) [root]
2. • /root/worker [child]
Press enter to confirm or esc to go back
"###
);
assert_eq!(take_backfill_counts(&requests), (0, 0));
tokio::time::timeout(Duration::from_secs(5), started_rx).await??;
app.chat_widget
.handle_key_event(KeyEvent::new(KeyCode::Esc, KeyModifiers::NONE));
futures::FutureExt::now_or_never(app.open_agent_picker(&mut app_server))
.expect("reopening the agent picker waited for the app server");
assert_eq!(
requests
.lock()
.expect("request recorder lock")
.iter()
.filter(|method| *method == "thread/list")
.count(),
1
);
app.chat_widget
.handle_key_event(KeyEvent::new(KeyCode::Down, KeyModifiers::NONE));
app.chat_widget.handle_server_request(
exec_approval_request(
root_thread_id,
"turn",
"item",
/*approval_id*/ None,
),
/*replay_kind*/ None,
);
app.agent_navigation.mark_stopped(child_thread_id);
release_tx.send(()).expect("release blocked thread list");
let discovered_thread_id = ThreadId::new();
let mut completion = tokio::time::timeout(Duration::from_secs(5), async {
loop {
let event = app_event_rx.recv().await.expect("app event channel");
if matches!(event, AppEvent::AgentPickerThreadsLoaded { .. }) {
break event;
}
}
})
.await?;
if let AppEvent::AgentPickerThreadsLoaded {
result: Ok(threads),
..
} = &mut completion
{
let child = threads
.iter_mut()
.find(|thread| thread.id == child_thread_id.to_string())
.expect("root-scoped response includes the cached child");
let mut discovered = child.clone();
discovered.id = discovered_thread_id.to_string();
discovered.can_accept_direct_input = None;
child.status = ThreadStatus::Active {
active_flags: Vec::new(),
};
threads.push(discovered);
}
app.handle_event(&mut tui, &mut app_server, completion)
.await?;
assert_eq!(
app.agent_navigation
.ordered_threads()
.last()
.map(|(thread_id, _)| *thread_id),
Some(discovered_thread_id)
);
assert!(!app.agent_navigation.is_parent_owned(discovered_thread_id));
assert_eq!(
app.chat_widget.selected_index_for_present_view(
super::super::agent_picker::AGENT_PICKER_VIEW_ID
),
Some(1)
);
assert!(render_bottom_popup(&app.chat_widget, /*width*/ 80).contains("echo hello"));
assert!(
app.agent_navigation
.get(&child_thread_id)
.is_some_and(|entry| !entry.is_running)
);
app_server.shutdown().await?;
proxy.await??;
Ok(())
+8
View File
@@ -30,6 +30,7 @@ use codex_app_server_protocol::PluginReadParams;
use codex_app_server_protocol::PluginReadResponse;
use codex_app_server_protocol::PluginUninstallResponse;
use codex_app_server_protocol::SkillsListResponse;
use codex_app_server_protocol::Thread;
use codex_app_server_protocol::ThreadGoalStatus;
use codex_connectors::AppInfo;
use codex_file_search::FileMatch;
@@ -38,6 +39,7 @@ use codex_protocol::ThreadId;
use codex_protocol::openai_models::ModelPreset;
use codex_utils_absolute_path::AbsolutePathBuf;
use codex_utils_approval_presets::ApprovalPreset;
use uuid::Uuid;
use crate::app_command::AppCommand;
use crate::app_server_session::AppServerStartedThread;
@@ -179,6 +181,12 @@ pub(crate) enum KeymapEditIntent {
pub(crate) enum AppEvent {
/// Open the agent picker for switching active threads.
OpenAgentPicker,
/// Merge a completed root-scoped agent-picker refresh without blocking terminal input.
AgentPickerThreadsLoaded {
primary_thread_id: ThreadId,
request_id: Uuid,
result: Result<Vec<Thread>, String>,
},
/// Switch the active thread to the selected agent.
SelectAgentThread(ThreadId),
@@ -486,6 +486,11 @@ impl ListSelectionView {
let previously_selected = self
.selected_actual_idx()
.filter(|actual_idx| self.enabled_actual_idx(*actual_idx).is_some())
.or_else(|| {
self.initial_selected_idx
.take()
.filter(|actual_idx| self.enabled_actual_idx(*actual_idx).is_some())
})
.or_else(|| {
(!self.is_searchable)
.then(|| {
@@ -494,11 +499,6 @@ impl ListSelectionView {
.position(|item| item.is_current && Self::item_is_enabled(item))
})
.flatten()
})
.or_else(|| {
self.initial_selected_idx
.take()
.filter(|actual_idx| self.enabled_actual_idx(*actual_idx).is_some())
});
if self.is_searchable && !self.search_query.is_empty() {
+7
View File
@@ -1236,6 +1236,13 @@ impl BottomPane {
.and_then(|view| view.selected_index())
}
pub(crate) fn selected_index_for_present_view(&self, view_id: &'static str) -> Option<usize> {
self.view_stack
.iter()
.rfind(|view| view.view_id() == Some(view_id))
.and_then(|view| view.selected_index())
}
pub(crate) fn active_tab_id_for_active_view(&self, view_id: &'static str) -> Option<&str> {
self.view_stack
.last()
@@ -233,6 +233,24 @@ impl ChatWidget {
self.request_redraw();
}
pub(crate) fn selected_index_for_present_view(&self, view_id: &'static str) -> Option<usize> {
self.bottom_pane.selected_index_for_present_view(view_id)
}
pub(crate) fn replace_selection_view_if_present(
&mut self,
view_id: &'static str,
params: SelectionViewParams,
) -> bool {
let replaced = self
.bottom_pane
.replace_selection_view_if_present(view_id, params);
if replaced {
self.refresh_plan_mode_nudge();
}
replaced
}
pub(crate) fn no_modal_or_popup_active(&self) -> bool {
self.bottom_pane.no_modal_or_popup_active()
}