From 3a2788e6fc897fc4dd40185225167af9a9f13754 Mon Sep 17 00:00:00 2001 From: Eric Traut Date: Wed, 4 Nov 2026 05:14:07 +0000 Subject: [PATCH] Restore background terminals from the displayed thread's live state (#59319) ## Why Replaying a fork's inherited command history can make the TUI show background terminals belonging to its parent thread. ## What changed Fetch the displayed thread's background terminals after history replay instead of treating replayed commands as live terminals. Read all pages, merge without duplicating tracked processes, and exclude terminals that exit while the request is pending. Clear terminal tracking and invalidate pending restoration when resetting thread state or stopping background terminals. ## Testing Add a regression test verifying that `/ps` shows a resumed fork's own terminal, excludes inherited commands, and does not restore a terminal that exited during the list request. GitOrigin-RevId: aca9ae9cde877f03143f8fbeea11affcc34dc372 --- codex-rs/tui/src/app/event_dispatch.rs | 14 +++++ codex-rs/tui/src/app/session_lifecycle.rs | 2 + codex-rs/tui/src/app/thread_routing.rs | 2 + codex-rs/tui/src/app_event.rs | 7 +++ codex-rs/tui/src/app_server_session.rs | 30 +++++++++++ codex-rs/tui/src/chatwidget.rs | 5 +- .../tui/src/chatwidget/command_lifecycle.rs | 52 ++++++++++++++++++- codex-rs/tui/src/chatwidget/constructor.rs | 1 + codex-rs/tui/src/chatwidget/protocol.rs | 3 +- codex-rs/tui/src/chatwidget/replay.rs | 2 +- codex-rs/tui/src/chatwidget/session_flow.rs | 1 + .../tui/src/chatwidget/tests/exec_flow.rs | 44 ++++++++++++++++ 12 files changed, 157 insertions(+), 6 deletions(-) diff --git a/codex-rs/tui/src/app/event_dispatch.rs b/codex-rs/tui/src/app/event_dispatch.rs index a3ea592f4b..a4c856e122 100644 --- a/codex-rs/tui/src/app/event_dispatch.rs +++ b/codex-rs/tui/src/app/event_dispatch.rs @@ -1974,6 +1974,20 @@ impl App { self.refresh_rate_limits(app_server, RateLimitRefreshOrigin::Periodic); } } + AppEvent::RestoreBackgroundTerminals(thread_id) => { + if self.chat_widget.thread_id() == Some(thread_id) { + let request_id = self.chat_widget.begin_background_terminal_restore(); + let mut server = app_server.request_view(); + let tx = self.app_event_tx.clone(); + tokio::spawn(async move { + let terminals = server.list_background_terminals(thread_id).await.unwrap_or_default(); + tx.send(AppEvent::BackgroundTerminalsLoaded { request_id, terminals }); + }); + } + } + AppEvent::BackgroundTerminalsLoaded { request_id, terminals } => { + self.chat_widget.restore_background_terminals(request_id, terminals); + } AppEvent::RefreshThreadUsage { thread_id, request_id, diff --git a/codex-rs/tui/src/app/session_lifecycle.rs b/codex-rs/tui/src/app/session_lifecycle.rs index abc35c7875..8be1dd6186 100644 --- a/codex-rs/tui/src/app/session_lifecycle.rs +++ b/codex-rs/tui/src/app/session_lifecycle.rs @@ -820,6 +820,8 @@ impl App { AppEvent::ResetTranscriptForThreadSwitch }); self.replay_thread_snapshot(snapshot, resume_restored_queue); + self.app_event_tx + .send(AppEvent::RestoreBackgroundTerminals(thread_id)); if let Some(thread_id) = self.chat_widget.thread_id() && let Some(active) = self .chat_widget diff --git a/codex-rs/tui/src/app/thread_routing.rs b/codex-rs/tui/src/app/thread_routing.rs index ccef8cf989..016b1e9900 100644 --- a/codex-rs/tui/src/app/thread_routing.rs +++ b/codex-rs/tui/src/app/thread_routing.rs @@ -1559,6 +1559,8 @@ impl App { 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, diff --git a/codex-rs/tui/src/app_event.rs b/codex-rs/tui/src/app_event.rs index 3444d65bd0..952dcc84d5 100644 --- a/codex-rs/tui/src/app_event.rs +++ b/codex-rs/tui/src/app_event.rs @@ -855,6 +855,13 @@ pub(crate) enum AppEvent { result: Result, }, + /// Restore the displayed thread's live terminals after replaying its history. + RestoreBackgroundTerminals(ThreadId), + BackgroundTerminalsLoaded { + request_id: Uuid, + terminals: Vec, + }, + /// Fetch backend-estimated usage for the currently visible enterprise thread. RefreshThreadUsage { thread_id: ThreadId, diff --git a/codex-rs/tui/src/app_server_session.rs b/codex-rs/tui/src/app_server_session.rs index 38ada25c33..61fb54f4eb 100644 --- a/codex-rs/tui/src/app_server_session.rs +++ b/codex-rs/tui/src/app_server_session.rs @@ -81,8 +81,11 @@ use codex_app_server_protocol::ThreadApproveGuardianDeniedActionParams; use codex_app_server_protocol::ThreadApproveGuardianDeniedActionResponse; use codex_app_server_protocol::ThreadArchiveParams; use codex_app_server_protocol::ThreadArchiveResponse; +use codex_app_server_protocol::ThreadBackgroundTerminal; use codex_app_server_protocol::ThreadBackgroundTerminalsCleanParams; use codex_app_server_protocol::ThreadBackgroundTerminalsCleanResponse; +use codex_app_server_protocol::ThreadBackgroundTerminalsListParams; +use codex_app_server_protocol::ThreadBackgroundTerminalsListResponse; use codex_app_server_protocol::ThreadCompactStartParams; use codex_app_server_protocol::ThreadCompactStartResponse; use codex_app_server_protocol::ThreadDeleteParams; @@ -1951,6 +1954,33 @@ impl AppServerSession { Ok(()) } + pub(crate) async fn list_background_terminals( + &mut self, + thread_id: ThreadId, + ) -> Result> { + let mut terminals = Vec::new(); + let mut cursor = None; + loop { + let request_id = self.next_request_id(); + let page: ThreadBackgroundTerminalsListResponse = self + .request_handle + .request_typed(ClientRequest::ThreadBackgroundTerminalsList { + request_id, + params: ThreadBackgroundTerminalsListParams { + thread_id: thread_id.to_string(), + cursor, + limit: None, + }, + }) + .await?; + terminals.extend(page.data); + cursor = page.next_cursor; + if cursor.is_none() { + return Ok(terminals); + } + } + } + pub(crate) async fn thread_background_terminals_clean( &mut self, thread_id: ThreadId, diff --git a/codex-rs/tui/src/chatwidget.rs b/codex-rs/tui/src/chatwidget.rs index 310f1e3b92..773825c487 100644 --- a/codex-rs/tui/src/chatwidget.rs +++ b/codex-rs/tui/src/chatwidget.rs @@ -629,6 +629,8 @@ pub(crate) struct ChatWidget { safety_buffering: SafetyBufferingState, task_complete_pending: bool, unified_exec_processes: Vec, + /// Exits observed during the pending list request take precedence over its response. + pending_terminal_restore: Option<(uuid::Uuid, HashSet)>, /// Tracks per-server MCP startup state while startup is in progress. /// /// The map is `Some(_)` from the first startup status update until the @@ -1524,8 +1526,7 @@ impl ChatWidget { fn clean_background_terminals(&mut self) { self.submit_op(AppCommand::clean_background_terminals()); - self.unified_exec_processes.clear(); - self.sync_unified_exec_footer(); + self.clear_background_terminals(); self.add_info_message( "Stopping all background terminals.".to_string(), /*hint*/ None, diff --git a/codex-rs/tui/src/chatwidget/command_lifecycle.rs b/codex-rs/tui/src/chatwidget/command_lifecycle.rs index c65cae1fec..2e3840e55a 100644 --- a/codex-rs/tui/src/chatwidget/command_lifecycle.rs +++ b/codex-rs/tui/src/chatwidget/command_lifecycle.rs @@ -18,7 +18,11 @@ impl ChatWidget { self.restore_reasoning_status_header(); } - pub(super) fn on_command_execution_started(&mut self, item: ThreadItemPayload) { + pub(super) fn on_command_execution_started( + &mut self, + item: ThreadItemPayload, + render_source: ThreadItemRenderSource, + ) { let ThreadItemPayload::CommandExecution { id, command, @@ -33,7 +37,7 @@ impl ChatWidget { let (_command, parsed_cmd) = command_execution_command_and_parsed(command, command_actions); self.flush_answer_stream_with_separator(); if is_unified_exec_source(*source) { - if *source == ExecCommandSource::UnifiedExecStartup { + if *source == ExecCommandSource::UnifiedExecStartup && !render_source.is_replay() { self.track_unified_exec_process_begin(id, process_id.as_deref(), command); } if !self.bottom_pane.is_task_running() { @@ -191,11 +195,55 @@ impl ChatWidget { self.sync_unified_exec_footer(); } + pub(super) fn clear_background_terminals(&mut self) { + self.pending_terminal_restore = None; + self.unified_exec_processes.clear(); + self.sync_unified_exec_footer(); + } + + pub(crate) fn begin_background_terminal_restore(&mut self) -> uuid::Uuid { + let request_id = uuid::Uuid::new_v4(); + self.pending_terminal_restore = Some((request_id, HashSet::new())); + request_id + } + + pub(crate) fn restore_background_terminals( + &mut self, + request_id: uuid::Uuid, + terminals: Vec, + ) { + let Some((_, exited)) = self + .pending_terminal_restore + .take_if(|(id, _)| *id == request_id) + else { + return; + }; + for terminal in terminals { + if !exited.contains(&terminal.item_id) + && !self + .unified_exec_processes + .iter() + .any(|process| process.key == terminal.process_id) + { + self.unified_exec_processes.push(UnifiedExecProcessSummary { + key: terminal.process_id, + call_id: terminal.item_id, + command_display: terminal.command, + recent_chunks: Vec::new(), + }); + } + } + self.sync_unified_exec_footer(); + } + pub(super) fn track_unified_exec_process_end( &mut self, call_id: &str, process_id: Option<&str>, ) { + if let Some((_, exited)) = &mut self.pending_terminal_restore { + exited.insert(call_id.to_owned()); + } let key = process_id.unwrap_or(call_id); let before = self.unified_exec_processes.len(); self.unified_exec_processes diff --git a/codex-rs/tui/src/chatwidget/constructor.rs b/codex-rs/tui/src/chatwidget/constructor.rs index e5e74417cc..9d1d4fa6f5 100644 --- a/codex-rs/tui/src/chatwidget/constructor.rs +++ b/codex-rs/tui/src/chatwidget/constructor.rs @@ -206,6 +206,7 @@ impl ChatWidget { safety_buffering: SafetyBufferingState::default(), task_complete_pending: false, unified_exec_processes: Vec::new(), + pending_terminal_restore: None, mcp_startup_status: None, mcp_startup_expected_servers: None, mcp_startup_ignore_updates_until_next_start: false, diff --git a/codex-rs/tui/src/chatwidget/protocol.rs b/codex-rs/tui/src/chatwidget/protocol.rs index 1c9d69844e..17b23819bf 100644 --- a/codex-rs/tui/src/chatwidget/protocol.rs +++ b/codex-rs/tui/src/chatwidget/protocol.rs @@ -632,7 +632,8 @@ impl ChatWidget { self.on_context_compaction_started(id, elapsed); } item @ ThreadItemPayload::CommandExecution { .. } => { - self.on_command_execution_started(item) + // Buffered notifications were received from this thread's live subscription. + self.on_command_execution_started(item, ThreadItemRenderSource::Live) } ThreadItemPayload::FileChange { id: _, changes, .. } => { self.on_patch_apply_begin(file_update_changes_to_display(changes)); diff --git a/codex-rs/tui/src/chatwidget/replay.rs b/codex-rs/tui/src/chatwidget/replay.rs index fbe40068bc..e3e95210be 100644 --- a/codex-rs/tui/src/chatwidget/replay.rs +++ b/codex-rs/tui/src/chatwidget/replay.rs @@ -391,7 +391,7 @@ impl ChatWidget { item @ ThreadItemPayload::CommandExecution { status: codex_app_server_protocol::CommandExecutionStatus::InProgress, .. - } => self.on_command_execution_started(item), + } => self.on_command_execution_started(item, render_source), item @ ThreadItemPayload::CommandExecution { source: ExecCommandSource::Agent | ExecCommandSource::UnifiedExecStartup, status: diff --git a/codex-rs/tui/src/chatwidget/session_flow.rs b/codex-rs/tui/src/chatwidget/session_flow.rs index a1ff412d4d..ce6c1b604d 100644 --- a/codex-rs/tui/src/chatwidget/session_flow.rs +++ b/codex-rs/tui/src/chatwidget/session_flow.rs @@ -343,6 +343,7 @@ impl ChatWidget { retained_turns: &[Turn], ) { self.current_rollout_path = rollout_path; + self.clear_background_terminals(); self.input_queue.clear(); self.reset_realtime_conversation(); // Finish any output not yet drained from the old runtime without live completion actions. diff --git a/codex-rs/tui/src/chatwidget/tests/exec_flow.rs b/codex-rs/tui/src/chatwidget/tests/exec_flow.rs index 5d2f4693dc..c5327a5375 100644 --- a/codex-rs/tui/src/chatwidget/tests/exec_flow.rs +++ b/codex-rs/tui/src/chatwidget/tests/exec_flow.rs @@ -47,6 +47,50 @@ async fn external_writer_snapshot_freezes_active_command_and_mcp_rows() { insta::assert_snapshot!(rendered.join("\n---\n")); } +#[tokio::test] +async fn replayed_background_terminals_belong_to_the_displayed_thread() { + let (mut parent, _events, _ops) = make_chatwidget_manual(/*model_override*/ None).await; + let inherited = begin_unified_exec_startup(&mut parent, "parent-call", "42", "sleep 100"); + let (mut fork, mut events, _ops) = make_chatwidget_manual(/*model_override*/ None).await; + fork.replay_thread_item( + inherited, + "turn-1".into(), + ReplayKind::ResumeInitialMessages, + ); + + // A resumed fork can have terminals of its own alongside the inherited command history. + let request_id = fork.begin_background_terminal_restore(); + let live = codex_app_server_protocol::ThreadBackgroundTerminal { + item_id: "child-call".into(), + process_id: "43".into(), + command: "sleep 200".into(), + cwd: fork.config.cwd.clone().into(), + os_pid: None, + cpu_percent: None, + rss_kb: None, + }; + let exited = codex_app_server_protocol::ThreadBackgroundTerminal { + item_id: "exited-call".into(), + process_id: "44".into(), + command: "sleep 300".into(), + ..live.clone() + }; + fork.track_unified_exec_process_end(&exited.item_id, Some(&exited.process_id)); + fork.restore_background_terminals(request_id, vec![live, exited]); + fork.add_ps_output(); + let history = drain_insert_history(&mut events) + .iter() + .map(|lines| lines_to_single_string(lines)) + .collect::(); + insta::assert_snapshot!(history, @r" + /ps + + Background terminals + + • sleep 200 + "); +} + #[tokio::test] async fn replayed_command_completion_preserves_tracking_without_duplicate_starts() { let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await;