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

feat(tui): integrate table rendering into streaming pipeline with reflow

Browse Source 
Wire the Unicode box-drawing table renderer into the live streaming
path with holdback-aware commit logic, post-stream cell consolidation,
and debounced resize reflow.

Key additions:
- StreamCore: shared bookkeeping deduplicating Stream/PlanStreamController
- Table holdback: fence-aware state machine keeps buffer mutable during table detection
- AgentMarkdownCell: stores raw markdown source, re-renders at any width on demand
- ConsolidateAgentMessage: backward-walk replaces streamed cells with single consolidated cell
- Resize reflow: debounced (75ms) re-render of all transcript cells after terminal resize
This commit is contained in:
Felipe Coury
2026-04-06 13:12:29 -03:00
committed by Felipe Coury
parent ff310bb9e0
commit 004e85e09e
10 changed files with 2048 additions and 165 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/tui/src/app.rs
+168 -15
View File
@@ -275,6 +275,7 @@ fn guardian_approvals_mode() -> GuardianApprovalsMode {
/// Smooth-mode streaming drains one line per tick, so this interval controls
/// perceived typing speed for non-backlogged output.
const COMMIT_ANIMATION_TICK: Duration = tui::TARGET_FRAME_INTERVAL;
const RESIZE_REFLOW_DEBOUNCE: Duration = Duration::from_millis(75);
#[derive(Debug, Clone)]
pub struct AppExitInfo {
@@ -956,6 +957,17 @@ pub(crate) struct App {
pub(crate) overlay: Option<Overlay>,
pub(crate) deferred_history_lines: Vec<Line<'static>>,
has_emitted_history_lines: bool,
last_transcript_render_width: Option<u16>,
resize_reflow_pending_until: Option<Instant>,
/// Set to `true` when a resize reflow runs during an active agent stream (`SetWidth` handler),
/// so that `ConsolidateAgentMessage` can schedule a re-reflow to pick up the
/// `AgentMarkdownCell` rendering.
///
/// Cleared in three places:
/// 1. `ConsolidateAgentMessage` - normal completion (cells consolidated).
/// 2. `ConsolidateAgentMessage` - no-cells-to-consolidate `else` branch.
/// 3. `clear_thread_state()` - thread switch / abort.
reflow_ran_during_stream: bool,
pub(crate) enhanced_keys_supported: bool,
@@ -3241,6 +3253,9 @@ impl App {
self.transcript_cells.clear();
self.deferred_history_lines.clear();
self.has_emitted_history_lines = false;
self.last_transcript_render_width = None;
self.resize_reflow_pending_until = None;
self.reflow_ran_during_stream = false;
self.backtrack = BacktrackState::default();
self.backtrack_render_pending = false;
tui.terminal.clear_scrollback()?;
@@ -3248,6 +3263,78 @@ impl App {
Ok(())
}
fn display_lines_for_history_insert(
&mut self,
cell: &dyn HistoryCell,
width: u16,
) -> Vec<Line<'static>> {
let mut display = cell.display_lines(width);
if !display.is_empty() && !cell.is_stream_continuation() {
if self.has_emitted_history_lines {
display.insert(0, Line::from(""));
} else {
self.has_emitted_history_lines = true;
}
}
display
}
fn insert_history_cell_lines(
&mut self,
tui: &mut tui::Tui,
cell: &dyn HistoryCell,
width: u16,
) {
let display = self.display_lines_for_history_insert(cell, width);
if display.is_empty() {
return;
}
if self.overlay.is_some() {
self.deferred_history_lines.extend(display);
} else {
tui.insert_history_lines(display);
}
}
fn schedule_resize_reflow(&mut self) {
self.resize_reflow_pending_until = Some(Instant::now() + RESIZE_REFLOW_DEBOUNCE);
}
fn maybe_run_resize_reflow(&mut self, tui: &mut tui::Tui) -> Result<()> {
let Some(deadline) = self.resize_reflow_pending_until else {
return Ok(());
};
if Instant::now() < deadline || self.overlay.is_some() {
return Ok(());
}
self.resize_reflow_pending_until = None;
if self.transcript_cells.is_empty() {
self.has_emitted_history_lines = false;
self.deferred_history_lines.clear();
return Ok(());
}
tui.terminal.clear_scrollback()?;
tui.terminal.clear()?;
self.has_emitted_history_lines = false;
self.deferred_history_lines.clear();
// Track that a reflow happened durin an active agent stream so ConsolidateAgentMessage can
// schedule a follow-up reflow with the correct AgentMarkdownCell rendering.
if self.chat_widget.has_active_agent_stream() {
self.reflow_ran_during_stream = true;
}
let width = tui.terminal.size()?.width;
for cell in self.transcript_cells.clone() {
self.insert_history_cell_lines(tui, cell.as_ref(), width);
}
Ok(())
}
fn reset_thread_event_state(&mut self) {
self.abort_all_thread_event_listeners();
self.thread_event_channels.clear();
@@ -3769,6 +3856,9 @@ impl App {
overlay: None,
deferred_history_lines: Vec::new(),
has_emitted_history_lines: false,
last_transcript_render_width: None,
resize_reflow_pending_until: None,
reflow_ran_during_stream: false,
commit_anim_running: Arc::new(AtomicBool::new(false)),
status_line_invalid_items_warned: status_line_invalid_items_warned.clone(),
terminal_title_invalid_items_warned: terminal_title_invalid_items_warned.clone(),
@@ -3950,6 +4040,15 @@ impl App {
) -> Result<AppRunControl> {
if matches!(event, TuiEvent::Draw) {
let size = tui.terminal.size()?;
let previous_width = self.last_transcript_render_width.replace(size.width);
if previous_width.is_some_and(|width| width != size.width) {
self.chat_widget.on_terminal_resize(size.width);
self.schedule_resize_reflow();
tui.frame_requester()
.schedule_frame_in(RESIZE_REFLOW_DEBOUNCE);
} else if previous_width.is_none() {
self.chat_widget.on_terminal_resize(size.width);
}
if size != tui.terminal.last_known_screen_size {
self.refresh_status_line();
}
@@ -4212,23 +4311,70 @@ impl App {
tui.frame_requester().schedule_frame();
}
self.transcript_cells.push(cell.clone());
let mut display = cell.display_lines(tui.terminal.last_known_screen_size.width);
if !display.is_empty() {
// Only insert a separating blank line for new cells that are not
// part of an ongoing stream. Streaming continuations should not
// accrue extra blank lines between chunks.
if !cell.is_stream_continuation() {
if self.has_emitted_history_lines {
display.insert(0, Line::from(""));
} else {
self.has_emitted_history_lines = true;
}
self.insert_history_cell_lines(
tui,
cell.as_ref(),
tui.terminal.last_known_screen_size.width,
);
}
AppEvent::ConsolidateAgentMessage { source, cwd } => {
// Walk backward to find the contiguous run of streaming AgentMarkdownCells that
// belong to the just-finalized stream
let end = self.transcript_cells.len();
tracing::debug!(
"ConsolidateAgentMessage: transcript_cells.len()={end}, source_len={}",
source.len()
);
let mut start = end;
// Include continuation cells that are AgentMarkdownCells.
while start > 0
&& self.transcript_cells[start - 1].is_stream_continuation()
&& self.transcript_cells[start - 1]
.as_any()
.is::<history_cell::AgentMessageCell>()
{
start -= 1;
}
// Include the head cell (is_stream_continuation == false) that started this agent
// message run
if start > 0
&& self.transcript_cells[start - 1]
.as_any()
.is::<history_cell::AgentMessageCell>()
&& !self.transcript_cells[start - 1].is_stream_continuation()
{
start -= 1;
}
if start < end {
tracing::debug!(
"ConsolidateAgentMessage: replacing cells [{start}..{end}] with AgentMarkdownCell"
);
let consolidated: Arc<dyn HistoryCell> =
Arc::new(history_cell::AgentMarkdownCell::new(source, &cwd));
self.transcript_cells
.splice(start..end, std::iter::once(consolidated.clone()));
// Keep the transcript overlay in sync.
if let Some(Overlay::Transcript(t)) = &mut self.overlay {
t.consolidate_cells(start..end, consolidated.clone());
tui.frame_requester().schedule_frame();
}
if self.overlay.is_some() {
self.deferred_history_lines.extend(display);
} else {
tui.insert_history_lines(display);
// If a resize reflow ran while these cells were still unconsolidated
// AgentMessageCells, the scrollback has stale table rendering. Schedule a
// re-reflow.
if self.reflow_ran_during_stream {
self.schedule_resize_reflow();
tui.frame_requester()
.schedule_frame_in(RESIZE_REFLOW_DEBOUNCE);
}
self.reflow_ran_during_stream = false;
} else {
tracing::debug!(
"ConsolidateAgentMessage: no cells to consolidate(start={start}, end={end})",
);
self.reflow_ran_during_stream = false;
}
}
AppEvent::ApplyThreadRollback { num_turns } => {
@@ -4257,6 +4403,7 @@ impl App {
}
AppEvent::CommitTick => {
self.chat_widget.on_commit_tick();
self.maybe_run_resize_reflow(tui)?;
}
AppEvent::Exit(mode) => {
return Ok(self.handle_exit_mode(app_server, mode).await);
@@ -9077,6 +9224,9 @@ guardian_approval = true
overlay: None,
deferred_history_lines: Vec::new(),
has_emitted_history_lines: false,
last_transcript_render_width: None,
resize_reflow_pending_until: None,
reflow_ran_during_stream: false,
enhanced_keys_supported: false,
commit_anim_running: Arc::new(AtomicBool::new(false)),
status_line_invalid_items_warned: Arc::new(AtomicBool::new(false)),
@@ -9131,6 +9281,9 @@ guardian_approval = true
overlay: None,
deferred_history_lines: Vec::new(),
has_emitted_history_lines: false,
last_transcript_render_width: None,
resize_reflow_pending_until: None,
reflow_ran_during_stream: false,
enhanced_keys_supported: false,
commit_anim_running: Arc::new(AtomicBool::new(false)),
status_line_invalid_items_warned: Arc::new(AtomicBool::new(false)),
codex-rs/tui/src/app_event.rs
+4
View File
@@ -266,6 +266,10 @@ pub(crate) enum AppEvent {
InsertHistoryCell(Box<dyn HistoryCell>),
/// Replace the contiguous run of streaming `AgentMessageCell`s at the end of the transcript
/// with a single `AgentMarkdownCell` that re-renders from the raw markdown source on resize.
ConsolidateAgentMessage { source: String, cwd: PathBuf },
/// Apply rollback semantics to local transcript cells.
///
/// This is emitted when rollback was not initiated by the current
codex-rs/tui/src/chatwidget.rs
+104 -12
View File
@@ -226,6 +226,7 @@ use crossterm::event::KeyCode;
use crossterm::event::KeyEvent;
use crossterm::event::KeyEventKind;
use crossterm::event::KeyModifiers;
use crossterm::terminal;
use rand::Rng;
use ratatui::buffer::Buffer;
use ratatui::layout::Rect;
@@ -1693,18 +1694,32 @@ impl ChatWidget {
}
fn flush_answer_stream_with_separator(&mut self) {
if let Some(mut controller) = self.stream_controller.take()
&& let Some(cell) = controller.finalize()
{
self.add_boxed_history(cell);
let had_stream_controller = self.stream_controller.is_some();
if let Some(mut controller) = self.stream_controller.take() {
self.clear_active_stream_tail();
let (cell, source) = controller.finalize();
if let Some(cell) = cell {
self.add_boxed_history(cell);
}
// Consolidate the run of streaming AgentMessageCells into a single AgentMarkdownCell
// that can re-render from source on resize.
if let Some(source) = source {
self.app_event_tx.send(AppEvent::ConsolidateAgentMessage {
source,
cwd: self.config.cwd.to_path_buf(),
});
}
}
self.adaptive_chunking.reset();
if had_stream_controller {
self.app_event_tx.send(AppEvent::StopCommitAnimation);
}
}
fn stream_controllers_idle(&self) -> bool {
self.stream_controller
.as_ref()
.map(|controller| controller.queued_lines() == 0)
.map(|controller| controller.queued_lines() == 0 && !controller.has_live_tail())
.unwrap_or(true)
&& self
.plan_stream_controller
@@ -2177,7 +2192,7 @@ impl ChatWidget {
if self.plan_stream_controller.is_none() {
self.plan_stream_controller = Some(PlanStreamController::new(
self.last_rendered_width.get().map(|w| w.saturating_sub(4)),
self.current_stream_width(4),
&self.config.cwd,
));
}
@@ -4087,6 +4102,7 @@ impl ChatWidget {
self.bottom_pane.hide_status_indicator();
self.add_boxed_history(cell);
}
self.sync_active_stream_tail();
if outcome.has_controller && outcome.all_idle {
self.maybe_restore_status_indicator_after_stream_idle();
@@ -4131,11 +4147,11 @@ impl ChatWidget {
#[inline]
fn handle_streaming_delta(&mut self, delta: String) {
// Before streaming agent content, flush any active exec cell group.
self.flush_unified_exec_wait_streak();
self.flush_active_cell();
if self.stream_controller.is_none() {
// Before streaming agent content, flush any active exec cell group.
self.flush_unified_exec_wait_streak();
self.flush_active_cell();
// If the previous turn inserted non-stream history (exec output, patch status, MCP
// calls), render a separator before starting the next streamed assistant message.
if self.needs_final_message_separator && self.had_work_activity {
@@ -4155,7 +4171,7 @@ impl ChatWidget {
self.needs_final_message_separator = false;
}
self.stream_controller = Some(StreamController::new(
self.last_rendered_width.get().map(|w| w.saturating_sub(2)),
self.current_stream_width(2),
&self.config.cwd,
));
}
@@ -4165,9 +4181,30 @@ impl ChatWidget {
self.app_event_tx.send(AppEvent::StartCommitAnimation);
self.run_catch_up_commit_tick();
}
self.sync_active_stream_tail();
self.request_redraw();
}
fn current_stream_width(&self, reserved_cols: usize) -> Option<usize> {
self.last_rendered_width
.get()
.or_else(|| terminal::size().ok().map(|(w, _)| usize::from(w)))
.map(|w| w.saturating_sub(reserved_cols))
}
pub(crate) fn on_terminal_resize(&mut self, width: u16) {
self.last_rendered_width.set(Some(width as usize));
let stream_width = self.current_stream_width(2);
let plan_stream_width = self.current_stream_width(4);
if let Some(controller) = self.stream_controller.as_mut() {
controller.set_width(stream_width);
}
if let Some(controller) = self.plan_stream_controller.as_mut() {
controller.set_width(plan_stream_width);
}
self.sync_active_stream_tail();
}
fn worked_elapsed_from(&mut self, current_elapsed: u64) -> u64 {
let baseline = match self.last_separator_elapsed_secs {
Some(last) if current_elapsed < last => 0,
@@ -5530,6 +5567,15 @@ impl ChatWidget {
}
fn flush_active_cell(&mut self) {
if self.stream_controller.is_some()
&& self
.active_cell
.as_ref()
.is_some_and(|cell| cell.as_any().is::<history_cell::StreamingAgentTailCell>())
{
return;
}
if let Some(active) = self.active_cell.take() {
self.needs_final_message_separator = true;
self.app_event_tx.send(AppEvent::InsertHistoryCell(active));
@@ -5551,12 +5597,53 @@ impl ChatWidget {
if !keep_placeholder_header_active && !cell.display_lines(u16::MAX).is_empty() {
// Only break exec grouping if the cell renders visible lines.
self.flush_active_cell();
let keep_stream_tail_active = self.stream_controller.is_some()
&& self.active_cell.as_ref().is_some_and(|active| {
active.as_any().is::<history_cell::StreamingAgentTailCell>()
});
if !keep_stream_tail_active {
self.flush_active_cell();
}
self.needs_final_message_separator = true;
}
self.app_event_tx.send(AppEvent::InsertHistoryCell(cell));
}
fn sync_active_stream_tail(&mut self) {
let Some((tail_lines, tail_starts_stream)) =
self.stream_controller.as_ref().map(|controller| {
(
controller.current_tail_lines(),
controller.tail_starts_stream(),
)
})
else {
return;
};
if tail_lines.is_empty() {
self.clear_active_stream_tail();
return;
}
self.bottom_pane.hide_status_indicator();
let tail_cell = history_cell::StreamingAgentTailCell::new(tail_lines, tail_starts_stream);
self.active_cell = Some(Box::new(tail_cell));
self.bump_active_cell_revision();
}
fn clear_active_stream_tail(&mut self) {
if self
.active_cell
.as_ref()
.is_some_and(|cell| cell.as_any().is::<history_cell::StreamingAgentTailCell>())
{
self.active_cell = None;
self.bump_active_cell_revision();
}
}
fn queue_user_message(&mut self, user_message: UserMessage) {
if !self.is_session_configured() || self.bottom_pane.is_task_running() {
self.queued_user_messages.push_back(user_message);
@@ -10299,6 +10386,11 @@ impl ChatWidget {
self.plan_stream_controller.is_some()
}
/// Whether an agent message stream is active (not a plan stream).
pub(crate) fn has_active_agent_stream(&self) -> bool {
self.stream_controller.is_some()
}
pub(crate) fn composer_is_empty(&self) -> bool {
self.bottom_pane.composer_is_empty()
}
codex-rs/tui/src/history_cell.rs
+245
View File
@@ -38,6 +38,7 @@ use crate::version::CODEX_CLI_VERSION;
use crate::wrapping::RtOptions;
use crate::wrapping::adaptive_wrap_line;
use crate::wrapping::adaptive_wrap_lines;
use crate::wrapping::word_wrap_lines;
use base64::Engine;
use codex_app_server_protocol::McpServerStatus;
use codex_config::types::McpServerTransportConfig;
@@ -480,6 +481,85 @@ impl HistoryCell for AgentMessageCell {
}
}
/// A consolidated agent message cell taht stores raw markdown source and re-renders from it at any
/// width. This replaces the run of `AgentMessageCell`s after a stream finalizes, so tables and
/// other width-sensitive content review correctly on terminal resize.
#[derive(Debug)]
pub(crate) struct AgentMarkdownCell {
markdown_source: String,
cwd: PathBuf,
}
impl AgentMarkdownCell {
pub(crate) fn new(markdown_source: String, cwd: &Path) -> Self {
Self {
markdown_source,
cwd: cwd.to_path_buf(),
}
}
}
impl HistoryCell for AgentMarkdownCell {
fn display_lines(&self, width: u16) -> Vec<Line<'static>> {
let mut lines: Vec<Line<'static>> = Vec::new();
// Re-render markdown from source at the current width. Reserve 2 columns for the "• " /
// " " prefix prepended below.
let wrap_width = (width as usize).saturating_sub(2);
crate::markdown::append_markdown_agent_with_cwd(
&self.markdown_source,
Some(wrap_width),
Some(self.cwd.as_path()),
&mut lines,
);
// Use prefix_lines (not word_wrap_lines) so table rows with box-drawing characters are not
// broken by word-wrapping. The markdown renderer already output to wrap_width.
prefix_lines(lines, "• ".dim(), " ".into())
}
}
/// Transient active-cell representation of the mutable tail of an agent stream.
///
/// During streaming, lines that have not yet been committed to scrollback (because they belong to
/// an in-progress table or are the last incomplete line) are displayed via this cell in the
/// `active_cell` slot. It is replaced on every delta and cleared when the stream finalizes.
///
/// Unlike `AgentMessageCell`, this cell is never committed to the transcript. It exists only as a
/// live preview of content that will eventually be emitted as stable `AgentMessageCell`s or
/// consolidated into an `AgentMarkdownCell`.
#[derive(Debug)]
pub(crate) struct StreamingAgentTailCell {
lines: Vec<Line<'static>>,
is_first_line: bool,
}
impl StreamingAgentTailCell {
pub(crate) fn new(lines: Vec<Line<'static>>, is_first_line: bool) -> Self {
Self {
lines,
is_first_line,
}
}
}
impl HistoryCell for StreamingAgentTailCell {
fn display_lines(&self, width: u16) -> Vec<Line<'static>> {
word_wrap_lines(
&self.lines,
RtOptions::new(width as usize)
.initial_indent(if self.is_first_line {
"• ".dim().into()
} else {
" ".into()
})
.subsequent_indent(" ".into()),
)
}
fn is_stream_continuation(&self) -> bool {
!self.is_first_line
}
}
#[derive(Debug)]
pub(crate) struct PlainHistoryCell {
lines: Vec<Line<'static>>,
@@ -4679,4 +4759,169 @@ mod tests {
]
);
}
#[test]
fn agent_markdown_cell_renders_table_at_different_widths() {
let source = "| Name | Role |\n|------|------|\n| Alice | Engineer |\n| Bob | Designer |\n";
let cell = AgentMarkdownCell::new(source.to_string(), &test_cwd());
// At width 80 the table should render with box-drawing characters.
let lines_80 = render_lines(&cell.display_lines(80));
assert!(
lines_80.iter().any(|l| l.contains('┌')),
"expected box-drawing table at width 80: {lines_80:?}"
);
// Verify the "• " leader is present on the first line.
assert!(
lines_80[0].starts_with("• "),
"first line should start with bullet prefix: {:?}",
lines_80[0]
);
// At width 40 the table should also render correctly (re-rendered from
// source, not just word-wrapped).
let lines_40 = render_lines(&cell.display_lines(40));
assert!(
lines_40.iter().any(|l| l.contains('┌')),
"expected box-drawing table at width 40: {lines_40:?}"
);
// Verify table borders are intact (not broken by naive word-wrapping).
// Every line with a box char should have matching left/right borders.
for line in &lines_40 {
let trimmed = line.trim();
if trimmed.starts_with('│') {
assert!(
trimmed.ends_with('│'),
"table row should have matching right border: {line:?}"
);
}
}
}
#[test]
fn agent_markdown_cell_survives_insert_history_rewrap() {
let source = "\
| Milestone | Target Date | Outcome | Extended Context |
|-----------|-------------|---------|------------------|
| Canary Rollout | 2026-01-15 | Completed | Canary remained at limited traffic longer than planned because p95 latency briefly regressed during
cold-cache periods |
| Regional Expansion | 2026-01-29 | Completed | Expansion succeeded with stable error rates, though internal analytics lagged temporarily |
";
let cell = AgentMarkdownCell::new(source.to_string(), &test_cwd());
let width: u16 = 80;
let lines = cell.display_lines(width);
// Simulate what insert_history_lines does: word_wrap_lines with
// the terminal width and no indent.
let rewrapped = word_wrap_lines(&lines, width as usize);
let before = render_lines(&lines);
let after = render_lines(&rewrapped);
assert_eq!(
before, after,
"word_wrap_lines should not alter lines that already fit within width"
);
}
#[test]
fn agent_markdown_cell_table_fits_within_narrow_width() {
let source = "\
| Milestone | Target Date | Outcome | Extended Context |
|-----------|-------------|---------|------------------|
| Canary Rollout | 2026-01-15 | Completed | Canary remained at limited traffic longer than planned because p95 latency briefly regressed during
cold-cache periods |
| Regional Expansion | 2026-01-29 | Completed | Expansion succeeded with stable error rates, though internal analytics lagged temporarily |
| Legacy Decommission | 2026-02-10 | In Progress | Most legacy jobs are drained, but final shutdown is blocked by one compliance export workflow |
";
let cell = AgentMarkdownCell::new(source.to_string(), &test_cwd());
// Render at a narrow width (simulating terminal resize).
let narrow_width: u16 = 80;
let lines = cell.display_lines(narrow_width);
let rendered = render_lines(&lines);
// Every rendered line must fit within the target width.
for line in &rendered {
let display_width = unicode_width::UnicodeWidthStr::width(line.as_str());
assert!(
display_width <= narrow_width as usize,
"line exceeds width {narrow_width}: ({display_width} chars) {line:?}"
);
}
// Table should still have box-drawing characters.
assert!(
rendered.iter().any(|l| l.contains('┌')),
"expected box-drawing table: {rendered:?}"
);
}
/// Simulate the consolidation backward-walk logic from `App::handle_event`
/// to verify it correctly identifies and replaces `AgentMessageCell` runs.
#[test]
fn consolidation_walker_replaces_agent_message_cells() {
use std::sync::Arc;
// Build a transcript with: [UserCell, AgentMsg(head), AgentMsg(cont), AgentMsg(cont)]
let user = Arc::new(UserHistoryCell {
message: "hello".to_string(),
text_elements: Vec::new(),
local_image_paths: Vec::new(),
}) as Arc<dyn HistoryCell>;
let head = Arc::new(AgentMessageCell::new(vec![Line::from("line 1")], true))
as Arc<dyn HistoryCell>;
let cont1 = Arc::new(AgentMessageCell::new(vec![Line::from("line 2")], false))
as Arc<dyn HistoryCell>;
let cont2 = Arc::new(AgentMessageCell::new(vec![Line::from("line 3")], false))
as Arc<dyn HistoryCell>;
let mut transcript_cells: Vec<Arc<dyn HistoryCell>> =
vec![user.clone(), head, cont1, cont2];
// Run the same consolidation logic as the handler.
let source = "line 1\nline 2\nline 3\n".to_string();
let end = transcript_cells.len();
let mut start = end;
while start > 0
&& transcript_cells[start - 1].is_stream_continuation()
&& transcript_cells[start - 1]
.as_any()
.is::<AgentMessageCell>()
{
start -= 1;
}
if start > 0
&& transcript_cells[start - 1]
.as_any()
.is::<AgentMessageCell>()
&& !transcript_cells[start - 1].is_stream_continuation()
{
start -= 1;
}
assert_eq!(
start, 1,
"should find all 3 agent cells starting at index 1"
);
assert_eq!(end, 4);
// Splice.
let consolidated: Arc<dyn HistoryCell> =
Arc::new(AgentMarkdownCell::new(source, &test_cwd()));
transcript_cells.splice(start..end, std::iter::once(consolidated));
assert_eq!(transcript_cells.len(), 2, "should be [user, consolidated]");
// Verify first cell is still the user cell.
assert!(
transcript_cells[0].as_any().is::<UserHistoryCell>(),
"first cell should be UserHistoryCell"
);
// Verify second cell is AgentMarkdownCell.
assert!(
transcript_cells[1].as_any().is::<AgentMarkdownCell>(),
"second cell should be AgentMarkdownCell"
);
}
}
codex-rs/tui/src/markdown.rs
+273
View File
@@ -1,6 +1,18 @@
//! Markdown-to-ratatui rendering entry points.
//!
//! This module provides the public API surface that the rest of the TUI uses to turn markdown
//! source into `Vec<Line<'static>>`. Three variants exist:
//!
//! - [`append_markdown`] -- general-purpose, used for plan blocks and history cells that already
//! hold pre-processed markdown.
//! - [`append_markdown_agent`] -- for agent responses. Runs [`unwrap_markdown_fences`] first so
//! that `` ```md ``/`` ```markdown `` fences containing tables are stripped and the table
//! parser sees raw markdown instead of fenced code.
use ratatui::text::Line;
use std::path::Path;
use crate::table_detect;
/// Render markdown into `lines` while resolving local file-link display relative to `cwd`.
///
/// Callers that already know the session working directory should pass it here so streamed and
@@ -19,6 +31,202 @@ pub(crate) fn append_markdown(
crate::render::line_utils::push_owned_lines(&rendered.lines, lines);
}
/// Render an agent message to styled ratatui lines.
///
/// Before rendering, the source is passed through [`unwrap_markdown_fences`] so that tables
/// wrapped in `` ```md `` fences are rendered as native tables rather than code blocks.
/// Non-markdown fences (e.g. `rust`, `sh`) are left
/// intact.
#[cfg(test)]
pub(crate) fn append_markdown_agent(
markdown_source: &str,
width: Option<usize>,
lines: &mut Vec<Line<'static>>,
) {
append_markdown_agent_with_cwd(markdown_source, width, /*cwd*/ None, lines);
}
/// Render an agent message while resolving local file links relative to `cwd`.
pub(crate) fn append_markdown_agent_with_cwd(
markdown_source: &str,
width: Option<usize>,
cwd: Option<&Path>,
lines: &mut Vec<Line<'static>>,
) {
let normalized = unwrap_markdown_fences(markdown_source);
append_markdown(&normalized, width, cwd, lines);
}
/// Strip `` ```md ``/`` ```markdown `` fences that contain tables, emitting their content as bare
/// markdown so `pulldown-cmark` parses the tables natively.
///
/// Fences whose info string is not `md` or `markdown` are passed through unchanged. Markdown
/// fences that do *not* contain a table (detected by checking for a header row + delimiter row)
/// are also passed through so that non-table markdown inside a fence still renders as a code
/// block.
///
/// The fence unwrapping is intentionally conservative: it buffers the entire fence body before
/// deciding, and an unclosed fence at end-of-input is re-emitted with its opening line so partial
/// streams degrade to code display.
fn unwrap_markdown_fences(markdown_source: &str) -> String {
#[derive(Clone, Copy)]
struct Fence {
marker: char,
len: usize,
is_markdown: bool,
}
fn parse_open_fence(line: &str) -> Option<Fence> {
let without_newline = line.strip_suffix('\n').unwrap_or(line);
let leading_ws = without_newline
.as_bytes()
.iter()
.take_while(|byte| **byte == b' ')
.count();
if leading_ws > 3 {
return None;
}
let trimmed = &without_newline[leading_ws..];
let marker = trimmed.chars().next()?;
if marker != '`' && marker != '~' {
return None;
}
let len = trimmed.chars().take_while(|ch| *ch == marker).count();
if len < 3 {
return None;
}
let rest = trimmed[len..].trim();
let info = rest.split_whitespace().next().unwrap_or_default();
let is_markdown = info.eq_ignore_ascii_case("md") || info.eq_ignore_ascii_case("markdown");
Some(Fence {
marker,
len,
is_markdown,
})
}
fn is_close_fence(line: &str, fence: Fence) -> bool {
let without_newline = line.strip_suffix('\n').unwrap_or(line);
let leading_ws = without_newline
.as_bytes()
.iter()
.take_while(|byte| **byte == b' ')
.count();
if leading_ws > 3 {
return false;
}
let trimmed = &without_newline[leading_ws..];
if !trimmed.starts_with(fence.marker) {
return false;
}
let len = trimmed.chars().take_while(|ch| *ch == fence.marker).count();
if len < fence.len {
return false;
}
trimmed[len..].trim().is_empty()
}
fn markdown_fence_contains_table(content: &str) -> bool {
let mut saw_candidate = false;
let mut saw_delimiter = false;
for line in content.lines() {
let trimmed = line.trim();
if trimmed.is_empty() {
continue;
}
if !table_detect::is_table_header_line(trimmed) {
continue;
}
if table_detect::is_table_delimiter_line(trimmed) {
saw_delimiter = true;
} else {
saw_candidate = true;
}
}
saw_candidate && saw_delimiter
}
enum ActiveFence {
Passthrough(Fence),
MarkdownCandidate {
fence: Fence,
opening_line: String,
content: String,
},
}
let mut out = String::with_capacity(markdown_source.len());
let mut active_fence: Option<ActiveFence> = None;
for line in markdown_source.split_inclusive('\n') {
if let Some(active) = active_fence.take() {
match active {
ActiveFence::Passthrough(fence) => {
out.push_str(line);
if !is_close_fence(line, fence) {
active_fence = Some(ActiveFence::Passthrough(fence));
}
}
ActiveFence::MarkdownCandidate {
fence,
opening_line,
mut content,
} => {
if is_close_fence(line, fence) {
if markdown_fence_contains_table(&content) {
out.push_str(&content);
} else {
out.push_str(&opening_line);
out.push_str(&content);
out.push_str(line);
}
} else {
content.push_str(line);
active_fence = Some(ActiveFence::MarkdownCandidate {
fence,
opening_line,
content,
});
}
}
}
continue;
}
if let Some(fence) = parse_open_fence(line) {
if fence.is_markdown {
active_fence = Some(ActiveFence::MarkdownCandidate {
fence,
opening_line: line.to_string(),
content: String::new(),
});
} else {
out.push_str(line);
active_fence = Some(ActiveFence::Passthrough(fence));
}
continue;
}
out.push_str(line);
}
if let Some(active) = active_fence {
match active {
ActiveFence::Passthrough(_) => {}
ActiveFence::MarkdownCandidate {
opening_line,
content,
..
} => {
out.push_str(&opening_line);
out.push_str(&content);
}
}
}
out
}
#[cfg(test)]
mod tests {
use super::*;
@@ -118,4 +326,69 @@ mod tests {
"did not expect a split into ['1.', 'Tight item']; got: {lines:?}"
);
}
#[test]
fn append_markdown_agent_unwraps_markdown_fences_for_table_rendering() {
let src = "```markdown\n| A | B |\n|---|---|\n| 1 | 2 |\n```\n";
let mut out = Vec::new();
append_markdown_agent(src, None, &mut out);
let rendered = lines_to_strings(&out);
assert!(rendered.iter().any(|line| line.contains("┌")));
assert!(rendered.iter().any(|line| line.contains("│ 1 │ 2 │")));
}
#[test]
fn append_markdown_agent_unwraps_markdown_fences_for_no_outer_table_rendering() {
let src = "```md\nCol A | Col B | Col C\n--- | --- | ---\nx | y | z\n10 | 20 | 30\n```\n";
let mut out = Vec::new();
append_markdown_agent(src, None, &mut out);
let rendered = lines_to_strings(&out);
assert!(rendered.iter().any(|line| line.contains("┌")));
assert!(
rendered
.iter()
.any(|line| line.contains("│ Col A │ Col B │ Col C │"))
);
assert!(
!rendered
.iter()
.any(|line| line.trim() == "Col A | Col B | Col C")
);
}
#[test]
fn append_markdown_agent_unwraps_markdown_fences_for_two_column_no_outer_table() {
let src = "```md\nA | B\n--- | ---\nleft | right\n```\n";
let mut out = Vec::new();
append_markdown_agent(src, None, &mut out);
let rendered = lines_to_strings(&out);
assert!(rendered.iter().any(|line| line.contains("┌")));
assert!(rendered.iter().any(|line| line.contains("│ A")));
assert!(!rendered.iter().any(|line| line.trim() == "A | B"));
}
#[test]
fn append_markdown_agent_keeps_non_markdown_fences_as_code() {
let src = "```rust\n| A | B |\n|---|---|\n| 1 | 2 |\n```\n";
let mut out = Vec::new();
append_markdown_agent(src, None, &mut out);
let rendered = lines_to_strings(&out);
assert_eq!(
rendered,
vec![
"| A | B |".to_string(),
"|---|---|".to_string(),
"| 1 | 2 |".to_string(),
]
);
}
#[test]
fn append_markdown_agent_keeps_markdown_fence_when_content_is_not_table() {
let src = "```markdown\n**bold**\n```\n";
let mut out = Vec::new();
append_markdown_agent(src, None, &mut out);
let rendered = lines_to_strings(&out);
assert_eq!(rendered, vec!["**bold**".to_string()]);
}
}
codex-rs/tui/src/markdown_stream.rs
+125 -13
View File
@@ -1,15 +1,36 @@
//! Newline-gated streaming accumulator for markdown source.
//!
//! `MarkdownStreamCollector` buffers incoming token deltas and exposes a commit boundary at each
//! newline. The stream controllers (`streaming/controller.rs`) call `commit_complete_source()`
//! after each newline-bearing delta to obtain the completed prefix for re-rendering, leaving the
//! trailing incomplete line in the buffer for the next delta.
//!
//! On finalization, `finalize_and_drain_source()` flushes whatever remains (the last line, which
//! may lack a trailing newline).
#[cfg(test)]
use ratatui::text::Line;
use std::path::Path;
#[cfg(test)]
use std::path::PathBuf;
#[cfg(test)]
use crate::markdown;
/// Newline-gated accumulator that renders markdown and commits only fully
/// completed logical lines.
/// Newline-gated accumulator that buffers raw markdown source and commits only completed lines
/// (terminated by `\n`).
///
/// The buffer tracks how many source bytes have already been committed via
/// `committed_source_len`, so each `commit_complete_source()` call returns only the newly
/// completed portion. This design lets the stream controller re-render the entire accumulated
/// source while only appending new content.
pub(crate) struct MarkdownStreamCollector {
buffer: String,
committed_source_len: usize,
#[cfg(test)]
committed_line_count: usize,
width: Option<usize>,
#[cfg(test)]
cwd: PathBuf,
}
@@ -21,17 +42,31 @@ impl MarkdownStreamCollector {
/// different working directories within one stream would make the same link render with
/// different path prefixes across incremental commits.
pub fn new(width: Option<usize>, cwd: &Path) -> Self {
#[cfg(not(test))]
let _ = cwd;
Self {
buffer: String::new(),
committed_source_len: 0,
#[cfg(test)]
committed_line_count: 0,
width,
#[cfg(test)]
cwd: cwd.to_path_buf(),
}
}
pub fn set_width(&mut self, width: Option<usize>) {
self.width = width;
}
pub fn clear(&mut self) {
self.buffer.clear();
self.committed_line_count = 0;
self.committed_source_len = 0;
#[cfg(test)]
{
self.committed_line_count = 0;
}
}
pub fn push_delta(&mut self, delta: &str) {
@@ -39,17 +74,48 @@ impl MarkdownStreamCollector {
self.buffer.push_str(delta);
}
/// Commit newly completed raw markdown source up to the last newline.
pub fn commit_complete_source(&mut self) -> Option<String> {
let commit_end = self.buffer.rfind('\n').map(|idx| idx + 1)?;
if commit_end <= self.committed_source_len {
return None;
}
let out = self.buffer[self.committed_source_len..commit_end].to_string();
self.committed_source_len = commit_end;
Some(out)
}
/// Finalize the stream and return any remaining raw source.
///
/// Ensures the returned source chunk is newline-terminated when non-empty so callers can
/// safely run markdown block parsing on the final chunk.
pub fn finalize_and_drain_source(&mut self) -> String {
if self.committed_source_len >= self.buffer.len() {
self.clear();
return String::new();
}
let mut out = self.buffer[self.committed_source_len..].to_string();
if !out.ends_with('\n') {
out.push('\n');
}
self.clear();
out
}
/// Render the full buffer and return only the newly completed logical lines
/// since the last commit. When the buffer does not end with a newline, the
/// final rendered line is considered incomplete and is not emitted.
#[cfg(test)]
pub fn commit_complete_lines(&mut self) -> Vec<Line<'static>> {
let source = self.buffer.clone();
let last_newline_idx = source.rfind('\n');
let source = if let Some(last_newline_idx) = last_newline_idx {
source[..=last_newline_idx].to_string()
} else {
let Some(commit_end) = self.buffer.rfind('\n').map(|idx| idx + 1) else {
return Vec::new();
};
if commit_end <= self.committed_source_len {
return Vec::new();
}
let source = self.buffer[..commit_end].to_string();
let mut rendered: Vec<Line<'static>> = Vec::new();
markdown::append_markdown(&source, self.width, Some(self.cwd.as_path()), &mut rendered);
let mut complete_line_count = rendered.len();
@@ -68,6 +134,7 @@ impl MarkdownStreamCollector {
let out_slice = &rendered[self.committed_line_count..complete_line_count];
let out = out_slice.to_vec();
self.committed_source_len = commit_end;
self.committed_line_count = complete_line_count;
out
}
@@ -76,17 +143,21 @@ impl MarkdownStreamCollector {
/// If the buffer does not end with a newline, a temporary one is appended
/// for rendering. Optionally unwraps ```markdown language fences in
/// non-test builds.
#[cfg(test)]
pub fn finalize_and_drain(&mut self) -> Vec<Line<'static>> {
let raw_buffer = self.buffer.clone();
let mut source: String = raw_buffer.clone();
let mut source = self.buffer.clone();
if source.is_empty() {
self.clear();
return Vec::new();
}
if !source.ends_with('\n') {
source.push('\n');
}
};
tracing::debug!(
raw_len = raw_buffer.len(),
raw_len = self.buffer.len(),
source_len = source.len(),
"markdown finalize (raw length: {}, rendered length: {})",
raw_buffer.len(),
self.buffer.len(),
source.len()
);
tracing::trace!("markdown finalize (raw source):\n---\n{source}\n---");
@@ -416,6 +487,42 @@ mod tests {
.collect()
}
#[tokio::test]
async fn table_header_commits_without_holdback() {
let mut c = super::MarkdownStreamCollector::new(None);
c.push_delta("| A | B |\n");
let out1 = c.commit_complete_lines();
let out1_str = lines_to_plain_strings(&out1);
assert_eq!(out1_str, vec!["| A | B |".to_string()]);
c.push_delta("| --- | --- |\n");
let out = c.commit_complete_lines();
let out_str = lines_to_plain_strings(&out);
assert!(
!out_str.is_empty(),
"expected output to continue committing after delimiter: {out_str:?}"
);
c.push_delta("| 1 | 2 |\n");
let out2 = c.commit_complete_lines();
assert!(
!out2.is_empty(),
"expected output to continue committing after body row"
);
c.push_delta("\n");
let _ = c.commit_complete_lines();
}
#[tokio::test]
async fn pipe_text_without_table_prefix_is_not_delayed() {
let mut c = super::MarkdownStreamCollector::new(None);
c.push_delta("Escaped pipe in text: a | b | c\n");
let out = c.commit_complete_lines();
let out_str = lines_to_plain_strings(&out);
assert_eq!(out_str, vec!["Escaped pipe in text: a | b | c".to_string()]);
}
#[tokio::test]
async fn lists_and_fences_commit_without_duplication() {
// List case
@@ -722,4 +829,9 @@ mod tests {
])
.await;
}
#[tokio::test]
async fn table_like_lines_inside_fenced_code_are_not_held() {
assert_streamed_equals_full(&["```\n", "| a | b |\n", "```\n"]).await;
}
}
codex-rs/tui/src/pager_overlay.rs
+24
View File
@@ -566,6 +566,30 @@ impl TranscriptOverlay {
}
}
/// Replace a range of committed cells with a single consolidated cell.
///
/// This mirros the splice performed on `App::transcript_cells` during `ConsolidateAgentMessage`
/// so the overlay stays in sync with the main transcript.
pub(crate) fn consolidate_cells(
&mut self,
range: std::ops::Range<usize>,
consolidated: Arc<dyn HistoryCell>,
) {
let follow_bottom = self.view.is_scrolled_to_bottom();
// Clamp the range to the overlay's cell count to avoid panic if the overlay has fewer
// cells than the main transcript (e.g. cells were inserted after the overlay has opened).
let clamped_end = range.end.min(self.cells.len());
let clamped_start = range.start.min(clamped_end);
if clamped_start < clamped_end {
self.cells
.splice(clamped_start..clamped_end, std::iter::once(consolidated));
self.rebuild_renderables();
}
if follow_bottom {
self.view.scroll_offset = usize::MAX;
}
}
/// Sync the active-cell live tail with the current width and cell state.
///
/// Recomputes the tail only when the cache key changes, preserving scroll
codex-rs/tui/src/render/line_utils.rs
+1
View File
@@ -26,6 +26,7 @@ pub fn push_owned_lines<'a>(src: &[Line<'a>], out: &mut Vec<Line<'static>>) {
/// Consider a line blank if it has no spans or only spans whose contents are
/// empty or consist solely of spaces (no tabs/newlines).
#[cfg(test)]
pub fn is_blank_line_spaces_only(line: &Line<'_>) -> bool {
if line.spans.is_empty() {
return true;
codex-rs/tui/src/streaming/controller.rs
+1101 -119
View File
File diff suppressed because it is too large Load Diff
codex-rs/tui/src/streaming/mod.rs
+3 -6
View File
@@ -70,12 +70,9 @@ impl StreamState {
.map(|queued| queued.line)
.collect()
}
/// Drains all queued lines from the front of the queue.
pub(crate) fn drain_all(&mut self) -> Vec<Line<'static>> {
self.queued_lines
.drain(..)
.map(|queued| queued.line)
.collect()
/// Clears queued lines while keeping collector/turn lifecycle state intact.
pub(crate) fn clear_queue(&mut self) {
self.queued_lines.clear();
}
/// Returns whether no lines are queued for commit.
pub(crate) fn is_idle(&self) -> bool {