diff --git a/src/main/ipc/pty.test.ts b/src/main/ipc/pty.test.ts index 2da7757f0..52f72ee7d 100644 --- a/src/main/ipc/pty.test.ts +++ b/src/main/ipc/pty.test.ts @@ -381,6 +381,17 @@ describe('registerPtyHandlers', () => { return writeCall[1] as (event: unknown, args: { id: string; data: string }) => void } + function getPtyAckDataListener(): ( + event: unknown, + args: { id: string; charCount: number } + ) => void { + const ackCall = onMock.mock.calls.find((call: unknown[]) => call[0] === 'pty:ackData') + if (!ackCall) { + throw new Error('missing pty:ackData listener') + } + return ackCall[1] as (event: unknown, args: { id: string; charCount: number }) => void + } + /** Helper: trigger pty:spawn and return the env passed to node-pty. */ async function spawnAndGetEnv( argsEnv?: Record, @@ -4168,6 +4179,147 @@ describe('registerPtyHandlers', () => { } }) + it('waits for renderer ACKs before sending more output for a saturated PTY', async () => { + vi.useFakeTimers() + const firstProc = createMockProc() + const secondProc = createMockProc() + spawnMock.mockReturnValueOnce(firstProc.proc).mockReturnValueOnce(secondProc.proc) + + try { + registerPtyHandlers(mainWindow as never) + const firstSpawn = (await handlers.get('pty:spawn')!(null, { + cols: 80, + rows: 24, + cwd: '/tmp' + })) as { id: string } + const secondSpawn = (await handlers.get('pty:spawn')!(null, { + cols: 80, + rows: 24, + cwd: '/tmp' + })) as { id: string } + const ackData = getPtyAckDataListener() + mainWindow.webContents.send.mockClear() + + firstProc.emitData('x'.repeat(600 * 1024)) + vi.advanceTimersByTime(8) + for (let index = 0; index < 31; index++) { + vi.advanceTimersByTime(1) + } + + expect(mainWindow.webContents.send).toHaveBeenCalledTimes(32) + vi.advanceTimersByTime(1) + expect(mainWindow.webContents.send).toHaveBeenCalledTimes(32) + expect(vi.getTimerCount()).toBe(0) + + secondProc.emitData('second-terminal-output') + vi.advanceTimersByTime(8) + + expect(mainWindow.webContents.send).toHaveBeenCalledTimes(33) + expect(mainWindow.webContents.send).toHaveBeenNthCalledWith(33, 'pty:data', { + id: secondSpawn.id, + data: 'second-terminal-output' + }) + + ackData(null, { id: firstSpawn.id, charCount: 16 * 1024 }) + vi.advanceTimersByTime(1) + + expect(mainWindow.webContents.send).toHaveBeenCalledTimes(34) + expect(mainWindow.webContents.send).toHaveBeenNthCalledWith(34, 'pty:data', { + id: firstSpawn.id, + data: 'x'.repeat(16 * 1024) + }) + } finally { + vi.useRealTimers() + } + }) + + it('keeps interactive output buffered when the renderer budget is saturated', async () => { + vi.useFakeTimers() + const bulkProcs = Array.from({ length: 16 }, () => createMockProc()) + const interactiveProc = createMockProc() + for (const proc of [...bulkProcs, interactiveProc]) { + spawnMock.mockReturnValueOnce(proc.proc) + } + + try { + registerPtyHandlers(mainWindow as never) + for (const _proc of bulkProcs) { + await handlers.get('pty:spawn')!(null, { + cols: 80, + rows: 24, + cwd: '/tmp' + }) + } + const interactiveSpawn = (await handlers.get('pty:spawn')!(null, { + cols: 80, + rows: 24, + cwd: '/tmp' + })) as { id: string } + const writeListener = getPtyWriteListener() + mainWindow.webContents.send.mockClear() + + for (const proc of bulkProcs) { + proc.emitData('x'.repeat(600 * 1024)) + } + vi.advanceTimersByTime(8) + for (let index = 0; index < 400; index++) { + vi.advanceTimersByTime(1) + } + expect(mainWindow.webContents.send).toHaveBeenCalledTimes(512) + expect(vi.getTimerCount()).toBe(0) + + writeListener(null, { + id: interactiveSpawn.id, + data: 'a' + }) + interactiveProc.emitData('\x1b[20;2Hredraw') + + expect(mainWindow.webContents.send).toHaveBeenCalledTimes(512) + } finally { + vi.useRealTimers() + } + }) + + it('caps total renderer in-flight output across many PTYs', async () => { + vi.useFakeTimers() + const procs = Array.from({ length: 17 }, () => createMockProc()) + for (const proc of procs) { + spawnMock.mockReturnValueOnce(proc.proc) + } + + try { + registerPtyHandlers(mainWindow as never) + const spawns: { id: string }[] = [] + for (const _proc of procs) { + spawns.push( + (await handlers.get('pty:spawn')!(null, { + cols: 80, + rows: 24, + cwd: '/tmp' + })) as { id: string } + ) + } + const ackData = getPtyAckDataListener() + mainWindow.webContents.send.mockClear() + + for (const proc of procs) { + proc.emitData('x'.repeat(600 * 1024)) + } + vi.advanceTimersByTime(8) + for (let index = 0; index < 400; index++) { + vi.advanceTimersByTime(1) + } + + expect(mainWindow.webContents.send).toHaveBeenCalledTimes(512) + ackData(null, { id: spawns[0].id, charCount: 16 * 1024 }) + vi.advanceTimersByTime(1) + + expect(mainWindow.webContents.send).toHaveBeenCalledTimes(513) + } finally { + vi.useRealTimers() + } + }) + it('batches stale PTY output after the interactive window expires', async () => { vi.useFakeTimers() const mockProc = createMockProc() diff --git a/src/main/ipc/pty.ts b/src/main/ipc/pty.ts index 38bee37b4..6c96b45ed 100644 --- a/src/main/ipc/pty.ts +++ b/src/main/ipc/pty.ts @@ -936,6 +936,7 @@ export function registerPtyHandlers( ipcMain.removeHandler('pty:writeAccepted') ipcMain.removeAllListeners('pty:write') ipcMain.removeAllListeners('pty:ackColdRestore') + ipcMain.removeAllListeners('pty:ackData') ipcMain.removeAllListeners('pty:serializeBuffer:response') // Configure the local provider with app-specific hooks. @@ -1012,12 +1013,16 @@ export function registerPtyHandlers( } const pendingData = new Map() + const rendererInFlightCharsByPty = new Map() const trustedTerminalHandleEnv = new Set() let flushTimer: ReturnType | null = null + let rendererInFlightTotalChars = 0 const PTY_BATCH_INTERVAL_MS = 8 const PTY_BATCH_DRAIN_CONTINUE_MS = 1 const PTY_BATCH_FLUSH_CHUNK_CHARS = 16 * 1024 const PTY_BATCH_FLUSH_MAX_WRITES = 2 + const PTY_RENDERER_IN_FLIGHT_HIGH_WATER_CHARS = 512 * 1024 + const PTY_RENDERER_TOTAL_IN_FLIGHT_HIGH_WATER_CHARS = 8 * 1024 * 1024 // Why: keep the immediate path bounded to keystroke-sized TUI redraws; // large output and non-interactive output must still use the batcher. const INTERACTIVE_OUTPUT_WINDOW_MS = 100 @@ -1071,6 +1076,27 @@ export function registerPtyHandlers( return payload } + function getPtyPayloadCharCount(payload: { data: string; rawLength?: number }): number { + return Math.max(0, payload.rawLength ?? payload.data.length) + } + + function canSendPtyDataToRenderer(id: string): boolean { + return ( + (rendererInFlightCharsByPty.get(id) ?? 0) < PTY_RENDERER_IN_FLIGHT_HIGH_WATER_CHARS && + rendererInFlightTotalChars < PTY_RENDERER_TOTAL_IN_FLIGHT_HIGH_WATER_CHARS + ) + } + + function sendPtyDataToRenderer( + id: string, + payload: { id: string; data: string; seq?: number; rawLength?: number } + ): void { + const charCount = getPtyPayloadCharCount(payload) + rendererInFlightCharsByPty.set(id, (rendererInFlightCharsByPty.get(id) ?? 0) + charCount) + rendererInFlightTotalChars += charCount + mainWindow.webContents.send('pty:data', payload) + } + function appendPendingPtyData( existing: PendingPtyData | undefined, data: string, @@ -1099,15 +1125,18 @@ export function registerPtyHandlers( flushTimer = null if (mainWindow.isDestroyed()) { pendingData.clear() + rendererInFlightCharsByPty.clear() + rendererInFlightTotalChars = 0 return } let writes = 0 - while (pendingData.size > 0 && writes < PTY_BATCH_FLUSH_MAX_WRITES) { - const next = pendingData.entries().next().value - if (!next) { + for (const [id, pending] of Array.from(pendingData.entries())) { + if (writes >= PTY_BATCH_FLUSH_MAX_WRITES) { break } - const [id, pending] = next + if (!canSendPtyDataToRenderer(id)) { + continue + } pendingData.delete(id) const { data } = pending const chunk = data.slice(0, PTY_BATCH_FLUSH_CHUNK_CHARS) @@ -1119,10 +1148,10 @@ export function registerPtyHandlers( } pendingData.set(id, nextPending) } - mainWindow.webContents.send('pty:data', makePtyDataPayload(id, chunk, pending.startSeq)) + sendPtyDataToRenderer(id, makePtyDataPayload(id, chunk, pending.startSeq)) writes++ } - if (pendingData.size > 0) { + if (pendingData.size > 0 && writes > 0) { // Why: a background terminal can dump megabytes at once. Yield between // small IPC slices so keystroke writes are not stuck behind one flush. schedulePendingDataFlush(PTY_BATCH_DRAIN_CONTINUE_MS) @@ -1167,6 +1196,8 @@ export function registerPtyHandlers( flushTimer = null } pendingData.clear() + rendererInFlightCharsByPty.clear() + rendererInFlightTotalChars = 0 return } const existing = pendingData.get(payload.id) @@ -1178,11 +1209,15 @@ export function registerPtyHandlers( performance.now() ) if (isInteractiveOutput) { + if (!canSendPtyDataToRenderer(payload.id)) { + pendingData.set(payload.id, pending) + return + } pendingData.delete(payload.id) clearFlushTimerIfIdle() // Why: agent TUIs redraw small prompt regions after every keystroke. // Waiting for the throughput batch timer adds visible input latency. - mainWindow.webContents.send('pty:data', { + sendPtyDataToRenderer(payload.id, { id: payload.id, data: nextData, ...(typeof pending.startSeq === 'number' @@ -1209,14 +1244,19 @@ export function registerPtyHandlers( // tears down the terminal on pty:exit before the batch timer fires. const remaining = pendingData.get(payload.id) if (remaining) { - mainWindow.webContents.send( - 'pty:data', + sendPtyDataToRenderer( + payload.id, makePtyDataPayload(payload.id, remaining.data, remaining.startSeq) ) pendingData.delete(payload.id) } lastInputAtByPty.delete(payload.id) interactiveOutputCharsByPty.delete(payload.id) + rendererInFlightTotalChars = Math.max( + 0, + rendererInFlightTotalChars - (rendererInFlightCharsByPty.get(payload.id) ?? 0) + ) + rendererInFlightCharsByPty.delete(payload.id) mainWindow.webContents.send('pty:exit', payload) } }) @@ -2395,6 +2435,26 @@ export function registerPtyHandlers( } }) + // Why: renderer ACKs bound main→renderer terminal delivery without stopping + // PTY ingestion. Agent/status consumers still see every chunk through the + // provider/runtime path while background renderer writes wait their turn. + ipcMain.on('pty:ackData', (_event, args: { id: string; charCount: number }) => { + const charCount = Number.isFinite(args.charCount) ? Math.max(0, args.charCount) : 0 + const current = rendererInFlightCharsByPty.get(args.id) ?? 0 + const acknowledged = Math.min(current, charCount) + const next = Math.max(0, current - charCount) + rendererInFlightTotalChars = Math.max(0, rendererInFlightTotalChars - acknowledged) + if (next === 0) { + rendererInFlightCharsByPty.delete(args.id) + } else { + rendererInFlightCharsByPty.set(args.id, next) + } + tryGetProviderForPty(args.id)?.acknowledgeDataEvent(args.id, charCount) + if (pendingData.size > 0 && !flushTimer) { + schedulePendingDataFlush(0) + } + }) + ipcMain.removeAllListeners('pty:signal') ipcMain.on('pty:signal', (_event, args: { id: string; signal: string }) => { tryGetProviderForPty(args.id) diff --git a/src/preload/api-types.ts b/src/preload/api-types.ts index 61eb67635..a3bec977a 100644 --- a/src/preload/api-types.ts +++ b/src/preload/api-types.ts @@ -901,6 +901,7 @@ export type PreloadApi = { signal: (id: string, signal: string) => void kill: (id: string, opts?: { keepHistory?: boolean }) => Promise ackColdRestore: (id: string) => void + ackData: (id: string, charCount: number) => void hasChildProcesses: (id: string) => Promise getForegroundProcess: (id: string) => Promise getCwd: (id: string) => Promise diff --git a/src/preload/index.ts b/src/preload/index.ts index f048724cd..f55b9af1c 100644 --- a/src/preload/index.ts +++ b/src/preload/index.ts @@ -683,6 +683,9 @@ const api = { ackColdRestore: (id: string): void => { ipcRenderer.send('pty:ackColdRestore', { id }) }, + ackData: (id: string, charCount: number): void => { + ipcRenderer.send('pty:ackData', { id, charCount }) + }, kill: (id: string, opts?: { keepHistory?: boolean }): Promise => ipcRenderer.invoke('pty:kill', { id, keepHistory: opts?.keepHistory ?? false }), diff --git a/src/renderer/src/components/terminal-pane/pty-dispatcher-pi-routing.test.ts b/src/renderer/src/components/terminal-pane/pty-dispatcher-pi-routing.test.ts index e49860259..e1bf44f05 100644 --- a/src/renderer/src/components/terminal-pane/pty-dispatcher-pi-routing.test.ts +++ b/src/renderer/src/components/terminal-pane/pty-dispatcher-pi-routing.test.ts @@ -21,7 +21,9 @@ describe('dispatcher → transport → onTitleChange for Pi spinner', () => { // The singleton dispatcher subscribes a SINGLE global `window.api.pty.onData` // callback on first `ensurePtyDispatcher()`. We simulate the main process // delivering IPC events by invoking that captured callback directly. - let dispatcherCallback: ((payload: { id: string; data: string }) => void) | null = null + let dispatcherCallback: + | ((payload: { id: string; data: string; rawLength?: number }) => void) + | null = null beforeEach(() => { vi.resetModules() @@ -36,17 +38,20 @@ describe('dispatcher → transport → onTitleChange for Pi spinner', () => { write: vi.fn(), resize: vi.fn(), kill: vi.fn(), - onData: vi.fn((cb: (payload: { id: string; data: string }) => void) => { - // Only the first subscriber wins in production — the dispatcher - // calls onData exactly once (ensurePtyDispatcher guards with the - // `ptyDispatcherAttached` flag). Subsequent transport calls go - // through the same cached subscription, so we capture the first - // one and ignore the rest. - if (!dispatcherCallback) { - dispatcherCallback = cb + ackData: vi.fn(), + onData: vi.fn( + (cb: (payload: { id: string; data: string; rawLength?: number }) => void) => { + // Only the first subscriber wins in production — the dispatcher + // calls onData exactly once (ensurePtyDispatcher guards with the + // `ptyDispatcherAttached` flag). Subsequent transport calls go + // through the same cached subscription, so we capture the first + // one and ignore the rest. + if (!dispatcherCallback) { + dispatcherCallback = cb + } + return () => {} } - return () => {} - }), + ), onReplay: vi.fn(() => () => {}), onExit: vi.fn(() => () => {}) } @@ -62,6 +67,20 @@ describe('dispatcher → transport → onTitleChange for Pi spinner', () => { } }) + it('ACKs PTY data after dispatcher consumers accept the chunk', async () => { + const { ensurePtyDispatcher, ptyDataHandlers } = await import('./pty-dispatcher') + const handler = vi.fn() + + ensurePtyDispatcher() + ptyDataHandlers.set('pty-pi', handler) + + dispatcherCallback?.({ id: 'pty-pi', data: 'chunk', rawLength: 10 } as never) + + expect(handler).toHaveBeenCalledWith('chunk', { rawLength: 10 }) + expect(window.api.pty.ackData).toHaveBeenCalledWith('pty-pi', 10) + ptyDataHandlers.delete('pty-pi') + }) + it('routes Pi OSC title frames from pty:data → onTitleChange via the dispatcher', async () => { const { createIpcPtyTransport } = await import('./pty-transport') const onTitleChange = vi.fn() diff --git a/src/renderer/src/components/terminal-pane/pty-dispatcher.ts b/src/renderer/src/components/terminal-pane/pty-dispatcher.ts index 0673a7a5d..691b599c2 100644 --- a/src/renderer/src/components/terminal-pane/pty-dispatcher.ts +++ b/src/renderer/src/components/terminal-pane/pty-dispatcher.ts @@ -94,29 +94,36 @@ export function ensurePtyDispatcher(): void { } ptyDispatcherAttached = true window.api.pty.onData((payload) => { - let meta: PtyDataMeta | undefined - if (typeof payload.seq === 'number') { - meta ??= {} - meta.seq = payload.seq - } - if (typeof payload.rawLength === 'number') { - meta ??= {} - meta.rawLength = payload.rawLength - } - ptyDataHandlers.get(payload.id)?.(payload.data, meta) - const sidecars = ptyDataSidecars.get(payload.id) - if (sidecars && sidecars.size > 0) { - // Why: snapshot the Set before iterating because watchers commonly - // unsubscribe themselves on the very chunk that satisfies them - // (e.g. agent-paste-draft resolves on DECSET 2004 and immediately - // tears down). Iterating the live Set in that case can skip a - // watcher or — if a watcher synchronously subscribes a sibling — - // double-fire. The Set is never large (one watcher per active - // ready-wait), so the array allocation is cheap. - const snapshot = Array.from(sidecars) - for (const watcher of snapshot) { - watcher(payload.data) + try { + let meta: PtyDataMeta | undefined + if (typeof payload.seq === 'number') { + meta ??= {} + meta.seq = payload.seq } + if (typeof payload.rawLength === 'number') { + meta ??= {} + meta.rawLength = payload.rawLength + } + ptyDataHandlers.get(payload.id)?.(payload.data, meta) + const sidecars = ptyDataSidecars.get(payload.id) + if (sidecars && sidecars.size > 0) { + // Why: snapshot the Set before iterating because watchers commonly + // unsubscribe themselves on the very chunk that satisfies them + // (e.g. agent-paste-draft resolves on DECSET 2004 and immediately + // tears down). Iterating the live Set in that case can skip a + // watcher or — if a watcher synchronously subscribes a sibling — + // double-fire. The Set is never large (one watcher per active + // ready-wait), so the array allocation is cheap. + const snapshot = Array.from(sidecars) + for (const watcher of snapshot) { + watcher(payload.data) + } + } + } finally { + // Why: main budgets renderer-bound terminal output by bytes accepted + // into this dispatcher. ACK in finally so a bad sidecar cannot leave + // a PTY permanently backpressured. + window.api.pty.ackData?.(payload.id, payload.rawLength ?? payload.data.length) } }) window.api.pty.onReplay((payload) => { diff --git a/src/renderer/src/web/web-preload-api.ts b/src/renderer/src/web/web-preload-api.ts index ee544644a..7089e399c 100644 --- a/src/renderer/src/web/web-preload-api.ts +++ b/src/renderer/src/web/web-preload-api.ts @@ -2153,6 +2153,7 @@ function createPtyApi(): NonNullable['pty']> { signal: () => {}, kill: () => Promise.resolve(), ackColdRestore: () => {}, + ackData: () => {}, hasChildProcesses: () => Promise.resolve(false), getForegroundProcess: () => Promise.resolve(null), getCwd: () => Promise.resolve('~'),