fix(tui): prepare app-level interrupts locally

Run the same local stream cleanup for app-level interrupt commands
as for direct widget submissions, clearing queued stream output and
transient tails before the backend response arrives.

Also ensure consolidated proposed-plan stream cells force the required
scrollback reflow when replacing an existing streamed cell run.
This commit is contained in:
Felipe Coury
2026-04-07 12:42:08 -03:00
parent 7a4c131e89
commit 42efa1a80a
3 changed files with 75 additions and 21 deletions
+17 -8
View File
@@ -3356,6 +3356,15 @@ impl App {
tui.frame_requester().schedule_frame();
}
fn finish_required_stream_reflow(&mut self, tui: &mut tui::Tui) -> Result<()> {
self.schedule_immediate_resize_reflow(tui);
self.maybe_run_resize_reflow(tui)?;
if self.resize_reflow_pending_until.is_none() {
self.reflow_ran_during_stream = false;
}
Ok(())
}
fn handle_draw_size_change(
&mut self,
size: ratatui::layout::Size,
@@ -3475,11 +3484,7 @@ impl App {
self.maybe_finish_stream_reflow(tui);
}
ConsolidationScrollbackReflow::Required => {
self.schedule_immediate_resize_reflow(tui);
self.maybe_run_resize_reflow(tui)?;
if self.resize_reflow_pending_until.is_none() {
self.reflow_ran_during_stream = false;
}
self.finish_required_stream_reflow(tui)?;
}
}
@@ -4527,6 +4532,8 @@ impl App {
t.consolidate_cells(start..end, consolidated.clone());
tui.frame_requester().schedule_frame();
}
self.finish_required_stream_reflow(tui)?;
} else {
self.transcript_cells.push(consolidated.clone());
if let Some(Overlay::Transcript(t)) = &mut self.overlay {
@@ -4538,9 +4545,9 @@ impl App {
consolidated.as_ref(),
tui.terminal.last_known_screen_size.width,
);
}
self.maybe_finish_stream_reflow(tui);
self.maybe_finish_stream_reflow(tui);
}
}
AppEvent::ApplyThreadRollback { num_turns } => {
if self.apply_non_pending_thread_rollback(num_turns) {
@@ -4577,7 +4584,9 @@ impl App {
return Ok(AppRunControl::Exit(ExitReason::Fatal(message)));
}
AppEvent::CodexOp(op) => {
self.submit_active_thread_op(app_server, op.into()).await?;
let op: AppCommand = op.into();
self.chat_widget.prepare_local_op_submission(&op);
self.submit_active_thread_op(app_server, op).await?;
}
AppEvent::SubmitThreadOp { thread_id, op } => {
self.submit_thread_op(app_server, thread_id, op.into())
+17 -13
View File
@@ -10618,19 +10618,7 @@ impl ChatWidget {
T: Into<AppCommand>,
{
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 let Some(controller) = self.stream_controller.as_mut() {
controller.clear_queue();
}
if let Some(controller) = self.plan_stream_controller.as_mut() {
controller.clear_queue();
}
self.clear_active_stream_tail();
self.request_redraw();
}
self.prepare_local_op_submission(&op);
if op.is_review() && !self.bottom_pane.is_task_running() {
self.bottom_pane.set_task_running(/*running*/ true);
}
@@ -10649,6 +10637,22 @@ impl ChatWidget {
true
}
pub(crate) fn prepare_local_op_submission(&mut self, op: &AppCommand) {
if matches!(op.view(), crate::app_command::AppCommandView::Interrupt)
&& self.agent_turn_running
{
self.interrupt_requested_for_turn = true;
if let Some(controller) = self.stream_controller.as_mut() {
controller.clear_queue();
}
if let Some(controller) = self.plan_stream_controller.as_mut() {
controller.clear_queue();
}
self.clear_active_stream_tail();
self.request_redraw();
}
}
#[cfg(test)]
fn on_list_mcp_tools(&mut self, ev: McpListToolsResponseEvent) {
self.add_to_history(history_cell::new_mcp_tools_output(
@@ -1127,6 +1127,47 @@ async fn interrupt_drops_stream_deltas_until_turn_aborted() {
let _ = drain_insert_history(&mut rx);
}
#[tokio::test]
async fn app_event_interrupt_prepares_local_stream_cleanup() {
let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(None).await;
let cwd = chat.config.cwd.to_path_buf();
let mut controller =
crate::streaming::controller::StreamController::new(Some(80), cwd.as_path(), false);
assert!(controller.push("stale backlog\n"));
chat.agent_turn_running = true;
chat.bottom_pane.set_task_running(/*running*/ true);
chat.bottom_pane.hide_status_indicator();
chat.stream_controller = Some(controller);
chat.active_cell = Some(Box::new(crate::history_cell::StreamingAgentTailCell::new(
vec![ratatui::text::Line::from("tail")],
true,
)));
chat.handle_key_event(KeyEvent::new(KeyCode::Esc, KeyModifiers::NONE));
let command = loop {
match rx.try_recv() {
Ok(AppEvent::CodexOp(op)) => break AppCommand::from(op),
Ok(_) => continue,
Err(err) => panic!("expected app-level interrupt event, got {err:?}"),
}
};
chat.prepare_local_op_submission(&command);
assert!(chat.interrupt_requested_for_turn);
assert!(
chat.active_cell.is_none(),
"interrupt cleanup should clear the active stream tail",
);
assert_eq!(
chat.stream_controller
.as_ref()
.map(crate::streaming::controller::StreamController::queued_lines),
Some(0),
);
}
#[tokio::test]
async fn interrupt_remains_responsive_during_resized_table_stream() {
let (mut chat, mut rx, mut op_rx) = make_chatwidget_manual(None).await;