diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 557466d7..9270921f 100644 --- a/.github/workflows/ci.yml +++ b/.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. diff --git a/AGENTS.md b/AGENTS.md index 8787e255..d21e150e 100644 --- a/AGENTS.md +++ b/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. diff --git a/playwright.fidelity.config.ts b/playwright.fidelity.config.ts index 4515da39..87210a26 100644 --- a/playwright.fidelity.config.ts +++ b/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 diff --git a/src/main/broker.test.ts b/src/main/broker.test.ts index 428bb263..8733d4b5 100644 --- a/src/main/broker.test.ts +++ b/src/main/broker.test.ts @@ -1436,7 +1436,12 @@ describe('BrokerManager local + cloud coexistence', () => { 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 @@ describe('BrokerManager local + cloud coexistence', () => { 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', '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).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).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).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).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'] + ]) await manager.shutdown() }) diff --git a/src/main/broker.ts b/src/main/broker.ts index de2aec5b..f493396e 100644 --- a/src/main/broker.ts +++ b/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(promise: Promise, label: string): Promise { 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() @@ -1664,7 +1647,10 @@ export class BrokerManager { const { cwd, name, channels } = session const brokerPid = session.client.brokerPid - let promise!: Promise + // `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 | 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) diff --git a/src/main/integrations.test.ts b/src/main/integrations.test.ts index 999d55a8..9d637cf1 100644 --- a/src/main/integrations.test.ts +++ b/src/main/integrations.test.ts @@ -305,7 +305,10 @@ describe('IntegrationsManager', () => { mock.fetch.mockClear() mock.browserWindow.getAllWindows.mockClear() mock.resolveCloudAuth.mockClear() - mock.getAccountWorkspaceId.mockClear() + // Full reset (not just calls): tests that block cloud hydration override + // the implementation with a rejection, which must not leak forward. + mock.getAccountWorkspaceId.mockReset() + mock.getAccountWorkspaceId.mockImplementation(async () => 'account-workspace-id') mock.accountWorkspaceReadyRetryOptions.mockClear() mock.loadStore.mockClear() mock.saveStore.mockClear() @@ -977,9 +980,203 @@ describe('IntegrationsManager', () => { warnSpy.mockRestore() }) + // Typed view over the untyped sendMessage mock: the integrations-update + // send inputs in call order. + const integrationUpdateSendInputs = (): Array<{ to: string; text: string }> => + (mock.brokerManager.sendMessage as unknown as { + mock: { calls: Array<[unknown, { to: string; text: string; data?: { kind?: string } }]> } + }).mock.calls + .map((call) => call[1]) + .filter((input) => input?.data?.kind === 'integrations-update') + + it('retries a partially failed broadcast only to the agents that missed it', async () => { + vi.useFakeTimers() + // Hermetic snippet text: background cloud hydration re-adds integrations + // to the store and changes the rendered text between broadcasts (the + // separate staged-churn case the cooldown test covers). Block hydration + // and pin the store empty so the text is identical across reconciles. + mock.store.projects[0].integrations = [] + mock.getAccountWorkspaceId.mockRejectedValue(new Error('offline')) + mock.brokerManager.listAgents.mockResolvedValue([ + { name: 'claude-1', projectId: 'project-1' }, + { name: 'codex-1', projectId: 'project-1' } + ] as never) + const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => undefined) + // First broadcast: codex-1's send fails, claude-1's succeeds. + mock.brokerManager.sendMessage.mockImplementation((async (_projectId: string, input: { to: string }) => { + if (input.to === 'codex-1') throw new Error('input stream wedged') + return undefined + }) as never) + const manager = new IntegrationsManager() + + await manager.notifyAgentState('project-1') + await vi.advanceTimersByTimeAsync(1_000) + const sentTo = (): string[] => integrationUpdateSendInputs().map((input) => input.to) + expect(sentTo()).toEqual(['claude-1', 'codex-1']) + + // Retry reconcile: ONLY codex-1 (which missed it) may be re-sent — + // claude-1 already received and acknowledged this update. Re-sending to + // everyone is the duplicated-transcript bug. + mock.brokerManager.sendMessage.mockResolvedValue(undefined) + await vi.advanceTimersByTimeAsync(60_000) // clear the broadcast cooldown + await manager.notifyAgentState('project-1') + await vi.advanceTimersByTimeAsync(61_000) + expect(sentTo()).toEqual(['claude-1', 'codex-1', 'codex-1']) + warnSpy.mockRestore() + }) + + it('sends an unchanged update to a newly arrived agent without re-sending to existing agents', async () => { + vi.useFakeTimers() + mock.store.projects[0].integrations = [] + mock.getAccountWorkspaceId.mockRejectedValue(new Error('offline')) + mock.brokerManager.listAgents.mockResolvedValue([{ name: 'claude-1', projectId: 'project-1' }] as never) + const manager = new IntegrationsManager() + + await manager.notifyAgentState('project-1') + await vi.advanceTimersByTimeAsync(1_000) + + mock.brokerManager.listAgents.mockResolvedValue([ + { name: 'claude-1', projectId: 'project-1' }, + { name: 'codex-1', projectId: 'project-1' } + ] as never) + await vi.advanceTimersByTimeAsync(60_000) + await manager.notifyAgentState('project-1') + await vi.advanceTimersByTimeAsync(61_000) + + expect(integrationUpdateSendInputs().map((input) => input.to)).toEqual(['claude-1', 'codex-1']) + }) + + it('skips the broadcast for an agent whose spawn task already embedded the snippet', async () => { + vi.useFakeTimers() + mock.store.projects[0].integrations = [{ + id: 'linear-integration-1', + name: 'Linear', + type: 'linear', + provider: 'linear', + integrationId: 'linear-integration-1', + scope: {}, + mountPaths: ['/linear'], + connectedAt: '2026-06-05T00:00:00.000Z', + notifyAgent: true, + subscribeAgent: false, + downloadHistoricalData: false, + visibleInProject: true + }] + mock.brokerManager.listAgents.mockResolvedValue([{ name: 'claude-1', projectId: 'project-1' }] as never) + const manager = new IntegrationsManager() + + // Spawn path: the snippet rides inside the spawn task, and the spawn + // handler records delivery for the new agent. + const instructions = manager.initialSpawnInstructions('project-1') + expect(instructions).toContain('') + manager.recordSpawnInstructionDelivery('project-1', 'claude-1') + + // The post-spawn broadcast of the same snippet must be a no-op for the + // agent that just received it in its spawn task ("Setup context + // received" acknowledged twice was this bug). + await manager.notifyAgentState('project-1') + await vi.advanceTimersByTimeAsync(1_000) + expect(integrationUpdateSendInputs()).toHaveLength(0) + }) + + it('does not re-notify a fresh spawn when the broadcast text differs only by staged churn', async () => { + // The codex-double-acknowledgment case: an agent spawns with the snippet + // embedded in its task; moments later the project broadcast renders a + // slightly different text (readiness clause / paths hydrating) and used + // to deliver a near-identical second setup message. The spawn grace + // skips it; the follow-up after the grace delivers a still-different + // text exactly once. + vi.useFakeTimers() + mock.store.projects[0].integrations = [{ + id: 'linear-integration-1', + name: 'Linear', + type: 'linear', + provider: 'linear', + integrationId: 'linear-integration-1', + scope: {}, + mountPaths: ['/linear'], + connectedAt: '2026-06-05T00:00:00.000Z', + notifyAgent: true, + subscribeAgent: false, + downloadHistoricalData: false, + visibleInProject: true + }] + mock.getAccountWorkspaceId.mockRejectedValue(new Error('offline')) + mock.brokerManager.listAgents.mockResolvedValue([{ name: 'codex-1', projectId: 'project-1' }] as never) + const manager = new IntegrationsManager() + + expect(manager.initialSpawnInstructions('project-1')).toContain('') + manager.recordSpawnInstructionDelivery('project-1', 'codex-1') + + // Simulate staged churn: the store changes right after the spawn, so the + // broadcast text differs from the embedded snippet. + mock.store.projects[0].integrations = [{ + ...mock.store.projects[0].integrations[0], + mountPaths: ['/linear/issues'] + }] + await manager.notifyAgentState('project-1') + await vi.advanceTimersByTimeAsync(5_000) + // Grace holds: no near-identical second setup message right after spawn. + expect(integrationUpdateSendInputs()).toHaveLength(0) + + // After the grace lapses the still-different text arrives exactly once. + await vi.advanceTimersByTimeAsync(61_000) + expect(integrationUpdateSendInputs()).toHaveLength(1) + expect(integrationUpdateSendInputs()[0].to).toBe('codex-1') + + // And an identical reconcile afterwards stays silent. + await manager.notifyAgentState('project-1') + await vi.advanceTimersByTimeAsync(61_000) + expect(integrationUpdateSendInputs()).toHaveLength(1) + }) + + it('holds a changed update inside the rebroadcast cooldown and sends the latest text once', async () => { + vi.useFakeTimers() + mock.store.projects[0].integrations = [] + mock.getAccountWorkspaceId.mockRejectedValue(new Error('offline')) + mock.brokerManager.listAgents.mockResolvedValue([{ name: 'claude-1', projectId: 'project-1' }] as never) + const manager = new IntegrationsManager() + const updateTexts = (): string[] => integrationUpdateSendInputs().map((input) => input.text) + + await manager.notifyAgentState('project-1') + await vi.advanceTimersByTimeAsync(1_000) + expect(updateTexts()).toHaveLength(1) + + // Integration state converges in stages right after the first broadcast + // (channel-name discovery renames mount paths, readiness flips). The + // changed text must NOT go out immediately — it waits out the cooldown, + // and the trailing broadcast carries the latest text exactly once. + mock.store.projects[0].integrations = [{ + id: 'linear-integration-1', + name: 'Linear', + type: 'linear', + provider: 'linear', + integrationId: 'linear-integration-1', + scope: {}, + mountPaths: ['/linear'], + connectedAt: '2026-06-05T00:00:00.000Z', + notifyAgent: true, + subscribeAgent: false, + downloadHistoricalData: false, + visibleInProject: true + }] + await manager.notifyAgentState('project-1') + await vi.advanceTimersByTimeAsync(5_000) + expect(updateTexts()).toHaveLength(1) // held by the cooldown + + await vi.advanceTimersByTimeAsync(60_000) + expect(updateTexts()).toHaveLength(2) + expect(updateTexts()[1]).toContain('linear') + }) + it('does not re-send an unchanged integrations update after the dedupe window lapses', async () => { vi.useFakeTimers() - mock.brokerManager.listAgents.mockResolvedValue([{ name: 'claude-1', projectId: 'project-1' }]) + // Pin the state so the rendered text is identical across reconciles + // (background hydration would otherwise change it, which is the + // legitimate-rebroadcast case). + mock.store.projects[0].integrations = [] + mock.getAccountWorkspaceId.mockRejectedValue(new Error('offline')) + mock.brokerManager.listAgents.mockResolvedValue([{ name: 'claude-1', projectId: 'project-1' }] as never) const manager = new IntegrationsManager() const integrationsUpdateSends = (): number => mock.brokerManager.sendMessage.mock.calls.filter( @@ -989,19 +1186,16 @@ describe('IntegrationsManager', () => { await manager.notifyAgentState('project-1') await vi.advanceTimersByTimeAsync(1_000) expect(integrationsUpdateSends()).toBe(1) - await vi.waitFor(() => expect(mock.integrationMountManager.ensureMounted).toHaveBeenCalled()) - mock.brokerManager.sendMessage.mockClear() // A later reconcile with identical integration state must not re-broadcast // regardless of elapsed time — duplicate messages // were observed minutes apart with the old 30s text-TTL dedupe. await manager.notifyAgentState('project-1') - await vi.advanceTimersByTimeAsync(1_000) + await vi.advanceTimersByTimeAsync(61_000) expect(integrationsUpdateSends()).toBe(1) - await vi.advanceTimersByTimeAsync(60_000) await manager.notifyAgentState('project-1') - await vi.advanceTimersByTimeAsync(1_000) + await vi.advanceTimersByTimeAsync(61_000) expect(integrationsUpdateSends()).toBe(1) }) diff --git a/src/main/integrations.ts b/src/main/integrations.ts index aa38d678..9f6586b8 100644 --- a/src/main/integrations.ts +++ b/src/main/integrations.ts @@ -226,7 +226,18 @@ type IntegrationSystemMessageOptions = { const POLL_INTERVAL_MS = 2_000 const POLL_TIMEOUT_MS = 5 * 60_000 const CATALOG_CACHE_MS = 5 * 60_000 +interface AgentSystemMessageRecord { + text: string + // Set when the record came from a spawn-embedded snippet rather than a + // broadcast send; grants the grace window below. + fromSpawnAt: number | null +} + const SYSTEM_MESSAGE_DEBOUNCE_MS = 1_000 +// Minimum gap between integrations-update broadcasts to the same project. +// Staged state convergence (channel-name discovery, subscription readiness) +// otherwise re-broadcasts a slightly-different full update every few seconds. +const SYSTEM_MESSAGE_REBROADCAST_COOLDOWN_MS = 60_000 const SYSTEM_MESSAGE_AGENT_WAIT_TIMEOUT_MS = 8_000 const SYSTEM_MESSAGE_AGENT_WAIT_INTERVAL_MS = 500 const LOCAL_MOUNT_CLOUD_HYDRATION_THROTTLE_MS = 30_000 @@ -800,10 +811,28 @@ export class IntegrationsManager { private sessionMetadata = new Map() private pollTimers = new Map() private systemMessageTimers = new Map() - // Last successfully broadcast system message per project. Content-keyed (not - // TTL-keyed): an unchanged integrations update is never re-broadcast, no - // matter how often reconcile runs; any content change re-sends immediately. - private lastBroadcastSystemMessages = new Map() + // Last integrations-update text each agent has CONFIRMED-SENT, per project. + // Content-keyed and per-agent (not per-project): a project-level signature + // recorded only when every send succeeded meant one flaky agent send + // re-broadcast the identical update to EVERY agent on the next reconcile — + // the agents that already received it replied again (the duplicated + // "…" / duplicated acknowledgment transcripts). With + // per-agent records, retries target only the agents that missed it, and a + // newly spawned agent receives the current update without re-sending it to + // everyone else. + private lastAgentSystemMessages = new Map>() + // Subscription readiness from the last reconcile, used so spawn-embedded + // snippets render with the same readiness clause the next broadcast will + // use (a hardcoded `true` made the texts differ and re-notified the agent). + private lastSubscriptionsReady = new Map() + // When each project's last broadcast went out. Mount-path discovery and + // subscription readiness converge in stages (e.g. bare → suffixed Slack + // channel ids), each stage changing the rendered text slightly; without a + // cooldown every stage re-broadcast the full update mid-task. + private lastSystemMessageBroadcastAt = new Map() + // Snippet most recently embedded into spawn instructions, per project, so + // the spawn path can record it as already-delivered for the new agent. + private lastSpawnInstructionSnippets = new Map() private catalogCache: IntegrationAdapter[] | null = null private catalogFetchedAt = 0 private localMountCloudHydrationStartedAt = 0 @@ -2037,8 +2066,9 @@ export class IntegrationsManager { notifyAgent: boolean, systemMessageOptions: IntegrationSystemMessageOptions = {} ): Promise { - let integrations = this.visibleIntegrationsForProject(projectId) + const integrations = this.visibleIntegrationsForProject(projectId) const subscriptionsReady = await this.syncEventSubscriptions(projectId) + this.lastSubscriptionsReady.set(projectId, subscriptionsReady) if (notifyAgent) { this.scheduleSystemMessage( @@ -2138,12 +2168,36 @@ export class IntegrationsManager { initialSpawnInstructions(projectId: string): string | undefined { const integrations = this.visibleIntegrationsForProject(projectId) if (integrations.length === 0) return undefined + const snippet = this.buildSystemMessageSnippet( + integrations, + this.lastSubscriptionsReady.get(projectId) ?? true + ) + // Remember what this spawn embedded so recordSpawnInstructionDelivery + // can mark the new agent as already-notified — without it, the next + // project broadcast re-delivered the same setup context to the agent + // that just received it in its spawn task ("Setup context received" + // acknowledged twice). + this.lastSpawnInstructionSnippets.set(projectId, snippet) return [ 'Initial project integration context. Treat this as setup context, not as the user task.', - this.buildSystemMessageSnippet(integrations, true) + snippet ].join('\n') } + // Called after a spawn whose task embedded initialSpawnInstructions: the + // new agent already has the current integrations snippet, so per-agent + // delivery records treat it as up to date. + recordSpawnInstructionDelivery(projectId: string, agentName: string): void { + const snippet = this.lastSpawnInstructionSnippets.get(projectId) + if (!snippet) return + let delivered = this.lastAgentSystemMessages.get(projectId) + if (!delivered) { + delivered = new Map() + this.lastAgentSystemMessages.set(projectId, delivered) + } + delivered.set(agentName, { text: snippet, fromSpawnAt: Date.now() }) + } + private mountPathsForAgentWorkspace(integration: ConnectedIntegration): string[] { return dedupeStrings([ discoveryMountPathForProvider(integration.provider), @@ -2198,34 +2252,88 @@ export class IntegrationsManager { options: IntegrationSystemMessageOptions = {} ): Promise { try { - if (this.lastBroadcastSystemMessages.get(projectId) === message) return - const bridge = brokerManager as unknown as IntegrationSystemMessageBridge const agents = await this.listSystemMessageAgents(bridge, projectId, options.waitForAgent === true) if (agents.length === 0) return - await Promise.all( - agents - .map((agent) => { - const input = { - to: agent.name, - from: 'system', - text: message, - priority: 0, - mode: 'steer', - data: { - kind: 'integrations-update', - system: true - } - } as const - // Delivery confirmations can hang behind inactive PTYs. Integration - // updates are setup context, so broker send acceptance is the - // reliable boundary for idempotent notification. - return bridge.sendMessage(projectId, input) - }) + + let delivered = this.lastAgentSystemMessages.get(projectId) + if (!delivered) { + delivered = new Map() + this.lastAgentSystemMessages.set(projectId, delivered) + } + // Only agents that have not already received THIS text. A retry after + // a partial failure reaches just the agents that missed it; agents + // that already acknowledged the update never see it twice. + const now = Date.now() + let graceRemainingMs = 0 + const pending = agents.filter((agent) => { + const record = delivered.get(agent.name) + if (!record) return true + if (record.text === message) return false + // Spawn grace: right after a spawn the broadcast text routinely + // differs only cosmetically from the snippet embedded in the spawn + // task (readiness clause, paths still hydrating). Skip the + // near-identical second setup message; the follow-up below delivers + // any text that still differs once the grace lapses. + if (record.fromSpawnAt !== null) { + const remaining = record.fromSpawnAt + SYSTEM_MESSAGE_REBROADCAST_COOLDOWN_MS - now + if (remaining > 0) { + graceRemainingMs = Math.max(graceRemainingMs, remaining) + return false + } + } + return true + }) + if (pending.length === 0) { + if (graceRemainingMs > 0 && !this.systemMessageTimers.has(projectId)) { + // Re-run after the grace lapses so a real post-spawn change still + // reaches the fresh agent. Never clobbers a newer pending text — + // only scheduled when no timer is waiting. + this.scheduleSystemMessage(projectId, message, options, graceRemainingMs) + } + return + } + + const results = await Promise.allSettled( + pending.map((agent) => { + const input = { + to: agent.name, + from: 'system', + text: message, + priority: 0, + mode: 'steer', + data: { + kind: 'integrations-update', + system: true + } + } as const + // Delivery confirmations can hang behind inactive PTYs. Integration + // updates are setup context, so broker send acceptance is the + // reliable boundary for idempotent notification. + return bridge.sendMessage(projectId, input) + }) ) - // Recorded only after every send was accepted: a failed broadcast leaves - // no signature, so the next reconcile retries the identical message. - this.lastBroadcastSystemMessages.set(projectId, message) + let anyDelivered = false + results.forEach((result, index) => { + if (result.status === 'fulfilled') { + delivered.set(pending[index].name, { text: message, fromSpawnAt: null }) + anyDelivered = true + // Low-noise send telemetry: integrations-update sends are rare by + // design now, and each one triggers a visible agent turn — keep a + // breadcrumb so duplicate-transcript reports are diagnosable. + console.info( + `[integrations] sent integrations-update to ${pending[index].name} (project ${projectId}, ${message.length} chars)` + ) + } else { + console.warn( + `[integrations] Failed to inject integration system message for ${pending[index].name}:`, + toErrorMessage(result.reason) + ) + } + }) + if (anyDelivered) { + this.lastSystemMessageBroadcastAt.set(projectId, Date.now()) + } } catch (error) { console.warn('[integrations] Failed to inject integration system message:', toErrorMessage(error)) } @@ -2234,15 +2342,28 @@ export class IntegrationsManager { private scheduleSystemMessage( projectId: string, message: string, - options: IntegrationSystemMessageOptions = {} + options: IntegrationSystemMessageOptions = {}, + minDelayMs = 0 ): void { const existing = this.systemMessageTimers.get(projectId) if (existing) clearTimeout(existing) + // Cooldown after a real broadcast: integration text converges in stages + // (channel-name discovery, subscription readiness), and each stage used + // to re-broadcast the whole update to every agent mid-task. Later + // schedules within the cooldown replace the timer, so the trailing + // broadcast carries the latest text. Per-agent delivery records make + // the eventual send a no-op for agents already up to date. + const lastBroadcastAt = this.lastSystemMessageBroadcastAt.get(projectId) + const cooldownRemaining = lastBroadcastAt === undefined + ? 0 + : Math.max(0, lastBroadcastAt + SYSTEM_MESSAGE_REBROADCAST_COOLDOWN_MS - Date.now()) + const delay = Math.max(SYSTEM_MESSAGE_DEBOUNCE_MS, cooldownRemaining, minDelayMs) + const timer = setTimeout(() => { this.systemMessageTimers.delete(projectId) void this.safeInjectSystemMessage(projectId, message, options) - }, SYSTEM_MESSAGE_DEBOUNCE_MS) + }, delay) this.systemMessageTimers.set(projectId, timer) } diff --git a/src/main/ipc-handlers.ts b/src/main/ipc-handlers.ts index c6220bde..1e3d11da 100644 --- a/src/main/ipc-handlers.ts +++ b/src/main/ipc-handlers.ts @@ -28,7 +28,7 @@ import { getIntegrationEventTelemetrySnapshot, integrationEventBridge } from './ import { aiHistManager } from './ai-hist' import { burnManager, type BurnAgentInput, type BurnProjectInput, type BurnSessionBreakdownInput, type BurnFingerprintInput } from './burn' import { resetRelayWorkspaceManager } from './relay-workspace' -import { assertDirectory, isDirectory } from './path-utils' +import { isDirectory } from './path-utils' import { findProjectForPath, projectContainsPath } from './cli' import type { BrokerReconcileMessagesInput, BrokerSpawnAgentResult } from '../shared/types/ipc' import type { ProactiveAgentDraft } from './proactive-agent.types' @@ -261,6 +261,12 @@ export function registerIpcHandlers(): void { : integrationInstructions } : input) + if (integrationInstructions && result?.name) { + // The spawn task embedded the current integrations snippet — mark the + // new agent as already notified so the next project broadcast doesn't + // deliver the same setup context to it a second time. + integrationsManager.recordSpawnInstructionDelivery(projectId, result.name) + } integrationEventBridge.invalidateProjectAgentCache(projectId) return toBrokerSpawnAgentResult(result) }) diff --git a/src/main/pty-dedup.test.ts b/src/main/pty-dedup.test.ts index 854dbc86..dea27910 100644 --- a/src/main/pty-dedup.test.ts +++ b/src/main/pty-dedup.test.ts @@ -9,19 +9,24 @@ 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 +// 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 now: number + let infoSpy: ReturnType beforeEach(() => { - now = 1_000_000 - vi.spyOn(Date, 'now').mockImplementation(() => now) + infoSpy = vi.spyOn(console, 'info').mockImplementation(() => {}) }) afterEach(() => { - vi.restoreAllMocks() + infoSpy.mockRestore() }) - it('suppresses a repeat of the same event identity', () => { + 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) @@ -34,51 +39,66 @@ describe('PtyChunkDeduper', () => { expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', chunkEvent({ name: 'claude-1', chunk: 'tick\n', seq: 2 }))).toBe(false) }) - it('lets the identity window survive past the short content window', () => { + 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', event_id: 'e1' }) + 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) - // Beyond the 1s content TTL but well within the 60s identity TTL: still a dup. - now += 5_000 - expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', event)).toBe(true) }) - it('expires an identity entry after the 60s window', () => { + it('logs the identity-less blind spot once per stream', () => { 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) - now += 60_001 - expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', event)).toBe(false) + 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('dedupes identity-less chunks within the 1s content window', () => { + it('delivers a chunk whose seq repeats with different bytes (daemon seq reset)', () => { 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(true) + 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('re-admits an identity-less chunk once the content window expires', () => { + it('delivers low-seq chunks after a watermark reset instead of dropping a window', () => { const deduper = new PtyChunkDeduper() - const event = chunkEvent({ name: 'claude-1', chunk: 'pong\n' }) - expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', event)).toBe(false) - now += 1_001 - expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', event)).toBe(false) + 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('only remembers the latest identity-less chunk per agent, so a repeat after intervening output is admitted', () => { + it('tracks replay windows per session and agent stream', () => { const deduper = new PtyChunkDeduper() - expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', chunkEvent({ name: 'claude-1', chunk: '>' }))).toBe(false) - expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', chunkEvent({ name: 'claude-1', chunk: ' ' }))).toBe(false) - // The first '>' content key was evicted when ' ' arrived, so '>' is fresh again. - expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', chunkEvent({ name: 'claude-1', chunk: '>' }))).toBe(false) + 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('scopes content dedup per session and agent', () => { + it('prefers event_id over seq for identity', () => { const deduper = new PtyChunkDeduper() - expect(deduper.isDuplicatePtyChunk('proj', 'claude-1', chunkEvent({ name: 'claude-1', chunk: 'pong\n' }))).toBe(false) - expect(deduper.isDuplicatePtyChunk('proj', 'claude-2', chunkEvent({ name: 'claude-2', chunk: 'pong\n' }))).toBe(false) - expect(deduper.isDuplicatePtyChunk('other', 'claude-1', chunkEvent({ name: 'claude-1', chunk: 'pong\n' }))).toBe(false) + 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', () => { @@ -87,4 +107,12 @@ describe('PtyChunkDeduper', () => { 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) + }) }) diff --git a/src/main/pty-dedup.ts b/src/main/pty-dedup.ts index 9fca9a7b..85c57423 100644 --- a/src/main/pty-dedup.ts +++ b/src/main/pty-dedup.ts @@ -1,65 +1,157 @@ import type { BrokerEvent } from '@agent-relay/harness-driver' -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 +// 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". +const PTY_CHUNK_REPLAY_WINDOW = 512 +const MAX_PTY_STREAM_DEDUPE_STATES = 2_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 + // PTY_CHUNK_REPLAY_WINDOW chunks, insertion-ordered for FIFO eviction. + recentByIdentity: Map + // Telemetry: replayed chunks suppressed for this stream. + suppressedReplays: number +} + +// Cheap, allocation-free content fingerprint (FNV-1a 32-bit + length). A +// false "duplicate" verdict requires an identity match AND a hash collision +// AND identical length — and only ever suppresses one chunk that renders the +// same bytes we already delivered for that identity. +function ptyChunkHash(chunk: string): string { + let hash = 0x811c9dc5 + for (let i = 0; i < chunk.length; i += 1) { + hash ^= chunk.charCodeAt(i) + hash = Math.imul(hash, 0x01000193) + } + return `${chunk.length}:${hash >>> 0}` +} -// Suppresses duplicate PTY chunks delivered more than once (overlapping event -// streams, reconnect replay). Prefers stable event identity (event_id / id / -// seq) with a 60s window, and falls back to a short 1s content window for -// chunks the broker emitted without identity (AGENTS.md duplicate-event -// hardening). State is a single `${prefix}:${sessionKey}:${name}:...` -> seenAt -// map, pruned lazily on each call. export class PtyChunkDeduper { - private recentPtyChunks = new Map() + private ptyStreamDedupe = new Map() + // Streams already flagged for identity-less PTY chunks (one log per stream). + private ptyStreamsMissingIdentity = new Set() + private readonly debugEnabled: () => boolean - 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 > ttl || - this.recentPtyChunks.size > MAX_PTY_CHUNK_DEDUPE_ENTRIES - ) { - this.recentPtyChunks.delete(key) - } - } + constructor(opts: { debugEnabled?: () => boolean } = {}) { + this.debugEnabled = opts.debugEnabled ?? (() => false) + } + // 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 + // repaint, which is strictly worse than rendering a rare duplicate. + isDuplicatePtyChunk(sessionKey: string, name: string, event: BrokerEvent): boolean { const eventRecord = event as Record - const seq = typeof eventRecord.seq === 'number' || typeof eventRecord.seq === 'string' - ? String(eventRecord.seq) - : '' - const eventId = typeof eventRecord.event_id === 'string' + 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' + : typeof eventRecord.id === 'string' && eventRecord.id ? eventRecord.id : '' - const identity = eventId || (seq ? `seq:${seq}` : '') - const chunk = typeof eventRecord.chunk === 'string' ? eventRecord.chunk : '' - if (!identity && !chunk) return false + 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` + ) + } + 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 <= ttl) { - return true + 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) } - if (!identity) { - for (const previousKey of this.recentPtyChunks.keys()) { - if (previousKey.startsWith(contentKeyPrefix)) { - this.recentPtyChunks.delete(previousKey) + const remembered = state.recentByIdentity.get(identity) + if (remembered !== undefined) { + if (remembered === hash) { + 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. + if (this.debugEnabled()) { + console.info('[broker] suppressed replayed PTY chunk', { + name, + identity, + suppressedTotal: state.suppressedReplays + }) } + 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() + } + + 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 + if (oldest !== undefined) state.recentByIdentity.delete(oldest) } - this.recentPtyChunks.set(key, now) return false } + + // Drop per-stream state when an agent session goes away so a respawned + // agent (fresh PTY, fresh seq space) starts with a clean window. + forgetStream(sessionKey: string, name: string): void { + this.ptyStreamDedupe.delete(`${sessionKey}:${name}`) + this.ptyStreamsMissingIdentity.delete(`${sessionKey}:${name}`) + } } diff --git a/src/main/relay-workspace.ts b/src/main/relay-workspace.ts index 4c946afe..a461ef18 100644 --- a/src/main/relay-workspace.ts +++ b/src/main/relay-workspace.ts @@ -109,8 +109,7 @@ export class RelayWorkspaceManager { authContext: RelayWorkspaceAuthContext, generation: number ): Promise { - let pendingHandle!: Promise - pendingHandle = this.bootstrap(auth) + const pendingHandle: Promise = this.bootstrap(auth) .then((result) => { if (!this.isCurrentBootstrap(pendingHandle, authContext, generation)) { return this.getCurrentOrNextHandle() diff --git a/src/renderer/index.html b/src/renderer/index.html index 4a22a3c2..38b49b09 100644 --- a/src/renderer/index.html +++ b/src/renderer/index.html @@ -6,7 +6,7 @@ Pear by Agent Relay diff --git a/src/renderer/src/hooks/use-terminal.ts b/src/renderer/src/hooks/use-terminal.ts index 9d793017..f27670e7 100644 --- a/src/renderer/src/hooks/use-terminal.ts +++ b/src/renderer/src/hooks/use-terminal.ts @@ -117,9 +117,11 @@ export function useTerminal( const sendInput = useCallback((data: string): void => { if (!agentName || terminalModeRef.current === 'view') return - // Optimistically echo before the round trip; the engine reconciles - // against authoritative output and stays dormant on fast local links. - runtimeRef.current?.getPredictiveEcho()?.onUserInput(data) + // Optimistically echo before the round trip; the runtime routes through + // the predictive-echo engine only when measured latency warrants it and + // reconciles against authoritative output, staying dormant (and + // zero-overhead) on fast local links. + runtimeRef.current?.noteUserInput(data) recordKeystrokeSent(data) pear.broker.sendInputFast(projectId, agentName, data) }, [agentName, projectId]) diff --git a/src/renderer/src/lib/echo-router.test.ts b/src/renderer/src/lib/echo-router.test.ts new file mode 100644 index 00000000..cf458979 --- /dev/null +++ b/src/renderer/src/lib/echo-router.test.ts @@ -0,0 +1,530 @@ +// Real-engine equivalence tests for the echo router. +// +// The router decides whether server bytes reach the live terminal directly +// or through the predictive-echo engine, and owns the direct→engine +// transition (model reseed at a write-queue quiescent point, with chunks +// held during the capture). The property every scenario must preserve: the +// final screen TEXT equals a reference terminal fed the same byte stream in +// one write — across route flips, held-chunk windows, flood bypasses, and +// reseed aborts. Runs against the REAL PredictiveEchoEngine and the REAL +// xterm VT parser (@xterm/headless), so an engine or parser behavior change +// that breaks the routing assumptions fails here, not on a user's screen. + +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { Terminal } from '@xterm/headless' +import { + PredictiveEchoEngine, + type ScreenModel +} from '@agent-relay/harness-driver/predictive-echo' +import { + buildModelSeedFromTerminal, + createEchoRouter, + ENGINE_FLOOD_BYPASS_BYTES, + PREDICTION_QUIET_MS, + RESEED_CAPTURE_TIMEOUT_MS, + type EchoRouter, + type SeedSource +} from './echo-router' +import type { PredictiveEchoWithStatus } from './predictive-echo' + +const ROWS = 24 +const COLS = 80 + +class HeadlessModel implements ScreenModel { + private readonly term: Terminal + // Test hook: when set, model bytes sit in a backlog for this many ms + // BEFORE being parsed — the model's cursor/rows stay stale meanwhile. + // This is what a busy renderer looks like: xterm's WriteBuffer queues + // the chunk and the parse (cursor advance included) happens ticks later. + writeDelayMs = 0 + + constructor(cols: number, rows: number) { + this.term = new Terminal({ cols, rows, allowProposedApi: true, scrollback: 0 }) + } + + write(data: string): Promise { + if (this.writeDelayMs <= 0) { + return new Promise((resolve) => this.term.write(data, resolve)) + } + const delay = this.writeDelayMs + return new Promise((resolve) => + setTimeout(() => this.term.write(data, resolve), delay) + ) + } + + cursor(): { row: number; col: number } { + const buffer = this.term.buffer.active + return { row: buffer.cursorY, col: buffer.cursorX } + } + + rowText(row: number): string { + const buffer = this.term.buffer.active + const line = buffer.getLine(buffer.baseY + row) + return line ? line.translateToString(true) : '' + } + + isAltScreen(): boolean { + return this.term.buffer.active.type === 'alternate' + } + + cols(): number { + return this.term.cols + } + + resize(cols: number, rows: number): void { + this.term.resize(cols, rows) + } + + dispose(): void { + this.term.dispose() + } +} + +function readScreen(term: Terminal): string { + const buf = term.buffer.active + const lines: string[] = [] + for (let i = 0; i < term.rows; i += 1) { + const line = buf.getLine(buf.baseY + i) + lines.push((line ? line.translateToString(true) : '').replace(/[ \t]+$/, '')) + } + return lines.join('\n') +} + +function drain(term: Terminal): Promise { + return new Promise((resolve) => term.write('', () => resolve())) +} + +function buildRedrawFrame(frame: number): string { + let out = '\x1b[H' + for (let r = 1; r <= ROWS; r += 1) { + out += `\x1b[${r};1H\x1b[2K\x1b[36mframe ${frame + .toString() + .padStart(3, '0')} row ${r.toString().padStart(2, '0')}\x1b[0m` + } + return out +} + +interface Harness { + router: EchoRouter + live: Terminal + engine: PredictiveEchoEngine + model: HeadlessModel + setSrtt(value: number | null): void + dispose(): void +} + +function makeHarness(): Harness { + const live = new Terminal({ cols: COLS, rows: ROWS, allowProposedApi: true }) + const model = new HeadlessModel(COLS, ROWS) + let srtt: number | null = null + const engine = new PredictiveEchoEngine({ + model, + write: (data: string) => { + live.write(data) + }, + now: () => performance.now(), + getInputSrtt: () => srtt + }) + const router = createEchoRouter({ + write: (data, callback) => live.write(data, callback), + getEngine: () => engine as unknown as PredictiveEchoWithStatus, + buildModelSeed: () => buildModelSeedFromTerminal(live as unknown as SeedSource), + getInputSrtt: () => srtt, + isViewportPinned: () => false, + scrollToBottom: () => {} + }) + return { + router, + live, + engine, + model, + setSrtt: (value) => { + srtt = value + }, + dispose: () => { + router.dispose() + engine.reset() + live.dispose() + } + } +} + +// Settle the engine's internal promise chain AND the live terminal's write +// queue (each engine step enqueues live writes; interleave a few rounds). +async function settle(h: Harness): Promise { + for (let i = 0; i < 4; i += 1) { + await h.engine.onServerOutput('') + await drain(h.live) + } +} + +async function reference(stream: string): Promise { + const term = new Terminal({ cols: COLS, rows: ROWS, allowProposedApi: true }) + await new Promise((resolve) => term.write(stream, () => resolve())) + const screen = readScreen(term) + term.dispose() + return screen +} + +// Predictions require the stream to have been quiet; streaming-sized chunks +// arm a real-time window (Date.now-based). +function waitForQuietWindow(): Promise { + return new Promise((resolve) => setTimeout(resolve, PREDICTION_QUIET_MS + 50)) +} + +describe('echo-router — real engine + real parser equivalence', () => { + let harness: Harness + + beforeEach(() => { + harness = makeHarness() + }) + + afterEach(() => { + harness.dispose() + }) + + it('starts on the direct route and passes bytes through byte-for-byte', async () => { + const frames = [1, 2, 3].map(buildRedrawFrame) + for (const frame of frames) harness.router.onServerOutput(frame) + await settle(harness) + + expect(harness.router.route()).toBe('direct') + expect(readScreen(harness.live)).toBe(await reference(frames.join(''))) + }) + + it('stays direct (no engine work, no transition) while latency is low', async () => { + harness.setSrtt(5) + const seedSpy = vi.spyOn(harness.engine, 'seed') + const outputSpy = vi.spyOn(harness.engine, 'onServerOutput') + + harness.router.onUserInput('a') + harness.router.onServerOutput(buildRedrawFrame(1)) + await drain(harness.live) + + expect(harness.router.route()).toBe('direct') + expect(seedSpy).not.toHaveBeenCalled() + expect(outputSpy).not.toHaveBeenCalled() + }) + + it('transitions to the engine route on an engaged keystroke and preserves the screen, including chunks held during the capture', async () => { + const pre = [1, 2].map(buildRedrawFrame) + for (const frame of pre) harness.router.onServerOutput(frame) + + // Latency spike → keystroke triggers the transition. Chunks that arrive + // while the quiescent-point capture is in flight are held and must land + // exactly once. (Transition requires a quiet stream.) + await waitForQuietWindow() + harness.setSrtt(80) + harness.router.onUserInput('x') + const during = [3, 4].map(buildRedrawFrame) + for (const frame of during) harness.router.onServerOutput(frame) + + await drain(harness.live) // capture callback fires in write-queue order + await settle(harness) + + expect(harness.router.route()).toBe('engine') + const post = buildRedrawFrame(5) + harness.router.onServerOutput(post) + await settle(harness) + + expect(readScreen(harness.live)).toBe( + await reference([...pre, ...during, post].join('')) + ) + }) + + it('renders a correct optimistic prediction after the transition reseed', async () => { + // Paint a prompt, cursor parked at end of row 0. + harness.router.onServerOutput('\x1b[2J\x1b[H$ ') + harness.setSrtt(80) + harness.router.onUserInput('q') // triggers transition; no prediction yet + await drain(harness.live) + await settle(harness) + expect(harness.router.route()).toBe('engine') + + // Now predictions are live against the reseeded model. + harness.router.onUserInput('a') + await settle(harness) + expect((harness.engine as PredictiveEchoEngine).hasPredictions).toBe(true) + expect(readScreen(harness.live).split('\n')[0]).toBe('$ a') + + // Authoritative echo confirms the glyph; the screen text is unchanged. + harness.router.onServerOutput('a') + await settle(harness) + expect(readScreen(harness.live).split('\n')[0]).toBe('$ a') + }) + + it('drops back to direct on an output flood with no outstanding predictions', async () => { + harness.setSrtt(80) + harness.router.onUserInput('x') + await drain(harness.live) + await settle(harness) + expect(harness.router.route()).toBe('engine') + + // A flood frame larger than the bypass threshold with nothing predicted. + let flood = '' + let i = 0 + while (flood.length <= ENGINE_FLOOD_BYPASS_BYTES) { + flood += buildRedrawFrame(i % 1000) + i += 1 + } + harness.router.onServerOutput(flood) + await settle(harness) + + expect(harness.router.route()).toBe('direct') + expect(readScreen(harness.live)).toBe(await reference(flood)) + }) + + it('rolls back on-screen predictions when streaming starts and stays byte-faithful', async () => { + harness.router.onServerOutput('\x1b[2J\x1b[H$ ') + harness.setSrtt(80) + harness.router.onUserInput('q') + await drain(harness.live) + await settle(harness) + harness.router.onUserInput('a') // outstanding prediction + await settle(harness) + expect((harness.engine as PredictiveEchoEngine).hasPredictions).toBe(true) + + let flood = '' + let i = 0 + while (flood.length <= ENGINE_FLOOD_BYPASS_BYTES) { + flood += buildRedrawFrame(i % 1000) + i += 1 + } + harness.router.onServerOutput(flood) + await settle(harness) + + // Streaming started with a glyph on screen: the router rolled it back + // at still-valid positions instead of letting the engine erase/repaint + // it around every frame. The screen equals pure pass-through of the + // confirmed bytes, and no prediction survives the storm. + expect((harness.engine as PredictiveEchoEngine).hasPredictions).toBe(false) + expect(readScreen(harness.live)).toBe( + await reference('\x1b[2J\x1b[H$ ' + flood) + ) + + // New keystrokes during the active stream stay prediction-free. + harness.router.onUserInput('b') + await settle(harness) + expect((harness.engine as PredictiveEchoEngine).hasPredictions).toBe(false) + expect(readScreen(harness.live)).toBe( + await reference('\x1b[2J\x1b[H$ ' + flood) + ) + }) + + it('preserves byte order when a flood bypass races chunks still queued in the engine tail', async () => { + // The interleaved-rows corruption class: chunk B is queued inside the + // engine's async tail (model write of A still in flight) when flood + // chunk C trips the bypass. A direct write of C MUST NOT overtake B — + // out-of-order bytes shear escape sequences and overlay row text. + harness.setSrtt(80) + harness.router.onUserInput('x') + await drain(harness.live) + await settle(harness) + expect(harness.router.route()).toBe('engine') + + harness.model.writeDelayMs = 30 // back up the engine tail + const a = buildRedrawFrame(1) + const b = buildRedrawFrame(2) + let flood = '' + let i = 0 + while (flood.length <= ENGINE_FLOOD_BYPASS_BYTES) { + flood += buildRedrawFrame(100 + (i % 100)) + i += 1 + } + harness.router.onServerOutput(a) + harness.router.onServerOutput(b) // queued behind A's slow model write + harness.router.onServerOutput(flood) // bypass: must still land after B + + await new Promise((resolve) => setTimeout(resolve, 250)) + harness.model.writeDelayMs = 0 + await settle(harness) + + expect(readScreen(harness.live)).toBe(await reference(a + b + flood)) + }) + + it('skips optimistic echo while a chunk is mid-flight in the engine tail (stale-cursor strand)', async () => { + // Get on the engine route with a prompt at row 0. + harness.router.onServerOutput('\x1b[2J\x1b[H$ ') + harness.setSrtt(80) + harness.router.onUserInput('q') + await drain(harness.live) + await settle(harness) + expect(harness.router.route()).toBe('engine') + + // Back up the model parse, deliver a chunk that moves the cursor to a + // new row, and type while that chunk's model parse is still backlogged. + // The live terminal already shows the chunk (pass-through precedes the + // model parse) but the model cursor is still on the OLD row — a + // prediction placed now renders at the stale row, overwriting real + // content, and the engine's cleanup skips erasing glyphs whose row no + // longer matches its prediction row, stranding them on screen. + harness.model.writeDelayMs = 30 + harness.router.onServerOutput('hello\r\nworld$ ') + // Let the op start (live write issued, model parse backlogged) … + await new Promise((resolve) => setTimeout(resolve, 5)) + harness.router.onUserInput('a') // mid-flight: optimistic echo must be skipped + harness.model.writeDelayMs = 0 + await new Promise((resolve) => setTimeout(resolve, 80)) + await settle(harness) + expect((harness.engine as PredictiveEchoEngine).hasPredictions).toBe(false) + + // The authoritative echo arrives later and lands at the real cursor. + harness.router.onServerOutput('a') + await settle(harness) + + expect(readScreen(harness.live)).toBe( + await reference('\x1b[2J\x1b[H$ hello\r\nworld$ a') + ) + }) + + it('re-engages predictions after a flood bypass via a fresh reseed', async () => { + harness.setSrtt(80) + harness.router.onUserInput('x') + await drain(harness.live) + await settle(harness) + + let flood = '' + let i = 0 + while (flood.length <= ENGINE_FLOOD_BYPASS_BYTES) { + flood += buildRedrawFrame(i % 1000) + i += 1 + } + harness.router.onServerOutput(flood) + await settle(harness) + expect(harness.router.route()).toBe('direct') + + // Paint a fresh prompt directly (model is stale now), wait out the + // streaming quiet window, then type. + harness.router.onServerOutput('\x1b[2J\x1b[H$ ') + await waitForQuietWindow() + harness.router.onUserInput('q') // reseeds from the live screen + await drain(harness.live) + await settle(harness) + expect(harness.router.route()).toBe('engine') + + harness.router.onUserInput('z') + await settle(harness) + expect(readScreen(harness.live).split('\n')[0]).toBe('$ z') + }) +}) + +describe('echo-router — reseed capture timeout', () => { + beforeEach(() => { + vi.useFakeTimers() + }) + + afterEach(() => { + vi.useRealTimers() + }) + + it('aborts a stalled capture, releases held chunks directly, and stays direct', () => { + const written: string[] = [] + const engine = { + seed: vi.fn(() => Promise.resolve()), + onUserInput: vi.fn(), + onServerOutput: vi.fn(() => Promise.resolve()), + rollback: vi.fn(), + onResize: vi.fn(), + reset: vi.fn(), + hasPredictions: false + } satisfies PredictiveEchoWithStatus + + const router = createEchoRouter({ + // Never invoke the capture callback — a stalled live terminal. + write: (data) => { + if (data) written.push(data) + }, + getEngine: () => engine, + buildModelSeed: () => '\x1bc', + getInputSrtt: () => 80, + isViewportPinned: () => false, + scrollToBottom: () => {} + }) + + router.onUserInput('a') // begins the transition; capture never completes + router.onServerOutput('held-1') + router.onServerOutput('held-2') + expect(written).toEqual([]) // held while the capture is pending + + vi.advanceTimersByTime(RESEED_CAPTURE_TIMEOUT_MS) + expect(written).toEqual(['held-1held-2']) + expect(router.route()).toBe('direct') + expect(engine.seed).not.toHaveBeenCalled() + + // Later output flows normally on the direct route. + router.onServerOutput('after') + expect(written).toEqual(['held-1held-2', 'after']) + router.dispose() + }) + + it('releases held chunks directly when the engine changes during the capture', () => { + const written: string[] = [] + let captureCallback: (() => void) | null = null + const makeEngine = (): PredictiveEchoWithStatus => ({ + seed: vi.fn(() => Promise.resolve()), + onUserInput: vi.fn(), + onServerOutput: vi.fn(() => Promise.resolve()), + rollback: vi.fn(), + onResize: vi.fn(), + reset: vi.fn(), + hasPredictions: false + }) + let engine = makeEngine() + const firstEngine = engine + + const router = createEchoRouter({ + write: (data, callback) => { + if (callback) captureCallback = callback + else if (data) written.push(data) + }, + getEngine: () => engine, + buildModelSeed: () => '\x1bc', + getInputSrtt: () => 80, + isViewportPinned: () => false, + scrollToBottom: () => {} + }) + + router.onUserInput('a') // begins the transition + router.onServerOutput('held-1') + router.onServerOutput('held-2') + + // The engine is swapped out before the capture completes; the held + // bytes are live PTY output and must still render. + engine = makeEngine() + captureCallback?.() + + expect(written).toEqual(['held-1held-2']) + expect(router.route()).toBe('direct') + expect(firstEngine.seed).not.toHaveBeenCalled() + expect(engine.seed).not.toHaveBeenCalled() + router.dispose() + }) + + it('dispose drops held chunks and cancels the capture timeout', () => { + const written: string[] = [] + const router = createEchoRouter({ + write: (data) => { + if (data) written.push(data) + }, + getEngine: () => ({ + seed: () => Promise.resolve(), + onUserInput: () => {}, + onServerOutput: () => Promise.resolve(), + rollback: () => {}, + onResize: () => {}, + reset: () => {}, + hasPredictions: false + }), + buildModelSeed: () => '\x1bc', + getInputSrtt: () => 80, + isViewportPinned: () => false, + scrollToBottom: () => {} + }) + + router.onUserInput('a') + router.onServerOutput('held') + router.dispose() + vi.advanceTimersByTime(RESEED_CAPTURE_TIMEOUT_MS * 2) + expect(written).toEqual([]) + }) +}) diff --git a/src/renderer/src/lib/echo-router.ts b/src/renderer/src/lib/echo-router.ts new file mode 100644 index 00000000..ac6a4ac4 --- /dev/null +++ b/src/renderer/src/lib/echo-router.ts @@ -0,0 +1,304 @@ +// Echo routing between the live terminal and the predictive-echo engine. +// +// The predictive-echo engine parses every server byte a second time into its +// headless confirmed model. On a fast local link the engine never engages +// (it needs >= PREDICTION_ENGAGE_THRESHOLD_MS echo SRTT), so that second +// parse is pure overhead — under heavy multi-agent streaming it is renderer +// main-thread time stolen from keystroke echo (the "typing lag" class). +// +// Server output therefore flows DIRECT to the terminal until measured +// latency says predictions are worth having. The direct→engine transition +// must not let the engine predict against a stale model: it reseeds the +// model from the live screen at a write-queue quiescent point (an empty +// write's callback fires only after everything already queued has parsed), +// holding any chunks that arrive during the capture so every byte lands +// either in the captured seed or in the model replay — exactly once, never +// both, never neither. +// +// Invariant that must hold for any input sequence: the bytes reaching the +// live terminal are exactly the bytes received, in order (the engine's own +// pass-through guarantees this on the engine route). The equivalence tests +// in echo-router.test.ts assert this against the real engine + real parser. + +import type { PredictiveEchoWithStatus } from '@/lib/predictive-echo' + +// Mirrors the engine's DEFAULT_PREDICTION_CONFIG.engageThresholdMs. +export const PREDICTION_ENGAGE_THRESHOLD_MS = 30 +// Predictions are a quiet-prompt tool (mosh's regime: you type, the line +// echoes back). While a TUI is actively streaming repaint frames, the engine +// would erase/re-render its optimistic glyphs around EVERY frame — any +// model/live divergence then erases real rows and strands glyphs inside +// repainted panels (the interleaved-rows corruption class). So: a server +// chunk bigger than PREDICTION_SAFE_CHUNK_BYTES marks the stream "active", +// outstanding predictions are rolled back at that moment, and no new +// predictions render until the stream has been quiet for +// PREDICTION_QUIET_MS. Keystroke echo (tiny chunks) never trips this. +export const PREDICTION_SAFE_CHUNK_BYTES = 1_024 +export const PREDICTION_QUIET_MS = 300 +// On the engine route, one coalesced frame bigger than this with no +// outstanding predictions drops back to direct: during output floods the +// model double-parse only adds drain time, and there is nothing on screen +// for the engine to reconcile. +export const ENGINE_FLOOD_BYPASS_BYTES = 32_768 +// Reseed capture safety net: if the live terminal's write callback hasn't +// fired by then (disposed mid-flight, renderer stalled), abort the +// transition and release any held chunks directly. +export const RESEED_CAPTURE_TIMEOUT_MS = 1_000 + +// Structural slice of an xterm Terminal sufficient to rebuild the engine's +// confirmed-model state: full reset, alt-screen flag, viewport row text, +// cursor position. The model only ever reads text/cursor/alt-screen (never +// colors), so this is sufficient fidelity for safe predictions. Accepting +// the slice (rather than Terminal) lets tests drive @xterm/headless. +export interface SeedSource { + readonly rows: number + readonly buffer: { + readonly active: { + readonly type: string + readonly cursorX: number + readonly cursorY: number + readonly baseY: number + getLine(y: number): { translateToString(trimRight: boolean): string } | undefined + } + } +} + +export function buildModelSeedFromTerminal(term: SeedSource): string { + const buffer = term.buffer.active + let seed = '\x1bc' + if (buffer.type === 'alternate') seed += '\x1b[?1049h' + const lines: string[] = [] + for (let row = 0; row < term.rows; row += 1) { + const line = buffer.getLine(buffer.baseY + row) + lines.push(line ? line.translateToString(true) : '') + } + seed += lines.join('\r\n') + seed += `\x1b[${buffer.cursorY + 1};${buffer.cursorX + 1}H` + return seed +} + +export interface EchoRouterDeps { + // Write to the live terminal. The optional callback fires after the data + // (and everything queued before it) has been parsed. + write(data: string, callback?: () => void): void + // The engine, or null while it hasn't been constructed yet. + getEngine(): PredictiveEchoWithStatus | null + buildModelSeed(): string + getInputSrtt(): number | null + isViewportPinned(): boolean + scrollToBottom(): void +} + +export interface EchoRouter { + // One coalesced frame of server output. Owns the pinned-viewport scroll. + onServerOutput(combined: string): void + // User keystrokes destined for the PTY (visual echo only — the input send + // path is independent). + onUserInput(data: string): void + // Which sink server bytes currently take. Exposed for tests/diagnostics. + route(): 'direct' | 'engine' + dispose(): void +} + +export function createEchoRouter(deps: EchoRouterDeps): EchoRouter { + let route: 'direct' | 'engine' = 'direct' + let reseedPending = false + let heldChunks: string[] = [] + let reseedTimeout: ReturnType | null = null + let disposed = false + + // Single ordering domain for everything that reaches the live terminal. + // Engine writes happen inside the engine's ASYNC tail (its model parse + // awaits between chunks), so a direct write issued while engine work is + // still queued — e.g. a flood-bypass route flip — would overtake bytes + // that arrived earlier. Out-of-order bytes shear escape sequences and + // overlay row text (character-interleaved lines). Any operation enqueued + // while the chain is non-empty chains behind it; the chain collapses back + // to synchronous direct writes once it drains. + let queuedOps = 0 + let opChain: Promise = Promise.resolve() + + const enqueueOp = (op: () => Promise | void): void => { + queuedOps += 1 + opChain = opChain + .then(() => (disposed ? undefined : op())) + .catch(() => {}) + .then(() => { + queuedOps -= 1 + }) + } + + const writePinnedAware = (data: string, sink: (data: string) => void): void => { + const wasPinned = deps.isViewportPinned() + sink(data) + if (wasPinned) deps.scrollToBottom() + } + + // Engine route: always enqueued (the engine's tail is asynchronous). + const writeViaEngine = (engine: PredictiveEchoWithStatus, data: string): void => { + enqueueOp(() => { + if (deps.getEngine() !== engine) { + // Engine torn down while queued; the bytes still must render. + writePinnedAware(data, (chunk) => deps.write(chunk)) + return + } + let done: Promise = Promise.resolve() + writePinnedAware(data, (chunk) => { + done = engine.onServerOutput(chunk) + }) + return done + }) + } + + // Direct route: synchronous while the chain is empty (the common case — + // zero overhead on local sessions), ordered behind it otherwise. + const writeDirectOrdered = (data: string, callback?: () => void): void => { + if (queuedOps === 0) { + writePinnedAware(data, (chunk) => deps.write(chunk, callback)) + return + } + enqueueOp(() => { + writePinnedAware(data, (chunk) => deps.write(chunk, callback)) + }) + } + + const releaseHeldDirect = (): void => { + const held = heldChunks + heldChunks = [] + if (held.length === 0 || disposed) return + writeDirectOrdered(held.join('')) + } + + const beginEngineTransition = (): void => { + const engine = deps.getEngine() + if (disposed || reseedPending || !engine) return + reseedPending = true + heldChunks = [] + reseedTimeout = setTimeout(() => { + reseedTimeout = null + if (!reseedPending) return + // Capture never completed — stay on the direct route; a later + // keystroke retries the transition. + reseedPending = false + releaseHeldDirect() + }, RESEED_CAPTURE_TIMEOUT_MS) + // Quiescent-point capture: the callback fires after everything already + // queued to the live terminal has parsed, so the visible grid equals + // the byte stream consumed so far. Ordered, so it also lands after any + // writes still chained from a recent flood bypass. + writeDirectOrdered('', () => { + if (!reseedPending) return // timed out and aborted + reseedPending = false + if (reseedTimeout) { + clearTimeout(reseedTimeout) + reseedTimeout = null + } + if (disposed) { + heldChunks = [] + return + } + if (deps.getEngine() !== engine) { + // The engine changed mid-transition (today only possible via + // dispose, but defend anyway): the terminal is still live, so the + // held bytes must render — release them directly and stay on the + // direct route rather than dropping live PTY output. + releaseHeldDirect() + return + } + void engine.seed(deps.buildModelSeed()) + route = 'engine' + const held = heldChunks + heldChunks = [] + if (held.length > 0) { + writeViaEngine(engine, held.join('')) + } + }) + } + + // Timestamp of the last streaming-sized server chunk. 0 = never. + let lastStreamOutputAt = 0 + const streamIsActive = (): boolean => + lastStreamOutputAt !== 0 && Date.now() - lastStreamOutputAt < PREDICTION_QUIET_MS + + return { + onServerOutput(combined: string): void { + if (disposed || combined.length === 0) return + if (reseedPending) { + heldChunks.push(combined) + return + } + const streamingChunk = combined.length > PREDICTION_SAFE_CHUNK_BYTES + if (streamingChunk) lastStreamOutputAt = Date.now() + const engine = route === 'engine' ? deps.getEngine() : null + if (engine) { + if (streamingChunk && engine.hasPredictions) { + // The TUI started streaming with optimistic glyphs on screen. + // Erase them NOW, at positions that are still valid, instead of + // letting the engine erase/re-render them around every repaint + // frame — that cycle is what strands glyphs inside repainted + // panels and erases real rows when model and live drift. + enqueueOp(() => { + if (deps.getEngine() === engine) engine.rollback() + }) + } + if (combined.length > ENGINE_FLOOD_BYPASS_BYTES && !engine.hasPredictions) { + // Output flood with nothing predicted on screen: the model + // double-parse only delays the drain. Drop to direct; the next + // engaged keystroke reseeds the model before predicting again. + // Ordered: this write chains behind chunks still queued in the + // engine's tail — overtaking them would interleave row bytes. + route = 'direct' + writeDirectOrdered(combined) + } else { + writeViaEngine(engine, combined) + } + } else { + writeDirectOrdered(combined) + } + }, + onUserInput(data: string): void { + if (disposed) return + // While a reseed capture is in flight predictions stay off — the + // engine's model isn't authoritative yet. Only the optimistic echo is + // skipped; the keystroke still reaches the PTY via the send path. + if (reseedPending) return + // Predictions stay off while the server is actively streaming — the + // user sees output flowing (their echo included); optimistic glyphs + // only add an erase/re-render cycle per repaint frame. + if (streamIsActive()) return + const engine = deps.getEngine() + if (!engine) return + if (route === 'engine') { + // Only hand keystrokes to the engine at a chain-quiescent point. + // The engine places optimistic glyphs by reading its confirmed + // model's cursor SYNCHRONOUSLY, but while ops are queued a server + // chunk has already reached the live terminal whose model parse is + // still in flight — the model cursor is stale. A prediction placed + // then renders on the wrong row, and the engine's cleanup skips + // erasing glyphs whose row no longer matches its prediction row, + // stranding them on screen (the stacked-frame-remnant / displaced- + // echo corruption class). Skipping only costs the optimistic echo; + // the keystroke still reaches the PTY and echoes back. + if (queuedOps === 0) engine.onUserInput(data) + return + } + const srtt = deps.getInputSrtt() + if (srtt === null || srtt < PREDICTION_ENGAGE_THRESHOLD_MS) return + // Latency crossed the engage threshold while on the direct route: + // bring the engine's model up to date, then let predictions resume. + beginEngineTransition() + }, + route(): 'direct' | 'engine' { + return route + }, + dispose(): void { + disposed = true + reseedPending = false + heldChunks = [] + if (reseedTimeout) { + clearTimeout(reseedTimeout) + reseedTimeout = null + } + } + } +} diff --git a/src/renderer/src/lib/ipc-mock.ts b/src/renderer/src/lib/ipc-mock.ts index 47d896aa..fbc489b2 100644 --- a/src/renderer/src/lib/ipc-mock.ts +++ b/src/renderer/src/lib/ipc-mock.ts @@ -129,6 +129,11 @@ export interface PearMockHarness { injectBrokerEvent: (event: BrokerEventLike) => void injectBrokerEvents: (events: BrokerEventLike[]) => void injectPtyChunk: (projectId: string, name: string, chunk: string) => void + // Rendering-harness knobs: force the input SRTT the renderer polls (so the + // predictive-echo engine route engages) and echo typed bytes back through + // the PTY stream after a delay (so predictions reconcile end-to-end). + setInputSrtt: (ms: number | null) => void + setInputEcho: (options: { delayMs: number } | null) => void spawnAgents: (count: number, options?: { projectId?: string; channel?: string; namePrefix?: string }) => void openChannel: (projectId: string, channelName: string) => void openAgents: (projectId?: string) => void @@ -229,6 +234,10 @@ function createState(): MockState { let state = createState() let seq = 0 +// Rendering-harness knobs (setInputSrtt / setInputEcho): survive reset() so +// a spec can configure them before booting agents. +let mockInputSrttMs: number | null = null +let mockInputEchoDelayMs: number | null = null const mockNow = new Date('2026-06-09T09:42:00.000Z').getTime() @@ -827,7 +836,16 @@ export const pearMock: PearAPI = { snapshot: { rows: input.rows || 24, cols: input.cols || 80, cursor: [0, 0], screen: '' } }), sendInput: async (_projectId: string | undefined, name: string, data: string) => ({ name, bytes_written: data.length }), - sendInputFast: () => undefined, + sendInputFast: (projectId: string | undefined, name: string, data: string) => { + // Optional raw echo back through the PTY stream (setInputEcho) so the + // rendering harnesses can exercise predictive echo end-to-end: type → + // optimistic glyph → delayed authoritative echo → reconcile. + if (mockInputEchoDelayMs === null) return + const resolvedProject = projectId || state.activeId || defaultProject.id + setTimeout(() => { + pearMockHarness.injectPtyChunk(resolvedProject, name, data) + }, mockInputEchoDelayMs) + }, setTerminalMode: async (projectId: string | undefined, name: string, mode: TerminalAttachMode): Promise => { state.terminalModes.set(key(projectId, name), mode) return { name, mode: mode === 'drive' ? 'manual_flush' : 'auto_inject', flushed: 0, pending: 0 } @@ -835,7 +853,7 @@ export const pearMock: PearAPI = { getPending: async (): Promise => [], flushPending: async () => ({ flushed: 0 }), resizePty: async () => undefined, - inputSrtt: async () => null, + inputSrtt: async () => mockInputSrttMs, sendMessage: async (projectId: string | undefined, input: BrokerSendMessageInput) => { handleInjectedBrokerEvent({ kind: 'relay_inbound', @@ -1126,6 +1144,12 @@ export const pearMockHarness: PearMockHarness = { state.ptyChunks[ptyKey] = [...(state.ptyChunks[ptyKey] || []), chunk] for (const listener of [...state.ptyChunkListeners]) listener(projectId, name, chunk) }, + setInputSrtt: (ms: number | null) => { + mockInputSrttMs = ms + }, + setInputEcho: (options: { delayMs: number } | null) => { + mockInputEchoDelayMs = options ? options.delayMs : null + }, spawnAgents: (count: number, options?: { projectId?: string; channel?: string; namePrefix?: string }) => { const projectId = options?.projectId || state.activeId || defaultProject.id const channel = options?.channel || 'general' diff --git a/src/renderer/src/lib/predictive-echo.ts b/src/renderer/src/lib/predictive-echo.ts index f4097893..37823ff0 100644 --- a/src/renderer/src/lib/predictive-echo.ts +++ b/src/renderer/src/lib/predictive-echo.ts @@ -57,8 +57,16 @@ class HeadlessScreenModel implements ScreenModel { } } +// The published PredictiveEcho interface omits the engine's hasPredictions +// getter; the runtime's echo router needs it to decide whether an output +// flood can bypass the model double-parse (only safe with no optimistic +// glyphs on screen to reconcile). +export interface PredictiveEchoWithStatus extends PredictiveEcho { + readonly hasPredictions: boolean +} + export interface PredictiveEchoHandle { - engine: PredictiveEcho + engine: PredictiveEchoWithStatus dispose: () => void } diff --git a/src/renderer/src/lib/pty-size-sync.test.ts b/src/renderer/src/lib/pty-size-sync.test.ts new file mode 100644 index 00000000..acf1fec5 --- /dev/null +++ b/src/renderer/src/lib/pty-size-sync.test.ts @@ -0,0 +1,159 @@ +// Invariants for the retrying PTY size synchronizer. The property that +// matters: the LAST desired size always eventually reaches the PTY — a +// failed resize IPC must never be recorded as delivered (that left the PTY +// permanently mismatched with the grid, making TUI repaints land mid-screen +// and stack in scrollback). + +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { createPtySizeSync } from './pty-size-sync' + +type Deferred = { + rows: number + cols: number + resolve: () => void + reject: () => void +} + +function makeHarness(): { sync: ReturnType; sends: Deferred[] } { + const sends: Deferred[] = [] + const sync = createPtySizeSync( + (rows, cols) => + new Promise((resolve, reject) => { + sends.push({ rows, cols, resolve: () => resolve(), reject: () => reject(new Error('ipc failed')) }) + }) + ) + return { sync, sends } +} + +// Let the promise chain settle between steps. +const tick = (): Promise => Promise.resolve().then(() => {}) + +describe('pty-size-sync', () => { + beforeEach(() => { + vi.useFakeTimers() + }) + + afterEach(() => { + vi.useRealTimers() + }) + + it('sends the desired size once and skips no-op re-asserts after the ack', async () => { + const { sync, sends } = makeHarness() + sync.setDesired(24, 80) + expect(sends).toHaveLength(1) + sends[0].resolve() + await tick() + + // Same size again (ResizeObserver spam): no new IPC. + sync.setDesired(24, 80) + expect(sends).toHaveLength(1) + + // A real change round-trips. + sync.setDesired(30, 100) + expect(sends).toHaveLength(2) + expect(sends[1]).toMatchObject({ rows: 30, cols: 100 }) + }) + + it('retries a failed send until it lands — failure is never recorded as delivered', async () => { + const { sync, sends } = makeHarness() + sync.setDesired(24, 80) + sends[0].reject() + await tick() + expect(sends).toHaveLength(1) + + // Backoff elapses → retry fires with the same size. + vi.advanceTimersByTime(250) + expect(sends).toHaveLength(2) + expect(sends[1]).toMatchObject({ rows: 24, cols: 80 }) + + // Second failure backs off longer, then still retries. + sends[1].reject() + await tick() + vi.advanceTimersByTime(500) + expect(sends).toHaveLength(3) + sends[2].resolve() + await tick() + + // Delivered now — re-asserting the same size is a no-op. + sync.setDesired(24, 80) + expect(sends).toHaveLength(3) + }) + + it('re-sends when desired moved while a send was in flight', async () => { + const { sync, sends } = makeHarness() + sync.setDesired(24, 80) + sync.setDesired(30, 100) // supersedes while #0 in flight + expect(sends).toHaveLength(1) + + sends[0].resolve() + await tick() + expect(sends).toHaveLength(2) + expect(sends[1]).toMatchObject({ rows: 30, cols: 100 }) + sends[1].resolve() + await tick() + expect(sends).toHaveLength(2) + }) + + it('a stale failed send never clobbers a newer delivered size', async () => { + const { sync, sends } = makeHarness() + sync.setDesired(24, 80) + sends[0].reject() + await tick() + + sync.setDesired(30, 100) + expect(sends).toHaveLength(2) + sends[1].resolve() + await tick() + + // The old failure's backoff retry fires — but desired === acked now, so + // no stale 24×80 is ever re-sent. + vi.advanceTimersByTime(10_000) + expect(sends).toHaveLength(2) + }) + + it('noteAcked suppresses the IPC but re-asserts a diverged desired size', async () => { + const { sync, sends } = makeHarness() + // Attach carried 24×80 to the PTY out-of-band. + sync.noteAcked(24, 80) + expect(sends).toHaveLength(0) + + // Grid measured something else while attach was in flight → re-assert. + sync.setDesired(26, 90) + expect(sends).toHaveLength(1) + expect(sends[0]).toMatchObject({ rows: 26, cols: 90 }) + }) + + it('invalidateAck forces a re-send of an unchanged size (SIGWINCH bounce restore)', async () => { + const { sync, sends } = makeHarness() + sync.setDesired(24, 80) + sends[0].resolve() + await tick() + + sync.invalidateAck() + sync.setDesired(24, 80) + expect(sends).toHaveLength(2) + expect(sends[1]).toMatchObject({ rows: 24, cols: 80 }) + }) + + it('dispose cancels pending retries', async () => { + const { sync, sends } = makeHarness() + sync.setDesired(24, 80) + sends[0].reject() + await tick() + + sync.dispose() + vi.advanceTimersByTime(60_000) + expect(sends).toHaveLength(1) + + sync.setDesired(30, 100) + expect(sends).toHaveLength(1) + }) + + it('ignores degenerate sizes', () => { + const { sync, sends } = makeHarness() + sync.setDesired(0, 80) + sync.setDesired(24, 0) + sync.setDesired(-1, -1) + expect(sends).toHaveLength(0) + }) +}) diff --git a/src/renderer/src/lib/pty-size-sync.ts b/src/renderer/src/lib/pty-size-sync.ts new file mode 100644 index 00000000..b2549966 --- /dev/null +++ b/src/renderer/src/lib/pty-size-sync.ts @@ -0,0 +1,103 @@ +// Retrying desired/acked PTY size synchronizer. +// +// The terminal grid (xterm) and the PTY winsize must converge to the same +// rows×cols. When they diverge persistently, a TUI computes its repaint +// cursor positions against a row count the renderer doesn't have — repaints +// land mid-screen and stack in scrollback instead of overwriting in place. +// +// The failure mode this kills: a resizePty IPC fails transiently (broker +// busy, stream rebind, agent restarting) while the caller records the size +// as "sent" — the gate then suppresses every subsequent identical resize and +// the mismatch becomes permanent. Here a size only graduates from `desired` +// to `acked` when the IPC RESOLVES; failures retry with capped backoff until +// the current desired size lands or a newer one supersedes it. + +export interface PtySizeSync { + /** The grid measured rows×cols; sync (or queue a retrying sync) to the PTY. */ + setDesired(rows: number, cols: number): void + /** + * Record that the PTY is known to be at rows×cols via a side channel + * (e.g. the size carried on attachTerminal), without sending an IPC. + * Re-asserts `desired` if it diverged while the side channel was in flight. + */ + noteAcked(rows: number, cols: number): void + /** + * Forget the ack after a deliberate out-of-band resize (the SIGWINCH + * bounce), so the next setDesired re-sends even an unchanged size. + */ + invalidateAck(): void + /** Cancel any pending retry. Further calls are no-ops. */ + dispose(): void +} + +const INITIAL_RETRY_DELAY_MS = 250 +const MAX_RETRY_DELAY_MS = 5_000 + +export function createPtySizeSync( + send: (rows: number, cols: number) => Promise +): PtySizeSync { + let desiredRows = -1 + let desiredCols = -1 + let ackedRows = -1 + let ackedCols = -1 + let inFlight = false + let disposed = false + let retryTimer: ReturnType | null = null + let retryDelayMs = INITIAL_RETRY_DELAY_MS + + const sync = (): void => { + if (disposed || inFlight) return + if (desiredRows <= 0 || desiredCols <= 0) return + if (desiredRows === ackedRows && desiredCols === ackedCols) return + const rows = desiredRows + const cols = desiredCols + inFlight = true + send(rows, cols) + .then(() => { + inFlight = false + ackedRows = rows + ackedCols = cols + retryDelayMs = INITIAL_RETRY_DELAY_MS + // Desired may have moved while this send was in flight. + sync() + }) + .catch(() => { + inFlight = false + if (disposed) return + if (retryTimer) clearTimeout(retryTimer) + retryTimer = setTimeout(() => { + retryTimer = null + sync() + }, retryDelayMs) + retryDelayMs = Math.min(retryDelayMs * 2, MAX_RETRY_DELAY_MS) + }) + } + + return { + setDesired(rows: number, cols: number): void { + desiredRows = rows + desiredCols = cols + sync() + }, + noteAcked(rows: number, cols: number): void { + ackedRows = rows + ackedCols = cols + if (desiredRows <= 0) { + desiredRows = rows + desiredCols = cols + } + sync() + }, + invalidateAck(): void { + ackedRows = -1 + ackedCols = -1 + }, + dispose(): void { + disposed = true + if (retryTimer) { + clearTimeout(retryTimer) + retryTimer = null + } + } + } +} diff --git a/src/renderer/src/lib/terminal-runtime-registry.ts b/src/renderer/src/lib/terminal-runtime-registry.ts index 1ff66b3a..add889d2 100644 --- a/src/renderer/src/lib/terminal-runtime-registry.ts +++ b/src/renderer/src/lib/terminal-runtime-registry.ts @@ -26,11 +26,14 @@ import { clearPtyBuffer, diagPtyEnabled, flushPtyChunksNow, - getPtyChunks, + getPtyChunkTotal, + getPtyChunksSinceTotal, subscribePtyBuffer } from '@/stores/pty-buffer-store' import { recordChunkEchoed } from '@/lib/typing-trace' -import { createPredictiveEcho } from '@/lib/predictive-echo' +import { createPredictiveEcho, type PredictiveEchoWithStatus } from '@/lib/predictive-echo' +import { createPtySizeSync } from '@/lib/pty-size-sync' +import { buildModelSeedFromTerminal, createEchoRouter } from '@/lib/echo-router' import type { PredictiveEcho } from '@agent-relay/harness-driver/predictive-echo' import { awaitFontSettle } from '@/lib/font-settle' import type { Theme } from '@/stores/ui-store' @@ -147,6 +150,10 @@ export interface TerminalRuntime { // through a runtime-owned slot to allow rebinding on each hook effect. setInputSrttGetter(getter: () => number | null): void getPredictiveEcho(): PredictiveEcho | null + // Forward user keystrokes for optimistic echo. Routes through the + // predictive-echo engine only when measured latency warrants predictions, + // transitioning (and reseeding the engine's confirmed model) on demand. + noteUserInput(data: string): void // Install a handler for `term.onData`. Returns the previous handler so the // caller can re-install it later (e.g. on unmount while keeping the // runtime alive). The runtime forwards via an internal mutable slot, so @@ -288,22 +295,45 @@ function createRuntime( let disposed = false let currentToken: symbol | null = null let webglAddon: WebglAddon | null = null - let predictiveEcho: PredictiveEcho | null = null + let predictiveEcho: PredictiveEchoWithStatus | null = null let disposePredictiveEcho: (() => void) | null = null let unsubBuffer: (() => void) | null = null - let writtenChunks = 0 + // Replay baseline: the buffer's monotonic chunk total at the moment the + // attach snapshot was painted. Chunks at or below this total are already + // on screen (inside the snapshot); only chunks past it get replayed. + let writtenTotal = 0 let attachSeeded = false let attachInFlight = false let pendingInitFrame: number | null = null - // Last rows/cols actually sent to the PTY. fitAndSync drops the IPC when - // the size hasn't changed — observers fire on every dragged pixel and the - // backend reacts to no-op resizes by reflowing. - let lastSentRows = -1 - let lastSentCols = -1 + // PTY size sync: a size only counts as delivered when the resizePty IPC + // RESOLVES; failures retry with backoff until the current desired size + // lands. The previous change-gate recorded a size as "sent" even when the + // IPC failed, leaving the PTY permanently mismatched with the rendered + // grid — a TUI then positions repaints against the wrong row count, so + // content lands mid-screen and repaints stack in scrollback instead of + // overwriting. See pty-size-sync.ts for the invariants (unit-tested). + const sizeSync = createPtySizeSync((rows, cols) => + pear.broker.resizePty(opts.projectId, opts.agentName, rows, cols) + ) // Holder for the current input-SRTT getter. The predictive echo engine // captures this once on construction, so we wrap it in a trampoline and // let setInputSrttGetter swap the underlying getter on each effect run. let currentSrttGetter: () => number | null = opts.getInputSrtt + // Routes server output direct-to-xterm while the predictive-echo engine + // is dormant and through the engine once latency warrants predictions. + // See echo-router.ts for the invariants (equivalence-tested against the + // real engine + parser). + const echoRouter = createEchoRouter({ + write: (data, callback) => { + if (disposed || !term) return + term.write(data, callback) + }, + getEngine: () => (disposed ? null : predictiveEcho), + buildModelSeed: () => (term ? buildModelSeedFromTerminal(term) : '\x1bc'), + getInputSrtt: () => currentSrttGetter(), + isViewportPinned: () => (term ? isViewportPinnedToBottom(term) : false), + scrollToBottom: () => term?.scrollToBottom() + }) const cancelPendingInit = (): void => { if (pendingInitFrame !== null) { @@ -371,7 +401,6 @@ function createRuntime( const seedBufferSubscription = (): void => { if (unsubBuffer || !term || disposed) return - const liveTerm = term const writeChunks = (newChunks: string[]): void => { if (disposed || !term) return @@ -383,7 +412,6 @@ function createRuntime( // eslint-disable-next-line no-console console.log(`[diag:runtime:writeChunks] key=${key} count=${newChunks.length} firstPreview="${newChunks[0]?.slice(0, 80).replace(/\n/g, '\\n').replace(/\r/g, '\\r').replace(/\x1b/g, '\\e')}"`) } - const wasPinned = isViewportPinnedToBottom(liveTerm) // Typing-trace accounting stays per-chunk: recordChunkEchoed consumes // one pending keystroke per call, so coalescing it would under-count. for (const chunk of newChunks) recordChunkEchoed(chunk) @@ -396,21 +424,16 @@ function createRuntime( // keep up and input lagged. Byte content and order are unchanged, so the // one-write-per-byte invariant holds. const combined = newChunks.length === 1 ? newChunks[0] : newChunks.join('') - if (predictiveEcho) { - void predictiveEcho.onServerOutput(combined) - } else { - liveTerm.write(combined) - } - if (wasPinned) liveTerm.scrollToBottom() + echoRouter.onServerOutput(combined) } unsubBuffer = subscribePtyBuffer(key, writeChunks) // Initial replay: pull whatever is already in the buffer past the - // snapshot baseline (writtenChunks). The listener only receives tails + // snapshot baseline (writtenTotal). The listener only receives tails // from this point on, so we have to do the catch-up explicitly here. - const buffered = getPtyChunks(key) - if (writtenChunks > buffered.length) writtenChunks = 0 - writeChunks(buffered.slice(writtenChunks)) + // The monotonic-total slice can never replay pre-baseline chunks on top + // of the snapshot, even if the buffer trimmed during the attach window. + writeChunks(getPtyChunksSinceTotal(key, writtenTotal)) } const attachAndSeed = async ( @@ -443,10 +466,16 @@ function createRuntime( // them into the buffer AFTER we capture writtenChunks, and the // subsequent subscribe would replay them on top of the snapshot. flushPtyChunksNow(key) - writtenChunks = getPtyChunks(key).length + writtenTotal = getPtyChunkTotal(key) shouldReplay = false } attachSeeded = true + if (initialSize) { + // attachTerminal carried rows/cols, so the PTY is at initialSize + // now. Record the ack; if a refit moved `desired` during the IPC + // roundtrip, the sync loop re-asserts it immediately. + sizeSync.noteAcked(initialSize.rows, initialSize.cols) + } } catch (err) { console.error('[terminal] attachTerminal failed:', err) // Don't latch attachSeeded — the next init/mount cycle should retry. @@ -457,7 +486,7 @@ function createRuntime( if (disposed || !term) return if (shouldReplay) { - writtenChunks = 0 + writtenTotal = 0 await predictiveEcho?.seed('') } @@ -479,19 +508,26 @@ function createRuntime( if (disposed || !liveTerm) return const { rows, cols } = liveTerm if (rows <= 1 || cols <= 0) return - pear.broker + // The (rows-1) intermediate is best-effort and deliberately bypasses + // the sync loop; the restore below ALWAYS runs (even when the + // intermediate fails) and goes through the retrying sync loop, so a + // dropped restore IPC can never strand the PTY one row short — the + // exact mismatch that makes TUI repaints stack down the screen. + void pear.broker .resizePty(opts.projectId, opts.agentName, rows - 1, cols) + .catch(() => {}) .then(() => { - if (disposed || !liveTerm) return - // Re-read dimensions across the async boundary — the user may - // have resized the pane between the first and second IPC. - // Sending stale dims would regress the PTY size. + if (disposed) return + // The PTY may now be at the bounce size; invalidate the ack so the + // sync loop re-asserts the real size even if it matches the last + // acked value. Re-read dimensions across the async boundary — the + // user may have resized the pane since the first IPC. + sizeSync.invalidateAck() const currentRows = liveTerm.rows const currentCols = liveTerm.cols if (currentRows <= 0 || currentCols <= 0) return - return pear.broker.resizePty(opts.projectId, opts.agentName, currentRows, currentCols) + sizeSync.setDesired(currentRows, currentCols) }) - .catch(() => {}) }, 200) } @@ -606,6 +642,8 @@ function createRuntime( // writeFromBuffer notification triggered by clearPtyBuffer (with []) // runs while the closure is still consistent. cancelPendingInit() + sizeSync.dispose() + echoRouter.dispose() clearPtyBuffer(key) disposed = true currentToken = null @@ -654,15 +692,10 @@ function createRuntime( const size = tryFit() if (size) { predictiveEcho?.onResize(size.cols, size.rows) - // Fix #9: ResizeObserver fires on every dragged pixel; only - // round-trip to the backend when the cell grid actually changed. - if (size.rows !== lastSentRows || size.cols !== lastSentCols) { - lastSentRows = size.rows - lastSentCols = size.cols - pear.broker - .resizePty(opts.projectId, opts.agentName, size.rows, size.cols) - .catch(() => {}) - } + // Fix #9: ResizeObserver fires on every dragged pixel; the sync + // loop only round-trips when the grid differs from the last ACKED + // size, and keeps retrying until the PTY confirms it. + sizeSync.setDesired(size.rows, size.cols) } return size }, @@ -681,6 +714,10 @@ function createRuntime( getPredictiveEcho(): PredictiveEcho | null { return predictiveEcho }, + noteUserInput(data: string): void { + if (disposed || !term) return + echoRouter.onUserInput(data) + }, setOnData(handler: ((data: string) => void) | null): void { onDataHandler = handler }, @@ -729,8 +766,22 @@ export function readRuntimeViewportText(projectId: string | undefined, name: str return lines.join('\n') } +export function readRuntimeViewportDims( + projectId: string | undefined, + name: string +): { rows: number; cols: number } | null { + const record = runtimes.get(getAgentKey(projectId, name)) + if (!record) return null + const term = record.runtime.term + return { rows: term.rows, cols: term.cols } +} + if (import.meta.env.VITE_PEAR_MOCK_IPC === 'true' && typeof window !== 'undefined') { ;(window as unknown as { __pearReadTerminalViewport?: (projectId: string | undefined, name: string) => string | null + __pearReadTerminalDims?: (projectId: string | undefined, name: string) => { rows: number; cols: number } | null }).__pearReadTerminalViewport = readRuntimeViewportText + ;(window as unknown as { + __pearReadTerminalDims?: (projectId: string | undefined, name: string) => { rows: number; cols: number } | null + }).__pearReadTerminalDims = readRuntimeViewportDims } diff --git a/src/renderer/src/stores/pty-buffer-store.test.ts b/src/renderer/src/stores/pty-buffer-store.test.ts index a2a2ddd4..fa397f9d 100644 --- a/src/renderer/src/stores/pty-buffer-store.test.ts +++ b/src/renderer/src/stores/pty-buffer-store.test.ts @@ -4,6 +4,8 @@ import { clearPtyBuffer, flushPtyChunksNow, getPtyChunks, + getPtyChunkTotal, + getPtyChunksSinceTotal, subscribePtyBuffer } from './pty-buffer-store' @@ -163,3 +165,113 @@ describe('pty-buffer-store', () => { expect(getPtyChunks('k1')).toEqual(['dup', 'dup']) }) }) + +// Invariants for the monotonic replay-baseline API (getPtyChunkTotal / +// getPtyChunksSinceTotal). The terminal runtime paints an attach snapshot, +// captures a baseline, then replays only chunks that arrived after it. The +// baseline must stay valid across the 10k buffer trim: a length-based +// baseline goes stale when the buffer slides, and the old recovery path +// (reset to 0, replay the whole buffer) painted duplicate screen content on +// top of the snapshot. Totals are monotonic per key for the renderer's +// lifetime — including across clearPtyBuffer — so these tests use unique +// keys instead of clearing shared ones. +describe('pty-buffer-store replay baselines', () => { + const MAX_PTY_BUFFER_CHUNKS = 10_000 + let keySerial = 0 + const testKeys: string[] = [] + + function trackedKey(label: string): string { + keySerial += 1 + const key = `baseline:${label}:${keySerial}` + testKeys.push(key) + return key + } + + beforeEach(() => { + vi.useFakeTimers() + }) + + afterEach(() => { + for (const key of testKeys.splice(0)) clearPtyBuffer(key) + vi.useRealTimers() + }) + + it('counts flushed chunks monotonically with and without subscribers', () => { + const key = trackedKey('totals') + expect(getPtyChunkTotal(key)).toBe(0) + + // No subscriber: inline flush. + appendPtyChunk(key, 'a') + appendPtyChunk(key, 'b') + expect(getPtyChunkTotal(key)).toBe(2) + + // With subscriber: rAF-batched flush counts at flush time. + const unsub = subscribePtyBuffer(key, () => {}) + appendPtyChunk(key, 'c') + expect(getPtyChunkTotal(key)).toBe(2) + flushRaf() + expect(getPtyChunkTotal(key)).toBe(3) + unsub() + }) + + it('returns only post-baseline chunks', () => { + const key = trackedKey('since') + appendPtyChunk(key, 'before-1') + appendPtyChunk(key, 'before-2') + const baseline = getPtyChunkTotal(key) + + appendPtyChunk(key, 'after-1') + appendPtyChunk(key, 'after-2') + + expect(getPtyChunksSinceTotal(key, baseline)).toEqual(['after-1', 'after-2']) + expect(getPtyChunksSinceTotal(key, getPtyChunkTotal(key))).toEqual([]) + }) + + it('never replays pre-baseline chunks after the buffer trims', () => { + const key = trackedKey('trim') + appendPtyChunk(key, 'snapshot-covered') + const baseline = getPtyChunkTotal(key) + + // Overflow the buffer so the pre-baseline chunk is trimmed away. + const appended = MAX_PTY_BUFFER_CHUNKS + 50 + for (let i = 0; i < appended; i += 1) { + appendPtyChunk(key, `post-${i}`) + } + + const replay = getPtyChunksSinceTotal(key, baseline) + // Everything retained is post-baseline; the trimmed-away middle is + // unrecoverable, but nothing from before the baseline is re-returned. + expect(replay.length).toBe(MAX_PTY_BUFFER_CHUNKS) + expect(replay[0]).toBe(`post-${appended - MAX_PTY_BUFFER_CHUNKS}`) + expect(replay.at(-1)).toBe(`post-${appended - 1}`) + expect(replay).not.toContain('snapshot-covered') + }) + + it('keeps totals monotonic across clearPtyBuffer', () => { + const key = trackedKey('clear') + appendPtyChunk(key, 'one') + const baseline = getPtyChunkTotal(key) + + clearPtyBuffer(key) + expect(getPtyChunks(key)).toEqual([]) + // A baseline captured before the clear stays valid: nothing new yet. + expect(getPtyChunksSinceTotal(key, baseline)).toEqual([]) + + appendPtyChunk(key, 'two') + expect(getPtyChunkTotal(key)).toBe(baseline + 1) + expect(getPtyChunksSinceTotal(key, baseline)).toEqual(['two']) + }) + + it('flushPtyChunksNow drains staged chunks into the total before a baseline capture', () => { + const key = trackedKey('drain') + const unsub = subscribePtyBuffer(key, () => {}) + appendPtyChunk(key, 'staged') + // Without the synchronous drain the staged chunk would be invisible to + // the baseline and replayed on top of the snapshot. + flushPtyChunksNow(key) + const baseline = getPtyChunkTotal(key) + expect(baseline).toBe(1) + expect(getPtyChunksSinceTotal(key, baseline)).toEqual([]) + unsub() + }) +}) diff --git a/src/renderer/src/stores/pty-buffer-store.ts b/src/renderer/src/stores/pty-buffer-store.ts index b3e564a1..dc7fa950 100644 --- a/src/renderer/src/stores/pty-buffer-store.ts +++ b/src/renderer/src/stores/pty-buffer-store.ts @@ -22,6 +22,13 @@ type Listener = (newChunks: string[]) => void const buffers = new Map() const listeners = new Map>() +// Monotonic count of chunks ever flushed into each key's buffer. Unlike +// `buffers.get(key).length` this never moves backwards on trim, so consumers +// can capture it as a replay baseline that stays valid across the 10k trim — +// comparing against buffer length instead made a trimmed baseline look +// "past the end", and the recovery path replayed the WHOLE buffer on top of +// an already-painted snapshot (the duplicated-screen-content bug class). +const totals = new Map() // Chunks staged for the next animation frame, keyed by agent key. const pending = new Map() @@ -72,6 +79,7 @@ function flushPending(key: string): void { ? combined.slice(combined.length - MAX_PTY_BUFFER_CHUNKS) : combined buffers.set(key, trimmed) + totals.set(key, (totals.get(key) ?? 0) + queued.length) const keyListeners = listeners.get(key) if (!keyListeners || keyListeners.size === 0) return @@ -88,6 +96,26 @@ export function getPtyChunks(key: string): string[] { return buffers.get(key) ?? [] } +// Monotonic count of chunks ever flushed into this key's buffer. Capture it +// (after flushPtyChunksNow) as a baseline, then read the chunks that arrived +// after the baseline with getPtyChunksSinceTotal. +export function getPtyChunkTotal(key: string): number { + return totals.get(key) ?? 0 +} + +// Chunks flushed after `baselineTotal` that are still in the buffer. When +// more chunks arrived than the buffer retains (trim during the window), the +// trimmed-away middle is unrecoverable — return everything retained rather +// than nothing, and never re-return chunks from before the baseline (no +// duplicate replay). +export function getPtyChunksSinceTotal(key: string, baselineTotal: number): string[] { + const buffer = buffers.get(key) ?? [] + const missed = (totals.get(key) ?? 0) - baselineTotal + if (missed <= 0) return [] + if (missed >= buffer.length) return buffer.slice() + return buffer.slice(buffer.length - missed) +} + // Synchronously drain any chunks staged for the next rAF into the buffer. // Used by the terminal runtime right before reading the buffer length as a // snapshot baseline — otherwise pending chunks would be replayed on top of @@ -127,7 +155,6 @@ function __previewChunk(chunk: string): string { export function appendPtyChunk(key: string, chunk: string): void { if (diagPtyEnabled()) { __appendSeq += 1 - // eslint-disable-next-line no-console console.log(`[diag:pty-append] #${__appendSeq} key=${key} bytes=${chunk.length} preview="${__previewChunk(chunk)}"`) } const queue = pending.get(key) diff --git a/src/renderer/src/stores/ui-store.ts b/src/renderer/src/stores/ui-store.ts index d302360e..0309e86a 100644 --- a/src/renderer/src/stores/ui-store.ts +++ b/src/renderer/src/stores/ui-store.ts @@ -239,8 +239,16 @@ function tabInputForViewMode(mode: ViewMode): AppTabInput { // The browser demo build has no real projects, so default straight to the // Attention Inbox (the web-first surface) instead of the empty Agents view. +// The Playwright rendering harnesses (fidelity/redraw/stress) need the +// Agents view to mount terminals — they opt in via this localStorage key in +// an init script, the same way they seed pear-terminal-layout. const isWebMock = import.meta.env.VITE_PEAR_MOCK_IPC === 'true' -const initialTab = createTab({ kind: isWebMock ? 'issues' : 'agents' }) +const webMockInitialTab = readStored( + 'pear-web-initial-tab', + z.enum(['agents', 'issues']), + 'issues' +) +const initialTab = createTab({ kind: isWebMock ? webMockInitialTab : 'agents' }) export const useUIStore = create((set, get) => ({ viewMode: viewModeForTab(initialTab), diff --git a/tests/playwright/REDRAW-DRAIN-RESULTS.md b/tests/playwright/REDRAW-DRAIN-RESULTS.md index 9b07f8d7..008651bb 100644 --- a/tests/playwright/REDRAW-DRAIN-RESULTS.md +++ b/tests/playwright/REDRAW-DRAIN-RESULTS.md @@ -35,10 +35,14 @@ grid then keeps up — full drain, lag bounded to the current burst, longest frame nearly halved. Numbers are from a single dev-machine run (macOS, headless Chrome) and are -indicative, not thresholds — the suite asserts only that the harness ran -end-to-end. The byte-for-byte correctness of coalescing is proven separately by -the real-parser equivalence unit test -(`src/renderer/src/lib/pty-coalesce-equivalence.test.ts`). +indicative, not thresholds. The suite GATES on the machine-independent drain +invariants (`drained === true` within the deadline and +`finalFrameRendered === true`) so a parser-backlog regression fails CI, while +the timing metrics stay informational. Byte-for-byte correctness of +coalescing is proven separately by the real-parser equivalence unit test +(`src/renderer/src/lib/pty-coalesce-equivalence.test.ts`), and routing +correctness (direct/engine transitions, reseed, flood bypass) by +`src/renderer/src/lib/echo-router.test.ts`. ## Knobs diff --git a/tests/playwright/fidelity-no-duplication.spec.ts b/tests/playwright/fidelity-no-duplication.spec.ts index 8e884907..4d3e8552 100644 --- a/tests/playwright/fidelity-no-duplication.spec.ts +++ b/tests/playwright/fidelity-no-duplication.spec.ts @@ -60,6 +60,9 @@ async function bootWithAgents( ): Promise { await page.addInitScript((initialLayout) => { window.localStorage.setItem('pear-terminal-layout', initialLayout) + // The web-mock build defaults to the Attention Inbox tab; the fidelity + // harness needs the Agents view so terminal runtimes mount. + window.localStorage.setItem('pear-web-initial-tab', 'agents') }, layout) await page.goto('/') @@ -71,6 +74,8 @@ async function bootWithAgents( mock.spawnAgents(agentCount, { projectId: 'mock-project', namePrefix, channel: 'general' }) }, { agentCount: count, namePrefix: prefix }) + // Select the spawned agent so its terminal runtime mounts (the mock build + // boots with demo agents; spawning does not auto-select). const firstAgent = agentName(prefix) await page.getByText('general').click() // Role-scoped but not exact: the button's accessible name includes status @@ -329,6 +334,10 @@ test.describe('terminal output fidelity under high-rate PTY streaming', () => { const name = agentName('remount') await startStream(page, [{ name, markerPrefix: 'remount-a', total: VARIANT_TOTAL }]) + // Leave the Agents view entirely (Issues tab) and come back mid-stream — + // the cross-view unmount/remount that used to replay duplicate buffer + // content on top of the parked runtime. (The old 'agent graph' layout + // this test used no longer exists.) await page.waitForTimeout(120) await page.getByRole('button', { name: 'Show split terminal pages' }).click() await expectSplitTerminalVisible(page, name) diff --git a/tests/playwright/redraw-drain.spec.ts b/tests/playwright/redraw-drain.spec.ts index e509891e..1f1a5ce1 100644 --- a/tests/playwright/redraw-drain.spec.ts +++ b/tests/playwright/redraw-drain.spec.ts @@ -257,10 +257,14 @@ test.describe('renderer redraw drain', () => { appendFileSync(resultPath, `${JSON.stringify({ ...result, consoleErrorCount: consoleErrors.length, recordedAt: new Date().toISOString() })}\n`) } - // Non-gating: assert only that the harness ran end-to-end. The drain / - // backpressure numbers are recorded for the before/after comparison, not - // enforced as thresholds yet. expect(result.framesInjected, 'no redraw frames were injected').toBeGreaterThan(0) expect(result.finalRenderedSeq, 'terminal grid never rendered any redraw frame').toBeGreaterThan(0) + // GATING drain invariants (machine-independent booleans, not timing + // thresholds): after the burst stops, the grid must fully catch up + // within the drain deadline and the final repaint must be the one on + // screen. A regression that reintroduces parser backlog (the typing-lag + // class) shows up here as drained=false long before users feel it. + expect(result.drained, `terminal never drained within ${config.drainTimeoutMs}ms — render backlog regression: ${JSON.stringify(result)}`).toBe(true) + expect(result.finalFrameRendered, `final repaint never rendered: ${JSON.stringify(result)}`).toBe(true) }) }) diff --git a/tests/playwright/rendering-corruption.spec.ts b/tests/playwright/rendering-corruption.spec.ts new file mode 100644 index 00000000..fe456a1b --- /dev/null +++ b/tests/playwright/rendering-corruption.spec.ts @@ -0,0 +1,243 @@ +// Rendering-corruption reproduction harness. +// +// Recreates the live-session workload that produced the interleaved-rows / +// stacked-lines corruption reports: a Claude Code-style Ink TUI that +// repaints a panel in place with cursor-up + erase-down sequences, large +// integration-update text dumps, >32KB single-chunk floods (which trip the +// echo router's engine bypass), and a human typing mid-flood with the +// predictive-echo engine engaged (forced high input SRTT + delayed echo). +// +// Verification is byte-faithfulness, not markers: after the storm settles, +// the live terminal's parsed viewport must equal a reference @xterm/headless +// terminal fed the exact canonical chunk stream the renderer received. ANY +// dropped, duplicated, reordered, or misparsed byte — or a stranded +// prediction glyph — shows up as a diff. A screenshot is saved for human +// inspection either way. +// +// Run: npm run test:fidelity (shares the fidelity Playwright config). + +import { expect, test } from '@playwright/test' +import { Terminal } from '@xterm/headless' + +const PROJECT_ID = 'mock-project' +const AGENT = 'storm-0001' +const PANEL_ROWS = 12 + +// Deterministic PRNG so chunk boundaries are reproducible across runs. +function mulberry32(seed: number): () => number { + let a = seed + return () => { + a |= 0 + a = (a + 0x6d2b79f5) | 0 + let t = Math.imul(a ^ (a >>> 15), 1 | a) + t = (t + Math.imul(t ^ (t >>> 7), 61 | t)) ^ t + return ((t ^ (t >>> 14)) >>> 0) / 4294967296 + } +} + +function integrationDump(blocks: number): string { + let out = '' + for (let b = 0; b < blocks; b += 1) { + out += `\r\n` + out += 'Writeback discovery schemas and examples are mounted through ' + out += '.integrations/discovery// for connected integrations. ' + out += 'Local historical provider records are not broadly downloaded.\r\n' + out += `channels/C0B9F02GU9__gtm-world-cup-soccer/messages entry ${b} ` + out += 'live thread context roots are mounted on demand or through incoming events.\r\n' + } + return out +} + +function panelFrame(frame: number, first: boolean): string { + const lines: string[] = [] + for (let row = 0; row < PANEL_ROWS; row += 1) { + if (row === 0) { + lines.push(`\x1b[36mFRAME=${String(frame).padStart(5, '0')}\x1b[0m ── pear corruption repro panel ──`) + } else { + const spinner = '|/-\\'[frame % 4] + lines.push(`\x1b[2m│\x1b[0m ${spinner} row ${String(row).padStart(2, '0')} ${'█'.repeat((frame + row) % 40)}`) + } + } + const body = lines.join('\r\n') + // First paint appends; repaints move up over the previous panel and erase + // down — the Ink in-place redraw shape. + return first ? body : `\x1b[${PANEL_ROWS - 1}A\r\x1b[J${body}` +} + +function chunkRandomly(data: string, rng: () => number): string[] { + const chunks: string[] = [] + let i = 0 + while (i < data.length) { + const size = 64 + Math.floor(rng() * 2048) + chunks.push(data.slice(i, i + size)) + i += size + } + return chunks +} + +function readReferenceViewport(stream: string, rows: number, cols: number): Promise { + const term = new Terminal({ cols, rows, allowProposedApi: true, scrollback: 3000 }) + return new Promise((resolve) => { + term.write(stream, () => { + const buf = term.buffer.active + const lines: string[] = [] + for (let row = 0; row < rows; row += 1) { + const line = buf.getLine(buf.baseY + row) + lines.push(line ? line.translateToString(true) : '') + } + term.dispose() + resolve(lines.join('\n')) + }) + }) +} + +// Bisect knobs for iterating on a divergence: +// STORM_SRTT=0 → engine never engages (direct route only) +// STORM_TYPE=0 → no typing (no predictions, no echo traffic) +// STORM_CPU=4 → 4x CDP CPU throttling (reproduces slow-CI timing) +const STORM_SRTT = process.env.STORM_SRTT === '0' ? null : 80 +const STORM_TYPE = process.env.STORM_TYPE !== '0' +const STORM_CPU = Number(process.env.STORM_CPU || '1') + +test.describe('rendering corruption repro — flood + typing + engine routing', () => { + test('screen stays byte-faithful through an Ink-style storm with engaged predictions', async ({ page }, testInfo) => { + const consoleErrors: string[] = [] + page.on('console', (message) => { + if (message.type() === 'error') consoleErrors.push(message.text()) + }) + page.on('pageerror', (error) => consoleErrors.push(error.message)) + + await page.addInitScript(() => { + window.localStorage.setItem('pear-web-initial-tab', 'agents') + window.localStorage.setItem('pear-terminal-layout', 'tabs') + }) + if (STORM_CPU > 1) { + const cdp = await page.context().newCDPSession(page) + await cdp.send('Emulation.setCPUThrottlingRate', { rate: STORM_CPU }) + } + await page.goto('/') + await expect(page.getByText('Mock Project')).toBeVisible() + + await page.evaluate(({ projectId, srtt }) => { + const mock = window.__pearMock + if (!mock) throw new Error('window.__pearMock is not available') + // Engage the predictive-echo engine route: high input SRTT + delayed + // raw echo so optimistic glyphs render and later reconcile. + mock.setInputSrtt(srtt) + mock.setInputEcho(srtt === null ? null : { delayMs: 60 }) + mock.spawnAgents(1, { projectId, namePrefix: 'storm' }) + }, { projectId: PROJECT_ID, srtt: STORM_SRTT }) + + await page.locator('button').filter({ hasText: AGENT }).first().click() + const terminal = page.locator(`[data-testid="terminal-instance"][data-agent-name="${AGENT}"]`) + await expect(terminal).toBeVisible() + // Attach + SIGWINCH bounce + font settle. + await page.waitForTimeout(900) + await terminal.click() + + // Build the storm byte stream in the TEST (so the node-side reference + // sees the identical bytes), then drive it into the page in phases with + // typing interleaved between them. + const rng = mulberry32(0xc0ffee) + const phaseA = integrationDump(40) // scrolling text dump + let panel = panelFrame(0, true) + const repaintBatch = (from: number, count: number): string => { + let out = '' + for (let f = from; f < from + count; f += 1) out += panelFrame(f, false) + return out + } + const phaseB = panel + repaintBatch(1, 60) + // One chunk big enough to trip the engine flood bypass. + let flood = repaintBatch(61, 1) + let floodFrame = 62 + while (flood.length <= 40_000) { + flood += panelFrame(floodFrame, false) + floodFrame += 1 + } + const phaseD = repaintBatch(floodFrame, 40) + const FINAL_FRAME = `FRAME=${String(floodFrame + 39).padStart(5, '0')}` + + const inject = async (chunks: string[]): Promise => { + await page.evaluate(async ({ projectId, agent, payload }) => { + const mock = window.__pearMock + if (!mock) throw new Error('mock missing') + for (let i = 0; i < payload.length; i += 1) { + mock.injectPtyChunk(projectId, agent, payload[i]) + if (i % 6 === 5) { + await new Promise((resolve) => requestAnimationFrame(() => resolve())) + } + } + }, { projectId: PROJECT_ID, agent: AGENT, payload: chunks }) + } + + // Phase A: text dump while the user types (predictions engage here). + const typingA = STORM_TYPE ? page.keyboard.type('status check', { delay: 45 }) : Promise.resolve() + await inject(chunkRandomly(phaseA, rng)) + await typingA + + // Phase B: in-place repaints, randomly split mid-sequence. + const typingB = STORM_TYPE ? page.keyboard.type('and more', { delay: 35 }) : Promise.resolve() + await inject(chunkRandomly(phaseB, rng)) + await typingB + + // Phase C: the >32KB flood lands as ONE chunk (engine bypass), with a + // keystroke racing it. + const typingC = STORM_TYPE ? page.keyboard.type('x', { delay: 10 }) : Promise.resolve() + await inject([flood]) + await typingC + + // Phase D: post-flood repaints on the direct route. + await inject(chunkRandomly(phaseD, rng)) + + // Settle: final repaint visible, echoes drained, engine tail flushed. + await expect + .poll( + () => + page.evaluate( + ({ projectId, agent }) => + (window as unknown as { + __pearReadTerminalViewport?: (p: string, n: string) => string | null + }).__pearReadTerminalViewport?.(projectId, agent) ?? '', + { projectId: PROJECT_ID, agent: AGENT } + ), + { timeout: 20_000, message: 'final repaint never rendered' } + ) + .toContain(FINAL_FRAME) + + // Phase E: the stream is quiet now — this is the predictions regime. + // Typing here engages the engine route (transition + reseed from the + // live screen), renders optimistic glyphs, and reconciles them against + // the delayed echo. The final screen must still be byte-faithful. + if (STORM_TYPE) { + await page.waitForTimeout(500) // clear the streaming quiet window + await terminal.click() + await page.keyboard.type('done', { delay: 60 }) + } + await page.waitForTimeout(700) + + const screenshotPath = testInfo.outputPath('storm-final.png') + await page.screenshot({ path: screenshotPath, fullPage: false }) + await testInfo.attach('storm-final', { path: screenshotPath, contentType: 'image/png' }) + + const { viewport, dims, canonical } = await page.evaluate(({ projectId, agent }) => { + const probe = window as unknown as { + __pearReadTerminalViewport?: (p: string, n: string) => string | null + __pearReadTerminalDims?: (p: string, n: string) => { rows: number; cols: number } | null + } + const mock = window.__pearMock + return { + viewport: probe.__pearReadTerminalViewport?.(projectId, agent) ?? null, + dims: probe.__pearReadTerminalDims?.(projectId, agent) ?? null, + canonical: mock?.getState().ptyChunks[`${projectId}:${agent}`]?.join('') ?? null + } + }, { projectId: PROJECT_ID, agent: AGENT }) + + expect(viewport, 'viewport probe missing').not.toBeNull() + expect(dims, 'dims probe missing').not.toBeNull() + expect(canonical, 'canonical chunk stream missing').not.toBeNull() + + const reference = await readReferenceViewport(canonical!, dims!.rows, dims!.cols) + expect(viewport!, 'live screen diverged from the canonical byte stream — rendering corruption').toBe(reference) + expect(consoleErrors, 'renderer console errors during the storm').toEqual([]) + }) +})