From edc5008dd0cc1c24f5ec00c70a5e8e867cc3a957 Mon Sep 17 00:00:00 2001 From: Felipe Coury Date: Tue, 17 Feb 2026 17:45:23 -0300 Subject: [PATCH] tui: fix interrupt behavior during heavy streaming --- codex-rs/tui/src/bottom_pane/mod.rs | 41 ++++++++++- codex-rs/tui/src/chatwidget.rs | 70 +++++++++++++++++++ .../chatwidget/tests/composer_submission.rs | 67 ++++++++++++++++++ codex-rs/tui/src/chatwidget/tests/helpers.rs | 1 + 4 files changed, 176 insertions(+), 3 deletions(-) diff --git a/codex-rs/tui/src/bottom_pane/mod.rs b/codex-rs/tui/src/bottom_pane/mod.rs index da5e2421f7..82a86745a3 100644 --- a/codex-rs/tui/src/bottom_pane/mod.rs +++ b/codex-rs/tui/src/bottom_pane/mod.rs @@ -15,6 +15,7 @@ //! hint. The pane schedules redraws so those hints can expire even when the UI is otherwise idle. use std::path::PathBuf; +use crate::app_event::AppEvent; use crate::app_event::ConnectorsSnapshot; use crate::app_event_sender::AppEventSender; use crate::bottom_pane::pending_input_preview::PendingInputPreview; @@ -31,6 +32,7 @@ use codex_core::plugins::PluginCapabilitySummary; use codex_core::skills::model::SkillMetadata; use codex_features::Features; use codex_file_search::FileMatch; +use codex_protocol::protocol::Op; use codex_protocol::request_user_input::RequestUserInputEvent; use codex_protocol::user_input::TextElement; use crossterm::event::KeyCode; @@ -429,10 +431,13 @@ impl BottomPane { && self.is_task_running && !is_agent_command && !self.composer.popup_active() - && let Some(status) = &self.status { - // Send Op::Interrupt - status.interrupt(); + if let Some(status) = &self.status { + // Send Op::Interrupt + status.interrupt(); + } else { + self.app_event_tx.send(AppEvent::CodexOp(Op::Interrupt)); + } self.request_redraw(); return InputResult::None; } @@ -1511,6 +1516,36 @@ mod tests { assert!(rendered.contains("background terminal running ยท /ps to view")); } + #[test] + fn esc_interrupts_running_task_when_status_hidden() { + let (tx_raw, mut rx) = unbounded_channel::(); + let tx = AppEventSender::new(tx_raw); + let mut pane = BottomPane::new(BottomPaneParams { + app_event_tx: tx, + frame_requester: FrameRequester::test_dummy(), + has_input_focus: true, + enhanced_keys_supported: false, + placeholder_text: "Ask Codex to do anything".to_string(), + disable_paste_burst: false, + animations_enabled: true, + skills: Some(Vec::new()), + }); + + pane.set_task_running(true); + pane.hide_status_indicator(); + assert!( + !pane.status_indicator_visible(), + "status indicator must be hidden for this repro" + ); + + pane.handle_key_event(KeyEvent::new(KeyCode::Esc, KeyModifiers::NONE)); + + assert!( + matches!(rx.try_recv(), Ok(AppEvent::CodexOp(Op::Interrupt))), + "expected Esc to send Op::Interrupt while status is hidden" + ); + } + #[test] fn status_with_details_and_queued_messages_snapshot() { let (tx_raw, _rx) = unbounded_channel::(); diff --git a/codex-rs/tui/src/chatwidget.rs b/codex-rs/tui/src/chatwidget.rs index ff9c5bb02c..303475482a 100644 --- a/codex-rs/tui/src/chatwidget.rs +++ b/codex-rs/tui/src/chatwidget.rs @@ -162,6 +162,7 @@ use codex_protocol::protocol::ErrorEvent; #[cfg(test)] use codex_protocol::protocol::Event; #[cfg(test)] +#[cfg(test)] use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::ExecApprovalRequestEvent; use codex_protocol::protocol::ExecCommandBeginEvent; @@ -177,6 +178,8 @@ use codex_protocol::protocol::ImageGenerationBeginEvent; use codex_protocol::protocol::ImageGenerationEndEvent; use codex_protocol::protocol::ListSkillsResponseEvent; #[cfg(test)] +use codex_protocol::protocol::EventMsg; +#[cfg(test)] use codex_protocol::protocol::McpListToolsResponseEvent; #[cfg(test)] use codex_protocol::protocol::McpStartupCompleteEvent; @@ -524,6 +527,43 @@ impl RateLimitWarningState { } } +#[cfg(test)] +fn is_interrupt_droppable_stream_event(msg: &EventMsg) -> bool { + match msg { + EventMsg::AgentMessage(_) + | EventMsg::AgentMessageDelta(_) + | EventMsg::PlanDelta(_) + | EventMsg::AgentReasoning(_) + | EventMsg::AgentReasoningDelta(_) + | EventMsg::AgentReasoningRawContent(_) + | EventMsg::AgentReasoningRawContentDelta(_) + | EventMsg::AgentReasoningSectionBreak(_) + | EventMsg::AgentMessageContentDelta(_) + | EventMsg::ReasoningContentDelta(_) + | EventMsg::ReasoningRawContentDelta(_) => true, + EventMsg::ItemCompleted(event) => matches!( + &event.item, + codex_protocol::items::TurnItem::AgentMessage(_) + | codex_protocol::items::TurnItem::Plan(_) + ), + _ => false, + } +} + +fn is_interrupt_droppable_server_notification(notification: &ServerNotification) -> bool { + match notification { + ServerNotification::AgentMessageDelta(_) + | ServerNotification::PlanDelta(_) + | ServerNotification::ReasoningSummaryTextDelta(_) + | ServerNotification::ReasoningTextDelta(_) => true, + ServerNotification::ItemCompleted(notification) => matches!( + ¬ification.item, + ThreadItem::AgentMessage { .. } | ThreadItem::Plan { .. } + ), + _ => false, + } +} + pub(crate) fn get_limits_duration(windows_minutes: i64) -> String { const MINUTES_PER_HOUR: i64 = 60; const MINUTES_PER_DAY: i64 = 24 * MINUTES_PER_HOUR; @@ -803,6 +843,11 @@ pub(crate) struct ChatWidget { /// bottom pane is treated as "running" while this is populated, even if no agent turn is /// currently executing. mcp_startup_status: Option>, + /// True once an interrupt was requested for the active turn. + /// + /// While this is set, inbound stream deltas are dropped locally so a large + /// queued backlog cannot keep rendering stale content after user interrupt. + interrupt_requested_for_turn: bool, /// Expected MCP servers for the current startup round, seeded from enabled local config. mcp_startup_expected_servers: Option>, /// After startup settles, ignore stale updates until enough notifications confirm a new round. @@ -2290,6 +2335,7 @@ impl ChatWidget { self.agent_turn_running = true; self.turn_sleep_inhibitor .set_turn_running(/*turn_running*/ true); + self.interrupt_requested_for_turn = false; self.saw_plan_update_this_turn = false; self.saw_plan_item_this_turn = false; self.last_plan_progress = None; @@ -2360,6 +2406,7 @@ impl ChatWidget { self.agent_turn_running = false; self.turn_sleep_inhibitor .set_turn_running(/*turn_running*/ false); + self.interrupt_requested_for_turn = false; self.update_task_running_state(); self.running_commands.clear(); self.suppressed_exec_calls.clear(); @@ -2723,6 +2770,7 @@ impl ChatWidget { self.agent_turn_running = false; self.turn_sleep_inhibitor .set_turn_running(/*turn_running*/ false); + self.interrupt_requested_for_turn = false; self.update_task_running_state(); self.running_commands.clear(); self.suppressed_exec_calls.clear(); @@ -4695,6 +4743,7 @@ impl ChatWidget { agent_turn_running: false, mcp_startup_status: None, pending_turn_copyable_output: None, + interrupt_requested_for_turn: false, mcp_startup_expected_servers: None, mcp_startup_ignore_updates_until_next_start: false, mcp_startup_allow_terminal_only_next_round: false, @@ -6396,6 +6445,13 @@ impl ChatWidget { if !is_resume_initial_replay && !is_retry_error { self.restore_retry_status_header_if_present(); } + if !from_replay + && self.interrupt_requested_for_turn + && is_interrupt_droppable_server_notification(¬ification) + { + tracing::trace!("dropping app-server stream notification while interrupt is pending"); + return; + } match notification { ServerNotification::ThreadTokenUsageUpdated(notification) => { self.set_token_info(Some(token_usage_info_from_app_server( @@ -6898,6 +6954,15 @@ impl ChatWidget { let from_replay = replay_kind.is_some(); let is_resume_initial_replay = matches!(replay_kind, Some(ReplayKind::ResumeInitialMessages)); + + if !from_replay + && self.interrupt_requested_for_turn + && is_interrupt_droppable_stream_event(&msg) + { + tracing::trace!("dropping stream event while interrupt is pending"); + return; + } + let is_stream_error = matches!(&msg, EventMsg::StreamError(_)); if !is_resume_initial_replay && !is_stream_error { self.restore_retry_status_header_if_present(); @@ -10519,6 +10584,11 @@ impl ChatWidget { T: Into, { let op: AppCommand = op.into(); + if matches!(op.view(), crate::app_command::AppCommandView::Interrupt) + && self.agent_turn_running + { + self.interrupt_requested_for_turn = true; + } if op.is_review() && !self.bottom_pane.is_task_running() { self.bottom_pane.set_task_running(/*running*/ true); } diff --git a/codex-rs/tui/src/chatwidget/tests/composer_submission.rs b/codex-rs/tui/src/chatwidget/tests/composer_submission.rs index a5678a6b93..09abc72da2 100644 --- a/codex-rs/tui/src/chatwidget/tests/composer_submission.rs +++ b/codex-rs/tui/src/chatwidget/tests/composer_submission.rs @@ -1060,6 +1060,73 @@ async fn interrupt_restores_queued_messages_into_composer() { let _ = drain_insert_history(&mut rx); } +#[tokio::test] +async fn interrupt_drops_stream_deltas_until_turn_aborted() { + let (mut chat, mut rx, mut op_rx) = make_chatwidget_manual(None).await; + + chat.handle_codex_event(Event { + id: "turn-1".into(), + msg: EventMsg::TurnStarted(TurnStartedEvent { + turn_id: "turn-1".to_string(), + model_context_window: None, + collaboration_mode_kind: ModeKind::Default, + }), + }); + + chat.handle_key_event(KeyEvent::new(KeyCode::Char('c'), KeyModifiers::CONTROL)); + let mut saw_interrupt = false; + while let Ok(op) = op_rx.try_recv() { + if matches!(op, Op::Interrupt) { + saw_interrupt = true; + break; + } + } + assert!(saw_interrupt, "expected Ctrl+C to submit Op::Interrupt"); + + // Simulate stale stream backlog that arrives before TurnAborted. + chat.stream_controller = None; + chat.handle_codex_event(Event { + id: "delta-stale".into(), + msg: EventMsg::AgentMessageDelta(AgentMessageDeltaEvent { + delta: "stale row\n".to_string(), + }), + }); + assert!( + chat.stream_controller.is_none(), + "expected stale delta to be dropped while interrupt is pending", + ); + + chat.handle_codex_event(Event { + id: "abort-1".into(), + msg: EventMsg::TurnAborted(codex_protocol::protocol::TurnAbortedEvent { + turn_id: Some("turn-1".to_string()), + reason: TurnAbortReason::Interrupted, + }), + }); + + // A subsequent turn should stream normally. + chat.handle_codex_event(Event { + id: "turn-2".into(), + msg: EventMsg::TurnStarted(TurnStartedEvent { + turn_id: "turn-2".to_string(), + model_context_window: None, + collaboration_mode_kind: ModeKind::Default, + }), + }); + chat.handle_codex_event(Event { + id: "delta-fresh".into(), + msg: EventMsg::AgentMessageDelta(AgentMessageDeltaEvent { + delta: "fresh row\n".to_string(), + }), + }); + assert!( + chat.stream_controller.is_some(), + "expected new-turn delta to stream after interrupt completion", + ); + + let _ = drain_insert_history(&mut rx); +} + #[tokio::test] async fn interrupt_prepends_queued_messages_before_existing_composer_text() { let (mut chat, mut rx, mut op_rx) = make_chatwidget_manual(/*model_override*/ None).await; diff --git a/codex-rs/tui/src/chatwidget/tests/helpers.rs b/codex-rs/tui/src/chatwidget/tests/helpers.rs index 46a4129a45..9b355de151 100644 --- a/codex-rs/tui/src/chatwidget/tests/helpers.rs +++ b/codex-rs/tui/src/chatwidget/tests/helpers.rs @@ -221,6 +221,7 @@ pub(super) async fn make_chatwidget_manual( unified_exec_processes: Vec::new(), agent_turn_running: false, mcp_startup_status: None, + interrupt_requested_for_turn: false, mcp_startup_expected_servers: None, mcp_startup_ignore_updates_until_next_start: false, mcp_startup_allow_terminal_only_next_round: false,