authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-09-04 09:27:24-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-04 09:27:24-07:00
logda7e46d08e85bcb07ecd78721a40f7b612fac2f2
treedd14e64aa9b665d52f3c654f6495ff638c118236
parent7839140e5e93d3f401d7eb45b86cf1a234eb3609
signature Signed by PGP key B5690EEEBB952194

perf(web): stop replaying terminal buffers on rollover (#9707)

Keep bounded, byte-counted terminal chunks and append only unread output. Use UTF-16 cursors so compaction preserves live terminal replies. Reset on lifecycle changes or a real retained-data gap. Keep the existing wire protocol, native buffer interface, and client retention limit. Native streaming and strict server replay byte bounds remain separate. Continue the client helpers from https://github.com/pingdotgg/t3code/pull/9027 at source head 9391da2b48439d1d7a2b01d169e785682bf8abb8. Created with GPT-6 Astra (preview) in Codex. Co-Authored-By: Wout Stiens <71498452+StiensWout@users.noreply.github.com> Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

8 files changed, 928 insertions(+), 67 deletions(-)

apps/mobile/src/features/terminal/terminalMenu.test.ts+5-2
...@@ -1,6 +1,9 @@...@@ -1,6 +1,9 @@
1import { describe, expect, it } from "vite-plus/test";1import { describe, expect, it } from "vite-plus/test";
22
3import { type KnownTerminalSession } from "@t3tools/client-runtime/state/terminal";3import {
4 EMPTY_TERMINAL_BUFFER_STATE,
5 type KnownTerminalSession,
6} from "@t3tools/client-runtime/state/terminal";
4import { DEFAULT_TERMINAL_ID, EnvironmentId, ThreadId } from "@t3tools/contracts";7import { DEFAULT_TERMINAL_ID, EnvironmentId, ThreadId } from "@t3tools/contracts";
58
6import { getTerminalLabel } from "@t3tools/shared/terminalLabels";9import { getTerminalLabel } from "@t3tools/shared/terminalLabels";
...@@ -55,7 +58,7 @@ function makeKnownSession(input: {...@@ -55,7 +58,7 @@ function makeKnownSession(input: {
55 updatedAt: input.updatedAt ?? "2026-04-15T20:00:00.000Z",58 updatedAt: input.updatedAt ?? "2026-04-15T20:00:00.000Z",
56 }59 }
57 : null,60 : null,
58 buffer: "",61 output: EMPTY_TERMINAL_BUFFER_STATE.output,
59 status: input.status,62 status: input.status,
60 error: null,63 error: null,
61 hasRunningSubprocess: false,64 hasRunningSubprocess: false,
apps/mobile/src/state/use-terminal-session.ts+18-4
...@@ -2,6 +2,7 @@ import {...@@ -2,6 +2,7 @@ import {
2 combineTerminalSessionState,2 combineTerminalSessionState,
3 EMPTY_TERMINAL_BUFFER_STATE,3 EMPTY_TERMINAL_BUFFER_STATE,
4 EMPTY_TERMINAL_SESSION_STATE,4 EMPTY_TERMINAL_SESSION_STATE,
5 terminalOutputText,
5 type KnownTerminalSession,6 type KnownTerminalSession,
6 type TerminalSessionState,7 type TerminalSessionState,
7} from "@t3tools/client-runtime/state/terminal";8} from "@t3tools/client-runtime/state/terminal";
...@@ -11,10 +12,16 @@ import { useMemo } from "react";...@@ -11,10 +12,16 @@ import { useMemo } from "react";
11import { useEnvironmentQuery } from "./query";12import { useEnvironmentQuery } from "./query";
12import { terminalEnvironment } from "./terminal";13import { terminalEnvironment } from "./terminal";
1314
15type LegacyTerminalSessionState = TerminalSessionState & { readonly buffer: string };
16const EMPTY_LEGACY_TERMINAL_SESSION_STATE: LegacyTerminalSessionState = {
17 ...EMPTY_TERMINAL_SESSION_STATE,
18 buffer: "",
19};
20
14export function useAttachedTerminalSession(input: {21export function useAttachedTerminalSession(input: {
15 readonly environmentId: EnvironmentId | null;22 readonly environmentId: EnvironmentId | null;
16 readonly terminal: TerminalAttachInput | null;23 readonly terminal: TerminalAttachInput | null;
17}): TerminalSessionState {24}): LegacyTerminalSessionState {
18 const attach = useEnvironmentQuery(25 const attach = useEnvironmentQuery(
19 input.environmentId !== null && input.terminal !== null26 input.environmentId !== null && input.terminal !== null
20 ? terminalEnvironment.attach({27 ? terminalEnvironment.attach({
...@@ -31,10 +38,14 @@ export function useAttachedTerminalSession(input: {...@@ -31,10 +38,14 @@ export function useAttachedTerminalSession(input: {
31 input: null,38 input: null,
32 }),39 }),
33 );40 );
41 const output = attach.data?.output ?? EMPTY_TERMINAL_BUFFER_STATE.output;
42 // Installed native binaries still accept initialBuffer. Keep materialization
43 // at this mobile boundary until the native streaming API is released.
44 const buffer = useMemo(() => terminalOutputText(output), [output]);
3445
35 return useMemo(() => {46 return useMemo(() => {
36 if (input.environmentId === null || input.terminal === null) {47 if (input.environmentId === null || input.terminal === null) {
37 return EMPTY_TERMINAL_SESSION_STATE;48 return EMPTY_LEGACY_TERMINAL_SESSION_STATE;
38 }49 }
39 const summary =50 const summary =
40 metadata.data?.find(51 metadata.data?.find(
...@@ -42,9 +53,12 @@ export function useAttachedTerminalSession(input: {...@@ -42,9 +53,12 @@ export function useAttachedTerminalSession(input: {
42 terminal.threadId === input.terminal?.threadId &&53 terminal.threadId === input.terminal?.threadId &&
43 terminal.terminalId === input.terminal?.terminalId,54 terminal.terminalId === input.terminal?.terminalId,
44 ) ?? null;55 ) ?? null;
45 const state = combineTerminalSessionState(summary, attach.data ?? EMPTY_TERMINAL_BUFFER_STATE);56 const state = {
57 ...combineTerminalSessionState(summary, attach.data ?? EMPTY_TERMINAL_BUFFER_STATE),
58 buffer,
59 };
46 return attach.error === null ? state : { ...state, error: attach.error, status: "error" };60 return attach.error === null ? state : { ...state, error: attach.error, status: "error" };
47 }, [attach.data, attach.error, input.environmentId, input.terminal, metadata.data]);61 }, [attach.data, attach.error, buffer, input.environmentId, input.terminal, metadata.data]);
48}62}
4963
50export function useKnownTerminalSessions(input: {64export function useKnownTerminalSessions(input: {
apps/web/src/components/ThreadTerminalDrawer.tsx+34-18
...@@ -3,7 +3,13 @@ import {...@@ -3,7 +3,13 @@ import {
3 isAtomCommandInterrupted,3 isAtomCommandInterrupted,
4 squashAtomCommandFailure,4 squashAtomCommandFailure,
5} from "@t3tools/client-runtime/state/runtime";5} from "@t3tools/client-runtime/state/runtime";
6import { type TerminalSessionState } from "@t3tools/client-runtime/state/terminal";6import {
7 INITIAL_TERMINAL_OUTPUT_CURSOR,
8 readTerminalOutputUpdate,
9 type TerminalOutputCursor,
10 type TerminalOutputUpdate,
11 type TerminalSessionState,
12} from "@t3tools/client-runtime/state/terminal";
7import {13import {
8 Plus,14 Plus,
9 Square,15 Square,
...@@ -99,8 +105,15 @@ function writeSystemMessage(terminal: GhosttyTerminalSurface, message: string):...@@ -99,8 +105,15 @@ function writeSystemMessage(terminal: GhosttyTerminalSurface, message: string):
99 terminal.write(`\r\n[terminal] ${message}\r\n`);105 terminal.write(`\r\n[terminal] ${message}\r\n`);
100}106}
101107
102function writeTerminalBuffer(terminal: GhosttyTerminalSurface, buffer: string): void {108export function writeTerminalOutputUpdate(
103 terminal.resetAndWrite(buffer);109 terminal: Pick<GhosttyTerminalSurface, "resetAndWrite" | "write">,
110 update: TerminalOutputUpdate,
111): void {
112 if (update.type === "reset") {
113 terminal.resetAndWrite(update.data);
114 } else if (update.type === "append") {
115 terminal.write(update.data);
116 }
104}117}
105118
106function parseTerminalColor(value: string, fallback: GhosttyColor): GhosttyColor {119function parseTerminalColor(value: string, fallback: GhosttyColor): GhosttyColor {
...@@ -380,9 +393,10 @@ export function TerminalViewport({...@@ -380,9 +393,10 @@ export function TerminalViewport({
380 input: { threadId, terminalId, cols, rows },393 input: { threadId, terminalId, cols, rows },
381 }),394 }),
382 );395 );
383 const terminalBuffer = terminalSession.buffer;396 const terminalOutput = terminalSession.output;
384 const terminalError = terminalSession.error;397 const terminalError = terminalSession.error;
385 const terminalStatus = terminalSession.status;398 const terminalStatus = terminalSession.status;
399 const outputCursorRef = useRef<TerminalOutputCursor>(INITIAL_TERMINAL_OUTPUT_CURSOR);
386 const synchronizedStatusRef = useRef<TerminalSessionState["status"]>("closed");400 const synchronizedStatusRef = useRef<TerminalSessionState["status"]>("closed");
387 const synchronizeTerminalStatus = useEffectEvent(401 const synchronizeTerminalStatus = useEffectEvent(
388 (terminal: GhosttyTerminalSurface, status: TerminalSessionState["status"]) => {402 (terminal: GhosttyTerminalSurface, status: TerminalSessionState["status"]) => {
...@@ -403,14 +417,14 @@ export function TerminalViewport({...@@ -403,14 +417,14 @@ export function TerminalViewport({
403 );417 );
404 const terminalVersion = terminalSession.version;418 const terminalVersion = terminalSession.version;
405 const previousSessionRef = useRef({419 const previousSessionRef = useRef({
406 buffer: terminalBuffer,420 output: terminalOutput,
407 error: terminalError,421 error: terminalError,
408 status: terminalStatus,422 status: terminalStatus,
409 version: terminalVersion,423 version: terminalVersion,
410 });424 });
411 const latestSessionRef = useRef(previousSessionRef.current);425 const latestSessionRef = useRef(previousSessionRef.current);
412 latestSessionRef.current = {426 latestSessionRef.current = {
413 buffer: terminalBuffer,427 output: terminalOutput,
414 error: terminalError,428 error: terminalError,
415 status: terminalStatus,429 status: terminalStatus,
416 version: terminalVersion,430 version: terminalVersion,
...@@ -474,7 +488,14 @@ export function TerminalViewport({...@@ -474,7 +488,14 @@ export function TerminalViewport({
474 }488 }
475 const latestSession = latestSessionRef.current;489 const latestSession = latestSessionRef.current;
476 previousSessionRef.current = latestSession;490 previousSessionRef.current = latestSession;
477 if (latestSession.buffer.length > 0) terminal.resetAndWrite(latestSession.buffer);491 const initialOutput = readTerminalOutputUpdate(
492 latestSession.output,
493 INITIAL_TERMINAL_OUTPUT_CURSOR,
494 );
495 if (initialOutput.type === "reset" && initialOutput.data.length > 0) {
496 writeTerminalOutputUpdate(terminal, initialOutput);
497 }
498 outputCursorRef.current = initialOutput.cursor;
478 if (latestSession.error !== null) writeSystemMessage(terminal, latestSession.error);499 if (latestSession.error !== null) writeSystemMessage(terminal, latestSession.error);
479 // Attaching to a session that already exited must still run exit handling500 // Attaching to a session that already exited must still run exit handling
480 // once, so mount synchronization starts from the empty "closed" state.501 // once, so mount synchronization starts from the empty "closed" state.
...@@ -845,7 +866,7 @@ export function TerminalViewport({...@@ -845,7 +866,7 @@ export function TerminalViewport({
845 useEffect(() => {866 useEffect(() => {
846 const terminal = terminalRef.current;867 const terminal = terminalRef.current;
847 const current = {868 const current = {
848 buffer: terminalBuffer,869 output: terminalOutput,
849 error: terminalError,870 error: terminalError,
850 status: terminalStatus,871 status: terminalStatus,
851 version: terminalVersion,872 version: terminalVersion,
...@@ -857,18 +878,13 @@ export function TerminalViewport({...@@ -857,18 +878,13 @@ export function TerminalViewport({
857878
858 const previous = previousSessionRef.current;879 const previous = previousSessionRef.current;
859 synchronizeTerminalStatus(terminal, current.status);880 synchronizeTerminalStatus(terminal, current.status);
860 if (current.version === previous.version) {881 if (current.version === previous.version && current.output === previous.output) {
861 return;882 return;
862 }883 }
863884
864 if (885 const outputUpdate = readTerminalOutputUpdate(current.output, outputCursorRef.current);
865 current.buffer.length >= previous.buffer.length &&886 writeTerminalOutputUpdate(terminal, outputUpdate);
866 current.buffer.startsWith(previous.buffer)887 outputCursorRef.current = outputUpdate.cursor;
867 ) {
868 terminal.write(current.buffer.slice(previous.buffer.length));
869 } else {
870 writeTerminalBuffer(terminal, current.buffer);
871 }
872 terminal.clearSelection();888 terminal.clearSelection();
873889
874 if (current.error !== null && current.error !== previous.error) {890 if (current.error !== null && current.error !== previous.error) {
...@@ -881,7 +897,7 @@ export function TerminalViewport({...@@ -881,7 +897,7 @@ export function TerminalViewport({
881 });897 });
882 }898 }
883 previousSessionRef.current = current;899 previousSessionRef.current = current;
884 }, [autoFocus, terminalBuffer, terminalError, terminalStatus, terminalVersion]);900 }, [autoFocus, terminalOutput, terminalError, terminalStatus, terminalVersion]);
885901
886 useEffect(() => {902 useEffect(() => {
887 if (!autoFocus) return;903 if (!autoFocus) return;
apps/web/src/terminal/ghostty/core.test.ts+234-2
...@@ -1,5 +1,15 @@...@@ -1,5 +1,15 @@
1import { afterEach, describe, expect, it, vi } from "vite-plus/test";1import { afterEach, describe, expect, it, vi } from "vite-plus/test";
2import {
3 applyTerminalAttachStreamEvent,
4 DEFAULT_MAX_TERMINAL_BUFFER_BYTES,
5 INITIAL_TERMINAL_OUTPUT_CURSOR,
6 nextTerminalAttachSeedState,
7 readTerminalOutputUpdate,
8 terminalOutputText,
9 type TerminalBufferState,
10} from "@t3tools/client-runtime/state/terminal";
211
12import { writeTerminalOutputUpdate } from "../../components/ThreadTerminalDrawer";
3import { GHOSTTY_CELL_WIDE, GhosttyTerminalCore, ghosttyCellText } from "./core";13import { GHOSTTY_CELL_WIDE, GhosttyTerminalCore, ghosttyCellText } from "./core";
4import { loadGhosttyRuntime } from "./runtime";14import { loadGhosttyRuntime } from "./runtime";
515
...@@ -50,7 +60,7 @@ describe("ghosttyCellText", () => {...@@ -50,7 +60,7 @@ describe("ghosttyCellText", () => {
50describe("GhosttyTerminalCore snapshots", () => {60describe("GhosttyTerminalCore snapshots", () => {
51 const cores = new Set<GhosttyTerminalCore>();61 const cores = new Set<GhosttyTerminalCore>();
5262
53 async function createCore() {63 async function createCore(onData: (data: string) => void = () => {}) {
54 const core = await GhosttyTerminalCore.create(64 const core = await GhosttyTerminalCore.create(
55 12,65 12,
56 3,66 3,
...@@ -61,12 +71,40 @@ describe("GhosttyTerminalCore snapshots", () => {...@@ -61,12 +71,40 @@ describe("GhosttyTerminalCore snapshots", () => {
61 background: { r: 0, g: 0, b: 0 },71 background: { r: 0, g: 0, b: 0 },
62 cursor: { r: 255, g: 255, b: 255 },72 cursor: { r: 255, g: 255, b: 255 },
63 },73 },
64 () => {},74 onData,
65 );75 );
66 cores.add(core);76 cores.add(core);
67 return core;77 return core;
68 }78 }
6979
80 function createSession(history: string) {
81 return applyTerminalAttachStreamEvent(nextTerminalAttachSeedState(), {
82 type: "snapshot",
83 snapshot: {
84 threadId: "terminal-stream-test",
85 terminalId: "term-1",
86 cwd: "/repo",
87 worktreePath: null,
88 status: "running",
89 pid: 123,
90 history,
91 exitCode: null,
92 exitSignal: null,
93 label: "Terminal",
94 updatedAt: "2026-09-04T00:00:00.000Z",
95 },
96 });
97 }
98
99 function append(state: TerminalBufferState, data: string) {
100 return applyTerminalAttachStreamEvent(state, {
101 type: "output",
102 threadId: "terminal-stream-test",
103 terminalId: "term-1",
104 data,
105 });
106 }
107
70 afterEach(() => {108 afterEach(() => {
71 for (const core of cores) core.dispose();109 for (const core of cores) core.dispose();
72 cores.clear();110 cores.clear();
...@@ -145,4 +183,198 @@ describe("GhosttyTerminalCore snapshots", () => {...@@ -145,4 +183,198 @@ describe("GhosttyTerminalCore snapshots", () => {
145 core.dispose();183 core.dispose();
146 expect(free).toHaveBeenCalledWith(buffer, capacity);184 expect(free).toHaveBeenCalledWith(buffer, capacity);
147 });185 });
186
187 it.each(["varied", "identical"] as const)(
188 "preserves Ghostty state through a full MiB of %s output without rollover resets",
189 async (kind) => {
190 const [core, reference] = await Promise.all([createCore(), createCore()]);
191 const initial = "\x1b[31m";
192 let state = createSession(initial);
193 const first = readTerminalOutputUpdate(state.output, INITIAL_TERMINAL_OUTPUT_CURSOR);
194 writeTerminalOutputUpdate(core, first);
195 reference.resetAndWrite(initial);
196 let cursor = first.cursor;
197 const reset = vi.spyOn(core, "resetAndWrite");
198 const inputs: string[] = [];
199 let receivedCharacters = 0;
200
201 for (let index = 0; index < 128; index += 1) {
202 const data =
203 kind === "identical"
204 ? "x".repeat(8192)
205 : `${index.toString().padStart(4, "0")}\r\n${"x".repeat(8186)}`;
206 inputs.push(data);
207 state = append(state, data);
208 const update = readTerminalOutputUpdate(state.output, cursor);
209 if (update.type !== "append") throw new Error(`Expected append, received ${update.type}`);
210 receivedCharacters += update.data.length;
211 writeTerminalOutputUpdate(core, update);
212 cursor = update.cursor;
213 }
214
215 reference.write(inputs.join(""));
216 expect(receivedCharacters).toBe(1024 * 1024);
217 expect(state.output.retainedBytes).toBe(DEFAULT_MAX_TERMINAL_BUFFER_BYTES);
218 expect(reset).not.toHaveBeenCalled();
219 expect(core.snapshot()).toEqual(reference.snapshot());
220 },
221 );
222
223 it("preserves Unicode and split ANSI parser state across batched renderer reads", async () => {
224 const [core, reference] = await Promise.all([createCore(), createCore()]);
225 let state = createSession("");
226 const first = readTerminalOutputUpdate(state.output, INITIAL_TERMINAL_OUTPUT_CURSOR);
227 writeTerminalOutputUpdate(core, first);
228 let cursor = first.cursor;
229 const reset = vi.spyOn(core, "resetAndWrite");
230 const inputs = [
231 `${"a".repeat(16_383)}πŸ™‚`,
232 "\x1b[3",
233 "1m",
234 "e",
235 "\u0301η•ŒπŸ™‚",
236 "\x1b[0",
237 "m\r\n",
238 "\x1b]8;;https://t3.codes\x1b",
239 "\\link",
240 "\x1b]8;;\x1b",
241 "\\\x1b[?1049h",
242 "alternate",
243 "\x1b[?1049l",
244 "\r\nend",
245 ];
246 let received = "";
247 for (const [index, data] of inputs.entries()) {
248 state = append(state, data);
249 if (index % 3 !== 0 && index !== inputs.length - 1) continue;
250 const update = readTerminalOutputUpdate(state.output, cursor);
251 if (update.type !== "append") throw new Error(`Expected append, received ${update.type}`);
252 received += update.data;
253 writeTerminalOutputUpdate(core, update);
254 cursor = update.cursor;
255 }
256
257 reference.write(inputs.join(""));
258 expect(received).toBe(inputs.join(""));
259 expect(reset).not.toHaveBeenCalled();
260 expect(core.snapshot()).toEqual(reference.snapshot());
261 });
262
263 it("answers a live VT query when batched reads cross a chunk compaction", async () => {
264 const replies: string[] = [];
265 const core = await createCore((data) => replies.push(data));
266 core.write("\x1b[5n");
267 expect(replies).toEqual(["\x1b[0n"]);
268 replies.length = 0;
269
270 let state = createSession("");
271 const initial = readTerminalOutputUpdate(state.output, INITIAL_TERMINAL_OUTPUT_CURSOR);
272 writeTerminalOutputUpdate(core, initial);
273 let cursor = initial.cursor;
274 for (let index = 0; index < 1000; index += 1) {
275 state = append(state, "x");
276 const update = readTerminalOutputUpdate(state.output, cursor);
277 writeTerminalOutputUpdate(core, update);
278 cursor = update.cursor;
279 }
280 for (let index = 0; index < 24; index += 1) state = append(state, "x");
281 state = append(state, "\x1b[5n");
282 const update = readTerminalOutputUpdate(state.output, cursor);
283 writeTerminalOutputUpdate(core, update);
284
285 expect({ type: update.type, replies }).toEqual({ type: "append", replies: ["\x1b[0n"] });
286 });
287
288 it("recovers a lagging renderer once from bounded output and resumes appending", async () => {
289 const [core, reference] = await Promise.all([createCore(), createCore()]);
290 let state = createSession("\x1b[31mold");
291 const initial = readTerminalOutputUpdate(state.output, INITIAL_TERMINAL_OUTPUT_CURSOR);
292 writeTerminalOutputUpdate(core, initial);
293 const reset = vi.spyOn(core, "resetAndWrite");
294 const data = "line\r\n".repeat(8192);
295 for (let index = 0; index < 16; index += 1) state = append(state, data);
296
297 const recovery = readTerminalOutputUpdate(state.output, initial.cursor);
298 if (recovery.type !== "reset") throw new Error(`Expected reset, received ${recovery.type}`);
299 expect(new TextEncoder().encode(recovery.data).byteLength).toBe(
300 DEFAULT_MAX_TERMINAL_BUFFER_BYTES,
301 );
302 writeTerminalOutputUpdate(core, recovery);
303 reference.resetAndWrite(recovery.data);
304 expect(core.snapshot()).toEqual(reference.snapshot());
305
306 state = append(state, "\r\nlatest");
307 const next = readTerminalOutputUpdate(state.output, recovery.cursor);
308 expect(next.type).toBe("append");
309 writeTerminalOutputUpdate(core, next);
310 reference.write("\r\nlatest");
311 expect(reset).toHaveBeenCalledTimes(1);
312 expect(core.snapshot()).toEqual(reference.snapshot());
313 });
314
315 it("replays the latest retained output when WASM arrives after several events", async () => {
316 const pendingCore = createCore();
317 let state = createSession("before");
318 state = append(state, "\r\nduring ");
319 state = append(state, "πŸ™‚ load");
320 const core = await pendingCore;
321 const first = readTerminalOutputUpdate(state.output, INITIAL_TERMINAL_OUTPUT_CURSOR);
322 writeTerminalOutputUpdate(core, first);
323 const reference = await createCore();
324 reference.resetAndWrite(terminalOutputText(state.output));
325 const reset = vi.spyOn(core, "resetAndWrite");
326
327 state = append(state, "\r\nafter");
328 const next = readTerminalOutputUpdate(state.output, first.cursor);
329 expect(next).toMatchObject({ type: "append", data: "\r\nafter" });
330 writeTerminalOutputUpdate(core, next);
331 reference.write("\r\nafter");
332 expect(reset).not.toHaveBeenCalled();
333 expect(core.snapshot()).toEqual(reference.snapshot());
334 });
335
336 it("resets real Ghostty for a repeated snapshot, clear, restart, and a fresh attach", async () => {
337 const [core, reference] = await Promise.all([createCore(), createCore()]);
338 let state = createSession("hello");
339 const initial = readTerminalOutputUpdate(state.output, INITIAL_TERMINAL_OUTPUT_CURSOR);
340 writeTerminalOutputUpdate(core, initial);
341 let cursor = initial.cursor;
342
343 const snapshot = {
344 threadId: "terminal-stream-test",
345 terminalId: "term-1",
346 cwd: "/repo",
347 worktreePath: null,
348 status: "running" as const,
349 pid: 456,
350 history: "hello",
351 exitCode: null,
352 exitSignal: null,
353 label: "Terminal",
354 updatedAt: "2026-09-04T00:00:01.000Z",
355 };
356 const events = [
357 { type: "snapshot", snapshot },
358 { type: "cleared", threadId: snapshot.threadId, terminalId: snapshot.terminalId },
359 { type: "restarted", threadId: snapshot.threadId, terminalId: snapshot.terminalId, snapshot },
360 ] as const;
361 for (const event of events) {
362 core.write("\r\nstale local text");
363 state = applyTerminalAttachStreamEvent(state, event);
364 const update = readTerminalOutputUpdate(state.output, cursor);
365 expect(update.type).toBe("reset");
366 writeTerminalOutputUpdate(core, update);
367 cursor = update.cursor;
368 reference.resetAndWrite(event.type === "cleared" ? "" : "hello");
369 expect(core.snapshot()).toEqual(reference.snapshot());
370 }
371
372 core.write("\r\nstale local text");
373 state = createSession("hello");
374 const reattached = readTerminalOutputUpdate(state.output, cursor);
375 expect(reattached.type).toBe("reset");
376 writeTerminalOutputUpdate(core, reattached);
377 reference.resetAndWrite("hello");
378 expect(core.snapshot()).toEqual(reference.snapshot());
379 });
148});380});
packages/client-runtime/src/state/terminal.ts+5-3
...@@ -13,7 +13,7 @@ import { subscribe, type EnvironmentRpcInput } from "../rpc/client.ts";...@@ -13,7 +13,7 @@ import { subscribe, type EnvironmentRpcInput } from "../rpc/client.ts";
13import {13import {
14 applyTerminalAttachStreamEvent,14 applyTerminalAttachStreamEvent,
15 applyTerminalMetadataStreamEvent,15 applyTerminalMetadataStreamEvent,
16 EMPTY_TERMINAL_BUFFER_STATE,16 nextTerminalAttachSeedState,
17} from "./terminalSession.ts";17} from "./terminalSession.ts";
1818
19export function createTerminalEnvironmentAtoms<R, E>(19export function createTerminalEnvironmentAtoms<R, E>(
...@@ -40,8 +40,10 @@ export function createTerminalEnvironmentAtoms<R, E>(...@@ -40,8 +40,10 @@ export function createTerminalEnvironmentAtoms<R, E>(
40 attach: createEnvironmentSubscriptionAtomFamily(runtime, {40 attach: createEnvironmentSubscriptionAtomFamily(runtime, {
41 label: "environment-data:terminal:attach",41 label: "environment-data:terminal:attach",
42 subscribe: (input: EnvironmentRpcInput<typeof WS_METHODS.terminalAttach>) =>42 subscribe: (input: EnvironmentRpcInput<typeof WS_METHODS.terminalAttach>) =>
43 subscribe(WS_METHODS.terminalAttach, input).pipe(43 Stream.suspend(() =>
44 Stream.scan(EMPTY_TERMINAL_BUFFER_STATE, applyTerminalAttachStreamEvent),44 subscribe(WS_METHODS.terminalAttach, input).pipe(
45 Stream.scan(nextTerminalAttachSeedState(), applyTerminalAttachStreamEvent),
46 ),
45 ),47 ),
46 }),48 }),
47 events: createEnvironmentRpcSubscriptionAtomFamily(runtime, {49 events: createEnvironmentRpcSubscriptionAtomFamily(runtime, {
packages/client-runtime/src/state/terminalOutput.ts created+320
...@@ -0,0 +1,320 @@
1export interface TerminalOutputChunk {
2 /** UTF-16 string offset within this generation and reset. */
3 readonly startOffset: number;
4 readonly data: string;
5 readonly byteLength: number;
6}
7
8export interface TerminalOutputState {
9 readonly generation: number;
10 readonly chunks: ReadonlyArray<TerminalOutputChunk>;
11 readonly retainedBytes: number;
12 readonly resetVersion: number;
13 readonly nextOffset: number;
14}
15
16export interface TerminalOutputCursor {
17 readonly generation: number;
18 readonly resetVersion: number;
19 readonly offset: number;
20}
21
22/** Forces the first `readTerminalOutputUpdate` to resynchronize from a reset. */
23export const INITIAL_TERMINAL_OUTPUT_CURSOR = Object.freeze<TerminalOutputCursor>({
24 generation: -1,
25 resetVersion: -1,
26 offset: 0,
27});
28
29export type TerminalOutputUpdate =
30 | {
31 readonly type: "none";
32 readonly cursor: TerminalOutputCursor;
33 }
34 | {
35 readonly type: "reset";
36 readonly data: string;
37 readonly cursor: TerminalOutputCursor;
38 }
39 | {
40 readonly type: "append";
41 readonly cursor: TerminalOutputCursor;
42 readonly data: string;
43 };
44
45export const DEFAULT_MAX_TERMINAL_BUFFER_BYTES = 512 * 1024;
46const DEFAULT_TERMINAL_CHUNK_BYTES = 16 * 1024;
47const MAX_TERMINAL_OUTPUT_CHUNKS = 1_024;
48const textEncoder = new TextEncoder();
49// A BOM at a retained chunk boundary is terminal data, not an encoding marker.
50const textDecoder = new TextDecoder("utf-8", { ignoreBOM: true });
51
52export const EMPTY_TERMINAL_OUTPUT_STATE = Object.freeze<TerminalOutputState>({
53 generation: 0,
54 chunks: Object.freeze([]),
55 retainedBytes: 0,
56 resetVersion: 0,
57 nextOffset: 0,
58});
59
60interface Utf8Chunk {
61 readonly data: string;
62 readonly byteLength: number;
63}
64
65/**
66 * Split a string into chunks of at most `maxBytes` UTF-8 bytes without cutting
67 * a code point in half. The retained-output budget always supplies a positive
68 * size. Only new output is encoded on live updates.
69 *
70 * A chunk that fits whole is returned as the original string, so the common
71 * small-write path pays one encode and no decode.
72 */
73function splitStringByUtf8Bytes(data: string, maxBytes: number): ReadonlyArray<Utf8Chunk> {
74 if (data.length === 0) return [];
75
76 const encoded = textEncoder.encode(data);
77 if (encoded.byteLength <= maxBytes) {
78 return [{ data, byteLength: encoded.byteLength }];
79 }
80
81 const chunks: Utf8Chunk[] = [];
82 let offset = 0;
83 while (offset < encoded.byteLength) {
84 let end = Math.min(offset + maxBytes, encoded.byteLength);
85 while (end < encoded.byteLength && ((encoded[end] ?? 0) & 0xc0) === 0x80) {
86 end -= 1;
87 }
88 // A degenerate budget smaller than one code point still has to advance:
89 // include the whole code point rather than looping forever.
90 if (end === offset) {
91 end = Math.min(offset + maxBytes, encoded.byteLength);
92 while (end < encoded.byteLength && ((encoded[end] ?? 0) & 0xc0) === 0x80) {
93 end += 1;
94 }
95 }
96 const bytes = encoded.subarray(offset, end);
97 chunks.push({ data: textDecoder.decode(bytes), byteLength: bytes.byteLength });
98 offset = end;
99 }
100
101 return chunks;
102}
103
104function trimBufferToBytes(buffer: string, maxBufferBytes: number): string {
105 if (maxBufferBytes <= 0) {
106 return "";
107 }
108
109 const encoded = textEncoder.encode(buffer);
110 if (encoded.byteLength <= maxBufferBytes) {
111 return buffer;
112 }
113
114 let start = encoded.byteLength - maxBufferBytes;
115 while (start < encoded.length) {
116 const byte = encoded[start];
117 if (byte === undefined || (byte & 0b1100_0000) !== 0b1000_0000) {
118 break;
119 }
120 start += 1;
121 }
122
123 return textDecoder.decode(encoded.subarray(start));
124}
125
126function splitOutputChunks(
127 data: string,
128 firstOffset: number,
129 maxChunkBytes = DEFAULT_TERMINAL_CHUNK_BYTES,
130): {
131 readonly chunks: ReadonlyArray<TerminalOutputChunk>;
132 readonly nextOffset: number;
133 readonly byteLength: number;
134} {
135 const split = splitStringByUtf8Bytes(data, maxChunkBytes);
136 let byteLength = 0;
137 let nextOffset = firstOffset;
138 const chunks = split.map((chunk) => {
139 byteLength += chunk.byteLength;
140 const startOffset = nextOffset;
141 nextOffset += chunk.data.length;
142 return {
143 startOffset,
144 data: chunk.data,
145 byteLength: chunk.byteLength,
146 };
147 });
148
149 return {
150 chunks,
151 nextOffset,
152 byteLength,
153 };
154}
155
156/**
157 * Merge adjacent chunks without changing their string positions. A reader can
158 * still append the unread suffix when its cursor falls inside a merged chunk.
159 */
160function compactRetainedChunks(chunks: ReadonlyArray<TerminalOutputChunk>) {
161 const compacted: TerminalOutputChunk[] = [];
162 for (const chunk of chunks) {
163 const previous = compacted.at(-1);
164 if (
165 previous !== undefined &&
166 previous.startOffset + previous.data.length === chunk.startOffset &&
167 previous.byteLength + chunk.byteLength <= DEFAULT_TERMINAL_CHUNK_BYTES
168 ) {
169 compacted[compacted.length - 1] = {
170 startOffset: previous.startOffset,
171 data: `${previous.data}${chunk.data}`,
172 byteLength: previous.byteLength + chunk.byteLength,
173 };
174 } else {
175 compacted.push(chunk);
176 }
177 }
178 return compacted;
179}
180
181// Scan only the removed prefix instead of encoding retained output again.
182function trimOutputChunkStart(
183 chunk: TerminalOutputChunk,
184 bytesToDrop: number,
185): TerminalOutputChunk {
186 let offset = 0;
187 let droppedBytes = 0;
188 while (droppedBytes < bytesToDrop && offset < chunk.data.length) {
189 const codepoint = chunk.data.codePointAt(offset)!;
190 droppedBytes += codepoint <= 0x7f ? 1 : codepoint <= 0x7ff ? 2 : codepoint <= 0xffff ? 3 : 4;
191 offset += codepoint <= 0xffff ? 1 : 2;
192 }
193 return {
194 ...chunk,
195 startOffset: chunk.startOffset + offset,
196 data: chunk.data.slice(offset),
197 byteLength: chunk.byteLength - droppedBytes,
198 };
199}
200
201function appendOutput(
202 current: TerminalOutputState,
203 data: string,
204 maxBufferBytes: number,
205): TerminalOutputState {
206 if (data.length === 0) return current;
207 if (maxBufferBytes <= 0) {
208 return {
209 generation: current.generation,
210 chunks: [],
211 retainedBytes: 0,
212 resetVersion: current.resetVersion + 1,
213 nextOffset: current.nextOffset + data.length,
214 };
215 }
216 const appended = splitOutputChunks(
217 data,
218 current.nextOffset,
219 Math.min(DEFAULT_TERMINAL_CHUNK_BYTES, Math.max(1, maxBufferBytes)),
220 );
221
222 const chunks = [...current.chunks, ...appended.chunks];
223 let retainedBytes = current.retainedBytes + appended.byteLength;
224 let firstRetainedIndex = 0;
225 while (retainedBytes > maxBufferBytes && firstRetainedIndex < chunks.length) {
226 const first = chunks[firstRetainedIndex]!;
227 const bytesToDrop = retainedBytes - maxBufferBytes;
228 if (bytesToDrop < first.byteLength) {
229 const trimmed = trimOutputChunkStart(first, bytesToDrop);
230 retainedBytes -= first.byteLength - trimmed.byteLength;
231 if (trimmed.byteLength > 0) {
232 chunks[firstRetainedIndex] = trimmed;
233 } else {
234 firstRetainedIndex += 1;
235 }
236 break;
237 }
238 retainedBytes -= first.byteLength;
239 firstRetainedIndex += 1;
240 }
241
242 let retainedChunks = firstRetainedIndex === 0 ? chunks : chunks.slice(firstRetainedIndex);
243 if (retainedChunks.length > MAX_TERMINAL_OUTPUT_CHUNKS) {
244 retainedChunks = compactRetainedChunks(retainedChunks);
245 const excessChunks = retainedChunks.length - MAX_TERMINAL_OUTPUT_CHUNKS;
246 if (excessChunks > 0) {
247 for (const chunk of retainedChunks.slice(0, excessChunks)) {
248 retainedBytes -= chunk.byteLength;
249 }
250 retainedChunks = retainedChunks.slice(excessChunks);
251 }
252 }
253
254 return {
255 generation: current.generation,
256 chunks: retainedChunks,
257 retainedBytes,
258 resetVersion: current.resetVersion,
259 nextOffset: appended.nextOffset,
260 };
261}
262
263function resetOutput(
264 current: TerminalOutputState,
265 data: string,
266 maxBufferBytes: number,
267): TerminalOutputState {
268 const retained = trimBufferToBytes(data, maxBufferBytes);
269 const reset = splitOutputChunks(
270 retained,
271 0,
272 Math.min(DEFAULT_TERMINAL_CHUNK_BYTES, Math.max(1, maxBufferBytes)),
273 );
274 return {
275 generation: current.generation,
276 chunks: reset.chunks,
277 retainedBytes: reset.byteLength,
278 resetVersion: current.resetVersion + 1,
279 nextOffset: reset.nextOffset,
280 };
281}
282
283export function terminalOutputText(output: TerminalOutputState): string {
284 return output.chunks.map((chunk) => chunk.data).join("");
285}
286
287export function readTerminalOutputUpdate(
288 output: TerminalOutputState,
289 cursor: TerminalOutputCursor,
290): TerminalOutputUpdate {
291 const nextCursor = {
292 generation: output.generation,
293 resetVersion: output.resetVersion,
294 offset: output.nextOffset,
295 };
296 const firstChunk = output.chunks[0];
297 if (
298 cursor.generation !== output.generation ||
299 cursor.resetVersion !== output.resetVersion ||
300 cursor.offset < (firstChunk?.startOffset ?? output.nextOffset)
301 ) {
302 return { type: "reset", data: terminalOutputText(output), cursor: nextCursor };
303 }
304
305 const appended = output.chunks.filter(
306 (chunk) => chunk.startOffset + chunk.data.length > cursor.offset,
307 );
308 if (appended.length === 0) {
309 return { type: "none", cursor: nextCursor };
310 }
311 return {
312 type: "append",
313 data: appended
314 .map((chunk) => chunk.data.slice(Math.max(0, cursor.offset - chunk.startOffset)))
315 .join(""),
316 cursor: nextCursor,
317 };
318}
319
320export { appendOutput, resetOutput };
packages/client-runtime/src/state/terminalSession.test.ts+272-3
...@@ -1,4 +1,4 @@...@@ -1,4 +1,4 @@
1import { describe, expect, it } from "vite-plus/test";1import { describe, expect, it, vi } from "vite-plus/test";
22
3import { EnvironmentId, TerminalSessionSnapshot, ThreadId } from "@t3tools/contracts";3import { EnvironmentId, TerminalSessionSnapshot, ThreadId } from "@t3tools/contracts";
44
...@@ -6,8 +6,13 @@ import {...@@ -6,8 +6,13 @@ import {
6 applyTerminalAttachStreamEvent,6 applyTerminalAttachStreamEvent,
7 applyTerminalMetadataStreamEvent,7 applyTerminalMetadataStreamEvent,
8 combineTerminalSessionState,8 combineTerminalSessionState,
9 DEFAULT_MAX_TERMINAL_BUFFER_BYTES,
9 EMPTY_TERMINAL_BUFFER_STATE,10 EMPTY_TERMINAL_BUFFER_STATE,
11 INITIAL_TERMINAL_OUTPUT_CURSOR,
12 nextTerminalAttachSeedState,
13 readTerminalOutputUpdate,
10 selectRunningSubprocessTerminalIds,14 selectRunningSubprocessTerminalIds,
15 terminalOutputText,
11} from "./terminalSession.ts";16} from "./terminalSession.ts";
1217
13const TARGET = {18const TARGET = {
...@@ -126,11 +131,11 @@ describe("terminal session reducers", () => {...@@ -126,11 +131,11 @@ describe("terminal session reducers", () => {
126 );131 );
127132
128 expect(output).toMatchObject({133 expect(output).toMatchObject({
129 buffer: "lo world",
130 status: "running",134 status: "running",
131 error: null,135 error: null,
132 version: 2,136 version: 2,
133 });137 });
138 expect(terminalOutputText(output.output)).toBe("lo world");
134 });139 });
135140
136 it("does not advance the lifecycle for the initial attach snapshot", () => {141 it("does not advance the lifecycle for the initial attach snapshot", () => {
...@@ -220,6 +225,270 @@ describe("terminal session reducers", () => {...@@ -220,6 +225,270 @@ describe("terminal session reducers", () => {
220 4,225 4,
221 );226 );
222227
223 expect(state.buffer).toBe("πŸ™‚");228 expect(terminalOutputText(state.output)).toBe("πŸ™‚");
229 });
230
231 it("preserves a BOM code point at a new output chunk boundary", () => {
232 const initial = applyTerminalAttachStreamEvent(EMPTY_TERMINAL_BUFFER_STATE, {
233 type: "snapshot",
234 snapshot: { ...BASE_SNAPSHOT, history: "" },
235 });
236 const cursor = readTerminalOutputUpdate(initial.output, INITIAL_TERMINAL_OUTPUT_CURSOR).cursor;
237 const data = `${"x".repeat(16_384)}\uFEFF${"y".repeat(20)}`;
238 const state = applyTerminalAttachStreamEvent(initial, {
239 type: "output",
240 threadId: TARGET.threadId,
241 terminalId: TARGET.terminalId,
242 data,
243 });
244
245 expect(terminalOutputText(state.output)).toBe(data);
246 expect(readTerminalOutputUpdate(state.output, cursor)).toMatchObject({ type: "append", data });
247 expect(state.output.retainedBytes).toBe(new TextEncoder().encode(data).byteLength);
248 });
249
250 it("preserves a leading BOM in the retained snapshot tail", () => {
251 const state = applyTerminalAttachStreamEvent(
252 EMPTY_TERMINAL_BUFFER_STATE,
253 { type: "snapshot", snapshot: { ...BASE_SNAPSHOT, history: "discard\uFEFFtail" } },
254 7,
255 );
256
257 expect(terminalOutputText(state.output)).toBe("\uFEFFtail");
258 expect(state.output.retainedBytes).toBe(7);
259 });
260
261 it("trims whole Unicode code points from a partially retained chunk", () => {
262 let state = applyTerminalAttachStreamEvent(
263 EMPTY_TERMINAL_BUFFER_STATE,
264 { type: "snapshot", snapshot: { ...BASE_SNAPSHOT, history: "éη•ŒπŸ™‚end" } },
265 12,
266 );
267 let cursor = readTerminalOutputUpdate(state.output, INITIAL_TERMINAL_OUTPUT_CURSOR).cursor;
268 for (const [data, expected, retainedBytes] of [
269 ["x", "η•ŒπŸ™‚endx", 11],
270 ["yz", "πŸ™‚endxyz", 10],
271 ["abc", "endxyzabc", 9],
272 ] as const) {
273 state = applyTerminalAttachStreamEvent(
274 state,
275 { type: "output", threadId: TARGET.threadId, terminalId: TARGET.terminalId, data },
276 12,
277 );
278 const update = readTerminalOutputUpdate(state.output, cursor);
279 expect(update).toMatchObject({ type: "append", data });
280 expect(terminalOutputText(state.output)).toBe(expected);
281 expect(state.output.retainedBytes).toBe(retainedBytes);
282 cursor = update.cursor;
283 }
284 });
285
286 it("delivers all output when several events arrive before a renderer reads", () => {
287 const initial = applyTerminalAttachStreamEvent(EMPTY_TERMINAL_BUFFER_STATE, {
288 type: "snapshot",
289 snapshot: BASE_SNAPSHOT,
290 });
291 const cursor = readTerminalOutputUpdate(initial.output, INITIAL_TERMINAL_OUTPUT_CURSOR).cursor;
292 let state = initial;
293 for (const data of [" one", " two", " three"]) {
294 state = applyTerminalAttachStreamEvent(state, {
295 type: "output",
296 threadId: TARGET.threadId,
297 terminalId: TARGET.terminalId,
298 data,
299 });
300 }
301 const update = readTerminalOutputUpdate(state.output, cursor);
302 expect(update).toMatchObject({ type: "append", data: " one two three" });
303 expect(readTerminalOutputUpdate(state.output, update.cursor).type).toBe("none");
304 });
305
306 it("preserves the byte-limited tail and resets a cursor before a partially trimmed chunk", () => {
307 const initial = applyTerminalAttachStreamEvent(EMPTY_TERMINAL_BUFFER_STATE, {
308 type: "snapshot",
309 snapshot: { ...BASE_SNAPSHOT, history: "" },
310 });
311 const staleCursor = readTerminalOutputUpdate(
312 initial.output,
313 INITIAL_TERMINAL_OUTPUT_CURSOR,
314 ).cursor;
315 const first = applyTerminalAttachStreamEvent(initial, {
316 type: "output",
317 threadId: TARGET.threadId,
318 terminalId: TARGET.terminalId,
319 data: "hello",
320 });
321 const caughtUpCursor = readTerminalOutputUpdate(first.output, staleCursor).cursor;
322 const state = applyTerminalAttachStreamEvent(
323 first,
324 { type: "output", threadId: TARGET.threadId, terminalId: TARGET.terminalId, data: " world" },
325 8,
326 );
327
328 expect(readTerminalOutputUpdate(state.output, caughtUpCursor)).toMatchObject({
329 type: "append",
330 data: " world",
331 });
332 expect(readTerminalOutputUpdate(state.output, staleCursor)).toMatchObject({
333 type: "reset",
334 data: "lo world",
335 });
336 expect(state.output.retainedBytes).toBe(8);
337 });
338
339 it("does not encode retained history again when appending at the byte limit", () => {
340 let state = applyTerminalAttachStreamEvent(EMPTY_TERMINAL_BUFFER_STATE, {
341 type: "snapshot",
342 snapshot: { ...BASE_SNAPSHOT, history: "x".repeat(DEFAULT_MAX_TERMINAL_BUFFER_BYTES) },
343 });
344 const data = "y".repeat(8192);
345 const encode = vi.spyOn(TextEncoder.prototype, "encode");
346 try {
347 for (let index = 0; index < 100; index += 1) {
348 state = applyTerminalAttachStreamEvent(state, {
349 type: "output",
350 threadId: TARGET.threadId,
351 terminalId: TARGET.terminalId,
352 data,
353 });
354 }
355 expect(encode.mock.calls.reduce((total, [text]) => total + (text?.length ?? 0), 0)).toBe(
356 data.length * 100,
357 );
358 expect(state.output.retainedBytes).toBe(DEFAULT_MAX_TERMINAL_BUFFER_BYTES);
359 } finally {
360 encode.mockRestore();
361 }
362 });
363
364 it.each([0, -1])("discards output without encoding when the byte budget is %s", (maxBytes) => {
365 const initial = applyTerminalAttachStreamEvent(EMPTY_TERMINAL_BUFFER_STATE, {
366 type: "snapshot",
367 snapshot: BASE_SNAPSHOT,
368 });
369 const cursor = readTerminalOutputUpdate(initial.output, INITIAL_TERMINAL_OUTPUT_CURSOR).cursor;
370 const data = "x".repeat(65_536);
371 const encode = vi.spyOn(TextEncoder.prototype, "encode");
372 try {
373 const discarded = applyTerminalAttachStreamEvent(
374 initial,
375 { type: "output", threadId: TARGET.threadId, terminalId: TARGET.terminalId, data },
376 maxBytes,
377 );
378 expect(encode).not.toHaveBeenCalled();
379 expect(discarded.output.nextOffset).toBe(initial.output.nextOffset + data.length);
380 expect(discarded.output.retainedBytes).toBe(0);
381 expect(readTerminalOutputUpdate(discarded.output, cursor)).toMatchObject({
382 type: "reset",
383 data: "",
384 });
385
386 const empty = applyTerminalAttachStreamEvent(
387 initial,
388 { type: "output", threadId: TARGET.threadId, terminalId: TARGET.terminalId, data: "" },
389 maxBytes,
390 );
391 expect(empty.output).toBe(initial.output);
392 expect(encode).not.toHaveBeenCalled();
393 } finally {
394 encode.mockRestore();
395 }
396 });
397
398 it("resets for repeated snapshots, clear, and restart even when output text repeats", () => {
399 let state = applyTerminalAttachStreamEvent(EMPTY_TERMINAL_BUFFER_STATE, {
400 type: "snapshot",
401 snapshot: BASE_SNAPSHOT,
402 });
403 let cursor = readTerminalOutputUpdate(state.output, INITIAL_TERMINAL_OUTPUT_CURSOR).cursor;
404 state = applyTerminalAttachStreamEvent(state, { type: "snapshot", snapshot: BASE_SNAPSHOT });
405 const repeated = readTerminalOutputUpdate(state.output, cursor);
406 expect(repeated).toMatchObject({ type: "reset", data: "hello" });
407 expect(state.version).toBe(2);
408 cursor = repeated.cursor;
409
410 state = applyTerminalAttachStreamEvent(state, {
411 type: "cleared",
412 threadId: TARGET.threadId,
413 terminalId: TARGET.terminalId,
414 });
415 const cleared = readTerminalOutputUpdate(state.output, cursor);
416 expect(cleared).toMatchObject({ type: "reset", data: "" });
417 state = applyTerminalAttachStreamEvent(state, {
418 type: "restarted",
419 threadId: TARGET.threadId,
420 terminalId: TARGET.terminalId,
421 snapshot: BASE_SNAPSHOT,
422 });
423 expect(readTerminalOutputUpdate(state.output, cleared.cursor)).toMatchObject({
424 type: "reset",
425 data: "hello",
426 });
427 expect(state.lifecycleVersion).toBe(2);
428 });
429
430 it("does not reuse a renderer cursor when a fresh attach has matching counters", () => {
431 const first = applyTerminalAttachStreamEvent(nextTerminalAttachSeedState(), {
432 type: "snapshot",
433 snapshot: BASE_SNAPSHOT,
434 });
435 const cursor = readTerminalOutputUpdate(first.output, INITIAL_TERMINAL_OUTPUT_CURSOR).cursor;
436 const next = applyTerminalAttachStreamEvent(nextTerminalAttachSeedState(), {
437 type: "snapshot",
438 snapshot: { ...BASE_SNAPSHOT, history: "other" },
439 });
440 expect(next.output.resetVersion).toBe(first.output.resetVersion);
441 expect(next.output.nextOffset).toBe(first.output.nextOffset);
442 expect(readTerminalOutputUpdate(next.output, cursor)).toMatchObject({
443 type: "reset",
444 data: "other",
445 });
446 expect(next.lifecycleVersion).toBe(0);
447 });
448
449 it("keeps appending while compacting metadata for many small writes", () => {
450 let state = applyTerminalAttachStreamEvent(EMPTY_TERMINAL_BUFFER_STATE, {
451 type: "snapshot",
452 snapshot: { ...BASE_SNAPSHOT, history: "" },
453 });
454 let cursor = readTerminalOutputUpdate(state.output, INITIAL_TERMINAL_OUTPUT_CURSOR).cursor;
455 let received = "";
456 for (let index = 0; index < 2500; index += 1) {
457 state = applyTerminalAttachStreamEvent(state, {
458 type: "output",
459 threadId: TARGET.threadId,
460 terminalId: TARGET.terminalId,
461 data: "x",
462 });
463 const update = readTerminalOutputUpdate(state.output, cursor);
464 if (update.type !== "append") throw new Error(`Expected append, received ${update.type}`);
465 received += update.data;
466 cursor = update.cursor;
467 }
468 expect(received).toBe("x".repeat(2500));
469 expect(state.output.chunks.length).toBeLessThan(1024);
470 expect(terminalOutputText(state.output)).toBe(received);
471 });
472
473 it("appends every unread character across a compaction boundary", () => {
474 let state = applyTerminalAttachStreamEvent(EMPTY_TERMINAL_BUFFER_STATE, {
475 type: "snapshot",
476 snapshot: { ...BASE_SNAPSHOT, history: "" },
477 });
478 let cursor = readTerminalOutputUpdate(state.output, INITIAL_TERMINAL_OUTPUT_CURSOR).cursor;
479 const writes = Array.from({ length: 1200 }, (_, index) => ["x", "é", "η•Œ", "πŸ™‚"][index % 4]!);
480 for (const [index, data] of writes.entries()) {
481 state = applyTerminalAttachStreamEvent(state, {
482 type: "output",
483 threadId: TARGET.threadId,
484 terminalId: TARGET.terminalId,
485 data,
486 });
487 if (index === 99) cursor = readTerminalOutputUpdate(state.output, cursor).cursor;
488 }
489 expect(readTerminalOutputUpdate(state.output, cursor)).toMatchObject({
490 type: "append",
491 data: writes.slice(100).join(""),
492 });
224 });493 });
225});494});
packages/client-runtime/src/state/terminalSession.ts+40-35
...@@ -6,10 +6,27 @@ import type {...@@ -6,10 +6,27 @@ import type {
6 TerminalSummary,6 TerminalSummary,
7 ThreadId,7 ThreadId,
8} from "@t3tools/contracts";8} from "@t3tools/contracts";
9import {
10 appendOutput,
11 DEFAULT_MAX_TERMINAL_BUFFER_BYTES,
12 EMPTY_TERMINAL_OUTPUT_STATE,
13 resetOutput,
14 type TerminalOutputState,
15} from "./terminalOutput.ts";
16
17export {
18 DEFAULT_MAX_TERMINAL_BUFFER_BYTES,
19 INITIAL_TERMINAL_OUTPUT_CURSOR,
20 readTerminalOutputUpdate,
21 terminalOutputText,
22 type TerminalOutputCursor,
23 type TerminalOutputState,
24 type TerminalOutputUpdate,
25} from "./terminalOutput.ts";
926
10export interface TerminalSessionState {27export interface TerminalSessionState {
11 readonly summary: TerminalSummary | null;28 readonly summary: TerminalSummary | null;
12 readonly buffer: string;29 readonly output: TerminalOutputState;
13 readonly status: TerminalSessionSnapshot["status"] | "closed";30 readonly status: TerminalSessionSnapshot["status"] | "closed";
14 readonly error: string | null;31 readonly error: string | null;
15 readonly hasRunningSubprocess: boolean;32 readonly hasRunningSubprocess: boolean;
...@@ -19,7 +36,7 @@ export interface TerminalSessionState {...@@ -19,7 +36,7 @@ export interface TerminalSessionState {
19}36}
2037
21export interface TerminalBufferState {38export interface TerminalBufferState {
22 readonly buffer: string;39 readonly output: TerminalOutputState;
23 readonly status: TerminalSessionSnapshot["status"] | "closed";40 readonly status: TerminalSessionSnapshot["status"] | "closed";
24 readonly error: string | null;41 readonly error: string | null;
25 readonly updatedAt: string | null;42 readonly updatedAt: string | null;
...@@ -47,7 +64,7 @@ export function selectRunningSubprocessTerminalIds(...@@ -47,7 +64,7 @@ export function selectRunningSubprocessTerminalIds(
47}64}
4865
49export const EMPTY_TERMINAL_BUFFER_STATE = Object.freeze<TerminalBufferState>({66export const EMPTY_TERMINAL_BUFFER_STATE = Object.freeze<TerminalBufferState>({
50 buffer: "",67 output: EMPTY_TERMINAL_OUTPUT_STATE,
51 status: "closed",68 status: "closed",
52 error: null,69 error: null,
53 updatedAt: null,70 updatedAt: null,
...@@ -57,7 +74,7 @@ export const EMPTY_TERMINAL_BUFFER_STATE = Object.freeze<TerminalBufferState>({...@@ -57,7 +74,7 @@ export const EMPTY_TERMINAL_BUFFER_STATE = Object.freeze<TerminalBufferState>({
5774
58export const EMPTY_TERMINAL_SESSION_STATE = Object.freeze<TerminalSessionState>({75export const EMPTY_TERMINAL_SESSION_STATE = Object.freeze<TerminalSessionState>({
59 summary: null,76 summary: null,
60 buffer: "",77 output: EMPTY_TERMINAL_OUTPUT_STATE,
61 status: "closed",78 status: "closed",
62 error: null,79 error: null,
63 hasRunningSubprocess: false,80 hasRunningSubprocess: false,
...@@ -66,43 +83,31 @@ export const EMPTY_TERMINAL_SESSION_STATE = Object.freeze<TerminalSessionState>(...@@ -66,43 +83,31 @@ export const EMPTY_TERMINAL_SESSION_STATE = Object.freeze<TerminalSessionState>(
66 lifecycleVersion: 0,83 lifecycleVersion: 0,
67});84});
6885
69export const DEFAULT_MAX_TERMINAL_BUFFER_BYTES = 512 * 1024;86let terminalAttachGeneration = 0;
70const textEncoder = new TextEncoder();
71const textDecoder = new TextDecoder();
7287
73function trimBufferToBytes(buffer: string, maxBufferBytes: number): string {88/** A reinstalled attach stream must not reuse an old renderer's output cursor. */
74 if (maxBufferBytes <= 0) {89export function nextTerminalAttachSeedState(): TerminalBufferState {
75 return "";90 return {
76 }91 ...EMPTY_TERMINAL_BUFFER_STATE,
7792 output: {
78 const encoded = textEncoder.encode(buffer);93 ...EMPTY_TERMINAL_OUTPUT_STATE,
79 if (encoded.byteLength <= maxBufferBytes) {94 generation: ++terminalAttachGeneration,
80 return buffer;95 },
81 }96 };
82
83 let start = encoded.byteLength - maxBufferBytes;
84 while (start < encoded.length) {
85 const byte = encoded[start];
86 if (byte === undefined || (byte & 0b1100_0000) !== 0b1000_0000) {
87 break;
88 }
89 start += 1;
90 }
91
92 return textDecoder.decode(encoded.subarray(start));
93}97}
9498
95export function terminalBufferStateFromSnapshot(99export function terminalBufferStateFromSnapshot(
96 snapshot: TerminalSessionSnapshot,100 snapshot: TerminalSessionSnapshot,
97 maxBufferBytes: number,101 maxBufferBytes: number,
102 current: TerminalBufferState = EMPTY_TERMINAL_BUFFER_STATE,
98): TerminalBufferState {103): TerminalBufferState {
99 return {104 return {
100 buffer: trimBufferToBytes(snapshot.history, maxBufferBytes),105 output: resetOutput(current.output, snapshot.history, maxBufferBytes),
101 status: snapshot.status,106 status: snapshot.status,
102 error: null,107 error: null,
103 updatedAt: snapshot.updatedAt,108 updatedAt: snapshot.updatedAt,
104 version: 1,109 version: current.version + 1,
105 lifecycleVersion: 0,110 lifecycleVersion: current.lifecycleVersion,
106 };111 };
107}112}
108113
...@@ -118,7 +123,7 @@ export function combineTerminalSessionState(...@@ -118,7 +123,7 @@ export function combineTerminalSessionState(
118): TerminalSessionState {123): TerminalSessionState {
119 return {124 return {
120 summary,125 summary,
121 buffer: buffer.buffer,126 output: buffer.output,
122 status: buffer.version > 0 ? buffer.status : (summary?.status ?? buffer.status),127 status: buffer.version > 0 ? buffer.status : (summary?.status ?? buffer.status),
123 error: buffer.error,128 error: buffer.error,
124 hasRunningSubprocess: summary?.hasRunningSubprocess ?? false,129 hasRunningSubprocess: summary?.hasRunningSubprocess ?? false,
...@@ -136,19 +141,19 @@ export function applyTerminalAttachStreamEvent(...@@ -136,19 +141,19 @@ export function applyTerminalAttachStreamEvent(
136 switch (event.type) {141 switch (event.type) {
137 case "snapshot":142 case "snapshot":
138 return {143 return {
139 ...terminalBufferStateFromSnapshot(event.snapshot, maxBufferBytes),144 ...terminalBufferStateFromSnapshot(event.snapshot, maxBufferBytes, current),
140 lifecycleVersion:145 lifecycleVersion:
141 current.version === 0 ? current.lifecycleVersion : current.lifecycleVersion + 1,146 current.version === 0 ? current.lifecycleVersion : current.lifecycleVersion + 1,
142 };147 };
143 case "restarted":148 case "restarted":
144 return {149 return {
145 ...terminalBufferStateFromSnapshot(event.snapshot, maxBufferBytes),150 ...terminalBufferStateFromSnapshot(event.snapshot, maxBufferBytes, current),
146 lifecycleVersion: current.lifecycleVersion + 1,151 lifecycleVersion: current.lifecycleVersion + 1,
147 };152 };
148 case "output":153 case "output":
149 return {154 return {
150 ...current,155 ...current,
151 buffer: trimBufferToBytes(`${current.buffer}${event.data}`, maxBufferBytes),156 output: appendOutput(current.output, event.data, maxBufferBytes),
152 status: current.status === "closed" ? "running" : current.status,157 status: current.status === "closed" ? "running" : current.status,
153 error: null,158 error: null,
154 version: current.version + 1,159 version: current.version + 1,
...@@ -156,7 +161,7 @@ export function applyTerminalAttachStreamEvent(...@@ -156,7 +161,7 @@ export function applyTerminalAttachStreamEvent(
156 case "cleared":161 case "cleared":
157 return {162 return {
158 ...current,163 ...current,
159 buffer: "",164 output: resetOutput(current.output, "", maxBufferBytes),
160 error: null,165 error: null,
161 version: current.version + 1,166 version: current.version + 1,
162 };167 };