Fix p1 remote terminal resync (#10089)

* fix(terminal): resume output after recovery resync

* fix(terminal): harden recovery resync boundaries

* 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

@@ -39,6 +39,9 @@ class FakeMultiplexServer {
   dropNextOutput = false
   droppedFrames = 0
   holdNextManualSnapshot = false
+  truncateNextRecoverySnapshot = false
+  dropNextRecoverySnapshotEnd = false
+  holdNextRecoverySnapshot = false
   snapshotRequests: (number | undefined)[] = []
   private heldManualRequestId: number | null = null
   private snapshotData = 'INITIAL'
@@ -71,6 +74,21 @@ 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 })
+        return
+      }
+      if (typeof payload?.requestId !== 'number' && this.dropNextRecoverySnapshotEnd) {
+        this.dropNextRecoverySnapshotEnd = false
+        this.sendSnapshot(undefined, { omitEnd: true })
+        return
+      }
       this.sendSnapshot(payload?.requestId)
     }
   }
@@ -79,14 +97,27 @@ class FakeMultiplexServer {
     this.toClient(encodeTerminalStreamFrame({ opcode, streamId: this.streamId, seq, payload }))
   }
 
-  private sendSnapshot(requestId?: number): void {
+  private sendSnapshot(
+    requestId?: number,
+    options?: { truncated?: boolean; omitEnd?: boolean }
+  ): void {
     this.send(
       TerminalStreamOpcode.SnapshotStart,
-      encodeTerminalStreamJson({ cols: 80, rows: 24, seq: this.cursorUnits, requestId }),
+      encodeTerminalStreamJson({
+        cols: 80,
+        rows: 24,
+        seq: options?.truncated ? undefined : this.cursorUnits,
+        requestId,
+        truncated: options?.truncated
+      }),
       0
     )
-    this.send(TerminalStreamOpcode.SnapshotChunk, encodeTerminalStreamText(this.snapshotData), 0)
-    this.send(TerminalStreamOpcode.SnapshotEnd, new Uint8Array(), 0)
+    if (!options?.truncated) {
+      this.send(TerminalStreamOpcode.SnapshotChunk, encodeTerminalStreamText(this.snapshotData), 0)
+    }
+    if (!options?.omitEnd) {
+      this.send(TerminalStreamOpcode.SnapshotEnd, new Uint8Array(), 0)
+    }
   }
 
   /** Emit an Output chunk, honoring simulated websocket backpressure. */
@@ -123,6 +154,10 @@ class FakeMultiplexServer {
     )
   }
 
+  replaySnapshotCoveredOutput(text: string): void {
+    this.send(TerminalStreamOpcode.Output, encodeTerminalStreamText(text), this.cursorUnits)
+  }
+
   flushHeldManualSnapshot(): void {
     if (this.heldManualRequestId === null) {
       throw new Error('No manual snapshot is held')
@@ -209,6 +244,109 @@ describe('remote terminal frame-drop resync', () => {
     expect(server.droppedFrames).toBe(1)
     // Instead, a fresh authoritative snapshot recovers the terminal.
     expect(snapshots).toEqual(['INITIAL', '\x1b[2J\x1b[3J\x1b[HRECOVERED'])
+
+    server.replaySnapshotCoveredOutput('ccc')
+    server.output('ddd')
+    expect(data).toEqual(['aaa', 'ddd'])
+  })
+
+  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')
+
+    expect(server.snapshotRequests).toEqual([undefined, undefined])
+    expect(snapshots).toEqual(['INITIAL', '\x1b[2J\x1b[3J\x1b[HRECOVERED'])
+    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 {
+      const { data, snapshots } = await subscribeClient()
+      server.dropNextRecoverySnapshotEnd = true
+
+      server.output('aaa')
+      server.dropNextOutput = true
+      server.output('bbb')
+      server.output('ccc')
+      await vi.advanceTimersByTimeAsync(10_000)
+      server.output('ddd')
+      server.output('eee')
+
+      expect(server.snapshotRequests).toEqual([undefined, undefined])
+      expect(snapshots).toEqual(['INITIAL', '\x1b[2J\x1b[3J\x1b[HRECOVERED'])
+      expect(data).toEqual(['aaa', 'eee'])
+    } finally {
+      vi.useRealTimers()
+    }
   })
 
   it('passes contiguous output straight through without resyncing', async () => {
@@ -352,5 +490,8 @@ describe('remote terminal frame-drop resync', () => {
     await expect(manualSnapshot).resolves.toMatchObject({ data: 'MANUAL' })
     expect(server.snapshotRequests).toHaveLength(2)
     expect(snapshots).toEqual(['INITIAL', '\x1b[2J\x1b[3J\x1b[HRECOVERED'])
+
+    server.output('ddd')
+    expect(data).toEqual(['aaa', 'ddd'])
   })
 })

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

