diff --git a/config/reliability-gates.jsonc b/config/reliability-gates.jsonc index bd90a3370b8..a407e218aed 100644 --- a/config/reliability-gates.jsonc +++ b/config/reliability-gates.jsonc @@ -1516,7 +1516,7 @@ "id": "terminal-performance.daemon-stream-backpressure", "title": "Daemon terminal streams respect socket backpressure under output floods", "maturity": "experimental", - "protection": "none", + "protection": "partial", "owner": "terminal-performance", "layer": "daemon-provider-contract", "surfaces": [ @@ -1534,19 +1534,46 @@ "providers": [ "daemon" ], - "coveredPlatforms": [], + "coveredPlatforms": [ + "macos" + ], "coveredProviders": [], - "coverageNotes": "Registered gap on main. The daemon batcher write(false)/drain contracts exist only on the pending reliability stack. It registers here with its owning split PR.", + "coverageNotes": "Deterministic daemon socket write/drain contracts with a fake socket; local macOS evidence. Live hidden-output floods, active key latency artifacts, and the renderer ACK slice remain gaps (pending perf slice B).", "motivatingLinks": [ "https://github.com/stablyai/orca/pull/6836", "https://github.com/stablyai/orca/pull/6858" ], "invariant": "Daemon terminal output floods must pause when socket writes return false, resume on drain, keep buffered bytes bounded, and not starve focused input.", "oracle": "The current executable slice injects a slow socket and asserts write(false)/drain ordering, bounded queued daemon stream bytes, close/error listener cleanup, newest-tail preservation under sustained pressure, and flush-immediate cross-session priority ahead of unrelated background backlog on drain. Active input latency remains part of the broader perf gate.", - "commands": [], - "testFiles": [], - "assertionRefs": [], - "evidenceRuns": [], + "commands": [ + "pnpm exec vitest run --config config/vitest.config.ts src/main/daemon/daemon-stream-data-batcher.test.ts" + ], + "testFiles": [ + "src/main/daemon/daemon-stream-data-batcher.test.ts" + ], + "assertionRefs": [ + { + "file": "src/main/daemon/daemon-stream-data-batcher.test.ts", + "assertions": [ + "stream writes pause when the socket applies backpressure and resume on drain", + "later stream data queues behind a backpressured socket until drain", + "interactive session output is prioritized over unrelated backpressured backlog on drain", + "queued stream data stays bounded while a socket remains backpressured", + "global cleanup clears clients that only have backpressured stream writes" + ] + } + ], + "evidenceRuns": [ + { + "date": "2026-07-03", + "runner": "local", + "platform": "macos", + "command": "pnpm exec vitest run --config config/vitest.config.ts src/main/daemon/daemon-stream-data-batcher.test.ts", + "result": "passed", + "durationSeconds": 0.2, + "summary": "1 test file passed, 10 tests passed; write/drain pause, queue-until-drain, interactive drain priority, bounded queued bytes, and backpressure-only cleanup contracts." + } + ], "runtimeBudget": { "p95Seconds": 60, "scope": "daemon provider-contract gate" @@ -1569,9 +1596,8 @@ "Pair with the broader output-backpressure budget before blocking." ], "knownGaps": [ - "No executable coverage on main yet; the slice lives on the pending fix-terminal-reliability stack.", - "Current command is daemon batcher contract coverage, not live daemon/Electron perf.", - "Active input latency is not proven in live daemon/Electron perf." + "No live flood or latency metric artifact yet; deterministic byte-accounting only.", + "The renderer/main ACK and pending-cap extensions arrive with the second perf slice (pty.ts / pty-connection.ts hunks)." ], "demotionRule": "Cannot promote if backpressure is inferred only from renderer output success." }, @@ -1834,7 +1860,7 @@ "macos" ], "coveredProviders": [], - "coverageNotes": "Local macOS evidence over the runtime-RPC stream budgets on main@1282f5c2d. The pending stack adds byte-exact 512KB/2MB/256KB/48KB budget assertions; legacy JSON subscribe parity remains undecided.", + "coverageNotes": "Local macOS evidence over the runtime-RPC stream budgets on main@1282f5c2d. The 512KB mobile initial-snapshot budget is now byte-exact; remaining byte-exact 2MB/256KB/48KB assertions arrive with the pending stack; legacy JSON subscribe parity remains undecided.", "motivatingLinks": [ "https://github.com/stablyai/orca/pull/6951", "https://github.com/stablyai/orca/pull/6955", @@ -1856,7 +1882,8 @@ "assertions": [ "legacy binary output queued during initial snapshot serialization stays bounded", "stale mobile resize re-stream completions are dropped for legacy binary streams", - "aborted stream signals do not register stale listeners" + "aborted stream signals do not register stale listeners", + "mobile initial snapshots downgrade until they fit the 512KB byte budget" ] }, { @@ -1884,6 +1911,15 @@ "result": "passed", "durationSeconds": 3.8, "summary": "3 test file(s) passed, 28 tests passed on main@1282f5c2d in a clean checkout." + }, + { + "date": "2026-07-03", + "runner": "local", + "platform": "macos", + "command": "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/rpc/terminal-subscribe-buffer.test.ts src/main/runtime/rpc/terminal-output-batching.test.ts src/main/runtime/rpc/terminal-multiplex.test.ts", + "result": "passed", + "durationSeconds": 3.3, + "summary": "3 test files passed, 29 tests passed, including the new byte-exact 512KB mobile initial-snapshot budget test." } ], "runtimeBudget": { diff --git a/src/main/daemon/daemon-stream-backpressure-queue.ts b/src/main/daemon/daemon-stream-backpressure-queue.ts new file mode 100644 index 00000000000..d044f171cfa --- /dev/null +++ b/src/main/daemon/daemon-stream-backpressure-queue.ts @@ -0,0 +1,156 @@ +import type { Socket } from 'node:net' + +export type BackpressuredStreamLine = { + sessionId: string + line: string + priority?: boolean +} + +type BackpressuredStreamWrites = { + socket: Socket + lines: BackpressuredStreamLine[] + queuedBytes: number + onDrain: () => void + onClose: () => void +} + +export type WriteStreamLinesOptions = { + prioritizeBackpressuredSession?: boolean +} + +export class BackpressuredStreamWriteQueue { + private byClient = new Map() + private maxQueuedBytes: number + + constructor(maxQueuedBytes: number) { + this.maxQueuedBytes = Math.max(1, maxQueuedBytes) + } + + clientIds(): Iterable { + return this.byClient.keys() + } + + write( + clientId: string, + streamSocket: Socket, + lines: BackpressuredStreamLine[], + opts: WriteStreamLinesOptions = {} + ): void { + const existing = this.byClient.get(clientId) + if (existing) { + if (existing.socket === streamSocket && !streamSocket.destroyed) { + this.appendLines(existing, lines, opts) + return + } + this.clear(clientId) + } + + for (let index = 0; index < lines.length; index += 1) { + // Why: only Node's explicit `false` return signals backpressure; duck-typed + // sockets in tests or wrappers may return undefined and must keep the + // legacy direct-write path. + if (streamSocket.write(lines[index].line) === false) { + this.deferUntilDrain(clientId, streamSocket, lines.slice(index + 1)) + return + } + } + } + + clear(clientId: string): void { + const pending = this.byClient.get(clientId) + if (!pending) { + return + } + this.byClient.delete(clientId) + pending.socket.off('drain', pending.onDrain) + pending.socket.off('close', pending.onClose) + pending.socket.off('error', pending.onClose) + } + + private deferUntilDrain( + clientId: string, + streamSocket: Socket, + lines: BackpressuredStreamLine[] + ): void { + const onDrain = (): void => { + const pending = this.byClient.get(clientId) + if (!pending || pending.socket !== streamSocket) { + return + } + this.byClient.delete(clientId) + streamSocket.off('close', pending.onClose) + streamSocket.off('error', pending.onClose) + if (!streamSocket.destroyed) { + this.write(clientId, streamSocket, pending.lines) + } + } + const onClose = (): void => { + this.clear(clientId) + } + + const pending: BackpressuredStreamWrites = { + socket: streamSocket, + lines: [], + queuedBytes: 0, + onDrain, + onClose + } + this.appendLines(pending, lines) + this.byClient.set(clientId, pending) + streamSocket.once('drain', onDrain) + streamSocket.once('close', onClose) + streamSocket.once('error', onClose) + } + + private appendLines( + pending: BackpressuredStreamWrites, + lines: BackpressuredStreamLine[], + opts: WriteStreamLinesOptions = {} + ): void { + if (opts.prioritizeBackpressuredSession && lines.length > 0) { + // Why: input-triggered daemon output should not sit behind an unrelated + // hidden-output flood once the socket drains, but each session's bytes + // must still keep their own order. + const sessionId = lines[0].sessionId + let lastSameSessionIndex = -1 + for (let index = pending.lines.length - 1; index >= 0; index -= 1) { + if (pending.lines[index].sessionId === sessionId) { + lastSameSessionIndex = index + break + } + } + if (lastSameSessionIndex >= 0) { + pending.lines.splice(lastSameSessionIndex + 1, 0, ...lines) + } else { + const firstNonPriorityIndex = pending.lines.findIndex((line) => !line.priority) + pending.lines.splice( + firstNonPriorityIndex === -1 ? pending.lines.length : firstNonPriorityIndex, + 0, + ...lines + ) + } + } else { + pending.lines.push(...lines) + } + pending.queuedBytes += this.measureLinesBytes(lines) + // Why: a wedged stream socket can otherwise retain unbounded hidden-output + // floods below the renderer ACK budget. Preserve the newest bounded tail. + while (pending.queuedBytes > this.maxQueuedBytes && pending.lines.length > 1) { + const dropIndex = pending.lines.findIndex((line) => !line.priority) + const [dropped] = pending.lines.splice(dropIndex === -1 ? 0 : dropIndex, 1) + pending.queuedBytes -= dropped ? this.measureLineBytes(dropped) : 0 + } + } + + private measureLinesBytes(lines: BackpressuredStreamLine[]): number { + let bytes = 0 + for (const line of lines) { + bytes += this.measureLineBytes(line) + } + return bytes + } + + private measureLineBytes(line: BackpressuredStreamLine): number { + return Buffer.byteLength(line.line, 'utf8') + } +} diff --git a/src/main/daemon/daemon-stream-data-batcher.test.ts b/src/main/daemon/daemon-stream-data-batcher.test.ts index 8777434e1d9..08f00677419 100644 --- a/src/main/daemon/daemon-stream-data-batcher.test.ts +++ b/src/main/daemon/daemon-stream-data-batcher.test.ts @@ -1,13 +1,14 @@ import { describe, expect, it, vi } from 'vitest' import type { Socket } from 'node:net' +import { EventEmitter } from 'node:events' import { DaemonStreamDataBatcher } from './daemon-stream-data-batcher' import { createNdjsonParser } from './ndjson' function createBatcher(options?: ConstructorParameters[1]) { - const streamSocket = { + const streamSocket = Object.assign(new EventEmitter(), { destroyed: false, - write: vi.fn() - } as unknown as Socket & { write: ReturnType } + write: vi.fn(() => true) + }) as unknown as Socket & { write: ReturnType } const batcher = new DaemonStreamDataBatcher(() => ({ streamSocket }), options) return { batcher, streamSocket } } @@ -129,4 +130,143 @@ describe('DaemonStreamDataBatcher', () => { vi.useRealTimers() } }) + + it('pauses stream writes when the socket applies backpressure', () => { + vi.useFakeTimers() + try { + const maxLineBytes = 256 + const { batcher, streamSocket } = createBatcher({ maxLineBytes }) + streamSocket.write.mockReturnValueOnce(false).mockReturnValue(true) + + batcher.enqueue('client-1', 'session-1', 'x'.repeat(maxLineBytes * 3)) + vi.advanceTimersByTime(8) + + expect(streamSocket.write).toHaveBeenCalledTimes(1) + + streamSocket.emit('drain') + + expect(streamSocket.write.mock.calls.length).toBeGreaterThan(1) + } finally { + vi.useRealTimers() + } + }) + + it('queues later stream data behind a backpressured socket until drain', () => { + vi.useFakeTimers() + try { + const maxLineBytes = 256 + const { batcher, streamSocket } = createBatcher({ maxLineBytes }) + streamSocket.write.mockReturnValueOnce(false).mockReturnValue(true) + + batcher.enqueue('client-1', 'session-1', 'x'.repeat(maxLineBytes * 3)) + vi.advanceTimersByTime(8) + batcher.enqueue('client-1', 'session-2', 'interactive', { + flushImmediately: true, + flushMaxChars: 1024 + }) + + expect(streamSocket.write).toHaveBeenCalledTimes(1) + + streamSocket.emit('drain') + + expect( + streamSocket.write.mock.calls.some(([line]) => + String(line).includes('"sessionId":"session-2"') + ) + ).toBe(true) + } finally { + vi.useRealTimers() + } + }) + + it('globally clears clients that only have backpressured stream writes', () => { + vi.useFakeTimers() + try { + const maxLineBytes = 256 + const { batcher, streamSocket } = createBatcher({ maxLineBytes }) + streamSocket.write.mockReturnValueOnce(false).mockReturnValue(true) + + batcher.enqueue('client-1', 'session-1', 'x'.repeat(maxLineBytes * 3)) + vi.advanceTimersByTime(8) + + expect(streamSocket.write).toHaveBeenCalledTimes(1) + expect(streamSocket.listenerCount('drain')).toBe(1) + + batcher.clear() + expect(streamSocket.listenerCount('drain')).toBe(0) + expect(streamSocket.listenerCount('close')).toBe(0) + expect(streamSocket.listenerCount('error')).toBe(0) + + streamSocket.write.mockClear() + streamSocket.emit('drain') + + expect(streamSocket.write).not.toHaveBeenCalled() + } finally { + vi.useRealTimers() + } + }) + + it('prioritizes interactive session output over unrelated backpressured backlog on drain', () => { + vi.useFakeTimers() + try { + const maxLineBytes = 256 + const { batcher, streamSocket } = createBatcher({ maxLineBytes }) + streamSocket.write.mockReturnValueOnce(false).mockReturnValue(true) + + batcher.enqueue('client-1', 'session-background', 'x'.repeat(maxLineBytes * 3)) + vi.advanceTimersByTime(8) + batcher.enqueue('client-1', 'session-interactive', 'prompt-redraw', { + flushImmediately: true, + flushMaxChars: 1024 + }) + + expect(streamSocket.write).toHaveBeenCalledTimes(1) + streamSocket.write.mockClear() + + streamSocket.emit('drain') + + expect(String(streamSocket.write.mock.calls[0]?.[0])).toContain( + '"sessionId":"session-interactive"' + ) + expect(String(streamSocket.write.mock.calls[0]?.[0])).toContain('"data":"prompt-redraw"') + expect( + streamSocket.write.mock.calls.some(([line]) => + String(line).includes('"sessionId":"session-background"') + ) + ).toBe(true) + } finally { + vi.useRealTimers() + } + }) + + it('bounds queued stream data while a socket remains backpressured', () => { + vi.useFakeTimers() + try { + const { batcher, streamSocket } = createBatcher({ + maxLineBytes: 256, + maxBackpressuredBytes: 1 + }) + streamSocket.write.mockReturnValueOnce(false).mockReturnValue(true) + + batcher.enqueue('client-1', 'session-old', 'old-output'.repeat(200)) + vi.advanceTimersByTime(8) + + for (let index = 0; index < 5; index += 1) { + batcher.enqueue('client-1', `session-new-${index}`, `new-output-${index}`, { + flushImmediately: true, + flushMaxChars: 1024 + }) + } + + expect(streamSocket.write).toHaveBeenCalledTimes(1) + streamSocket.write.mockClear() + + streamSocket.emit('drain') + + expect(streamSocket.write).toHaveBeenCalledTimes(1) + expect(String(streamSocket.write.mock.calls[0]?.[0])).toContain('"sessionId":"session-new-4"') + } finally { + vi.useRealTimers() + } + }) }) diff --git a/src/main/daemon/daemon-stream-data-batcher.ts b/src/main/daemon/daemon-stream-data-batcher.ts index 3374222f17f..596b61128e8 100644 --- a/src/main/daemon/daemon-stream-data-batcher.ts +++ b/src/main/daemon/daemon-stream-data-batcher.ts @@ -1,5 +1,7 @@ import type { Socket } from 'node:net' -import { encodeNdjson, NDJSON_MAX_LINE_BYTES } from './ndjson' +import { NDJSON_MAX_LINE_BYTES } from './ndjson' +import { BackpressuredStreamWriteQueue } from './daemon-stream-backpressure-queue' +import { encodeStreamDataEvent, splitStreamDataForNdjson } from './daemon-stream-ndjson-frames' type StreamDataClient = { streamSocket: Socket | null @@ -14,6 +16,7 @@ type PendingStreamDataBatch = { // Why: match main-process PTY IPC batching to avoid adding latency while // removing daemon socket writes and JSON framing during bursty output. const STREAM_DATA_BATCH_INTERVAL_MS = 8 +const DEFAULT_MAX_BACKPRESSURED_STREAM_BYTES = 8 * 1024 * 1024 type EnqueueOptions = { flushImmediately?: boolean @@ -22,88 +25,12 @@ type EnqueueOptions = { type DaemonStreamDataBatcherOptions = { maxLineBytes?: number -} - -function encodeStreamDataEvent(sessionId: string, data: string): string { - return encodeNdjson({ - type: 'event', - event: 'data', - sessionId, - payload: { data } - }) -} - -function streamDataEventLineBytes(sessionId: string, data: string): number { - return Buffer.byteLength(encodeStreamDataEvent(sessionId, data), 'utf8') -} - -function isHighSurrogate(value: number): boolean { - return value >= 0xd800 && value <= 0xdbff -} - -function isLowSurrogate(value: number): boolean { - return value >= 0xdc00 && value <= 0xdfff -} - -function clampToSafeSplitIndex(value: string, start: number, end: number): number { - if (end <= start || end >= value.length) { - return end - } - const prev = value.charCodeAt(end - 1) - const next = value.charCodeAt(end) - return isHighSurrogate(prev) && isLowSurrogate(next) ? end - 1 : end -} - -function nextSafeSplitIndex(value: string, start: number): number { - const next = Math.min(value.length, start + 1) - if ( - next < value.length && - isHighSurrogate(value.charCodeAt(start)) && - isLowSurrogate(value.charCodeAt(next)) - ) { - return next + 1 - } - return next -} - -function splitStreamDataForNdjson(sessionId: string, data: string, maxLineBytes: number): string[] { - if (streamDataEventLineBytes(sessionId, data) <= maxLineBytes) { - return [data] - } - - const chunks: string[] = [] - let start = 0 - while (start < data.length) { - let low = start + 1 - let high = data.length - let best = start - - while (low <= high) { - const rawMid = Math.floor((low + high) / 2) - const mid = clampToSafeSplitIndex(data, start, rawMid) - if (mid <= start) { - low = rawMid + 1 - continue - } - - if (streamDataEventLineBytes(sessionId, data.slice(start, mid)) <= maxLineBytes) { - best = mid - low = rawMid + 1 - } else { - high = rawMid - 1 - } - } - - const end = best > start ? best : nextSafeSplitIndex(data, start) - chunks.push(data.slice(start, end)) - start = end - } - - return chunks + maxBackpressuredBytes?: number } export class DaemonStreamDataBatcher { private pendingByClient = new Map() + private backpressureQueue: BackpressuredStreamWriteQueue private getClient: (clientId: string) => StreamDataClient | undefined private maxLineBytes: number @@ -113,6 +40,9 @@ export class DaemonStreamDataBatcher { ) { this.getClient = getClient this.maxLineBytes = Math.max(1, options.maxLineBytes ?? NDJSON_MAX_LINE_BYTES) + this.backpressureQueue = new BackpressuredStreamWriteQueue( + options.maxBackpressuredBytes ?? DEFAULT_MAX_BACKPRESSURED_STREAM_BYTES + ) } enqueue(clientId: string, sessionId: string, data: string, options: EnqueueOptions = {}): void { @@ -166,7 +96,7 @@ export class DaemonStreamDataBatcher { } for (const entry of batch.queue) { - this.writeStreamDataEvent(client.streamSocket, entry.sessionId, entry.data) + this.writeStreamDataEvent(clientId, client.streamSocket, entry.sessionId, entry.data) } } @@ -217,30 +147,43 @@ export class DaemonStreamDataBatcher { } for (const entry of flushed) { - this.writeStreamDataEvent(client.streamSocket, entry.sessionId, entry.data) + this.writeStreamDataEvent(clientId, client.streamSocket, entry.sessionId, entry.data, { + prioritizeBackpressuredSession: true + }) } } clear(clientId?: string): void { - const batches = + const clientIds = clientId === undefined - ? Array.from(this.pendingByClient.entries()) - : [[clientId, this.pendingByClient.get(clientId)] as const] + ? new Set([...this.pendingByClient.keys(), ...this.backpressureQueue.clientIds()]) + : new Set([clientId]) - for (const [id, batch] of batches) { + for (const id of clientIds) { + const batch = this.pendingByClient.get(id) if (batch?.timer) { clearTimeout(batch.timer) } this.pendingByClient.delete(id) + this.backpressureQueue.clear(id) } } - private writeStreamDataEvent(streamSocket: Socket, sessionId: string, data: string): void { + private writeStreamDataEvent( + clientId: string, + streamSocket: Socket, + sessionId: string, + data: string, + opts: { prioritizeBackpressuredSession?: boolean } = {} + ): void { // Why: createNdjsonParser rejects oversized lines. Terminal output can // burst faster than the batch interval, so writer-side chunking prevents // the daemon from dropping its own stream events at the receiver. - for (const chunk of splitStreamDataForNdjson(sessionId, data, this.maxLineBytes)) { - streamSocket.write(encodeStreamDataEvent(sessionId, chunk)) - } + const lines = splitStreamDataForNdjson(sessionId, data, this.maxLineBytes).map((chunk) => ({ + sessionId, + line: encodeStreamDataEvent(sessionId, chunk), + ...(opts.prioritizeBackpressuredSession ? { priority: true } : {}) + })) + this.backpressureQueue.write(clientId, streamSocket, lines, opts) } } diff --git a/src/main/daemon/daemon-stream-ndjson-frames.ts b/src/main/daemon/daemon-stream-ndjson-frames.ts new file mode 100644 index 00000000000..9cc9f8dd044 --- /dev/null +++ b/src/main/daemon/daemon-stream-ndjson-frames.ts @@ -0,0 +1,85 @@ +import { encodeNdjson } from './ndjson' + +// NDJSON framing for daemon stream-data events: encoding, byte measurement, and +// surrogate-safe splitting so oversized chunks never break a UTF-16 pair mid-frame. +export function encodeStreamDataEvent(sessionId: string, data: string): string { + return encodeNdjson({ + type: 'event', + event: 'data', + sessionId, + payload: { data } + }) +} + +export function streamDataEventLineBytes(sessionId: string, data: string): number { + return Buffer.byteLength(encodeStreamDataEvent(sessionId, data), 'utf8') +} + +function isHighSurrogate(value: number): boolean { + return value >= 0xd800 && value <= 0xdbff +} + +function isLowSurrogate(value: number): boolean { + return value >= 0xdc00 && value <= 0xdfff +} + +function clampToSafeSplitIndex(value: string, start: number, end: number): number { + if (end <= start || end >= value.length) { + return end + } + const prev = value.charCodeAt(end - 1) + const next = value.charCodeAt(end) + return isHighSurrogate(prev) && isLowSurrogate(next) ? end - 1 : end +} + +function nextSafeSplitIndex(value: string, start: number): number { + const next = Math.min(value.length, start + 1) + if ( + next < value.length && + isHighSurrogate(value.charCodeAt(start)) && + isLowSurrogate(value.charCodeAt(next)) + ) { + return next + 1 + } + return next +} + +export function splitStreamDataForNdjson( + sessionId: string, + data: string, + maxLineBytes: number +): string[] { + if (streamDataEventLineBytes(sessionId, data) <= maxLineBytes) { + return [data] + } + + const chunks: string[] = [] + let start = 0 + while (start < data.length) { + let low = start + 1 + let high = data.length + let best = start + + while (low <= high) { + const rawMid = Math.floor((low + high) / 2) + const mid = clampToSafeSplitIndex(data, start, rawMid) + if (mid <= start) { + low = rawMid + 1 + continue + } + + if (streamDataEventLineBytes(sessionId, data.slice(start, mid)) <= maxLineBytes) { + best = mid + low = rawMid + 1 + } else { + high = rawMid - 1 + } + } + + const end = best > start ? best : nextSafeSplitIndex(data, start) + chunks.push(data.slice(start, end)) + start = end + } + + return chunks +} diff --git a/src/main/runtime/rpc/terminal-output-batching.test.ts b/src/main/runtime/rpc/terminal-output-batching.test.ts index 2df3dbc7173..73068cdd598 100644 --- a/src/main/runtime/rpc/terminal-output-batching.test.ts +++ b/src/main/runtime/rpc/terminal-output-batching.test.ts @@ -246,6 +246,9 @@ describe('terminal output batching', () => { .map((frame) => decodeTerminalStreamFrame(frame)) .filter((frame) => frame?.opcode === TerminalStreamOpcode.Output) expect(outputFrames.length).toBeGreaterThan(1) + for (const frame of outputFrames) { + expect(frame?.payload.byteLength ?? 0).toBeLessThanOrEqual(48 * 1024) + } expect(firstOutputEncodeCount).toBe(1) expect( outputFrames.map((frame) => (frame ? decodeTerminalStreamText(frame.payload) : '')).join('') diff --git a/src/main/runtime/rpc/terminal-subscribe-buffer.test.ts b/src/main/runtime/rpc/terminal-subscribe-buffer.test.ts index dfd185b33b4..4c4c101a736 100644 --- a/src/main/runtime/rpc/terminal-subscribe-buffer.test.ts +++ b/src/main/runtime/rpc/terminal-subscribe-buffer.test.ts @@ -430,6 +430,88 @@ describe('terminal subscribe buffering', () => { } }) + it('downgrades mobile initial snapshots until they fit the 512KB byte budget', async () => { + const messages: string[] = [] + const binaryFrames: Uint8Array[] = [] + const cleanups = new Map void>() + const oversizedSnapshot = 'x'.repeat(600 * 1024) + const budgetedSnapshot = 'tail'.repeat(40 * 1024) + const runtime = stubRuntime({ + resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }), + readTerminal: vi.fn().mockResolvedValue({ tail: [], truncated: false }), + serializeTerminalBuffer: vi.fn(async (_ptyId: string, opts?: { scrollbackRows?: number }) => { + if ((opts?.scrollbackRows ?? 0) > 250) { + return { data: oversizedSnapshot, cols: 100, rows: 30 } + } + return { data: budgetedSnapshot, cols: 100, rows: 30, seq: 42 } + }), + getTerminalSize: vi.fn().mockReturnValue({ cols: 80, rows: 24 }), + getMobileDisplayMode: vi.fn().mockReturnValue('mobile-fit'), + getLayout: vi.fn().mockReturnValue({ seq: 1 }), + isTerminalAlternateScreen: vi.fn().mockReturnValue(false), + handleMobileSubscribe: vi.fn().mockResolvedValue(undefined), + handleMobileUnsubscribe: vi.fn(), + subscribeToTerminalData: vi.fn().mockReturnValue(vi.fn()), + subscribeToTerminalResize: vi.fn().mockReturnValue(vi.fn()), + subscribeToFitOverrideChanges: vi.fn().mockReturnValue(vi.fn()), + registerSubscriptionCleanup: vi.fn((id: string, cleanup: () => void) => { + cleanups.set(id, cleanup) + }), + cleanupSubscription: vi.fn((id: string) => { + const cleanup = cleanups.get(id) + cleanups.delete(id) + cleanup?.() + }), + waitForTerminal: vi.fn(() => new Promise(() => {})), + sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), + updateMobileViewport: vi.fn().mockResolvedValue(false) + }) + const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + + const dispatchPromise = dispatcher.dispatchStreaming( + makeRequest('terminal.subscribe', { + terminal: 'terminal-1', + client: { id: 'phone-1', type: 'mobile' }, + capabilities: { terminalBinaryStream: 1 } + }), + (msg) => messages.push(msg), + { + connectionId: 'conn-mobile-budget', + sendBinary: (bytes) => binaryFrames.push(bytes) + } + ) + + await vi.waitFor(() => + expect(messages.some((msg) => JSON.parse(msg).result?.type === 'subscribed')).toBe(true) + ) + + expect(runtime.serializeTerminalBuffer).toHaveBeenNthCalledWith(1, 'pty-1', { + scrollbackRows: 1000 + }) + expect(runtime.serializeTerminalBuffer).toHaveBeenNthCalledWith(2, 'pty-1', { + scrollbackRows: 500 + }) + expect(runtime.serializeTerminalBuffer).toHaveBeenNthCalledWith(3, 'pty-1', { + scrollbackRows: 250 + }) + const decodedFrames = binaryFrames.map((frame) => decodeTerminalStreamFrame(frame)) + const snapshotStart = decodedFrames.find( + (frame) => frame?.opcode === TerminalStreamOpcode.SnapshotStart + ) + expect(snapshotStart && decodeTerminalStreamJson(snapshotStart.payload)).toMatchObject({ + truncatedByByteBudget: true + }) + const snapshotData = decodedFrames + .filter((frame) => frame?.opcode === TerminalStreamOpcode.SnapshotChunk) + .map((frame) => (frame ? decodeTerminalStreamText(frame.payload) : '')) + .join('') + expect(snapshotData).toBe(budgetedSnapshot) + expect(Buffer.byteLength(snapshotData, 'utf8')).toBeLessThanOrEqual(512 * 1024) + + runtime.cleanupSubscription('terminal-1:phone-1') + await dispatchPromise + }) + it('drops stale mobile resize re-stream completions for legacy binary streams', async () => { const messages: string[] = [] const binaryFrames: Uint8Array[] = []