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;