fix(terminal): answer queries in output a paused paired view discards (#33783)

* fix(terminal): answer queries in ACK-pending output a paused paired view discards

Refs #32944

* fix(terminal): capture yielded-query ownership at ingestion and cap reply age

Refs #32944

* fix(terminal): settle yielded queries by their own end seq across split frames

Refs #32944

* test: use a fixed timestamp in the yielded-query responder test

* test(cross-version-wire): stub yielded-query runtime methods on the host runtime

Refs #32944

src/main/runtime/rpc/methods/terminal/terminal-multiplex-cleanup.ts

@@ -1,4 +1,5 @@
 import type { TerminalMultiplexConnection } from './terminal-multiplex-connection'
+import { answerUndeliveredYieldedQueries } from './terminal-multiplex-yielded-queries'
 
 export function installMultiplexCleanup(state: TerminalMultiplexConnection): void {
   const { runtime, streams, pendingPtyWaitControllers, emit, signal } = state
@@ -17,6 +18,7 @@ export function installMultiplexCleanup(state: TerminalMultiplexConnection): voi
     stream.ackRecoverySnapshotInFlight = false
     stream.ackRecoveryHistoryOwed = false
     stream.unsubscribeData()
+    answerUndeliveredYieldedQueries(runtime, stream)
     stream.unsubscribeResize()
     stream.unsubscribeFit()
     stream.unsubscribeDriver()

src/main/runtime/rpc/methods/terminal/terminal-multiplex-frame-delivery.ts

@@ -7,6 +7,7 @@ import {
 } from '../../../../../shared/terminal-stream-protocol'
 import { appendAckPendingOutput } from './terminal-stream-replay'
 import { notePausedOutputDropped } from './terminal-multiplex-paused-output-drop'
+import { settleDeliveredYieldedQueries } from './terminal-multiplex-yielded-queries'
 import {
   isMultiplexStreamAttached,
   type TerminalMultiplexConnection
@@ -112,6 +113,7 @@ export function installMultiplexFrameDelivery(state: TerminalMultiplexConnection
     if (!sent) {
       return false
     }
+    settleDeliveredYieldedQueries(stream, chunk.seq)
     if (stream.ackOutput) {
       stream.ackInFlightBytes += chunk.bytes.byteLength
       state.ackTotalInFlightBytes += chunk.bytes.byteLength

src/main/runtime/rpc/methods/terminal/terminal-multiplex-slot-frames.ts

@@ -19,6 +19,7 @@ import {
 } from './terminal-snapshot-publication'
 import { updateViewportForClient } from './terminal-viewport-update'
 import { notePausedOutputDropped } from './terminal-multiplex-paused-output-drop'
+import { answerUndeliveredYieldedQueries } from './terminal-multiplex-yielded-queries'
 import {
   isMultiplexStreamAttached,
   type MultiplexSnapshotRequest,
@@ -91,6 +92,7 @@ export function installMultiplexSlotFrames(state: TerminalMultiplexConnection):
         stream.ackPendingOutput = []
         stream.ackPendingOutputBytes = 0
         stream.ackPendingOutputOverflowed = false
+        answerUndeliveredYieldedQueries(runtime, stream)
         if (discardsQueuedOutput) {
           notePausedOutputDropped(state, stream)
         }

src/main/runtime/rpc/methods/terminal/terminal-multiplex-stream-initialization.ts

@@ -2,6 +2,8 @@ import {
   TerminalStreamOpcode,
   encodeTerminalStreamJson
 } from '../../../../../shared/terminal-stream-protocol'
+import { recordYieldedTerminalQueries } from './terminal-multiplex-yielded-queries'
+import { EMPTY_TERMINAL_REPLY_QUERY_SCAN_STATE } from '../../../../../shared/terminal-reply-query-scan'
 import { iterateTerminalOutputFrameChunks } from '../../terminal-output-frame-chunks'
 import { TERMINAL_MULTIPLEX_ACK_STREAM_INITIAL_WINDOW_BYTES } from '../../../../../shared/terminal-multiplex-flow-control'
 import { createTerminalOutputBatcher } from './terminal-output-batcher'
@@ -54,6 +56,8 @@ export async function initializeMultiplexStream(
     ackPendingOutputBytes: 0,
     ackPendingOutputOverflowed: false,
     ackRecoverySnapshotInFlight: false,
+    yieldedQueries: [],
+    yieldedQueryScan: EMPTY_TERMINAL_REPLY_QUERY_SCAN_STATE,
     ackRecoveryHistoryOwed: false,
     pendingOutput: [],
     pendingOutputBytes: 0,
@@ -94,6 +98,7 @@ export async function initializeMultiplexStream(
       notePausedOutputDropped(state, stream)
       return
     }
+    recordYieldedTerminalQueries(runtime, stream, data, meta)
     if (stream.buffering) {
       appendPendingMultiplexOutput(stream, data, meta)
       return

src/main/runtime/rpc/methods/terminal/terminal-multiplex-yielded-queries.ts

@@ -0,0 +1,74 @@
+import { isStatelessRendererReplyCsiQuery } from '../../../../../shared/terminal-reply-query-extraction'
+import {
+  EMPTY_TERMINAL_REPLY_QUERY_SCAN_STATE,
+  scanTerminalReplyQuerySequences
+} from '../../../../../shared/terminal-reply-query-scan'
+import type { RuntimeTerminalDataMeta } from '../../../orca-runtime'
+import type { TerminalMultiplexConnection } from './terminal-multiplex-connection'
+import type { TerminalMultiplexStream } from './terminal-stream-types'
+
+// Why: half of fish's 10s DA1 wait (fish-shell src/input/decode.rs LONG_READ_TIMEOUT); older replies risk landing at a prompt as junk.
+const YIELDED_QUERY_MAX_AGE_MS = 5_000
+const YIELDED_QUERY_MAX_ENTRIES = 16
+
+export type YieldedTerminalQuery = { seq: number; data: string; at: number }
+
+/** Record stateless queries main yielded solely to this stream at ingestion, so a later discard can hand them back. */
+export function recordYieldedTerminalQueries(
+  runtime: TerminalMultiplexConnection['runtime'],
+  stream: TerminalMultiplexStream,
+  data: string,
+  meta: RuntimeTerminalDataMeta | undefined
+): void {
+  if (
+    stream.isMobile ||
+    typeof meta?.seq !== 'number' ||
+    !runtime.isSoleRemoteTerminalQueryOwner(stream.ptyId)
+  ) {
+    stream.yieldedQueryScan = EMPTY_TERMINAL_REPLY_QUERY_SCAN_STATE
+    return
+  }
+  // Why: frames split large events with per-slice seqs, so settle by each query's own end seq; transformed text can't map offsets to seq.
+  const mapsToSeq = (meta.rawLength ?? data.length) === data.length
+  const { queries, state } = scanTerminalReplyQuerySequences(
+    data,
+    meta.seq - data.length,
+    mapsToSeq ? stream.yieldedQueryScan : EMPTY_TERMINAL_REPLY_QUERY_SCAN_STATE
+  )
+  stream.yieldedQueryScan = mapsToSeq ? state : EMPTY_TERMINAL_REPLY_QUERY_SCAN_STATE
+  const at = Date.now()
+  for (const query of queries) {
+    if (query.data.startsWith('\x1b[') && isStatelessRendererReplyCsiQuery(query.data)) {
+      stream.yieldedQueries.push({ seq: mapsToSeq ? query.endSeq : meta.seq, data: query.data, at })
+    }
+  }
+  if (stream.yieldedQueries.length > YIELDED_QUERY_MAX_ENTRIES) {
+    stream.yieldedQueries.splice(0, stream.yieldedQueries.length - YIELDED_QUERY_MAX_ENTRIES)
+  }
+}
+
+export function settleDeliveredYieldedQueries(
+  stream: TerminalMultiplexStream,
+  deliveredSeq: number | undefined
+): void {
+  if (typeof deliveredSeq === 'number' && stream.yieldedQueries.length > 0) {
+    stream.yieldedQueries = stream.yieldedQueries.filter((query) => query.seq > deliveredSeq)
+  }
+}
+
+/** Main answers recent yielded queries the view will now never receive; each was owned by this stream alone. */
+export function answerUndeliveredYieldedQueries(
+  runtime: TerminalMultiplexConnection['runtime'],
+  stream: TerminalMultiplexStream
+): void {
+  const cutoff = Date.now() - YIELDED_QUERY_MAX_AGE_MS
+  const data = stream.yieldedQueries
+    .filter((query) => query.at >= cutoff)
+    .map((query) => query.data)
+    .join('')
+  stream.yieldedQueries = []
+  stream.yieldedQueryScan = EMPTY_TERMINAL_REPLY_QUERY_SCAN_STATE
+  if (data) {
+    runtime.answerYieldedTerminalQueries(stream.ptyId, data)
+  }
+}

src/main/runtime/rpc/methods/terminal/terminal-stream-types.ts

@@ -5,6 +5,8 @@ import type {
   TerminalOutputMeta
 } from '../../terminal-output-frame-chunks'
 import type { TerminalOutputBatcher } from './terminal-output-batcher'
+import type { TerminalReplyQueryScanState } from '../../../../../shared/terminal-reply-query-scan'
+import type { YieldedTerminalQuery } from './terminal-multiplex-yielded-queries'
 
 export type SnapshotFrameOptions = {
   kind: 'scrollback' | 'resized'
@@ -83,6 +85,8 @@ export type TerminalMultiplexStream = {
   ackPendingOutputBytes: number
   ackPendingOutputOverflowed: boolean
   ackRecoverySnapshotInFlight: boolean
+  yieldedQueries: YieldedTerminalQuery[]
+  yieldedQueryScan: TerminalReplyQueryScanState
   /** A recovery went out screen-only while the link was saturated; history follows once it drains. */
   ackRecoveryHistoryOwed: boolean
   pendingOutput: TerminalOutputChunk[]

src/main/runtime/rpc/terminal-multiplex-paused-output-drop.test.ts

@@ -8,10 +8,17 @@ import {
   encodeTerminalStreamFrame,
   encodeTerminalStreamJson
 } from '../../../shared/terminal-stream-protocol'
+import {
+  TERMINAL_MULTIPLEX_ACK_STREAM_INITIAL_WINDOW_BYTES,
+  TERMINAL_STREAM_CHUNK_BYTES
+} from '../../../shared/terminal-multiplex-flow-control'
 import { startDesktopMultiplexSubscribe } from './terminal-multiplex-test-harness'
 import { RuntimeTerminalViewSubscribers } from '../runtime-terminal-view-subscribers'
 
-async function subscribePausable(capabilities: Record<string, 1>) {
+async function subscribePausable(
+  capabilities: Record<string, 1>,
+  answerYieldedTerminalQueries = vi.fn()
+) {
   const listeners: ((data: string, meta?: RuntimeTerminalDataMeta) => void)[] = []
   const viewSubscribers = new RuntimeTerminalViewSubscribers({
     notifyPresenceChanged: () => {},
@@ -25,7 +32,10 @@ async function subscribePausable(capabilities: Record<string, 1>) {
       return vi.fn()
     }),
     registerRemoteTerminalViewSubscriber: (ptyId, isOutputPaused) =>
-      viewSubscribers.registerRemote(ptyId, isOutputPaused)
+      viewSubscribers.registerRemote(ptyId, isOutputPaused),
+    isSoleRemoteTerminalQueryOwner: (ptyId) =>
+      viewSubscribers.remoteReceivingOutputCount(ptyId) === 1,
+    answerYieldedTerminalQueries
   })
   await vi.waitFor(() => expect(harness.handlers.has(0)).toBe(true))
   harness.handlers.get(0)?.(
@@ -89,6 +99,7 @@ async function subscribePausable(capabilities: Record<string, 1>) {
     dropNotices,
     outputText,
     remoteViewHoldsAuthority,
+    viewSubscribers,
     close
   }
 }
@@ -127,6 +138,59 @@ describe('terminal multiplex paused-output drop notice', () => {
     await host.close()
   })
 
+  const fillAckWindow = (host: { emitOutput: (text: string) => void }): void => {
+    for (let i = 0; i <= TERMINAL_MULTIPLEX_ACK_STREAM_INITIAL_WINDOW_BYTES / (64 * 1024); i++) {
+      host.emitOutput('x'.repeat(64 * 1024))
+    }
+  }
+
+  // #32944: output parked behind a full ACK window is discarded on pause; main must answer its queries.
+  it('hands queries in discarded ACK-pending output back to main', async () => {
+    const answerYieldedTerminalQueries = vi.fn()
+    const host = await subscribePausable(
+      { ackOutput: 1, outputPause: 1, outputPauseDropNotice: 1 },
+      answerYieldedTerminalQueries
+    )
+    fillAckWindow(host)
+    host.emitOutput('prompt\x1b[')
+    host.emitOutput('c')
+    host.setPaused(true)
+    expect(host.outputText()).not.toContain('\x1b[c')
+    expect(answerYieldedTerminalQueries).toHaveBeenCalledWith('pty-1', '\x1b[c')
+    await host.close()
+  })
+
+  it('does not re-answer a query whose sub-frame of a large event was delivered', async () => {
+    const answerYieldedTerminalQueries = vi.fn()
+    const host = await subscribePausable(
+      { ackOutput: 1, outputPause: 1, outputPauseDropNotice: 1 },
+      answerYieldedTerminalQueries
+    )
+    for (let i = 1; i < TERMINAL_MULTIPLEX_ACK_STREAM_INITIAL_WINDOW_BYTES / (64 * 1024); i++) {
+      host.emitOutput('x'.repeat(64 * 1024))
+    }
+    host.emitOutput(`prompt\x1b[c${'y'.repeat(2 * TERMINAL_STREAM_CHUNK_BYTES)}`)
+    host.setPaused(true)
+    expect(host.outputText()).toContain('prompt\x1b[c')
+    expect(answerYieldedTerminalQueries).not.toHaveBeenCalled()
+    await host.close()
+  })
+
+  it('leaves a query to the other view that also received it at ingestion', async () => {
+    const answerYieldedTerminalQueries = vi.fn()
+    const host = await subscribePausable(
+      { ackOutput: 1, outputPause: 1, outputPauseDropNotice: 1 },
+      answerYieldedTerminalQueries
+    )
+    const releaseOtherView = host.viewSubscribers.registerRemote('pty-1', () => false)
+    fillAckWindow(host)
+    host.emitOutput('prompt\x1b[c')
+    releaseOtherView()
+    host.setPaused(true)
+    await host.close()
+    expect(answerYieldedTerminalQueries).not.toHaveBeenCalled()
+  })
+
   // #32944: the dropped output can carry a query, so main must answer it while this view is paused.
   it('yields query authority while paused and takes it back on resume and close', async () => {
     const host = await subscribePausable({ ackOutput: 1, outputPause: 1, outputPauseDropNotice: 1 })

src/main/runtime/rpc/terminal-multiplex-resync-replay-trim.test.ts

@@ -62,6 +62,7 @@ async function setupMultiplexStream(): Promise<{
     getMobileDisplayMode: vi.fn().mockReturnValue('auto'),
     getLayout: vi.fn().mockReturnValue({ seq: 1 }),
     registerRemoteTerminalViewSubscriber: vi.fn(() => () => {}),
+    isSoleRemoteTerminalQueryOwner: () => false,
     subscribeToTerminalData: vi.fn(
       (_ptyId: string, cb: (data: string, meta?: OutputMeta) => void) => {
         emitOutput = cb

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

@@ -30,6 +30,8 @@ export function stubRuntime(overrides: Partial<OrcaRuntimeService> = {}): OrcaRu
     subscribeToPtyExit: vi.fn(() => vi.fn()),
     // View registration suppresses competing query responders (terminal-query-authority.md).
     registerRemoteTerminalViewSubscriber: () => () => {},
+    isSoleRemoteTerminalQueryOwner: () => false,
+    answerYieldedTerminalQueries: vi.fn(),
     // Default guarded handle lookup to live PTY for legacy harness stubs; null/stale cases override it (#7718).
     resolveLiveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }),
     requestRendererTerminalTabMount: vi.fn().mockReturnValue(true),

src/main/runtime/runtime-terminal-view-subscribers.ts

@@ -73,16 +73,19 @@ export class RuntimeTerminalViewSubscribers {
 
   /** A remote view that will receive output now; a paused one drops it, so cannot answer its queries. */
   hasRemoteReceivingOutput(ptyId: string): boolean {
-    if (this.deps.hasMobileSubscribers(ptyId)) {
-      return true
-    }
+    return this.remoteReceivingOutputCount(ptyId) > 0
+  }
+
+  /** Mobile subscribers count once: callers only distinguish none, one, or several. */
+  remoteReceivingOutputCount(ptyId: string): number {
     let paused = 0
     for (const view of this.pausableRemoteViews.get(ptyId) ?? []) {
       if (view.isOutputPaused()) {
         paused += 1
       }
     }
-    return (this.remoteCounts.get(ptyId) ?? 0) > paused
+    const mobile = this.deps.hasMobileSubscribers(ptyId) ? 1 : 0
+    return (this.remoteCounts.get(ptyId) ?? 0) - paused + mobile
   }
 
   isKnownUnattachedLocal(ptyId: string): boolean {

src/main/runtime/terminal-query-responder.test.ts

@@ -283,6 +283,23 @@ describe('reply ownership matrix', () => {
     expect(replies).toHaveLength(2)
   })
 
+  // #32944: a paused view's discarded queries fall back to main only when that view alone owned them.
+  it('reports sole remote query ownership and answers yielded stateless queries', async () => {
+    const { runtime, replies } = createResponderRuntime()
+    markHiddenRendererPty('pty-yield')
+    const releaseA = runtime.registerRemoteTerminalViewSubscriber('pty-yield', () => false)
+    expect(runtime.isSoleRemoteTerminalQueryOwner('pty-yield')).toBe(true)
+    const releaseB = runtime.registerRemoteTerminalViewSubscriber('pty-yield', () => false)
+    expect(runtime.isSoleRemoteTerminalQueryOwner('pty-yield')).toBe(false)
+    releaseB()
+    releaseA()
+
+    runtime.onPtyData('pty-yield', 'prompt', 0)
+    runtime.answerYieldedTerminalQueries('pty-yield', '\x1b[c')
+    await settle(runtime, 'pty-yield')
+    expect(replies.map((reply) => reply.data)).toEqual(['\x1b[?1;2c'])
+  })
+
   it('keeps raw preview presence separate from terminal query authority', async () => {
     const { runtime, replies } = createResponderRuntime()
     markHiddenRendererPty('pty-preview')

src/main/runtime/terminal/output/headless-terminal-hydration.ts

@@ -148,6 +148,34 @@ export abstract class OrcaRuntimeWithMaybeHydrateHeadlessFromRenderer extends Or
     })
   }
 
+  /** True when this PTY's queries go unanswered unless its single receiving remote view answers them. */
+  isSoleRemoteTerminalQueryOwner(ptyId: string): boolean {
+    return (
+      this.terminalViewSubscribers.remoteReceivingOutputCount(ptyId) === 1 &&
+      shouldModelAnswerHiddenPtyQueries({
+        ptyId,
+        settings: this.store?.getSettings(),
+        hasRemoteViewSubscriber: false
+      })
+    )
+  }
+
+  /** Answer stateless queries a sole remote view owned at ingestion but will never receive. */
+  answerYieldedTerminalQueries(ptyId: string, statelessQueryData: string): void {
+    const state = this.headlessTerminals.get(ptyId)
+    if (!state) {
+      return
+    }
+    state.writeChain = state.writeChain
+      .then(async () => {
+        // Why drop, not retry: injecting mid-escape corrupts live parsing, and a deferred reply risks arriving after the query timed out.
+        if (state.emulator.partialEscapeTailAnsi === '') {
+          await state.emulator.write(statelessQueryData, { forwardQueryReplies: true })
+        }
+      })
+      .catch(() => {})
+  }
+
   protected trackHeadlessTerminalData(
     ptyId: string,
     data: string,

tests/e2e/cross-version-wire/host-terminal-runtime-stub.ts

@@ -113,6 +113,8 @@ export function createHostTerminalRuntimeStub(
     resolveLiveLeafForHandle: (handle: string) => (handle === terminalHandle ? { ptyId } : null),
     resolveLeafForHandle: (handle: string) => (handle === terminalHandle ? { ptyId } : null),
     registerRemoteTerminalViewSubscriber: () => () => {},
+    isSoleRemoteTerminalQueryOwner: () => false,
+    answerYieldedTerminalQueries: () => {},
     requestRendererTerminalTabMount: () => true,
     updateRemoteDesktopViewer: async () => true,
     unregisterRemoteDesktopViewer: async () => true,