diff --git a/src/main/broker.test.ts b/src/main/broker.test.ts index ba2559de..38cfc4b7 100644 --- a/src/main/broker.test.ts +++ b/src/main/broker.test.ts @@ -492,7 +492,7 @@ describe('BrokerManager local + cloud coexistence', () => { await manager.shutdown() }) - it('keeps repeated PTY chunks when broker events have no identity', async () => { + it('dedupes repeated PTY chunks when broker events have no identity', async () => { const manager = new BrokerManager() const win = createMockWindow() const local = await startLocalWithWindow(manager, win) @@ -512,7 +512,35 @@ describe('BrokerManager local + cloud coexistence', () => { const ptyCalls = (win.webContents.send as ReturnType).mock.calls .filter(([channel]) => channel === 'broker:pty-chunk') - expect(ptyCalls).toHaveLength(2) + expect(ptyCalls).toEqual([['broker:pty-chunk', PROJECT_ID, 'claude-1', 'pong\n']]) + + await manager.shutdown() + }) + + it('keeps distinct PTY chunks when broker events have no identity', async () => { + const manager = new BrokerManager() + const win = createMockWindow() + const local = await startLocalWithWindow(manager, win) + const listener = local.onEvent.mock.calls.at(-1)?.[0] + expect(listener).toBeTypeOf('function') + + listener?.({ + kind: 'worker_stream', + name: 'claude-1', + chunk: 'po' + }) + listener?.({ + kind: 'worker_stream', + name: 'claude-1', + chunk: 'ng\n' + }) + + const ptyCalls = (win.webContents.send as ReturnType).mock.calls + .filter(([channel]) => channel === 'broker:pty-chunk') + expect(ptyCalls).toEqual([ + ['broker:pty-chunk', PROJECT_ID, 'claude-1', 'po'], + ['broker:pty-chunk', PROJECT_ID, 'claude-1', 'ng\n'] + ]) await manager.shutdown() }) @@ -625,6 +653,40 @@ describe('BrokerManager local + cloud coexistence', () => { await manager.shutdown() }) + + it('keeps repeated no-identity PTY chunks after intervening output', async () => { + const manager = new BrokerManager() + const win = createMockWindow() + const local = await startLocalWithWindow(manager, win) + const listener = local.onEvent.mock.calls.at(-1)?.[0] + expect(listener).toBeTypeOf('function') + + listener?.({ + kind: 'worker_stream', + name: 'claude-1', + chunk: '>' + }) + listener?.({ + kind: 'worker_stream', + name: 'claude-1', + chunk: ' ' + }) + listener?.({ + kind: 'worker_stream', + name: 'claude-1', + chunk: '>' + }) + + const ptyCalls = (win.webContents.send as ReturnType).mock.calls + .filter(([channel]) => channel === 'broker:pty-chunk') + expect(ptyCalls).toEqual([ + ['broker:pty-chunk', PROJECT_ID, 'claude-1', '>'], + ['broker:pty-chunk', PROJECT_ID, 'claude-1', ' '], + ['broker:pty-chunk', PROJECT_ID, 'claude-1', '>'] + ]) + + await manager.shutdown() + }) }) describe('BrokerManager spawnAgent CLI preflight', () => { diff --git a/src/main/broker.ts b/src/main/broker.ts index 726f8cba..dcdad0af 100644 --- a/src/main/broker.ts +++ b/src/main/broker.ts @@ -344,6 +344,7 @@ const DEFAULT_RECONCILE_MESSAGE_LIMIT = 50 const MAX_RECONCILE_MESSAGE_LIMIT = 100 const EVENT_STREAM_REBIND_COOLDOWN_MS = 5_000 const PTY_CHUNK_IDENTITY_DEDUPE_TTL_MS = 60_000 +const PTY_CHUNK_CONTENT_DEDUPE_TTL_MS = 1_000 const MAX_PTY_CHUNK_DEDUPE_ENTRIES = 2_000 // After this many consecutive failures to open a PTY input stream, give up on // the WS fast path for that agent briefly and send over HTTP while it cools down. @@ -621,8 +622,12 @@ function isBrokerUnreachableError(err: unknown): boolean { return someInCauseChain(err, (node) => { const code = node.code if (code === 'ECONNREFUSED' || code === 'ECONNRESET') return true + if (code === 'UND_ERR_SOCKET') { + const message = node.message + return typeof message === 'string' && /other side closed|socket closed/i.test(message) + } const message = node.message - return typeof message === 'string' && /ECONNREFUSED|ECONNRESET/.test(message) + return typeof message === 'string' && /ECONNREFUSED|ECONNRESET|UND_ERR_SOCKET/.test(message) }) } @@ -1952,8 +1957,11 @@ export class BrokerManager { private isDuplicatePtyChunk(sessionKey: string, name: string, event: BrokerEvent): boolean { const now = Date.now() for (const [key, seenAt] of this.recentPtyChunks) { + const ttl = key.startsWith('chunk:') + ? PTY_CHUNK_CONTENT_DEDUPE_TTL_MS + : PTY_CHUNK_IDENTITY_DEDUPE_TTL_MS if ( - now - seenAt > PTY_CHUNK_IDENTITY_DEDUPE_TTL_MS || + now - seenAt > ttl || this.recentPtyChunks.size > MAX_PTY_CHUNK_DEDUPE_ENTRIES ) { this.recentPtyChunks.delete(key) @@ -1970,14 +1978,28 @@ export class BrokerManager { ? eventRecord.id : '' const identity = eventId || (seq ? `seq:${seq}` : '') - if (!identity) return false - - const key = `${sessionKey}:${name}:${identity}` + const chunk = typeof eventRecord.chunk === 'string' ? eventRecord.chunk : '' + if (!identity && !chunk) return false + + const contentKeyPrefix = `chunk:${sessionKey}:${name}:` + const key = identity + ? `identity:${sessionKey}:${name}:${identity}` + : `${contentKeyPrefix}${chunk}` + const ttl = identity + ? PTY_CHUNK_IDENTITY_DEDUPE_TTL_MS + : PTY_CHUNK_CONTENT_DEDUPE_TTL_MS const previous = this.recentPtyChunks.get(key) - if (previous !== undefined && now - previous <= PTY_CHUNK_IDENTITY_DEDUPE_TTL_MS) { + if (previous !== undefined && now - previous <= ttl) { return true } + if (!identity) { + for (const previousKey of this.recentPtyChunks.keys()) { + if (previousKey.startsWith(contentKeyPrefix)) { + this.recentPtyChunks.delete(previousKey) + } + } + } this.recentPtyChunks.set(key, now) return false } @@ -2669,6 +2691,7 @@ export class BrokerManager { // and the broker's transient "internal channel closed" while it's // tearing down. Both surface as noise that's not actionable. if (isMissingAgentError(err)) return + if (isBrokerUnreachableError(err)) return const message = toErrorMessage(err) if (/internal channel closed|internal reply dropped/i.test(message)) return console.warn(`[broker] sendInputFireAndForget failed for ${trimmedName}:`, err)