test(tui): harden table scrollback invariants

Add debug guardrails around stream stable/tail accounting and prevent live stream tail cells from being committed through generic history paths.

Add regressions for confirmed table holdback, resize with a mutable table tail, and finalizing live table tails exactly once.

codex-rs/Cargo.toml

@@ -312,7 +312,10 @@ starlark = "0.13.0"
 strum = "0.27.2"
 strum_macros = "0.28.0"
 supports-color = "3.0.2"
-syntect = "5"
+# Keep syntect pinned while we carry explicit deny.toml exceptions for
+# RUSTSEC-2024-0320 / RUSTSEC-2025-0141; re-evaluate replacement/upgrade
+# once transitive dependencies are maintained.
+syntect = "=5.3.0"
 sys-locale = "0.3.2"
 tempfile = "3.23.0"
 test-log = "0.2.19"

codex-rs/core/config.schema.json

@@ -482,6 +482,9 @@
             "steer": {
               "type": "boolean"
             },
+            "stream_table_live_tail_reflow": {
+              "type": "boolean"
+            },
             "tool_call_mcp_elicitation": {
               "type": "boolean"
             },
@@ -2189,6 +2192,9 @@
         "steer": {
           "type": "boolean"
         },
+        "stream_table_live_tail_reflow": {
+          "type": "boolean"
+        },
         "tool_call_mcp_elicitation": {
           "type": "boolean"
         },

codex-rs/core/src/agent/builtins/awaiter.toml

@@ -19,7 +19,7 @@ Behavior rules:
    - If the task is still running, continue polling using tool calls.
    - Use repeated tool calls if necessary.
    - Do not hallucinate completion.
-   - Use long timeouts when awaiting for something. If you need multiple awaits, increase the timeouts/yield times exponentially.
+   - Use long timeouts when awaiting something. If you need multiple awaits, increase the timeouts/yield times exponentially.
 
 4. If asked for status:
    - Return the current known status.

codex-rs/core/src/codex.rs

