fix(terminal): credit output after parse or discard

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

@@ -42,17 +42,20 @@ import {
   navigationTargetsHost,
   resolveRuntimeNavigationTarget
 } from '../../../../shared/runtime-navigation'
+import {
+  TERMINAL_MULTIPLEX_ACK_STREAM_INITIAL_WINDOW_BYTES,
+  TERMINAL_MULTIPLEX_ACK_STREAM_MAX_WINDOW_BYTES,
+  TERMINAL_MULTIPLEX_ACK_TOTAL_INITIAL_WINDOW_BYTES,
+  TERMINAL_MULTIPLEX_ACK_TOTAL_MAX_WINDOW_BYTES,
+  TERMINAL_MULTIPLEX_MAX_STREAMS_PER_CONNECTION,
+  TERMINAL_MULTIPLEX_PENDING_MAX_BYTES,
+  TERMINAL_OUTPUT_BATCH_MAX_BYTES,
+  TERMINAL_STREAM_CHUNK_BYTES
+} from '../../../../shared/terminal-multiplex-flow-control'
+import { drainTerminalMultiplexRoundRobin } from '../terminal-multiplex-round-robin'
 
 const REQUESTED_SNAPSHOT_BYTE_BUDGET = 2 * 1024 * 1024
-const TERMINAL_STREAM_CHUNK_BYTES = 48 * 1024
 const TERMINAL_OUTPUT_FLUSH_MS = 5
-// Why: output batches become binary stream payloads; byte size is the transport cost.
-const TERMINAL_OUTPUT_BATCH_MAX_BYTES = 64 * 1024
-// Why: remote clients can apply output pressure without pausing runtime PTY ingestion.
-const TERMINAL_MULTIPLEX_ACK_STREAM_HIGH_WATER_BYTES = 512 * 1024
-const TERMINAL_MULTIPLEX_ACK_TOTAL_HIGH_WATER_BYTES = 2 * 1024 * 1024
-// Why: pending output becomes binary frames, so cap encoded payload bytes, not UTF-16 code units.
-const TERMINAL_MULTIPLEX_PENDING_MAX_BYTES = 256 * 1024
 const TERMINAL_QUERY_REPLAY_MAX_CHARS = 16 * 1024
 // Why: bound initial subscribe latency; readiness after this deadline triggers an in-stream recovery snapshot.
 const MOBILE_RENDERER_MOUNT_READY_TIMEOUT_MS = 3_000
@@ -102,6 +105,7 @@ type TerminalMultiplexStream = {
   isMobile: boolean
   ackOutput: boolean
   ackInFlightBytes: number
+  ackWindowBytes: number
   supportsDesktopViewportClaims: boolean
   desktopClaimTail: Promise<boolean>
   // Whether THIS stream registered the width driver, so detach won't release a peer stream's floor.
@@ -1598,6 +1602,8 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
       const streams = new Map<number, TerminalMultiplexStream>()
       const pendingPtyWaitControllers = new Map<number, Set<AbortController>>()
       let ackTotalInFlightBytes = 0
+      let ackTotalWindowBytes = TERMINAL_MULTIPLEX_ACK_TOTAL_INITIAL_WINDOW_BYTES
+      let ackFlushCursorStreamId: number | null = null
       let resolveMultiplex = (): void => {}
       const multiplexClosed = new Promise<void>((resolve) => {
         resolveMultiplex = resolve
@@ -1614,10 +1620,21 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
         // Why: a seq-less Output chunk must carry sentinel 0, not the control-frame cursor, or it poisons the client's frame-drop tracker.
         const resolvedSeq =
           typeof seq === 'number' ? seq : opcode === TerminalStreamOpcode.Output ? 0 : cursor++
-        const sent = sendBinary(
-          encodeTerminalStreamFrame({ opcode, streamId, seq: resolvedSeq, payload })
-        )
-        return sent !== false
+        let sent: boolean | void
+        try {
+          sent = sendBinary(
+            encodeTerminalStreamFrame({ opcode, streamId, seq: resolvedSeq, payload })
+          )
+        } catch {
+          closeMultiplex()
+          return false
+        }
+        if (sent === false) {
+          // Why: false means the transport discarded this frame; reconnect is the only available retry boundary with an authoritative snapshot.
+          closeMultiplex()
+          return false
+        }
+        return true
       }
       const sendStreamError = (streamId: number, message: string): void => {
         sendFrame(streamId, TerminalStreamOpcode.Error, encodeTerminalStreamText(message))
@@ -1645,24 +1662,28 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
           return true
         }
         return (
-          stream.ackInFlightBytes + bytes <= TERMINAL_MULTIPLEX_ACK_STREAM_HIGH_WATER_BYTES &&
-          ackTotalInFlightBytes + bytes <= TERMINAL_MULTIPLEX_ACK_TOTAL_HIGH_WATER_BYTES
+          stream.ackInFlightBytes + bytes <= stream.ackWindowBytes &&
+          ackTotalInFlightBytes + bytes <= ackTotalWindowBytes
         )
       }
       const sendAckGatedOutput = (
         stream: TerminalMultiplexStream,
         chunk: TerminalOutputFrameChunk
-      ): void => {
-        sendFrame(
+      ): boolean => {
+        const sent = sendFrame(
           stream.streamId,
           chunk.opcode ?? TerminalStreamOpcode.Output,
           chunk.bytes,
           chunk.seq
         )
+        if (!sent) {
+          return false
+        }
         if (stream.ackOutput) {
           stream.ackInFlightBytes += chunk.bytes.byteLength
           ackTotalInFlightBytes += chunk.bytes.byteLength
         }
+        return true
       }
       const queueOrSendOutput = (
         stream: TerminalMultiplexStream,
@@ -1729,24 +1750,32 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
             stream.streamId,
             error instanceof Error ? error.message : 'Remote terminal recovery snapshot failed.'
           )
+          // Why: retrying the same failed recovery from finally creates an unbounded error loop.
+          detachStream(stream.streamId, true)
         } finally {
           if (streams.get(stream.streamId) === stream) {
             stream.ackRecoverySnapshotInFlight = false
-            flushAckPendingOutput(stream)
+            flushAllAckPendingOutput()
           }
         }
       }
-      const flushAckPendingOutput = (stream: TerminalMultiplexStream): void => {
+      const flushAckPendingOutput = (
+        stream: TerminalMultiplexStream,
+        maxChunks = Number.POSITIVE_INFINITY
+      ): number => {
         if (stream.ackPendingOutputOverflowed) {
           void sendAckRecoverySnapshot(stream)
-          return
+          return 0
         }
         let flushed = 0
         while (
           flushed < stream.ackPendingOutput.length &&
+          flushed < maxChunks &&
           canSendAckGatedOutput(stream, stream.ackPendingOutput[flushed]!.bytes.byteLength)
         ) {
-          sendAckGatedOutput(stream, stream.ackPendingOutput[flushed]!)
+          if (!sendAckGatedOutput(stream, stream.ackPendingOutput[flushed]!)) {
+            return flushed
+          }
           flushed += 1
         }
         if (flushed > 0) {
@@ -1756,17 +1785,38 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
             0
           )
         }
+        return flushed
       }
       const flushAllAckPendingOutput = (): void => {
-        for (const stream of streams.values()) {
-          flushAckPendingOutput(stream)
-        }
+        const ordered = Array.from(streams.values())
+        ackFlushCursorStreamId = drainTerminalMultiplexRoundRobin({
+          streams: ordered,
+          cursorStreamId: ackFlushCursorStreamId,
+          canContinue: () => !closed,
+          drainOne: (stream) => {
+            if (streams.get(stream.streamId) !== stream) {
+              return false
+            }
+            if (flushAckPendingOutput(stream, 1) > 0) {
+              return true
+            }
+            return false
+          }
+        })
       }
       const acknowledgeOutput = (stream: TerminalMultiplexStream, bytes: number): void => {
         if (!stream.ackOutput || bytes <= 0) {
           return
         }
         const acknowledged = Math.min(stream.ackInFlightBytes, bytes)
+        stream.ackWindowBytes = Math.min(
+          TERMINAL_MULTIPLEX_ACK_STREAM_MAX_WINDOW_BYTES,
+          stream.ackWindowBytes + acknowledged
+        )
+        ackTotalWindowBytes = Math.min(
+          TERMINAL_MULTIPLEX_ACK_TOTAL_MAX_WINDOW_BYTES,
+          ackTotalWindowBytes + acknowledged
+        )
         stream.ackInFlightBytes -= acknowledged
         ackTotalInFlightBytes = Math.max(0, ackTotalInFlightBytes - acknowledged)
         flushAllAckPendingOutput()
@@ -2103,6 +2153,15 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
         }
         const request = parsed.data
         detachStream(request.streamId, false)
