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

codex: simplify side-thread cleanup

Browse Source
This commit is contained in:
Eric Traut
2026-06-06 18:15:27 -07:00
parent 5cb56085d4
commit 30c9fd3151
7 changed files with 67 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/tui/src/app.rs
+9 -13
View File
@@ -1253,20 +1253,16 @@ See the Codex keymap documentation for supported actions and examples."
// Keep rendering while /side prepares, but freeze input so drafts and shortcuts cannot
// race the pending transition. Esc remains available to cancel the preparation.
if self.pending_side_start.is_some() {
match &event {
TuiEvent::Key(key_event) => {
if matches!(key_event.code, KeyCode::Esc)
&& matches!(key_event.kind, KeyEventKind::Press | KeyEventKind::Repeat)
{
self.cancel_pending_side_start();
tui.frame_requester().schedule_frame();
}
return Ok(AppRunControl::Continue);
}
TuiEvent::Paste(_) => return Ok(AppRunControl::Continue),
TuiEvent::Draw | TuiEvent::Resize => {}
if self.pending_side_start.is_some() && !matches!(event, TuiEvent::Draw | TuiEvent::Resize)
{
if let TuiEvent::Key(key_event) = &event
&& matches!(key_event.code, KeyCode::Esc)
&& matches!(key_event.kind, KeyEventKind::Press | KeyEventKind::Repeat)
{
self.cancel_pending_side_start();
tui.frame_requester().schedule_frame();
}
return Ok(AppRunControl::Continue);
}
if self.overlay.is_some() {
codex-rs/tui/src/app/session_lifecycle.rs
+6 -6
View File
@@ -138,6 +138,9 @@ impl App {
app_server: &mut AppServerSession,
thread_id: ThreadId,
) -> bool {
if self.side_threads.contains_key(&thread_id) {
return true;
}
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
@@ -290,12 +293,9 @@ impl App {
return Ok(());
}
// Side-thread preparation already attached the listener and cached the full snapshot.
// Avoid a backend round trip when switching to that freshly prepared local state.
if !self.side_threads.contains_key(&thread_id)
&& !self
.refresh_agent_picker_thread_liveness(app_server, thread_id)
.await
if !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."));
codex-rs/tui/src/app/side.rs
+36 -30
View File
@@ -128,18 +128,22 @@ mod tests {
#[test]
fn side_start_error_message_explains_missing_first_prompt() {
let err = "thread/fork failed during TUI bootstrap: thread/fork failed: no rollout found for thread id 019da1a1-bed9-7a43-88a2-b49d43915021";
let err = color_eyre::eyre::eyre!(
"thread/fork failed during TUI bootstrap: thread/fork failed: no rollout found for thread id 019da1a1-bed9-7a43-88a2-b49d43915021"
);
assert_eq!(
App::side_start_error_message(err),
App::side_start_error_message(&err),
"'/side' is unavailable until the current conversation has started. Send a message first, then try /side again."
);
}
#[test]
fn side_start_error_message_uses_generic_start_wording() {
let err = color_eyre::eyre::eyre!("transport disconnected");
assert_eq!(
App::side_start_error_message("transport disconnected"),
App::side_start_error_message(&err),
"Failed to start side conversation: transport disconnected"
);
}
@@ -482,10 +486,12 @@ impl App {
}
}
pub(super) fn side_start_error_message(err: &str) -> String {
if err.contains("no rollout found for thread id")
|| err.contains("includeTurns is unavailable before first user message")
{
pub(super) fn side_start_error_message(err: &color_eyre::Report) -> String {
if err.chain().any(|cause| {
let message = cause.to_string();
message.contains("no rollout found for thread id")
|| message.contains("includeTurns is unavailable before first user message")
}) {
SIDE_NO_STARTED_CONVERSATION_MESSAGE.to_string()
} else {
format!("Failed to start side conversation: {err}")
@@ -602,14 +608,15 @@ impl App {
app_server: &AppServerSession,
thread_id: ThreadId,
) {
// Hide the canceled thread while cleanup is pending, but retain enough local state to
// restore it in /agent if the server refuses to unsubscribe it.
// Hide the canceled thread while cleanup runs; failures restore it to /agent.
self.agent_navigation.remove(thread_id);
let request_handle = app_server.request_handle();
let app_event_tx = self.app_event_tx.clone();
tokio::spawn(async move {
let result = super::side_server::cleanup_side_thread(request_handle, thread_id).await;
app_event_tx.send(AppEvent::SideThreadCleanupFinished(thread_id, result));
app_event_tx.send(AppEvent::SideThreadCleanupFinished(
thread_id,
super::side_server::cleanup_side_thread(request_handle, thread_id).await,
));
});
}
@@ -618,16 +625,18 @@ impl App {
thread_id: ThreadId,
result: std::result::Result<(), String>,
) {
self.discard_thread_local_state(thread_id).await;
if let Err(err) = result {
self.upsert_agent_picker_thread(
thread_id, /*agent_nickname*/ None, /*agent_role*/ None,
/*is_closed*/ false,
);
self.chat_widget.add_error_message(format!(
"Failed to close side conversation {thread_id}; it remains available in /agent: \
{err}"
));
match result {
Ok(()) => self.discard_thread_local_state(thread_id).await,
Err(err) => {
self.agent_navigation.upsert(
thread_id, /*agent_nickname*/ None, /*agent_role*/ None,
/*is_closed*/ false,
);
self.chat_widget.add_error_message(format!(
"Failed to close side conversation {thread_id}; it remains available in \
/agent: {err}"
));
}
}
}
@@ -636,10 +645,7 @@ impl App {
tui: &mut tui::Tui,
app_server: &mut AppServerSession,
request_id: Uuid,
result: std::result::Result<
AppServerStartedThread,
crate::app_event::SideThreadPrepareError,
>,
result: std::result::Result<ThreadSessionState, crate::app_event::SideThreadPrepareError>,
) -> Result<()> {
let Some(PendingSideStart {
side_state,
@@ -650,7 +656,7 @@ impl App {
.take_if(|pending| pending.request_id == request_id)
else {
if let Some(thread_id) = match result {
Ok(started) => Some(started.session.thread_id),
Ok(session) => Some(session.thread_id),
Err(err) => err.thread_id,
} {
self.cleanup_side_thread_in_background(app_server, thread_id);
@@ -660,12 +666,12 @@ impl App {
let parent_thread_id = side_state.parent_thread_id;
self.chat_widget.set_side_start_pending(/*pending*/ false);
match result {
Ok(started) => {
let child_thread_id = started.session.thread_id;
Ok(session) => {
let child_thread_id = session.thread_id;
let channel = self.ensure_thread_channel(child_thread_id);
{
let mut store = channel.store.lock().await;
Self::install_side_thread_snapshot(&mut store, started.session, started.turns);
Self::install_side_thread_snapshot(&mut store, session, Vec::new());
}
self.side_threads.insert(child_thread_id, side_state);
// Prepared side threads switch through the selector's local replay path.
@@ -707,7 +713,7 @@ impl App {
self.chat_widget
.set_side_conversation_context_label(/*label*/ None);
self.chat_widget
.add_error_message(Self::side_start_error_message(&err.message));
.add_error_message(Self::side_start_error_message(&err.error));
}
}
Ok(())
codex-rs/tui/src/app/side_server.rs
+12 -19
View File
@@ -15,11 +15,11 @@ pub(super) async fn prepare_side_thread(
parent_thread_id: ThreadId,
thread_params_mode: ThreadParamsMode,
remote_cwd_override: Option<PathBuf>,
) -> std::result::Result<AppServerStartedThread, SideThreadPrepareError> {
) -> std::result::Result<ThreadSessionState, SideThreadPrepareError> {
let boundary_item = serde_json::to_value(App::side_boundary_prompt_item()).map_err(|err| {
SideThreadPrepareError {
thread_id: None,
message: format!("failed to encode thread/inject_items payload: {err}"),
error: color_eyre::eyre::eyre!("failed to encode thread/inject_items payload: {err}"),
}
})?;
let started = crate::app_server_session::fork_thread_with_request_handle(
@@ -32,7 +32,7 @@ pub(super) async fn prepare_side_thread(
.await
.map_err(|err| SideThreadPrepareError {
thread_id: None,
message: format!("{err:#}"),
error: err,
})?;
let child_thread_id = started.session.thread_id;
@@ -50,19 +50,19 @@ pub(super) async fn prepare_side_thread(
if let Err(err) = inject_result {
return Err(SideThreadPrepareError {
thread_id: Some(child_thread_id),
message: format!(
error: color_eyre::eyre::eyre!(
"thread/inject_items failed during TUI side conversation setup: {err}"
),
});
}
Ok(started)
Ok(started.session)
}
pub(super) async fn cleanup_side_thread(
request_handle: AppServerRequestHandle,
thread_id: ThreadId,
) -> std::result::Result<(), String> {
let interrupt_result = request_handle
request_handle
.request_typed::<TurnInterruptResponse>(ClientRequest::TurnInterrupt {
request_id: RequestId::String(format!("side-thread-interrupt-{}", Uuid::new_v4())),
params: TurnInterruptParams {
@@ -70,23 +70,16 @@ pub(super) async fn cleanup_side_thread(
turn_id: String::new(),
},
})
.await;
let unsubscribe_result = request_handle
.await
.map_err(|err| format!("turn/interrupt failed: {err}"))?;
request_handle
.request_typed::<ThreadUnsubscribeResponse>(ClientRequest::ThreadUnsubscribe {
request_id: RequestId::String(format!("side-thread-unsubscribe-{}", Uuid::new_v4())),
params: ThreadUnsubscribeParams {
thread_id: thread_id.to_string(),
},
})
.await;
match (interrupt_result, unsubscribe_result) {
(Ok(_), Ok(_)) => Ok(()),
(Err(interrupt_err), Ok(_)) => Err(format!("turn/interrupt failed: {interrupt_err}")),
(Ok(_), Err(unsubscribe_err)) => {
Err(format!("thread/unsubscribe failed: {unsubscribe_err}"))
}
(Err(interrupt_err), Err(unsubscribe_err)) => Err(format!(
"turn/interrupt failed: {interrupt_err}; thread/unsubscribe failed: {unsubscribe_err}"
)),
}
.await
.map(|_| ())
.map_err(|err| format!("thread/unsubscribe failed: {err}"))
}
codex-rs/tui/src/app/tests.rs
-2
View File
@@ -3405,12 +3405,10 @@ async fn cancelling_pending_side_start_restores_message() {
async fn failed_side_cleanup_restores_thread_to_agent_picker() {
let mut app = make_test_app().await;
let thread_id = ThreadId::new();
app.ensure_thread_channel(thread_id);
app.handle_side_thread_cleanup_finished(thread_id, Err("transport disconnected".to_string()))
.await;
assert!(!app.thread_event_channels.contains_key(&thread_id));
assert!(app.agent_navigation.get(&thread_id).is_some());
}
codex-rs/tui/src/app_event.rs
+3 -2
View File
@@ -38,6 +38,7 @@ use crate::bottom_pane::ApprovalRequest;
use crate::bottom_pane::StatusLineItem;
use crate::bottom_pane::TerminalTitleItem;
use crate::chatwidget::UserMessage;
use crate::session_state::ThreadSessionState;
use codex_app_server_protocol::AskForApproval;
use codex_config::types::ApprovalsReviewer;
use codex_features::Feature;
@@ -55,7 +56,7 @@ use crate::history_cell::HistoryCell;
#[derive(Debug)]
pub(crate) struct SideThreadPrepareError {
pub(crate) thread_id: Option<ThreadId>,
pub(crate) message: String,
pub(crate) error: color_eyre::Report,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -154,7 +155,7 @@ pub(crate) enum AppEvent {
user_message: Option<UserMessage>,
},
/// Finish preparing a transient side conversation off the TUI event loop.
SideThreadPrepared(Uuid, Result<AppServerStartedThread, SideThreadPrepareError>),
SideThreadPrepared(Uuid, Result<ThreadSessionState, SideThreadPrepareError>),
/// Finish removing a canceled or failed side conversation off the TUI event loop.
SideThreadCleanupFinished(ThreadId, Result<(), String>),
codex-rs/tui/src/app_server_session.rs
+1 -9
View File
@@ -190,20 +190,12 @@ impl ThreadParamsMode {
}
}
#[derive(Debug)]
pub(crate) struct AppServerStartedThread {
pub(crate) session: ThreadSessionState,
pub(crate) turns: Vec<Turn>,
}
impl std::fmt::Debug for AppServerStartedThread {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("AppServerStartedThread")
.field("session", &self.session)
.field("turn_count", &self.turns.len())
.finish()
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum TurnPermissionsOverride {
/// Leave the app-server thread's sticky permission profile unchanged.