From db3765fc17a67c03a76c3442593d63335ce0be95 Mon Sep 17 00:00:00 2001 From: Jinwoo Hong <73622457+Jinwoo-H@users.noreply.github.com> Date: Tue, 19 May 2026 21:04:42 -0400 Subject: [PATCH] Revert terminal stream backpressure changes (#2380) --- 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 | 58 ++-- .../daemon/daemon-stream-data-batch-state.ts | 100 ------ ...on-stream-data-batcher-interactive.test.ts | 174 ---------- ...daemon-stream-data-batcher-test-harness.ts | 83 ----- .../daemon/daemon-stream-data-batcher.test.ts | 249 -------------- src/main/daemon/daemon-stream-data-batcher.ts | 309 +++--------------- 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 +---- 18 files changed, 98 insertions(+), 1445 deletions(-) delete mode 100644 src/main/daemon/daemon-stream-data-batch-state.ts delete mode 100644 src/main/daemon/daemon-stream-data-batcher-interactive.test.ts delete mode 100644 src/main/daemon/daemon-stream-data-batcher-test-harness.ts delete 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 18a49c0a6ff..ea3ff88bfee 100644 --- a/src/main/daemon/client.test.ts +++ b/src/main/daemon/client.test.ts @@ -50,14 +50,10 @@ 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() @@ -77,22 +73,10 @@ 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) { @@ -268,30 +252,6 @@ 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 49f50c42db0..007afd5ee68 100644 --- a/src/main/daemon/client.ts +++ b/src/main/daemon/client.ts @@ -31,7 +31,6 @@ 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 @@ -75,20 +74,15 @@ 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 @@ -96,18 +90,11 @@ export class DaemonClient { this.connectionGeneration++ const gen = this.connectionGeneration - 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}`) - ) + 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) } catch (error) { this.controlSocket?.destroy() this.streamSocket?.destroy() @@ -143,19 +130,14 @@ export class DaemonClient { }) } - notify(type: string, payload: unknown): boolean { + notify(type: string, payload: unknown): void { if (!this.connected || !this.controlSocket) { - return false + return } const id = `${NOTIFY_PREFIX}${++this.requestCounter}` const msg = { id, type, ...(payload !== undefined ? { payload } : {}) } - try { - this.controlSocket.write(encodeNdjson(msg)) - return true - } catch { - return false - } + this.controlSocket.write(encodeNdjson(msg)) } onEvent(listener: (event: unknown) => void): () => void { @@ -179,10 +161,8 @@ 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) @@ -196,12 +176,6 @@ 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) @@ -233,11 +207,6 @@ 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') @@ -245,7 +214,7 @@ export class DaemonClient { return } - cleanup() + socket.removeListener('data', onData) const line = buffer.slice(0, newlineIdx) try { const response = JSON.parse(line) as HelloResponse @@ -260,18 +229,8 @@ 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)) }) } @@ -323,18 +282,13 @@ export class DaemonClient { this.streamSocket.on('data', (chunk) => parser.feed(chunk.toString())) } - private handleDisconnect(generation: number, reason: string): void { + private handleDisconnect(generation: number): 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 d25d76014c9..85454407dc3 100644 --- a/src/main/daemon/daemon-pty-adapter.test.ts +++ b/src/main/daemon/daemon-pty-adapter.test.ts @@ -114,22 +114,6 @@ 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 2ff8273e24d..ca492a9538f 100644 --- a/src/main/daemon/daemon-pty-adapter.ts +++ b/src/main/daemon/daemon-pty-adapter.ts @@ -31,12 +31,6 @@ 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) { @@ -59,10 +53,6 @@ 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 @@ -70,7 +60,6 @@ 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. @@ -95,11 +84,6 @@ 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 { @@ -167,7 +151,6 @@ 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 @@ -251,13 +234,12 @@ export class DaemonPtyAdapter implements IPtyProvider { write(id: string, data: string): void { this.markSessionDirty(id) - this.sendNotification('write', { sessionId: id, data }) + this.client.notify('write', { sessionId: id, data }) } resize(id: string, cols: number, rows: number): void { this.markSessionDirty(id) - this.sessionSizes.set(id, { cols, rows }) - this.sendNotification('resize', { sessionId: id, cols, rows }) + this.client.notify('resize', { sessionId: id, cols, rows }) } async shutdown(id: string, opts: { immediate?: boolean; keepHistory?: boolean }): Promise { @@ -265,7 +247,6 @@ 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 @@ -382,7 +363,6 @@ 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) } } @@ -427,7 +407,6 @@ 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 @@ -483,13 +462,11 @@ 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() @@ -501,9 +478,6 @@ 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. @@ -513,7 +487,6 @@ 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 @@ -534,9 +507,6 @@ export class DaemonPtyAdapter implements IPtyProvider { this.removeEventListener?.() this.removeEventListener = null this.client.disconnect() - this.removeDisconnectedListener?.() - this.removeDisconnectedListener = null - this.pendingNotifications = [] } private async ensureConnected(): Promise { @@ -563,72 +533,6 @@ 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 @@ -746,7 +650,6 @@ 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 c0fc94e809a..a58895f1d69 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, type Socket } from 'net' +import { connect } from 'net' import { tmpdir } from 'os' import { join } from 'path' import { mkdtempSync, rmSync, readFileSync } from 'fs' @@ -66,39 +66,6 @@ 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() @@ -155,22 +122,6 @@ 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 aaba11a0024..2998c16d582 100644 --- a/src/main/daemon/daemon-server.ts +++ b/src/main/daemon/daemon-server.ts @@ -41,12 +41,7 @@ export class DaemonServer { private tokenPath: string private clients = new Map() - private streamDataBatcher = new DaemonStreamDataBatcher( - (clientId) => this.clients.get(clientId), - { - onStreamFailure: (clientId) => this.disconnectClient(clientId) - } - ) + private streamDataBatcher = new DaemonStreamDataBatcher((clientId) => this.clients.get(clientId)) constructor(opts: DaemonServerOptions) { this.socketPath = opts.socketPath @@ -139,7 +134,6 @@ 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, @@ -150,8 +144,6 @@ 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) @@ -169,10 +161,8 @@ export class DaemonServer { socket.on('data', (chunk) => parser.feed(chunk.toString())) socket.on('close', () => { - const client = this.clients.get(clientId) - if (client?.controlSocket === socket) { - this.disconnectClient(clientId, client) - } + this.streamDataBatcher.clear(clientId) + this.clients.delete(clientId) }) } @@ -222,11 +212,19 @@ export class DaemonServer { this.streamDataBatcher.enqueue(clientId, p.sessionId, data) }, onExit: (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) - this.streamDataBatcher.clearSessionInput(clientId, p.sessionId) + // 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 } + }) + ) + } } } }) @@ -240,10 +238,8 @@ export class DaemonServer { case 'write': try { - this.streamDataBatcher.markInput(clientId, request.payload.sessionId) this.host.write(request.payload.sessionId, request.payload.data) } catch (err) { - this.streamDataBatcher.clearSessionInput(clientId, request.payload.sessionId) if (err instanceof SessionNotFoundError) { this.sendExitEvent(client, request.payload.sessionId, -1) } @@ -308,23 +304,19 @@ export class DaemonServer { sessionId: string, code: number ): void { - if (!client) { + if (!client?.streamSocket) { 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. - 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() + client.streamSocket.write( + encodeNdjson({ + type: 'event', + event: 'exit', + sessionId, + payload: { code } + }) + ) } } diff --git a/src/main/daemon/daemon-stream-data-batch-state.ts b/src/main/daemon/daemon-stream-data-batch-state.ts deleted file mode 100644 index cdb4fb90fe1..00000000000 --- a/src/main/daemon/daemon-stream-data-batch-state.ts +++ /dev/null @@ -1,100 +0,0 @@ -import type { Socket } from 'net' -import { encodeNdjson } from './ndjson' - -export type StreamDataClient = { - streamSocket: Socket | null -} - -export type PendingStreamEvent = - | { kind: 'data'; sessionId: string; data: string } - | { kind: 'exit'; sessionId: string; code: number } - -export type PendingStreamDataBatch = { - timer: ReturnType | null - 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. -export const STREAM_DATA_BATCH_INTERVAL_MS = 8 -export const STREAM_DATA_BACKPRESSURE_WARN_BYTES = 512 * 1024 -export const STREAM_DATA_DRAIN_TIMEOUT_MS = 30_000 -export const STREAM_DATA_MAX_QUEUED_BYTES = 8 * 1024 * 1024 -export const STREAM_DATA_MAX_PAYLOAD_CHARS = 64 * 1024 -export const STREAM_DATA_MAX_EVENTS_PER_FLUSH = 1024 -export const INTERACTIVE_OUTPUT_WINDOW_MS = 100 -export const INTERACTIVE_OUTPUT_MAX_CHARS = 1024 - -export function createPendingStreamDataBatch(): PendingStreamDataBatch { - return { - timer: null, - drainTimer: null, - cleanupWait: null, - queue: [], - queueHead: 0, - queuedDataBytes: 0, - waitingForDrain: false, - warnedBackpressure: false - } -} - -export function compactStreamDataBatch(batch: PendingStreamDataBatch): void { - if (batch.queueHead === 0) { - return - } - batch.queue = batch.queue.slice(batch.queueHead) - batch.queueHead = 0 -} - -export function streamInputKey(clientId: string, sessionId: string): string { - return `${clientId}\0${sessionId}` -} - -export function getQueuedDataForSession(batch: PendingStreamDataBatch, sessionId: string): string { - let data = '' - for (let index = batch.queueHead; index < batch.queue.length; index++) { - const entry = batch.queue[index]! - if (entry.kind === 'data' && entry.sessionId === sessionId) { - data += entry.data - } - } - return data -} - -export function removeQueuedDataForSession(batch: PendingStreamDataBatch, sessionId: string): void { - const remaining: PendingStreamEvent[] = [] - for (let index = batch.queueHead; index < batch.queue.length; index++) { - const entry = batch.queue[index]! - if (entry.kind === 'data' && entry.sessionId === sessionId) { - batch.queuedDataBytes -= Buffer.byteLength(entry.data, 'utf8') - } else { - remaining.push(entry) - } - } - batch.queue = remaining - batch.queueHead = 0 -} - -export function writeStreamEvent(streamSocket: Socket, entry: PendingStreamEvent): boolean { - 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 } - } - return streamSocket.write(encodeNdjson(payload)) -} diff --git a/src/main/daemon/daemon-stream-data-batcher-interactive.test.ts b/src/main/daemon/daemon-stream-data-batcher-interactive.test.ts deleted file mode 100644 index caf9aa56019..00000000000 --- a/src/main/daemon/daemon-stream-data-batcher-interactive.test.ts +++ /dev/null @@ -1,174 +0,0 @@ -import { describe, expect, it, vi } from 'vitest' -import { createHarness, parseWrite } from './daemon-stream-data-batcher-test-harness' - -describe('DaemonStreamDataBatcher interactive output', () => { - it('coalesces non-interactive output before writing to the stream socket', () => { - vi.useFakeTimers() - try { - const { batcher, fake } = createHarness() - - batcher.enqueue('client-1', 'session-1', 'a') - batcher.enqueue('client-1', 'session-1', 'b') - - expect(fake.write).not.toHaveBeenCalled() - vi.advanceTimersByTime(7) - expect(fake.write).not.toHaveBeenCalled() - vi.advanceTimersByTime(1) - - expect(fake.write).toHaveBeenCalledTimes(1) - expect(parseWrite(fake.write.mock.calls[0])).toMatchObject({ - type: 'event', - event: 'data', - sessionId: 'session-1', - payload: { data: 'ab' } - }) - } finally { - vi.useRealTimers() - } - }) - - it('sends small redraws immediately after terminal input', () => { - vi.useFakeTimers() - try { - const { batcher, fake, setNow } = createHarness() - - setNow(10) - batcher.markInput('client-1', 'session-1') - setNow(15) - batcher.enqueue('client-1', 'session-1', '\x1b[20;2Hredraw') - - expect(fake.write).toHaveBeenCalledTimes(1) - expect(parseWrite(fake.write.mock.calls[0])).toMatchObject({ - type: 'event', - event: 'data', - sessionId: 'session-1', - payload: { data: '\x1b[20;2Hredraw' } - }) - vi.advanceTimersByTime(8) - expect(fake.write).toHaveBeenCalledTimes(1) - } finally { - vi.useRealTimers() - } - }) - - it('flushes only the interactive session when another session has pending output', () => { - vi.useFakeTimers() - try { - const { batcher, fake, setNow } = createHarness() - - batcher.enqueue('client-1', 'background-session', 'background') - setNow(20) - batcher.markInput('client-1', 'interactive-session') - setNow(21) - batcher.enqueue('client-1', 'interactive-session', 'redraw') - - expect(fake.write).toHaveBeenCalledTimes(1) - expect(parseWrite(fake.write.mock.calls[0])).toMatchObject({ - sessionId: 'interactive-session', - payload: { data: 'redraw' } - }) - - vi.advanceTimersByTime(8) - expect(fake.write).toHaveBeenCalledTimes(2) - expect(parseWrite(fake.write.mock.calls[1])).toMatchObject({ - sessionId: 'background-session', - payload: { data: 'background' } - }) - } finally { - vi.useRealTimers() - } - }) - - it('waits for drain after an immediate interactive write backpressures', () => { - vi.useFakeTimers() - try { - const { batcher, fake, setNow } = createHarness([false, true]) - - setNow(10) - batcher.markInput('client-1', 'session-1') - setNow(11) - batcher.enqueue('client-1', 'session-1', 'redraw') - batcher.enqueue('client-1', 'session-2', 'queued') - batcher.flush('client-1') - - expect(fake.write).toHaveBeenCalledTimes(1) - fake.drain() - - expect(fake.write).toHaveBeenCalledTimes(2) - expect(parseWrite(fake.write.mock.calls[1])).toMatchObject({ - sessionId: 'session-2', - payload: { data: 'queued' } - }) - } finally { - vi.useRealTimers() - } - }) - - it('batches large output even after recent terminal input', () => { - vi.useFakeTimers() - try { - const { batcher, fake, setNow } = createHarness() - const largeOutput = 'x'.repeat(1025) - - setNow(10) - batcher.markInput('client-1', 'session-1') - setNow(11) - batcher.enqueue('client-1', 'session-1', largeOutput) - - expect(fake.write).not.toHaveBeenCalled() - vi.advanceTimersByTime(8) - expect(fake.write).toHaveBeenCalledTimes(1) - expect(parseWrite(fake.write.mock.calls[0])).toMatchObject({ - sessionId: 'session-1', - payload: { data: largeOutput } - }) - } finally { - vi.useRealTimers() - } - }) - - it('batches stale output after the interactive window expires', () => { - vi.useFakeTimers() - try { - const { batcher, fake, setNow } = createHarness() - - setNow(10) - batcher.markInput('client-1', 'session-1') - setNow(111) - batcher.enqueue('client-1', 'session-1', 'stale redraw') - - expect(fake.write).not.toHaveBeenCalled() - vi.advanceTimersByTime(8) - expect(fake.write).toHaveBeenCalledTimes(1) - expect(parseWrite(fake.write.mock.calls[0])).toMatchObject({ - sessionId: 'session-1', - payload: { data: 'stale redraw' } - }) - } finally { - vi.useRealTimers() - } - }) - - it('forgets recent input when a session is cleared', () => { - vi.useFakeTimers() - try { - const { batcher, fake, setNow } = createHarness() - - setNow(10) - batcher.markInput('client-1', 'session-1') - batcher.clearSessionInput('client-1', 'session-1') - setNow(11) - batcher.enqueue('client-1', 'session-1', 'redraw') - - expect(fake.write).not.toHaveBeenCalled() - vi.advanceTimersByTime(8) - expect(fake.write).toHaveBeenCalledTimes(1) - expect(parseWrite(fake.write.mock.calls[0])).toMatchObject({ - sessionId: 'session-1', - payload: { data: 'redraw' } - }) - } finally { - vi.useRealTimers() - } - }) -}) diff --git a/src/main/daemon/daemon-stream-data-batcher-test-harness.ts b/src/main/daemon/daemon-stream-data-batcher-test-harness.ts deleted file mode 100644 index 2003b98ad66..00000000000 --- a/src/main/daemon/daemon-stream-data-batcher-test-harness.ts +++ /dev/null @@ -1,83 +0,0 @@ -import type { Socket } from 'net' -import { vi } from 'vitest' -import { DaemonStreamDataBatcher } from './daemon-stream-data-batcher' - -export function parseWrite(call: unknown[]): unknown { - return JSON.parse(String(call[0]).trim()) -} - -export function createFakeSocket(writeResults: boolean[] = [true]): { - 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?.() - } -} - -export function createHarness(writeResults: boolean[] = [true]) { - let now = 0 - const fake = createFakeSocket(writeResults) - const batcher = new DaemonStreamDataBatcher(() => ({ streamSocket: fake.socket }), { - now: () => now - }) - - return { - batcher, - fake, - setNow(value: number) { - now = value - } - } -} diff --git a/src/main/daemon/daemon-stream-data-batcher.test.ts b/src/main/daemon/daemon-stream-data-batcher.test.ts deleted file mode 100644 index 0660fa0d8ad..00000000000 --- a/src/main/daemon/daemon-stream-data-batcher.test.ts +++ /dev/null @@ -1,249 +0,0 @@ -import { describe, expect, it, vi } from 'vitest' -import { DaemonStreamDataBatcher } from './daemon-stream-data-batcher' -import { createFakeSocket } from './daemon-stream-data-batcher-test-harness' - -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 c577e4e52fe..a286854401b 100644 --- a/src/main/daemon/daemon-stream-data-batcher.ts +++ b/src/main/daemon/daemon-stream-data-batcher.ts @@ -1,169 +1,49 @@ import type { Socket } from 'net' -import { performance } from 'node:perf_hooks' -import { - compactStreamDataBatch, - createPendingStreamDataBatch, - getQueuedDataForSession, - INTERACTIVE_OUTPUT_MAX_CHARS, - INTERACTIVE_OUTPUT_WINDOW_MS, - removeQueuedDataForSession, - STREAM_DATA_BACKPRESSURE_WARN_BYTES, - STREAM_DATA_BATCH_INTERVAL_MS, - STREAM_DATA_DRAIN_TIMEOUT_MS, - STREAM_DATA_MAX_EVENTS_PER_FLUSH, - STREAM_DATA_MAX_PAYLOAD_CHARS, - STREAM_DATA_MAX_QUEUED_BYTES, - streamInputKey, - type PendingStreamDataBatch, - type PendingStreamEvent, - type StreamDataClient, - writeStreamEvent -} from './daemon-stream-data-batch-state' +import { encodeNdjson } from './ndjson' + +type StreamDataClient = { + streamSocket: Socket | null +} + +type PendingStreamDataBatch = { + timer: ReturnType | null + queue: { sessionId: string; data: string }[] +} + +// 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 export class DaemonStreamDataBatcher { private pendingByClient = new Map() - private lastInputAtBySession = new Map() private getClient: (clientId: string) => StreamDataClient | undefined - private onStreamFailure: (clientId: string) => void - private now: () => number - constructor( - getClient: (clientId: string) => StreamDataClient | undefined, - opts: { - onStreamFailure?: (clientId: string) => void - now?: () => number - } = {} - ) { + constructor(getClient: (clientId: string) => StreamDataClient | undefined) { this.getClient = getClient - this.onStreamFailure = opts.onStreamFailure ?? (() => {}) - this.now = opts.now ?? (() => performance.now()) - } - - markInput(clientId: string, sessionId: string): void { - this.lastInputAtBySession.set(streamInputKey(clientId, sessionId), this.now()) - } - - clearSessionInput(clientId: string, sessionId: string): void { - this.lastInputAtBySession.delete(streamInputKey(clientId, sessionId)) } 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.clearSessionInput(clientId, sessionId) - 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 false - } - - const batch = this.getOrCreateBatch(clientId) - compactStreamDataBatch(batch) - - if (event.kind === 'data' && !batch.waitingForDrain) { - const queuedSessionData = getQueuedDataForSession(batch, event.sessionId) - const nextSessionData = queuedSessionData + event.data - if (this.shouldSendImmediately(clientId, event.sessionId, nextSessionData)) { - removeQueuedDataForSession(batch, event.sessionId) - const ok = writeStreamEvent(client.streamSocket, { - kind: 'data', - sessionId: event.sessionId, - data: nextSessionData - }) - if (!ok) { - this.handleBackpressure(clientId, batch, client.streamSocket) - return true - } - this.deleteBatchIfIdle(clientId, batch) - return true - } - } - - return this.enqueueForBatch(clientId, batch, event) - } - - private getOrCreateBatch(clientId: string): PendingStreamDataBatch { - let batch = this.pendingByClient.get(clientId) - if (!batch) { - batch = createPendingStreamDataBatch() - this.pendingByClient.set(clientId, batch) - } - return batch - } - - private enqueueForBatch( - clientId: string, - batch: PendingStreamDataBatch, - event: PendingStreamEvent - ): boolean { - const last = batch.queue.at(-1) - 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(event) - if (event.kind === 'data') { - batch.queuedDataBytes += Buffer.byteLength(event.data, 'utf8') - } - } - - 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 - } - - private shouldSendImmediately(clientId: string, sessionId: string, data: string): boolean { - const lastInputAt = this.lastInputAtBySession.get(streamInputKey(clientId, sessionId)) - return ( - data.length <= INTERACTIVE_OUTPUT_MAX_CHARS && - lastInputAt !== undefined && - this.now() - lastInputAt <= INTERACTIVE_OUTPUT_WINDOW_MS - ) - } - - private deleteBatchIfIdle(clientId: string, batch: PendingStreamDataBatch): void { - if (batch.waitingForDrain || batch.queueHead < batch.queue.length) { return } - if (batch.timer) { - clearTimeout(batch.timer) - batch.timer = null + + let batch = this.pendingByClient.get(clientId) + if (!batch) { + batch = { timer: null, queue: [] } + this.pendingByClient.set(clientId, batch) + } + + const last = batch.queue.at(-1) + if (last?.sessionId === sessionId) { + last.data += data + } else { + batch.queue.push({ sessionId, data }) + } + + if (!batch.timer) { + batch.timer = setTimeout(() => this.flush(clientId), STREAM_DATA_BATCH_INTERVAL_MS) } - this.pendingByClient.delete(clientId) } flush(clientId: string): void { @@ -176,41 +56,23 @@ export class DaemonStreamDataBatcher { clearTimeout(batch.timer) batch.timer = null } - if (batch.waitingForDrain) { - return - } + this.pendingByClient.delete(clientId) 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 ok = writeStreamEvent(streamSocket, entry) - if (!ok) { - this.handleBackpressure(clientId, batch, streamSocket) - return - } - if ( - flushedEvents >= STREAM_DATA_MAX_EVENTS_PER_FLUSH && - batch.queueHead < batch.queue.length - ) { - compactStreamDataBatch(batch) - batch.timer = setTimeout(() => this.flush(clientId), 0) - batch.timer.unref?.() - return - } + for (const entry of batch.queue) { + client.streamSocket.write( + encodeNdjson({ + type: 'event', + event: 'data', + sessionId: entry.sessionId, + payload: { data: entry.data } + }) + ) } - this.pendingByClient.delete(clientId) } clear(clientId?: string): void { @@ -220,99 +82,10 @@ 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) } - if (clientId === undefined) { - this.lastInputAtBySession.clear() - } else { - for (const key of this.lastInputAtBySession.keys()) { - if (key.startsWith(`${clientId}\0`)) { - this.lastInputAtBySession.delete(key) - } - } - } - } - - private handleBackpressure( - clientId: string, - batch: PendingStreamDataBatch, - streamSocket: Socket - ): void { - batch.waitingForDrain = true - if (batch.timer) { - clearTimeout(batch.timer) - batch.timer = null - } - compactStreamDataBatch(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) - } - - 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 c30d7e4e6b8..ce9200c0305 100644 --- a/src/main/ipc/pty.test.ts +++ b/src/main/ipc/pty.test.ts @@ -2452,41 +2452,6 @@ 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 77383870653..3407a64b6d8 100644 --- a/src/main/ipc/pty.ts +++ b/src/main/ipc/pty.ts @@ -650,12 +650,6 @@ 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 @@ -663,30 +657,10 @@ export function registerPtyHandlers( pendingData.clear() return } - - 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) + for (const [id, data] of pendingData) { + mainWindow.webContents.send('pty:data', { id, data }) } + pendingData.clear() } const clearFlushTimerIfIdle = (): void => { @@ -740,7 +714,10 @@ export function registerPtyHandlers( clearFlushTimerIfIdle() // Why: agent TUIs redraw small prompt regions after every keystroke. // Waiting for the throughput batch timer adds visible input latency. - sendPtyData(payload.id, nextData) + mainWindow.webContents.send('pty:data', { + id: payload.id, + data: nextData + }) return } pendingData.set(payload.id, nextData) @@ -761,7 +738,7 @@ export function registerPtyHandlers( // tears down the terminal on pty:exit before the batch timer fires. const remaining = pendingData.get(payload.id) if (remaining) { - sendPtyData(payload.id, remaining) + mainWindow.webContents.send('pty:data', { id: payload.id, data: 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 5110d24148d..e405225b512 100644 --- a/src/renderer/src/components/terminal-pane/pty-connection.test.ts +++ b/src/renderer/src/components/terminal-pane/pty-connection.test.ts @@ -2009,51 +2009,7 @@ describe('connectPanePty', () => { } }) - 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 () => { + it('writes visible split-pane PTY bytes immediately even when the tab is not active', async () => { const { connectPanePty } = await import('./pty-connection') const transport = createMockTransport() const capturedDataCallback: { current: ((data: string) => void) | null } = { current: null } @@ -2064,9 +2020,9 @@ describe('connectPanePty', () => { transportFactoryQueue.push(transport) const pane = createPane(1) - const manager = createManager(2) - manager.getActivePane.mockReturnValue({ id: 1 }) + const manager = createManager(1) const deps = createDeps({ + isActiveRef: { current: false }, isVisibleRef: { current: true } }) @@ -2074,9 +2030,9 @@ describe('connectPanePty', () => { await flushAsyncTicks(6) expect(capturedDataCallback.current).not.toBeNull() - capturedDataCallback.current?.('active split output\r\n') + capturedDataCallback.current?.('visible split output\r\n') - expect(pane.terminal.write).toHaveBeenCalledWith('active split output\r\n') + expect(pane.terminal.write).toHaveBeenCalledWith('visible 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 71c4e3ec1cf..feb125f8dd5 100644 --- a/src/renderer/src/components/terminal-pane/pty-connection.ts +++ b/src/renderer/src/components/terminal-pane/pty-connection.ts @@ -1139,12 +1139,11 @@ export function connectPanePty( manager.markPaneHasComplexScriptOutput(pane.id) } recordTerminalOutput(pane.terminal) - // 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 + // 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. writeTerminalOutput(pane.terminal, data, { - foreground: deps.isVisibleRef.current && activePaneId === pane.id + foreground: deps.isVisibleRef.current }) 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 5c4438615c9..0ebbf111580 100644 --- a/src/renderer/src/store/slices/agent-status.test.ts +++ b/src/renderer/src/store/slices/agent-status.test.ts @@ -192,46 +192,6 @@ 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 344623a40eb..d61aa81cc3e 100644 --- a/src/renderer/src/store/slices/agent-status.ts +++ b/src/renderer/src/store/slices/agent-status.ts @@ -123,27 +123,6 @@ 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 @@ -305,23 +284,6 @@ 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 c5e931ca51b..df8974ae5a4 100644 --- a/tests/e2e/terminal-output-scheduler.spec.ts +++ b/tests/e2e/terminal-output-scheduler.spec.ts @@ -15,13 +15,7 @@ import { waitForActiveWorktree, waitForSessionReady } from './helpers/store' -import { - getTerminalContent, - splitActiveTerminalPane, - waitForActiveTerminalManager, - waitForPaneCount, - waitForPaneIdentitySnapshot -} from './helpers/terminal' +import { getTerminalContent, waitForActiveTerminalManager } from './helpers/terminal' type SchedulerDebugSnapshot = { backgroundEnqueueCount: number @@ -255,75 +249,4 @@ 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) - }) })