github-featured/hmbown--codewhale
2
0
Fork 0
Code Issues Pull Requests Actions 16 Deployments Models Agent Notes Packages Projects Releases Wiki Activity

docs(mcp): correct the /mcp reload hot-reload claim (#4068)

Browse Source 
docs/MCP.md still said the model-visible MCP tool pool "is not hot-reloaded"
and that the manager marks it restart-required until the TUI is restarted.
That stopped being true when 7429bf440 routed `/mcp reload` through the
engine-owned pool; the code change shipped without the doc update, and #4068
was filed against the doc's description rather than the binary's behavior.

Actual behavior: `McpUiAction::Reload` calls `EngineHandle::reload_mcp`
(tui/src/tui/ui.rs:12903), which reaches `Engine::reload_mcp_pool`
(tui/src/core/engine.rs:5052) and re-reads config plus reconnects on the same
pool `mcp_tools()` builds the model catalog from. The manager pager says
"run /mcp reload to rebuild the live model-visible tool pool"
(tui/src/tui/mcp_routing.rs:13) — reload-required, never restart-required.

Also corrected the two other stale restart instructions (Hugging Face setup
step 4, `mcp_config_path` change), which the same commit had already fixed in
the binary's own messages, and recorded the one place a restart is still
required: the headless `ConfigReload` app-server request does not refresh MCP
connections.

No code change; this only makes the doc match the shipped binary.
This commit is contained in:
Hmbown
2026-08-04 00:44:55 -07:00
parent d6fa973675
commit 7d425158ce
4 changed files with 21 additions and 945 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/tui/src/tui/streaming/chunking.rs
-457
View File
@@ -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<Duration>,
}
#[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<Instant>,
last_catch_up_exit_at: Option<Instant>,
/// 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);
}
}
crates/tui/src/tui/streaming/commit_tick.rs
-256
View File
@@ -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<QueuedChunk>,
}
#[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<Duration> {
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());
}
}
crates/tui/src/tui/streaming/line_buffer.rs
-223
View File
@@ -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(), "");
}
}
docs/MCP.md
+21 -9
View File
@@ -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`.