fix(remote): bound outgoing queues and recover after overflow

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.

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<string, unknown>
+    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() }
+  })
+})

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<void> {
+    // 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<boolean>(resolve => {
+          if (!sendBoundedRemoteOutput(ws, encoded, resolve)) resolve(false)
+        })
+        if (!sent) return
+      }
+    }
   }
 
   private async onMessage(ws: WebSocket, raw: string): Promise<void> {
@@ -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)
   }
 }
 

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()
+  })
+})

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,
+  }
+}

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<typeof vi.fn>).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<typeof vi.fn>).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)
+})