fix(terminal): harden the rendering pipeline against dropped bytes, size drift, and flood lag

.github/workflows/ci.yml

@@ -44,10 +44,13 @@ jobs:
       # Both configs run `npm run build:web` themselves (via test:fidelity / test:redraw),
       # but build once up front so a build break fails fast and is shared by both suites.
       - run: npm run build:web
-      # Fidelity now opens the terminal view explicitly, so duplicate-output
-      # regressions block CI instead of soft-failing.
+      # fidelity GATES: the spec seeds pear-web-initial-tab=agents and selects
+      # the spawned agent, so terminal runtimes mount in the web-mock build.
+      # These six specs are the no-duplication / no-loss fence for the
+      # renderer pipeline — a failure here is a rendering regression.
       - run: npx playwright test --config playwright.fidelity.config.ts
-      # redraw passes today and gates as a hard requirement.
+      # redraw GATES on the drain invariants (drained + finalFrameRendered) —
+      # the typing-lag / parser-backlog fence. Timing metrics stay informational.
       - run: npx playwright test --config playwright.redraw.config.ts
       - name: Upload Playwright traces
         # always() so traces from any failed Playwright gate are captured.

AGENTS.md

@@ -6,7 +6,7 @@ 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 that identity first and use short content-based windows only as a fallback.
+- 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.
 - 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.

playwright.fidelity.config.ts

@@ -2,7 +2,7 @@ import { defineConfig, devices } from '@playwright/test'
 
 export default defineConfig({
   testDir: './tests/playwright',
-  testMatch: /fidelity-no-duplication\.spec\.ts/,
+  testMatch: /(fidelity-no-duplication|rendering-corruption)\.spec\.ts/,
   timeout: 90_000,
   expect: {
     timeout: 15_000

src/main/broker.test.ts

@@ -119,7 +119,7 @@
   }
 
   class HarnessDriverClient {
     static spawn = vi.fn(async (_options: unknown) => {
       const client = createMockClient(state.nextLocalAgents.splice(0))
       state.spawnedClients.push(client)
       return client
@@ -1436,7 +1436,12 @@
     await manager.shutdown()
   })
 
-  it('dedupes repeated PTY chunks when broker events have no identity', async () => {
+  it('delivers identical PTY chunks when broker events have no identity', 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,
+    // and dropping real bytes mangles escape sequences — the stacked
+    // "duplicate line" rendering corruption. Never drop on content alone.
     const manager = new BrokerManager()
     const win = createMockWindow()
     const local = await startLocalWithWindow(manager, win)
@@ -1456,7 +1461,141 @@
 
     const ptyCalls = (win.webContents.send as ReturnType<typeof vi.fn>).mock.calls
       .filter(([channel]) => channel === 'broker:pty-chunk')
-    expect(ptyCalls).toEqual([['broker:pty-chunk', PROJECT_ID, 'claude-1', 'pong\n']])
+    expect(ptyCalls).toEqual([
+      ['broker:pty-chunk', PROJECT_ID, 'claude-1', 'pong\n'],
+      ['broker:pty-chunk', PROJECT_ID, 'claude-1', 'pong\n']
+    ])
+
+    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(() => {})
+    try {
+      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' })
+      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')
+      )
+      expect(blindSpotLogs).toHaveLength(2) // once for claude-1, once for codex-1
+      expect(blindSpotLogs[0][0]).toContain('claude-1')
+      expect(blindSpotLogs[1][0]).toContain('codex-1')
+
+      // Telemetry never suppresses delivery.
+      const ptyCalls = (win.webContents.send as ReturnType<typeof vi.fn>).mock.calls
+        .filter(([channel]) => channel === 'broker:pty-chunk')
+      expect(ptyCalls).toHaveLength(3)
+
+      await manager.shutdown()
+    } finally {
+      infoSpy.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.
+    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 })
+
+    const ptyCalls = (win.webContents.send as ReturnType<typeof vi.fn>).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']
+    ])
+
+    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.
+    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 })
+
+    const ptyCalls = (win.webContents.send as ReturnType<typeof vi.fn>).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']
+    ])
+
+    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.
+    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 })
+
+    const ptyCalls = (win.webContents.send as ReturnType<typeof vi.fn>).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<typeof vi.fn>).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']
+    ])
 
     await manager.shutdown()
   })

src/main/broker.ts

@@ -950,18 +950,6 @@ function isInputStreamClosedError(err: unknown): boolean {
   return (err as { code?: unknown } | null | undefined)?.code === 'input_stream_closed'
 }
 
-function getBrokerEventName(event: BrokerEvent): string | undefined {
-  return 'name' in event && typeof event.name === 'string' && event.name.trim()
-    ? event.name
-    : undefined
-}
-
-function getBrokerEventFrom(event: BrokerEvent): string | undefined {
-  return 'from' in event && typeof event.from === 'string' && event.from.trim()
-    ? event.from
-    : undefined
-}
-
 function withBrokerDetailsTimeout<T>(promise: Promise<T>, label: string): Promise<T> {
   return new Promise((resolve, reject) => {
     const timer = setTimeout(() => {
@@ -1070,11 +1058,6 @@ function brokerConnectionPathCandidates(cwd: string): string[] {
   ]
 }
 
-function resolveBrokerConnectionPath(cwd: string): string {
-  const candidates = brokerConnectionPathCandidates(cwd)
-  return candidates.find((candidate) => existsSync(candidate)) ?? candidates[0]
-}
-
 function toInboundDeliveryMode(mode?: TerminalAttachMode): InboundDeliveryMode {
   return mode === 'drive' ? 'manual_flush' : 'auto_inject'
 }
@@ -1414,7 +1397,7 @@ export class BrokerManager {
   // prune) plus the replay query the renderer rehydrates from. BrokerManager
   // keeps event publishing / IPC fan-out inline and delegates storage here.
   private brokerEventHistory = new BrokerEventHistory()
-  private ptyDeduper = new PtyChunkDeduper()
+  private ptyDeduper = new PtyChunkDeduper({ debugEnabled: isBrokerDebugEnabled })
   // Ingress-validation telemetry: last warn time per malformed `kind`
   // (throttled) and the set of unknown `kind`s already logged (once each).
   private malformedEventWarnedAt = new Map<string, number>()
@@ -1664,7 +1647,10 @@ export class BrokerManager {
 
     const { cwd, name, channels } = session
     const brokerPid = session.client.brokerPid
-    let promise!: Promise<boolean>
+    // `let` + undefined: the closure reads `promise` after its first await,
+    // by which point the assignment below has run; TS can't prove that.
+    let promise: Promise<boolean> | undefined
+    // eslint-disable-next-line prefer-const
     promise = (async () => {
       console.warn(`[broker] Broker for project ${projectId} is unreachable; restarting on a fresh port`)
       this.dropSession(projectId, { disconnectOnly: true })
@@ -2474,6 +2460,7 @@ export class BrokerManager {
   }
 
   private forgetAgentSession(name: string, sessionKey: string): void {
+    this.ptyDeduper.forgetStream(sessionKey, name)
     const sessionKeys = this.agentSessions.get(name)
     if (!sessionKeys) return
     sessionKeys.delete(sessionKey)