+        cancelPendingPtyWaits(request.streamId)
+        if (
+          streams.size + pendingPtyWaitControllers.size >=
+          TERMINAL_MULTIPLEX_MAX_STREAMS_PER_CONNECTION
+        ) {
+          sendStreamError(request.streamId, 'terminal_stream_limit_exceeded')
+          emit({ type: 'end', streamId: request.streamId })
+          return
+        }
 
         const isMobile = request.client?.type === 'mobile'
         let leaf: { ptyId: string | null } | null
@@ -2175,6 +2234,7 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
           isMobile,
           ackOutput: request.capabilities?.ackOutput === 1,
           ackInFlightBytes: 0,
+          ackWindowBytes: TERMINAL_MULTIPLEX_ACK_STREAM_INITIAL_WINDOW_BYTES,
           supportsDesktopViewportClaims: request.capabilities?.desktopViewportClaims === 1,
           desktopClaimTail: Promise.resolve(true),
           registeredRemoteDesktopDriver: false,

src/main/runtime/rpc/terminal-multiplex-flow-control.bench.test.ts

@@ -0,0 +1,208 @@
+import { performance } from 'node:perf_hooks'
+import { describe, expect, it } from 'vitest'
+import {
+  TERMINAL_MULTIPLEX_ACK_BATCH_BYTES,
+  TERMINAL_MULTIPLEX_ACK_STREAM_INITIAL_WINDOW_BYTES,
+  TERMINAL_MULTIPLEX_ACK_STREAM_MAX_WINDOW_BYTES,
+  TERMINAL_MULTIPLEX_ACK_TOTAL_INITIAL_WINDOW_BYTES,
+  TERMINAL_MULTIPLEX_ACK_TOTAL_MAX_WINDOW_BYTES,
+  TERMINAL_STREAM_CHUNK_BYTES
+} from '../../../shared/terminal-multiplex-flow-control'
+import { drainTerminalMultiplexRoundRobin } from './terminal-multiplex-round-robin'
+
+const MIB = 1024 * 1024
+const PAYLOAD_BYTES_PER_STREAM = 64 * MIB
+const PARSER_PAYLOAD_BYTES_PER_VIEWER = 4 * MIB
+const benchEnabled = process.env.ORCA_TERMINAL_PERF_BENCH === '1'
+
+type SimulationResult = {
+  throughputMiBPerSecond: number
+  perStreamCompletionMs: number[]
+  maxInFlightBytes: number
+  outputFrames: number
+  ackFrames: number
+  loopIterations: number
+}
+
+type ParserMeasurement = {
+  aggregateMiBPerSecond: number
+  cpuMs: number
+  retainedHeapKiB: number
+  xtermWrites: number
+}
+
+async function measureHeadlessXtermParsing(viewers: number): Promise<ParserMeasurement> {
+  const { Terminal } = await import('@xterm/headless')
+  const sample = '\x1b[?25l\x1b[38;5;45mremote output | build | status | 0123456789\x1b[0m\r\n'
+  const chunk = sample
+    .repeat(Math.ceil(TERMINAL_STREAM_CHUNK_BYTES / sample.length))
+    .slice(0, TERMINAL_STREAM_CHUNK_BYTES)
+  const terminals = Array.from(
+    { length: viewers },
+    () => new Terminal({ cols: 120, rows: 40, scrollback: 5_000 })
+  )
+  const heapBefore = process.memoryUsage().heapUsed
+  const cpuBefore = process.cpuUsage()
+  const startedAt = performance.now()
+  let xtermWrites = 0
+  await Promise.all(
+    terminals.map(async (terminal) => {
+      let remaining = PARSER_PAYLOAD_BYTES_PER_VIEWER
+      while (remaining > 0) {
+        const data = remaining >= chunk.length ? chunk : chunk.slice(0, remaining)
+        xtermWrites += 1
+        await new Promise<void>((resolve) => terminal.write(data, resolve))
+        remaining -= data.length
+      }
+    })
+  )
+  const elapsedMs = performance.now() - startedAt
+  const cpu = process.cpuUsage(cpuBefore)
+  const heapAfter = process.memoryUsage().heapUsed
+  for (const terminal of terminals) {
+    terminal.dispose()
+  }
+  return {
+    aggregateMiBPerSecond: (PARSER_PAYLOAD_BYTES_PER_VIEWER * viewers) / MIB / (elapsedMs / 1_000),
+    cpuMs: (cpu.user + cpu.system) / 1_000,
+    retainedHeapKiB: Math.max(0, heapAfter - heapBefore) / 1_024,
+    xtermWrites
+  }
+}
+
+function simulateParsedCredit(streamCount: number, rttMs: number): SimulationResult {
+  const remaining = Array.from({ length: streamCount }, () => PAYLOAD_BYTES_PER_STREAM)
+  const inFlight = Array.from({ length: streamCount }, () => 0)
+  const windows = Array.from(
+    { length: streamCount },
+    () => TERMINAL_MULTIPLEX_ACK_STREAM_INITIAL_WINDOW_BYTES
+  )
+  const streams = Array.from({ length: streamCount }, (_, streamIndex) => ({
+    streamId: streamIndex + 1,
+    streamIndex
+  }))
+  const perStreamCompletionMs = Array.from({ length: streamCount }, () => 0)
+  const acknowledgements = new Map<number, { streamIndex: number; bytes: number }[]>()
+  let totalInFlight = 0
+  let totalWindow = TERMINAL_MULTIPLEX_ACK_TOTAL_INITIAL_WINDOW_BYTES
+  let maxInFlightBytes = 0
+  let outputFrames = 0
+  let ackFrames = 0
+  let nowMs = 0
+  let loopIterations = 0
+  let sendCursorStreamId: number | null = null
+  while (remaining.some((bytes) => bytes > 0) || totalInFlight > 0) {
+    for (const acknowledgement of acknowledgements.get(nowMs) ?? []) {
+      inFlight[acknowledgement.streamIndex] -= acknowledgement.bytes
+      totalInFlight -= acknowledgement.bytes
+      if (
+        remaining[acknowledgement.streamIndex] === 0 &&
+        inFlight[acknowledgement.streamIndex] === 0
+      ) {
+        perStreamCompletionMs[acknowledgement.streamIndex] = nowMs
+      }
+      windows[acknowledgement.streamIndex] = Math.min(
+        TERMINAL_MULTIPLEX_ACK_STREAM_MAX_WINDOW_BYTES,
+        windows[acknowledgement.streamIndex]! + acknowledgement.bytes
+      )
+      totalWindow = Math.min(
+        TERMINAL_MULTIPLEX_ACK_TOTAL_MAX_WINDOW_BYTES,
+        totalWindow + acknowledgement.bytes
+      )
+      ackFrames += Math.ceil(acknowledgement.bytes / TERMINAL_MULTIPLEX_ACK_BATCH_BYTES)
+    }
+    acknowledgements.delete(nowMs)
+    sendCursorStreamId = drainTerminalMultiplexRoundRobin({
+      streams,
+      cursorStreamId: sendCursorStreamId,
+      drainOne: ({ streamIndex }) => {
+        if (
+          remaining[streamIndex]! <= 0 ||
+          inFlight[streamIndex]! >= windows[streamIndex]! ||
+          totalInFlight >= totalWindow
+        ) {
+          return false
+        }
+        const bytes = Math.min(
+          TERMINAL_STREAM_CHUNK_BYTES,
+          remaining[streamIndex]!,
+          windows[streamIndex]! - inFlight[streamIndex]!,
+          totalWindow - totalInFlight
+        )
+        remaining[streamIndex] -= bytes
+        inFlight[streamIndex] += bytes
+        totalInFlight += bytes
+        outputFrames += 1
+        const due = nowMs + rttMs
+        const dueAcks = acknowledgements.get(due) ?? []
+        const existingAck = dueAcks.find((ack) => ack.streamIndex === streamIndex)
+        if (existingAck) {
+          existingAck.bytes += bytes
+        } else {
+          dueAcks.push({ streamIndex, bytes })
+        }
+        acknowledgements.set(due, dueAcks)
+        return true
+      }
+    })
+    maxInFlightBytes = Math.max(maxInFlightBytes, totalInFlight)
+    nowMs += 1
+    loopIterations += 1
+  }
+  return {
+    throughputMiBPerSecond: (PAYLOAD_BYTES_PER_STREAM * streamCount) / MIB / (nowMs / 1000),
+    perStreamCompletionMs,
+    maxInFlightBytes,
+    outputFrames,
+    ackFrames,
+    loopIterations
+  }
+}
+
+describe('terminal multiplex parsed-credit bounds', () => {
+  it('keeps aggregate memory bounded and streams fair at 100 ms RTT', () => {
+    const result = simulateParsedCredit(8, 100)
+    expect(result.maxInFlightBytes).toBeLessThanOrEqual(
+      TERMINAL_MULTIPLEX_ACK_TOTAL_MAX_WINDOW_BYTES
+    )
+    expect(result.throughputMiBPerSecond / 8).toBeGreaterThan(7)
+    expect(result.ackFrames).toBeLessThan(result.outputFrames / 3)
+    expect(
+      Math.max(...result.perStreamCompletionMs) - Math.min(...result.perStreamCompletionMs)
+    ).toBeLessThan(200)
+  })
+})
+
+describe.skipIf(!benchEnabled)('terminal multiplex parsed-credit benchmark', () => {
+  it('reports RTT, fairness, protocol allocations, and measured xterm parser cost', async () => {
+    const parserMeasurements = new Map<number, ParserMeasurement>()
+    for (const viewers of [1, 4, 8]) {
+      parserMeasurements.set(viewers, await measureHeadlessXtermParsing(viewers))
+    }
+    const rows = [1, 20, 100].flatMap((rttMs) =>
+      [1, 4, 8].map((viewers) => {
+        const startedAt = performance.now()
+        const result = simulateParsedCredit(viewers, rttMs)
+        const parser = parserMeasurements.get(viewers)!
+        return {
+          rttMs,
+          viewers,
+          aggregateMiBps: Number(result.throughputMiBPerSecond.toFixed(1)),
+          perViewerMiBps: Number((result.throughputMiBPerSecond / viewers).toFixed(1)),
+          schedulerCpuMs: Number((performance.now() - startedAt).toFixed(2)),
+          protocolFrameAllocations: result.outputFrames + result.ackFrames,
+          loopIterations: result.loopIterations,
+          maxInFlightKiB: result.maxInFlightBytes / 1024,
+          completionSpreadMs:
+            Math.max(...result.perStreamCompletionMs) - Math.min(...result.perStreamCompletionMs),
+          measuredParserMiBps: Number(parser.aggregateMiBPerSecond.toFixed(1)),
+          parserCpuMs: Number(parser.cpuMs.toFixed(1)),
+          parserRetainedHeapKiB: Number(parser.retainedHeapKiB.toFixed(0)),
+          xtermWriteAllocations: parser.xtermWrites
+        }
+      })
+    )
+    // eslint-disable-next-line no-console -- opt-in benchmark evidence
+    console.table(rows)
+  })
+})

