codewhale-tui 0.9.3

#![allow(dead_code)]

//! Markdown stream collector for display-clock rendering.
//!
//! This module implements the pattern from codex-rs where:
//! - Provider deltas are ingested without mutating the visible transcript
//! - Commit ticks coalesce pending deltas into one visible update
//! - Final content is emitted when the stream ends

use ratatui::style::{Modifier, Style};
use ratatui::text::{Line, Span};
use std::time::Duration;
use std::time::Instant;
use unicode_width::{UnicodeWidthChar, UnicodeWidthStr};

use crate::palette;

pub mod chunking;
pub mod commit_tick;
pub mod line_buffer;

pub use chunking::{AdaptiveChunkingPolicy, ChunkingMode};
pub use commit_tick::{StreamChunker, run_commit_tick};
pub use line_buffer::LineBuffer;

/// Default cadence for moving queued provider deltas into visible transcript
/// text. This intentionally tracks animation frames rather than upstream SSE
/// cadence, so tiny bursty deltas coalesce into one history/cache mutation.
///
/// ~30 FPS (33ms). Full motion may catch up sooner when backlog crosses
/// [`CATCH_UP_QUEUE_DEPTH`] / [`CATCH_UP_OLDEST_AGE`]; reduced motion never
/// accelerates — it stays on this steady clock (not a slow typewriter).
pub const DEFAULT_STREAM_COMMIT_INTERVAL: Duration = Duration::from_millis(33);

/// Queue-depth threshold that pulls the display clock forward (catch-up).
pub const CATCH_UP_QUEUE_DEPTH: usize = chunking::ENTER_QUEUE_DEPTH_LINES;

/// Oldest-chunk age that pulls the display clock forward (catch-up).
pub const CATCH_UP_OLDEST_AGE: Duration = chunking::ENTER_OLDEST_AGE;

/// Frame-clock gate for stream display commits.
///
/// Provider deltas may arrive in dozens of tiny chunks inside one event-loop
/// drain. This clock lets the TUI ingest those bytes cheaply, then mutate the
/// visible transcript at most once per display beat unless the stream is being
/// finalized or measured backlog demands catch-up.
#[derive(Debug, Clone)]
pub struct StreamDisplayClock {
    interval: Duration,
    pending: bool,
    next_due_at: Option<Instant>,
    last_commit_at: Option<Instant>,
    commit_count: u64,
    catch_up_count: u64,
    /// When true, backlog may pull `next_due_at` forward to `now`.
    allow_catch_up: bool,
}

impl Default for StreamDisplayClock {
    fn default() -> Self {
        Self::new(DEFAULT_STREAM_COMMIT_INTERVAL)
    }
}

impl StreamDisplayClock {
    pub fn new(interval: Duration) -> Self {
        Self {
            interval,
            pending: false,
            next_due_at: None,
            last_commit_at: None,
            commit_count: 0,
            catch_up_count: 0,
            allow_catch_up: true,
        }
    }

    /// Enable/disable catch-up acceleration. Reduced motion keeps the steady
    /// clock and never pulls beats forward.
    pub fn set_allow_catch_up(&mut self, allow: bool) {
        self.allow_catch_up = allow;
    }

    #[must_use]
    pub fn allow_catch_up(&self) -> bool {
        self.allow_catch_up
    }

    /// Note that at least one stream delta is waiting to become visible.
    pub fn note_delta(&mut self, now: Instant) {
        self.note_delta_with_backlog(now, 1, None);
    }

    /// Note a delta together with measured queue pressure.
    ///
    /// Normal motion coalesces onto the steady interval unless backlog crosses
    /// the catch-up thresholds, in which case the next beat is due immediately.
    pub fn note_delta_with_backlog(
        &mut self,
        now: Instant,
        queued: usize,
        oldest_age: Option<Duration>,
    ) {
        self.pending = true;
        let catch_up = self.allow_catch_up
            && (queued >= CATCH_UP_QUEUE_DEPTH
                || oldest_age.is_some_and(|age| age >= CATCH_UP_OLDEST_AGE));

        if catch_up {
            self.next_due_at = Some(now);
            self.catch_up_count = self.catch_up_count.saturating_add(1);
            return;
        }

        if self.next_due_at.is_some() {
            return;
        }
        self.next_due_at = Some(match self.last_commit_at {
            Some(last) => last.checked_add(self.interval).unwrap_or(now).max(now),
            None => now,
        });
    }

