From 70bc4fc22692fa80b9bf3eea07b8211493c000ff Mon Sep 17 00:00:00 2001 From: Felipe Coury Date: Sun, 22 Feb 2026 16:42:11 -0300 Subject: [PATCH] fix(tui): keep plan streams incremental during table output Disable table holdback for `PlanStreamController` by adding a per-stream `TableHoldbackMode` in `StreamCore`. Agent streams keep holdback enabled, while plan streams now enqueue table lines incrementally. Also treat plan live tail as active in `stream_controllers_idle()` and add regression coverage in `codex-rs/tui/src/streaming/controller.rs` and `codex-rs/tui/src/chatwidget/tests.rs` to prevent premature `StopCommitAnimation` and preserve streaming behavior. --- codex-rs/tui/src/chatwidget.rs | 2 +- .../src/chatwidget/tests/status_and_layout.rs | 38 +++++++ codex-rs/tui/src/streaming/controller.rs | 99 +++++++++++++++---- 3 files changed, 118 insertions(+), 21 deletions(-) diff --git a/codex-rs/tui/src/chatwidget.rs b/codex-rs/tui/src/chatwidget.rs index 107aaf5f86..6c059ab7fb 100644 --- a/codex-rs/tui/src/chatwidget.rs +++ b/codex-rs/tui/src/chatwidget.rs @@ -1768,7 +1768,7 @@ impl ChatWidget { && self .plan_stream_controller .as_ref() - .map(|controller| controller.queued_lines() == 0) + .map(|controller| controller.queued_lines() == 0 && !controller.has_live_tail()) .unwrap_or(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 df8a2281e1..c80d638769 100644 --- a/codex-rs/tui/src/chatwidget/tests/status_and_layout.rs +++ b/codex-rs/tui/src/chatwidget/tests/status_and_layout.rs @@ -267,6 +267,44 @@ async fn flush_answer_stream_does_not_stop_animation_while_plan_stream_is_active ); } +#[tokio::test] +async fn flush_answer_stream_does_not_stop_animation_while_plan_table_stream_is_active() { + let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(None).await; + let cwd = chat.config.cwd.to_path_buf(); + + let mut plan_controller = + crate::streaming::controller::PlanStreamController::new(Some(80), cwd.as_path()); + assert!( + plan_controller.push("| Step | Owner |\n"), + "expected table header to enqueue while plan stream is active", + ); + assert!( + plan_controller.queued_lines() > 0, + "expected queued table header lines for this repro", + ); + chat.plan_stream_controller = Some(plan_controller); + chat.stream_controller = Some(crate::streaming::controller::StreamController::new( + Some(80), + cwd.as_path(), + )); + + while rx.try_recv().is_ok() {} + + chat.flush_answer_stream_with_separator(); + + let mut saw_stop = false; + while let Ok(event) = rx.try_recv() { + if matches!(event, AppEvent::StopCommitAnimation) { + saw_stop = true; + } + } + + assert!( + !saw_stop, + "did not expect StopCommitAnimation while plan table stream is still active", + ); +} + #[tokio::test] async fn helpers_are_available_and_do_not_panic() { let (tx_raw, _rx) = unbounded_channel::(); diff --git a/codex-rs/tui/src/streaming/controller.rs b/codex-rs/tui/src/streaming/controller.rs index d49e80fe12..76486cd177 100644 --- a/codex-rs/tui/src/streaming/controller.rs +++ b/codex-rs/tui/src/streaming/controller.rs @@ -15,8 +15,10 @@ //! every column's width and reshape all prior rows. The holdback mechanism //! (`table_holdback_state`) detects pipe-table patterns (header + delimiter //! pair) in the accumulated source and keeps content from the table header -//! onward as mutable tail until the stream finalizes. Lines in `Outside` and `Markdown` -//! fence contexts are scanned; lines inside non-markdown fences are skipped. +//! onward as mutable tail until the stream finalizes. Holdback is enabled for +//! agent streams and disabled for proposed-plan streams so plan output +//! continues to stream incrementally. Lines in `Outside` and `Markdown` fence +//! contexts are scanned; lines inside non-markdown fences are skipped. //! //! ## Resize handling //! @@ -70,6 +72,8 @@ use super::StreamState; /// finalize return types. struct StreamCore { state: StreamState, + /// Whether table holdback is active for this stream type. + table_holdback_mode: TableHoldbackMode, /// Current rendering width (columns available for markdown content). width: Option, /// Accumulated raw markdown source for the current stream. @@ -94,10 +98,17 @@ struct StablePrefixLenCache { stable_prefix_len: usize, } +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum TableHoldbackMode { + Enabled, + Disabled, +} + impl StreamCore { - fn new(width: Option, cwd: &Path) -> Self { + fn new(width: Option, table_holdback_mode: TableHoldbackMode, cwd: &Path) -> Self { Self { state: StreamState::new(width, cwd), + table_holdback_mode, width, raw_source: String::with_capacity(1024), rendered_lines: Vec::with_capacity(64), @@ -122,7 +133,9 @@ impl StreamCore { && let Some(committed_source) = self.state.collector.commit_complete_source() { self.raw_source.push_str(&committed_source); - self.holdback_scanner.push_source_chunk(&committed_source); + if self.table_holdback_mode == TableHoldbackMode::Enabled { + self.holdback_scanner.push_source_chunk(&committed_source); + } self.recompute_streaming_render(); enqueued = self.sync_stable_queue(); } @@ -135,7 +148,9 @@ impl StreamCore { let remainder_source = self.state.collector.finalize_and_drain_source(); if !remainder_source.is_empty() { self.raw_source.push_str(&remainder_source); - self.holdback_scanner.push_source_chunk(&remainder_source); + if self.table_holdback_mode == TableHoldbackMode::Enabled { + self.holdback_scanner.push_source_chunk(&remainder_source); + } } let mut rendered = Vec::new(); append_markdown_agent_with_cwd( @@ -322,6 +337,9 @@ impl StreamCore { /// streaming. When no table is detected, everything flows directly to /// stable. This is the core decision point for the holdback mechanism. fn active_tail_budget_lines(&mut self) -> usize { + if self.table_holdback_mode == TableHoldbackMode::Disabled { + return 0; + } let scan_start = Instant::now(); let holdback_state = self.holdback_scanner.state(); let tail_budget = match holdback_state { @@ -403,7 +421,7 @@ impl StreamController { /// render against the same session cwd that was active when streaming started. pub(crate) fn new(width: Option, cwd: &Path) -> Self { Self { - core: StreamCore::new(width, cwd), + core: StreamCore::new(width, TableHoldbackMode::Enabled, cwd), header_emitted: false, } } @@ -510,7 +528,7 @@ impl PlanStreamController { /// render against the same session cwd that was active when streaming started. pub(crate) fn new(width: Option, cwd: &Path) -> Self { Self { - core: StreamCore::new(width, cwd), + core: StreamCore::new(width, TableHoldbackMode::Disabled, cwd), header_emitted: false, top_padding_emitted: false, } @@ -554,6 +572,11 @@ impl PlanStreamController { self.core.queued_lines() } + #[inline] + pub(crate) fn has_live_tail(&self) -> bool { + self.core.has_tail() + } + pub(crate) fn oldest_queued_age(&self, now: Instant) -> Option { self.core.oldest_queued_age(now) } @@ -968,6 +991,19 @@ mod tests { assert!(!ctrl.has_live_tail()); } + #[test] + fn plan_controller_has_live_tail_reflects_tail_presence() { + let mut ctrl = plan_stream_controller(Some(80)); + assert!(!ctrl.has_live_tail()); + + ctrl.core.rendered_lines = vec![Line::from("tail line")]; + ctrl.core.enqueued_stable_len = 0; + assert!(ctrl.has_live_tail()); + + ctrl.core.enqueued_stable_len = 1; + assert!(!ctrl.has_live_tail()); + } + #[test] fn controller_set_width_partial_drain_no_lost_lines() { let mut ctrl = stream_controller(Some(40)); @@ -1047,7 +1083,7 @@ mod tests { #[test] fn controller_set_width_preserves_table_tail_when_queue_is_empty() { - let mut ctrl = StreamController::new(Some(80)); + let mut ctrl = stream_controller(Some(80)); ctrl.push("intro line\n"); let (_cell, idle) = ctrl.on_commit_tick(); @@ -1098,6 +1134,19 @@ mod tests { ); } + #[test] + fn plan_controller_streams_table_header_incrementally() { + let mut ctrl = plan_stream_controller(Some(80)); + assert!(ctrl.push("Intro\n")); + let (_cell, idle) = ctrl.on_commit_tick_batch(usize::MAX); + assert!(idle, "intro line should fully drain"); + + assert!( + ctrl.push("| Step | Owner |\n"), + "table header should enqueue incrementally for plan streams", + ); + } + #[test] fn controller_loose_vs_tight_with_commit_ticks_matches_full() { let mut ctrl = stream_controller(/*width*/ None); @@ -1573,7 +1622,7 @@ mod tests { fn controller_live_view_matches_render_during_interleaved_table_streaming() { let source = "Project updates are easier to scan when narrative and structured data alternate.\n\n| Focus Area | Owner | Priority | Status |\n|---|---|---|---|\n| Authentication cleanup | Maya | High | 80% |\n| CLI error messages | Jordan | Medium | 55% |\n| Docs refresh | Lee | Low | 30% |\n\nThe first checkpoint shows progress, but we still have open risks.\n\n| Task | Command / Artifact | Due | State |\n|---|---|---|---|\n| Run unit tests | `cargo test -p codex-core` | Today | ✅ |\n| Snapshot review | `cargo insta pending-snapshots -p codex-tui` | Today | ⏳ |\n| Changelog draft | Release template (https://replacechangelog.com/) | Tomorrow | 📝 |\n\nFinal sign-off criteria are summarized below.\n"; let width = Some(72usize); - let mut ctrl = StreamController::new(width); + let mut ctrl = stream_controller(width); let mut emitted_lines: Vec> = Vec::new(); for delta in source.split_inclusive('\n') { @@ -1639,7 +1688,7 @@ mod tests { } #[test] - fn plan_controller_streamed_table_matches_finalize_render() { + fn plan_controller_streamed_table_emits_incremental_rows_before_finalize() { let deltas = vec![ "## Build plan\n\n", "| Step | Owner |\n", @@ -1653,21 +1702,23 @@ mod tests { let source: String = deltas.iter().copied().collect(); let baseline = collect_plan_streamed_lines(&[source.as_str()], Some(80)); - assert_eq!(streamed, baseline); + assert_ne!(streamed, baseline); assert!( - streamed.iter().any(|line| line.contains('┌')), - "expected unicode table border in plan streamed output: {streamed:?}" + streamed + .iter() + .any(|line| line.contains('│') || line.contains('└') || line.contains('┌')), + "expected unicode table box drawing chars in plan streamed output: {streamed:?}" ); assert!( - !streamed + streamed .iter() .any(|line| line.trim() == "| Step | Owner |"), - "did not expect raw table header line in plan output: {streamed:?}" + "expected incremental raw table header line in plan output: {streamed:?}" ); } #[test] - fn plan_controller_streamed_markdown_fenced_table_matches_finalize_render() { + fn plan_controller_streamed_markdown_fenced_table_emits_incremental_rows_before_finalize() { let deltas = vec![ "## Build plan\n\n", "```md\n", @@ -1683,10 +1734,18 @@ mod tests { let source: String = deltas.iter().copied().collect(); let baseline = collect_plan_streamed_lines(&[source.as_str()], Some(80)); - assert_eq!(streamed, baseline); + assert_ne!(streamed, baseline); assert!( - streamed.iter().any(|line| line.contains('┌')), - "expected unicode table border in fenced plan output: {streamed:?}" + streamed + .iter() + .any(|line| line.contains('│') || line.contains('└') || line.contains('┌')), + "expected unicode table box drawing chars in fenced plan output: {streamed:?}" + ); + assert!( + streamed + .iter() + .any(|line| line.trim() == "| Step | Owner |"), + "expected incremental raw table header line in fenced plan output: {streamed:?}" ); } @@ -1840,7 +1899,7 @@ mod tests { #[test] fn controller_set_width_partial_wrapped_emit_keeps_wrapped_remainder() { - let mut ctrl = StreamController::new(Some(18)); + let mut ctrl = stream_controller(Some(18)); ctrl.push("alpha beta gamma delta epsilon zeta eta theta iota kappa lambda mu\n"); let (first_emit, idle) = ctrl.on_commit_tick();