From 55869631994c16b6f388ad082b776c6d854ed1d0 Mon Sep 17 00:00:00 2001 From: Julius Olsson Date: Sat, 5 Sep 2026 02:26:47 -0700 Subject: [PATCH 1/2] docs(remote): plan bounded output and reconnect recovery Refs #804 --- .../2026-09-05-remote-output-backpressure.md | 39 +++++++++++++++++++ 1 file changed, 39 insertions(+) create mode 100644 docs/superpowers/plans/2026-09-05-remote-output-backpressure.md diff --git a/docs/superpowers/plans/2026-09-05-remote-output-backpressure.md b/docs/superpowers/plans/2026-09-05-remote-output-backpressure.md new file mode 100644 index 00000000..8dc56fbb --- /dev/null +++ b/docs/superpowers/plans/2026-09-05-remote-output-backpressure.md @@ -0,0 +1,39 @@ +# Bounded remote output and reconnect recovery + +Status: implementing. Issue #804. Base origin/main at 5d641845. + +## Outcome + +A slow paired client cannot queue unbounded output in main. Overflow ends that +connection and the phone rebuilds its live history window from durable history +on reconnect, preserving uncertainty for interrupted prompt requests. Healthy +clients continue receiving ordered output. No changes to external control #795, +A5's renderer/worktree paths, or any running app configuration. + +## Implementation + +1. Give broadcast, direct replies and bootstrap one per-socket byte budget. + Terminate an overflowing socket immediately instead of queuing a close frame + behind its backlog. Prefer this complete-prefix/reset contract to a second + custom snapshot queue with independent ordering/retention semantics. +2. Return bounded errors for individually oversized replies; history requests + may reduce page size, but an individual oversized record must remain an + explicit error rather than silent truncation or an infinite retry loop. +3. Make disconnect reject in-flight requests without replaying mutations. + Reject late old-socket events and stale async history completions. +4. Reset transcript windows at disconnect. On the next authoritative session + list, backfill viewed sessions even if history loaded before disconnect. + Ignore partial semantic deltas until a fresh turn begins; committed records + remain authoritative for the interrupted turn. Keep older pages available. +5. Test real paused/draining socket consumers, budget enforcement on replies, + reconnect with more than one missed page, stale responses, and prompt retry + uncertainty. Preserve existing remote integration and rendering contracts. + +## Validation and following work + +Run focused remote transport/store/system tests, typecheck, contract and client +build; full applicable repository gates run in CI. Compare bounded queued bytes +with the audit's synthetic 31.5 MiB backlog, not production memory/CPU claims. +Review, synchronize #804, open a complete PR, and address checks/review. Do not +merge. #805 will depend on this PR because it changes the same store lifecycle +and must preserve this reconnect/backfill contract while bounding retention. From 205eb9ddec16a89cbd876946539f9d55092f84a7 Mon Sep 17 00:00:00 2001 From: Julius Olsson Date: Sat, 5 Sep 2026 02:36:53 -0700 Subject: [PATCH 2/2] fix(remote): bound output queues and recover disconnected history Terminate slow sockets before the shared output budget is exceeded and rebuild remote transcript windows after reconnect. Pace large bootstraps, preserve contiguous history cursors, and never replay requests whose outcome is unknown. Refs #804 --- .../2026-09-05-remote-output-backpressure.md | 24 +++- .../remote/RemoteServer.integration.test.ts | 58 +++++++++ src/main/remote/RemoteServer.ts | 61 +++++++--- src/main/remote/outputBudget.test.ts | 41 +++++++ src/main/remote/outputBudget.ts | 44 +++++++ .../WebSocketSessionFeed.integration.test.ts | 56 +++++++++ src/remote-client/src/WebSocketSessionFeed.ts | 27 ++++- .../src/transcript/store.reconnect.test.ts | 114 ++++++++++++++++++ src/remote-client/src/transcript/store.ts | 40 +++++- src/remote-client/src/ui/SessionView.tsx | 3 + src/shared/remoteOutputLimits.ts | 6 + 11 files changed, 446 insertions(+), 28 deletions(-) create mode 100644 src/main/remote/outputBudget.test.ts create mode 100644 src/main/remote/outputBudget.ts create mode 100644 src/remote-client/src/transcript/store.reconnect.test.ts create mode 100644 src/shared/remoteOutputLimits.ts diff --git a/docs/superpowers/plans/2026-09-05-remote-output-backpressure.md b/docs/superpowers/plans/2026-09-05-remote-output-backpressure.md index 8dc56fbb..a3206483 100644 --- a/docs/superpowers/plans/2026-09-05-remote-output-backpressure.md +++ b/docs/superpowers/plans/2026-09-05-remote-output-backpressure.md @@ -1,6 +1,6 @@ # Bounded remote output and reconnect recovery -Status: implementing. Issue #804. Base origin/main at 5d641845. +Status: implemented and locally verified. Issue #804. Base origin/main at 5d641845. ## Outcome @@ -37,3 +37,25 @@ with the audit's synthetic 31.5 MiB backlog, not production memory/CPU claims. Review, synchronize #804, open a complete PR, and address checks/review. Do not merge. #805 will depend on this PR because it changes the same store lifecycle and must preserve this reconnect/backfill contract while bounding retention. + +## Evidence and decisions + +All remote transport/store tests pass: 105 tests in 12 files, including an +actual paused receiver bounded to 4 MiB while a healthy receiver receives every +frame, reconnect/backfill across 310 durable entries, and an interrupted prompt +sent exactly once. A healthy bootstrap totaling more than 4 MiB is paced by +write completion; cached values are read immediately before each write. +Typecheck, test contract, client production build and diff check pass. Existing +client bundle size/mixed import warnings remain. The audit previously measured +31.5 MiB queued for a synthetic paused receiver; these are reproducible bounds, +not a measurement of production memory savings. + +History pages retain a contiguous newest suffix up to 3 MiB with matching byte +offsets and an older-history cursor. Individual records above that budget fail +explicitly. Reconnect discards the disconnected history window and hides a +partial semantic turn until a fresh turn boundary or durable completion. This +trades temporary streaming continuity for correct history after dropped output. + +Refreshed base: rebased onto main f7507980 after toolkit #812 and MCP repair +#818 merged. Their files do not overlap this implementation; rerun checks on +the refreshed PR head before requesting merge approval. diff --git a/src/main/remote/RemoteServer.integration.test.ts b/src/main/remote/RemoteServer.integration.test.ts index 343f39b3..0571b407 100644 --- a/src/main/remote/RemoteServer.integration.test.ts +++ b/src/main/remote/RemoteServer.integration.test.ts @@ -4,6 +4,7 @@ import { mkdtemp, rm, writeFile } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' import { WebSocket } from 'ws' +import { REMOTE_OUTPUT_MAX_BYTES } from '@shared/remoteOutputLimits.js' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { DevicePairing } from './auth/DevicePairing.js' @@ -447,3 +448,60 @@ describe('inbound scope enforcement on a live socket', () => { ws.close() }) }) + + +describe('remote slow-consumer isolation', () => { + it('bounds a paused socket while a healthy consumer receives every ordered event', async () => { + const slow = await connect(await pairDevice('synthetic slow')) + const healthy = await connect(await pairDevice('synthetic healthy')) + // Inspect the real accepted sockets; do not substitute a buffering model + // for ws, whose OPEN state previously hid an ever-growing sender queue. + const accepted = [...Reflect.get(server, 'sockets')] as Array<{ ws: WebSocket }> + const slowSocket = accepted[0]!.ws + slow.ws.pause() + let peak = 0 + try { + for (let index = 0; index < 160; index++) { + manager.emit('screen', { sessionId: 'synthetic', recent: `${index}:` + 'x'.repeat(64 * 1024), index }) + peak = Math.max(peak, slowSocket.bufferedAmount) + await waitFor(healthy.frames, frames => frames.some(frame => { + const value = frame as { payload?: { index?: number } } + return value.payload?.index === index + })) + } + expect(peak).toBeLessThanOrEqual(REMOTE_OUTPUT_MAX_BYTES) + await vi.waitFor(() => expect(slowSocket.readyState).toBe(WebSocket.CLOSED)) + const indices = healthy.frames.flatMap(frame => { + const value = frame as { payload?: { index?: number } } + return typeof value.payload?.index === 'number' ? [value.payload.index] : [] + }) + expect(indices).toEqual(Array.from({ length: 160 }, (_, i) => i)) + expect(healthy.ws.readyState).toBe(WebSocket.OPEN) + } finally { + slow.ws.terminate(); healthy.ws.terminate() + } + }, 15000) + + it('paces a bootstrap larger than the queue budget for a healthy late joiner', async () => { + const cache = Reflect.get(server, 'lastScreen') as Map + for (let i = 0; i < 80; i++) cache.set(`s-${i}`, { sessionId: `s-${i}`, recent: 'x'.repeat(64 * 1024) }) + const client = await connect(await pairDevice('synthetic bootstrap')) + try { + await waitFor(client.frames, frames => frames.filter(frame => (frame as { channel?: string }).channel === 'screen').length === 80) + expect(client.ws.readyState).toBe(WebSocket.OPEN) + } finally { client.ws.terminate() } + }) + + it('uses the same budget for an oversized direct reply without reconnect looping', async () => { + const client = await connect(await pairDevice('synthetic reply')) + const accepted = [...Reflect.get(server, 'sockets')] as Array<{ ws: WebSocket }> + try { + Reflect.get(server, 'send').call(server, accepted[0]!.ws, { + type: 'reply', id: 'oversized', ok: true, result: 'x'.repeat(REMOTE_OUTPUT_MAX_BYTES), + }) + await waitFor(client.frames, frames => frames.some(frame => (frame as { id?: string }).id === 'oversized')) + expect(client.frames.find(frame => (frame as { id?: string }).id === 'oversized')).toMatchObject({ ok: false, error: 'Remote response exceeds the output limit.' }) + expect(client.ws.readyState).toBe(WebSocket.OPEN) + } finally { client.ws.terminate() } + }) +}) diff --git a/src/main/remote/RemoteServer.ts b/src/main/remote/RemoteServer.ts index 1b8667b0..c8b71dec 100644 --- a/src/main/remote/RemoteServer.ts +++ b/src/main/remote/RemoteServer.ts @@ -6,6 +6,8 @@ import { extname, join, normalize, sep } from 'node:path' import type { Duplex } from 'node:stream' import { WebSocketServer } from 'ws' +import { sendBoundedRemoteOutput, boundRemoteHistory } from './outputBudget.js' +import { REMOTE_OUTPUT_MAX_BYTES, REMOTE_HISTORY_TOO_LARGE } from '@shared/remoteOutputLimits.js' import type { WebSocket } from 'ws' import type { AppRunJournal } from '@main/incident/AppRunJournal.js' @@ -543,21 +545,7 @@ export class RemoteServer extends EventEmitter { sttAvailable: Boolean(this.deps.transcribeAudio) && (this.deps.isSttAvailable?.() ?? true), }) this.send(ws, { type: 'session-list', sessions: this.deps.feedSource.listSessions() }) - // Late-joiner replay — same channel shape as live events so the client - // needs no special bootstrap path. - for (const [, payload] of this.lastScreen) { - this.send(ws, { type: 'session-event', channel: 'screen', payload }) - } - for (const [, payload] of this.lastConditions) { - this.send(ws, { type: 'session-event', channel: 'conditions', payload }) - } - for (const [, payload] of this.lastProcessState) { - this.send(ws, { type: 'session-event', channel: 'process-state', payload }) - } - for (const [, payload] of this.lastInputReadiness) { - this.send(ws, { type: 'session-event', channel: 'input-readiness', payload }) - } - + ws.on('error', () => ws.terminate()) ws.on('message', data => { void this.onMessage(ws, String(data)) }) @@ -570,6 +558,33 @@ export class RemoteServer extends EventEmitter { data: { deviceId }, }) }) + void this.replaySnapshots(ws) + } + + private async replaySnapshots(ws: WebSocket): Promise { + // A healthy late joiner can need more than the socket budget in TOTAL. + // Pace bootstrap by write completion rather than enqueueing every cached + // pane synchronously and disconnecting it again on every reconnect. Live + // broadcasts retain their ordering and budget; read each cache value just + // before sending so intervening live updates cannot be overwritten by an + // older snapshot captured at connection time. + const caches = [ + ['screen', this.lastScreen], + ['conditions', this.lastConditions], + ['process-state', this.lastProcessState], + ['input-readiness', this.lastInputReadiness], + ] as const + for (const [channel, cache] of caches) { + for (const [sessionId] of cache) { + const payload = cache.get(sessionId) + if (!payload) continue + const encoded = JSON.stringify({ type: 'session-event', channel, payload }) + const sent = await new Promise(resolve => { + if (!sendBoundedRemoteOutput(ws, encoded, resolve)) resolve(false) + }) + if (!sent) return + } + } } private async onMessage(ws: WebSocket, raw: string): Promise { @@ -686,7 +701,10 @@ export class RemoteServer extends EventEmitter { // durable line, and a backfill served from the OLD file must be // discardable client-side by comparing against the file the live // frames carry. - return { ok: true, result: { ...chunk, file } } + const bounded = boundRemoteHistory(chunk) + return bounded + ? { ok: true, result: { ...bounded, file } } + : { ok: false, error: REMOTE_HISTORY_TOO_LARGE } } } } @@ -754,12 +772,19 @@ export class RemoteServer extends EventEmitter { // rate times N phones would otherwise multiply stringify cost. const encoded = JSON.stringify(frame) for (const client of this.sockets) { - if (client.ws.readyState === client.ws.OPEN) client.ws.send(encoded) + sendBoundedRemoteOutput(client.ws, encoded) } } private send(ws: WebSocket, frame: OutboundFrame): void { - if (ws.readyState === ws.OPEN) ws.send(JSON.stringify(frame)) + let encoded = JSON.stringify(frame) + if (frame.type === 'reply' && Buffer.byteLength(encoded) + 16 > REMOTE_OUTPUT_MAX_BYTES) { + // A single oversized reply cannot recover by reconnecting and repeating + // the identical request. Return a bounded failure without replaying any + // mutation (the caller must treat absent delivery evidence as uncertain). + encoded = JSON.stringify({ type: 'reply', id: frame.id, ok: false, error: 'Remote response exceeds the output limit.' }) + } + sendBoundedRemoteOutput(ws, encoded) } } diff --git a/src/main/remote/outputBudget.test.ts b/src/main/remote/outputBudget.test.ts new file mode 100644 index 00000000..2874b7fd --- /dev/null +++ b/src/main/remote/outputBudget.test.ts @@ -0,0 +1,41 @@ +import { describe, expect, it, vi } from 'vitest' +import type { WebSocket } from 'ws' +import { boundRemoteHistory, sendBoundedRemoteOutput } from './outputBudget.js' +import { REMOTE_HISTORY_MAX_BYTES, REMOTE_OUTPUT_MAX_BYTES } from '@shared/remoteOutputLimits.js' + +function socket(bufferedAmount = 0) { + return { OPEN: 1, readyState: 1, bufferedAmount, send: vi.fn(), terminate: vi.fn() } +} + +describe('remote output byte budget', () => { + it('includes UTF-8 payload bytes and existing queued bytes before enqueueing', () => { + const ws = socket(REMOTE_OUTPUT_MAX_BYTES - 20) + expect(sendBoundedRemoteOutput(ws as unknown as WebSocket, '🌱')).toBe(true) + expect(ws.send).toHaveBeenCalledOnce() + expect(sendBoundedRemoteOutput(ws as unknown as WebSocket, '🌱x')).toBe(false) + expect(ws.terminate).toHaveBeenCalledOnce() + expect(ws.send).toHaveBeenCalledOnce() + }) + + it('terminates a single oversized frame without ever queueing it', () => { + const ws = socket() + expect(sendBoundedRemoteOutput(ws as unknown as WebSocket, 'x'.repeat(REMOTE_OUTPUT_MAX_BYTES))).toBe(false) + expect(ws.send).not.toHaveBeenCalled() + expect(ws.terminate).toHaveBeenCalledOnce() + }) + + it('keeps a contiguous history suffix and matching byte offsets for further pagination', () => { + const entries = Array.from({ length: 4 }, (_, id) => ({ id, text: 'x'.repeat(1024 * 1024) })) + const chunk = boundRemoteHistory({ entries, offsets: [0, 10, 20, 30], hasMore: false, totalEntries: 4 })! + expect(chunk.entries.map(e => e.id)).toEqual([2, 3]) + expect(chunk.offsets).toEqual([20, 30]) + expect(chunk.hasMore).toBe(true) + expect(chunk.totalEntries).toBe(4) + const older = boundRemoteHistory({ entries: entries.slice(0, 2), offsets: [0, 10], hasMore: false })! + expect([...older.entries, ...chunk.entries].map(e => e.id)).toEqual([0, 1, 2, 3]) + }) + + it('reports an individually oversized record rather than silently truncating it', () => { + expect(boundRemoteHistory({ entries: [{ text: 'x'.repeat(REMOTE_HISTORY_MAX_BYTES) }], hasMore: false })).toBeNull() + }) +}) diff --git a/src/main/remote/outputBudget.ts b/src/main/remote/outputBudget.ts new file mode 100644 index 00000000..a4fc5a59 --- /dev/null +++ b/src/main/remote/outputBudget.ts @@ -0,0 +1,44 @@ +import type { WebSocket } from 'ws' +import type { HistoryChunk } from '@main/sessions/historyLoader.js' +import { REMOTE_OUTPUT_MAX_BYTES, REMOTE_HISTORY_MAX_BYTES } from '@shared/remoteOutputLimits.js' + +export function sendBoundedRemoteOutput(ws: WebSocket, encoded: string, flushed?: (ok: boolean) => void): boolean { + if (ws.readyState !== ws.OPEN) return false + // bufferedAmount covers both the socket and ws sender queue. Reserve the + // maximum frame header too. All writers run on this event loop, so nothing + // can enqueue between this check and send. Do not queue a graceful close + // behind the very backlog we're trying to release: terminate immediately. + // Reconnection resets/backfills the client rather than hiding missing deltas. + if (ws.bufferedAmount + Buffer.byteLength(encoded) + 16 > REMOTE_OUTPUT_MAX_BYTES) { + ws.terminate() + return false + } + if (flushed) { + ws.send(encoded, error => { + if (error) ws.terminate() + flushed(!error) + }) + } else ws.send(encoded) + return true +} + +export function boundRemoteHistory(chunk: HistoryChunk): HistoryChunk | null { + let size = 2 + let start = chunk.entries.length + // Keep the suffix nearest the requested cursor. Throwing away the newest + // end of an older page would create an unpageable gap; moving the oldest + // boundary instead lets the next request retrieve everything left behind. + while (start > 0) { + const bytes = Buffer.byteLength(JSON.stringify(chunk.entries[start - 1])) + 1 + if (size + bytes > REMOTE_HISTORY_MAX_BYTES) break + size += bytes + start-- + } + if (chunk.entries.length > 0 && start === chunk.entries.length) return null + return { + ...chunk, + entries: chunk.entries.slice(start), + offsets: chunk.offsets?.slice(start), + hasMore: chunk.hasMore || start > 0, + } +} diff --git a/src/remote-client/src/WebSocketSessionFeed.integration.test.ts b/src/remote-client/src/WebSocketSessionFeed.integration.test.ts index 35948464..c8ffe773 100644 --- a/src/remote-client/src/WebSocketSessionFeed.integration.test.ts +++ b/src/remote-client/src/WebSocketSessionFeed.integration.test.ts @@ -367,3 +367,59 @@ describe('WebSocketSessionFeed against a live RemoteServer', () => { }) }) }) + + +describe('remote reconnect after bounded output overflow', () => { + it('backfills a previously loaded session after overflow and retains access to every missed page', async () => { + const transcript = join(dir, 'reconnect.jsonl') + const entry = (i: number) => ({ type: 'user', uuid: `reconnect-${i}`, message: { role: 'user', content: `synthetic-${i}` } }) + await writeFile(transcript, Array.from({ length: 10 }, (_, i) => JSON.stringify(entry(i))).join('\n') + '\n') + ;(manager.resolveTranscriptFile as ReturnType).mockResolvedValue(transcript) + const f = makeFeed() + const store = new TranscriptStore(f) + const unsub = store.subscribe('s1', () => {}) + try { + await waitForOpen(f) + manager.emit('started', { sessionId: 's1', kind: 'claude', projectDir: '/synthetic' }) + await vi.waitFor(() => expect(store.getSnapshot('s1').entries).toHaveLength(10)) + const oldSocket = Reflect.get(f, 'socket') as NodeWebSocket + const accepted = [...Reflect.get(server, 'sockets')] as Array<{ ws: NodeWebSocket }> + const serverSocket = accepted[0]!.ws + oldSocket.pause() + await writeFile(transcript, Array.from({ length: 310 }, (_, i) => JSON.stringify(entry(i))).join('\n') + '\n') + for (let i = 0; i < 160; i++) { + manager.emit('screen', { sessionId: 's1', recent: `${i}:` + 'x'.repeat(64 * 1024) }) + if (i % 8 === 0) await new Promise(setImmediate) + } + await vi.waitFor(() => expect(serverSocket.readyState).toBe(NodeWebSocket.CLOSED)) + oldSocket.resume() + await vi.waitFor(() => { + expect(Reflect.get(f, 'socket')).not.toBe(oldSocket) + expect(store.getSnapshot('s1').entries[0]?.uuid).toBe('reconnect-190') + expect(store.getSnapshot('s1').entries).toHaveLength(120) + }, { timeout: 6000 }) + await store.loadOlderHistory('s1') + expect(store.getSnapshot('s1').entries.map(e => e.uuid)).toEqual(Array.from({ length: 310 }, (_, i) => `reconnect-${i}`)) + } finally { unsub(); store.dispose() } + }, 15000) + + it('rejects an interrupted prompt as uncertain and never replays it on reconnect', async () => { + let finish!: (value: unknown) => void + const result = new Promise(resolve => { finish = resolve }) + ;(manager.deliverPromptToAgent as ReturnType).mockReturnValue(result) + const f = makeFeed() + await waitForOpen(f) + const delivery = f.deliverPrompt('s1', 'synthetic prompt') + await vi.waitFor(() => expect(manager.deliverPromptToAgent).toHaveBeenCalledTimes(1)) + const oldSocket = Reflect.get(f, 'socket') as NodeWebSocket + oldSocket.close() + expect(await delivery).toMatchObject({ ok: false, retrySafe: false, disposition: 'do-not-retry' }) + finish({ ok: true, acceptance: { kind: 'transport', acceptedAt: 1 } }) + await vi.waitFor(() => { + const current = Reflect.get(f, 'socket') as NodeWebSocket | null + expect(current).not.toBe(oldSocket) + expect(current?.readyState).toBe(NodeWebSocket.OPEN) + }, { timeout: 6000 }) + expect(manager.deliverPromptToAgent).toHaveBeenCalledTimes(1) + }, 10000) +}) diff --git a/src/remote-client/src/WebSocketSessionFeed.ts b/src/remote-client/src/WebSocketSessionFeed.ts index 3a4e6f0a..f4661742 100644 --- a/src/remote-client/src/WebSocketSessionFeed.ts +++ b/src/remote-client/src/WebSocketSessionFeed.ts @@ -50,7 +50,8 @@ function applyRemoteThemeSettings(settings: Record | null | und // // Reconnect: the socket redials with capped backoff until dispose(). On // each (re)connect the server replays session-list + cached per-session -// state, so listeners self-heal without a client-side resync protocol. +// state. TranscriptStore resets its loaded window on disconnect and backfills +// after the next authoritative list; cached state alone cannot repair a gap. const BRACKETED_PASTE = /^\x1b\[200~([\s\S]*)\x1b\[201~$/ // WHY this exceeds the provider protocol: Claude may spend 2s proving paste @@ -114,6 +115,7 @@ export class WebSocketSessionFeed implements SessionFeed { private socket: WebSocketLike | null = null private disposed = false private reconnectAttempt = 0 + private reconnectTimer: ReturnType | null = null private nextRequestId = 1 private lastSessionList: RemoteSessionSummary[] = [] /** Latest hello-declared STT capability. null = unknown (no hello yet, or @@ -159,6 +161,8 @@ export class WebSocketSessionFeed implements SessionFeed { dispose(): void { this.disposed = true + if (this.reconnectTimer) clearTimeout(this.reconnectTimer) + this.reconnectTimer = null for (const [, p] of this.pending) { clearTimeout(p.timer) p.resolve({ ok: false, error: 'feed disposed' }) @@ -352,14 +356,24 @@ export class WebSocketSessionFeed implements SessionFeed { this.socket = socket socket.addEventListener('open', () => { + if (this.socket !== socket || this.disposed) return this.reconnectAttempt = 0 this.emitConnection('open') }) socket.addEventListener('message', event => { - this.onFrame(String(event.data)) + if (this.socket === socket && !this.disposed) this.onFrame(String(event.data)) }) socket.addEventListener('close', () => { - if (this.socket === socket) this.socket = null + if (this.socket !== socket || this.disposed) return + this.socket = null + // The remote command may already have reached the provider. Reject + // pending requests promptly, but NEVER replay them on reconnect. The + // prompt API's conservative fallback preserves duplicate-send safety. + for (const pending of this.pending.values()) { + clearTimeout(pending.timer) + pending.resolve({ ok: false, error: 'Connection lost; request outcome is unknown.' }) + } + this.pending.clear() this.emitConnection('closed') this.scheduleReconnect() }) @@ -369,13 +383,16 @@ export class WebSocketSessionFeed implements SessionFeed { } private scheduleReconnect(): void { - if (this.disposed) return + if (this.disposed || this.reconnectTimer) return const delay = Math.min( RECONNECT_BASE_MS * 2 ** this.reconnectAttempt, RECONNECT_CAP_MS, ) this.reconnectAttempt += 1 - setTimeout(() => this.dial(), delay) + this.reconnectTimer = setTimeout(() => { + this.reconnectTimer = null + this.dial() + }, delay) } private onFrame(raw: string): void { diff --git a/src/remote-client/src/transcript/store.reconnect.test.ts b/src/remote-client/src/transcript/store.reconnect.test.ts new file mode 100644 index 00000000..7b5b5420 --- /dev/null +++ b/src/remote-client/src/transcript/store.reconnect.test.ts @@ -0,0 +1,114 @@ +import { describe, expect, it, vi } from 'vitest' +import type { WebSocketSessionFeed } from '../WebSocketSessionFeed' +import type { HistoryChunkResult } from '../wire' +import { TranscriptStore } from './store' +import { REMOTE_HISTORY_TOO_LARGE } from '@shared/remoteOutputLimits' + +// The real store/mapper/reducers consume a synthetic transport. Tests control +// response ordering directly so old-history-versus-reconnect races don't depend +// on kernel scheduling or on a running provider. +function fixture() { + const listeners = new Map void>>() + const list = [{ sessionId: 's', kind: 'claude', alive: true, cwd: '/synthetic', lastActivityAt: 0 }] + const getHistory = vi.fn<(...args: unknown[]) => Promise<{ ok: true; chunk: HistoryChunkResult } | { ok: false; error: string }>>() + const methods = { getHistory, getSessionList: () => list } + const feed = new Proxy(methods, { + get(target, key: string) { + if (key in target) return target[key as keyof typeof target] + return (cb: (value: unknown) => void) => { + let set = listeners.get(key) + if (!set) listeners.set(key, set = new Set()) + set.add(cb) + return () => set.delete(cb) + } + }, + }) as unknown as WebSocketSessionFeed + const store = new TranscriptStore(feed) + return { store, getHistory, list, emit: (name: string, value: unknown) => { for (const cb of listeners.get(name) ?? []) cb(value) } } +} +const entry = (i: number) => ({ type: 'user', uuid: `u-${i}`, message: { role: 'user', content: `synthetic-${i}` } }) +const chunk = (start: number, end: number, hasMore = false): HistoryChunkResult => ({ + entries: Array.from({ length: end - start }, (_, i) => entry(start + i)), + file: '/synthetic/transcript.jsonl', hasMore, totalEntries: end, +}) + +function deferred() { + let resolve!: (value: T) => void + const promise = new Promise(r => { resolve = r }) + return { promise, resolve } +} + +describe('remote transcript reconnect recovery', () => { + it('replaces a disconnected window, backfills a previously loaded view and pages across a gap larger than one page', async () => { + const f = fixture() + const unsub = f.store.subscribe('s', () => {}) + try { + f.getHistory.mockResolvedValueOnce({ ok: true, chunk: chunk(0, 10) }) + await f.store.loadInitialHistory('s') + f.emit('onConnectionState', 'closed') + expect(f.store.getSnapshot('s').entries).toEqual([]) + f.getHistory.mockResolvedValueOnce({ ok: true, chunk: chunk(190, 310, true) }) + f.emit('onSessionList', f.list) + await vi.waitFor(() => expect(f.store.getSnapshot('s').entries).toHaveLength(120)) + expect(f.store.getSnapshot('s').entries[0]?.uuid).toBe('u-190') + // No stale prefix joined across the missing 180 records. The entire + // durable range is reachable through normal older-history pagination. + f.getHistory.mockResolvedValueOnce({ ok: true, chunk: chunk(0, 190) }) + await f.store.loadOlderHistory('s') + expect(f.getHistory).toHaveBeenLastCalledWith('s', { beforeMarker: 'u-190', limit: 200 }) + expect(f.store.getSnapshot('s').entries.map(e => e.uuid)).toEqual(Array.from({ length: 310 }, (_, i) => `u-${i}`)) + } finally { unsub(); f.store.dispose() } + }) + + it('ignores an in-flight old history reply after disconnect, even for the same file', async () => { + const f = fixture() + const old = deferred<{ ok: true; chunk: HistoryChunkResult }>() + try { + f.getHistory.mockReturnValueOnce(old.promise) + const loading = f.store.loadInitialHistory('s') + f.emit('onConnectionState', 'closed') + f.getHistory.mockResolvedValueOnce({ ok: true, chunk: chunk(200, 210) }) + await f.store.loadInitialHistory('s') + old.resolve({ ok: true, chunk: chunk(0, 10) }) + await loading + expect(f.store.getSnapshot('s').entries.map(e => e.uuid)).toEqual(Array.from({ length: 10 }, (_, i) => `u-${200 + i}`)) + } finally { f.store.dispose() } + }) + + it('uses committed content after reconnect until a complete new semantic turn starts', () => { + const f = fixture() + try { + f.store.getSnapshot('s') + f.emit('onConnectionState', 'closed') + f.emit('onSessionSemanticEvent', { sessionId: 's', event: { type: 'block_started', source: 'proxy', turnId: 'lost-prefix', blockId: 'b', blockType: 'text' } }) + expect(f.store.getSnapshot('s').semanticTurn).toBeNull() + f.emit('onSessionJsonlEntries', { sessionId: 's', entries: [{ entry: entry(1), file: '/synthetic/transcript.jsonl' }] }) + expect(f.store.getSnapshot('s').entries).toHaveLength(1) + f.emit('onSessionSemanticEvent', { sessionId: 's', event: { type: 'turn_started', turnId: 'new-turn' } }) + expect(f.store.getSnapshot('s').semanticTurn?.turnId).toBe('new-turn') + } finally { f.store.dispose() } + }) + + it('does not retry an oversized history record on every subsequent activity event', async () => { + const f = fixture() + const unsub = f.store.subscribe('s', () => {}) + try { + f.getHistory.mockResolvedValue({ ok: false, error: REMOTE_HISTORY_TOO_LARGE }) + await f.store.loadInitialHistory('s') + for (let i = 0; i < 50; i++) f.emit('onSessionList', f.list) + expect(f.getHistory).toHaveBeenCalledTimes(1) + expect(f.store.getSnapshot('s').historyError).toBe(REMOTE_HISTORY_TOO_LARGE) + } finally { unsub(); f.store.dispose() } + }) + + it('does not re-backfill a view that unsubscribed before reconnect', async () => { + const f = fixture() + const unsub = f.store.subscribe('s', () => {}) + f.store.getSnapshot('s') + unsub() + f.emit('onConnectionState', 'closed') + f.emit('onSessionList', f.list) + expect(f.getHistory).not.toHaveBeenCalled() + f.store.dispose() + }) +}) diff --git a/src/remote-client/src/transcript/store.ts b/src/remote-client/src/transcript/store.ts index f0ced390..35385fa5 100644 --- a/src/remote-client/src/transcript/store.ts +++ b/src/remote-client/src/transcript/store.ts @@ -1,3 +1,4 @@ +import { REMOTE_HISTORY_TOO_LARGE } from '@shared/remoteOutputLimits' import { getRendererProviderCapabilities } from '@providers/registry.renderer.capabilities' import { foldSemanticEvent } from '@renderer/session-runtime/semantic/foldEvent' import { reduceStreamPhase } from '@renderer/session-runtime/semantic/streamPhaseMachine' @@ -91,6 +92,7 @@ type SessionState = { historyOldestMarker: string | null historyLoaded: boolean historyLoading: boolean + awaitingSemanticStart: boolean } type Mapper = ReturnType< @@ -136,6 +138,17 @@ export class TranscriptStore { constructor(private readonly feed: WebSocketSessionFeed) { this.unsubs.push( + feed.onConnectionState(connection => { + if (connection !== 'closed') return + // A reconnect can have missed MORE than one page. Prepending a fresh + // tail to the old window would silently join two disconnected ranges. + // Drop the window and backfill anew on the next server session-list; + // older pages remain available through the durable cursor contract. + for (const id of this.sessions.keys()) { + this.resetTranscript(id) + this.state(id).awaitingSemanticStart = true + } + }), feed.onSessionJsonlEntries(e => { this.ingestLiveEntries(e.sessionId, e.entries as Array<{ entry: unknown; file: string }>) }), @@ -190,7 +203,7 @@ export class TranscriptStore { // loadInitialHistory once per mount (review finding). The list // frame doubles as the "connection is live again" signal. for (const [sessionId, state] of this.sessions) { - if (!state.historyLoaded && !state.historyLoading && this.listeners.has(sessionId)) { + if (!state.historyLoaded && !state.historyLoading && (this.listeners.get(sessionId)?.size ?? 0) > 0) { void this.loadInitialHistory(sessionId) } } @@ -237,10 +250,14 @@ export class TranscriptStore { * hook above and from live-entry arrival. */ async loadInitialHistory(sessionId: string): Promise { const state = this.state(sessionId) - if (state.historyLoaded || state.historyLoading) return + if (state.historyLoaded || state.historyLoading || state.transcript.historyError === REMOTE_HISTORY_TOO_LARGE) return state.historyLoading = true this.mutate(sessionId, t => ({ ...t, loadingOlderHistory: true })) const result = await this.feed.getHistory(sessionId, { limit: 120 }) + // Disconnect, transcript roll, removal or disposal may replace this state + // while the network request is pending. Its reply has no authority over + // the new window, even when it names the same transcript file. + if (this.sessions.get(sessionId) !== state) return state.historyLoading = false if (!result.ok) { // "No transcript yet" is normal for a brand-new session — live frames @@ -297,8 +314,14 @@ export class TranscriptStore { } this.mutate(sessionId, t => ({ ...t, loadingOlderHistory: true })) const result = await this.feed.getHistory(sessionId, { beforeMarker, limit: 200 }) + if (this.sessions.get(sessionId) !== state) return if (!result.ok || this.chunkFileConflicts(state, result.chunk?.file)) { - this.mutate(sessionId, t => ({ ...t, loadingOlderHistory: false })) + this.mutate(sessionId, t => ({ + ...t, + loadingOlderHistory: false, + historyError: result.ok ? t.historyError : result.error, + hasOlderHistory: !result.ok && result.error === REMOTE_HISTORY_TOO_LARGE ? false : t.hasOlderHistory, + })) return } const marker = this.ingestRawEntries( @@ -333,6 +356,7 @@ export class TranscriptStore { historyOldestMarker: null, historyLoaded: false, historyLoading: false, + awaitingSemanticStart: false, } this.sessions.set(sessionId, state) } @@ -414,7 +438,7 @@ export class TranscriptStore { // Live-entry arrival is also the backfill retry trigger for sessions // whose initial get-history failed with "no transcript on disk yet". const fresh = this.state(sessionId) - if (!fresh.historyLoaded && !fresh.historyLoading && this.listeners.has(sessionId)) { + if (!fresh.historyLoaded && !fresh.historyLoading && (this.listeners.get(sessionId)?.size ?? 0) > 0) { void this.loadInitialHistory(sessionId) } } @@ -459,6 +483,7 @@ export class TranscriptStore { historyOldestMarker: null, historyLoaded: false, historyLoading: false, + awaitingSemanticStart: false, }) const set = this.listeners.get(sessionId) if (set) for (const cb of [...set]) cb() @@ -524,6 +549,13 @@ export class TranscriptStore { const state = this.state(sessionId) const record = asRecord(event) if (!record) return + if (state.awaitingSemanticStart) { + // We did not see the prefix of the interrupted semantic turn. Rendering + // its suffix as a complete live answer is misleading. Durable JSONL + // still flows; live semantic painting resumes at the next turn boundary. + if (record.type !== 'turn_started') return + state.awaitingSemanticStart = false + } // Desktop order: fold first, then the shared phase machine over the // POST-fold turn (reduceStreamPhase's caller contract). diff --git a/src/remote-client/src/ui/SessionView.tsx b/src/remote-client/src/ui/SessionView.tsx index a036f044..4b6cacc0 100644 --- a/src/remote-client/src/ui/SessionView.tsx +++ b/src/remote-client/src/ui/SessionView.tsx @@ -378,6 +378,9 @@ export function SessionView({ )} + {transcript.historyError && transcript.entries.length > 0 && ( +
{transcript.historyError}
+ )} {working &&
● {working}
} {/* The REAL desktop condition rendering. The generic core outlet routes diff --git a/src/shared/remoteOutputLimits.ts b/src/shared/remoteOutputLimits.ts new file mode 100644 index 00000000..94b0956a --- /dev/null +++ b/src/shared/remoteOutputLimits.ts @@ -0,0 +1,6 @@ +// One connection must not pin an unbounded backlog in main. The history slice +// leaves headroom for frame/file metadata; a single larger record is an explicit +// mobile error, never silently truncated transcript content. +export const REMOTE_OUTPUT_MAX_BYTES = 4 * 1024 * 1024 +export const REMOTE_HISTORY_MAX_BYTES = 3 * 1024 * 1024 +export const REMOTE_HISTORY_TOO_LARGE = 'This history record is too large for the remote client.'