fix(remote-terminal): keep live output flowing during resync and retry an unanswered one (#29236)

* fix(remote-terminal): keep live output flowing during resync and retry an unanswered one

Co-authored-by: Claude <noreply@anthropic.com>

* fix(remote-terminal): escalate to stall recovery when a resync is never answered

Co-authored-by: Claude <noreply@anthropic.com>

---------

Co-authored-by: Claude <noreply@anthropic.com>

src/renderer/src/runtime/remote-runtime-terminal-binary-controller.ts

@@ -97,11 +97,6 @@ export abstract class RemoteRuntimeTerminalBinaryController extends RemoteRuntim
         frame.opcode === TerminalStreamOpcode.OutputSpan && typeof span?.rawLength === 'number'
           ? span.rawLength
           : data.length
-      // Why: a resync snapshot is authoritative; discard live output while
-      // it is in flight, but still return transport credit in finally.
-      if (stream.resyncInFlight) {
-        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.
@@ -112,11 +107,15 @@ export abstract class RemoteRuntimeTerminalBinaryController extends RemoteRuntim
       ) {
         return
       }
-      if (this.detectOutputGap(stream, seq, rawLength)) {
+      const resyncInFlight = stream.resyncInFlight
+      // Why: keep painting while recovery is in flight — gating froze the pane for
+      // the whole round trip (#18488); the snapshot replaces the screen when it lands.
+      // expectedSeq stays at the gap so a recovery that never lands can still be retried.
+      if (!resyncInFlight && this.detectOutputGap(stream, seq, rawLength)) {
         this.requestResyncSnapshot(stream)
         return
       }
-      if (typeof seq === 'number') {
+      if (typeof seq === 'number' && !resyncInFlight) {
         stream.expectedSeq = seq
         stream.commandProbeBaselineSeq = undefined
       }

src/renderer/src/runtime/remote-runtime-terminal-binary-snapshots.ts

@@ -38,6 +38,9 @@ export abstract class RemoteRuntimeTerminalBinarySnapshots extends RemoteRuntime
           : stream.initialSnapshotReceived
             ? 'recovery'
             : 'initial'
+      if (stream.snapshotTarget === 'recovery') {
+        this.noteResyncSnapshotProgress(stream)
+      }
       return
     }
     if (frame.opcode === TerminalStreamOpcode.SnapshotChunk) {
@@ -53,6 +56,9 @@ export abstract class RemoteRuntimeTerminalBinarySnapshots extends RemoteRuntime
         return
       }
       stream.snapshotChunks.push(frame.payload)
+      if (stream.snapshotTarget === 'recovery') {
+        this.noteResyncSnapshotProgress(stream)
+      }
       return
     }
     if (frame.opcode === TerminalStreamOpcode.SnapshotEnd) {

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

@@ -43,9 +43,11 @@ class FakeMultiplexServer {
   truncateNextRecoverySnapshot = false
   dropNextRecoverySnapshotEnd = false
   holdNextRecoverySnapshot = false
+  streamNextRecoverySnapshot = false
   snapshotRequests: (number | undefined)[] = []
   snapshotRequestRows: (number | undefined)[] = []
   private heldManualRequestId: number | null = null
+  private heldRecoverySnapshot = false
   private snapshotData = 'INITIAL'
 
   constructor(
@@ -82,6 +84,17 @@ class FakeMultiplexServer {
       if (typeof payload?.requestId !== 'number' && this.holdNextRecoverySnapshot) {
         // The reply's binary frames were all dropped under backpressure.
         this.holdNextRecoverySnapshot = false
+        this.heldRecoverySnapshot = true
+        return
+      }
+      if (typeof payload?.requestId !== 'number' && this.streamNextRecoverySnapshot) {
+        // A slow link: the reply starts now and its chunks trickle in later.
+        this.streamNextRecoverySnapshot = false
+        this.send(
+          TerminalStreamOpcode.SnapshotStart,
+          encodeTerminalStreamJson({ cols: 80, rows: 24, seq: this.cursorUnits }),
+          0
+        )
         return
       }
       if (typeof payload?.requestId !== 'number' && this.truncateNextRecoverySnapshot) {
@@ -172,6 +185,22 @@ class FakeMultiplexServer {
     this.snapshotData = 'MANUAL'
     this.sendSnapshot(requestId)
   }
+
+  streamRecoveryChunk(text: string): void {
+    this.send(TerminalStreamOpcode.SnapshotChunk, encodeTerminalStreamText(text), 0)
+  }
+
+  endStreamedRecovery(): void {
+    this.send(TerminalStreamOpcode.SnapshotEnd, new Uint8Array(), 0)
+  }
+
+  flushHeldRecoverySnapshot(): void {
+    if (!this.heldRecoverySnapshot) {
+      throw new Error('No recovery snapshot is held')
+    }
+    this.heldRecoverySnapshot = false
+    this.sendSnapshot()
+  }
 }
 
 describe('remote terminal frame-drop resync', () => {
@@ -242,7 +271,8 @@ describe('remote terminal frame-drop resync', () => {
     await Promise.resolve()
     await Promise.resolve()
 
-    // The corrupt tail ('ccc', which followed a gap) is NOT rendered as live data.
+    // The immediate snapshot covers the corrupt tail ('ccc'), so it is not
+    // rendered separately or duplicated after the recovery reset.
     expect(data).toEqual(['aaa'])
     expect(server.droppedFrames).toBe(1)
     // Instead, a fresh authoritative snapshot recovers the terminal.
@@ -255,7 +285,7 @@ describe('remote terminal frame-drop resync', () => {
     expect(data).toEqual(['aaa', 'ddd'])
   })
 
-  it('retries a truncated recovery on a backoff without accepting output across the gap', async () => {
+  it('retries a truncated recovery on a backoff while keeping live output flowing', async () => {
     vi.useFakeTimers()
     try {
       const { data, snapshots } = await subscribeClient()
@@ -265,8 +295,8 @@ describe('remote terminal frame-drop resync', () => {
       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.
+      // Recovery retries are backoff-limited, but the live path stays open;
+      // output must not freeze while a jittery host serializes the snapshot.
       server.output('ddd')
       expect(server.snapshotRequests).toEqual([undefined])
 
@@ -276,12 +306,59 @@ describe('remote terminal frame-drop resync', () => {
 
       server.output('eee')
       expect(snapshots).toEqual(['INITIAL', '\x1b[?2026l\x1b[2J\x1b[3J\x1b[HRECOVERED'])
-      expect(data).toEqual(['aaa', 'eee'])
+      expect(data).toEqual(['aaa', 'ddd', 'eee'])
     } finally {
       vi.useRealTimers()
     }
   })
 
+  it('keeps painting live output while a recovery snapshot is delayed', async () => {
+    const { data, snapshots, stream } = await subscribeClient()
+    try {
+      server.holdNextRecoverySnapshot = true
+
+      server.output('aaa')
+      server.dropNextOutput = true
+      server.output('bbb')
+      server.output('ccc') // exposes the gap and starts the delayed recovery
+      server.output('ddd')
+      server.output('eee')
+
+      // A slow snapshot must not become a renderer backpressure gate.
+      expect(data).toEqual(['aaa', 'ddd', 'eee'])
+      expect(snapshots).toEqual(['INITIAL'])
+    } finally {
+      stream.close()
+    }
+  })
+
+  it('reconciles painted output when a delayed recovery snapshot arrives', async () => {
+    const { data, snapshots, stream } = await subscribeClient()
+    try {
+      server.holdNextRecoverySnapshot = true
+
+      server.output('aaa')
+      server.dropNextOutput = true
+      server.output('bbb')
+      server.output('ccc')
+      server.output('ddd')
+      server.output('eee')
+
+      expect(data).toEqual(['aaa', 'ddd', 'eee'])
+      expect(snapshots).toEqual(['INITIAL'])
+
+      server.flushHeldRecoverySnapshot()
+      expect(snapshots).toEqual(['INITIAL', '\x1b[?2026l\x1b[2J\x1b[3J\x1b[HRECOVERED'])
+
+      // A replay covered by the authoritative snapshot must not duplicate.
+      server.replaySnapshotCoveredOutput('eee')
+      server.output('fff')
+      expect(data).toEqual(['aaa', 'ddd', 'eee', 'fff'])
+    } finally {
+      stream.close()
+    }
+  })
+
   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
@@ -332,7 +409,7 @@ describe('remote terminal frame-drop resync', () => {
     expect(data).toEqual(['aaa', 'ddd'])
   })
 
-  it('times out a dropped recovery end and retries on the next sequence gap', async () => {
+  it('retries a recovery whose end was dropped without waiting for more output', async () => {
     vi.useFakeTimers()
     try {
       const { data, snapshots } = await subscribeClient()
@@ -342,13 +419,95 @@ describe('remote terminal frame-drop resync', () => {
       server.dropNextOutput = true
       server.output('bbb')
       server.output('ccc')
+      server.output('ddd')
       await vi.advanceTimersByTimeAsync(10_000)
+      expect(server.snapshotRequests).toEqual([undefined])
+
+      // #18488: the timeout itself retries; an idle pane still heals.
+      await vi.advanceTimersByTimeAsync(500)
+      expect(server.snapshotRequests).toEqual([undefined, undefined])
+      expect(snapshots).toEqual(['INITIAL', '\x1b[?2026l\x1b[2J\x1b[3J\x1b[HRECOVERED'])
+
+      server.output('eee')
+      expect(data).toEqual(['aaa', 'ddd', 'eee'])
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+
+  it('does not re-gate on every timeout while recovery cannot complete', async () => {
+    vi.useFakeTimers()
+    try {
+      const { data } = await subscribeClient()
+      server.holdNextRecoverySnapshot = true
+
+      server.output('aaa')
+      server.dropNextOutput = true
+      server.output('bbb')
+      server.output('ccc')
+      await vi.advanceTimersByTimeAsync(10_000)
+      server.holdNextRecoverySnapshot = true
+      // Before #18488 the frame after each timeout re-tripped the stale gap and was dropped.
       server.output('ddd')
       server.output('eee')
 
+      expect(data).toEqual(['aaa', 'ddd', 'eee'])
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+
+  it('escalates to stall recovery when the host never answers a recovery', async () => {
+    vi.useFakeTimers()
+    try {
+      const onTransportClose = vi.fn()
+      await getRemoteRuntimeTerminalMultiplexer('env-1').subscribeTerminal({
+        terminal: 'terminal-1',
+        client: { id: 'desktop-1', type: 'desktop' },
+        callbacks: { onData: vi.fn(), onSnapshot: vi.fn(), onTransportClose }
+      })
+      await Promise.resolve()
+      await Promise.resolve()
+      server.holdNextRecoverySnapshot = true
+      server.output('aaa')
+      server.dropNextOutput = true
+      server.output('bbb')
+      server.output('ccc')
+      await vi.advanceTimersByTimeAsync(10_000)
+      server.holdNextRecoverySnapshot = true
+      await vi.advanceTimersByTimeAsync(500)
       expect(server.snapshotRequests).toEqual([undefined, undefined])
-      expect(snapshots).toEqual(['INITIAL', '\x1b[?2026l\x1b[2J\x1b[3J\x1b[HRECOVERED'])
-      expect(data).toEqual(['aaa', 'eee'])
+      expect(onTransportClose).not.toHaveBeenCalled()
+
+      await vi.advanceTimersByTimeAsync(10_000)
+      expect(onTransportClose).toHaveBeenCalledWith({ recoverable: true })
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+
+  it('does not re-request a recovery that is still streaming on a slow link', async () => {
+    vi.useFakeTimers()
+    try {
+      const { data, snapshots } = await subscribeClient()
+      server.streamNextRecoverySnapshot = true
+
+      server.output('aaa')
+      server.dropNextOutput = true
+      server.output('bbb')
+      server.output('ccc')
+      await vi.advanceTimersByTimeAsync(6_000)
+      server.streamRecoveryChunk('SLOW-')
+      await vi.advanceTimersByTimeAsync(6_000)
+      server.streamRecoveryChunk('RECOVERY')
+      await vi.advanceTimersByTimeAsync(6_000)
+      server.endStreamedRecovery()
+      await vi.advanceTimersByTimeAsync(10_000)
+
+      expect(server.snapshotRequests).toEqual([undefined])
+      expect(snapshots).toEqual(['INITIAL', '\x1b[?2026l\x1b[2J\x1b[3J\x1b[HSLOW-RECOVERY'])
+      server.output('ddd')
+      expect(data).toEqual(['aaa', 'ddd'])
     } finally {
       vi.useRealTimers()
     }

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

@@ -56,6 +56,7 @@ export class RemoteRuntimeTerminalMultiplexer extends RemoteRuntimeTerminalBinar
       resyncPendingSend: false,
       resyncTimer: null,
       resyncAttempts: 0,
+      resyncUnansweredTimeouts: 0,
       capacityRejected: false,
       watchdog: createRemoteTerminalStreamWatchdog((stall) => {
         if (e2eDisableRemoteTerminalStallRecovery) {

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

@@ -193,6 +193,8 @@ export type RemoteRuntimeMultiplexedTerminalState = {
   resyncPendingSend: boolean
   resyncTimer: ReturnType<typeof setTimeout> | null
   resyncAttempts: number
+  /** Consecutive resync timeouts with no recovery frame at all. */
+  resyncUnansweredTimeouts: number
   capacityRejected: boolean
   watchdog: RemoteTerminalStreamWatchdog
 }

src/renderer/src/runtime/remote-runtime-terminal-snapshot-controller.ts

@@ -6,6 +6,7 @@ import { TERMINAL_MULTIPLEX_RECOVERY_SCROLLBACK_ROWS } from '../../../shared/ter
 import { recordRendererCrashBreadcrumb } from '@/lib/crash-breadcrumb-recorder'
 import { RemoteRuntimeTerminalMultiplexerBase } from './remote-runtime-terminal-multiplexer-base'
 import {
+  REMOTE_TERMINAL_RESYNC_MAX_UNANSWERED_TIMEOUTS,
   REMOTE_TERMINAL_RESYNC_RETRY_BASE_MS,
   REMOTE_TERMINAL_RESYNC_RETRY_MAX_MS,
   REMOTE_TERMINAL_RESYNC_TIMEOUT_MS,
@@ -38,10 +39,9 @@ export abstract class RemoteRuntimeTerminalSnapshotController extends RemoteRunt
     return startSeq > stream.expectedSeq
   }
 
-  // Why: on a detected gap, discard the corrupt tail and pull a fresh
-  // authoritative snapshot. The request carries no requestId so the server
-  // reply renders through the initial-snapshot path (full reset), self-healing
-  // without surfacing an error to the user.
+  // Why: on a detected gap, pull a fresh authoritative snapshot. The request
+  // carries no requestId so the server reply uses the recovery path (full
+  // reset), self-healing without surfacing an error to the user.
   protected requestResyncSnapshot(stream: RemoteRuntimeMultiplexedTerminalState): void {
     if (stream.resyncInFlight) {
       return
@@ -51,7 +51,7 @@ export abstract class RemoteRuntimeTerminalSnapshotController extends RemoteRunt
       // 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.
+      // request without re-dispatching cannot leave recovery pending forever.
       stream.resyncPendingSend = true
       this.startResyncTimer(stream)
       return
@@ -82,8 +82,8 @@ export abstract class RemoteRuntimeTerminalSnapshotController extends RemoteRunt
     }
   }
 
-  // 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.
+  // Why: keep recovery pending across the backoff and heal even if the flood
+  // ends with no further output.
   protected scheduleResyncRetry(stream: RemoteRuntimeMultiplexedTerminalState): void {
     stream.resyncAttempts += 1
     const delay = Math.min(
@@ -113,6 +113,15 @@ export abstract class RemoteRuntimeTerminalSnapshotController extends RemoteRunt
     stream.resyncTimer = timer
   }
 
+  // Why: a large recovery on a slow link can take longer than the timeout to
+  // stream; expiring mid-transfer would request a second one and repaint twice.
+  protected noteResyncSnapshotProgress(stream: RemoteRuntimeMultiplexedTerminalState): void {
+    stream.resyncUnansweredTimeouts = 0
+    if (stream.resyncInFlight && !stream.resyncPendingSend) {
+      this.startResyncTimer(stream)
+    }
+  }
+
   private startResyncTimer(stream: RemoteRuntimeMultiplexedTerminalState): void {
     clearResyncTimer(stream)
     const timer = setTimeout(() => {
@@ -124,8 +133,15 @@ export abstract class RemoteRuntimeTerminalSnapshotController extends RemoteRunt
         return
       }
       stream.resyncTimer = null
-      stream.resyncInFlight = false
-      stream.resyncPendingSend = false
+      stream.resyncUnansweredTimeouts += 1
+      if (stream.resyncUnansweredTimeouts >= REMOTE_TERMINAL_RESYNC_MAX_UNANSWERED_TIMEOUTS) {
+        // Why: pending recovery refuses the watchdog's probe, so escalate here instead.
+        this.recoverStalledStream(stream)
+        return
+      }
+      // Why retry, not reopen: reopening left expectedSeq before the gap, so the
+      // next frame re-tripped it and recovery ran on the timeout's cadence (#18488).
+      this.scheduleResyncRetry(stream)
     }, REMOTE_TERMINAL_RESYNC_TIMEOUT_MS)
     if (typeof timer.unref === 'function') {
       timer.unref()

src/renderer/src/runtime/remote-runtime-terminal-snapshot-state.ts

@@ -13,6 +13,8 @@ export const CONTROL_STREAM_ID = 0
 export const MAX_REMOTE_TERMINAL_SNAPSHOT_BYTES = 2 * 1024 * 1024
 export const REMOTE_TERMINAL_SNAPSHOT_REQUEST_TIMEOUT_MS = 10_000
 export const REMOTE_TERMINAL_RESYNC_TIMEOUT_MS = 10_000
+// Why: a host that never answers a recovery must reach stall recovery, not retry forever.
+export const REMOTE_TERMINAL_RESYNC_MAX_UNANSWERED_TIMEOUTS = 2
 // Why: a truncated recovery means the server is too flooded to serialize;
 // retrying once per incoming chunk would stampede it, so back off instead.
 export const REMOTE_TERMINAL_RESYNC_RETRY_BASE_MS = 500