From 35fee387cf2aba3b7eec9003fa9a99f979db6f17 Mon Sep 17 00:00:00 2001 From: Felipe Coury Date: Sat, 30 May 2026 14:12:55 -0300 Subject: [PATCH] fix(tui): consolidate held plan stream tails --- codex-rs/tui/src/app/event_dispatch.rs | 27 +- codex-rs/tui/src/app_event.rs | 8 +- codex-rs/tui/src/chatwidget/streaming.rs | 57 ++-- .../src/chatwidget/tests/status_and_layout.rs | 295 +++++++++++++++++- codex-rs/tui/src/chatwidget/turn_runtime.rs | 19 +- codex-rs/tui/src/streaming/controller.rs | 16 + 6 files changed, 377 insertions(+), 45 deletions(-) diff --git a/codex-rs/tui/src/app/event_dispatch.rs b/codex-rs/tui/src/app/event_dispatch.rs index f9683e26ea..fa6092fd13 100644 --- a/codex-rs/tui/src/app/event_dispatch.rs +++ b/codex-rs/tui/src/app/event_dispatch.rs @@ -5,6 +5,7 @@ use super::resize_reflow::trailing_run_start; use super::*; +use crate::app_event::ConsolidationScrollbackReflow; const SHUTDOWN_FIRST_EXIT_TIMEOUT: Duration = Duration::from_secs(/*secs*/ 2); @@ -234,11 +235,10 @@ impl App { deferred_history_cell, )?; } - AppEvent::ConsolidateProposedPlan(source) => { - if !self.terminal_resize_reflow_enabled() { - self.transcript_reflow.clear(); - return Ok(AppRunControl::Continue); - } + AppEvent::ConsolidateProposedPlan { + source, + scrollback_reflow, + } => { let end = self.transcript_cells.len(); let start = trailing_run_start::( &self.transcript_cells, @@ -255,7 +255,22 @@ impl App { tui.frame_requester().schedule_frame(); } - self.finish_required_stream_reflow(tui)?; + match scrollback_reflow { + ConsolidationScrollbackReflow::IfResizeReflowRan => { + self.maybe_finish_stream_reflow(tui)?; + } + ConsolidationScrollbackReflow::Required + if self.terminal_resize_reflow_enabled() => + { + self.finish_required_stream_reflow(tui)?; + } + ConsolidationScrollbackReflow::Required => { + // The already-emitted stream prefix cannot be removed without reflow. + // Keep terminal scrollback as-is instead of appending a duplicate full + // plan if the feature was disabled before this queued event arrived. + self.maybe_finish_stream_reflow(tui)?; + } + } } else { self.transcript_cells.push(consolidated.clone()); if let Some(Overlay::Transcript(t)) = &mut self.overlay { diff --git a/codex-rs/tui/src/app_event.rs b/codex-rs/tui/src/app_event.rs index f1e612dd58..e3d50f0b87 100644 --- a/codex-rs/tui/src/app_event.rs +++ b/codex-rs/tui/src/app_event.rs @@ -602,8 +602,12 @@ pub(crate) enum AppEvent { /// end of the transcript with a single source-backed `ProposedPlanCell`. /// /// Emitted by `ChatWidget::on_plan_item_completed` after plan stream - /// finalization. - ConsolidateProposedPlan(String), + /// finalization. `scrollback_reflow` lets table-tail finalization and + /// authoritative plan corrections refresh already-emitted scrollback. + ConsolidateProposedPlan { + source: String, + scrollback_reflow: ConsolidationScrollbackReflow, + }, /// Apply rollback semantics to local transcript cells. /// diff --git a/codex-rs/tui/src/chatwidget/streaming.rs b/codex-rs/tui/src/chatwidget/streaming.rs index 9fef57d6b9..81a9428589 100644 --- a/codex-rs/tui/src/chatwidget/streaming.rs +++ b/codex-rs/tui/src/chatwidget/streaming.rs @@ -4,6 +4,7 @@ //! cells, commit ticks, and interrupt deferral. use super::*; +use crate::app_event::ConsolidationScrollbackReflow; impl ChatWidget { pub(super) fn restore_reasoning_status_header(&mut self) { @@ -19,7 +20,7 @@ impl ChatWidget { pub(super) fn flush_answer_stream_with_separator(&mut self) { let had_stream_controller = self.stream_controller.is_some(); if let Some(mut controller) = self.stream_controller.take() { - let scrollback_reflow = if controller.has_live_tail() + let scrollback_reflow = if controller.requires_forced_reflow_on_finalize() && self.config.features.enabled(Feature::TerminalResizeReflow) { crate::app_event::ConsolidationScrollbackReflow::Required @@ -140,10 +141,11 @@ impl ChatWidget { pub(super) fn on_plan_item_completed(&mut self, text: String) { let streamed_plan = self.transcript.plan_delta_buffer.trim().to_string(); - let plan_text = if text.trim().is_empty() { - streamed_plan - } else { + let has_authoritative_plan_text = !text.trim().is_empty(); + let plan_text = if has_authoritative_plan_text { text + } else { + streamed_plan }; if !plan_text.trim().is_empty() { self.record_agent_markdown(&plan_text); @@ -155,32 +157,49 @@ impl ChatWidget { self.transcript.plan_delta_buffer.clear(); self.transcript.plan_item_active = false; self.transcript.saw_plan_item_this_turn = true; - let (finalized_streamed_cell, consolidated_plan_source) = + let (finalized_streamed_cell, consolidated_plan_source, scrollback_reflow) = if let Some(mut controller) = self.plan_stream_controller.take() { - let had_live_tail = controller.has_live_tail(); + let requires_forced_reflow = controller.requires_forced_reflow_on_finalize() + && self.config.features.enabled(Feature::TerminalResizeReflow); self.clear_active_stream_tail(); let (cell, source) = controller.finalize(); - if had_live_tail { - (None, source) + let requires_authoritative_refresh = source + .as_ref() + .is_some_and(|source| has_authoritative_plan_text && source != &plan_text) + && self.config.features.enabled(Feature::TerminalResizeReflow); + let scrollback_reflow = if requires_forced_reflow || requires_authoritative_refresh + { + ConsolidationScrollbackReflow::Required } else { - (cell, source) + ConsolidationScrollbackReflow::IfResizeReflowRan + }; + let source = source.map(|source| { + if has_authoritative_plan_text { + plan_text.clone() + } else { + source + } + }); + let hold_live_tail_for_reflow = + scrollback_reflow == ConsolidationScrollbackReflow::Required; + if hold_live_tail_for_reflow { + (None, source, scrollback_reflow) + } else { + (cell, source, scrollback_reflow) } } else { - (None, None) + (None, None, ConsolidationScrollbackReflow::IfResizeReflowRan) }; if let Some(cell) = finalized_streamed_cell { self.add_boxed_history(cell); - // TODO: Replace streamed output with the final plan item text if plan streaming is - // removed or if we need to reconcile mismatches between streamed and final content. - if let Some(source) = consolidated_plan_source { - self.app_event_tx - .send(AppEvent::ConsolidateProposedPlan(source)); - } + } + if let Some(source) = consolidated_plan_source { + self.app_event_tx.send(AppEvent::ConsolidateProposedPlan { + source, + scrollback_reflow, + }); } else if !plan_text.is_empty() { self.add_to_history(history_cell::new_proposed_plan(plan_text, &self.config.cwd)); - } else if let Some(source) = consolidated_plan_source { - self.app_event_tx - .send(AppEvent::ConsolidateProposedPlan(source)); } if should_restore_after_stream { self.status_state.pending_status_indicator_restore = true; diff --git a/codex-rs/tui/src/chatwidget/tests/status_and_layout.rs b/codex-rs/tui/src/chatwidget/tests/status_and_layout.rs index 01e609e660..9a6bfdc668 100644 --- a/codex-rs/tui/src/chatwidget/tests/status_and_layout.rs +++ b/codex-rs/tui/src/chatwidget/tests/status_and_layout.rs @@ -408,7 +408,60 @@ async fn flush_answer_stream_inserts_live_list_tail_when_resize_reflow_disabled( } #[tokio::test] -async fn completed_plan_table_tail_skips_provisional_history_insert() { +async fn flush_answer_stream_inserts_live_list_tail_without_forced_reflow() { + let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await; + let cwd = chat.config.cwd.to_path_buf(); + + let mut controller = crate::streaming::controller::StreamController::new( + Some(/*width*/ 80), + cwd.as_path(), + HistoryRenderMode::Rich, + ); + controller.push("- first\n"); + controller.push("- second\n"); + assert!( + controller.has_live_tail(), + "expected trailing list holdback to leave a live tail for this regression", + ); + chat.stream_controller = Some(controller); + + while rx.try_recv().is_ok() {} + + chat.flush_answer_stream_with_separator(); + + let mut saw_consolidate = false; + let mut saw_insert_history = false; + while let Ok(event) = rx.try_recv() { + match event { + AppEvent::InsertHistoryCell(_) => saw_insert_history = true, + AppEvent::ConsolidateAgentMessage { + scrollback_reflow, + deferred_history_cell, + .. + } => { + saw_consolidate = true; + assert_eq!( + scrollback_reflow, + crate::app_event::ConsolidationScrollbackReflow::IfResizeReflowRan + ); + assert!(deferred_history_cell.is_none()); + } + _ => {} + } + } + + assert!( + saw_consolidate, + "expected stream finalization to consolidate" + ); + assert!( + saw_insert_history, + "live list tail should insert history directly without forced reflow" + ); +} + +#[tokio::test] +async fn completed_plan_table_tail_uses_consolidation_without_provisional_history_insert() { let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await; let cwd = chat.config.cwd.to_path_buf(); @@ -432,27 +485,241 @@ async fn completed_plan_table_tail_skips_provisional_history_insert() { chat.on_plan_item_completed(String::new()); - let mut saw_source_backed_plan = false; + let mut saw_consolidate = false; + let mut saw_stream_plan = false; + while let Ok(event) = rx.try_recv() { + match event { + AppEvent::ConsolidateProposedPlan { + source, + scrollback_reflow, + } => { + saw_consolidate = true; + assert!(source.contains("| Verify | Codex |")); + assert_eq!( + scrollback_reflow, + crate::app_event::ConsolidationScrollbackReflow::Required + ); + } + AppEvent::InsertHistoryCell(cell) => { + saw_stream_plan |= cell.as_any().is::(); + } + _ => {} + } + } + + assert!(saw_consolidate, "expected source-backed plan consolidation"); + assert!( + !saw_stream_plan, + "live plan table tail should not be inserted provisionally" + ); +} + +#[tokio::test] +async fn completed_plan_list_tail_consolidates_emitted_prefix() { + let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await; + let cwd = chat.config.cwd.to_path_buf(); + let completed_plan = "Plan intro.\n\n- Revised first step\n- Revised second step\n"; + + let mut controller = crate::streaming::controller::PlanStreamController::new( + Some(/*width*/ 80), + cwd.as_path(), + HistoryRenderMode::Rich, + ); + controller.push("Plan intro.\n\n"); + let (prefix, _) = controller.on_commit_tick_batch(usize::MAX); + chat.add_boxed_history(prefix.expect("expected emitted plan prefix")); + controller.push("- First step\n"); + controller.push("- Second step\n"); + assert!( + controller.has_live_tail(), + "expected trailing list holdback to leave a live tail", + ); + chat.plan_stream_controller = Some(controller); + chat.transcript.plan_delta_buffer = "Plan intro.\n\n- First step\n- Second step\n".to_string(); + + while rx.try_recv().is_ok() {} + + chat.on_plan_item_completed(completed_plan.to_string()); + + let mut saw_consolidate = false; + let mut saw_direct_plan_insert = false; + while let Ok(event) = rx.try_recv() { + match event { + AppEvent::ConsolidateProposedPlan { + source, + scrollback_reflow, + } => { + saw_consolidate = true; + assert_eq!(source, completed_plan); + assert_eq!( + scrollback_reflow, + crate::app_event::ConsolidationScrollbackReflow::Required + ); + } + AppEvent::InsertHistoryCell(cell) => { + saw_direct_plan_insert |= cell.as_any().is::(); + } + _ => {} + } + } + + assert!(saw_consolidate, "expected emitted prefix consolidation"); + assert!( + !saw_direct_plan_insert, + "held list should consolidate instead of duplicating the emitted prefix", + ); +} + +#[tokio::test] +async fn completed_plan_list_tail_inserts_stream_tail_without_forced_reflow() { + let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await; + let cwd = chat.config.cwd.to_path_buf(); + + let mut controller = crate::streaming::controller::PlanStreamController::new( + Some(/*width*/ 80), + cwd.as_path(), + HistoryRenderMode::Rich, + ); + controller.push("- First step\n"); + controller.push("- Second step\n"); + assert!( + controller.has_live_tail(), + "expected trailing list holdback to leave a live tail", + ); + chat.plan_stream_controller = Some(controller); + chat.transcript.plan_delta_buffer = "- First step\n- Second step\n".to_string(); + + while rx.try_recv().is_ok() {} + + chat.on_plan_item_completed(String::new()); + + let mut saw_consolidate = false; + let mut saw_stream_plan = false; + while let Ok(event) = rx.try_recv() { + match event { + AppEvent::ConsolidateProposedPlan { + source, + scrollback_reflow, + } => { + saw_consolidate = true; + assert_eq!(source, "- First step\n- Second step\n"); + assert_eq!( + scrollback_reflow, + crate::app_event::ConsolidationScrollbackReflow::IfResizeReflowRan + ); + } + AppEvent::InsertHistoryCell(cell) => { + saw_stream_plan |= cell.as_any().is::(); + } + _ => {} + } + } + + assert!(saw_consolidate, "expected source-backed plan consolidation"); + assert!( + saw_stream_plan, + "live plan list tail should insert history directly without forced reflow", + ); +} + +#[tokio::test] +async fn completed_plan_list_tail_inserts_stream_tail_when_resize_reflow_disabled() { + let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await; + let cwd = chat.config.cwd.to_path_buf(); + let completed_plan = "Plan intro.\n\n- Revised first step\n- Revised second step\n"; + chat.set_feature_enabled(Feature::TerminalResizeReflow, /*enabled*/ false); + + let mut controller = crate::streaming::controller::PlanStreamController::new( + Some(/*width*/ 80), + cwd.as_path(), + HistoryRenderMode::Rich, + ); + controller.push("Plan intro.\n\n"); + let (prefix, _) = controller.on_commit_tick_batch(usize::MAX); + chat.add_boxed_history(prefix.expect("expected emitted plan prefix")); + controller.push("- First step\n"); + controller.push("- Second step\n"); + assert!( + controller.has_live_tail(), + "expected trailing list holdback to leave a live tail", + ); + chat.plan_stream_controller = Some(controller); + chat.transcript.plan_delta_buffer = "Plan intro.\n\n- First step\n- Second step\n".to_string(); + + while rx.try_recv().is_ok() {} + + chat.on_plan_item_completed(completed_plan.to_string()); + + let mut saw_consolidate = false; + let mut saw_stream_plan = false; + let mut saw_direct_plan_insert = false; + while let Ok(event) = rx.try_recv() { + match event { + AppEvent::ConsolidateProposedPlan { + source, + scrollback_reflow, + } => { + saw_consolidate = true; + assert_eq!(source, completed_plan); + assert_eq!( + scrollback_reflow, + crate::app_event::ConsolidationScrollbackReflow::IfResizeReflowRan + ); + } + AppEvent::InsertHistoryCell(cell) => { + saw_stream_plan |= cell.as_any().is::(); + saw_direct_plan_insert |= cell.as_any().is::(); + } + _ => {} + } + } + + assert!(saw_consolidate, "expected source-backed plan consolidation"); + assert!( + saw_stream_plan, + "authoritative correction should finish the streamed plan when resize reflow is disabled", + ); + assert!( + !saw_direct_plan_insert, + "held list should not duplicate the emitted prefix with a full plan insert", + ); +} + +#[tokio::test] +async fn task_completion_inserts_live_plan_list_tail_when_resize_reflow_disabled() { + let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await; + let cwd = chat.config.cwd.to_path_buf(); + chat.set_feature_enabled(Feature::TerminalResizeReflow, /*enabled*/ false); + + let mut controller = crate::streaming::controller::PlanStreamController::new( + Some(/*width*/ 80), + cwd.as_path(), + HistoryRenderMode::Rich, + ); + controller.push("- First step\n"); + controller.push("- Second step\n"); + assert!( + controller.has_live_tail(), + "expected trailing list holdback to leave a live tail", + ); + chat.plan_stream_controller = Some(controller); + + while rx.try_recv().is_ok() {} + + chat.on_task_complete( + /*last_agent_message*/ None, /*duration_ms*/ None, /*from_replay*/ false, + ); + let mut saw_stream_plan = false; - let mut rendered_plan = String::new(); while let Ok(event) = rx.try_recv() { if let AppEvent::InsertHistoryCell(cell) = event { - if cell.as_any().is::() { - saw_source_backed_plan = true; - rendered_plan = lines_to_single_string(&cell.display_lines(/*width*/ 80)); - } saw_stream_plan |= cell.as_any().is::(); } } - assert!(saw_source_backed_plan, "expected source-backed plan insert"); assert!( - rendered_plan.contains('━'), - "expected completed plan table to render with separators, got: {rendered_plan:?}" - ); - assert!( - !saw_stream_plan, - "live plan table tail should not be inserted provisionally" + saw_stream_plan, + "live plan list tail should insert history directly without resize reflow", ); } diff --git a/codex-rs/tui/src/chatwidget/turn_runtime.rs b/codex-rs/tui/src/chatwidget/turn_runtime.rs index 4868d6a25c..01995acdf2 100644 --- a/codex-rs/tui/src/chatwidget/turn_runtime.rs +++ b/codex-rs/tui/src/chatwidget/turn_runtime.rs @@ -4,6 +4,7 @@ //! and final-message separator handling. use super::*; +use crate::app_event::ConsolidationScrollbackReflow; impl ChatWidget { /// Synchronize the bottom-pane "task running" indicator with the current lifecycles. @@ -117,15 +118,25 @@ impl ChatWidget { // If a stream is currently active, finalize it. self.flush_answer_stream_with_separator(); if let Some(mut controller) = self.plan_stream_controller.take() { - let had_live_tail = controller.has_live_tail(); + let scrollback_reflow = if controller.requires_forced_reflow_on_finalize() + && self.config.features.enabled(Feature::TerminalResizeReflow) + { + ConsolidationScrollbackReflow::Required + } else { + ConsolidationScrollbackReflow::IfResizeReflowRan + }; + let hold_live_tail_for_reflow = + scrollback_reflow == ConsolidationScrollbackReflow::Required; self.clear_active_stream_tail(); let (cell, source) = controller.finalize(); - if !had_live_tail && let Some(cell) = cell { + if !hold_live_tail_for_reflow && let Some(cell) = cell { self.add_boxed_history(cell); } if let Some(source) = source { - self.app_event_tx - .send(AppEvent::ConsolidateProposedPlan(source)); + self.app_event_tx.send(AppEvent::ConsolidateProposedPlan { + source, + scrollback_reflow, + }); } } self.flush_unified_exec_wait_streak(); diff --git a/codex-rs/tui/src/streaming/controller.rs b/codex-rs/tui/src/streaming/controller.rs index 9246f6736a..8bfcaf1255 100644 --- a/codex-rs/tui/src/streaming/controller.rs +++ b/codex-rs/tui/src/streaming/controller.rs @@ -221,6 +221,10 @@ impl StreamCore { self.enqueued_stable_len < self.rendered_lines.len() } + fn requires_forced_reflow_on_finalize(&self) -> bool { + self.has_tail() && !matches!(self.holdback_scanner.state(), TableHoldbackState::None) + } + /// Update rendering width and rebuild queued stable lines for the new layout. /// /// Re-renders once at the new width and rebuilds queue state from the @@ -542,11 +546,17 @@ impl StreamController { !self.header_emitted && self.core.enqueued_stable_len == 0 } + #[cfg(test)] #[inline] pub(crate) fn has_live_tail(&self) -> bool { self.core.has_tail() } + #[inline] + pub(crate) fn requires_forced_reflow_on_finalize(&self) -> bool { + self.core.requires_forced_reflow_on_finalize() + } + pub(crate) fn clear_queue(&mut self) { self.core.state.clear_queue(); self.core.enqueued_stable_len = self.core.emitted_stable_len; @@ -645,11 +655,17 @@ impl PlanStreamController { self.core.queued_lines() } + #[cfg(test)] #[inline] pub(crate) fn has_live_tail(&self) -> bool { self.core.has_tail() } + #[inline] + pub(crate) fn requires_forced_reflow_on_finalize(&self) -> bool { + self.core.requires_forced_reflow_on_finalize() + } + #[inline] pub(crate) fn current_tail_lines(&self) -> Vec { self.core.current_tail_lines()