mirror of
https://github.com/openai/codex.git
synced 2026-11-05 20:10:33 +00:00
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
This commit is contained in:
@@ -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,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -855,6 +855,13 @@ pub(crate) enum AppEvent {
|
||||
result: Result<ConsumeAccountRateLimitResetCreditResponse, String>,
|
||||
},
|
||||
|
||||
/// Restore the displayed thread's live terminals after replaying its history.
|
||||
RestoreBackgroundTerminals(ThreadId),
|
||||
BackgroundTerminalsLoaded {
|
||||
request_id: Uuid,
|
||||
terminals: Vec<codex_app_server_protocol::ThreadBackgroundTerminal>,
|
||||
},
|
||||
|
||||
/// Fetch backend-estimated usage for the currently visible enterprise thread.
|
||||
RefreshThreadUsage {
|
||||
thread_id: ThreadId,
|
||||
|
||||
@@ -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<Vec<ThreadBackgroundTerminal>> {
|
||||
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,
|
||||
|
||||
@@ -629,6 +629,8 @@ pub(crate) struct ChatWidget {
|
||||
safety_buffering: SafetyBufferingState,
|
||||
task_complete_pending: bool,
|
||||
unified_exec_processes: Vec<UnifiedExecProcessSummary>,
|
||||
/// Exits observed during the pending list request take precedence over its response.
|
||||
pending_terminal_restore: Option<(uuid::Uuid, HashSet<String>)>,
|
||||
/// 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,
|
||||
|
||||
@@ -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<codex_app_server_protocol::ThreadBackgroundTerminal>,
|
||||
) {
|
||||
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
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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));
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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::<String>();
|
||||
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;
|
||||
|
||||
Reference in New Issue
Block a user