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 000000000..a32064830 --- /dev/null +++ b/docs/superpowers/plans/2026-09-05-remote-output-backpressure.md @@ -0,0 +1,61 @@ +# Bounded remote output and reconnect recovery + +Status: implemented and locally verified. 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. + +## 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 343f39b3f..0571b4072 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 1b8667b0f..c8b71decb 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 000000000..2874b7fd8 --- /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 000000000..a4fc5a593 --- /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 359484645..c8ffe773a 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 3a4e6f0a6..f46617420 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 000000000..7b5b5420e --- /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 f0ced3907..35385fa51 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 a036f044b..4b6cacc02 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 000000000..94b0956a2 --- /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.'