Fix paired remote terminal parse backpressure (#9683)

* fix(terminal): defer remote output ACKs until parse

* test(terminal): document synchronous credit claims

---------

Co-authored-by: OrcaWin <293788423+OrcaWin@users.noreply.github.com>

src/renderer/src/components/terminal-pane/pty-connection.test.ts

@@ -13463,6 +13463,8 @@ describe('connectPanePty', () => {
 
   it('holds newer live bytes until a later replay frame has fully parsed', async () => {
     const { connectPanePty } = await import('./pty-connection')
+    const { deliverTerminalDataWithDeferredCredit } =
+      await import('@/lib/pane-manager/terminal-delivery-credit')
     enableActiveRuntimeEnvironment()
     const transport = createMockTransport('remote:env-1@@terminal-live-order')
     const callbacksRef: {
@@ -13484,8 +13486,12 @@ describe('connectPanePty', () => {
     await flushAsyncTicks(8)
     expect(writes).toEqual(['\x1b[2J\x1b[3J\x1b[H'])
 
-    callbacksRef.data?.('NEWER-LIVE\r\n')
+    const acknowledgeLiveFrame = vi.fn()
+    deliverTerminalDataWithDeferredCredit(acknowledgeLiveFrame, () => {
+      callbacksRef.data?.('NEWER-LIVE\r\n')
+    })
     expect(writes).not.toContain('NEWER-LIVE\r\n')
+    expect(acknowledgeLiveFrame).not.toHaveBeenCalled()
     for (let index = 0; index < 12 && parseCallbacks.length > 0; index += 1) {
       parseCallbacks.shift()?.()
       await flushAsyncTicks(4)
@@ -13498,7 +13504,17 @@ describe('connectPanePty', () => {
     expect(replayIndex).toBeGreaterThan(0)
     expect(resetIndex).toBeGreaterThan(replayIndex)
     expect(liveIndex).toBeGreaterThan(resetIndex)
+    expect(acknowledgeLiveFrame).toHaveBeenCalledOnce()
+
+    callbacksRef.replay?.('stalled replay')
+    await flushAsyncTicks(8)
+    const acknowledgeDisposedFrame = vi.fn()
+    deliverTerminalDataWithDeferredCredit(acknowledgeDisposedFrame, () => {
+      callbacksRef.data?.('LIVE-BEHIND-STALLED-REPLAY')
+    })
+    expect(acknowledgeDisposedFrame).not.toHaveBeenCalled()
     binding.dispose()
+    expect(acknowledgeDisposedFrame).toHaveBeenCalledOnce()
   })
 
   it('drops a queued relay replay instead of retagging it for a replacement PTY', async () => {

src/renderer/src/components/terminal-pane/pty-connection.ts

@@ -23,7 +23,10 @@ import {
   isStatefulRendererReplyCsiQuery,
   isStatelessRendererReplyCsiQuery
 } from '../../../../shared/terminal-reply-query-extraction'
-import { takeCurrentPtyDeliveryAckCredit } from './terminal-pty-ack-gate'
+import {
+  deliverTerminalDataWithDeferredCredit,
+  takeCurrentTerminalDeliveryCredit
+} from '@/lib/pane-manager/terminal-delivery-credit'
 import { serializeWithAbsoluteCursor } from '../../../../shared/terminal-serialize-absolute-cursor'
 import { isTerminalQueryReply } from '../../../../shared/terminal-query-reply'
 import type { PtyBufferSnapshot, PtyConnectResult } from './pty-transport'
@@ -3312,7 +3315,13 @@ export function connectPanePty(
   let lastTerminalInputAt = Number.NEGATIVE_INFINITY
   let hasReceivedPtyOutput = false
   let deferredReattachLiveData:
-    | { data: string; ptyId: string | null; streamGeneration: number; meta?: PtyDataMeta }[]
+    | {
+        data: string
+        ptyId: string | null
+        streamGeneration: number
+        meta?: PtyDataMeta
+        ackCredit?: () => void
+      }[]
     | null = null
   let deferredReattachLiveDataChars = 0
   let reattachLiveDataDeferralDepth = 0
@@ -5787,7 +5796,7 @@ export function connectPanePty(
         foreground: foregroundOutput,
         beforeWrite: beforeTerminalOutputWrite,
         // Why: claim the delivery's parse-deferred ACK credit (null outside a delivery); the FIRST scheduler write carries it all and fires when bytes are consumed.
-        ackCredit: takeCurrentPtyDeliveryAckCredit() ?? undefined,
+        ackCredit: takeCurrentTerminalDeliveryCredit() ?? undefined,
         onBackgroundBacklogDropped: markHiddenOutputRestoreNeeded,
         latencySensitive:
           !foreground || parseHiddenStartupOutput
@@ -6881,20 +6890,26 @@ export function connectPanePty(
       }
       if (deferredReattachLiveData !== null) {
         // Why: a replacement stream must not inherit bytes or a gap marker from the replay owner it superseded.
-        deferredReattachLiveData = deferredReattachLiveData.filter(
-          (chunk) => chunk.streamGeneration === streamGeneration
-        )
+        deferredReattachLiveData = deferredReattachLiveData.filter((chunk) => {
+          const keep = chunk.streamGeneration === streamGeneration
+          if (!keep) {
+            chunk.ackCredit?.()
+          }
+          return keep
+        })
         deferredReattachLiveDataChars = deferredReattachLiveData.reduce(
           (total, chunk) => total + chunk.data.length,
           0
         )
         const oversized = data.length > MAX_DEFERRED_REATTACH_LIVE_CHARS
         const deferredData = oversized ? data.slice(-MAX_DEFERRED_REATTACH_LIVE_CHARS) : data
+        const ackCredit = takeCurrentTerminalDeliveryCredit()
         deferredReattachLiveData.push({
           data: deferredData,
           ptyId: transport.getPtyId(),
           streamGeneration,
-          ...(meta ? { meta } : {})
+          ...(meta ? { meta } : {}),
+          ...(ackCredit ? { ackCredit } : {})
         })
         deferredReattachLiveDataChars += deferredData.length
         // Why: one huge IPC frame would bypass the queue's memory bound; mark a stream gap so snapshot recovery replaces it, not a partial ANSI frame.
@@ -6906,6 +6921,7 @@ export function connectPanePty(
         ) {
           const removed = deferredReattachLiveData.shift()
           deferredReattachLiveDataChars -= removed?.data.length ?? 0
+          removed?.ackCredit?.()
           dropped = true
         }
         if (dropped && deferredReattachLiveData[0]) {
@@ -7111,6 +7127,9 @@ export function connectPanePty(
       const currentOwner = deferredReattachLiveDataOwners.get(currentGeneration)
       deferredReattachLiveDataOwners = new Map()
       if (disposed || !chunks) {
+        for (const chunk of chunks ?? []) {
+          chunk.ackCredit?.()
+        }
         return
       }
       // Why: paint the authoritative replay first, then admit deferred live chunks so the replay clear can't erase newer output.
@@ -7121,9 +7140,16 @@ export function connectPanePty(
           chunk.streamGeneration !== currentGeneration ||
           currentOwner?.failed === true
         ) {
+          chunk.ackCredit?.()
           continue
         }
-        dataCallback(chunk.data, chunk.meta, chunk.streamGeneration)
+        if (chunk.ackCredit) {
+          deliverTerminalDataWithDeferredCredit(chunk.ackCredit, () => {
+            dataCallback(chunk.data, chunk.meta, chunk.streamGeneration)
+          })
+        } else {
+          dataCallback(chunk.data, chunk.meta, chunk.streamGeneration)
+        }
         deliveredDeferredChunks += 1
       }
       if (deliveredDeferredChunks > 0) {
@@ -8157,6 +8183,14 @@ export function connectPanePty(
     reconcileIfSessionMissing,
     dispose() {
       disposed = true
+      // Why: a stalled xterm replay may never reach its finally; release live-frame credit when this renderer no longer owns the stream.
+      for (const chunk of deferredReattachLiveData ?? []) {
+        chunk.ackCredit?.()
+      }
+      deferredReattachLiveData = null
+      deferredReattachLiveDataChars = 0
+      reattachLiveDataDeferralDepth = 0
+      deferredReattachLiveDataOwners = new Map()
       cancelPendingSafeFitContinuations(pane)
       pendingHiddenSnapshotFit = null
       pendingReattachFit = null

src/renderer/src/components/terminal-pane/terminal-pty-ack-gate.ts

@@ -1,4 +1,8 @@
 import { e2eConfig } from '@/lib/e2e-config'
+import {
+  deliverTerminalDataWithDeferredCredit,
+  takeCurrentTerminalDeliveryCredit
+} from '@/lib/pane-manager/terminal-delivery-credit'
 
 type E2eTerminalPtyAckGateSnapshot = {
   gatedPtyCount: number
@@ -89,25 +93,6 @@ export function ackPtyData(ptyId: string, chars: number): void {
 // true parse backpressure and main's producer flow control pauses the shell
 // instead of dropping.
 
-type DeferredPtyAckCredit = {
-  ptyId: string
-  chars: number
-  claimed: boolean
-  credited: boolean
-}
-
-let currentDeliveryCredit: DeferredPtyAckCredit | null = null
-
-function creditDeferredPtyAck(credit: DeferredPtyAckCredit): void {
-  // Why fire-once: split queue chunks and discard paths may both touch the
-  // same delivery; the invariant is exactly one credit per delivered chunk.
-  if (credit.credited) {
-    return
-  }
-  credit.credited = true
-  ackPtyData(credit.ptyId, credit.chars)
-}
-
 /** Runs one pty:data delivery with a parse-deferred ACK credit. If the
  *  handler hands bytes to the output scheduler, the claimed credit fires when
  *  the scheduler consumes (writes or discards) them; any credit left
@@ -118,28 +103,14 @@ export function deliverPtyDataWithDeferredAck(
   chars: number,
   deliver: () => void
 ): void {
-  const credit: DeferredPtyAckCredit = { ptyId, chars, claimed: false, credited: false }
-  currentDeliveryCredit = credit
-  try {
-    deliver()
-  } finally {
-    currentDeliveryCredit = null
-    if (!credit.claimed) {
-      creditDeferredPtyAck(credit)
-    }
-  }
+  deliverTerminalDataWithDeferredCredit(() => ackPtyData(ptyId, chars), deliver)
 }
 
 /** Claims the in-progress delivery's credit for the output scheduler. Returns
  *  a fire-once callback, or null when outside a delivery or already claimed
  *  (only the FIRST scheduler write of a delivery carries the credit). */
 export function takeCurrentPtyDeliveryAckCredit(): (() => void) | null {
-  const credit = currentDeliveryCredit
-  if (!credit || credit.claimed) {
-    return null
-  }
-  credit.claimed = true
-  return () => creditDeferredPtyAck(credit)
+  return takeCurrentTerminalDeliveryCredit()
 }
 
 export function getProcessedPtyCharTotals(): Record<string, number> {

src/renderer/src/lib/pane-manager/terminal-delivery-credit.test.ts

@@ -0,0 +1,44 @@
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+describe('terminal delivery credit', () => {
+  beforeEach(() => {
+    vi.resetModules()
+  })
+
+  it('restores an outer delivery after a nested delivery returns', async () => {
+    const { deliverTerminalDataWithDeferredCredit, takeCurrentTerminalDeliveryCredit } =
+      await import('./terminal-delivery-credit')
+    const completeOuter = vi.fn()
+    const completeInner = vi.fn()
+    let outerCredit: (() => void) | null = null
+    let innerCredit: (() => void) | null = null
+
+    deliverTerminalDataWithDeferredCredit(completeOuter, () => {
+      deliverTerminalDataWithDeferredCredit(completeInner, () => {
+        innerCredit = takeCurrentTerminalDeliveryCredit()
+      })
+      outerCredit = takeCurrentTerminalDeliveryCredit()
+    })
+
+    expect(completeOuter).not.toHaveBeenCalled()
+    expect(completeInner).not.toHaveBeenCalled()
+    innerCredit!()
+    outerCredit!()
+    expect(completeInner).toHaveBeenCalledOnce()
+    expect(completeOuter).toHaveBeenCalledOnce()
+  })
+
+  it('auto-settles before a deferred consumer can claim the delivery', async () => {
+    const { deliverTerminalDataWithDeferredCredit, takeCurrentTerminalDeliveryCredit } =
+      await import('./terminal-delivery-credit')
+    const complete = vi.fn()
+    let claimLater: (() => (() => void) | null) | null = null
+
+    deliverTerminalDataWithDeferredCredit(complete, () => {
+      claimLater = takeCurrentTerminalDeliveryCredit
+    })
+
+    expect(complete).toHaveBeenCalledOnce()
+    expect(claimLater!()).toBeNull()
+  })
+})

src/renderer/src/lib/pane-manager/terminal-delivery-credit.ts

@@ -0,0 +1,49 @@
+type TerminalDeliveryCredit = {
+  complete: () => void
+  claimed: boolean
+  credited: boolean
+}
+
+// Why: consumers must claim during deliver(); after it returns this synchronous slot is restored and unclaimed credit settles.
+let currentDeliveryCredit: TerminalDeliveryCredit | null = null
+
+function completeTerminalDeliveryCredit(credit: TerminalDeliveryCredit): void {
+  // Why: queue splitting and discard paths can both settle one delivery.
+  if (credit.credited) {
+    return
+  }
+  credit.credited = true
+  credit.complete()
+}
+
+/** Defers producer credit until the output scheduler consumes or discards the delivery. */
+export function deliverTerminalDataWithDeferredCredit(
+  complete: () => void,
+  deliver: () => void
+): void {
+  const credit: TerminalDeliveryCredit = {
+    complete,
+    claimed: false,
+    credited: false
+  }
+  const previousCredit = currentDeliveryCredit
+  currentDeliveryCredit = credit
+  try {
+    deliver()
+  } finally {
+    currentDeliveryCredit = previousCredit
+    if (!credit.claimed) {
+      completeTerminalDeliveryCredit(credit)
+    }
+  }
+}
+
+/** Claims the current delivery for parse-deferred settlement by the output scheduler. */
+export function takeCurrentTerminalDeliveryCredit(): (() => void) | null {
+  const credit = currentDeliveryCredit
+  if (!credit || credit.claimed) {
+    return null
+  }
+  credit.claimed = true
+  return () => completeTerminalDeliveryCredit(credit)
+}

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

@@ -11,6 +11,7 @@ import {
   encodeTerminalStreamText
 } from '../../../shared/terminal-stream-protocol'
 import { e2eConfig } from '@/lib/e2e-config'
+import { deliverTerminalDataWithDeferredCredit } from '@/lib/pane-manager/terminal-delivery-credit'
 import { unwrapRuntimeRpcResult } from './runtime-rpc-client'
 
 type RuntimeEnvironmentSubscriptionHandle = {
@@ -472,7 +473,7 @@ class RemoteRuntimeTerminalMultiplexer {
             ? (span!.data as string)
             : ''
           : decodeTerminalStreamText(frame.payload)
-      try {
+      const deliverOutput = (): void => {
         if (!validSpan) {
           // Why: rendering malformed span JSON would expose protocol framing
           // as terminal text and lose its raw sequence accounting.
@@ -501,15 +502,18 @@ class RemoteRuntimeTerminalMultiplexer {
           rawLength,
           ...(frame.opcode === TerminalStreamOpcode.OutputSpan ? { transformed: true } : {})
         })
-      } finally {
-        if (stream.acknowledgeOutput) {
-          if (shouldHoldE2eRemoteTerminalAck(stream.terminal)) {
-            stream.heldAckBytes += frame.payload.byteLength
-          } else {
-            this.acknowledgeOutput(stream, frame.payload.byteLength)
-          }
-        }
       }
+      if (!stream.acknowledgeOutput) {
+        deliverOutput()
+        return
+      }
+      deliverTerminalDataWithDeferredCredit(() => {
+        if (shouldHoldE2eRemoteTerminalAck(stream.terminal)) {
+          stream.heldAckBytes += frame.payload.byteLength
+        } else {
+          this.acknowledgeOutput(stream, frame.payload.byteLength)
+        }
+      }, deliverOutput)
       return
     }
     if (frame.opcode === TerminalStreamOpcode.SnapshotStart) {

src/renderer/src/runtime/remote-runtime-terminal-parse-backpressure.test.ts

@@ -0,0 +1,118 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import {
+  TerminalStreamOpcode,
+  decodeTerminalStreamFrame,
+  decodeTerminalStreamJson,
+  encodeTerminalStreamFrame,
+  encodeTerminalStreamJson,
+  encodeTerminalStreamText
+} from '../../../shared/terminal-stream-protocol'
+
+describe('remote terminal renderer backpressure', () => {
+  const sendBinary = vi.fn()
+  let callbacks: {
+    onResponse: (response: unknown) => void
+    onBinary: (bytes: Uint8Array<ArrayBufferLike>) => void
+  } | null = null
+
+  beforeEach(() => {
+    vi.resetModules()
+    sendBinary.mockReset()
+    callbacks = null
+    vi.stubGlobal('window', {
+      api: {
+        runtimeEnvironments: {
+          subscribe: vi.fn(async (_args, nextCallbacks) => {
+            callbacks = nextCallbacks
+            queueMicrotask(() => {
+              callbacks?.onResponse({ ok: true, result: { type: 'ready' } })
+            })
+            return { unsubscribe: vi.fn(), sendBinary }
+          })
+        }
+      }
+    })
+  })
+
+  afterEach(() => {
+    vi.unstubAllGlobals()
+  })
+
+  it('withholds server credit until xterm consumes the output frame', async () => {
+    const { getRemoteRuntimeTerminalMultiplexer } =
+      await import('./remote-runtime-terminal-multiplexer')
+    const { takeCurrentTerminalDeliveryCredit } =
+      await import('../lib/pane-manager/terminal-delivery-credit')
+    const { writeTerminalOutput } =
+      await import('../lib/pane-manager/pane-terminal-output-scheduler')
+    const parsedCallbacks: (() => void)[] = []
+    const terminal = {
+      write: vi.fn((_data: string, parsed?: () => void) => {
+        if (parsed) {
+          parsedCallbacks.push(parsed)
+        }
+      })
+    }
+    const stream = await getRemoteRuntimeTerminalMultiplexer('windows-test').subscribeTerminal({
+      terminal: 'term-codex',
+      client: { id: 'mac-viewer', type: 'desktop' },
+      callbacks: {
+        onData: (data) => {
+          writeTerminalOutput(terminal, data, {
+            foreground: true,
+            ackCredit: takeCurrentTerminalDeliveryCredit() ?? undefined
+          })
+        },
+        onSnapshot: vi.fn()
+      }
+    })
+
+    callbacks?.onBinary(
+      encodeTerminalStreamFrame({
+        opcode: TerminalStreamOpcode.SnapshotStart,
+        streamId: stream.streamId,
+        seq: 1,
+        payload: encodeTerminalStreamJson({ kind: 'scrollback', seq: 0 })
+      })
+    )
+    callbacks?.onBinary(
+      encodeTerminalStreamFrame({
+        opcode: TerminalStreamOpcode.SnapshotEnd,
+        streamId: stream.streamId,
+        seq: 2,
+        payload: new Uint8Array()
+      })
+    )
+    sendBinary.mockClear()
+
+    const text = '\x1b[?1049h\x1b[?2026h\x1b[2J\x1b[H\x1b[31m-red 🙂 界\x1b[0m\x1b[?2026l'
+    const output = encodeTerminalStreamText(text)
+    callbacks?.onBinary(
+      encodeTerminalStreamFrame({
+        opcode: TerminalStreamOpcode.Output,
+        streamId: stream.streamId,
+        seq: text.length,
+        payload: output
+      })
+    )
+
+    expect(terminal.write).toHaveBeenCalledWith(text, expect.any(Function))
+    expect(parsedCallbacks).toHaveLength(1)
+    expect(sentAckBytes()).toEqual([])
+
+    parsedCallbacks.shift()?.()
+    expect(sentAckBytes()).toEqual([output.byteLength])
+    stream.close()
+  })
+
+  function sentAckBytes(): number[] {
+    return sendBinary.mock.calls.flatMap(([bytes]) => {
+      const frame = decodeTerminalStreamFrame(bytes)
+      if (frame?.opcode !== TerminalStreamOpcode.Ack) {
+        return []
+      }
+      const payload = decodeTerminalStreamJson<{ bytes?: number }>(frame.payload)
+      return typeof payload?.bytes === 'number' ? [payload.bytes] : []
+    })
+  }
+})