src/main/runtime/rpc/terminal-multiplex-round-robin.test.ts

@@ -0,0 +1,67 @@
+import { describe, expect, it } from 'vitest'
+import { drainTerminalMultiplexRoundRobin } from './terminal-multiplex-round-robin'
+
+describe('terminal multiplex round-robin drain', () => {
+  it('admits a later interactive stream before older bulk queues refill the window', () => {
+    const streams = Array.from({ length: 8 }, (_, index) => ({
+      streamId: index + 1,
+      pendingChunks: index === 7 ? 1 : 8
+    }))
+    const order: number[] = []
+    let remainingSlots = 8
+
+    const cursor = drainTerminalMultiplexRoundRobin({
+      streams,
+      cursorStreamId: null,
+      canContinue: () => remainingSlots > 0,
+      drainOne: (stream) => {
+        if (stream.pendingChunks === 0) {
+          return false
+        }
+        stream.pendingChunks -= 1
+        remainingSlots -= 1
+        order.push(stream.streamId)
+        return true
+      }
+    })
+
+    expect(order).toEqual([1, 2, 3, 4, 5, 6, 7, 8])
+    expect(cursor).toBe(8)
+  })
+
+  it('resumes after the previous sender on the next release', () => {
+    const streams = [1, 2, 3].map((streamId) => ({ streamId, pendingChunks: 2 }))
+    let slots = 2
+    const firstOrder: number[] = []
+    const cursor = drainTerminalMultiplexRoundRobin({
+      streams,
+      cursorStreamId: null,
+      canContinue: () => slots > 0,
+      drainOne: (stream) => {
+        stream.pendingChunks -= 1
+        slots -= 1
+        firstOrder.push(stream.streamId)
+        return true
+      }
+    })
+    slots = 2
+    const secondOrder: number[] = []
+    drainTerminalMultiplexRoundRobin({
+      streams,
+      cursorStreamId: cursor,
+      canContinue: () => slots > 0,
+      drainOne: (stream) => {
+        if (stream.pendingChunks === 0) {
+          return false
+        }
+        stream.pendingChunks -= 1
+        slots -= 1
+        secondOrder.push(stream.streamId)
+        return true
+      }
+    })
+
+    expect(firstOrder).toEqual([1, 2])
+    expect(secondOrder).toEqual([3, 1])
+  })
+})

src/main/runtime/rpc/terminal-multiplex-round-robin.ts

@@ -0,0 +1,41 @@
+type TerminalMultiplexDrainStream = { streamId: number }
+
+export function drainTerminalMultiplexRoundRobin<T extends TerminalMultiplexDrainStream>(args: {
+  streams: readonly T[]
+  cursorStreamId: number | null
+  drainOne: (stream: T) => boolean
+  canContinue?: () => boolean
+}): number | null {
+  const { streams, drainOne } = args
+  if (streams.length === 0) {
+    return null
+  }
+  const canContinue = args.canContinue ?? (() => true)
+  let cursorStreamId = args.cursorStreamId
+  let startIndex = getStartIndex(streams, cursorStreamId)
+  while (canContinue()) {
+    let progressed = false
+    for (let offset = 0; offset < streams.length && canContinue(); offset += 1) {
+      const stream = streams[(startIndex + offset) % streams.length]!
+      if (drainOne(stream)) {
+        cursorStreamId = stream.streamId
+        progressed = true
+      }
+    }
+    if (!progressed) {
+      break
+    }
+    startIndex = getStartIndex(streams, cursorStreamId)
+  }
+  return cursorStreamId
+}
+
+function getStartIndex<T extends TerminalMultiplexDrainStream>(
+  streams: readonly T[],
+  cursorStreamId: number | null
+): number {
+  if (cursorStreamId === null) {
+    return 0
+  }
+  return (streams.findIndex((stream) => stream.streamId === cursorStreamId) + 1) % streams.length
+}

src/main/runtime/rpc/terminal-multiplex.test.ts

