diff --git a/CHANGELOG.md b/CHANGELOG.md index 528bf4a82..4c240118b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -80,6 +80,7 @@ This release upgrades the engine and hardens the terminal agent. The three engin - Restored the streaming display coalescer. The raw engine wrapper that precedes every derived text and thinking delta was classified as a synchronous event, so each provider chunk cancelled the pending 16 ms display frame and forced an immediate render request; the coalescer was defeated exactly while a response streamed, which is the one time it exists for. Wrappers carrying text or thinking deltas, which the panel has always ignored, now stop at the renderer instead of reaching it; tool-call formation keeps its synchronous path, and every event the panel acts on arrives in the same order as before. - Enabled Node's V8 compile cache for the boot-path module graphs: interactive chat, `run`, `acp`, and native fleet workers. Locally spawned workers inherit the directory through their spawn environment and consume it before any of their children can see it; SSH-placed workers enable the remote install's own cache in-process at the worker entry, with nothing exported to their children. Read-only and dry-run commands never enable it, `paths` and bare `doctor` keep their promise that nothing is created, and the cache enables only on an initialized install so a home Clio never set up stays untouched. `NODE_COMPILE_CACHE` and `NODE_DISABLE_COMPILE_CACHE` always win, and a cache failure never affects a command. The import-only and real-PTY fill/hit observations, with their deliberately narrower endpoint names, are recorded in `docs/performance-methodology.md` rather than generalized into a per-command saving. - Corrected boot and TUI performance instrumentation before using it for 0.3.2 decisions. `first TUI paint` is now marked only after the first real frame has issued all of its stdout writes; explicit frame ids group diff, ANSI, hardware-cursor, and IME writes; canonical event and input ids correlate queue, panel, and committed-frame high-water marks; and stdout return, backpressure, and drain are recorded. The trace writer is bounded, asynchronous, awaited at shutdown, and nonfatal on storage failure. A deterministic contract plus a real built-CLI PTY harness cover first frame, input, resize, grouped writes, paused output, and process reaping. The documented metrics are stdout/PTY endpoints, never an in-process claim of literal glass latency. +- Added the adaptive stream pacer as `terminal.smoothStreaming: off | auto | on`, with the conservative 0.3.2 default left at `off`. Only derived visible text and thinking are paced, through one ordered generation queue; raw text/thinking wrappers remain transparent and public events, transcript persistence, tool formation, cumulative tool state, and ordered boundaries remain synchronous. Pacing is grapheme-safe, arrival-credit and oldest-age bounded, self-stopping, folded-thinking aware, and forced to settle before abort, retry, submit, mode change, final return, or teardown. Fullscreen frozen scrolling and resize survive paced updates, and stdout backpressure gates frame construction rather than creating a second unbounded SSH buffer. `auto` bypasses non-TTY, remote/multiplexed, CI, accessibility-marked, and backpressured sessions; `CLIO_CODER_SMOOTH_STREAM=off` is the immediate per-process escape hatch. Deterministic fake-clock, ordering, grapheme, reset, folded-thinking, scroll/resize, no-drain, and built-CLI PTY contracts cover the rollout. - Removed the eager userland Undici graph from `web_fetch` and use the Node 22.19+ built-in Fetch, Headers, Request, Response, stream, and abort implementation. Localhost contracts preserve request headers/body, redirects, UTF-8 streaming and cancellation at the byte ceiling, external abort, timeout, HTTP previews, binary rejection, and transport errors on both supported Node lines; an installed-tarball turn invokes the real tool from a foreign working directory. The built graph contains no Undici source or dependency, while the measured size and deliberately mixed import-time observations are recorded without turning host noise into a boot claim. - Made codewiki's tree-sitter graph genuinely lazy and moved runtime indexing off the interactive event loop. Lightweight schema, artifact, and path modules preserve synchronous cached reads without evaluating the builder; actual full, stale, and incremental builds run in a dedicated worker and load only the required grammars. Session startup, parallel `code_nav` demand, mutation batches, explicit index/refresh, bootstrap, wiki grounding, and reset now share one per-workspace generation queue and cross-process lease, so an older build cannot overwrite newer state or resurrect a reset artifact, and shutdown drains admitted work. Built-source and installed-tarball coverage prove nested help is tree-sitter-free and write-free while a real foreign-cwd build loads the runtime and its vendored grammar; measurements and deliberately scoped import observations are in `docs/performance-methodology.md`. - Split `context`, `code_nav`, `verify`, `web_fetch`, `dispatch`, `monitor`, and `steer` into immutable lightweight tool surfaces and first-use implementation chunks. Registration order, provider schemas and descriptions, policy metadata, execution modes, argument normalization, admission, permissions, middleware, result shaping, and worker surface attestation remain registry-owned and eager; only an admitted runner imports code. Dispatch keeps its trusted plan and capacity-reservation identities in one synchronous admission controller shared with the lazy runner, deeply freezes every execution-affecting control before middleware or approval can observe it, binds the `apply_winner` repository destination into the approval hash and text, and releases a guard-blocked prepared admission exactly once. Workers import only the core tool bootstrap, so their built entry never evaluates the three orchestrator-only runners. Concurrent first calls share one import, an implementation whose surface drifted fails closed, unrelated tools remain absent, and a missing Clio-owned chunk carries the same named reinstall guidance as a missing command chunk. V8 coverage repeats the absent-before-use/present-on-invocation proof for every tool and the worker exclusion against both the source build and an installed tarball from a foreign working directory. diff --git a/docs/architecture.md b/docs/architecture.md index f295c8d84..f61094e13 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -272,6 +272,18 @@ decoding and before the application controller mutates editor, overlay, scroll, or submit state. The first frame whose input high water includes that id is the input-to-stdout-commit endpoint. +Adaptive streaming remains inside that presentation boundary. One semantic +classifier drops only transparent raw text/thinking mirrors, sends derived +visible content through one generation/epoch FIFO, and treats every other +transcript mutation as an ordered drain boundary. Pacer slices mutate the +panel directly and are never re-emitted on the public bus, so session storage, +replay/export, tool-call formation, and cumulative tool-result behavior keep +their canonical synchronous inputs. Abort, retry, interrupt, submit, mode +change, and teardown drain the queue and can await the containing committed +frame. The stdout gate stops later frame construction after a false write and +coalesces to current model state until `drain`; it is not installed for the +default `off` path and therefore cannot become a second unbounded SSH buffer. + Tracing is opt-in and content-free. Its bounded asynchronous writer never does filesystem append I/O on the render stack, and shutdown awaits a bounded flush. See [performance-methodology.md](performance-methodology.md) for vocabulary, diff --git a/docs/configuration-and-targets.md b/docs/configuration-and-targets.md index 847555f90..a91ceb0d5 100644 --- a/docs/configuration-and-targets.md +++ b/docs/configuration-and-targets.md @@ -207,6 +207,7 @@ terminal: outputVerbosity: default tuiMode: regular # regular terminal scrollback or fullscreen sticky layout fullscreenScrollbar: auto # hidden, auto, or always in fullscreen mode + smoothStreaming: off # off, conservative auto, or explicit on skills: trustProjectCompatRoots: false delegation: @@ -446,7 +447,7 @@ The Settings Center organizes all configuration under four non-selectable group | **RUNTIME** | Budget (`budget`) | `budget.sessionCeilingUsd`, `defaults.maxTokens`, and `budget.concurrency` (restart required). | | **RUNTIME** | Compaction (`compaction`) | `compaction.auto`, `compaction.threshold`, and `compaction.excludeLastTurns`. | | **RUNTIME** | Retry (`retry`) | `retry.enabled`, `retry.maxRetries`, `retry.baseDelayMs`, and `retry.maxDelayMs`. | -| **EXPERIENCE** | Terminal (`terminal`) | `terminal.showTerminalProgress`, `terminal.outputVerbosity` (`minimal`, `default`, `verbose`), `terminal.tuiMode` (`regular`, `fullscreen`), `terminal.fullscreenScrollbar` (`hidden`, `auto`, `always`), and `theme`. | +| **EXPERIENCE** | Terminal (`terminal`) | `terminal.showTerminalProgress`, `terminal.outputVerbosity` (`minimal`, `default`, `verbose`), `terminal.tuiMode` (`regular`, `fullscreen`), `terminal.fullscreenScrollbar` (`hidden`, `auto`, `always`), `terminal.smoothStreaming` (`off`, `auto`, `on`), and `theme`. | | **EXPERIENCE** | Advanced (`advanced`) | `runtimePlugins`, `compaction.model`, `compaction.systemPrompt`, `delegation.defaults.connectTimeoutMs`, `delegation.defaults.turnTimeoutMs`, `delegation.defaults.permissionTimeoutMs`, `keybindings`, and `delegation.agents`. | `retry.streamStallMs` has no Settings Center row; edit it in `settings.yaml`. @@ -495,6 +496,7 @@ Label to config path mapping: | Output detail | `terminal.outputVerbosity` (`minimal`, `default`, or `verbose`) | | TUI mode | `terminal.tuiMode` (`regular` or `fullscreen`, restart required) | | Fullscreen scrollbar | `terminal.fullscreenScrollbar` (`hidden`, `auto`, or `always`, restart required) | +| Smooth streaming | `terminal.smoothStreaming` (`off`, `auto`, or `on`, live) | | Theme | `theme` | | Runtime plugins | `runtimePlugins` | | Compaction model | `compaction.model` | @@ -608,6 +610,7 @@ Generic provider and transport errors are classified by transient retry rules, i | `terminal.outputVerbosity` | `default` | `minimal`, `default`, `verbose` | immediately | | `terminal.tuiMode` | `regular` | `regular`, `fullscreen` | restart | | `terminal.fullscreenScrollbar` | `auto` | `hidden`, `auto`, `always` | restart | +| `terminal.smoothStreaming` | `off` | `off`, `auto`, `on` | immediately | | `modelSelector.favorites` | `[]` | list of strings | immediately | | `modelSelector.recentLimit` | `12` | integer ≥ 1 | immediately | | `keybindings` | `{}` | map of binding id to a key string or list of them | restart | diff --git a/docs/environment-variables.md b/docs/environment-variables.md index 482b5085e..cb8beede3 100644 --- a/docs/environment-variables.md +++ b/docs/environment-variables.md @@ -42,6 +42,9 @@ Durable values live in the `guardrails:` section of settings.yaml (see [configur | `CLIO_CODER_SKILL_MARKETPLACE_INDEX` | unset | Skill-marketplace index path override (`src/domains/resources/skills/marketplace.ts`). | | `CLIO_CODER_MODEL_CATALOG_DIRS` | unset | Extra model-catalog directories (`src/domains/providers/knowledge-base-path.ts`). | | `CLIO_CODER_NO_NETWORK_TOOLS` | off | `1` strips network tools from every registry in the process; the skills-eval harness sets it for hermetic arms; `--allow-network` clears it (`src/tools/network-policy.ts`). | +| `CLIO_CODER_SMOOTH_STREAM` | settings value | Per-process override for `terminal.smoothStreaming`: `0`/`off`/`false`, `auto`, or `1`/`on`/`true`. A valid value wins over settings; an invalid value fails safely to `off`. | +| `CLIO_CODER_REDUCE_MOTION` | off | `1` makes smooth-streaming `auto` use the immediate coalescer. Explicit `on` remains an operator request, while stdout backpressure still pauses frame production. | +| `CLIO_CODER_SCREEN_READER` | off | `1` makes smooth-streaming `auto` use the immediate coalescer so a screen reader receives the existing low-motion update behavior. | ## Directory and install layout diff --git a/docs/installation-and-lifecycle.md b/docs/installation-and-lifecycle.md index 153f7c324..ca736dfde 100644 --- a/docs/installation-and-lifecycle.md +++ b/docs/installation-and-lifecycle.md @@ -232,7 +232,7 @@ clio-coder upgrade Key lifecycle and operational updates in v0.3.2: - Upgraded the underlying engine SDK libraries to 0.84.0 with signal-aware OAuth cancellation. - Hardened migration resilience: damaged `credentials.yaml` files no longer block upgrades when no renames are needed (#121); `--skip-migrations` is available as a recovery override. -- Fullscreen TUI mode (`terminal.tuiMode`, `terminal.fullscreenScrollbar`) is available via Settings → Terminal (restart required). +- Fullscreen TUI mode (`terminal.tuiMode`, `terminal.fullscreenScrollbar`) is available via Settings → Terminal (restart required). Adaptive presentation pacing is the live `terminal.smoothStreaming` setting; 0.3.2 defaults it to `off`, with conservative `auto` and explicit `on` available from the same section. - Turn settlement is enforced on `/new`, `/resume`, `/tree`, and `/fork` to cleanly commit in-flight streams before session writer replacement (#114). - Resumed and forked session entry replays standardize message prefixes through `src/engine/messages.ts`. - `AI_AGENT=clio-coder` is set on all child processes for system attribution. diff --git a/docs/performance-methodology.md b/docs/performance-methodology.md index 916d6250c..d87a3fc19 100644 --- a/docs/performance-methodology.md +++ b/docs/performance-methodology.md @@ -56,10 +56,12 @@ node dist/cli/index.js The deterministic contracts are `tests/contracts/render-pipeline-trace.test.ts`. The real built-CLI acceptance harness is `tests/smoke/render-trace-pty.test.ts`; it covers first frame, input -correlation, resize, grouped writes, paused PTY output, and bounded process -cleanup. Set `CLIO_CODER_PERF_REPORT=1` while running that test to print its -observation record. Real PTY acceptance is currently unavailable on Windows; -the fake-stream frame/backpressure contract remains cross-platform. +correlation, resize, grouped writes, paused PTY output, adaptive pacing against +a chunked hermetic provider, a controlled `stdout.write() === false`/`drain` +boundary, final-frame settlement, and bounded process cleanup. Set +`CLIO_CODER_PERF_REPORT=1` while running that test to print its observation +records. Real PTY acceptance is currently unavailable on Windows; the +fake-stream frame/backpressure contracts remain cross-platform. ## Import-graph method @@ -367,6 +369,52 @@ justified by the deterministic absence/presence and worker-exclusion contracts, the smaller evaluated graph, exact registration-order and surface contracts, and the dispatch reservation, approval, gate, detach, monitor, and steer suites. +## Adaptive stream-pacer observations + +`terminal.smoothStreaming` is presentation-only. `off` is the exact existing +16 ms coalescer and remains the 0.3.2 default. `auto` uses the pacer only on a +capable local TTY with no accessibility, remote/multiplexer, CI, or observed +backpressure signal. `on` requests pacing, but frame construction still stops +behind stdout backpressure. The pacer never republishes slices on the public +event bus: canonical events, persistence, replay/export, tool formation, and +cumulative tool state remain synchronous while one presentation queue owns +only derived visible text/thinking mutations. + +The deterministic fake-clock contracts cover semantic classification, FIFO +generation/epoch ordering, abort and stale-admission rejection, grapheme +clusters, fractional arrival credit, event-loop suspension, catch-up, the +oldest-visible deadline, absolute queue byte/grapheme bounds, idle shutdown, +folded-thinking fidelity, reset/discard accounting, mode changes, final-frame +settlement, fullscreen frozen scrolling through resize, and bounded no-drain +cleanup. The PTY arm uses a built CLI, a four-delta localhost provider, a +4 KiB reply, an 80x24 `xterm-256color` PTY whose reader is paused, and a +test-only writable shim that makes exactly one real child `stdout.write()` +return `false` before emitting a delayed `drain`. This is deterministic +backpressure acceptance, not a claim about a particular SSH kernel buffer. + +Five independent processes per supported Node line were measured on the same +2026-08-19 WSL2 host as the corrected baseline, with the operating-system page +cache warm and V8 compile caching disabled. Values are median / largest of the +five observations; they are diagnostic observations, not timing gates. + +| Node | Input-to-stdout commit | First ingress-to-stdout commit | Final ingress-to-stdout commit | Controlled backpressure wait | +| --- | ---: | ---: | ---: | ---: | +| 22.22.3 | 5.530 / 8.896 ms | 62.909 / 70.274 ms | 37.798 / 56.412 ms | 403.809 / 426.570 ms | +| 24.9.0 | 5.468 / 7.225 ms | 67.878 / 74.021 ms | 27.901 / 35.750 ms | 402.582 / 418.783 ms | + +The command was: + +```bash +NODE_DISABLE_COMPILE_CACHE=1 CLIO_CODER_PERF_REPORT=1 \ + node --import tsx --import ./tests/harness/tmp-root.ts --test \ + --test-name-pattern 'paces provider deltas' \ + tests/smoke/render-trace-pty.test.ts +``` + +These endpoints end at stdout commit. Even the PTY reader assertion stops at +the pseudo-terminal boundary; none of these values is literal token-to-glass +latency. + ## Reporting checklist Every published observation records: diff --git a/docs/tui-design.md b/docs/tui-design.md index a605d4212..4aef629aa 100644 --- a/docs/tui-design.md +++ b/docs/tui-design.md @@ -152,6 +152,8 @@ The Clio screen maintains a responsive, four-zone structure: the launchpad / ses In fullscreen mode, `PageUp` and `PageDown` scroll one viewport, `Home` and `End` jump to its bounds, `Ctrl+Shift+Up` and `Ctrl+Shift+Down` jump between semantic prompts, and the mouse wheel scrolls the transcript. Dragging the scrollbar thumb moves the viewport directly. `terminal.fullscreenScrollbar` is `hidden`, `auto` (visible during interaction), or `always`. Manual scrolling suspends follow-end so new output does not steal the operator's position; returning to the bottom resumes it. Both fullscreen settings are restart-scoped because Clio constructs its terminal renderer and component graph once at startup. +`terminal.smoothStreaming` controls presentation-only pacing of derived assistant text and thinking. `off`, the 0.3.2 release default, is the existing immediate 16 ms coalescer. `auto` paces only on a capable local TTY and bypasses pacing for non-TTY, SSH, multiplexers, CI, screen-reader/reduced-motion markers, or observed stdout backpressure. `on` explicitly requests grapheme-safe pacing, while still stopping frame production behind stdout backpressure. Raw provider wrappers never enter the panel, canonical events and persistence remain synchronous, and tool/message/turn/abort/retry/submit/teardown boundaries drain visible state before they continue. `CLIO_CODER_SMOOTH_STREAM` is the one-process escape hatch and takes precedence over settings; invalid values resolve to `off`. + ### 5.1 Welcome Launchpad & Session Header - **Pre-Submit Launchpad**: Before the first prompt, renders a compact launchpad at line 0 with bold CAPS section tags (`WORKSPACE`, `ROUTE`, `NEXT`), honest readiness state, and a context-sensitive next action (e.g. `ctx missing · /context init`, `ctx checking…`, or `ctx ready · type a task`). Asynchronous repository probes use height-stable dim placeholders (`factsPending`). diff --git a/src/core/config.ts b/src/core/config.ts index 01dfb5794..e248b3a70 100644 --- a/src/core/config.ts +++ b/src/core/config.ts @@ -1315,6 +1315,7 @@ export function validateSettings(raw: unknown): SettingsValidationResult { "outputVerbosity", "tuiMode", "fullscreenScrollbar", + "smoothStreaming", ]); if ("showTerminalProgress" in raw.terminal) { const v = expectBoolean(issues, "terminal.showTerminalProgress", raw.terminal.showTerminalProgress); @@ -1335,6 +1336,11 @@ export function validateSettings(raw: unknown): SettingsValidationResult { if (v === "hidden" || v === "auto" || v === "always") settings.terminal.fullscreenScrollbar = v; else if (v !== undefined) issues.add("terminal.fullscreenScrollbar", "expected hidden, auto, or always"); } + if ("smoothStreaming" in raw.terminal) { + const v = expectString(issues, "terminal.smoothStreaming", raw.terminal.smoothStreaming); + if (v === "off" || v === "auto" || v === "on") settings.terminal.smoothStreaming = v; + else if (v !== undefined) issues.add("terminal.smoothStreaming", "expected off, auto, or on"); + } } } diff --git a/src/core/defaults.ts b/src/core/defaults.ts index 9c6d99d77..0864aaa62 100644 --- a/src/core/defaults.ts +++ b/src/core/defaults.ts @@ -115,6 +115,7 @@ export interface RetrySettings { export type OutputVerbosity = "minimal" | "default" | "verbose"; export type TuiMode = "regular" | "fullscreen"; export type FullscreenScrollbar = "hidden" | "auto" | "always"; +export type SmoothStreaming = "off" | "auto" | "on"; export interface TerminalSettings { showTerminalProgress: boolean; @@ -124,6 +125,8 @@ export interface TerminalSettings { tuiMode: TuiMode; /** Fullscreen transcript scrollbar visibility. */ fullscreenScrollbar: FullscreenScrollbar; + /** Presentation-only pacing for streamed assistant text and thinking. */ + smoothStreaming: SmoothStreaming; } export interface ModelSelectorSettings { @@ -322,6 +325,7 @@ export const DEFAULT_SETTINGS = { outputVerbosity: "default", tuiMode: "regular", fullscreenScrollbar: "auto", + smoothStreaming: "off", } as TerminalSettings, skills: { trustProjectCompatRoots: false, @@ -538,6 +542,10 @@ terminal: tuiMode: regular # hidden, auto (visible while scrolling), or always in fullscreen mode. fullscreenScrollbar: auto + # off preserves immediate 16ms coalescing; auto paces only on a capable, + # accessibility-safe local TTY; on requests pacing but still honors stdout + # backpressure. CLIO_CODER_SMOOTH_STREAM overrides this for one process. + smoothStreaming: off # Skills are local prompt resources. Project-local compatibility roots such as # .agents/skills, .claude/skills, .codex/skills, .github/skills, and diff --git a/src/domains/config/classify.ts b/src/domains/config/classify.ts index f8e46cdd5..bd6ae1a0d 100644 --- a/src/domains/config/classify.ts +++ b/src/domains/config/classify.ts @@ -21,7 +21,13 @@ export interface ConfigDiff { restartRequired: string[]; } -const HOT_RELOAD_FIELDS = new Set(["theme", "keybindings", "autonomy", "modelSelector"]); +const HOT_RELOAD_FIELDS = new Set([ + "theme", + "keybindings", + "autonomy", + "modelSelector", + "terminal.smoothStreaming", +]); const NEXT_TURN_FIELDS = new Set([ "identity", diff --git a/src/engine/instrumented-tui.ts b/src/engine/instrumented-tui.ts index b32a73445..4ba40a391 100644 --- a/src/engine/instrumented-tui.ts +++ b/src/engine/instrumented-tui.ts @@ -21,17 +21,64 @@ export interface TuiRenderObserver { endPhase(frame: unknown, phase: TuiRenderPhase, phaseToken: unknown): void; } +export interface TuiRenderAdmission { + readonly blocked: boolean; + onWritable(listener: () => void): () => void; +} + +class DeferredRenderAdmission { + private pending = false; + private force = false; + private release: (() => void) | null = null; + + constructor( + private readonly admission: TuiRenderAdmission | undefined, + private readonly render: (force: boolean) => void, + ) {} + + request(force: boolean): boolean { + if (!this.admission?.blocked) return false; + this.pending = true; + this.force ||= force; + this.release ??= this.admission.onWritable(() => { + this.release?.(); + this.release = null; + if (!this.pending) return; + const pendingForce = this.force; + this.pending = false; + this.force = false; + this.render(pendingForce); + }); + return true; + } + + /** A direct/final render supersedes any older request held behind the gate. */ + settled(): void { + this.pending = false; + this.force = false; + this.release?.(); + this.release = null; + } +} + export class InstrumentedTuiMainScreen extends TuiMainScreen { constructor( terminal: Terminal, private readonly renderObserver: TuiRenderObserver, showHardwareCursor?: boolean, logDirectory?: string, + renderAdmission?: TuiRenderAdmission, ) { super(terminal, showHardwareCursor, logDirectory); + this.deferredAdmission = new DeferredRenderAdmission(renderAdmission, (force) => super.requestRender(force)); + } + + override requestRender(force = false): void { + if (!this.deferredAdmission.request(force)) super.requestRender(force); } protected override doRender(): void { + this.deferredAdmission.settled(); if (this.renderObserver.isEnabled?.() === false) { super.doRender(); return; @@ -75,6 +122,7 @@ export class InstrumentedTuiMainScreen extends TuiMainScreen { } private currentFrame: unknown | undefined; + private readonly deferredAdmission: DeferredRenderAdmission; } export class InstrumentedTuiAltScreen extends TuiAltScreen { @@ -86,11 +134,18 @@ export class InstrumentedTuiAltScreen extends TuiAltScreen { showHardwareCursor?: boolean, logDirectory?: string, options?: TuiAltScreenOptions, + renderAdmission?: TuiRenderAdmission, ) { super(terminal, showHardwareCursor, logDirectory, options); + this.deferredAdmission = new DeferredRenderAdmission(renderAdmission, (force) => super.requestRender(force)); + } + + override requestRender(force = false): void { + if (!this.deferredAdmission.request(force)) super.requestRender(force); } protected override doRender(): void { + this.deferredAdmission.settled(); if (this.renderObserver.isEnabled?.() === false) { super.doRender(); return; @@ -132,4 +187,6 @@ export class InstrumentedTuiAltScreen extends TuiAltScreen { this.renderObserver.endPhase(frame, phase, phaseToken); } } + + private readonly deferredAdmission: DeferredRenderAdmission; } diff --git a/src/interactive/application-controller.ts b/src/interactive/application-controller.ts index 465ea466b..81e1394c6 100644 --- a/src/interactive/application-controller.ts +++ b/src/interactive/application-controller.ts @@ -112,6 +112,8 @@ export interface ApplicationControllerDeps { toggleLastThinking: () => boolean; toggleAllThinking: () => boolean; shutdownDisposers: ReadonlyArray<() => void>; + /** Settle the last presentation mutation into an accepted/drained frame. */ + beforeStopUi?: () => Promise; stopUi: () => void; cancelParkedCalls: (reason: string) => void; onShutdown: () => Promise; @@ -132,7 +134,7 @@ export interface ApplicationControllerDeps { * hidden, the last private-mode sequence on the wire being `?25l`, where * `/quit` on the same build ended with `?25h`. */ - registerTerminalTeardown?: (teardown: () => void) => void; + registerTerminalTeardown?: (teardown: () => void | Promise) => void; } export type ApplicationInputResult = { consume: true } | undefined; @@ -255,6 +257,13 @@ export function createApplicationController(deps: ApplicationControllerDeps): Ap release("owned interval", () => deps.intervals.clearInterval(interval)); } for (const dispose of deps.shutdownDisposers) release("shutdown disposer", dispose); + if (deps.beforeStopUi) { + try { + await deps.beforeStopUi(); + } catch (error) { + failures.push({ step: "final frame", error }); + } + } release("terminal stop", () => deps.stopUi()); release("parked calls", () => deps.cancelParkedCalls("Clio Coder shutting down")); // Reported here rather than as each step fails: the terminal is down by @@ -396,7 +405,14 @@ export function createApplicationController(deps: ApplicationControllerDeps): Ap // Every exit gives the terminal back, including the ones that never reach // this controller. Stopping an already-stopped terminal is a no-op, so the // ordinary path running both is harmless. - deps.registerTerminalTeardown?.(() => { + deps.registerTerminalTeardown?.(async () => { + if (deps.beforeStopUi) { + try { + await deps.beforeStopUi(); + } catch (error) { + failures.push({ step: "final frame", error }); + } + } release("terminal stop", () => deps.stopUi()); reportFailures(); }); diff --git a/src/interactive/chat-panel.ts b/src/interactive/chat-panel.ts index 554897c10..6ca67bb01 100644 --- a/src/interactive/chat-panel.ts +++ b/src/interactive/chat-panel.ts @@ -279,6 +279,8 @@ export interface ChatPanel extends Component { */ toggleLastThinking(): boolean; toggleAllThinking(): boolean; + /** Current panel-level live-thinking visibility used by presentation pacing. */ + isThinkingExpanded(): boolean; /** Toggle whether expanded live tool bodies include cumulative partial output. */ toggleLiveToolOutput(): boolean; /** Clears the visible transcript. /new uses this after rotating the session. */ @@ -1450,6 +1452,7 @@ export function createChatPanel(options: ChatPanelOptions = {}): ChatPanel { markDirty(); return true; }, + isThinkingExpanded: () => thinkingExpanded, toggleLiveToolOutput(): boolean { liveToolOutput = !liveToolOutput; markDirty(); diff --git a/src/interactive/chat-renderer.ts b/src/interactive/chat-renderer.ts index f30bd09be..fad59e2eb 100644 --- a/src/interactive/chat-renderer.ts +++ b/src/interactive/chat-renderer.ts @@ -46,6 +46,13 @@ import { renderCompactionSummaryEntry } from "./renderers/compaction-summary.js" import { styleTaggedNotice } from "./renderers/notice.js"; import { formatRetryStatus } from "./renderers/retry-status.js"; import { renderBashTranscriptExecution, renderToolResultOnly } from "./renderers/tool-execution.js"; +import { + classifyStreamEvent, + createStreamPacer, + type SmoothStreamingMode, + type StreamPacer, + type StreamPacerSlice, +} from "./stream-pacer.js"; import { readWorkerReceiptFactsForReplay } from "./worker-receipts.js"; import { workerEntriesFromRunEntries } from "./worker-replay.js"; import type { WorkerReceiptReader } from "./worker-stream.js"; @@ -89,18 +96,35 @@ export interface CreateCoalescingChatRendererDeps { setTimer?: (cb: () => void, ms: number) => unknown; /** Override for tests. Mirrors the clearTimeout signature. */ clearTimer?: (id: unknown) => void; + /** Monotonic clock shared with the pacer; injectable for deterministic tests. */ + now?: () => number; /** Sequence captured at canonical projection ingress before this panel consumer runs. */ visibleEventSequence?: (event: ChatLoopEvent) => number | null; onQueue?: (eventSeq: number, action: "admit" | "dequeue") => void; onPanelApplied?: (eventSeq: number) => void; /** Legacy aggregate callback retained for non-text cumulative tool-update observations. */ onDelta?: () => void; + /** Canonical presentation-ingress identity; absent keeps the exact legacy coalescer. */ + streamIngress?: (event: ChatLoopEvent) => { sequence: number; generation: string | number; ingressAt: number } | null; + getSmoothStreamingMode?: () => SmoothStreamingMode; + isAutoPacingAllowed?: () => boolean; + /** Force and await one actual frame plus any stdout drain. */ + commitFrame?: (reason?: string) => Promise; } export interface CoalescingChatRenderer { applyEvent(event: ChatLoopEvent): void; /** Cancel the pending coalesce timer (if any) and request one synchronous render. */ flush(): void; + /** Ordered barrier for replay, worker, command-output, and other panel mutations. */ + mutate(mutation: () => void, reason?: string): void; + /** Drop queued presentation content before replacing/resetting the transcript. */ + reset(mutation: () => void): void; + /** Drain paced content and await the first committed frame containing it. */ + flushAndCommit(reason?: string): Promise; + /** Apply a live mode change as an immediate ordered drain boundary. */ + setSmoothStreamingMode(mode: SmoothStreamingMode): void; + dispose(): void; } export function createCoalescingChatRenderer(deps: CreateCoalescingChatRendererDeps): CoalescingChatRenderer { @@ -113,8 +137,13 @@ export function createCoalescingChatRenderer(deps: CreateCoalescingChatRendererD const coalesceMs = deps.coalesceMs ?? DEFAULT_COALESCE_MS; let pendingTimer: unknown = null; + let mutationDepth = 0; + let transactionNeedsRender = false; + let disposed = false; + let pacer: StreamPacer | null = null; const fireCoalesced = (): void => { + if (disposed) return; pendingTimer = null; deps.requestRender(); }; @@ -126,31 +155,165 @@ export function createCoalescingChatRenderer(deps: CreateCoalescingChatRendererD return true; }; - return { + const requestTransactionalRender = (coalesce: boolean): void => { + if (mutationDepth > 0) { + transactionNeedsRender = true; + return; + } + if (!coalesce) { + cancelPending(); + deps.requestRender(); + return; + } + deps.onDelta?.(); + if (pendingTimer === null) pendingTimer = setTimer(fireCoalesced, coalesceMs); + }; + const transaction = (operation: () => void, coalesce = false): void => { + mutationDepth += 1; + try { + operation(); + } finally { + mutationDepth -= 1; + if (mutationDepth === 0 && transactionNeedsRender) { + transactionNeedsRender = false; + requestTransactionalRender(coalesce); + } + } + }; + const applySlice = (slice: StreamPacerSlice): void => { + const event = + slice.kind === "text" + ? ({ type: "text_delta", contentIndex: slice.contentIndex, delta: slice.text, partialText: "" } as const) + : ({ type: "thinking_delta", contentIndex: slice.contentIndex, delta: slice.text, partialThinking: "" } as const); + deps.chatPanel.applyEvent(event); + if (slice.finalForItem) { + deps.onPanelApplied?.(slice.sequence); + deps.onQueue?.(slice.sequence, "dequeue"); + } + requestTransactionalRender(true); + }; + if (deps.streamIngress && deps.getSmoothStreamingMode) { + pacer = createStreamPacer({ + mode: deps.getSmoothStreamingMode(), + onSlice: applySlice, + onDiscard: (sequence) => deps.onQueue?.(sequence, "dequeue"), + ...(deps.now ? { now: deps.now } : {}), + ...(deps.setTimer ? { setTimer: deps.setTimer } : {}), + ...(deps.clearTimer ? { clearTimer: deps.clearTimer } : {}), + ...(deps.isAutoPacingAllowed ? { isAutoPacingAllowed: deps.isAutoPacingAllowed } : {}), + }); + } + const syncPacerMode = (): SmoothStreamingMode => { + const mode = deps.getSmoothStreamingMode?.() ?? "off"; + if (pacer && pacer.mode !== mode) transaction(() => pacer?.setMode(mode)); + return mode; + }; + const drainPacer = (reason: string): void => { + if (pacer?.snapshot().queuedItems) pacer.flush(reason); + }; + const applyLegacy = (event: ChatLoopEvent): void => { + const visibleEventSeq = deps.visibleEventSequence?.(event) ?? null; + if (visibleEventSeq !== null) deps.onQueue?.(visibleEventSeq, "admit"); + deps.chatPanel.applyEvent(event); + if (visibleEventSeq !== null) { + deps.onPanelApplied?.(visibleEventSeq); + deps.onQueue?.(visibleEventSeq, "dequeue"); + } + if (DELTA_TYPES.has(event.type)) { + requestTransactionalRender(true); + return; + } + requestTransactionalRender(false); + }; + + const renderer: CoalescingChatRenderer = { applyEvent(event) { + if (disposed) return; if (isTransparentAssistantWrapper(event)) return; - const visibleEventSeq = deps.visibleEventSequence?.(event) ?? null; - if (visibleEventSeq !== null) deps.onQueue?.(visibleEventSeq, "admit"); - deps.chatPanel.applyEvent(event); - if (visibleEventSeq !== null) { - deps.onPanelApplied?.(visibleEventSeq); - deps.onQueue?.(visibleEventSeq, "dequeue"); + const mode = syncPacerMode(); + const ingress = deps.streamIngress?.(event) ?? null; + if (!pacer || mode === "off") { + applyLegacy(event); + return; } - if (DELTA_TYPES.has(event.type)) { - deps.onDelta?.(); - if (pendingTimer !== null) return; - pendingTimer = setTimer(fireCoalesced, coalesceMs); + const classification = classifyStreamEvent(event); + if (classification === "paced-display-content") { + if (ingress === null) { + applyLegacy(event); + return; + } + const delta = event as Extract; + if (delta.delta.length === 0) { + applyLegacy(event); + return; + } + deps.onQueue?.(ingress.sequence, "admit"); + transaction(() => { + pacer?.enqueue({ + sequence: ingress.sequence, + generation: ingress.generation, + kind: delta.type === "text_delta" ? "text" : "thinking", + contentIndex: delta.contentIndex, + text: delta.delta, + ingressAt: ingress.ingressAt, + folded: delta.type === "thinking_delta" && !deps.chatPanel.isThinkingExpanded(), + }); + }, true); return; } - cancelPending(); - deps.requestRender(); + if (classification === "cumulative-live-state") { + transaction(() => { + drainPacer("cumulative-live-state"); + applyLegacy(event); + }, true); + return; + } + transaction(() => { + drainPacer(`boundary:${event.type}`); + applyLegacy(event); + }); }, flush() { + if (disposed) return; + transaction(() => drainPacer("explicit-flush")); const wasPending = cancelPending(); - if (!wasPending) return; - deps.requestRender(); + if (wasPending) deps.requestRender(); + }, + mutate(mutation, reason = "panel-mutation") { + if (disposed) return; + transaction(() => { + drainPacer(reason); + mutation(); + requestTransactionalRender(false); + }); + }, + reset(mutation) { + if (disposed) return; + transaction(() => { + pacer?.invalidateEpoch(); + mutation(); + requestTransactionalRender(false); + }); + }, + async flushAndCommit(reason = "final-frame") { + if (disposed) return; + transaction(() => drainPacer(reason)); + cancelPending(); + if (deps.commitFrame) await deps.commitFrame(reason); + else deps.requestRender(); + }, + setSmoothStreamingMode(mode) { + if (disposed || !pacer || pacer.mode === mode) return; + transaction(() => pacer?.setMode(mode)); + }, + dispose() { + if (disposed) return; + transaction(() => pacer?.dispose("renderer-dispose")); + cancelPending(); + disposed = true; }, }; + return renderer; } /** diff --git a/src/interactive/editor-submit.ts b/src/interactive/editor-submit.ts index ed142b7e8..61966ecdf 100644 --- a/src/interactive/editor-submit.ts +++ b/src/interactive/editor-submit.ts @@ -86,6 +86,8 @@ export interface EditorSubmitDeps { session?: EditorSubmitSession; sessionTranscript: EditorSubmitSessionTranscript; chatPanel: Pick; + beforeSemanticBoundary?: (reason: string) => void; + settleVisibleFrame?: (reason: string) => Promise; dispatchCommand: (text: string) => void; /** Idempotently collapses a fresh-session launchpad before any handler can append output. */ collapseLaunchpadBeforeSubmit?: () => void; @@ -346,6 +348,7 @@ export function createEditorSubmitController(deps: EditorSubmitDeps): EditorSubm deps.io.stderr("[follow-up] image references cannot be queued while a response is streaming\n"); return; } + deps.beforeSemanticBoundary?.("follow-up-submit"); if (!deps.chat.queueFollowUp(submitted.text)) { deps.io.stderr("[follow-up] no active response to queue against\n"); return; @@ -390,11 +393,13 @@ export function createEditorSubmitController(deps: EditorSubmitDeps): EditorSubm deps.sessionTranscript.recordSubmittedTurn(); const paths = submitted.workingContextPaths ?? []; const skillRequests = submitted.pendingSkillRequests ?? []; + deps.beforeSemanticBoundary?.("interrupt-submit"); await deps.chat.submit(submitted.text, { steering: "interrupt", ...(paths.length > 0 ? { workingContextPaths: paths } : {}), ...(skillRequests.length > 0 ? { pendingSkillRequests: skillRequests } : {}), }); + await deps.settleVisibleFrame?.("interrupt-submit-return"); })().catch((err) => { const msg = err instanceof Error ? err.message : String(err); deps.io.stderr(`[interrupt] ${msg}\n`); diff --git a/src/interactive/interactive-application.ts b/src/interactive/interactive-application.ts index f21df3a0c..d47c974a5 100644 --- a/src/interactive/interactive-application.ts +++ b/src/interactive/interactive-application.ts @@ -45,6 +45,7 @@ import type { RunIo, TaskMemorySeedCommandResult, } from "./slash-commands.js"; +import { processAutoPacingAllowed, resolveSmoothStreamingMode } from "./stream-pacing-policy.js"; import { createWorkspaceFacts } from "./workspace-facts.js"; export { @@ -404,26 +405,44 @@ export function routeInteractiveKey(data: string, deps: KeyBindingDeps): boolean } export async function createInteractiveApplication(deps: InteractiveDeps): Promise { + const initialSmoothStreaming = resolveSmoothStreamingMode(deps.getSettings?.().terminal.smoothStreaming ?? "off"); + const initialAutoPacingAllowed = processAutoPacingAllowed(false); const shell = createProcessInteractiveShell({ tuiMode: deps.getSettings?.().terminal.tuiMode ?? "regular", + streamPacingActive: + initialSmoothStreaming === "on" || (initialSmoothStreaming === "auto" && initialAutoPacingAllowed), ...(deps.onFirstFrameCommit ? { onFirstFrameCommit: deps.onFirstFrameCommit } : {}), }); const { terminal, tui } = shell; const renderTrace = getActiveRenderTrace(); - const visibleEventSequences = new WeakMap(); + const visibleEventIngress = new WeakMap< + object, + { sequence: number; traceSequence: number; generation: number; ingressAt: number } + >(); + let visibleEventSequence = 0; + let visibleEventGeneration = 0; const recordChatEventIngress = (event: ChatLoopEvent): void => { - if (!renderTrace) return; if (event.type === "agent_start") { - renderTrace.beginGeneration(); + visibleEventGeneration += 1; + renderTrace?.beginGeneration(); return; } if (event.type !== "text_delta" && event.type !== "thinking_delta") return; - const sequence = renderTrace.recordVisibleEvent({ - kind: event.type === "text_delta" ? "text" : "thinking", - contentIndex: event.contentIndex, - delta: event.delta, - }); - visibleEventSequences.set(event, sequence); + visibleEventSequence += 1; + const ingress = { + sequence: visibleEventSequence, + traceSequence: visibleEventSequence, + generation: visibleEventGeneration, + ingressAt: performance.now(), + }; + visibleEventIngress.set(event, ingress); + if (renderTrace) { + ingress.traceSequence = renderTrace.recordVisibleEvent({ + kind: event.type === "text_delta" ? "text" : "thinking", + contentIndex: event.contentIndex, + delta: event.delta, + }); + } }; let applicationController: ApplicationController; const workspaceFacts = createWorkspaceFacts({ @@ -459,7 +478,11 @@ export async function createInteractiveApplication(deps: InteractiveDeps): Promi bus: deps.bus, getLeaderArmed: () => leaderArmed, getShutdownArmed: () => shutdownArmed, - resolveVisibleEventSequence: (event) => visibleEventSequences.get(event) ?? null, + resolveVisibleEventSequence: (event) => visibleEventIngress.get(event)?.traceSequence ?? null, + resolveStreamIngress: (event) => visibleEventIngress.get(event) ?? null, + commitFrame: (reason) => shell.commitCurrentFrame(reason === "teardown" ? 300 : 30_000), + hasObservedBackpressure: () => shell.hasObservedBackpressure(), + onSmoothStreamingMode: (_mode, pacingActive) => shell.setStreamPacingActive(pacingActive), providers: deps.providers, dispatch: deps.dispatch, observability: deps.observability, @@ -499,8 +522,8 @@ export async function createInteractiveApplication(deps: InteractiveDeps): Promi const agentProgress = createAgentProgress(terminal); const busNoticeSink = { appendReplayBlock: (renderBlock: Parameters[0]) => - chatPanel.appendReplayBlock(renderBlock), - requestRender: () => tui.requestRender(), + chatRenderer.mutate(() => chatPanel.appendReplayBlock(renderBlock), "bus-notice"), + requestRender: () => {}, }; const eventProjection = createInteractiveEventProjection({ bus: deps.bus, @@ -529,6 +552,12 @@ export async function createInteractiveApplication(deps: InteractiveDeps): Promi dismissNotification: (key) => notifications.dismiss(key), appendTranscriptNotice: (level, text) => appendNotice(level, text, busNoticeSink), refreshSettingsOverlay: () => overlayLifecycle.refreshSettingsOverlay(), + onConfigHotReload: (settings) => { + const mode = resolveSmoothStreamingMode(settings.terminal.smoothStreaming); + const autoAllowed = processAutoPacingAllowed(shell.hasObservedBackpressure()); + chatRenderer.setSmoothStreamingMode(mode); + shell.setStreamPacingActive(mode === "on" || (mode === "auto" && autoAllowed)); + }, }); // The overlay reads the report this process already produced at boot; it // never probes on a keystroke. @@ -541,7 +570,12 @@ export async function createInteractiveApplication(deps: InteractiveDeps): Promi ...(deps.session ? { session: deps.session } : {}), providers: deps.providers, chat: deps.chat, - chatPanel, + chatPanel: { + appendReplayBlock: (...args) => chatRenderer.mutate(() => chatPanel.appendReplayBlock(...args), "slash-output"), + appendUser: (text) => chatRenderer.mutate(() => chatPanel.appendUser(text), "user-submit"), + }, + beforeSemanticSubmit: () => chatRenderer.flush(), + settleVisibleFrame: (reason) => chatRenderer.flushAndCommit(reason), ...(deps.resources ? { resources: deps.resources } : {}), ...(deps.extensions ? { extensions: deps.extensions } : {}), ...(interopSurface ? { interop: interopSurface } : {}), @@ -600,7 +634,12 @@ export async function createInteractiveApplication(deps: InteractiveDeps): Promi dispatch: deps.dispatch, ...(deps.session ? { session: deps.session } : {}), sessionTranscript, - chatPanel, + chatPanel: { + appendReplayBlock: (...args) => + chatRenderer.mutate(() => chatPanel.appendReplayBlock(...args), "editor-command-output"), + }, + beforeSemanticBoundary: () => chatRenderer.flush(), + settleVisibleFrame: (reason) => chatRenderer.flushAndCommit(reason), dispatchCommand: slashRuntime.dispatchCommand, collapseLaunchpadBeforeSubmit: () => presentation.collapseWelcomeDashboard(), expandSubmit: (text) => expandInteractiveSubmitAsync(text, deps.resources), @@ -625,8 +664,10 @@ export async function createInteractiveApplication(deps: InteractiveDeps): Promi * because the subscriptions that own the fold are built below. */ const resetTranscript = (): void => { - chatPanel.reset(); - interactiveSubscriptions.workers.reset(); + chatRenderer.reset(() => { + chatPanel.reset(); + interactiveSubscriptions.workers.reset(); + }); }; overlayLifecycle = createOverlayLifecycle({ app: deps, @@ -690,6 +731,7 @@ export async function createInteractiveApplication(deps: InteractiveDeps): Promi const current = editor.getText(); editor.setText([restored.join("\n\n"), current].filter((part) => part.trim().length > 0).join("\n\n")); } + chatRenderer.flush(); deps.chat.cancel(); deps.toolRegistry?.cancelParkedCalls("run cancelled by operator"); footer.refresh(); @@ -745,7 +787,7 @@ export async function createInteractiveApplication(deps: InteractiveDeps): Promi // The reducer mutates one state object per assignment, so the panel is // handed that object rather than a copy: a streamed delta reaches the // screen by invalidating a cached render, not by rebuilding the entry. - applyWorkerState: (state) => chatPanel.applyWorkerState(state), + applyWorkerState: (state) => chatRenderer.mutate(() => chatPanel.applyWorkerState(state), "worker-state"), recordWorkerRun: (fields) => { // A `/run` is an operator action that produced durable state, so it // opens a session the same way a local `!bash` line does. Persistence @@ -794,7 +836,25 @@ export async function createInteractiveApplication(deps: InteractiveDeps): Promi editorSubmit, requestRender: () => tui.requestRender(), notifications, - chatPanel, + chatPanel: { + toggleLastToolExpanded: () => chatPanel.toggleLastToolExpanded(), + toggleAllToolsExpanded: () => chatPanel.toggleAllToolsExpanded(), + toggleLiveToolOutput: () => chatPanel.toggleLiveToolOutput(), + toggleLastThinking: () => { + let changed = false; + chatRenderer.mutate(() => { + changed = chatPanel.toggleLastThinking(); + }, "thinking-visibility"); + return changed; + }, + toggleAllThinking: () => { + let changed = false; + chatRenderer.mutate(() => { + changed = chatPanel.toggleAllThinking(); + }, "thinking-visibility"); + return changed; + }, + }, shutdown: { stopTickers: presentation.stopTickers, disposeInteractiveTickers: interactiveTickers.dispose, @@ -807,6 +867,13 @@ export async function createInteractiveApplication(deps: InteractiveDeps): Promi disposeChat: () => deps.chat.dispose(), disposeSubscriptions: interactiveSubscriptions.dispose, }, + beforeStopUi: (() => { + let settlement: Promise | null = null; + return () => { + settlement ??= chatRenderer.flushAndCommit("teardown").finally(() => chatRenderer.dispose()); + return settlement; + }; + })(), stopUi: () => shell.stop(), cancelParkedCalls: (reason) => deps.toolRegistry?.cancelParkedCalls(reason), onShutdown: async () => { diff --git a/src/interactive/interactive-event-projection.ts b/src/interactive/interactive-event-projection.ts index d7015d5d4..8e95e5450 100644 --- a/src/interactive/interactive-event-projection.ts +++ b/src/interactive/interactive-event-projection.ts @@ -77,6 +77,7 @@ export interface InteractiveEventProjectionDeps { dismissNotification: (key: string) => void; appendTranscriptNotice: (level: InteractiveTranscriptNoticeLevel, text: string) => void; refreshSettingsOverlay: () => void; + onConfigHotReload?: (settings: Readonly) => void; } export interface InteractiveEventProjection { @@ -313,7 +314,8 @@ export function createInteractiveEventProjection(deps: InteractiveEventProjectio deps.refreshFooter(); deps.requestRender(); }), - deps.bus.on(BusChannels.ConfigHotReload, () => { + deps.bus.on(BusChannels.ConfigHotReload, (payload) => { + deps.onConfigHotReload?.(payload.settings); deps.refreshSettingsOverlay(); }), deps.bus.on(BusChannels.ConfigReloadFailed, (payload) => { diff --git a/src/interactive/interactive-input-runtime.ts b/src/interactive/interactive-input-runtime.ts index b23d76b9c..c84806881 100644 --- a/src/interactive/interactive-input-runtime.ts +++ b/src/interactive/interactive-input-runtime.ts @@ -115,11 +115,12 @@ export interface InteractiveInputRuntimeDeps { disposeSubscriptions(): void; }; stopUi: () => void; + beforeStopUi?: () => Promise; cancelParkedCalls: (reason: string) => void; onShutdown: () => Promise; reportShutdownFailure?: (step: string, error: unknown) => void; /** Defaults to the process termination coordinator's drain phase. */ - registerTerminalTeardown?: (teardown: () => void) => void; + registerTerminalTeardown?: (teardown: () => void | Promise) => void; registerInputListener: (listener: (data: string) => ApplicationInputResult) => void; onInputIngress?: (action: RenderInputAction, data: string) => void; intervalsToClear?: ReadonlyArray; @@ -265,6 +266,7 @@ export function createInteractiveInputRuntime(deps: InteractiveInputRuntimeDeps) () => deps.shutdown.disposeSubscriptions(), ], stopUi: deps.stopUi, + ...(deps.beforeStopUi ? { beforeStopUi: deps.beforeStopUi } : {}), cancelParkedCalls: deps.cancelParkedCalls, onShutdown: deps.onShutdown, registerTerminalTeardown: diff --git a/src/interactive/interactive-presentation.ts b/src/interactive/interactive-presentation.ts index b97b5f192..0b21a2054 100644 --- a/src/interactive/interactive-presentation.ts +++ b/src/interactive/interactive-presentation.ts @@ -29,6 +29,8 @@ import type { SessionTranscript } from "./session-transcript.js"; import { createSlashCommandAutocompleteProvider } from "./slash-autocomplete.js"; import { parseSlashCommand, type RunIo } from "./slash-commands.js"; import { createStatusController, type StatusController, type TurnSummary } from "./status/index.js"; +import type { SmoothStreamingMode } from "./stream-pacer.js"; +import { processAutoPacingAllowed, resolveSmoothStreamingMode } from "./stream-pacing-policy.js"; import { formatTargetLabel } from "./theme/index.js"; import { createWelcomeDashboard, type WelcomeDashboardComponent } from "./welcome-dashboard.js"; import type { WorkspaceFacts } from "./workspace-facts.js"; @@ -84,6 +86,12 @@ export interface InteractivePresentationDeps { getShutdownArmed?: () => boolean; getCwd?: () => string; resolveVisibleEventSequence?: (event: ChatLoopEvent) => number | null; + resolveStreamIngress?: ( + event: ChatLoopEvent, + ) => { sequence: number; generation: string | number; ingressAt: number } | null; + commitFrame?: (reason?: string) => Promise; + hasObservedBackpressure?: () => boolean; + onSmoothStreamingMode?: (mode: SmoothStreamingMode, pacingActive: boolean) => void; scheduleInterval?: (callback: () => void, intervalMs: number) => PresentationTickerHandle; clearScheduledInterval?: (handle: PresentationTickerHandle) => void; factories?: Partial; @@ -339,6 +347,19 @@ export function createInteractivePresentation(deps: InteractivePresentationDeps) const chatRenderer = factories.createChatRenderer({ chatPanel, requestRender, + ...(deps.resolveStreamIngress && deps.getSettings + ? { + streamIngress: deps.resolveStreamIngress, + getSmoothStreamingMode: () => { + const mode = resolveSmoothStreamingMode(deps.getSettings?.().terminal.smoothStreaming ?? "off"); + const autoAllowed = processAutoPacingAllowed(deps.hasObservedBackpressure?.() ?? false); + deps.onSmoothStreamingMode?.(mode, mode === "on" || (mode === "auto" && autoAllowed)); + return mode; + }, + isAutoPacingAllowed: () => processAutoPacingAllowed(deps.hasObservedBackpressure?.() ?? false), + ...(deps.commitFrame ? { commitFrame: deps.commitFrame } : {}), + } + : {}), ...(renderTrace ? { visibleEventSequence: (event) => deps.resolveVisibleEventSequence?.(event) ?? null, @@ -348,8 +369,9 @@ export function createInteractivePresentation(deps: InteractivePresentationDeps) : {}), }); const io = factories.createIo({ - appendReplayBlock: (renderBlock) => chatPanel.appendReplayBlock(renderBlock), - requestRender, + appendReplayBlock: (renderBlock) => + chatRenderer.mutate(() => chatPanel.appendReplayBlock(renderBlock), "command-output"), + requestRender: () => {}, }); const root = factories.buildLayout( { diff --git a/src/interactive/interactive-shell.ts b/src/interactive/interactive-shell.ts index 01b13f50d..7028c7a12 100644 --- a/src/interactive/interactive-shell.ts +++ b/src/interactive/interactive-shell.ts @@ -1,12 +1,5 @@ import type { Component, Terminal, TuiMode, TuiRenderObserver } from "../engine/tui.js"; -import { - InstrumentedTuiAltScreen, - InstrumentedTuiMainScreen, - ProcessTerminal, - type TUI, - TuiAltScreen, - TuiMainScreen, -} from "../engine/tui.js"; +import { InstrumentedTuiAltScreen, InstrumentedTuiMainScreen, ProcessTerminal, type TUI } from "../engine/tui.js"; import { createRenderTrace, type RenderTrace, @@ -14,6 +7,8 @@ import { traceComponentRenders, traceProcessStdout, } from "./render-trace.js"; +import type { StdoutBackpressureGate } from "./stdout-backpressure.js"; +import { installStdoutBackpressureGate } from "./stdout-backpressure.js"; export interface InteractiveShellTui { readonly mode?: TuiMode; @@ -23,6 +18,7 @@ export interface InteractiveShellTui { start(): void; stop(): void; requestRender(): void; + renderNow?(force?: boolean): void; } export interface InteractiveShellInterval { @@ -51,6 +47,12 @@ export interface InteractiveShell< anchor(): Promise; releaseAnchor(): void; stop(): void; + /** Issue the latest model frame and wait a finite bound for stdout drain. */ + commitCurrentFrame(timeoutMs?: number): Promise; + /** True after this process has observed stdout saturation at least once. */ + hasObservedBackpressure(): boolean; + /** Enable admission control only while presentation pacing is actually active. */ + setStreamPacingActive(active: boolean): void; /** Await asynchronous teardown work started by stop(), including trace-file settlement. */ settle(): Promise; complete(code: number): void; @@ -128,6 +130,16 @@ export function createInteractiveShell { + if (!tui.renderNow) { + tui.requestRender(); + return null; + } + tui.renderNow(false); + return null; + }, + hasObservedBackpressure: () => false, + setStreamPacingActive: () => {}, settle(): Promise { return stopSettlement ?? Promise.resolve(); }, @@ -150,14 +162,32 @@ export function getActiveRenderTrace(): RenderTrace | null { return activeRenderTrace; } +/** + * Wait for ordinary admission, but never let a missing drain erase the final + * model state. After the finite pre-render bound, exactly one caller-supplied + * final frame is still issued; a post-render drain wait is likewise bounded. + */ +export async function settleLatestInteractiveFrame( + gate: Pick | null, + timeoutMs: number, + renderFrame: () => Promise, +): Promise { + const writableBeforeRender = gate ? await gate.whenWritable(timeoutMs) : true; + const frameId = await renderFrame(); + if (writableBeforeRender && gate) await gate.whenWritable(timeoutMs); + return frameId; +} + /** Production factories stay here so the composition root does not own them. */ export function createProcessInteractiveShell( - options: { tuiMode?: TuiMode; onFirstFrameCommit?: (frameId: number) => void } = {}, + options: { tuiMode?: TuiMode; onFirstFrameCommit?: (frameId: number) => void; streamPacingActive?: boolean } = {}, ): InteractiveShell { const tracePath = renderTracePath(); let restoreStdout: (() => void) | null = null; let restoreFirstFrameStdout: (() => void) | null = null; let restoreRoot: (() => void) | null = null; + let backpressure: StdoutBackpressureGate | null = null; + let observedBackpressure = false; if (tracePath) { try { activeRenderTrace = createRenderTrace(tracePath); @@ -205,21 +235,62 @@ export function createProcessInteractiveShell( beginPhase: () => null, endPhase: () => {}, }; - const renderObserver = trace ?? firstFrameOnlyObserver; - return createInteractiveShell({ + const primaryObserver = trace ?? (options.onFirstFrameCommit ? firstFrameOnlyObserver : null); + let committedFrameSequence = 0; + const frameWaiters: Array<{ resolve: (frameId: number) => void }> = []; + const renderObserver: TuiRenderObserver = { + isEnabled: () => frameWaiters.length > 0 || (primaryObserver !== null && (primaryObserver.isEnabled?.() ?? true)), + beginFrame: (fields) => ({ + primary: primaryObserver?.beginFrame(fields), + primaryActive: primaryObserver !== null && (primaryObserver.isEnabled?.() ?? true), + frameId: ++committedFrameSequence, + }), + endFrame: (token) => { + const frame = token as { primary: unknown; primaryActive: boolean; frameId: number }; + if (frame.primaryActive) primaryObserver?.endFrame(frame.primary); + for (const waiter of frameWaiters.splice(0)) waiter.resolve(frame.frameId); + }, + beginPhase: (token, phase) => { + const frame = token as { primary: unknown; primaryActive: boolean }; + return frame.primaryActive ? primaryObserver?.beginPhase(frame.primary, phase) : null; + }, + endPhase: (token, phase, phaseToken) => { + const frame = token as { primary: unknown; primaryActive: boolean }; + if (frame.primaryActive) primaryObserver?.endPhase(frame.primary, phase, phaseToken); + }, + }; + const renderAdmission = { + get blocked() { + return backpressure?.blocked ?? false; + }, + onWritable(listener: () => void) { + if (backpressure) return backpressure.onWritable(listener); + queueMicrotask(listener); + return () => {}; + }, + }; + const setStreamPacingActive = (active: boolean): void => { + if (active) { + backpressure ??= installStdoutBackpressureGate(); + return; + } + if (!backpressure) return; + observedBackpressure ||= backpressure.observed; + backpressure.restore(); + backpressure = null; + }; + setStreamPacingActive(options.streamPacingActive === true); + const shell = createInteractiveShell({ createTerminal: () => new ProcessTerminal(), createTui: (terminal) => options.tuiMode === "fullscreen" - ? options.onFirstFrameCommit || trace - ? new InstrumentedTuiAltScreen(terminal, renderObserver) - : new TuiAltScreen(terminal) - : options.onFirstFrameCommit || trace - ? new InstrumentedTuiMainScreen(terminal, renderObserver) - : new TuiMainScreen(terminal), + ? new InstrumentedTuiAltScreen(terminal, renderObserver, undefined, undefined, undefined, renderAdmission) + : new InstrumentedTuiMainScreen(terminal, renderObserver, undefined, undefined, renderAdmission), prepareRoot: (root) => { if (trace) restoreRoot = traceComponentRenders(root, trace); }, onStop: async () => { + setStreamPacingActive(false); restoreRoot?.(); restoreRoot = null; restoreStdout?.(); @@ -230,4 +301,33 @@ export function createProcessInteractiveShell( if (trace) await trace.close(); }, }); + return { + ...shell, + async commitCurrentFrame(timeoutMs = 30_000): Promise { + return await settleLatestInteractiveFrame(backpressure, timeoutMs, async () => { + const tui = shell.tui as TUI & { renderNow(force?: boolean): void }; + const waiter: { resolve: (frameId: number) => void } = { resolve: () => {} }; + const frame = new Promise((resolve) => { + waiter.resolve = resolve; + frameWaiters.push(waiter); + }); + try { + // Even when the writable never drains, issue exactly one bounded + // final frame. Preserve the renderer's cursor/history state: forcing + // a reset here makes TuiMainScreen repaint from the current physical + // column and can concatenate the full frame onto the footer. A direct + // differential render is still synchronous and commits every newest + // component mutation without reopening ordinary frame production. + tui.renderNow(false); + } catch (error) { + const index = frameWaiters.indexOf(waiter); + if (index >= 0) frameWaiters.splice(index, 1); + throw error; + } + return await frame; + }); + }, + hasObservedBackpressure: () => observedBackpressure || backpressure?.observed === true, + setStreamPacingActive, + }; } diff --git a/src/interactive/interactive-slash-runtime.ts b/src/interactive/interactive-slash-runtime.ts index 7dad380c3..1c52258bd 100644 --- a/src/interactive/interactive-slash-runtime.ts +++ b/src/interactive/interactive-slash-runtime.ts @@ -89,6 +89,8 @@ export interface InteractiveSlashRuntimeDeps { stateDir: string; shutdown: () => void | Promise; requestRender: () => void; + beforeSemanticSubmit?: () => void; + settleVisibleFrame?: (reason: string) => Promise; refreshFooter: () => void; dismissContextBootstrapNotices: () => void; recordSubmittedTurn: () => void; @@ -196,11 +198,13 @@ export function createInteractiveSlashRuntime(deps: InteractiveSlashRuntimeDeps) deps.refreshFooter(); if (!willQueue) deps.chatPanel.appendUser(sub.text); deps.requestRender(); + deps.beforeSemanticSubmit?.(); await deps.chat.submit(sub.text, { ...(sub.images.length > 0 ? { images: sub.images } : {}), ...(sub.workingContextPaths.length > 0 ? { workingContextPaths: sub.workingContextPaths } : {}), ...(sub.pendingSkillRequests.length > 0 ? { pendingSkillRequests: sub.pendingSkillRequests } : {}), }); + await deps.settleVisibleFrame?.("submit-return"); } catch (err) { const msg = err instanceof Error ? err.message : String(err); deps.io.stderr(`[interactive] chat failed: ${msg}\n`); diff --git a/src/interactive/overlays/settings.ts b/src/interactive/overlays/settings.ts index c95b6d65b..eb87c21d5 100644 --- a/src/interactive/overlays/settings.ts +++ b/src/interactive/overlays/settings.ts @@ -197,6 +197,7 @@ export const SETTINGS_LABELS_BY_ID = { "terminal.outputVerbosity": "Output detail", "terminal.tuiMode": "TUI mode", "terminal.fullscreenScrollbar": "Fullscreen scrollbar", + "terminal.smoothStreaming": "Smooth streaming", theme: "Theme", runtimePlugins: "Runtime plugins", "compaction.model": "Compaction model", @@ -266,6 +267,7 @@ export const SETTINGS_SECTION_ROWS = { "terminal.outputVerbosity", "terminal.tuiMode", "terminal.fullscreenScrollbar", + "terminal.smoothStreaming", "theme", ], advanced: [ @@ -322,6 +324,7 @@ const SETTINGS_DESCRIPTIONS_BY_ID = { "terminal.outputVerbosity": "How much reasoning, tool input, and live tool output appears in the transcript.", "terminal.tuiMode": "Use regular terminal scrollback or a fullscreen transcript with a sticky composer and footer.", "terminal.fullscreenScrollbar": "When the draggable transcript scrollbar is visible in fullscreen mode.", + "terminal.smoothStreaming": "Presentation-only pacing for streamed assistant text and thinking.", theme: "Color palette. Clio ships a single tuned palette.", runtimePlugins: "npm packages exporting clioRuntimes: RuntimeDescriptor[].", "compaction.model": "Dedicated summarization model; blank uses the orchestrator.", @@ -409,6 +412,11 @@ const SETTINGS_VALUE_HELP_BY_ID: Partial; @@ -1389,6 +1397,9 @@ export function buildSettingItems( settingItem("terminal.fullscreenScrollbar", terminal.fullscreenScrollbar, { values: ["hidden", "auto", "always"], }), + settingItem("terminal.smoothStreaming", terminal.smoothStreaming, { + values: ["off", "auto", "on"], + }), settingItem("theme", settings.theme, { affordance: "single clio-coder palette", readOnly: true, @@ -1870,6 +1881,9 @@ export function applySettingChange(settings: ClioSettings, id: string, value: st settings.terminal.fullscreenScrollbar = value; } return; + case "terminal.smoothStreaming": + if (value === "off" || value === "auto" || value === "on") settings.terminal.smoothStreaming = value; + return; case "runtimePlugins": settings.runtimePlugins = value .split(",") diff --git a/src/interactive/stdout-backpressure.ts b/src/interactive/stdout-backpressure.ts new file mode 100644 index 000000000..2d63bffef --- /dev/null +++ b/src/interactive/stdout-backpressure.ts @@ -0,0 +1,99 @@ +/** + * Process-wide stdout admission gate for interactive frame production. + * + * A terminal frame whose write returns false was accepted by Node, but the + * writable is now saturated. The current frame may finish its cursor/IME + * writes; later frames wait for drain and then render the newest model state. + */ + +export interface StdoutBackpressureGate { + readonly blocked: boolean; + readonly observed: boolean; + onWritable(listener: () => void): () => void; + /** Resolve false when a finite wait expires; no listener survives the bound. */ + whenWritable(timeoutMs?: number): Promise; + restore(): void; +} + +interface WritableStdoutBoundary { + write: typeof process.stdout.write; + once(event: "drain", listener: () => void): unknown; + off(event: "drain", listener: () => void): unknown; +} + +export function installStdoutBackpressureGate(stdout: WritableStdoutBoundary = process.stdout): StdoutBackpressureGate { + const original = stdout.write; + const listeners = new Set<() => void>(); + let blocked = false; + let observed = false; + let restored = false; + let drainListening = false; + + const notifyWritable = (): void => { + drainListening = false; + if (!blocked) return; + blocked = false; + for (const listener of [...listeners]) listener(); + }; + const listenForDrain = (): void => { + if (drainListening || restored) return; + drainListening = true; + stdout.once("drain", notifyWritable); + }; + const wrapped = function (this: typeof stdout, ...args: unknown[]): boolean { + const returned = Reflect.apply(original, this, args) as boolean; + if (!returned) { + observed = true; + blocked = true; + listenForDrain(); + } + return returned; + } as typeof stdout.write; + stdout.write = wrapped; + + return { + get blocked() { + return blocked; + }, + get observed() { + return observed; + }, + onWritable(listener) { + if (restored || !blocked) { + queueMicrotask(listener); + return () => {}; + } + listeners.add(listener); + return () => listeners.delete(listener); + }, + whenWritable(timeoutMs?: number) { + if (restored || !blocked) return Promise.resolve(true); + return new Promise((resolve) => { + let timer: ReturnType | undefined; + const release = this.onWritable(() => { + if (timer) clearTimeout(timer); + release(); + resolve(true); + }); + if (timeoutMs !== undefined && Number.isFinite(timeoutMs) && timeoutMs >= 0) { + timer = setTimeout(() => { + release(); + resolve(false); + }, timeoutMs); + } + }); + }, + restore() { + if (restored) return; + restored = true; + blocked = false; + if (drainListening) { + stdout.off("drain", notifyWritable); + drainListening = false; + } + if (stdout.write === wrapped) stdout.write = original; + for (const listener of [...listeners]) listener(); + listeners.clear(); + }, + }; +} diff --git a/src/interactive/stream-pacer.ts b/src/interactive/stream-pacer.ts new file mode 100644 index 000000000..5ec8c18d1 --- /dev/null +++ b/src/interactive/stream-pacer.ts @@ -0,0 +1,530 @@ +/** + * Presentation-local adaptive pacing for streamed assistant text. + * + * This module deliberately knows nothing about the event bus, ChatPanel, or + * the terminal. It owns only the ordered display-mutation queue. Callers keep + * canonical events and persistence synchronous, then apply the slices emitted + * here directly to their presentation model. + */ + +export type SmoothStreamingMode = "off" | "auto" | "on"; + +export type StreamEventClass = + | "transparent-mirror" + | "paced-display-content" + | "ordered-content-boundary" + | "cumulative-live-state" + | "non-transcript-input"; + +export interface StreamSemanticEvent { + type: string; + assistantMessageEvent?: { type?: unknown }; +} + +const NON_TRANSCRIPT_INPUT_TYPES: ReadonlySet = new Set([ + "cursor_move", + "editor_mutation", + "input", + "overlay_action", + "scroll", +]); + +/** + * Classify an ingress event by the ordering action its presentation consumer + * must take. Unknown events are boundaries: an extra synchronous drain is + * safer than allowing delayed assistant text to cross a new mutation kind. + */ +export function classifyStreamEvent(event: StreamSemanticEvent): StreamEventClass { + if (event.type === "message_update") { + const innerType = event.assistantMessageEvent?.type; + if (innerType === "text_delta" || innerType === "thinking_delta") return "transparent-mirror"; + } + if (event.type === "text_delta" || event.type === "thinking_delta") return "paced-display-content"; + if (event.type === "tool_execution_update") return "cumulative-live-state"; + if (NON_TRANSCRIPT_INPUT_TYPES.has(event.type)) return "non-transcript-input"; + return "ordered-content-boundary"; +} + +export type StreamContentKind = "text" | "thinking"; +export type StreamPacerGeneration = string | number; +export type StreamPacerSliceReason = "first" | "tick" | "deadline" | "capacity" | "flush" | "off" | "folded"; + +export interface StreamPacerAdmission { + /** Canonical ingress sequence. Values must increase for accepted items. */ + sequence: number; + generation: StreamPacerGeneration; + kind: StreamContentKind; + contentIndex: number; + text: string; + /** Canonical monotonic ingress time. Defaults to the injected clock. */ + ingressAt?: number; + /** + * Folded thinking still mutates the panel's complete backing state and live + * counter, but is consumed as one item instead of spending animation ticks. + */ + folded?: boolean; + /** Reject work captured before a boundary/reset instead of crossing epochs. */ + epoch?: number; +} + +export interface StreamPacerSlice { + sequence: number; + generation: StreamPacerGeneration; + epoch: number; + kind: StreamContentKind; + contentIndex: number; + text: string; + ingressAt: number; + reason: StreamPacerSliceReason; + graphemes: number; + bytes: number; + remainingGraphemes: number; + finalForItem: boolean; +} + +export interface StreamPacerSnapshot { + mode: SmoothStreamingMode; + epoch: number; + queuedItems: number; + queuedGraphemes: number; + queuedBytes: number; + oldestAgeMs: number; + credit: number; + timerPending: boolean; + disposed: boolean; +} + +export interface StreamPacerFlushResult { + reason: string; + fromEpoch: number; + toEpoch: number; + items: number; + graphemes: number; + bytes: number; +} + +export interface StreamPacerOptions { + mode: SmoothStreamingMode; + onSlice: (slice: StreamPacerSlice) => void; + /** Balances external queue accounting when a reset intentionally discards content. */ + onDiscard?: (sequence: number) => void; + now?: () => number; + setTimer?: (callback: () => void, delayMs: number) => unknown; + clearTimer?: (handle: unknown) => void; + /** Re-evaluated at admission/tick time. Defaults false for conservative auto. */ + isAutoPacingAllowed?: () => boolean; + tickMs?: number; + baseGraphemesPerSecond?: number; + /** Fraction of measured arrivals granted as fractional display credit. */ + arrivalCreditRatio?: number; + /** Additional convergence rate, expressed as one backlog per this window. */ + catchUpWindowMs?: number; + maxElapsedMs?: number; + maxOldestAgeMs?: number; + maxQueueGraphemes?: number; + maxQueueBytes?: number; + maxSliceGraphemes?: number; + maxSliceBytes?: number; + firstSliceGraphemes?: number; +} + +export interface StreamPacer { + readonly epoch: number; + readonly mode: SmoothStreamingMode; + enqueue(admission: StreamPacerAdmission): { accepted: boolean; epoch: number }; + /** Consume at most one bounded visible slice (a folded item is intentionally whole). */ + dequeue(maxGraphemes?: number, reason?: StreamPacerSliceReason): StreamPacerSlice | null; + /** Invalidate scheduled callbacks, then synchronously drain all prior-epoch content. */ + flush(reason: string): StreamPacerFlushResult; + /** Drop queued content after a reset/session switch, making captured work stale. */ + invalidateEpoch(): number; + /** Changing to a bypass mode synchronously settles existing content first. */ + setMode(mode: SmoothStreamingMode): void; + snapshot(): StreamPacerSnapshot; + dispose(reason?: string): StreamPacerFlushResult; +} + +interface QueueItem { + sequence: number; + generation: StreamPacerGeneration; + epoch: number; + kind: StreamContentKind; + contentIndex: number; + ingressAt: number; + folded: boolean; + graphemes: string[]; + graphemeBytes: number[]; + offset: number; +} + +const DEFAULT_TICK_MS = 25; +const DEFAULT_RATE = 40; +const DEFAULT_ARRIVAL_CREDIT_RATIO = 0.25; +const DEFAULT_CATCH_UP_WINDOW_MS = 250; +const DEFAULT_MAX_ELAPSED_MS = 100; +const DEFAULT_MAX_AGE_MS = 60; +const DEFAULT_MAX_QUEUE_GRAPHEMES = 8_192; +const DEFAULT_MAX_QUEUE_BYTES = 512 * 1_024; +const DEFAULT_MAX_SLICE_GRAPHEMES = 64; +const DEFAULT_MAX_SLICE_BYTES = 16 * 1_024; +const DEFAULT_FIRST_SLICE_GRAPHEMES = 1; + +function positiveFinite(value: number | undefined, fallback: number): number { + return value !== undefined && Number.isFinite(value) && value > 0 ? value : fallback; +} + +function nonnegativeFinite(value: number | undefined, fallback: number): number { + return value !== undefined && Number.isFinite(value) && value >= 0 ? value : fallback; +} + +export function createStreamPacer(options: StreamPacerOptions): StreamPacer { + const now = options.now ?? (() => performance.now()); + const setTimer = options.setTimer ?? ((callback, delayMs) => setTimeout(callback, delayMs)); + const clearTimer = options.clearTimer ?? ((handle) => clearTimeout(handle as ReturnType)); + const tickMs = positiveFinite(options.tickMs, DEFAULT_TICK_MS); + const baseRate = positiveFinite(options.baseGraphemesPerSecond, DEFAULT_RATE); + const arrivalCreditRatio = nonnegativeFinite(options.arrivalCreditRatio, DEFAULT_ARRIVAL_CREDIT_RATIO); + const catchUpWindowMs = positiveFinite(options.catchUpWindowMs, DEFAULT_CATCH_UP_WINDOW_MS); + const maxElapsedMs = positiveFinite(options.maxElapsedMs, DEFAULT_MAX_ELAPSED_MS); + const maxOldestAgeMs = positiveFinite(options.maxOldestAgeMs, DEFAULT_MAX_AGE_MS); + const maxQueueGraphemes = Math.max( + 1, + Math.floor(positiveFinite(options.maxQueueGraphemes, DEFAULT_MAX_QUEUE_GRAPHEMES)), + ); + const maxQueueBytes = Math.max(1, Math.floor(positiveFinite(options.maxQueueBytes, DEFAULT_MAX_QUEUE_BYTES))); + const maxSliceGraphemes = Math.max( + 1, + Math.floor(positiveFinite(options.maxSliceGraphemes, DEFAULT_MAX_SLICE_GRAPHEMES)), + ); + const maxSliceBytes = Math.max(1, Math.floor(positiveFinite(options.maxSliceBytes, DEFAULT_MAX_SLICE_BYTES))); + const firstSliceGraphemes = Math.max( + 0, + Math.floor(nonnegativeFinite(options.firstSliceGraphemes, DEFAULT_FIRST_SLICE_GRAPHEMES)), + ); + const segmenter = new Intl.Segmenter(undefined, { granularity: "grapheme" }); + + let mode = options.mode; + let epoch = 0; + let disposed = false; + let timer: unknown = null; + let timerToken = 0; + let queue: QueueItem[] = []; + let queuedGraphemes = 0; + let queuedBytes = 0; + let credit = 0; + let lastTickAt = now(); + let lastSequence = -Infinity; + let activeGeneration: StreamPacerGeneration | null = null; + let emittedInEpoch = false; + + const pacingAllowed = (): boolean => mode === "on" || (mode === "auto" && (options.isAutoPacingAllowed?.() ?? false)); + + const cancelTimer = (): void => { + if (timer === null) return; + clearTimer(timer); + timer = null; + timerToken += 1; + }; + + const oldestAge = (at: number): number => { + const head = queue[0]; + return head ? Math.max(0, at - head.ingressAt) : 0; + }; + + const schedule = (): void => { + if (disposed || timer !== null || queue.length === 0 || !pacingAllowed()) return; + const capturedEpoch = epoch; + const capturedTimerToken = ++timerToken; + const untilDeadline = Math.max(0, maxOldestAgeMs - oldestAge(now())); + const delay = !emittedInEpoch && firstSliceGraphemes > 0 ? 0 : Math.min(tickMs, untilDeadline); + timer = setTimer(() => { + if (disposed || capturedEpoch !== epoch || capturedTimerToken !== timerToken) return; + timer = null; + if (!pacingAllowed()) { + flush("auto-bypass"); + return; + } + if (!emittedInEpoch && firstSliceGraphemes > 0) { + const first = dequeueInternal(firstSliceGraphemes, "first"); + if (first && first.reason !== "folded") credit = Math.max(0, credit - first.graphemes); + consumeFoldedHeads(); + schedule(); + return; + } + const tickAt = now(); + const elapsedMs = Math.min(maxElapsedMs, Math.max(0, tickAt - lastTickAt)); + lastTickAt = tickAt; + const catchUpRate = queuedGraphemes * (1_000 / catchUpWindowMs); + credit += (elapsedMs / 1_000) * (baseRate + catchUpRate); + + // Every overdue item is settled now. Per-slice limits remain intact, so + // mutation consumers never receive an unexpectedly giant visible chunk. + while (queue[0] && oldestAge(tickAt) >= maxOldestAgeMs) { + const head = queue[0]; + if (!head) break; + const before = head.offset; + const overdue = dequeueInternal(maxSliceGraphemes, "deadline"); + if (overdue && overdue.reason !== "folded") credit = Math.max(0, credit - overdue.graphemes); + if (queue[0] === head && head.offset === before) break; + } + + const budget = Math.min(maxSliceGraphemes, Math.floor(credit)); + if (budget > 0 && queue.length > 0) { + const slice = dequeueInternal(budget, "tick"); + if (slice && slice.reason !== "folded") credit = Math.max(0, credit - slice.graphemes); + } + consumeFoldedHeads(); + schedule(); + }, delay); + }; + + const dequeueInternal = (requestedGraphemes: number, reason: StreamPacerSliceReason): StreamPacerSlice | null => { + const head = queue[0]; + if (!head) return null; + const remaining = head.graphemes.length - head.offset; + if (remaining <= 0) return null; + + let take = head.folded + ? remaining + : Math.max(1, Math.min(remaining, maxSliceGraphemes, Math.floor(requestedGraphemes))); + let bytes = 0; + if (!head.folded) { + take = 0; + const limit = Math.max(1, Math.min(remaining, maxSliceGraphemes, Math.floor(requestedGraphemes))); + while (take < limit) { + const nextBytes = head.graphemeBytes[head.offset + take]; + if (nextBytes === undefined) break; + if (take > 0 && bytes + nextBytes > maxSliceBytes) break; + bytes += nextBytes; + take += 1; + } + } else { + for (let index = head.offset; index < head.graphemes.length; index += 1) { + bytes += head.graphemeBytes[index] ?? 0; + } + } + + const start = head.offset; + const end = start + take; + const text = head.graphemes.slice(start, end).join(""); + head.offset = end; + queuedGraphemes -= take; + queuedBytes -= bytes; + const finalForItem = head.offset === head.graphemes.length; + if (finalForItem) queue.shift(); + if (queue.length === 0) { + activeGeneration = null; + credit = 0; + } + + const slice: StreamPacerSlice = { + sequence: head.sequence, + generation: head.generation, + epoch: head.epoch, + kind: head.kind, + contentIndex: head.contentIndex, + text, + ingressAt: head.ingressAt, + reason: head.folded ? "folded" : reason, + graphemes: take, + bytes, + remainingGraphemes: head.graphemes.length - head.offset, + finalForItem, + }; + emittedInEpoch = true; + options.onSlice(slice); + return slice; + }; + + const consumeFoldedHeads = (): void => { + while (queue[0]?.folded) dequeueInternal(Number.MAX_SAFE_INTEGER, "folded"); + }; + + const drainDetached = (detached: QueueItem[], flushReason: string): StreamPacerFlushResult => { + const fromEpoch = detached[0]?.epoch ?? epoch - 1; + let items = 0; + let graphemes = 0; + let bytes = 0; + for (const item of detached) { + const start = item.offset; + if (start >= item.graphemes.length) continue; + const text = item.graphemes.slice(start).join(""); + let itemBytes = 0; + for (let index = start; index < item.graphemeBytes.length; index += 1) itemBytes += item.graphemeBytes[index] ?? 0; + const count = item.graphemes.length - start; + items += 1; + graphemes += count; + bytes += itemBytes; + options.onSlice({ + sequence: item.sequence, + generation: item.generation, + epoch: item.epoch, + kind: item.kind, + contentIndex: item.contentIndex, + text, + ingressAt: item.ingressAt, + reason: item.folded ? "folded" : "flush", + graphemes: count, + bytes: itemBytes, + remainingGraphemes: 0, + finalForItem: true, + }); + } + return { reason: flushReason, fromEpoch, toEpoch: epoch, items, graphemes, bytes }; + }; + + const flush = (reason: string): StreamPacerFlushResult => { + cancelTimer(); + const detached = queue; + queue = []; + queuedGraphemes = 0; + queuedBytes = 0; + credit = 0; + activeGeneration = null; + epoch += 1; + emittedInEpoch = false; + lastTickAt = now(); + return drainDetached(detached, reason); + }; + + const api: StreamPacer = { + get epoch() { + return epoch; + }, + get mode() { + return mode; + }, + enqueue(admission) { + if (disposed || (admission.epoch !== undefined && admission.epoch !== epoch) || admission.text.length === 0) { + return { accepted: false, epoch }; + } + if (!Number.isFinite(admission.sequence) || admission.sequence <= lastSequence) { + throw new Error(`stream pacer sequence must increase (received ${admission.sequence} after ${lastSequence})`); + } + if (activeGeneration !== null && activeGeneration !== admission.generation) { + throw new Error( + "stream pacer generation changed with queued content; flush or invalidate the prior generation first", + ); + } + if (!pacingAllowed()) { + if (queue.length > 0) flush("mode-bypass"); + lastSequence = admission.sequence; + const ingressAt = admission.ingressAt ?? now(); + const graphemes = Array.from(segmenter.segment(admission.text), ({ segment }) => segment); + const bytes = Buffer.byteLength(admission.text); + options.onSlice({ + sequence: admission.sequence, + generation: admission.generation, + epoch, + kind: admission.kind, + contentIndex: admission.contentIndex, + text: admission.text, + ingressAt, + reason: admission.folded ? "folded" : "off", + graphemes: graphemes.length, + bytes, + remainingGraphemes: 0, + finalForItem: true, + }); + return { accepted: true, epoch }; + } + + const graphemes = Array.from(segmenter.segment(admission.text), ({ segment }) => segment); + if (graphemes.length === 0) return { accepted: false, epoch }; + const graphemeBytes = graphemes.map((grapheme) => Buffer.byteLength(grapheme)); + const itemBytes = graphemeBytes.reduce((sum, value) => sum + value, 0); + const queueWasEmpty = queue.length === 0; + const item: QueueItem = { + sequence: admission.sequence, + generation: admission.generation, + epoch, + kind: admission.kind, + contentIndex: admission.contentIndex, + ingressAt: admission.ingressAt ?? now(), + folded: admission.kind === "thinking" && admission.folded === true, + graphemes, + graphemeBytes, + offset: 0, + }; + lastSequence = admission.sequence; + activeGeneration = admission.generation; + if (queueWasEmpty) lastTickAt = now(); + queue.push(item); + queuedGraphemes += graphemes.length; + queuedBytes += itemBytes; + if (!item.folded) credit += graphemes.length * arrivalCreditRatio; + + while (queuedGraphemes > maxQueueGraphemes || queuedBytes > maxQueueBytes) { + const capacitySlice = dequeueInternal(maxSliceGraphemes, "capacity"); + if (!capacitySlice) break; + if (capacitySlice.reason !== "folded") credit = Math.max(0, credit - capacitySlice.graphemes); + } + if (queue.length === 0) cancelTimer(); + else schedule(); + return { accepted: true, epoch }; + }, + dequeue(maxGraphemes = maxSliceGraphemes, reason = "tick") { + if (disposed) return null; + const slice = dequeueInternal(Math.max(1, Math.floor(maxGraphemes)), reason); + if (slice && slice.reason !== "folded") credit = Math.max(0, credit - slice.graphemes); + consumeFoldedHeads(); + if (queue.length === 0) cancelTimer(); + return slice; + }, + flush, + invalidateEpoch() { + cancelTimer(); + for (const item of queue) options.onDiscard?.(item.sequence); + queue = []; + queuedGraphemes = 0; + queuedBytes = 0; + credit = 0; + activeGeneration = null; + epoch += 1; + emittedInEpoch = false; + lastTickAt = now(); + return epoch; + }, + setMode(nextMode) { + if (nextMode === mode) return; + if ( + queue.length > 0 && + (nextMode === "off" || (nextMode === "auto" && !(options.isAutoPacingAllowed?.() ?? false))) + ) { + flush("mode-change"); + } + mode = nextMode; + if (queue.length > 0) schedule(); + }, + snapshot() { + return { + mode, + epoch, + queuedItems: queue.length, + queuedGraphemes, + queuedBytes, + oldestAgeMs: oldestAge(now()), + credit, + timerPending: timer !== null, + disposed, + }; + }, + dispose(reason = "dispose") { + if (disposed) return { reason, fromEpoch: epoch, toEpoch: epoch, items: 0, graphemes: 0, bytes: 0 }; + cancelTimer(); + const detached = queue; + queue = []; + queuedGraphemes = 0; + queuedBytes = 0; + credit = 0; + activeGeneration = null; + epoch += 1; + emittedInEpoch = false; + lastTickAt = now(); + disposed = true; + return drainDetached(detached, reason); + }, + }; + + return api; +} diff --git a/src/interactive/stream-pacing-policy.ts b/src/interactive/stream-pacing-policy.ts new file mode 100644 index 000000000..90bbbe825 --- /dev/null +++ b/src/interactive/stream-pacing-policy.ts @@ -0,0 +1,66 @@ +import type { SmoothStreaming } from "../core/defaults.js"; + +export const SMOOTH_STREAM_ENV = "CLIO_CODER_SMOOTH_STREAM"; + +/** + * One-process override precedence: a valid environment value wins, while an + * invalid value can never accidentally enable animation. `0` is the immediate + * escape hatch; `1` is accepted as the explicit `on` spelling. + */ +export function resolveSmoothStreamingMode( + configured: SmoothStreaming, + env: NodeJS.ProcessEnv = process.env, +): SmoothStreaming { + const raw = env[SMOOTH_STREAM_ENV]?.trim().toLowerCase(); + if (!raw) return configured; + if (raw === "0" || raw === "off" || raw === "false") return "off"; + if (raw === "auto") return "auto"; + if (raw === "1" || raw === "on" || raw === "true") return "on"; + return "off"; +} + +export interface AutoPacingEnvironment { + isTTY: boolean; + term?: string; + termProgram?: string; + sshConnection?: string; + sshTty?: string; + tmux?: string; + sty?: string; + ci?: string; + reducedMotion?: string; + screenReader?: string; + backpressureObserved: boolean; +} + +/** + * Auto is intentionally conservative. Terminal/screen-reader detection is not + * reliable, so persisted `off` remains the release default and explicit + * accessibility markers always bypass pacing. Remote/multiplexed and already + * saturated outputs keep the proven coalescer behavior. + */ +export function autoPacingAllowed(environment: AutoPacingEnvironment): boolean { + if (!environment.isTTY || environment.backpressureObserved) return false; + const term = environment.term?.trim().toLowerCase(); + if (!term || term === "dumb" || term === "unknown") return false; + if (environment.sshConnection || environment.sshTty || environment.tmux || environment.sty || environment.ci) + return false; + if (environment.reducedMotion === "1" || environment.screenReader === "1") return false; + return true; +} + +export function processAutoPacingAllowed(backpressureObserved: boolean, env: NodeJS.ProcessEnv = process.env): boolean { + return autoPacingAllowed({ + isTTY: process.stdout.isTTY === true, + ...(env.TERM === undefined ? {} : { term: env.TERM }), + ...(env.TERM_PROGRAM === undefined ? {} : { termProgram: env.TERM_PROGRAM }), + ...(env.SSH_CONNECTION === undefined ? {} : { sshConnection: env.SSH_CONNECTION }), + ...(env.SSH_TTY === undefined ? {} : { sshTty: env.SSH_TTY }), + ...(env.TMUX === undefined ? {} : { tmux: env.TMUX }), + ...(env.STY === undefined ? {} : { sty: env.STY }), + ...(env.CI === undefined ? {} : { ci: env.CI }), + ...(env.CLIO_CODER_REDUCE_MOTION === undefined ? {} : { reducedMotion: env.CLIO_CODER_REDUCE_MOTION }), + ...(env.CLIO_CODER_SCREEN_READER === undefined ? {} : { screenReader: env.CLIO_CODER_SCREEN_READER }), + backpressureObserved, + }); +} diff --git a/tests/contracts/chat-renderer-coalescing.test.ts b/tests/contracts/chat-renderer-coalescing.test.ts index 61a048b24..9e9551ad3 100644 --- a/tests/contracts/chat-renderer-coalescing.test.ts +++ b/tests/contracts/chat-renderer-coalescing.test.ts @@ -1,7 +1,7 @@ -import { deepStrictEqual, strictEqual } from "node:assert/strict"; +import { deepStrictEqual, ok, strictEqual } from "node:assert/strict"; import { describe, it } from "node:test"; import type { ChatLoopEvent } from "../../src/interactive/chat-loop.js"; -import type { ChatPanel } from "../../src/interactive/chat-panel.js"; +import { type ChatPanel, createChatPanel } from "../../src/interactive/chat-panel.js"; import { createCoalescingChatRenderer } from "../../src/interactive/chat-renderer.js"; describe("contracts/chat renderer coalescing", () => { @@ -134,4 +134,300 @@ describe("contracts/chat renderer coalescing", () => { deepStrictEqual(log, ["queue:admit:17", "panel:apply", "panel:high-water:17", "queue:dequeue:17"]); }); + + it("paces only derived content and drains it atomically before a raw tool boundary", () => { + const applied: string[] = []; + const timers = new Map void>(); + let timerId = 0; + let renders = 0; + let sequence = 0; + const ingress = new WeakMap(); + const renderer = createCoalescingChatRenderer({ + chatPanel: { + applyEvent: (event: ChatLoopEvent) => + applied.push( + event.type === "text_delta" || event.type === "thinking_delta" ? `${event.type}:${event.delta}` : event.type, + ), + isThinkingExpanded: () => false, + } as unknown as ChatPanel, + requestRender: () => { + renders += 1; + }, + streamIngress: (event) => ingress.get(event) ?? null, + getSmoothStreamingMode: () => "on", + setTimer: (callback) => { + const id = ++timerId; + timers.set(id, callback); + return id; + }, + clearTimer: (id) => void timers.delete(id as number), + now: () => 0, + }); + const delta = { type: "text_delta", contentIndex: 0, delta: "A👩‍🔬B", partialText: "A👩‍🔬B" } as ChatLoopEvent; + ingress.set(delta, { sequence: ++sequence, generation: 1, ingressAt: 0 }); + renderer.applyEvent(delta); + renderer.applyEvent({ + type: "message_update", + assistantMessageEvent: { type: "thinking_delta", delta: "hidden" }, + } as unknown as ChatLoopEvent); + strictEqual(applied.length, 0, "admission itself neither mutates nor renders paced visible text"); + + renderer.applyEvent({ + type: "message_update", + assistantMessageEvent: { + type: "toolcall_start", + contentIndex: 1, + partial: { content: [null, { type: "toolCall", id: "t", name: "read", arguments: {} }] }, + }, + } as unknown as ChatLoopEvent); + deepStrictEqual(applied, ["text_delta:A👩‍🔬B", "message_update"]); + strictEqual(renders, 1, "drain plus boundary is one render transaction"); + strictEqual(timers.size, 0); + }); + + it("settles queued graphemes before cumulative state and awaits the final committed frame", async () => { + const log: string[] = []; + let sequence = 0; + const ingress = new WeakMap(); + const renderer = createCoalescingChatRenderer({ + chatPanel: { + applyEvent: (event: ChatLoopEvent) => log.push(event.type === "text_delta" ? `text:${event.delta}` : event.type), + isThinkingExpanded: () => false, + } as unknown as ChatPanel, + requestRender: () => log.push("render"), + streamIngress: (event) => ingress.get(event) ?? null, + getSmoothStreamingMode: () => "on", + setTimer: () => 1, + clearTimer: () => {}, + commitFrame: async () => { + log.push("commit"); + }, + }); + const first = { type: "text_delta", contentIndex: 0, delta: "before", partialText: "before" } as ChatLoopEvent; + ingress.set(first, { sequence: ++sequence, generation: 1, ingressAt: 0 }); + renderer.applyEvent(first); + renderer.applyEvent({ + type: "tool_execution_update", + toolCallId: "tool-1", + toolName: "bash", + partialResult: { content: [{ type: "text", text: "cumulative" }] }, + } as ChatLoopEvent); + const last = { type: "text_delta", contentIndex: 2, delta: "after", partialText: "after" } as ChatLoopEvent; + ingress.set(last, { sequence: ++sequence, generation: 1, ingressAt: 1 }); + renderer.applyEvent(last); + await renderer.flushAndCommit("message-end"); + + deepStrictEqual( + log.filter((entry) => entry !== "render"), + ["text:before", "tool_execution_update", "text:after", "commit"], + ); + }); + + it("advances canonical panel high-water only after the complete paced ingress is present", () => { + const applied: string[] = []; + const highWater: number[] = []; + const timers = new Map void>(); + let timerId = 0; + const event = { type: "text_delta", contentIndex: 0, delta: "abc", partialText: "abc" } as ChatLoopEvent; + const renderer = createCoalescingChatRenderer({ + chatPanel: { + applyEvent: (candidate: ChatLoopEvent) => { + if (candidate.type === "text_delta") applied.push(candidate.delta); + }, + isThinkingExpanded: () => true, + } as unknown as ChatPanel, + requestRender: () => {}, + streamIngress: (candidate) => (candidate === event ? { sequence: 1, generation: 1, ingressAt: 0 } : null), + getSmoothStreamingMode: () => "on", + onPanelApplied: (sequence) => highWater.push(sequence), + setTimer: (callback) => { + const id = ++timerId; + timers.set(id, callback); + return id; + }, + clearTimer: (id) => void timers.delete(id as number), + now: () => 0, + }); + + renderer.applyEvent(event); + const firstSlice = [...timers.values()][0]; + ok(firstSlice); + firstSlice(); + deepStrictEqual(applied, ["a"]); + deepStrictEqual(highWater, [], "a partial synthetic slice cannot claim the complete canonical event"); + renderer.flush(); + deepStrictEqual(applied, ["a", "bc"]); + deepStrictEqual(highWater, [1]); + }); + + it("drains queued content immediately when a live mode change arrives", () => { + const applied: string[] = []; + const timers = new Map void>(); + let timerId = 0; + let configuredMode: "off" | "auto" | "on" = "on"; + const event = { type: "text_delta", contentIndex: 0, delta: "backlog", partialText: "backlog" } as ChatLoopEvent; + const renderer = createCoalescingChatRenderer({ + chatPanel: { + applyEvent: (candidate: ChatLoopEvent) => { + if (candidate.type === "text_delta") applied.push(candidate.delta); + }, + isThinkingExpanded: () => true, + } as unknown as ChatPanel, + requestRender: () => {}, + streamIngress: (candidate) => (candidate === event ? { sequence: 1, generation: 1, ingressAt: 0 } : null), + getSmoothStreamingMode: () => configuredMode, + setTimer: (callback) => { + const id = ++timerId; + timers.set(id, callback); + return id; + }, + clearTimer: (id) => void timers.delete(id as number), + now: () => 0, + }); + + renderer.applyEvent(event); + strictEqual(applied.length, 0); + configuredMode = "off"; + renderer.setSmoothStreamingMode("off"); + deepStrictEqual(applied, ["backlog"]); + strictEqual(timers.size, 0); + }); + + it("keeps retry and operator-abort settlement behind all earlier visible content", () => { + const applied: string[] = []; + let sequence = 0; + const ingress = new WeakMap(); + const renderer = createCoalescingChatRenderer({ + chatPanel: { + applyEvent: (event: ChatLoopEvent) => + applied.push(event.type === "text_delta" ? `text:${event.delta}` : event.type), + isThinkingExpanded: () => true, + } as unknown as ChatPanel, + requestRender: () => {}, + streamIngress: (event) => ingress.get(event) ?? null, + getSmoothStreamingMode: () => "on", + setTimer: () => 1, + clearTimer: () => {}, + }); + const beforeRetry = { + type: "text_delta", + contentIndex: 0, + delta: "before retry", + partialText: "before retry", + } as ChatLoopEvent; + ingress.set(beforeRetry, { sequence: ++sequence, generation: 1, ingressAt: 0 }); + renderer.applyEvent(beforeRetry); + renderer.applyEvent({ + type: "retry_status", + status: { phase: "retrying", attempt: 1, maxAttempts: 2 }, + } as ChatLoopEvent); + const beforeAbort = { + type: "text_delta", + contentIndex: 1, + delta: "before abort", + partialText: "before abort", + } as ChatLoopEvent; + ingress.set(beforeAbort, { sequence: ++sequence, generation: 1, ingressAt: 1 }); + renderer.applyEvent(beforeAbort); + renderer.applyEvent({ + type: "agent_end", + messages: [{ role: "assistant", content: [], stopReason: "aborted" }], + } as unknown as ChatLoopEvent); + + deepStrictEqual(applied, ["text:before retry", "retry_status", "text:before abort", "agent_end"]); + }); + + it("drops old-session paced content before a reset mutation", () => { + const applied: string[] = []; + const queue: string[] = []; + const panelHighWater: number[] = []; + let renders = 0; + const event = { type: "text_delta", contentIndex: 0, delta: "stale", partialText: "stale" } as ChatLoopEvent; + const renderer = createCoalescingChatRenderer({ + chatPanel: { + applyEvent: (candidate: ChatLoopEvent) => applied.push(candidate.type), + isThinkingExpanded: () => false, + } as unknown as ChatPanel, + requestRender: () => { + renders += 1; + }, + streamIngress: (candidate) => (candidate === event ? { sequence: 1, generation: "old", ingressAt: 0 } : null), + getSmoothStreamingMode: () => "on", + onQueue: (sequence, action) => queue.push(`${action}:${sequence}`), + onPanelApplied: (sequence) => panelHighWater.push(sequence), + setTimer: () => 1, + clearTimer: () => {}, + }); + renderer.applyEvent(event); + renderer.reset(() => applied.push("reset")); + renderer.flush(); + deepStrictEqual(applied, ["reset"]); + deepStrictEqual(queue, ["admit:1", "dequeue:1"], "discard balances render-queue trace accounting"); + deepStrictEqual(panelHighWater, [], "discarded state never advances the panel high-water mark"); + strictEqual(renders, 1, "only the replacement state renders"); + }); + + it("balances an empty derived delta through the exact legacy path", () => { + const queue: string[] = []; + const applied: ChatLoopEvent[] = []; + const event = { type: "text_delta", contentIndex: 0, delta: "", partialText: "" } as ChatLoopEvent; + const renderer = createCoalescingChatRenderer({ + chatPanel: { + applyEvent: (candidate: ChatLoopEvent) => applied.push(candidate), + isThinkingExpanded: () => true, + } as unknown as ChatPanel, + requestRender: () => {}, + streamIngress: (candidate) => (candidate === event ? { sequence: 8, generation: 1, ingressAt: 0 } : null), + visibleEventSequence: (candidate) => (candidate === event ? 8 : null), + getSmoothStreamingMode: () => "on", + onQueue: (sequence, action) => queue.push(`${action}:${sequence}`), + setTimer: () => 1, + }); + + renderer.applyEvent(event); + strictEqual(applied.length, 1); + deepStrictEqual(queue, ["admit:8", "dequeue:8"]); + }); + + it("retains folded thinking in one panel mutation and expands it without pacing invisible graphemes", () => { + const panel = createChatPanel({ now: () => 0 }); + const timers = new Map void>(); + let timerId = 0; + let sequence = 0; + const queue: string[] = []; + const ingress = new WeakMap(); + const renderer = createCoalescingChatRenderer({ + chatPanel: panel, + requestRender: () => {}, + streamIngress: (event) => ingress.get(event) ?? null, + getSmoothStreamingMode: () => "on", + onQueue: (eventSequence, action) => queue.push(`${action}:${eventSequence}`), + setTimer: (callback) => { + const id = ++timerId; + timers.set(id, callback); + return id; + }, + clearTimer: (id) => void timers.delete(id as number), + now: () => 0, + }); + const text = { type: "text_delta", contentIndex: 0, delta: "visible", partialText: "visible" } as ChatLoopEvent; + const thinking = { + type: "thinking_delta", + contentIndex: 1, + delta: "private chain retained exactly", + partialThinking: "private chain retained exactly", + } as ChatLoopEvent; + for (const event of [text, thinking]) { + ingress.set(event, { sequence: ++sequence, generation: 1, ingressAt: 0 }); + renderer.applyEvent(event); + } + + renderer.mutate(() => panel.toggleLastThinking(), "thinking-visibility"); + const expanded = panel.render(80).join("\n"); + ok(expanded.includes("visible")); + ok(expanded.includes("private chain retained exactly"), expanded); + strictEqual(expanded.split("private chain retained exactly").length - 1, 1); + deepStrictEqual(queue, ["admit:1", "admit:2", "dequeue:1", "dequeue:2"]); + strictEqual(timers.size, 0, "visibility mutation drains folded state without spending later ticks"); + }); }); diff --git a/tests/contracts/config.test.ts b/tests/contracts/config.test.ts index dc47eda3b..41fa9a55d 100644 --- a/tests/contracts/config.test.ts +++ b/tests/contracts/config.test.ts @@ -240,7 +240,7 @@ describe("contracts/config", () => { budget: { concurrency: 0 }, targets: [{ runtime: "openai-compat" }], retry: { maxRetries: 1.5 }, - terminal: { tuiMode: "windowed", fullscreenScrollbar: "sometimes" }, + terminal: { tuiMode: "windowed", fullscreenScrollbar: "sometimes", smoothStreaming: "sometimes" }, }); const paths = result.issues.map((issue) => issue.path).sort(); deepStrictEqual(paths, [ @@ -249,6 +249,7 @@ describe("contracts/config", () => { "retry.maxRetries", "targets[0].id", "terminal.fullscreenScrollbar", + "terminal.smoothStreaming", "terminal.tuiMode", ]); // Invalid fields fall back to defaults on the built settings. @@ -256,6 +257,21 @@ describe("contracts/config", () => { strictEqual(result.settings.budget.concurrency, "auto"); }); + it("validates smooth streaming and classifies it as a live presentation setting", () => { + for (const mode of ["off", "auto", "on"] as const) { + const result = validateSettings({ terminal: { smoothStreaming: mode } }); + deepStrictEqual(result.issues, []); + strictEqual(result.settings.terminal.smoothStreaming, mode); + } + const next = structuredClone(DEFAULT_SETTINGS); + next.terminal.smoothStreaming = "on"; + deepStrictEqual(diffSettings(DEFAULT_SETTINGS, next), { + hotReload: ["terminal.smoothStreaming"], + nextTurn: [], + restartRequired: [], + }); + }); + it("validates active routing roles and postures as strict unique lists", () => { const valid = validateSettings({ routing: { diff --git a/tests/contracts/fullscreen-layout.test.ts b/tests/contracts/fullscreen-layout.test.ts index a97f72679..cf8f7c4a9 100644 --- a/tests/contracts/fullscreen-layout.test.ts +++ b/tests/contracts/fullscreen-layout.test.ts @@ -2,7 +2,9 @@ import { ok, strictEqual } from "node:assert/strict"; import { describe, it } from "node:test"; import type { Component, Terminal } from "../../src/engine/tui.js"; import { TuiAltScreen } from "../../src/engine/tui.js"; +import type { ChatLoopEvent } from "../../src/interactive/chat-loop.js"; import { createChatPanel } from "../../src/interactive/chat-panel.js"; +import { createCoalescingChatRenderer } from "../../src/interactive/chat-renderer.js"; import { buildFullscreenLayout, buildLayout } from "../../src/interactive/layout.js"; function linesComponent(lines: string[]): Component { @@ -14,8 +16,8 @@ function linesComponent(lines: string[]): Component { class CaptureTerminal implements Terminal { readonly writes: string[] = []; - readonly columns = 30; - readonly rows = 10; + columns = 30; + rows = 10; readonly kittyProtocolActive = false; private onInput: (data: string) => void = () => {}; @@ -130,4 +132,63 @@ describe("fullscreen transcript layout", () => { tui.stop({ preserveScreen: true }); }); + + it("preserves a frozen transcript viewport through paced output and a mid-queue resize", () => { + const terminal = new CaptureTerminal(); + const chat = createChatPanel({ now: () => 0 }); + for (let index = 1; index <= 20; index += 1) chat.appendUser(`prompt-${index}`); + const layout = buildFullscreenLayout({ + banner: linesComponent(["banner"]), + chat, + editor: linesComponent(["editor"]), + footer: linesComponent(["footer"]), + }); + const tui = new TuiAltScreen(terminal); + tui.setLayoutRoot(layout.root); + tui.start(); + tui.renderNow(true); + tui.scrollToTop(); + tui.renderNow(true); + strictEqual(layout.transcript.scrollTop, 0); + + const timers = new Map void>(); + let timerId = 0; + const event = { + type: "text_delta", + contentIndex: 0, + delta: "paced output remains ordered after resize", + partialText: "paced output remains ordered after resize", + } as ChatLoopEvent; + const renderer = createCoalescingChatRenderer({ + chatPanel: chat, + requestRender: () => tui.renderNow(true), + streamIngress: (candidate) => (candidate === event ? { sequence: 1, generation: "turn", ingressAt: 0 } : null), + getSmoothStreamingMode: () => "on", + setTimer: (callback) => { + const id = ++timerId; + timers.set(id, callback); + return id; + }, + clearTimer: (id) => void timers.delete(id as number), + now: () => 0, + }); + renderer.applyEvent(event); + + terminal.columns = 20; + layout.root.invalidate(); + tui.renderNow(true); + strictEqual(layout.transcript.scrollTop, 0, "resize cannot re-enable follow mode while output is queued"); + const firstTick = [...timers.values()][0]; + ok(firstTick); + firstTick(); + strictEqual(layout.transcript.scrollTop, 0, "the first paced slice cannot steal a manually frozen viewport"); + + renderer.applyEvent({ type: "agent_end", messages: [] } as ChatLoopEvent); + strictEqual(layout.transcript.scrollTop, 0, "the final synchronous drain also preserves scroll position"); + chat.toggleLastThinking(); + const transcript = chat.render(terminal.columns).join("\n"); + ok(transcript.includes("paced output"), transcript); + ok(transcript.includes("after resize"), transcript); + tui.stop({ preserveScreen: true }); + }); }); diff --git a/tests/contracts/interactive-event-projection.test.ts b/tests/contracts/interactive-event-projection.test.ts index 3e5d9555a..15516d599 100644 --- a/tests/contracts/interactive-event-projection.test.ts +++ b/tests/contracts/interactive-event-projection.test.ts @@ -1,6 +1,7 @@ import { deepStrictEqual, match, strictEqual } from "node:assert/strict"; import { describe, it } from "node:test"; import { BusChannels, type LoopBlockedPayload } from "../../src/core/bus-events.js"; +import { DEFAULT_SETTINGS } from "../../src/core/defaults.js"; import { createSafeEventBus } from "../../src/core/event-bus.js"; import type { ChatCancelOptions, ChatLoopEvent } from "../../src/interactive/chat-loop.js"; import { @@ -79,6 +80,21 @@ describe("interactive event projection", () => { deepStrictEqual(log.slice(0, 2), ["ingress:text_delta", "chat:text_delta"]); }); + it("applies a hot presentation setting before refreshing the overlay", () => { + const log: string[] = []; + const harness = createHarness(log); + harness.deps.onConfigHotReload = (settings) => log.push(`hot:${settings.terminal.smoothStreaming}`); + createInteractiveEventProjection(harness.deps); + const settings = structuredClone(DEFAULT_SETTINGS); + settings.terminal.smoothStreaming = "on"; + harness.deps.bus.emit(BusChannels.ConfigHotReload, { + diff: { hotReload: ["terminal.smoothStreaming"], nextTurn: [], restartRequired: [] }, + settings, + }); + + deepStrictEqual(log, ["hot:on", "settings:refresh"]); + }); + it("preserves startup, queue, tool, and chat-render ordering", () => { const log: string[] = []; const harness = createHarness(log); diff --git a/tests/contracts/interactive-presentation.test.ts b/tests/contracts/interactive-presentation.test.ts index d461a90a2..8475e7537 100644 --- a/tests/contracts/interactive-presentation.test.ts +++ b/tests/contracts/interactive-presentation.test.ts @@ -138,7 +138,15 @@ function harness() { setAutocompleteProvider: () => log.push("editor.autocomplete"), } as unknown as ClioEditor; const dispatchBoard = component() as DispatchBoardView; - const chatRenderer = { applyEvent: () => {}, flush: () => {} } as CoalescingChatRenderer; + const chatRenderer = { + applyEvent: () => {}, + flush: () => {}, + mutate: (mutation: () => void) => mutation(), + reset: (mutation: () => void) => mutation(), + flushAndCommit: async () => {}, + setSmoothStreamingMode: () => {}, + dispose: () => {}, + } as CoalescingChatRenderer; const io = {} as RunIo; const root = component(); diff --git a/tests/contracts/interactive-shell.test.ts b/tests/contracts/interactive-shell.test.ts index 08eae7426..497669472 100644 --- a/tests/contracts/interactive-shell.test.ts +++ b/tests/contracts/interactive-shell.test.ts @@ -9,6 +9,7 @@ import { createProcessInteractiveShell, type InteractiveShellInterval, type InteractiveShellTui, + settleLatestInteractiveFrame, } from "../../src/interactive/interactive-shell.js"; interface TestTerminal extends Terminal { @@ -182,4 +183,26 @@ describe("interactive shell ownership", () => { rmSync(dir, { recursive: true, force: true }); } }); + + it("issues one final frame after a permanent no-drain bound instead of abandoning model state", async () => { + const waits: number[] = []; + let renders = 0; + const frameId = await settleLatestInteractiveFrame( + { + whenWritable: async (timeoutMs) => { + waits.push(timeoutMs ?? -1); + return false; + }, + }, + 7, + async () => { + renders += 1; + return 41; + }, + ); + + strictEqual(frameId, 41); + strictEqual(renders, 1); + deepStrictEqual(waits, [7], "a failed pre-render wait does not add a second unbounded wait"); + }); }); diff --git a/tests/contracts/render-pipeline-trace.test.ts b/tests/contracts/render-pipeline-trace.test.ts index 642373d89..8799fa428 100644 --- a/tests/contracts/render-pipeline-trace.test.ts +++ b/tests/contracts/render-pipeline-trace.test.ts @@ -201,6 +201,78 @@ describe("contracts/render pipeline trace", () => { strictEqual(drain.waitMs, 7); }); + it("defers later frame construction while stdout is blocked and coalesces to the newest state", async () => { + const harness = createHarness(); + const terminal = new TraceTerminal(harness.trace, []); + const writable = { release: undefined as (() => void) | undefined }; + const admission = { + blocked: true, + onWritable(listener: () => void) { + writable.release = listener; + return () => { + if (writable.release === listener) writable.release = undefined; + }; + }, + }; + const tui = new InstrumentedTuiMainScreen(terminal, harness.trace, undefined, undefined, admission); + let state = "first"; + let renders = 0; + tui.addChild({ + render: () => { + renders += 1; + return [state]; + }, + invalidate: () => {}, + }); + + tui.requestRender(); + state = "latest"; + tui.requestRender(); + await new Promise((resolve) => setImmediate(resolve)); + strictEqual(renders, 0, "no differential frame is built while the writable is saturated"); + admission.blocked = false; + writable.release?.(); + await new Promise((resolve) => setTimeout(resolve, 25)); + strictEqual(renders, 1, "one released render observes the latest presentation model"); + await harness.close(); + }); + + it("lets one direct final render supersede a deferred blocked request", async () => { + const harness = createHarness(); + const terminal = new TraceTerminal(harness.trace, []); + let release: (() => void) | undefined; + const admission = { + blocked: true, + onWritable(listener: () => void) { + release = listener; + return () => { + if (release === listener) release = undefined; + }; + }, + }; + const tui = new InstrumentedTuiMainScreen(terminal, harness.trace, undefined, undefined, admission); + let renders = 0; + tui.addChild({ + render: () => { + renders += 1; + return ["latest final state"]; + }, + invalidate: () => {}, + }); + tui.requestRender(); + strictEqual(renders, 0); + const staleRelease = release; + ok(staleRelease); + tui.renderNow(true); + strictEqual(renders, 1); + strictEqual(release, undefined, "the direct final frame removes the older gate listener"); + admission.blocked = false; + staleRelease(); + await new Promise((resolve) => setImmediate(resolve)); + strictEqual(renders, 1, "gate restoration cannot schedule a duplicate frame after teardown settlement"); + await harness.close(); + }); + it("buffers asynchronously, reports bounded drops, and treats write failure as nonfatal", async () => { let appendCalls = 0; const payloads: string[] = []; diff --git a/tests/contracts/settings-center.test.ts b/tests/contracts/settings-center.test.ts index d1937cac2..1a9113f66 100644 --- a/tests/contracts/settings-center.test.ts +++ b/tests/contracts/settings-center.test.ts @@ -264,6 +264,7 @@ describe("contracts/settings center", () => { strictEqual(byId.get("runtimePlugins")?.scope, "restart"); strictEqual(byId.get("terminal.tuiMode")?.scope, "restart"); strictEqual(byId.get("terminal.fullscreenScrollbar")?.scope, "restart"); + strictEqual(byId.get("terminal.smoothStreaming")?.scope, "live"); strictEqual(byId.get("autonomy")?.scope, "live"); strictEqual(byId.get("retry.maxRetries")?.scope, "live"); }); @@ -826,6 +827,11 @@ describe("contracts/settings center", () => { value: "always", assert: (s) => strictEqual(s.terminal.fullscreenScrollbar, "always"), }, + { + id: "terminal.smoothStreaming", + value: "on", + assert: (s) => strictEqual(s.terminal.smoothStreaming, "on"), + }, { id: "runtimePlugins", value: "@scope/a, @scope/b", diff --git a/tests/contracts/settings-write-footprint.test.ts b/tests/contracts/settings-write-footprint.test.ts index dd6dbbc6a..d6669977d 100644 --- a/tests/contracts/settings-write-footprint.test.ts +++ b/tests/contracts/settings-write-footprint.test.ts @@ -68,11 +68,13 @@ describe("contracts/settings-write-footprint", () => { for (const testCase of [ { leaf: "tuiMode", value: "fullscreen" }, { leaf: "fullscreenScrollbar", value: "always" }, + { leaf: "smoothStreaming", value: "on" }, ] as const) { const before = seedMinimalFile(); updateSettings((settings) => { if (testCase.leaf === "tuiMode") settings.terminal.tuiMode = testCase.value; - else settings.terminal.fullscreenScrollbar = testCase.value; + else if (testCase.leaf === "fullscreenScrollbar") settings.terminal.fullscreenScrollbar = testCase.value; + else settings.terminal.smoothStreaming = testCase.value; }); const after = readFileSync(settingsPath(), "utf8"); @@ -83,10 +85,9 @@ describe("contracts/settings-write-footprint", () => { strictEqual(loaded[testCase.leaf], testCase.value); strictEqual(loaded.showTerminalProgress, false); strictEqual(loaded.outputVerbosity, "default"); - strictEqual( - testCase.leaf === "tuiMode" ? loaded.fullscreenScrollbar : loaded.tuiMode, - testCase.leaf === "tuiMode" ? "auto" : "regular", - ); + strictEqual(loaded.tuiMode, testCase.leaf === "tuiMode" ? "fullscreen" : "regular"); + strictEqual(loaded.fullscreenScrollbar, testCase.leaf === "fullscreenScrollbar" ? "always" : "auto"); + strictEqual(loaded.smoothStreaming, testCase.leaf === "smoothStreaming" ? "on" : "off"); } }); diff --git a/tests/contracts/stdout-backpressure.test.ts b/tests/contracts/stdout-backpressure.test.ts new file mode 100644 index 000000000..abb8b25aa --- /dev/null +++ b/tests/contracts/stdout-backpressure.test.ts @@ -0,0 +1,65 @@ +import { deepStrictEqual, strictEqual } from "node:assert/strict"; +import { EventEmitter } from "node:events"; +import { describe, it } from "node:test"; +import { installStdoutBackpressureGate } from "../../src/interactive/stdout-backpressure.js"; + +class FakeStdout extends EventEmitter { + returns: boolean[] = []; + writes: string[] = []; + + write = ((chunk: unknown): boolean => { + this.writes.push(String(chunk)); + return this.returns.shift() ?? true; + }) as typeof process.stdout.write; +} + +describe("stdout backpressure gate", () => { + it("blocks after false, releases every waiter on drain, and restores exactly once", async () => { + const stdout = new FakeStdout(); + const original = stdout.write; + stdout.returns.push(false, false, true); + const gate = installStdoutBackpressureGate(stdout); + const calls: string[] = []; + + strictEqual(stdout.write("frame-1"), false); + strictEqual(gate.blocked, true); + strictEqual(gate.observed, true); + gate.onWritable(() => calls.push("listener")); + const settled = gate.whenWritable().then(() => calls.push("promise")); + strictEqual(stdout.write("cursor"), false, "the current frame may finish its cursor write"); + stdout.emit("drain"); + await settled; + + strictEqual(gate.blocked, false); + deepStrictEqual(calls, ["listener", "promise"]); + strictEqual(stdout.listenerCount("drain"), 0); + gate.restore(); + gate.restore(); + strictEqual(stdout.write, original); + deepStrictEqual(stdout.writes, ["frame-1", "cursor"]); + }); + + it("settles pending waiters when teardown happens before drain", async () => { + const stdout = new FakeStdout(); + stdout.returns.push(false); + const gate = installStdoutBackpressureGate(stdout); + stdout.write("frame"); + const settled = gate.whenWritable(); + strictEqual(stdout.listenerCount("drain"), 1); + gate.restore(); + await settled; + strictEqual(gate.blocked, false); + strictEqual(stdout.listenerCount("drain"), 0); + }); + + it("bounds a missing drain and releases the native listener", async () => { + const stdout = new FakeStdout(); + stdout.returns.push(false); + const gate = installStdoutBackpressureGate(stdout); + stdout.write("stalled frame"); + strictEqual(await gate.whenWritable(5), false); + strictEqual(stdout.listenerCount("drain"), 1, "the gate still owns one observer for later writes"); + gate.restore(); + strictEqual(stdout.listenerCount("drain"), 0); + }); +}); diff --git a/tests/contracts/stream-pacer.test.ts b/tests/contracts/stream-pacer.test.ts new file mode 100644 index 000000000..8ef394c86 --- /dev/null +++ b/tests/contracts/stream-pacer.test.ts @@ -0,0 +1,396 @@ +import { deepStrictEqual, ok, strictEqual, throws } from "node:assert/strict"; +import { describe, it } from "node:test"; +import { + classifyStreamEvent, + createStreamPacer, + type StreamPacerOptions, + type StreamPacerSlice, +} from "../../src/interactive/stream-pacer.js"; + +class FakeClock { + nowMs = 0; + private nextId = 1; + private readonly timers = new Map void }>(); + + readonly now = (): number => this.nowMs; + + readonly setTimer = (callback: () => void, delayMs: number): number => { + const id = this.nextId++; + this.timers.set(id, { at: this.nowMs + delayMs, callback }); + return id; + }; + + readonly clearTimer = (handle: unknown): void => { + this.timers.delete(handle as number); + }; + + peekCallback(): (() => void) | null { + return [...this.timers.values()].sort((left, right) => left.at - right.at)[0]?.callback ?? null; + } + + takeCallback(): (() => void) | null { + const next = [...this.timers.entries()].sort((left, right) => left[1].at - right[1].at || left[0] - right[0])[0]; + if (!next) return null; + this.timers.delete(next[0]); + return next[1].callback; + } + + advance(ms: number): void { + const target = this.nowMs + ms; + for (;;) { + const due = [...this.timers.entries()] + .filter(([, timer]) => timer.at <= target) + .sort((left, right) => left[1].at - right[1].at || left[0] - right[0])[0]; + if (!due) break; + const [id, timer] = due; + this.timers.delete(id); + this.nowMs = timer.at; + timer.callback(); + } + this.nowMs = target; + } + + get pendingTimers(): number { + return this.timers.size; + } +} + +function harness(overrides: Partial = {}): { + clock: FakeClock; + slices: StreamPacerSlice[]; + pacer: ReturnType; +} { + const clock = new FakeClock(); + const slices: StreamPacerSlice[] = []; + const pacer = createStreamPacer({ + mode: "on", + onSlice: (slice) => slices.push(slice), + now: clock.now, + setTimer: clock.setTimer, + clearTimer: clock.clearTimer, + tickMs: 25, + baseGraphemesPerSecond: 40, + ...overrides, + }); + return { clock, slices, pacer }; +} + +function admission(sequence: number, text: string, overrides: Record = {}) { + return { + sequence, + generation: "turn-1", + kind: "text" as const, + contentIndex: 0, + text, + ...overrides, + }; +} + +describe("stream pacer semantic classifier", () => { + it("drops only transparent raw text/thinking mirrors", () => { + strictEqual( + classifyStreamEvent({ type: "message_update", assistantMessageEvent: { type: "text_delta" } }), + "transparent-mirror", + ); + strictEqual( + classifyStreamEvent({ type: "message_update", assistantMessageEvent: { type: "thinking_delta" } }), + "transparent-mirror", + ); + strictEqual( + classifyStreamEvent({ type: "message_update", assistantMessageEvent: { type: "toolcall_delta" } }), + "ordered-content-boundary", + ); + }); + + it("separates derived display content, cumulative tool state, input, and conservative boundaries", () => { + deepStrictEqual( + ["text_delta", "thinking_delta", "tool_execution_update", "scroll", "message_end", "new_event"].map((type) => + classifyStreamEvent({ type }), + ), + [ + "paced-display-content", + "paced-display-content", + "cumulative-live-state", + "non-transcript-input", + "ordered-content-boundary", + "ordered-content-boundary", + ], + ); + }); +}); + +describe("stream pacer queue", () => { + it("makes off exact and keeps auto conservative unless explicitly admitted", () => { + const off = harness({ mode: "off" }); + off.pacer.enqueue(admission(1, "whole reply")); + deepStrictEqual( + off.slices.map(({ text, reason }) => [text, reason]), + [["whole reply", "off"]], + ); + strictEqual(off.clock.pendingTimers, 0); + + const automatic = harness({ mode: "auto" }); + automatic.pacer.enqueue(admission(1, "also whole")); + deepStrictEqual( + automatic.slices.map(({ text, reason }) => [text, reason]), + [["also whole", "off"]], + ); + + const enabled = harness({ mode: "auto", isAutoPacingAllowed: () => true, firstSliceGraphemes: 1 }); + enabled.pacer.enqueue(admission(1, "paced")); + enabled.clock.advance(0); + deepStrictEqual( + enabled.slices.map(({ text, reason }) => [text, reason]), + [["p", "first"]], + ); + strictEqual(enabled.clock.pendingTimers, 1); + }); + + it("preserves one FIFO across text, thinking, content indices, and a boundary flush", () => { + const { pacer, slices, clock } = harness({ firstSliceGraphemes: 1 }); + pacer.enqueue(admission(10, "abc")); + clock.advance(0); + pacer.enqueue(admission(11, "THINK", { kind: "thinking" as const, contentIndex: 1 })); + const result = pacer.flush("tool-boundary"); + + deepStrictEqual( + slices.map(({ sequence, kind, contentIndex, text, reason }) => ({ sequence, kind, contentIndex, text, reason })), + [ + { sequence: 10, kind: "text", contentIndex: 0, text: "a", reason: "first" }, + { sequence: 10, kind: "text", contentIndex: 0, text: "bc", reason: "flush" }, + { sequence: 11, kind: "thinking", contentIndex: 1, text: "THINK", reason: "flush" }, + ], + ); + deepStrictEqual(result, { + reason: "tool-boundary", + fromEpoch: 0, + toEpoch: 1, + items: 2, + graphemes: 7, + bytes: 7, + }); + }); + + it("never splits extended grapheme clusters and preserves RTL order", () => { + const { pacer, slices } = harness({ firstSliceGraphemes: 0, maxSliceGraphemes: 1, maxSliceBytes: 4 }); + const clusters = ["e\u0301", "👩‍🔬", "🇺🇳", "שָ", "ל", "ו", "ם"]; + pacer.enqueue(admission(1, clusters.join(""))); + while (pacer.snapshot().queuedGraphemes > 0) pacer.dequeue(1); + + deepStrictEqual( + slices.map(({ text }) => text), + clusters, + ); + strictEqual(slices.map(({ text }) => text).join(""), clusters.join("")); + ok( + slices.some(({ bytes }) => bytes > 4), + "one indivisible ZWJ grapheme may exceed the byte slice cap", + ); + }); + + it("carries fractional arrival/time credit instead of rounding every tick", () => { + const { pacer, slices, clock } = harness({ + firstSliceGraphemes: 0, + baseGraphemesPerSecond: 10, + arrivalCreditRatio: 0.25, + catchUpWindowMs: Number.MAX_VALUE, + maxOldestAgeMs: 1_000, + }); + pacer.enqueue(admission(1, "abc")); + strictEqual(pacer.snapshot().credit, 0.75); + clock.advance(24); + strictEqual(slices.length, 0); + clock.advance(1); + deepStrictEqual( + slices.map(({ text }) => text), + ["a"], + ); + ok(pacer.snapshot().credit < 0.01, `fractional remainder was ${pacer.snapshot().credit}`); + }); + + it("clamps elapsed credit after event-loop suspension", () => { + const { pacer, slices, clock } = harness({ + firstSliceGraphemes: 0, + baseGraphemesPerSecond: 10, + arrivalCreditRatio: 0, + catchUpWindowMs: Number.MAX_VALUE, + maxElapsedMs: 100, + maxOldestAgeMs: 20_000, + }); + pacer.enqueue(admission(1, "abcdefghij")); + const delayedTick = clock.takeCallback(); + ok(delayedTick); + clock.nowMs = 10_000; + delayedTick(); + + deepStrictEqual( + slices.map(({ text }) => text), + ["a"], + ); + strictEqual(pacer.snapshot().queuedGraphemes, 9); + }); + + it("enforces the oldest-grapheme deadline even below the nominal rate", () => { + const { pacer, slices, clock } = harness({ + firstSliceGraphemes: 0, + baseGraphemesPerSecond: 0.01, + arrivalCreditRatio: 0, + catchUpWindowMs: Number.MAX_VALUE, + maxOldestAgeMs: 50, + maxSliceGraphemes: 2, + }); + pacer.enqueue(admission(1, "abcdef")); + clock.advance(49); + strictEqual(slices.length, 0); + clock.advance(1); + strictEqual(slices.map(({ text }) => text).join(""), "abcdef"); + ok(slices.every(({ reason, graphemes }) => reason === "deadline" && graphemes <= 2)); + strictEqual(pacer.snapshot().queuedGraphemes, 0); + strictEqual(clock.pendingTimers, 0); + }); + + it("self-stops while idle and restarts without banking elapsed-time credit", () => { + const { pacer, slices, clock } = harness({ + firstSliceGraphemes: 0, + arrivalCreditRatio: 0, + catchUpWindowMs: Number.MAX_VALUE, + maxOldestAgeMs: 1_000, + }); + pacer.enqueue(admission(1, "a")); + clock.advance(25); + strictEqual(clock.pendingTimers, 0); + clock.advance(10_000); + pacer.enqueue(admission(2, "bcdef")); + clock.advance(25); + + deepStrictEqual( + slices.map(({ text }) => text), + ["a", "b"], + ); + strictEqual(clock.pendingTimers, 1, "only live backlog owns a timer"); + }); + + it("drains toward absolute grapheme and byte bounds without reordering", () => { + const { pacer, slices } = harness({ + firstSliceGraphemes: 0, + arrivalCreditRatio: 0, + maxQueueGraphemes: 3, + maxQueueBytes: 3, + maxSliceGraphemes: 2, + maxSliceBytes: 2, + }); + pacer.enqueue(admission(1, "abcdef")); + deepStrictEqual( + slices.map(({ text, reason }) => [text, reason]), + [ + ["ab", "capacity"], + ["cd", "capacity"], + ], + ); + deepStrictEqual(pacer.snapshot(), { + mode: "on", + epoch: 0, + queuedItems: 1, + queuedGraphemes: 2, + queuedBytes: 2, + oldestAgeMs: 0, + credit: 0, + timerPending: true, + disposed: false, + }); + pacer.flush("finish"); + strictEqual(slices.map(({ text }) => text).join(""), "abcdef"); + }); + + it("consumes folded thinking as one backing-state mutation after older text", () => { + const { pacer, slices } = harness({ firstSliceGraphemes: 0, maxSliceGraphemes: 3 }); + pacer.enqueue(admission(1, "abc")); + pacer.enqueue(admission(2, "invisible but retained", { kind: "thinking" as const, folded: true })); + strictEqual(slices.length, 0, "folded state cannot overtake older visible text"); + pacer.dequeue(3); + + deepStrictEqual( + slices.map(({ sequence, text, reason }) => [sequence, text, reason]), + [ + [1, "abc", "tick"], + [2, "invisible but retained", "folded"], + ], + ); + strictEqual(slices[1]?.finalForItem, true); + strictEqual(pacer.snapshot().queuedGraphemes, 0); + }); + + it("invalidates captured epochs and makes a queued stale timer harmless", () => { + const { pacer, slices, clock } = harness({ firstSliceGraphemes: 0 }); + const oldEpoch = pacer.epoch; + pacer.enqueue(admission(1, "stale", { epoch: oldEpoch })); + const staleCallback = clock.peekCallback(); + ok(staleCallback); + strictEqual(pacer.invalidateEpoch(), oldEpoch + 1); + strictEqual(pacer.enqueue(admission(2, "rejected", { epoch: oldEpoch })).accepted, false); + staleCallback(); + deepStrictEqual(slices, []); + strictEqual(pacer.snapshot().queuedItems, 0); + }); + + it("does not let a cleared same-epoch timer consume or unschedule replacement work", () => { + const { pacer, slices, clock } = harness({ firstSliceGraphemes: 0, arrivalCreditRatio: 0 }); + pacer.enqueue(admission(1, "old")); + const staleCallback = clock.peekCallback(); + ok(staleCallback); + pacer.dequeue(3); + pacer.enqueue(admission(2, "new")); + strictEqual(clock.pendingTimers, 1); + staleCallback(); + + deepStrictEqual( + slices.map(({ text }) => text), + ["old"], + ); + strictEqual(clock.pendingTimers, 1, "stale callback cannot clear the replacement timer"); + clock.advance(25); + deepStrictEqual( + slices.map(({ text }) => text), + ["old", "n"], + ); + }); + + it("flushes abort content in exact order and rejects every post-abort stale admission", () => { + const log: string[] = []; + const clock = new FakeClock(); + const pacer = createStreamPacer({ + mode: "on", + onSlice: ({ text, reason }) => log.push(`${reason}:${text}`), + now: clock.now, + setTimer: clock.setTimer, + clearTimer: clock.clearTimer, + firstSliceGraphemes: 1, + }); + const epoch = pacer.epoch; + pacer.enqueue(admission(1, "abc", { epoch })); + clock.advance(0); + pacer.enqueue(admission(2, "def", { epoch })); + pacer.flush("abort"); + log.push("boundary:abort"); + strictEqual(pacer.enqueue(admission(3, "late", { epoch })).accepted, false); + + deepStrictEqual(log, ["first:a", "flush:bc", "flush:def", "boundary:abort"]); + strictEqual(clock.pendingTimers, 0); + }); + + it("requires a flush between generations and settles exactly once on dispose", () => { + const { pacer, slices, clock } = harness({ firstSliceGraphemes: 0 }); + pacer.enqueue(admission(1, "one")); + throws( + () => pacer.enqueue(admission(2, "two", { generation: "turn-2" })), + /error changed with queued content|generation changed/, + ); + const first = pacer.dispose("shutdown"); + const second = pacer.dispose("shutdown-again"); + strictEqual(slices.map(({ text }) => text).join(""), "one"); + strictEqual(first.graphemes, 3); + strictEqual(second.graphemes, 0); + strictEqual(clock.pendingTimers, 0); + strictEqual(pacer.enqueue(admission(3, "never")).accepted, false); + }); +}); diff --git a/tests/contracts/stream-pacing-policy.test.ts b/tests/contracts/stream-pacing-policy.test.ts new file mode 100644 index 000000000..e009a6fa2 --- /dev/null +++ b/tests/contracts/stream-pacing-policy.test.ts @@ -0,0 +1,33 @@ +import { deepStrictEqual, strictEqual } from "node:assert/strict"; +import { describe, it } from "node:test"; +import { autoPacingAllowed, resolveSmoothStreamingMode } from "../../src/interactive/stream-pacing-policy.js"; + +describe("stream pacing policy", () => { + it("gives valid process overrides precedence and makes invalid values fail off", () => { + deepStrictEqual( + [undefined, "0", "off", "auto", "1", "on", "unexpected"].map((value) => + resolveSmoothStreamingMode("auto", value === undefined ? {} : { CLIO_CODER_SMOOTH_STREAM: value }), + ), + ["auto", "off", "off", "auto", "on", "on", "off"], + ); + }); + + it("admits auto only for a local capable TTY with no accessibility or output risk", () => { + const base = { isTTY: true, term: "xterm-256color", backpressureObserved: false }; + strictEqual(autoPacingAllowed(base), true); + for (const unsafe of [ + { isTTY: false }, + { term: "dumb" }, + { sshConnection: "host" }, + { sshTty: "/dev/pts/1" }, + { tmux: "/tmp/tmux" }, + { sty: "screen" }, + { ci: "1" }, + { reducedMotion: "1" }, + { screenReader: "1" }, + { backpressureObserved: true }, + ]) { + strictEqual(autoPacingAllowed({ ...base, ...unsafe }), false, JSON.stringify(unsafe)); + } + }); +}); diff --git a/tests/harness/openai-compat-fixture.ts b/tests/harness/openai-compat-fixture.ts index 2012808ed..3a3b3f27c 100644 --- a/tests/harness/openai-compat-fixture.ts +++ b/tests/harness/openai-compat-fixture.ts @@ -44,6 +44,10 @@ export interface OpenAICompatToolCallScript { export interface OpenAICompatFixtureOptions { models?: Array & { id: string }>; + /** Split a text reply into provider-visible SSE deltas for pacing/ordering tests. */ + replyChunks?: readonly string[]; + /** Optional deterministic delay between text chunks. */ + chunkDelayMs?: number; /** * When set, answer the tool-free completion of every turn with this tool * call instead of text. Absent, the fixture behaves exactly as before. @@ -150,15 +154,22 @@ export async function startOpenAICompatFixture( res.end("data: [DONE]\n\n"); return; } - res.write( - `data: ${JSON.stringify({ - id: "chatcmpl-clio-print", - object: "chat.completion.chunk", - created: 1, - model: "mock-model", - choices: [{ index: 0, delta: { content: reply } }], - })}\n\n`, - ); + const replyChunks = options.replyChunks ?? [reply]; + for (let index = 0; index < replyChunks.length; index += 1) { + const chunk = replyChunks[index] ?? ""; + res.write( + `data: ${JSON.stringify({ + id: "chatcmpl-clio-print", + object: "chat.completion.chunk", + created: 1, + model: "mock-model", + choices: [{ index: 0, delta: { content: chunk } }], + })}\n\n`, + ); + if ((options.chunkDelayMs ?? 0) > 0 && index + 1 < replyChunks.length) { + await new Promise((resolve) => setTimeout(resolve, options.chunkDelayMs)); + } + } res.write( `data: ${JSON.stringify({ id: "chatcmpl-clio-print", diff --git a/tests/smoke/render-trace-pty.test.ts b/tests/smoke/render-trace-pty.test.ts index 7650e06fa..c7ac5fc22 100644 --- a/tests/smoke/render-trace-pty.test.ts +++ b/tests/smoke/render-trace-pty.test.ts @@ -5,7 +5,7 @@ * not call an in-process timestamp "glass" latency: a terminal emulator or * external observation harness is required to measure a displayed pixel. */ -import { ok, strictEqual } from "node:assert/strict"; +import { deepStrictEqual, ok, strictEqual } from "node:assert/strict"; import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; @@ -13,6 +13,7 @@ import { describe, it } from "node:test"; import { stringify } from "yaml"; import { DEFAULT_SETTINGS } from "../../src/core/defaults.js"; import type { RenderTraceFrameRecord, RenderTraceRecord } from "../../src/interactive/render-trace.js"; +import { closeServer, startOpenAICompatFixture } from "../harness/openai-compat-fixture.js"; import { openPty, ptySupported, stripAnsi } from "../harness/pty.js"; const REPO_ROOT = new URL("../..", import.meta.url).pathname; @@ -22,30 +23,63 @@ const READY = /ctx /; interface Scratch { dir: string; tracePath: string; + backpressureArmPath: string; env: Record; cleanup(): void; } -function makeScratch(): Scratch { +function makeScratch( + options: { providerUrl?: string; smoothStreaming?: "off" | "auto" | "on"; simulateBackpressure?: boolean } = {}, +): Scratch { const dir = mkdtempSync(join(tmpdir(), "clio-render-trace-pty-")); const configDir = join(dir, "config"); mkdirSync(configDir, { recursive: true }); const settings = structuredClone(DEFAULT_SETTINGS) as Record; + const targetId = options.providerUrl ? "mock-chat" : "declared"; + const modelId = options.providerUrl ? "mock-model" : "declared-model"; settings.targets = [ { - id: "declared", + id: targetId, runtime: "openai-compat", - url: "http://127.0.0.1:9", - defaultModel: "declared-model", + url: options.providerUrl ?? "http://127.0.0.1:9", + defaultModel: modelId, lifecycle: "user-managed", + ...(options.providerUrl ? { auth: { apiKeyEnvVar: "CLIO_CODER_TEST_OPENAI_KEY" }, wireModels: [modelId] } : {}), }, ]; - settings.orchestrator = { target: "declared", model: "declared-model", thinkingLevel: "off" }; + settings.orchestrator = { target: targetId, model: modelId, thinkingLevel: "off" }; + (settings.terminal as Record).smoothStreaming = options.smoothStreaming ?? "off"; writeFileSync(join(configDir, "settings.yaml"), stringify(settings), "utf8"); const tracePath = join(dir, "render.jsonl"); + const backpressureArmPath = join(dir, "arm-backpressure"); + const backpressurePreload = join(dir, "stdout-backpressure.mjs"); + if (options.simulateBackpressure) { + writeFileSync( + backpressurePreload, + [ + "import { existsSync } from 'node:fs';", + "const stdout = process.stdout;", + "const originalWrite = stdout.write.bind(stdout);", + "let injected = false;", + `const armPath = ${JSON.stringify(backpressureArmPath)};`, + "stdout.write = function controlledWrite(chunk, encoding, callback) {", + " const returned = originalWrite(chunk, encoding, callback);", + " if (!injected && existsSync(armPath)) {", + " injected = true;", + " setTimeout(() => stdout.emit('drain'), 250);", + " return false;", + " }", + " return returned;", + "};", + ].join("\n"), + "utf8", + ); + } + const inheritedNodeOptions = process.env.NODE_OPTIONS?.trim(); return { dir, tracePath, + backpressureArmPath, env: { ...process.env, CLIO_CODER_HOME: dir, @@ -56,7 +90,11 @@ function makeScratch(): Scratch { CLIO_CODER_RESIDENCY: "observe", CLIO_CODER_RENDER_TRACE: tracePath, CLIO_CODER_TRACE_BOOT: "1", + CLIO_CODER_TEST_OPENAI_KEY: "sk-test", TERM: "xterm-256color", + ...(options.simulateBackpressure + ? { NODE_OPTIONS: [inheritedNodeOptions, `--import=${backpressurePreload}`].filter(Boolean).join(" ") } + : {}), } as Record, cleanup() { rmSync(dir, { recursive: true, force: true }); @@ -220,6 +258,12 @@ describe("render trace through a real PTY", { })}\n`, ); } + } catch (error) { + const traceTypes = readTrace(scratch.tracePath).map((record) => record.type); + throw new Error( + `${error instanceof Error ? error.message : String(error)}; trace=${JSON.stringify(traceTypes.slice(-40))}; output=${JSON.stringify(stripAnsi(session.output).slice(-800))}`, + { cause: error }, + ); } finally { if (!session.exited) { session.resumeOutput(); @@ -249,4 +293,128 @@ describe("render trace through a real PTY", { strictEqual(settledAgain.exitCode, first.exitCode); strictEqual(settledAgain.signal, first.signal); }); + + it("paces provider deltas to a final committed frame and recovers from real PTY backpressure", async () => { + const pacedMiddle = "x".repeat(4 * 1_024); + const replyChunks = ["paced-start-", "👩‍🔬", pacedMiddle, "-paced-final"]; + const fixture = await startOpenAICompatFixture(replyChunks.join(""), { replyChunks, chunkDelayMs: 2 }); + const scratch = makeScratch({ + providerUrl: fixture.url, + smoothStreaming: "on", + simulateBackpressure: true, + }); + const session = await openPty( + process.execPath, + [join(REPO_ROOT, "dist", "cli", "index.js"), "--no-context-files", "--no-skills"], + { + cols: 80, + rows: 24, + cwd: REPO_ROOT, + env: scratch.env, + }, + ); + try { + await session.waitForOutput((output) => READY.test(stripAnsi(output)), 30_000); + session.pauseOutput(); + writeFileSync(scratch.backpressureArmPath, "armed\n", "utf8"); + session.write("exercise paced PTY output"); + await waitForTrace( + scratch.tracePath, + (records) => records.find((record) => record.type === "input_ingress" && record.action === "editor"), + "editor ingress before paced submit", + ); + session.write("\r"); + const submitIngress = await waitForTrace( + scratch.tracePath, + (records) => records.find((record) => record.type === "input_ingress" && record.action === "submit"), + "semantic submit ingress", + ); + ok(submitIngress.type === "input_ingress"); + + const backpressuredWrite = await waitForTrace( + scratch.tracePath, + (records) => records.find((record) => record.type === "terminal_write" && record.returned === false), + "stdout.write() backpressure against a paused PTY", + 30_000, + ); + ok(backpressuredWrite.type === "terminal_write" && backpressuredWrite.backpressured); + const ingress = await waitForTrace( + scratch.tracePath, + (records) => { + const visible = records.filter((record) => record.type === "event_ingress"); + return visible.length >= 2 ? visible : undefined; + }, + "multiple provider delta ingress records", + ); + const lastIngress = ingress.at(-1); + ok(lastIngress?.type === "event_ingress"); + + session.resumeOutput(); + await session.waitForOutput((output) => stripAnsi(output).includes("paced-final"), 30_000); + const drained = await waitForTrace( + scratch.tracePath, + (records) => records.find((record) => record.type === "terminal_drain"), + "matching stdout drain", + ); + ok(drained.type === "terminal_drain"); + const finalFrame = await waitForTrace( + scratch.tracePath, + (records) => + frames(records).find((frame) => frame.panelHighWater >= lastIngress.eventSeq && frame.commits.length > 0), + "first committed frame containing the final paced delta", + ); + ok(finalFrame.panelHighWater >= lastIngress.eventSeq); + const records = readTrace(scratch.tracePath); + for (const event of ingress) { + const actions = records.flatMap((record) => + record.type === "queue" && record.eventSeq === event.eventSeq ? [record.action] : [], + ); + deepStrictEqual(actions, ["admit", "dequeue"], `event ${event.eventSeq} settles exactly once`); + } + if (process.env.CLIO_CODER_PERF_REPORT === "1") { + const submitFrame = frames(records).find( + (frame) => frame.inputHighWater >= submitIngress.inputSeq && frame.commits.length > 0, + ); + const firstVisibleFrame = frames(records).find( + (frame) => frame.panelHighWater >= (ingress[0]?.eventSeq ?? 0) && frame.commits.length > 0, + ); + process.stdout.write( + `${JSON.stringify({ + node: process.versions.node, + mode: "on", + providerIngressEvents: ingress.length, + inputToStdoutCommitMs: + submitFrame && submitIngress.type === "input_ingress" + ? (submitFrame.commits[0]?.at ?? submitFrame.endAt) - submitIngress.at + : null, + firstIngressToStdoutCommitMs: + firstVisibleFrame && ingress[0]?.type === "event_ingress" + ? (firstVisibleFrame.commits[0]?.at ?? firstVisibleFrame.endAt) - ingress[0].at + : null, + finalIngressToStdoutCommitMs: (finalFrame.commits[0]?.at ?? finalFrame.endAt) - lastIngress.at, + controlledBackpressureWaitMs: drained.at - backpressuredWrite.at, + })}\n`, + ); + } + + session.write(CTRL_C); + await new Promise((resolve) => setTimeout(resolve, 75)); + session.write(CTRL_C); + const exit = await session.waitForExit(10_000); + strictEqual(exit.exitCode, 0, `clean PTY exit; output tail: ${stripAnsi(session.output).slice(-400)}`); + } catch (error) { + const traceTypes = readTrace(scratch.tracePath).map((record) => record.type); + throw new Error( + `${error instanceof Error ? error.message : String(error)}; requests=${fixture.requests.length}; trace=${JSON.stringify(traceTypes.slice(-40))}; output=${JSON.stringify(stripAnsi(session.output).slice(-800))}`, + { cause: error }, + ); + } finally { + if (!session.exited) { + session.resumeOutput(); + await session.killAndWaitForExit(); + } + scratch.cleanup(); + await closeServer(fixture.server); + } + }); });