fix(pty): bound cooked reply queue (#13422)

* fix(pty): bound cooked reply queue

* fix(pty): bound cooked reply queue

Implement bounded storage for cooked-echo-safe replies: 64 pending
replies and 4096 UTF-16 code units. Shed oldest replies on overflow,
never ordinary input. Add drain failure containment with generation
fencing to prevent stale operations from clearing fresh input after
clear() reuse.

* fix(pty): report pty id in drain failures

When the async drain yields, the owner may rebind to a different PTY
before the failure surfaces. Pass the failing pty id so the transport can
ignore stale failures from a rebound owner. Also tighten the pending
reply queue size bound to prevent half-written entries.

(cherry picked from commit c0c893d171)

src/renderer/src/components/terminal-pane/pty-input-write-queue.test.ts

@@ -1,5 +1,8 @@
 import { describe, expect, it, vi } from 'vitest'
+import { Terminal } from '@xterm/headless'
 import {
+  PTY_INPUT_WRITE_QUEUE_MAX_PENDING_REPLIES,
+  PTY_INPUT_WRITE_QUEUE_MAX_PENDING_REPLY_CODE_UNITS,
   TERMINAL_INPUT_COALESCE_MAX_CODE_UNITS,
   createPtyInputWriteQueue
 } from './pty-input-write-queue'
@@ -7,9 +10,14 @@ import {
   TERMINAL_INPUT_CHUNK_MAX_BYTES,
   TERMINAL_INPUT_MAX_BYTES
 } from '../../../../shared/terminal-input'
+import { CLIPBOARD_TEXT_MEASURE_YIELD_CODE_UNITS } from '../../../../shared/clipboard-text'
 import { mode2031SequenceFor } from '../../../../shared/terminal-color-scheme-protocol'
 import { PtyStartupIngress, type PtyIngressEmission } from '../../../../shared/pty-startup-ingress'
-import { extractOnlyCookedEchoSafeQueryReplies } from '../../../../shared/terminal-query-reply'
+import {
+  extractOnlyCookedEchoSafeQueryReplies,
+  needsCookedEchoSafeQueryReply
+} from '../../../../shared/terminal-query-reply'
+import { installTerminalCapabilityReplyHandlers } from './terminal-capability-replies'
 
 const WHEEL_UP_REPORT = '\x1b[<64;60;20M'
 
@@ -27,6 +35,44 @@ function createRecordingQueue(options: { writable?: () => boolean } = {}): {
   return { writes, queue }
 }
 
+function createParkedQueue(): {
+  writes: WriteRecord[]
+  pendingYields: (() => void)[]
+  queue: ReturnType<typeof createPtyInputWriteQueue>
+} {
+  const writes: WriteRecord[] = []
+  const pendingYields: (() => void)[] = []
+  const queue = createPtyInputWriteQueue({
+    isWritable: () => true,
+    write: (id, data) => writes.push({ id, data }),
+    yieldBetweenWrites: () =>
+      new Promise<void>((resolve) => {
+        pendingYields.push(resolve)
+      })
+  })
+  return { writes, pendingYields, queue }
+}
+
+async function releaseNextWrite(
+  writes: WriteRecord[],
+  pendingYields: (() => void)[]
+): Promise<void> {
+  const before = writes.length
+  const release = pendingYields.shift()
+  expect(release).toBeDefined()
+  release?.()
+  await Promise.resolve()
+  expect(writes.length).toBeGreaterThan(before)
+}
+
+function writeTerminal(term: Terminal, data: string): Promise<void> {
+  return new Promise((resolve) => term.write(data, resolve))
+}
+
+function extractReplyWrites(writes: WriteRecord[]): string[] {
+  return writes.flatMap((write) => extractOnlyCookedEchoSafeQueryReplies(write.data) ?? [])
+}
+
 describe('pty input write queue', () => {
   it('coalesces a dense burst of wheel reports instead of one write per macrotask turn', async () => {
     const { writes, queue } = createRecordingQueue()
@@ -136,26 +182,21 @@ describe('pty input write queue', () => {
     const { writes, queue } = createRecordingQueue()
     const reply = mode2031SequenceFor('dark')
 
-    expect(queue.enqueue('pty-1', reply)).toBe(true)
-    expect(queue.enqueue('pty-1', reply)).toBe(true)
+    expect(queue.enqueueQueryReply('pty-1', reply)).toBe(true)
+    expect(queue.enqueueQueryReply('pty-1', reply)).toBe(true)
     await queue.waitForDrain()
 
-    // Atomic writes so each host intercept sees a single extractable reply —
-    // not one coalesced `?997;1n?997;1n` payload.
     expect(writes).toEqual([
       { id: 'pty-1', data: reply },
       { id: 'pty-1', data: reply }
     ])
-    expect(writes).toHaveLength(2)
-    expect(writes[0]?.data).toBe(reply)
-    expect(writes[1]?.data).toBe(reply)
   })
 
   it('does not coalesce a color-scheme reply with a following keystroke', async () => {
     const { writes, queue } = createRecordingQueue()
     const reply = mode2031SequenceFor('dark')
 
-    queue.enqueue('pty-1', reply)
+    queue.enqueueQueryReply('pty-1', reply)
     queue.enqueue('pty-1', 'y')
     await queue.waitForDrain()
 
@@ -165,6 +206,268 @@ describe('pty input write queue', () => {
     ])
   })
 
+  it('bounds an OSC 10/11 reply flood and drains a following keystroke', async () => {
+    const { writes, pendingYields, queue } = createParkedQueue()
+    const replies: string[] = []
+    let allAccepted = true
+
+    for (let index = 0; index < 10_000; index += 1) {
+      const color = index.toString(16).padStart(4, '0')
+      const reply = `\x1b]${index % 2 === 0 ? 10 : 11};rgb:${color}/0000/0000\x1b\\`
+      replies.push(reply)
+      allAccepted = queue.enqueueQueryReply('pty-1', reply) && allAccepted
+    }
+    expect(allAccepted).toBe(true)
+    expect(queue.enqueue('pty-1', 'k')).toBe(true)
+
+    await Promise.resolve()
+    expect(pendingYields).toHaveLength(1)
+    for (let turn = 0; turn < PTY_INPUT_WRITE_QUEUE_MAX_PENDING_REPLIES; turn += 1) {
+      await releaseNextWrite(writes, pendingYields)
+    }
+    await queue.waitForDrain()
+
+    const replyWrites = writes.filter((write) => write.data.startsWith('\x1b]'))
+    expect(replyWrites.map((write) => write.data)).toEqual([
+      replies[0],
+      ...replies.slice(-PTY_INPUT_WRITE_QUEUE_MAX_PENDING_REPLIES)
+    ])
+    expect(replyWrites.every((write) => needsCookedEchoSafeQueryReply(write.data))).toBe(true)
+    expect(writes.at(-1)?.data).toBe('k')
+  })
+
+  it('drops the oldest reply-only payload when the reply text budget fills', async () => {
+    const reply = (slot: 10 | 11, marker: string): string =>
+      `\x1b]${slot};${marker.repeat(1_400)}\x1b\\`
+    const first = reply(10, '1')
+    const dropped = reply(11, '2')
+    const second = reply(10, '3')
+    const third = reply(11, '4')
+    const { writes, pendingYields, queue } = createParkedQueue()
+
+    expect(second.length * 2).toBeLessThanOrEqual(
+      PTY_INPUT_WRITE_QUEUE_MAX_PENDING_REPLY_CODE_UNITS
+    )
+    expect(second.length * 3).toBeGreaterThan(PTY_INPUT_WRITE_QUEUE_MAX_PENDING_REPLY_CODE_UNITS)
+    queue.enqueueQueryReply('pty-1', first)
+    queue.enqueueQueryReply('pty-1', dropped)
+    queue.enqueue('pty-1', 'x')
+    queue.enqueueQueryReply('pty-1', second)
+    queue.enqueueQueryReply('pty-1', third)
+    queue.enqueue('pty-1', 'k')
+
+    await Promise.resolve()
+    expect(pendingYields).toHaveLength(1)
+    for (let turn = 0; turn < 3; turn += 1) {
+      await releaseNextWrite(writes, pendingYields)
+    }
+    await queue.waitForDrain()
+
+    expect(writes).toEqual([
+      { id: 'pty-1', data: first },
+      { id: 'pty-1', data: 'x' },
+      { id: 'pty-1', data: second },
+      { id: 'pty-1', data: third },
+      { id: 'pty-1', data: 'k' }
+    ])
+  })
+
+  it('bounds replies generated by a real xterm OSC query flood', async () => {
+    const term = new Terminal({ cols: 80, rows: 24, allowProposedApi: true })
+    term.options.theme = { foreground: '#2e3434', background: '#ffffff' }
+    const { writes, queue } = createRecordingQueue()
+    const generated: string[] = []
+    const disposable = installTerminalCapabilityReplyHandlers({
+      terminal: {
+        cols: term.cols,
+        rows: term.rows,
+        element: undefined,
+        options: term.options
+      },
+      parser: term.parser,
+      sendInput: (data) => {
+        generated.push(data)
+        return queue.enqueueQueryReply('pty-1', data)
+      },
+      isReplaying: () => false
+    })
+
+    try {
+      await writeTerminal(term, '\x1b]10;?;?\x1b\\'.repeat(34))
+      await queue.waitForDrain()
+
+      const foreground = '\x1b]10;rgb:2e2e/3434/3434\x1b\\'
+      const background = '\x1b]11;rgb:ffff/ffff/ffff\x1b\\'
+      expect(generated).toEqual(Array.from({ length: 34 }, () => [foreground, background]).flat())
+      expect(extractReplyWrites(writes)).toEqual([
+        foreground,
+        ...generated.slice(-PTY_INPUT_WRITE_QUEUE_MAX_PENDING_REPLIES)
+      ])
+    } finally {
+      disposable.dispose()
+      term.dispose()
+    }
+  })
+
+  it('rejects one reply larger than the total reply retention budget', async () => {
+    const { writes, queue } = createRecordingQueue()
+    const prefix = '\x1b]10;'
+    const suffix = '\x1b\\'
+    const input = `${prefix}${'x'.repeat(
+      TERMINAL_INPUT_COALESCE_MAX_CODE_UNITS + 1 - prefix.length - suffix.length
+    )}${suffix}`
+
+    expect(input).toHaveLength(TERMINAL_INPUT_COALESCE_MAX_CODE_UNITS + 1)
+    expect(queue.enqueueQueryReply('pty-1', input)).toBe(false)
+    await queue.waitForDrain()
+
+    expect(writes).toEqual([])
+  })
+
+  it('preserves an ordinary backlog while bounding later replies', async () => {
+    const { writes, queue } = createRecordingQueue()
+    const ordinary = Array.from({ length: 10_000 }, (_, index) => String(index % 10))
+    const replies = Array.from({ length: 10_000 }, (_, index) => `\x1b[?${index};1n`)
+
+    for (const input of ordinary) {
+      expect(queue.enqueue('pty-1', input)).toBe(true)
+    }
+    for (const reply of replies) {
+      expect(queue.enqueueQueryReply('pty-1', reply)).toBe(true)
+    }
+    await queue.waitForDrain()
+
+    expect(writes.map((write) => write.data).join('')).toBe(
+      [...ordinary, ...replies.slice(-PTY_INPUT_WRITE_QUEUE_MAX_PENDING_REPLIES)].join('')
+    )
+    expect(extractReplyWrites(writes)).toEqual(
+      replies.slice(-PTY_INPUT_WRITE_QUEUE_MAX_PENDING_REPLIES)
+    )
+  })
+
+  it('never sheds reply-shaped bytes sent through the ordinary input path', async () => {
+    const { writes, queue } = createRecordingQueue()
+    const ordinary = '\x1b]10;user-ordinary-marker\x1b\\'
+    const replies = Array.from({ length: 10_000 }, (_, index) => `\x1b[?${index};1n`)
+
+    expect(queue.enqueueQueryReply('pty-1', '\x1b[?10000;1n')).toBe(true)
+    expect(queue.enqueue('pty-1', ordinary)).toBe(true)
+    for (const reply of replies) {
+      expect(queue.enqueueQueryReply('pty-1', reply)).toBe(true)
+    }
+    await queue.waitForDrain()
+
+    expect(writes.map((write) => write.data).join('')).toContain(ordinary)
+  })
+
+  it('contains drain failures and rejects input until queue reuse', async () => {
+    const failure = new Error('write failed')
+    const writes: WriteRecord[] = []
+    let shouldThrow = true
+    const onDrainFailure = vi.fn()
+    const warn = vi.spyOn(console, 'warn').mockImplementation(() => undefined)
+    const queue = createPtyInputWriteQueue({
+      isWritable: () => true,
+      write: (id, data) => {
+        if (shouldThrow) {
+          throw failure
+        }
+        writes.push({ id, data })
+      },
+      onDrainFailure
+    })
+
+    try {
+      expect(queue.enqueue('pty-1', 'stale')).toBe(true)
+      expect(queue.enqueue('pty-1', 'also-stale')).toBe(false)
+      await expect(queue.waitForDrain()).resolves.toBeUndefined()
+
+      shouldThrow = false
+      expect(queue.enqueue('pty-1', 'still-stale')).toBe(false)
+      queue.clear()
+      expect(queue.enqueue('pty-1', 'fresh')).toBe(true)
+      await queue.waitForDrain()
+
+      expect(writes).toEqual([{ id: 'pty-1', data: 'fresh' }])
+      expect(warn).toHaveBeenCalledWith('[pty-input-write-queue] drain failed:', failure)
+      expect(onDrainFailure).toHaveBeenCalledOnce()
+    } finally {
+      warn.mockRestore()
+    }
+  })
+
+  it('reports the pty id that failed so a rebound owner can ignore the drain failure', async () => {
+    const failure = new Error('yield failed')
+    const onDrainFailure = vi.fn()
+    let rejectYield: ((error: Error) => void) | undefined
+    const warn = vi.spyOn(console, 'warn').mockImplementation(() => undefined)
+    const queue = createPtyInputWriteQueue({
+      isWritable: () => true,
+      write: () => undefined,
+      yieldBetweenWrites: () =>
+        new Promise<void>((_resolve, reject) => {
+          rejectYield = reject
+        }),
+      onDrainFailure
+    })
+
+    try {
+      expect(queue.enqueue('pty-1', 'x'.repeat(TERMINAL_INPUT_CHUNK_MAX_BYTES * 2))).toBe(true)
+      expect(rejectYield).toBeDefined()
+      rejectYield?.(failure)
+      await queue.waitForDrain()
+
+      expect(onDrainFailure).toHaveBeenCalledExactlyOnceWith('pty-1')
+    } finally {
+      warn.mockRestore()
+    }
+  })
+
+  it('keeps a pending reply small enough that one drain step writes and removes it', () => {
+    // Larger replies would span chunks, so admitReply could evict a half-written entry.
+    expect(PTY_INPUT_WRITE_QUEUE_MAX_PENDING_REPLY_CODE_UNITS * 3).toBeLessThanOrEqual(
+      TERMINAL_INPUT_CHUNK_MAX_BYTES
+    )
+  })
+
+  it('does not let a stale drain failure clear input queued after reuse', async () => {
+    const failure = new Error('stale yield failed')
+    const writes: WriteRecord[] = []
+    let rejectYield: ((error: Error) => void) | undefined
+    let rejectNextYield = true
+    const onDrainFailure = vi.fn()
+    const warn = vi.spyOn(console, 'warn').mockImplementation(() => undefined)
+    const queue = createPtyInputWriteQueue({
+      isWritable: () => true,
+      write: (id, data) => writes.push({ id, data }),
+      yieldBetweenWrites: () => {
+        if (!rejectNextYield) {
+          return Promise.resolve()
+        }
+        return new Promise<void>((_resolve, reject) => {
+          rejectYield = reject
+        })
+      },
+      onDrainFailure
+    })
+
+    try {
+      expect(queue.enqueue('pty-1', 'x'.repeat(TERMINAL_INPUT_CHUNK_MAX_BYTES * 2))).toBe(true)
+      expect(rejectYield).toBeDefined()
+      queue.clear()
+      rejectNextYield = false
+      expect(queue.enqueue('pty-2', 'fresh')).toBe(true)
+      rejectYield?.(failure)
+      await queue.waitForDrain()
+
+      expect(writes.at(-1)).toEqual({ id: 'pty-2', data: 'fresh' })
+      expect(warn).toHaveBeenCalledWith('[pty-input-write-queue] drain failed:', failure)
+      expect(onDrainFailure).not.toHaveBeenCalled()
+    } finally {
+      warn.mockRestore()
+    }
+  })
+
   it('dual mode-2031 enqueues through the real queue never paint 997 under host echo-safe write', async () => {
     // Issue path: xterm onData + mode-2031 scan each enqueue mode2031SequenceFor.
     // Drive the real write queue → host intercept (extract + answerLiveQueryReply)
@@ -198,8 +501,8 @@ describe('pty input write queue', () => {
       write: hostWrite
     })
 
-    expect(queue.enqueue('pty-1', reply)).toBe(true)
-    expect(queue.enqueue('pty-1', reply)).toBe(true)
+    expect(queue.enqueueQueryReply('pty-1', reply)).toBe(true)
+    expect(queue.enqueueQueryReply('pty-1', reply)).toBe(true)
     await queue.waitForDrain()
     await vi.advanceTimersByTimeAsync(0)
 
@@ -243,4 +546,37 @@ describe('pty input write queue', () => {
     expect(writes.length).toBe(1)
     expect(writes.map((write) => write.data).join('')).not.toContain('tail')
   })
+
+  it('clear() fences an in-flight validation before queue reuse', async () => {
+    vi.useFakeTimers()
+    const { writes, queue } = createRecordingQueue()
+    const stale = 'é'.repeat(CLIPBOARD_TEXT_MEASURE_YIELD_CODE_UNITS + 1)
+
+    try {
+      expect(queue.enqueue('pty-1', stale)).toBe(true)
+      queue.clear()
+      expect(queue.enqueue('pty-1', 'fresh')).toBe(true)
+
+      await vi.runAllTimersAsync()
+      await queue.waitForDrain()
+
+      expect(writes).toEqual([{ id: 'pty-1', data: 'fresh' }])
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+
+  it('clear() resets reply capacity for later input', async () => {
+    const { writes, queue } = createRecordingQueue()
+    const reply = mode2031SequenceFor('dark')
+
+    for (let index = 0; index <= PTY_INPUT_WRITE_QUEUE_MAX_PENDING_REPLIES; index += 1) {
+      expect(queue.enqueueQueryReply('pty-1', reply)).toBe(true)
+    }
+    queue.clear()
+    expect(queue.enqueueQueryReply('pty-1', reply)).toBe(true)
+    await queue.waitForDrain()
+
+    expect(extractReplyWrites(writes)).toEqual([reply, reply])
+  })
 })

src/renderer/src/components/terminal-pane/pty-input-write-queue.ts

@@ -9,10 +9,18 @@ import { needsCookedEchoSafeQueryReply } from '../../../../shared/terminal-query
 // 16KB TERMINAL_INPUT_CHUNK_MAX_BYTES cap without paying byte measurement on
 // the hot input path.
 export const TERMINAL_INPUT_COALESCE_MAX_CODE_UNITS = 4096
+// Match host delivery's reply ceiling while keeping all retained reply text under one PTY chunk.
+export const PTY_INPUT_WRITE_QUEUE_MAX_PENDING_REPLIES = 64
+// Keep ≤ TERMINAL_INPUT_CHUNK_MAX_BYTES/3 so a reply is written and dropped in one drain step:
+// admitReply evicts the head, and a half-written entry would truncate. Guarded by a unit test.
+export const PTY_INPUT_WRITE_QUEUE_MAX_PENDING_REPLY_CODE_UNITS =
+  TERMINAL_INPUT_COALESCE_MAX_CODE_UNITS
 
 type PendingPtyInputWrite = {
+  sequence: number
   id: string
   text: string
+  replyOnly: boolean
   tooLarge: boolean | Promise<boolean>
   chunks?: Iterator<string>
   nextChunk?: string
@@ -20,6 +28,7 @@ type PendingPtyInputWrite = {
 
 export type PtyInputWriteQueue = {
   enqueue: (id: string, data: string) => boolean
+  enqueueQueryReply: (id: string, data: string) => boolean
   waitForDrain: () => Promise<void>
   clear: () => void
 }
@@ -28,90 +37,209 @@ export type PtyInputWriteQueueDeps = {
   isWritable: (id: string) => boolean
   write: (id: string, data: string) => void
   yieldBetweenWrites?: () => Promise<void>
+  onDrainFailure?: (id: string) => void
 }
 
-function isCoalescibleText(text: string): boolean {
+function isCoalescibleInput(input: PendingPtyInputWrite): boolean {
   // Echo-risk replies stay atomic so host classifiers cannot miss them (#13137).
-  return (
-    text.length <= TERMINAL_INPUT_COALESCE_MAX_CODE_UNITS && !needsCookedEchoSafeQueryReply(text)
-  )
+  return input.text.length <= TERMINAL_INPUT_COALESCE_MAX_CODE_UNITS && !input.replyOnly
 }
 
 export function createPtyInputWriteQueue(deps: PtyInputWriteQueueDeps): PtyInputWriteQueue {
   const yieldBetweenWrites = deps.yieldBetweenWrites ?? yieldToEventLoop
-  let pending: PendingPtyInputWrite[] = []
+  let pendingOrdinary: (PendingPtyInputWrite | undefined)[] = []
+  let pendingOrdinaryHead = 0
+  let pendingReplies: (PendingPtyInputWrite | undefined)[] = []
+  let pendingReplyHead = 0
+  let pendingReplyCount = 0
+  let pendingReplyCodeUnits = 0
+  let nextSequence = 0
+  let generation = 0
+  let failedGeneration: number | null = null
   let drainPromise: Promise<void> | null = null
 
+  function compactOrdinary(): void {
+    if (pendingOrdinaryHead === pendingOrdinary.length) {
+      pendingOrdinary = []
+      pendingOrdinaryHead = 0
+    } else if (pendingOrdinaryHead >= 1024 && pendingOrdinaryHead * 2 >= pendingOrdinary.length) {
+      pendingOrdinary = pendingOrdinary.slice(pendingOrdinaryHead)
+      pendingOrdinaryHead = 0
+    }
+  }
+
+  function compactReplies(): void {
+    if (pendingReplyHead === pendingReplies.length) {
+      pendingReplies = []
+      pendingReplyHead = 0
+    } else if (pendingReplyHead >= 1024 && pendingReplyHead * 2 >= pendingReplies.length) {
+      pendingReplies = pendingReplies.slice(pendingReplyHead)
+      pendingReplyHead = 0
+    }
+  }
+
+  function resetSequenceIfEmpty(): void {
+    if (pendingOrdinary.length === 0 && pendingReplies.length === 0) {
+      nextSequence = 0
+    }
+  }
+
+  function firstPending(): PendingPtyInputWrite | undefined {
+    const ordinary = pendingOrdinary[pendingOrdinaryHead]
+    const reply = pendingReplies[pendingReplyHead]
+    if (!ordinary) {
+      return reply
+    }
+    if (!reply) {
+      return ordinary
+    }
+    return ordinary.sequence < reply.sequence ? ordinary : reply
+  }
+
+  function shiftOrdinary(): PendingPtyInputWrite | undefined {
+    const removed = pendingOrdinary[pendingOrdinaryHead]
+    pendingOrdinary[pendingOrdinaryHead] = undefined
+    pendingOrdinaryHead += 1
+    compactOrdinary()
+    resetSequenceIfEmpty()
+    return removed
+  }
+
+  function shiftReply(): PendingPtyInputWrite | undefined {
+    const removed = pendingReplies[pendingReplyHead]
+    pendingReplies[pendingReplyHead] = undefined
+    pendingReplyHead += 1
+    if (removed) {
+      pendingReplyCount -= 1
+      pendingReplyCodeUnits -= removed.text.length
+    }
+    compactReplies()
+    resetSequenceIfEmpty()
+    return removed
+  }
+
+  function removePending(item: PendingPtyInputWrite): void {
+    if (item.replyOnly) {
+      shiftReply()
+    } else {
+      shiftOrdinary()
+    }
+  }
+
+  function admitReply(text: string): boolean {
+    if (text.length > PTY_INPUT_WRITE_QUEUE_MAX_PENDING_REPLY_CODE_UNITS) {
+      return false
+    }
+    while (
+      pendingReplyCount >= PTY_INPUT_WRITE_QUEUE_MAX_PENDING_REPLIES ||
+      pendingReplyCodeUnits + text.length > PTY_INPUT_WRITE_QUEUE_MAX_PENDING_REPLY_CODE_UNITS
+    ) {
+      if (!shiftReply()) {
+        return false
+      }
+    }
+    return true
+  }
+
+  function clearPending(): void {
+    pendingOrdinary = []
+    pendingOrdinaryHead = 0
+    pendingReplies = []
+    pendingReplyHead = 0
+    pendingReplyCount = 0
+    pendingReplyCodeUnits = 0
+    nextSequence = 0
+  }
+
   async function drain(): Promise<void> {
-    while (pending.length > 0) {
-      const next = pending[0]
-      if (!next) {
-        pending.shift()
-        continue
-      }
-      if (!deps.isWritable(next.id)) {
-        pending.shift()
-        continue
-      }
-      if (next.tooLarge !== false) {
-        next.tooLarge = await Promise.resolve(next.tooLarge).catch(() => true)
-        if (next.tooLarge) {
-          pending.shift()
-          continue
-        }
+    let failureGeneration = generation
+    // Why: the drain yields, so the owner may rebind before the failure surfaces; report the id that actually failed.
+    let failingId: string | null = null
+    try {
+      let next: PendingPtyInputWrite | undefined
+      while ((next = firstPending())) {
+        failureGeneration = generation
+        failingId = next.id
         if (!deps.isWritable(next.id)) {
-          pending.shift()
+          removePending(next)
           continue
         }
-      }
-      // Why: dense input streams (SGR wheel reports during trackpad momentum,
-      // key auto-repeat) enqueue one tiny item per event. Writing one item per
-      // macrotask turn lets Chromium's nested-timer clamp pace the drain at
-      // ≥4ms per item, so a fast gesture's reports reach the PTY seconds after
-      // the gesture ended and the TUI visibly replays them one by one.
-      // Coalescing consecutive validated small items into a single write keeps
-      // the PTY byte stream identical while draining the backlog in one turn.
-      if (next.chunks === undefined && isCoalescibleText(next.text)) {
-        let payload = next.text
-        pending.shift()
-        while (pending.length > 0) {
-          const peek = pending[0]
-          if (
-            !peek ||
-            peek.id !== next.id ||
-            peek.tooLarge !== false ||
-            peek.chunks !== undefined ||
-            !isCoalescibleText(peek.text) ||
-            payload.length + peek.text.length > TERMINAL_INPUT_COALESCE_MAX_CODE_UNITS
-          ) {
-            break
+        if (next.tooLarge !== false) {
+          next.tooLarge = await Promise.resolve(next.tooLarge).catch(() => true)
+          if (firstPending() !== next) {
+            continue
+          }
+          if (next.tooLarge) {
+            removePending(next)
+            continue
+          }
+          if (!deps.isWritable(next.id)) {
+            removePending(next)
+            continue
           }
-          payload += peek.text
-          pending.shift()
         }
-        deps.write(next.id, payload)
-        if (pending.length > 0) {
+        // Why: dense input streams (SGR wheel reports during trackpad momentum,
+        // key auto-repeat) enqueue one tiny item per event. Writing one item per
+        // macrotask turn lets Chromium's nested-timer clamp pace the drain at
+        // ≥4ms per item, so a fast gesture's reports reach the PTY seconds after
+        // the gesture ended and the TUI visibly replays them one by one.
+        // Coalescing consecutive validated small items into a single write keeps
+        // the PTY byte stream identical while draining the backlog in one turn.
+        if (next.chunks === undefined && isCoalescibleInput(next)) {
+          let payload = next.text
+          removePending(next)
+          let peek: PendingPtyInputWrite | undefined
+          while ((peek = firstPending())) {
+            if (
+              peek.id !== next.id ||
+              peek.tooLarge !== false ||
+              peek.chunks !== undefined ||
+              !isCoalescibleInput(peek) ||
+              payload.length + peek.text.length > TERMINAL_INPUT_COALESCE_MAX_CODE_UNITS
+            ) {
+              break
+            }
+            payload += peek.text
+            removePending(peek)
+          }
+          deps.write(next.id, payload)
+          if (firstPending()) {
+            await yieldBetweenWrites()
+          }
+          continue
+        }
+        next.chunks ??= iterateTerminalInputChunks(next.text)
+        const chunk =
+          next.nextChunk === undefined ? next.chunks.next() : { done: false, value: next.nextChunk }
+        next.nextChunk = undefined
+        if (chunk.done) {
+          removePending(next)
+          continue
+        }
+        deps.write(next.id, chunk.value)
+        const following = next.chunks.next()
+        if (following.done) {
+          removePending(next)
+        } else {
+          next.nextChunk = following.value
+        }
+        if (firstPending()) {
           await yieldBetweenWrites()
         }
-        continue
       }
-      next.chunks ??= iterateTerminalInputChunks(next.text)
-      const chunk =
-        next.nextChunk === undefined ? next.chunks.next() : { done: false, value: next.nextChunk }
-      next.nextChunk = undefined
-      if (chunk.done) {
-        pending.shift()
-        continue
+    } catch (error) {
+      const failureIsCurrent = generation === failureGeneration
+      if (failureIsCurrent) {
+        clearPending()
+        failedGeneration = generation
       }
-      deps.write(next.id, chunk.value)
-      const following = next.chunks.next()
-      if (following.done) {
-        pending.shift()
-      } else {
-        next.nextChunk = following.value
-      }
-      if (pending.length > 0) {
-        await yieldBetweenWrites()
+      console.warn('[pty-input-write-queue] drain failed:', error)
+      if (failureIsCurrent && failingId !== null) {
+        try {
+          deps.onDrainFailure?.(failingId)
+        } catch (recoveryError) {
+          console.warn('[pty-input-write-queue] failure handler failed:', recoveryError)
+        }
       }
     }
   }
@@ -120,27 +248,51 @@ export function createPtyInputWriteQueue(deps: PtyInputWriteQueueDeps): PtyInput
     if (drainPromise) {
       return
     }
-    drainPromise = drain().finally(() => {
+    const finishDrain = (): void => {
       drainPromise = null
-      if (pending.length > 0) {
+      if (firstPending()) {
         scheduleDrain()
       }
-    })
+    }
+    drainPromise = drain().finally(finishDrain)
+  }
+
+  function enqueueInput(id: string, data: string, queryReply: boolean): boolean {
+    try {
+      if (failedGeneration === generation) {
+        return false
+      }
+      const replyOnly = queryReply && needsCookedEchoSafeQueryReply(data)
+      if (replyOnly && !admitReply(data)) {
+        return false
+      }
+      const tooLarge = replyOnly ? false : isTerminalInputTooLargeWithDeferredMeasurement(data)
+      if (tooLarge === true) {
+        return false
+      }
+      const item = { sequence: nextSequence, id, text: data, replyOnly, tooLarge }
+      nextSequence += 1
+      if (replyOnly) {
+        pendingReplies.push(item)
+        pendingReplyCount += 1
+        pendingReplyCodeUnits += data.length
+      } else {
+        pendingOrdinary.push(item)
+      }
+      scheduleDrain()
+      return true
+    } catch {
+      return false
+    }
   }
 
   return {
     enqueue(id: string, data: string): boolean {
-      try {
-        const tooLarge = isTerminalInputTooLargeWithDeferredMeasurement(data)
-        if (tooLarge === true) {
-          return false
-        }
-        pending.push({ id, text: data, tooLarge })
-        scheduleDrain()
-        return true
-      } catch {
-        return false
-      }
+      return enqueueInput(id, data, false)
+    },
+
+    enqueueQueryReply(id: string, data: string): boolean {
+      return enqueueInput(id, data, true)
     },
 
     async waitForDrain(): Promise<void> {
@@ -150,7 +302,9 @@ export function createPtyInputWriteQueue(deps: PtyInputWriteQueueDeps): PtyInput
     },
 
     clear(): void {
-      pending = []
+      generation += 1
+      failedGeneration = null
+      clearPending()
     }
   }
 }

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

@@ -14,6 +14,7 @@ import {
   TERMINAL_INPUT_MAX_BYTES
 } from '../../../../shared/terminal-input'
 import { CLIPBOARD_TEXT_MEASURE_YIELD_CODE_UNITS } from '../../../../shared/clipboard-text'
+import { PTY_INPUT_WRITE_QUEUE_MAX_PENDING_REPLIES } from './pty-input-write-queue'
 
 describe('createIpcPtyTransport', () => {
   const originalWindow = (globalThis as { window?: typeof window }).window
@@ -107,6 +108,29 @@ describe('createIpcPtyTransport', () => {
     transport.disconnect()
   })
 
+  it('routes a thrown renderer write to the owning transport recovery callback', async () => {
+    const failure = new Error('ipc write failed')
+    vi.mocked(window.api.pty.write).mockImplementation(() => {
+      throw failure
+    })
+    const warn = vi.spyOn(console, 'warn').mockImplementation(() => undefined)
+
+    try {
+      const { createIpcPtyTransport } = await import('./pty-transport')
+      const recovery = vi.fn()
+      const transport = createIpcPtyTransport({})
+      await transport.connect({ url: '', callbacks: { onWriteUnavailable: recovery } })
+
+      expect(transport.sendInput('input')).toBe(true)
+      expect(transport.sendInput('later-input')).toBe(false)
+
+      expect(recovery).toHaveBeenCalledOnce()
+      expect(warn).toHaveBeenCalledWith('[pty-input-write-queue] drain failed:', failure)
+    } finally {
+      warn.mockRestore()
+    }
+  })
+
   it('does not create a second kill authority when a mounted pane detaches', async () => {
     const { createIpcPtyTransport } = await import('./pty-transport')
     const kill = window.api.pty.kill as unknown as ReturnType<typeof vi.fn>
@@ -1132,6 +1156,36 @@ describe('createIpcPtyTransport', () => {
     }
   })
 
+  it('bounds immediate cooked replies without shedding ordinary-path lookalikes', async () => {
+    vi.useFakeTimers()
+    try {
+      const { createIpcPtyTransport } = await import('./pty-transport')
+      const transport = createIpcPtyTransport({})
+      const first = '\x1b[?10000;1n'
+      const ordinary = '\x1b]10;user-ordinary-marker\x1b\\'
+      const replies = Array.from({ length: 10_000 }, (_, index) => `\x1b[?${index};1n`)
+
+      await transport.connect({ url: '', callbacks: {} })
+      expect(transport.sendInputImmediate(first)).toBe(true)
+      expect(transport.sendInput(ordinary)).toBe(true)
+      for (const reply of replies) {
+        expect(transport.sendInputImmediate(reply)).toBe(true)
+      }
+
+      await vi.runAllTimersAsync()
+
+      expect(vi.mocked(window.api.pty.write).mock.calls).toEqual([
+        ['pty-1', first],
+        ['pty-1', ordinary],
+        ...replies
+          .slice(-PTY_INPUT_WRITE_QUEUE_MAX_PENDING_REPLIES)
+          .map((reply) => ['pty-1', reply])
+      ])
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+
   it('yields while validating accepted large local IPC terminal input before renderer-to-main writes', async () => {
     vi.useFakeTimers()
     try {

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

@@ -577,9 +577,17 @@ export function createIpcPtyTransport(opts: IpcPtyTransportOptions = {}): PtyTra
   let ptyId: string | null = null
   // Why: replayed eager-buffer data (often from a prior app session) must not fire fresh bells, unread marks, or notifications on reconnect.
   let suppressAttentionEvents = false
+  let storedCallbacks: Parameters<PtyTransport['connect']>[0]['callbacks'] = {}
   const inputWriteQueue = createPtyInputWriteQueue({
     isWritable: (id) => connected && ptyId === id,
-    write: (id, data) => window.api.pty.write(id, data)
+    write: (id, data) => window.api.pty.write(id, data),
+    // Guard like the registered writeUnavailable handler: a rebind during the async drain
+    // must not tell the new pane that its write failed.
+    onDrainFailure: (id) => {
+      if (ptyId === id) {
+        storedCallbacks.onWriteUnavailable?.()
+      }
+    }
   })
   const outputProcessor = createPtyOutputProcessor({
     onTitleChange,
@@ -593,8 +601,6 @@ export function createIpcPtyTransport(opts: IpcPtyTransportOptions = {}): PtyTra
     onAgentExited,
     onAgentStatus
   })
-  let storedCallbacks: Parameters<PtyTransport['connect']>[0]['callbacks'] = {}
-
   // Why: a new pane can attach to the same ptyId before the old instance's detach() runs; track owned handlers so unregister never deletes the live one.
   const ownedDataAndReplayHandlers = new Map<
     string,
@@ -1051,7 +1057,7 @@ export function createIpcPtyTransport(opts: IpcPtyTransportOptions = {}): PtyTra
       if (!connected || !ptyId) {
         return false
       }
-      return inputWriteQueue.enqueue(ptyId, data)
+      return inputWriteQueue.enqueueQueryReply(ptyId, data)
     },
 
     ...(connectionId