    /// Returns the time until the pending commit is due, if any.
    pub fn due_in(&self, now: Instant) -> Option<Duration> {
        let due = self.next_due_at?;
        Some(due.saturating_duration_since(now))
    }

    /// Consume a due commit beat.
    pub fn take_due(&mut self, now: Instant) -> bool {
        if !self.pending {
            self.next_due_at = None;
            return false;
        }
        let Some(due) = self.next_due_at else {
            return false;
        };
        if now < due {
            return false;
        }
        self.pending = false;
        self.next_due_at = None;
        self.last_commit_at = Some(now);
        self.commit_count = self.commit_count.saturating_add(1);
        true
    }

    /// Force a commit beat, used when the stream is being finalized.
    pub fn flush_now(&mut self, now: Instant) -> bool {
        let had_pending = self.pending;
        self.pending = false;
        self.next_due_at = None;
        if had_pending {
            self.last_commit_at = Some(now);
            self.commit_count = self.commit_count.saturating_add(1);
        }
        had_pending
    }

    pub fn reset(&mut self) {
        self.pending = false;
        self.next_due_at = None;
        self.last_commit_at = None;
        self.commit_count = 0;
        self.catch_up_count = 0;
    }

    pub fn commit_count(&self) -> u64 {
        self.commit_count
    }

    pub fn catch_up_count(&self) -> u64 {
        self.catch_up_count
    }
}
/// Collects streaming text and commits complete lines.
#[derive(Debug, Clone)]
pub struct MarkdownStreamCollector {
    /// Buffer for incoming text
    buffer: String,
    /// Number of lines already committed
    committed_line_count: usize,
    /// Terminal width for wrapping
    width: Option<usize>,
    /// Whether the stream is still active
    is_streaming: bool,
    /// Whether this is a thinking block
    is_thinking: bool,
}

impl Default for MarkdownStreamCollector {
    fn default() -> Self {
        // `is_streaming: true` matches `MarkdownStreamCollector::new` so a
        // freshly-default block behaves like a freshly-started stream.
        Self::new(None, false)
    }
}

impl MarkdownStreamCollector {
    /// Create a new collector
    pub fn new(width: Option<usize>, is_thinking: bool) -> Self {
        Self {
            buffer: String::new(),
            committed_line_count: 0,
            width,
            is_streaming: true,
            is_thinking,
        }
    }

    /// Push new content to the buffer
    pub fn push(&mut self, content: &str) {
        self.buffer.push_str(content);
    }

    /// Get the current buffer content (for display during streaming)
    pub fn current_content(&self) -> &str {
        &self.buffer
    }

    /// Check if there are complete lines to commit
    pub fn has_complete_lines(&self) -> bool {
        self.buffer.contains('\n')
    }

