diff --git a/codex-rs/tui/src/chatwidget/streaming.rs b/codex-rs/tui/src/chatwidget/streaming.rs index d1d17f8cd5..02491ee761 100644 --- a/codex-rs/tui/src/chatwidget/streaming.rs +++ b/codex-rs/tui/src/chatwidget/streaming.rs @@ -86,7 +86,7 @@ impl ChatWidget { fn flush_answer_stream(&mut self, completed_message: Option<&str>) { let had_stream_controller = self.stream_controller.is_some(); if let Some(mut controller) = self.stream_controller.take() { - let had_live_tail = controller.has_live_tail(); + let needs_scrollback_reflow = controller.needs_scrollback_reflow(); self.clear_active_stream_tail(); let (cell, streamed_source) = controller.finalize(); let completed_message_differs = completed_message.is_some_and(|completed| { @@ -96,7 +96,7 @@ impl ChatWidget { // Stream finalization supplies one trailing newline when the last delta omitted it. streamed != completed && streamed.strip_suffix('\n') != Some(completed) }); - let scrollback_reflow = if had_live_tail || completed_message_differs { + let scrollback_reflow = if needs_scrollback_reflow || completed_message_differs { crate::app_event::ConsolidationScrollbackReflow::Required } else { crate::app_event::ConsolidationScrollbackReflow::IfResizeReflowRan 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 b0330e69df..3856860036 100644 --- a/codex-rs/tui/src/chatwidget/tests/status_and_layout.rs +++ b/codex-rs/tui/src/chatwidget/tests/status_and_layout.rs @@ -450,60 +450,74 @@ async fn flush_answer_stream_keeps_default_reflow_for_plain_text_tail() { } #[tokio::test] -async fn flush_answer_stream_requests_scrollback_reflow_for_live_table_tail() { - let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await; - let cwd = chat.config.cwd.to_path_buf(); +async fn flush_answer_stream_requests_scrollback_reflow_for_tables() { + for definition in [ + None, + Some("[ref]: https://example.com\n"), + Some("[ref]: https://example.com"), + ] { + 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(80), - cwd.as_path(), - HistoryRenderMode::Rich, - ); - controller.push("| Name | Notes |\n"); - controller.push("| --- | --- |\n"); - controller.push("| alpha | tail held until final table render |\n"); - assert!( - controller.has_live_tail(), - "expected table 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::Required - ); - assert!( - deferred_history_cell.is_some(), - "live table tail should be staged for consolidation", - ); - } - _ => {} + let mut controller = crate::streaming::controller::StreamController::new( + Some(80), + cwd.as_path(), + HistoryRenderMode::Rich, + ); + controller.push("| Name | Notes |\n"); + controller.push("| --- | --- |\n"); + controller.push("| [A][ref] | tail held until final table render |\n"); + assert!( + controller.has_live_tail(), + "expected table holdback to leave a live tail for this regression", + ); + if let Some(definition) = definition { + controller.on_commit_tick_batch(usize::MAX); + controller.push("\nAfter table.\n\n"); + controller.on_commit_tick_batch(usize::MAX); + controller.push(definition); + controller.on_commit_tick_batch(usize::MAX); + assert!(!controller.has_live_tail()); } - } + chat.stream_controller = Some(controller); - assert!( - saw_consolidate, - "expected stream finalization to consolidate" - ); - assert!( - !saw_insert_history, - "live table tail should not be inserted before canonical reflow" - ); + 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::Required + ); + assert!( + deferred_history_cell.is_some() == definition.is_none(), + "only the uncommitted tail should be staged for consolidation", + ); + } + _ => {} + } + } + + assert!( + saw_consolidate, + "expected stream finalization to consolidate" + ); + assert!( + !saw_insert_history, + "table rows should not be inserted before canonical reflow" + ); + } } #[tokio::test] diff --git a/codex-rs/tui/src/streaming/controller.rs b/codex-rs/tui/src/streaming/controller.rs index 02fd4e6a04..50d9326910 100644 --- a/codex-rs/tui/src/streaming/controller.rs +++ b/codex-rs/tui/src/streaming/controller.rs @@ -15,9 +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. Holdback is enabled for -//! agent and proposed-plan streams. Lines in `Outside` and `Markdown` fence -//! contexts are scanned; lines inside non-markdown fences are skipped. +//! onward as mutable tail until its containing top-level block is stable. +//! Holdback is enabled for agent and proposed-plan streams. Lines in `Outside` +//! and `Markdown` fence contexts are scanned; lines inside non-markdown fences +//! are skipped. //! //! Transformable fences stay mutable while their containing top-level block is last. Markdown //! fences may gain literal delimiters when table rendering is disabled. For Mermaid, the closing fence replaces @@ -161,7 +162,6 @@ impl StreamCore { { let source = self.state.collector.committed_source(); let committed_source = &source[range]; - self.holdback_scanner.push_source_chunk(committed_source); self.render.append( source, committed_source, @@ -170,6 +170,8 @@ impl StreamCore { self.render_mode, self.inline_visualization_context.as_ref(), ); + self.holdback_scanner + .scan(source, self.render.completed_source_len); enqueued = self.sync_stable_queue(); } let preview_changed = self.refresh_preview(); @@ -385,6 +387,8 @@ impl StreamCore { self.render_mode, self.inline_visualization_context.as_ref(), ); + self.holdback_scanner + .scan(source, self.render.completed_source_len); if let Some(start) = previous_tail_start.or(self.active_tail_source_start(self.render_mode)) { let prefix_len = |width, mode| { @@ -680,6 +684,28 @@ impl StreamController { self.core.has_tail() } + /// Completion must repair emitted rows when reference definitions can change earlier blocks. + pub(crate) fn needs_scrollback_reflow(&self) -> bool { + if self.has_live_tail() || self.core.render.has_reference_link_definition { + return true; + } + if self.core.render_mode == HistoryRenderMode::Raw + || self.core.state.collector.pending_source().is_empty() + { + return false; + } + let source = format!( + "{}{}", + self.core.state.collector.committed_source(), + self.core.state.collector.pending_source(), + ); + pulldown_cmark::Parser::new(&source) + .reference_definitions() + .iter() + .next() + .is_some() + } + pub(crate) fn clear_queue(&mut self) { self.core.state.clear_queue(); self.core.enqueued_stable_len = self.core.emitted_stable_len; @@ -1441,6 +1467,65 @@ mod tests { "expected pre-table line to commit independently: {committed:?}", ); assert!(idle, "only pre-table content should have been queued"); + + ctrl.push( + "| first | row |\n\nAfter table.\n\n| Next | Table |\n| --- | --- |\n| second | row |\n", + ); + let (cell, idle) = ctrl.on_commit_tick_batch(usize::MAX); + let committed = cell.expect("completed table and prose should enter scrollback"); + assert!(idle); + assert!(ctrl.has_live_tail(), "the later table must stay mutable"); + insta::assert_debug_snapshot!( + "completed_table_scrollback", + lines_to_plain_strings(&committed.transcript_lines(u16::MAX)), + ); + + let context = InlineVisualizationContext::new(&test_cwd(), codex_protocol::ThreadId::new()) + .expect("visualization context"); + for (context, definition) in [ + (None, "[ref]: https://example.com\n"), + (Some(context), "[ref]: https://example.com\n"), + (None, "[ref]: https://example.com"), + ] { + let mut ctrl = StreamController::new_with_inline_visualizations( + Some(80), + &test_cwd(), + HistoryRenderMode::Rich, + context, + ); + let mut committed = Vec::new(); + for delta in [ + "| Key | Value |\n| --- | --- |\n| [A][ref] | row |\n", + "\nAfter table.\n\n", + definition, + ] { + ctrl.push(delta); + if let (Some(cell), _) = ctrl.on_commit_tick_batch(usize::MAX) { + committed.extend(lines_to_plain_strings(&cell.transcript_lines(u16::MAX))); + } + } + assert!(committed.iter().any(|line| line.contains("[A][ref]"))); + assert!( + !ctrl.has_live_tail(), + "the completed table should be released" + ); + assert!( + ctrl.needs_scrollback_reflow(), + "late definitions must repair scrollback" + ); + let (_, source) = ctrl.finalize(); + let rendered = render_source( + source.as_deref().unwrap(), + Some(80), + &test_cwd(), + HistoryRenderMode::Rich, + /*inline_visualization_context*/ None, + ); + insta::assert_debug_snapshot!( + "completed_table_reference", + lines_to_plain_strings(&visible_lines(rendered)) + ); + } } #[test] @@ -1778,6 +1863,10 @@ mod tests { "live view diverged after delta: {delta:?}" ); } + assert!( + !ctrl.has_live_tail(), + "completed tables and prose should enter scrollback before finalization", + ); } #[test] @@ -1984,7 +2073,7 @@ mod tests { let mut source = String::new(); for chunk in chunks { source.push_str(chunk); - scanner.push_source_chunk(chunk); + scanner.scan(&source, /*completed_source_len*/ 0); assert_eq!( scanner.state(), table_holdback_state(&source), @@ -1996,12 +2085,15 @@ mod tests { #[test] fn incremental_holdback_detects_header_delimiter_across_chunk_boundary() { let mut scanner = TableHoldbackScanner::new(); - scanner.push_source_chunk("| A | B |\n"); + scanner.scan("| A | B |\n", /*completed_source_len*/ 0); assert_eq!( scanner.state(), TableHoldbackState::PendingHeader { header_start: 0 } ); - scanner.push_source_chunk("| --- | --- |\n"); + scanner.scan( + "| A | B |\n| --- | --- |\n", + /*completed_source_len*/ 0, + ); assert_eq!( scanner.state(), TableHoldbackState::Confirmed { table_start: 0 } diff --git a/codex-rs/tui/src/streaming/render.rs b/codex-rs/tui/src/streaming/render.rs index 1322895dce..53ee13ce3b 100644 --- a/codex-rs/tui/src/streaming/render.rs +++ b/codex-rs/tui/src/streaming/render.rs @@ -26,12 +26,14 @@ pub(super) struct StreamingRender { pub(super) lines: Vec, pub(super) pending_math_start: Option, pub(super) mutable_fence_start: Option, + /// Source boundary before the final top-level block, including after full re-renders. + pub(super) completed_source_len: usize, /// Source prefix containing only completed top-level markdown blocks. stable_source_len: usize, /// Rendered-line boundary corresponding to `stable_source_len`. stable_rendered_len: usize, /// Reference-style link definitions can affect any earlier or later markdown block. - has_reference_link_definition: bool, + pub(super) has_reference_link_definition: bool, /// Inline visualization directives require source-wide rewriting once one is committed. has_inline_visualization_directive: bool, /// Parser state for a directly appendable, open top-level code fence. @@ -45,6 +47,7 @@ impl StreamingRender { lines: Vec::with_capacity(64), pending_math_start: None, mutable_fence_start: None, + completed_source_len: 0, stable_source_len: 0, stable_rendered_len: 0, has_reference_link_definition: false, @@ -57,6 +60,7 @@ impl StreamingRender { self.lines.clear(); self.pending_math_start = None; self.mutable_fence_start = None; + self.completed_source_len = 0; self.stable_source_len = 0; self.stable_rendered_len = 0; self.has_reference_link_definition = false; @@ -79,6 +83,7 @@ impl StreamingRender { self.open_code_fence = None; self.pending_math_start = None; self.mutable_fence_start = None; + self.completed_source_len = 0; self.has_inline_visualization_directive = contains_inline_visualization(source); self.lines = match (render_mode, inline_visualization_context) { (HistoryRenderMode::Rich, None) if !self.has_inline_visualization_directive => { @@ -89,6 +94,7 @@ impl StreamingRender { self.list_spacing, ); self.has_reference_link_definition = rendered.has_reference_link_definition; + self.completed_source_len = rendered.last_top_level_block_start.unwrap_or(0); self.pending_math_start = rendered.pending_math_start; self.mutable_fence_start = rendered.mutable_fence_start; rendered.lines @@ -102,6 +108,8 @@ impl StreamingRender { Some(cwd), self.list_spacing, ); + self.has_reference_link_definition = rendered.has_reference_link_definition; + self.completed_source_len = rendered.last_top_level_block_start.unwrap_or(0); self.pending_math_start = rendered.pending_math_start; self.mutable_fence_start = rendered.mutable_fence_start; } @@ -200,6 +208,7 @@ impl StreamingRender { } let final_block_start = pending.last_top_level_block_start.unwrap_or(/*default*/ 0); + self.completed_source_len = self.stable_source_len + final_block_start; self.open_code_fence = OpenCodeFence::detect( &pending_source[final_block_start..], raw_source.len(), diff --git a/codex-rs/tui/src/streaming/snapshots/codex_tui__streaming__controller__tests__completed_table_reference.snap b/codex-rs/tui/src/streaming/snapshots/codex_tui__streaming__controller__tests__completed_table_reference.snap new file mode 100644 index 0000000000..b4c81cf270 --- /dev/null +++ b/codex-rs/tui/src/streaming/snapshots/codex_tui__streaming__controller__tests__completed_table_reference.snap @@ -0,0 +1,11 @@ +--- +source: tui/src/streaming/controller.rs +expression: lines_to_plain_strings(&visible_lines(rendered)) +--- +[ + " Key Value", + "━━━━━━━━━━━━━━━━━━━━━━━━━ ━━━━━━━", + " A (https://example.com) row", + "", + "After table.", +] diff --git a/codex-rs/tui/src/streaming/snapshots/codex_tui__streaming__controller__tests__completed_table_scrollback.snap b/codex-rs/tui/src/streaming/snapshots/codex_tui__streaming__controller__tests__completed_table_scrollback.snap new file mode 100644 index 0000000000..749fd2e220 --- /dev/null +++ b/codex-rs/tui/src/streaming/snapshots/codex_tui__streaming__controller__tests__completed_table_scrollback.snap @@ -0,0 +1,13 @@ +--- +source: tui/src/streaming/controller.rs +assertion_line: 1456 +expression: "lines_to_plain_strings(&committed.transcript_lines(u16::MAX))" +--- +[ + " ", + " Key Value", + " ━━━━━━━ ━━━━━━━", + " first row", + " ", + " After table.", +] diff --git a/codex-rs/tui/src/streaming/table_holdback.rs b/codex-rs/tui/src/streaming/table_holdback.rs index 7a7f92eb69..482e7fa28d 100644 --- a/codex-rs/tui/src/streaming/table_holdback.rs +++ b/codex-rs/tui/src/streaming/table_holdback.rs @@ -45,10 +45,9 @@ struct PreviousLineState { /// Incremental scanner for table holdback state on append-only source streams. /// -/// `push_source_chunk` must receive source in the same order it will be -/// appended to the stream. The scanner stores byte offsets into that logical -/// source buffer, so feeding chunks out of order would make later tail -/// boundaries point at the wrong rendered region. +/// `scan` receives the append-only source and the renderer's completed-block +/// boundary. Completed tables are released only when their containing top-level +/// block is stable; the remaining tail is rescanned to find later tables. pub(super) struct TableHoldbackScanner { source_offset: usize, fence_tracker: FenceTracker, @@ -81,7 +80,7 @@ impl TableHoldbackScanner { /// header but the delimiter row has not arrived yet, so callers should /// optimistically keep that region mutable. `Confirmed` means the header /// and delimiter pair has been seen and all subsequent body rows remain in - /// the live tail until finalization. + /// the live tail until their containing top-level block is stable. pub(super) fn state(&self) -> TableHoldbackState { if let Some(table_start) = self.confirmed_table_start { TableHoldbackState::Confirmed { table_start } @@ -99,14 +98,24 @@ impl TableHoldbackScanner { && self.fence_tracker.kind() == FenceKind::Outside } - /// Advance the scanner with newly committed source. + /// Scan newly committed source, releasing holdback for completed blocks. /// - /// Chunks are expected to contain only source that is now safe to commit - /// into `raw_source`, typically newline-terminated lines from the - /// streaming collector. Partial rows are intentionally excluded so the - /// scanner never treats an unfinished table row as a stable structural - /// signal. - pub(super) fn push_source_chunk(&mut self, source_chunk: &str) { + /// Source must contain only complete lines. Releasing a table rescans from + /// the final top-level block, where fence state can safely start outside. + /// This also detects later tables that arrived in the same delta. + pub(super) fn scan(&mut self, source: &str, completed_source_len: usize) { + let holdback_start = match self.state() { + TableHoldbackState::Confirmed { table_start } => Some(table_start), + TableHoldbackState::PendingHeader { header_start } => Some(header_start), + TableHoldbackState::None => None, + }; + if self.source_offset < completed_source_len + || holdback_start.is_some_and(|start| start < completed_source_len) + { + self.reset(); + self.source_offset = completed_source_len; + } + let source_chunk = &source[self.source_offset..]; if source_chunk.is_empty() { return; }