From b38f8a8bd9ce6e5978c91ff536e8a0d0480d1434 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Fri, 17 Jul 2026 17:59:37 +0200 Subject: [PATCH] fix(#404): re-key inert pty-dedup on (generation, offset) + content-hash MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit worker_stream PTY chunks never carry seq/event_id — those are ephemeral and excluded from the daemon replay buffer, so the previous seq/event_id identity could never match: the layer was structurally inert (it logged "no seq/event_id" once per stream and suppressed nothing). worker_stream chunks DO carry a cumulative per-worker byte `offset` (present 5/5 on PTY streams), so re-key identity to `${generation}:${offset}` where generation is the broker event-stream generation the delivering listener attached under. This matches the renderer-side guardrail (pty-buffer-store.ts) exactly, which already keyed on (generation, offset)+hash. Invariants preserved (never drop a real byte): - suppress ONLY on full identity (generation + offset) AND content-hash match - never drop on content alone (identical chunks are normal terminal traffic) - never drop on identity alone (repeated offset + different bytes = fresh output) - generation scopes the offset so a fresh stream's low offset can't collide with a stale remembered one → no cross-generation false suppression - offset absent → DELIVER + rate-limited loud console.warn + running counter (no more silently claiming a protection that isn't running) Updates AGENTS.md "Duplicate Event Hardening" to describe the real mechanism. Rewrites pty-dedup.test.ts (suppress/deliver/loud paths) and the broker.test.ts e2e PTY suite from seq-based to offset-based semantics. Co-Authored-By: Claude Opus 4.8 --- .gitignore | 1 + AGENTS.md | 4 +- src/main/broker.test.ts | 119 ++++++++----------- src/main/broker.ts | 2 +- src/main/pty-dedup.test.ts | 235 +++++++++++++++++++++++-------------- src/main/pty-dedup.ts | 186 +++++++++++++++-------------- 6 files changed, 297 insertions(+), 250 deletions(-) diff --git a/.gitignore b/.gitignore index 14a0517b..b8a9fa78 100644 --- a/.gitignore +++ b/.gitignore @@ -1,4 +1,5 @@ node_modules/ +node_modules out/ dist/ bin/relayfile-mount diff --git a/AGENTS.md b/AGENTS.md index d04b75bb..e1dfcdda 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -6,12 +6,12 @@ Pear broker work must treat duplicate delivery as a normal failure mode. Rendere - Make lifecycle operations idempotent. `broker:start`, agent registration, integration notifications, and mount/link setup should return or record whether they actually changed state before triggering side effects. - Coalesce concurrent starts or attaches with keyed in-flight promises. Repeated UI calls for the same project/root/channels should wait on the existing operation instead of starting another broker or event stream. -- Prefer stable event identity over content matching. PTY and broker events should carry `event_id`, `id`, or `seq`; dedupe by identity AND content hash together. Never drop a PTY chunk on content alone (identical consecutive chunks are normal terminal traffic — repeated keystroke echoes, byte-identical TUI repaint frames) and never on seq alone (a daemon restart resets seq; dropping fresh low-seq chunks corrupts the escape-sequence stream — the stacked-duplicate-lines rendering corruption). When in doubt, deliver: a rare duplicate repaints once; a dropped chunk mangles the screen until the next full repaint. +- Prefer stable event identity over content matching, and use the identity fields each event *actually* carries. Chat/broker events carry `event_id`, `id`, or `seq`; dedupe those by identity AND content hash together. `worker_stream` PTY chunks do **not** carry `seq`/`event_id` — those are ephemeral and excluded from the daemon replay buffer, so any dedup keyed on them is structurally inert (it can never match). PTY chunks instead carry a cumulative per-worker byte `offset`; dedupe them by `(generation, offset)` identity AND content hash, where `generation` is the broker event-stream generation the delivering listener was attached under (it scopes the offset so a fresh worker stream's low offset can't collide with a stale remembered one). Never drop a PTY chunk on content alone (identical consecutive chunks are normal terminal traffic — repeated keystroke echoes, byte-identical TUI repaint frames) and never on identity alone (a repeated offset with different bytes is fresh output, not a replay — dropping it corrupts the escape-sequence stream into the stacked-duplicate-lines rendering corruption). When correlation metadata is absent (no `offset`), deliver every chunk and log the blind spot loudly — do not silently claim a protection that isn't running. When in doubt, deliver: a rare duplicate repaints once; a dropped chunk mangles the screen until the next full repaint. - Scope live event listeners with a generation token when reconnecting or refreshing streams. Stale callbacks from an older listener must bail before publishing IPC events. - Keep PTY stream delivery separately guarded in main and renderer code. Main should suppress duplicate `worker_stream` chunks before `broker:pty-chunk`; renderer buffers should tolerate repeated chunk metadata as a final guardrail. - Do not post integration or launch metadata on reused broker sessions. Notify agents only after a real broker start, reconnect, or state transition, and make repeated payloads no-ops when possible. - Add regression tests when touching broker start, event streaming, PTY buffering, spawned personas, or integration notifications. Include duplicate/replay cases, not just the happy path. -- Add low-noise telemetry for suppressed duplicates and missing event identity so replay issues are visible without flooding the terminal. +- Add low-noise telemetry for suppressed duplicates and a rate-limited *loud* warning (with a running count) when correlation metadata is missing, so both real replays and dedup blind spots are visible without flooding the terminal. ## Terminal Screen Convergence diff --git a/src/main/broker.test.ts b/src/main/broker.test.ts index 43ccbaee..e63799e8 100644 --- a/src/main/broker.test.ts +++ b/src/main/broker.test.ts @@ -2039,24 +2039,27 @@ exit 2 await manager.shutdown() }) - it('keeps legitimate repeated PTY chunks with different broker sequences', async () => { + it('keeps legitimate repeated PTY chunks with different offsets', 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') + // Byte-identical output at advancing offsets is normal terminal traffic + // (a repeated keystroke echo). Distinct offsets are distinct identities; + // both must render — never drop on content alone. listener?.({ kind: 'worker_stream', name: 'claude-1', chunk: 'pong\n', - seq: 23 + offset: 23 }) listener?.({ kind: 'worker_stream', name: 'claude-1', chunk: 'pong\n', - seq: 24 + offset: 24 }) const ptyCalls = (win.webContents.send as ReturnType).mock.calls @@ -2066,10 +2069,10 @@ exit 2 await manager.shutdown() }) - it('delivers identical PTY chunks when broker events have no identity', async () => { + it('delivers identical PTY chunks when broker events have no offset', async () => { // Identical consecutive chunks are NORMAL terminal traffic (the same // keystroke echoed twice, byte-identical TUI repaint frames). Without an - // identity there is no way to tell a transport replay from real output, + // offset there is no way to tell a transport replay from real output, // and dropping real bytes mangles escape sequences — the stacked // "duplicate line" rendering corruption. Never drop on content alone. const manager = new BrokerManager() @@ -2099,11 +2102,11 @@ exit 2 await manager.shutdown() }) - it('logs the identity-less PTY stream blind spot once per stream while delivering', async () => { - // AGENTS.md: low-noise telemetry for missing event identity. One line per - // stream, not per chunk — an identity-less stream hits the branch on - // every chunk and per-chunk logging would flood. - const infoSpy = vi.spyOn(console, 'info').mockImplementation(() => {}) + it('loudly warns about the offset-less PTY stream blind spot per stream while delivering', async () => { + // AGENTS.md: loud, rate-limited telemetry when correlation metadata is + // absent. Without an offset the deduper is blind — it warns (at most once + // per interval per stream, carrying a running count) but always delivers. + const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {}) try { const manager = new BrokerManager() const win = createMockWindow() @@ -2115,10 +2118,12 @@ exit 2 listener?.({ kind: 'worker_stream', name: 'claude-1', chunk: 'two\n' }) listener?.({ kind: 'worker_stream', name: 'codex-1', chunk: 'three\n' }) - const blindSpotLogs = infoSpy.mock.calls.filter(([first]) => - typeof first === 'string' && first.includes('no seq/event_id') + const blindSpotLogs = warnSpy.mock.calls.filter(([first]) => + typeof first === 'string' && first.includes('no offset') ) - expect(blindSpotLogs).toHaveLength(2) // once for claude-1, once for codex-1 + // claude-1's two chunks land in one rate-limit interval → one warn; + // codex-1 is a distinct stream → its own warn. + expect(blindSpotLogs).toHaveLength(2) expect(blindSpotLogs[0][0]).toContain('claude-1') expect(blindSpotLogs[1][0]).toContain('codex-1') @@ -2129,108 +2134,78 @@ exit 2 await manager.shutdown() } finally { - infoSpy.mockRestore() + warnSpy.mockRestore() } }) - it('delivers chunks whose seq repeats with different bytes (daemon seq reset)', async () => { - // A daemon restart resets its event seq counter. The replacement stream - // reuses seq numbers we have already seen — but the bytes are new output - // and must render. Only an (identity AND content) match is a replay. + it('delivers chunks whose offset repeats with different bytes (never drop on identity alone)', async () => { + // An offset that repeats with different bytes is an anomaly, not a replay + // (offset is meant to be unique within a generation+worker). The bytes are + // real output and must render — only an (identity AND content) match drops. 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: 'before restart\n', seq: 90 }) - listener?.({ kind: 'worker_stream', name: 'claude-1', chunk: 'after restart\n', seq: 90 }) + listener?.({ kind: 'worker_stream', name: 'claude-1', chunk: 'before\n', offset: 90 }) + listener?.({ kind: 'worker_stream', name: 'claude-1', chunk: 'after\n', offset: 90 }) 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', 'before restart\n'], - ['broker:pty-chunk', PROJECT_ID, 'claude-1', 'after restart\n'] - ]) + expect(ptyCalls.map((call) => call[3])).toEqual(['before\n', 'after\n']) await manager.shutdown() }) - it('delivers low-seq chunks after a watermark reset instead of dropping a window', async () => { - // After the daemon restarts, fresh chunks arrive with seqs far below the - // old watermark. They must all be delivered immediately — the previous - // TTL-based dedup dropped them for up to 60s, losing repaint bytes. + it('drops a delayed replay of an earlier offset with identical bytes', async () => { + // A double-emit re-delivers recent chunks: same offset, same bytes, + // possibly long after first delivery. These are the one provable-duplicate + // case — full identity (generation + offset) AND content match. 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: 'old-high\n', seq: 5_000 }) - listener?.({ kind: 'worker_stream', name: 'claude-1', chunk: 'fresh-1\n', seq: 1 }) - listener?.({ kind: 'worker_stream', name: 'claude-1', chunk: 'fresh-2\n', seq: 2 }) + listener?.({ kind: 'worker_stream', name: 'claude-1', chunk: 'a\n', offset: 10 }) + listener?.({ kind: 'worker_stream', name: 'claude-1', chunk: 'b\n', offset: 11 }) + // The tail of the stream is re-emitted. + listener?.({ kind: 'worker_stream', name: 'claude-1', chunk: 'a\n', offset: 10 }) + listener?.({ kind: 'worker_stream', name: 'claude-1', chunk: 'b\n', offset: 11 }) 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', 'old-high\n'], - ['broker:pty-chunk', PROJECT_ID, 'claude-1', 'fresh-1\n'], - ['broker:pty-chunk', PROJECT_ID, 'claude-1', 'fresh-2\n'] - ]) + expect(ptyCalls.map((call) => call[3])).toEqual(['a\n', 'b\n']) await manager.shutdown() }) - it('drops a delayed replay of an older seq with identical bytes', async () => { - // Rebind replay re-delivers recent events: same seq, same bytes, possibly - // long after first delivery. These are the one provable-duplicate case. + it('tracks PTY dedup identity per agent stream', async () => { + // Two agents share the session event stream; one agent's offset window must + // not swallow the other's chunks even at a colliding offset. 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: 'a\n', seq: 10 }) - listener?.({ kind: 'worker_stream', name: 'claude-1', chunk: 'b\n', seq: 11 }) - // Rebind replays the tail of the stream. - listener?.({ kind: 'worker_stream', name: 'claude-1', chunk: 'a\n', seq: 10 }) - listener?.({ kind: 'worker_stream', name: 'claude-1', chunk: 'b\n', seq: 11 }) + listener?.({ kind: 'worker_stream', name: 'claude-1', chunk: 'one\n', offset: 100 }) + listener?.({ kind: 'worker_stream', name: 'codex-1', chunk: 'one\n', offset: 100 }) + listener?.({ kind: 'worker_stream', name: 'codex-1', chunk: 'two\n', offset: 101 }) 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', 'a\n'], - ['broker:pty-chunk', PROJECT_ID, 'claude-1', 'b\n'] - ]) - - await manager.shutdown() - }) - - it('tracks PTY dedup watermarks per agent stream', async () => { - // Two agents share the session event-stream seq space; one agent's - // watermark must not swallow the other's chunks. - 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: 'one\n', seq: 100 }) - listener?.({ kind: 'worker_stream', name: 'codex-1', chunk: 'one\n', seq: 100 }) - listener?.({ kind: 'worker_stream', name: 'codex-1', chunk: 'two\n', seq: 101 }) - - 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', 'one\n'], - ['broker:pty-chunk', PROJECT_ID, 'codex-1', 'one\n'], - ['broker:pty-chunk', PROJECT_ID, 'codex-1', 'two\n'] + expect(ptyCalls.map((call) => [call[2], call[3]])).toEqual([ + ['claude-1', 'one\n'], + ['codex-1', 'one\n'], + ['codex-1', 'two\n'] ]) await manager.shutdown() }) - it('keeps distinct PTY chunks when broker events have no identity', async () => { + it('keeps distinct PTY chunks when broker events have no offset', async () => { const manager = new BrokerManager() const win = createMockWindow() const local = await startLocalWithWindow(manager, win) diff --git a/src/main/broker.ts b/src/main/broker.ts index ea6eeaab..6ffe6041 100644 --- a/src/main/broker.ts +++ b/src/main/broker.ts @@ -2510,7 +2510,7 @@ export class BrokerManager { 'name' in event && typeof event.name === 'string' && 'chunk' in event && typeof event.chunk === 'string' ) { - if (this.ptyDeduper.isDuplicatePtyChunk(sessionKey, event.name, event)) { + if (this.ptyDeduper.isDuplicatePtyChunk(sessionKey, event.name, event, eventStreamGeneration)) { return } const targetWindow = this.windowForSession(sessionKey, win) diff --git a/src/main/pty-dedup.test.ts b/src/main/pty-dedup.test.ts index dea27910..2dcb5ada 100644 --- a/src/main/pty-dedup.test.ts +++ b/src/main/pty-dedup.test.ts @@ -2,117 +2,174 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import type { BrokerEvent } from '@agent-relay/harness-driver' import { PtyChunkDeduper } from './pty-dedup' -// The deduper reads chunk/identity fields dynamically off the event; the suite -// hands it loosely-shaped worker_stream records cast to BrokerEvent, mirroring -// how broker.ts forwards raw SDK events. +// The deduper reads chunk/offset off the event and takes the broker +// event-stream `generation` from the caller; the suite hands it loosely-shaped +// worker_stream records cast to BrokerEvent, mirroring how broker.ts forwards +// raw SDK events. Identity = `${generation}:${offset}`. function chunkEvent(fields: Record): BrokerEvent { return { kind: 'worker_stream', stream: 'stdout', ...fields } as unknown as BrokerEvent } -// Hardened invariants (AGENTS.md "Duplicate Event Hardening"): drop ONLY on -// identity (seq/event_id) AND content-hash match within the replay window. -// Never on content alone (identical consecutive chunks are normal terminal -// traffic), never on identity alone (a daemon restart resets seq). The +// Hardened invariants (AGENTS.md "Duplicate Event Hardening"): drop ONLY on a +// full identity (generation + offset) AND content-hash match within the replay +// window. Never on content alone (identical consecutive chunks are normal +// terminal traffic), never when correlation metadata (offset) is absent. The // broker.test.ts PTY suite covers the same invariants end-to-end through the // IPC path; this suite pins the deduper unit. describe('PtyChunkDeduper', () => { let infoSpy: ReturnType + let warnSpy: ReturnType beforeEach(() => { infoSpy = vi.spyOn(console, 'info').mockImplementation(() => {}) + warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {}) }) afterEach(() => { infoSpy.mockRestore() + warnSpy.mockRestore() }) - it('suppresses a replay with the same identity and the same bytes', () => { - const deduper = new PtyChunkDeduper() - const event = chunkEvent({ name: 'claude-1', chunk: 'pong\n', event_id: 'e1' }) - expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', event)).toBe(false) - expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', event)).toBe(true) + describe('suppress path', () => { + it('suppresses a replay with the same generation+offset and the same bytes', () => { + const deduper = new PtyChunkDeduper() + const event = chunkEvent({ name: 'claude-1', chunk: 'pong\n', offset: 42 }) + expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', event, 1)).toBe(false) + expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', event, 1)).toBe(true) + }) + + it('drops a delayed replay of an earlier offset with identical bytes', () => { + const deduper = new PtyChunkDeduper() + expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', chunkEvent({ name: 'claude-1', chunk: 'a\n', offset: 10 }), 1)).toBe(false) + expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', chunkEvent({ name: 'claude-1', chunk: 'b\n', offset: 11 }), 1)).toBe(false) + // A double-emit replays the tail with the same offsets and bytes. + expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', chunkEvent({ name: 'claude-1', chunk: 'a\n', offset: 10 }), 1)).toBe(true) + expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', chunkEvent({ name: 'claude-1', chunk: 'b\n', offset: 11 }), 1)).toBe(true) + }) }) - it('keeps chunks that share content but carry distinct broker sequences', () => { - const deduper = new PtyChunkDeduper() - expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', chunkEvent({ name: 'claude-1', chunk: 'tick\n', seq: 1 }))).toBe(false) - expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', chunkEvent({ name: 'claude-1', chunk: 'tick\n', seq: 2 }))).toBe(false) + describe('deliver path', () => { + it('keeps chunks that share content but carry distinct offsets', () => { + const deduper = new PtyChunkDeduper() + expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', chunkEvent({ name: 'claude-1', chunk: 'tick\n', offset: 1 }), 1)).toBe(false) + expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', chunkEvent({ name: 'claude-1', chunk: 'tick\n', offset: 2 }), 1)).toBe(false) + }) + + it('delivers a chunk whose offset repeats with different bytes (never drop on identity alone)', () => { + const deduper = new PtyChunkDeduper() + expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', chunkEvent({ name: 'claude-1', chunk: 'before\n', offset: 90 }), 1)).toBe(false) + expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', chunkEvent({ name: 'claude-1', chunk: 'after\n', offset: 90 }), 1)).toBe(false) + }) + + it('does not suppress across a generation bump even at the same offset with the same bytes', () => { + // offset is cumulative WITHIN a (generation, worker); scoping identity by + // generation prevents a fresh stream's low offset from colliding with a + // stale remembered one. When in doubt: DELIVER. + const deduper = new PtyChunkDeduper() + const event = chunkEvent({ name: 'claude-1', chunk: 'pong\n', offset: 42 }) + expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', event, 1)).toBe(false) + expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', event, 2)).toBe(false) + }) + + it('tracks replay windows per session and agent stream', () => { + const deduper = new PtyChunkDeduper() + const event = chunkEvent({ name: 'claude-1', chunk: 'pong\n', offset: 100 }) + expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', event, 1)).toBe(false) + expect(deduper.isDuplicatePtyChunk('proj', 'claude-2', chunkEvent({ name: 'claude-2', chunk: 'pong\n', offset: 100 }), 1)).toBe(false) + expect(deduper.isDuplicatePtyChunk('other', 'claude-1', event, 1)).toBe(false) + }) + + it('forgetStream resets the replay window so a respawned agent replays cleanly', () => { + const deduper = new PtyChunkDeduper() + const event = chunkEvent({ name: 'claude-1', chunk: 'pong\n', offset: 5 }) + expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', event, 1)).toBe(false) + deduper.forgetStream('proj', 'claude-1') + expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', event, 1)).toBe(false) + }) }) - it('always delivers identity-less chunks, even byte-identical ones', () => { - // Without identity a transport replay is indistinguishable from real - // repeated output (same keystroke echoed twice, byte-identical repaint - // frames). Dropping real bytes mangles escape sequences — never dedupe - // on content alone. - const deduper = new PtyChunkDeduper() - const event = chunkEvent({ name: 'claude-1', chunk: 'pong\n' }) - expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', event)).toBe(false) - expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', event)).toBe(false) + describe('loud (no-offset) path', () => { + it('always delivers offset-less chunks, even byte-identical ones', () => { + // Without an offset a transport replay is indistinguishable from real + // repeated output (same keystroke echoed twice, byte-identical repaint + // frames). Dropping real bytes mangles escape sequences — never dedupe + // on content alone. + const deduper = new PtyChunkDeduper() + const event = chunkEvent({ name: 'claude-1', chunk: 'pong\n' }) + expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', event, 1)).toBe(false) + expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', event, 1)).toBe(false) + }) + + it('loudly warns the first time a stream delivers an offset-less chunk, with a running count', () => { + const deduper = new PtyChunkDeduper() + deduper.isDuplicatePtyChunk('proj', 'claude-1', chunkEvent({ name: 'claude-1', chunk: 'one\n' }), 1) + const warns = warnSpy.mock.calls.filter(([first]: unknown[]) => + typeof first === 'string' && first.includes('no offset') + ) + expect(warns).toHaveLength(1) + expect(warns[0][0]).toContain('claude-1') + expect(warns[0][0]).toContain('1 delivered') + }) + + it('rate-limits the loud warning per stream but keeps counting', () => { + // Deterministic clock: all chunks land inside one interval, so only the + // first warns, but every one increments the running counter. + let clock = 1_000_000 + const deduper = new PtyChunkDeduper({ now: () => clock }) + for (let i = 0; i < 5; i += 1) { + deduper.isDuplicatePtyChunk('proj', 'claude-1', chunkEvent({ name: 'claude-1', chunk: `x${i}\n` }), 1) + clock += 1_000 // < 60s interval + } + const warns = warnSpy.mock.calls.filter(([first]: unknown[]) => + typeof first === 'string' && first.includes('no offset') + ) + expect(warns).toHaveLength(1) + + // Cross the interval → warn again, now reporting the full running count. + clock += 60_000 + deduper.isDuplicatePtyChunk('proj', 'claude-1', chunkEvent({ name: 'claude-1', chunk: 'late\n' }), 1) + const warnsAfter = warnSpy.mock.calls.filter(([first]: unknown[]) => + typeof first === 'string' && first.includes('no offset') + ) + expect(warnsAfter).toHaveLength(2) + expect(warnsAfter[1][0]).toContain('6 delivered') + }) + + it('warns independently per stream', () => { + const deduper = new PtyChunkDeduper() + deduper.isDuplicatePtyChunk('proj', 'claude-1', chunkEvent({ name: 'claude-1', chunk: 'one\n' }), 1) + deduper.isDuplicatePtyChunk('proj', 'codex-1', chunkEvent({ name: 'codex-1', chunk: 'two\n' }), 1) + const warns = warnSpy.mock.calls.filter(([first]: unknown[]) => + typeof first === 'string' && first.includes('no offset') + ) + expect(warns).toHaveLength(2) + expect(warns[0][0]).toContain('claude-1') + expect(warns[1][0]).toContain('codex-1') + }) }) - it('logs the identity-less blind spot once per stream', () => { - const deduper = new PtyChunkDeduper() - deduper.isDuplicatePtyChunk('proj', 'claude-1', chunkEvent({ name: 'claude-1', chunk: 'one\n' })) - deduper.isDuplicatePtyChunk('proj', 'claude-1', chunkEvent({ name: 'claude-1', chunk: 'two\n' })) - deduper.isDuplicatePtyChunk('proj', 'codex-1', chunkEvent({ name: 'codex-1', chunk: 'three\n' })) - - const blindSpotLogs = infoSpy.mock.calls.filter(([first]: unknown[]) => - typeof first === 'string' && first.includes('no seq/event_id') - ) - expect(blindSpotLogs).toHaveLength(2) - expect(blindSpotLogs[0][0]).toContain('claude-1') - expect(blindSpotLogs[1][0]).toContain('codex-1') - }) - - it('delivers a chunk whose seq repeats with different bytes (daemon seq reset)', () => { - const deduper = new PtyChunkDeduper() - expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', chunkEvent({ name: 'claude-1', chunk: 'before restart\n', seq: 90 }))).toBe(false) - expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', chunkEvent({ name: 'claude-1', chunk: 'after restart\n', seq: 90 }))).toBe(false) - }) - - it('delivers low-seq chunks after a watermark reset instead of dropping a window', () => { - const deduper = new PtyChunkDeduper() - expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', chunkEvent({ name: 'claude-1', chunk: 'old-high\n', seq: 5_000 }))).toBe(false) - expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', chunkEvent({ name: 'claude-1', chunk: 'fresh-1\n', seq: 1 }))).toBe(false) - expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', chunkEvent({ name: 'claude-1', chunk: 'fresh-2\n', seq: 2 }))).toBe(false) - }) - - it('drops a delayed replay of an older seq with identical bytes', () => { - const deduper = new PtyChunkDeduper() - expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', chunkEvent({ name: 'claude-1', chunk: 'a\n', seq: 10 }))).toBe(false) - expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', chunkEvent({ name: 'claude-1', chunk: 'b\n', seq: 11 }))).toBe(false) - // Rebind replays the tail of the stream. - expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', chunkEvent({ name: 'claude-1', chunk: 'a\n', seq: 10 }))).toBe(true) - expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', chunkEvent({ name: 'claude-1', chunk: 'b\n', seq: 11 }))).toBe(true) - }) - - it('tracks replay windows per session and agent stream', () => { - const deduper = new PtyChunkDeduper() - const event = chunkEvent({ name: 'claude-1', chunk: 'pong\n', seq: 100 }) - expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', event)).toBe(false) - expect(deduper.isDuplicatePtyChunk('proj', 'claude-2', chunkEvent({ name: 'claude-2', chunk: 'pong\n', seq: 100 }))).toBe(false) - expect(deduper.isDuplicatePtyChunk('other', 'claude-1', event)).toBe(false) - }) - - it('prefers event_id over seq for identity', () => { - const deduper = new PtyChunkDeduper() - expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', chunkEvent({ name: 'claude-1', chunk: 'x\n', event_id: 'e1', seq: 1 }))).toBe(false) - // Same event_id, same bytes, different seq: still the same event. - expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', chunkEvent({ name: 'claude-1', chunk: 'x\n', event_id: 'e1', seq: 2 }))).toBe(true) - }) - - it('treats an event with neither identity nor chunk as never-duplicate', () => { - const deduper = new PtyChunkDeduper() - const event = chunkEvent({ name: 'claude-1' }) - expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', event)).toBe(false) - expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', event)).toBe(false) - }) - - it('forgetStream resets the replay window so a respawned agent replays cleanly', () => { - const deduper = new PtyChunkDeduper() - const event = chunkEvent({ name: 'claude-1', chunk: 'pong\n', seq: 5 }) - expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', event)).toBe(false) - deduper.forgetStream('proj', 'claude-1') - expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', event)).toBe(false) + describe('debug replay telemetry', () => { + it('logs a suppressed replay only when debug is enabled, with a running total', () => { + const deduper = new PtyChunkDeduper({ debugEnabled: () => true }) + const event = chunkEvent({ name: 'claude-1', chunk: 'pong\n', offset: 7 }) + deduper.isDuplicatePtyChunk('proj', 'claude-1', event, 1) + deduper.isDuplicatePtyChunk('proj', 'claude-1', event, 1) + const drops = infoSpy.mock.calls.filter(([first]: unknown[]) => + typeof first === 'string' && first.includes('suppressed replayed PTY chunk') + ) + expect(drops).toHaveLength(1) + expect(drops[0][1]).toMatchObject({ name: 'claude-1', suppressedTotal: 1 }) + }) + + it('stays silent about suppressed replays when debug is disabled', () => { + const deduper = new PtyChunkDeduper() + const event = chunkEvent({ name: 'claude-1', chunk: 'pong\n', offset: 7 }) + deduper.isDuplicatePtyChunk('proj', 'claude-1', event, 1) + expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', event, 1)).toBe(true) + const drops = infoSpy.mock.calls.filter(([first]: unknown[]) => + typeof first === 'string' && first.includes('suppressed replayed PTY chunk') + ) + expect(drops).toHaveLength(0) + }) }) }) diff --git a/src/main/pty-dedup.ts b/src/main/pty-dedup.ts index 85c57423..0d581d8c 100644 --- a/src/main/pty-dedup.ts +++ b/src/main/pty-dedup.ts @@ -1,29 +1,48 @@ import type { BrokerEvent } from '@agent-relay/harness-driver' -// PTY chunk replay-dedupe. An event-stream rebind reconnects with -// `sinceSeq = lastEventSeq`, so the daemon may re-deliver the most recent -// events; those replays carry the SAME seq and the SAME bytes as the first -// delivery. We remember (identity → content hash) for the last -// PTY_CHUNK_REPLAY_WINDOW chunks per agent stream and drop a chunk only when -// BOTH match — a chunk is never dropped on content alone (consecutive -// identical chunks are normal terminal traffic: repeated keystroke echoes, -// byte-identical TUI repaint frames) and never on seq alone (a daemon -// restart resets seq; dropping fresh low-seq chunks for a TTL corrupts the -// stream — dropped repaint bytes are exactly the "stacked duplicate lines / -// stray escape-sequence fragments" rendering corruption class). See -// AGENTS.md "Duplicate Event Hardening". +// PTY chunk replay-dedupe, keyed on the correlation metadata that relay +// `worker_stream` chunks ACTUALLY carry: the cumulative per-worker byte +// `offset` (present 5/5 on PTY streams; absent only on headless workers and +// pre-offset brokers) plus the broker event-stream `generation` supplied by +// the caller. seq/event_id are NOT on worker_stream chunks — they are excluded +// from the daemon replay buffer, so an earlier identity keyed on them was +// structurally inert (it could never match) and this layer said so once and +// suppressed nothing. See AGENTS.md "Duplicate Event Hardening". +// +// Identity = `${generation}:${offset}`. We remember (identity → content hash) +// for the last PTY_CHUNK_REPLAY_WINDOW chunks per agent stream and drop a chunk +// only when BOTH the identity AND the content hash match a remembered entry. +// Invariants (never violated): +// - Never drop on content alone. Consecutive byte-identical chunks are normal +// terminal traffic (repeated keystroke echoes, byte-identical TUI repaint +// frames); dropping them mangles the screen. A duplicate repaint is safe; +// a dropped chunk corrupts until the next full repaint. +// - Never drop when correlation metadata is absent. No `offset` → no identity +// → we cannot tell a transport replay from genuinely repeated output, so we +// DELIVER and surface the blind spot with a rate-limited loud log + counter. +// - `generation` is part of the identity so an `offset` that resets across a +// new worker stream / generation (offset is cumulative WITHIN a worker's +// lifetime) cannot collide with a stale remembered offset and cause a +// false suppression. When in doubt: DELIVER. const PTY_CHUNK_REPLAY_WINDOW = 512 const MAX_PTY_STREAM_DEDUPE_STATES = 2_000 +// Rate-limit for the loud "no offset — dedup is blind" warning, per stream. +// The first blind chunk warns immediately; further blind chunks on the same +// stream warn at most once per interval, always carrying the running count so +// a burst is visible from a single line without flooding the terminal. +const MISSING_OFFSET_LOG_INTERVAL_MS = 60_000 interface PtyStreamDedupeState { - // Highest numeric seq delivered for this agent stream. -Infinity until the - // first numeric seq arrives. - lastSeq: number - // identity key (seq or event_id) → content hash for the most recent + // identity key (`${generation}:${offset}`) → content hash for the most recent // PTY_CHUNK_REPLAY_WINDOW chunks, insertion-ordered for FIFO eviction. recentByIdentity: Map // Telemetry: replayed chunks suppressed for this stream. suppressedReplays: number + // Telemetry: chunks delivered with no `offset` (dedup blind for them). + identityLessDelivered: number + // Timestamp of the last loud missing-offset warning for this stream (0 until + // the first one), for rate limiting. + lastMissingOffsetLogAt: number } // Cheap, allocation-free content fingerprint (FNV-1a 32-bit + length). A @@ -41,79 +60,69 @@ function ptyChunkHash(chunk: string): string { export class PtyChunkDeduper { private ptyStreamDedupe = new Map() - // Streams already flagged for identity-less PTY chunks (one log per stream). - private ptyStreamsMissingIdentity = new Set() private readonly debugEnabled: () => boolean + private readonly now: () => number - constructor(opts: { debugEnabled?: () => boolean } = {}) { + constructor(opts: { debugEnabled?: () => boolean; now?: () => number } = {}) { this.debugEnabled = opts.debugEnabled ?? (() => false) + this.now = opts.now ?? (() => Date.now()) } // Drop a PTY chunk ONLY when it is provably a replay of one we already - // delivered: same identity (seq or event_id) AND same content hash, within - // the remembered window. Everything else is delivered — losing real bytes - // mangles escape sequences and corrupts the screen until the next full + // delivered: same identity (`${generation}:${offset}`) AND same content hash, + // within the remembered window. Everything else is delivered — losing real + // bytes mangles escape sequences and corrupts the screen until the next full // repaint, which is strictly worse than rendering a rare duplicate. - isDuplicatePtyChunk(sessionKey: string, name: string, event: BrokerEvent): boolean { + // + // `generation` is the broker event-stream generation for the delivering + // listener (BrokerManager.eventStreamGeneration). It scopes the offset so a + // fresh worker stream (offset restarts near zero) can never be mistaken for a + // replay of an earlier stream that happened to reach the same offset. + isDuplicatePtyChunk( + sessionKey: string, + name: string, + event: BrokerEvent, + generation?: number + ): boolean { const eventRecord = event as Record - const seqRaw = eventRecord.seq - const numericSeq = - typeof seqRaw === 'number' && Number.isFinite(seqRaw) ? seqRaw : undefined - const seqIdentity = - numericSeq !== undefined - ? `seq:${numericSeq}` - : typeof seqRaw === 'string' && seqRaw.trim() - ? `seq:${seqRaw}` - : '' - const eventId = typeof eventRecord.event_id === 'string' && eventRecord.event_id - ? eventRecord.event_id - : typeof eventRecord.id === 'string' && eventRecord.id - ? eventRecord.id - : '' - const identity = eventId ? `id:${eventId}` : seqIdentity - // No identity → we cannot distinguish a transport replay from genuinely - // repeated output (e.g. the same keystroke echoed twice, byte-identical - // repaint frames). Always deliver, but surface the blind spot once per - // stream (AGENTS.md low-noise telemetry: per-chunk logging would flood — - // an identity-less stream hits this branch on every chunk). - if (!identity) { - const streamKey = `${sessionKey}:${name}` - if (!this.ptyStreamsMissingIdentity.has(streamKey)) { - if (this.ptyStreamsMissingIdentity.size >= MAX_PTY_STREAM_DEDUPE_STATES) { - this.ptyStreamsMissingIdentity.clear() - } - this.ptyStreamsMissingIdentity.add(streamKey) - console.info( - `[broker] PTY chunks for ${name} carry no seq/event_id — replay dedup is blind for this stream` + const offsetRaw = eventRecord.offset + const offset = + typeof offsetRaw === 'number' && Number.isFinite(offsetRaw) ? offsetRaw : undefined + const streamKey = `${sessionKey}:${name}` + + // No offset → no correlation metadata → we cannot distinguish a transport + // replay from genuinely repeated output. Always deliver, but surface the + // blind spot with a rate-limited loud log + running counter (per stream). + if (offset === undefined) { + const state = this.ensureState(streamKey) + state.identityLessDelivered += 1 + const at = this.now() + if ( + state.lastMissingOffsetLogAt === 0 || + at - state.lastMissingOffsetLogAt >= MISSING_OFFSET_LOG_INTERVAL_MS + ) { + state.lastMissingOffsetLogAt = at + console.warn( + `[broker] PTY chunks for ${name} carry no offset — replay dedup is blind for this stream ` + + `(${state.identityLessDelivered} delivered without correlation metadata)` ) } return false } + const identity = `${generation ?? 'g?'}:${offset}` const chunk = typeof eventRecord.chunk === 'string' ? eventRecord.chunk : '' const hash = ptyChunkHash(chunk) - const streamKey = `${sessionKey}:${name}` - let state = this.ptyStreamDedupe.get(streamKey) - if (!state) { - if (this.ptyStreamDedupe.size >= MAX_PTY_STREAM_DEDUPE_STATES) { - const oldest = this.ptyStreamDedupe.keys().next().value - if (oldest !== undefined) this.ptyStreamDedupe.delete(oldest) - } - state = { - lastSeq: Number.NEGATIVE_INFINITY, - recentByIdentity: new Map(), - suppressedReplays: 0 - } - this.ptyStreamDedupe.set(streamKey, state) - } + const state = this.ensureState(streamKey) const remembered = state.recentByIdentity.get(identity) if (remembered !== undefined) { if (remembered === hash) { + // Full identity + content match within the window: a provable replay. state.suppressedReplays += 1 - // Low-noise replay telemetry (AGENTS.md): only the debug channel - // sees per-drop logs; the running count rides along so a burst is - // visible from a single line. + // Low-noise replay telemetry (AGENTS.md): only the debug channel sees + // per-drop logs; the running count rides along so a burst is visible + // from a single line. if (this.debugEnabled()) { console.info('[broker] suppressed replayed PTY chunk', { name, @@ -123,23 +132,11 @@ export class PtyChunkDeduper { } return true } - // Same identity, different bytes: the daemon restarted and its seq - // counter reset. This is a fresh stream — forget the stale window so - // the new seq range isn't misjudged against the old one. - state.lastSeq = Number.NEGATIVE_INFINITY - state.recentByIdentity.clear() - } else if (numericSeq !== undefined && numericSeq <= state.lastSeq) { - // Seq at or below the watermark but outside the remembered window: - // either a replay deeper than the window (rare) or a daemon seq reset. - // Both must be delivered — dropping fresh bytes corrupts the screen, - // while a deep-replay duplicate at worst repaints content once. - state.lastSeq = Number.NEGATIVE_INFINITY - state.recentByIdentity.clear() + // Same identity, DIFFERENT bytes. offset is meant to be unique within a + // (generation, worker) — this is an anomaly, not a replay. DELIVER (never + // drop real bytes) and refresh the remembered hash to the latest. } - if (numericSeq !== undefined && numericSeq > state.lastSeq) { - state.lastSeq = numericSeq - } state.recentByIdentity.set(identity, hash) if (state.recentByIdentity.size > PTY_CHUNK_REPLAY_WINDOW) { const oldest = state.recentByIdentity.keys().next().value @@ -148,10 +145,27 @@ export class PtyChunkDeduper { return false } + private ensureState(streamKey: string): PtyStreamDedupeState { + let state = this.ptyStreamDedupe.get(streamKey) + if (!state) { + if (this.ptyStreamDedupe.size >= MAX_PTY_STREAM_DEDUPE_STATES) { + const oldest = this.ptyStreamDedupe.keys().next().value + if (oldest !== undefined) this.ptyStreamDedupe.delete(oldest) + } + state = { + recentByIdentity: new Map(), + suppressedReplays: 0, + identityLessDelivered: 0, + lastMissingOffsetLogAt: 0 + } + this.ptyStreamDedupe.set(streamKey, state) + } + return state + } + // Drop per-stream state when an agent session goes away so a respawned - // agent (fresh PTY, fresh seq space) starts with a clean window. + // agent (fresh PTY, fresh offset space) starts with a clean window. forgetStream(sessionKey: string, name: string): void { this.ptyStreamDedupe.delete(`${sessionKey}:${name}`) - this.ptyStreamsMissingIdentity.delete(`${sessionKey}:${name}`) } }