    /// Commit complete lines and return them.
    /// Only lines ending with '\n' are committed.
    /// Returns the newly committed lines since last call.
    pub fn commit_complete_lines(&mut self) -> Vec<Line<'static>> {
        let committed = self.commit_complete_text();
        if committed.is_empty() {
            return Vec::new();
        }
        self.render_lines(&committed)
    }

    /// Commit complete text chunks ending in a newline.
    /// Returns the raw text that became visible since the last call.
    pub fn commit_complete_text(&mut self) -> String {
        if self.buffer.is_empty() {
            return String::new();
        }

        // Find the last newline - only process up to there
        let Some(last_newline_idx) = self.buffer.rfind('\n') else {
            return String::new(); // No complete lines yet
        };

        // Extract the complete portion (up to and including last newline)
        let complete_portion = self.buffer[..=last_newline_idx].to_string();

        // Remove the committed portion from the buffer so finalize only emits the remainder
        self.buffer = self.buffer[last_newline_idx + 1..].to_string();
        self.committed_line_count = 0;

        complete_portion
    }

    /// Finalize the stream and return any remaining content.
    /// Call this when the stream ends to emit the final incomplete line.
    pub fn finalize(&mut self) -> Vec<Line<'static>> {
        let remaining = self.finalize_text();
        if remaining.is_empty() {
            return Vec::new();
        }
        self.render_lines(&remaining)
    }

    /// Finalize the stream and return any remaining raw text.
    pub fn finalize_text(&mut self) -> String {
        self.is_streaming = false;

        if self.buffer.is_empty() {
            return String::new();
        }

        let remaining = self.buffer.clone();
        self.buffer.clear();
        self.committed_line_count = 0;
        remaining
    }

    /// Get all rendered lines (for final display after stream ends)
    pub fn all_lines(&self) -> Vec<Line<'static>> {
        self.render_lines(&self.buffer)
    }

    /// Render content into styled lines
    fn render_lines(&self, content: &str) -> Vec<Line<'static>> {
        let width = self.width.unwrap_or(80);
        let style = if self.is_thinking {
            Style::default()
                .fg(palette::STATUS_WARNING)
                .add_modifier(Modifier::DIM | Modifier::ITALIC)
        } else {
            Style::default()
        };

        let mut lines = Vec::new();

        for line in content.lines() {
            // Wrap long lines
            let wrapped = wrap_line(line, width);
            for wrapped_line in wrapped {
                lines.push(Line::from(Span::styled(wrapped_line, style)));
            }
        }

        // Handle trailing newline (add empty line)
        if content.ends_with('\n') {
            lines.push(Line::from(""));
        }

        lines
    }

    /// Check if the stream is still active
    pub fn is_streaming(&self) -> bool {
        self.is_streaming
    }

    /// Get the raw buffer length
    pub fn buffer_len(&self) -> usize {
        self.buffer.len()
    }

    /// Clear the buffer
    pub fn clear(&mut self) {
        self.buffer.clear();
        self.committed_line_count = 0;
    }
}

/// Wrap a single line to fit within the given width
fn wrap_line(line: &str, width: usize) -> Vec<String> {
    if line.is_empty() {
        return vec![String::new()];
    }
    if width == 0 {
        return vec![line.to_string()];
    }

    let mut result = Vec::new();
    let mut current_line = String::new();
    let mut current_width = 0;

    for word in line.split_whitespace() {
        let word_width = word.width();

        if word_width > width {
            if !current_line.is_empty() {
                result.push(std::mem::take(&mut current_line));
                current_width = 0;
            }
            push_word_breaking_chars(
                word,
                width,
                &mut current_line,
                &mut current_width,
                &mut result,
            );
            continue;
        }

        if current_width == 0 {
            // First word on line
            current_line = word.to_string();
            current_width = word_width;
        } else if current_width + 1 + word_width <= width {
            // Word fits with space
            current_line.push(' ');
            current_line.push_str(word);
            current_width += 1 + word_width;
        } else {
            // Word doesn't fit, start new line
            result.push(current_line);
            current_line = word.to_string();
            current_width = word_width;
        }
    }

    if !current_line.is_empty() {
        result.push(current_line);
    }

    if result.is_empty() {
        vec![String::new()]
    } else {
        result
    }
}

fn push_word_breaking_chars(
    word: &str,
    width: usize,
    current_line: &mut String,
    current_width: &mut usize,
    result: &mut Vec<String>,
) {
    for ch in word.chars() {
        let ch_width = ch.width().unwrap_or(1);
        if *current_width + ch_width > width && *current_width > 0 {
            result.push(std::mem::take(current_line));
            *current_width = 0;
        }
        current_line.push(ch);
        *current_width += ch_width;
    }
}

/// Per-block streaming substate: optional line-buffer feeding a collector +
/// chunker/policy for two-gear pacing.
///
/// Pipeline:
/// ```text
/// raw delta -> LineBuffer.push -> take_committable -> collector + chunker -> commit tick
/// ```
///
/// The [`LineBuffer`] remains available for line-sensitive modes. Normal
/// assistant prose and thinking blocks bypass it so text can stream in live
/// micro-chunks instead of waiting for newline boundaries.
#[derive(Debug, Default)]
struct BlockState {
    /// Newline gate: holds back trailing partial-line text between deltas.
    /// Bypassed when `bypass_gate` is true (thinking blocks).
    line_buffer: LineBuffer,
    /// Whether to bypass the [`LineBuffer`] (thinking blocks stream live).
    bypass_gate: bool,
    collector: MarkdownStreamCollector,
    chunker: StreamChunker,
    policy: AdaptiveChunkingPolicy,
}

