From f010bb861900549adefcb1686fda8ef654620812 Mon Sep 17 00:00:00 2001 From: Jinwoo Hong <73622457+Jinwoo-H@users.noreply.github.com> Date: Tue, 19 May 2026 16:59:01 -0400 Subject: [PATCH] Fix terminal stream backpressure and split-pane input lag (#2341) Co-authored-by: Orca --- src/main/daemon/client.test.ts | 42 ++- src/main/daemon/client.ts | 66 +++- src/main/daemon/daemon-pty-adapter.test.ts | 16 + src/main/daemon/daemon-pty-adapter.ts | 101 +++++- src/main/daemon/daemon-server.test.ts | 51 ++- src/main/daemon/daemon-server.ts | 55 ++-- .../daemon/daemon-stream-data-batcher.test.ts | 309 ++++++++++++++++++ src/main/daemon/daemon-stream-data-batcher.ts | 223 +++++++++++-- src/main/ipc/pty.test.ts | 35 ++ src/main/ipc/pty.ts | 39 ++- .../terminal-pane/pty-connection.test.ts | 54 ++- .../terminal-pane/pty-connection.ts | 9 +- .../src/store/slices/agent-status.test.ts | 40 +++ src/renderer/src/store/slices/agent-status.ts | 38 +++ tests/e2e/terminal-output-scheduler.spec.ts | 79 ++++- 15 files changed, 1079 insertions(+), 78 deletions(-) create mode 100644 src/main/daemon/daemon-stream-data-batcher.test.ts diff --git a/src/main/daemon/client.test.ts b/src/main/daemon/client.test.ts index ea3ff88bf..18a49c0a6 100644 --- a/src/main/daemon/client.test.ts +++ b/src/main/daemon/client.test.ts @@ -50,10 +50,14 @@ describe('DaemonClient', () => { function startMockDaemon(opts?: { onControlMessage?: (msg: unknown) => string | null onStreamHello?: (msg: HelloMessage) => void + streamHelloGate?: Promise rejectVersion?: boolean }): Promise { return new Promise((resolve) => { server = createServer((socket) => { + socket.on('error', () => { + /* tests intentionally destroy sockets mid-handshake */ + }) let buffer = '' socket.on('data', (chunk) => { buffer += chunk.toString() @@ -73,10 +77,22 @@ describe('DaemonClient', () => { socket.write(encodeNdjson({ type: 'hello', ok: false, error: 'Version mismatch' })) return } - socket.write(encodeNdjson({ type: 'hello', ok: true })) if (hello.role === 'stream') { opts?.onStreamHello?.(hello) + if (opts?.streamHelloGate) { + void opts.streamHelloGate.then(() => { + if (!socket.destroyed) { + try { + socket.write(encodeNdjson({ type: 'hello', ok: true })) + } catch { + /* client intentionally disconnected during handshake */ + } + } + }) + return + } } + socket.write(encodeNdjson({ type: 'hello', ok: true })) } else if (opts?.onControlMessage) { const response = opts.onControlMessage(msg) if (response) { @@ -252,6 +268,30 @@ describe('DaemonClient', () => { client = new DaemonClient({ socketPath, tokenPath }) expect(() => client.disconnect()).not.toThrow() }) + + it('disconnect() cancels an in-flight connection attempt', async () => { + let releaseStreamHello!: () => void + const streamHelloGate = new Promise((resolve) => { + releaseStreamHello = resolve + }) + let sawStreamHello = false + await startMockDaemon({ + streamHelloGate, + onStreamHello: () => { + sawStreamHello = true + } + }) + + client = new DaemonClient({ socketPath, tokenPath }) + const connectPromise = client.ensureConnected() + await waitFor(() => sawStreamHello) + + client.disconnect() + releaseStreamHello() + + await expect(connectPromise).rejects.toThrow() + expect(client.isConnected()).toBe(false) + }) }) describe('notify (fire-and-forget)', () => { diff --git a/src/main/daemon/client.ts b/src/main/daemon/client.ts index 007afd5ee..49f50c42d 100644 --- a/src/main/daemon/client.ts +++ b/src/main/daemon/client.ts @@ -31,6 +31,7 @@ export class DaemonClient { private streamSocket: Socket | null = null private connected = false private disconnectArmed = false + private disconnectGeneration = 0 // Why: after a disconnect + reconnect (daemon respawn), a stale 'close' // event from the old sockets can fire. Without a generation check, that // event would tear down the fresh connection. Each doConnect() increments @@ -74,15 +75,20 @@ export class DaemonClient { private async doConnect(): Promise { const token = readFileSync(this.tokenPath, 'utf-8').trim() + const disconnectGeneration = this.disconnectGeneration try { // Sequential: control first, then stream this.controlSocket = await this.connectSocket() + this.assertConnectNotCancelled(disconnectGeneration) await this.sendHello(this.controlSocket, token, 'control') + this.assertConnectNotCancelled(disconnectGeneration) this.setupControlParser() this.streamSocket = await this.connectSocket() + this.assertConnectNotCancelled(disconnectGeneration) await this.sendHello(this.streamSocket, token, 'stream') + this.assertConnectNotCancelled(disconnectGeneration) this.setupStreamParser() this.connected = true @@ -90,11 +96,18 @@ export class DaemonClient { this.connectionGeneration++ const gen = this.connectionGeneration - const handleClose = () => this.handleDisconnect(gen) - this.controlSocket.on('close', handleClose) - this.controlSocket.on('error', handleClose) - this.streamSocket.on('close', handleClose) - this.streamSocket.on('error', handleClose) + this.controlSocket.on('close', (hadError) => + this.handleDisconnect(gen, `control socket closed${hadError ? ' after error' : ''}`) + ) + this.controlSocket.on('error', (error) => + this.handleDisconnect(gen, `control socket error: ${error.message}`) + ) + this.streamSocket.on('close', (hadError) => + this.handleDisconnect(gen, `stream socket closed${hadError ? ' after error' : ''}`) + ) + this.streamSocket.on('error', (error) => + this.handleDisconnect(gen, `stream socket error: ${error.message}`) + ) } catch (error) { this.controlSocket?.destroy() this.streamSocket?.destroy() @@ -130,14 +143,19 @@ export class DaemonClient { }) } - notify(type: string, payload: unknown): void { + notify(type: string, payload: unknown): boolean { if (!this.connected || !this.controlSocket) { - return + return false } const id = `${NOTIFY_PREFIX}${++this.requestCounter}` const msg = { id, type, ...(payload !== undefined ? { payload } : {}) } - this.controlSocket.write(encodeNdjson(msg)) + try { + this.controlSocket.write(encodeNdjson(msg)) + return true + } catch { + return false + } } onEvent(listener: (event: unknown) => void): () => void { @@ -161,8 +179,10 @@ export class DaemonClient { } disconnect(): void { + this.disconnectGeneration++ this.connected = false this.disconnectArmed = false + this.connectingPromise = null for (const [id, pending] of this.pendingRequests) { clearTimeout(pending.timer) @@ -176,6 +196,12 @@ export class DaemonClient { this.streamSocket = null } + private assertConnectNotCancelled(disconnectGeneration: number): void { + if (disconnectGeneration !== this.disconnectGeneration) { + throw new DaemonProtocolError('Connection cancelled') + } + } + private connectSocket(): Promise { return new Promise((resolve, reject) => { const socket = connect(this.socketPath) @@ -207,6 +233,11 @@ export class DaemonClient { } let buffer = '' + const cleanup = (): void => { + socket.removeListener('data', onData) + socket.removeListener('error', onError) + socket.removeListener('close', onClose) + } const onData = (chunk: Buffer): void => { buffer += chunk.toString() const newlineIdx = buffer.indexOf('\n') @@ -214,7 +245,7 @@ export class DaemonClient { return } - socket.removeListener('data', onData) + cleanup() const line = buffer.slice(0, newlineIdx) try { const response = JSON.parse(line) as HelloResponse @@ -229,8 +260,18 @@ export class DaemonClient { reject(new DaemonProtocolError('Invalid hello response')) } } + const onError = (error: Error): void => { + cleanup() + reject(error) + } + const onClose = (): void => { + cleanup() + reject(new DaemonProtocolError('Connection closed during hello')) + } socket.on('data', onData) + socket.once('error', onError) + socket.once('close', onClose) socket.write(encodeNdjson(hello)) }) } @@ -282,13 +323,18 @@ export class DaemonClient { this.streamSocket.on('data', (chunk) => parser.feed(chunk.toString())) } - private handleDisconnect(generation: number): void { + private handleDisconnect(generation: number, reason: string): void { if (!this.disconnectArmed || generation !== this.connectionGeneration) { return } this.disconnectArmed = false this.connected = false + console.warn('[daemon-client] disconnected', { + reason, + pendingRequests: this.pendingRequests.size + }) + for (const [id, pending] of this.pendingRequests) { clearTimeout(pending.timer) pending.reject(new DaemonProtocolError('Connection lost')) diff --git a/src/main/daemon/daemon-pty-adapter.test.ts b/src/main/daemon/daemon-pty-adapter.test.ts index 85454407d..d25d76014 100644 --- a/src/main/daemon/daemon-pty-adapter.test.ts +++ b/src/main/daemon/daemon-pty-adapter.test.ts @@ -114,6 +114,22 @@ describe('DaemonPtyAdapter (IPtyProvider)', () => { await new Promise((r) => setTimeout(r, 50)) expect(lastSubprocess.write).toHaveBeenCalledWith('ls\n') }) + + it('reattaches active sessions and flushes writes after daemon stream failure', async () => { + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}) + try { + const { id } = await adapter.spawn({ cols: 80, rows: 24 }) + + lastSubprocess._simulateData('x'.repeat(8 * 1024 * 1024 + 1)) + await new Promise((resolve) => setTimeout(resolve, 100)) + adapter.write(id, 'after-reconnect\n') + + await waitFor(() => vi.mocked(lastSubprocess.write).mock.calls.length > 0, 3000) + expect(lastSubprocess.write).toHaveBeenCalledWith('after-reconnect\n') + } finally { + warn.mockRestore() + } + }) }) describe('resize', () => { diff --git a/src/main/daemon/daemon-pty-adapter.ts b/src/main/daemon/daemon-pty-adapter.ts index 96ac6932a..c7f114623 100644 --- a/src/main/daemon/daemon-pty-adapter.ts +++ b/src/main/daemon/daemon-pty-adapter.ts @@ -31,6 +31,12 @@ export type DaemonPtyAdapterOptions = { } const MAX_TOMBSTONES = 1000 +const MAX_PENDING_DAEMON_NOTIFICATIONS = 512 + +type PendingDaemonNotification = { + type: 'write' | 'resize' + payload: unknown +} export class TerminalKilledError extends Error { constructor(sessionId: string) { @@ -53,6 +59,10 @@ export class DaemonPtyAdapter implements IPtyProvider { private dataListeners: ((payload: { id: string; data: string }) => void)[] = [] private exitListeners: ((payload: { id: string; code: number }) => void)[] = [] private removeEventListener: (() => void) | null = null + private removeDisconnectedListener: (() => void) | null = null + private recoveryPromise: Promise | null = null + private pendingNotifications: PendingDaemonNotification[] = [] + private disposed = false private initialCwds = new Map() // Why: React re-renders and StrictMode double-mounts can call createOrAttach // for a session the user just killed. Without tombstones, the daemon would @@ -60,6 +70,7 @@ export class DaemonPtyAdapter implements IPtyProvider { // Uses a Map so eviction removes the oldest by insertion order, // matching terminal-host.ts tombstone semantics. private killedSessionTombstones = new Map() + private sessionSizes = new Map() // Why: React StrictMode double-mounts: mount → cold restore → unmount → // mount → ??? The sticky cache returns the same cold restore data on the // second mount until the renderer explicitly acknowledges it. @@ -84,6 +95,11 @@ export class DaemonPtyAdapter implements IPtyProvider { this.historyReader = opts.historyPath ? new HistoryReader(opts.historyPath) : null this.respawnFn = opts.respawn ?? null this.supportsCheckpoints = this.protocolVersion >= 4 + this.removeDisconnectedListener = this.client.onDisconnected(() => { + void this.recoverActiveSessionsAfterDisconnect().catch((err) => + console.warn('[daemon] reconnect after stream failure failed:', err) + ) + }) } getHistoryManager(): HistoryManager | null { @@ -151,6 +167,7 @@ export class DaemonPtyAdapter implements IPtyProvider { } this.activeSessionIds.add(sessionId) + this.sessionSizes.set(sessionId, { cols: effectiveCols, rows: effectiveRows }) // Cold restore: daemon created a new session but disk history shows // an unclean shutdown → return saved scrollback so the renderer can @@ -230,12 +247,13 @@ export class DaemonPtyAdapter implements IPtyProvider { write(id: string, data: string): void { this.markSessionDirty(id) - this.client.notify('write', { sessionId: id, data }) + this.sendNotification('write', { sessionId: id, data }) } resize(id: string, cols: number, rows: number): void { this.markSessionDirty(id) - this.client.notify('resize', { sessionId: id, cols, rows }) + this.sessionSizes.set(id, { cols, rows }) + this.sendNotification('resize', { sessionId: id, cols, rows }) } async shutdown(id: string, opts: { immediate?: boolean; keepHistory?: boolean }): Promise { @@ -243,6 +261,7 @@ export class DaemonPtyAdapter implements IPtyProvider { this.activeSessionIds.delete(id) this.dirtySessionVersions.delete(id) this.initialCwds.delete(id) + this.sessionSizes.delete(id) // Why: history removal is for the "user explicitly closed this terminal" // path. Sleep also calls shutdown but expects scrollback to survive — wake // re-spawns and the cold-restore reader needs the dir intact. Caller @@ -359,6 +378,7 @@ export class DaemonPtyAdapter implements IPtyProvider { // set, disconnectOnly()'s final checkpoint would skip them, leaving // stale recovery data if the daemon later crashes. this.activeSessionIds.add(session.sessionId) + this.sessionSizes.set(session.sessionId, { cols: session.cols, rows: session.rows }) this.historyManager?.registerWriter(session.sessionId) } } @@ -403,6 +423,7 @@ export class DaemonPtyAdapter implements IPtyProvider { const ids = [...this.activeSessionIds] this.activeSessionIds.clear() this.dirtySessionVersions.clear() + this.sessionSizes.clear() for (const id of ids) { // Why: listener throws are intentionally *not* caught — matches the // natural onExit fanout in setupEventRouting, so synthetic exits don't @@ -458,11 +479,13 @@ export class DaemonPtyAdapter implements IPtyProvider { } dispose(): void { + this.disposed = true if (this.checkpointInterval) { clearInterval(this.checkpointInterval) this.checkpointInterval = null } this.dirtySessionVersions.clear() + this.sessionSizes.clear() this.removeEventListener?.() this.removeEventListener = null // Why: final checkpoints are written daemon-side in TerminalHost.dispose() @@ -474,6 +497,9 @@ export class DaemonPtyAdapter implements IPtyProvider { .catch((err) => console.warn('[history] dispose failed:', err)) } this.client.disconnect() + this.removeDisconnectedListener?.() + this.removeDisconnectedListener = null + this.pendingNotifications = [] } // Why: for in-process daemon mode, disconnect without flushing history. @@ -483,6 +509,7 @@ export class DaemonPtyAdapter implements IPtyProvider { // We write a final checkpoint before disconnecting so that if the daemon // later crashes while Orca is closed, checkpoint.json has recovery data. async disconnectOnly(): Promise { + this.disposed = true if (this.checkpointInterval) { clearInterval(this.checkpointInterval) this.checkpointInterval = null @@ -503,6 +530,9 @@ export class DaemonPtyAdapter implements IPtyProvider { this.removeEventListener?.() this.removeEventListener = null this.client.disconnect() + this.removeDisconnectedListener?.() + this.removeDisconnectedListener = null + this.pendingNotifications = [] } private async ensureConnected(): Promise { @@ -529,6 +559,72 @@ export class DaemonPtyAdapter implements IPtyProvider { }, DaemonPtyAdapter.CHECKPOINT_INTERVAL_MS) } + private sendNotification(type: PendingDaemonNotification['type'], payload: unknown): void { + if (this.recoveryPromise) { + this.queueNotification(type, payload) + return + } + if (this.client.notify(type, payload)) { + return + } + this.queueNotification(type, payload) + void this.recoverActiveSessionsAfterDisconnect().catch((err) => + console.warn('[daemon] reconnect after notification failure failed:', err) + ) + } + + private queueNotification(type: PendingDaemonNotification['type'], payload: unknown): void { + this.pendingNotifications.push({ type, payload }) + if (this.pendingNotifications.length > MAX_PENDING_DAEMON_NOTIFICATIONS) { + this.pendingNotifications.splice( + 0, + this.pendingNotifications.length - MAX_PENDING_DAEMON_NOTIFICATIONS + ) + } + } + + private async recoverActiveSessionsAfterDisconnect(): Promise { + if (this.disposed || this.activeSessionIds.size === 0) { + return + } + if (!this.recoveryPromise) { + this.recoveryPromise = this.reattachActiveSessions().finally(() => { + this.recoveryPromise = null + }) + } + await this.recoveryPromise + } + + private async reattachActiveSessions(): Promise { + await this.ensureConnected() + // Why: daemon stream failure only breaks the renderer socket pair; the + // backing PTYs stay alive in TerminalHost. Reattach active sessions so + // stream events resume instead of letting panes black-hole input. + for (const sessionId of this.activeSessionIds) { + const size = this.sessionSizes.get(sessionId) ?? { cols: 80, rows: 24 } + await this.client.request('createOrAttach', { + sessionId, + cols: size.cols, + rows: size.rows + }) + } + this.flushPendingNotifications() + } + + private flushPendingNotifications(): void { + const pending = this.pendingNotifications + this.pendingNotifications = [] + for (const notification of pending) { + if (!this.client.notify(notification.type, notification.payload)) { + this.queueNotification(notification.type, notification.payload) + void this.recoverActiveSessionsAfterDisconnect().catch((err) => + console.warn('[daemon] reconnect after pending notification failed:', err) + ) + return + } + } + } + private markSessionDirty(sessionId: string): void { if (!this.activeSessionIds.has(sessionId)) { return @@ -646,6 +742,7 @@ export class DaemonPtyAdapter implements IPtyProvider { } else if (event.event === 'exit') { this.activeSessionIds.delete(event.sessionId) this.dirtySessionVersions.delete(event.sessionId) + this.sessionSizes.delete(event.sessionId) if (this.historyManager) { void this.historyManager .closeSession(event.sessionId, event.payload.code) diff --git a/src/main/daemon/daemon-server.test.ts b/src/main/daemon/daemon-server.test.ts index a58895f1d..c0fc94e80 100644 --- a/src/main/daemon/daemon-server.test.ts +++ b/src/main/daemon/daemon-server.test.ts @@ -1,5 +1,5 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' -import { connect } from 'net' +import { connect, type Socket } from 'net' import { tmpdir } from 'os' import { join } from 'path' import { mkdtempSync, rmSync, readFileSync } from 'fs' @@ -66,6 +66,39 @@ describe('DaemonServer', () => { return client } + async function connectRawSocket(role: 'control' | 'stream', clientId: string): Promise { + const socket = connect(socketPath) + await new Promise((resolve) => socket.on('connect', resolve)) + socket.write( + encodeNdjson({ + type: 'hello', + version: PROTOCOL_VERSION, + token: readFileSync(tokenPath, 'utf-8').trim(), + clientId, + role + }) + ) + const response = await readSocketLine(socket) + expect(JSON.parse(response)).toMatchObject({ ok: true }) + return socket + } + + function readSocketLine(socket: Socket): Promise { + return new Promise((resolve) => { + let buffer = '' + const onData = (chunk: Buffer): void => { + buffer += chunk.toString() + const newlineIdx = buffer.indexOf('\n') + if (newlineIdx === -1) { + return + } + socket.removeListener('data', onData) + resolve(buffer.slice(0, newlineIdx)) + } + socket.on('data', onData) + }) + } + describe('startup', () => { it('creates token file and starts listening', async () => { await startServer() @@ -122,6 +155,22 @@ describe('DaemonServer', () => { expect(result).toEqual({ pong: true }) }) + it('keeps a replacement control socket alive when the old socket closes later', async () => { + await startServer() + const firstControl = await connectRawSocket('control', 'same-client') + const secondControl = await connectRawSocket('control', 'same-client') + + await new Promise((resolve) => setTimeout(resolve, 20)) + secondControl.write(encodeNdjson({ id: 'req-1', type: 'ping' })) + + await expect(readSocketLine(secondControl)).resolves.toMatch( + /"id":"req-1".*"payload":\{"pong":true\}/ + ) + + firstControl.destroy() + secondControl.destroy() + }) + it('handles write (fire-and-forget)', async () => { await startServer() const c = await connectClient() diff --git a/src/main/daemon/daemon-server.ts b/src/main/daemon/daemon-server.ts index 2998c16d5..1193cf4ae 100644 --- a/src/main/daemon/daemon-server.ts +++ b/src/main/daemon/daemon-server.ts @@ -41,7 +41,12 @@ export class DaemonServer { private tokenPath: string private clients = new Map() - private streamDataBatcher = new DaemonStreamDataBatcher((clientId) => this.clients.get(clientId)) + private streamDataBatcher = new DaemonStreamDataBatcher( + (clientId) => this.clients.get(clientId), + { + onStreamFailure: (clientId) => this.disconnectClient(clientId) + } + ) constructor(opts: DaemonServerOptions) { this.socketPath = opts.socketPath @@ -134,6 +139,7 @@ export class DaemonServer { socket.write(encodeNdjson({ type: 'hello', ok: true })) if (hello.role === 'control') { + this.disconnectClient(hello.clientId) const client: ConnectedClient = { clientId: hello.clientId, controlSocket: socket, @@ -144,6 +150,8 @@ export class DaemonServer { } else if (hello.role === 'stream') { const client = this.clients.get(hello.clientId) if (client) { + this.streamDataBatcher.clear(hello.clientId) + client.streamSocket?.destroy() client.streamSocket = socket } // Stream socket is receive-only from daemon's perspective (for events) @@ -161,8 +169,10 @@ export class DaemonServer { socket.on('data', (chunk) => parser.feed(chunk.toString())) socket.on('close', () => { - this.streamDataBatcher.clear(clientId) - this.clients.delete(clientId) + const client = this.clients.get(clientId) + if (client?.controlSocket === socket) { + this.disconnectClient(clientId, client) + } }) } @@ -212,19 +222,10 @@ export class DaemonServer { this.streamDataBatcher.enqueue(clientId, p.sessionId, data) }, onExit: (code) => { - // Why: exit tears down renderer handlers; flush final output first - // so the last few milliseconds of PTY data are not stranded. - this.streamDataBatcher.flush(clientId) - if (client?.streamSocket) { - client.streamSocket.write( - encodeNdjson({ - type: 'event', - event: 'exit', - sessionId: p.sessionId, - payload: { code } - }) - ) - } + // Why: exit tears down renderer handlers; queue it behind any + // pending data so the final PTY bytes cannot be overtaken under + // stream backpressure. + this.streamDataBatcher.enqueueExit(clientId, p.sessionId, code) } } }) @@ -304,19 +305,23 @@ export class DaemonServer { sessionId: string, code: number ): void { - if (!client?.streamSocket) { + if (!client) { return } // Why: write/resize are notification-heavy and intentionally do not wait // for replies. If their target session is gone, this synthetic exit is the // only signal the renderer gets to clear stale terminal pane bindings. - client.streamSocket.write( - encodeNdjson({ - type: 'event', - event: 'exit', - sessionId, - payload: { code } - }) - ) + this.streamDataBatcher.enqueueExit(client.clientId, sessionId, code) + } + + private disconnectClient(clientId: string, expectedClient?: ConnectedClient): void { + const client = this.clients.get(clientId) + if (expectedClient && client !== expectedClient) { + return + } + this.streamDataBatcher.clear(clientId) + this.clients.delete(clientId) + client?.streamSocket?.destroy() + client?.controlSocket.destroy() } } diff --git a/src/main/daemon/daemon-stream-data-batcher.test.ts b/src/main/daemon/daemon-stream-data-batcher.test.ts new file mode 100644 index 000000000..891c2f032 --- /dev/null +++ b/src/main/daemon/daemon-stream-data-batcher.test.ts @@ -0,0 +1,309 @@ +import { describe, expect, it, vi } from 'vitest' +import type { Socket } from 'net' +import { DaemonStreamDataBatcher } from './daemon-stream-data-batcher' + +function createFakeSocket(writeResults: boolean[]): { + socket: Socket + write: ReturnType + removeListener: ReturnType + drain: () => void + close: () => void + error: () => void + staleDrain: () => void + staleClose: () => void + staleError: () => void +} { + let drainHandler: (() => void) | null = null + let closeHandler: (() => void) | null = null + let errorHandler: (() => void) | null = null + let removedDrainHandler: (() => void) | null = null + let removedCloseHandler: (() => void) | null = null + let removedErrorHandler: (() => void) | null = null + const write = vi.fn(() => writeResults.shift() ?? true) + const removeListener = vi.fn((event: string, handler: () => void) => { + if (event === 'drain' && drainHandler === handler) { + removedDrainHandler = handler + drainHandler = null + } else if (event === 'close' && closeHandler === handler) { + removedCloseHandler = handler + closeHandler = null + } else if (event === 'error' && errorHandler === handler) { + removedErrorHandler = handler + errorHandler = null + } + return socket + }) + const socket = { + destroyed: false, + write, + removeListener, + once: vi.fn((event: string, handler: () => void) => { + if (event === 'drain') { + drainHandler = handler + } else if (event === 'close') { + closeHandler = handler + } else if (event === 'error') { + errorHandler = handler + } + return socket + }) + } as unknown as Socket + + return { + socket, + write, + removeListener, + drain: () => drainHandler?.(), + close: () => closeHandler?.(), + error: () => errorHandler?.(), + staleDrain: () => removedDrainHandler?.(), + staleClose: () => removedCloseHandler?.(), + staleError: () => removedErrorHandler?.() + } +} + +describe('DaemonStreamDataBatcher', () => { + it('drops queued output when the backpressured stream errors before drain', () => { + const fake = createFakeSocket([false, true]) + const batcher = new DaemonStreamDataBatcher(() => ({ streamSocket: fake.socket })) + + batcher.enqueue('client-1', 'session-a', 'first') + batcher.enqueue('client-1', 'session-b', 'second') + + batcher.flush('client-1') + fake.error() + fake.drain() + + expect(fake.write).toHaveBeenCalledTimes(1) + }) + + it('cleans up unused backpressure listeners after drain', () => { + const fake = createFakeSocket([false, true]) + const batcher = new DaemonStreamDataBatcher(() => ({ streamSocket: fake.socket })) + + batcher.enqueue('client-1', 'session-a', 'first') + batcher.enqueue('client-1', 'session-b', 'second') + + batcher.flush('client-1') + fake.drain() + + expect(fake.removeListener).toHaveBeenCalledWith('close', expect.any(Function)) + expect(fake.removeListener).toHaveBeenCalledWith('error', expect.any(Function)) + fake.close() + + expect(fake.write).toHaveBeenCalledTimes(2) + }) + + it('drops queued output when the backpressured stream closes before drain', () => { + const fake = createFakeSocket([false, true]) + const batcher = new DaemonStreamDataBatcher(() => ({ streamSocket: fake.socket })) + + batcher.enqueue('client-1', 'session-a', 'first') + batcher.enqueue('client-1', 'session-b', 'second') + + batcher.flush('client-1') + fake.close() + fake.drain() + + expect(fake.write).toHaveBeenCalledTimes(1) + }) + + it('does not keep queued output forever when drain never arrives', () => { + vi.useFakeTimers() + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}) + try { + const fake = createFakeSocket([false, true]) + const onStreamFailure = vi.fn() + const batcher = new DaemonStreamDataBatcher(() => ({ streamSocket: fake.socket }), { + onStreamFailure + }) + + batcher.enqueue('client-1', 'session-a', 'first') + batcher.enqueue('client-1', 'session-b', 'second') + batcher.flush('client-1') + + vi.advanceTimersByTime(30_000) + fake.drain() + + expect(fake.write).toHaveBeenCalledTimes(1) + expect(warn).toHaveBeenCalledWith( + '[daemon] PTY stream socket drain timed out', + expect.objectContaining({ clientId: 'client-1' }) + ) + expect(onStreamFailure).toHaveBeenCalledWith('client-1') + } finally { + warn.mockRestore() + vi.useRealTimers() + } + }) + + it('orders exit behind queued data while waiting for drain', () => { + const fake = createFakeSocket([false, true, true]) + const batcher = new DaemonStreamDataBatcher(() => ({ streamSocket: fake.socket })) + + batcher.enqueue('client-1', 'session-a', 'first') + batcher.flush('client-1') + batcher.enqueue('client-1', 'session-b', 'second') + batcher.enqueueExit('client-1', 'session-a', 0) + + expect(fake.write).toHaveBeenCalledTimes(1) + fake.drain() + + expect(fake.write).toHaveBeenCalledTimes(3) + expect(fake.write.mock.calls[1]?.[0]).toContain('"data"') + expect(fake.write.mock.calls[1]?.[0]).toContain('second') + expect(fake.write.mock.calls[2]?.[0]).toContain('"exit"') + }) + + it('waits for socket drain before continuing after stream backpressure', () => { + const fake = createFakeSocket([false, true]) + const batcher = new DaemonStreamDataBatcher(() => ({ streamSocket: fake.socket })) + + batcher.enqueue('client-1', 'session-a', 'first') + batcher.enqueue('client-1', 'session-b', 'second') + + batcher.flush('client-1') + expect(fake.write).toHaveBeenCalledTimes(1) + expect(fake.write.mock.calls[0]?.[0]).toContain('first') + + fake.drain() + + expect(fake.write).toHaveBeenCalledTimes(2) + expect(fake.write.mock.calls[1]?.[0]).toContain('second') + }) + + it('queues additional output while waiting for drain', () => { + const fake = createFakeSocket([false, true]) + const batcher = new DaemonStreamDataBatcher(() => ({ streamSocket: fake.socket })) + + batcher.enqueue('client-1', 'session-a', 'first') + batcher.flush('client-1') + + batcher.enqueue('client-1', 'session-b', 'second') + batcher.flush('client-1') + expect(fake.write).toHaveBeenCalledTimes(1) + + fake.drain() + + expect(fake.write).toHaveBeenCalledTimes(2) + expect(fake.write.mock.calls[1]?.[0]).toContain('second') + }) + + it('ignores stale drain callbacks after clearing an old client stream', () => { + const oldStream = createFakeSocket([false, true]) + const newStream = createFakeSocket([true]) + let streamSocket = oldStream.socket + const batcher = new DaemonStreamDataBatcher(() => ({ streamSocket })) + + batcher.enqueue('client-1', 'session-a', 'first') + batcher.enqueue('client-1', 'session-b', 'second') + batcher.flush('client-1') + + batcher.clear('client-1') + streamSocket = newStream.socket + batcher.enqueue('client-1', 'session-c', 'third') + batcher.flush('client-1') + oldStream.staleDrain() + + expect(oldStream.write).toHaveBeenCalledTimes(1) + expect(newStream.write).toHaveBeenCalledTimes(1) + expect(newStream.write.mock.calls[0]?.[0]).toContain('third') + }) + + it('ignores stale close and error callbacks after clearing an old client stream', () => { + const oldStream = createFakeSocket([false, true]) + const newStream = createFakeSocket([true]) + let streamSocket = oldStream.socket + const batcher = new DaemonStreamDataBatcher(() => ({ streamSocket })) + + batcher.enqueue('client-1', 'session-a', 'first') + batcher.enqueue('client-1', 'session-b', 'second') + batcher.flush('client-1') + + batcher.clear('client-1') + streamSocket = newStream.socket + batcher.enqueue('client-1', 'session-c', 'third') + batcher.flush('client-1') + oldStream.staleClose() + oldStream.staleError() + + expect(oldStream.write).toHaveBeenCalledTimes(1) + expect(newStream.write).toHaveBeenCalledTimes(1) + expect(newStream.write.mock.calls[0]?.[0]).toContain('third') + }) + + it('fails the stream when queued output exceeds the hard cap', () => { + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}) + try { + const fake = createFakeSocket([true]) + const onStreamFailure = vi.fn() + const batcher = new DaemonStreamDataBatcher(() => ({ streamSocket: fake.socket }), { + onStreamFailure + }) + + batcher.enqueue('client-1', 'session-a', 'x'.repeat(8 * 1024 * 1024 + 1)) + + expect(fake.write).not.toHaveBeenCalled() + expect(onStreamFailure).toHaveBeenCalledWith('client-1') + expect(warn).toHaveBeenCalledWith( + '[daemon] PTY stream socket queue exceeded limit', + expect.objectContaining({ clientId: 'client-1' }) + ) + } finally { + warn.mockRestore() + } + }) + + it('fails the stream when queued output cumulatively exceeds the hard cap while backpressured', () => { + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}) + try { + const fake = createFakeSocket([false]) + const onStreamFailure = vi.fn() + const batcher = new DaemonStreamDataBatcher(() => ({ streamSocket: fake.socket }), { + onStreamFailure + }) + + batcher.enqueue('client-1', 'session-a', 'first') + batcher.flush('client-1') + batcher.enqueue('client-1', 'session-a', 'x'.repeat(8 * 1024 * 1024 + 1)) + + expect(fake.write).toHaveBeenCalledTimes(1) + expect(onStreamFailure).toHaveBeenCalledWith('client-1') + expect(warn).toHaveBeenCalledWith( + '[daemon] PTY stream socket queue exceeded limit', + expect.objectContaining({ clientId: 'client-1' }) + ) + } finally { + warn.mockRestore() + } + }) + + it('splits very large output into bounded stream frames', () => { + const fake = createFakeSocket([true, true]) + const batcher = new DaemonStreamDataBatcher(() => ({ streamSocket: fake.socket })) + + batcher.enqueue('client-1', 'session-a', 'x'.repeat(65 * 1024)) + batcher.flush('client-1') + + expect(fake.write).toHaveBeenCalledTimes(2) + }) + + it('yields between large queued flushes', () => { + vi.useFakeTimers() + try { + const fake = createFakeSocket(Array.from({ length: 1025 }, () => true)) + const batcher = new DaemonStreamDataBatcher(() => ({ streamSocket: fake.socket })) + + for (let i = 0; i < 1025; i++) { + batcher.enqueue('client-1', `session-${i}`, `${i}`) + } + batcher.flush('client-1') + + expect(fake.write).toHaveBeenCalledTimes(1024) + vi.advanceTimersByTime(0) + expect(fake.write).toHaveBeenCalledTimes(1025) + } finally { + vi.useRealTimers() + } + }) +}) diff --git a/src/main/daemon/daemon-stream-data-batcher.ts b/src/main/daemon/daemon-stream-data-batcher.ts index a28685440..64336443a 100644 --- a/src/main/daemon/daemon-stream-data-batcher.ts +++ b/src/main/daemon/daemon-stream-data-batcher.ts @@ -5,45 +5,115 @@ type StreamDataClient = { streamSocket: Socket | null } +type PendingStreamEvent = + | { kind: 'data'; sessionId: string; data: string } + | { kind: 'exit'; sessionId: string; code: number } + type PendingStreamDataBatch = { timer: ReturnType | null - queue: { sessionId: string; data: string }[] + drainTimer: ReturnType | null + cleanupWait: (() => void) | null + queue: PendingStreamEvent[] + queueHead: number + queuedDataBytes: number + waitingForDrain: boolean + warnedBackpressure: boolean } // 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 STREAM_DATA_BACKPRESSURE_WARN_BYTES = 512 * 1024 +const STREAM_DATA_DRAIN_TIMEOUT_MS = 30_000 +const STREAM_DATA_MAX_QUEUED_BYTES = 8 * 1024 * 1024 +const STREAM_DATA_MAX_PAYLOAD_CHARS = 64 * 1024 +const STREAM_DATA_MAX_EVENTS_PER_FLUSH = 1024 export class DaemonStreamDataBatcher { private pendingByClient = new Map() private getClient: (clientId: string) => StreamDataClient | undefined + private onStreamFailure: (clientId: string) => void - constructor(getClient: (clientId: string) => StreamDataClient | undefined) { + constructor( + getClient: (clientId: string) => StreamDataClient | undefined, + opts: { onStreamFailure?: (clientId: string) => void } = {} + ) { this.getClient = getClient + this.onStreamFailure = opts.onStreamFailure ?? (() => {}) } enqueue(clientId: string, sessionId: string, data: string): void { + for (let offset = 0; offset < data.length; offset += STREAM_DATA_MAX_PAYLOAD_CHARS) { + const shouldContinue = this.enqueueEvent(clientId, { + kind: 'data', + sessionId, + data: data.slice(offset, offset + STREAM_DATA_MAX_PAYLOAD_CHARS) + }) + if (!shouldContinue) { + return + } + } + } + + enqueueExit(clientId: string, sessionId: string, code: number): void { + this.enqueueEvent(clientId, { kind: 'exit', sessionId, code }) + this.flush(clientId) + } + + private enqueueEvent(clientId: string, event: PendingStreamEvent): boolean { const client = this.getClient(clientId) if (!client?.streamSocket || client.streamSocket.destroyed) { - return + return false } let batch = this.pendingByClient.get(clientId) if (!batch) { - batch = { timer: null, queue: [] } + batch = { + timer: null, + drainTimer: null, + cleanupWait: null, + queue: [], + queueHead: 0, + queuedDataBytes: 0, + waitingForDrain: false, + warnedBackpressure: false + } this.pendingByClient.set(clientId, batch) } const last = batch.queue.at(-1) - if (last?.sessionId === sessionId) { - last.data += data + if ( + event.kind === 'data' && + last?.kind === 'data' && + last.sessionId === event.sessionId && + last.data.length + event.data.length <= STREAM_DATA_MAX_PAYLOAD_CHARS + ) { + last.data += event.data + batch.queuedDataBytes += Buffer.byteLength(event.data, 'utf8') } else { - batch.queue.push({ sessionId, data }) + batch.queue.push(event) + if (event.kind === 'data') { + batch.queuedDataBytes += Buffer.byteLength(event.data, 'utf8') + } } - if (!batch.timer) { + if (batch.queuedDataBytes > STREAM_DATA_MAX_QUEUED_BYTES) { + console.warn('[daemon] PTY stream socket queue exceeded limit', { + clientId, + queuedEvents: batch.queue.length - batch.queueHead, + queuedBytes: batch.queuedDataBytes + }) + // Why: backpressure is on the renderer's single stream socket. Once that + // socket is unhealthy, per-PTY recovery cannot make progress until the + // client reconnects and reattaches its active sessions. + this.failStream(clientId) + return false + } + + if (!batch.timer && !batch.waitingForDrain) { batch.timer = setTimeout(() => this.flush(clientId), STREAM_DATA_BATCH_INTERVAL_MS) } + return true } flush(clientId: string): void { @@ -56,23 +126,117 @@ export class DaemonStreamDataBatcher { clearTimeout(batch.timer) batch.timer = null } - this.pendingByClient.delete(clientId) - - const client = this.getClient(clientId) - if (!client?.streamSocket || client.streamSocket.destroyed) { + if (batch.waitingForDrain) { return } - for (const entry of batch.queue) { - client.streamSocket.write( - encodeNdjson({ - type: 'event', - event: 'data', - sessionId: entry.sessionId, - payload: { data: entry.data } - }) - ) + const client = this.getClient(clientId) + if (!client?.streamSocket || client.streamSocket.destroyed) { + this.pendingByClient.delete(clientId) + return } + const streamSocket = client.streamSocket + + let flushedEvents = 0 + while (batch.queueHead < batch.queue.length) { + const entry = batch.queue[batch.queueHead]! + batch.queueHead += 1 + flushedEvents += 1 + if (entry.kind === 'data') { + batch.queuedDataBytes -= Buffer.byteLength(entry.data, 'utf8') + } + const payload = + entry.kind === 'data' + ? { + type: 'event', + event: 'data', + sessionId: entry.sessionId, + payload: { data: entry.data } + } + : { + type: 'event', + event: 'exit', + sessionId: entry.sessionId, + payload: { code: entry.code } + } + const ok = streamSocket.write(encodeNdjson(payload)) + if (!ok) { + batch.waitingForDrain = true + this.compactQueue(batch) + if ( + batch.queuedDataBytes >= STREAM_DATA_BACKPRESSURE_WARN_BYTES && + !batch.warnedBackpressure + ) { + batch.warnedBackpressure = true + console.warn('[daemon] PTY stream socket backpressure', { + clientId, + queuedEvents: batch.queue.length - batch.queueHead, + queuedBytes: batch.queuedDataBytes + }) + } + let settled = false + const handleDrain = (): void => { + if (settled) { + return + } + cleanupWait() + const current = this.pendingByClient.get(clientId) + if (current !== batch) { + return + } + current.waitingForDrain = false + this.flush(clientId) + } + const handleTerminal = (): void => { + if (settled) { + return + } + cleanupWait() + if (this.pendingByClient.get(clientId) === batch) { + this.pendingByClient.delete(clientId) + } + } + const cleanupWait = (): void => { + settled = true + if (batch.drainTimer) { + clearTimeout(batch.drainTimer) + batch.drainTimer = null + } + batch.cleanupWait = null + streamSocket.removeListener('drain', handleDrain) + streamSocket.removeListener('close', handleTerminal) + streamSocket.removeListener('error', handleTerminal) + } + batch.cleanupWait = cleanupWait + batch.drainTimer = setTimeout(() => { + if (settled) { + return + } + console.warn('[daemon] PTY stream socket drain timed out', { + clientId, + queuedEvents: batch.queue.length - batch.queueHead, + queuedBytes: batch.queuedDataBytes + }) + cleanupWait() + this.failStream(clientId) + }, STREAM_DATA_DRAIN_TIMEOUT_MS) + batch.drainTimer.unref?.() + streamSocket.once('close', handleTerminal) + streamSocket.once('error', handleTerminal) + streamSocket.once('drain', handleDrain) + return + } + if ( + flushedEvents >= STREAM_DATA_MAX_EVENTS_PER_FLUSH && + batch.queueHead < batch.queue.length + ) { + this.compactQueue(batch) + batch.timer = setTimeout(() => this.flush(clientId), 0) + batch.timer.unref?.() + return + } + } + this.pendingByClient.delete(clientId) } clear(clientId?: string): void { @@ -82,10 +246,27 @@ export class DaemonStreamDataBatcher { : [[clientId, this.pendingByClient.get(clientId)] as const] for (const [id, batch] of batches) { + batch?.cleanupWait?.() if (batch?.timer) { clearTimeout(batch.timer) } + if (batch?.drainTimer) { + clearTimeout(batch.drainTimer) + } this.pendingByClient.delete(id) } } + + private compactQueue(batch: PendingStreamDataBatch): void { + if (batch.queueHead === 0) { + return + } + batch.queue = batch.queue.slice(batch.queueHead) + batch.queueHead = 0 + } + + private failStream(clientId: string): void { + this.clear(clientId) + this.onStreamFailure(clientId) + } } diff --git a/src/main/ipc/pty.test.ts b/src/main/ipc/pty.test.ts index 38f15fe68..d07b7857c 100644 --- a/src/main/ipc/pty.test.ts +++ b/src/main/ipc/pty.test.ts @@ -2434,6 +2434,41 @@ describe('registerPtyHandlers', () => { } }) + it('splits very large batched PTY output across IPC flush turns', async () => { + vi.useFakeTimers() + const mockProc = createMockProc() + spawnMock.mockReturnValue(mockProc.proc) + + try { + registerPtyHandlers(mainWindow as never) + const spawnResult = (await handlers.get('pty:spawn')!(null, { + cols: 80, + rows: 24, + cwd: '/tmp' + })) as { id: string } + mainWindow.webContents.send.mockClear() + + const largeOutput = 'x'.repeat(9 * 64 * 1024) + mockProc.emitData(largeOutput) + + vi.advanceTimersByTime(8) + expect(mainWindow.webContents.send).toHaveBeenCalledTimes(8) + expect(mainWindow.webContents.send).toHaveBeenLastCalledWith('pty:data', { + id: spawnResult.id, + data: 'x'.repeat(64 * 1024) + }) + + vi.advanceTimersByTime(8) + expect(mainWindow.webContents.send).toHaveBeenCalledTimes(9) + expect(mainWindow.webContents.send).toHaveBeenLastCalledWith('pty:data', { + id: spawnResult.id, + data: 'x'.repeat(64 * 1024) + }) + } finally { + vi.useRealTimers() + } + }) + it('batches combined pending output that exceeds the interactive size limit', async () => { vi.useFakeTimers() const mockProc = createMockProc() diff --git a/src/main/ipc/pty.ts b/src/main/ipc/pty.ts index 4a1454292..53531e4ce 100644 --- a/src/main/ipc/pty.ts +++ b/src/main/ipc/pty.ts @@ -646,6 +646,12 @@ export function registerPtyHandlers( // large output and non-interactive output must still use the batcher. const INTERACTIVE_OUTPUT_WINDOW_MS = 100 const INTERACTIVE_OUTPUT_MAX_CHARS = 1024 + const PTY_IPC_CHUNK_CHARS = 64 * 1024 + const PTY_MAX_IPC_CHUNKS_PER_FLUSH = 8 + + const sendPtyData = (id: string, data: string): void => { + mainWindow.webContents.send('pty:data', { id, data }) + } const flushPendingData = (): void => { flushTimer = null @@ -653,10 +659,30 @@ export function registerPtyHandlers( pendingData.clear() return } - for (const [id, data] of pendingData) { - mainWindow.webContents.send('pty:data', { id, data }) + + let sentChunks = 0 + while (pendingData.size > 0 && sentChunks < PTY_MAX_IPC_CHUNKS_PER_FLUSH) { + const entry = pendingData.entries().next().value + if (!entry) { + break + } + const [id, data] = entry + pendingData.delete(id) + if (data.length <= PTY_IPC_CHUNK_CHARS) { + sendPtyData(id, data) + } else { + sendPtyData(id, data.slice(0, PTY_IPC_CHUNK_CHARS)) + // Why: a single noisy PTY can otherwise serialize megabytes of IPC in + // one main-process turn. Requeue the remainder behind other PTYs so + // active terminal redraws and control IPC keep getting turns. + pendingData.set(id, data.slice(PTY_IPC_CHUNK_CHARS)) + } + sentChunks++ + } + + if (pendingData.size > 0) { + flushTimer = setTimeout(flushPendingData, PTY_BATCH_INTERVAL_MS) } - pendingData.clear() } const clearFlushTimerIfIdle = (): void => { @@ -710,10 +736,7 @@ export function registerPtyHandlers( 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', { - id: payload.id, - data: nextData - }) + sendPtyData(payload.id, nextData) return } pendingData.set(payload.id, nextData) @@ -734,7 +757,7 @@ 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', { id: payload.id, data: remaining }) + sendPtyData(payload.id, remaining) pendingData.delete(payload.id) } lastInputAtByPty.delete(payload.id) diff --git a/src/renderer/src/components/terminal-pane/pty-connection.test.ts b/src/renderer/src/components/terminal-pane/pty-connection.test.ts index 060f6ea82..584e6c0da 100644 --- a/src/renderer/src/components/terminal-pane/pty-connection.test.ts +++ b/src/renderer/src/components/terminal-pane/pty-connection.test.ts @@ -990,7 +990,51 @@ describe('connectPanePty', () => { } }) - it('writes visible split-pane PTY bytes immediately even when the tab is not active', async () => { + it('queues visible inactive split-pane PTY bytes so active pane input stays responsive', async () => { + const pendingTimeouts: (() => void)[] = [] + const originalSetTimeout = globalThis.setTimeout + globalThis.setTimeout = vi.fn((fn: () => void) => { + pendingTimeouts.push(fn) + return 999 as unknown as ReturnType + }) as unknown as typeof setTimeout + + try { + const { connectPanePty } = await import('./pty-connection') + const transport = createMockTransport() + const capturedDataCallback: { current: ((data: string) => void) | null } = { current: null } + transport.connect.mockImplementation( + async ({ callbacks }: { callbacks: ConnectCallbacks }) => { + capturedDataCallback.current = callbacks.onData ?? null + return 'pty-id' + } + ) + transportFactoryQueue.push(transport) + + const pane = createPane(1) + const manager = createManager(2) + manager.getActivePane.mockReturnValue({ id: 2 }) + const deps = createDeps({ + isVisibleRef: { current: true } + }) + + connectPanePty(pane as never, manager as never, deps as never) + await flushAsyncTicks(6) + + expect(capturedDataCallback.current).not.toBeNull() + capturedDataCallback.current?.('visible split output\r\n') + expect(pane.terminal.write).not.toHaveBeenCalledWith('visible split output\r\n') + + for (const fn of pendingTimeouts) { + fn() + } + + expect(pane.terminal.write).toHaveBeenCalledWith('visible split output\r\n') + } finally { + globalThis.setTimeout = originalSetTimeout + } + }) + + it('writes active visible split-pane PTY bytes immediately', async () => { const { connectPanePty } = await import('./pty-connection') const transport = createMockTransport() const capturedDataCallback: { current: ((data: string) => void) | null } = { current: null } @@ -1001,9 +1045,9 @@ describe('connectPanePty', () => { transportFactoryQueue.push(transport) const pane = createPane(1) - const manager = createManager(1) + const manager = createManager(2) + manager.getActivePane.mockReturnValue({ id: 1 }) const deps = createDeps({ - isActiveRef: { current: false }, isVisibleRef: { current: true } }) @@ -1011,9 +1055,9 @@ describe('connectPanePty', () => { await flushAsyncTicks(6) expect(capturedDataCallback.current).not.toBeNull() - capturedDataCallback.current?.('visible split output\r\n') + capturedDataCallback.current?.('active split output\r\n') - expect(pane.terminal.write).toHaveBeenCalledWith('visible split output\r\n') + expect(pane.terminal.write).toHaveBeenCalledWith('active split output\r\n') }) it('marks panes that receive Arabic output for DOM rendering', async () => { diff --git a/src/renderer/src/components/terminal-pane/pty-connection.ts b/src/renderer/src/components/terminal-pane/pty-connection.ts index 8a89b9b9d..f504d1737 100644 --- a/src/renderer/src/components/terminal-pane/pty-connection.ts +++ b/src/renderer/src/components/terminal-pane/pty-connection.ts @@ -877,11 +877,12 @@ export function connectPanePty( if (terminalOutputPrefersDomRenderer(data)) { manager.markPaneHasComplexScriptOutput(pane.id) } - // Why: visibility is the right gate — split-pane layouts have multiple - // visible-but-inactive panes whose output the user is watching. Only - // hidden panes (background tabs) should be throttled. + // Why: the active split pane owns keyboard latency. Visible inactive + // panes still drain, but through the shared scheduler so a build log in + // another split cannot monopolize xterm writes while the user types. + const activePaneId = manager.getActivePane()?.id ?? pane.id writeTerminalOutput(pane.terminal, data, { - foreground: deps.isVisibleRef.current + foreground: deps.isVisibleRef.current && activePaneId === pane.id }) if (pendingStartupCommand) { diff --git a/src/renderer/src/store/slices/agent-status.test.ts b/src/renderer/src/store/slices/agent-status.test.ts index 0ebbf1115..5c4438615 100644 --- a/src/renderer/src/store/slices/agent-status.test.ts +++ b/src/renderer/src/store/slices/agent-status.test.ts @@ -192,6 +192,46 @@ describe('agent status tool + assistant fields', () => { expect(store.getState().sortEpoch).toBe(firstSortEpoch + 1) }) + it('throttles unchanged fresh same-state heartbeats to avoid status-map churn', () => { + vi.useFakeTimers() + const store = createTestStore() + store + .getState() + .setAgentStatus( + 'tab-1:1', + { state: 'working', prompt: 'p1', agentType: 'claude', toolName: 'Read' }, + 'claude', + { updatedAt: 1_000, stateStartedAt: 1_000 } + ) + const firstMap = store.getState().agentStatusByPaneKey + const firstEntry = firstMap['tab-1:1'] + + store + .getState() + .setAgentStatus( + 'tab-1:1', + { state: 'working', prompt: 'p1', agentType: 'claude', toolName: 'Read' }, + 'claude', + { updatedAt: 1_500, stateStartedAt: 1_000 } + ) + + expect(store.getState().agentStatusByPaneKey).toBe(firstMap) + expect(store.getState().agentStatusByPaneKey['tab-1:1']).toBe(firstEntry) + expect(store.getState().agentStatusByPaneKey['tab-1:1'].updatedAt).toBe(1_000) + + store + .getState() + .setAgentStatus( + 'tab-1:1', + { state: 'working', prompt: 'p1', agentType: 'claude', toolName: 'Read' }, + 'claude', + { updatedAt: 2_000, stateStartedAt: 1_000 } + ) + + expect(store.getState().agentStatusByPaneKey).not.toBe(firstMap) + expect(store.getState().agentStatusByPaneKey['tab-1:1'].updatedAt).toBe(2_000) + }) + it('bumps global epochs when a stale same-state entry refreshes', () => { vi.useFakeTimers() const store = createTestStore() diff --git a/src/renderer/src/store/slices/agent-status.ts b/src/renderer/src/store/slices/agent-status.ts index d61aa81cc..344623a40 100644 --- a/src/renderer/src/store/slices/agent-status.ts +++ b/src/renderer/src/store/slices/agent-status.ts @@ -123,6 +123,27 @@ function paneKeyMatchesAnyTabPrefix(paneKey: string, tabPrefixes: string[]): boo return false } +const UNCHANGED_AGENT_STATUS_UPDATE_MIN_INTERVAL_MS = 1_000 + +function isUnchangedAgentStatusHeartbeat( + previous: AgentStatusEntry, + next: AgentStatusEntry +): boolean { + return ( + previous.state === next.state && + previous.prompt === next.prompt && + previous.stateStartedAt === next.stateStartedAt && + previous.agentType === next.agentType && + previous.paneKey === next.paneKey && + previous.terminalTitle === next.terminalTitle && + previous.stateHistory === next.stateHistory && + previous.toolName === next.toolName && + previous.toolInput === next.toolInput && + previous.lastAssistantMessage === next.lastAssistantMessage && + previous.interrupted === next.interrupted + ) +} + function pruneMigrationUnsupportedEntries( entries: Record, predicate: (entry: MigrationUnsupportedPtyEntry) => boolean @@ -284,6 +305,23 @@ export const createAgentStatusSlice: StateCreator entry.paneKey === paneKey + ) + // Why: Codex/Claude hook heartbeats can arrive many times per second + // with only `updatedAt` changed. Rewriting the whole status map for + // those pings wakes sidebar/runtime subscribers without changing what + // the user sees, so keep freshness accurate at human-scale cadence. + if ( + existing && + !sortRelevantChange && + !hasSuppressor && + !hasMigrationUnsupportedForPaneKey && + updatedAt - existing.updatedAt < UNCHANGED_AGENT_STATUS_UPDATE_MIN_INTERVAL_MS && + isUnchangedAgentStatusHeartbeat(existing, entry) + ) { + return s + } let nextRetentionSuppressedPaneKeys = s.retentionSuppressedPaneKeys if (hasSuppressor) { nextRetentionSuppressedPaneKeys = { ...s.retentionSuppressedPaneKeys } diff --git a/tests/e2e/terminal-output-scheduler.spec.ts b/tests/e2e/terminal-output-scheduler.spec.ts index df8974ae5..c5e931ca5 100644 --- a/tests/e2e/terminal-output-scheduler.spec.ts +++ b/tests/e2e/terminal-output-scheduler.spec.ts @@ -15,7 +15,13 @@ import { waitForActiveWorktree, waitForSessionReady } from './helpers/store' -import { getTerminalContent, waitForActiveTerminalManager } from './helpers/terminal' +import { + getTerminalContent, + splitActiveTerminalPane, + waitForActiveTerminalManager, + waitForPaneCount, + waitForPaneIdentitySnapshot +} from './helpers/terminal' type SchedulerDebugSnapshot = { backgroundEnqueueCount: number @@ -249,4 +255,75 @@ test.describe('Terminal output scheduler', () => { }) .toBe(true) }) + + test('visible inactive split-pane output uses the shared drain @headful', async ({ + orcaPage + }) => { + await waitForSessionReady(orcaPage) + await waitForActiveWorktree(orcaPage) + await ensureTerminalVisible(orcaPage) + await waitForActiveTerminalManager(orcaPage, 30_000) + await waitForPaneCount(orcaPage, 1, 30_000) + + await splitActiveTerminalPane(orcaPage, 'horizontal') + const snapshot = await waitForPaneIdentitySnapshot(orcaPage, 2) + const activePane = snapshot.panes[0] + const inactivePane = snapshot.panes[1] + if (!activePane.ptyId || !inactivePane.ptyId) { + throw new Error('Split pane PTY ids were unavailable') + } + + await orcaPage.evaluate( + ({ tabId, paneId }) => { + const manager = window.__paneManagers?.get(tabId) + if (!manager) { + throw new Error('Active terminal PaneManager is not mounted') + } + // Why: the active split pane owns keyboard latency even though both + // split panes are visible in this headful repro. + manager.setActivePane(paneId, { focus: true }) + }, + { tabId: snapshot.tabId, paneId: activePane.numericPaneId } + ) + + await resetSchedulerDebug(orcaPage) + + const runId = Date.now() + const activeMarker = `ACTIVE_SPLIT_SCHED_${runId}` + const inactiveMarker = `INACTIVE_SPLIT_SCHED_${runId}` + await sendPtyCommands(orcaPage, [ + { + ptyId: inactivePane.ptyId, + command: nodeConsoleCommand(`'x'.repeat(120000) + ':${inactiveMarker}'`) + }, + { + ptyId: activePane.ptyId, + command: nodeConsoleCommand(`'${activeMarker}'`) + } + ]) + + await expect + .poll(async () => (await getTerminalContent(orcaPage)).includes(activeMarker), { + timeout: 5_000, + message: 'Active split pane did not render foreground output during inactive burst' + }) + .toBe(true) + + await expect + .poll(async () => (await getSchedulerDebug(orcaPage)).backgroundEnqueueCount, { + timeout: 5_000, + message: 'Visible inactive split-pane output bypassed the shared scheduler' + }) + .toBeGreaterThanOrEqual(1) + + await expect + .poll(async () => (await getSchedulerDebug(orcaPage)).backgroundWriteCount, { + timeout: 10_000, + message: 'Visible inactive split-pane output did not drain from the shared scheduler' + }) + .toBeGreaterThanOrEqual(1) + + const debug = await getSchedulerDebug(orcaPage) + expect(debug.foregroundWriteCount).toBeGreaterThan(0) + }) })