diff --git a/crates/tui/src/tui/streaming/chunking.rs b/crates/tui/src/tui/streaming/chunking.rs deleted file mode 100644 index 5de7ee517..000000000 --- a/crates/tui/src/tui/streaming/chunking.rs +++ /dev/null @@ -1,457 +0,0 @@ -//! Adaptive stream chunking policy for two-gear streaming. -//! -//! Ported from `codex-rs/tui/src/streaming/chunking.rs`, adapted for codewhale's -//! text-based streaming pipeline. The policy is queue-pressure driven and -//! source-agnostic. -//! -//! # Mental model -//! -//! Two gears: -//! - [`ChunkingMode::Smooth`]: normal pressure. -//! - [`ChunkingMode::CatchUp`]: elevated pressure. -//! -//! Every caller drains all currently available chunks so the display follows -//! the upstream SSE delta cadence. Low motion affects decorative animation and -//! redraw frequency, never the apparent speed of model text. -//! -//! # Hysteresis -//! -//! - Enter `CatchUp` when `queued_lines >= ENTER_QUEUE_DEPTH_LINES` OR -//! the oldest queued chunk is at least [`ENTER_OLDEST_AGE`]. -//! - Exit `CatchUp` only after pressure stays below [`EXIT_QUEUE_DEPTH_LINES`] -//! AND [`EXIT_OLDEST_AGE`] for at least [`EXIT_HOLD`]. -//! - After exit, suppress immediate re-entry for [`REENTER_CATCH_UP_HOLD`] -//! unless backlog is "severe" (queue >= [`SEVERE_QUEUE_DEPTH_LINES`] or -//! oldest >= [`SEVERE_OLDEST_AGE`]). - -use std::time::Duration; -use std::time::Instant; - -/// Queue-depth threshold that allows entering catch-up mode. -pub(crate) const ENTER_QUEUE_DEPTH_LINES: usize = 160; - -/// Oldest-chunk age threshold that allows entering catch-up mode. -pub(crate) const ENTER_OLDEST_AGE: Duration = Duration::from_millis(1_200); - -/// Queue-depth threshold used when evaluating catch-up exit hysteresis. -pub(crate) const EXIT_QUEUE_DEPTH_LINES: usize = 32; - -/// Oldest-chunk age threshold used when evaluating catch-up exit hysteresis. -pub(crate) const EXIT_OLDEST_AGE: Duration = Duration::from_millis(300); - -/// Minimum duration queue pressure must stay below exit thresholds to leave catch-up mode. -pub(crate) const EXIT_HOLD: Duration = Duration::from_millis(250); - -/// Cooldown window after a catch-up exit that suppresses immediate re-entry. -pub(crate) const REENTER_CATCH_UP_HOLD: Duration = Duration::from_millis(250); - -/// Queue-depth cutoff that marks backlog as severe (bypasses re-entry hold). -pub(crate) const SEVERE_QUEUE_DEPTH_LINES: usize = 640; - -/// Oldest-line age cutoff that marks backlog as severe. -pub(crate) const SEVERE_OLDEST_AGE: Duration = Duration::from_millis(4_000); - -#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] -pub enum ChunkingMode { - /// Drain one display chunk per baseline commit tick. - #[default] - Smooth, - /// Drain the queued backlog according to queue pressure. - CatchUp, -} - -/// Captures queue pressure inputs used by adaptive chunking decisions. -#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] -pub struct QueueSnapshot { - /// Number of queued stream chunks waiting to be displayed. - pub queued_lines: usize, - /// Age of the oldest queued chunk at decision time. - pub oldest_age: Option, -} - -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -pub enum DrainPlan { - /// Emit all queued chunks available at this tick. - Available, - /// Emit exactly one queued line. - Single, -} - -/// Represents one policy decision for a specific queue snapshot. -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -pub struct ChunkingDecision { - /// Mode after applying hysteresis transitions for this decision. - pub mode: ChunkingMode, - /// Whether this decision transitioned from `Smooth` into `CatchUp`. - pub entered_catch_up: bool, - /// Drain plan to execute for the current commit tick. - pub drain_plan: DrainPlan, -} - -/// Maintains adaptive chunking mode and hysteresis state across ticks. -#[derive(Debug, Default, Clone)] -pub struct AdaptiveChunkingPolicy { - mode: ChunkingMode, - below_exit_threshold_since: Option, - last_catch_up_exit_at: Option, - /// When true, the policy never enters `CatchUp` — it stays in `Smooth` - /// regardless of queue pressure, keeping the display calm for users who - /// prefer reduced visual churn. - low_motion: bool, -} - -impl AdaptiveChunkingPolicy { - pub fn new() -> Self { - Self::default() - } - - /// Returns the policy mode used by the most recent decision. - pub fn mode(&self) -> ChunkingMode { - self.mode - } - - /// Resets state to baseline smooth mode. - pub fn reset(&mut self) { - self.mode = ChunkingMode::Smooth; - self.below_exit_threshold_since = None; - self.last_catch_up_exit_at = None; - } - - /// When true, the policy never enters `CatchUp` — it stays in `Smooth` - /// regardless of queue pressure. - pub fn set_low_motion(&mut self, low_motion: bool) { - self.low_motion = low_motion; - if low_motion { - self.mode = ChunkingMode::Smooth; - self.below_exit_threshold_since = None; - self.last_catch_up_exit_at = None; - } - } - - /// Computes a drain decision from the current queue snapshot. - pub fn decide(&mut self, snapshot: QueueSnapshot, now: Instant) -> ChunkingDecision { - // Low motion stays in Smooth mode, but text still follows upstream - // cadence. Dripping one grapheme per redraw creates an artificial - // typewriter followed by a large final flush. - if self.low_motion { - self.mode = ChunkingMode::Smooth; - self.below_exit_threshold_since = None; - return ChunkingDecision { - mode: self.mode, - entered_catch_up: false, - drain_plan: DrainPlan::Available, - }; - } - - if snapshot.queued_lines == 0 { - self.note_catch_up_exit(now); - self.mode = ChunkingMode::Smooth; - self.below_exit_threshold_since = None; - return ChunkingDecision { - mode: self.mode, - entered_catch_up: false, - drain_plan: DrainPlan::Available, - }; - } - - let entered_catch_up = match self.mode { - ChunkingMode::Smooth => self.maybe_enter_catch_up(snapshot, now), - ChunkingMode::CatchUp => { - self.maybe_exit_catch_up(snapshot, now); - false - } - }; - - ChunkingDecision { - mode: self.mode, - entered_catch_up, - drain_plan: DrainPlan::Available, - } - } - - fn maybe_enter_catch_up(&mut self, snapshot: QueueSnapshot, now: Instant) -> bool { - if !should_enter_catch_up(snapshot) { - return false; - } - if self.reentry_hold_active(now) && !is_severe_backlog(snapshot) { - return false; - } - self.mode = ChunkingMode::CatchUp; - self.below_exit_threshold_since = None; - self.last_catch_up_exit_at = None; - true - } - - fn maybe_exit_catch_up(&mut self, snapshot: QueueSnapshot, now: Instant) { - if !should_exit_catch_up(snapshot) { - self.below_exit_threshold_since = None; - return; - } - - match self.below_exit_threshold_since { - Some(since) if now.saturating_duration_since(since) >= EXIT_HOLD => { - self.mode = ChunkingMode::Smooth; - self.below_exit_threshold_since = None; - self.last_catch_up_exit_at = Some(now); - } - Some(_) => {} - None => { - self.below_exit_threshold_since = Some(now); - } - } - } - - fn note_catch_up_exit(&mut self, now: Instant) { - if self.mode == ChunkingMode::CatchUp { - self.last_catch_up_exit_at = Some(now); - } - } - - fn reentry_hold_active(&self, now: Instant) -> bool { - self.last_catch_up_exit_at - .is_some_and(|exit| now.saturating_duration_since(exit) < REENTER_CATCH_UP_HOLD) - } -} - -/// Returns whether current queue pressure warrants entering catch-up mode. -fn should_enter_catch_up(snapshot: QueueSnapshot) -> bool { - snapshot.queued_lines >= ENTER_QUEUE_DEPTH_LINES - || snapshot - .oldest_age - .is_some_and(|oldest| oldest >= ENTER_OLDEST_AGE) -} - -/// Returns whether queue pressure is low enough to begin exit hysteresis. -fn should_exit_catch_up(snapshot: QueueSnapshot) -> bool { - snapshot.queued_lines <= EXIT_QUEUE_DEPTH_LINES - && snapshot - .oldest_age - .is_some_and(|oldest| oldest <= EXIT_OLDEST_AGE) -} - -/// Returns whether backlog is severe enough to bypass the re-entry hold. -fn is_severe_backlog(snapshot: QueueSnapshot) -> bool { - snapshot.queued_lines >= SEVERE_QUEUE_DEPTH_LINES - || snapshot - .oldest_age - .is_some_and(|oldest| oldest >= SEVERE_OLDEST_AGE) -} - -#[cfg(test)] -mod tests { - use super::*; - - fn snap(queued_lines: usize, oldest_age_ms: u64) -> QueueSnapshot { - QueueSnapshot { - queued_lines, - oldest_age: Some(Duration::from_millis(oldest_age_ms)), - } - } - - fn empty_snap() -> QueueSnapshot { - QueueSnapshot { - queued_lines: 0, - oldest_age: None, - } - } - - #[test] - fn smooth_only_burst_drains_available_chunks_in_normal_motion() { - // Five slowly-arriving lines, each well below enter thresholds, never - // flip the policy out of `Smooth`. Normal motion still drains what is - // already available so display pacing follows upstream deltas. - let mut policy = AdaptiveChunkingPolicy::new(); - let t0 = Instant::now(); - - for i in 0..5 { - // 1 queued line, age 10 ms — far below ENTER thresholds. - let decision = policy.decide(snap(1, 10), t0 + Duration::from_millis(50 * i)); - assert_eq!(decision.mode, ChunkingMode::Smooth); - assert!(!decision.entered_catch_up); - assert_eq!(decision.drain_plan, DrainPlan::Available); - } - } - - #[test] - fn deep_burst_flips_to_catch_up_and_drains_backlog() { - // A burst crossing ENTER_QUEUE_DEPTH_LINES enters CatchUp. With - // single-grapheme chunks, the threshold stays high enough that - // ordinary prose still drips in visibly before catch-up engages. - // The policy should enter `CatchUp`, while normal-motion draining still - // preserves the already-arrived upstream burst. - let mut policy = AdaptiveChunkingPolicy::new(); - let now = Instant::now(); - - let decision = policy.decide(snap(ENTER_QUEUE_DEPTH_LINES, 10), now); - assert_eq!(decision.mode, ChunkingMode::CatchUp); - assert!(decision.entered_catch_up); - assert_eq!(decision.drain_plan, DrainPlan::Available); - - // Larger backlog requested next tick: still CatchUp, batch grows to match. - let larger_backlog = ENTER_QUEUE_DEPTH_LINES + 80; - let decision = policy.decide(snap(larger_backlog, 30), now + Duration::from_millis(10)); - assert_eq!(decision.mode, ChunkingMode::CatchUp); - assert!(!decision.entered_catch_up, "no second transition signal"); - assert_eq!(decision.drain_plan, DrainPlan::Available); - } - - #[test] - fn age_threshold_alone_triggers_catch_up() { - // Queue depth is small, but the oldest chunk has crossed the age threshold. - // Either condition is sufficient to enter catch-up. - let mut policy = AdaptiveChunkingPolicy::new(); - let now = Instant::now(); - - let decision = policy.decide(snap(2, ENTER_OLDEST_AGE.as_millis() as u64), now); - assert_eq!(decision.mode, ChunkingMode::CatchUp); - assert!(decision.entered_catch_up); - assert_eq!(decision.drain_plan, DrainPlan::Available); - } - - #[test] - fn catch_up_exits_after_low_activity_hold() { - // Enter CatchUp via depth burst, then drop pressure below exit - // thresholds. Policy must hold for >=EXIT_HOLD before returning to Smooth. - let mut policy = AdaptiveChunkingPolicy::new(); - let t0 = Instant::now(); - - let _ = policy.decide(snap(ENTER_QUEUE_DEPTH_LINES, 20), t0); - assert_eq!(policy.mode(), ChunkingMode::CatchUp); - - // Pressure drops to the exit thresholds. - // Hold begins; not yet 250ms. - let pre_hold = policy.decide( - snap(EXIT_QUEUE_DEPTH_LINES, EXIT_OLDEST_AGE.as_millis() as u64), - t0 + Duration::from_millis(50), - ); - assert_eq!(pre_hold.mode, ChunkingMode::CatchUp); - - // Still under hold. - let mid_hold = policy.decide( - snap(EXIT_QUEUE_DEPTH_LINES, EXIT_OLDEST_AGE.as_millis() as u64), - t0 + Duration::from_millis(200), - ); - assert_eq!(mid_hold.mode, ChunkingMode::CatchUp); - - // Past EXIT_HOLD (250 ms) → return to Smooth. - let post_hold = policy.decide( - snap(EXIT_QUEUE_DEPTH_LINES, EXIT_OLDEST_AGE.as_millis() as u64), - t0 + Duration::from_millis(320), - ); - assert_eq!(post_hold.mode, ChunkingMode::Smooth); - assert_eq!(post_hold.drain_plan, DrainPlan::Available); - } - - #[test] - fn idle_resets_to_smooth_immediately() { - // An empty queue forces Smooth regardless of prior mode. - let mut policy = AdaptiveChunkingPolicy::new(); - let now = Instant::now(); - - let _ = policy.decide(snap(ENTER_QUEUE_DEPTH_LINES, 20), now); - assert_eq!(policy.mode(), ChunkingMode::CatchUp); - - let decision = policy.decide(empty_snap(), now + Duration::from_millis(10)); - assert_eq!(decision.mode, ChunkingMode::Smooth); - assert_eq!(decision.drain_plan, DrainPlan::Available); - } - - #[test] - fn reentry_hold_blocks_immediate_flip_back() { - // After exiting CatchUp via idle, a threshold-sized burst that arrives within - // the re-entry hold window should not immediately re-enter CatchUp. - let mut policy = AdaptiveChunkingPolicy::new(); - let t0 = Instant::now(); - - let _ = policy.decide(snap(ENTER_QUEUE_DEPTH_LINES, 20), t0); - let _ = policy.decide(empty_snap(), t0 + Duration::from_millis(10)); - - // Within REENTER_CATCH_UP_HOLD (250 ms): hold blocks re-entry. - let held = policy.decide( - snap(ENTER_QUEUE_DEPTH_LINES, 20), - t0 + Duration::from_millis(100), - ); - assert_eq!(held.mode, ChunkingMode::Smooth); - assert_eq!(held.drain_plan, DrainPlan::Available); - - // Past the hold: re-entry permitted. - let reentered = policy.decide( - snap(ENTER_QUEUE_DEPTH_LINES, 20), - t0 + Duration::from_millis(400), - ); - assert_eq!(reentered.mode, ChunkingMode::CatchUp); - assert_eq!(reentered.drain_plan, DrainPlan::Available); - } - - #[test] - fn severe_backlog_bypasses_reentry_hold() { - // Even within the hold window, a "severe" backlog bypasses - // the gate so display lag doesn't unbounded-grow. - let mut policy = AdaptiveChunkingPolicy::new(); - let t0 = Instant::now(); - - let _ = policy.decide(snap(ENTER_QUEUE_DEPTH_LINES, 20), t0); - let _ = policy.decide(empty_snap(), t0 + Duration::from_millis(10)); - - let severe = policy.decide( - snap(SEVERE_QUEUE_DEPTH_LINES, 20), - t0 + Duration::from_millis(100), - ); - assert_eq!(severe.mode, ChunkingMode::CatchUp); - assert_eq!(severe.drain_plan, DrainPlan::Available); - } - - #[test] - fn low_motion_always_smooth_regardless_of_pressure() { - let mut policy = AdaptiveChunkingPolicy::new(); - policy.set_low_motion(true); - let t0 = Instant::now(); - - // Queue depth far above ENTER threshold. - let d1 = policy.decide(snap(ENTER_QUEUE_DEPTH_LINES + 80, 10), t0); - assert_eq!(d1.mode, ChunkingMode::Smooth); - assert!(!d1.entered_catch_up); - assert_eq!(d1.drain_plan, DrainPlan::Available); - - // Oldest age far above ENTER threshold. - let d2 = policy.decide( - snap(5, ENTER_OLDEST_AGE.as_millis() as u64), - t0 + Duration::from_millis(100), - ); - assert_eq!(d2.mode, ChunkingMode::Smooth); - assert!(!d2.entered_catch_up); - assert_eq!(d2.drain_plan, DrainPlan::Available); - - // Severe backlog — still Smooth. - let d3 = policy.decide( - snap( - SEVERE_QUEUE_DEPTH_LINES + 80, - SEVERE_OLDEST_AGE.as_millis() as u64, - ), - t0 + Duration::from_millis(200), - ); - assert_eq!(d3.mode, ChunkingMode::Smooth); - assert_eq!(d3.drain_plan, DrainPlan::Available); - } - - #[test] - fn low_motion_reset_resumes_normal_operation() { - let mut policy = AdaptiveChunkingPolicy::new(); - policy.set_low_motion(true); - let t0 = Instant::now(); - - // Low motion blocks catch-up. - let d1 = policy.decide(snap(ENTER_QUEUE_DEPTH_LINES + 80, 10), t0); - assert_eq!(d1.mode, ChunkingMode::Smooth); - - // Turn off low motion — next burst should enter CatchUp. - policy.set_low_motion(false); - let d2 = policy.decide( - snap(ENTER_QUEUE_DEPTH_LINES + 80, 10), - t0 + Duration::from_millis(10), - ); - assert_eq!(d2.mode, ChunkingMode::CatchUp); - assert!(d2.entered_catch_up); - assert_eq!(d2.drain_plan, DrainPlan::Available); - } -} diff --git a/crates/tui/src/tui/streaming/commit_tick.rs b/crates/tui/src/tui/streaming/commit_tick.rs deleted file mode 100644 index c0beea551..000000000 --- a/crates/tui/src/tui/streaming/commit_tick.rs +++ /dev/null @@ -1,256 +0,0 @@ -//! Commit-tick scheduler that drains a stream chunker according to policy. -//! -//! Bridges [`AdaptiveChunkingPolicy`] with a concrete [`StreamChunker`] queue. -//! Callers feed raw text deltas via [`StreamChunker::push_delta`], then call -//! [`run_commit_tick`] on every commit beat to obtain text to flush to the -//! transcript on this beat. Normal motion drains all text received since the -//! prior tick so the display follows the upstream delta cadence. Low-motion -//! mode may coalesce redraws, but never fabricates a one-grapheme typewriter. -//! -//! The chunker is the unit of streaming — one per active block (assistant / -//! thinking). Tool output is unbuffered and bypasses this path. - -use std::collections::VecDeque; -use std::time::Duration; -use std::time::Instant; - -use super::chunking::AdaptiveChunkingPolicy; -use super::chunking::ChunkingDecision; -use super::chunking::DrainPlan; -use super::chunking::QueueSnapshot; - -/// Buffers raw model deltas and emits them on display-clock commits. -/// -/// A queue entry is one provider delta, not one grapheme. The previous path -/// split every delta into one-grapheme `String` allocations and then drained -/// the entire queue on the same beat. That preserved burstiness while paying -/// maximum allocation and queue overhead. -#[derive(Debug, Default)] -pub struct StreamChunker { - /// Bytes received but not yet split into display chunks. Normally empty; - /// retained so `drain_remaining` has a lossless place to pull from if we - /// ever decide to hold a tail for a future markdown-sensitive mode. - pending: String, - /// Provider deltas waiting to be flushed to the transcript. - queue: VecDeque, -} - -#[derive(Debug, Clone)] -struct QueuedChunk { - text: String, - enqueued_at: Instant, -} - -impl StreamChunker { - pub fn new() -> Self { - Self::default() - } - - /// Append a raw model delta. Returns whether at least one new display chunk was queued. - pub fn push_delta(&mut self, delta: &str) -> bool { - if delta.is_empty() { - return false; - } - let now = Instant::now(); - self.queue.push_back(QueuedChunk { - text: delta.to_string(), - enqueued_at: now, - }); - true - } - - /// Number of display chunks currently queued for commit. - pub fn queued_lines(&self) -> usize { - self.queue.len() - } - - /// Age of the oldest queued chunk, if any. - pub fn oldest_queued_age(&self, now: Instant) -> Option { - self.queue - .front() - .map(|q| now.saturating_duration_since(q.enqueued_at)) - } - - /// Whether the queue is empty AND no buffered partial line remains. - pub fn is_idle(&self) -> bool { - self.queue.is_empty() && self.pending.is_empty() - } - - /// Snapshot for policy decisions. - pub fn snapshot(&self, now: Instant) -> QueueSnapshot { - QueueSnapshot { - queued_lines: self.queue.len(), - oldest_age: self.oldest_queued_age(now), - } - } - - /// Drain `max_lines` queued chunks and return them as concatenated text. - pub fn drain_lines(&mut self, max_lines: usize) -> String { - let n = max_lines.min(self.queue.len()); - let mut out = String::new(); - for queued in self.queue.drain(..n) { - out.push_str(&queued.text); - } - out - } - - /// Drain any remaining pending bytes (called at stream finalize). - /// This includes both queued complete lines AND the tail partial line. - pub fn drain_remaining(&mut self) -> String { - let mut out = String::new(); - while let Some(q) = self.queue.pop_front() { - out.push_str(&q.text); - } - if !self.pending.is_empty() { - out.push_str(&self.pending); - self.pending.clear(); - } - out - } - - /// Reset internal state. - pub fn reset(&mut self) { - self.pending.clear(); - self.queue.clear(); - } -} - -/// One commit-tick decision plus the text that should be flushed on this tick. -pub struct CommitTickOutput { - pub committed_text: String, - pub decision: ChunkingDecision, - pub is_idle: bool, -} - -/// Run a single commit tick: ask the policy, drain the chunker accordingly. -pub fn run_commit_tick( - policy: &mut AdaptiveChunkingPolicy, - chunker: &mut StreamChunker, - now: Instant, -) -> CommitTickOutput { - let snapshot = chunker.snapshot(now); - let prior_mode = policy.mode(); - let decision = policy.decide(snapshot, now); - - if decision.mode != prior_mode { - tracing::trace!( - prior_mode = ?prior_mode, - new_mode = ?decision.mode, - queued_lines = snapshot.queued_lines, - oldest_queued_age_ms = snapshot.oldest_age.map(|age| age.as_millis() as u64), - entered_catch_up = decision.entered_catch_up, - "stream chunking mode transition" - ); - } - - let max = match decision.drain_plan { - DrainPlan::Available => snapshot.queued_lines, - DrainPlan::Single => 1, - }; - - // Drain through the chunker; an empty queue under Smooth produces "". - let committed_text = chunker.drain_lines(max); - - CommitTickOutput { - committed_text, - decision, - is_idle: chunker.is_idle(), - } -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::tui::streaming::chunking::ChunkingMode; - - #[test] - fn prose_streams_before_newline() { - let mut chunker = StreamChunker::new(); - let mut policy = AdaptiveChunkingPolicy::new(); - let now = Instant::now(); - - chunker.push_delta("hello world"); - let out = run_commit_tick(&mut policy, &mut chunker, now); - assert_eq!(out.committed_text, "hello world"); - assert!( - chunker.is_idle(), - "normal motion should preserve upstream pacing" - ); - - let out = run_commit_tick(&mut policy, &mut chunker, now + Duration::from_millis(5)); - assert_eq!(out.committed_text, ""); - } - - #[test] - fn low_motion_preserves_upstream_delta_pacing() { - let mut chunker = StreamChunker::new(); - let mut policy = AdaptiveChunkingPolicy::new(); - policy.set_low_motion(true); - let now = Instant::now(); - - chunker.push_delta("hello world"); - let out = run_commit_tick(&mut policy, &mut chunker, now); - assert_eq!(out.committed_text, "hello world"); - assert!(chunker.is_idle()); - - let out = run_commit_tick(&mut policy, &mut chunker, now + Duration::from_millis(20)); - assert_eq!(out.committed_text, ""); - } - - #[test] - fn normal_motion_burst_drains_available_backlog() { - let mut chunker = StreamChunker::new(); - let mut policy = AdaptiveChunkingPolicy::new(); - let t0 = Instant::now(); - - chunker.push_delta("abc"); - let out1 = run_commit_tick(&mut policy, &mut chunker, t0); - assert_eq!(out1.decision.mode, ChunkingMode::Smooth); - assert_eq!(out1.committed_text, "abc"); - assert!(out1.is_idle); - - let out2 = run_commit_tick(&mut policy, &mut chunker, t0 + Duration::from_millis(20)); - assert_eq!(out2.committed_text, ""); - } - - #[test] - fn low_motion_stream_keeps_combining_marks_with_base_letter() { - let mut chunker = StreamChunker::new(); - let mut policy = AdaptiveChunkingPolicy::new(); - policy.set_low_motion(true); - let t0 = Instant::now(); - - chunker.push_delta("e\u{301}x"); - let out1 = run_commit_tick(&mut policy, &mut chunker, t0); - assert_eq!(out1.committed_text, "e\u{301}x"); - let out2 = run_commit_tick(&mut policy, &mut chunker, t0 + Duration::from_millis(20)); - assert_eq!(out2.committed_text, ""); - } - - #[test] - fn large_burst_preserves_upstream_burst_in_normal_motion() { - // A large text burst arriving "at once" should be displayed at the - // same cadence instead of being synthetically dripped and then flushed - // at the end of the turn. - let mut chunker = StreamChunker::new(); - let mut policy = AdaptiveChunkingPolicy::new(); - let now = Instant::now(); - - let burst = "abcdefghijklmnopqrstuvwxyz".repeat(8); - chunker.push_delta(&burst); - let out = run_commit_tick(&mut policy, &mut chunker, now); - assert_eq!(out.decision.mode, ChunkingMode::Smooth); - assert_eq!(out.committed_text, burst); - assert!(out.is_idle); - } - - #[test] - fn finalize_drains_partial_tail() { - // The final, possibly-incomplete line must be flushed by drain_remaining. - let mut chunker = StreamChunker::new(); - chunker.push_delta("done\nno-newline-here"); - let drained = chunker.drain_remaining(); - assert_eq!(drained, "done\nno-newline-here"); - assert!(chunker.is_idle()); - } -} diff --git a/crates/tui/src/tui/streaming/line_buffer.rs b/crates/tui/src/tui/streaming/line_buffer.rs deleted file mode 100644 index caea2848e..000000000 --- a/crates/tui/src/tui/streaming/line_buffer.rs +++ /dev/null @@ -1,223 +0,0 @@ -//! Newline-boundary gate for streaming text. -//! -//! `LineBuffer` is an upstream-of-the-chunker safety layer that holds back any -//! text after the LAST `\n` until the next newline arrives. This prevents -//! partial multi-character markdown — most importantly partial code fences -//! (` ``` `) whose meaning flips depending on what follows on the same line — -//! from ever becoming visible state in the renderer. -//! -//! Mental model: -//! - `push(delta)` appends raw stream text to an internal pending buffer. -//! - `take_committable()` returns only the prefix up to and including the -//! LAST `\n` and clears that prefix. Whatever follows the last `\n` stays -//! in the buffer for the next push. -//! - `flush()` returns whatever is left, used at end-of-stream when the model -//! signals the turn is done. (The contract upstream of the chunker is that -//! only complete-line text is committed; `flush()` is the explicit escape -//! hatch when we know no more text will arrive.) -//! -//! See `cx5_chx5_newline_gate.md` in the task brief for full rationale. - -/// Holds streaming text until a newline boundary is reached. -/// -/// This is upstream of [`StreamChunker`](super::commit_tick::StreamChunker) -/// in the streaming pipeline: -/// -/// ```text -/// raw delta -> LineBuffer.push -> take_committable -> StreamChunker.push_delta -> commit tick -/// ``` -/// -/// The chunker also enforces a "drain-up-to-last-newline" rule on its pending -/// buffer, but `LineBuffer` exists as a *separate* layer so that: -/// 1. The contract is explicit and locally testable. -/// 2. Future downstream consumers (e.g. live preview that renders queued lines -/// optimistically) cannot accidentally see a partial fence. -/// 3. End-of-turn flush semantics are owned by the gate, not the policy. -#[derive(Debug, Default, Clone)] -pub struct LineBuffer { - /// Pending text not yet released because no terminating `\n` has been seen - /// since the last commit. - pending: String, -} - -impl LineBuffer { - /// Create an empty buffer. - pub fn new() -> Self { - Self::default() - } - - /// Append a raw delta. - pub fn push(&mut self, delta: &str) { - if delta.is_empty() { - return; - } - self.pending.push_str(delta); - } - - /// Return the prefix of the pending buffer up to and including the LAST - /// `\n`. Whatever follows that newline (if anything) stays buffered. - /// - /// Returns an empty string when the buffer is empty or contains no - /// newline yet — callers can treat the empty-string case as "nothing - /// committable on this push". - pub fn take_committable(&mut self) -> String { - let Some(last_nl) = self.pending.rfind('\n') else { - return String::new(); - }; - // Drain everything up to and including the last newline. The remaining - // tail (post-newline) stays in `pending` and is concatenated with the - // next `push` before the next commit decision is made. - self.pending.drain(..=last_nl).collect() - } - - /// Return whatever is left in the buffer, even if it is not newline - /// terminated. Used when the stream ends so we don't strand the final - /// partial line. - pub fn flush(&mut self) -> String { - std::mem::take(&mut self.pending) - } - - /// Whether the buffer holds any uncommitted text. - pub fn is_empty(&self) -> bool { - self.pending.is_empty() - } - - /// Length of the pending tail in bytes (testing/observability). - pub fn pending_len(&self) -> usize { - self.pending.len() - } - - /// Reset the buffer (e.g. on stream restart). - pub fn reset(&mut self) { - self.pending.clear(); - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn push_without_newline_holds_everything() { - // Cornerstone invariant: nothing escapes the gate until a newline - // terminates the line. This is what protects partial code fences - // (e.g. ``` arriving in chunk N, language tag in chunk N+1). - let mut buf = LineBuffer::new(); - buf.push("hello"); - assert_eq!(buf.take_committable(), ""); - assert_eq!(buf.pending_len(), 5); - assert!(!buf.is_empty()); - } - - #[test] - fn push_with_trailing_partial_returns_only_prefix() { - let mut buf = LineBuffer::new(); - buf.push("hello\nwo"); - assert_eq!(buf.take_committable(), "hello\n"); - // Tail is held for next call. - assert_eq!(buf.pending_len(), 2); - assert!(!buf.is_empty()); - } - - #[test] - fn next_push_is_concatenated_with_held_tail() { - let mut buf = LineBuffer::new(); - buf.push("hello\nwo"); - assert_eq!(buf.take_committable(), "hello\n"); - // The held "wo" is concatenated with "rld\n", and the whole line - // becomes committable. - buf.push("rld\n"); - assert_eq!(buf.take_committable(), "world\n"); - assert!(buf.is_empty()); - } - - #[test] - fn flush_returns_unterminated_tail() { - let mut buf = LineBuffer::new(); - buf.push("trailing without newline"); - // No newline → nothing committable. - assert_eq!(buf.take_committable(), ""); - // End-of-stream flush returns it raw. - assert_eq!(buf.flush(), "trailing without newline"); - assert!(buf.is_empty()); - } - - #[test] - fn flush_is_empty_when_buffer_drained() { - let mut buf = LineBuffer::new(); - buf.push("a\n"); - assert_eq!(buf.take_committable(), "a\n"); - assert_eq!(buf.flush(), ""); - } - - #[test] - fn multi_line_burst_returns_prefix_through_last_newline() { - // Multiple newlines in one push: the entire prefix up through the - // last newline is committable in one go; only the unterminated tail - // is held. - let mut buf = LineBuffer::new(); - buf.push("a\nb\nc\nd"); - assert_eq!(buf.take_committable(), "a\nb\nc\n"); - assert_eq!(buf.pending_len(), 1); - // Finishing "d" with a newline releases it on the next take. - buf.push("\n"); - assert_eq!(buf.take_committable(), "d\n"); - } - - #[test] - fn partial_code_fence_never_escapes_the_gate() { - // Acceptance scenario from CX#5: a fenced code block whose opener - // arrives split across deltas must never expose "foo```rust" without - // a terminating newline. We assert that on every intermediate - // commit, the *committed* text either contains a newline or is empty - // — i.e. the pre-language partial fence never leaks. - let mut buf = LineBuffer::new(); - - // Chunk 1: a paragraph fragment ending with the fence opener. - buf.push("foo```"); - let c1 = buf.take_committable(); - assert!( - c1.is_empty() || c1.ends_with('\n'), - "partial fence leaked: {c1:?}" - ); - assert!( - !c1.contains("foo```"), - "fence opener escaped without newline: {c1:?}" - ); - - // Chunk 2: language tag + start of body. The fence line is now - // newline-terminated, so it can commit; the post-newline body is - // held. - buf.push("rust\nlet x"); - let c2 = buf.take_committable(); - assert!( - c2.ends_with('\n'), - "expected newline-terminated commit: {c2:?}" - ); - assert_eq!(c2, "foo```rust\n"); - - // Chunk 3: rest of body and the fence closer. - buf.push("= 1;\n```\n"); - let c3 = buf.take_committable(); - assert_eq!(c3, "let x= 1;\n```\n"); - assert!(buf.is_empty()); - } - - #[test] - fn empty_push_is_a_noop() { - let mut buf = LineBuffer::new(); - buf.push(""); - assert!(buf.is_empty()); - assert_eq!(buf.take_committable(), ""); - } - - #[test] - fn reset_clears_pending_tail() { - let mut buf = LineBuffer::new(); - buf.push("partial"); - assert_eq!(buf.pending_len(), 7); - buf.reset(); - assert!(buf.is_empty()); - assert_eq!(buf.flush(), ""); - } -} diff --git a/docs/MCP.md b/docs/MCP.md index 577133630..a62c53bef 100644 --- a/docs/MCP.md +++ b/docs/MCP.md @@ -115,10 +115,20 @@ Supported in-TUI actions: /mcp reload ``` -`/mcp validate` and `/mcp reload` reconnect for UI discovery and refresh the -manager snapshot. Config edits made from the TUI are written immediately, but -the model-visible MCP tool pool is not hot-reloaded; the manager marks this as -restart-required until the TUI is restarted. +`/mcp validate` (alias `/mcp doctor`) reconnects for UI discovery only: it +refreshes the manager snapshot you see in the pager, not the catalog the model +gets. + +`/mcp reload` (aliases `/mcp reconnect`, `/mcp restart`) is the hot-reload path. +It re-reads the MCP config sources and reconnects through the engine-owned pool, +so the rebuilt catalog is the exact one the next model turn uses — no TUI +restart. Config edits made from the TUI are written immediately and the manager +marks the snapshot reload-required until you run it; a failed reload leaves the +previous live pool intact and says so. + +Headless surfaces are the exception: the `ConfigReload` app-server request does +**not** refresh MCP connections, so a headless runtime still needs a restart +after MCP config changes. ## Remote HTTP Auth @@ -202,8 +212,7 @@ The recommended setup path is Hugging Face's settings-generated configuration: 2. Choose the MCP client closest to your Codewhale config shape and copy the generated server snippet. 3. Paste the Hugging Face server entry into your resolved MCP config file. -4. Restart Codewhale, or run `/mcp reload` for the manager snapshot and restart - if the model-visible tool pool still needs to rebuild. +4. Run `/mcp reload` to rebuild the live model-visible tool pool. Codewhale reads both `servers` and `mcpServers`, so settings-generated snippets can be adapted without changing the rest of the MCP file. A placeholder-only @@ -253,10 +262,11 @@ Overrides: `codewhale-tui mcp init` (and `codewhale-tui setup --mcp`) writes to this resolved path. The interactive `/config` editor also exposes `mcp_config_path`. Changing it in -the TUI updates the path used by `/mcp`, and requires a restart before the -model-visible MCP tool pool is rebuilt. +the TUI updates the path used by `/mcp` and marks the pool reload-required; +`/mcp reload` then switches the live pool to the new config source. -After editing the file or changing `mcp_config_path`, restart the TUI. +After editing the MCP file or changing `mcp_config_path`, run `/mcp reload`. No +TUI restart is needed. ## Tool Naming @@ -414,5 +424,7 @@ Avoid committing literal `Authorization` headers. Prefer `env_headers`, - Run `codewhale-tui doctor` to confirm the MCP config path it resolved and whether it exists. - In the TUI, run `/mcp validate` to refresh the visible server/tool snapshot. +- If tools are missing from the model's catalog after a config or credential + change, run `/mcp reload` — `/mcp validate` only refreshes the UI snapshot. - If the MCP config is missing, run `codewhale-tui mcp init --force` to regenerate it. - If tools don’t appear, verify the server command works from your shell and that the server supports MCP `tools/list`.