@@ -2164,6 +2164,9 @@ impl Session {
     }
 
     pub(crate) async fn route_realtime_text_input(self: &Arc<Self>, text: String) {
+        if text.trim().is_empty() {
+            return;
+        }
         handlers::user_input_or_turn(
             self,
             self.next_internal_sub_id(),
@@ -7021,12 +7024,71 @@ fn agent_message_text(item: &codex_protocol::items::AgentMessageItem) -> String
 }
 
 fn realtime_text_for_event(msg: &EventMsg) -> Option<String> {
+    const REALTIME_MIRROR_MAX_CHARS: usize = 2_000;
+    const REALTIME_MIRROR_MAX_FILES: usize = 50;
+
     match msg {
         EventMsg::AgentMessage(event) => Some(event.message.clone()),
         EventMsg::ItemCompleted(event) => match &event.item {
             TurnItem::AgentMessage(item) => Some(agent_message_text(item)),
             _ => None,
         },
+        EventMsg::ExecCommandBegin(event) => {
+            let command = event.command.join(" ");
+            Some(format!(
+                "Exec command started: {command}\nWorking directory: {}",
+                event.cwd.display()
+            ))
+        }
+        EventMsg::PatchApplyBegin(event) => {
+            let mut files: Vec<String> = event
+                .changes
+                .keys()
+                .map(|path| path.display().to_string())
+                .collect();
+            files.sort();
+            let total_file_count = files.len();
+            let truncated_file_count = total_file_count.saturating_sub(REALTIME_MIRROR_MAX_FILES);
+            if truncated_file_count > 0 {
+                files.truncate(REALTIME_MIRROR_MAX_FILES);
+            }
+            let file_list = if files.is_empty() {
+                "none".to_string()
+            } else {
+                files.join(", ")
+            };
+            let file_list = if truncated_file_count > 0 {
+                format!("{file_list} ...(truncated, +{truncated_file_count} more)")
+            } else {
+                file_list
+            };
+            Some(format!(
+                "apply_patch started ({total_file_count} file change(s))\nFiles: {file_list}"
+            ))
+        }
+        EventMsg::PatchApplyEnd(event) => {
+            let status = match event.status {
+                codex_protocol::protocol::PatchApplyStatus::Completed => "completed",
+                codex_protocol::protocol::PatchApplyStatus::Failed => "failed",
+                codex_protocol::protocol::PatchApplyStatus::Declined => "declined",
+            };
+            let mut text = format!("apply_patch {status}");
+            if !event.stdout.is_empty() {
+                text.push_str(&format!("\nstdout:\n{}", event.stdout));
+            }
+            if !event.stderr.is_empty() {
+                text.push_str(&format!("\nstderr:\n{}", event.stderr));
+            }
+            if text.len() > REALTIME_MIRROR_MAX_CHARS {
+                let mut end = REALTIME_MIRROR_MAX_CHARS;
+                while !text.is_char_boundary(end) {
+                    end -= 1;
+                }
+                text.truncate(end);
+                text.push_str("\n...(truncated)");
+            }
+            Some(text)
+        }
         EventMsg::Error(_)
         | EventMsg::Warning(_)
         | EventMsg::RealtimeConversationStarted(_)
@@ -7053,12 +7115,9 @@ fn realtime_text_for_event(msg: &EventMsg) -> Option<String> {
         | EventMsg::McpToolCallEnd(_)
         | EventMsg::WebSearchBegin(_)
         | EventMsg::WebSearchEnd(_)
-        | EventMsg::ExecCommandBegin(_)
         | EventMsg::ExecCommandOutputDelta(_)
         | EventMsg::TerminalInteraction(_)
         | EventMsg::ExecCommandEnd(_)
-        | EventMsg::PatchApplyBegin(_)
-        | EventMsg::PatchApplyEnd(_)
         | EventMsg::ViewImageToolCall(_)
         | EventMsg::ImageGenerationBegin(_)
         | EventMsg::ImageGenerationEnd(_)

codex-rs/core/src/config/mod.rs

@@ -1270,6 +1270,7 @@ pub struct ConfigToml {
 
     /// Maximum poll window for background terminal output (`write_stdin`), in milliseconds.
     /// Default: `300000` (5 minutes).
+    #[serde(alias = "background_terminal_timeout")]
     pub background_terminal_max_timeout: Option<u64>,
 
     /// Optional absolute path to the Node runtime used by `js_repl`.

codex-rs/deny.toml

@@ -81,8 +81,8 @@ ignore = [
     { id = "RUSTSEC-2026-0048", reason = "aws-lc-rs/aws-lc-sys are pulled in transitively via rustls stack dependencies; upgrade will be handled separately from this hooks PR" },
     { id = "RUSTSEC-2026-0049", reason = "aws-lc-rs/aws-lc-sys are pulled in transitively via rustls stack dependencies; upgrade will be handled separately from this hooks PR" },
     # TODO(fcoury): remove this exception when syntect drops yaml-rust and bincode, or updates to versions that have fixed the vulnerabilities.
-    { id = "RUSTSEC-2024-0320", reason = "yaml-rust is unmaintained; pulled in via syntect v5.3.0 used by codex-tui for syntax highlighting; no fixed release yet" },
-    { id = "RUSTSEC-2025-0141", reason = "bincode is unmaintained; pulled in via syntect v5.3.0 used by codex-tui for syntax highlighting; no fixed release yet" },
+    { id = "RUSTSEC-2024-0320", reason = "yaml-rust is unmaintained; pulled in via syntect v5.3.0 used by codex-tui for syntax highlighting; no fixed release yet; review by 2026-06-30" },
+    { id = "RUSTSEC-2025-0141", reason = "bincode is unmaintained; pulled in via syntect v5.3.0 used by codex-tui for syntax highlighting; no fixed release yet; review by 2026-06-30" },
 ]
 # If this is true, then cargo deny will use the git executable to fetch advisory database.
 # If this is false, then it uses a built-in git library.

codex-rs/features/src/lib.rs

@@ -180,6 +180,8 @@ pub enum Feature {
     TuiAppServer,
     /// Prevent idle system sleep while a turn is actively running.
     PreventIdleSleep,
+    /// Speculatively re-render table holdback tail with uncommitted source.
+    StreamTableLiveTailReflow,
     /// Legacy rollout flag for Responses API WebSocket transport experiments.
     ResponsesWebsockets,
     /// Legacy rollout flag for Responses API WebSocket transport v2 experiments.
@@ -849,6 +851,12 @@ pub const FEATURES: &[FeatureSpec] = &[
         },
         default_enabled: false,
     },
+    FeatureSpec {
+        id: Feature::StreamTableLiveTailReflow,
+        key: "stream_table_live_tail_reflow",
+        stage: Stage::UnderDevelopment,
+        default_enabled: false,
+    },
     FeatureSpec {
         id: Feature::ResponsesWebsockets,
         key: "responses_websockets",

codex-rs/scripts/tui-resize-stream-repro.sh

@@ -0,0 +1,256 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+usage() {
+  cat <<'EOF'
+Usage: scripts/tui-resize-stream-repro.sh [options]
+
+Start Codex in tmux, send a long table-heavy prompt, resize repeatedly while
+the response streams, and capture scrollback after each resize.
+
+Options:
+  --session NAME      tmux session name (default: codex-resize-repro-$$)
+  --out DIR           output directory (default: /tmp/codex-resize-repro-*)
+  --duration SECONDS  resize duration after prompt submission (default: 20)
+  --sleep SECONDS     delay between resizes (default: 0.20)
+  --startup SECONDS   wait after launching Codex before sending prompt (default: 8)
+  --expect MODE       either, artifact, or clean (default: either)
+  --keep-session      leave the tmux session running for manual inspection
+  --skip-prebuild     do not run cargo build --bin codex before launching tmux
+  --prompt TEXT       override the default prompt
+  -h, --help          show this help
+
+Examples:
+  scripts/tui-resize-stream-repro.sh --keep-session
+  scripts/tui-resize-stream-repro.sh --expect clean --duration 30
+EOF
+}
+
+session="codex-resize-repro-$$"
+out_dir=""
+duration=20
+sleep_interval=0.20
+startup_wait=8
+expect="either"
+keep_session=0
+prebuild=1
+prompt="Produce 6 long Markdown tables, 25 rows each. Include emojis, bold, italic, strikethrough, inline code, markdown links, code-like values, short cells, wrapped cells, and pipe characters escaped inside cells. Include the token RESIZE_REPRO_SENTINEL in several table cells. Stream the answer directly. Do not use tools."
+
+while [[ $# -gt 0 ]]; do
+  case "$1" in
+    --session)
+      session="$2"
+      shift 2
+      ;;
+    --out)
+      out_dir="$2"
+      shift 2
+      ;;
+    --duration)
+      duration="$2"
+      shift 2
+      ;;
+    --sleep)
+      sleep_interval="$2"
+      shift 2
+      ;;
+    --startup)
+      startup_wait="$2"
+      shift 2
+      ;;
+    --expect)
+      expect="$2"
+      shift 2
+      ;;
+    --keep-session)
+      keep_session=1
+      shift
+      ;;
+    --skip-prebuild)
+      prebuild=0
+      shift
+      ;;
+    --prompt)
+      prompt="$2"
+      shift 2
+      ;;
+    -h|--help)
+      usage
+      exit 0
+      ;;
+    *)
+      echo "unknown option: $1" >&2
+      usage >&2
+      exit 2
+      ;;
+  esac
+done
+
+case "$expect" in
+  either|artifact|clean) ;;
+  *)
+    echo "--expect must be one of: either, artifact, clean" >&2
+    exit 2
+    ;;
+esac
+
+if ! command -v tmux >/dev/null 2>&1; then
+  echo "tmux is required for this repro" >&2
+  exit 2
+fi
+
+if [[ "$prebuild" -eq 1 ]]; then
+  cargo build --bin codex
+fi
+
+if [[ -z "$out_dir" ]]; then
+  out_dir="$(mktemp -d "${TMPDIR:-/tmp}/codex-resize-repro-XXXXXX")"
+else
+  mkdir -p "$out_dir"
+fi
+
+captures_dir="$out_dir/captures"
+log_dir="$out_dir/logs"
+mkdir -p "$captures_dir" "$log_dir"
+
+if tmux has-session -t "$session" 2>/dev/null; then
+  echo "tmux session already exists: $session" >&2
+  exit 2
+fi
+
+cleanup() {
+  if [[ "$keep_session" -eq 0 ]] && tmux has-session -t "$session" 2>/dev/null; then
+    tmux kill-session -t "$session"
+  fi
+}
+trap cleanup EXIT
+
+capture() {
+  local idx="$1"
+  local width="$2"
+  local height="$3"
+  local stem
+  stem="$(printf '%03d_%sx%s' "$idx" "$width" "$height")"
+  tmux capture-pane -t "$session" -p -S -5000 >"$captures_dir/$stem.txt"
+  tmux capture-pane -t "$session" -p -e -S -5000 >"$captures_dir/$stem.ansi"
+}
+
+wait_for_tui_ready() {
+  local deadline=$((SECONDS + startup_wait))
+  while [[ "$SECONDS" -lt "$deadline" ]]; do
+    if tmux capture-pane -t "$session" -p | grep -q 'OpenAI Codex'; then
+      return 0
+    fi
+    sleep 0.25
+  done
+  return 1
+}
+
+detect_artifacts() {
+  local report="$out_dir/artifacts.txt"
+  : >"$report"
+
+  for capture in "$captures_dir"/*.txt; do
+    [[ -e "$capture" ]] || continue
+    local raw_count box_count sentinel_count
+    raw_count="$(grep -Ec '^[[:space:]]*\|.*\|' "$capture" || true)"
+    box_count="$(grep -Ec '[┌┬┐└┴┘├┼┤│]' "$capture" || true)"
+    sentinel_count="$(grep -Ec 'RESIZE_REPRO_SENTINEL' "$capture" || true)"
+
+    if [[ "$raw_count" -ge 3 && "$box_count" -ge 3 ]]; then
+      printf '%s mixed_raw_and_boxed_table raw=%s boxed=%s sentinel=%s\n' \
+        "$(basename "$capture")" "$raw_count" "$box_count" "$sentinel_count" >>"$report"
+    fi
+
+    local prompt_count
+    prompt_count="$(grep -Ec '^[[:space:]]*› ' "$capture" || true)"
+    if [[ "$prompt_count" -ge 2 && "$box_count" -ge 3 ]]; then
+      printf '%s duplicated_inline_prompt_with_table prompts=%s boxed=%s sentinel=%s\n' \
+        "$(basename "$capture")" "$prompt_count" "$box_count" "$sentinel_count" >>"$report"
+    fi
+
+    if awk '
+      /^[[:space:]]*\|.*\|[[:space:]]*$/ {
+        pipe_rows++
+        next
+      }
+      pipe_rows >= 3 && /[┌┬┐└┴┘├┼┤│]/ {
+        found = 1
+      }
+      END { exit found ? 0 : 1 }
+    ' "$capture"; then
+      printf '%s raw_table_precedes_boxed_table\n' "$(basename "$capture")" >>"$report"
+    fi
+  done
+
+  [[ -s "$report" ]]
+}
+
+tmux new-session -d -s "$session" -x 100 -y 34
+
+launch_cmd="RUST_LOG=trace ./target/debug/codex --no-alt-screen -C '$PWD' -c 'log_dir=$log_dir'"
+tmux send-keys -t "$session" -l "$launch_cmd"
+tmux send-keys -t "$session" Enter
+if ! wait_for_tui_ready; then
+  capture 0 100 34
+  echo "Codex TUI did not become ready within ${startup_wait}s" >&2
+  echo "captures: $captures_dir" >&2
+  exit 1
+fi
+
+capture 0 100 34
+
+tmux send-keys -t "$session" -l "$prompt"
+sleep 0.2
+tmux send-keys -t "$session" Enter
+
+widths=(110 62 132 74 120 56 140)
+heights=(36 28 38 24 34 22 40)
+idx=1
+deadline=$((SECONDS + duration))
+
+while [[ "$SECONDS" -lt "$deadline" ]]; do
+  for i in "${!widths[@]}"; do
+    width="${widths[$i]}"
+    height="${heights[$i]}"
+    tmux resize-window -t "$session" -x "$width" -y "$height"
+    sleep "$sleep_interval"
+    capture "$idx" "$width" "$height"
+    idx=$((idx + 1))
+    if [[ "$SECONDS" -ge "$deadline" ]]; then
+      break
+    fi
+  done
+done
+
+sleep 2
+capture "$idx" "final" "final"
+
+cat >"$out_dir/metadata.txt" <<EOF
+session=$session
+cwd=$PWD
+log_dir=$log_dir
+duration=$duration
+sleep_interval=$sleep_interval
+startup_wait=$startup_wait
+expect=$expect
+keep_session=$keep_session
+prebuild=$prebuild
+prompt=$prompt
+EOF
+
+if detect_artifacts; then
+  echo "resize-stream repro captured possible artifacts:"
+  cat "$out_dir/artifacts.txt"
+  echo "captures: $captures_dir"
+  if [[ "$expect" == "clean" ]]; then
+    exit 1
+  fi
+  exit 0
+fi
+
+echo "resize-stream repro found no text artifacts"
+echo "captures: $captures_dir"
+if [[ "$expect" == "artifact" ]]; then
+  exit 1
+fi

codex-rs/tui/src/app.rs

@@ -2,6 +2,7 @@ use crate::app_backtrack::BacktrackState;
 use crate::app_command::AppCommand;
 use crate::app_command::AppCommandView;
 use crate::app_event::AppEvent;
+use crate::app_event::ConsolidationScrollbackReflow;
 use crate::app_event::ExitMode;
 use crate::app_event::FeedbackCategory;
 use crate::app_event::RealtimeAudioDeviceKind;
@@ -275,6 +276,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 {
@@ -482,6 +484,27 @@ fn emit_system_bwrap_warning(app_event_tx: &AppEventSender, config: &Config) {
     )));
 }
 
+fn trailing_run_start<T: 'static>(transcript_cells: &[Arc<dyn HistoryCell>]) -> usize {
+    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::<T>()
+    {
+        start -= 1;
+    }
+
+    if start > 0
+        && transcript_cells[start - 1].as_any().is::<T>()
+        && !transcript_cells[start - 1].is_stream_continuation()
+    {
+        start -= 1;
+    }
+
+    start
+}
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 struct SessionSummary {
     usage_line: String,
@@ -956,6 +979,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,
 
@@ -3240,7 +3274,11 @@ impl App {
         self.overlay = None;
         self.transcript_cells.clear();
         self.deferred_history_lines.clear();
+        tui.clear_pending_history_lines();
         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 +3286,216 @@ impl App {
         Ok(())
     }
 
+    fn reset_history_emission_state(&mut self) {
+        self.has_emitted_history_lines = false;
+        self.deferred_history_lines.clear();
+    }
+
+    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) -> bool {
+        let now = Instant::now();
+        let due_now = self
+            .resize_reflow_pending_until
+            .is_some_and(|deadline| now >= deadline);
+        self.resize_reflow_pending_until = Some(now + RESIZE_REFLOW_DEBOUNCE);
+        due_now
+    }
+
+    /// After stream consolidation, schedule a follow-up reflow if one ran mid-stream.
+    fn maybe_finish_stream_reflow(&mut self, tui: &mut tui::Tui) {
+        if self.reflow_ran_during_stream {
+            if self.schedule_resize_reflow() {
+                tui.frame_requester().schedule_frame();
+            } else {
+                tui.frame_requester()
+                    .schedule_frame_in(RESIZE_REFLOW_DEBOUNCE);
+            }
+        } else if self
+            .resize_reflow_pending_until
+            .is_some_and(|deadline| Instant::now() >= deadline)
+        {
+            tui.frame_requester().schedule_frame();
+        }
+        self.reflow_ran_during_stream = false;
+    }
+
+    fn schedule_immediate_resize_reflow(&mut self, tui: &mut tui::Tui) {
+        self.resize_reflow_pending_until = Some(Instant::now());
+        tui.frame_requester().schedule_frame();
+    }
+
+    fn finish_required_stream_reflow(&mut self, tui: &mut tui::Tui) -> Result<()> {
+        self.schedule_immediate_resize_reflow(tui);
+        self.maybe_run_resize_reflow(tui)?;
+        if self.resize_reflow_pending_until.is_none() {
+            self.reflow_ran_during_stream = false;
+        }
+        Ok(())
+    }
+
+    fn handle_draw_size_change(
+        &mut self,
+        size: ratatui::layout::Size,
+        last_known_screen_size: ratatui::layout::Size,
+        frame_requester: &tui::FrameRequester,
+    ) -> bool {
+        let previous_width = self.last_transcript_render_width.replace(size.width);
+        let width_changed = previous_width.is_some_and(|width| width != size.width);
+        if width_changed {
+            self.chat_widget.on_terminal_resize(size.width);
+            if self.schedule_resize_reflow() {
+                frame_requester.schedule_frame();
+            } else {
+                frame_requester.schedule_frame_in(RESIZE_REFLOW_DEBOUNCE);
+            }
+        } else if previous_width.is_none() {
+            self.chat_widget.on_terminal_resize(size.width);
+        }
+        if size != last_known_screen_size {
+            self.refresh_status_line();
+        }
+        self.maybe_clear_resize_reflow_without_terminal();
+        width_changed
+    }
+
+    fn maybe_clear_resize_reflow_without_terminal(&mut self) {
+        let Some(deadline) = self.resize_reflow_pending_until else {
+            return;
+        };
+        if Instant::now() < deadline || self.overlay.is_some() || !self.transcript_cells.is_empty()
+        {
+            return;
+        }
+
+        self.resize_reflow_pending_until = None;
+        self.reset_history_emission_state();
+    }
+
+    fn handle_draw_pre_render(&mut self, tui: &mut tui::Tui) -> Result<()> {
+        let size = tui.terminal.size()?;
+        let width_changed = self.handle_draw_size_change(
+            size,
+            tui.terminal.last_known_screen_size,
+            &tui.frame_requester(),
+        );
+        if width_changed {
+            // Width-sensitive history inserts queued before this frame may be wrapped for the old
+            // viewport. Drop them and let resize reflow rebuild from transcript cells.
+            tui.clear_pending_history_lines();
+        }
+        self.maybe_run_resize_reflow(tui)?;
+        Ok(())
+    }
+
+    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;
+
+        // Track that a reflow happened during an active stream or while trailing
+        // unconsolidated AgentMessageCells are still pending consolidation so
+        // ConsolidateAgentMessage can schedule a follow-up reflow.
+        let reflow_ran_during_stream =
+            !self.transcript_cells.is_empty() && self.should_mark_reflow_as_stream_time();
+
+        self.reflow_transcript_now(tui)?;
+
+        if reflow_ran_during_stream {
+            self.reflow_ran_during_stream = true;
+        }
+
+        Ok(())
+    }
+
+    fn reflow_transcript_now(&mut self, tui: &mut tui::Tui) -> Result<()> {
+        // Drop any queued pre-resize/pre-consolidation inserts before rebuilding from cells.
+        tui.clear_pending_history_lines();
+        if self.transcript_cells.is_empty() {
+            self.reset_history_emission_state();
+            return Ok(());
+        }
+
+        if tui.is_alt_screen_active() {
+            tui.terminal.clear_visible_screen()?;
+        } else {
+            tui.terminal.clear_scrollback_and_visible_screen_ansi()?;
+        }
+
+        self.reset_history_emission_state();
+
+        let width = tui.terminal.size()?.width;
+        // Iterate by index to avoid cloning the Vec and bumping Arc refcounts.
+        for i in 0..self.transcript_cells.len() {
+            let cell = self.transcript_cells[i].clone();
+            self.insert_history_cell_lines(tui, cell.as_ref(), width);
+        }
+
+        Ok(())
+    }
+
+    fn finish_agent_message_consolidation(
+        &mut self,
+        tui: &mut tui::Tui,
+        scrollback_reflow: ConsolidationScrollbackReflow,
+    ) -> Result<()> {
+        match scrollback_reflow {
+            ConsolidationScrollbackReflow::IfResizeReflowRan => {
+                self.maybe_finish_stream_reflow(tui);
+            }
+            ConsolidationScrollbackReflow::Required => {
+                self.finish_required_stream_reflow(tui)?;
+            }
+        }
+
+        Ok(())
+    }
+
+    fn should_mark_reflow_as_stream_time(&self) -> bool {
+        self.chat_widget.has_active_agent_stream()
+            || self.chat_widget.has_active_plan_stream()
+            || trailing_run_start::<history_cell::AgentMessageCell>(&self.transcript_cells)
+                < self.transcript_cells.len()
+            || trailing_run_start::<history_cell::ProposedPlanStreamCell>(&self.transcript_cells)
+                < self.transcript_cells.len()
+    }
+
     fn reset_thread_event_state(&mut self) {
         self.abort_all_thread_event_listeners();
         self.thread_event_channels.clear();
@@ -3769,6 +4017,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(),
@@ -3949,10 +4200,7 @@ impl App {
         event: TuiEvent,
     ) -> Result<AppRunControl> {
         if matches!(event, TuiEvent::Draw) {
-            let size = tui.terminal.size()?;
-            if size != tui.terminal.last_known_screen_size {
-                self.refresh_status_line();
-            }
+            self.handle_draw_pre_render(tui)?;
         }
 
         if self.overlay.is_some() {
@@ -4212,23 +4460,89 @@ 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,
+                scrollback_reflow,
+                deferred_history_cell,
+            } => {
+                if let Some(cell) = deferred_history_cell {
+                    let cell: Arc<dyn HistoryCell> = cell.into();
+                    if let Some(Overlay::Transcript(t)) = &mut self.overlay {
+                        t.insert_cell(cell.clone());
                     }
-                    if self.overlay.is_some() {
-                        self.deferred_history_lines.extend(display);
-                    } else {
-                        tui.insert_history_lines(display);
+                    self.transcript_cells.push(cell);
+                }
+
+                // Walk backward to find the contiguous run of streaming AgentMessageCells 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 start =
+                    trailing_run_start::<history_cell::AgentMessageCell>(&self.transcript_cells);
+                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();
                     }
+
+                    self.finish_agent_message_consolidation(tui, scrollback_reflow)?;
+                } else {
+                    tracing::debug!(
+                        "ConsolidateAgentMessage: no cells to consolidate(start={start}, end={end})",
+                    );
+                    self.maybe_finish_stream_reflow(tui);
+                }
+            }
+            AppEvent::ConsolidateProposedPlan(source) => {
+                let end = self.transcript_cells.len();
+                let start = trailing_run_start::<history_cell::ProposedPlanStreamCell>(
+                    &self.transcript_cells,
+                );
+                let consolidated: Arc<dyn HistoryCell> =
+                    Arc::new(history_cell::new_proposed_plan(source, &self.config.cwd));
+
+                if start < end {
+                    self.transcript_cells
+                        .splice(start..end, std::iter::once(consolidated.clone()));
+
+                    if let Some(Overlay::Transcript(t)) = &mut self.overlay {
+                        t.consolidate_cells(start..end, consolidated.clone());
+                        tui.frame_requester().schedule_frame();
+                    }
+
+                    self.finish_required_stream_reflow(tui)?;
+                } else {
+                    self.transcript_cells.push(consolidated.clone());
+                    if let Some(Overlay::Transcript(t)) = &mut self.overlay {
+                        t.insert_cell(consolidated.clone());
+                        tui.frame_requester().schedule_frame();
+                    }
+                    self.insert_history_cell_lines(
+                        tui,
+                        consolidated.as_ref(),
+                        tui.terminal.last_known_screen_size.width,
+                    );
+
+                    self.maybe_finish_stream_reflow(tui);
                 }
             }
             AppEvent::ApplyThreadRollback { num_turns } => {
@@ -4257,6 +4571,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);
@@ -4265,7 +4580,9 @@ impl App {
                 return Ok(AppRunControl::Exit(ExitReason::Fatal(message)));
             }
             AppEvent::CodexOp(op) => {
-                self.submit_active_thread_op(app_server, op.into()).await?;
+                let op: AppCommand = op.into();
+                self.chat_widget.prepare_local_op_submission(&op);
+                self.submit_active_thread_op(app_server, op).await?;
             }
             AppEvent::SubmitThreadOp { thread_id, op } => {
                 self.submit_thread_op(app_server, thread_id, op.into())
@@ -6287,6 +6604,7 @@ mod tests {
     use crossterm::event::KeyModifiers;
     use insta::assert_snapshot;
     use pretty_assertions::assert_eq;
+    use ratatui::layout::Size;
     use ratatui::prelude::Line;
     use std::path::PathBuf;
     use std::sync::Arc;
@@ -9054,6 +9372,43 @@ guardian_approval = true
         assert_snapshot!("clear_ui_header_fast_status_gpt54_only", rendered);
     }
 
+    #[tokio::test]
+    async fn resize_reflow_repro_draw_should_drain_pending_without_commit_tick() {
+        let mut app = make_test_app().await;
+        let frame_requester = crate::tui::FrameRequester::test_dummy();
+        let size = Size::new(120, 40);
+
+        app.last_transcript_render_width = Some(100);
+        app.handle_draw_size_change(size, size, &frame_requester);
+
+        app.resize_reflow_pending_until = Some(Instant::now() - Duration::from_millis(1));
+
+        app.handle_draw_size_change(size, size, &frame_requester);
+
+        assert!(
+            app.resize_reflow_pending_until.is_none(),
+            "resize reflow should drain on draw even when commit animation is idle",
+        );
+    }
+
+    #[tokio::test]
+    async fn resize_reflow_repro_marks_stream_time_before_consolidation() {
+        let mut app = make_test_app().await;
+        app.transcript_cells.push(Arc::new(AgentMessageCell::new(
+            vec![Line::from("| Key | Value |")],
+            false,
+        )));
+
+        assert!(
+            !app.chat_widget.has_active_agent_stream(),
+            "repro requires stream controller to be cleared before consolidate event",
+        );
+        assert!(
+            app.should_mark_reflow_as_stream_time(),
+            "reflow in the pre-consolidation window should still be treated as stream-time",
+        );
+    }
+
     async fn make_test_app() -> App {
         let (chat_widget, app_event_tx, _rx, _op_rx) = make_chatwidget_manual_with_sender().await;
         let config = chat_widget.config_ref().clone();
@@ -9077,6 +9432,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 +9489,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

@@ -62,6 +62,12 @@ impl RealtimeAudioDeviceKind {
     }
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum ConsolidationScrollbackReflow {
+    IfResizeReflowRan,
+    Required,
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 #[cfg_attr(not(target_os = "windows"), allow(dead_code))]
 pub(crate) enum WindowsSandboxEnableMode {
@@ -266,6 +272,32 @@ 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 stores the raw
+    /// markdown source and re-renders from it on resize.
+    ///
+    /// Emitted by `ChatWidget::flush_answer_stream_with_separator` after stream
+    /// finalization. The `App` handler walks backward through `transcript_cells`
+    /// to find the `AgentMessageCell` run and splices in the consolidated cell.
+    /// The `cwd` keeps local file-link display stable across the final re-render.
+    /// `scrollback_reflow` lets table-tail finalization force the already-emitted
+    /// terminal scrollback to be rebuilt from the consolidated source-backed cell.
+    /// `deferred_history_cell` lets callers add the final stream tail to the
+    /// transcript without first writing its provisional render to scrollback.
+    ConsolidateAgentMessage {
+        source: String,
+        cwd: PathBuf,
+        scrollback_reflow: ConsolidationScrollbackReflow,
+        deferred_history_cell: Option<Box<dyn HistoryCell>>,
+    },
+
+    /// Replace the contiguous run of streaming `ProposedPlanStreamCell`s at the
+    /// end of the transcript with a single source-backed `ProposedPlanCell`.
+    ///
+    /// Emitted by `ChatWidget::on_plan_item_completed` after plan stream
+    /// finalization.
+    ConsolidateProposedPlan(String),
+
     /// Apply rollback semantics to local transcript cells.
     ///
     /// This is emitted when rollback was not initiated by the current

codex-rs/tui/src/bottom_pane/mod.rs

@@ -15,6 +15,7 @@
 //! hint. The pane schedules redraws so those hints can expire even when the UI is otherwise idle.
 use std::path::PathBuf;
 
+use crate::app_event::AppEvent;
 use crate::app_event::ConnectorsSnapshot;
 use crate::app_event_sender::AppEventSender;
 use crate::bottom_pane::pending_input_preview::PendingInputPreview;
@@ -31,6 +32,7 @@ use codex_core::plugins::PluginCapabilitySummary;
 use codex_core::skills::model::SkillMetadata;
 use codex_features::Features;
 use codex_file_search::FileMatch;
+use codex_protocol::protocol::Op;
 use codex_protocol::request_user_input::RequestUserInputEvent;
 use codex_protocol::user_input::TextElement;
 use crossterm::event::KeyCode;
@@ -429,10 +431,13 @@ impl BottomPane {
                 && self.is_task_running
                 && !is_agent_command
                 && !self.composer.popup_active()
-                && let Some(status) = &self.status
             {
-                // Send Op::Interrupt
-                status.interrupt();
+                if let Some(status) = &self.status {
+                    // Send Op::Interrupt
+                    status.interrupt();
+                } else {
+                    self.app_event_tx.send(AppEvent::CodexOp(Op::Interrupt));
+                }
                 self.request_redraw();
                 return InputResult::None;
             }
@@ -1511,6 +1516,36 @@ mod tests {
         assert!(rendered.contains("background terminal running · /ps to view"));
     }
 
+    #[test]
+    fn esc_interrupts_running_task_when_status_hidden() {
+        let (tx_raw, mut rx) = unbounded_channel::<AppEvent>();
+        let tx = AppEventSender::new(tx_raw);
+        let mut pane = BottomPane::new(BottomPaneParams {
+            app_event_tx: tx,
+            frame_requester: FrameRequester::test_dummy(),
+            has_input_focus: true,
+            enhanced_keys_supported: false,
+            placeholder_text: "Ask Codex to do anything".to_string(),
+            disable_paste_burst: false,
+            animations_enabled: true,
+            skills: Some(Vec::new()),
+        });
+
+        pane.set_task_running(true);
+        pane.hide_status_indicator();
+        assert!(
+            !pane.status_indicator_visible(),
+            "status indicator must be hidden for this repro"
+        );
+
+        pane.handle_key_event(KeyEvent::new(KeyCode::Esc, KeyModifiers::NONE));
+
+        assert!(
+            matches!(rx.try_recv(), Ok(AppEvent::CodexOp(Op::Interrupt))),
+            "expected Esc to send Op::Interrupt while status is hidden"
+        );
+    }
+
     #[test]
     fn status_with_details_and_queued_messages_snapshot() {
         let (tx_raw, _rx) = unbounded_channel::<AppEvent>();

codex-rs/tui/src/chatwidget.rs

@@ -523,6 +523,43 @@ impl RateLimitWarningState {
     }
 }
 
+#[cfg(test)]
+fn is_interrupt_droppable_stream_event(msg: &EventMsg) -> bool {
+    match msg {
+        EventMsg::AgentMessage(_)
+        | EventMsg::AgentMessageDelta(_)
+        | EventMsg::PlanDelta(_)
+        | EventMsg::AgentReasoning(_)
+        | EventMsg::AgentReasoningDelta(_)
+        | EventMsg::AgentReasoningRawContent(_)
+        | EventMsg::AgentReasoningRawContentDelta(_)
+        | EventMsg::AgentReasoningSectionBreak(_)
+        | EventMsg::AgentMessageContentDelta(_)
+        | EventMsg::ReasoningContentDelta(_)
+        | EventMsg::ReasoningRawContentDelta(_) => true,
+        EventMsg::ItemCompleted(event) => matches!(
+            &event.item,
+            codex_protocol::items::TurnItem::AgentMessage(_)
+                | codex_protocol::items::TurnItem::Plan(_)
+        ),
+        _ => false,
+    }
+}
+
+fn is_interrupt_droppable_server_notification(notification: &ServerNotification) -> bool {
+    match notification {
+        ServerNotification::AgentMessageDelta(_)
+        | ServerNotification::PlanDelta(_)
+        | ServerNotification::ReasoningSummaryTextDelta(_)
+        | ServerNotification::ReasoningTextDelta(_) => true,
+        ServerNotification::ItemCompleted(notification) => matches!(
+            &notification.item,
+            ThreadItem::AgentMessage { .. } | ThreadItem::Plan { .. }
+        ),
+        _ => false,
+    }
+}
+
 pub(crate) fn get_limits_duration(windows_minutes: i64) -> String {
     const MINUTES_PER_HOUR: i64 = 60;
     const MINUTES_PER_DAY: i64 = 24 * MINUTES_PER_HOUR;
@@ -802,6 +839,11 @@ pub(crate) struct ChatWidget {
     /// bottom pane is treated as "running" while this is populated, even if no agent turn is
     /// currently executing.
     mcp_startup_status: Option<HashMap<String, McpStartupStatus>>,
+    /// True once an interrupt was requested for the active turn.
+    ///
+    /// While this is set, inbound stream deltas are dropped locally so a large
+    /// queued backlog cannot keep rendering stale content after user interrupt.
+    interrupt_requested_for_turn: bool,
     /// Expected MCP servers for the current startup round, seeded from enabled local config.
     mcp_startup_expected_servers: Option<HashSet<String>>,
     /// After startup settles, ignore stale updates until enough notifications confirm a new round.
@@ -1693,23 +1735,60 @@ 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() {
+            let scrollback_reflow = if controller.has_live_tail() {
+                crate::app_event::ConsolidationScrollbackReflow::Required
+            } else {
+                crate::app_event::ConsolidationScrollbackReflow::IfResizeReflowRan
+            };
+            self.clear_active_stream_tail();
+            let (cell, source) = controller.finalize();
+            let deferred_history_cell =
+                if scrollback_reflow == crate::app_event::ConsolidationScrollbackReflow::Required {
+                    cell
+                } else {
+                    if let Some(cell) = cell {
+                        self.add_boxed_history(cell);
+                    }
+                    None
+                };
+            if let Some(cell) = deferred_history_cell.as_ref() {
+                debug_assert!(
+                    cell.as_any()
+                        .downcast_ref::<history_cell::AgentMessageCell>()
+                        .is_some(),
+                    "only agent message stream tails should be deferred for consolidation",
+                );
+            }
+            // 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(),
+                    scrollback_reflow,
+                    deferred_history_cell,
+                });
+            } else if let Some(cell) = deferred_history_cell {
+                self.add_boxed_history(cell);
+            }
         }
         self.adaptive_chunking.reset();
+        if had_stream_controller && self.stream_controllers_idle() {
+            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
                 .as_ref()
-                .map(|controller| controller.queued_lines() == 0)
+                .map(|controller| controller.queued_lines() == 0 && !controller.has_live_tail())
                 .unwrap_or(true)
     }
 
@@ -2177,7 +2256,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,
             ));
         }
@@ -2206,18 +2285,25 @@ impl ChatWidget {
         self.plan_delta_buffer.clear();
         self.plan_item_active = false;
         self.saw_plan_item_this_turn = true;
-        let finalized_streamed_cell =
+        let (finalized_streamed_cell, consolidated_plan_source) =
             if let Some(mut controller) = self.plan_stream_controller.take() {
                 controller.finalize()
             } else {
-                None
+                (None, None)
             };
         if let Some(cell) = finalized_streamed_cell {
             self.add_boxed_history(cell);
             // TODO: Replace streamed output with the final plan item text if plan streaming is
             // removed or if we need to reconcile mismatches between streamed and final content.
+            if let Some(source) = consolidated_plan_source {
+                self.app_event_tx
+                    .send(AppEvent::ConsolidateProposedPlan(source));
+            }
         } else if !plan_text.is_empty() {
             self.add_to_history(history_cell::new_proposed_plan(plan_text, &self.config.cwd));
+        } else if let Some(source) = consolidated_plan_source {
+            self.app_event_tx
+                .send(AppEvent::ConsolidateProposedPlan(source));
         }
         if should_restore_after_stream {
             self.pending_status_indicator_restore = true;
@@ -2275,6 +2361,7 @@ impl ChatWidget {
         self.agent_turn_running = true;
         self.turn_sleep_inhibitor
             .set_turn_running(/*turn_running*/ true);