@@ -90,8 +90,11 @@ type RemoteRuntimeMultiplexedTerminalState = {
   // Track it so a gap triggers a self-healing snapshot resync instead of
   // silently rendering corrupt/missing output (frame-drop resync).
   expectedSeq: number | undefined
+  recoverySnapshotSeq: number | undefined
   resyncInFlight: boolean
   resyncPendingSend: boolean
+  resyncTimer: ReturnType<typeof setTimeout> | null
+  resyncAttempts: number
 }
 
 type RemoteRuntimeSnapshotInfo = {
@@ -126,6 +129,11 @@ type RemoteRuntimeSnapshotRequest = {
 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 =
@@ -247,8 +255,11 @@ class RemoteRuntimeTerminalMultiplexer {
       initialSnapshotReceived: false,
       pendingSnapshotRequest: null,
       expectedSeq: undefined,
+      recoverySnapshotSeq: undefined,
       resyncInFlight: false,
-      resyncPendingSend: false
+      resyncPendingSend: false,
+      resyncTimer: null,
+      resyncAttempts: 0
     }
     this.streams.set(streamId, state)
 
@@ -282,6 +293,7 @@ class RemoteRuntimeTerminalMultiplexer {
       close: () => {
         if (this.streams.get(streamId) === state) {
           this.sendFrame(streamId, TerminalStreamOpcode.Unsubscribe)
+          clearResyncTimer(state)
           rejectPendingSnapshotRequest(state, 'Remote terminal stream closed.')
           this.streams.delete(streamId)
           this.closeIfIdle()
@@ -418,6 +430,7 @@ class RemoteRuntimeTerminalMultiplexer {
     }
     if (event.type === 'end') {
       clearSnapshot(stream)
+      clearResyncTimer(stream)
       rejectPendingSnapshotRequest(stream, 'Remote terminal stream ended.')
       this.streams.delete(event.streamId)
       stream.callbacks.onEnd?.()
@@ -428,6 +441,15 @@ class RemoteRuntimeTerminalMultiplexer {
         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.'
       )
@@ -509,6 +531,15 @@ class RemoteRuntimeTerminalMultiplexer {
           return
         }
         const seq = typeof frame.seq === 'number' && frame.seq > 0 ? frame.seq : undefined
+        // Why: older servers replay snapshot-covered buffered chunks after a
+        // requested recovery; rendering them would duplicate the recovered tail.
+        if (
+          typeof seq === 'number' &&
+          typeof stream.recoverySnapshotSeq === 'number' &&
+          seq <= stream.recoverySnapshotSeq
+        ) {
+          return
+        }
         if (this.detectOutputGap(stream, seq, rawLength)) {
           this.requestResyncSnapshot(stream)
           return
@@ -570,13 +601,14 @@ class RemoteRuntimeTerminalMultiplexer {
       const target = stream.snapshotTarget
       const info = stream.snapshotInfo
       const pendingRequest = stream.pendingSnapshotRequest
+      const snapshotApplied = !stream.snapshotOverflowed && info?.truncated !== true
       const matchesPendingRequest =
         target === 'request' &&
         pendingRequest &&
         (typeof info?.requestId === 'number'
           ? info.requestId === pendingRequest.requestId
           : stream.initialSnapshotReceived)
-      if (!stream.snapshotOverflowed && info?.truncated !== true) {
+      if (snapshotApplied) {
         if (matchesPendingRequest) {
           pendingRequest.resolve({
             data: data ?? '',
@@ -605,15 +637,31 @@ class RemoteRuntimeTerminalMultiplexer {
         clearPendingSnapshotRequest(stream)
       }
       clearSnapshot(stream)
-      // Why: the snapshot is the new authoritative output high-water; align the
-      // gap detector to it and re-open the live path (used by both the initial
-      // snapshot and a frame-drop resync, which reuses the 'initial' target).
       if (target === 'initial') {
+        clearResyncTimer(stream)
         stream.expectedSeq = typeof info?.seq === 'number' ? info.seq : undefined
         stream.resyncInFlight = false
         stream.resyncPendingSend = false
         stream.initialSnapshotReceived = true
         stream.callbacks.onSubscribed?.()
+      } else if (target === 'recovery') {
+        // Why: only an applied recovery is authoritative; retaining the prior
+        // 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)
+        }
       } else {
         this.sendDeferredResyncSnapshot(stream)
       }
@@ -629,6 +677,7 @@ class RemoteRuntimeTerminalMultiplexer {
         return
       }
       // Why: a failed resync must re-open the live path or output stalls forever.
+      clearResyncTimer(stream)
       stream.resyncInFlight = false
       stream.resyncPendingSend = false
       stream.callbacks.onError?.(decodeTerminalStreamText(frame.payload))
@@ -660,11 +709,13 @@ class RemoteRuntimeTerminalMultiplexer {
       return
     }
     stream.resyncInFlight = true
-    stream.expectedSeq = undefined
     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)
@@ -679,6 +730,7 @@ class RemoteRuntimeTerminalMultiplexer {
 
   private sendResyncSnapshot(stream: RemoteRuntimeMultiplexedTerminalState): void {
     stream.resyncPendingSend = false
+    this.startResyncTimer(stream)
     const sent = this.sendFrame(
       stream.streamId,
       TerminalStreamOpcode.SnapshotRequest,
@@ -686,10 +738,62 @@ class RemoteRuntimeTerminalMultiplexer {
     )
     if (!sent) {
       // Transport is down; the reconnect path re-subscribes from scratch.
+      clearResyncTimer(stream)
       stream.resyncInFlight = false
     }
   }
 
+  // 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(() => {
+      if (
+        stream.resyncTimer !== timer ||
+        this.streams.get(stream.streamId) !== stream ||
+        !stream.resyncInFlight
+      ) {
+        return
+      }
+      stream.resyncTimer = null
+      stream.resyncInFlight = false
+      stream.resyncPendingSend = false
+    }, REMOTE_TERMINAL_RESYNC_TIMEOUT_MS)
+    if (typeof timer.unref === 'function') {
+      timer.unref()
+    }
+    stream.resyncTimer = timer
+  }
+
   private requestSnapshot(
     stream: RemoteRuntimeMultiplexedTerminalState,
     opts?: { scrollbackRows?: number }
@@ -820,6 +924,7 @@ class RemoteRuntimeTerminalMultiplexer {
     this.releaseIfCurrent(this.environmentId, this)
     for (const stream of streams) {
       clearSnapshot(stream)
+      clearResyncTimer(stream)
       rejectPendingSnapshotRequest(stream, message ?? 'Remote runtime connection closed.')
       const canHandleClose = Boolean(stream.callbacks.onTransportClose)
       stream.callbacks.onTransportClose?.({ recoverable })
@@ -909,6 +1014,14 @@ function clearPendingSnapshotRequest(stream: RemoteRuntimeMultiplexedTerminalSta
   }
 }
 
+function clearResyncTimer(stream: RemoteRuntimeMultiplexedTerminalState): void {
+  const timer = stream.resyncTimer
+  stream.resyncTimer = null
+  if (timer) {
+    clearTimeout(timer)
+  }
+}
+
 function rejectPendingSnapshotRequest(
   stream: RemoteRuntimeMultiplexedTerminalState,
   message: string