perf(terminal): coalesce PTY output into one xterm.write per animation frame (v2 transport)

apps/desktop/src/renderer/lib/terminal/terminal-ws-transport.test.ts

@@ -119,6 +119,117 @@ afterEach(() => {
 	jest.useRealTimers();
 });
 
+describe("PTY output write coalescing", () => {
+	let frameCallbacks: Map<number, FrameRequestCallback>;
+	let nextFrameId: number;
+	const originalRaf = globalThis.requestAnimationFrame;
+	const originalCancelRaf = globalThis.cancelAnimationFrame;
+
+	function fireFrame() {
+		const callbacks = [...frameCallbacks.values()];
+		frameCallbacks.clear();
+		for (const callback of callbacks) {
+			callback(performance.now());
+		}
+	}
+
+	beforeEach(() => {
+		frameCallbacks = new Map();
+		nextFrameId = 1;
+		globalThis.requestAnimationFrame = (callback: FrameRequestCallback) => {
+			const id = nextFrameId++;
+			frameCallbacks.set(id, callback);
+			return id;
+		};
+		globalThis.cancelAnimationFrame = (id: number) => {
+			frameCallbacks.delete(id);
+		};
+	});
+
+	afterEach(() => {
+		globalThis.requestAnimationFrame = originalRaf;
+		globalThis.cancelAnimationFrame = originalCancelRaf;
+	});
+
+	function connectWithRecordingTerminal() {
+		const transport = createTransport();
+		const terminal = createMockTerminal();
+		const writes: string[] = [];
+		const events: string[] = [];
+		(terminal as unknown as { write: (d: Uint8Array) => void }).write = (
+			data: Uint8Array,
+		) => {
+			const text = new TextDecoder().decode(data);
+			writes.push(text);
+			events.push(`write:${text}`);
+		};
+		(terminal as unknown as { writeln: (s: string) => void }).writeln = (
+			line: string,
+		) => {
+			events.push(`writeln:${line}`);
+		};
+		connect(transport, terminal, "ws://host/terminal/t1");
+		const socket = MockWebSocket.instances[0];
+		if (!socket) throw new Error("expected websocket instance");
+		socket.open();
+		socket.message(JSON.stringify({ type: "attached", terminalId: "t1" }));
+		return { transport, socket, writes, events };
+	}
+
+	function binaryFrame(text: string): ArrayBuffer {
+		const bytes = new TextEncoder().encode(text);
+		return bytes.buffer.slice(
+			bytes.byteOffset,
+			bytes.byteOffset + bytes.byteLength,
+		) as ArrayBuffer;
+	}
+
+	test("coalesces binary frames into one terminal.write per frame", () => {
+		const { writes, socket } = connectWithRecordingTerminal();
+
+		socket.message(binaryFrame("chunk1"));
+		socket.message(binaryFrame("chunk2"));
+		socket.message(binaryFrame("chunk3"));
+		expect(writes).toEqual([]);
+
+		fireFrame();
+		expect(writes).toEqual(["chunk1chunk2chunk3"]);
+	});
+
+	test("flushes pending PTY bytes before writing the exit notice", () => {
+		const { events, socket } = connectWithRecordingTerminal();
+
+		socket.message(binaryFrame("final output"));
+		socket.message(JSON.stringify({ type: "exit", exitCode: 0, signal: 0 }));
+
+		expect(events).toEqual([
+			"write:final output",
+			"writeln:\r\n[terminal] exited with code 0 (signal 0)",
+		]);
+	});
+
+	test("does not flush pending PTY bytes for non-writing control messages", () => {
+		const { writes, socket } = connectWithRecordingTerminal();
+
+		socket.message(binaryFrame("prompt"));
+		socket.message(JSON.stringify({ type: "title", title: "agent" }));
+		socket.message(JSON.stringify({ type: "attached", terminalId: "t1" }));
+		expect(writes).toEqual([]);
+
+		fireFrame();
+		expect(writes).toEqual(["prompt"]);
+	});
+
+	test("flushes pending PTY bytes when the socket closes", () => {
+		const { writes, socket } = connectWithRecordingTerminal();
+
+		socket.message(binaryFrame("tail"));
+		socket.close(1006, "host restart");
+
+		expect(writes).toEqual(["tail"]);
+	});
+});
+
 describe("terminal-ws-transport", () => {
 	test("server-sent error routes to logs, not xterm, and stops reconnect", () => {
 		const transport = createTransport();

apps/desktop/src/renderer/lib/terminal/terminal-ws-transport.ts

@@ -1,5 +1,6 @@
 import { primeRelayAffinity } from "@superset/workspace-client";
 import type { Terminal as XTerm } from "@xterm/xterm";
+import { createWriteCoalescer, type WriteCoalescer } from "./write-coalescer";
 
 export type ConnectionState = "disconnected" | "connecting" | "open" | "closed";
 
@@ -64,6 +65,12 @@ export interface TerminalTransport {
 	/** Internal: bound resume handler shared by the online/focus/visibility
 	 * listeners, so they can be removed on teardown. */
 	_resumeListener: (() => void) | null;
+	/**
+	 * Internal: batches PTY output into one xterm.write per animation frame.
+	 * Agent CLIs emit repaints as many small chunks; per-chunk writes trigger
+	 * a parse/render cycle each and overwhelm the renderer (#2241, #2244).
+	 */
+	_writeCoalescer: WriteCoalescer | null;
 }
 
 const MAX_LOG_ENTRIES = 200;
@@ -163,6 +170,7 @@ export function createTransport(): TerminalTransport {
 		_livenessTimer: null,
 		_lastLivenessTick: 0,
 		_resumeListener: null,
+		_writeCoalescer: null,
 	};
 }
 
@@ -337,6 +345,12 @@ export function connect(
 	cancelReconnect(transport);
 	transport.currentUrl = wsUrl;
 	transport._terminal = terminal;
+	// Recreate per connect so the coalescer always targets the current
+	// terminal; dispose flushes anything the previous socket left pending.
+	transport._writeCoalescer?.dispose();
+	transport._writeCoalescer = createWriteCoalescer((data) =>
+		terminal.write(data),
+	);
 	transport._terminated = false;
 	setupLiveness(transport);
 	setConnectionState(transport, "connecting");
@@ -411,12 +425,13 @@ function attachSocketListeners(
 		// channel; renderer treats them identically). Pipe straight into
 		// xterm without any decoding step.
 		if (event.data instanceof ArrayBuffer) {
-			// Pipe PTY bytes straight into xterm. There's no output ACK back to
-			// host-service: back-pressure lives entirely on the host side, which
-			// bounds this socket's send buffer and drops us (we reconnect and
-			// replay) if we fall hopelessly behind. That means a slow/stalled
-			// renderer can never wedge the shell — it just loses some scrollback.
-			terminal.write(new Uint8Array(event.data));
+			// Queue PTY bytes; the coalescer batches them into one xterm.write
+			// per animation frame. There's no output ACK back to host-service:
+			// back-pressure lives entirely on the host side, which bounds this
+			// socket's send buffer and drops us (we reconnect and replay) if we
+			// fall hopelessly behind. That means a slow/stalled renderer can
+			// never wedge the shell — it just loses some scrollback.
+			transport._writeCoalescer?.push(new Uint8Array(event.data));
 			transport._hasReceivedBytes = true;
 			return;
 		}
@@ -425,6 +440,7 @@ function attachSocketListeners(
 		try {
 			message = JSON.parse(String(event.data)) as TerminalServerMessage;
 		} catch {
+			transport._writeCoalescer?.flushSync();
 			terminal.writeln("\r\n[terminal] invalid server payload");
 			return;
 		}
@@ -449,6 +465,7 @@ function attachSocketListeners(
 		}
 
 		if (message.type === "exit") {
+			transport._writeCoalescer?.flushSync();
 			transport._terminated = true;
 			cancelReconnect(transport);
 			terminal.writeln(
@@ -459,6 +476,9 @@ function attachSocketListeners(
 
 	socket.addEventListener("close", (event) => {
 		if (transport.socket !== socket) return;
+		// Render whatever arrived before the close instead of holding it for a
+		// frame that may never come (e.g. hidden window).
+		transport._writeCoalescer?.flushSync();
 		setConnectionState(transport, "closed");
 		transport.socket = null;
 		if (!transport._terminated && event.code !== 1000) {
@@ -500,6 +520,8 @@ export function disconnect(transport: TerminalTransport) {
 		transport.socket.close();
 		transport.socket = null;
 	}
+	transport._writeCoalescer?.dispose();
+	transport._writeCoalescer = null;
 	transport.currentUrl = null;
 	transport._terminal = null;
 	transport._reconnectAttempt = 0;
@@ -540,6 +562,8 @@ export function disposeTransport(transport: TerminalTransport) {
 		transport.socket.close();
 		transport.socket = null;
 	}
+	transport._writeCoalescer?.dispose();
+	transport._writeCoalescer = null;
 	transport.currentUrl = null;
 	transport._terminal = null;
 	transport._reconnectAttempt = 0;

apps/desktop/src/renderer/lib/terminal/write-coalescer.test.ts

@@ -0,0 +1,131 @@
+import { afterEach, beforeEach, describe, expect, test } from "bun:test";
+import { createWriteCoalescer, MAX_PENDING_BYTES } from "./write-coalescer";
+
+// Capture rAF callbacks so tests control frame timing deterministically.
+let frameCallbacks: Map<number, FrameRequestCallback>;
+let nextFrameId: number;
+
+const originalRaf = globalThis.requestAnimationFrame;
+const originalCancelRaf = globalThis.cancelAnimationFrame;
+
+function fireFrame() {
+	const callbacks = [...frameCallbacks.values()];
+	frameCallbacks.clear();
+	for (const callback of callbacks) {
+		callback(performance.now());
+	}
+}
+
+beforeEach(() => {
+	frameCallbacks = new Map();
+	nextFrameId = 1;
+	globalThis.requestAnimationFrame = (callback: FrameRequestCallback) => {
+		const id = nextFrameId++;
+		frameCallbacks.set(id, callback);
+		return id;
+	};
+	globalThis.cancelAnimationFrame = (id: number) => {
+		frameCallbacks.delete(id);
+	};
+});
+
+afterEach(() => {
+	globalThis.requestAnimationFrame = originalRaf;
+	globalThis.cancelAnimationFrame = originalCancelRaf;
+});
+
+function bytes(text: string): Uint8Array {
+	return new TextEncoder().encode(text);
+}
+
+describe("createWriteCoalescer", () => {
+	test("coalesces chunks arriving in the same frame into one write", () => {
+		const writes: Uint8Array[] = [];
+		const coalescer = createWriteCoalescer((data) => writes.push(data));
+
+		coalescer.push(bytes("foo"));
+		coalescer.push(bytes("bar"));
+		coalescer.push(bytes("baz"));
+		expect(writes).toHaveLength(0);
+
+		fireFrame();
+		expect(writes).toHaveLength(1);
+		expect(new TextDecoder().decode(writes[0])).toBe("foobarbaz");
+	});
+
+	test("schedules a new frame for data arriving after a flush", () => {
+		const writes: Uint8Array[] = [];
+		const coalescer = createWriteCoalescer((data) => writes.push(data));
+
+		coalescer.push(bytes("first"));
+		fireFrame();
+		coalescer.push(bytes("second"));
+		fireFrame();
+
+		expect(writes).toHaveLength(2);
+		expect(new TextDecoder().decode(writes[1])).toBe("second");
+	});
+
+	test("flushSync writes pending bytes immediately and cancels the scheduled frame", () => {
+		const writes: Uint8Array[] = [];
+		const coalescer = createWriteCoalescer((data) => writes.push(data));
+
+		coalescer.push(bytes("pending"));
+		coalescer.flushSync();
+		expect(writes).toHaveLength(1);
+		expect(new TextDecoder().decode(writes[0])).toBe("pending");
+
+		// The previously scheduled frame must not produce a second write.
+		fireFrame();
+		expect(writes).toHaveLength(1);
+	});
+
+	test("flushSync with nothing pending writes nothing", () => {
+		const writes: Uint8Array[] = [];
+		const coalescer = createWriteCoalescer((data) => writes.push(data));
+
+		coalescer.flushSync();
+		expect(writes).toHaveLength(0);
+	});
+
+	test("flushes immediately when pending bytes exceed the cap", () => {
+		const writes: Uint8Array[] = [];
+		const coalescer = createWriteCoalescer((data) => writes.push(data));
+
+		coalescer.push(new Uint8Array(MAX_PENDING_BYTES + 1));
+		expect(writes).toHaveLength(1);
+		expect(writes[0]).toHaveLength(MAX_PENDING_BYTES + 1);
+
+		// Nothing left for the frame to write.
+		fireFrame();
+		expect(writes).toHaveLength(1);
+	});
+
+	test("dispose flushes pending bytes and ignores later pushes", () => {
+		const writes: Uint8Array[] = [];
+		const coalescer = createWriteCoalescer((data) => writes.push(data));
+
+		coalescer.push(bytes("tail"));
+		coalescer.dispose();
+		expect(writes).toHaveLength(1);
+		expect(new TextDecoder().decode(writes[0])).toBe("tail");
+
+		coalescer.push(bytes("ignored"));
+		fireFrame();
+		expect(writes).toHaveLength(1);
+	});
+
+	test("preserves byte order across many small chunks", () => {
+		const writes: Uint8Array[] = [];
+		const coalescer = createWriteCoalescer((data) => writes.push(data));
+
+		const parts = Array.from({ length: 100 }, (_, i) => `${i},`);
+		for (const part of parts) {
+			coalescer.push(bytes(part));
+		}
+		fireFrame();
+
+		expect(writes).toHaveLength(1);
+		expect(new TextDecoder().decode(writes[0])).toBe(parts.join(""));
+	});
+});