diff --git a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-cleanup.ts b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-cleanup.ts index f8f35fb8ae2..789d20e9d77 100644 --- a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-cleanup.ts +++ b/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() diff --git a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-frame-delivery.ts b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-frame-delivery.ts index 126d6ea6c85..83cb0519b8b 100644 --- a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-frame-delivery.ts +++ b/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 diff --git a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-slot-frames.ts b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-slot-frames.ts index e7202ed7dfc..aab5dd26eca 100644 --- a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-slot-frames.ts +++ b/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) } diff --git a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-stream-initialization.ts b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-stream-initialization.ts index 7e1a743bfe6..279b74dd6f9 100644 --- a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-stream-initialization.ts +++ b/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 diff --git a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-yielded-queries.ts b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-yielded-queries.ts new file mode 100644 index 00000000000..709e3a30325 --- /dev/null +++ b/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) + } +} diff --git a/src/main/runtime/rpc/methods/terminal/terminal-stream-types.ts b/src/main/runtime/rpc/methods/terminal/terminal-stream-types.ts index 12155202431..3040ad9e444 100644 --- a/src/main/runtime/rpc/methods/terminal/terminal-stream-types.ts +++ b/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[] diff --git a/src/main/runtime/rpc/terminal-multiplex-paused-output-drop.test.ts b/src/main/runtime/rpc/terminal-multiplex-paused-output-drop.test.ts index 3815112c918..52816834a44 100644 --- a/src/main/runtime/rpc/terminal-multiplex-paused-output-drop.test.ts +++ b/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) { +async function subscribePausable( + capabilities: Record, + answerYieldedTerminalQueries = vi.fn() +) { const listeners: ((data: string, meta?: RuntimeTerminalDataMeta) => void)[] = [] const viewSubscribers = new RuntimeTerminalViewSubscribers({ notifyPresenceChanged: () => {}, @@ -25,7 +32,10 @@ async function subscribePausable(capabilities: Record) { 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) { 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 }) diff --git a/src/main/runtime/rpc/terminal-multiplex-resync-replay-trim.test.ts b/src/main/runtime/rpc/terminal-multiplex-resync-replay-trim.test.ts index 973c9b7b461..5319da8ba1b 100644 --- a/src/main/runtime/rpc/terminal-multiplex-resync-replay-trim.test.ts +++ b/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 diff --git a/src/main/runtime/rpc/terminal-multiplex-test-harness.ts b/src/main/runtime/rpc/terminal-multiplex-test-harness.ts index c3e62e7de85..ef12d36f3ec 100644 --- a/src/main/runtime/rpc/terminal-multiplex-test-harness.ts +++ b/src/main/runtime/rpc/terminal-multiplex-test-harness.ts @@ -30,6 +30,8 @@ export function stubRuntime(overrides: Partial = {}): 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), diff --git a/src/main/runtime/runtime-terminal-view-subscribers.ts b/src/main/runtime/runtime-terminal-view-subscribers.ts index e1d676be0c4..2775a93def0 100644 --- a/src/main/runtime/runtime-terminal-view-subscribers.ts +++ b/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 { diff --git a/src/main/runtime/terminal-query-responder.test.ts b/src/main/runtime/terminal-query-responder.test.ts index 451bac0c5e8..5166225a9ce 100644 --- a/src/main/runtime/terminal-query-responder.test.ts +++ b/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') diff --git a/src/main/runtime/terminal/output/headless-terminal-hydration.ts b/src/main/runtime/terminal/output/headless-terminal-hydration.ts index d89e708ba64..6c8549cee89 100644 --- a/src/main/runtime/terminal/output/headless-terminal-hydration.ts +++ b/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, diff --git a/tests/e2e/cross-version-wire/host-terminal-runtime-stub.ts b/tests/e2e/cross-version-wire/host-terminal-runtime-stub.ts index 45a9ee7f9ab..8beae856ab4 100644 --- a/tests/e2e/cross-version-wire/host-terminal-runtime-stub.ts +++ b/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,