/// State for managing multiple stream collectors (one per content block)
#[derive(Debug, Default)]
pub struct StreamingState {
    /// Per-block state by index (collector + chunker + policy).
    blocks: Vec<Option<BlockState>>,
    /// Whether any stream is currently active
    pub is_active: bool,
    /// Accumulated text for display
    pub accumulated_text: String,
    /// Accumulated thinking for display
    pub accumulated_thinking: String,
}

impl StreamingState {
    /// Create a new streaming state
    pub fn new() -> Self {
        Self::default()
    }

    /// Start a new text block. Assistant prose is buffered until the next
    /// display-clock beat so provider bursts produce one visible mutation.
    pub fn start_text(&mut self, index: usize, width: Option<usize>) {
        self.ensure_capacity(index);
        self.blocks[index] = Some(BlockState {
            line_buffer: LineBuffer::new(),
            bypass_gate: true,
            collector: MarkdownStreamCollector::new(width, false),
            chunker: StreamChunker::new(),
            policy: AdaptiveChunkingPolicy::new(),
        });
        self.is_active = true;
    }

    /// Start a new thinking block. Thinking deltas bypass the newline gate so
    /// they remain visually live — long reasoning often arrives as a single
    /// paragraph without intermediate newlines, and gating it would create
    /// long pauses where the user sees nothing.
    pub fn start_thinking(&mut self, index: usize, width: Option<usize>) {
        self.ensure_capacity(index);
        self.blocks[index] = Some(BlockState {
            line_buffer: LineBuffer::new(),
            bypass_gate: true,
            collector: MarkdownStreamCollector::new(width, true),
            chunker: StreamChunker::new(),
            policy: AdaptiveChunkingPolicy::new(),
        });
        self.is_active = true;
    }

    /// Push content to a block. Routing depends on the block kind:
    ///
    /// - Assistant text blocks: incoming bytes normally bypass [`LineBuffer`]
    ///   and are split into small display chunks downstream.
    /// - Thinking blocks: bytes bypass the gate and go straight to the
    ///   collector/chunker so reasoning stays visually live (long thoughts
    ///   often have no intermediate newlines).
    ///
    /// `accumulated_text` / `accumulated_thinking` always track the full raw
    /// stream so callers building API messages or doing retries see exactly
    /// what the model emitted, regardless of UI gating.
    pub fn push_content(&mut self, index: usize, content: &str) {
        if let Some(Some(block)) = self.blocks.get_mut(index) {
            // Always update the raw accumulator first — UI gating must not
            // affect what we send back to the model on retry/continuation.
            if block.collector.is_thinking {
                self.accumulated_thinking.push_str(content);
            } else {
                self.accumulated_text.push_str(content);
            }

            // Determine what bytes are safe to expose downstream on this push.
            let downstream: String = if block.bypass_gate {
                // Thinking: forward verbatim to collector + chunker.
                content.to_string()
            } else {
                // Assistant text: gate at the last-newline boundary.
                block.line_buffer.push(content);
                block.line_buffer.take_committable()
            };

            if downstream.is_empty() {
                return;
            }

            if block.bypass_gate {
                block.chunker.push_delta(&downstream);
            } else {
                block.collector.push(&downstream);
                let committed = block.collector.commit_complete_text();
                if !committed.is_empty() {
                    block.chunker.push_delta(&committed);
                }
            }
        }
    }

