diff --git a/src/main/runtime/rpc/methods/terminal.ts b/src/main/runtime/rpc/methods/terminal.ts index 10f775bf024..5e870b8f509 100644 --- a/src/main/runtime/rpc/methods/terminal.ts +++ b/src/main/runtime/rpc/methods/terminal.ts @@ -25,7 +25,8 @@ const MOBILE_SNAPSHOT_BYTE_BUDGET = 512 * 1024 const REQUESTED_SNAPSHOT_BYTE_BUDGET = 2 * 1024 * 1024 const TERMINAL_STREAM_CHUNK_BYTES = 48 * 1024 const TERMINAL_OUTPUT_FLUSH_MS = 5 -const TERMINAL_OUTPUT_BATCH_MAX_CHARS = 64 * 1024 +// Why: output batches become binary stream payloads; byte size is the transport cost. +const TERMINAL_OUTPUT_BATCH_MAX_BYTES = 64 * 1024 // Why: pending output is held for later binary frames, so cap the encoded // payload bytes rather than UTF-16 code units. const TERMINAL_MULTIPLEX_PENDING_MAX_BYTES = 256 * 1024 @@ -92,7 +93,7 @@ function createTerminalOutputBatcher( dispose: () => void } { let chunks: string[] = [] - let chars = 0 + let bytes = 0 let lastSeq: number | undefined let timer: ReturnType | null = null @@ -112,7 +113,7 @@ function createTerminalOutputBatcher( const data = chunks.length === 1 ? chunks[0]! : chunks.join('') const meta = typeof lastSeq === 'number' ? { seq: lastSeq, rawLength: data.length } : undefined chunks = [] - chars = 0 + bytes = 0 lastSeq = undefined onFlush(data, meta) } @@ -123,11 +124,11 @@ function createTerminalOutputBatcher( return } chunks.push(data) - chars += data.length + bytes += terminalStreamByteLength(data) if (typeof meta?.seq === 'number') { lastSeq = meta.seq } - if (chars >= TERMINAL_OUTPUT_BATCH_MAX_CHARS) { + if (bytes >= TERMINAL_OUTPUT_BATCH_MAX_BYTES) { flush() return } @@ -144,7 +145,7 @@ function createTerminalOutputBatcher( dispose(): void { clearTimer() chunks = [] - chars = 0 + bytes = 0 } } } diff --git a/src/main/runtime/rpc/terminal-multiplex.test.ts b/src/main/runtime/rpc/terminal-multiplex.test.ts index 36930b4c5f8..b5581e0aaca 100644 --- a/src/main/runtime/rpc/terminal-multiplex.test.ts +++ b/src/main/runtime/rpc/terminal-multiplex.test.ts @@ -232,6 +232,100 @@ describe('terminal multiplex RPC', () => { } }) + it('flushes multibyte live output when encoded bytes reach the batch budget', async () => { + vi.useFakeTimers() + try { + const messages: string[] = [] + const binaryFrames: Uint8Array[] = [] + const handlers = new Map< + number, + (frame: NonNullable>) => void + >() + const cleanups = new Map void>() + const dataListenerRef: { current?: (data: string) => void } = {} + const runtime = stubRuntime({ + resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }), + readTerminal: vi.fn().mockResolvedValue({ tail: [], truncated: false }), + serializeTerminalBuffer: vi.fn().mockResolvedValue({ + data: 'snapshot', + cols: 120, + rows: 40 + }), + getTerminalSize: vi.fn().mockReturnValue({ cols: 120, rows: 40 }), + getMobileDisplayMode: vi.fn().mockReturnValue('auto'), + getLayout: vi.fn().mockReturnValue({ seq: 1 }), + subscribeToTerminalData: vi.fn((_: string, listener: (data: string) => void) => { + dataListenerRef.current = listener + return vi.fn() + }), + subscribeToTerminalResize: vi.fn().mockReturnValue(vi.fn()), + subscribeToFitOverrideChanges: vi.fn().mockReturnValue(vi.fn()), + subscribeToDriverChanges: vi.fn().mockReturnValue(vi.fn()), + getTerminalFitOverride: vi.fn().mockReturnValue(null), + getDriver: vi.fn().mockReturnValue({ kind: 'idle' }), + registerSubscriptionCleanup: vi.fn((id: string, cleanup: () => void) => { + cleanups.set(id, cleanup) + }), + waitForTerminal: vi.fn(() => new Promise(() => {})), + sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), + updateDesktopViewport: vi.fn().mockResolvedValue(true) + }) + const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + + const dispatchPromise = dispatcher.dispatchStreaming( + makeRequest('terminal.multiplex', {}), + (msg) => messages.push(msg), + { + connectionId: 'conn-multibyte-output-batch', + sendBinary: (bytes) => binaryFrames.push(bytes), + registerBinaryStreamHandler: (streamId, handler) => { + handlers.set(streamId, handler) + return () => handlers.delete(streamId) + } + } + ) + + await vi.waitFor(() => + expect(messages.some((msg) => JSON.parse(msg).result?.type === 'ready')).toBe(true) + ) + handlers.get(0)?.( + decodeTerminalStreamFrame( + encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.Subscribe, + streamId: 0, + seq: 1, + payload: encodeTerminalStreamJson({ + streamId: 6, + terminal: 'terminal-1', + client: { id: 'desktop-1', type: 'desktop' }, + viewport: { cols: 120, rows: 40 } + }) + }) + )! + ) + await vi.waitFor(() => + expect(messages.some((msg) => JSON.parse(msg).result?.type === 'subscribed')).toBe(true) + ) + binaryFrames.splice(0) + + const multibyteOutput = '界'.repeat(22_000) + dataListenerRef.current?.(multibyteOutput) + + const outputFrames = binaryFrames + .map((frame) => decodeTerminalStreamFrame(frame)) + .filter((frame) => frame?.opcode === TerminalStreamOpcode.Output) + expect(outputFrames).toHaveLength(1) + expect(outputFrames[0] ? decodeTerminalStreamText(outputFrames[0].payload) : '').toBe( + multibyteOutput + ) + + cleanups.get('terminal-multiplex:conn-multibyte-output-batch')?.() + await dispatchPromise + } finally { + vi.useRealTimers() + } + }) + it('marks multiplex fallback snapshots truncated when the uncursored read is limited', async () => { const messages: string[] = [] const binaryFrames: Uint8Array[] = []