+        self.interrupt_requested_for_turn = false;
         self.saw_plan_update_this_turn = false;
         self.saw_plan_item_this_turn = false;
         self.last_plan_progress = None;
@@ -2310,10 +2397,15 @@ impl ChatWidget {
         }
         // If a stream is currently active, finalize it.
         self.flush_answer_stream_with_separator();
-        if let Some(mut controller) = self.plan_stream_controller.take()
-            && let Some(cell) = controller.finalize()
-        {
-            self.add_boxed_history(cell);
+        if let Some(mut controller) = self.plan_stream_controller.take() {
+            let (cell, source) = controller.finalize();
+            if let Some(cell) = cell {
+                self.add_boxed_history(cell);
+            }
+            if let Some(source) = source {
+                self.app_event_tx
+                    .send(AppEvent::ConsolidateProposedPlan(source));
+            }
         }
         self.flush_unified_exec_wait_streak();
         if !from_replay {
@@ -2345,6 +2437,7 @@ impl ChatWidget {
         self.agent_turn_running = false;
         self.turn_sleep_inhibitor
             .set_turn_running(/*turn_running*/ false);
+        self.interrupt_requested_for_turn = false;
         self.update_task_running_state();
         self.running_commands.clear();
         self.suppressed_exec_calls.clear();
@@ -2699,12 +2792,16 @@ impl ChatWidget {
     /// This does not clear MCP startup tracking, because MCP startup can overlap with turn cleanup
     /// and should continue to drive the bottom-pane running indicator while it is in progress.
     fn finalize_turn(&mut self) {
+        // Drop preview-only stream tail content on any termination path before
+        // failed-cell finalization, so transient tail cells are never persisted.
+        self.clear_active_stream_tail();
         // Ensure any spinner is replaced by a red ✗ and flushed into history.
         self.finalize_active_cell_as_failed();
         // Reset running state and clear streaming buffers.
         self.agent_turn_running = false;
         self.turn_sleep_inhibitor
             .set_turn_running(/*turn_running*/ false);
+        self.interrupt_requested_for_turn = false;
         self.update_task_running_state();
         self.running_commands.clear();
         self.suppressed_exec_calls.clear();
@@ -4087,6 +4184,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 +4229,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,8 +4253,11 @@ 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,
+                self.config
+                    .features
+                    .enabled(Feature::StreamTableLiveTailReflow),
             ));
         }
         if let Some(controller) = self.stream_controller.as_mut()
@@ -4165,9 +4266,39 @@ 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().and_then(|w| {
+            if w == 0 {
+                None
+            } else {
+                // Keep a 1-column minimum for active stream controllers so they can
+                // continue accepting deltas on ultra-narrow layouts.
+                Some(crate::width::usable_content_width(w, reserved_cols).unwrap_or(1))
+            }
+        })
+    }
+
+    pub(crate) fn on_terminal_resize(&mut self, width: u16) {
+        let had_rendered_width = self.last_rendered_width.get().is_some();
+        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();
+        if !had_rendered_width {
+            self.request_redraw();
+        }
+    }
+
     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,
@@ -4649,6 +4780,7 @@ impl ChatWidget {
             agent_turn_running: false,
             mcp_startup_status: None,
             pending_turn_copyable_output: None,
+            interrupt_requested_for_turn: false,
             mcp_startup_expected_servers: None,
             mcp_startup_ignore_updates_until_next_start: false,
             mcp_startup_allow_terminal_only_next_round: false,
@@ -4879,7 +5011,7 @@ impl ChatWidget {
                         return;
                     };
                     let should_submit_now =
-                        self.is_session_configured() && !self.is_plan_streaming_in_tui();
+                        self.is_session_configured() && !self.has_active_plan_stream();
                     if should_submit_now {
                         // Submitted is emitted when user submits.
                         // Reset any reasoning header only when we are actually submitting a turn.
@@ -5530,7 +5662,21 @@ impl ChatWidget {
     }
 
     fn flush_active_cell(&mut self) {
+        if self.active_cell_is_stream_tail() {
+            if self.stream_controller.is_some() {
+                return;
+            }
+            // If stream cleanup already cleared the controller, drop the transient tail instead
+            // of committing preview-only content to transcript history.
+            self.active_cell.take();
+            return;
+        }
+
         if let Some(active) = self.active_cell.take() {
+            debug_assert!(
+                !active.as_any().is::<history_cell::StreamingAgentTailCell>(),
+                "stream tail active cells are preview-only and must not enter transcript history"
+            );
             self.needs_final_message_separator = true;
             self.app_event_tx.send(AppEvent::InsertHistoryCell(active));
         }
@@ -5541,6 +5687,11 @@ impl ChatWidget {
     }
 
     fn add_boxed_history(&mut self, cell: Box<dyn HistoryCell>) {
+        debug_assert!(
+            !cell.as_any().is::<history_cell::StreamingAgentTailCell>(),
+            "stream tail active cells are preview-only and must not enter transcript history"
+        );
+
         // Keep the placeholder session header as the active cell until real session info arrives,
         // so we can merge headers instead of committing a duplicate box to history.
         let keep_placeholder_header_active = !self.is_session_configured()
@@ -5551,12 +5702,47 @@ 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_is_stream_tail();
+            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_mut().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_is_stream_tail() {
+            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);
@@ -6295,6 +6481,13 @@ impl ChatWidget {
         if !is_resume_initial_replay && !is_retry_error {
             self.restore_retry_status_header_if_present();
         }
+        if !from_replay
+            && self.interrupt_requested_for_turn
+            && is_interrupt_droppable_server_notification(&notification)
+        {
+            tracing::trace!("dropping app-server stream notification while interrupt is pending");
+            return;
+        }
         match notification {
             ServerNotification::ThreadTokenUsageUpdated(notification) => {
                 self.set_token_info(Some(token_usage_info_from_app_server(
@@ -6797,6 +6990,15 @@ impl ChatWidget {
         let from_replay = replay_kind.is_some();
         let is_resume_initial_replay =
             matches!(replay_kind, Some(ReplayKind::ResumeInitialMessages));
+
+        if !from_replay
+            && self.interrupt_requested_for_turn
+            && is_interrupt_droppable_stream_event(&msg)
+        {
+            tracing::trace!("dropping stream event while interrupt is pending");
+            return;
+        }
+
         let is_stream_error = matches!(&msg, EventMsg::StreamError(_));
         if !is_resume_initial_replay && !is_stream_error {
             self.restore_retry_status_header_if_present();
@@ -10295,10 +10497,22 @@ impl ChatWidget {
         self.bottom_pane.is_task_running() || self.is_review_mode
     }
 
-    fn is_plan_streaming_in_tui(&self) -> bool {
+    /// 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 has_active_plan_stream(&self) -> bool {
         self.plan_stream_controller.is_some()
     }
 
+    /// Whether the active cell is a transient streaming tail preview.
+    fn active_cell_is_stream_tail(&self) -> bool {
+        self.active_cell
+            .as_ref()
+            .is_some_and(|cell| cell.as_any().is::<history_cell::StreamingAgentTailCell>())
+    }
+
     pub(crate) fn composer_is_empty(&self) -> bool {
         self.bottom_pane.composer_is_empty()
     }
@@ -10327,7 +10541,7 @@ impl ChatWidget {
             return;
         }
         self.set_collaboration_mask(collaboration_mode);
-        let should_queue = self.is_plan_streaming_in_tui();
+        let should_queue = self.has_active_plan_stream();
         let user_message = UserMessage {
             text,
             local_images: Vec::new(),
@@ -10413,6 +10627,7 @@ impl ChatWidget {
         T: Into<AppCommand>,
     {
         let op: AppCommand = op.into();
+        self.prepare_local_op_submission(&op);
         if op.is_review() && !self.bottom_pane.is_task_running() {
             self.bottom_pane.set_task_running(/*running*/ true);
         }
@@ -10431,6 +10646,22 @@ impl ChatWidget {
         true
     }
 
+    pub(crate) fn prepare_local_op_submission(&mut self, op: &AppCommand) {
+        if matches!(op.view(), crate::app_command::AppCommandView::Interrupt)
+            && self.agent_turn_running
+        {
+            self.interrupt_requested_for_turn = true;
+            if let Some(controller) = self.stream_controller.as_mut() {
+                controller.clear_queue();
+            }
+            if let Some(controller) = self.plan_stream_controller.as_mut() {
+                controller.clear_queue();
+            }
+            self.clear_active_stream_tail();
+            self.request_redraw();
+        }
+    }
+
     #[cfg(test)]
     fn on_list_mcp_tools(&mut self, ev: McpListToolsResponseEvent) {
         self.add_to_history(history_cell::new_mcp_tools_output(

codex-rs/tui/src/chatwidget/tests/composer_submission.rs

@@ -1060,6 +1060,202 @@ async fn interrupt_restores_queued_messages_into_composer() {
     let _ = drain_insert_history(&mut rx);
 }
 
+#[tokio::test]
+async fn interrupt_drops_stream_deltas_until_turn_aborted() {
+    let (mut chat, mut rx, mut op_rx) = make_chatwidget_manual(None).await;
+
+    chat.handle_codex_event(Event {
+        id: "turn-1".into(),
+        msg: EventMsg::TurnStarted(TurnStartedEvent {
+            turn_id: "turn-1".to_string(),
+            model_context_window: None,
+            collaboration_mode_kind: ModeKind::Default,
+        }),
+    });
+
+    chat.handle_key_event(KeyEvent::new(KeyCode::Char('c'), KeyModifiers::CONTROL));
+    let mut saw_interrupt = false;
+    while let Ok(op) = op_rx.try_recv() {
+        if matches!(op, Op::Interrupt) {
+            saw_interrupt = true;
+            break;
+        }
+    }
+    assert!(saw_interrupt, "expected Ctrl+C to submit Op::Interrupt");
+
+    // Simulate stale stream backlog that arrives before TurnAborted.
+    chat.stream_controller = None;
+    chat.handle_codex_event(Event {
+        id: "delta-stale".into(),
+        msg: EventMsg::AgentMessageDelta(AgentMessageDeltaEvent {
+            delta: "stale row\n".to_string(),
+        }),
+    });
+    assert!(
+        chat.stream_controller.is_none(),
+        "expected stale delta to be dropped while interrupt is pending",
+    );
+
+    chat.handle_codex_event(Event {
+        id: "abort-1".into(),
+        msg: EventMsg::TurnAborted(codex_protocol::protocol::TurnAbortedEvent {
+            turn_id: Some("turn-1".to_string()),
+            reason: TurnAbortReason::Interrupted,
+        }),
+    });
+
+    // A subsequent turn should stream normally.
+    chat.handle_codex_event(Event {
+        id: "turn-2".into(),
+        msg: EventMsg::TurnStarted(TurnStartedEvent {
+            turn_id: "turn-2".to_string(),
+            model_context_window: None,
+            collaboration_mode_kind: ModeKind::Default,
+        }),
+    });
+    chat.handle_codex_event(Event {
+        id: "delta-fresh".into(),
+        msg: EventMsg::AgentMessageDelta(AgentMessageDeltaEvent {
+            delta: "fresh row\n".to_string(),
+        }),
+    });
+    assert!(
+        chat.stream_controller.is_some(),
+        "expected new-turn delta to stream after interrupt completion",
+    );
+
+    let _ = drain_insert_history(&mut rx);
+}
+
+#[tokio::test]
+async fn app_event_interrupt_prepares_local_stream_cleanup() {
+    let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(None).await;
+    let cwd = chat.config.cwd.to_path_buf();
+    let mut controller =
+        crate::streaming::controller::StreamController::new(Some(80), cwd.as_path(), false);
+    assert!(controller.push("stale backlog\n"));
+
+    chat.agent_turn_running = true;
+    chat.bottom_pane.set_task_running(/*running*/ true);
+    chat.bottom_pane.hide_status_indicator();
+    chat.stream_controller = Some(controller);
+    chat.active_cell = Some(Box::new(crate::history_cell::StreamingAgentTailCell::new(
+        vec![ratatui::text::Line::from("tail")],
+        true,
+    )));
+
+    chat.handle_key_event(KeyEvent::new(KeyCode::Esc, KeyModifiers::NONE));
+
+    let command = loop {
+        match rx.try_recv() {
+            Ok(AppEvent::CodexOp(op)) => break AppCommand::from(op),
+            Ok(_) => continue,
+            Err(err) => panic!("expected app-level interrupt event, got {err:?}"),
+        }
+    };
+    chat.prepare_local_op_submission(&command);
+
+    assert!(chat.interrupt_requested_for_turn);
+    assert!(
+        chat.active_cell.is_none(),
+        "interrupt cleanup should clear the active stream tail",
+    );
+    assert_eq!(
+        chat.stream_controller
+            .as_ref()
+            .map(crate::streaming::controller::StreamController::queued_lines),
+        Some(0),
+    );
+}
+
+#[tokio::test]
+async fn interrupt_remains_responsive_during_resized_table_stream() {
+    let (mut chat, mut rx, mut op_rx) = make_chatwidget_manual(None).await;
+
+    chat.last_rendered_width.set(Some(120));
+    chat.handle_codex_event(Event {
+        id: "turn-resize".into(),
+        msg: EventMsg::TurnStarted(TurnStartedEvent {
+            turn_id: "turn-resize".to_string(),
+            model_context_window: None,
+            collaboration_mode_kind: ModeKind::Default,
+        }),
+    });
+
+    chat.handle_codex_event(Event {
+        id: "table-head".into(),
+        msg: EventMsg::AgentMessageDelta(AgentMessageDeltaEvent {
+            delta:
+                "| Row | Initiative Summary | Extended Notes | URL |\n| --- | --- | --- | --- |\n"
+                    .to_string(),
+        }),
+    });
+
+    for idx in 0..40 {
+        chat.handle_codex_event(Event {
+            id: format!("table-row-{idx}"),
+            msg: EventMsg::AgentMessageDelta(AgentMessageDeltaEvent {
+                delta: format!(
+                    "| {idx:03} | Workstream {idx:03} provides a long narrative about scope boundaries, sequencing assumptions, contingency paths, stakeholder dependencies, and quality criteria to keep complex coordination readable under pressure. | Record {idx:03} stores extended execution commentary including risk signals, approvals, rollback conditions, evidence links, and checkpoint outcomes so auditors and new contributors can understand context without reopening old threads. | https://example.com/program/workstream-{idx:03}-detailed-governance-and-delivery-context |\n",
+                ),
+            }),
+        });
+        chat.on_terminal_resize(if idx % 2 == 0 { 72 } else { 116 });
+        chat.on_commit_tick();
+        let _ = drain_insert_history(&mut rx);
+    }
+
+    assert!(
+        chat.stream_controller.is_some(),
+        "expected active stream during table tail stress",
+    );
+
+    chat.handle_key_event(KeyEvent::new(KeyCode::Char('c'), KeyModifiers::CONTROL));
+    let mut saw_interrupt = false;
+    while let Ok(op) = op_rx.try_recv() {
+        if matches!(op, Op::Interrupt) {
+            saw_interrupt = true;
+            break;
+        }
+    }
+    assert!(saw_interrupt, "expected Ctrl+C to submit Op::Interrupt");
+
+    chat.on_terminal_resize(64);
+    let resized_tail = {
+        let controller = chat
+            .stream_controller
+            .as_mut()
+            .expect("expected stream controller after resize");
+        lines_to_single_string(&controller.current_tail_lines())
+    };
+
+    chat.handle_codex_event(Event {
+        id: "table-row-stale-after-interrupt".into(),
+        msg: EventMsg::AgentMessageDelta(AgentMessageDeltaEvent {
+            delta: "| 999 | INTERRUPT_STALE_SENTINEL | INTERRUPT_STALE_SENTINEL | https://example.com/stale |\n"
+                .to_string(),
+        }),
+    });
+
+    let tail_after_stale_delta = {
+        let controller = chat
+            .stream_controller
+            .as_mut()
+            .expect("expected stream controller after stale delta");
+        lines_to_single_string(&controller.current_tail_lines())
+    };
+    assert_eq!(
+        tail_after_stale_delta, resized_tail,
+        "expected stale table delta to be dropped while interrupt is pending",
+    );
+    assert!(
+        !tail_after_stale_delta.contains("INTERRUPT_STALE_SENTINEL"),
+        "stale sentinel should never reach the active stream tail",
+    );
+
+    let _ = drain_insert_history(&mut rx);
+}
+
 #[tokio::test]
 async fn interrupt_prepends_queued_messages_before_existing_composer_text() {
     let (mut chat, mut rx, mut op_rx) = make_chatwidget_manual(/*model_override*/ None).await;

codex-rs/tui/src/chatwidget/tests/helpers.rs

@@ -221,6 +221,7 @@ pub(super) async fn make_chatwidget_manual(
         unified_exec_processes: Vec::new(),
         agent_turn_running: false,
         mcp_startup_status: None,
+        interrupt_requested_for_turn: false,
         mcp_startup_expected_servers: None,
         mcp_startup_ignore_updates_until_next_start: false,
         mcp_startup_allow_terminal_only_next_round: false,

codex-rs/tui/src/chatwidget/tests/status_and_layout.rs

@@ -111,6 +111,305 @@ async fn turn_started_uses_runtime_context_window_before_first_token_count() {
         "expected /status to avoid raw config context window, got: {context_line}"
     );
 }
+
+#[tokio::test]
+async fn current_stream_width_clamps_to_minimum_when_reserved_columns_exhaust_width() {
+    let (chat, _rx, _op_rx) = make_chatwidget_manual(None).await;
+
+    chat.last_rendered_width.set(None);
+    assert_eq!(chat.current_stream_width(2), None);
+
+    chat.last_rendered_width.set(Some(2));
+    assert_eq!(chat.current_stream_width(2), Some(1));
+
+    chat.last_rendered_width.set(Some(4));
+    assert_eq!(chat.current_stream_width(4), Some(1));
+
+    chat.last_rendered_width.set(Some(5));
+    assert_eq!(chat.current_stream_width(4), Some(1));
+}
+
+#[tokio::test]
+async fn on_terminal_resize_initial_width_requests_redraw() {
+    let (mut chat, _rx, _op_rx) = make_chatwidget_manual(None).await;
+    let (draw_tx, mut draw_rx) = tokio::sync::broadcast::channel(8);
+    chat.frame_requester = FrameRequester::new(draw_tx);
+    chat.last_rendered_width.set(None);
+
+    chat.on_terminal_resize(120);
+
+    let draw = tokio::time::timeout(std::time::Duration::from_millis(200), draw_rx.recv())
+        .await
+        .expect("timed out waiting for redraw request");
+    assert!(draw.is_ok(), "expected redraw notification to be sent");
+}
+
+#[tokio::test]
+async fn add_to_history_does_not_commit_transient_stream_tail_after_controller_clear() {
+    let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(None).await;
+
+    chat.active_cell = Some(Box::new(crate::history_cell::StreamingAgentTailCell::new(
+        vec![ratatui::text::Line::from("transient table tail preview")],
+        true,
+    )));
+    // Interrupt/error cleanup paths clear the controller before appending history.
+    chat.stream_controller = None;
+
+    chat.add_to_history(crate::history_cell::new_error_event(
+        "stream interrupted".to_string(),
+    ));
+
+    let mut inserted_count = 0usize;
+    let mut saw_transient_tail = false;
+    let mut saw_error = false;
+    while let Ok(event) = rx.try_recv() {
+        if let AppEvent::InsertHistoryCell(cell) = event {
+            inserted_count += 1;
+            if cell
+                .as_any()
+                .is::<crate::history_cell::StreamingAgentTailCell>()
+            {
+                saw_transient_tail = true;
+            }
+            let rendered = lines_to_single_string(&cell.display_lines(80));
+            if rendered.contains("stream interrupted") {
+                saw_error = true;
+            }
+        }
+    }
+
+    assert!(saw_error, "expected error history cell to be emitted");
+    assert!(
+        !saw_transient_tail,
+        "did not expect transient stream-tail cell to be committed",
+    );
+    assert_eq!(
+        inserted_count, 1,
+        "expected only one committed history cell after cleanup"
+    );
+}
+
+#[tokio::test]
+async fn on_error_does_not_persist_transient_stream_tail_during_finalize_turn() {
+    let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(None).await;
+
+    chat.stream_controller = Some(crate::streaming::controller::StreamController::new(
+        Some(80),
+        chat.config.cwd.as_path(),
+        false,
+    ));
+    chat.active_cell = Some(Box::new(crate::history_cell::StreamingAgentTailCell::new(
+        vec![ratatui::text::Line::from("transient stream tail preview")],
+        true,
+    )));
+
+    chat.on_error("stream failed".to_string());
+
+    let mut saw_transient_tail = false;
+    let mut saw_error = false;
+    while let Ok(event) = rx.try_recv() {
+        if let AppEvent::InsertHistoryCell(cell) = event {
+            if cell
+                .as_any()
+                .is::<crate::history_cell::StreamingAgentTailCell>()
+            {
+                saw_transient_tail = true;
+            }
+            let rendered = lines_to_single_string(&cell.display_lines(80));
+            if rendered.contains("stream failed") {
+                saw_error = true;
+            }
+        }
+    }
+
+    assert!(saw_error, "expected error history cell to be emitted");
+    assert!(
+        !saw_transient_tail,
+        "did not expect transient stream-tail cell to be committed during finalize_turn",
+    );
+}
+
+#[tokio::test]
+async fn flush_answer_stream_does_not_stop_animation_while_plan_stream_is_active() {
+    let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(None).await;
+    let cwd = chat.config.cwd.to_path_buf();
+
+    let mut plan_controller =
+        crate::streaming::controller::PlanStreamController::new(Some(80), cwd.as_path());
+    assert!(plan_controller.push("- Step 1\n"));
+    assert!(
+        plan_controller.queued_lines() > 0,
+        "expected plan stream to have queued lines for this repro",
+    );
+    chat.plan_stream_controller = Some(plan_controller);
+    chat.stream_controller = Some(crate::streaming::controller::StreamController::new(
+        Some(80),
+        cwd.as_path(),
+        false,
+    ));
+
+    while rx.try_recv().is_ok() {}
+
+    chat.flush_answer_stream_with_separator();
+
+    let mut saw_stop = false;
+    while let Ok(event) = rx.try_recv() {
+        if matches!(event, AppEvent::StopCommitAnimation) {
+            saw_stop = true;
+        }
+    }
+
+    assert!(
+        !saw_stop,
+        "did not expect StopCommitAnimation while plan stream still has queued lines",
+    );
+}
+
+#[tokio::test]
+async fn flush_answer_stream_requests_scrollback_reflow_for_live_table_tail() {
+    let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(None).await;
+    let cwd = chat.config.cwd.to_path_buf();
+
+    let mut controller =
+        crate::streaming::controller::StreamController::new(Some(80), cwd.as_path(), false);
+    controller.push("| Name | Notes |\n");
+    controller.push("| --- | --- |\n");
+    controller.push("| alpha | FINAL_DUPLICATE_SENTINEL |\n");
+    assert!(
+        controller.has_live_tail(),
+        "expected table holdback to leave a live tail for this regression",
+    );
+    chat.stream_controller = Some(controller);
+
+    while rx.try_recv().is_ok() {}
+
+    chat.flush_answer_stream_with_separator();
+
+    let mut saw_consolidate = false;
+    let mut saw_insert_history = false;
+    while let Ok(event) = rx.try_recv() {
+        match event {
+            AppEvent::InsertHistoryCell(_) => saw_insert_history = true,
+            AppEvent::ConsolidateAgentMessage {
+                scrollback_reflow,
+                deferred_history_cell,
+                ..
+            } => {
+                saw_consolidate = true;
+                assert_eq!(
+                    scrollback_reflow,
+                    crate::app_event::ConsolidationScrollbackReflow::Required
+                );
+                assert!(
+                    deferred_history_cell.is_some(),
+                    "live table tail should be staged for consolidation without provisional insert",
+                );
+            }
+            _ => {}
+        }
+    }
+
+    assert!(
+        saw_consolidate,
+        "expected stream finalization to consolidate"
+    );
+    assert!(
+        !saw_insert_history,
+        "live table tail should not be inserted before canonical reflow"
+    );
+}
+
+#[tokio::test]
+async fn flush_answer_stream_keeps_default_reflow_for_plain_text_tail() {
+    let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(None).await;
+    let cwd = chat.config.cwd.to_path_buf();
+
+    let mut controller =
+        crate::streaming::controller::StreamController::new(Some(80), cwd.as_path(), false);
+    assert!(controller.push("plain response line\n"));
+    assert!(
+        !controller.has_live_tail(),
+        "plain completed text should not force scrollback reflow",
+    );
+    chat.stream_controller = Some(controller);
+
+    while rx.try_recv().is_ok() {}
+
+    chat.flush_answer_stream_with_separator();
+
+    let mut saw_consolidate = false;
+    let mut saw_insert_history = false;
+    while let Ok(event) = rx.try_recv() {
+        match event {
+            AppEvent::InsertHistoryCell(_) => saw_insert_history = true,
+            AppEvent::ConsolidateAgentMessage {
+                scrollback_reflow,
+                deferred_history_cell,
+                ..
+            } => {
+                saw_consolidate = true;
+                assert_eq!(
+                    scrollback_reflow,
+                    crate::app_event::ConsolidationScrollbackReflow::IfResizeReflowRan
+                );
+                assert!(
+                    deferred_history_cell.is_none(),
+                    "plain text should keep the normal provisional insert path",
+                );
+            }
+            _ => {}
+        }
+    }
+
+    assert!(
+        saw_consolidate,
+        "expected stream finalization to consolidate"
+    );
+    assert!(
+        saw_insert_history,
+        "plain text should still insert history before consolidation"
+    );
+}
+
+#[tokio::test]
+async fn flush_answer_stream_does_not_stop_animation_while_plan_table_stream_is_active() {
+    let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(None).await;
+    let cwd = chat.config.cwd.to_path_buf();
+
+    let mut plan_controller =
+        crate::streaming::controller::PlanStreamController::new(Some(80), cwd.as_path());
+    assert!(
+        plan_controller.push("| Step | Owner |\n"),
+        "expected table header to enqueue while plan stream is active",
+    );
+    assert!(
+        plan_controller.queued_lines() > 0,
+        "expected queued table header lines for this repro",
+    );
+    chat.plan_stream_controller = Some(plan_controller);
+    chat.stream_controller = Some(crate::streaming::controller::StreamController::new(
+        Some(80),
+        cwd.as_path(),
+        false,
+    ));
+
+    while rx.try_recv().is_ok() {}
+
+    chat.flush_answer_stream_with_separator();
+
+    let mut saw_stop = false;
+    while let Ok(event) = rx.try_recv() {
+        if matches!(event, AppEvent::StopCommitAnimation) {
+            saw_stop = true;
+        }
+    }
+
+    assert!(
+        !saw_stop,
+        "did not expect StopCommitAnimation while plan table stream is still active",
+    );
+}
+
 #[tokio::test]
 async fn helpers_are_available_and_do_not_panic() {
     let (tx_raw, _rx) = unbounded_channel::<AppEvent>();

codex-rs/tui/src/history_cell.rs

@@ -78,6 +78,7 @@ use ratatui::style::Modifier;
 use ratatui::style::Style;
 use ratatui::style::Styled;
 use ratatui::style::Stylize;
+use ratatui::widgets::Clear;
 use ratatui::widgets::Paragraph;
 use ratatui::widgets::Wrap;
 use std::any::Any;
@@ -189,6 +190,9 @@ impl Renderable for Box<dyn HistoryCell> {
                 .saturating_sub(usize::from(area.height));
             u16::try_from(overflow).unwrap_or(u16::MAX)
         };
+        // Active-cell content can reflow dramatically during resize/stream updates. Clear the
+        // entire draw area first so stale glyphs from previous frames never linger.
+        Clear.render(area, buf);
         paragraph.scroll((y, 0)).render(area, buf);
     }
     fn desired_height(&self, width: u16) -> u16 {
@@ -406,7 +410,7 @@ impl ReasoningSummaryCell {
         let mut lines: Vec<Line<'static>> = Vec::new();
         append_markdown(
             &self.content,
-            Some((width as usize).saturating_sub(2)),
+            crate::width::usable_content_width_u16(width, 2),
             Some(self.cwd.as_path()),
             &mut lines,
         );
@@ -480,6 +484,98 @@ impl HistoryCell for AgentMessageCell {
     }
 }
 
+/// A consolidated agent message cell that stores raw markdown source and
+/// re-renders from it at any width.
+///
+/// After a stream finalizes, the `ConsolidateAgentMessage` handler in `App`
+/// replaces the contiguous run of `AgentMessageCell`s with a single
+/// `AgentMarkdownCell`. On terminal resize, `display_lines(width)` re-renders
+/// from source via `append_markdown_agent`, producing correctly-sized tables
+/// with box-drawing borders.
+///
+/// Uses `prefix_lines` (not `word_wrap_lines`) so table rows with box-drawing
+/// characters pass through without re-wrapping.
+#[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 Some(wrap_width) = crate::width::usable_content_width_u16(width, 2) else {
+            return prefix_lines(vec![Line::default()], "• ".dim(), "  ".into());
+        };
+
+        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.
+        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>> {
+        // Tail lines are already rendered at the controller's current stream width.
+        // Re-wrapping them here can split table borders and produce malformed
+        // in-flight table rows.
+        prefix_lines(
+            self.lines.clone(),
+            if self.is_first_line {
+                "• ".dim()
+            } else {
+                "  ".into()
+            },
+            "  ".into(),
+        )
+    }
+
+    fn is_stream_continuation(&self) -> bool {
+        !self.is_first_line
+    }
+}
+
 #[derive(Debug)]
 pub(crate) struct PlainHistoryCell {
     lines: Vec<Line<'static>>,
@@ -2768,6 +2864,7 @@ mod tests {
     use crate::exec_cell::CommandOutput;
     use crate::exec_cell::ExecCall;
     use crate::exec_cell::ExecCell;
+    use crate::wrapping::word_wrap_lines;
     use codex_config::types::McpServerConfig;
     use codex_config::types::McpServerDisabledReason;
     use codex_core::config::Config;
@@ -2784,6 +2881,8 @@ mod tests {
     use codex_protocol::protocol::SessionConfiguredEvent;
     use dirs::home_dir;
     use pretty_assertions::assert_eq;
+    use ratatui::buffer::Buffer;
+    use ratatui::layout::Rect;
     use serde_json::json;
     use std::collections::HashMap;
     use std::path::PathBuf;
@@ -4679,4 +4778,280 @@ 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_narrow_width_shows_prefix_only() {
+        let source = "| Name | Role |\n|------|------|\n| Alice | Engineer |\n";
+        let cell = AgentMarkdownCell::new(source.to_string(), &test_cwd());
+
+        let lines = render_lines(&cell.display_lines(2));
+        assert_eq!(lines, vec!["• ".to_string()]);
+    }
+
+    #[test]
+    fn wrapped_and_prefixed_cells_handle_tiny_widths() {
+        let user_cell = UserHistoryCell {
+            message: "tiny width coverage for wrapped user history".to_string(),
+            text_elements: Vec::new(),
+            local_image_paths: Vec::new(),
+            remote_image_urls: Vec::new(),
+        };
+        let agent_message_cell = AgentMessageCell::new(vec!["tiny width agent line".into()], true);
+        let reasoning_cell = ReasoningSummaryCell::new(
+            "Plan".to_string(),
+            "Reasoning summary content for tiny widths.".to_string(),
+            &test_cwd(),
+            false,
+        );
+        let agent_markdown_cell =
+            AgentMarkdownCell::new("| A | B |\n|---|---|\n| x | y |\n".to_string(), &test_cwd());
+
+        for width in 1..=4 {
+            assert!(
+                !user_cell.display_lines(width).is_empty(),
+                "user cell should render at width {width}",
+            );
+            assert!(
+                !agent_message_cell.display_lines(width).is_empty(),
+                "agent message cell should render at width {width}",
+            );
+            assert!(
+                !reasoning_cell.display_lines(width).is_empty(),
+                "reasoning cell should render at width {width}",
+            );
+            assert!(
+                !agent_markdown_cell.display_lines(width).is_empty(),
+                "agent markdown cell should render at width {width}",
+            );
+        }
+    }
+
+    #[test]
+    fn render_clears_area_when_cell_content_shrinks() {
+        let area = Rect::new(0, 0, 40, 6);
+        let mut buf = Buffer::empty(area);
+
+        let first: Box<dyn HistoryCell> = Box::new(PlainHistoryCell::new(vec![
+            Line::from("STALE ROW 1"),
+            Line::from("STALE ROW 2"),
+            Line::from("STALE ROW 3"),
+            Line::from("STALE ROW 4"),
+        ]));
+        first.render(area, &mut buf);
+
+        let second: Box<dyn HistoryCell> =
+            Box::new(PlainHistoryCell::new(vec![Line::from("fresh")]));
+        second.render(area, &mut buf);
+
+        let mut rendered_rows: Vec<String> = Vec::new();
+        for y in 0..area.height {
+            let mut row = String::new();
+            for x in 0..area.width {
+                row.push_str(buf.cell((x, y)).expect("cell should exist").symbol());
+            }
+            rendered_rows.push(row);
+        }
+
+        assert!(
+            rendered_rows.iter().all(|row| !row.contains("STALE")),
+            "rendered buffer should not retain stale glyphs: {rendered_rows:?}",
+        );
+        assert!(
+            rendered_rows
+                .first()
+                .is_some_and(|row| row.contains("fresh")),
+            "expected fresh content in first row: {rendered_rows:?}",
+        );
+    }
+
+    #[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:?}"
+        );
+    }
+
+    #[test]
+    fn streaming_agent_tail_cell_does_not_rewrap_table_rows() {
+        let source = "\
+| # | Type | Example | Details | Score |\n\
+| --- | --- | --- | --- | --- |\n\
+| 5 | URL link | Rust (https://www.rust-lang.org) | external | 93 |\n\
+| 6 | GitHub link | Repo (https://github.com/openai) | external | 89 |\n";
+        let mut rendered_lines = Vec::new();
+        crate::markdown::append_markdown_agent(source, Some(120), &mut rendered_lines);
+        let cell = StreamingAgentTailCell::new(rendered_lines, true);
+        let lines = render_lines(&cell.display_lines(72));
+
+        // Ensure wrapped spillover lines were not introduced by a second wrap pass.
+        for line in &lines {
+            let content = line.chars().skip(2).collect::<String>();
+            let trimmed = content.trim();
+            if trimmed.starts_with('│') {
+                assert!(
+                    trimmed.ends_with('│'),
+                    "table row should preserve right border while streaming: {line:?}",
+                );
+            }
+        }
+    }
+
+    /// 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(),
+            remote_image_urls: 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/insert_history.rs

@@ -185,8 +185,15 @@ where
         // fetch/restore the cursor position. insert_history_lines should be cursor-position-neutral :)
         queue!(writer, MoveTo(0, cursor_top))?;
 
+        let scroll_bottom = area.top().saturating_sub(1);
+        let mut advance_row = cursor_top;
         for line in &wrapped {
-            queue!(writer, Print("\r\n"))?;
+            // Explicitly anchor before each line advance to avoid terminal wrap-pending drift when
+            // prior content reached the right edge.
+            queue!(writer, MoveTo(0, advance_row), Print("\n"))?;
+            if advance_row < scroll_bottom {
+                advance_row += 1;
+            }
             write_history_line(writer, line, wrap_width)?;
         }
 
@@ -821,4 +828,58 @@ mod tests {
         assert_eq!(term.viewport_area, Rect::new(0, 5, width, 2));
         assert_eq!(term.visible_history_rows(), 1);
     }
+
+    #[test]
+    fn vt100_exact_width_rows_keep_stable_line_progression() {
+        let width: u16 = 10;
+        let height: u16 = 7;
+        let backend = VT100Backend::new(width, height);
+        let mut term = crate::custom_terminal::Terminal::with_options(backend).expect("terminal");
+        let viewport = Rect::new(0, height - 1, width, 1);
+        term.set_viewport_area(viewport);
+
+        let lines = vec![
+            Line::from("1234567890"),
+            Line::from("abcdefghij"),
+            Line::from("KLMNOPQRST"),
+        ];
+        insert_history_lines(&mut term, lines).expect("insert_history_lines should succeed");
+
+        let screen = term.backend().vt100().screen();
+        let mut non_empty_rows: Vec<(u16, String)> = Vec::new();
+        for row in 0..height.saturating_sub(1) {
+            let row_text = (0..width)
+                .map(|col| {
+                    screen
+                        .cell(row, col)
+                        .map(|cell| cell.contents().to_string())
+                        .unwrap_or_default()
+                })
+                .collect::<String>();
+            if !row_text.trim().is_empty() {
+                non_empty_rows.push((row, row_text));
+            }
+        }
+
+        assert_eq!(
+            non_empty_rows.len(),
+            3,
+            "expected exactly three populated rows, got {non_empty_rows:?}",
+        );
+
+        let expected = ["1234567890", "abcdefghij", "KLMNOPQRST"];
+        for (idx, (row, row_text)) in non_empty_rows.iter().enumerate() {
+            assert_eq!(
+                row_text, expected[idx],
+                "unexpected row content at y={row}: {row_text:?}",
+            );
+        }
+        for pair in non_empty_rows.windows(2) {
+            assert_eq!(
+                pair[1].0,
+                pair[0].0 + 1,
+                "expected contiguous row progression, got {non_empty_rows:?}",
+            );
+        }
+    }
 }

codex-rs/tui/src/lib.rs

@@ -208,9 +208,11 @@ mod voice {
         pub(crate) fn clear(&self) {}
     }
 }
+mod width;
 
 mod wrapping;
 
+mod table_detect;
 #[cfg(test)]
 pub mod test_backend;
 #[cfg(test)]

codex-rs/tui/src/markdown.rs

@@ -1,7 +1,32 @@
+//! 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>>`.  Two variants exist:
+//!
+//! - [`append_markdown`] -- general-purpose, used for plan blocks and history
+//!   cells that already hold pre-processed markdown (no fence unwrapping).
+//! - [`append_markdown_agent`] -- for agent responses.  Runs
+//!   [`unwrap_markdown_fences`] first so that `` ```md ``/`` ```markdown ``
+//!   fences containing tables are stripped and `pulldown-cmark` sees raw
+//!   table syntax instead of fenced code.
+//!
+//! ## Why fence unwrapping exists
+//!
+//! LLM agents frequently wrap tables in `` ```markdown `` fences, treating
+//! them as code.  Without unwrapping, `pulldown-cmark` parses those lines
+//! as a fenced code block and renders them as monospace code rather than a
+//! structured table.  The unwrapper is intentionally conservative: it
+//! buffers the entire fence body before deciding, only unwraps fences whose
+//! info string is `md` or `markdown` AND whose body contains a
+//! header+delimiter pair, and degrades gracefully on unclosed fences.
 use ratatui::text::Line;
+use std::borrow::Cow;
+use std::ops::Range;
 use std::path::Path;
 
-/// Render markdown into `lines` while resolving local file-link display relative to `cwd`.
+use crate::table_detect;
+
+/// Render markdown source to styled ratatui lines and append them to `lines`.
 ///
 /// Callers that already know the session working directory should pass it here so streamed and
 /// non-streamed rendering show the same relative path text even if the process cwd differs.
@@ -19,6 +44,253 @@ 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);
+    let rendered =
+        crate::markdown_render::render_markdown_text_with_width_and_cwd(&normalized, width, cwd);
+    crate::render::line_utils::push_owned_lines(&rendered.lines, 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<'a>(markdown_source: &'a str) -> Cow<'a, str> {
+    // Zero-copy fast path: most messages contain no fences at all.
+    if !markdown_source.contains("```") && !markdown_source.contains("~~~") {
+        return Cow::Borrowed(markdown_source);
+    }
+
+    #[derive(Clone, Copy)]
+    struct Fence {
+        marker: u8,
+        len: usize,
+        is_blockquoted: bool,
+    }
+
+    // Strip a trailing newline and up to 3 leading spaces, returning the
+    // trimmed slice.  Returns `None` when the line has 4+ leading spaces
+    // (which makes it an indented code line per CommonMark).
+    fn strip_line_indent(line: &str) -> Option<&str> {
+        let without_newline = line.strip_suffix('\n').unwrap_or(line);
+        let mut byte_idx = 0usize;
+        let mut column = 0usize;
+        for b in without_newline.as_bytes() {
+            match b {
+                b' ' => {
+                    byte_idx += 1;
+                    column += 1;
+                }
+                b'\t' => {
+                    byte_idx += 1;
+                    column += 4;
+                }
+                _ => break,
+            }
+            if column >= 4 {
+                return None;
+            }
+        }
+        Some(&without_newline[byte_idx..])
+    }
+
+    // Parse an opening fence line, returning the fence metadata and whether
+    // the fence info string indicates markdown content.
+    fn parse_open_fence(line: &str) -> Option<(Fence, bool)> {
+        let trimmed = strip_line_indent(line)?;
+        let is_blockquoted = trimmed.trim_start().starts_with('>');
+        let fence_scan_text = table_detect::strip_blockquote_prefix(trimmed);
+        let (marker, len) = table_detect::parse_fence_marker(fence_scan_text)?;
+        let is_markdown = table_detect::is_markdown_fence_info(fence_scan_text, len);
+        Some((
+            Fence {
+                marker: marker as u8,
+                len,
+                is_blockquoted,
+            },
+            is_markdown,
+        ))
+    }
+
+    fn is_close_fence(line: &str, fence: Fence) -> bool {
+        let Some(trimmed) = strip_line_indent(line) else {
+            return false;
+        };
+        let fence_scan_text = if fence.is_blockquoted {
+            if !trimmed.trim_start().starts_with('>') {
+                return false;
+            }
+            table_detect::strip_blockquote_prefix(trimmed)
+        } else {
+            trimmed
+        };
+        if let Some((marker, len)) = table_detect::parse_fence_marker(fence_scan_text) {
+            marker as u8 == fence.marker
+                && len >= fence.len
+                && fence_scan_text[len..].trim().is_empty()
+        } else {
+            false
+        }
+    }
+
+    fn markdown_fence_contains_table(content: &str, is_blockquoted_fence: bool) -> bool {
+        let mut previous_line: Option<&str> = None;
+        for line in content.lines() {
+            let text = if is_blockquoted_fence {
+                table_detect::strip_blockquote_prefix(line)
+            } else {
+                line
+            };
+            let trimmed = text.trim();
+            if trimmed.is_empty() {
+                previous_line = None;
+                continue;
+            }
+
+            if let Some(previous) = previous_line
+                && table_detect::is_table_header_line(previous)
+                && !table_detect::is_table_delimiter_line(previous)
+                && table_detect::is_table_delimiter_line(trimmed)
+            {
+                return true;
+            }
+
+            previous_line = Some(trimmed);
+        }
+        false
+    }
+
+    fn content_from_ranges(source: &str, ranges: &[Range<usize>]) -> String {
+        let total_len: usize = ranges.iter().map(ExactSizeIterator::len).sum();
+        let mut content = String::with_capacity(total_len);
+        for range in ranges {
+            content.push_str(&source[range.start..range.end]);
+        }
+        content
+    }
+
+    struct MarkdownCandidateData {
+        fence: Fence,
+        opening_range: Range<usize>,
+        content_ranges: Vec<Range<usize>>,
+    }
+
+    // Box the large variant to keep ActiveFence small (~pointer-sized).
+    enum ActiveFence {
+        Passthrough(Fence),
+        MarkdownCandidate(Box<MarkdownCandidateData>),
+    }
+
+    let mut out = String::with_capacity(markdown_source.len());
+    let mut active_fence: Option<ActiveFence> = None;
+    let mut source_offset = 0usize;
+
+    let mut push_source_range = |range: Range<usize>| {
+        if !range.is_empty() {
+            out.push_str(&markdown_source[range]);
+        }
+    };
+
+    for line in markdown_source.split_inclusive('\n') {
+        let line_start = source_offset;
+        source_offset += line.len();
+        let line_range = line_start..source_offset;
+
+        if let Some(active) = active_fence.take() {
+            match active {
+                ActiveFence::Passthrough(fence) => {
+                    push_source_range(line_range);
+                    if !is_close_fence(line, fence) {
+                        active_fence = Some(ActiveFence::Passthrough(fence));
+                    }
+                }
+                ActiveFence::MarkdownCandidate(mut data) => {
+                    if is_close_fence(line, data.fence) {
+                        if markdown_fence_contains_table(
+                            &content_from_ranges(markdown_source, &data.content_ranges),
+                            data.fence.is_blockquoted,
+                        ) {
+                            for range in data.content_ranges {
+                                push_source_range(range);
+                            }
+                        } else {
+                            push_source_range(data.opening_range);
+                            for range in data.content_ranges {
+                                push_source_range(range);
+                            }
+                            push_source_range(line_range);
+                        }
+                    } else {
+                        data.content_ranges.push(line_range);
+                        active_fence = Some(ActiveFence::MarkdownCandidate(data));
+                    }
+                }
+            }
+            continue;
+        }
+
+        if let Some((fence, is_markdown)) = parse_open_fence(line) {
+            if is_markdown {
+                active_fence = Some(ActiveFence::MarkdownCandidate(Box::new(
+                    MarkdownCandidateData {
+                        fence,
+                        opening_range: line_range,
+                        content_ranges: Vec::new(),
+                    },
+                )));
+            } else {
+                push_source_range(line_range);
+                active_fence = Some(ActiveFence::Passthrough(fence));
+            }
+            continue;
+        }
+
+        push_source_range(line_range);
+    }
+
+    if let Some(active) = active_fence {
+        match active {
+            ActiveFence::Passthrough(_) => {}
+            ActiveFence::MarkdownCandidate(data) => {
+                push_source_range(data.opening_range);
+                for range in data.content_ranges {
+                    push_source_range(range);
+                }
+            }
+        }
+    }
+
+    Cow::Owned(out)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -118,4 +390,110 @@ 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_unwraps_markdown_fences_for_single_column_table() {
+        let src = "```md\n| Only |\n|---|\n| value |\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.trim() == "| Only |"));
+    }
+
+    #[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_unwraps_blockquoted_markdown_fence_table() {
+        let src = "> ```markdown\n> | A | B |\n> |---|---|\n> | 1 | 2 |\n> ```\n";
+        let rendered = unwrap_markdown_fences(src);
+        assert!(
+            !rendered.contains("```"),
+            "expected markdown fence markers to be removed: {rendered:?}"
+        );
+    }
+
+    #[test]
+    fn append_markdown_agent_keeps_non_blockquoted_markdown_fence_with_blockquote_table_example() {
+        let src = "```markdown\n> | A | B |\n> |---|---|\n> | 1 | 2 |\n```\n";
+        let normalized = unwrap_markdown_fences(src);
+        assert_eq!(normalized, src);
+    }
+
+    #[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()]);
+    }
+
+    #[test]
+    fn unwrap_markdown_fences_repro_keeps_fence_without_header_delimiter_pair() {
+        let src = "```markdown\n| A | B |\nnot a delimiter row\n| --- | --- |\n# Heading\n```\n";
+        let normalized = unwrap_markdown_fences(src);
+        assert_eq!(normalized, src);
+    }
+
+    #[test]
+    fn append_markdown_agent_keeps_markdown_fence_with_blank_line_between_header_and_delimiter() {
+        let src = "```markdown\n| A | B |\n\n|---|---|\n| 1 | 2 |\n```\n";
+        let rendered = unwrap_markdown_fences(src);
+        assert_eq!(rendered, src);
+    }
 }

codex-rs/tui/src/markdown_render_tests.rs

@@ -1111,23 +1111,6 @@ fn code_block_multiple_lines_inside_unordered_list() {
     assert_eq!(lines, vec!["- Item", "", "  first", "  second"]);
 }
 
-#[test]
-fn code_block_inside_unordered_list_item_multiple_lines() {
-    let md = "- Item\n\n  ```\n  first\n  second\n  ```\n";
-    let text = render_markdown_text(md);
-    let lines: Vec<String> = text
-        .lines
-        .iter()
-        .map(|l| {
-            l.spans
-                .iter()
-                .map(|s| s.content.clone())
-                .collect::<String>()
-        })
-        .collect();
-    assert_eq!(lines, vec!["- Item", "", "  first", "  second"]);
-}
-
 #[test]
 fn markdown_render_complex_snapshot() {
     let md = r#"# H1: Markdown Streaming Test
@@ -1368,3 +1351,503 @@ fn code_block_preserves_trailing_blank_lines() {
         "trailing blank line inside code fence was lost: {content:?}"
     );
 }
+
+#[test]
+fn table_renders_unicode_box() {
+    let md = "| A | B |\n|---|---|\n| 1 | 2 |\n";
+    let text = render_markdown_text(md);
+    let lines: Vec<String> = text
+        .lines
+        .iter()
+        .map(|line| line.spans.iter().map(|span| span.content.clone()).collect())
+        .collect();
+    assert_eq!(
+        lines,
+        vec![
+            "┌─────┬─────┐".to_string(),
+            "│ A   │ B   │".to_string(),
+            "├─────┼─────┤".to_string(),
+            "│ 1   │ 2   │".to_string(),
+            "└─────┴─────┘".to_string(),
+        ]
+    );
+}
+
+#[test]
+fn table_alignment_respects_markers() {
+    let md = "| Left | Center | Right |\n|:-----|:------:|------:|\n| a | b | c |\n";
+    let text = render_markdown_text(md);
+    let lines: Vec<String> = text
+        .lines
+        .iter()
+        .map(|line| line.spans.iter().map(|span| span.content.clone()).collect())
+        .collect();
+    assert_eq!(lines[1], "│ Left │ Center │ Right │");
+    assert_eq!(lines[3], "│ a    │   b    │     c │");
+}
+
+#[test]
+fn table_wraps_cell_content_when_width_is_narrow() {
+    let md = "| Key | Description |\n| --- | --- |\n| -v | Enable very verbose logging output for debugging |\n";
+    let text = crate::markdown_render::render_markdown_text_with_width(md, Some(30));
+    let lines: Vec<String> = text
+        .lines
+        .iter()
+        .map(|line| line.spans.iter().map(|span| span.content.clone()).collect())
+        .collect();
+    assert!(lines[0].starts_with('┌') && lines[0].ends_with('┐'));
+    assert!(
+        lines
+            .iter()
+            .any(|line| line.contains("Enable very verbose"))
+            && lines.iter().any(|line| line.contains("logging output")),
+        "expected wrapped row content: {lines:?}"
+    );
+}
+
+#[test]
+fn table_inside_blockquote_has_quote_prefix() {
+    let md = "> | A | B |\n> |---|---|\n> | 1 | 2 |\n";
+    let text = render_markdown_text(md);
+    let lines: Vec<String> = text
+        .lines
+        .iter()
+        .map(|line| line.spans.iter().map(|span| span.content.clone()).collect())
+        .collect();
+    assert!(lines.iter().all(|line| line.starts_with("> ")));
+    assert!(lines.iter().any(|line| line.contains("┌─────┬─────┐")));
+}
+
+#[test]
+fn escaped_pipes_render_in_table_cells() {
+    let md = "| Col |\n| --- |\n| a \\| b |\n";
+    let text = render_markdown_text(md);
+    let lines: Vec<String> = text
+        .lines
+        .iter()
+        .map(|line| line.spans.iter().map(|span| span.content.clone()).collect())
+        .collect();
+    assert!(lines.iter().any(|line| line.contains("a | b")));
+}
+
+#[test]
+fn table_with_emoji_cells_renders_boxed_table() {
+    let md = "| Task | State |\n|---|---|\n| Unit tests | ✅ |\n| Release notes | 📝 |\n";
+    let text = crate::markdown_render::render_markdown_text_with_width(md, Some(80));
+    let lines: Vec<String> = text
+        .lines
+        .iter()
+        .map(|line| line.spans.iter().map(|span| span.content.clone()).collect())
+        .collect();
+    assert!(
+        lines.iter().any(|line| line.contains('┌')),
+        "expected boxed table for emoji content: {lines:?}"
+    );
+    assert!(
+        !lines.iter().any(|line| line.starts_with("|:---")),
+        "did not expect pipe-delimiter fallback for emoji content: {lines:?}"
+    );
+}
+
+#[test]
+fn table_falls_back_to_pipe_rendering_if_it_cannot_fit() {
+    let md = "| c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 | c9 | c10 |\n|---|---|---|---|---|---|---|---|---|---|\n| 1 | 2 | 3 | 4 | 5 | 6 | 7 | 8 | 9 |
+  10 |\n";
+    let text = crate::markdown_render::render_markdown_text_with_width(md, Some(20));
+    let lines: Vec<String> = text
+        .lines
+        .iter()
+        .map(|line| line.spans.iter().map(|span| span.content.clone()).collect())
+        .collect();
+    assert!(lines.first().is_some_and(|line| line.starts_with('|')));
+    assert!(!lines.iter().any(|line| line.contains('┌')));
+}
+
+#[test]
+fn table_pipe_fallback_rows_wrap_in_narrow_width() {
+    let md = "| c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 | c9 | c10 |\n|---|---|---|---|---|---|---|---|---|---|\n| 111111 | 222222 | 333333 | 444444 | 555555 | 666666 | 777777 | 888888 | 999999 | 101010 |\n";
+    let text = crate::markdown_render::render_markdown_text_with_width(md, Some(20));
+    let lines: Vec<String> = text
+        .lines
+        .iter()
+        .map(|line| line.spans.iter().map(|span| span.content.clone()).collect())
+        .collect();
+
+    assert!(lines.first().is_some_and(|line| line.starts_with('|')));
+    assert!(
+        lines.len() > 3,
+        "expected wrapped pipe-fallback rows at narrow width, got {lines:?}"
+    );
+}
+
+#[test]
+fn table_pipe_fallback_escapes_literal_pipes_in_cell_content() {
+    let md = "| c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 | c9 | c10 |\n|---|---|---|---|---|---|---|---|---|---|\n| keep | keep | keep | keep | keep | keep | keep | keep | a \\| b | keep |\n";
+    let text = crate::markdown_render::render_markdown_text_with_width(md, Some(20));
+    let lines: Vec<String> = text
+        .lines
+        .iter()
+        .map(|line| line.spans.iter().map(|span| span.content.clone()).collect())
+        .collect();
+
+    assert!(lines.first().is_some_and(|line| line.starts_with('|')));
+    assert!(
+        lines.iter().any(|line| line.contains("\\|")),
+        "expected escaped pipe marker to be preserved in wrapped fallback rows: {lines:?}"
+    );
+}
+
+#[test]
+fn table_link_keeps_url_suffix_inside_cell() {
+    let md = "| Site |\n|---|\n| [OpenAI](https://openai.com) |\n";
+    let text = render_markdown_text(md);
+    let lines: Vec<String> = text
+        .lines
+        .iter()
+        .map(|line| line.spans.iter().map(|span| span.content.clone()).collect())
+        .collect();
+
+    assert!(
+        lines
+            .iter()
+            .any(|line| line.contains("OpenAI (https://openai.com)")),
+        "expected link suffix inside table cell: {lines:?}"
+    );
+    assert!(
+        !lines.iter().any(|line| line.trim() == "(https://openai.com)"),
+        "did not expect stray url suffix line outside table: {lines:?}"
+    );
+}
+
+#[test]
+fn table_does_not_absorb_trailing_html_block_label_line() {
+    let md = "| Left | Center | Right |\n|:-----|:------:|------:|\n| a    |   b    |     c |\nInline HTML: <sup>sup</sup> and <sub>sub</sub>.\nHTML block:\n<div style=\"border:1px solid #ccc;padding:2px\">inline block</div>\n";
+    let text = render_markdown_text(md);
+    let lines: Vec<String> = text
+        .lines
+        .iter()
+        .map(|line| line.spans.iter().map(|span| span.content.clone()).collect())
+        .collect();
+
+    assert!(
+        lines
+            .iter()
+            .any(|line| line.trim() == "HTML block:"),
+        "expected 'HTML block:' as plain prose line: {lines:?}"
+    );
+    assert!(
+        !lines.iter().any(|line| line.contains("│ HTML block:")),
+        "did not expect 'HTML block:' inside table grid: {lines:?}"
+    );
+}
+
+#[test]
+fn table_spillover_prose_wraps_in_narrow_width() {
+    let long_label = "This html spillover prose line should wrap on narrow widths to avoid clipping:";
+    let md = format!(
+        "| Left | Center | Right |\n|:-----|:------:|------:|\n| a    |   b    |     c |\n{long_label}\n<div style=\"border:1px solid #ccc;padding:2px\">inline block</div>\n"
+    );
+    let text = crate::markdown_render::render_markdown_text_with_width(&md, Some(40));
+    let lines: Vec<String> = text
+        .lines
+        .iter()
+        .map(|line| line.spans.iter().map(|span| span.content.clone()).collect())
+        .collect();
+
+    assert!(
+        lines.iter().any(|line| line.contains("This html spillover prose")),
+        "expected spillover prose to be present: {lines:?}"
+    );
+    assert!(
+        !lines.iter().any(|line| line.contains(long_label)),
+        "did not expect spillover prose to remain as one long clipped line: {lines:?}"
+    );
+}
+
+#[test]
+fn table_keeps_sparse_comparison_row_inside_grid() {
+    let md = "| A | B | C |\n|---|---|---|\n| x < y > z | | |\n";
+    let text = render_markdown_text(md);
+    let lines: Vec<String> = text
+        .lines
+        .iter()
+        .map(|line| line.spans.iter().map(|span| span.content.clone()).collect())
+        .collect();
+
+    assert!(
+        lines
+            .iter()
+            .any(|line| line.contains("│ x < y > z") && line.ends_with('│')),
+        "expected sparse comparison row to remain inside table grid: {lines:?}"
+    );
+    assert!(
+        !lines.iter().any(|line| line.trim() == "x < y > z"),
+        "did not expect sparse comparison row to spill outside table: {lines:?}"
+    );
+}
+
+#[test]
+fn table_keeps_sparse_rows_with_empty_trailing_cells() {
+    let md = "| A | B | C |\n|---|---|---|\n| a | | |\n";
+    let text = render_markdown_text(md);
+    let lines: Vec<String> = text
+        .lines
+        .iter()
+        .map(|line| line.spans.iter().map(|span| span.content.clone()).collect())
+        .collect();
+
+    assert!(
+        lines
+            .iter()
+            .any(|line| line.contains("│ a") && line.ends_with('│')),
+        "expected sparse row to remain inside table grid: {lines:?}"
+    );
+    assert!(
+        !lines.iter().any(|line| line == "a"),
+        "did not expect sparse row content to spill outside the table: {lines:?}"
+    );
+}
+
+#[test]
+fn table_keeps_single_cell_pipe_row_inside_grid() {
+    let md = "| A | B |\n|---|---|\n| value |\n";
+    let text = render_markdown_text(md);
+    let lines: Vec<String> = text
+        .lines
+        .iter()
+        .map(|line| line.spans.iter().map(|span| span.content.clone()).collect())
+        .collect();
+
+    assert!(
+        lines
+            .iter()
+            .any(|line| line.contains("│ value") && line.ends_with('│')),
+        "expected single-cell pipe row to remain inside table grid: {lines:?}"
+    );
+    assert!(
+        !lines.iter().any(|line| line.trim() == "value"),
+        "did not expect single-cell pipe row to spill outside the table: {lines:?}"
+    );
+}
+
+#[test]
+fn table_keeps_single_cell_row_with_leading_pipe_inside_grid() {
+    let md = "| A | B |\n|---|---|\n| value\n";
+    let text = render_markdown_text(md);
+    let lines: Vec<String> = text
+        .lines
+        .iter()
+        .map(|line| line.spans.iter().map(|span| span.content.clone()).collect())
+        .collect();
+
+    assert!(
+        lines
+            .iter()
+            .any(|line| line.contains("│ value") && line.ends_with('│')),
+        "expected leading-pipe sparse row to remain inside table grid: {lines:?}"
+    );
+    assert!(
+        !lines.iter().any(|line| line.trim() == "value"),
+        "did not expect leading-pipe sparse row to spill outside the table: {lines:?}"
+    );
+}
+
+#[test]
+fn table_normalizes_uneven_row_column_counts() {
+    let md = "| A | B | C |\n|---|---|---|\n| 1 | 2 |\n| 3 | 4 | 5 | 6 |\n";
+    let text = render_markdown_text(md);
+    let lines: Vec<String> = text
+        .lines
+        .iter()
+        .map(|line| line.spans.iter().map(|span| span.content.clone()).collect())
+        .collect();
+
+    assert!(
+        lines.iter().any(|line| line.starts_with('┌'))
+            && lines.iter().any(|line| line.starts_with('└')),
+        "expected normalized uneven rows to remain in boxed table output: {lines:?}"
+    );
+    assert!(
+        lines
+            .iter()
+            .any(|line| line.contains("│ 1") && line.ends_with('│')),
+        "expected shorter row to be padded inside grid: {lines:?}"
+    );
+    assert!(
+        lines
+            .iter()
+            .any(|line| line.contains("│ 3") && line.ends_with('│')),
+        "expected longer row to be truncated to grid width: {lines:?}"
+    );
+}
+
+#[test]
+fn table_keeps_sparse_sentence_row_inside_grid() {
+    let md = "| A | B | C |\n|---|---|---|\n| This is done. | | |\n";
+    let text = render_markdown_text(md);
+    let lines: Vec<String> = text
+        .lines
+        .iter()
+        .map(|line| line.spans.iter().map(|span| span.content.clone()).collect())
+        .collect();
+
+    assert!(
+        lines
+            .iter()
+            .any(|line| line.contains("│ This is done.") && line.ends_with('│')),
+        "expected sparse sentence row to remain inside table grid: {lines:?}"
+    );
+    assert!(
+        !lines.iter().any(|line| line.trim() == "This is done."),
+        "did not expect sparse sentence row to spill outside table: {lines:?}"
+    );
+}
+
+#[test]
+fn table_keeps_label_only_sparse_row_inside_grid() {
+    let md = "| A | B | C |\n|---|---|---|\n| Status: | | |\n| ok | | |\n";
+    let text = render_markdown_text(md);
+    let lines: Vec<String> = text
+        .lines
+        .iter()
+        .map(|line| line.spans.iter().map(|span| span.content.clone()).collect())
+        .collect();
+
+    assert!(
+        lines
+            .iter()
+            .any(|line| line.contains("│ Status:") && line.ends_with('│')),
+        "expected label-only sparse row to remain inside table grid: {lines:?}"
+    );
+    assert!(
+        !lines.iter().any(|line| line.trim() == "Status:"),
+        "did not expect label-only sparse row to spill outside table: {lines:?}"
+    );
+}
+
+#[test]
+fn table_keeps_single_word_label_row_at_end_inside_grid() {
+    let md = "| A | B | C |\n|---|---|---|\n| Status: | | |\n";
+    let text = render_markdown_text(md);
+    let lines: Vec<String> = text
+        .lines
+        .iter()
+        .map(|line| line.spans.iter().map(|span| span.content.clone()).collect())
+        .collect();
+
+    assert!(
+        lines
+            .iter()
+            .any(|line| line.contains("│ Status:") && line.ends_with('│')),
+        "expected single-word trailing label row to remain inside table grid: {lines:?}"
+    );
+    assert!(
+        !lines.iter().any(|line| line.trim() == "Status:"),
+        "did not expect single-word trailing label row to spill outside table: {lines:?}"
+    );
+}
+
+#[test]
+fn table_keeps_multi_word_label_row_at_end_inside_grid() {
+    let md = "| A | B | C |\n|---|---|---|\n| Build status: | | |\n";
+    let text = render_markdown_text(md);
+    let lines: Vec<String> = text
+        .lines
+        .iter()
+        .map(|line| line.spans.iter().map(|span| span.content.clone()).collect())
+        .collect();
+
+    assert!(
+        lines
+            .iter()
+            .any(|line| line.contains("│ Build status:") && line.ends_with('│')),
+        "expected multi-word trailing label row to remain inside table grid: {lines:?}"
+    );
+    assert!(
+        !lines.iter().any(|line| line.trim() == "Build status:"),
+        "did not expect multi-word trailing label row to spill outside table: {lines:?}"
+    );
+}
+
+#[test]
+fn table_preserves_structured_leading_columns_when_last_column_is_long() {
+    let md = "| Milestone | Planned Date | Outcome | Retrospective Summary |\n|---|---|---|---|\n| Canary rollout | 2026-01-10 | Completed | Canary
+  traffic was held at 5% longer than planned due to latency regressions tied to cold cache behavior; after pre-warming and query plan hints, p95
+  returned to baseline and rollout resumed safely. |\n| Full region cutover | 2026-01-24 | Completed | Cutover succeeded with no customer-visible
+  downtime, though internal dashboards lagged for approximately 18 minutes because ingestion workers autoscaled slower than forecast under burst load.
+  |\n";
+    let text = crate::markdown_render::render_markdown_text_with_width(md, Some(160));
+    let lines: Vec<String> = text
+        .lines
+        .iter()
+        .map(|line| line.spans.iter().map(|span| span.content.clone()).collect())
+        .collect();
+
+    assert!(
+        lines.iter().any(|line| line.contains("Milestone")),
+        "expected first structured header to remain readable: {lines:?}"
+    );
+    assert!(
+        lines.iter().any(|line| line.contains("Planned Date")),
+        "expected date header to remain readable: {lines:?}"
+    );
+    assert!(
+        lines.iter().any(|line| line.contains("2026-01-10")),
+        "expected date values to avoid forced mid-token wraps: {lines:?}"
+    );
+}
+
+#[test]
+fn table_preserves_status_column_with_long_notes() {
+    let md = "| Service | Status | Notes |\n|---|---|---|\n| Auth API | Stable | Handles login and token refresh with no major incidents in the last
+  30 days. |\n| Billing Worker | Monitoring | Throughput is good, but we still see occasional retry storms when upstream settlement providers return
+  partial failures. |\n| Search Indexer | Tuning | Performance improved after shard balancing, yet memory usage remains elevated during full rebuild
+  windows. |\n";
+    let text = crate::markdown_render::render_markdown_text_with_width(md, Some(150));
+    let lines: Vec<String> = text
+        .lines
+        .iter()
+        .map(|line| line.spans.iter().map(|span| span.content.clone()).collect())
+        .collect();
+
+    assert!(
+        lines.iter().any(|line| line.contains("Status")),
+        "expected status header to remain readable: {lines:?}"
+    );
+    assert!(
+        lines.iter().any(|line| line.contains("Monitoring")),
+        "expected status values to avoid mid-word wraps: {lines:?}"
+    );
+}
+
+#[test]
+fn table_keeps_long_body_rows_inside_grid_instead_of_spilling_raw_pipe_rows() {
+    let md = "| Milestone | Planned Date | Outcome | Retrospective Summary |\n|---|---|---|---|\n| Canary rollout | 2026-01-10 | Completed | Canary
+  traffic was held at 5% longer than planned due to latency regressions tied to cold cache behavior; after pre-warming and query plan hints, p95
+  returned to baseline and rollout resumed safely. |\n| Full region cutover | 2026-01-24 | Completed | Cutover succeeded with no customer-visible
+  downtime, though internal dashboards lagged for approximately 18 minutes because ingestion workers autoscaled slower than forecast under burst load.
+  |\n| Legacy decommission | 2026-02-07 | In progress | Most workloads have been drained, but final decommission is blocked by one compliance export
+  task that still depends on a deprecated storage path and requires legal sign-off before removal. |\n";
+    let text = crate::markdown_render::render_markdown_text_with_width(md, Some(200));
+    let lines: Vec<String> = text
+        .lines
+        .iter()
+        .map(|line| line.spans.iter().map(|span| span.content.clone()).collect())
+        .collect();
+
+    assert!(
+        lines.iter().any(|line| line.starts_with('┌'))
+            && lines.iter().any(|line| line.starts_with('└')),
+        "expected boxed table output: {lines:?}"
+    );
+    assert!(
+        lines.iter().any(|line| line.contains("│ Canary rollout")),
+        "expected first body row to stay inside table grid: {lines:?}"
+    );
+    assert!(
+        !lines
+            .iter()
+            .any(|line| line.trim_start().starts_with("| Canary rollout |")),
+        "did not expect raw pipe-form body rows outside table: {lines:?}"
+    );
+}

codex-rs/tui/src/markdown_stream.rs

@@ -1,55 +1,132 @@
+//! 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,
 }
 
 impl MarkdownStreamCollector {
-    /// Create a collector that renders markdown using `cwd` for local file-link display.
+    /// Create a collector that accumulates raw markdown deltas.
     ///
-    /// The collector snapshots `cwd` into owned state because stream commits can happen long after
-    /// construction. The same `cwd` should be reused for the entire stream lifecycle; mixing
-    /// different working directories within one stream would make the same link render with
-    /// different path prefixes across incremental commits.
+    /// `width` and `cwd` are only used by test-only rendering helpers; production stream commits
+    /// operate on raw source boundaries. The collector snapshots `cwd` so test rendering keeps
+    /// local file-link display stable 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 clear(&mut self) {
-        self.buffer.clear();
-        self.committed_line_count = 0;
+    /// Update the rendering width used by test-only line-commit helpers.
+    pub fn set_width(&mut self, width: Option<usize>) {
+        self.width = width;
     }
 
+    /// Reset all buffered source and commit bookkeeping.
+    pub fn clear(&mut self) {
+        self.buffer.clear();
+        self.committed_source_len = 0;
+        #[cfg(test)]
+        {
+            self.committed_line_count = 0;
+        }
+    }
+
+    /// Append a raw streaming delta to the internal source buffer.
     pub fn push_delta(&mut self, delta: &str) {
         tracing::trace!("push_delta: {delta:?}");
         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)
+    }
+
+    /// Peek at uncommitted source content beyond the latest commit boundary.
+    pub fn peek_uncommitted(&self) -> &str {
+        &self.buffer[self.committed_source_len..]
+    }
+
+    /// 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.
+    ///
+    /// This helper intentionally uses `append_markdown` (not
+    /// `append_markdown_agent`) so tests can isolate collector newline boundary
+    /// behavior without stream-controller holdback semantics.
+    #[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,25 +145,29 @@ 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
     }
 
     /// Finalize the stream: emit all remaining lines beyond the last commit.
     /// 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.
+    /// for rendering.
+    #[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---");
@@ -156,6 +237,21 @@ mod tests {
         assert_eq!(out.len(), 1);
     }
 
+    #[tokio::test]
+    async fn peek_uncommitted_tracks_buffer_after_commits() {
+        let mut c = super::MarkdownStreamCollector::new(None, &super::test_cwd());
+
+        c.push_delta("alpha");
+        assert_eq!(c.peek_uncommitted(), "alpha");
+
+        c.push_delta("\n");
+        assert_eq!(c.commit_complete_source(), Some("alpha\n".to_string()));
+        assert_eq!(c.peek_uncommitted(), "");
+
+        c.push_delta("beta");
+        assert_eq!(c.peek_uncommitted(), "beta");
+    }
+
     #[tokio::test]
     async fn e2e_stream_blockquote_simple_is_green() {
         let out = super::simulate_stream_markdown_for_tests(&["> Hello\n"], /*finalize*/ true);
@@ -416,6 +512,42 @@ mod tests {
             .collect()
     }
 
+    #[tokio::test]
+    async fn table_header_commits_without_holdback() {
+        let mut c = super::MarkdownStreamCollector::new(None, &super::test_cwd());
+        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, &super::test_cwd());
+        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 +854,45 @@ 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;
+    }
+
+    #[tokio::test]
+    async fn collector_source_chunks_round_trip_into_agent_fence_unwrapping() {
+        let deltas = [
+            "```md\n",
+            "| A | B |\n",
+            "|---|---|\n",
+            "| 1 | 2 |\n",
+            "```\n",
+        ];
+        let mut collector = super::MarkdownStreamCollector::new(None, &super::test_cwd());
+        let mut raw_source = String::new();
+
+        for delta in deltas {
+            collector.push_delta(delta);
+            if delta.contains('\n')
+                && let Some(chunk) = collector.commit_complete_source()
+            {
+                raw_source.push_str(&chunk);
+            }
+        }
+        raw_source.push_str(&collector.finalize_and_drain_source());
+
+        let mut rendered = Vec::new();
+        crate::markdown::append_markdown_agent(&raw_source, None, &mut rendered);
+        let rendered_strs = lines_to_plain_strings(&rendered);
+
+        assert!(
+            rendered_strs.iter().any(|line| line.contains('┌')),
+            "expected markdown-fenced table to render as boxed table: {rendered_strs:?}"
+        );
+        assert!(
+            !rendered_strs.iter().any(|line| line.trim() == "| A | B |"),
+            "did not expect raw table header after markdown-fence unwrapping: {rendered_strs:?}"
+        );
+    }
 }

codex-rs/tui/src/pager_overlay.rs

@@ -566,6 +566,49 @@ impl TranscriptOverlay {
         }
     }
 
+    /// Replace a range of committed cells with a single consolidated cell.
+    ///
+    /// Mirrors the splice performed on `App::transcript_cells` during
+    /// `ConsolidateAgentMessage` so the Ctrl+T overlay stays in sync with the
+    /// main transcript. The range is clamped defensively: cells may have been
+    /// inserted after the overlay opened, leaving it with fewer entries than
+    /// 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 {
+            let removed = clamped_end - clamped_start;
+            if let Some(highlight_cell) = self.highlight_cell.as_mut()
+                && *highlight_cell >= clamped_start
+            {
+                if *highlight_cell < clamped_end {
+                    *highlight_cell = clamped_start;
+                } else {
+                    *highlight_cell = highlight_cell.saturating_sub(removed.saturating_sub(1));
+                }
+            }
+            self.cells
+                .splice(clamped_start..clamped_end, std::iter::once(consolidated));
+            if self
+                .highlight_cell
+                .is_some_and(|highlight_cell| highlight_cell >= self.cells.len())
+            {
+                self.highlight_cell = None;
+            }
+            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
@@ -1090,6 +1133,60 @@ mod tests {
         assert_eq!(overlay.view.scroll_offset, 0);
     }
 
+    #[test]
+    fn transcript_overlay_consolidation_remaps_highlight_inside_range() {
+        let mut overlay = TranscriptOverlay::new(
+            (0..6)
+                .map(|i| {
+                    Arc::new(TestCell {
+                        lines: vec![Line::from(format!("line{i}"))],
+                    }) as Arc<dyn HistoryCell>
+                })
+                .collect(),
+        );
+        overlay.set_highlight_cell(Some(3));
+
+        overlay.consolidate_cells(
+            2..5,
+            Arc::new(TestCell {
+                lines: vec![Line::from("consolidated")],
+            }),
+        );
+
+        assert_eq!(
+            overlay.highlight_cell,
+            Some(2),
+            "highlight inside consolidated range should point to replacement cell",
+        );
+    }
+
+    #[test]
+    fn transcript_overlay_consolidation_remaps_highlight_after_range() {
+        let mut overlay = TranscriptOverlay::new(
+            (0..7)
+                .map(|i| {
+                    Arc::new(TestCell {
+                        lines: vec![Line::from(format!("line{i}"))],
+                    }) as Arc<dyn HistoryCell>
+                })
+                .collect(),
+        );
+        overlay.set_highlight_cell(Some(6));
+
+        overlay.consolidate_cells(
+            2..5,
+            Arc::new(TestCell {
+                lines: vec![Line::from("consolidated")],
+            }),
+        );
+
+        assert_eq!(
+            overlay.highlight_cell,
+            Some(4),
+            "highlight after consolidated range should shift left by removed cells",
+        );
+    }
+
     #[test]
     fn static_overlay_snapshot_basic() {
         // Prepare a static overlay with a few lines and a title

codex-rs/tui/src/render/line_utils.rs

@@ -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/snapshots/codex_tui__markdown_render__markdown_render_tests__markdown_render_complex_snapshot.snap

@@ -28,9 +28,12 @@ Image: alt text
 ———
 
 Table below (alignment test):
-| Left | Center | Right |
-|:-----|:------:|------:|
-| a    |   b    |     c |
+
+┌──────┬────────┬───────┐
+│ Left │ Center │ Right │
+├──────┼────────┼───────┤
+│ a    │   b    │     c │
+└──────┴────────┴───────┘
 Inline HTML: <sup>sup</sup> and <sub>sub</sub>.
 HTML block:
 <div style="border:1px solid #ccc;padding:2px">inline block</div>

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<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 {

codex-rs/tui/src/table_detect.rs

@@ -0,0 +1,479 @@
+//! Canonical pipe-table structure detection and fenced-code-block tracking for
+//! raw markdown source.
+//!
+//! Both the streaming controller (`streaming/controller.rs`) and the
+//! markdown-fence unwrapper (`markdown.rs`) need to identify pipe-table
+//! structure and fenced code blocks in raw markdown source.  This module
+//! provides the canonical implementations so fixes only need to happen in one
+//! place.
+//!
+//! ## Concepts
+//!
+//! A GFM pipe table is a sequence of lines where:
+//! - A **header line** contains pipe-separated segments with at least one
+//!   non-empty cell.
+//! - A **delimiter line** immediately follows the header and contains only
+//!   alignment markers (`---`, `:---`, `---:`, `:---:`), each with at least
+//!   three dashes.
+//! - **Body rows** follow the delimiter.
+//!
+//! A **fenced code block** starts with 3+ backticks or tildes and ends with a
+//! matching close marker.  [`FenceTracker`] classifies each line as
+//! [`FenceKind::Outside`], [`FenceKind::Markdown`], or [`FenceKind::Other`]
+//! so callers can skip pipe characters that appear inside non-markdown fences.
+//!
+//! The table functions operate on single lines and do not maintain cross-line
+//! state.  Callers (the streaming controller and fence unwrapper) are
+//! responsible for pairing consecutive lines to confirm a table.
+
+/// Split a pipe-delimited line into trimmed segments.
+///
+/// Returns `None` if the line is empty or has no unescaped separator marker.
+/// Leading/trailing pipes are stripped before splitting.
+pub(crate) fn parse_table_segments(line: &str) -> Option<Vec<&str>> {
+    let trimmed = line.trim();
+    if trimmed.is_empty() {
+        return None;
+    }
+
+    let has_outer_pipe = trimmed.starts_with('|') || trimmed.ends_with('|');
+    let content = trimmed.strip_prefix('|').unwrap_or(trimmed);
+    let content = content.strip_suffix('|').unwrap_or(content);
+    let raw_segments = split_unescaped_pipe(content);
+    if !has_outer_pipe && raw_segments.len() <= 1 {
+        return None;
+    }
+
+    let segments: Vec<&str> = raw_segments.into_iter().map(str::trim).collect();
+    (!segments.is_empty()).then_some(segments)
+}
+
+/// Split `content` on unescaped `|` characters.
+///
+/// A pipe preceded by `\` is treated as literal text, not a column separator.
+/// The backslash remains in the segment (this is structure detection, not
+/// rendering).
+fn split_unescaped_pipe(content: &str) -> Vec<&str> {
+    let mut segments = Vec::with_capacity(8);
+    let mut start = 0;
+    let bytes = content.as_bytes();
+    let mut i = 0;
+    while i < bytes.len() {
+        if bytes[i] == b'\\' {
+            // Skip the escaped character.
+            i += 2;
+        } else if bytes[i] == b'|' {
+            segments.push(&content[start..i]);
+            start = i + 1;
+            i += 1;
+        } else {
+            i += 1;
+        }
+    }
+    segments.push(&content[start..]);
+    segments
+}
+
+// Small table-detection helpers inlined for the streaming hot path — they are
+// called on every source line during incremental holdback scanning.
+
+/// Whether `line` looks like a table header row (has pipe-separated
+/// segments with at least one non-empty cell).
+#[inline]
+pub(crate) fn is_table_header_line(line: &str) -> bool {
+    parse_table_segments(line).is_some_and(|segments| segments.iter().any(|s| !s.is_empty()))
+}
+
+/// Whether a single segment matches the `---`, `:---`, `---:`, or `:---:`
+/// alignment-colon syntax used in markdown table delimiter rows.
+#[inline]
+fn is_table_delimiter_segment(segment: &str) -> bool {
+    let trimmed = segment.trim();
+    if trimmed.is_empty() {
+        return false;
+    }
+    let without_leading = trimmed.strip_prefix(':').unwrap_or(trimmed);
+    let without_ends = without_leading.strip_suffix(':').unwrap_or(without_leading);
+    without_ends.len() >= 3 && without_ends.chars().all(|c| c == '-')
+}
+
+/// Whether `line` is a valid table delimiter row (every segment passes
+/// [`is_table_delimiter_segment`]).
+#[inline]
+pub(crate) fn is_table_delimiter_line(line: &str) -> bool {
+    parse_table_segments(line)
+        .is_some_and(|segments| segments.into_iter().all(is_table_delimiter_segment))
+}
+
+// ---------------------------------------------------------------------------
+// Fenced code block tracking
+// ---------------------------------------------------------------------------
+
+/// 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)]
+pub(crate) enum FenceKind {
+    /// Not inside any fenced code block.
+    Outside,
+    /// Inside a `` ```md `` or `` ```markdown `` fence.
+    Markdown,
+    /// Inside a fence with a non-markdown info string.
+    Other,
+}
+
+/// Incremental tracker for fenced-code-block open/close transitions.
+///
+/// Feed lines one at a time via [`advance`](Self::advance); query the current
+/// context with [`kind`](Self::kind).  The tracker handles leading-whitespace
+/// limits (>3 spaces → not a fence), blockquote prefix stripping, and
+/// backtick/tilde marker matching.
+pub(crate) struct FenceTracker {
+    state: Option<(char, usize, FenceKind)>,
+}
+
+impl FenceTracker {
+    #[inline]
+    pub(crate) fn new() -> Self {
+        Self { state: None }
+    }
+
+    /// Process one raw source line and update fence state.
+    ///
+    /// Lines with >3 leading spaces are ignored (indented code blocks, not
+    /// fences).  Blockquote prefixes (`>`) are stripped before scanning.
+    pub(crate) fn advance(&mut self, raw_line: &str) {
+        let leading_spaces = raw_line
+            .as_bytes()
+            .iter()
+            .take_while(|byte| **byte == b' ')
+            .count();
+        if leading_spaces > 3 {
+            return;
+        }
+
+        let trimmed = &raw_line[leading_spaces..];
+        let fence_scan_text = strip_blockquote_prefix(trimmed);
+        if let Some((marker, len)) = parse_fence_marker(fence_scan_text) {
+            if let Some((open_char, open_len, _)) = self.state {
+                // Close the current fence if the marker matches.
+                if marker == open_char
+                    && len >= open_len
+                    && fence_scan_text[len..].trim().is_empty()
+                {
+                    self.state = None;
+                }
+            } else {
+                // Opening a new fence.
+                let kind = if is_markdown_fence_info(fence_scan_text, len) {
+                    FenceKind::Markdown
+                } else {
+                    FenceKind::Other
+                };
+                self.state = Some((marker, len, kind));
+            }
+        }
+    }
+
+    /// Current fence context for the most-recently-advanced line.
+    #[inline]
+    pub(crate) fn kind(&self) -> FenceKind {
+        self.state.map_or(FenceKind::Outside, |(_, _, k)| k)
+    }
+}
+
+/// Return fence marker character and run length for a potential fence line.
+///
+/// Recognises backtick and tilde fences with a minimum run of 3.
+/// The input should already have leading whitespace and blockquote prefixes
+/// stripped.
+#[inline]
+pub(crate) fn parse_fence_marker(line: &str) -> Option<(char, usize)> {
+    let first = line.as_bytes().first().copied()?;
+    if first != b'`' && first != b'~' {
+        return None;
+    }
+    let len = line.bytes().take_while(|&b| b == first).count();
+    if len < 3 {
+        return None;
+    }
+    Some((first as char, len))
+}
+
+/// Whether the info string after a fence marker indicates markdown content.
+///
+/// Matches `md` and `markdown` (case-insensitive).
+#[inline]
+pub(crate) fn is_markdown_fence_info(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")
+}
+
+/// Peel all leading `>` blockquote markers from a line.
+///
+/// Tables can appear inside blockquotes (`> | A | B |`), so the holdback
+/// scanner must strip these markers before checking for table syntax.
+#[inline]
+pub(crate) 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();
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn parse_table_segments_basic() {
+        assert_eq!(
+            parse_table_segments("| A | B | C |"),
+            Some(vec!["A", "B", "C"])
+        );
+    }
+
+    #[test]
+    fn parse_table_segments_no_outer_pipes() {
+        assert_eq!(parse_table_segments("A | B | C"), Some(vec!["A", "B", "C"]));
+    }
+
+    #[test]
+    fn parse_table_segments_no_leading_pipe() {
+        assert_eq!(
+            parse_table_segments("A | B | C |"),
+            Some(vec!["A", "B", "C"])
+        );
+    }
+
+    #[test]
+    fn parse_table_segments_no_trailing_pipe() {
+        assert_eq!(
+            parse_table_segments("| A | B | C"),
+            Some(vec!["A", "B", "C"])
+        );
+    }
+
+    #[test]
+    fn parse_table_segments_single_segment_is_allowed() {
+        assert_eq!(parse_table_segments("| only |"), Some(vec!["only"]));
+    }
+
+    #[test]
+    fn parse_table_segments_without_pipe_returns_none() {
+        assert_eq!(parse_table_segments("just text"), None);
+    }
+
+    #[test]
+    fn parse_table_segments_empty_returns_none() {
+        assert_eq!(parse_table_segments(""), None);
+        assert_eq!(parse_table_segments("   "), None);
+    }
+
+    #[test]
+    fn parse_table_segments_escaped_pipe() {
+        // Escaped pipe should NOT split — stays inside the segment.
+        assert_eq!(
+            parse_table_segments(r"| A \| B | C |"),
+            Some(vec![r"A \| B", "C"])
+        );
+    }
+
+    #[test]
+    fn is_table_delimiter_segment_valid() {
+        assert!(is_table_delimiter_segment("---"));
+        assert!(is_table_delimiter_segment(":---"));
+        assert!(is_table_delimiter_segment("---:"));
+        assert!(is_table_delimiter_segment(":---:"));
+        assert!(is_table_delimiter_segment(":-------:"));
+    }
+
+    #[test]
+    fn is_table_delimiter_segment_invalid() {
+        assert!(!is_table_delimiter_segment(""));
+        assert!(!is_table_delimiter_segment("--"));
+        assert!(!is_table_delimiter_segment("abc"));
+        assert!(!is_table_delimiter_segment(":--"));
+    }
+
+    #[test]
+    fn is_table_delimiter_line_valid() {
+        assert!(is_table_delimiter_line("| --- | --- |"));
+        assert!(is_table_delimiter_line("|:---:|---:|"));
+        assert!(is_table_delimiter_line("--- | --- | ---"));
+    }
+
+    #[test]
+    fn is_table_delimiter_line_invalid() {
+        assert!(!is_table_delimiter_line("| A | B |"));
+        assert!(!is_table_delimiter_line("| -- | -- |"));
+    }
+
+    #[test]
+    fn is_table_header_line_valid() {
+        assert!(is_table_header_line("| A | B |"));
+        assert!(is_table_header_line("Name | Value"));
+    }
+
+    #[test]
+    fn is_table_header_line_all_empty_segments() {
+        assert!(!is_table_header_line("| | |"));
+    }
+
+    // -----------------------------------------------------------------------
+    // FenceTracker tests
+    // -----------------------------------------------------------------------
+
+    #[test]
+    fn fence_tracker_outside_by_default() {
+        let tracker = FenceTracker::new();
+        assert_eq!(tracker.kind(), FenceKind::Outside);
+    }
+
+    #[test]
+    fn fence_tracker_opens_and_closes_backtick_fence() {
+        let mut tracker = FenceTracker::new();
+        tracker.advance("```rust");
+        assert_eq!(tracker.kind(), FenceKind::Other);
+
+        tracker.advance("let x = 1;");
+        assert_eq!(tracker.kind(), FenceKind::Other);
+
+        tracker.advance("```");
+        assert_eq!(tracker.kind(), FenceKind::Outside);
+    }
+
+    #[test]
+    fn fence_tracker_opens_and_closes_tilde_fence() {
+        let mut tracker = FenceTracker::new();
+        tracker.advance("~~~python");
+        assert_eq!(tracker.kind(), FenceKind::Other);
+        tracker.advance("~~~");
+        assert_eq!(tracker.kind(), FenceKind::Outside);
+    }
+
+    #[test]
+    fn fence_tracker_markdown_fence() {
+        let mut tracker = FenceTracker::new();
+        tracker.advance("```md");
+        assert_eq!(tracker.kind(), FenceKind::Markdown);
+        tracker.advance("| A | B |");
+        assert_eq!(tracker.kind(), FenceKind::Markdown);
+        tracker.advance("```");
+        assert_eq!(tracker.kind(), FenceKind::Outside);
+    }
+
+    #[test]
+    fn fence_tracker_markdown_case_insensitive() {
+        let mut tracker = FenceTracker::new();
+        tracker.advance("```Markdown");
+        assert_eq!(tracker.kind(), FenceKind::Markdown);
+        tracker.advance("```");
+        assert_eq!(tracker.kind(), FenceKind::Outside);
+    }
+
+    #[test]
+    fn fence_tracker_nested_shorter_marker_does_not_close() {
+        let mut tracker = FenceTracker::new();
+        tracker.advance("````sh");
+        assert_eq!(tracker.kind(), FenceKind::Other);
+        // Shorter marker inside should not close.
+        tracker.advance("```");
+        assert_eq!(tracker.kind(), FenceKind::Other);
+        // Matching length closes.
+        tracker.advance("````");
+        assert_eq!(tracker.kind(), FenceKind::Outside);
+    }
+
+    #[test]
+    fn fence_tracker_mismatched_char_does_not_close() {
+        let mut tracker = FenceTracker::new();
+        tracker.advance("```sh");
+        assert_eq!(tracker.kind(), FenceKind::Other);
+        // Tilde marker should not close a backtick fence.
+        tracker.advance("~~~");
+        assert_eq!(tracker.kind(), FenceKind::Other);
+        tracker.advance("```");
+        assert_eq!(tracker.kind(), FenceKind::Outside);
+    }
+
+    #[test]
+    fn fence_tracker_indented_4_spaces_ignored() {
+        let mut tracker = FenceTracker::new();
+        tracker.advance("    ```sh");
+        assert_eq!(tracker.kind(), FenceKind::Outside);
+    }
+
+    #[test]
+    fn fence_tracker_blockquote_prefix_stripped() {
+        let mut tracker = FenceTracker::new();
+        tracker.advance("> ```sh");
+        assert_eq!(tracker.kind(), FenceKind::Other);
+        tracker.advance("> ```");
+        assert_eq!(tracker.kind(), FenceKind::Outside);
+    }
+
+    #[test]
+    fn fence_tracker_close_with_trailing_content_does_not_close() {
+        let mut tracker = FenceTracker::new();
+        tracker.advance("```sh");
+        assert_eq!(tracker.kind(), FenceKind::Other);
+        // Trailing content prevents closing.
+        tracker.advance("``` extra");
+        assert_eq!(tracker.kind(), FenceKind::Other);
+        tracker.advance("```");
+        assert_eq!(tracker.kind(), FenceKind::Outside);
+    }
+
+    // -----------------------------------------------------------------------
+    // Fence helper function tests
+    // -----------------------------------------------------------------------
+
+    #[test]
+    fn parse_fence_marker_backtick() {
+        assert_eq!(parse_fence_marker("```rust"), Some(('`', 3)));
+        assert_eq!(parse_fence_marker("````"), Some(('`', 4)));
+    }
+
+    #[test]
+    fn parse_fence_marker_tilde() {
+        assert_eq!(parse_fence_marker("~~~python"), Some(('~', 3)));
+    }
+
+    #[test]
+    fn parse_fence_marker_too_short() {
+        assert_eq!(parse_fence_marker("``"), None);
+        assert_eq!(parse_fence_marker("~~"), None);
+    }
+
+    #[test]
+    fn parse_fence_marker_not_fence() {
+        assert_eq!(parse_fence_marker("hello"), None);
+        assert_eq!(parse_fence_marker(""), None);
+    }
+
+    #[test]
+    fn is_markdown_fence_info_basic() {
+        assert!(is_markdown_fence_info("```md", 3));
+        assert!(is_markdown_fence_info("```markdown", 3));
+        assert!(is_markdown_fence_info("```MD", 3));
+        assert!(!is_markdown_fence_info("```rust", 3));
+        assert!(!is_markdown_fence_info("```", 3));
+    }
+
+    #[test]
+    fn strip_blockquote_prefix_basic() {
+        assert_eq!(strip_blockquote_prefix("> hello"), "hello");
+        assert_eq!(strip_blockquote_prefix("> > nested"), "nested");
+        assert_eq!(strip_blockquote_prefix("no prefix"), "no prefix");
+    }
+}

codex-rs/tui/src/tui.rs

@@ -449,6 +449,7 @@ impl Tui {
         self.frame_requester().schedule_frame();
     }
 
+    /// Drop any queued history lines that have not yet been flushed to the terminal.
     pub fn clear_pending_history_lines(&mut self) {
         self.pending_history_lines.clear();
     }

codex-rs/tui/src/width.rs

@@ -0,0 +1,54 @@
+//! Width guards for transcript rendering with fixed prefix columns.
+//!
+//! Several rendering paths reserve a fixed number of columns for bullets,
+//! gutters, or labels before laying out content.  When the terminal is very
+//! narrow, those reserved columns can consume the entire width, leaving zero
+//! or negative space for content.
+//!
+//! These helpers centralise the subtraction and enforce a strict-positive
+//! contract: they return `Some(n)` where `n > 0`, or `None` when no usable
+//! content width remains.  Callers treat `None` as "render prefix-only
+//! fallback" rather than attempting wrapped rendering at zero width, which
+//! would produce empty or unstable output.
+
+/// Returns usable content width after reserving fixed columns.
+///
+/// Guarantees a strict positive width (`Some(n)` where `n > 0`) or `None` when
+/// the reserved columns consume the full width.
+///
+/// Treat `None` as "render prefix-only fallback". Coercing it to `0` and still
+/// attempting wrapped rendering often produces empty or unstable output at very
+/// narrow terminal widths.
+pub(crate) fn usable_content_width(total_width: usize, reserved_cols: usize) -> Option<usize> {
+    total_width
+        .checked_sub(reserved_cols)
+        .filter(|remaining| *remaining > 0)
+}
+
+/// `u16` convenience wrapper around [`usable_content_width`].
+///
+/// This keeps width math at callsites that receive terminal dimensions as
+/// `u16` while preserving the same `None` contract for exhausted width.
+pub(crate) fn usable_content_width_u16(total_width: u16, reserved_cols: u16) -> Option<usize> {
+    usable_content_width(usize::from(total_width), usize::from(reserved_cols))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use pretty_assertions::assert_eq;
+
+    #[test]
+    fn usable_content_width_returns_none_when_reserved_exhausts_width() {
+        assert_eq!(usable_content_width(0, 0), None);
+        assert_eq!(usable_content_width(2, 2), None);
+        assert_eq!(usable_content_width(3, 4), None);
+        assert_eq!(usable_content_width(5, 4), Some(1));
+    }
+
+    #[test]
+    fn usable_content_width_u16_matches_usize_variant() {
+        assert_eq!(usable_content_width_u16(2, 2), None);
+        assert_eq!(usable_content_width_u16(5, 4), Some(1));
+    }
+}