    /// Get newly committed lines from a block. (Legacy entry point that maps
    /// onto the chunker.)
    pub fn commit_lines(&mut self, index: usize) -> Vec<Line<'static>> {
        let text = self.commit_text(index);
        if text.is_empty() {
            return Vec::new();
        }
        // Re-render the text through the same path the collector used.
        let style = if self
            .blocks
            .get(index)
            .and_then(|b| b.as_ref())
            .is_some_and(|b| b.collector.is_thinking)
        {
            Style::default()
                .fg(palette::STATUS_WARNING)
                .add_modifier(Modifier::DIM | Modifier::ITALIC)
        } else {
            Style::default()
        };
        let mut lines = Vec::new();
        for line in text.lines() {
            lines.push(Line::from(Span::styled(line.to_string(), style)));
        }
        if text.ends_with('\n') {
            lines.push(Line::from(""));
        }
        lines
    }

    /// Run one commit-tick of the chunker policy and return any text safe to
    /// flush to the transcript on this tick. May be empty (Smooth-mode tick
    /// against an empty queue) or contain anywhere from one line up to the
    /// full backlog (CatchUp-mode burst drain).
    pub fn commit_text(&mut self, index: usize) -> String {
        if let Some(Some(block)) = self.blocks.get_mut(index) {
            let now = Instant::now();
            let out = run_commit_tick(&mut block.policy, &mut block.chunker, now);
            out.committed_text
        } else {
            String::new()
        }
    }

    /// Inspect the current chunking mode for a block (testing/observability).
    pub fn chunking_mode(&self, index: usize) -> Option<ChunkingMode> {
        self.blocks
            .get(index)
            .and_then(|b| b.as_ref())
            .map(|b| b.policy.mode())
    }

    /// Whether the chunker has queued content waiting to be flushed by the
    /// next commit tick. Useful for callers that want to drive an extra tick
    /// while the queue drains under Smooth-mode pacing.
    pub fn has_pending_chunker_lines(&self, index: usize) -> bool {
        self.blocks
            .get(index)
            .and_then(|b| b.as_ref())
            .is_some_and(|b| b.chunker.queued_lines() > 0)
    }

    /// Finalize a block and get remaining lines
    pub fn finalize_block(&mut self, index: usize) -> Vec<Line<'static>> {
        let text = self.finalize_block_text(index);
        if text.is_empty() {
            return Vec::new();
        }
        let style = if self
            .blocks
            .get(index)
            .and_then(|b| b.as_ref())
            .is_some_and(|b| b.collector.is_thinking)
        {
            Style::default()
                .fg(palette::STATUS_WARNING)
                .add_modifier(Modifier::DIM | Modifier::ITALIC)
        } else {
            Style::default()
        };
        let mut lines = Vec::new();
        for line in text.lines() {
            lines.push(Line::from(Span::styled(line.to_string(), style)));
        }
        if text.ends_with('\n') {
            lines.push(Line::from(""));
        }
        lines
    }

    /// Finalize a block and get remaining raw text. Drains the full pipeline
    /// in upstream-to-downstream order:
    ///
    /// 1. [`LineBuffer::flush`] returns any post-newline tail held by the gate.
    ///    For gated blocks this is critical — without it, a final partial
    ///    line (e.g. text the model emitted without a trailing newline before
    ///    the turn ended) would otherwise be stranded in the gate.
    /// 2. The collector's `finalize_text` releases any partial line it still
    ///    holds (relevant for the bypass path where the collector receives
    ///    raw deltas directly).
    /// 3. The chunker's `drain_remaining` releases queued whole-line text
    ///    that the policy hadn't yet committed.
    pub fn finalize_block_text(&mut self, index: usize) -> String {
        if let Some(Some(block)) = self.blocks.get_mut(index) {
            // Flush the gate first so any held tail rejoins the stream
            // before the collector/chunker drain. For thinking blocks the
            // gate is unused, so this is a no-op.
            let gate_tail = block.line_buffer.flush();
            if !gate_tail.is_empty() {
                block.collector.push(&gate_tail);
            }
            // Any newly committable text after the gate flush feeds the
            // chunker so drain order remains "queued-lines, then partial-tail".
            let post_flush = block.collector.commit_complete_text();
            if !post_flush.is_empty() {
                block.chunker.push_delta(&post_flush);
            }
            // Any unterminated tail still in the collector is returned raw.
            let tail = block.collector.finalize_text();
            // Any whole-line text held by the chunker is safe to emit now.
            let mut out = block.chunker.drain_remaining();
            if !tail.is_empty() {
                out.push_str(&tail);
            }
            self.check_active();
            out
        } else {
            String::new()
        }
    }

    /// Finalize all blocks
    pub fn finalize_all(&mut self) -> Vec<(usize, Vec<Line<'static>>)> {
        let mut result = Vec::new();
        let len = self.blocks.len();
        for i in 0..len {
            let lines = self.finalize_block(i);
            if !lines.is_empty() {
                result.push((i, lines));
            }
        }
        self.is_active = false;
        result
    }

    /// Propagate the low-motion flag to every block's chunking policy.
    /// When true, all policies stay in `Smooth` regardless of queue pressure,
    /// preventing CatchUp burst drains that would create sudden visual jumps.
    pub fn set_low_motion(&mut self, low_motion: bool) {
        for block in self.blocks.iter_mut().flatten() {
            block.policy.set_low_motion(low_motion);
        }
    }

    /// Check if any stream is still active
    fn check_active(&mut self) {
        self.is_active = self.blocks.iter().any(|b| {
            b.as_ref()
                .is_some_and(|state| state.collector.is_streaming())
        });
    }

    /// Ensure capacity for the given index
    fn ensure_capacity(&mut self, index: usize) {
        while self.blocks.len() <= index {
            self.blocks.push(None);
        }
    }

    /// Reset the streaming state
    pub fn reset(&mut self) {
        self.blocks.clear();
        self.is_active = false;
        self.accumulated_text.clear();
        self.accumulated_thinking.clear();
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use unicode_width::UnicodeWidthStr;

    #[test]
    fn test_commit_complete_lines() {
        let mut collector = MarkdownStreamCollector::new(Some(80), false);

        // Push incomplete line
        collector.push("Hello ");
        let lines = collector.commit_complete_lines();
        assert!(lines.is_empty()); // No complete lines yet

        // Complete the line
        collector.push("World\n");
        let lines = collector.commit_complete_lines();
        assert_eq!(lines.len(), 2); // "Hello World" + empty line from trailing \n

        // Push more content
        collector.push("Second line");
        let lines = collector.commit_complete_lines();
        assert!(lines.is_empty()); // No new complete lines

        // Finalize
        let lines = collector.finalize();
        assert_eq!(lines.len(), 1); // "Second line"
    }

    #[test]
    fn test_wrap_line() {
        let result = wrap_line("This is a long line that should be wrapped", 20);
        assert!(result.len() > 1);
    }

    #[test]
    fn wrap_line_breaks_no_whitespace_cjk_text() {
        let text = "这是一个没有任何空格的中文长段落".repeat(12);
        let result = wrap_line(&text, 40);

        assert!(result.len() > 1);
        assert!(result.iter().all(|line| line.width() <= 40));
        assert_eq!(result.join(""), text);
    }

    #[test]
    fn wrap_line_breaks_first_overlong_word() {
        let text = "x".repeat(120);
        let result = wrap_line(&text, 40);

        assert_eq!(result.len(), 3);
        assert!(result.iter().all(|line| line.width() <= 40));
        assert_eq!(result.join(""), text);
    }

    #[test]
    fn assistant_text_streams_before_newline() {
        let mut state = StreamingState::new();
        state.start_text(0, None);
        state.push_content(0, "hello world");

        assert_eq!(state.commit_text(0), "hello world");
        assert!(!state.has_pending_chunker_lines(0));
    }

    #[test]
    fn thinking_text_streams_before_newline() {
        let mut state = StreamingState::new();
        state.start_thinking(0, None);
        state.push_content(0, "thinking deeply");

        assert_eq!(state.commit_text(0), "thinking deeply");
        assert!(!state.has_pending_chunker_lines(0));
    }

    #[test]
    fn finalize_preserves_uncommitted_micro_chunks() {
        let mut state = StreamingState::new();
        state.start_text(0, None);
        state.set_low_motion(true);
        state.push_content(0, "abc");
        assert_eq!(state.commit_text(0), "abc");
        assert_eq!(state.finalize_block_text(0), "");
    }

    #[test]
    fn stream_display_clock_coalesces_bursty_tiny_deltas() {
        let interval = Duration::from_millis(33);
        let mut clock = StreamDisplayClock::new(interval);
        let t0 = Instant::now();

        for _ in 0..100 {
            clock.note_delta(t0);
        }

        assert_eq!(clock.due_in(t0), Some(Duration::ZERO));
        assert!(clock.take_due(t0));
        assert_eq!(clock.commit_count(), 1);

        for _ in 0..25 {
            clock.note_delta(t0 + Duration::from_millis(5));
        }
        assert!(!clock.take_due(t0 + Duration::from_millis(5)));
        assert_eq!(
            clock.due_in(t0 + Duration::from_millis(5)),
            Some(Duration::from_millis(28))
        );
        assert!(clock.take_due(t0 + interval));
        assert_eq!(clock.commit_count(), 2);
    }

    #[test]
    fn stream_display_clock_bounds_long_reasoning_commit_count() {
        let interval = Duration::from_millis(33);
        let mut clock = StreamDisplayClock::new(interval);
        let t0 = Instant::now();
        let mut commits = 0u64;

        for millis in 0..300 {
            let now = t0 + Duration::from_millis(millis);
            clock.note_delta(now);
            if clock.take_due(now) {
                commits += 1;
            }
        }

        assert!(commits > 1, "long streams should keep advancing visibly");
        assert!(
            commits <= 11,
            "300 one-ms deltas should not commit on provider cadence: {commits}"
        );
        assert_eq!(commits, clock.commit_count());
    }

    #[test]
    fn stream_display_clock_final_flush_consumes_pending_delta() {
        let mut clock = StreamDisplayClock::new(Duration::from_millis(33));
        let t0 = Instant::now();

        clock.note_delta(t0);
        assert!(clock.take_due(t0));
        clock.note_delta(t0 + Duration::from_millis(4));

        assert!(!clock.take_due(t0 + Duration::from_millis(4)));
        assert!(clock.flush_now(t0 + Duration::from_millis(5)));
        assert_eq!(clock.due_in(t0 + Duration::from_millis(5)), None);
        assert!(!clock.take_due(t0 + Duration::from_millis(33)));
        assert_eq!(clock.commit_count(), 2);
    }

    #[test]
    fn bursty_stream_state_has_no_text_loss_after_coalesced_flushes() {
        let mut state = StreamingState::new();
        state.start_text(0, None);
        let mut expected = String::new();

        for idx in 0..250 {
            let chunk = format!("{idx}.");
            expected.push_str(&chunk);
            state.push_content(0, &chunk);
        }

        let first_flush = state.commit_text(0);
        assert_eq!(first_flush, expected);
        assert_eq!(state.finalize_block_text(0), "");
    }

    #[test]
    fn normal_clock_catch_up_only_when_backlog_crosses_threshold() {
        let interval = Duration::from_millis(33);
        let mut clock = StreamDisplayClock::new(interval);
        let t0 = Instant::now();

        clock.note_delta(t0);
        assert!(clock.take_due(t0));
        // Small backlog after a commit stays on the steady interval.
        clock.note_delta_with_backlog(
            t0 + Duration::from_millis(1),
            3,
            Some(Duration::from_millis(5)),
        );
        assert!(!clock.take_due(t0 + Duration::from_millis(1)));
        assert_eq!(clock.catch_up_count(), 0);

        // Measured backlog crosses the catch-up threshold → due immediately.
        clock.note_delta_with_backlog(
            t0 + Duration::from_millis(2),
            CATCH_UP_QUEUE_DEPTH,
            Some(Duration::from_millis(10)),
        );
        assert!(clock.take_due(t0 + Duration::from_millis(2)));
        assert!(clock.catch_up_count() >= 1);
    }

    #[test]
    fn reduced_motion_keeps_steady_clock_without_catch_up_or_typewriter() {
        let interval = Duration::from_millis(33);
        let mut clock = StreamDisplayClock::new(interval);
        clock.set_allow_catch_up(false);
        let t0 = Instant::now();

        clock.note_delta(t0);
        assert!(clock.take_due(t0));
        clock.note_delta_with_backlog(
            t0 + Duration::from_millis(1),
            CATCH_UP_QUEUE_DEPTH * 2,
            Some(CATCH_UP_OLDEST_AGE),
        );
        // Reduced motion must not pull the beat forward.
        assert!(!clock.take_due(t0 + Duration::from_millis(1)));
        assert_eq!(clock.catch_up_count(), 0);
        assert_eq!(
            clock.due_in(t0 + Duration::from_millis(1)),
            Some(Duration::from_millis(32))
        );
        assert!(clock.take_due(t0 + interval));
        // Same interval as full motion — not a slower artificial typewriter.
        assert_eq!(clock.commit_count(), 2);
    }
}