Preserve sub-agent liveness in the agent picker (#33921)

## Why

Opening the agent picker could clear a successful spawn hint before the new
sub-agent had emitted a turn event, making a running agent appear stopped.

## What changed

- Treat active turns and matching completion events as authoritative liveness
  transitions while leaving an empty event store unchanged.
- Remember terminal liveness so delayed activity cannot revive a stopped agent;
  only a new turn can mark it running again.
- Ignore `Interacted` activity when deriving agent liveness.

## Testing

Added coverage for empty stores, stale and current completion events, completed
snapshots, delayed activity, and subsequent turns.

GitOrigin-RevId: 462539d8134a5d3520dbd41edb44b3251f50ead5
This commit is contained in:
Eric Traut
2026-07-17 23:07:40 +00:00
committed by copyberry
parent 3a06748458
commit d4982d0a33
5 changed files with 185 additions and 23 deletions
+30 -2
View File
@@ -44,6 +44,8 @@ pub(crate) struct AgentNavigationState {
threads: HashMap<ThreadId, AgentPickerThreadEntry>,
/// Stable first-seen traversal order for picker rows and keyboard cycling.
order: Vec<ThreadId>,
/// Threads with observed terminal liveness that must not be revived by delayed activity.
stopped_threads: HashSet<ThreadId>,
/// Spawned child threads whose instructions are owned by their parent agent.
parent_owned_threads: HashSet<ThreadId>,
}
@@ -132,8 +134,32 @@ impl AgentNavigationState {
is_closed: false,
});
entry.agent_path = Some(activity.agent_path);
entry.is_running = activity.is_running_hint;
entry.is_closed = false;
if activity.is_running_hint
&& !entry.is_closed
&& !self.stopped_threads.contains(&activity.thread_id)
{
entry.is_running = true;
} else {
entry.is_running = false;
self.stopped_threads.insert(activity.thread_id);
}
}
pub(crate) fn mark_running(&mut self, thread_id: ThreadId) {
if self
.threads
.get(&thread_id)
.is_some_and(|entry| entry.is_closed)
{
return;
}
self.stopped_threads.remove(&thread_id);
self.set_running(thread_id, /*is_running*/ true);
}
pub(crate) fn mark_stopped(&mut self, thread_id: ThreadId) {
self.stopped_threads.insert(thread_id);
self.set_running(thread_id, /*is_running*/ false);
}
pub(crate) fn set_running(&mut self, thread_id: ThreadId, is_running: bool) {
@@ -175,6 +201,7 @@ impl AgentNavigationState {
pub(crate) fn clear(&mut self) {
self.threads.clear();
self.order.clear();
self.stopped_threads.clear();
self.parent_owned_threads.clear();
}
@@ -186,6 +213,7 @@ impl AgentNavigationState {
pub(crate) fn remove(&mut self, thread_id: ThreadId) {
self.threads.remove(&thread_id);
self.order.retain(|candidate| *candidate != thread_id);
self.stopped_threads.remove(&thread_id);
self.parent_owned_threads.remove(&thread_id);
}
+23 -5
View File
@@ -18,8 +18,8 @@ impl App {
pub(super) async fn open_agent_picker(&mut self, app_server: &mut AppServerSession) {
self.backfill_loaded_subagent_threads(app_server).await;
// V2 subagents are identified by canonical paths observed from activity events or loaded
// thread metadata. Prefer local buffered turn state for liveness, and fall back to
// thread/read only when no local event channel exists.
// 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.
let path_backed_thread_ids: Vec<_> = self
.agent_navigation
.ordered_path_backed_subagent_threads(self.primary_thread_id)
@@ -30,8 +30,21 @@ impl App {
if let Some(channel) = self.thread_event_channels.get(&thread_id)
&& channel.attachment() == ThreadEventAttachment::Live
{
let is_running = channel.store.lock().await.active_turn_id().is_some();
self.agent_navigation.set_running(thread_id, is_running);
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)),
)
};
if has_active_turn {
self.agent_navigation.mark_running(thread_id);
} else if has_terminal_snapshot {
self.agent_navigation.mark_stopped(thread_id);
}
} else {
self.refresh_agent_picker_thread_liveness(app_server, thread_id)
.await;
@@ -241,7 +254,12 @@ impl App {
self.agent_navigation.mark_parent_owned(thread_id);
}
self.agent_navigation.set_agent_path(thread_id, agent_path);
self.agent_navigation.set_running(thread_id, is_running);
if is_running {
self.agent_navigation.mark_running(thread_id);
} else {
self.agent_navigation
.set_running(thread_id, /*is_running*/ false);
}
true
}
Err(err) => {
+96 -12
View File
@@ -1418,7 +1418,94 @@ async fn open_agent_picker_preserves_cached_metadata_for_replay_threads() -> Res
}
#[tokio::test]
async fn open_agent_picker_clears_completed_path_backed_agent_running_state() -> Result<()> {
async fn open_agent_picker_preserves_running_hints_until_observed_completion() -> Result<()> {
let (mut app, mut app_event_rx, _op_rx) = make_test_app_with_channels().await;
let mut app_server = Box::pin(crate::start_embedded_app_server_for_picker(
app.chat_widget.config_ref(),
))
.await
.expect("embedded app server");
let thread_id = ThreadId::new();
app.thread_event_channels
.insert(thread_id, ThreadEventChannel::new(/*capacity*/ 4));
app.agent_navigation
.record_sub_agent_activity(SubAgentActivityDisplay {
thread_id,
agent_path: "/root/child".to_string(),
is_running_hint: true,
});
Box::pin(app.open_agent_picker(&mut app_server)).await;
let mut expected_entry = AgentPickerThreadEntry {
agent_nickname: None,
agent_role: None,
agent_path: Some("/root/child".to_string()),
is_running: true,
is_closed: false,
};
assert_eq!(app.agent_navigation.get(&thread_id), Some(&expected_entry));
let status = loop {
let event = app_event_rx.try_recv().expect("agent status history cell");
if let AppEvent::InsertHistoryCell(cell) = event {
let rendered = lines_to_single_string(&cell.display_lines(/*width*/ 80));
if rendered.contains("/agent") {
break rendered;
}
}
};
assert_snapshot!(status, @r###"
/agent
Sub-agents running
• `/root/child`
No recent activity yet.
"###);
app.enqueue_thread_notification(
thread_id,
turn_completed_notification(thread_id, "turn-older", TurnStatus::Completed),
)
.await?;
Box::pin(app.open_agent_picker(&mut app_server)).await;
assert_eq!(app.agent_navigation.get(&thread_id), Some(&expected_entry));
app.enqueue_thread_notification(thread_id, turn_started_notification(thread_id, "turn-1"))
.await?;
app.enqueue_thread_notification(
thread_id,
turn_completed_notification(thread_id, "turn-1", TurnStatus::Completed),
)
.await?;
Box::pin(app.open_agent_picker(&mut app_server)).await;
expected_entry.is_running = false;
assert_eq!(app.agent_navigation.get(&thread_id), Some(&expected_entry));
app.agent_navigation
.record_sub_agent_activity(SubAgentActivityDisplay {
thread_id,
agent_path: "/root/child".to_string(),
is_running_hint: true,
});
Box::pin(app.open_agent_picker(&mut app_server)).await;
expected_entry.is_running = false;
assert_eq!(app.agent_navigation.get(&thread_id), Some(&expected_entry));
app.enqueue_thread_notification(thread_id, turn_started_notification(thread_id, "turn-2"))
.await?;
Box::pin(app.open_agent_picker(&mut app_server)).await;
expected_entry.is_running = true;
assert_eq!(app.agent_navigation.get(&thread_id), Some(&expected_entry));
Ok(())
}
#[tokio::test]
async fn open_agent_picker_clears_running_hint_from_completed_snapshot() -> Result<()> {
let mut app = Box::pin(make_test_app()).await;
let mut app_server = Box::pin(crate::start_embedded_app_server_for_picker(
app.chat_widget.config_ref(),
@@ -1426,17 +1513,14 @@ async fn open_agent_picker_clears_completed_path_backed_agent_running_state() ->
.await
.expect("embedded app server");
let thread_id = ThreadId::new();
let channel = ThreadEventChannel::new(/*capacity*/ 4);
{
let mut store = channel.store.lock().await;
store.push_notification(turn_started_notification(thread_id, "turn-1"));
store.push_notification(turn_completed_notification(
thread_id,
"turn-1",
TurnStatus::Completed,
));
}
app.thread_event_channels.insert(thread_id, channel);
app.thread_event_channels.insert(
thread_id,
ThreadEventChannel::new_with_session(
THREAD_EVENT_CHANNEL_CAPACITY,
test_thread_session(thread_id, test_path_buf("/tmp/project")),
vec![test_turn("turn-1", TurnStatus::Completed, Vec::new())],
),
);
app.agent_navigation
.record_sub_agent_activity(SubAgentActivityDisplay {
thread_id,
+18 -3
View File
@@ -930,17 +930,32 @@ impl App {
let channel = self.ensure_thread_channel(thread_id);
(channel.sender.clone(), Arc::clone(&channel.store))
};
let (should_send, pending_status) = {
let (should_send, pending_status, turn_stopped) = {
let mut guard = store.lock().await;
if guard.session.is_none()
&& let Some(session) = inferred_session
{
guard.session = Some(session);
}
let turn_stopped = match &notification {
ServerNotification::TurnCompleted(notification) => {
guard.active_turn_id() == Some(notification.turn.id.as_str())
}
ServerNotification::ThreadClosed(_) => true,
_ => false,
};
guard.push_notification(notification.clone());
(guard.active, guard.side_parent_pending_status())
(
guard.active,
guard.side_parent_pending_status(),
turn_stopped,
)
};
if matches!(notification, ServerNotification::TurnStarted(_)) {
self.agent_navigation.mark_running(thread_id);
} else if turn_stopped {
self.agent_navigation.mark_stopped(thread_id);
}
let notification_status_change = SideParentStatusChange::for_notification(&notification);
if should_send {
+18 -1
View File
@@ -288,10 +288,15 @@ pub(crate) fn sub_agent_activity_display(item: &ThreadItem) -> Option<SubAgentAc
else {
return None;
};
let is_running_hint = match kind {
SubAgentActivityKind::Started => true,
SubAgentActivityKind::Interacted => return None,
SubAgentActivityKind::Interrupted => false,
};
Some(SubAgentActivityDisplay {
thread_id: parse_thread_id(agent_thread_id)?,
agent_path: agent_path.clone(),
is_running_hint: !matches!(kind, SubAgentActivityKind::Interrupted),
is_running_hint,
})
}
@@ -674,6 +679,18 @@ mod tests {
use ratatui::style::Modifier;
use std::collections::HashMap;
#[test]
fn interacted_sub_agent_activity_does_not_change_liveness() {
let item = ThreadItem::SubAgentActivity {
id: "activity-1".to_string(),
kind: SubAgentActivityKind::Interacted,
agent_thread_id: ThreadId::new().to_string(),
agent_path: "/root/child".to_string(),
};
assert_eq!(sub_agent_activity_display(&item), None);
}
#[test]
fn collab_events_snapshot() {
let sender_thread_id = ThreadId::from_string("00000000-0000-0000-0000-000000000001")