@@ -43,7 +43,8 @@ function makeRequest(method: string, params?: unknown): RpcRequest {
 
 function startDesktopMultiplexSubscribe(
   overrides: Partial<OrcaRuntimeService> = {},
-  trace?: string[]
+  trace?: string[],
+  sendBinaryOverride?: (bytes: Uint8Array<ArrayBufferLike>) => boolean | void
 ) {
   const messages: string[] = []
   const binaryFrames: Uint8Array<ArrayBufferLike>[] = []
@@ -87,6 +88,10 @@ function startDesktopMultiplexSubscribe(
     {
       connectionId: 'conn-desktop-first-paint',
       sendBinary: (bytes) => {
+        const sent = sendBinaryOverride?.(bytes)
+        if (sent === false) {
+          return false
+        }
         binaryFrames.push(bytes)
         const opcode = decodeTerminalStreamFrame(bytes)?.opcode
         if (
@@ -96,10 +101,15 @@ function startDesktopMultiplexSubscribe(
         ) {
           trace?.push('snapshot')
         }
+        return sent
       },
       registerBinaryStreamHandler: (streamId, handler) => {
         handlers.set(streamId, handler)
-        return () => handlers.delete(streamId)
+        return () => {
+          if (handlers.get(streamId) === handler) {
+            handlers.delete(streamId)
+          }
+        }
       }
     }
   )
@@ -128,6 +138,65 @@ function sendDesktopMultiplexSubscribe(
 }
 
 describe('terminal multiplex RPC', () => {
+  it.each(['refuses', 'throws'] as const)(
+    'closes without reserving ACK debt when the transport %s an output frame',
+    async (failureMode) => {
+      let dataListener:
+        | ((data: string, meta?: { seq?: number; rawLength?: number }) => void)
+        | null = null
+      let rejectOutput = false
+      const unsubscribeData = vi.fn()
+      const harness = startDesktopMultiplexSubscribe(
+        {
+          subscribeToTerminalData: vi.fn((_ptyId, listener) => {
+            dataListener = listener
+            return unsubscribeData
+          })
+        },
+        undefined,
+        (bytes) => {
+          const frame = decodeTerminalStreamFrame(bytes)
+          if (!rejectOutput || frame?.opcode !== TerminalStreamOpcode.Output) {
+            return true
+          }
+          if (failureMode === 'throws') {
+            throw new Error('socket closed')
+          }
+          return false
+        }
+      )
+
+      await vi.waitFor(() =>
+        expect(harness.messages.some((msg) => JSON.parse(msg).result?.type === 'ready')).toBe(true)
+      )
+      sendDesktopMultiplexSubscribe(harness.handlers)
+      await vi.waitFor(() =>
+        expect(harness.messages.some((msg) => JSON.parse(msg).result?.type === 'subscribed')).toBe(
+          true
+        )
+      )
+      await vi.waitFor(() => expect(dataListener).not.toBeNull())
+      harness.binaryFrames.splice(0)
+      rejectOutput = true
+
+      const output = 'x'.repeat(64 * 1024)
+      const deliverData = dataListener as unknown as (
+        data: string,
+        meta?: { seq?: number; rawLength?: number }
+      ) => void
+      deliverData(output, { seq: output.length, rawLength: output.length })
+
+      await vi.waitFor(() => expect(unsubscribeData).toHaveBeenCalledOnce())
+      await harness.dispatchPromise
+      expect(
+        harness.binaryFrames
+          .map((bytes) => decodeTerminalStreamFrame(bytes))
+          .filter((frame) => frame?.opcode === TerminalStreamOpcode.Output)
+      ).toEqual([])
+      expect(harness.handlers.size).toBe(0)
+    }
+  )
+
   it('multiplexes terminal streams and routes desktop resize to the source PTY', async () => {
     vi.useFakeTimers()
     try {
@@ -1197,7 +1266,7 @@ describe('terminal multiplex RPC', () => {
     await dispatchPromise
   })
 
-  it('releases shared ACK budget to other stalled multiplex streams', async () => {
+  it('round-robins released ACK budget to a later interactive stream', async () => {
     const messages: string[] = []
     const binaryFrames: Uint8Array<ArrayBufferLike>[] = []
     const handlers = new Map<
@@ -1272,7 +1341,7 @@ describe('terminal multiplex RPC', () => {
       expect(messages.some((msg) => JSON.parse(msg).result?.type === 'ready')).toBe(true)
     )
 
-    const streamIds = [21, 22, 23, 24, 25, 26]
+    const streamIds = [21, 22, 23, 24, 25, 26, 27, 28]
     for (const streamId of streamIds) {
       handlers.get(0)?.(
         decodeTerminalStreamFrame(
@@ -1302,22 +1371,33 @@ describe('terminal multiplex RPC', () => {
     await vi.waitFor(() => expect(dataListeners.size).toBe(streamIds.length))
     binaryFrames.splice(0)
 
-    const fillerOutput = 'f'.repeat(480 * 1024)
+    const fillerOutput = 'f'.repeat(512 * 1024)
     for (let index = 1; index <= 4; index += 1) {
       dataListeners.get(`pty-${index}`)?.(fillerOutput, {
         seq: fillerOutput.length,
         rawLength: fillerOutput.length
       })
     }
-    const stalledOutput = 's'.repeat(700 * 1024)
-    dataListeners.get('pty-5')?.(stalledOutput, {
-      seq: stalledOutput.length,
-      rawLength: stalledOutput.length
-    })
-    dataListeners.get('pty-6')?.(stalledOutput, {
-      seq: stalledOutput.length,
-      rawLength: stalledOutput.length
+    const queuedFillerOutput = 'q'.repeat(256 * 1024)
+    for (let index = 1; index <= 4; index += 1) {
+      dataListeners.get(`pty-${index}`)?.(queuedFillerOutput, {
+        seq: fillerOutput.length + queuedFillerOutput.length,
+        rawLength: queuedFillerOutput.length
+      })
+    }
+    const stalledOutput = 's'.repeat(256 * 1024)
+    for (let index = 5; index <= 7; index += 1) {
+      dataListeners.get(`pty-${index}`)?.(stalledOutput, {
+        seq: stalledOutput.length,
+        rawLength: stalledOutput.length
+      })
+    }
+    const interactiveOutput = 'interactive-output\r\n'
+    dataListeners.get('pty-8')?.(interactiveOutput, {
+      seq: interactiveOutput.length,
+      rawLength: interactiveOutput.length
     })
+    await new Promise((resolve) => setTimeout(resolve, 10))
 
     const initialOutputFrames = binaryFrames
       .map((frame) => decodeTerminalStreamFrame(frame))
@@ -1337,25 +1417,27 @@ describe('terminal multiplex RPC', () => {
       0
     )
     expect(initialBytes).toBeLessThanOrEqual(2 * 1024 * 1024)
-    expect(initialBytesByStream.get(21)).toBe(480 * 1024)
-    expect(initialBytesByStream.get(22)).toBe(480 * 1024)
-    expect(initialBytesByStream.get(23)).toBe(480 * 1024)
-    expect(initialBytesByStream.get(24)).toBe(480 * 1024)
-    expect(initialBytesByStream.get(25)).toBeGreaterThan(0)
+    expect(initialBytesByStream.get(21)).toBe(512 * 1024)
+    expect(initialBytesByStream.get(22)).toBe(512 * 1024)
+    expect(initialBytesByStream.get(23)).toBe(512 * 1024)
+    expect(initialBytesByStream.get(24)).toBe(512 * 1024)
+    expect(initialBytesByStream.get(25) ?? 0).toBe(0)
     expect(initialBytesByStream.get(26) ?? 0).toBe(0)
+    expect(initialBytesByStream.get(27) ?? 0).toBe(0)
+    expect(initialBytesByStream.get(28) ?? 0).toBe(0)
 
-    handlers.get(26)?.(
+    handlers.get(28)?.(
       decodeTerminalStreamFrame(
         encodeTerminalStreamFrame({
           opcode: TerminalStreamOpcode.Input,
-          streamId: 26,
+          streamId: 28,
           seq: 200,
           payload: encodeTerminalStreamText('remote-still-interactive\r')
         })
       )!
     )
     await vi.waitFor(() =>
-      expect(runtime.sendTerminal).toHaveBeenCalledWith('terminal-6', {
+      expect(runtime.sendTerminal).toHaveBeenCalledWith('terminal-8', {
         text: 'remote-still-interactive\r',
         enter: false,
         interrupt: false
@@ -1379,25 +1461,17 @@ describe('terminal multiplex RPC', () => {
         binaryFrames
           .slice(frameCountBeforeAck)
           .map((frame) => decodeTerminalStreamFrame(frame))
-          .some((frame) => {
-            if (frame?.streamId !== 25 || frame.opcode !== TerminalStreamOpcode.SnapshotStart) {
-              return false
-            }
-            const payload = decodeTerminalStreamJson<{ reason?: string }>(frame.payload)
-            return payload?.reason === 'ack-pending-overflow'
-          })
+          .some(
+            (frame) =>
+              frame?.streamId === 28 &&
+              frame.opcode === TerminalStreamOpcode.Output &&
+              decodeTerminalStreamText(frame.payload) === interactiveOutput
+          )
       ).toBe(true)
     )
     const framesAfterAck = binaryFrames
       .slice(frameCountBeforeAck)
       .map((frame) => decodeTerminalStreamFrame(frame))
-    const snapshotStartIndex = framesAfterAck.findIndex((frame) => {
-      if (frame?.streamId !== 25 || frame.opcode !== TerminalStreamOpcode.SnapshotStart) {
-        return false
-      }
-      const payload = decodeTerminalStreamJson<{ reason?: string }>(frame.payload)
-      return payload?.reason === 'ack-pending-overflow'
-    })
     const outputFramesAfterAck = framesAfterAck.filter(
       (frame) => frame?.opcode === TerminalStreamOpcode.Output
     )
@@ -1411,17 +1485,14 @@ describe('terminal multiplex RPC', () => {
         (bytesAfterAckByStream.get(frame.streamId) ?? 0) + frame.payload.byteLength
       )
     }
-    expect(snapshotStartIndex).toBeGreaterThanOrEqual(0)
-    expect(
-      framesAfterAck
-        .filter((frame) => frame?.streamId === 25 && frame.opcode === TerminalStreamOpcode.Output)
-        .every((frame) => framesAfterAck.indexOf(frame) > snapshotStartIndex)
-    ).toBe(true)
     expect(bytesAfterAckByStream.get(25) ?? 0).toBeGreaterThan(0)
-    expect(bytesAfterAckByStream.get(21) ?? 0).toBe(0)
+    expect(bytesAfterAckByStream.get(26) ?? 0).toBeGreaterThan(0)
+    expect(bytesAfterAckByStream.get(27) ?? 0).toBeGreaterThan(0)
+    expect(bytesAfterAckByStream.get(28) ?? 0).toBe(interactiveOutput.length)
+    expect(bytesAfterAckByStream.get(21) ?? 0).toBeGreaterThan(0)
     expect(
       outputFramesAfterAck.reduce((total, frame) => total + (frame?.payload.byteLength ?? 0), 0)
-    ).toBeLessThanOrEqual(initialBytesByStream.get(21) ?? 0)
+    ).toBeLessThanOrEqual((initialBytesByStream.get(21) ?? 0) * 2)
 
     runtime.cleanupSubscription('terminal-multiplex:conn-ack-shared-budget')
     await dispatchPromise
@@ -1605,6 +1676,72 @@ describe('terminal multiplex RPC', () => {
     await dispatchPromise
   })
 
+  it('ends a stream when ACK overflow recovery serialization fails', async () => {
+    const dataListenerRef: {
+      current?: (data: string, meta?: { seq?: number; rawLength?: number }) => void
+    } = {}
+    const serializeTerminalBuffer = vi
+      .fn()
+      .mockResolvedValueOnce({ data: 'initial snapshot', cols: 120, rows: 40 })
+      .mockRejectedValue(new Error('snapshot unavailable'))
+    const harness = startDesktopMultiplexSubscribe({
+      serializeTerminalBuffer,
+      subscribeToTerminalData: vi.fn(
+        (
+          _: string,
+          listener: (data: string, meta?: { seq?: number; rawLength?: number }) => void
+        ) => {
+          dataListenerRef.current = listener
+          return vi.fn()
+        }
+      )
+    })
+
+    await vi.waitFor(() =>
+      expect(harness.messages.some((message) => JSON.parse(message).result?.type === 'ready')).toBe(
+        true
+      )
+    )
+    sendDesktopMultiplexSubscribe(harness.handlers)
+    await vi.waitFor(() =>
+      expect(
+        harness.messages.some((message) => JSON.parse(message).result?.type === 'subscribed')
+      ).toBe(true)
+    )
+    harness.binaryFrames.splice(0)
+
+    const output = 'x'.repeat(3 * 1024 * 1024)
+    dataListenerRef.current?.(output, { seq: output.length, rawLength: output.length })
+    const inFlightBytes = harness.binaryFrames
+      .map((bytes) => decodeTerminalStreamFrame(bytes))
+      .filter((frame) => frame?.opcode === TerminalStreamOpcode.Output)
+      .reduce((total, frame) => total + (frame?.payload.byteLength ?? 0), 0)
+    harness.handlers.get(7)?.(
+      decodeTerminalStreamFrame(
+        encodeTerminalStreamFrame({
+          opcode: TerminalStreamOpcode.Ack,
+          streamId: 7,
+          seq: 2,
+          payload: encodeTerminalStreamJson({ bytes: inFlightBytes })
+        })
+      )!
+    )
+
+    await vi.waitFor(() => {
+      const eventTypes = harness.messages.map((message) => JSON.parse(message).result?.type)
+      expect(eventTypes).toContain('error')
+      expect(eventTypes).toContain('end')
+    })
+    expect(harness.handlers.has(7)).toBe(false)
+    expect(serializeTerminalBuffer).toHaveBeenCalledTimes(2)
+    await Promise.resolve()
+    await Promise.resolve()
+    expect(serializeTerminalBuffer).toHaveBeenCalledTimes(2)
+
+    harness.cleanups.get('terminal-multiplex:conn-desktop-first-paint')?.()
+    await harness.dispatchPromise
+  })
+
   it('trims recovery-covered ACK pending output instead of replaying it', async () => {
     const messages: string[] = []
     const binaryFrames: Uint8Array<ArrayBufferLike>[] = []
@@ -2881,6 +3018,81 @@ describe('terminal multiplex RPC', () => {
     await harness.dispatchPromise
   })
 
+  it('cancels an older pending PTY wait when the same multiplex slot resubscribes', async () => {
+    const waitSignals: AbortSignal[] = []
+    const waitForLeafPtyId = vi.fn(
+      (_handle: string, _timeoutMs?: number, signal?: AbortSignal) =>
+        new Promise<string>((_resolve, reject) => {
+          if (signal) {
+            waitSignals.push(signal)
+          }
+          signal?.addEventListener('abort', () => reject(new Error('request_aborted')), {
+            once: true
+          })
+        })
+    )
+    const harness = startDesktopMultiplexSubscribe({
+      resolveLiveLeafForHandle: vi.fn().mockReturnValue({ ptyId: null }),
+      waitForLeafPtyId
+    })
+    await vi.waitFor(() =>
+      expect(harness.messages.some((message) => JSON.parse(message).result?.type === 'ready')).toBe(
+        true
+      )
+    )
+
+    sendDesktopMultiplexSubscribe(harness.handlers)
+    await vi.waitFor(() => expect(waitSignals).toHaveLength(1))
+    sendDesktopMultiplexSubscribe(harness.handlers)
+    await vi.waitFor(() => expect(waitSignals).toHaveLength(2))
+
+    expect(waitSignals[0]?.aborted).toBe(true)
+    expect(waitSignals[1]?.aborted).toBe(false)
+    harness.cleanups.get('terminal-multiplex:conn-desktop-first-paint')?.()
+    await vi.waitFor(() => expect(waitSignals[1]?.aborted).toBe(true))
+    await harness.dispatchPromise
+  })
+
+  it('caps multiplex stream slots so aggregate pending output stays bounded', async () => {
+    const harness = startDesktopMultiplexSubscribe()
+    await vi.waitFor(() =>
+      expect(harness.messages.some((message) => JSON.parse(message).result?.type === 'ready')).toBe(
+        true
+      )
+    )
+    for (let streamId = 1; streamId <= 33; streamId += 1) {
+      harness.handlers.get(0)?.(
+        decodeTerminalStreamFrame(
+          encodeTerminalStreamFrame({
+            opcode: TerminalStreamOpcode.Subscribe,
+            streamId: 0,
+            seq: streamId,
+            payload: encodeTerminalStreamJson({
+              streamId,
+              terminal: 'terminal-1',
+              client: { id: 'desktop-1', type: 'desktop' },
+              capabilities: { ackOutput: 1 }
+            })
+          })
+        )!
+      )
+    }
+
+    await vi.waitFor(() => {
+      const results = harness.messages.map((message) => JSON.parse(message).result)
+      expect(results.filter((result) => result?.type === 'subscribed')).toHaveLength(32)
+      expect(results).toContainEqual({
+        type: 'error',
+        streamId: 33,
+        message: 'terminal_stream_limit_exceeded'
+      })
+      expect(results).toContainEqual({ type: 'end', streamId: 33 })
+    })
+
+    harness.cleanups.get('terminal-multiplex:conn-desktop-first-paint')?.()
+    await harness.dispatchPromise
+  })
+
   it("still reports no_connected_pty when a desktop multiplex subscriber's PTY never appears", async () => {
     const runtime = stubRuntime({
       resolveLiveLeafForHandle: vi.fn().mockReturnValue({ ptyId: null }),

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

@@ -13560,6 +13560,8 @@ describe('connectPanePty', () => {
 
   it('requests snapshot recovery for one oversized live frame deferred by replay', async () => {
     const { connectPanePty } = await import('./pty-connection')
+    const { deliverTerminalDataWithDeferredCredit } =
+      await import('@/lib/pane-manager/terminal-delivery-credit')
     const transport = createMockTransport('pty-large-live')
     const callbacksRef: {
       replay: ((data: string) => void) | null
@@ -13584,7 +13586,11 @@ describe('connectPanePty', () => {
     callbacksRef.replay?.('authoritative replay')
     await flushAsyncTicks(8)
     const oversizedLiveFrame = 'L'.repeat(512 * 1024 + 1)
-    callbacksRef.data?.(oversizedLiveFrame)
+    const acknowledgeDroppedFrame = vi.fn()
+    deliverTerminalDataWithDeferredCredit(acknowledgeDroppedFrame, () => {
+      callbacksRef.data?.(oversizedLiveFrame)
+    })
+    expect(acknowledgeDroppedFrame).not.toHaveBeenCalled()
     while (parseCallbacks.length > 0) {
       parseCallbacks.shift()?.()
       await flushAsyncTicks(4)
@@ -13595,6 +13601,7 @@ describe('connectPanePty', () => {
       scrollbackRows: 5000
     })
     expect(writes.some((write) => write.startsWith('L'))).toBe(false)
+    expect(acknowledgeDroppedFrame).toHaveBeenCalledOnce()
     binding.dispose()
   })
 

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

@@ -5823,7 +5823,7 @@ export function connectPanePty(
       writeTerminalOutput(pane.terminal, data, {
         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.
+        // Why: every scheduler write claims one child so a split delivery is credited only after all children parse or discard.
         ackCredit: takeCurrentTerminalDeliveryCredit() ?? undefined,
         onBackgroundBacklogDropped: markHiddenOutputRestoreNeeded,
         latencySensitive:

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

@@ -91,7 +91,7 @@ describe('terminal-pty-ack-gate parse-deferred crediting', () => {
     expect(ackDataMock).toHaveBeenCalledTimes(1)
   })
 
-  it('hands out the credit only once per delivery', async () => {
+  it('waits for every scheduler write produced by one delivery', async () => {
     const { deliverPtyDataWithDeferredAck, takeCurrentPtyDeliveryAckCredit } = await loadAckGate()
     let first: (() => void) | null = null
     let second: (() => void) | null = null
@@ -102,7 +102,11 @@ describe('terminal-pty-ack-gate parse-deferred crediting', () => {
     })
 
     expect(first).not.toBeNull()
-    expect(second).toBeNull()
+    expect(second).not.toBeNull()
+    first!()
+    expect(ackDataMock).not.toHaveBeenCalled()
+    second!()
+    expect(ackDataMock).toHaveBeenCalledWith('pty-a', 5, 5)
   })
 
   it('returns null outside a delivery', async () => {

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

@@ -106,9 +106,6 @@ export function deliverPtyDataWithDeferredAck(
   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 {
   return takeCurrentTerminalDeliveryCredit()
 }

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

@@ -41,4 +41,26 @@ describe('terminal delivery credit', () => {
     expect(complete).toHaveBeenCalledOnce()
     expect(claimLater!()).toBeNull()
   })
+  it('settles only after every scheduler write claimed by one delivery completes', async () => {
+    const { deliverTerminalDataWithDeferredCredit, takeCurrentTerminalDeliveryCredit } =
+      await import('./terminal-delivery-credit')
+    const complete = vi.fn()
+    let first: (() => void) | null = null
+    let second: (() => void) | null = null
+
+    deliverTerminalDataWithDeferredCredit(complete, () => {
+      first = takeCurrentTerminalDeliveryCredit()
+      second = takeCurrentTerminalDeliveryCredit()
+    })
+
+    expect(first).not.toBeNull()
+    expect(second).not.toBeNull()
+    first!()
+    expect(complete).not.toHaveBeenCalled()
+    second!()
+    expect(complete).toHaveBeenCalledOnce()
+    first!()
+    second!()
+    expect(complete).toHaveBeenCalledOnce()
+  })
 })

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

@@ -1,30 +1,31 @@
 type TerminalDeliveryCredit = {
   complete: () => void
-  claimed: boolean
-  credited: boolean
+  open: boolean
+  pendingClaims: number
+  completed: boolean
 }
 
-// Why: consumers must claim during deliver(); after it returns this synchronous slot is restored and unclaimed credit settles.
+// Why: claims are synchronous; nesting restores an outer delivery after an inner callback returns.
 let currentDeliveryCredit: TerminalDeliveryCredit | null = null
 
 function completeTerminalDeliveryCredit(credit: TerminalDeliveryCredit): void {
-  // Why: queue splitting and discard paths can both settle one delivery.
-  if (credit.credited) {
+  if (credit.completed || credit.open || credit.pendingClaims > 0) {
     return
   }
-  credit.credited = true
+  credit.completed = true
   credit.complete()
 }
 
-/** Defers producer credit until the output scheduler consumes or discards the delivery. */
+/** Defers producer credit until every output scheduler consumer parses or discards it. */
 export function deliverTerminalDataWithDeferredCredit(
   complete: () => void,
   deliver: () => void
 ): void {
   const credit: TerminalDeliveryCredit = {
     complete,
-    claimed: false,
-    credited: false
+    open: true,
+    pendingClaims: 0,
+    completed: false
   }
   const previousCredit = currentDeliveryCredit
   currentDeliveryCredit = credit
@@ -32,18 +33,24 @@ export function deliverTerminalDataWithDeferredCredit(
     deliver()
   } finally {
     currentDeliveryCredit = previousCredit
-    if (!credit.claimed) {
-      completeTerminalDeliveryCredit(credit)
-    }
+    credit.open = false
+    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) {
+  if (!credit || !credit.open) {
     return null
   }
-  credit.claimed = true
-  return () => completeTerminalDeliveryCredit(credit)
+  credit.pendingClaims += 1
+  let settled = false
+  return () => {
+    if (settled) {
+      return
+    }
+    settled = true
+    credit.pendingClaims -= 1
+    completeTerminalDeliveryCredit(credit)
+  }
 }

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

@@ -14,6 +14,10 @@ import { e2eConfig } from '@/lib/e2e-config'
 import { deliverTerminalDataWithDeferredCredit } from '@/lib/pane-manager/terminal-delivery-credit'
 import { unwrapRuntimeRpcResult } from './runtime-rpc-client'
 import { getRuntimeEnvironmentRevision } from './runtime-environment-revision'
+import {
+  TERMINAL_MULTIPLEX_ACK_BATCH_BYTES,
+  TERMINAL_MULTIPLEX_ACK_FLUSH_MS
+} from '../../../shared/terminal-multiplex-flow-control'
 
 type RuntimeEnvironmentSubscriptionHandle = {
   unsubscribe: () => void
@@ -78,6 +82,8 @@ type RemoteRuntimeMultiplexedTerminalState = {
   subscriptionRequested: boolean
   acknowledgeOutput: boolean
   heldAckBytes: number
+  pendingAckBytes: number
+  ackFlushTimer: ReturnType<typeof setTimeout> | null
   snapshotChunks: Uint8Array<ArrayBufferLike>[]
   snapshotBytes: number
   snapshotOverflowed: boolean
@@ -247,6 +253,8 @@ class RemoteRuntimeTerminalMultiplexer {
       subscriptionRequested: false,
       acknowledgeOutput: args.client.type === 'desktop',
       heldAckBytes: 0,
+      pendingAckBytes: 0,
+      ackFlushTimer: null,
       snapshotChunks: [],
       snapshotBytes: 0,
       snapshotOverflowed: false,
@@ -292,6 +300,7 @@ class RemoteRuntimeTerminalMultiplexer {
       serializeBuffer: (opts) => this.requestSnapshot(state, opts),
       close: () => {
         if (this.streams.get(streamId) === state) {
+          discardOutputAcknowledgements(state)
           this.sendFrame(streamId, TerminalStreamOpcode.Unsubscribe)
           clearResyncTimer(state)
           rejectPendingSnapshotRequest(state, 'Remote terminal stream closed.')
@@ -429,6 +438,7 @@ class RemoteRuntimeTerminalMultiplexer {
       return
     }
     if (event.type === 'end') {
+      discardOutputAcknowledgements(stream)
       clearSnapshot(stream)
       clearResyncTimer(stream)
       rejectPendingSnapshotRequest(stream, 'Remote terminal stream ended.')
@@ -483,10 +493,19 @@ class RemoteRuntimeTerminalMultiplexer {
     }
     const frame = decodeTerminalStreamFrame(bytes)
     if (!frame) {
+      // Why: malformed framing cannot be credited safely; closing makes the server release every stream window.
+      this.failConnection(new Error('Remote terminal stream received a malformed frame.'))
       return
     }
     const stream = this.streams.get(frame.streamId)
     if (!stream) {
+      if (
+        frame.opcode === TerminalStreamOpcode.Output ||
+        frame.opcode === TerminalStreamOpcode.OutputSpan
+      ) {
+        // Why: the renderer already disposed this stream; unsubscribe releases server credit that cannot reach a parser.
+        this.sendFrame(frame.streamId, TerminalStreamOpcode.Unsubscribe)
+      }
       return
     }
     if (
@@ -557,13 +576,19 @@ class RemoteRuntimeTerminalMultiplexer {
         deliverOutput()
         return
       }
-      deliverTerminalDataWithDeferredCredit(() => {
-        if (shouldHoldE2eRemoteTerminalAck(stream.terminal)) {
-          stream.heldAckBytes += frame.payload.byteLength
-        } else {
-          this.acknowledgeOutput(stream, frame.payload.byteLength)
-        }
-      }, deliverOutput)
+      try {
+        deliverTerminalDataWithDeferredCredit(() => {
+          if (shouldHoldE2eRemoteTerminalAck(stream.terminal)) {
+            stream.heldAckBytes += frame.payload.byteLength
+          } else {
+            this.queueOutputAcknowledgement(stream, frame.payload.byteLength)
+          }
+        }, deliverOutput)
+      } catch (error) {
+        this.failConnection(
+          error instanceof Error ? error : new Error('Remote terminal output delivery failed.')
+        )
+      }
       return
     }
     if (frame.opcode === TerminalStreamOpcode.SnapshotStart) {
@@ -856,6 +881,33 @@ class RemoteRuntimeTerminalMultiplexer {
     )
   }
 
+  private queueOutputAcknowledgement(
+    stream: RemoteRuntimeMultiplexedTerminalState,
+    bytes: number
+  ): boolean {
+    if (this.streams.get(stream.streamId) !== stream) {
+      return true
+    }
+    stream.pendingAckBytes += bytes
+    if (stream.pendingAckBytes >= TERMINAL_MULTIPLEX_ACK_BATCH_BYTES) {
+      return this.flushOutputAcknowledgement(stream)
+    }
+    if (stream.ackFlushTimer === null) {
+      stream.ackFlushTimer = setTimeout(() => {
+        stream.ackFlushTimer = null
+        this.flushOutputAcknowledgement(stream)
+      }, TERMINAL_MULTIPLEX_ACK_FLUSH_MS)
+    }
+    return true
+  }
+
+  private flushOutputAcknowledgement(stream: RemoteRuntimeMultiplexedTerminalState): boolean {
+    clearAckFlushTimer(stream)
+    const bytes = stream.pendingAckBytes
+    stream.pendingAckBytes = 0
+    return bytes <= 0 || this.acknowledgeOutput(stream, bytes)
+  }
+
   getStreamsForE2e(): Iterable<RemoteRuntimeMultiplexedTerminalState> {
     return this.streams.values()
   }
@@ -868,7 +920,7 @@ class RemoteRuntimeTerminalMultiplexer {
       }
       const bytes = stream.heldAckBytes
       stream.heldAckBytes = 0
-      if (this.acknowledgeOutput(stream, bytes)) {
+      if (this.queueOutputAcknowledgement(stream, bytes)) {
         released += bytes
       }
     }
@@ -883,8 +935,15 @@ class RemoteRuntimeTerminalMultiplexer {
     if (!this.matchesCurrentEnvironmentRevision() || !this.ready || !this.subscription) {
       return false
     }
-    this.subscription.sendBinary(encodeTerminalStreamFrame({ opcode, streamId, seq: 0, payload }))
-    return true
+    try {
+      this.subscription.sendBinary(encodeTerminalStreamFrame({ opcode, streamId, seq: 0, payload }))
+      return true
+    } catch (error) {
+      this.handleClose(
+        error instanceof Error ? error.message : 'Remote terminal transport write failed.'
+      )
+      return false
+    }
   }
 
   private resolveReadyIfConnected(): void {
@@ -923,6 +982,7 @@ class RemoteRuntimeTerminalMultiplexer {
     // Why: close callbacks may resubscribe synchronously; release first so every replacement shares the new environment multiplexer.
     this.releaseIfCurrent(this.environmentId, this)
     for (const stream of streams) {
+      discardOutputAcknowledgements(stream)
       clearSnapshot(stream)
       clearResyncTimer(stream)
       rejectPendingSnapshotRequest(stream, message ?? 'Remote runtime connection closed.')
@@ -1006,6 +1066,19 @@ function clearSnapshot(stream: RemoteRuntimeMultiplexedTerminalState): void {
   stream.snapshotInfo = null
 }
 
+function clearAckFlushTimer(stream: RemoteRuntimeMultiplexedTerminalState): void {
+  if (stream.ackFlushTimer !== null) {
+    clearTimeout(stream.ackFlushTimer)
+    stream.ackFlushTimer = null
+  }
+}
+
+function discardOutputAcknowledgements(stream: RemoteRuntimeMultiplexedTerminalState): void {
+  clearAckFlushTimer(stream)
+  stream.pendingAckBytes = 0
+  stream.heldAckBytes = 0
+}
+
 function clearPendingSnapshotRequest(stream: RemoteRuntimeMultiplexedTerminalState): void {
   const request = stream.pendingSnapshotRequest
   stream.pendingSnapshotRequest = null

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

@@ -10,6 +10,7 @@ import {
 
 describe('remote terminal renderer backpressure', () => {
   const sendBinary = vi.fn()
+  const unsubscribe = vi.fn()
   let callbacks: {
     onResponse: (response: unknown) => void
     onBinary: (bytes: Uint8Array<ArrayBufferLike>) => void
@@ -18,6 +19,7 @@ describe('remote terminal renderer backpressure', () => {
   beforeEach(() => {
     vi.resetModules()
     sendBinary.mockReset()
+    unsubscribe.mockReset()
     callbacks = null
     vi.stubGlobal('window', {
       api: {
@@ -27,7 +29,7 @@ describe('remote terminal renderer backpressure', () => {
             queueMicrotask(() => {
               callbacks?.onResponse({ ok: true, result: { type: 'ready' } })
             })
-            return { unsubscribe: vi.fn(), sendBinary }
+            return { unsubscribe, sendBinary }
           })
         }
       }
@@ -101,10 +103,265 @@ describe('remote terminal renderer backpressure', () => {
     expect(sentAckBytes()).toEqual([])
 
     parsedCallbacks.shift()?.()
-    expect(sentAckBytes()).toEqual([output.byteLength])
+    await vi.waitFor(() => expect(sentAckBytes()).toEqual([output.byteLength]))
     stream.close()
   })
 
+  it('batches parsed bulk output credit up to the byte threshold', 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-bulk',
+      client: { id: 'mac-viewer', type: 'desktop' },
+      callbacks: {
+        onData: (data) => {
+          writeTerminalOutput(terminal, data, {
+            foreground: true,
+            ackCredit: takeCurrentTerminalDeliveryCredit() ?? undefined
+          })
+        },
+        onSnapshot: vi.fn()
+      }
+    })
+    sendBinary.mockClear()
+    const output = encodeTerminalStreamText('x'.repeat(64 * 1024))
+
+    for (let index = 0; index < 3; index += 1) {
+      callbacks?.onBinary(
+        encodeTerminalStreamFrame({
+          opcode: TerminalStreamOpcode.Output,
+          streamId: stream.streamId,
+          seq: index + 1,
+          payload: output
+        })
+      )
+    }
+
+    expect(sentAckBytes()).toEqual([])
+    for (const parsed of parsedCallbacks) {
+      parsed()
+    }
+    expect(sentAckBytes()).toEqual([output.byteLength * 3])
+    stream.close()
+  })
+
+  it('releases unknown streams and closes malformed connections instead of leaking credit', async () => {
+    const { getRemoteRuntimeTerminalMultiplexer } =
+      await import('./remote-runtime-terminal-multiplexer')
+    const multiplexer = getRemoteRuntimeTerminalMultiplexer('windows-test')
+    const stream = await multiplexer.subscribeTerminal({
+      terminal: 'term-codex',
+      client: { id: 'mac-viewer', type: 'desktop' },
+      callbacks: { onData: vi.fn(), onSnapshot: vi.fn() }
+    })
+    sendBinary.mockClear()
+
+    callbacks?.onBinary(
+      encodeTerminalStreamFrame({
+        opcode: TerminalStreamOpcode.Output,
+        streamId: stream.streamId + 100,
+        seq: 1,
+        payload: encodeTerminalStreamText('x')
+      })
+    )
+    expect(
+      sendBinary.mock.calls.some(([bytes]) => {
+        const frame = decodeTerminalStreamFrame(bytes)
+        return (
+          frame?.opcode === TerminalStreamOpcode.Unsubscribe &&
+          frame.streamId === stream.streamId + 100
+        )
+      })
+    ).toBe(true)
+
+    callbacks?.onBinary(new Uint8Array([1, 2, 3]))
+    expect(unsubscribe).toHaveBeenCalledOnce()
+  })
+
+  it('credits malformed transformed output only after intentionally discarding it', async () => {
+    const { getRemoteRuntimeTerminalMultiplexer } =
+      await import('./remote-runtime-terminal-multiplexer')
+    const onData = vi.fn()
+    const stream = await getRemoteRuntimeTerminalMultiplexer('windows-test').subscribeTerminal({
+      terminal: 'term-codex',
+      client: { id: 'mac-viewer', type: 'desktop' },
+      callbacks: { onData, onSnapshot: vi.fn() }
+    })
+    sendBinary.mockClear()
+    const malformed = encodeTerminalStreamJson({ data: 42, rawLength: 'wrong' })
+
+    callbacks?.onBinary(
+      encodeTerminalStreamFrame({
+        opcode: TerminalStreamOpcode.OutputSpan,
+        streamId: stream.streamId,
+        seq: 4,
+        payload: malformed
+      })
+    )
+
+    expect(onData).not.toHaveBeenCalled()
+    await vi.waitFor(() => expect(sentAckBytes()).toEqual([malformed.byteLength]))
+    expect(
+      sendBinary.mock.calls.some(([bytes]) => {
+        const frame = decodeTerminalStreamFrame(bytes)
+        return frame?.opcode === TerminalStreamOpcode.SnapshotRequest
+      })
+    ).toBe(true)
+    stream.close()
+  })
+
+  it('passes transformed sequence metadata and cancels pending credit on disposal', async () => {
+    const { getRemoteRuntimeTerminalMultiplexer } =
+      await import('./remote-runtime-terminal-multiplexer')
+    const onData = vi.fn()
+    const stream = await getRemoteRuntimeTerminalMultiplexer('windows-test').subscribeTerminal({
+      terminal: 'term-codex',
+      client: { id: 'mac-viewer', type: 'desktop' },
+      callbacks: { onData, onSnapshot: vi.fn() }
+    })
+    sendBinary.mockClear()
+    const transformed = encodeTerminalStreamJson({
+      data: 'visible',
+      rawLength: 11,
+      transformed: true
+    })
+    callbacks?.onBinary(
+      encodeTerminalStreamFrame({
+        opcode: TerminalStreamOpcode.OutputSpan,
+        streamId: stream.streamId,
+        seq: 21,
+        payload: transformed
+      })
+    )
+
+    expect(onData).toHaveBeenCalledWith('visible', {
+      seq: 21,
+      rawLength: 11,
+      transformed: true
+    })
+    stream.close()
+    await new Promise((resolve) => setTimeout(resolve, 10))
+    expect(sentAckBytes()).toEqual([])
+  })
+
+  it('settles a late parser callback locally after the server ends the stream', async () => {
+    const { getRemoteRuntimeTerminalMultiplexer } =
+      await import('./remote-runtime-terminal-multiplexer')
+    const { takeCurrentTerminalDeliveryCredit } =
+      await import('../lib/pane-manager/terminal-delivery-credit')
+    const parsedCredits: (() => void)[] = []
+    const stream = await getRemoteRuntimeTerminalMultiplexer('windows-test').subscribeTerminal({
+      terminal: 'term-codex',
+      client: { id: 'mac-viewer', type: 'desktop' },
+      callbacks: {
+        onData: () => {
+          const credit = takeCurrentTerminalDeliveryCredit()
+          if (credit) {
+            parsedCredits.push(credit)
+          }
+        },
+        onSnapshot: vi.fn()
+      }
+    })
+    sendBinary.mockClear()
+    callbacks?.onBinary(
+      encodeTerminalStreamFrame({
+        opcode: TerminalStreamOpcode.Output,
+        streamId: stream.streamId,
+        seq: 1,
+        payload: encodeTerminalStreamText('x')
+      })
+    )
+    callbacks?.onResponse({ ok: true, result: { type: 'end', streamId: stream.streamId } })
+
+    parsedCredits[0]?.()
+    await new Promise((resolve) => setTimeout(resolve, 10))
+    expect(sentAckBytes()).toEqual([])
+  })
+
+  it('closes without ACKing when the renderer delivery callback throws', async () => {
+    const { getRemoteRuntimeTerminalMultiplexer } =
+      await import('./remote-runtime-terminal-multiplexer')
+    const stream = await getRemoteRuntimeTerminalMultiplexer('windows-test').subscribeTerminal({
+      terminal: 'term-codex',
+      client: { id: 'mac-viewer', type: 'desktop' },
+      callbacks: {
+        onData: () => {
+          throw new Error('renderer delivery failed')
+        },
+        onSnapshot: vi.fn()
+      }
+    })
+    sendBinary.mockClear()
+
+    callbacks?.onBinary(
+      encodeTerminalStreamFrame({
+        opcode: TerminalStreamOpcode.Output,
+        streamId: stream.streamId,
+        seq: 1,
+        payload: encodeTerminalStreamText('x')
+      })
+    )
+
+    expect(unsubscribe).toHaveBeenCalledOnce()
+    await new Promise((resolve) => setTimeout(resolve, 10))
+    expect(sentAckBytes()).toEqual([])
+  })
+
+  it('closes and releases server debt when an ACK transport write throws', async () => {
+    const { getRemoteRuntimeTerminalMultiplexer } =
+      await import('./remote-runtime-terminal-multiplexer')
+    const { takeCurrentTerminalDeliveryCredit } =
+      await import('../lib/pane-manager/terminal-delivery-credit')
+    const parseCredits: (() => void)[] = []
+    const stream = await getRemoteRuntimeTerminalMultiplexer('windows-test').subscribeTerminal({
+      terminal: 'term-codex',
+      client: { id: 'mac-viewer', type: 'desktop' },
+      callbacks: {
+        onData: () => {
+          const credit = takeCurrentTerminalDeliveryCredit()
+          if (credit) {
+            parseCredits.push(credit)
+          }
+        },
+        onSnapshot: vi.fn()
+      }
+    })
+    sendBinary.mockClear()
+    sendBinary.mockImplementation((bytes) => {
+      const frame = decodeTerminalStreamFrame(bytes)
+      if (frame?.opcode === TerminalStreamOpcode.Ack) {
+        throw new Error('socket closed')
+      }
+    })
+    callbacks?.onBinary(
+      encodeTerminalStreamFrame({
+        opcode: TerminalStreamOpcode.Output,
+        streamId: stream.streamId,
+        seq: 1,
+        payload: encodeTerminalStreamText('x')
+      })
+    )
+
+    expect(parseCredits).toHaveLength(1)
+    parseCredits[0]?.()
+
+    await vi.waitFor(() => expect(unsubscribe).toHaveBeenCalledOnce())
+    expect(sentAckBytes()).toEqual([1])
+  })
+
   function sentAckBytes(): number[] {
     return sendBinary.mock.calls.flatMap(([bytes]) => {
       const frame = decodeTerminalStreamFrame(bytes)

src/renderer/src/runtime/runtime-terminal-stream.test.ts

@@ -120,6 +120,14 @@ describe('remote runtime terminal data subscriptions', () => {
     )
 
     expect(watcher).toHaveBeenCalledWith('live')
+    await vi.waitFor(() =>
+      expect(
+        sendBinary.mock.calls
+          .slice(1)
+          .map((call) => decodeTerminalStreamFrame(call[0]))
+          .some((frame) => frame?.opcode === TerminalStreamOpcode.Ack)
+      ).toBe(true)
+    )
     const ackFrame = sendBinary.mock.calls
       .slice(1)
       .map((call) => decodeTerminalStreamFrame(call[0]))
@@ -362,6 +370,13 @@ describe('remote runtime terminal multiplex ACK gate', () => {
       })
     )
 
+    await vi.waitFor(() =>
+      expect(
+        sendBinary.mock.calls
+          .map((call) => decodeTerminalStreamFrame(call[0]))
+          .filter((frame) => frame?.opcode === TerminalStreamOpcode.Ack)
+      ).toHaveLength(1)
+    )
     const immediateAckFrames = sendBinary.mock.calls
       .map((call) => decodeTerminalStreamFrame(call[0]))
       .filter((frame) => frame?.opcode === TerminalStreamOpcode.Ack)
@@ -374,6 +389,13 @@ describe('remote runtime terminal multiplex ACK gate', () => {
     })
 
     gate?.release()
+    await vi.waitFor(() =>
+      expect(
+        sendBinary.mock.calls
+          .map((call) => decodeTerminalStreamFrame(call[0]))
+          .filter((frame) => frame?.opcode === TerminalStreamOpcode.Ack)
+      ).toHaveLength(2)
+    )
     const allAckFrames = sendBinary.mock.calls
       .map((call) => decodeTerminalStreamFrame(call[0]))
       .filter((frame) => frame?.opcode === TerminalStreamOpcode.Ack)

src/shared/terminal-multiplex-flow-control.ts

@@ -0,0 +1,10 @@
+export const TERMINAL_STREAM_CHUNK_BYTES = 48 * 1024
+export const TERMINAL_OUTPUT_BATCH_MAX_BYTES = 64 * 1024
+export const TERMINAL_MULTIPLEX_ACK_STREAM_INITIAL_WINDOW_BYTES = 512 * 1024
+export const TERMINAL_MULTIPLEX_ACK_STREAM_MAX_WINDOW_BYTES = 2 * 1024 * 1024
+export const TERMINAL_MULTIPLEX_ACK_TOTAL_INITIAL_WINDOW_BYTES = 2 * 1024 * 1024
+export const TERMINAL_MULTIPLEX_ACK_TOTAL_MAX_WINDOW_BYTES = 8 * 1024 * 1024
+export const TERMINAL_MULTIPLEX_PENDING_MAX_BYTES = 256 * 1024
+export const TERMINAL_MULTIPLEX_ACK_BATCH_BYTES = 192 * 1024
+export const TERMINAL_MULTIPLEX_ACK_FLUSH_MS = 4
+export const TERMINAL_MULTIPLEX_MAX_STREAMS_PER_CONNECTION = 32