From 85b1070a235a904da60d72f4b4c48819ed6f4df4 Mon Sep 17 00:00:00 2001 From: Leo Date: Thu, 10 Sep 2026 14:16:44 -0400 Subject: [PATCH 1/2] perf(client): stop replaying terminal buffers on rollover Keep terminal attach output as incremental chunks with a 512KiB bound and a cursor so renderers append instead of rewriting the whole buffer on every event. Mobile still materializes a string at the native surface boundary. --- .../features/terminal/terminalMenu.test.ts | 7 +- apps/mobile/src/state/use-terminal-session.ts | 22 +- .../src/components/ThreadTerminalDrawer.tsx | 52 ++- packages/client-runtime/src/state/terminal.ts | 22 +- .../src/state/terminalOutput.ts | 320 ++++++++++++++++++ .../src/state/terminalSession.test.ts | 273 ++++++++++++++- .../src/state/terminalSession.ts | 71 ++-- 7 files changed, 705 insertions(+), 62 deletions(-) create mode 100644 packages/client-runtime/src/state/terminalOutput.ts diff --git a/apps/mobile/src/features/terminal/terminalMenu.test.ts b/apps/mobile/src/features/terminal/terminalMenu.test.ts index 966312270951..3e09dc8721f6 100644 --- a/apps/mobile/src/features/terminal/terminalMenu.test.ts +++ b/apps/mobile/src/features/terminal/terminalMenu.test.ts @@ -1,6 +1,9 @@ import { describe, expect, it } from "vite-plus/test"; -import { type KnownTerminalSession } from "@t3tools/client-runtime/state/terminal"; +import { + EMPTY_TERMINAL_BUFFER_STATE, + type KnownTerminalSession, +} from "@t3tools/client-runtime/state/terminal"; import { DEFAULT_TERMINAL_ID, EnvironmentId, ThreadId } from "@t3tools/contracts"; import { getTerminalLabel } from "@t3tools/shared/terminalLabels"; @@ -55,7 +58,7 @@ function makeKnownSession(input: { updatedAt: input.updatedAt ?? "2026-04-15T20:00:00.000Z", } : null, - buffer: "", + output: EMPTY_TERMINAL_BUFFER_STATE.output, status: input.status, error: null, hasRunningSubprocess: false, diff --git a/apps/mobile/src/state/use-terminal-session.ts b/apps/mobile/src/state/use-terminal-session.ts index 328557a2005d..6be57007a60f 100644 --- a/apps/mobile/src/state/use-terminal-session.ts +++ b/apps/mobile/src/state/use-terminal-session.ts @@ -2,6 +2,7 @@ import { combineTerminalSessionState, EMPTY_TERMINAL_BUFFER_STATE, EMPTY_TERMINAL_SESSION_STATE, + terminalOutputText, type KnownTerminalSession, type TerminalSessionState, } from "@t3tools/client-runtime/state/terminal"; @@ -11,10 +12,16 @@ import { useMemo } from "react"; import { useEnvironmentQuery } from "./query"; import { terminalEnvironment } from "./terminal"; +type LegacyTerminalSessionState = TerminalSessionState & { readonly buffer: string }; +const EMPTY_LEGACY_TERMINAL_SESSION_STATE: LegacyTerminalSessionState = { + ...EMPTY_TERMINAL_SESSION_STATE, + buffer: "", +}; + export function useAttachedTerminalSession(input: { readonly environmentId: EnvironmentId | null; readonly terminal: TerminalAttachInput | null; -}): TerminalSessionState { +}): LegacyTerminalSessionState { const attach = useEnvironmentQuery( input.environmentId !== null && input.terminal !== null ? terminalEnvironment.attach({ @@ -31,10 +38,14 @@ export function useAttachedTerminalSession(input: { input: null, }), ); + const output = attach.data?.output ?? EMPTY_TERMINAL_BUFFER_STATE.output; + // Installed native binaries still accept initialBuffer. Keep materialization + // at this mobile boundary until the native streaming API is released. + const buffer = useMemo(() => terminalOutputText(output), [output]); return useMemo(() => { if (input.environmentId === null || input.terminal === null) { - return EMPTY_TERMINAL_SESSION_STATE; + return EMPTY_LEGACY_TERMINAL_SESSION_STATE; } const summary = metadata.data?.find( @@ -42,9 +53,12 @@ export function useAttachedTerminalSession(input: { terminal.threadId === input.terminal?.threadId && terminal.terminalId === input.terminal?.terminalId, ) ?? null; - const state = combineTerminalSessionState(summary, attach.data ?? EMPTY_TERMINAL_BUFFER_STATE); + const state = { + ...combineTerminalSessionState(summary, attach.data ?? EMPTY_TERMINAL_BUFFER_STATE), + buffer, + }; return attach.error === null ? state : { ...state, error: attach.error, status: "error" }; - }, [attach.data, attach.error, input.environmentId, input.terminal, metadata.data]); + }, [attach.data, attach.error, buffer, input.environmentId, input.terminal, metadata.data]); } export function useKnownTerminalSessions(input: { diff --git a/apps/web/src/components/ThreadTerminalDrawer.tsx b/apps/web/src/components/ThreadTerminalDrawer.tsx index 64a377b7c39f..274d4ee6f492 100644 --- a/apps/web/src/components/ThreadTerminalDrawer.tsx +++ b/apps/web/src/components/ThreadTerminalDrawer.tsx @@ -3,7 +3,13 @@ import { isAtomCommandInterrupted, squashAtomCommandFailure, } from "@t3tools/client-runtime/state/runtime"; -import { type TerminalSessionState } from "@t3tools/client-runtime/state/terminal"; +import { + INITIAL_TERMINAL_OUTPUT_CURSOR, + readTerminalOutputUpdate, + type TerminalOutputCursor, + type TerminalOutputUpdate, + type TerminalSessionState, +} from "@t3tools/client-runtime/state/terminal"; import { Plus, Square, @@ -95,8 +101,15 @@ function writeSystemMessage(terminal: GhosttyTerminalSurface, message: string): terminal.write(`\r\n[terminal] ${message}\r\n`); } -function writeTerminalBuffer(terminal: GhosttyTerminalSurface, buffer: string): void { - terminal.resetAndWrite(buffer); +export function writeTerminalOutputUpdate( + terminal: Pick, + update: TerminalOutputUpdate, +): void { + if (update.type === "reset") { + terminal.resetAndWrite(update.data); + } else if (update.type === "append") { + terminal.write(update.data); + } } function parseTerminalColor(value: string, fallback: GhosttyColor): GhosttyColor { @@ -425,9 +438,10 @@ export function TerminalViewport({ input: { threadId, terminalId, cols, rows }, }), ); - const terminalBuffer = terminalSession.buffer; + const terminalOutput = terminalSession.output; const terminalError = terminalSession.error; const terminalStatus = terminalSession.status; + const outputCursorRef = useRef(INITIAL_TERMINAL_OUTPUT_CURSOR); const synchronizedStatusRef = useRef("closed"); const synchronizeTerminalStatus = useEffectEvent( (terminal: GhosttyTerminalSurface, status: TerminalSessionState["status"]) => { @@ -448,14 +462,14 @@ export function TerminalViewport({ ); const terminalVersion = terminalSession.version; const previousSessionRef = useRef({ - buffer: terminalBuffer, + output: terminalOutput, error: terminalError, status: terminalStatus, version: terminalVersion, }); const latestSessionRef = useRef(previousSessionRef.current); latestSessionRef.current = { - buffer: terminalBuffer, + output: terminalOutput, error: terminalError, status: terminalStatus, version: terminalVersion, @@ -518,7 +532,14 @@ export function TerminalViewport({ } const latestSession = latestSessionRef.current; previousSessionRef.current = latestSession; - if (latestSession.buffer.length > 0) terminal.resetAndWrite(latestSession.buffer); + const initialOutput = readTerminalOutputUpdate( + latestSession.output, + INITIAL_TERMINAL_OUTPUT_CURSOR, + ); + if (initialOutput.type === "reset" && initialOutput.data.length > 0) { + writeTerminalOutputUpdate(terminal, initialOutput); + } + outputCursorRef.current = initialOutput.cursor; if (latestSession.error !== null) writeSystemMessage(terminal, latestSession.error); // Attaching to a session that already exited must still run exit handling // once, so mount synchronization starts from the empty "closed" state. @@ -910,7 +931,7 @@ export function TerminalViewport({ useEffect(() => { const terminal = terminalRef.current; const current = { - buffer: terminalBuffer, + output: terminalOutput, error: terminalError, status: terminalStatus, version: terminalVersion, @@ -922,18 +943,13 @@ export function TerminalViewport({ const previous = previousSessionRef.current; synchronizeTerminalStatus(terminal, current.status); - if (current.version === previous.version) { + if (current.version === previous.version && current.output === previous.output) { return; } - if ( - current.buffer.length >= previous.buffer.length && - current.buffer.startsWith(previous.buffer) - ) { - terminal.write(current.buffer.slice(previous.buffer.length)); - } else { - writeTerminalBuffer(terminal, current.buffer); - } + const outputUpdate = readTerminalOutputUpdate(current.output, outputCursorRef.current); + writeTerminalOutputUpdate(terminal, outputUpdate); + outputCursorRef.current = outputUpdate.cursor; terminal.clearSelection(); if (current.error !== null && current.error !== previous.error) { @@ -946,7 +962,7 @@ export function TerminalViewport({ }); } previousSessionRef.current = current; - }, [autoFocus, terminalBuffer, terminalError, terminalStatus, terminalVersion]); + }, [autoFocus, terminalOutput, terminalError, terminalStatus, terminalVersion]); useEffect(() => { if (!autoFocus) return; diff --git a/packages/client-runtime/src/state/terminal.ts b/packages/client-runtime/src/state/terminal.ts index 028f7a8c6609..cc96dbe4b812 100644 --- a/packages/client-runtime/src/state/terminal.ts +++ b/packages/client-runtime/src/state/terminal.ts @@ -13,7 +13,23 @@ import { subscribe, type EnvironmentRpcInput } from "../rpc/client.ts"; import { applyTerminalAttachStreamEvent, applyTerminalMetadataStreamEvent, + nextTerminalAttachSeedState, +} from "./terminalSession.ts"; + +export { + applyTerminalAttachStreamEvent, + combineTerminalSessionState, + DEFAULT_MAX_TERMINAL_BUFFER_BYTES, EMPTY_TERMINAL_BUFFER_STATE, + EMPTY_TERMINAL_SESSION_STATE, + INITIAL_TERMINAL_OUTPUT_CURSOR, + nextTerminalAttachSeedState, + readTerminalOutputUpdate, + selectRunningSubprocessTerminalIds, + terminalOutputText, + type KnownTerminalSession, + type TerminalBufferState, + type TerminalSessionState, } from "./terminalSession.ts"; export function createTerminalEnvironmentAtoms( @@ -40,8 +56,10 @@ export function createTerminalEnvironmentAtoms( attach: createEnvironmentSubscriptionAtomFamily(runtime, { label: "environment-data:terminal:attach", subscribe: (input: EnvironmentRpcInput) => - subscribe(WS_METHODS.terminalAttach, input).pipe( - Stream.scan(EMPTY_TERMINAL_BUFFER_STATE, applyTerminalAttachStreamEvent), + Stream.suspend(() => + subscribe(WS_METHODS.terminalAttach, input).pipe( + Stream.scan(nextTerminalAttachSeedState(), applyTerminalAttachStreamEvent), + ), ), }), events: createEnvironmentRpcSubscriptionAtomFamily(runtime, { diff --git a/packages/client-runtime/src/state/terminalOutput.ts b/packages/client-runtime/src/state/terminalOutput.ts new file mode 100644 index 000000000000..fcb0389c434b --- /dev/null +++ b/packages/client-runtime/src/state/terminalOutput.ts @@ -0,0 +1,320 @@ +export interface TerminalOutputChunk { + /** UTF-16 string offset within this generation and reset. */ + readonly startOffset: number; + readonly data: string; + readonly byteLength: number; +} + +export interface TerminalOutputState { + readonly generation: number; + readonly chunks: ReadonlyArray; + readonly retainedBytes: number; + readonly resetVersion: number; + readonly nextOffset: number; +} + +export interface TerminalOutputCursor { + readonly generation: number; + readonly resetVersion: number; + readonly offset: number; +} + +/** Forces the first `readTerminalOutputUpdate` to resynchronize from a reset. */ +export const INITIAL_TERMINAL_OUTPUT_CURSOR = Object.freeze({ + generation: -1, + resetVersion: -1, + offset: 0, +}); + +export type TerminalOutputUpdate = + | { + readonly type: "none"; + readonly cursor: TerminalOutputCursor; + } + | { + readonly type: "reset"; + readonly data: string; + readonly cursor: TerminalOutputCursor; + } + | { + readonly type: "append"; + readonly cursor: TerminalOutputCursor; + readonly data: string; + }; + +export const DEFAULT_MAX_TERMINAL_BUFFER_BYTES = 512 * 1024; +const DEFAULT_TERMINAL_CHUNK_BYTES = 16 * 1024; +const MAX_TERMINAL_OUTPUT_CHUNKS = 1_024; +const textEncoder = new TextEncoder(); +// A BOM at a retained chunk boundary is terminal data, not an encoding marker. +const textDecoder = new TextDecoder("utf-8", { ignoreBOM: true }); + +export const EMPTY_TERMINAL_OUTPUT_STATE = Object.freeze({ + generation: 0, + chunks: Object.freeze([]), + retainedBytes: 0, + resetVersion: 0, + nextOffset: 0, +}); + +interface Utf8Chunk { + readonly data: string; + readonly byteLength: number; +} + +/** + * Split a string into chunks of at most `maxBytes` UTF-8 bytes without cutting + * a code point in half. The retained-output budget always supplies a positive + * size. Only new output is encoded on live updates. + * + * A chunk that fits whole is returned as the original string, so the common + * small-write path pays one encode and no decode. + */ +function splitStringByUtf8Bytes(data: string, maxBytes: number): ReadonlyArray { + if (data.length === 0) return []; + + const encoded = textEncoder.encode(data); + if (encoded.byteLength <= maxBytes) { + return [{ data, byteLength: encoded.byteLength }]; + } + + const chunks: Utf8Chunk[] = []; + let offset = 0; + while (offset < encoded.byteLength) { + let end = Math.min(offset + maxBytes, encoded.byteLength); + while (end < encoded.byteLength && ((encoded[end] ?? 0) & 0xc0) === 0x80) { + end -= 1; + } + // A degenerate budget smaller than one code point still has to advance: + // include the whole code point rather than looping forever. + if (end === offset) { + end = Math.min(offset + maxBytes, encoded.byteLength); + while (end < encoded.byteLength && ((encoded[end] ?? 0) & 0xc0) === 0x80) { + end += 1; + } + } + const bytes = encoded.subarray(offset, end); + chunks.push({ data: textDecoder.decode(bytes), byteLength: bytes.byteLength }); + offset = end; + } + + return chunks; +} + +function trimBufferToBytes(buffer: string, maxBufferBytes: number): string { + if (maxBufferBytes <= 0) { + return ""; + } + + const encoded = textEncoder.encode(buffer); + if (encoded.byteLength <= maxBufferBytes) { + return buffer; + } + + let start = encoded.byteLength - maxBufferBytes; + while (start < encoded.length) { + const byte = encoded[start]; + if (byte === undefined || (byte & 0b1100_0000) !== 0b1000_0000) { + break; + } + start += 1; + } + + return textDecoder.decode(encoded.subarray(start)); +} + +function splitOutputChunks( + data: string, + firstOffset: number, + maxChunkBytes = DEFAULT_TERMINAL_CHUNK_BYTES, +): { + readonly chunks: ReadonlyArray; + readonly nextOffset: number; + readonly byteLength: number; +} { + const split = splitStringByUtf8Bytes(data, maxChunkBytes); + let byteLength = 0; + let nextOffset = firstOffset; + const chunks = split.map((chunk) => { + byteLength += chunk.byteLength; + const startOffset = nextOffset; + nextOffset += chunk.data.length; + return { + startOffset, + data: chunk.data, + byteLength: chunk.byteLength, + }; + }); + + return { + chunks, + nextOffset, + byteLength, + }; +} + +/** + * Merge adjacent chunks without changing their string positions. A reader can + * still append the unread suffix when its cursor falls inside a merged chunk. + */ +function compactRetainedChunks(chunks: ReadonlyArray) { + const compacted: TerminalOutputChunk[] = []; + for (const chunk of chunks) { + const previous = compacted.at(-1); + if ( + previous !== undefined && + previous.startOffset + previous.data.length === chunk.startOffset && + previous.byteLength + chunk.byteLength <= DEFAULT_TERMINAL_CHUNK_BYTES + ) { + compacted[compacted.length - 1] = { + startOffset: previous.startOffset, + data: `${previous.data}${chunk.data}`, + byteLength: previous.byteLength + chunk.byteLength, + }; + } else { + compacted.push(chunk); + } + } + return compacted; +} + +// Scan only the removed prefix instead of encoding retained output again. +function trimOutputChunkStart( + chunk: TerminalOutputChunk, + bytesToDrop: number, +): TerminalOutputChunk { + let offset = 0; + let droppedBytes = 0; + while (droppedBytes < bytesToDrop && offset < chunk.data.length) { + const codepoint = chunk.data.codePointAt(offset)!; + droppedBytes += codepoint <= 0x7f ? 1 : codepoint <= 0x7ff ? 2 : codepoint <= 0xffff ? 3 : 4; + offset += codepoint <= 0xffff ? 1 : 2; + } + return { + ...chunk, + startOffset: chunk.startOffset + offset, + data: chunk.data.slice(offset), + byteLength: chunk.byteLength - droppedBytes, + }; +} + +function appendOutput( + current: TerminalOutputState, + data: string, + maxBufferBytes: number, +): TerminalOutputState { + if (data.length === 0) return current; + if (maxBufferBytes <= 0) { + return { + generation: current.generation, + chunks: [], + retainedBytes: 0, + resetVersion: current.resetVersion + 1, + nextOffset: current.nextOffset + data.length, + }; + } + const appended = splitOutputChunks( + data, + current.nextOffset, + Math.min(DEFAULT_TERMINAL_CHUNK_BYTES, Math.max(1, maxBufferBytes)), + ); + + const chunks = [...current.chunks, ...appended.chunks]; + let retainedBytes = current.retainedBytes + appended.byteLength; + let firstRetainedIndex = 0; + while (retainedBytes > maxBufferBytes && firstRetainedIndex < chunks.length) { + const first = chunks[firstRetainedIndex]!; + const bytesToDrop = retainedBytes - maxBufferBytes; + if (bytesToDrop < first.byteLength) { + const trimmed = trimOutputChunkStart(first, bytesToDrop); + retainedBytes -= first.byteLength - trimmed.byteLength; + if (trimmed.byteLength > 0) { + chunks[firstRetainedIndex] = trimmed; + } else { + firstRetainedIndex += 1; + } + break; + } + retainedBytes -= first.byteLength; + firstRetainedIndex += 1; + } + + let retainedChunks = firstRetainedIndex === 0 ? chunks : chunks.slice(firstRetainedIndex); + if (retainedChunks.length > MAX_TERMINAL_OUTPUT_CHUNKS) { + retainedChunks = compactRetainedChunks(retainedChunks); + const excessChunks = retainedChunks.length - MAX_TERMINAL_OUTPUT_CHUNKS; + if (excessChunks > 0) { + for (const chunk of retainedChunks.slice(0, excessChunks)) { + retainedBytes -= chunk.byteLength; + } + retainedChunks = retainedChunks.slice(excessChunks); + } + } + + return { + generation: current.generation, + chunks: retainedChunks, + retainedBytes, + resetVersion: current.resetVersion, + nextOffset: appended.nextOffset, + }; +} + +function resetOutput( + current: TerminalOutputState, + data: string, + maxBufferBytes: number, +): TerminalOutputState { + const retained = trimBufferToBytes(data, maxBufferBytes); + const reset = splitOutputChunks( + retained, + 0, + Math.min(DEFAULT_TERMINAL_CHUNK_BYTES, Math.max(1, maxBufferBytes)), + ); + return { + generation: current.generation, + chunks: reset.chunks, + retainedBytes: reset.byteLength, + resetVersion: current.resetVersion + 1, + nextOffset: reset.nextOffset, + }; +} + +export function terminalOutputText(output: TerminalOutputState): string { + return output.chunks.map((chunk) => chunk.data).join(""); +} + +export function readTerminalOutputUpdate( + output: TerminalOutputState, + cursor: TerminalOutputCursor, +): TerminalOutputUpdate { + const nextCursor = { + generation: output.generation, + resetVersion: output.resetVersion, + offset: output.nextOffset, + }; + const firstChunk = output.chunks[0]; + if ( + cursor.generation !== output.generation || + cursor.resetVersion !== output.resetVersion || + cursor.offset < (firstChunk?.startOffset ?? output.nextOffset) + ) { + return { type: "reset", data: terminalOutputText(output), cursor: nextCursor }; + } + + const appended = output.chunks.filter( + (chunk) => chunk.startOffset + chunk.data.length > cursor.offset, + ); + if (appended.length === 0) { + return { type: "none", cursor: nextCursor }; + } + return { + type: "append", + data: appended + .map((chunk) => chunk.data.slice(Math.max(0, cursor.offset - chunk.startOffset))) + .join(""), + cursor: nextCursor, + }; +} + +export { appendOutput, resetOutput }; diff --git a/packages/client-runtime/src/state/terminalSession.test.ts b/packages/client-runtime/src/state/terminalSession.test.ts index 85c57592d118..d9438860ddab 100644 --- a/packages/client-runtime/src/state/terminalSession.test.ts +++ b/packages/client-runtime/src/state/terminalSession.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, it } from "vite-plus/test"; +import { describe, expect, it, vi } from "vite-plus/test"; import { EnvironmentId, TerminalSessionSnapshot, ThreadId } from "@t3tools/contracts"; @@ -6,8 +6,13 @@ import { applyTerminalAttachStreamEvent, applyTerminalMetadataStreamEvent, combineTerminalSessionState, + DEFAULT_MAX_TERMINAL_BUFFER_BYTES, EMPTY_TERMINAL_BUFFER_STATE, + INITIAL_TERMINAL_OUTPUT_CURSOR, + nextTerminalAttachSeedState, + readTerminalOutputUpdate, selectRunningSubprocessTerminalIds, + terminalOutputText, } from "./terminalSession.ts"; const TARGET = { @@ -126,11 +131,11 @@ describe("terminal session reducers", () => { ); expect(output).toMatchObject({ - buffer: "lo world", status: "running", error: null, version: 2, }); + expect(terminalOutputText(output.output)).toBe("lo world"); }); it("reduces terminal metadata snapshots, upserts, and removals", () => { @@ -182,6 +187,268 @@ describe("terminal session reducers", () => { 4, ); - expect(state.buffer).toBe("๐Ÿ™‚"); + expect(terminalOutputText(state.output)).toBe("๐Ÿ™‚"); + }); + + it("preserves a BOM code point at a new output chunk boundary", () => { + const initial = applyTerminalAttachStreamEvent(EMPTY_TERMINAL_BUFFER_STATE, { + type: "snapshot", + snapshot: { ...BASE_SNAPSHOT, history: "" }, + }); + const cursor = readTerminalOutputUpdate(initial.output, INITIAL_TERMINAL_OUTPUT_CURSOR).cursor; + const data = `${"x".repeat(16_384)}\uFEFF${"y".repeat(20)}`; + const state = applyTerminalAttachStreamEvent(initial, { + type: "output", + threadId: TARGET.threadId, + terminalId: TARGET.terminalId, + data, + }); + + expect(terminalOutputText(state.output)).toBe(data); + expect(readTerminalOutputUpdate(state.output, cursor)).toMatchObject({ type: "append", data }); + expect(state.output.retainedBytes).toBe(new TextEncoder().encode(data).byteLength); + }); + + it("preserves a leading BOM in the retained snapshot tail", () => { + const state = applyTerminalAttachStreamEvent( + EMPTY_TERMINAL_BUFFER_STATE, + { type: "snapshot", snapshot: { ...BASE_SNAPSHOT, history: "discard\uFEFFtail" } }, + 7, + ); + + expect(terminalOutputText(state.output)).toBe("\uFEFFtail"); + expect(state.output.retainedBytes).toBe(7); + }); + + it("trims whole Unicode code points from a partially retained chunk", () => { + let state = applyTerminalAttachStreamEvent( + EMPTY_TERMINAL_BUFFER_STATE, + { type: "snapshot", snapshot: { ...BASE_SNAPSHOT, history: "รฉ็•Œ๐Ÿ™‚end" } }, + 12, + ); + let cursor = readTerminalOutputUpdate(state.output, INITIAL_TERMINAL_OUTPUT_CURSOR).cursor; + for (const [data, expected, retainedBytes] of [ + ["x", "็•Œ๐Ÿ™‚endx", 11], + ["yz", "๐Ÿ™‚endxyz", 10], + ["abc", "endxyzabc", 9], + ] as const) { + state = applyTerminalAttachStreamEvent( + state, + { type: "output", threadId: TARGET.threadId, terminalId: TARGET.terminalId, data }, + 12, + ); + const update = readTerminalOutputUpdate(state.output, cursor); + expect(update).toMatchObject({ type: "append", data }); + expect(terminalOutputText(state.output)).toBe(expected); + expect(state.output.retainedBytes).toBe(retainedBytes); + cursor = update.cursor; + } + }); + + it("delivers all output when several events arrive before a renderer reads", () => { + const initial = applyTerminalAttachStreamEvent(EMPTY_TERMINAL_BUFFER_STATE, { + type: "snapshot", + snapshot: BASE_SNAPSHOT, + }); + const cursor = readTerminalOutputUpdate(initial.output, INITIAL_TERMINAL_OUTPUT_CURSOR).cursor; + let state = initial; + for (const data of [" one", " two", " three"]) { + state = applyTerminalAttachStreamEvent(state, { + type: "output", + threadId: TARGET.threadId, + terminalId: TARGET.terminalId, + data, + }); + } + const update = readTerminalOutputUpdate(state.output, cursor); + expect(update).toMatchObject({ type: "append", data: " one two three" }); + expect(readTerminalOutputUpdate(state.output, update.cursor).type).toBe("none"); + }); + + it("preserves the byte-limited tail and resets a cursor before a partially trimmed chunk", () => { + const initial = applyTerminalAttachStreamEvent(EMPTY_TERMINAL_BUFFER_STATE, { + type: "snapshot", + snapshot: { ...BASE_SNAPSHOT, history: "" }, + }); + const staleCursor = readTerminalOutputUpdate( + initial.output, + INITIAL_TERMINAL_OUTPUT_CURSOR, + ).cursor; + const first = applyTerminalAttachStreamEvent(initial, { + type: "output", + threadId: TARGET.threadId, + terminalId: TARGET.terminalId, + data: "hello", + }); + const caughtUpCursor = readTerminalOutputUpdate(first.output, staleCursor).cursor; + const state = applyTerminalAttachStreamEvent( + first, + { type: "output", threadId: TARGET.threadId, terminalId: TARGET.terminalId, data: " world" }, + 8, + ); + + expect(readTerminalOutputUpdate(state.output, caughtUpCursor)).toMatchObject({ + type: "append", + data: " world", + }); + expect(readTerminalOutputUpdate(state.output, staleCursor)).toMatchObject({ + type: "reset", + data: "lo world", + }); + expect(state.output.retainedBytes).toBe(8); + }); + + it("does not encode retained history again when appending at the byte limit", () => { + let state = applyTerminalAttachStreamEvent(EMPTY_TERMINAL_BUFFER_STATE, { + type: "snapshot", + snapshot: { ...BASE_SNAPSHOT, history: "x".repeat(DEFAULT_MAX_TERMINAL_BUFFER_BYTES) }, + }); + const data = "y".repeat(8192); + const encode = vi.spyOn(TextEncoder.prototype, "encode"); + try { + for (let index = 0; index < 100; index += 1) { + state = applyTerminalAttachStreamEvent(state, { + type: "output", + threadId: TARGET.threadId, + terminalId: TARGET.terminalId, + data, + }); + } + expect(encode.mock.calls.reduce((total, [text]) => total + (text?.length ?? 0), 0)).toBe( + data.length * 100, + ); + expect(state.output.retainedBytes).toBe(DEFAULT_MAX_TERMINAL_BUFFER_BYTES); + } finally { + encode.mockRestore(); + } + }); + + it.each([0, -1])("discards output without encoding when the byte budget is %s", (maxBytes) => { + const initial = applyTerminalAttachStreamEvent(EMPTY_TERMINAL_BUFFER_STATE, { + type: "snapshot", + snapshot: BASE_SNAPSHOT, + }); + const cursor = readTerminalOutputUpdate(initial.output, INITIAL_TERMINAL_OUTPUT_CURSOR).cursor; + const data = "x".repeat(65_536); + const encode = vi.spyOn(TextEncoder.prototype, "encode"); + try { + const discarded = applyTerminalAttachStreamEvent( + initial, + { type: "output", threadId: TARGET.threadId, terminalId: TARGET.terminalId, data }, + maxBytes, + ); + expect(encode).not.toHaveBeenCalled(); + expect(discarded.output.nextOffset).toBe(initial.output.nextOffset + data.length); + expect(discarded.output.retainedBytes).toBe(0); + expect(readTerminalOutputUpdate(discarded.output, cursor)).toMatchObject({ + type: "reset", + data: "", + }); + + const empty = applyTerminalAttachStreamEvent( + initial, + { type: "output", threadId: TARGET.threadId, terminalId: TARGET.terminalId, data: "" }, + maxBytes, + ); + expect(empty.output).toBe(initial.output); + expect(encode).not.toHaveBeenCalled(); + } finally { + encode.mockRestore(); + } + }); + + it("resets for repeated snapshots, clear, and restart even when output text repeats", () => { + let state = applyTerminalAttachStreamEvent(EMPTY_TERMINAL_BUFFER_STATE, { + type: "snapshot", + snapshot: BASE_SNAPSHOT, + }); + let cursor = readTerminalOutputUpdate(state.output, INITIAL_TERMINAL_OUTPUT_CURSOR).cursor; + state = applyTerminalAttachStreamEvent(state, { type: "snapshot", snapshot: BASE_SNAPSHOT }); + const repeated = readTerminalOutputUpdate(state.output, cursor); + expect(repeated).toMatchObject({ type: "reset", data: "hello" }); + expect(state.version).toBe(2); + cursor = repeated.cursor; + + state = applyTerminalAttachStreamEvent(state, { + type: "cleared", + threadId: TARGET.threadId, + terminalId: TARGET.terminalId, + }); + const cleared = readTerminalOutputUpdate(state.output, cursor); + expect(cleared).toMatchObject({ type: "reset", data: "" }); + state = applyTerminalAttachStreamEvent(state, { + type: "restarted", + threadId: TARGET.threadId, + terminalId: TARGET.terminalId, + snapshot: BASE_SNAPSHOT, + }); + expect(readTerminalOutputUpdate(state.output, cleared.cursor)).toMatchObject({ + type: "reset", + data: "hello", + }); + }); + + it("does not reuse a renderer cursor when a fresh attach has matching counters", () => { + const first = applyTerminalAttachStreamEvent(nextTerminalAttachSeedState(), { + type: "snapshot", + snapshot: BASE_SNAPSHOT, + }); + const cursor = readTerminalOutputUpdate(first.output, INITIAL_TERMINAL_OUTPUT_CURSOR).cursor; + const next = applyTerminalAttachStreamEvent(nextTerminalAttachSeedState(), { + type: "snapshot", + snapshot: { ...BASE_SNAPSHOT, history: "other" }, + }); + expect(next.output.resetVersion).toBe(first.output.resetVersion); + expect(next.output.nextOffset).toBe(first.output.nextOffset); + expect(readTerminalOutputUpdate(next.output, cursor)).toMatchObject({ + type: "reset", + data: "other", + }); + }); + + it("keeps appending while compacting metadata for many small writes", () => { + let state = applyTerminalAttachStreamEvent(EMPTY_TERMINAL_BUFFER_STATE, { + type: "snapshot", + snapshot: { ...BASE_SNAPSHOT, history: "" }, + }); + let cursor = readTerminalOutputUpdate(state.output, INITIAL_TERMINAL_OUTPUT_CURSOR).cursor; + let received = ""; + for (let index = 0; index < 2500; index += 1) { + state = applyTerminalAttachStreamEvent(state, { + type: "output", + threadId: TARGET.threadId, + terminalId: TARGET.terminalId, + data: "x", + }); + const update = readTerminalOutputUpdate(state.output, cursor); + if (update.type !== "append") throw new Error(`Expected append, received ${update.type}`); + received += update.data; + cursor = update.cursor; + } + expect(received).toBe("x".repeat(2500)); + expect(state.output.chunks.length).toBeLessThan(1024); + expect(terminalOutputText(state.output)).toBe(received); + }); + + it("appends every unread character across a compaction boundary", () => { + let state = applyTerminalAttachStreamEvent(EMPTY_TERMINAL_BUFFER_STATE, { + type: "snapshot", + snapshot: { ...BASE_SNAPSHOT, history: "" }, + }); + let cursor = readTerminalOutputUpdate(state.output, INITIAL_TERMINAL_OUTPUT_CURSOR).cursor; + const writes = Array.from({ length: 1200 }, (_, index) => ["x", "รฉ", "็•Œ", "๐Ÿ™‚"][index % 4]!); + for (const [index, data] of writes.entries()) { + state = applyTerminalAttachStreamEvent(state, { + type: "output", + threadId: TARGET.threadId, + terminalId: TARGET.terminalId, + data, + }); + if (index === 99) cursor = readTerminalOutputUpdate(state.output, cursor).cursor; + } + expect(readTerminalOutputUpdate(state.output, cursor)).toMatchObject({ + type: "append", + data: writes.slice(100).join(""), + }); }); }); diff --git a/packages/client-runtime/src/state/terminalSession.ts b/packages/client-runtime/src/state/terminalSession.ts index ee444e36db41..70a7ea2bc093 100644 --- a/packages/client-runtime/src/state/terminalSession.ts +++ b/packages/client-runtime/src/state/terminalSession.ts @@ -6,10 +6,27 @@ import type { TerminalSummary, ThreadId, } from "@t3tools/contracts"; +import { + appendOutput, + DEFAULT_MAX_TERMINAL_BUFFER_BYTES, + EMPTY_TERMINAL_OUTPUT_STATE, + resetOutput, + type TerminalOutputState, +} from "./terminalOutput.ts"; + +export { + DEFAULT_MAX_TERMINAL_BUFFER_BYTES, + INITIAL_TERMINAL_OUTPUT_CURSOR, + readTerminalOutputUpdate, + terminalOutputText, + type TerminalOutputCursor, + type TerminalOutputState, + type TerminalOutputUpdate, +} from "./terminalOutput.ts"; export interface TerminalSessionState { readonly summary: TerminalSummary | null; - readonly buffer: string; + readonly output: TerminalOutputState; readonly status: TerminalSessionSnapshot["status"] | "closed"; readonly error: string | null; readonly hasRunningSubprocess: boolean; @@ -18,7 +35,7 @@ export interface TerminalSessionState { } export interface TerminalBufferState { - readonly buffer: string; + readonly output: TerminalOutputState; readonly status: TerminalSessionSnapshot["status"] | "closed"; readonly error: string | null; readonly updatedAt: string | null; @@ -45,7 +62,7 @@ export function selectRunningSubprocessTerminalIds( } export const EMPTY_TERMINAL_BUFFER_STATE = Object.freeze({ - buffer: "", + output: EMPTY_TERMINAL_OUTPUT_STATE, status: "closed", error: null, updatedAt: null, @@ -54,7 +71,7 @@ export const EMPTY_TERMINAL_BUFFER_STATE = Object.freeze({ export const EMPTY_TERMINAL_SESSION_STATE = Object.freeze({ summary: null, - buffer: "", + output: EMPTY_TERMINAL_OUTPUT_STATE, status: "closed", error: null, hasRunningSubprocess: false, @@ -62,42 +79,30 @@ export const EMPTY_TERMINAL_SESSION_STATE = Object.freeze( version: 0, }); -export const DEFAULT_MAX_TERMINAL_BUFFER_BYTES = 512 * 1024; -const textEncoder = new TextEncoder(); -const textDecoder = new TextDecoder(); +let terminalAttachGeneration = 0; -function trimBufferToBytes(buffer: string, maxBufferBytes: number): string { - if (maxBufferBytes <= 0) { - return ""; - } - - const encoded = textEncoder.encode(buffer); - if (encoded.byteLength <= maxBufferBytes) { - return buffer; - } - - let start = encoded.byteLength - maxBufferBytes; - while (start < encoded.length) { - const byte = encoded[start]; - if (byte === undefined || (byte & 0b1100_0000) !== 0b1000_0000) { - break; - } - start += 1; - } - - return textDecoder.decode(encoded.subarray(start)); +/** A reinstalled attach stream must not reuse an old renderer's output cursor. */ +export function nextTerminalAttachSeedState(): TerminalBufferState { + return { + ...EMPTY_TERMINAL_BUFFER_STATE, + output: { + ...EMPTY_TERMINAL_OUTPUT_STATE, + generation: ++terminalAttachGeneration, + }, + }; } export function terminalBufferStateFromSnapshot( snapshot: TerminalSessionSnapshot, maxBufferBytes: number, + current: TerminalBufferState = EMPTY_TERMINAL_BUFFER_STATE, ): TerminalBufferState { return { - buffer: trimBufferToBytes(snapshot.history, maxBufferBytes), + output: resetOutput(current.output, snapshot.history, maxBufferBytes), status: snapshot.status, error: null, updatedAt: snapshot.updatedAt, - version: 1, + version: current.version + 1, }; } @@ -113,7 +118,7 @@ export function combineTerminalSessionState( ): TerminalSessionState { return { summary, - buffer: buffer.buffer, + output: buffer.output, status: buffer.version > 0 ? buffer.status : (summary?.status ?? buffer.status), error: buffer.error, hasRunningSubprocess: summary?.hasRunningSubprocess ?? false, @@ -130,11 +135,11 @@ export function applyTerminalAttachStreamEvent( switch (event.type) { case "snapshot": case "restarted": - return terminalBufferStateFromSnapshot(event.snapshot, maxBufferBytes); + return terminalBufferStateFromSnapshot(event.snapshot, maxBufferBytes, current); case "output": return { ...current, - buffer: trimBufferToBytes(`${current.buffer}${event.data}`, maxBufferBytes), + output: appendOutput(current.output, event.data, maxBufferBytes), status: current.status === "closed" ? "running" : current.status, error: null, version: current.version + 1, @@ -142,7 +147,7 @@ export function applyTerminalAttachStreamEvent( case "cleared": return { ...current, - buffer: "", + output: resetOutput(current.output, "", maxBufferBytes), error: null, version: current.version + 1, }; From 92495df0c0c80fa3ee1b072ea5767cb3cc01ef42 Mon Sep 17 00:00:00 2001 From: Leo Date: Sun, 13 Sep 2026 22:23:58 -0400 Subject: [PATCH 2/2] fix(client): reset terminal viewport after rollover --- packages/client-runtime/src/state/terminalOutput.ts | 9 +++++++-- .../client-runtime/src/state/terminalSession.test.ts | 8 ++++---- 2 files changed, 11 insertions(+), 6 deletions(-) diff --git a/packages/client-runtime/src/state/terminalOutput.ts b/packages/client-runtime/src/state/terminalOutput.ts index fcb0389c434b..3da12ecd778e 100644 --- a/packages/client-runtime/src/state/terminalOutput.ts +++ b/packages/client-runtime/src/state/terminalOutput.ts @@ -16,6 +16,8 @@ export interface TerminalOutputState { export interface TerminalOutputCursor { readonly generation: number; readonly resetVersion: number; + /** Start of the retained output already rendered into the viewport. */ + readonly retainedStartOffset: number; readonly offset: number; } @@ -23,6 +25,7 @@ export interface TerminalOutputCursor { export const INITIAL_TERMINAL_OUTPUT_CURSOR = Object.freeze({ generation: -1, resetVersion: -1, + retainedStartOffset: 0, offset: 0, }); @@ -288,16 +291,18 @@ export function readTerminalOutputUpdate( output: TerminalOutputState, cursor: TerminalOutputCursor, ): TerminalOutputUpdate { + const retainedStartOffset = output.chunks[0]?.startOffset ?? output.nextOffset; const nextCursor = { generation: output.generation, resetVersion: output.resetVersion, + retainedStartOffset, offset: output.nextOffset, }; - const firstChunk = output.chunks[0]; if ( cursor.generation !== output.generation || cursor.resetVersion !== output.resetVersion || - cursor.offset < (firstChunk?.startOffset ?? output.nextOffset) + cursor.retainedStartOffset !== retainedStartOffset || + cursor.offset < retainedStartOffset ) { return { type: "reset", data: terminalOutputText(output), cursor: nextCursor }; } diff --git a/packages/client-runtime/src/state/terminalSession.test.ts b/packages/client-runtime/src/state/terminalSession.test.ts index d9438860ddab..055a999a9df6 100644 --- a/packages/client-runtime/src/state/terminalSession.test.ts +++ b/packages/client-runtime/src/state/terminalSession.test.ts @@ -238,7 +238,7 @@ describe("terminal session reducers", () => { 12, ); const update = readTerminalOutputUpdate(state.output, cursor); - expect(update).toMatchObject({ type: "append", data }); + expect(update).toMatchObject({ type: "reset", data: expected }); expect(terminalOutputText(state.output)).toBe(expected); expect(state.output.retainedBytes).toBe(retainedBytes); cursor = update.cursor; @@ -265,7 +265,7 @@ describe("terminal session reducers", () => { expect(readTerminalOutputUpdate(state.output, update.cursor).type).toBe("none"); }); - it("preserves the byte-limited tail and resets a cursor before a partially trimmed chunk", () => { + it("preserves the byte-limited tail and resets every viewport after rollover", () => { const initial = applyTerminalAttachStreamEvent(EMPTY_TERMINAL_BUFFER_STATE, { type: "snapshot", snapshot: { ...BASE_SNAPSHOT, history: "" }, @@ -288,8 +288,8 @@ describe("terminal session reducers", () => { ); expect(readTerminalOutputUpdate(state.output, caughtUpCursor)).toMatchObject({ - type: "append", - data: " world", + type: "reset", + data: "lo world", }); expect(readTerminalOutputUpdate(state.output, staleCursor)).toMatchObject({ type: "reset",