openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-07 20:30:33 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
57e64682f8ef83c09cc868c39010733dd6760805
codex/codex-rs/tui/src/app/thread_routing.rs
T
Felipe Coury 57e64682f8 Persist local workspace side chats and add tab close controls (#60152) 
## What changed

- Save side chats as attachments to their parent conversation when local workspace history supports them, and let users reopen them from the sidebar after a restart.
- Show chat names or first-prompt previews in sidebar labels, and allow saved side chats to be renamed with `/rename`.
- Add tab close buttons, middle-click closing, and the configurable `close_sidebar_tab` shortcut, defaulting to `Ctrl-X W`. Closing a saved chat keeps it available in the picker; archiving removes it from Side chats.
- Confirm actions that stop running work, discard unsent input, or lose temporary history. Delete saved TUI side chats when deleting their parent conversation.

## Testing

Add coverage for reopening saved side chats after restart, archive handling, close confirmations, and preventing close-confirmation keys from accepting a replacement approval dialog.

GitOrigin-RevId: 3d47533d007286a7354660c08df13a83dd48d5d7
2026-11-07 17:31:27 +00:00

2274 lines
92 KiB
Rust
Raw Blame History

//! Thread routing, buffering, and app-server operation submission for the TUI app.
//!
//! This module manages active thread channels, routes server requests and notifications into those
//! channels, submits thread-scoped operations through the app server, and replays buffered events
//! when the visible thread changes.
use super::session_lifecycle::ThreadAttachPresentation;
use super::*;
use crate::app_event::ThreadTitleDestination;
use crate::chatwidget::ThreadInputStateRestoreMode;
use codex_app_server_protocol::ThreadStartedNotification;
use codex_app_server_protocol::TurnInterruptParams;
use codex_app_server_protocol::TurnInterruptResponse;
use codex_app_server_protocol::WarningNotification;
// Leave time for side-thread cleanup and unsubscribe inside the two-second exit budget.
const REALTIME_STOP_TIMEOUT: Duration = Duration::from_secs(/*secs*/ 1);
impl App {
pub(super) async fn stop_realtime_conversation(&mut self, app_server: &mut AppServerSession) {
let thread_id = self
.background_voice
.as_mut()
.and_then(|owner| owner.reset_realtime_conversation())
.or_else(|| self.chat_widget.reset_realtime_conversation());
self.retire_background_voice();
let Some(thread_id) = thread_id else {
return;
};
match tokio::time::timeout(
REALTIME_STOP_TIMEOUT,
app_server.thread_realtime_stop(thread_id),
)
.await
{
Ok(Ok(())) => {}
Ok(Err(error)) => {
tracing::warn!(%thread_id, %error, "failed to stop voice conversation");
}
Err(_) => {
tracing::warn!(%thread_id, "timed out stopping voice conversation");
}
}
}
pub(super) async fn shutdown_current_thread(&mut self, app_server: &mut AppServerSession) {
self.stop_realtime_conversation(app_server).await;
self.detach_current_thread_for_navigation(app_server, /*destination*/ None)
.await;
}
pub(super) async fn shutdown_side_threads(
&mut self,
app_server: &mut AppServerSession,
root: Option<ThreadId>,
) -> bool {
if self.side_threads.is_empty() {
return true;
}
let subtree = root.map(|id| self.known_thread_subtree(id));
let ids: Vec<_> = self
.side_threads
.iter()
.filter(|(_, side)| {
subtree
.as_ref()
.is_none_or(|ids| ids.contains(&side.parent_thread_id))
})
.map(|(id, _)| *id)
.collect();
let mut closed = true;
for id in ids {
closed &= self.discard_side_thread(app_server, id).await;
}
closed
}
pub(super) fn abort_thread_event_listener(&mut self, thread_id: ThreadId) {
if let Some(handle) = self.thread_event_listener_tasks.remove(&thread_id) {
handle.abort();
}
}
pub(super) fn abort_all_thread_event_listeners(&mut self) {
for handle in self
.thread_event_listener_tasks
.drain()
.map(|(_, handle)| handle)
{
handle.abort();
}
}
pub(super) fn ensure_thread_channel(&mut self, thread_id: ThreadId) -> &mut ThreadEventChannel {
self.thread_event_channels
.entry(thread_id)
.or_insert_with(|| ThreadEventChannel::new(THREAD_EVENT_CHANNEL_CAPACITY))
}
pub(super) async fn set_thread_active(&mut self, thread_id: ThreadId, active: bool) {
if let Some(channel) = self.thread_event_channels.get_mut(&thread_id) {
let mut store = channel.store.lock().await;
store.active = active;
}
}
pub(super) async fn activate_thread_channel(&mut self, thread_id: ThreadId) {
if self.active_thread_id.is_some() {
return;
}
self.set_thread_active(thread_id, /*active*/ true).await;
let receiver = if let Some(channel) = self.thread_event_channels.get_mut(&thread_id) {
channel.receiver.take()
} else {
None
};
if let Some(startup_pane) = self.sidebars.remove(&None)
&& let std::collections::hash_map::Entry::Vacant(entry) =
self.sidebars.entry(Some(thread_id))
{
entry.insert(startup_pane);
if let Some(mode) = &mut self.pane_resize
&& mode.parent.is_none()
{
mode.parent = Some(thread_id);
}
}
self.active_thread_id = Some(thread_id);
self.active_thread_rx = receiver;
self.refresh_pending_thread_approvals().await;
}
pub(super) async fn store_active_thread_receiver(&mut self) {
let Some(active_id) = self.active_thread_id else {
return;
};
let input_state = self.chat_widget.capture_thread_input_state();
let recap_progress = self.recap.progress();
if let Some(channel) = self.thread_event_channels.get_mut(&active_id) {
let receiver = self.active_thread_rx.take();
let mut store = channel.store.lock().await;
if store.active {
store.replay_may_have_output = false;
}
store.active = false;
store.input_state = input_state;
store.merge_recap_progress(recap_progress);
if let Some(receiver) = receiver {
channel.receiver = Some(receiver);
}
}
}
pub(super) fn retain_subscriptions_for_fork(&mut self) {
// Tasks can start while the fork RPC is pending.
for id in self
.thread_event_channels
.keys()
.copied()
.chain(self.agent_navigation.tracked_thread_ids())
.filter(|id| !self.side_threads.contains_key(id))
{
self.agents_overview
.dispatched_requests
.entry(id)
.or_default();
}
}
pub(super) async fn activate_thread_for_replay(
&mut self,
thread_id: ThreadId,
) -> Option<(mpsc::Receiver<ThreadBufferedEvent>, ThreadEventSnapshot)> {
let channel = self.thread_event_channels.get_mut(&thread_id)?;
let receiver = channel.receiver.take()?;
let mut store = channel.store.lock().await;
store.active = true;
let snapshot = store.snapshot();
Some((receiver, snapshot))
}
pub(super) async fn clear_active_thread(&mut self) {
if let Some(active_id) = self.active_thread_id.take() {
self.set_thread_active(active_id, /*active*/ false).await;
}
self.active_thread_rx = None;
self.refresh_pending_thread_approvals().await;
}
pub(super) async fn note_thread_outbound_op(&mut self, thread_id: ThreadId, op: &AppCommand) {
let Some(channel) = self.thread_event_channels.get(&thread_id) else {
return;
};
let mut store = channel.store.lock().await;
store.note_outbound_op(op);
}
pub(super) async fn note_active_thread_outbound_op(&mut self, op: &AppCommand) {
if !ThreadEventStore::op_can_change_pending_replay_state(op) {
return;
}
let Some(thread_id) = self.active_thread_id else {
return;
};
self.note_thread_outbound_op(thread_id, op).await;
}
pub(super) async fn active_turn_id_for_thread(&self, thread_id: ThreadId) -> Option<String> {
let channel = self.thread_event_channels.get(&thread_id)?;
let store = channel.store.lock().await;
store.active_turn_id().map(ToOwned::to_owned)
}
pub(super) fn thread_label(&self, thread_id: ThreadId) -> String {
let is_primary = self.primary_thread_id == Some(thread_id);
let anonymous = |label: &str| {
// Agents spawned together share the timestamp prefix of their UUIDv7 IDs.
let id = thread_id.to_string();
let suffix = id.rsplit('-').next().unwrap_or(&id);
format!("{label} ({suffix})")
};
if let Some(entry) = self.agent_navigation.get(&thread_id) {
if let Some(path) = entry.agent_path.as_deref().map(str::trim)
&& !is_primary
&& !path.is_empty()
{
return path.to_string();
}
let label = format_agent_picker_item_name(
entry.agent_nickname.as_deref(),
entry.agent_role.as_deref(),
is_primary,
);
if !is_primary
&& entry
.agent_nickname
.as_deref()
.is_none_or(|name| name.trim().is_empty())
{
anonymous(&label)
} else {
label
}
} else if is_primary {
"Main [default]".to_string()
} else {
anonymous("Agent")
}
}
/// Returns the thread whose transcript is currently on screen.
///
/// `active_thread_id` is the source of truth during steady state, but the widget can briefly
/// lag behind thread bookkeeping during transitions. The footer label and adjacent-thread
/// navigation both follow what the user is actually looking at, not whichever thread most
/// recently began switching.
pub(super) fn current_displayed_thread_id(&self) -> Option<ThreadId> {
self.active_thread_id.or(self.chat_widget.thread_id())
}
pub(super) fn ignore_same_thread_resume(
&mut self,
target_session: &crate::resume_picker::SessionTarget,
) -> bool {
if self.chat_widget.has_provider_error()
|| self.active_thread_id != Some(target_session.thread_id)
|| self
.thread_event_channels
.get(&target_session.thread_id)
.is_some_and(|channel| channel.attachment() != ThreadEventAttachment::Live)
{
return false;
};
self.chat_widget.add_info_message(
format!("Already viewing {}.", target_session.display_label()),
/*hint*/ None,
);
true
}
/// Mirrors the visible thread into the contextual footer row.
///
/// The footer sometimes shows ambient context instead of an instructional hint. In multi-agent
/// sessions, that contextual row includes the currently viewed agent label. The label is
/// intentionally hidden until there is more than one known thread so single-thread sessions do
/// not spend footer space restating that the user is already on the main conversation.
pub(super) fn sync_active_agent_label(&mut self) {
let label = self
.agent_navigation
.active_agent_label(self.current_displayed_thread_id(), self.primary_thread_id);
self.chat_widget.set_active_agent_label(label);
self.sync_side_thread_ui();
}
pub(super) async fn thread_cwd(&self, thread_id: ThreadId) -> Option<AbsolutePathBuf> {
let channel = self.thread_event_channels.get(&thread_id)?;
let store = channel.store.lock().await;
store.session.as_ref().map(|session| session.cwd.clone())
}
pub(super) async fn interactive_request_for_thread_request(
&self,
thread_id: ThreadId,
request: &ServerRequest,
) -> std::io::Result<Option<ThreadInteractiveRequest>> {
let thread_label = Some(self.thread_label(thread_id));
Ok(match request {
ServerRequest::CommandExecutionRequestApproval { params, .. } => {
let network_approval_context = params.network_approval_context.clone();
let additional_permissions = params.additional_permissions.clone();
let proposed_execpolicy_amendment = params.proposed_execpolicy_amendment.clone();
let proposed_network_policy_amendments =
params.proposed_network_policy_amendments.clone();
let approval = ExecApprovalRequest {
kind: params.kind,
thread_id,
thread_label,
id: params
.approval_id
.clone()
.unwrap_or_else(|| params.item_id.clone()),
environment_id: params.environment_id.clone(),
command: params
.command
.as_deref()
.map(split_command_string)
.unwrap_or_default(),
reason: params.reason.clone(),
available_decisions: params.available_decisions.clone().unwrap_or_else(|| {
default_exec_approval_decisions(
network_approval_context.as_ref(),
proposed_execpolicy_amendment.as_ref(),
proposed_network_policy_amendments.as_deref(),
additional_permissions.as_ref(),
)
}),
network_approval_context,
additional_permissions,
};
Some(ThreadInteractiveRequest::Approval(ApprovalRequest::Exec(
approval,
)))
}
ServerRequest::FileChangeRequestApproval { params, .. } => Some(
ThreadInteractiveRequest::Approval(ApprovalRequest::ApplyPatch(
ApplyPatchApprovalRequest {
thread_id,
thread_label,
id: params.item_id.clone(),
reason: params.reason.clone(),
cwd: self
.thread_cwd(thread_id)
.await
.unwrap_or_else(|| self.config.cwd.clone()),
changes: self
.thread_file_change_changes(thread_id, &params.turn_id, &params.item_id)
.await
.map(crate::app_server_approval_conversions::file_update_changes_to_display)
.unwrap_or_default(),
},
)),
),
ServerRequest::McpServerElicitationRequest { request_id, params } => {
if let Some(params) = AppLinkViewParams::from_url_app_server_request(
thread_id,
&params.server_name,
request_id.clone(),
&params.request,
) {
Some(ThreadInteractiveRequest::AppLink(params))
} else if let Some(request) =
McpServerElicitationFormRequest::from_app_server_request(
thread_id,
request_id.clone(),
params,
)
{
Some(ThreadInteractiveRequest::McpServerElicitation(request))
} else {
match &params.request {
codex_app_server_protocol::McpServerElicitationRequest::Form {
message,
..
} => Some(ThreadInteractiveRequest::Approval(
ApprovalRequest::McpElicitation(McpElicitationApprovalRequest {
thread_id,
thread_label,
server_name: params.server_name.clone(),
request_id: request_id.clone(),
message: message.clone(),
}),
)),
codex_app_server_protocol::McpServerElicitationRequest::UserVerification {
title, description, ..
} => Some(ThreadInteractiveRequest::UserVerification {
thread_id,
request: crate::bottom_pane::user_verification::UserVerificationRequest {
thread_label,
server_name: params.server_name.clone(),
request_id: request_id.clone(),
title: title.clone(),
description: description.clone(),
},
}),
codex_app_server_protocol::McpServerElicitationRequest::OpenAiForm {
..
}
| codex_app_server_protocol::McpServerElicitationRequest::OpenAiElicitationForm { .. }
| codex_app_server_protocol::McpServerElicitationRequest::Url { .. } => {
self.app_event_tx.resolve_elicitation(
thread_id,
params.server_name.clone(),
request_id.clone(),
codex_app_server_protocol::McpServerElicitationAction::Decline,
/*content*/ None,
/*meta*/ None,
);
None
}
}
}
}
ServerRequest::PermissionsRequestApproval { params, .. } => {
// TODO(anp): Remove this native-path localization error path once core permission
// paths remain PathUri after crossing the app-server boundary.
let permissions = params.permissions.clone().try_into().map_err(|err| {
std::io::Error::new(
std::io::ErrorKind::InvalidInput,
format!("failed to localize requested filesystem paths: {err}"),
)
})?;
Some(ThreadInteractiveRequest::Approval(
ApprovalRequest::Permissions(PermissionsApprovalRequest {
thread_id,
thread_label,
call_id: params.item_id.clone(),
environment_id: params.environment_id.clone(),
reason: params.reason.clone(),
permissions,
}),
))
}
_ => None,
})
}
pub(super) fn push_thread_interactive_request(&mut self, request: ThreadInteractiveRequest) {
if self.chat_widget.has_misalignment_policy_violation() {
return;
}
match request {
ThreadInteractiveRequest::AppLink(params) => {
self.chat_widget.open_app_link_view(params);
}
ThreadInteractiveRequest::Approval(request) => {
self.render_inactive_patch_preview(&request);
self.chat_widget.push_approval_request(request);
if self.startup_protected_input_boundary && !self.chat_widget.has_active_modal() {
self.startup_pending_protected_request = true;
}
}
ThreadInteractiveRequest::UserVerification { thread_id, request } => {
self.chat_widget
.push_user_verification_request(thread_id, request);
}
ThreadInteractiveRequest::McpServerElicitation(request) => {
self.chat_widget
.push_mcp_server_elicitation_request(request);
}
}
}
fn render_inactive_patch_preview(&mut self, request: &ApprovalRequest) {
let ApprovalRequest::ApplyPatch(request) = request else {
return;
};
if request.thread_label.is_none() || request.changes.is_empty() {
return;
}
self.chat_widget
.add_to_history(history_cell::new_patch_event(
request.changes.clone(),
&request.cwd,
));
}
pub(super) async fn pending_inactive_thread_requests(&self) -> Vec<(ThreadId, ServerRequest)> {
let channels: Vec<(ThreadId, Arc<Mutex<ThreadEventStore>>)> = self
.thread_event_channels
.iter()
.map(|(thread_id, channel)| (*thread_id, Arc::clone(&channel.store)))
.collect();
let mut requests = Vec::new();
for (thread_id, store) in channels {
if Some(thread_id) == self.active_thread_id {
continue;
}
let store = store.lock().await;
if store.side_parent_pending_status().is_none() {
continue;
}
requests.extend(
store
.pending_replay_requests()
.into_iter()
.map(|request| (thread_id, request)),
);
}
requests
}
pub(super) async fn surface_pending_inactive_thread_interactive_requests(
&mut self,
) -> Result<()> {
if self.active_side_parent_thread_id().is_some() {
return Ok(());
}
let requests = self.pending_inactive_thread_requests().await;
for (thread_id, request) in requests {
if let Some(request) = self
.interactive_request_for_thread_request(thread_id, &request)
.await?
{
self.push_thread_interactive_request(request);
}
}
Ok(())
}
pub(super) async fn submit_active_thread_op(
&mut self,
app_server: &mut AppServerSession,
op: AppCommand,
) -> Result<()> {
let Some(thread_id) = self.active_thread_id else {
if let AppCommand::RealtimeConversationSpeech { delivery_id, .. } = &op {
self.chat_widget
.restore_undelivered_realtime_speech(*delivery_id);
}
if let AppCommand::RealtimeConversationStart { thread_id, .. }
| AppCommand::RealtimeConversationStop { thread_id } = &op
&& self.chat_widget.thread_id() == Some(*thread_id)
{
self.chat_widget
.record_realtime_failure(crate::chatwidget::RealtimeFailureCause::LocalRequest);
self.chat_widget.reset_realtime_conversation();
}
self.chat_widget
.add_error_message("No active thread is available.".to_string());
return Ok(());
};
self.submit_thread_op(app_server, thread_id, op).await
}
pub(super) async fn submit_thread_op(
&mut self,
app_server: &mut AppServerSession,
thread_id: ThreadId,
op: AppCommand,
) -> Result<()> {
if self.thread_unavailable(thread_id) {
if let AppCommand::RealtimeConversationSpeech { delivery_id, .. } = &op {
self.chat_widget
.restore_undelivered_realtime_speech(*delivery_id);
}
if let AppCommand::RealtimeConversationStart { thread_id, .. }
| AppCommand::RealtimeConversationStop { thread_id } = &op
&& self.chat_widget.thread_id() == Some(*thread_id)
{
self.chat_widget
.record_realtime_failure(crate::chatwidget::RealtimeFailureCause::LocalRequest);
self.chat_widget.reset_realtime_conversation();
}
self.chat_widget.add_error_message(
"This conversation is read-only or unavailable; no operation was sent.".into(),
);
return Ok(());
}
if self.chat_widget.rejects_misalignment_policy_op(&op) {
if let AppCommand::RealtimeConversationSpeech { delivery_id, .. } = &op {
self.chat_widget
.restore_undelivered_realtime_speech(*delivery_id);
}
return Ok(());
}
crate::session_log::log_outbound_op(&op);
if self.try_resolve_app_server_request(thread_id, &op).await? {
return Ok(());
}
let submitted = match self
.try_submit_active_thread_op_via_app_server(app_server, thread_id, &op)
.await
{
Ok(submitted) => submitted,
Err(error) => {
if let Some(provider_error) =
crate::app_server_session::model_provider_error(&error)
&& self.current_displayed_thread_id() == Some(thread_id)
{
self.chat_widget.show_provider_error(
provider_error,
match &op {
AppCommand::UserTurn {
client_user_message_id,
..
} => Some(client_user_message_id.as_str()),
_ => None,
},
);
return Ok(());
}
return Err(error);
}
};
if submitted {
if ThreadEventStore::op_can_change_pending_replay_state(&op) {
self.note_thread_outbound_op(thread_id, &op).await;
self.refresh_pending_thread_approvals().await;
self.refresh_side_parent_status_from_store(thread_id).await;
}
return Ok(());
}
self.chat_widget
.add_error_message(format!("Not available in TUI yet for thread {thread_id}."));
Ok(())
}
/// Persist prompt text in the local cross-session message history.
pub(super) fn append_message_history_entry(&self, thread_id: ThreadId, text: String) {
let history_config = codex_message_history::HistoryConfig::new(
self.local_settings.codex_home.clone(),
&self.local_settings.history,
);
tokio::spawn(async move {
if let Err(err) =
codex_message_history::append_entry(&text, thread_id, &history_config).await
{
tracing::warn!(
thread_id = %thread_id,
error = %err,
"failed to append to message history"
);
}
});
}
/// Fetch one local cross-session message history entry for the requesting thread.
pub(super) async fn lookup_message_history_entry(
&mut self,
thread_id: ThreadId,
offset: usize,
log_id: u64,
) -> Result<()> {
let history_config = codex_message_history::HistoryConfig::new(
self.local_settings.codex_home.clone(),
&self.local_settings.history,
);
let app_event_tx = self.app_event_tx.clone();
tokio::spawn(async move {
let entry_opt = tokio::task::spawn_blocking(move || {
codex_message_history::lookup(log_id, offset, &history_config)
})
.await
.unwrap_or_else(|err| {
tracing::warn!(error = %err, "history lookup task failed");
None
});
app_event_tx.send(AppEvent::ThreadHistoryEntryResponse {
thread_id,
event: HistoryLookupResponse::Entry {
offset,
log_id,
entry: entry_opt.map(|entry| entry.text),
},
});
});
Ok(())
}
/// Fetch one bounded local cross-session history batch for the requesting thread.
pub(super) async fn lookup_message_history_batch(
&mut self,
thread_id: ThreadId,
cursor: codex_message_history::HistoryBatchCursor,
log_id: u64,
) -> Result<()> {
let history_config = codex_message_history::HistoryConfig::new(
self.local_settings.codex_home.clone(),
&self.local_settings.history,
);
let app_event_tx = self.app_event_tx.clone();
tokio::spawn(async move {
let event = match tokio::task::spawn_blocking(move || {
codex_message_history::lookup_batch(log_id, cursor, &history_config)
})
.await
{
Ok(Ok(batch)) => {
let entries = batch
.entries
.into_iter()
.map(|entry| crate::app_event::HistoryBatchEntryResponse {
offset: entry.offset,
entry: entry.entry.map(|entry| entry.text),
})
.collect();
HistoryLookupResponse::Batch {
cursor,
log_id,
entries,
next_older_cursor: batch.next_older_cursor,
}
}
Ok(Err(err)) => {
tracing::warn!(error = %err, "history batch lookup failed");
HistoryLookupResponse::BatchError { cursor, log_id }
}
Err(err) => {
tracing::warn!(error = %err, "history batch lookup task failed");
HistoryLookupResponse::BatchError { cursor, log_id }
}
};
app_event_tx.send(AppEvent::ThreadHistoryEntryResponse { thread_id, event });
});
Ok(())
}
pub(super) async fn try_submit_active_thread_op_via_app_server(
&mut self,
app_server: &mut AppServerSession,
thread_id: ThreadId,
op: &AppCommand,
) -> Result<bool> {
if user_turn_submission::try_submit_user_turn(self, app_server, thread_id, op).await? {
return Ok(true);
}
match op {
AppCommand::Interrupt => {
let mut turn_id = self
.active_turn_id_for_thread(thread_id)
.await
.unwrap_or_default();
let (thread_event_tx, thread_event_store) = {
let channel = self.ensure_thread_channel(thread_id);
(channel.sender.clone(), Arc::clone(&channel.store))
};
self.reset_backtrack_state();
if !turn_id.is_empty() {
let mut store = thread_event_store.lock().await;
if store.pending_interrupt_turn_id.as_deref() == Some(turn_id.as_str()) {
return Ok(true);
}
store.pending_interrupt_turn_id = Some(turn_id.clone());
}
let request_handle = app_server.request_handle();
let request_ids = [app_server.next_request_id(), app_server.next_request_id()];
tokio::spawn(async move {
for (attempt, request_id) in request_ids.into_iter().enumerate() {
let result = request_handle
.request_typed::<TurnInterruptResponse>(ClientRequest::TurnInterrupt {
request_id,
params: TurnInterruptParams {
thread_id: thread_id.to_string(),
turn_id: turn_id.clone(),
},
})
.await;
match result {
Ok(_) => break,
Err(error) => {
if attempt == 0
&& let Some(actual_turn_id) = active_turn_interrupt_race(&error)
&& actual_turn_id != turn_id
{
thread_event_store.lock().await.pending_interrupt_turn_id =
Some(actual_turn_id.clone());
turn_id = actual_turn_id;
continue;
}
tracing::warn!(error = %error, "turn/interrupt failed in TUI");
let notification =
ServerNotification::Warning(WarningNotification {
thread_id: Some(thread_id.to_string()),
message: format!("Failed to interrupt turn: {error}"),
});
let should_send = {
let mut store = thread_event_store.lock().await;
store.push_notification_ref(&notification);
store.active
};
if should_send
&& let Err(error) = thread_event_tx
.send(ThreadBufferedEvent::Notification(Box::new(
notification,
)))
.await
{
tracing::warn!(error = %error, "thread event channel closed");
}
break;
}
}
}
});
Ok(true)
}
AppCommand::ListSkills { cwds, force_reload } => {
// A queued refresh can outlive the workspace it was requested for.
if !cwds
.iter()
.any(|cwd| cwd == self.chat_widget.config_ref().cwd.as_path())
{
return Ok(true);
}
self.handle_skills_list_result(
app_server
.skills_list(codex_app_server_protocol::SkillsListParams {
cwds: cwds.clone(),
force_reload: *force_reload,
})
.await,
"failed to refresh skills",
);
Ok(true)
}
AppCommand::Compact => {
app_server.thread_compact_start(thread_id).await?;
Ok(true)
}
AppCommand::SetThreadName { name } => {
let name = name.to_string();
app_server.thread_set_name(thread_id, name.clone()).await?;
self.chat_widget.expect_manual_thread_name(thread_id, name);
self.cancel_thread_title_generation(thread_id);
Ok(true)
}
AppCommand::Review { target } => {
let response = app_server.review_start(thread_id, target.clone()).await?;
let review_thread_id = ThreadId::from_string(&response.review_thread_id)
.wrap_err("review/start returned invalid review thread id")?;
let store = Arc::clone(&self.ensure_thread_channel(review_thread_id).store);
let mut store = store.lock().await;
store.set_active_turn_id(response.turn.id);
self.chat_widget.on_review_started();
Ok(true)
}
AppCommand::CleanBackgroundTerminals => {
app_server
.thread_background_terminals_clean(thread_id)
.await?;
Ok(true)
}
AppCommand::RealtimeConversationStart {
thread_id: realtime_thread_id,
offer_sdp,
} => {
if *realtime_thread_id != thread_id {
return Ok(true);
}
let model = self
.chat_widget
.config_ref()
.experimental_realtime_ws_model
.clone();
let voices = self.realtime_voices(app_server).await?;
let voice = self.effective_realtime_voice(app_server, &voices).await?;
app_server
.thread_realtime_start(
*realtime_thread_id,
String::from(offer_sdp.clone()),
model,
voice,
)
.await?;
Ok(true)
}
AppCommand::RealtimeConversationStop {
thread_id: realtime_thread_id,
} => {
app_server.thread_realtime_stop(*realtime_thread_id).await?;
Ok(true)
}
AppCommand::RealtimeConversationSpeech {
thread_id: realtime_thread_id,
attempt_id,
input_generation,
delivery_id,
text,
} => {
if !self.chat_widget.has_pending_realtime_speech(*delivery_id) {
return Ok(true);
}
if *realtime_thread_id != thread_id
|| !self.chat_widget.is_current_realtime_attempt(
*realtime_thread_id,
*attempt_id,
*input_generation,
)
{
self.chat_widget
.restore_undelivered_realtime_speech(*delivery_id);
return Ok(true);
}
app_server
.thread_realtime_append_speech(*realtime_thread_id, text.as_str().to_owned())
.await?;
self.chat_widget.accept_realtime_speech(*delivery_id);
Ok(true)
}
AppCommand::RunUserShellCommand { command } => {
app_server
.thread_shell_command(thread_id, command.to_string())
.await?;
Ok(true)
}
AppCommand::ReloadUserConfig => {
app_server.reload_user_config().await?;
Ok(true)
}
AppCommand::OverrideTurnContext { .. } => {
self.sync_override_turn_context_settings(app_server, thread_id, op)
.await;
Ok(true)
}
AppCommand::ApproveGuardianDeniedAction { event } => {
app_server
.thread_approve_guardian_denied_action(thread_id, event)
.await?;
Ok(true)
}
_ => Ok(false),
}
}
pub(super) fn turn_permissions_override_from_config(
config: &Config,
active_permission_profile: Option<&ActivePermissionProfile>,
runtime_permission_profile_override: Option<&PermissionProfile>,
) -> TurnPermissionsOverride {
if let Some(active_permission_profile) = active_permission_profile {
return TurnPermissionsOverride::ActiveProfile(active_permission_profile.clone());
}
let effective_permission_profile = config.permissions.effective_permission_profile();
let runtime_permission_profile_override =
runtime_permission_profile_override.map(|profile| {
profile
.clone()
.materialize_project_roots_with_path_uris(&config.effective_workspace_roots())
});
if runtime_permission_profile_override
.as_ref()
.is_some_and(|profile| profile == &effective_permission_profile)
{
return TurnPermissionsOverride::LegacySandbox(effective_permission_profile);
}
TurnPermissionsOverride::Preserve
}
pub(super) fn handle_skills_list_result(
&mut self,
result: Result<SkillsListResponse>,
failure_message: &str,
) {
match result {
Ok(response) => self.handle_skills_list_response(response),
Err(err) => {
tracing::warn!("{failure_message}: {err:#}");
self.chat_widget
.add_error_message(format!("{failure_message}: {err:#}"));
}
}
}
pub(super) async fn try_resolve_app_server_request(
&mut self,
thread_id: ThreadId,
op: &AppCommand,
) -> Result<bool> {
let Some(resolution) = self
.pending_app_server_requests
.take_resolution_record(&thread_id.to_string(), op)
.map_err(|err| color_eyre::eyre::eyre!(err))?
else {
return Ok(false);
};
let Some(reply) = resolution.request.reply else {
return Err(color_eyre::eyre::eyre!(
"Missing original app-server connection for pending request"
));
};
if let Err(err) = reply.resolve(resolution.result).await {
self.chat_widget.add_error_message(format!(
"Failed to resolve app-server request for thread {thread_id}: {err}"
));
}
if ThreadEventStore::op_can_change_pending_replay_state(op) {
self.note_thread_outbound_op(thread_id, op).await;
self.refresh_pending_thread_approvals().await;
self.refresh_side_parent_status_from_store(thread_id).await;
}
Ok(true)
}
pub(super) async fn refresh_pending_thread_approvals(&mut self) {
let side_parent_thread_id = self.active_side_parent_thread_id();
let channels: Vec<(ThreadId, Arc<Mutex<ThreadEventStore>>)> = self
.thread_event_channels
.iter()
.map(|(thread_id, channel)| (*thread_id, Arc::clone(&channel.store)))
.collect();
let mut pending_thread_ids = HashSet::new();
for (thread_id, store) in channels {
if Some(thread_id) == self.active_thread_id || Some(thread_id) == side_parent_thread_id
{
continue;
}
if store.lock().await.has_pending_thread_approvals() {
pending_thread_ids.insert(thread_id);
// Approval requests can arrive before the thread's metadata.
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,
);
}
}
}
if self.agent_navigation.pending_approvals != pending_thread_ids {
self.agent_navigation.pending_approvals = pending_thread_ids;
self.refresh_agent_picker();
}
}
pub(super) async fn refresh_side_parent_status_from_store(&mut self, thread_id: ThreadId) {
let Some(channel) = self.thread_event_channels.get(&thread_id) else {
return;
};
let status = {
let store = channel.store.lock().await;
store.side_parent_pending_status()
};
if let Some(status) = status {
self.set_side_parent_status(thread_id, Some(status));
} else {
self.clear_side_parent_action_status(thread_id);
}
}
pub(super) async fn enqueue_thread_notification(
&mut self,
thread_id: ThreadId,
notification: ServerNotification,
) -> Result<()> {
self.deliver_background_voice_notification(thread_id, &notification);
if self.abandoned_side_threads.contains(&thread_id) {
return Ok(());
}
self.track_side_chat_title(thread_id, &notification);
if self.current_displayed_thread_id() == Some(thread_id)
&& let ServerNotification::TurnCompleted(notification) = &notification
{
let now = Instant::now();
self.recap
.note_turn_finished(&notification.turn.status, now);
self.schedule_recap_check(thread_id, now);
}
let misalignment_policy_violation =
match &notification {
ServerNotification::Error(notification) if !notification.will_retry => {
notification.error.codex_error_info.as_ref()
}
ServerNotification::TurnCompleted(notification) => notification
.turn
.error
.as_ref()
.and_then(|error| error.codex_error_info.as_ref()),
_ => None,
} == Some(&AppServerCodexErrorInfo::MisalignmentPolicyViolation);
if misalignment_policy_violation && self.active_side_parent_thread_id() == Some(thread_id) {
if self.chat_widget.is_user_turn_pending_or_running() {
self.chat_widget.submit_op(AppCommand::Interrupt);
}
self.chat_widget.on_misalignment_policy_violation();
}
if matches!(
notification,
ServerNotification::ThreadSettingsUpdated(_)
| ServerNotification::ThreadArchived(_)
| ServerNotification::ThreadAttachmentUpdated(_)
) && self.primary_thread_id.is_some()
&& self.primary_thread_id != Some(thread_id)
&& !self.thread_event_channels.contains_key(&thread_id)
{
return Ok(());
}
let mut confirmed_profile = None;
if let ServerNotification::ThreadSettingsUpdated(notification) = &notification {
self.apply_thread_settings_to_cached_session(thread_id, &notification.thread_settings)
.await;
if self
.pending_server_profiles
.get(&thread_id)
.or_else(|| {
self.agents_overview
.requested_permission_profiles
.get(&thread_id)
})
.is_some_and(|selected| {
notification
.thread_settings
.active_permission_profile
.as_ref()
.is_some_and(|active| active.id == selected.profile_id)
&& selected.approval_policy.is_none_or(|policy| {
notification.thread_settings.approval_policy == policy
})
&& selected.approvals_reviewer.is_none_or(|reviewer| {
notification.thread_settings.approvals_reviewer.to_core() == reviewer
})
})
{
confirmed_profile = self.pending_server_profiles.remove(&thread_id).or_else(|| {
self.agents_overview
.requested_permission_profiles
.remove(&thread_id)
});
}
}
let inferred_session = if let ServerNotification::ThreadStarted(started) = &notification
&& self.primary_session_configured.is_some()
{
self.upsert_agent_picker_thread(
thread_id,
started.thread.agent_nickname.clone(),
started.thread.agent_role.clone(),
/*is_closed*/ false,
);
self.agent_navigation.set_agent_path(
thread_id,
crate::app_server_session::source_agent_path(&started.thread.source),
);
self.sync_agent_picker_thread(thread_id);
// Lifecycle responses already contain authoritative session state. Their rollout may
// not exist until the first turn, so inferring it again can wait through reader retries.
let already_has_session = match self.thread_event_channels.get(&thread_id) {
Some(channel) => channel.store.lock().await.session.is_some(),
None => false,
};
if already_has_session {
None
} else {
self.infer_session_for_started_thread(thread_id, started)
.await
}
} else {
None
};
let is_turn_started = matches!(notification, ServerNotification::TurnStarted(_));
let is_thread_closed = matches!(notification, ServerNotification::ThreadClosed(_));
let notification_status_change = SideParentStatusChange::for_notification(&notification);
let (sender, store) = {
let channel = self.ensure_thread_channel(thread_id);
(channel.sender.clone(), Arc::clone(&channel.store))
};
let (mut notification, previous_pending_status, 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 previous_pending_status = guard.side_parent_pending_status();
let turn_stopped = match &notification {
ServerNotification::TurnCompleted(notification) => {
guard.active_turn_id() == Some(notification.turn.id.as_str())
}
_ => false,
};
let notification = if guard.active {
guard.push_notification_ref(&notification);
Some(notification)
} else {
if self
.background_voice
.as_ref()
.is_none_or(|owner| owner.thread_id() != Some(thread_id))
{
self.retain_inactive_realtime_transcript(thread_id, &notification);
}
if let Some(companion) = self.side_companions.get_mut(&thread_id)
&& companion.apply_notification(&notification, &guard, || {
self.chat_widget.new_status_indicator()
})
{
self.chat_widget.request_redraw();
}
// Status traffic alone does not change the displayed transcript.
guard.replay_may_have_output |= !matches!(
&notification,
ServerNotification::ThreadStarted(_)
| ServerNotification::ThreadStatusChanged(_)
| ServerNotification::ThreadReadStateChanged(_)
| ServerNotification::ThreadTokenUsageUpdated(_)
| ServerNotification::ThreadNameUpdated(_)
| ServerNotification::ThreadProjectUpdated(_)
| ServerNotification::ThreadQueueChanged(_)
| ServerNotification::ThreadPredictionUpdated(_)
| ServerNotification::ThreadAttachmentUpdated(_)
| ServerNotification::ThreadSettingsUpdated(_)
| ServerNotification::McpToolCallProgress(_)
| ServerNotification::FileChangePatchUpdated(_)
| ServerNotification::ServerRequestResolved(_)
| ServerNotification::ThreadRealtimeOutputAudioDelta(_)
| ServerNotification::ThreadRealtimeSdp(_)
| ServerNotification::ThreadRealtimeItemAdded(_)
| ServerNotification::RawResponseItemCompleted(_)
| ServerNotification::RawResponseCompleted(_)
);
guard.push_notification(notification);
None
};
(
notification,
previous_pending_status,
guard.side_parent_pending_status(),
turn_stopped,
)
};
if is_turn_started {
self.agent_navigation.mark_running(thread_id);
} else if is_thread_closed {
self.mark_agent_picker_thread_closed(thread_id);
} else if turn_stopped {
self.agent_navigation.mark_stopped(thread_id);
}
// Settings snapshots do not belong in the transcript queue: apply them in receive order.
if let Some(ServerNotification::ThreadSettingsUpdated(settings)) = notification.as_ref()
&& self.active_thread_id == Some(thread_id)
&& self.chat_widget.thread_id() == Some(thread_id)
{
self.chat_widget
.on_thread_settings_updated(settings.clone());
if confirmed_profile.is_some()
|| settings
.thread_settings
.active_permission_profile
.as_ref()
.is_some_and(|profile| {
self.agents_overview
.selected_permission_profiles
.get(&thread_id)
== Some(&profile.id)
})
{
// Also adopt changed custom definitions and settings from another client
// after a no-op re-selection that needed no pending request.
self.adopt_server_permissions();
}
notification = None;
}
if confirmed_profile.is_some() {
self.app_event_tx.send(AppEvent::SettingsSelectionSettled);
}
if let Some(notification) = notification {
match sender.try_send(ThreadBufferedEvent::Notification(Box::new(notification))) {
Ok(()) => {}
Err(TrySendError::Full(event)) => {
tokio::spawn(async move {
if let Err(err) = sender.send(event).await {
tracing::warn!("thread {thread_id} event channel closed: {err}");
}
});
}
Err(TrySendError::Closed(_)) => {
tracing::warn!("thread {thread_id} event channel closed");
}
}
}
if let Some(status) = pending_status {
self.set_side_parent_status(thread_id, Some(status));
} else if let Some(change) = notification_status_change {
self.apply_side_parent_status_change(thread_id, change);
} else if previous_pending_status.is_some() {
self.clear_side_parent_action_status(thread_id);
}
self.refresh_pending_thread_approvals().await;
Ok(())
}
/// Locally remembers receiver threads referenced by a collab notification.
///
/// This intentionally avoids app-server reads on the active-thread rendering path. During large
/// fan-outs the app-server can be saturated with spawn work, and blocking here would freeze the
/// TUI event loop. Metadata from `ThreadStarted` or explicit picker refreshes still fills in
/// names and roles later; until then, rendering falls back to the thread id.
pub(super) fn cache_collab_receiver_threads_for_notification(
&mut self,
notification: &ServerNotification,
) {
if let Some(activity) =
sub_agent_activity_item(notification).and_then(sub_agent_activity_display)
{
let thread_id = activity.thread_id;
self.agent_navigation.record_sub_agent_activity(activity);
self.sync_agent_picker_thread(thread_id);
return;
}
let Some(receiver_thread_ids) = collab_receiver_thread_ids(notification) else {
return;
};
for receiver_thread_id in receiver_thread_ids {
if collab_receiver_is_not_found(notification, receiver_thread_id) {
continue;
}
let Ok(thread_id) = ThreadId::from_string(receiver_thread_id) else {
tracing::warn!(
thread_id = receiver_thread_id,
"ignoring collab receiver with invalid thread id during local caching"
);
continue;
};
if self.agent_navigation.get(&thread_id).is_some() {
continue;
}
self.upsert_agent_picker_thread(
thread_id, /*agent_nickname*/ None, /*agent_role*/ None,
/*is_closed*/ false,
);
self.sync_agent_picker_thread(thread_id);
}
}
async fn infer_session_for_started_thread(
&self,
thread_id: ThreadId,
notification: &ThreadStartedNotification,
) -> Option<ThreadSessionState> {
let mut session = self.primary_session_configured.clone()?;
session.thread_id = thread_id;
session.daybreak_enabled = notification.thread.daybreak_enabled.unwrap_or(false);
session.windows_sandbox_host = crate::windows_sandbox::host_from_environments(
notification.thread.environments.as_deref(),
);
session.thread_name = notification.thread.name.clone();
session.model_provider_id = notification.thread.model_provider.clone();
session
.set_cwd_retargeting_implicit_runtime_workspace_root(notification.thread.cwd.clone());
let rollout_path = notification.thread.path.clone();
if let Some(model) = &notification.thread.model {
session.model = model.clone();
} else if rollout_path.is_some() {
session.model.clear();
}
session.message_history = None;
session.rollout_path = rollout_path;
Some(session)
}
pub(super) async fn enqueue_thread_request(
&mut self,
thread_id: ThreadId,
request: ServerRequest,
) -> Result<()> {
let inactive_interactive_request = if self.active_thread_id != Some(thread_id) {
self.interactive_request_for_thread_request(thread_id, &request)
.await?
} else {
None
};
let (sender, store) = {
let channel = self.ensure_thread_channel(thread_id);
(channel.sender.clone(), Arc::clone(&channel.store))
};
let (should_send, pending_status) = {
let mut guard = store.lock().await;
guard.push_request(request.clone());
(guard.active, guard.side_parent_pending_status())
};
let request_status = SideParentStatus::for_request(&request);
if should_send {
match sender.try_send(ThreadBufferedEvent::Request(Box::new(request))) {
Ok(()) => {}
Err(TrySendError::Full(event)) => {
tokio::spawn(async move {
if let Err(err) = sender.send(event).await {
tracing::warn!("thread {thread_id} event channel closed: {err}");
}
});
}
Err(TrySendError::Closed(_)) => {
tracing::warn!("thread {thread_id} event channel closed");
}
}
} else if self.active_side_parent_thread_id().is_none()
&& let Some(request) = inactive_interactive_request
{
self.push_thread_interactive_request(request);
}
if let Some(status) = pending_status.or(request_status) {
self.set_side_parent_status(thread_id, Some(status));
}
self.refresh_pending_thread_approvals().await;
Ok(())
}
pub(super) async fn enqueue_thread_history_entry_response(
&mut self,
thread_id: ThreadId,
event: HistoryLookupResponse,
) -> Result<()> {
let (sender, store) = {
let channel = self.ensure_thread_channel(thread_id);
(channel.sender.clone(), Arc::clone(&channel.store))
};
let should_send = {
let mut guard = store.lock().await;
let should_send = guard.active;
// Active batch responses remain queued in the receiver across a concurrent detach, so
// retaining another deep copy for thread replay only accumulates already-delivered
// history data. Inactive responses still need the buffer because they are not sent.
if !should_send || !matches!(&event, HistoryLookupResponse::Batch { .. }) {
guard.push_buffered_event(ThreadBufferedEvent::HistoryEntryResponse(event.clone()));
}
should_send
};
if should_send {
match sender.try_send(ThreadBufferedEvent::HistoryEntryResponse(event)) {
Ok(()) => {}
Err(TrySendError::Full(event)) => {
tokio::spawn(async move {
if let Err(err) = sender.send(event).await {
tracing::warn!("thread {thread_id} event channel closed: {err}");
}
});
}
Err(TrySendError::Closed(_)) => {
tracing::warn!("thread {thread_id} event channel closed");
}
}
}
Ok(())
}
pub(super) async fn enqueue_primary_thread_session(
&mut self,
session: ThreadSessionState,
turns: Vec<Turn>,
) -> Result<()> {
self.enqueue_primary_thread_session_with_presentation(
session,
turns,
ThreadAttachPresentation::SessionLineage,
)
.await
}
pub(super) async fn enqueue_primary_thread_session_with_presentation(
&mut self,
session: ThreadSessionState,
turns: Vec<Turn>,
presentation: ThreadAttachPresentation,
) -> Result<()> {
if let Err(err) = self
.config
.permissions
.approval_policy
.set(session.approval_policy.to_core())
{
tracing::warn!(%err, "failed to sync app approval policy from thread session");
}
if let Err(err) = self
.config
.permissions
.set_permission_profile_from_session_snapshot(
PermissionProfileSnapshot::from_session_snapshot(
session.permission_profile.clone(),
session.active_permission_profile.clone(),
),
)
{
tracing::warn!(%err, "failed to sync app permissions from thread session");
}
self.config.approvals_reviewer = session.approvals_reviewer;
if !self.app_server_target.uses_remote_workspace() {
self.config
.workspace_roots
.clone_from(&session.runtime_workspace_roots);
self.config
.permissions
.set_workspace_roots(session.runtime_workspace_roots.clone());
}
let thread_id = session.thread_id;
self.pending_server_profiles.remove(&thread_id);
self.agents_overview
.requested_permission_profiles
.remove(&thread_id);
if self.primary_thread_id != Some(thread_id) {
self.recap.reset_for_new_thread(Instant::now());
}
self.primary_thread_id = Some(thread_id);
self.agents_overview.hidden_threads.remove(&thread_id);
self.agents_overview.threads.entry(thread_id).or_default();
self.primary_session_configured = Some(session.clone());
self.upsert_agent_picker_thread(
thread_id, /*agent_nickname*/ None, /*agent_role*/ None,
/*is_closed*/ false,
);
self.sync_agent_picker_thread(thread_id);
let channel = self.ensure_thread_channel(thread_id);
{
let mut store = channel.store.lock().await;
store.set_session(session.clone(), turns.clone());
}
self.activate_thread_channel(thread_id).await;
self.chat_widget
.set_initial_user_message_submit_suppressed(/*suppressed*/ true);
if !turns.is_empty() {
self.chat_widget.set_token_info(/*info*/ None);
}
match presentation {
ThreadAttachPresentation::Fresh
| ThreadAttachPresentation::FreshAfterClear
| ThreadAttachPresentation::FreshWithDraft
| ThreadAttachPresentation::SessionLineage => {
self.chat_widget.handle_thread_session(session);
}
}
let should_buffer_initial_replay = !turns.is_empty();
let replayed_final_items = realtime_delivery::completed_agent_items_from_turns(&turns);
let replayed_voice_texts = realtime_delivery::replayed_voice_texts_from_turns(&turns);
let retained_assistant_captions =
self.prepare_realtime_transcript_replay(replayed_voice_texts);
if should_buffer_initial_replay {
self.app_event_tx
.send(AppEvent::BeginInitialHistoryReplayBuffer);
}
let now = Instant::now();
self.recap.seed_from_turns(&turns, now);
self.schedule_recap_check(thread_id, now);
self.chat_widget
.replay_thread_turns(turns, ReplayKind::ResumeInitialMessages);
if should_buffer_initial_replay {
self.app_event_tx
.send(AppEvent::EndInitialHistoryReplayBuffer);
self.app_event_tx
.send(AppEvent::RestoreBackgroundTerminals(thread_id));
}
self.restore_realtime_replay_state_after_replay(
&replayed_final_items,
retained_assistant_captions,
);
self.restore_voice_owner_after_replay();
let pending = std::mem::take(&mut self.pending_primary_events);
for pending_event in pending {
match pending_event {
ThreadBufferedEvent::Notification(notification) => {
self.enqueue_thread_notification(thread_id, *notification)
.await?;
}
ThreadBufferedEvent::Request(request) => {
self.enqueue_thread_request(thread_id, *request).await?;
}
ThreadBufferedEvent::HistoryEntryResponse(event) => {
self.enqueue_thread_history_entry_response(thread_id, event)
.await?;
}
ThreadBufferedEvent::FeedbackSubmission(event) => {
self.enqueue_thread_feedback_event(thread_id, event).await;
}
}
}
self.chat_widget
.set_initial_user_message_submit_suppressed(/*suppressed*/ false);
self.chat_widget.submit_initial_user_message_if_pending();
Ok(())
}
pub(super) async fn enqueue_primary_thread_notification(
&mut self,
notification: ServerNotification,
) -> Result<()> {
if let Some(thread_id) = self.primary_thread_id {
return self
.enqueue_thread_notification(thread_id, notification)
.await;
}
self.pending_primary_events
.push_back(ThreadBufferedEvent::Notification(Box::new(notification)));
Ok(())
}
pub(super) async fn enqueue_primary_thread_request(
&mut self,
request: ServerRequest,
) -> Result<()> {
if let Some(thread_id) = self.primary_thread_id {
return self.enqueue_thread_request(thread_id, request).await;
}
self.pending_primary_events
.push_back(ThreadBufferedEvent::Request(Box::new(request)));
Ok(())
}
pub(super) async fn refresh_snapshot_session_if_needed(
&mut self,
app_server: &mut AppServerSession,
thread_id: ThreadId,
is_replay_only: bool,
snapshot: &mut ThreadEventSnapshot,
) -> bool {
if !self.should_refresh_snapshot_session(thread_id, is_replay_only, snapshot) {
return true;
}
match app_server
.resume_thread(
&self.local_settings,
self.config.clone(),
thread_id,
crate::app_server_session::ResumeModelSettings::PreserveExistingThread,
)
.await
{
Ok(started) => {
self.apply_refreshed_snapshot_thread(thread_id, started, snapshot)
.await;
true
}
Err(err) => {
tracing::warn!(
thread_id = %thread_id,
error = %err,
"failed to refresh inferred thread session before replay"
);
false
}
}
}
pub(super) fn should_refresh_snapshot_session(
&self,
thread_id: ThreadId,
is_replay_only: bool,
snapshot: &ThreadEventSnapshot,
) -> bool {
!is_replay_only
&& !self.side_threads.contains_key(&thread_id)
&& snapshot.session.as_ref().is_none_or(|session| {
session.model.trim().is_empty()
|| session.rollout_path.is_none()
|| (self.primary_thread_id != Some(thread_id)
&& session
.active_permission_profile
.as_ref()
.is_some_and(|profile| !profile.id.starts_with(':')))
})
}
pub(super) async fn apply_refreshed_snapshot_thread(
&mut self,
thread_id: ThreadId,
started: AppServerStartedThread,
snapshot: &mut ThreadEventSnapshot,
) {
if started.blocks_direct_input {
self.agent_navigation.mark_parent_owned(thread_id);
}
let AppServerStartedThread { session, turns, .. } = started;
if let Some(channel) = self.thread_event_channels.get(&thread_id) {
let mut store = channel.store.lock().await;
store.replay_may_have_output |= snapshot.turns != turns;
store.set_session(session.clone(), turns.clone());
store.rebase_buffer_after_session_refresh();
snapshot.active_reasoning_item = store.active_reasoning_item.clone();
}
snapshot.session = Some(session);
snapshot.turns = turns;
snapshot
.events
.retain_mut(ThreadEventStore::event_survives_session_refresh);
}
/// Opens the `/subagents` picker after refreshing cached labels for known threads.
///
/// The picker state is derived from long-lived thread channels plus best-effort metadata
/// refreshes from the backend. Refresh failures are treated as "thread is only inspectable by
/// historical id now" and converted into closed picker entries instead of deleting them, so
/// the stable traversal order remains intact for review and keyboard navigation.
pub(super) async fn drain_active_thread_events(&mut self, tui: &mut tui::Tui) -> Result<()> {
let frame_deadline = Instant::now() + tui::TARGET_FRAME_INTERVAL;
self.drain_active_thread_events_until(tui, frame_deadline)
.await
}
pub(super) async fn drain_active_thread_events_until(
&mut self,
tui: &mut tui::Tui,
frame_deadline: Instant,
) -> Result<()> {
let Some(mut rx) = self.active_thread_rx.take() else {
return Ok(());
};
let mut disconnected = false;
loop {
match rx.try_recv() {
Ok(event) => {
self.handle_thread_event_now_recovering_file_changes(event)
.await
}
Err(TryRecvError::Empty) => break,
Err(TryRecvError::Disconnected) => {
disconnected = true;
break;
}
}
if Instant::now() >= frame_deadline {
break;
}
}
if !disconnected {
self.active_thread_rx = Some(rx);
} else {
self.clear_active_thread().await;
}
if self.backtrack_render_pending {
tui.frame_requester().schedule_frame();
}
Ok(())
}
/// Returns `(closed_thread_id, primary_thread_id)` when a non-primary active
/// thread has died and we should fail over to the primary thread.
///
/// A user-requested shutdown (`ExitMode::ShutdownFirst`) sets
/// `pending_shutdown_exit_thread_id`; matching shutdown completions are ignored
/// here so Ctrl+C-like exits don't accidentally resurrect the main thread.
///
/// Failover is only eligible when all of these are true:
/// 1. the event is `thread/closed`;
/// 2. the active thread differs from the primary thread;
/// 3. the active thread is not the pending shutdown-exit thread.
pub(super) fn active_non_primary_shutdown_target(
&self,
notification: &ServerNotification,
) -> Option<(ThreadId, ThreadId)> {
if !matches!(notification, ServerNotification::ThreadClosed(_)) {
return None;
}
let active_thread_id = self.active_thread_id?;
let primary_thread_id = self.primary_thread_id?;
if self.pending_shutdown_exit_thread_id == Some(active_thread_id) {
return None;
}
(active_thread_id != primary_thread_id).then_some((active_thread_id, primary_thread_id))
}
pub(super) fn replay_thread_snapshot(
&mut self,
mut snapshot: ThreadEventSnapshot,
resume_restored_queue: bool,
) {
let mut reasoning_replay = reasoning_replay::ReasoningReplay::new(&mut snapshot);
let replayed_final_items = realtime_delivery::completed_agent_items(&snapshot);
let replayed_voice_texts = realtime_delivery::replayed_voice_texts(&snapshot);
replay_filter::omit_completed_agent_deltas(&mut snapshot.events);
let request_changes = snapshot
.events
.iter()
.map(|event| {
let ThreadBufferedEvent::Request(request) = event else {
return None;
};
let ServerRequest::FileChangeRequestApproval { params, .. } = request.as_ref()
else {
return None;
};
file_change_changes(
snapshot.events.iter(),
&snapshot.turns,
&params.turn_id,
&params.item_id,
)
})
.collect::<Vec<_>>();
self.refresh_mcp_startup_expected_servers_from_config();
let should_buffer_replay = !snapshot.turns.is_empty() || !snapshot.events.is_empty();
if should_buffer_replay {
self.app_event_tx
.send(AppEvent::BeginThreadSwitchHistoryReplayBuffer);
}
let suppress_replay_notices =
replay_filter::snapshot_has_pending_interactive_request(&snapshot);
self.chat_widget
.set_queue_autosend_suppressed(/*suppressed*/ true);
let has_resumed_collaboration_mode = snapshot
.session
.as_ref()
.is_some_and(|session| session.collaboration_mode.is_some());
if let Some(session) = snapshot.session {
if session.reasoning_effort != Some(ReasoningEffortConfig::Ultra) {
self.chat_widget
.set_plan_mode_reasoning_effort(self.config.plan_mode_reasoning_effort.clone());
}
if self.side_threads.contains_key(&session.thread_id) {
self.chat_widget.handle_side_thread_session(session);
} else if suppress_replay_notices {
self.chat_widget.handle_thread_session_quiet(session);
} else {
self.chat_widget.handle_thread_session(session);
}
}
let retained_assistant_captions =
self.prepare_realtime_transcript_replay(replayed_voice_texts);
for turn_id in &snapshot.delegated_turns {
self.chat_widget
.remember_realtime_delegated_reasoning_turn(turn_id);
}
let confirmed_message_ids = super::thread_events::confirmed_user_message_ids(
&snapshot.turns,
snapshot.events.iter(),
);
let recovered_input = snapshot
.input_state
.as_ref()
.is_some_and(|input| {
input.recovered_queue || input.reconnect_pending || input.has_unconfirmed_messages()
})
.then(|| snapshot.input_state.take())
.flatten();
self.chat_widget.restore_thread_input_state(
snapshot.input_state,
ThreadInputStateRestoreMode {
preserve_in_flight_turn: true,
},
);
if !snapshot.turns.is_empty() {
self.chat_widget
.replay_thread_turns(snapshot.turns, ReplayKind::ThreadSnapshot);
}
for (event, changes) in snapshot.events.into_iter().zip(request_changes) {
reasoning_replay.before_event(&event, &mut self.chat_widget);
match (event, changes) {
(ThreadBufferedEvent::Request(request), Some(changes)) => {
self.handle_file_change_request(*request, changes)
}
(event, _) => self.handle_thread_event_replay(event),
}
}
reasoning_replay.restore(&mut self.chat_widget);
if should_buffer_replay {
self.app_event_tx
.send(AppEvent::EndInitialHistoryReplayBuffer);
}
let recovered = recovered_input.is_some();
if recovered {
let mode = has_resumed_collaboration_mode
.then(|| self.chat_widget.effective_collaboration_mode());
self.chat_widget
.restore_reconnected_input(recovered_input, &confirmed_message_ids);
if let Some(mode) = mode {
self.chat_widget.set_effective_collaboration_mode(mode);
}
}
self.restore_realtime_replay_state_after_replay(
&replayed_final_items,
retained_assistant_captions,
);
self.restore_voice_owner_after_replay();
self.chat_widget
.set_queue_autosend_suppressed(/*suppressed*/ false);
self.chat_widget
.set_initial_user_message_submit_suppressed(/*suppressed*/ false);
self.chat_widget.submit_initial_user_message_if_pending();
if resume_restored_queue {
self.chat_widget.maybe_send_next_queued_input();
}
self.refresh_status_line();
}
pub(super) fn should_wait_for_initial_session(session_selection: &SessionSelection) -> bool {
matches!(
session_selection,
SessionSelection::StartFresh | SessionSelection::Exit
)
}
pub(super) fn should_prompt_for_paused_goal_after_startup_resume(
session_selection: &SessionSelection,
initial_prompt: &Option<String>,
initial_images: &[PathBuf],
) -> bool {
matches!(session_selection, SessionSelection::Resume(_))
&& initial_prompt.is_none()
&& initial_images.is_empty()
}
pub(super) fn should_handle_active_thread_events(
waiting_for_initial_session_configured: bool,
has_active_thread_receiver: bool,
) -> bool {
has_active_thread_receiver && !waiting_for_initial_session_configured
}
pub(super) fn should_stop_waiting_for_initial_session(
waiting_for_initial_session_configured: bool,
primary_thread_id: Option<ThreadId>,
) -> bool {
waiting_for_initial_session_configured && primary_thread_id.is_some()
}
#[allow(clippy::too_many_arguments)]
pub(super) fn handle_skills_list_response(&mut self, response: SkillsListResponse) {
let cwd = self.chat_widget.config_ref().cwd.clone();
let errors = errors_for_cwd(&cwd, &response);
let errors = self.skill_load_warnings.newly_active_errors(&errors);
let warnings = skill_load_warning_messages(&errors);
if self.skill_load_warnings.startup_complete {
for warning in warnings {
self.chat_widget.add_warning_message(warning);
}
} else {
self.app_event_tx.send(AppEvent::InsertHistoryCell(Box::new(
// The per-file diagnostics already identify every affected skill.
history_cell::StartupWarningsCell::new(
warnings.into_iter().skip(/*n*/ 1).collect(),
),
)));
}
self.chat_widget.handle_skills_list_response(response);
}
fn startup_request_may_open_protected_view(&self, request: &ServerRequest) -> bool {
let ServerRequest::McpServerElicitationRequest { request_id, params } = request else {
return true;
};
match &params.request {
codex_app_server_protocol::McpServerElicitationRequest::UserVerification { .. } => true,
codex_app_server_protocol::McpServerElicitationRequest::Form { .. } => true,
codex_app_server_protocol::McpServerElicitationRequest::OpenAiForm { .. }
| codex_app_server_protocol::McpServerElicitationRequest::OpenAiElicitationForm {
..
} => false,
request @ codex_app_server_protocol::McpServerElicitationRequest::Url { .. } => {
let thread_id = ThreadId::from_string(&params.thread_id)
.unwrap_or_else(|_| self.chat_widget.thread_id().unwrap_or_default());
AppLinkViewParams::from_url_app_server_request(
thread_id,
&params.server_name,
request_id.clone(),
request,
)
.is_some()
}
}
}
pub(super) fn handle_thread_event_now(&mut self, event: ThreadBufferedEvent) {
// Replay has its own handler; suppress repeated notifications for the current turn.
if crate::terminal_size_telemetry::enabled(&self.config)
&& !self.terminal_size_tags.is_empty()
&& let ThreadBufferedEvent::Notification(notification) = &event
&& let ServerNotification::TurnStarted(started) = notification.as_ref()
&& self
.active_thread_id
.is_some_and(|id| id.to_string() == started.thread_id)
&& self.chat_widget.last_turn_id() != Some(started.turn.id.as_str())
{
crate::terminal_size_telemetry::record(
&self.terminal_size_tags,
&self.config,
"turn_start",
crate::terminal_size_telemetry::read_size(),
);
}
let needs_refresh = matches!(
&event,
ThreadBufferedEvent::Notification(notification)
if matches!(
notification.as_ref(),
ServerNotification::TurnStarted(_)
| ServerNotification::ThreadTokenUsageUpdated(_)
)
);
match event {
ThreadBufferedEvent::Notification(notification) => {
self.cache_collab_receiver_threads_for_notification(notification.as_ref());
let tip_ready = self.turn_tips.observe(&notification, Instant::now());
self.chat_widget
.handle_server_notification(*notification, /*replay_kind*/ None);
// History cells queued by completion must be applied before anchoring its tip.
if let Some(event) = tip_ready {
self.app_event_tx.send(event);
}
}
ThreadBufferedEvent::Request(request) => {
if self
.pending_app_server_requests
.contains_server_request(request.as_ref())
{
let may_open_protected_view =
self.startup_request_may_open_protected_view(request.as_ref());
self.chat_widget
.handle_server_request(*request, /*replay_kind*/ None);
if may_open_protected_view
&& self.startup_protected_input_boundary
&& !self.chat_widget.has_active_modal()
{
self.startup_pending_protected_request = true;
}
}
}
ThreadBufferedEvent::HistoryEntryResponse(event) => {
self.chat_widget.handle_history_entry_response(event);
}
ThreadBufferedEvent::FeedbackSubmission(event) => {
self.handle_feedback_thread_event(event);
}
}
if needs_refresh {
self.refresh_status_line();
}
}
pub(super) fn handle_thread_event_replay(&mut self, event: ThreadBufferedEvent) {
self.turn_tips.dismiss();
match event {
ThreadBufferedEvent::Notification(notification) => self
.chat_widget
.handle_server_notification(*notification, Some(ReplayKind::ThreadSnapshot)),
ThreadBufferedEvent::Request(request) => {
let may_open_protected_view =
self.startup_request_may_open_protected_view(request.as_ref());
self.chat_widget
.handle_server_request(*request, Some(ReplayKind::ThreadSnapshot));
if may_open_protected_view
&& self.startup_protected_input_boundary
&& !self.chat_widget.has_active_modal()
{
self.startup_pending_protected_request = true;
}
}
ThreadBufferedEvent::HistoryEntryResponse(event) => {
self.chat_widget.handle_history_entry_response(event)
}
ThreadBufferedEvent::FeedbackSubmission(event) => {
self.handle_feedback_thread_event(event);
}
}
}
/// Handles an event emitted by the currently active thread.
///
/// This function enforces shutdown intent routing: unexpected non-primary
/// thread shutdowns fail over to the primary thread, while user-requested
/// app exits consume only the tracked shutdown completion and then proceed.
pub(super) async fn handle_active_thread_event(
&mut self,
tui: &mut tui::Tui,
app_server: &mut AppServerSession,
event: ThreadBufferedEvent,
) -> Result<()> {
// Capture this before any potential thread switch: we only want to clear
// the exit marker when the currently active thread acknowledges shutdown.
let pending_shutdown_exit_completed = matches!(
&event,
ThreadBufferedEvent::Notification(notification)
if matches!(notification.as_ref(), ServerNotification::ThreadClosed(_))
) && self.pending_shutdown_exit_thread_id
== self.active_thread_id;
// Processing order matters:
//
// 1. handle unexpected non-primary shutdown failover first;
// 2. clear pending exit marker for matching shutdown;
// 3. forward the event through normal handling.
//
// This preserves the mental model that user-requested exits do not trigger
// failover, while true sub-agent deaths still do.
if let ThreadBufferedEvent::Notification(notification) = &event
&& let Some((closed_thread_id, primary_thread_id)) =
self.active_non_primary_shutdown_target(notification.as_ref())
{
self.mark_agent_picker_thread_closed(closed_thread_id);
if self.side_threads.contains_key(&closed_thread_id) {
self.discard_closed_side_thread(closed_thread_id).await;
}
self.select_agent_thread(tui, app_server, primary_thread_id)
.await?;
if self.active_thread_id == Some(primary_thread_id) {
self.chat_widget.add_info_message(
format!(
"Agent thread {closed_thread_id} closed. Switched back to main thread."
),
/*hint*/ None,
);
} else {
self.clear_active_thread().await;
self.chat_widget.add_error_message(format!(
"Agent thread {closed_thread_id} closed. Failed to switch back to main thread {primary_thread_id}.",
));
}
return Ok(());
}
if pending_shutdown_exit_completed {
// Clear only after seeing the shutdown completion for the tracked
// thread, so unrelated shutdowns cannot consume this marker.
self.pending_shutdown_exit_thread_id = None;
}
// Sidebar chats use their first prompt until they receive an explicit name.
let automatic_title_user_message = if self.chat_widget.thread_name().is_none()
&& !(self.side_retention_enabled(tui) && self.active_side_parent_thread_id().is_some())
&& let ThreadBufferedEvent::Notification(notification) = &event
&& let ServerNotification::ItemCompleted(notification) = notification.as_ref()
&& let ThreadItemPayload::UserMessage { content, .. } = &notification.item.payload
{
Some(
content
.iter()
.filter_map(|item| match item {
codex_app_server_protocol::UserInput::Text { text, .. } => {
Some(crate::user_prompt::extract_prompt_request_with_offset(text).0)
}
_ => None,
})
.collect::<Vec<_>>()
.join(" "),
)
} else {
None
};
let had_active_modal = self.chat_widget.has_active_modal();
self.handle_thread_event_now_recovering_file_changes(event)
.await;
if let Some(user_message) = automatic_title_user_message
&& !user_message.trim().is_empty()
&& let Some(thread_id) = self.active_thread_id
{
self.generate_thread_title(
app_server,
thread_id,
ThreadTitleDestination::Automatic,
super::thread_title::thread_title_prompt(&user_message),
);
}
if !had_active_modal
&& self.chat_widget.has_active_modal()
&& self.startup_protected_input_boundary
{
self.chat_widget.pre_draw_tick();
self.render_chat_widget_frame(tui, tui.terminal.last_known_screen_size)?;
self.discard_startup_modal_input(tui)?;
}
if self.backtrack_render_pending {
tui.frame_requester().schedule_frame();
}
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
use codex_protocol::models::ActivePermissionProfile;
use codex_protocol::models::BUILT_IN_PERMISSION_PROFILE_WORKSPACE;
async fn config_with_workspace_profile() -> Config {
let temp_dir = tempfile::tempdir().expect("tempdir");
ConfigBuilder::default()
.codex_home(temp_dir.path().to_path_buf())
.harness_overrides(ConfigOverrides {
default_permissions: Some(BUILT_IN_PERMISSION_PROFILE_WORKSPACE.to_string()),
..ConfigOverrides::default()
})
.build()
.await
.expect("config should build")
}
#[tokio::test]
async fn turn_permissions_use_active_profile_when_available() {
let config = config_with_workspace_profile().await;
let active_permission_profile = config.permissions.active_permission_profile();
assert_eq!(
App::turn_permissions_override_from_config(
&config,
active_permission_profile.as_ref(),
/*runtime_permission_profile_override*/ None,
),
TurnPermissionsOverride::ActiveProfile(ActivePermissionProfile::new(
BUILT_IN_PERMISSION_PROFILE_WORKSPACE
))
);
}
#[tokio::test]
async fn turn_permissions_preserve_server_snapshot_without_local_override() {
let mut config = config_with_workspace_profile().await;
config
.permissions
.set_permission_profile(PermissionProfile::read_only())
.expect("read-only profile should be allowed");
assert_eq!(
App::turn_permissions_override_from_config(
&config, /*active_permission_profile*/ None,
/*runtime_permission_profile_override*/ None,
),
TurnPermissionsOverride::Preserve
);
}
#[tokio::test]
async fn turn_permissions_send_legacy_sandbox_for_local_override() {
let mut config = config_with_workspace_profile().await;
let permission_profile = PermissionProfile::workspace_write();
config
.permissions
.set_permission_profile(permission_profile.clone())
.expect("workspace profile should be allowed");
let effective_permission_profile = config.permissions.effective_permission_profile();
assert_eq!(
App::turn_permissions_override_from_config(
&config,
/*active_permission_profile*/ None,
Some(&permission_profile),
),
TurnPermissionsOverride::LegacySandbox(effective_permission_profile)
);
}
}
Reference in New Issue View Git Blame Copy Permalink