Merge pull request #396 from changeroa/feat/adaptive-streaming-buffer

feat(coding-agent): smooth streamed response pacing

Author: code-yeongyu

packages/coding-agent/src/modes/interactive/changes.md

@@ -56,6 +56,34 @@
 - MEDIUM: `components/footer.ts` `render()` was rewritten around `FooterSegment` pairs; upstream footer layout
   changes will conflict textually. `components/footer-layout.ts` is additive.
 
+## Adaptive smooth-streaming buffer (2026-07-27)
+
+### What changed
+
+- `streaming-reveal.ts`, `streaming-reveal-pacing.ts`, and `streaming-reveal-content.ts`: smooth assistant output
+  waits for an 80ms startup buffer, estimates the provider's grapheme arrival rate with an EWMA, and follows that
+  learned base rate without a hard ceiling. Individual outlier samples are limited to four times the prior
+  estimate, extra catch-up is bounded independently, and signed backlog correction converges toward roughly
+  140ms of queued text across provider chunk cadences.
+- Fully drained bursts reset fractional progress so a later chunk cannot inherit reveal budget from an earlier
+  burst. Streaming tool arguments retain one-code-unit progress and parse in bounded 64-unit batches, preserving
+  surrogate pairs while sharing the assistant pacing helper.
+- `../../../test/streaming-reveal-{content,pacing}.test.ts`, `../../../test/streaming-reveal.test.ts`, and
+  `../../../test/helpers/streaming-reveal.ts`: split grapheme, pacing, and controller coverage into focused modules
+  and exercise timed 45/90/180/240/500-unit-per-second arrivals, multiple cadences, sustained fast streams,
+  convergence and final-tail bounds, lifecycle flushes, and drained-burst carry reset.
+
+### Why
+
+- The previous fixed 267ms catch-up policy drained each provider burst completely, while the first adaptive
+  implementation capped the total reveal rate at 240 graphemes per second. Providers above that rate accumulated
+  an unbounded tail that snapped onscreen at `message_end`; separating the learned base rate from bounded
+  correction keeps immediate completion flushes small without delaying lifecycle events.
+
+### Expected merge conflict zones
+
+- LOW: the fork-only streaming reveal modules, focused tests, and the shared pacing call in `tool-args-reveal.ts`.
+
 ## Runtime-error headline rendering (2026-07-27)
 
 ### What changed

packages/coding-agent/src/modes/interactive/streaming-reveal-content.ts

