diff --git a/src/main/daemon/daemon-server.ts b/src/main/daemon/daemon-server.ts index 1193cf4ae..aaba11a00 100644 --- a/src/main/daemon/daemon-server.ts +++ b/src/main/daemon/daemon-server.ts @@ -226,6 +226,7 @@ export class DaemonServer { // 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) } } }) @@ -239,8 +240,10 @@ 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) } diff --git a/src/main/daemon/daemon-stream-data-batcher.test.ts b/src/main/daemon/daemon-stream-data-batcher.test.ts index 891c2f032..0ddec7c6f 100644 --- a/src/main/daemon/daemon-stream-data-batcher.test.ts +++ b/src/main/daemon/daemon-stream-data-batcher.test.ts @@ -2,7 +2,11 @@ import { describe, expect, it, vi } from 'vitest' import type { Socket } from 'net' import { DaemonStreamDataBatcher } from './daemon-stream-data-batcher' -function createFakeSocket(writeResults: boolean[]): { +function parseWrite(call: unknown[]): unknown { + return JSON.parse(String(call[0]).trim()) +} + +function createFakeSocket(writeResults: boolean[] = [true]): { socket: Socket write: ReturnType removeListener: ReturnType @@ -62,7 +66,193 @@ function createFakeSocket(writeResults: boolean[]): { } } +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 + } + } +} + describe('DaemonStreamDataBatcher', () => { + 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() + } + }) + it('drops queued output when the backpressured stream errors before drain', () => { const fake = createFakeSocket([false, true]) const batcher = new DaemonStreamDataBatcher(() => ({ streamSocket: fake.socket })) diff --git a/src/main/daemon/daemon-stream-data-batcher.ts b/src/main/daemon/daemon-stream-data-batcher.ts index 64336443a..422102726 100644 --- a/src/main/daemon/daemon-stream-data-batcher.ts +++ b/src/main/daemon/daemon-stream-data-batcher.ts @@ -1,4 +1,5 @@ import type { Socket } from 'net' +import { performance } from 'node:perf_hooks' import { encodeNdjson } from './ndjson' type StreamDataClient = { @@ -28,18 +29,34 @@ 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 +const INTERACTIVE_OUTPUT_WINDOW_MS = 100 +const INTERACTIVE_OUTPUT_MAX_CHARS = 1024 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 } = {} + opts: { + onStreamFailure?: (clientId: string) => void + now?: () => number + } = {} ) { this.getClient = getClient this.onStreamFailure = opts.onStreamFailure ?? (() => {}) + this.now = opts.now ?? (() => performance.now()) + } + + markInput(clientId: string, sessionId: string): void { + this.lastInputAtBySession.set(this.inputKey(clientId, sessionId), this.now()) + } + + clearSessionInput(clientId: string, sessionId: string): void { + this.lastInputAtBySession.delete(this.inputKey(clientId, sessionId)) } enqueue(clientId: string, sessionId: string, data: string): void { @@ -56,6 +73,7 @@ export class DaemonStreamDataBatcher { } enqueueExit(clientId: string, sessionId: string, code: number): void { + this.clearSessionInput(clientId, sessionId) this.enqueueEvent(clientId, { kind: 'exit', sessionId, code }) this.flush(clientId) } @@ -66,6 +84,32 @@ export class DaemonStreamDataBatcher { return false } + const batch = this.getOrCreateBatch(clientId) + this.compactQueue(batch) + + if (event.kind === 'data' && !batch.waitingForDrain) { + const queuedSessionData = this.getQueuedDataForSession(batch, event.sessionId) + const nextSessionData = queuedSessionData + event.data + if (this.shouldSendImmediately(clientId, event.sessionId, nextSessionData)) { + this.removeQueuedDataForSession(batch, event.sessionId) + const ok = this.writeEvent(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 = { @@ -80,7 +124,14 @@ export class DaemonStreamDataBatcher { } 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' && @@ -116,6 +167,76 @@ export class DaemonStreamDataBatcher { return true } + private shouldSendImmediately(clientId: string, sessionId: string, data: string): boolean { + const lastInputAt = this.lastInputAtBySession.get(this.inputKey(clientId, sessionId)) + return ( + data.length <= INTERACTIVE_OUTPUT_MAX_CHARS && + lastInputAt !== undefined && + this.now() - lastInputAt <= INTERACTIVE_OUTPUT_WINDOW_MS + ) + } + + private 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 + } + + private 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 + } + + 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 + } + this.pendingByClient.delete(clientId) + } + + private writeEvent(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)) + } + + private inputKey(clientId: string, sessionId: string): string { + return `${clientId}\0${sessionId}` + } + flush(clientId: string): void { const batch = this.pendingByClient.get(clientId) if (!batch) { @@ -145,85 +266,9 @@ export class DaemonStreamDataBatcher { 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)) + const ok = this.writeEvent(streamSocket, entry) 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) + this.handleBackpressure(clientId, batch, streamSocket) return } if ( @@ -255,6 +300,15 @@ export class DaemonStreamDataBatcher { } 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 compactQueue(batch: PendingStreamDataBatch): void { @@ -265,6 +319,77 @@ export class DaemonStreamDataBatcher { batch.queueHead = 0 } + private handleBackpressure( + clientId: string, + batch: PendingStreamDataBatch, + streamSocket: Socket + ): void { + batch.waitingForDrain = true + if (batch.timer) { + clearTimeout(batch.timer) + batch.timer = null + } + 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) + } + private failStream(clientId: string): void { this.clear(clientId) this.onStreamFailure(clientId) 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 c940126d8..5110d2414 100644 --- a/src/renderer/src/components/terminal-pane/pty-connection.test.ts +++ b/src/renderer/src/components/terminal-pane/pty-connection.test.ts @@ -2102,6 +2102,31 @@ describe('connectPanePty', () => { expect(pane.terminal.write).toHaveBeenCalledWith('Arabic: السلام عليكم\r\n') }) + it('keeps panes on WebGL for terminal UI drawing glyphs', async () => { + 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(1) + const deps = createDeps() + + connectPanePty(pane as never, manager as never, deps as never) + await flushAsyncTicks(6) + + capturedDataCallback.current?.('⠋ Working ├─ file.ts █ progress \uE0B0 prompt\r\n') + + expect(manager.markPaneHasComplexScriptOutput).not.toHaveBeenCalled() + expect(pane.terminal.write).toHaveBeenCalledWith( + '⠋ Working ├─ file.ts █ progress \uE0B0 prompt\r\n' + ) + }) + it('reattaches via daemon sessionId when an in-session PTY is live', async () => { const { connectPanePty } = await import('./pty-connection') const transport = createMockTransport() diff --git a/src/renderer/src/lib/pane-manager/terminal-complex-script.test.ts b/src/renderer/src/lib/pane-manager/terminal-complex-script.test.ts index 5c66dc064..0384d2213 100644 --- a/src/renderer/src/lib/pane-manager/terminal-complex-script.test.ts +++ b/src/renderer/src/lib/pane-manager/terminal-complex-script.test.ts @@ -19,12 +19,15 @@ describe('terminalOutputPrefersDomRenderer', () => { expect(terminalOutputPrefersDomRenderer('Fullwidth: ABC123')).toBe(true) }) - it('detects glyph classes common in agent terminal UIs', () => { - expect(terminalOutputPrefersDomRenderer('⠋ Working')).toBe(true) - expect(terminalOutputPrefersDomRenderer('├─ file.ts')).toBe(true) - expect(terminalOutputPrefersDomRenderer('█ progress')).toBe(true) - expect(terminalOutputPrefersDomRenderer('◆ status')).toBe(true) - expect(terminalOutputPrefersDomRenderer('\uE0B0 prompt')).toBe(true) + it('keeps terminal drawing glyphs on WebGL', () => { + expect(terminalOutputPrefersDomRenderer('⠋ Working')).toBe(false) + expect(terminalOutputPrefersDomRenderer('├─ file.ts')).toBe(false) + expect(terminalOutputPrefersDomRenderer('█ progress')).toBe(false) + expect(terminalOutputPrefersDomRenderer('◆ status')).toBe(false) + expect(terminalOutputPrefersDomRenderer('\uE0B0 prompt')).toBe(false) + }) + + it('detects malformed replacement characters', () => { expect(terminalOutputPrefersDomRenderer('bad replacement �')).toBe(true) }) diff --git a/src/renderer/src/lib/pane-manager/terminal-complex-script.ts b/src/renderer/src/lib/pane-manager/terminal-complex-script.ts index beecca3cf..8132487ef 100644 --- a/src/renderer/src/lib/pane-manager/terminal-complex-script.ts +++ b/src/renderer/src/lib/pane-manager/terminal-complex-script.ts @@ -1,5 +1,6 @@ -// Why: xterm WebGL renders from a glyph atlas; agent TUIs often combine glyphs -// that are safer through the browser text path even when they are not RTL. +// Why: xterm WebGL renders from a glyph atlas; actual complex text is safer +// through the browser text path. Terminal UI drawing glyphs stay on WebGL +// because xterm's custom-glyph renderer is built for those ranges. const EMOJI_PRESENTATION_PATTERN = /\p{Emoji_Presentation}/u function isInRange(value: number, start: number, end: number): boolean { @@ -11,16 +12,12 @@ function isRendererRiskCodePoint(value: number): boolean { isInRange(value, 0x0590, 0x08ff) || value === 0x200d || isInRange(value, 0x1100, 0x11ff) || - isInRange(value, 0x2500, 0x259f) || - isInRange(value, 0x25a0, 0x25ff) || - isInRange(value, 0x2800, 0x28ff) || // Why: xterm WebGL can leave stale atlas cells for East Asian wide glyphs // on Windows; force browser text rendering before long CJK output paints. isInRange(value, 0x2e80, 0x9fff) || isInRange(value, 0xa960, 0xa97f) || isInRange(value, 0xac00, 0xd7ff) || isInRange(value, 0xd800, 0xdfff) || - isInRange(value, 0xe000, 0xf8ff) || isInRange(value, 0xf900, 0xfaff) || isInRange(value, 0xfe10, 0xfe1f) || isInRange(value, 0xfe30, 0xfe4f) ||