diff --git a/codex-rs/tui/src/chatwidget/streaming.rs b/codex-rs/tui/src/chatwidget/streaming.rs index 3dd9f26ee9..562dc4737d 100644 --- a/codex-rs/tui/src/chatwidget/streaming.rs +++ b/codex-rs/tui/src/chatwidget/streaming.rs @@ -19,7 +19,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_final_scrollback_reflow() { crate::app_event::ConsolidationScrollbackReflow::Required } else { crate::app_event::ConsolidationScrollbackReflow::IfResizeReflowRan @@ -158,10 +158,10 @@ impl ChatWidget { self.transcript.saw_plan_item_this_turn = true; let (finalized_streamed_cell, consolidated_plan_source) = if let Some(mut controller) = self.plan_stream_controller.take() { - let had_live_tail = controller.has_live_tail(); + let requires_scrollback_reflow = controller.requires_final_scrollback_reflow(); self.clear_active_stream_tail(); let (cell, source) = controller.finalize(); - if had_live_tail { + if requires_scrollback_reflow { (None, source) } else { (cell, source) 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 3d08daa427..fab814a6f6 100644 --- a/codex-rs/tui/src/chatwidget/tests/status_and_layout.rs +++ b/codex-rs/tui/src/chatwidget/tests/status_and_layout.rs @@ -259,7 +259,10 @@ async fn flush_answer_stream_keeps_default_reflow_for_plain_text_tail() { cwd.as_path(), HistoryRenderMode::Rich, ); - assert!(controller.push("plain response line\n")); + assert!( + !controller.push("plain response line\n"), + "short prose should stay in the ordinary mutable tail until finalization", + ); chat.stream_controller = Some(controller); while rx.try_recv().is_ok() {} diff --git a/codex-rs/tui/src/chatwidget/turn_runtime.rs b/codex-rs/tui/src/chatwidget/turn_runtime.rs index bbea783cf2..f135c93bea 100644 --- a/codex-rs/tui/src/chatwidget/turn_runtime.rs +++ b/codex-rs/tui/src/chatwidget/turn_runtime.rs @@ -117,10 +117,10 @@ 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 requires_scrollback_reflow = controller.requires_final_scrollback_reflow(); self.clear_active_stream_tail(); let (cell, source) = controller.finalize(); - if !had_live_tail && let Some(cell) = cell { + if !requires_scrollback_reflow && let Some(cell) = cell { self.add_boxed_history(cell); } if let Some(source) = source { diff --git a/codex-rs/tui/src/streaming/controller.rs b/codex-rs/tui/src/streaming/controller.rs index 95a3b72269..622a8717f3 100644 --- a/codex-rs/tui/src/streaming/controller.rs +++ b/codex-rs/tui/src/streaming/controller.rs @@ -89,6 +89,13 @@ struct StreamCore { stable_prefix_len_cache: Option, /// Incremental holdback scanner state for append-only source updates. holdback_scanner: TableHoldbackScanner, + /// Number of source lines to keep mutable even when no structural holdback is active. + /// + /// CommonMark can reinterpret the newest block after the next line arrives. For example, a + /// paragraph after a fenced code block inside a list can shift when the following blank/list + /// marker is parsed. Holding a small source suffix keeps that boundary out of scrollback until + /// it is stable. + default_tail_lines: usize, } struct StablePrefixLenCache { @@ -104,7 +111,12 @@ struct StablePrefixLenCache { } impl StreamCore { - fn new(width: Option, cwd: &Path, render_mode: HistoryRenderMode) -> Self { + fn new( + width: Option, + cwd: &Path, + render_mode: HistoryRenderMode, + default_tail_lines: usize, + ) -> Self { Self { state: StreamState::new(width, cwd), width, @@ -116,6 +128,7 @@ impl StreamCore { render_mode, stable_prefix_len_cache: None, holdback_scanner: TableHoldbackScanner::new(), + default_tail_lines, } } @@ -216,6 +229,7 @@ impl StreamCore { self.rendered_lines[start..].to_vec() } + #[cfg(test)] #[inline] fn has_tail(&self) -> bool { self.enqueued_stable_len < self.rendered_lines.len() @@ -235,7 +249,7 @@ impl StreamCore { return; } let had_pending_queue = self.state.queued_len() > 0; - let had_live_tail = self.has_tail(); + let had_structural_tail = self.requires_final_scrollback_reflow(); self.width = width; self.state.collector.set_width(width); if self.raw_source.is_empty() { @@ -244,21 +258,25 @@ impl StreamCore { self.recompute_streaming_render(); self.emitted_stable_len = self.emitted_stable_len.min(self.rendered_lines.len()); + let target_stable_len = self.compute_target_stable_len(); if had_pending_queue - && self.emitted_stable_len == self.rendered_lines.len() - && self.emitted_stable_len > 0 + && self.emitted_stable_len >= target_stable_len + && target_stable_len > 0 { // If wrapped remainder compresses into fewer lines at the new width, // keep at least one line un-emitted so pre-resize pending content is // not skipped permanently. - self.emitted_stable_len -= 1; + self.emitted_stable_len = target_stable_len - 1; } self.state.clear_queue(); - if self.emitted_stable_len > 0 && !had_pending_queue && !had_live_tail { + if self.emitted_stable_len > 0 && !had_pending_queue && !had_structural_tail { // Avoid replaying already-emitted content after resize when no // stable lines were waiting in the queue and there was no mutable - // tail to preserve. - self.enqueued_stable_len = self.rendered_lines.len(); + // structural tail to preserve. Ordinary source-line tails are + // remapped to the new render suffix and stay mutable. + let target_stable_len = self.compute_target_stable_len(); + self.emitted_stable_len = target_stable_len; + self.enqueued_stable_len = target_stable_len; return; } self.rebuild_stable_queue_from_render(); @@ -296,7 +314,7 @@ impl StreamCore { } let had_pending_queue = self.state.queued_len() > 0; - let had_live_tail = self.has_tail(); + let had_structural_tail = self.requires_final_scrollback_reflow(); self.render_mode = render_mode; if self.raw_source.is_empty() { return; @@ -304,15 +322,18 @@ impl StreamCore { self.recompute_streaming_render(); self.emitted_stable_len = self.emitted_stable_len.min(self.rendered_lines.len()); + let target_stable_len = self.compute_target_stable_len(); if had_pending_queue - && self.emitted_stable_len == self.rendered_lines.len() - && self.emitted_stable_len > 0 + && self.emitted_stable_len >= target_stable_len + && target_stable_len > 0 { - self.emitted_stable_len -= 1; + self.emitted_stable_len = target_stable_len - 1; } self.state.clear_queue(); - if self.emitted_stable_len > 0 && !had_pending_queue && !had_live_tail { - self.enqueued_stable_len = self.rendered_lines.len(); + if self.emitted_stable_len > 0 && !had_pending_queue && !had_structural_tail { + let target_stable_len = self.compute_target_stable_len(); + self.emitted_stable_len = target_stable_len; + self.enqueued_stable_len = target_stable_len; return; } self.rebuild_stable_queue_from_render(); @@ -389,7 +410,7 @@ impl StreamCore { | TableHoldbackState::PendingHeader { header_start: start, } => self.tail_budget_from_source_start(start), - TableHoldbackState::None => 0, + TableHoldbackState::None => self.default_tail_budget_lines(), }; tracing::trace!( state = ?holdback_state, @@ -454,6 +475,44 @@ impl StreamCore { }); stable_prefix_len } + + fn default_tail_budget_lines(&mut self) -> usize { + if self.default_tail_lines == 0 || self.rendered_lines.is_empty() { + return 0; + } + let Some(source_start) = tail_source_start(&self.raw_source, self.default_tail_lines) + else { + return 0; + }; + self.tail_budget_from_source_start(source_start) + } + + fn requires_final_scrollback_reflow(&self) -> bool { + self.render_mode != HistoryRenderMode::Raw + && !matches!(self.holdback_scanner.state(), TableHoldbackState::None) + } +} + +fn tail_source_start(source: &str, tail_lines: usize) -> Option { + if source.is_empty() || tail_lines == 0 { + return None; + } + + let mut end = source.len(); + if source.as_bytes().last() == Some(&b'\n') { + end = end.saturating_sub(1); + } + let content_end = end; + + let mut start = 0; + for _ in 0..tail_lines { + start = source[..end].rfind('\n').map(|idx| idx + 1).unwrap_or(0); + end = start.saturating_sub(1); + } + if start == content_end && start > 0 { + start = source[..end].rfind('\n').map(|idx| idx + 1).unwrap_or(0); + } + Some(start) } /// Controller for streaming agent message content with table-aware holdback. @@ -472,7 +531,7 @@ impl StreamController { /// the old viewport until app-level reflow repairs the finalized transcript. pub(crate) fn new(width: Option, cwd: &Path, render_mode: HistoryRenderMode) -> Self { Self { - core: StreamCore::new(width, cwd, render_mode), + core: StreamCore::new(width, cwd, render_mode, /*default_tail_lines*/ 1), header_emitted: false, } } @@ -532,11 +591,16 @@ 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() } + pub(crate) fn requires_final_scrollback_reflow(&self) -> bool { + self.core.requires_final_scrollback_reflow() + } + pub(crate) fn clear_queue(&mut self) { self.core.state.clear_queue(); self.core.enqueued_stable_len = self.core.emitted_stable_len; @@ -585,7 +649,7 @@ impl PlanStreamController { /// callers must update it when the terminal width changes. pub(crate) fn new(width: Option, cwd: &Path, render_mode: HistoryRenderMode) -> Self { Self { - core: StreamCore::new(width, cwd, render_mode), + core: StreamCore::new(width, cwd, render_mode, /*default_tail_lines*/ 0), header_emitted: false, top_padding_emitted: false, } @@ -635,11 +699,16 @@ impl PlanStreamController { self.core.queued_lines() } + #[cfg(test)] #[inline] pub(crate) fn has_live_tail(&self) -> bool { self.core.has_tail() } + pub(crate) fn requires_final_scrollback_reflow(&self) -> bool { + self.core.requires_final_scrollback_reflow() + } + #[inline] pub(crate) fn current_tail_lines(&self) -> Vec { self.core.current_tail_lines() @@ -808,7 +877,8 @@ mod tests { fn controller_set_width_rebuilds_queued_lines() { let mut ctrl = stream_controller(Some(120)); let delta = "This is a long line that should wrap into multiple rows when resized.\n"; - assert!(ctrl.push(delta)); + assert!(!ctrl.push(delta)); + assert!(ctrl.push("tail line\n")); assert_eq!(ctrl.queued_lines(), 1); ctrl.set_width(Some(24)); @@ -832,6 +902,7 @@ mod tests { let line = "This is a long line that definitely wraps when the terminal shrinks to 24 columns.\n"; ctrl.push(line); + ctrl.push("tail line\n"); let (cell, _) = ctrl.on_commit_tick_batch(usize::MAX); assert!(cell.is_some(), "expected emitted cell"); assert_eq!(ctrl.queued_lines(), 0); @@ -848,7 +919,8 @@ mod tests { #[test] fn controller_tick_batch_zero_is_noop() { let mut ctrl = stream_controller(Some(80)); - assert!(ctrl.push("line one\n")); + assert!(!ctrl.push("line one\n")); + assert!(ctrl.push("line two\n")); assert_eq!(ctrl.queued_lines(), 1); let (cell, idle) = ctrl.on_commit_tick_batch(/*max_lines*/ 0); @@ -913,6 +985,33 @@ mod tests { assert!(!ctrl.has_live_tail()); } + #[test] + fn controller_does_not_duplicate_list_continuation_after_fenced_code() { + let lines = collect_streamed_lines( + &[ + "**Happy Path**\n", + "\n", + "1. In the second terminal, make the smoke home non-writable:\n", + " ```fish\n", + " chmod a-w $CODEX_SMOKE_HOME\n", + " ```\n", + " Expected: the running TUI stays open.\n", + "\n", + "2. In the TUI, type `/model` and press Enter.\n", + ], + Some(/*width*/ 80), + ); + + let expected_count = lines + .iter() + .filter(|line| line.trim() == "Expected: the running TUI stays open.") + .count(); + assert_eq!( + expected_count, 1, + "expected list continuation to render once, got {lines:?}", + ); + } + #[test] fn controller_live_tail_rerenders_table_tail_after_resize() { let mut ctrl = stream_controller(Some(96)); @@ -997,8 +1096,17 @@ mod tests { } assert!( - drained.iter().any(|l| l.contains("second line")), - "pending lines should continue draining after resize; got {drained:?}", + drained.iter().any(|l| l.contains("IIII JJJJ")), + "pending wrapped lines should continue draining after resize; got {drained:?}", + ); + + let (cell, _source) = ctrl.finalize(); + let final_lines = cell + .map(|cell| lines_to_plain_strings(&cell.transcript_lines(u16::MAX))) + .unwrap_or_default(); + assert!( + final_lines.iter().any(|line| line.contains("second line")), + "ordinary mutable tail should flush on finalize; got {final_lines:?}", ); } @@ -1028,6 +1136,8 @@ mod tests { assert_eq!(ctrl.queued_lines(), 0, "expected empty queue before table"); ctrl.push("| A | B |\n"); + let (_cell, idle) = ctrl.on_commit_tick(); + assert!(idle, "pre-table intro should drain before resize"); assert_eq!( ctrl.queued_lines(), 0, @@ -1247,7 +1357,7 @@ mod tests { let mut ctrl = stream_controller(Some(80)); ctrl.push("Intro line before table.\n"); - assert_eq!(ctrl.queued_lines(), 1); + assert_eq!(ctrl.queued_lines(), 0); ctrl.push("| Key | Value |\n"); ctrl.push("| --- | --- |\n"); @@ -1898,6 +2008,7 @@ mod tests { fn controller_set_width_partial_wrapped_emit_keeps_wrapped_remainder() { let mut ctrl = stream_controller(Some(18)); ctrl.push("alpha beta gamma delta epsilon zeta eta theta iota kappa lambda mu\n"); + ctrl.push("tail line\n"); let (first_emit, idle) = ctrl.on_commit_tick(); assert!(first_emit.is_some(), "expected first wrapped line emission");