@@ -0,0 +1,152 @@
+import type { AssistantMessage } from "@earendil-works/pi-ai";
+import { getGraphemeSegmenter } from "@earendil-works/pi-tui";
+
+type AssistantContentBlock = AssistantMessage["content"][number];
+export type GraphemeCounter = (index: number, text: string) => number;
+export type GraphemeSlicer = (index: number, text: string, units: number) => string;
+
+function countGraphemesFrom(text: string, start: number): { count: number; tailStart: number } {
+	let count = 0;
+	let tailStart = start;
+	for (const segment of getGraphemeSegmenter().segment(start === 0 ? text : text.slice(start))) {
+		count += 1;
+		tailStart = start + segment.index;
+	}
+	return { count, tailStart };
+}
+
+function segmentFrom(text: string, start: number, clusters: number): { count: number; end: number; lastStart: number } {
+	let count = 0;
+	let end = start;
+	let lastStart = start;
+	for (const segment of getGraphemeSegmenter().segment(start === 0 ? text : text.slice(start))) {
+		count += 1;
+		lastStart = start + segment.index;
+		end = lastStart + segment.segment.length;
+		if (count >= clusters) break;
+	}
+	return { count, end, lastStart };
+}
+
+export class BlockUnitCounter {
+	readonly #entries = new Map<number, { text: string; count: number; tailStart: number }>();
+	readonly #sliceEntries = new Map<number, { text: string; units: number; end: number; lastStart: number }>();
+
+	count(index: number, text: string): number {
+		const entry = this.#entries.get(index);
+		if (entry !== undefined) {
+			if (entry.text === text) return entry.count;
+			if (entry.count > 0 && text.length > entry.text.length && text.startsWith(entry.text)) {
+				const tail = countGraphemesFrom(text, entry.tailStart);
+				const next = { text, count: entry.count - 1 + tail.count, tailStart: tail.tailStart };
+				this.#entries.set(index, next);
+				return next.count;
+			}
+		}
+		const full = countGraphemesFrom(text, 0);
+		this.#entries.set(index, { text, count: full.count, tailStart: full.tailStart });
+		return full.count;
+	}
+
+	slice(index: number, text: string, units: number): string {
+		const wholeUnits = Math.floor(units);
+		if (wholeUnits <= 0 || text.length === 0) return "";
+		const entry = this.#sliceEntries.get(index);
+		if (entry?.text === text && entry.units === wholeUnits) {
+			return entry.end >= text.length ? text : text.slice(0, entry.end);
+		}
+		if (entry !== undefined && (entry.text === text || text.startsWith(entry.text)) && wholeUnits >= entry.units) {
+			const segment = segmentFrom(text, entry.lastStart, wholeUnits - entry.units + 1);
+			this.#sliceEntries.set(index, {
+				text,
+				units: entry.units - 1 + segment.count,
+				end: segment.end,
+				lastStart: segment.lastStart,
+			});
+			return segment.end >= text.length ? text : text.slice(0, segment.end);
+		}
+		const segment = segmentFrom(text, 0, wholeUnits);
+		this.#sliceEntries.set(index, {
+			text,
+			units: segment.count,
+			end: segment.end,
+			lastStart: segment.lastStart,
+		});
+		return segment.end >= text.length ? text : text.slice(0, segment.end);
+	}
+
+	reset(): void {
+		this.#entries.clear();
+		this.#sliceEntries.clear();
+	}
+}
+
+function countGraphemes(text: string): number {
+	return countGraphemesFrom(text, 0).count;
+}
+
+function sliceGraphemes(text: string, units: number): string {
+	if (units <= 0 || text.length === 0) return "";
+	const segment = segmentFrom(text, 0, units);
+	return segment.end >= text.length ? text : text.slice(0, segment.end);
+}
+
+export function countVisibleUnits(message: AssistantMessage, hideThinking: boolean, countOf: GraphemeCounter): number {
+	let total = 0;
+	for (let index = 0; index < message.content.length; index++) {
+		const block = message.content[index];
+		if (block?.type === "text") {
+			total += countOf(index, block.text);
+		} else if (block?.type === "thinking" && !hideThinking) {
+			total += countOf(index, block.thinking);
+		}
+	}
+	return total;
+}
+
+export function visibleUnits(message: AssistantMessage, hideThinking: boolean): number {
+	return countVisibleUnits(message, hideThinking, (_index, text) => countGraphemes(text));
+}
+
+export function buildDisplayMessage(
+	target: AssistantMessage,
+	revealed: number,
+	hideThinking: boolean,
+	countOf: GraphemeCounter = (_index, text) => countGraphemes(text),
+	sliceOf: GraphemeSlicer = (_index, text, units) => sliceGraphemes(text, units),
+): AssistantMessage {
+	let remaining = Math.max(0, Math.floor(revealed));
+	const content: AssistantContentBlock[] = [];
+	for (let index = 0; index < target.content.length; index++) {
+		const block = target.content[index];
+		if (!block) continue;
+		if (block.type === "text") {
+			const units = countOf(index, block.text);
+			content.push(
+				remaining <= 0
+					? block.text.length === 0
+						? block
+						: { ...block, text: "" }
+					: remaining >= units
+						? block
+						: { ...block, text: sliceOf(index, block.text, remaining) },
+			);
+			remaining = Math.max(0, remaining - units);
+		} else if (block.type === "thinking" && !hideThinking) {
+			const units = countOf(index, block.thinking);
+			content.push(
+				remaining <= 0
+					? block.thinking.length === 0
+						? block
+						: { ...block, thinking: "" }
+					: remaining >= units
+						? block
+						: { ...block, thinking: sliceOf(index, block.thinking, remaining) },
+			);
+			remaining = Math.max(0, remaining - units);
+		} else {
+			content.push(block);
+		}
+	}
+	return { ...target, content };
+}

packages/coding-agent/src/modes/interactive/streaming-reveal-pacing.ts

@@ -0,0 +1,32 @@
+export const INITIAL_BUFFER_MS = 80;
+export const TARGET_BUFFER_MS = 140;
+export const CATCHUP_WINDOW_MS = 267;
+export const MIN_ARRIVAL_UNITS_PER_SEC = 45;
+export const MAX_ARRIVAL_SAMPLE_MULTIPLIER = 4;
+export const MAX_EXTRA_CATCHUP_UNITS_PER_SEC = 600;
+export const ARRIVAL_RATE_ALPHA = 0.25;
+export const MIN_SMOOTH_FPS = 30;
+export const MAX_SMOOTH_FPS = 120;
+export const DEFAULT_SMOOTH_FPS = 60;
+
+export function updateArrivalRate(currentRate: number, appendedUnits: number, elapsedMs: number): number {
+	if (appendedUnits <= 0 || elapsedMs <= 0) return currentRate;
+	const sampleRate = (appendedUnits * 1000) / elapsedMs;
+	const maximumSample = Math.max(MIN_ARRIVAL_UNITS_PER_SEC, currentRate * MAX_ARRIVAL_SAMPLE_MULTIPLIER);
+	const boundedSample = Math.min(maximumSample, Math.max(MIN_ARRIVAL_UNITS_PER_SEC, sampleRate));
+	return currentRate + ARRIVAL_RATE_ALPHA * (boundedSample - currentRate);
+}
+
+export function nextStep(backlog: number, dtMs: number, arrivalRate = 90): number {
+	if (backlog <= 0) return 0;
+	const dt = Math.min(Math.max(dtMs, 1), 100);
+	const baseArrivalRate = Math.max(MIN_ARRIVAL_UNITS_PER_SEC, arrivalRate);
+	const targetBacklog = (baseArrivalRate * TARGET_BUFFER_MS) / 1000;
+	const backlogError = backlog - targetBacklog;
+	const correctionRate = Math.min(
+		MAX_EXTRA_CATCHUP_UNITS_PER_SEC,
+		Math.max(-baseArrivalRate, backlogError * (1000 / CATCHUP_WINDOW_MS)),
+	);
+	const revealRate = Math.max(0, baseArrivalRate + correctionRate);
+	return Math.min(backlog, (revealRate * dt) / 1000);
+}