openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-18 17:18:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

Release stable Markdown tables into scrollback during streaming (#50207)

Browse Source 
## Why

Table holdback kept completed tables and subsequent content mutable until the response finished. Releasing completed tables earlier also requires a final scrollback reflow when late reference-link definitions change previously emitted content.

## What changed

- Release table holdback once the containing top-level Markdown block is stable, rescanning the remaining tail for later tables.
- Track completed-block boundaries across incremental and full renders.
- Request scrollback reflow at answer completion when reference-link definitions are present, including definitions pending without a trailing newline.

## Testing

Add regression assertions and snapshots for completed tables entering scrollback while later tables remain mutable, and for late reference-link definitions resolving correctly at completion. Cover reflow requests with and without a trailing newline and reference handling with inline visualizations enabled.

GitOrigin-RevId: a6b2aac74496281495b53840bbb1daf06bef60d2
This commit is contained in:
Eric Traut
2026-10-02 06:40:07 +00:00
committed by copyberry
parent a75987455a
commit e3c2a83937
7 changed files with 221 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/tui/src/chatwidget/streaming.rs
+2 -2
View File
@@ -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
codex-rs/tui/src/chatwidget/tests/status_and_layout.rs
+65 -51
View File
@@ -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]
codex-rs/tui/src/streaming/controller.rs
+99 -7
View File
@@ -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 }
codex-rs/tui/src/streaming/render.rs
+10 -1
View File
@@ -26,12 +26,14 @@ pub(super) struct StreamingRender {
pub(super) lines: Vec<HyperlinkLine>,
pub(super) pending_math_start: Option<usize>,
pub(super) mutable_fence_start: Option<usize>,
/// 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(),
codex-rs/tui/src/streaming/snapshots/codex_tui__streaming__controller__tests__completed_table_reference.snap
+11
View File
@@ -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.",
]
codex-rs/tui/src/streaming/snapshots/codex_tui__streaming__controller__tests__completed_table_scrollback.snap
+13
View File
@@ -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.",
]
codex-rs/tui/src/streaming/table_holdback.rs
+21 -12
View File
@@ -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;
}