fix(terminal): prevent double rendering on frame-drop resync

After an untagged resync snapshot, the client resets to the snapshot's output
high-water. Buffered output generated while serializing is already included in
the snapshot, so replaying the entire buffer afterward renders those bytes
twice. Trim to output after the snapshot seq.

Tagged snapshots feed side consumers and need all buffered output unchanged.
Add exponential backoff for truncated recovery retries to avoid stampeding
flooded servers.

src/main/runtime/rpc/methods/terminal.ts

@@ -1986,6 +1986,7 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
         stream.pendingOutputOverflowed = false
         stream.buffering = true
         const requestId = request.requestId
+        let sentSnapshotOutputSeq: number | undefined
         try {
           const scrollbackRows = normalizeMultiplexSnapshotScrollbackRows(request.scrollbackRows)
           let serialized = await serializeBudgetedRequestedSnapshot(
@@ -2027,6 +2028,7 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
               return
             }
           }
+          sentSnapshotOutputSeq = serialized?.seq
           sendSnapshotFrames((opcode, payload) => sendFrame(stream.streamId, opcode, payload), {
             kind: 'scrollback',
             cols: serialized?.cols ?? size?.cols ?? 80,
@@ -2054,7 +2056,17 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
             const pendingOutput = stream.pendingOutput.splice(0)
             if (shouldFlushPendingOutput) {
               for (const chunk of pendingOutput) {
-                stream.outputBatcher.push(chunk.data, chunk.meta)
+                // Why: an untagged reply resets the client to the snapshot's
+                // high-water, so covered bytes would render twice; tagged
+                // snapshots feed a side consumer and the live view still
+                // needs every buffered chunk.
+                const uncoveredData =
+                  typeof requestId === 'number'
+                    ? chunk.data
+                    : getOutputAfterSnapshotSeq(chunk, sentSnapshotOutputSeq)
+                if (uncoveredData) {
+                  stream.outputBatcher.push(uncoveredData, chunk.meta)
+                }
               }
             }
             stream.pendingOutputBytes = 0

src/main/runtime/rpc/terminal-multiplex-resync-replay-trim.test.ts

@@ -0,0 +1,226 @@
+import { describe, expect, it, vi } from 'vitest'
+import { RpcDispatcher } from './dispatcher'
+import type { RpcRequest } from './core'
+import type { OrcaRuntimeService } from '../orca-runtime'
+import { TERMINAL_METHODS } from './methods/terminal'
+import type { RuntimeTerminalWait } from '../../../shared/runtime-types'
+import {
+  TerminalStreamOpcode,
+  decodeTerminalStreamFrame,
+  decodeTerminalStreamJson,
+  decodeTerminalStreamText,
+  encodeTerminalStreamFrame,
+  encodeTerminalStreamJson
+} from '../../../shared/terminal-stream-protocol'
+
+// An untagged SnapshotRequest is the client's frame-drop resync: its reply
+// resets the client terminal to the snapshot's output high-water. Output
+// buffered while the snapshot serialized is already inside that snapshot up to
+// its seq, so replaying the covered bytes afterward renders the recovered tail
+// twice. A tagged (requestId) snapshot feeds a side consumer instead — the
+// live view still needs every buffered chunk, so that replay must stay whole.
+
+type OutputMeta = { seq?: number; rawLength?: number }
+
+async function setupMultiplexStream(): Promise<{
+  binaryFrames: Uint8Array<ArrayBufferLike>[]
+  sendClientFrame: (opcode: TerminalStreamOpcode, payload: Uint8Array<ArrayBufferLike>) => void
+  emitOutput: (data: string, meta?: OutputMeta) => void
+  setSnapshot: (snapshot: { data: string; seq?: number }) => void
+  deferNextSerialize: () => void
+  releaseSerialize: () => Promise<void>
+  finish: () => Promise<void>
+}> {
+  const messages: string[] = []
+  const binaryFrames: Uint8Array<ArrayBufferLike>[] = []
+  const handlers = new Map<
+    number,
+    (frame: NonNullable<ReturnType<typeof decodeTerminalStreamFrame>>) => void
+  >()
+  const cleanups = new Map<string, () => void>()
+  let emitOutput: ((data: string, meta?: OutputMeta) => void) | null = null
+  let snapshot: { data: string; seq?: number } = { data: 'INITIAL', seq: 0 }
+  let deferSerialize = false
+  let releaseDeferredSerialize: (() => void) | null = null
+
+  const runtime = {
+    getRuntimeId: () => 'test-runtime',
+    resolveLiveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }),
+    readTerminal: vi.fn().mockResolvedValue({ tail: [], truncated: false }),
+    serializeTerminalBuffer: vi.fn(async () => {
+      if (deferSerialize) {
+        deferSerialize = false
+        await new Promise<void>((resolve) => {
+          releaseDeferredSerialize = resolve
+        })
+      }
+      return { data: snapshot.data, cols: 80, rows: 24, seq: snapshot.seq }
+    }),
+    getTerminalSize: vi.fn().mockReturnValue({ cols: 80, rows: 24 }),
+    getMobileDisplayMode: vi.fn().mockReturnValue('auto'),
+    getLayout: vi.fn().mockReturnValue({ seq: 1 }),
+    registerRemoteTerminalViewSubscriber: vi.fn(() => () => {}),
+    subscribeToTerminalData: vi.fn(
+      (_ptyId: string, cb: (data: string, meta?: OutputMeta) => void) => {
+        emitOutput = cb
+        return vi.fn()
+      }
+    ),
+    subscribeToTerminalResize: vi.fn().mockReturnValue(vi.fn()),
+    subscribeToFitOverrideChanges: vi.fn().mockReturnValue(vi.fn()),
+    subscribeToDriverChanges: vi.fn().mockReturnValue(vi.fn()),
+    getTerminalFitOverride: vi.fn().mockReturnValue(null),
+    getDriver: vi.fn().mockReturnValue({ kind: 'idle' }),
+    registerSubscriptionCleanup: vi.fn((id: string, cleanup: () => void) => {
+      cleanups.set(id, cleanup)
+    }),
+    cleanupSubscription: vi.fn((id: string) => {
+      const cleanup = cleanups.get(id)
+      cleanups.delete(id)
+      cleanup?.()
+    }),
+    waitForTerminal: vi.fn(() => new Promise<RuntimeTerminalWait>(() => {})),
+    updateDesktopViewport: vi.fn().mockResolvedValue(true)
+  } as unknown as OrcaRuntimeService
+  const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
+
+  const request: RpcRequest = {
+    id: 'req-1',
+    authToken: 'tok',
+    method: 'terminal.multiplex',
+    params: {}
+  }
+  const dispatchPromise = dispatcher.dispatchStreaming(request, (msg) => messages.push(msg), {
+    connectionId: 'conn-1',
+    sendBinary: (bytes) => {
+      binaryFrames.push(bytes)
+    },
+    registerBinaryStreamHandler: (streamId, handler) => {
+      handlers.set(streamId, handler)
+      return () => handlers.delete(streamId)
+    }
+  })
+
+  await vi.runOnlyPendingTimersAsync()
+  expect(messages.some((msg) => JSON.parse(msg).result?.type === 'ready')).toBe(true)
+
+  handlers.get(0)?.(
+    decodeTerminalStreamFrame(
+      encodeTerminalStreamFrame({
+        opcode: TerminalStreamOpcode.Subscribe,
+        streamId: 0,
+        seq: 1,
+        payload: encodeTerminalStreamJson({
+          streamId: 5,
+          terminal: 'terminal-1',
+          client: { id: 'desktop-1', type: 'desktop' }
+        })
+      })
+    )!
+  )
+  for (let i = 0; i < 5; i += 1) {
+    await vi.runOnlyPendingTimersAsync()
+  }
+  expect(emitOutput).not.toBeNull()
+
+  return {
+    binaryFrames,
+    sendClientFrame: (opcode, payload) => {
+      handlers.get(5)?.(
+        decodeTerminalStreamFrame(
+          encodeTerminalStreamFrame({ opcode, streamId: 5, seq: 1, payload })
+        )!
+      )
+    },
+    emitOutput: (data, meta) => emitOutput!(data, meta),
+    setSnapshot: (next) => {
+      snapshot = next
+    },
+    deferNextSerialize: () => {
+      deferSerialize = true
+    },
+    releaseSerialize: async () => {
+      releaseDeferredSerialize?.()
+      releaseDeferredSerialize = null
+      for (let i = 0; i < 5; i += 1) {
+        await vi.runOnlyPendingTimersAsync()
+      }
+    },
+    finish: async () => {
+      runtime.cleanupSubscription('terminal-multiplex:conn-1')
+      await dispatchPromise
+    }
+  }
+}
+
+function outputTextsAfterLastSnapshotEnd(frames: Uint8Array<ArrayBufferLike>[]): string[] {
+  const decoded = frames.map((frame) => decodeTerminalStreamFrame(frame))
+  const lastEnd = decoded.reduce(
+    (last, frame, index) => (frame?.opcode === TerminalStreamOpcode.SnapshotEnd ? index : last),
+    -1
+  )
+  return decoded.slice(lastEnd + 1).flatMap((frame) => {
+    if (frame?.opcode === TerminalStreamOpcode.Output) {
+      return [decodeTerminalStreamText(frame.payload)]
+    }
+    if (frame?.opcode === TerminalStreamOpcode.OutputSpan) {
+      return [decodeTerminalStreamJson<{ data?: string }>(frame.payload)?.data ?? '']
+    }
+    return []
+  })
+}
+
+describe('terminal.multiplex requested-snapshot replay trim', () => {
+  it('drops snapshot-covered buffered output after an untagged resync reply', async () => {
+    vi.useFakeTimers()
+    try {
+      const harness = await setupMultiplexStream()
+
+      harness.setSnapshot({ data: 'RECOVERED', seq: 12 })
+      harness.deferNextSerialize()
+      harness.sendClientFrame(TerminalStreamOpcode.SnapshotRequest, encodeTerminalStreamJson({}))
+      // Buffered while the snapshot serialized: fully covered by seq 12, and a
+      // partial chunk straddling the boundary whose tail the client still needs.
+      harness.emitOutput('xxx', { seq: 9, rawLength: 3 })
+      harness.emitOutput('bbbccc', { seq: 15, rawLength: 6 })
+      await harness.releaseSerialize()
+
+      const snapshotStart = harness.binaryFrames
+        .map((frame) => decodeTerminalStreamFrame(frame))
+        .findLast((frame) => frame?.opcode === TerminalStreamOpcode.SnapshotStart)!
+      expect(decodeTerminalStreamJson(snapshotStart.payload)).toMatchObject({ seq: 12 })
+      expect(outputTextsAfterLastSnapshotEnd(harness.binaryFrames).join('')).toBe('ccc')
+
+      await harness.finish()
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+
+  it('replays all buffered output untouched after a tagged snapshot reply', async () => {
+    vi.useFakeTimers()
+    try {
+      const harness = await setupMultiplexStream()
+
+      harness.setSnapshot({ data: 'MANUAL', seq: 12 })
+      harness.deferNextSerialize()
+      harness.sendClientFrame(
+        TerminalStreamOpcode.SnapshotRequest,
+        encodeTerminalStreamJson({ requestId: 7 })
+      )
+      harness.emitOutput('xxx', { seq: 9, rawLength: 3 })
+      harness.emitOutput('bbbccc', { seq: 15, rawLength: 6 })
+      await harness.releaseSerialize()
+
+      const snapshotStart = harness.binaryFrames
+        .map((frame) => decodeTerminalStreamFrame(frame))
+        .findLast((frame) => frame?.opcode === TerminalStreamOpcode.SnapshotStart)!
+      expect(decodeTerminalStreamJson(snapshotStart.payload)).toMatchObject({ requestId: 7 })
+      expect(outputTextsAfterLastSnapshotEnd(harness.binaryFrames).join('')).toBe('xxxbbbccc')
+
+      await harness.finish()
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+})

src/renderer/src/runtime/remote-runtime-terminal-frame-drop-resync.test.ts

@@ -41,6 +41,7 @@ class FakeMultiplexServer {
   holdNextManualSnapshot = false
   truncateNextRecoverySnapshot = false
   dropNextRecoverySnapshotEnd = false
+  holdNextRecoverySnapshot = false
   snapshotRequests: (number | undefined)[] = []
   private heldManualRequestId: number | null = null
   private snapshotData = 'INITIAL'
@@ -73,6 +74,11 @@ class FakeMultiplexServer {
       // Resync request: the server serializes the *current* buffer, so recovery
       // includes everything the client missed.
       this.snapshotData = 'RECOVERED'
+      if (typeof payload?.requestId !== 'number' && this.holdNextRecoverySnapshot) {
+        // The reply's binary frames were all dropped under backpressure.
+        this.holdNextRecoverySnapshot = false
+        return
+      }
       if (typeof payload?.requestId !== 'number' && this.truncateNextRecoverySnapshot) {
         this.truncateNextRecoverySnapshot = false
         this.sendSnapshot(undefined, { truncated: true })
@@ -244,14 +250,50 @@ describe('remote terminal frame-drop resync', () => {
     expect(data).toEqual(['aaa', 'ddd'])
   })
 
-  it('retries after a truncated recovery without accepting output across the gap', async () => {
-    const { data, snapshots } = await subscribeClient()
-    server.truncateNextRecoverySnapshot = true
+  it('retries a truncated recovery on a backoff without accepting output across the gap', async () => {
+    vi.useFakeTimers()
+    try {
+      const { data, snapshots } = await subscribeClient()
+      server.truncateNextRecoverySnapshot = true
+
+      server.output('aaa')
+      server.dropNextOutput = true
+      server.output('bbb')
+      server.output('ccc')
+      // The gate stays shut across the backoff: the post-gap tail is corrupt,
+      // and retrying once per chunk would stampede a flooded server.
+      server.output('ddd')
+      expect(server.snapshotRequests).toEqual([undefined])
+
+      // The retry fires from the backoff timer alone — no further output needed.
+      await vi.advanceTimersByTimeAsync(500)
+      expect(server.snapshotRequests).toEqual([undefined, undefined])
+
+      server.output('eee')
+      expect(snapshots).toEqual(['INITIAL', '\x1b[2J\x1b[3J\x1b[HRECOVERED'])
+      expect(data).toEqual(['aaa', 'eee'])
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+
+  it('re-opens the live path when only the JSON error event for a resync survives', async () => {
+    const { data, snapshots, stream } = await subscribeClient()
+    server.holdNextRecoverySnapshot = true
 
     server.output('aaa')
     server.dropNextOutput = true
     server.output('bbb')
     server.output('ccc')
+    expect(server.snapshotRequests).toEqual([undefined])
+
+    // The paired binary Error frame was dropped under backpressure; only the
+    // reliable JSON error event arrives. It must release the resync gate.
+    subscriptionCallbacks.onResponse({
+      ok: true,
+      result: { type: 'error', streamId: stream.streamId, message: 'snapshot failed' }
+    })
+
     server.output('ddd')
     server.output('eee')
 
@@ -260,6 +302,31 @@ describe('remote terminal frame-drop resync', () => {
     expect(data).toEqual(['aaa', 'eee'])
   })
 
+  it('dispatches the deferred resync when a JSON error consumes the manual snapshot', async () => {
+    const { data, snapshots, stream } = await subscribeClient()
+    server.holdNextManualSnapshot = true
+    const manualSnapshot = stream.serializeBuffer({ scrollbackRows: 100 })
+    await Promise.resolve()
+
+    server.output('aaa')
+    server.dropNextOutput = true
+    server.output('bbb')
+    server.output('ccc')
+    expect(server.snapshotRequests).toHaveLength(1)
+
+    subscriptionCallbacks.onResponse({
+      ok: true,
+      result: { type: 'error', streamId: stream.streamId, message: 'stream failed' }
+    })
+    await expect(manualSnapshot).rejects.toThrow('stream failed')
+
+    expect(server.snapshotRequests).toEqual([expect.any(Number), undefined])
+    expect(snapshots).toEqual(['INITIAL', '\x1b[2J\x1b[3J\x1b[HRECOVERED'])
+
+    server.output('ddd')
+    expect(data).toEqual(['aaa', 'ddd'])
+  })
+
   it('times out a dropped recovery end and retries on the next sequence gap', async () => {
     vi.useFakeTimers()
     try {

src/renderer/src/runtime/remote-runtime-terminal-multiplexer.ts

@@ -94,6 +94,7 @@ type RemoteRuntimeMultiplexedTerminalState = {
   resyncInFlight: boolean
   resyncPendingSend: boolean
   resyncTimer: ReturnType<typeof setTimeout> | null
+  resyncAttempts: number
 }
 
 type RemoteRuntimeSnapshotInfo = {
@@ -129,6 +130,10 @@ const CONTROL_STREAM_ID = 0
 const MAX_REMOTE_TERMINAL_SNAPSHOT_BYTES = 2 * 1024 * 1024
 const REMOTE_TERMINAL_SNAPSHOT_REQUEST_TIMEOUT_MS = 10_000
 const REMOTE_TERMINAL_RESYNC_TIMEOUT_MS = 10_000
+// Why: a truncated recovery means the server is too flooded to serialize;
+// retrying once per incoming chunk would stampede it, so back off instead.
+const REMOTE_TERMINAL_RESYNC_RETRY_BASE_MS = 500
+const REMOTE_TERMINAL_RESYNC_RETRY_MAX_MS = 5_000
 // Why: exported so the transport can classify it as benign — the snapshot was
 // skipped but live output continues, so it must not surface a fatal red banner.
 export const REMOTE_TERMINAL_SNAPSHOT_TOO_LARGE =
@@ -253,7 +258,8 @@ class RemoteRuntimeTerminalMultiplexer {
       recoverySnapshotSeq: undefined,
       resyncInFlight: false,
       resyncPendingSend: false,
-      resyncTimer: null
+      resyncTimer: null,
+      resyncAttempts: 0
     }
     this.streams.set(streamId, state)
 
@@ -431,11 +437,19 @@ class RemoteRuntimeTerminalMultiplexer {
       this.closeIfIdle()
     } else if (event.type === 'error') {
       clearSnapshot(stream)
-      clearResyncTimer(stream)
       rejectPendingSnapshotRequest(
         stream,
         typeof event.message === 'string' ? event.message : 'Remote terminal stream failed.'
       )
+      // Why: the paired binary Error frame can be dropped under backpressure;
+      // this reliable event must also dispatch or release the resync gate, and
+      // must never disarm the watchdog while leaving the gate shut.
+      if (stream.resyncPendingSend) {
+        this.sendDeferredResyncSnapshot(stream)
+      } else {
+        clearResyncTimer(stream)
+        stream.resyncInFlight = false
+      }
       stream.callbacks.onError?.(
         typeof event.message === 'string' ? event.message : 'Remote terminal stream failed.'
       )
@@ -631,15 +645,23 @@ class RemoteRuntimeTerminalMultiplexer {
         stream.initialSnapshotReceived = true
         stream.callbacks.onSubscribed?.()
       } else if (target === 'recovery') {
-        clearResyncTimer(stream)
         // Why: only an applied recovery is authoritative; retaining the prior
-        // high-water after a discarded snapshot makes live output retry the gap.
+        // high-water after a discarded snapshot keeps the gap detectable.
         if (snapshotApplied) {
+          clearResyncTimer(stream)
           stream.expectedSeq = typeof info?.seq === 'number' ? info.seq : undefined
           stream.recoverySnapshotSeq = typeof info?.seq === 'number' ? info.seq : undefined
+          stream.resyncAttempts = 0
+          stream.resyncInFlight = false
+          stream.resyncPendingSend = false
+        } else if (stream.resyncInFlight) {
+          this.scheduleResyncRetry(stream)
+        } else {
+          // Why: a discarded server-pushed recovery leaves dropped output
+          // unrepresented; pull a fresh snapshot now instead of waiting for
+          // the next chunk to expose the gap.
+          this.requestResyncSnapshot(stream)
         }
-        stream.resyncInFlight = false
-        stream.resyncPendingSend = false
       } else {
         this.sendDeferredResyncSnapshot(stream)
       }
@@ -690,7 +712,10 @@ class RemoteRuntimeTerminalMultiplexer {
     if (stream.pendingSnapshotRequest) {
       // Why: snapshot frame groups are not multiplexed; wait for the manual
       // snapshot to finish so its response cannot be mistaken for recovery.
+      // Arm the watchdog now so a dispatch path that consumes the pending
+      // request without re-dispatching cannot hold the gate shut forever.
       stream.resyncPendingSend = true
+      this.startResyncTimer(stream)
       return
     }
     this.sendResyncSnapshot(stream)
@@ -718,6 +743,37 @@ class RemoteRuntimeTerminalMultiplexer {
     }
   }
 
+  // Why: keep the gate shut across the backoff — the post-gap tail is corrupt
+  // either way — and heal even if the flood ends with no further output.
+  private scheduleResyncRetry(stream: RemoteRuntimeMultiplexedTerminalState): void {
+    stream.resyncAttempts += 1
+    const delay = Math.min(
+      REMOTE_TERMINAL_RESYNC_RETRY_MAX_MS,
+      REMOTE_TERMINAL_RESYNC_RETRY_BASE_MS * 2 ** Math.min(stream.resyncAttempts - 1, 4)
+    )
+    clearResyncTimer(stream)
+    const timer = setTimeout(() => {
+      if (
+        stream.resyncTimer !== timer ||
+        this.streams.get(stream.streamId) !== stream ||
+        !stream.resyncInFlight
+      ) {
+        return
+      }
+      stream.resyncTimer = null
+      if (stream.pendingSnapshotRequest) {
+        stream.resyncPendingSend = true
+        this.startResyncTimer(stream)
+        return
+      }
+      this.sendResyncSnapshot(stream)
+    }, delay)
+    if (typeof timer.unref === 'function') {
+      timer.unref()
+    }
+    stream.resyncTimer = timer
+  }
+
   private startResyncTimer(stream: RemoteRuntimeMultiplexedTerminalState): void {
     clearResyncTimer(stream)
     const timer = setTimeout(() => {