diff --git a/codex-rs/tui/src/app.rs b/codex-rs/tui/src/app.rs index 4feefcac93..cf1061775d 100644 --- a/codex-rs/tui/src/app.rs +++ b/codex-rs/tui/src/app.rs @@ -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, pub(crate) deferred_history_lines: Vec>, has_emitted_history_lines: bool, + last_transcript_render_width: Option, + resize_reflow_pending_until: Option, + /// 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> { + 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 { 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::() + { + 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::() + && !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 = + 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)), diff --git a/codex-rs/tui/src/app_event.rs b/codex-rs/tui/src/app_event.rs index d727963f0d..e8e945a321 100644 --- a/codex-rs/tui/src/app_event.rs +++ b/codex-rs/tui/src/app_event.rs @@ -266,6 +266,10 @@ pub(crate) enum AppEvent { InsertHistoryCell(Box), + /// 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 diff --git a/codex-rs/tui/src/chatwidget.rs b/codex-rs/tui/src/chatwidget.rs index 34c0df7b4b..0a72278e43 100644 --- a/codex-rs/tui/src/chatwidget.rs +++ b/codex-rs/tui/src/chatwidget.rs @@ -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 { + 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::()) + { + 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::() + }); + 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::()) + { + 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() } diff --git a/codex-rs/tui/src/history_cell.rs b/codex-rs/tui/src/history_cell.rs index cf65f918e5..b20f73d695 100644 --- a/codex-rs/tui/src/history_cell.rs +++ b/codex-rs/tui/src/history_cell.rs @@ -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> { + let mut lines: Vec> = 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>, + is_first_line: bool, +} + +impl StreamingAgentTailCell { + pub(crate) fn new(lines: Vec>, is_first_line: bool) -> Self { + Self { + lines, + is_first_line, + } + } +} + +impl HistoryCell for StreamingAgentTailCell { + fn display_lines(&self, width: u16) -> Vec> { + 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>, @@ -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; + let head = Arc::new(AgentMessageCell::new(vec![Line::from("line 1")], true)) + as Arc; + let cont1 = Arc::new(AgentMessageCell::new(vec![Line::from("line 2")], false)) + as Arc; + let cont2 = Arc::new(AgentMessageCell::new(vec![Line::from("line 3")], false)) + as Arc; + + let mut transcript_cells: Vec> = + 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::() + { + start -= 1; + } + if start > 0 + && transcript_cells[start - 1] + .as_any() + .is::() + && !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 = + 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::(), + "first cell should be UserHistoryCell" + ); + + // Verify second cell is AgentMarkdownCell. + assert!( + transcript_cells[1].as_any().is::(), + "second cell should be AgentMarkdownCell" + ); + } } diff --git a/codex-rs/tui/src/markdown.rs b/codex-rs/tui/src/markdown.rs index e29246e76d..0d46d815ba 100644 --- a/codex-rs/tui/src/markdown.rs +++ b/codex-rs/tui/src/markdown.rs @@ -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>`. 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, + lines: &mut Vec>, +) { + 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, + cwd: Option<&Path>, + lines: &mut Vec>, +) { + 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 { + 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 = 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()]); + } } diff --git a/codex-rs/tui/src/markdown_stream.rs b/codex-rs/tui/src/markdown_stream.rs index 3eb37e345e..175cf4ca3d 100644 --- a/codex-rs/tui/src/markdown_stream.rs +++ b/codex-rs/tui/src/markdown_stream.rs @@ -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, + #[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, 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) { + 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 { + 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> { - 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> = 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> { - 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; + } } diff --git a/codex-rs/tui/src/pager_overlay.rs b/codex-rs/tui/src/pager_overlay.rs index bca5f1f360..d06846d700 100644 --- a/codex-rs/tui/src/pager_overlay.rs +++ b/codex-rs/tui/src/pager_overlay.rs @@ -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, + consolidated: Arc, + ) { + 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 diff --git a/codex-rs/tui/src/render/line_utils.rs b/codex-rs/tui/src/render/line_utils.rs index 175b79b2a8..54970f4486 100644 --- a/codex-rs/tui/src/render/line_utils.rs +++ b/codex-rs/tui/src/render/line_utils.rs @@ -26,6 +26,7 @@ pub fn push_owned_lines<'a>(src: &[Line<'a>], out: &mut Vec>) { /// 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; diff --git a/codex-rs/tui/src/streaming/controller.rs b/codex-rs/tui/src/streaming/controller.rs index d524e707a6..32e826b551 100644 --- a/codex-rs/tui/src/streaming/controller.rs +++ b/codex-rs/tui/src/streaming/controller.rs @@ -1,20 +1,272 @@ +//! Two-region streaming controllers for agent messages and proposed plans. +//! +//! Each stream partitions rendered markdown into a *stable region* (committed +//! to scrollback via the animation queue in `StreamState`) and a *tail region* +//! (mutable, displayed in the active-cell slot as `StreamingAgentTailCell`). +//! +//! `StreamCore` owns the shared bookkeeping: source accumulation, re-rendering, +//! stable/tail partitioning, commit-animation queue management, and terminal +//! resize handling. `StreamController` and `PlanStreamController` are thin +//! wrappers that add only their `emit()` styling and finalize return types. +//! +//! Table-aware holdback (`table_holdback_state`) keeps the entire buffer as +//! tail while a table is being streamed, because each new row can change column +//! widths and reshape every prior line. + use crate::history_cell::HistoryCell; use crate::history_cell::{self}; +use crate::markdown::append_markdown_agent_with_cwd; use crate::render::line_utils::prefix_lines; use crate::style::proposed_plan_style; use ratatui::prelude::Stylize; use ratatui::text::Line; use std::path::Path; +use std::path::PathBuf; use std::time::Duration; use std::time::Instant; +use crate::table_detect::is_table_delimiter_line; +use crate::table_detect::is_table_header_line; +use crate::table_detect::parse_table_segments; + use super::StreamState; -/// Controller that manages newline-gated streaming, header emission, and -/// commit animation across streams. -pub(crate) struct StreamController { +// --------------------------------------------------------------------------- +// StreamCore — shared bookkeeping for both stream controllers +// --------------------------------------------------------------------------- + +/// Shared state and logic for the two-region streaming model. +/// +/// Both [`StreamController`] (agent messages) and [`PlanStreamController`] +/// (proposed plans) delegate their core bookkeeping here: source +/// accumulation, re-rendering, stable/tail partitioning, commit-animation +/// queue management, and terminal resize handling. +/// +/// The wrapping controllers add only their own `emit()` styling and +/// finalize return types. +struct StreamCore { state: StreamState, - finishing_after_drain: bool, + /// Current rendering width (columns available for markdown content). + width: Option, + /// Accumulated raw markdown source for the current stream. + raw_source: String, + /// Full re-render of `raw_source` at `width`. Rebuilt on every committed delta. + rendered_lines: Vec>, + /// Lines enqueued into the commit-animation queue. + enqueued_stable_len: usize, + /// Lines actually emitted to scrollback. + emitted_stable_len: usize, + /// Session cwd used to keep local file-link display stable during stream re-renders. + cwd: PathBuf, +} + +impl StreamCore { + fn new(width: Option, cwd: &Path) -> Self { + Self { + state: StreamState::new(width, cwd), + width, + raw_source: String::new(), + rendered_lines: Vec::new(), + enqueued_stable_len: 0, + emitted_stable_len: 0, + cwd: cwd.to_path_buf(), + } + } + + /// Push a delta; if it contains a newline, commit completed lines and enqueue newly-stable + /// lines. Returns `true` if new lines were enqueued. + fn push_delta(&mut self, delta: &str) -> bool { + if !delta.is_empty() { + self.state.has_seen_delta = true; + } + self.state.collector.push_delta(delta); + + let mut enqueued = false; + if delta.contains('\n') + && let Some(committed_source) = self.state.collector.commit_complete_source() + { + self.raw_source.push_str(&committed_source); + self.recompute_streaming_render(); + enqueued = self.sync_stable_queue(); + } + enqueued + } + + /// Drain the collector, re-render, and return lines not yet emitted. + /// Does NOT reset state - the caller must call `reset()` afterward. + fn finalize_remaining(&mut self) -> Vec> { + let remainder_source = self.state.collector.finalize_and_drain_source(); + if !remainder_source.is_empty() { + self.raw_source.push_str(&remainder_source); + } + let mut rendered = Vec::new(); + append_markdown_agent_with_cwd( + &self.raw_source, + self.width, + Some(self.cwd.as_path()), + &mut rendered, + ); + if self.emitted_stable_len >= rendered.len() { + Vec::new() + } else { + rendered[self.emitted_stable_len..].to_vec() + } + } + + /// Step animation: dequeue one line, update the emitted count. + fn tick(&mut self) -> Vec> { + let step = self.state.step(); + self.emitted_stable_len += step.len(); + step + } + + /// Batch drain: dequeue up to `max_lines`, update the emitted count. + fn tick_batch(&mut self, max_lines: usize) -> Vec> { + let step = self.state.drain_n(max_lines.max(1)); + self.emitted_stable_len += step.len(); + step + } + + fn is_idle(&self) -> bool { + self.state.is_idle() + } + + fn queued_lines(&self) -> usize { + self.state.queued_len() + } + + fn oldest_queued_age(&self, now: Instant) -> Option { + self.state.oldest_queued_age(now) + } + + /// Mutable tail lines not yet queued into the stable region. + fn current_tail_lines(&self) -> Vec> { + let start = self.enqueued_stable_len.min(self.rendered_lines.len()); + self.rendered_lines[start..].to_vec() + } + + /// Update rendering width and rebuild queued stable lines for the new layout. + fn set_width(&mut self, width: Option) { + if self.width == width { + return; + } + let old_width = self.width; + self.width = width; + self.state.collector.set_width(width); + if self.raw_source.is_empty() { + return; + } + + // Recalculate emitted_stabe_len for the new width so we don't re-queue lines that were + // already emitted at the old width. + if self.emitted_stable_len > 0 { + let emitted_bytes = source_bytes_for_rendered_count( + &self.raw_source, + old_width, + self.cwd.as_path(), + self.emitted_stable_len, + ); + let mut emitted_at_new = Vec::new(); + append_markdown_agent_with_cwd( + &self.raw_source[..emitted_bytes], + self.width, + Some(self.cwd.as_path()), + &mut emitted_at_new, + ); + self.emitted_stable_len = emitted_at_new.len(); + } + + self.recompute_streaming_render(); + self.rebuild_stable_queue_from_render(); + } + + /// Clear all accumulated state for current stream. + fn reset(&mut self) { + self.state.clear(); + self.raw_source.clear(); + self.rendered_lines.clear(); + self.enqueued_stable_len = 0; + self.emitted_stable_len = 0; + } + + /// Re-render the full `raw_source` at current `width`. + fn recompute_streaming_render(&mut self) { + let mut rendered = Vec::new(); + append_markdown_agent_with_cwd( + &self.raw_source, + self.width, + Some(self.cwd.as_path()), + &mut rendered, + ); + self.rendered_lines = rendered; + } + + /// Advance `enqueued_stable_len` toward the target stable boundary and enqueue any + /// newly-stable lines. Returns `true` if new lines were enqueued. + fn sync_stable_queue(&mut self) -> bool { + let tail_budget = self.active_tail_budget_lines(); + let target_stable_len = self + .rendered_lines + .len() + .saturating_sub(tail_budget) + .max(self.emitted_stable_len); + + // A structural rewrite moved the stable boundary backward into enqueue-but-unemitted + // lines. Rebuild queue from the latest snapshot. + if target_stable_len < self.enqueued_stable_len { + self.state.clear_queue(); + if self.emitted_stable_len < target_stable_len { + self.state.enqueue( + self.rendered_lines[self.emitted_stable_len..target_stable_len].to_vec(), + ); + } + self.enqueued_stable_len = target_stable_len; + return self.state.queued_len() > 0; + } + + if target_stable_len == self.enqueued_stable_len { + return false; + } + + self.state + .enqueue(self.rendered_lines[self.enqueued_stable_len..target_stable_len].to_vec()); + self.enqueued_stable_len = target_stable_len; + true + } + + /// Rebuild the stable queue from the current render snapshot. Used after `set_width()` where + /// the old queue is stale. + fn rebuild_stable_queue_from_render(&mut self) { + let tail_budget = self.active_tail_budget_lines(); + let target_stable_len = self + .rendered_lines + .len() + .saturating_sub(tail_budget) + .max(self.emitted_stable_len); + self.state.clear_queue(); + if self.emitted_stable_len < target_stable_len { + self.state + .enqueue(self.rendered_lines[self.emitted_stable_len..target_stable_len].to_vec()); + } + self.enqueued_stable_len = target_stable_len; + } + + /// How many rendered lines to withhold as mutable tail. When a table is detected, the entire + /// buffer is tail; otherwise zero. + fn active_tail_budget_lines(&self) -> usize { + match table_holdback_state(&self.raw_source) { + TableHoldbackState::Confirmed => self.rendered_lines.len(), + TableHoldbackState::PendingHeader => self.rendered_lines.len(), + TableHoldbackState::None => 0, + } + } +} + +/// Controller for streaming agent message content with table-aware holdback. +/// +/// Wraps [`StreamCore`] and adds `AgentMessageCell` emission styling. +pub(crate) struct StreamController { + core: StreamCore, header_emitted: bool, } @@ -25,79 +277,66 @@ impl StreamController { /// render against the same session cwd that was active when streaming started. pub(crate) fn new(width: Option, cwd: &Path) -> Self { Self { - state: StreamState::new(width, cwd), - finishing_after_drain: false, + core: StreamCore::new(width, cwd), header_emitted: false, } } - /// Push a delta; if it contains a newline, commit completed lines and start animation. pub(crate) fn push(&mut self, delta: &str) -> bool { - let state = &mut self.state; - if !delta.is_empty() { - state.has_seen_delta = true; - } - state.collector.push_delta(delta); - if delta.contains('\n') { - let newly_completed = state.collector.commit_complete_lines(); - if !newly_completed.is_empty() { - state.enqueue(newly_completed); - return true; - } - } - false + self.core.push_delta(delta) } - /// Finalize the active stream. Drain and emit now. - pub(crate) fn finalize(&mut self) -> Option> { - // Finalize collector first. - let remaining = { - let state = &mut self.state; - state.collector.finalize_and_drain() - }; - // Collect all output first to avoid emitting headers when there is no content. - let mut out_lines = Vec::new(); - { - let state = &mut self.state; - if !remaining.is_empty() { - state.enqueue(remaining); - } - let step = state.drain_all(); - out_lines.extend(step); + /// Finalize the active stream. Returns the final cell (if any remaining lines) and the raw + /// markdown source for consolidation. + pub(crate) fn finalize(&mut self) -> (Option>, Option) { + let remaining = self.core.finalize_remaining(); + if self.core.raw_source.is_empty() { + self.core.reset(); + return (None, None); } - // Cleanup - self.state.clear(); - self.finishing_after_drain = false; - self.emit(out_lines) + // Capture the source before reset clears it. + let source = self.core.raw_source.clone(); + let out = self.emit(remaining); + self.core.reset(); + (out, Some(source)) } - /// Step animation: commit at most one queued line and handle end-of-drain cleanup. pub(crate) fn on_commit_tick(&mut self) -> (Option>, bool) { - let step = self.state.step(); - (self.emit(step), self.state.is_idle()) + let step = self.core.tick(); + (self.emit(step), self.core.is_idle()) } - /// Step animation: commit at most `max_lines` queued lines. - /// - /// This is intended for adaptive catch-up drains. Callers should keep `max_lines` bounded; a - /// very large value can collapse perceived animation into a single jump. pub(crate) fn on_commit_tick_batch( &mut self, max_lines: usize, ) -> (Option>, bool) { - let step = self.state.drain_n(max_lines.max(1)); - (self.emit(step), self.state.is_idle()) + let step = self.core.tick_batch(max_lines); + (self.emit(step), self.core.is_idle()) } - /// Returns the current number of queued lines waiting to be displayed. pub(crate) fn queued_lines(&self) -> usize { - self.state.queued_len() + self.core.queued_lines() } - /// Returns the age of the oldest queued line. pub(crate) fn oldest_queued_age(&self, now: Instant) -> Option { - self.state.oldest_queued_age(now) + self.core.oldest_queued_age(now) + } + + pub(crate) fn current_tail_lines(&self) -> Vec> { + self.core.current_tail_lines() + } + + pub(crate) fn tail_starts_stream(&self) -> bool { + !self.header_emitted && self.core.enqueued_stable_len == 0 + } + + pub(crate) fn has_live_tail(&self) -> bool { + !self.current_tail_lines().is_empty() + } + + pub(crate) fn set_width(&mut self, width: Option) { + self.core.set_width(width); } fn emit(&mut self, lines: Vec>) -> Option> { @@ -111,10 +350,16 @@ impl StreamController { }))) } } +// --------------------------------------------------------------------------- +// PlanStreamController — proposed plan streams +// --------------------------------------------------------------------------- /// Controller that streams proposed plan markdown into a styled plan block. +/// +/// Wraps [`StreamCore`] and adds plan-specific header, indentation, and +/// background styling. pub(crate) struct PlanStreamController { - state: StreamState, + core: StreamCore, header_emitted: bool, top_padding_emitted: bool, } @@ -127,81 +372,47 @@ impl PlanStreamController { /// render against the same session cwd that was active when streaming started. pub(crate) fn new(width: Option, cwd: &Path) -> Self { Self { - state: StreamState::new(width, cwd), + core: StreamCore::new(width, cwd), header_emitted: false, top_padding_emitted: false, } } - /// Push a delta; if it contains a newline, commit completed lines and start animation. pub(crate) fn push(&mut self, delta: &str) -> bool { - let state = &mut self.state; - if !delta.is_empty() { - state.has_seen_delta = true; - } - state.collector.push_delta(delta); - if delta.contains('\n') { - let newly_completed = state.collector.commit_complete_lines(); - if !newly_completed.is_empty() { - state.enqueue(newly_completed); - return true; - } - } - false + self.core.push_delta(delta) } /// Finalize the active stream. Drain and emit now. pub(crate) fn finalize(&mut self) -> Option> { - let remaining = { - let state = &mut self.state; - state.collector.finalize_and_drain() - }; - let mut out_lines = Vec::new(); - { - let state = &mut self.state; - if !remaining.is_empty() { - state.enqueue(remaining); - } - let step = state.drain_all(); - out_lines.extend(step); - } - - self.state.clear(); - self.emit(out_lines, /*include_bottom_padding*/ true) + let remaining = self.core.finalize_remaining(); + let out = self.emit(remaining, true); + self.core.reset(); + out } - /// Step animation: commit at most one queued line and handle end-of-drain cleanup. pub(crate) fn on_commit_tick(&mut self) -> (Option>, bool) { - let step = self.state.step(); - ( - self.emit(step, /*include_bottom_padding*/ false), - self.state.is_idle(), - ) + let step = self.core.tick(); + (self.emit(step, false), self.core.is_idle()) } - /// Step animation: commit at most `max_lines` queued lines. - /// - /// This is intended for adaptive catch-up drains. Callers should keep `max_lines` bounded; a - /// very large value can collapse perceived animation into a single jump. pub(crate) fn on_commit_tick_batch( &mut self, max_lines: usize, ) -> (Option>, bool) { - let step = self.state.drain_n(max_lines.max(1)); - ( - self.emit(step, /*include_bottom_padding*/ false), - self.state.is_idle(), - ) + let step = self.core.tick_batch(max_lines); + (self.emit(step, false), self.core.is_idle()) } - /// Returns the current number of queued plan lines waiting to be displayed. pub(crate) fn queued_lines(&self) -> usize { - self.state.queued_len() + self.core.queued_lines() } - /// Returns the age of the oldest queued plan line. pub(crate) fn oldest_queued_age(&self, now: Instant) -> Option { - self.state.oldest_queued_age(now) + self.core.oldest_queued_age(now) + } + + pub(crate) fn set_width(&mut self, width: Option) { + self.core.set_width(width); } fn emit( @@ -245,6 +456,203 @@ impl PlanStreamController { } } +// --------------------------------------------------------------------------- +// source_bytes_for_rendered_count — resize remapping helper +// --------------------------------------------------------------------------- + +/// Find the largest newline-terminated prefix of `raw_source` whose +/// rendering at `width` produces at most `target_count` lines. +/// +/// When the target falls exactly on a source-line boundary, the returned +/// offset covers that line. When it falls in the middle of a wrapped +/// source line (partial drain), the offset stops at the *previous* +/// newline to avoid overshooting — this may re-queue a few already-emitted +/// wrapped lines as duplicates, but never drops un-emitted content. +/// +/// For non-table content (the only case where `emitted_stable_len > 0`), +/// rendering a newline-terminated prefix produces a prefix of the full +/// rendering, so this converges correctly. +fn source_bytes_for_rendered_count( + raw_source: &str, + width: Option, + cwd: &Path, + target_count: usize, +) -> usize { + if target_count == 0 { + return 0; + } + let mut best_offset = 0; + for (i, _) in raw_source.match_indices('\n') { + let prefix = &raw_source[..=i]; + let mut lines = Vec::new(); + append_markdown_agent_with_cwd(prefix, width, Some(cwd), &mut lines); + if lines.len() <= target_count { + best_offset = i + 1; + } + if lines.len() >= target_count { + break; + } + } + best_offset +} + +// --------------------------------------------------------------------------- +// Table holdback infrastructure +// --------------------------------------------------------------------------- + +fn parse_fence_marker(line: &str) -> Option<(char, usize)> { + let mut chars = line.chars(); + let first = chars.next()?; + if first != '`' && first != '~' { + return None; + } + let mut len = 1usize; + for ch in chars { + if ch == first { + len += 1; + } else { + break; + } + } + if len < 3 { + return None; + } + Some((first, len)) +} + +/// A source line annotated with whether it falls inside a fenced code block. +struct ParsedLine<'a> { + text: &'a str, + fence_context: FenceContext, +} + +/// Where a source line sits relative to fenced code blocks. +/// +/// Table holdback only applies to lines that are `Outside` or inside a +/// `Markdown` fence. Lines inside `Other` fences (e.g. `sh`, `rust`) are +/// ignored by the table scanner because their pipe characters are code, not +/// table syntax. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum FenceContext { + /// Not inside any fenced code block. + Outside, + /// Inside a `` ```md `` or `` ```markdown `` fence. + Markdown, + /// Inside a fence with a non-markdown info string. + Other, +} + +fn is_markdown_fence(trimmed_line: &str, marker_len: usize) -> bool { + let info = trimmed_line[marker_len..] + .split_whitespace() + .next() + .unwrap_or_default(); + info.eq_ignore_ascii_case("md") || info.eq_ignore_ascii_case("markdown") +} + +fn parse_lines_with_fence_state(source: &str) -> Vec> { + let mut in_fence = false; + let mut fence_char = '\0'; + let mut fence_context = FenceContext::Other; + let mut lines = Vec::new(); + + for raw_line in source.split('\n') { + lines.push(ParsedLine { + text: raw_line, + fence_context: if in_fence { + fence_context + } else { + FenceContext::Outside + }, + }); + + let trimmed = raw_line.trim_start(); + if let Some((marker, len)) = parse_fence_marker(trimmed) { + if !in_fence { + in_fence = true; + fence_char = marker; + fence_context = if is_markdown_fence(trimmed, len) { + FenceContext::Markdown + } else { + FenceContext::Other + }; + } else if marker == fence_char && len >= 3 { + in_fence = false; + fence_context = FenceContext::Other; + } + } + } + + lines +} + +fn strip_blockquote_prefix(line: &str) -> &str { + let mut rest = line.trim_start(); + loop { + let Some(stripped) = rest.strip_prefix('>') else { + return rest; + }; + rest = stripped.strip_prefix(' ').unwrap_or(stripped).trim_start(); + } +} + +fn table_candidate_text(line: &str) -> Option<&str> { + let stripped = strip_blockquote_prefix(line).trim(); + parse_table_segments(stripped).map(|_| stripped) +} + +/// Whether the accumulated raw source contains a markdown table that requires +/// holdback of the mutable tail to prevent partial-table commits. +enum TableHoldbackState { + /// No table detected -- all rendered lines can flow into the stable queue. + None, + /// The last non-blank line looks like a table header row but no delimiter + /// row has followed yet. Hold back in case the next delta is a delimiter. + PendingHeader, + /// A header + delimiter pair was found -- the source contains a confirmed + /// table. The entire rendered buffer is held as mutable tail. + Confirmed, +} + +/// Scan `source` for pipe-table patterns (header row followed by delimiter row) +/// outside of non-markdown fenced code blocks. Used by the stream controllers +/// to decide the tail budget. +fn table_holdback_state(source: &str) -> TableHoldbackState { + let lines = parse_lines_with_fence_state(source); + for pair in lines.windows(2) { + let [header_line, delimiter_line] = pair else { + continue; + }; + if header_line.fence_context == FenceContext::Other + || delimiter_line.fence_context == FenceContext::Other + { + continue; + } + + let Some(header_text) = table_candidate_text(header_line.text) else { + continue; + }; + let Some(delimiter_text) = table_candidate_text(delimiter_line.text) else { + continue; + }; + + if is_table_header_line(header_text) && is_table_delimiter_line(delimiter_text) { + return TableHoldbackState::Confirmed; + } + } + + let pending_header = lines.iter().rev().find(|line| !line.text.trim().is_empty()); + let pending_header = pending_header.is_some_and(|line| { + line.fence_context != FenceContext::Other + && table_candidate_text(line.text).is_some_and(is_table_header_line) + }); + if pending_header { + TableHoldbackState::PendingHeader + } else { + TableHoldbackState::None + } +} + #[cfg(test)] mod tests { use super::*; @@ -256,6 +664,16 @@ mod tests { std::env::temp_dir() } + fn stream_controller(width: Option) -> StreamController { + let cwd = test_cwd(); + StreamController::new(width, &cwd) + } + + fn plan_stream_controller(width: Option) -> PlanStreamController { + let cwd = test_cwd(); + PlanStreamController::new(width, &cwd) + } + fn lines_to_plain_strings(lines: &[ratatui::text::Line<'_>]) -> Vec { lines .iter() @@ -269,12 +687,154 @@ mod tests { .collect() } + fn collect_streamed_lines(deltas: &[&str], width: Option) -> Vec { + let mut ctrl = stream_controller(width); + let mut lines = Vec::new(); + for d in deltas { + ctrl.push(d); + while let (Some(cell), idle) = ctrl.on_commit_tick() { + lines.extend(cell.transcript_lines(u16::MAX)); + if idle { + break; + } + } + } + if let (Some(cell), _source) = ctrl.finalize() { + lines.extend(cell.transcript_lines(u16::MAX)); + } + lines_to_plain_strings(&lines) + .into_iter() + .map(|s| s.chars().skip(2).collect::()) + .collect() + } + + fn collect_plan_streamed_lines(deltas: &[&str], width: Option) -> Vec { + let mut ctrl = plan_stream_controller(width); + let mut lines = Vec::new(); + for d in deltas { + ctrl.push(d); + while let (Some(cell), idle) = ctrl.on_commit_tick() { + lines.extend(cell.transcript_lines(u16::MAX)); + if idle { + break; + } + } + } + if let Some(cell) = ctrl.finalize() { + lines.extend(cell.transcript_lines(u16::MAX)); + } + lines_to_plain_strings(&lines) + } + + #[test] + 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_eq!(ctrl.queued_lines(), 1); + + ctrl.set_width(Some(24)); + let (cell, idle) = ctrl.on_commit_tick_batch(usize::MAX); + let rendered = lines_to_plain_strings( + &cell + .expect("expected resized queued lines") + .transcript_lines(u16::MAX), + ); + + assert!(idle); + assert!( + rendered.len() > 1, + "expected resized content to occupy multiple lines, got {rendered:?}", + ); + } + + #[test] + fn controller_set_width_no_duplicate_after_emit() { + let mut ctrl = stream_controller(Some(120)); + let line = + "This is a long line that definitely wraps when the terminal shrinks to 24 columns.\n"; + ctrl.push(line); + let (cell, _) = ctrl.on_commit_tick_batch(usize::MAX); + assert!(cell.is_some(), "expected emitted cell"); + assert_eq!(ctrl.queued_lines(), 0); + + ctrl.set_width(Some(24)); + + assert_eq!( + ctrl.queued_lines(), + 0, + "already-emitted content must not be re-queued after resize", + ); + } + + #[test] + fn controller_set_width_partial_drain_no_lost_lines() { + let mut ctrl = stream_controller(Some(40)); + ctrl.push("AAAA BBBB CCCC DDDD EEEE FFFF GGGG HHHH IIII JJJJ\n"); + ctrl.push("second line\n"); + + let (cell, idle) = ctrl.on_commit_tick(); + assert!(cell.is_some(), "expected 1 emitted line"); + assert!(!idle, "queue should still have lines"); + let remaining_before = ctrl.queued_lines(); + assert!(remaining_before > 0, "should have queued lines left"); + + ctrl.set_width(Some(20)); + + let (cell, source) = ctrl.finalize(); + let final_lines = cell + .map(|c| lines_to_plain_strings(&c.transcript_lines(u16::MAX))) + .unwrap_or_default(); + + assert!( + final_lines.iter().any(|l| l.contains("second line")), + "un-emitted 'second line' was lost after resize; got: {final_lines:?}", + ); + assert!(source.is_some(), "expected source from finalize"); + } + + #[test] + fn controller_set_width_preserves_in_flight_tail() { + let mut ctrl = stream_controller(Some(80)); + ctrl.push("tail without newline"); + ctrl.set_width(Some(24)); + + let (cell, _source) = ctrl.finalize(); + let rendered = lines_to_plain_strings( + &cell + .expect("expected finalized tail") + .transcript_lines(u16::MAX), + ); + + assert_eq!(rendered, vec!["• tail without newline".to_string()]); + } + + #[test] + fn plan_controller_set_width_preserves_in_flight_tail() { + let mut ctrl = plan_stream_controller(Some(80)); + ctrl.push("1. Item without newline"); + ctrl.set_width(Some(24)); + + let rendered = lines_to_plain_strings( + &ctrl + .finalize() + .expect("expected finalized tail") + .transcript_lines(u16::MAX), + ); + + assert!( + rendered + .iter() + .any(|line| line.contains("Item without newline")), + "expected finalized plan content after resize, got {rendered:?}", + ); + } + #[tokio::test] async fn controller_loose_vs_tight_with_commit_ticks_matches_full() { - let mut ctrl = StreamController::new(/*width*/ None, &test_cwd()); + let mut ctrl = stream_controller(/*width*/ None); let mut lines = Vec::new(); - // Exact deltas from the session log (section: Loose vs. tight list items) let deltas = vec![ "\n\n", "Loose", @@ -345,7 +905,6 @@ mod tests { "\n\n", ]; - // Simulate streaming with a commit tick attempt after each delta. for d in deltas.iter() { ctrl.push(d); while let (Some(cell), idle) = ctrl.on_commit_tick() { @@ -355,32 +914,22 @@ mod tests { } } } - // Finalize and flush remaining lines now. - if let Some(cell) = ctrl.finalize() { + if let (Some(cell), _source) = ctrl.finalize() { lines.extend(cell.transcript_lines(u16::MAX)); } let streamed: Vec<_> = lines_to_plain_strings(&lines) .into_iter() - // skip • and 2-space indentation .map(|s| s.chars().skip(2).collect::()) .collect(); - // Full render of the same source let source: String = deltas.iter().copied().collect(); let mut rendered: Vec> = Vec::new(); - let test_cwd = test_cwd(); - crate::markdown::append_markdown( - &source, - /*width*/ None, - Some(test_cwd.as_path()), - &mut rendered, - ); + crate::markdown::append_markdown_agent(&source, None, &mut rendered); let rendered_strs = lines_to_plain_strings(&rendered); assert_eq!(streamed, rendered_strs); - // Also assert exact expected plain strings for clarity. let expected = vec![ "Loose vs. tight list items:".to_string(), "".to_string(), @@ -398,4 +947,437 @@ mod tests { "expected exact rendered lines for loose/tight section" ); } + + #[tokio::test] + async fn controller_streamed_table_matches_full_render_widths() { + let deltas = vec![ + "| Key | Description |\n", + "| --- | --- |\n", + "| -v | Enable very verbose logging output for debugging |\n", + "\n", + ]; + + let streamed = collect_streamed_lines(&deltas, Some(80)); + + let source: String = deltas.iter().copied().collect(); + let mut rendered = Vec::new(); + crate::markdown::append_markdown_agent(&source, Some(80), &mut rendered); + let expected = lines_to_plain_strings(&rendered); + + assert_eq!(streamed, expected); + } + + #[tokio::test] + async fn controller_holds_blockquoted_table_tail_until_stable() { + let deltas = vec![ + "> | A | B |\n", + "> | --- | --- |\n", + "> | longvalue | ok |\n", + "\n", + ]; + + let streamed = collect_streamed_lines(&deltas, Some(80)); + + let source: String = deltas.iter().copied().collect(); + let mut rendered = Vec::new(); + crate::markdown::append_markdown_agent(&source, Some(80), &mut rendered); + let expected = lines_to_plain_strings(&rendered); + + assert_eq!(streamed, expected); + } + + #[tokio::test] + async fn controller_does_not_hold_back_pipe_prose_without_table_delimiter() { + let mut ctrl = stream_controller(Some(80)); + + ctrl.push("status | owner | note\n"); + let (_first_commit, first_idle) = ctrl.on_commit_tick(); + assert!(first_idle); + + ctrl.push("next line\n"); + let (second_commit, _second_idle) = ctrl.on_commit_tick(); + assert!( + second_commit.is_some(), + "expected prose lines to be released once no table delimiter follows" + ); + } + + #[tokio::test] + async fn controller_handles_table_immediately_after_heading() { + let deltas = vec![ + "### 1) Basic table\n", + "| Name | Role | Status |\n", + "|---|---|---|\n", + "| Alice | Admin | Active |\n", + "| Bob | Editor | Pending |\n", + "\n", + ]; + + let streamed = collect_streamed_lines(&deltas, Some(100)); + + let source: String = deltas.iter().copied().collect(); + let mut rendered = Vec::new(); + crate::markdown::append_markdown_agent(&source, Some(100), &mut rendered); + let expected = lines_to_plain_strings(&rendered); + + assert_eq!(streamed, expected); + } + + #[tokio::test] + async fn controller_renders_unicode_for_multi_table_response_shape() { + let source = "Absolutely. Here are several different Markdown table patterns you can use for rendering tests.\n\n| Name | Role | + Location |\n|-------|-----------|----------|\n| Ava | Engineer | NYC |\n| Malik | Designer | Berlin |\n| Priya | PM | Remote + |\n\n| Item | Qty | Price | In Stock |\n|:------------|----:|------:|:--------:|\n| Keyboard | 2 | 49.99 | Yes |\n| Mouse | 10 + | 19.50 | Yes |\n| Monitor | 1 | 219.0 | No |\n\n| Field | Example | Notes + |\n|---------------|----------------------------------|--------------------------|\n| Escaped pipe | `foo \\| bar` | Should stay + in one cell |\n| Inline code | `let x = value;` | Monospace inline content |\n| Link | [OpenAI](https://openai.com) | + Standard markdown link |\n"; + + let chunked = source + .split_inclusive('\n') + .map(ToString::to_string) + .collect::>(); + let deltas = chunked.iter().map(String::as_str).collect::>(); + let streamed = collect_streamed_lines(&deltas, Some(120)); + assert!( + streamed.iter().any(|line| line.contains('┌')), + "expected unicode table border in streamed output: {streamed:?}" + ); + } + + #[tokio::test] + async fn controller_renders_unicode_for_no_outer_pipes_table_shape() { + let source = "### 1) Basic\n\n| Name | Role | Active |\n|---|---|---|\n| Alice | Engineer | Yes |\n| Bob | Designer | No |\n\n### 2) No outer + pipes\n\nCol A | Col B | Col C\n--- | --- | ---\nx | y | z\n10 | 20 | 30\n\n### 3) Another table\n\n| Key | Value |\n|---|---|\n| a | b |\n"; + + let chunked = source + .split_inclusive('\n') + .map(ToString::to_string) + .collect::>(); + let deltas = chunked.iter().map(String::as_str).collect::>(); + let streamed = collect_streamed_lines(&deltas, Some(100)); + + let mut rendered = Vec::new(); + crate::markdown::append_markdown_agent(source, Some(100), &mut rendered); + let expected = lines_to_plain_strings(&rendered); + + assert_eq!(streamed, expected); + let has_raw_no_outer_header = streamed + .iter() + .any(|line| line.trim() == "Col A | Col B | Col C"); + assert!( + !has_raw_no_outer_header, + "no-outer-pipes header should not remain raw in final streamed output: {streamed:?}" + ); + } + + #[tokio::test] + async fn controller_stabilizes_first_no_outer_pipes_table_in_response() { + let deltas = vec![ + "### No outer pipes first\n\n", + "Col A | Col B | Col C\n", + "--- | --- | ---\n", + "x | y | z\n", + "10 | 20 | 30\n", + "\n", + "After table paragraph.\n", + ]; + let streamed = collect_streamed_lines(&deltas, Some(100)); + + let source: String = deltas.iter().copied().collect(); + let mut rendered = Vec::new(); + crate::markdown::append_markdown_agent(&source, Some(100), &mut rendered); + let expected = lines_to_plain_strings(&rendered); + + assert_eq!(streamed, expected); + assert!( + streamed.iter().any(|line| line.contains('┌')), + "expected unicode table border for no-outer-pipes streaming: {streamed:?}" + ); + assert!( + !streamed + .iter() + .any(|line| line.trim() == "Col A | Col B | Col C"), + "did not expect raw no-outer-pipes header in final streamed output: {streamed:?}" + ); + } + + #[tokio::test] + async fn controller_stabilizes_two_column_no_outer_table_in_response() { + let deltas = vec![ + "A | B\n", + "--- | ---\n", + "left | right\n", + "\n", + "After table paragraph.\n", + ]; + let streamed = collect_streamed_lines(&deltas, Some(80)); + + let source: String = deltas.iter().copied().collect(); + let mut rendered = Vec::new(); + crate::markdown::append_markdown_agent(&source, Some(80), &mut rendered); + let expected = lines_to_plain_strings(&rendered); + + assert_eq!(streamed, expected); + assert!( + streamed.iter().any(|line| line.contains('┌')), + "expected unicode table border for two-column no-outer table: {streamed:?}" + ); + assert!( + !streamed.iter().any(|line| line.trim() == "A | B"), + "did not expect raw two-column no-outer header in final streamed output: {streamed:?}" + ); + } + + #[tokio::test] + async fn controller_converts_no_outer_table_between_preboxed_sections() { + let source = " ┌───────┬──────────┬────────┐\n │ Name │ Role │ Active │\n ├───────┼──────────┼────────┤\n │ Alice │ Engineer │ Yes │\n │ Bob │ Designer │ No │\n │ Cara │ PM │ Yes │\n └───────┴──────────┴────────┘\n\n ### 3) No outer pipes\n\n Col A | Col B | Col C\n --- | --- | ---\n x | y | z\n 10 | 20 | 30\n\n ┌─────────────────┬────────┬────────────────────────┐\n │ Example │ Output │ Notes │\n ├─────────────────┼────────┼────────────────────────┤\n │ a | b │ `a │ b` │\n │ npm run test │ ok │ Inline code formatting │\n │ SELECT * FROM t │ 3 rows │ SQL snippet │\n └─────────────────┴────────┴────────────────────────┘\n"; + + let deltas = source + .split_inclusive('\n') + .map(ToString::to_string) + .collect::>(); + let streamed = collect_streamed_lines( + &deltas.iter().map(String::as_str).collect::>(), + Some(100), + ); + + let has_raw_no_outer_header = streamed + .iter() + .any(|line| line.trim() == "Col A | Col B | Col C"); + assert!( + !has_raw_no_outer_header, + "no-outer table header remained raw in streamed output: {streamed:?}" + ); + assert!( + streamed + .iter() + .any(|line| line.contains("┌───────┬───────┬───────┐")), + "expected converted no-outer table border in streamed output: {streamed:?}" + ); + } + + #[tokio::test] + async fn controller_keeps_markdown_fenced_tables_mutable_until_finalize() { + let source = "```md\n| A | B |\n|---|---|\n| 1 | 2 |\n```\n"; + let deltas = vec![ + "```md\n", + "| A | B |\n", + "|---|---|\n", + "| 1 | 2 |\n", + "```\n", + ]; + let streamed = collect_streamed_lines(&deltas, Some(80)); + + let mut rendered = Vec::new(); + crate::markdown::append_markdown_agent(source, Some(80), &mut rendered); + let expected = lines_to_plain_strings(&rendered); + + assert_eq!(streamed, expected); + assert!( + streamed.iter().any(|line| line.contains('┌')), + "expected unicode table border in streamed output: {streamed:?}" + ); + assert!( + !streamed.iter().any(|line| line.trim() == "| A | B |"), + "did not expect raw table header line after finalize: {streamed:?}" + ); + } + + #[tokio::test] + async fn controller_keeps_markdown_fenced_no_outer_tables_mutable_until_finalize() { + let source = + "```md\nCol A | Col B | Col C\n--- | --- | ---\nx | y | z\n10 | 20 | 30\n```\n"; + let deltas = vec![ + "```md\n", + "Col A | Col B | Col C\n", + "--- | --- | ---\n", + "x | y | z\n", + "10 | 20 | 30\n", + "```\n", + ]; + let streamed = collect_streamed_lines(&deltas, Some(100)); + + let mut rendered = Vec::new(); + crate::markdown::append_markdown_agent(source, Some(100), &mut rendered); + let expected = lines_to_plain_strings(&rendered); + + assert_eq!(streamed, expected); + assert!( + streamed.iter().any(|line| line.contains('┌')), + "expected unicode table border in streamed output: {streamed:?}" + ); + assert!( + !streamed + .iter() + .any(|line| line.trim() == "Col A | Col B | Col C"), + "did not expect raw no-outer-pipes header line after finalize: {streamed:?}" + ); + } + + #[tokio::test] + async fn controller_keeps_non_markdown_fenced_tables_as_code() { + let source = "```sh\n| A | B |\n|---|---|\n| 1 | 2 |\n```\n"; + let deltas = vec![ + "```sh\n", + "| A | B |\n", + "|---|---|\n", + "| 1 | 2 |\n", + "```\n", + ]; + let streamed = collect_streamed_lines(&deltas, Some(80)); + + let mut rendered = Vec::new(); + crate::markdown::append_markdown_agent(source, Some(80), &mut rendered); + let expected = lines_to_plain_strings(&rendered); + + assert_eq!(streamed, expected); + assert!( + streamed.iter().any(|line| line.trim() == "| A | B |"), + "expected code-fenced pipe line to remain raw: {streamed:?}" + ); + assert!( + !streamed.iter().any(|line| line.contains('┌')), + "did not expect unicode table border for non-markdown fence: {streamed:?}" + ); + } + + #[tokio::test] + async fn plan_controller_streamed_table_matches_finalize_render() { + let deltas = vec![ + "## Build plan\n\n", + "| Step | Owner |\n", + "|---|---|\n", + "| Write tests | Agent |\n", + "| Verify output | User |\n", + "\n", + ]; + let streamed = collect_plan_streamed_lines(&deltas, Some(80)); + + let source: String = deltas.iter().copied().collect(); + let baseline = collect_plan_streamed_lines(&[source.as_str()], Some(80)); + + assert_eq!(streamed, baseline); + assert!( + streamed.iter().any(|line| line.contains('┌')), + "expected unicode table border in plan streamed output: {streamed:?}" + ); + assert!( + !streamed + .iter() + .any(|line| line.trim() == "| Step | Owner |"), + "did not expect raw table header line in plan output: {streamed:?}" + ); + } + + #[tokio::test] + async fn plan_controller_streamed_markdown_fenced_table_matches_finalize_render() { + let deltas = vec![ + "## Build plan\n\n", + "```md\n", + "| Step | Owner |\n", + "|---|---|\n", + "| Write tests | Agent |\n", + "| Verify output | User |\n", + "```\n", + "\n", + ]; + let streamed = collect_plan_streamed_lines(&deltas, Some(80)); + + let source: String = deltas.iter().copied().collect(); + let baseline = collect_plan_streamed_lines(&[source.as_str()], Some(80)); + + assert_eq!(streamed, baseline); + assert!( + streamed.iter().any(|line| line.contains('┌')), + "expected unicode table border in fenced plan output: {streamed:?}" + ); + } + + #[test] + fn table_holdback_state_detects_header_plus_delimiter() { + let source = "| Key | Description |\n| --- | --- |\n"; + assert!(matches!( + table_holdback_state(source), + TableHoldbackState::Confirmed + )); + } + + // ----------------------------------------------------------------------- + // source_bytes_for_rendered_count — prefix-stability tests + // ----------------------------------------------------------------------- + + #[test] + fn source_bytes_plain_paragraphs() { + let src = "Alpha\n\nBravo\n\nCharlie\n"; + let width = Some(80); + let cwd = test_cwd(); + + let mut full = Vec::new(); + crate::markdown::append_markdown_agent(src, width, &mut full); + let total = full.len(); + assert!( + total >= 3, + "expected at least 3 rendered lines, got {total}" + ); + + assert_eq!(source_bytes_for_rendered_count(src, width, &cwd, 0), 0); + assert_eq!( + source_bytes_for_rendered_count(src, width, &cwd, total), + src.len() + ); + + let off = source_bytes_for_rendered_count(src, width, &cwd, 1); + assert!(off > 0 && off <= src.len()); + let mut partial = Vec::new(); + crate::markdown::append_markdown_agent(&src[..off], width, &mut partial); + assert!( + partial.len() <= 1, + "expected <=1 line, got {}", + partial.len() + ); + } + + #[test] + fn source_bytes_wrapped_lines() { + let src = "The quick brown fox jumps over the lazy dog near the riverbank.\n"; + let width = Some(20); + let cwd = test_cwd(); + + let mut full = Vec::new(); + crate::markdown::append_markdown_agent(src, width, &mut full); + let total = full.len(); + assert!( + total > 1, + "expected wrapping at width 20, got {total} lines" + ); + + let off = source_bytes_for_rendered_count(src, width, &cwd, 1); + assert_eq!(off, 0, "single wrapped source line should not be split"); + } + + #[test] + fn source_bytes_list_items() { + let src = "- First item\n- Second item\n- Third item\n"; + let width = Some(80); + let cwd = test_cwd(); + + let mut full = Vec::new(); + crate::markdown::append_markdown_agent(src, width, &mut full); + let total = full.len(); + assert!( + total >= 3, + "expected at least 3 rendered lines, got {total}" + ); + + let off = source_bytes_for_rendered_count(src, width, &cwd, 2); + assert!(off > 0); + let mut partial = Vec::new(); + crate::markdown::append_markdown_agent(&src[..off], width, &mut partial); + assert!( + partial.len() <= 2, + "expected <=2 lines from prefix, got {}", + partial.len() + ); + } } diff --git a/codex-rs/tui/src/streaming/mod.rs b/codex-rs/tui/src/streaming/mod.rs index ae3b68742a..ddbac2e4c5 100644 --- a/codex-rs/tui/src/streaming/mod.rs +++ b/codex-rs/tui/src/streaming/mod.rs @@ -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> { - 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 {