mirror of
https://github.com/openai/codex.git
synced 2026-10-27 18:48:37 +00:00
Eric Traut
1fb5158b34
## Why Local permission settings could overwrite a remote task's saved server settings when resuming or forking, including its named permission profile. ## What changed - Omit permission overrides from remote resume requests and remote fork requests that inherit saved settings. - Reject explicit permission overrides when resuming or forking remote tasks, including overrides from session flags and selected configuration profiles. - Refresh restored runtime permission and approval overrides after a remote resume so settings from the previous task do not persist. - Carry the selected fork permission mode through retries after unarchiving. ## Testing Add regression coverage for restoring saved server permission profiles, approval policies, and approval reviewers across resume and fork; rejecting explicit remote resume overrides; and clearing stale restored runtime overrides. GitOrigin-RevId: 465c0444df380553e50ddbeeceff9dc64e5bf9c1
1338 lines
56 KiB
Rust
1338 lines
56 KiB
Rust
//! Session, resume, fork, and subagent selection lifecycle for the TUI app.
|
|
//!
|
|
//! This module owns the high-level transitions between app-server threads: starting fresh sessions,
|
|
//! resuming/forking saved sessions, replacing ChatWidget instances, and maintaining the agent picker
|
|
//! cache used for multi-agent navigation.
|
|
|
|
use std::io;
|
|
|
|
use super::agent_picker::AGENT_PICKER_VIEW_ID;
|
|
use super::app_server_event_targets::ServerNotificationThreadTarget;
|
|
use super::app_server_event_targets::server_notification_thread_target;
|
|
use super::app_server_event_targets::server_request_thread_id;
|
|
use super::*;
|
|
use crate::app_server_session::source_agent_path;
|
|
use crate::app_server_session::thread_blocks_direct_input;
|
|
use crate::chatwidget::ThreadInputStateRestoreMode;
|
|
use std::collections::HashSet;
|
|
|
|
#[derive(Clone, Copy)]
|
|
pub(super) enum ThreadAttachPresentation {
|
|
SessionLineage,
|
|
PromptEdit,
|
|
}
|
|
|
|
/// Reports whether a loaded-thread backfill completed and which descendants already had their
|
|
/// liveness metadata refreshed, allowing the picker to skip duplicate `thread/read` requests.
|
|
#[derive(Default)]
|
|
pub(super) struct LoadedSubagentBackfill {
|
|
pub(super) completed: bool,
|
|
pub(super) refreshed_thread_ids: HashSet<ThreadId>,
|
|
}
|
|
|
|
impl App {
|
|
pub(super) async fn open_agent_picker(&mut self, app_server: &mut AppServerSession) {
|
|
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.
|
|
let path_backed_thread_ids: Vec<_> = self
|
|
.agent_navigation
|
|
.ordered_path_backed_subagent_threads(self.primary_thread_id)
|
|
.into_iter()
|
|
.map(|(thread_id, _)| thread_id)
|
|
.collect();
|
|
for thread_id in path_backed_thread_ids.iter().copied() {
|
|
if let Some(channel) = self.thread_event_channels.get(&thread_id)
|
|
&& channel.attachment() == ThreadEventAttachment::Live
|
|
{
|
|
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 self.primary_thread_id.is_none()
|
|
&& !backfill.refreshed_thread_ids.contains(&thread_id)
|
|
{
|
|
self.refresh_agent_picker_thread_liveness(app_server, thread_id)
|
|
.await;
|
|
}
|
|
}
|
|
let path_backed_threads = self
|
|
.agent_navigation
|
|
.ordered_path_backed_subagent_threads(self.primary_thread_id);
|
|
if !path_backed_threads.is_empty() {
|
|
let running_threads: Vec<_> = path_backed_threads
|
|
.into_iter()
|
|
.filter_map(|(thread_id, entry)| {
|
|
if !entry.is_running || entry.is_closed {
|
|
return None;
|
|
}
|
|
Some((thread_id, entry.agent_path.as_deref()?.trim().to_string()))
|
|
})
|
|
.collect();
|
|
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) {
|
|
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)
|
|
};
|
|
entries.push(preview);
|
|
}
|
|
|
|
self.chat_widget
|
|
.add_to_history(super::agent_status_feed::AgentStatusHistoryCell::new(
|
|
entries,
|
|
));
|
|
}
|
|
|
|
let mut thread_ids = self.agent_navigation.tracked_thread_ids();
|
|
for thread_id in self.thread_event_channels.keys().copied() {
|
|
if !thread_ids.contains(&thread_id) {
|
|
thread_ids.push(thread_id);
|
|
}
|
|
}
|
|
for thread_id in thread_ids {
|
|
if path_backed_thread_ids.contains(&thread_id)
|
|
|| self.side_threads.contains_key(&thread_id)
|
|
|| backfill.refreshed_thread_ids.contains(&thread_id)
|
|
{
|
|
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,
|
|
);
|
|
}
|
|
}
|
|
|
|
let has_non_primary_agent_thread = self
|
|
.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_feature_enable_prompt(Feature::Collab);
|
|
return;
|
|
}
|
|
|
|
if self.agent_navigation.is_empty() {
|
|
self.chat_widget
|
|
.add_info_message("No agents available yet.".to_string(), /*hint*/ None);
|
|
return;
|
|
}
|
|
|
|
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 initial_selected_idx.is_none() && self.active_thread_id == Some(thread_id) {
|
|
initial_selected_idx = Some(idx);
|
|
}
|
|
let id = thread_id;
|
|
let is_primary = self.primary_thread_id == Some(thread_id);
|
|
let name = entry
|
|
.agent_path
|
|
.as_deref()
|
|
.map(str::trim)
|
|
.filter(|agent_path| !is_primary && !agent_path.is_empty())
|
|
.map(ToOwned::to_owned)
|
|
.unwrap_or_else(|| {
|
|
format_agent_picker_item_name(
|
|
entry.agent_nickname.as_deref(),
|
|
entry.agent_role.as_deref(),
|
|
is_primary,
|
|
)
|
|
});
|
|
let uuid = thread_id.to_string();
|
|
SelectionItem {
|
|
name: name.clone(),
|
|
name_prefix_spans: agent_picker_status_dot_spans(entry.is_closed),
|
|
description: Some(uuid.clone()),
|
|
is_current: self.active_thread_id == Some(thread_id),
|
|
actions: vec![Box::new(move |tx| {
|
|
tx.send(AppEvent::SelectAgentThread(id));
|
|
})],
|
|
dismiss_on_select: true,
|
|
search_value: Some(format!("{name} {uuid}")),
|
|
..Default::default()
|
|
}
|
|
})
|
|
.collect();
|
|
|
|
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 {
|
|
err.chain()
|
|
.any(|cause| cause.to_string().contains("thread not loaded:"))
|
|
}
|
|
|
|
pub(super) fn closed_state_for_thread_read_error(
|
|
err: &color_eyre::Report,
|
|
existing_is_closed: Option<bool>,
|
|
) -> bool {
|
|
Self::is_terminal_thread_read_error(err) || existing_is_closed.unwrap_or(false)
|
|
}
|
|
|
|
pub(super) fn can_fallback_from_include_turns_error(err: &color_eyre::Report) -> bool {
|
|
err.chain().any(|cause| {
|
|
let message = cause.to_string();
|
|
message.contains("includeTurns is unavailable before first user message")
|
|
|| message.contains("thread/turns/list is unavailable before first user message")
|
|
|| message.contains("ephemeral threads do not support includeTurns")
|
|
})
|
|
}
|
|
|
|
/// Updates cached picker metadata and then mirrors any visible-label change into the footer.
|
|
///
|
|
/// These two writes stay paired so the picker rows and contextual footer continue to describe
|
|
/// the same displayed thread after nickname or role updates.
|
|
pub(super) fn upsert_agent_picker_thread(
|
|
&mut self,
|
|
thread_id: ThreadId,
|
|
agent_nickname: Option<String>,
|
|
agent_role: Option<String>,
|
|
is_closed: bool,
|
|
) {
|
|
self.chat_widget.set_collab_agent_metadata(
|
|
thread_id,
|
|
agent_nickname.clone(),
|
|
agent_role.clone(),
|
|
);
|
|
self.agent_navigation
|
|
.upsert(thread_id, agent_nickname, agent_role, is_closed);
|
|
self.sync_active_agent_label();
|
|
}
|
|
|
|
/// Persists the app-server's authoritative ownership flag and updates the active composer.
|
|
pub(super) fn mark_primary_thread_parent_owned(&mut self, thread_id: ThreadId) {
|
|
self.agent_navigation.mark_parent_owned(thread_id);
|
|
self.chat_widget.set_parent_owned_thread();
|
|
}
|
|
|
|
/// Marks a cached picker thread closed and recomputes the contextual footer label.
|
|
///
|
|
/// Closing a thread is not the same as removing it: users can still inspect finished agent
|
|
/// transcripts, and the stable next/previous traversal order should not collapse around them.
|
|
pub(super) fn mark_agent_picker_thread_closed(&mut self, thread_id: ThreadId) {
|
|
self.agent_navigation.mark_closed(thread_id);
|
|
self.sync_active_agent_label();
|
|
}
|
|
|
|
pub(super) async fn refresh_agent_picker_thread_liveness(
|
|
&mut self,
|
|
app_server: &mut AppServerSession,
|
|
thread_id: ThreadId,
|
|
) -> bool {
|
|
let existing_entry = self.agent_navigation.get(&thread_id).cloned();
|
|
let has_replay_channel = self.thread_event_channels.contains_key(&thread_id);
|
|
match app_server
|
|
.thread_read(thread_id, /*include_turns*/ false)
|
|
.await
|
|
{
|
|
Ok(thread) => {
|
|
let is_parent_owned = thread_blocks_direct_input(&thread);
|
|
let agent_path = source_agent_path(&thread.source);
|
|
let is_running = matches!(
|
|
thread.status,
|
|
codex_app_server_protocol::ThreadStatus::Active { .. }
|
|
);
|
|
let is_closed = matches!(
|
|
thread.status,
|
|
codex_app_server_protocol::ThreadStatus::NotLoaded
|
|
);
|
|
self.upsert_agent_picker_thread(
|
|
thread_id,
|
|
thread.agent_nickname.or_else(|| {
|
|
existing_entry
|
|
.as_ref()
|
|
.and_then(|entry| entry.agent_nickname.clone())
|
|
}),
|
|
thread.agent_role.or_else(|| {
|
|
existing_entry
|
|
.as_ref()
|
|
.and_then(|entry| entry.agent_role.clone())
|
|
}),
|
|
is_closed,
|
|
);
|
|
if is_parent_owned {
|
|
self.agent_navigation.mark_parent_owned(thread_id);
|
|
}
|
|
self.agent_navigation.set_agent_path(thread_id, agent_path);
|
|
if is_running {
|
|
self.agent_navigation.mark_running(thread_id);
|
|
} else {
|
|
self.agent_navigation
|
|
.set_running(thread_id, /*is_running*/ false);
|
|
}
|
|
true
|
|
}
|
|
Err(err) => {
|
|
if Self::is_terminal_thread_read_error(&err) && !has_replay_channel {
|
|
self.agent_navigation.remove(thread_id);
|
|
return false;
|
|
}
|
|
let is_closed = Self::closed_state_for_thread_read_error(
|
|
&err,
|
|
existing_entry.as_ref().map(|entry| entry.is_closed),
|
|
);
|
|
if let Some(entry) = existing_entry {
|
|
self.upsert_agent_picker_thread(
|
|
thread_id,
|
|
entry.agent_nickname,
|
|
entry.agent_role,
|
|
is_closed,
|
|
);
|
|
} else {
|
|
self.upsert_agent_picker_thread(
|
|
thread_id, /*agent_nickname*/ None, /*agent_role*/ None,
|
|
is_closed,
|
|
);
|
|
}
|
|
self.agent_navigation
|
|
.set_running(thread_id, /*is_running*/ false);
|
|
true
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Materializes a live thread into local replay state when the picker knows about it but the
|
|
/// TUI has not cached a local event channel yet.
|
|
///
|
|
/// Resume-time backfill intentionally avoids creating empty placeholder channels, because those
|
|
/// placeholders make stale `/subagents` entries open blank transcripts. When a user later
|
|
/// selects a still-live discovered thread, attach it on demand with a real resumed snapshot.
|
|
pub(super) async fn attach_live_thread_for_selection(
|
|
&mut self,
|
|
app_server: &mut AppServerSession,
|
|
thread_id: ThreadId,
|
|
) -> Result<bool> {
|
|
let was_external_writer = self
|
|
.thread_event_channels
|
|
.get(&thread_id)
|
|
.is_some_and(|channel| channel.attachment() == ThreadEventAttachment::ExternalWriter);
|
|
if self
|
|
.thread_event_channels
|
|
.get(&thread_id)
|
|
.is_some_and(|channel| channel.attachment() == ThreadEventAttachment::Live)
|
|
{
|
|
return Ok(true);
|
|
}
|
|
|
|
let (session, turns, live_attached) = match app_server
|
|
.resume_thread(
|
|
&self.local_settings,
|
|
self.config.clone(),
|
|
thread_id,
|
|
crate::app_server_session::ResumeModelSettings::PreserveExistingThread,
|
|
)
|
|
.await
|
|
{
|
|
Ok(started) => {
|
|
if let Some(entry) = self.agent_navigation.get(&thread_id).cloned() {
|
|
self.upsert_agent_picker_thread(
|
|
thread_id,
|
|
entry.agent_nickname,
|
|
entry.agent_role,
|
|
/*is_closed*/ false,
|
|
);
|
|
}
|
|
if started.blocks_direct_input {
|
|
self.agent_navigation.mark_parent_owned(thread_id);
|
|
}
|
|
(started.session, started.turns, true)
|
|
}
|
|
Err(resume_err) => {
|
|
tracing::warn!(
|
|
thread_id = %thread_id,
|
|
error = %resume_err,
|
|
"failed to resume live thread for selection; falling back to thread/read"
|
|
);
|
|
let mut thread = app_server
|
|
.thread_read(thread_id, /*include_turns*/ false)
|
|
.await?;
|
|
match app_server
|
|
.hydrate_initial_thread_history(
|
|
&mut thread,
|
|
/*turn_cursor*/ None,
|
|
/*item_cursor*/ None,
|
|
Some(&self.config),
|
|
Some(&self.local_settings),
|
|
crate::app_server_session::HistoryHydrationScope::Initial,
|
|
)
|
|
.await
|
|
{
|
|
Ok(()) => {}
|
|
Err(err) if Self::can_fallback_from_include_turns_error(&err) => {}
|
|
Err(err) => return Err(err),
|
|
}
|
|
let turns = thread.turns.clone();
|
|
if turns.is_empty() {
|
|
// A `thread/read` fallback without turns would create a blank local replay
|
|
// channel with no live listener attached, which blocks later real re-attach.
|
|
return Err(color_eyre::eyre::eyre!(
|
|
"Agent thread {thread_id} is not yet available for replay or live attach."
|
|
));
|
|
}
|
|
let mut session = self.session_state_for_thread_read(thread_id, &thread).await;
|
|
// Reads have no settings. Keep this cached thread's permissions rather than
|
|
// inferring them from the conversation that is currently displayed.
|
|
if let Some(channel) = self.thread_event_channels.get(&thread_id)
|
|
&& let Some(cached) = channel.store.lock().await.session.as_ref()
|
|
{
|
|
session.approval_policy = cached.approval_policy;
|
|
session.permission_profile = cached.permission_profile.clone();
|
|
session.active_permission_profile = cached.active_permission_profile.clone();
|
|
session.approvals_reviewer = cached.approvals_reviewer;
|
|
}
|
|
// `thread/read` can seed replay state, but it does not attach the app-server
|
|
// listener that `thread/resume` establishes, so treat this path as replay-only.
|
|
session.model.clear();
|
|
(session, turns, false)
|
|
}
|
|
};
|
|
self.agents_overview.activity.remove(&thread_id);
|
|
let recap_progress =
|
|
if live_attached && let Some(channel) = self.thread_event_channels.remove(&thread_id) {
|
|
let store = channel.store.lock().await;
|
|
if let Some(input) = store.input_state.clone() {
|
|
self.agents_overview.input_states.insert(thread_id, input);
|
|
}
|
|
store.recap_progress()
|
|
} else {
|
|
Default::default()
|
|
};
|
|
let channel = self.ensure_thread_channel(thread_id);
|
|
if !live_attached {
|
|
if was_external_writer {
|
|
channel.mark_external_writer();
|
|
} else {
|
|
channel.mark_replay_only();
|
|
}
|
|
}
|
|
let mut store = channel.store.lock().await;
|
|
store.set_session(session, turns);
|
|
store.merge_recap_progress(recap_progress);
|
|
store.rebase_buffer_after_session_refresh();
|
|
Ok(live_attached)
|
|
}
|
|
|
|
/// Replaces the chat widget and re-seeds the new widget's collab metadata from the navigation
|
|
/// cache.
|
|
///
|
|
/// Thread switches reconstruct the `ChatWidget`, which loses the `collab_agent_metadata` map.
|
|
/// This helper copies every known nickname/role from `AgentNavigationState` into the
|
|
/// replacement widget so that replayed collab items render agent names immediately.
|
|
pub(super) fn replace_chat_widget(&mut self, mut chat_widget: ChatWidget) {
|
|
chat_widget.cyber_policy_notice = self.chat_widget.cyber_policy_notice.clone();
|
|
self.commit_animation = None;
|
|
// Transfer the last-written terminal title to the replacement widget
|
|
// so it knows what OSC title is currently displayed. Without this, the
|
|
// new widget would redundantly clear and rewrite the same title, causing
|
|
// a visible flicker in some terminals.
|
|
let previous_terminal_title = self.chat_widget.last_terminal_title.take();
|
|
if chat_widget.last_terminal_title.is_none() {
|
|
chat_widget.last_terminal_title = previous_terminal_title;
|
|
}
|
|
chat_widget.remote_connection = self.chat_widget.remote_connection.clone();
|
|
chat_widget.set_local_worktree_operations(self.chat_widget.local_worktree_operations);
|
|
chat_widget.set_agents_navigation_enabled(matches!(
|
|
self.app_server_target,
|
|
AppServerTarget::LocalDaemon { .. }
|
|
));
|
|
chat_widget.inherit_backend_banner_state(&mut self.chat_widget);
|
|
for (thread_id, entry) in self.agent_navigation.ordered_threads() {
|
|
chat_widget.set_collab_agent_metadata(
|
|
thread_id,
|
|
entry.agent_nickname.clone(),
|
|
entry.agent_role.clone(),
|
|
);
|
|
}
|
|
self.chat_widget = chat_widget;
|
|
self.sync_active_agent_label();
|
|
}
|
|
|
|
pub(super) async fn select_agent_thread(
|
|
&mut self,
|
|
tui: &mut tui::Tui,
|
|
app_server: &mut AppServerSession,
|
|
thread_id: ThreadId,
|
|
) -> Result<()> {
|
|
let cached_session = if self.thread_unavailable(thread_id) {
|
|
self.thread_event_channels[&thread_id]
|
|
.store
|
|
.lock()
|
|
.await
|
|
.session
|
|
.clone()
|
|
} else {
|
|
None
|
|
};
|
|
if self.active_thread_id == Some(thread_id) {
|
|
if !self.thread_unavailable(thread_id) {
|
|
return Ok(());
|
|
}
|
|
// Detach the cached receiver before a successful attachment replaces its channel.
|
|
self.store_active_thread_receiver().await;
|
|
}
|
|
|
|
// A tracked side thread stays loaded until it is explicitly discarded and already has a
|
|
// replay channel, so another liveness read cannot add anything before selection.
|
|
if !(self.side_threads.contains_key(&thread_id)
|
|
&& self.thread_event_channels.contains_key(&thread_id)
|
|
|| self
|
|
.refresh_agent_picker_thread_liveness(app_server, thread_id)
|
|
.await)
|
|
{
|
|
self.chat_widget
|
|
.add_error_message(format!("Agent thread {thread_id} is no longer available."));
|
|
return Ok(());
|
|
}
|
|
let mut is_replay_only = self
|
|
.agent_navigation
|
|
.get(&thread_id)
|
|
.is_some_and(|entry| entry.is_closed);
|
|
let mut attached_replay_only = false;
|
|
if self.should_attach_live_thread_for_selection(thread_id) {
|
|
match self
|
|
.attach_live_thread_for_selection(app_server, thread_id)
|
|
.await
|
|
{
|
|
Ok(live_attached) => {
|
|
attached_replay_only = !live_attached;
|
|
is_replay_only = attached_replay_only;
|
|
}
|
|
Err(_) if self.thread_event_channels.contains_key(&thread_id) => {
|
|
is_replay_only = true;
|
|
attached_replay_only = true;
|
|
}
|
|
Err(err) => {
|
|
self.chat_widget.add_error_message(format!(
|
|
"Failed to attach to agent thread {thread_id}: {err}"
|
|
));
|
|
return Ok(());
|
|
}
|
|
}
|
|
} else if !self.thread_event_channels.contains_key(&thread_id) && is_replay_only {
|
|
self.chat_widget
|
|
.add_error_message(format!("Agent thread {thread_id} is no longer available."));
|
|
return Ok(());
|
|
}
|
|
let previous_thread_id = self.active_thread_id;
|
|
self.store_active_thread_receiver().await;
|
|
self.active_thread_id = None;
|
|
let Some((receiver, mut snapshot)) = self.activate_thread_for_replay(thread_id).await
|
|
else {
|
|
self.chat_widget
|
|
.add_error_message(format!("Agent thread {thread_id} is already active."));
|
|
if let Some(previous_thread_id) = previous_thread_id {
|
|
self.activate_thread_channel(previous_thread_id).await;
|
|
}
|
|
return Ok(());
|
|
};
|
|
|
|
self.refresh_snapshot_session_if_needed(
|
|
app_server,
|
|
thread_id,
|
|
is_replay_only,
|
|
&mut snapshot,
|
|
)
|
|
.await;
|
|
// Refreshing can merge restored turns into the store, so recap progress must be read only
|
|
// after the refresh while the activated thread channel is still retained.
|
|
let Some(channel) = self.thread_event_channels.get(&thread_id) else {
|
|
self.chat_widget
|
|
.add_error_message(format!("Agent thread {thread_id} is no longer available."));
|
|
return Ok(());
|
|
};
|
|
let recap_progress = {
|
|
let mut store = channel.store.lock().await;
|
|
if let (Some(cached), Some(session)) =
|
|
(cached_session.as_ref(), snapshot.session.as_mut())
|
|
{
|
|
self.restore_runtime_permissions(session, cached);
|
|
store.session = Some(session.clone());
|
|
if !is_replay_only && self.primary_thread_id == Some(thread_id) {
|
|
self.primary_session_configured = Some(session.clone());
|
|
}
|
|
}
|
|
store.recap_progress()
|
|
};
|
|
snapshot.input_state = snapshot
|
|
.input_state
|
|
.or(self.agents_overview.input_states.remove(&thread_id));
|
|
|
|
self.active_thread_id = Some(thread_id);
|
|
self.active_thread_rx = Some(receiver);
|
|
|
|
self.recap.note_focus_gained();
|
|
self.recap = recap::RecapState::default();
|
|
|
|
if !tui.is_terminal_focused() {
|
|
self.recap.note_focus_lost(Instant::now());
|
|
}
|
|
let now = Instant::now();
|
|
self.recap.seed_from_progress(recap_progress, now);
|
|
self.schedule_recap_check(thread_id, now);
|
|
|
|
self.render_thread_snapshot(tui, app_server, thread_id, snapshot, !is_replay_only)?;
|
|
if is_replay_only
|
|
&& self
|
|
.thread_event_channels
|
|
.get(&thread_id)
|
|
.is_none_or(|channel| channel.attachment() != ThreadEventAttachment::ExternalWriter)
|
|
{
|
|
self.chat_widget.pause_unavailable_thread();
|
|
let message = if attached_replay_only {
|
|
format!(
|
|
"Agent thread {thread_id} could not be resumed live. Replaying saved transcript."
|
|
)
|
|
} else {
|
|
format!("Agent thread {thread_id} is closed. Replaying saved transcript.")
|
|
};
|
|
self.chat_widget.add_info_message(message, /*hint*/ None);
|
|
}
|
|
self.refresh_pending_thread_approvals().await;
|
|
|
|
Ok(())
|
|
}
|
|
|
|
pub(super) fn render_thread_snapshot(
|
|
&mut self,
|
|
tui: &mut tui::Tui,
|
|
app_server: &AppServerSession,
|
|
thread_id: ThreadId,
|
|
snapshot: ThreadEventSnapshot,
|
|
resume_restored_queue: bool,
|
|
) -> Result<()> {
|
|
let init = self.chatwidget_init_for_forked_or_resumed_thread(
|
|
tui,
|
|
self.config.clone(),
|
|
/*initial_user_message*/ None,
|
|
);
|
|
self.replace_chat_widget(ChatWidget::new_with_app_event(init));
|
|
self.chat_widget
|
|
.set_task_mentions_enabled(app_server.task_tools_available(thread_id));
|
|
self.chat_widget
|
|
.note_rendered_width(tui.terminal.last_known_screen_size.width);
|
|
let external_writer = self
|
|
.thread_event_channels
|
|
.get(&thread_id)
|
|
.is_some_and(|channel| channel.attachment() == ThreadEventAttachment::ExternalWriter);
|
|
if external_writer {
|
|
self.chat_widget.show_external_writer_thread();
|
|
}
|
|
if self.agent_navigation.is_parent_owned(thread_id) {
|
|
self.chat_widget.set_parent_owned_thread();
|
|
}
|
|
self.reset_for_thread_switch(tui)?;
|
|
self.replay_thread_snapshot(snapshot, resume_restored_queue);
|
|
if external_writer {
|
|
self.chat_widget.show_external_writer_thread();
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
pub(super) fn should_attach_live_thread_for_selection(&self, thread_id: ThreadId) -> bool {
|
|
self.thread_event_channels
|
|
.get(&thread_id)
|
|
.is_none_or(|channel| channel.attachment() != ThreadEventAttachment::Live)
|
|
&& self
|
|
.agent_navigation
|
|
.get(&thread_id)
|
|
.is_none_or(|entry| !entry.is_closed || self.thread_unavailable(thread_id))
|
|
}
|
|
|
|
pub(super) fn reset_for_thread_switch(&mut self, tui: &mut tui::Tui) -> Result<()> {
|
|
self.reset_transcript_state_after_clear();
|
|
tui.clear_pending_history_lines();
|
|
Self::clear_terminal_for_thread_switch(&mut tui.terminal)?;
|
|
Ok(())
|
|
}
|
|
|
|
pub(super) fn clear_terminal_for_thread_switch<B>(
|
|
terminal: &mut crate::custom_terminal::Terminal<B>,
|
|
) -> Result<()>
|
|
where
|
|
B: Backend<Error = io::Error> + Write,
|
|
{
|
|
terminal.clear_scrollback_and_visible_screen_ansi()?;
|
|
let mut area = terminal.viewport_area;
|
|
if area.y > 0 {
|
|
area.y = 0;
|
|
terminal.set_viewport_area(area);
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
pub(super) fn reset_thread_event_state(&mut self) {
|
|
self.abort_all_thread_event_listeners();
|
|
self.thread_event_channels.clear();
|
|
self.agents_overview.activity.clear();
|
|
self.agent_navigation.clear();
|
|
self.side_threads.clear();
|
|
self.active_thread_id = None;
|
|
self.active_thread_rx = None;
|
|
self.primary_thread_id = None;
|
|
self.last_subagent_backfill_attempt = None;
|
|
self.primary_session_configured = None;
|
|
self.pending_primary_events.clear();
|
|
self.pending_app_server_requests.clear();
|
|
self.pending_startup_thread_start = false;
|
|
self.chat_widget.set_pending_thread_approvals(Vec::new());
|
|
self.sync_active_agent_label();
|
|
}
|
|
|
|
pub(super) async fn handle_startup_thread_started(
|
|
&mut self,
|
|
app_server: &mut AppServerSession,
|
|
result: Result<AppServerStartedThread>,
|
|
) -> Result<()> {
|
|
if !self.pending_startup_thread_start {
|
|
if let Ok(started) = result {
|
|
let thread_id = started.session.thread_id;
|
|
if let Err(err) = app_server.thread_unsubscribe(thread_id).await {
|
|
tracing::warn!(
|
|
thread_id = %thread_id,
|
|
"failed to unsubscribe stale startup thread: {err}"
|
|
);
|
|
}
|
|
self.discard_thread_local_state(thread_id).await;
|
|
}
|
|
return Ok(());
|
|
}
|
|
|
|
self.pending_startup_thread_start = false;
|
|
self.chat_widget
|
|
.set_queue_submissions_until_session_configured(/*queue*/ false);
|
|
match result {
|
|
Ok(started) => {
|
|
let thread_id = started.session.thread_id;
|
|
if started.task_tools_available {
|
|
app_server.remember_task_tool_thread(thread_id);
|
|
self.chat_widget.set_task_mentions_enabled(/*enabled*/ true);
|
|
}
|
|
self.pending_primary_events.retain(|event| match event {
|
|
ThreadBufferedEvent::Notification(notification) => matches!(
|
|
server_notification_thread_target(notification),
|
|
ServerNotificationThreadTarget::Thread(event_thread_id)
|
|
if event_thread_id == thread_id
|
|
),
|
|
ThreadBufferedEvent::Request(request) => {
|
|
server_request_thread_id(request) == Some(thread_id)
|
|
}
|
|
ThreadBufferedEvent::HistoryEntryResponse(_)
|
|
| ThreadBufferedEvent::FeedbackSubmission(_) => true,
|
|
});
|
|
self.pending_app_server_requests.clear();
|
|
let mut unsupported_requests = Vec::new();
|
|
for event in &self.pending_primary_events {
|
|
if let ThreadBufferedEvent::Request(request) = event
|
|
&& let Some(unsupported) = self
|
|
.pending_app_server_requests
|
|
.note_server_request(request)
|
|
{
|
|
unsupported_requests.push(unsupported);
|
|
}
|
|
}
|
|
for unsupported in unsupported_requests {
|
|
self.pending_primary_events.retain(|event| {
|
|
!matches!(event, ThreadBufferedEvent::Request(request)
|
|
if request.id() == &unsupported.request_id)
|
|
});
|
|
if let Err(error) = self
|
|
.reject_app_server_request(
|
|
app_server,
|
|
unsupported.request_id,
|
|
unsupported.message,
|
|
)
|
|
.await
|
|
{
|
|
tracing::warn!("{error}");
|
|
}
|
|
}
|
|
if started.blocks_direct_input {
|
|
self.mark_primary_thread_parent_owned(thread_id);
|
|
}
|
|
// A full usage read can finish before thread/start. Apply its cached fallback
|
|
// after attachment but before the initial prompt or queued draft is submitted.
|
|
let recovery_was_pending = self.chat_widget.hold_rate_limit_recovery();
|
|
self.enqueue_primary_thread_session(started.session, started.turns)
|
|
.await?;
|
|
self.apply_backend_banner_fallback(app_server).await;
|
|
if !recovery_was_pending {
|
|
self.chat_widget.finish_rate_limit_recovery();
|
|
}
|
|
self.chat_widget.maybe_send_next_queued_input();
|
|
}
|
|
Err(err) if self.recover_transport_error(&err) => {}
|
|
Err(err) => {
|
|
let warnings = self
|
|
.transcript_cells
|
|
.iter()
|
|
.find_map(|cell| {
|
|
cell.as_any()
|
|
.downcast_ref::<history_cell::StartupWarningsCell>()
|
|
})
|
|
.filter(|cell| cell.pending_header)
|
|
.map(|cell| format!("\n\nStartup warnings:\n{}", cell.messages.join("\n")))
|
|
.unwrap_or_default();
|
|
return Err(color_eyre::eyre::eyre!(
|
|
"Failed to start a fresh session through the app server: {err}{warnings}"
|
|
));
|
|
}
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
pub(super) async fn start_fresh_session_with_summary_hint(
|
|
&mut self,
|
|
tui: &mut tui::Tui,
|
|
app_server: &mut AppServerSession,
|
|
session_start_source: Option<ThreadStartSource>,
|
|
initial_user_message: Option<crate::chatwidget::UserMessage>,
|
|
new_thread_name: Option<String>,
|
|
) {
|
|
// Start a fresh in-memory session while preserving resumability via persisted rollout
|
|
// history. If an initial message is provided, `enqueue_primary_thread_session` suppresses it
|
|
// until the new session is configured and any replayed turns have been rendered.
|
|
let mut config = match self.load_new_session_config(app_server).await {
|
|
Ok(config) => config,
|
|
Err(err) => {
|
|
if let Some(message) = initial_user_message {
|
|
self.chat_widget.restore_user_message_to_composer(message);
|
|
}
|
|
self.chat_widget
|
|
.add_error_message(format!("Failed to read new session defaults: {err}"));
|
|
tui.frame_requester().schedule_frame();
|
|
return;
|
|
}
|
|
};
|
|
apply_managed_new_thread_defaults(
|
|
&mut config,
|
|
app_server.managed_new_thread_defaults(),
|
|
&self.cli_kv_overrides,
|
|
&self.harness_overrides,
|
|
);
|
|
let summary = session_summary(
|
|
self.chat_widget.token_usage(),
|
|
self.chat_widget.thread_id(),
|
|
self.chat_widget.thread_name(),
|
|
self.chat_widget.rollout_path().as_deref(),
|
|
);
|
|
match app_server
|
|
.start_thread_with_session_start_source(
|
|
&self.local_settings,
|
|
&config,
|
|
session_start_source,
|
|
/*remote_cwd_override*/ None,
|
|
)
|
|
.await
|
|
{
|
|
Ok(mut started) => {
|
|
self.shutdown_current_thread(app_server).await;
|
|
let tracked_thread_ids: Vec<ThreadId> =
|
|
self.thread_event_channels.keys().copied().collect();
|
|
for thread_id in tracked_thread_ids {
|
|
if let Err(err) = app_server.thread_unsubscribe(thread_id).await {
|
|
tracing::warn!("failed to unsubscribe tracked thread {thread_id}: {err}");
|
|
}
|
|
}
|
|
self.local_settings = crate::local_settings::LocalSettings::from(&config);
|
|
self.config = config;
|
|
|
|
let name_error = if let Some(name) = new_thread_name {
|
|
match app_server
|
|
.thread_set_name(started.session.thread_id, name.clone())
|
|
.await
|
|
{
|
|
Ok(()) => {
|
|
started.session.thread_name = Some(name);
|
|
None
|
|
}
|
|
Err(err) => Some(format!("Failed to name the new session: {err}")),
|
|
}
|
|
} else {
|
|
None
|
|
};
|
|
if let Err(err) = self
|
|
.replace_chat_widget_with_app_server_thread(
|
|
tui,
|
|
started,
|
|
ThreadAttachPresentation::SessionLineage,
|
|
initial_user_message,
|
|
)
|
|
.await
|
|
{
|
|
self.chat_widget.add_error_message(format!(
|
|
"Failed to attach to fresh app-server thread: {err}"
|
|
));
|
|
} else {
|
|
if let Some(err) = name_error {
|
|
self.chat_widget.add_error_message(err);
|
|
}
|
|
if let Some(summary) = summary {
|
|
let mut lines: Vec<Line<'static>> = Vec::new();
|
|
if let Some(usage_line) = summary.usage_line {
|
|
lines.push(usage_line.into());
|
|
}
|
|
if let Some(command) = summary.resume_hint {
|
|
let spans =
|
|
vec!["To continue this session, run ".into(), command.cyan()];
|
|
lines.push(spans.into());
|
|
}
|
|
self.chat_widget.add_plain_history_lines(lines);
|
|
}
|
|
}
|
|
}
|
|
Err(err) => {
|
|
self.chat_widget.add_error_message(format!(
|
|
"Failed to start a fresh session through the app server: {err}"
|
|
));
|
|
if let Some(message) = initial_user_message {
|
|
self.chat_widget.restore_user_message_to_composer(message);
|
|
}
|
|
}
|
|
}
|
|
tui.frame_requester().schedule_frame();
|
|
}
|
|
|
|
pub(super) async fn replace_chat_widget_with_app_server_thread(
|
|
&mut self,
|
|
tui: &mut tui::Tui,
|
|
started: AppServerStartedThread,
|
|
presentation: ThreadAttachPresentation,
|
|
initial_user_message: Option<crate::chatwidget::UserMessage>,
|
|
) -> Result<()> {
|
|
// Initial messages are for freshly attached primary threads only. Thread switches and
|
|
// resume/fork flows pass `None` so they cannot replay old history and then auto-submit a new
|
|
// user turn by accident.
|
|
self.reset_thread_event_state();
|
|
let init = self.chatwidget_init_for_forked_or_resumed_thread(
|
|
tui,
|
|
self.config.clone(),
|
|
initial_user_message,
|
|
);
|
|
self.replace_chat_widget(ChatWidget::new_with_app_event(init));
|
|
self.chat_widget
|
|
.set_task_mentions_enabled(started.task_tools_available);
|
|
self.chat_widget
|
|
.note_rendered_width(tui.terminal.last_known_screen_size.width);
|
|
if started.blocks_direct_input {
|
|
self.mark_primary_thread_parent_owned(started.session.thread_id);
|
|
}
|
|
self.enqueue_primary_thread_session_with_presentation(
|
|
started.session,
|
|
started.turns,
|
|
presentation,
|
|
)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Fetches all loaded threads from the app server and registers descendants of the primary
|
|
/// thread in the navigation cache and chat widget metadata.
|
|
///
|
|
/// Called when opening the `/subagents` picker and after resuming a thread so that the picker and
|
|
/// keyboard navigation are pre-populated even if the TUI did not witness the original spawn
|
|
/// events. Fresh and forked threads cannot have pre-existing descendants.
|
|
///
|
|
/// The loaded-thread list is fetched in full (no pagination) and the spawn tree is walked
|
|
/// by `find_loaded_subagent_threads_for_primary`. Each discovered subagent is registered via
|
|
/// `upsert_agent_picker_thread`, which writes to both `AgentNavigationState` and the
|
|
/// `ChatWidget` metadata map.
|
|
pub(super) async fn backfill_loaded_subagent_threads(
|
|
&mut self,
|
|
app_server: &mut AppServerSession,
|
|
) -> LoadedSubagentBackfill {
|
|
let Some(primary_thread_id) = self.primary_thread_id else {
|
|
return LoadedSubagentBackfill::default();
|
|
};
|
|
|
|
let loaded_thread_ids = match app_server
|
|
.thread_loaded_list(ThreadLoadedListParams {
|
|
cursor: None,
|
|
limit: None,
|
|
})
|
|
.await
|
|
{
|
|
Ok(response) => response.data,
|
|
Err(err) => {
|
|
tracing::warn!(%err, "failed to list loaded threads for subagent backfill");
|
|
return LoadedSubagentBackfill::default();
|
|
}
|
|
};
|
|
|
|
let mut threads = Vec::new();
|
|
let mut had_read_error = false;
|
|
for thread_id in loaded_thread_ids {
|
|
let Ok(thread_id) = ThreadId::from_string(&thread_id) else {
|
|
tracing::warn!("ignoring loaded thread with invalid id during subagent backfill");
|
|
continue;
|
|
};
|
|
|
|
if thread_id == primary_thread_id {
|
|
continue;
|
|
}
|
|
|
|
match app_server
|
|
.thread_read(thread_id, /*include_turns*/ false)
|
|
.await
|
|
{
|
|
Ok(thread) => threads.push(thread),
|
|
Err(err) => {
|
|
had_read_error = true;
|
|
tracing::warn!(thread_id = %thread_id, %err, "failed to read loaded thread");
|
|
}
|
|
}
|
|
}
|
|
|
|
let mut refreshed_thread_ids = HashSet::new();
|
|
for thread in find_loaded_subagent_threads_for_primary(threads, primary_thread_id) {
|
|
let agent_path = thread.agent_path;
|
|
let has_live_channel = self
|
|
.thread_event_channels
|
|
.get(&thread.thread_id)
|
|
.is_some_and(|channel| channel.attachment() == ThreadEventAttachment::Live);
|
|
let is_closed = !has_live_channel && thread.is_closed;
|
|
if thread.blocks_direct_input {
|
|
self.agent_navigation.mark_parent_owned(thread.thread_id);
|
|
}
|
|
self.upsert_agent_picker_thread(
|
|
thread.thread_id,
|
|
thread.agent_nickname,
|
|
thread.agent_role,
|
|
is_closed,
|
|
);
|
|
self.agent_navigation
|
|
.set_agent_path(thread.thread_id, agent_path);
|
|
// A live channel can have an empty store after a successful spawn. Only apply server
|
|
// status for channels that would otherwise need another liveness read.
|
|
if !has_live_channel {
|
|
if thread.is_running {
|
|
self.agent_navigation.mark_running(thread.thread_id);
|
|
} else {
|
|
self.agent_navigation
|
|
.set_running(thread.thread_id, /*is_running*/ false);
|
|
}
|
|
refreshed_thread_ids.insert(thread.thread_id);
|
|
}
|
|
}
|
|
self.sync_active_agent_label();
|
|
|
|
LoadedSubagentBackfill {
|
|
completed: !had_read_error,
|
|
refreshed_thread_ids,
|
|
}
|
|
}
|
|
|
|
/// Returns the adjacent thread id for keyboard navigation, backfilling from the server if the
|
|
/// local cache has no neighbor.
|
|
///
|
|
/// Tries the fast path first: ask `AgentNavigationState` directly. If it returns `None` (no
|
|
/// adjacent entry exists, typically because the cache was never populated with remote
|
|
/// subagents), performs a full `backfill_loaded_subagent_threads` and retries. This ensures the
|
|
/// first next/previous keypress in a resumed remote session discovers subagents on demand
|
|
/// without requiring the user to wait for a proactive fetch.
|
|
pub(super) async fn adjacent_thread_id_with_backfill(
|
|
&mut self,
|
|
app_server: &mut AppServerSession,
|
|
direction: AgentNavigationDirection,
|
|
) -> Option<ThreadId> {
|
|
let current_thread = self.current_displayed_thread_id();
|
|
if let Some(thread_id) = self
|
|
.agent_navigation
|
|
.adjacent_thread_id(current_thread, direction)
|
|
{
|
|
return Some(thread_id);
|
|
}
|
|
|
|
let primary_thread_id = self.primary_thread_id?;
|
|
if self.last_subagent_backfill_attempt == Some(primary_thread_id) {
|
|
return None;
|
|
}
|
|
|
|
if self
|
|
.backfill_loaded_subagent_threads(app_server)
|
|
.await
|
|
.completed
|
|
{
|
|
self.last_subagent_backfill_attempt = Some(primary_thread_id);
|
|
}
|
|
self.agent_navigation
|
|
.adjacent_thread_id(self.current_displayed_thread_id(), direction)
|
|
}
|
|
|
|
pub(super) fn fresh_session_config(&self) -> Config {
|
|
let mut config = self.config.clone();
|
|
config.service_tier = self.chat_widget.configured_service_tier();
|
|
config
|
|
}
|
|
pub(super) async fn resume_target_session(
|
|
&mut self,
|
|
tui: &mut tui::Tui,
|
|
app_server: &mut AppServerSession,
|
|
target_session: SessionTarget,
|
|
) -> Result<AppRunControl> {
|
|
if self.ignore_same_thread_resume(&target_session) {
|
|
tui.frame_requester().schedule_frame();
|
|
return Ok(AppRunControl::Continue);
|
|
}
|
|
|
|
let (mut resume_config, local_settings) = match self
|
|
.resume_config_for_target(tui, app_server, &target_session)
|
|
.await
|
|
{
|
|
Ok(config) => config,
|
|
Err(control) => return Ok(control),
|
|
};
|
|
if self.reject_remote_resume_permission_override(&resume_config) {
|
|
return Ok(AppRunControl::Continue);
|
|
}
|
|
let baseline_approval = resume_config.permissions.approval_policy.value();
|
|
let baseline_permissions = RuntimePermissionProfileOverride::from_config(&resume_config);
|
|
self.apply_runtime_policy_overrides(&mut resume_config, RuntimePolicyOverrideScope::All);
|
|
|
|
let summary = session_summary(
|
|
self.chat_widget.token_usage(),
|
|
self.chat_widget.thread_id(),
|
|
self.chat_widget.thread_name(),
|
|
self.chat_widget.rollout_path().as_deref(),
|
|
);
|
|
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(
|
|
&local_settings,
|
|
resume_config.clone(),
|
|
target_session.thread_id,
|
|
self.resume_model_settings(),
|
|
)
|
|
.await;
|
|
let (resumed, read_only) = match resumed {
|
|
Ok(resumed) => (resumed, false),
|
|
Err(err) if crate::app_server_session::is_active_writer_error(&err) => match app_server
|
|
.read_thread_for_viewing(&resume_config, &local_settings, target_session.thread_id)
|
|
.await
|
|
{
|
|
Ok(thread) => (thread, true),
|
|
Err(read_err) => {
|
|
self.add_session_picker_error(format!(
|
|
"Failed to view thread open elsewhere: {read_err}"
|
|
));
|
|
return Ok(AppRunControl::Continue);
|
|
}
|
|
},
|
|
Err(err) => {
|
|
let path_display = target_session.display_label();
|
|
self.add_session_picker_error(format!(
|
|
"Failed to resume session from {path_display}: {err}"
|
|
));
|
|
return Ok(AppRunControl::Continue);
|
|
}
|
|
};
|
|
let resumed_thread_id = resumed.session.thread_id;
|
|
let retained_input = (self.chat_widget.is_external_writer_view()
|
|
&& self.chat_widget.thread_id() == Some(resumed_thread_id))
|
|
.then(|| self.chat_widget.capture_thread_input_state())
|
|
.flatten();
|
|
if self.chat_widget.thread_id() == Some(resumed_thread_id) {
|
|
// The successful resume has already subscribed this connection to the same thread.
|
|
self.shutdown_side_threads(app_server).await;
|
|
} else {
|
|
self.shutdown_current_thread(app_server).await;
|
|
}
|
|
self.local_settings = local_settings;
|
|
self.config = resume_config;
|
|
tui.set_notification_settings(
|
|
self.local_settings.tui.notification_settings.method,
|
|
self.local_settings.tui.notification_settings.condition,
|
|
);
|
|
self.file_search
|
|
.update_search_dir(self.config.cwd.to_path_buf());
|
|
match self
|
|
.replace_chat_widget_with_app_server_thread(
|
|
tui,
|
|
resumed,
|
|
ThreadAttachPresentation::SessionLineage,
|
|
/*initial_user_message*/ None,
|
|
)
|
|
.await
|
|
{
|
|
Ok(()) => {
|
|
if let Some(input) = retained_input {
|
|
self.chat_widget.restore_thread_input_state(
|
|
Some(input),
|
|
ThreadInputStateRestoreMode {
|
|
preserve_in_flight_turn: false,
|
|
},
|
|
);
|
|
}
|
|
if read_only {
|
|
self.ensure_thread_channel(resumed_thread_id)
|
|
.mark_external_writer();
|
|
self.chat_widget.show_external_writer_thread();
|
|
}
|
|
if self.app_server_target.uses_remote_workspace() {
|
|
let config = self.chat_widget.config_ref();
|
|
let approval = config.permissions.approval_policy.value();
|
|
self.runtime_approval_policy_override = (approval != baseline_approval)
|
|
.then_some(RuntimeApprovalPolicyOverride::Restored(approval.into()));
|
|
self.runtime_permission_profile_override = (!baseline_permissions
|
|
.matches_config(config))
|
|
.then(|| RuntimePermissionProfileOverride::from_restored_config(config));
|
|
}
|
|
self.backfill_loaded_subagent_threads(app_server).await;
|
|
if !read_only {
|
|
self.replay_agents_overview_requests(app_server, resumed_thread_id)
|
|
.await;
|
|
}
|
|
if let Some(summary) = summary {
|
|
let mut lines: Vec<Line<'static>> = Vec::new();
|
|
if let Some(usage_line) = summary.usage_line {
|
|
lines.push(usage_line.into());
|
|
}
|
|
if let Some(command) = summary.resume_hint {
|
|
let spans = vec!["To continue this session, run ".into(), command.cyan()];
|
|
lines.push(spans.into());
|
|
}
|
|
self.chat_widget.add_plain_history_lines(lines);
|
|
}
|
|
if !read_only {
|
|
self.maybe_prompt_resume_paused_goal_after_resume(
|
|
app_server,
|
|
resumed_thread_id,
|
|
)
|
|
.await;
|
|
}
|
|
}
|
|
Err(err) => {
|
|
self.add_session_picker_error(format!(
|
|
"Failed to attach to resumed app-server thread: {err}"
|
|
));
|
|
}
|
|
}
|
|
|
|
Ok(AppRunControl::Continue)
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
#[test]
|
|
fn terminal_thread_read_error_detection_matches_not_loaded_errors() {
|
|
let err = color_eyre::eyre::eyre!(
|
|
"thread/read failed during TUI session lookup: thread/read failed: thread not loaded: thr_123"
|
|
);
|
|
|
|
assert!(App::is_terminal_thread_read_error(&err));
|
|
}
|
|
|
|
#[test]
|
|
fn terminal_thread_read_error_detection_ignores_transient_failures() {
|
|
let err = color_eyre::eyre::eyre!(
|
|
"thread/read failed during TUI session lookup: thread/read transport error: broken pipe"
|
|
);
|
|
|
|
assert!(!App::is_terminal_thread_read_error(&err));
|
|
}
|
|
|
|
#[test]
|
|
fn closed_state_for_thread_read_error_preserves_live_state_without_cache_on_transient_error() {
|
|
let err = color_eyre::eyre::eyre!(
|
|
"thread/read failed during TUI session lookup: thread/read transport error: broken pipe"
|
|
);
|
|
|
|
assert!(!App::closed_state_for_thread_read_error(
|
|
&err, /*existing_is_closed*/ None
|
|
));
|
|
}
|
|
|
|
#[test]
|
|
fn closed_state_for_thread_read_error_marks_terminal_uncached_threads_closed() {
|
|
let err = color_eyre::eyre::eyre!(
|
|
"thread/read failed during TUI session lookup: thread/read failed: thread not loaded: thr_123"
|
|
);
|
|
|
|
assert!(App::closed_state_for_thread_read_error(
|
|
&err, /*existing_is_closed*/ None
|
|
));
|
|
}
|
|
|
|
#[test]
|
|
fn include_turns_fallback_detection_handles_unmaterialized_and_ephemeral_threads() {
|
|
let unmaterialized = color_eyre::eyre::eyre!(
|
|
"thread/read failed during TUI session lookup: thread/read failed: thread thr_123 is not materialized yet; includeTurns is unavailable before first user message"
|
|
);
|
|
let ephemeral = color_eyre::eyre::eyre!(
|
|
"thread/read failed during TUI session lookup: thread/read failed: ephemeral threads do not support includeTurns"
|
|
);
|
|
|
|
assert!(App::can_fallback_from_include_turns_error(&unmaterialized));
|
|
assert!(App::can_fallback_from_include_turns_error(&ephemeral));
|
|
}
|
|
}
|