diff --git a/src/main/runtime/rpc/methods/terminal/stream-schemas.ts b/src/main/runtime/rpc/methods/terminal/stream-schemas.ts index d2d562b67e7..afca8964a31 100644 --- a/src/main/runtime/rpc/methods/terminal/stream-schemas.ts +++ b/src/main/runtime/rpc/methods/terminal/stream-schemas.ts @@ -18,6 +18,7 @@ export const TerminalMultiplexSubscribeFrame = TerminalHandle.extend({ ackOutputSourceRanges: z.literal(1).optional(), desktopViewportClaims: z.literal(1).optional(), outputPause: z.literal(1).optional(), + outputPauseDropNotice: z.literal(1).optional(), writeUnavailable: z.literal(1).optional(), inputAck: z.literal(1).optional() }) 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 7e931445cf0..e673379edde 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 @@ -6,6 +6,7 @@ import { encodeTerminalStreamText } from '../../../../../shared/terminal-stream-protocol' import { appendAckPendingOutput } from './terminal-stream-replay' +import { notePausedOutputDropped } from './terminal-multiplex-paused-output-drop' import { isMultiplexStreamAttached, type TerminalMultiplexConnection @@ -121,7 +122,11 @@ export function installMultiplexFrameDelivery(state: TerminalMultiplexConnection stream: TerminalMultiplexStream, chunk: TerminalOutputFrameChunk ): void => { - if (!isMultiplexStreamAttached(state, stream) || stream.outputPaused) { + if (!isMultiplexStreamAttached(state, stream)) { + return + } + if (stream.outputPaused) { + notePausedOutputDropped(state, stream) return } if ( diff --git a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-initial-snapshot.ts b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-initial-snapshot.ts index 7b56a66ce89..f027d5498dd 100644 --- a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-initial-snapshot.ts +++ b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-initial-snapshot.ts @@ -65,6 +65,7 @@ export async function publishMultiplexInitialSnapshot( ...((stream.supportsOutputPause || stream.inputSessionId !== null) && { capabilities: { ...(stream.supportsOutputPause ? { outputPause: 1 as const } : {}), + ...(stream.supportsOutputPauseDropNotice ? { outputPauseDropNotice: 1 as const } : {}), ...(stream.inputSessionId !== null ? { inputAck: 1 as const } : {}) } }), diff --git a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-paused-output-drop.ts b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-paused-output-drop.ts new file mode 100644 index 00000000000..c485e48d244 --- /dev/null +++ b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-paused-output-drop.ts @@ -0,0 +1,16 @@ +import { TerminalStreamOpcode } from '../../../../../shared/terminal-stream-protocol' +import type { TerminalMultiplexConnection } from './terminal-multiplex-connection' +import type { TerminalMultiplexStream } from './terminal-stream-types' + +/** Tells a negotiated client that this pause discarded output, so its reveal must restore. */ +export function notePausedOutputDropped( + state: Pick, + stream: TerminalMultiplexStream +): void { + if (!stream.supportsOutputPauseDropNotice || stream.pausedOutputDropNoticeSent) { + return + } + // Why mark before send: a failed send means the link is going down, and resubscribe repaints anyway. + stream.pausedOutputDropNoticeSent = true + state.sendFrame(stream.streamId, TerminalStreamOpcode.PausedOutputDropped) +} 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 d01e8dab317..ae7b9327863 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 @@ -18,6 +18,7 @@ import { serializeBudgetedRequestedSnapshot } from './terminal-snapshot-publication' import { updateViewportForClient } from './terminal-viewport-update' +import { notePausedOutputDropped } from './terminal-multiplex-paused-output-drop' import { isMultiplexStreamAttached, type MultiplexSnapshotRequest, @@ -80,10 +81,17 @@ export function installMultiplexSlotFrames(state: TerminalMultiplexConnection): } stream.outputPaused = payload.paused if (stream.outputPaused) { + stream.pausedOutputDropNoticeSent = false + const discardsQueuedOutput = + stream.ackPendingOutput.length > 0 || stream.ackPendingOutputOverflowed + // Why: batched bytes flush into the paused gate, which reports them as dropped. stream.outputBatcher.flush() stream.ackPendingOutput = [] stream.ackPendingOutputBytes = 0 stream.ackPendingOutputOverflowed = false + if (discardsQueuedOutput) { + notePausedOutputDropped(state, stream) + } } return } 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 6c0da16cf87..48df4c9e721 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 @@ -7,6 +7,7 @@ import { TERMINAL_MULTIPLEX_ACK_STREAM_INITIAL_WINDOW_BYTES } from '../../../../ import { createTerminalOutputBatcher } from './terminal-output-batcher' import { appendPendingMultiplexOutput } from './terminal-stream-replay' import { updateViewportForClient } from './terminal-viewport-update' +import { notePausedOutputDropped } from './terminal-multiplex-paused-output-drop' import { isMultiplexStreamAttached, type MultiplexSubscribeRequest, @@ -33,6 +34,9 @@ export async function initializeMultiplexStream( ackInFlightBytes: 0, ackWindowBytes: TERMINAL_MULTIPLEX_ACK_STREAM_INITIAL_WINDOW_BYTES, supportsOutputPause: request.capabilities?.outputPause === 1, + supportsOutputPauseDropNotice: + request.capabilities?.outputPause === 1 && request.capabilities.outputPauseDropNotice === 1, + pausedOutputDropNoticeSent: false, supportsWriteUnavailable: request.capabilities?.writeUnavailable === 1, inputSessionId: request.capabilities?.inputAck === 1 ? (request.inputSessionId ?? null) : null, outputPaused: false, @@ -84,6 +88,7 @@ export async function initializeMultiplexStream( return } if (stream.outputPaused) { + notePausedOutputDropped(state, stream) return } if (stream.buffering) { 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 fb102dd7666..8fdaccc6a93 100644 --- a/src/main/runtime/rpc/methods/terminal/terminal-stream-types.ts +++ b/src/main/runtime/rpc/methods/terminal/terminal-stream-types.ts @@ -63,6 +63,9 @@ export type TerminalMultiplexStream = { ackInFlightBytes: number ackWindowBytes: number supportsOutputPause: boolean + // Why: lets the client skip the reveal snapshot when a pause discarded nothing. + supportsOutputPauseDropNotice: boolean + pausedOutputDropNoticeSent: boolean supportsWriteUnavailable: boolean // Set only when the client negotiated `inputAck`; sequenced Input frames are then deduped and acked. inputSessionId: string | null 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 new file mode 100644 index 00000000000..361404e4d42 --- /dev/null +++ b/src/main/runtime/rpc/terminal-multiplex-paused-output-drop.test.ts @@ -0,0 +1,112 @@ +import './unused-default-rpc-methods.test-fixture' +import { describe, expect, it, vi } from 'vitest' +import type { RuntimeTerminalDataMeta } from '../orca-runtime' +import { + TerminalStreamOpcode, + decodeTerminalStreamFrame, + encodeTerminalStreamFrame, + encodeTerminalStreamJson +} from '../../../shared/terminal-stream-protocol' +import { startDesktopMultiplexSubscribe } from './terminal-multiplex-test-harness' + +async function subscribePausable(capabilities: Record) { + const listeners: ((data: string, meta?: RuntimeTerminalDataMeta) => void)[] = [] + const harness = startDesktopMultiplexSubscribe({ + subscribeToTerminalData: vi.fn((_ptyId, listener) => { + listeners.push(listener) + return vi.fn() + }) + }) + await vi.waitFor(() => expect(harness.handlers.has(0)).toBe(true)) + harness.handlers.get(0)?.( + decodeTerminalStreamFrame( + encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.Subscribe, + streamId: 0, + seq: 1, + payload: encodeTerminalStreamJson({ + streamId: 1, + terminal: 'terminal-1', + client: { id: 'desktop-1', type: 'desktop' }, + capabilities + }) + }) + )! + ) + const subscribedEvent = () => + harness.messages + .map((message) => JSON.parse(message).result) + .find((event) => event?.type === 'subscribed' && event.streamId === 1) + await vi.waitFor(() => expect(subscribedEvent()).toBeDefined()) + let seq = 0 + let frameSeq = 10 + const setPaused = (paused: boolean): void => + harness.handlers.get(1)?.( + decodeTerminalStreamFrame( + encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.SetOutputPaused, + streamId: 1, + seq: frameSeq++, + payload: encodeTerminalStreamJson({ paused }) + }) + )! + ) + const emitOutput = (text: string): void => { + seq += text.length + listeners[0]?.(text, { seq, rawLength: text.length }) + } + const dropNotices = (): number => + harness.binaryFrames.filter( + (bytes) => + decodeTerminalStreamFrame(bytes)?.opcode === TerminalStreamOpcode.PausedOutputDropped + ).length + const close = async (): Promise => { + harness.registry.cleanupSubscription('terminal-multiplex:conn-desktop-first-paint') + await harness.dispatchPromise + } + return { subscribed: subscribedEvent(), setPaused, emitOutput, dropNotices, close } +} + +describe('terminal multiplex paused-output drop notice', () => { + it('announces a pause only when it actually discarded output', async () => { + const host = await subscribePausable({ ackOutput: 1, outputPause: 1, outputPauseDropNotice: 1 }) + expect(host.subscribed.capabilities).toMatchObject({ outputPause: 1, outputPauseDropNotice: 1 }) + + host.setPaused(true) + host.setPaused(false) + expect(host.dropNotices()).toBe(0) + + host.setPaused(true) + host.emitOutput('hidden-1') + host.emitOutput('hidden-2') + expect(host.dropNotices()).toBe(1) + host.setPaused(false) + host.emitOutput('visible') + expect(host.dropNotices()).toBe(1) + + // Each pause reports independently. + host.setPaused(true) + host.emitOutput('hidden-3') + expect(host.dropNotices()).toBe(2) + await host.close() + }) + + it('announces output still batched when the pause lands', async () => { + const host = await subscribePausable({ ackOutput: 1, outputPause: 1, outputPauseDropNotice: 1 }) + host.emitOutput('batched-before-pause') + host.setPaused(true) + expect(host.dropNotices()).toBe(1) + await host.close() + }) + + it('never sends the frame to a client that did not negotiate it', async () => { + const host = await subscribePausable({ ackOutput: 1, outputPause: 1 }) + expect(host.subscribed.capabilities?.outputPauseDropNotice).toBeUndefined() + + host.setPaused(true) + host.emitOutput('hidden') + host.setPaused(false) + expect(host.dropNotices()).toBe(0) + await host.close() + }) +}) diff --git a/src/renderer/src/components/terminal-pane/pty-connection-hidden-replay.test.ts b/src/renderer/src/components/terminal-pane/pty-connection-hidden-replay.test.ts index ad9d2cd240f..66c177b6a90 100644 --- a/src/renderer/src/components/terminal-pane/pty-connection-hidden-replay.test.ts +++ b/src/renderer/src/components/terminal-pane/pty-connection-hidden-replay.test.ts @@ -9,6 +9,7 @@ import { registerPtyConnectionHiddenTuiSnapshotReplayTests } from './pty-connect import { registerPtyConnectionStalledHiddenRestoreTests } from './pty-connection-stalled-hidden-restore.spec' import { registerPtyConnectionDaemonSnapshotReplayTests } from './pty-connection-daemon-snapshot-replay.spec' import { registerPtyConnectionDeferredReattachLiveOutputTests } from './pty-connection-deferred-reattach-live-output.spec' +import { registerPtyConnectionRemotePauseDropNoticeTests } from './pty-connection-remote-pause-drop-notice.spec' describe('pty-connection-hidden-atlas-recovery', registerPtyConnectionHiddenAtlasRecoveryTests) describe( @@ -38,3 +39,4 @@ describe( 'pty-connection-deferred-reattach-live-output', registerPtyConnectionDeferredReattachLiveOutputTests ) +describe('pty-connection-remote-pause-drop-notice', registerPtyConnectionRemotePauseDropNoticeTests) diff --git a/src/renderer/src/components/terminal-pane/pty-connection-remote-pause-drop-notice.spec.ts b/src/renderer/src/components/terminal-pane/pty-connection-remote-pause-drop-notice.spec.ts new file mode 100644 index 00000000000..40307366d2e --- /dev/null +++ b/src/renderer/src/components/terminal-pane/pty-connection-remote-pause-drop-notice.spec.ts @@ -0,0 +1,144 @@ +import { bindPtyConnectionTestMocks } from './pty-connection-mocks.spec' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { flushAsyncTicks } from './pty-connection-test-async' +import { + createMockTransport, + createPane, + createManager, + type ConnectCallbacks, + type MockTransport +} from './pty-connection-test-pane-fixtures' +import { buildPaneConnectionDeps } from './pty-connection-test-deps' +import { createInitialStoreState } from './pty-connection-test-store-fixtures' +import type { StoreState } from './pty-connection-test-store-state' +import { + installTerminalTestGlobals, + restoreTerminalTestGlobals +} from './pty-connection-test-environment' + +// Why: switching back to a remote pane cost a 5000-row snapshot even when the host dropped nothing (#24863). +export function registerPtyConnectionRemotePauseDropNoticeTests(): void { + let mockStoreState: StoreState + let transportFactoryQueue: MockTransport[] = [] + let createdTransportOptions: Record[] = [] + let storeSubscribers: ((state: StoreState) => void)[] = [] + + describe('connectPanePty remote pause drop notice', () => { + beforeEach(async () => { + vi.resetModules() + vi.clearAllMocks() + transportFactoryQueue = [] + createdTransportOptions = [] + storeSubscribers = [] + mockStoreState = createInitialStoreState(() => mockStoreState) + bindPtyConnectionTestMocks({ + getState: () => mockStoreState, + getTransportQueue: () => transportFactoryQueue, + getTransportOptions: () => createdTransportOptions, + getStoreSubscribers: () => storeSubscribers, + setStoreSubscribers: (subscribers) => { + storeSubscribers = subscribers + } + }) + await installTerminalTestGlobals() + }) + + afterEach(async () => { + await restoreTerminalTestGlobals() + }) + + async function connectRemotePane(dropsReported: boolean) { + const { connectPanePty } = await import('./pty-connection') + const remotePtyId = 'remote:env-1@@terminal-1' + const transport = createMockTransport(remotePtyId) + const callbacks: { current: ConnectCallbacks } = { current: {} } + transport.connect.mockImplementation(async (options: { callbacks?: ConnectCallbacks }) => { + callbacks.current = options.callbacks ?? {} + return remotePtyId + }) + transport.setOutputPaused = vi.fn((paused: boolean) => { + callbacks.current.onOutputPauseChanged?.(paused, true, dropsReported) + return true + }) + transport.serializeBuffer = vi.fn().mockResolvedValue({ + data: 'host snapshot\r\n', + cols: 100, + rows: 30, + seq: 10, + source: 'headless' + }) + transportFactoryQueue.push(transport) + mockStoreState.repos = [ + { id: 'repo1', connectionId: null, displayName: 'orca', executionHostId: 'runtime:env-1' } + ] + mockStoreState.worktreesByRepo.repo1[0].runtimeOwnerEnvironmentId = 'env-1' + const pane = createPane(1) + const deps = buildPaneConnectionDeps(() => mockStoreState, { + isVisibleRef: { current: true } + }) + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the fixtures model the pane, manager and deps surface connectPanePty reads. + const binding = connectPanePty(pane as never, createManager(1) as never, deps as never) + await flushAsyncTicks(6) + const setVisible = async (visible: boolean): Promise => { + deps.isVisibleRef.current = visible + binding.syncProcessTracking() + await flushAsyncTicks(20) + } + const written = (): string => + pane.terminal.write.mock.calls.map((call) => String(call[0])).join('') + return { transport, callbacks, binding, setVisible, written } + } + + it('switches back to a quiet paused pane without fetching a snapshot', async () => { + const { transport, callbacks, binding, setVisible, written } = await connectRemotePane(true) + + for (let switchIndex = 0; switchIndex < 3; switchIndex += 1) { + await setVisible(false) + expect(transport.setOutputPaused).toHaveBeenLastCalledWith(true) + await setVisible(true) + expect(transport.setOutputPaused).toHaveBeenLastCalledWith(false) + } + const live = 'live after reveal\r\n' + callbacks.current.onData?.(live, { seq: 10 + live.length, rawLength: live.length }) + await flushAsyncTicks(20) + + expect(transport.serializeBuffer).not.toHaveBeenCalled() + expect(written()).toContain(live) + binding.dispose() + }) + + it('restores exactly once when the host reports a paused drop', async () => { + const { transport, callbacks, binding, setVisible, written } = await connectRemotePane(true) + + await setVisible(false) + callbacks.current.onPausedOutputDropped?.() + await flushAsyncTicks(6) + expect(transport.serializeBuffer).not.toHaveBeenCalled() + await setVisible(true) + const live = 'live after restore\r\n' + callbacks.current.onData?.(live, { seq: 10 + live.length, rawLength: live.length }) + await flushAsyncTicks(20) + + expect(transport.serializeBuffer).toHaveBeenCalledTimes(1) + const text = written() + expect(text.split('host snapshot').length - 1).toBe(1) + expect(text.split(live).length - 1).toBe(1) + + // The next quiet switch is free again. + await setVisible(false) + await setVisible(true) + expect(transport.serializeBuffer).toHaveBeenCalledTimes(1) + binding.dispose() + }) + + it('still restores on every reveal when the host cannot report drops', async () => { + const { transport, binding, setVisible } = await connectRemotePane(false) + + await setVisible(false) + await setVisible(true) + + expect(transport.serializeBuffer).toHaveBeenCalledTimes(1) + binding.dispose() + }) + }) +} diff --git a/src/renderer/src/components/terminal-pane/pty-connection-test-pane-fixtures.ts b/src/renderer/src/components/terminal-pane/pty-connection-test-pane-fixtures.ts index 0959c996bfd..9a6ec1be734 100644 --- a/src/renderer/src/components/terminal-pane/pty-connection-test-pane-fixtures.ts +++ b/src/renderer/src/components/terminal-pane/pty-connection-test-pane-fixtures.ts @@ -21,7 +21,8 @@ export type ConnectCallbacks = { onError?: (msg: string) => void onErrorCleared?: (msg: string) => void onWriteUnavailable?: () => void - onOutputPauseChanged?: (paused: boolean, supported: boolean) => void + onOutputPauseChanged?: (paused: boolean, supported: boolean, dropsReported?: boolean) => void + onPausedOutputDropped?: () => void } export type MockTransport = { diff --git a/src/renderer/src/components/terminal-pane/pty-connection/connect-pane-pty.ts b/src/renderer/src/components/terminal-pane/pty-connection/connect-pane-pty.ts index e18f1186396..4df6a4e9caa 100644 --- a/src/renderer/src/components/terminal-pane/pty-connection/connect-pane-pty.ts +++ b/src/renderer/src/components/terminal-pane/pty-connection/connect-pane-pty.ts @@ -111,6 +111,7 @@ export function connectPanePty( session.syncHiddenRendererPtyDelivery = () => {} session.releaseHiddenRendererPtyDelivery = () => {} session.handleRemoteOutputPauseChanged = () => {} + session.handleRemotePausedOutputDropped = () => {} session.handleRendererOwnedAgentStatus = () => {} session.remoteOutputGatedPtyId = null session.remoteOutputFactConsumerPtyId = null diff --git a/src/renderer/src/components/terminal-pane/pty-connection/foreground-output-refresh.ts b/src/renderer/src/components/terminal-pane/pty-connection/foreground-output-refresh.ts index fb4acf69654..662399532d5 100644 --- a/src/renderer/src/components/terminal-pane/pty-connection/foreground-output-refresh.ts +++ b/src/renderer/src/components/terminal-pane/pty-connection/foreground-output-refresh.ts @@ -102,7 +102,7 @@ export function bindForegroundOutputRefresh(session: ConnectPanePtySession): voi ) } - session.handleRemoteOutputPauseChanged = (paused, supported): void => { + session.handleRemoteOutputPauseChanged = (paused, supported, dropsReported): void => { const ptyId = session.transport.getPtyId() if (!ptyId || !isRemoteRuntimePtyId(ptyId)) { return @@ -127,7 +127,24 @@ export function bindForegroundOutputRefresh(session: ConnectPanePtySession): voi session.registerSideEffectFactConsumerForPty(ptyId, true) session.remoteOutputFactConsumerPtyId = ptyId } + // Why: a host that reports drops lets a quiet pane reveal with zero snapshot fetches; others may have dropped anything. + if (!(supported && dropsReported)) { + session.markHiddenOutputRestoreNeeded() + } + } + + session.handleRemotePausedOutputDropped = (): void => { + const ptyId = session.transport.getPtyId() + if (session.disposed || !ptyId || !isRemoteRuntimePtyId(ptyId)) { + return + } + // Why: the gap invalidates cross-chunk carry, exactly like a local drop marker. + session.transport.resetCrossChunkParserState?.() + const restoreWasInFlight = session.hiddenOutputRestoreInFlight !== null session.markHiddenOutputRestoreNeeded() + if (restoreWasInFlight) { + session.hiddenOutputRestoreFreshSnapshotNeeded = true + } } session.syncHiddenRendererPtyDelivery = (): void => { diff --git a/src/renderer/src/components/terminal-pane/pty-connection/transport-output-callbacks.ts b/src/renderer/src/components/terminal-pane/pty-connection/transport-output-callbacks.ts index 46598014fc8..6d142305be7 100644 --- a/src/renderer/src/components/terminal-pane/pty-connection/transport-output-callbacks.ts +++ b/src/renderer/src/components/terminal-pane/pty-connection/transport-output-callbacks.ts @@ -95,9 +95,18 @@ export function bindCaptureTransportOutputCallbacks(session: ConnectPanePtySessi session.deps.onPtyRecoveryStateRef?.current?.(session.pane.id, state) } }, - onOutputPauseChanged: (paused: boolean, supported: boolean): void => { + onOutputPauseChanged: ( + paused: boolean, + supported: boolean, + dropsReported?: boolean + ): void => { if (isCurrent()) { - session.handleRemoteOutputPauseChanged(paused, supported) + session.handleRemoteOutputPauseChanged(paused, supported, dropsReported === true) + } + }, + onPausedOutputDropped: (): void => { + if (isCurrent()) { + session.handleRemotePausedOutputDropped() } }, onSpawnDeferred: (reason: SpawnDeferralReason): void => { diff --git a/src/renderer/src/components/terminal-pane/pty-transport-types.ts b/src/renderer/src/components/terminal-pane/pty-transport-types.ts index 956e9ebad30..f81f4c3e332 100644 --- a/src/renderer/src/components/terminal-pane/pty-transport-types.ts +++ b/src/renderer/src/components/terminal-pane/pty-transport-types.ts @@ -142,7 +142,9 @@ type PtyCallbacks = { onExit?: (code: number) => void onWriteUnavailable?: () => void onRecoveryStateChange?: (state: PtyTransportRecoveryState) => void - onOutputPauseChanged?: (paused: boolean, supported: boolean) => void + /** `dropsReported`: the host sends onPausedOutputDropped, so a pause without one lost nothing. */ + onOutputPauseChanged?: (paused: boolean, supported: boolean, dropsReported?: boolean) => void + onPausedOutputDropped?: () => void /** The host kept this restore asleep; no process started and the resume handle is intact. */ onSpawnDeferred?: (reason: SpawnDeferralReason) => void } diff --git a/src/renderer/src/components/terminal-pane/remote-runtime-pty-transport-attach-subscription.test.ts b/src/renderer/src/components/terminal-pane/remote-runtime-pty-transport-attach-subscription.test.ts index 5982b63a897..55105015ce3 100644 --- a/src/renderer/src/components/terminal-pane/remote-runtime-pty-transport-attach-subscription.test.ts +++ b/src/renderer/src/components/terminal-pane/remote-runtime-pty-transport-attach-subscription.test.ts @@ -77,6 +77,7 @@ describe('createRemoteRuntimePtyTransport', () => { ackOutputSourceRanges: 1, desktopViewportClaims: 1, outputPause: 1, + outputPauseDropNotice: 1, writeUnavailable: 1, inputAck: 1 }) diff --git a/src/renderer/src/components/terminal-pane/remote-runtime-pty-transport-stream-reconnect.test.ts b/src/renderer/src/components/terminal-pane/remote-runtime-pty-transport-stream-reconnect.test.ts index a0dc82f3b5a..100390c9387 100644 --- a/src/renderer/src/components/terminal-pane/remote-runtime-pty-transport-stream-reconnect.test.ts +++ b/src/renderer/src/components/terminal-pane/remote-runtime-pty-transport-stream-reconnect.test.ts @@ -247,7 +247,7 @@ describe('createRemoteRuntimePtyTransport', () => { 'RECONNECT_SNAPSHOT' ]) expect(onData.mock.calls.map((call) => call[0])).toEqual(['LIVE_AFTER_RECONNECT']) - expect(onOutputPauseChanged).toHaveBeenLastCalledWith(false, true) + expect(onOutputPauseChanged).toHaveBeenLastCalledWith(false, true, false) transport.destroy?.() }) diff --git a/src/renderer/src/components/terminal-pane/remote-runtime-pty-transport.ts b/src/renderer/src/components/terminal-pane/remote-runtime-pty-transport.ts index f96931ce509..f36d365147c 100644 --- a/src/renderer/src/components/terminal-pane/remote-runtime-pty-transport.ts +++ b/src/renderer/src/components/terminal-pane/remote-runtime-pty-transport.ts @@ -2414,10 +2414,16 @@ export function createRemoteRuntimePtyTransport( if (isCurrentSubscription()) { storedCallbacks.onOutputPauseChanged?.( desiredOutputPaused, - nextStream.setOutputPaused(desiredOutputPaused) + nextStream.setOutputPaused(desiredOutputPaused), + nextStream.reportsPausedOutputDrops?.() === true ) } }, + onPausedOutputDropped: () => { + if (isCurrentSubscription()) { + storedCallbacks.onPausedOutputDropped?.() + } + }, onSubscribeAcknowledged: () => { if (isCurrentSubscription()) { inputReadySubscription = generation @@ -2430,7 +2436,8 @@ export function createRemoteRuntimePtyTransport( } storedCallbacks.onOutputPauseChanged?.( desiredOutputPaused, - nextStream.setOutputPaused(desiredOutputPaused) + nextStream.setOutputPaused(desiredOutputPaused), + nextStream.reportsPausedOutputDrops?.() === true ) if (!subscriptionAttached && sameHandleEndRecovery) { recordSameHandleEndReuse(subscribedHandle) @@ -3129,8 +3136,13 @@ export function createRemoteRuntimePtyTransport( if (!connected || !handle) { return false } - const supported = getCurrentMultiplexedStream(handle)?.setOutputPaused(paused) === true - storedCallbacks.onOutputPauseChanged?.(paused, supported) + const stream = getCurrentMultiplexedStream(handle) + const supported = stream?.setOutputPaused(paused) === true + storedCallbacks.onOutputPauseChanged?.( + paused, + supported, + stream?.reportsPausedOutputDrops?.() === true + ) return supported }, diff --git a/src/renderer/src/runtime/remote-runtime-terminal-binary-controller.ts b/src/renderer/src/runtime/remote-runtime-terminal-binary-controller.ts index 14d5b7d3d91..b182e6329a2 100644 --- a/src/renderer/src/runtime/remote-runtime-terminal-binary-controller.ts +++ b/src/renderer/src/runtime/remote-runtime-terminal-binary-controller.ts @@ -47,6 +47,14 @@ export abstract class RemoteRuntimeTerminalBinaryController extends RemoteRuntim stream.callbacks.onWriteUnavailable?.() return } + if (frame.opcode === TerminalStreamOpcode.PausedOutputDropped) { + if (!stream.supportsPausedOutputDropNotice) { + return + } + stream.pausedOutputGapExpected = true + stream.callbacks.onPausedOutputDropped?.() + return + } if (frame.opcode === TerminalStreamOpcode.InputAck) { if (!stream.supportsInputAck) { return @@ -111,11 +119,16 @@ export abstract class RemoteRuntimeTerminalBinaryController extends RemoteRuntim // Why: keep painting while recovery is in flight — gating froze the pane for // the whole round trip (#18488); the snapshot replaces the screen when it lands. // expectedSeq stays at the gap so a recovery that never lands can still be retried. - if (!resyncInFlight && this.detectOutputGap(stream, seq, rawLength)) { + if ( + !resyncInFlight && + !stream.pausedOutputGapExpected && + this.detectOutputGap(stream, seq, rawLength) + ) { this.requestResyncSnapshot(stream) return } if (typeof seq === 'number' && !resyncInFlight) { + stream.pausedOutputGapExpected = false stream.expectedSeq = seq stream.commandProbeBaselineSeq = undefined } diff --git a/src/renderer/src/runtime/remote-runtime-terminal-multiplexer-implementation.ts b/src/renderer/src/runtime/remote-runtime-terminal-multiplexer-implementation.ts index 8ad2f2759d9..5cf67d463ee 100644 --- a/src/renderer/src/runtime/remote-runtime-terminal-multiplexer-implementation.ts +++ b/src/renderer/src/runtime/remote-runtime-terminal-multiplexer-implementation.ts @@ -34,6 +34,8 @@ export class RemoteRuntimeTerminalMultiplexer extends RemoteRuntimeTerminalBinar acknowledgeOutput: true, acknowledgeOutputSourceRanges: false, supportsOutputPause: false, + supportsPausedOutputDropNotice: false, + pausedOutputGapExpected: false, supportsInputAck: false, inputLedgerId: null, outputPaused: false, @@ -117,6 +119,8 @@ export class RemoteRuntimeTerminalMultiplexer extends RemoteRuntimeTerminalBinar return claimed && resized }, setOutputPaused: (paused) => this.setOutputPaused(state, paused), + reportsPausedOutputDrops: () => + this.isRegisteredStream(state) && state.supportsPausedOutputDropNotice, serializeBuffer: (opts) => this.requestSnapshot(state, opts), serializeBufferOutcome: (opts) => this.requestSnapshotOutcome(state, opts), close: () => { @@ -147,6 +151,7 @@ export class RemoteRuntimeTerminalMultiplexer extends RemoteRuntimeTerminalBinar ackOutput: 1, ackOutputSourceRanges: 1, outputPause: 1, + outputPauseDropNotice: 1, writeUnavailable: 1, ...(args.client.type === 'desktop' ? { desktopViewportClaims: 1 } : {}), ...(args.inputSessionId ? { inputAck: 1 } : {}) diff --git a/src/renderer/src/runtime/remote-runtime-terminal-multiplexer-types.ts b/src/renderer/src/runtime/remote-runtime-terminal-multiplexer-types.ts index 9961fafc844..219e1f4e490 100644 --- a/src/renderer/src/runtime/remote-runtime-terminal-multiplexer-types.ts +++ b/src/renderer/src/runtime/remote-runtime-terminal-multiplexer-types.ts @@ -14,7 +14,12 @@ export type TerminalMultiplexEvent = type: 'subscribed' streamId: number streamGeneration?: string - capabilities?: { ackOutputSourceRanges?: 1; outputPause?: 1; inputAck?: 1 } + capabilities?: { + ackOutputSourceRanges?: 1 + outputPause?: 1 + outputPauseDropNotice?: 1 + inputAck?: 1 + } inputLedgerId?: string } | { type: 'end'; streamId: number; verdict?: TerminalStreamEndVerdict } @@ -55,6 +60,8 @@ export type RemoteRuntimeMultiplexedTerminalCallbacks = { /** The host installed the stream and named its capabilities; its initial snapshot is still on the way. */ onSubscribeAcknowledged?: () => void onOutputPauseCapability?: () => void + /** The host discarded output while this stream was paused (negotiated streams only). */ + onPausedOutputDropped?: () => void onEnd?: (verdict: TerminalStreamEndVerdict) => void onError?: (message: string) => void onFitOverrideChanged?: (event: { @@ -148,6 +155,8 @@ export type RemoteRuntimeMultiplexedTerminal = { resize: (cols: number, rows: number) => boolean claimViewport: (cols: number, rows: number) => boolean setOutputPaused: (paused: boolean) => boolean + // Why: only then does a pause with no drop notice prove nothing was lost. + reportsPausedOutputDrops?: () => boolean serializeBuffer: (opts?: { scrollbackRows?: number }) => Promise @@ -166,6 +175,9 @@ export type RemoteRuntimeMultiplexedTerminalState = { acknowledgeOutput: boolean acknowledgeOutputSourceRanges: boolean supportsOutputPause: boolean + supportsPausedOutputDropNotice: boolean + // Why: the host announced the seq jump the pane's own restore covers, so it must not also trigger a resync. + pausedOutputGapExpected: boolean supportsInputAck: boolean inputLedgerId: string | null outputPaused: boolean diff --git a/src/renderer/src/runtime/remote-runtime-terminal-paused-output-drop.test.ts b/src/renderer/src/runtime/remote-runtime-terminal-paused-output-drop.test.ts new file mode 100644 index 00000000000..9a5ca73cd83 --- /dev/null +++ b/src/renderer/src/runtime/remote-runtime-terminal-paused-output-drop.test.ts @@ -0,0 +1,137 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest' +import { + TerminalStreamOpcode, + decodeTerminalStreamFrame, + decodeTerminalStreamJson, + encodeTerminalStreamFrame, + encodeTerminalStreamJson, + encodeTerminalStreamText +} from '../../../shared/terminal-stream-protocol' +import { + getRemoteRuntimeTerminalMultiplexer, + resetRemoteRuntimeTerminalMultiplexersForTests +} from './remote-runtime-terminal-multiplexer' +import { replaceRuntimeEnvironmentRevisions } from './runtime-environment-revision' +import { stubRuntimeEnvironmentBridge } from './runtime-environment-bridge-fixture' + +type SubscribeCallbacks = { + onResponse: (response: unknown) => void + onBinary?: (bytes: Uint8Array) => void +} + +type HostCapabilities = { outputPause?: 1; outputPauseDropNotice?: 1 } + +describe('remote terminal paused-output drop notice', () => { + let callbacks: SubscribeCallbacks + let streamId = 0 + let cursor = 0 + let hostCapabilities: HostCapabilities + let subscribeCapabilities: Record | undefined + let snapshotRequests: { requestId?: number }[] + + const send = (opcode: TerminalStreamOpcode, payload: Uint8Array, seq = 0): void => + callbacks.onBinary?.(encodeTerminalStreamFrame({ opcode, streamId, seq, payload })) + const output = (text: string): void => { + cursor += text.length + send(TerminalStreamOpcode.Output, encodeTerminalStreamText(text), cursor) + } + + beforeEach(() => { + vi.clearAllMocks() + resetRemoteRuntimeTerminalMultiplexersForTests() + replaceRuntimeEnvironmentRevisions([]) + cursor = 0 + snapshotRequests = [] + const subscribe = vi.fn(async (_args: unknown, next: SubscribeCallbacks) => { + callbacks = next + queueMicrotask(() => next.onResponse({ ok: true, result: { type: 'ready' } })) + return { + unsubscribe: vi.fn(), + sendBinary: (bytes: Uint8Array) => { + const frame = decodeTerminalStreamFrame(bytes) + if (frame?.opcode === TerminalStreamOpcode.Subscribe) { + const payload = decodeTerminalStreamJson<{ + streamId: number + capabilities?: Record + }>(frame.payload) + streamId = payload?.streamId ?? 0 + subscribeCapabilities = payload?.capabilities + next.onResponse({ + ok: true, + result: { type: 'subscribed', streamId, capabilities: hostCapabilities } + }) + send(TerminalStreamOpcode.SnapshotStart, encodeTerminalStreamJson({ seq: cursor })) + send(TerminalStreamOpcode.SnapshotEnd, new Uint8Array()) + } else if (frame?.opcode === TerminalStreamOpcode.SnapshotRequest) { + snapshotRequests.push(decodeTerminalStreamJson(frame.payload) ?? {}) + } + } + } + }) + stubRuntimeEnvironmentBridge({ subscribe }) + }) + + async function subscribeClient() { + const data: string[] = [] + const onPausedOutputDropped = vi.fn() + const stream = await getRemoteRuntimeTerminalMultiplexer('env-1').subscribeTerminal({ + terminal: 'terminal-1', + client: { id: 'desktop-1', type: 'desktop' }, + callbacks: { + onData: (chunk) => data.push(chunk), + onSnapshot: () => {}, + onPausedOutputDropped + } + }) + await Promise.resolve() + await Promise.resolve() + return { data, onPausedOutputDropped, stream } + } + + /** Paused host discards `dropped`, optionally announcing it, then resumes with `live`. */ + function pauseDropAndResume( + stream: { setOutputPaused: (paused: boolean) => boolean }, + announce: boolean + ): void { + stream.setOutputPaused(true) + cursor += 'dropped-while-hidden'.length + if (announce) { + send(TerminalStreamOpcode.PausedOutputDropped, new Uint8Array()) + } + stream.setOutputPaused(false) + output('live') + } + + it('offers the capability and treats the announced pause gap as covered, not a resync', async () => { + hostCapabilities = { outputPause: 1, outputPauseDropNotice: 1 } + const { data, onPausedOutputDropped, stream } = await subscribeClient() + expect(subscribeCapabilities).toMatchObject({ outputPause: 1, outputPauseDropNotice: 1 }) + expect(stream.reportsPausedOutputDrops?.()).toBe(true) + + output('before') + pauseDropAndResume(stream, true) + + expect(onPausedOutputDropped).toHaveBeenCalledTimes(1) + expect(data).toEqual(['before', 'live']) + expect(snapshotRequests).toEqual([]) + + // The exemption covers only that pause: a later unannounced gap still resyncs. + cursor += 5 + output('after-real-gap') + expect(snapshotRequests).toHaveLength(1) + expect(snapshotRequests[0]?.requestId).toBeUndefined() + }) + + it('keeps the resync for an older host that does not negotiate drop notices', async () => { + hostCapabilities = { outputPause: 1 } + const { data, onPausedOutputDropped, stream } = await subscribeClient() + expect(stream.reportsPausedOutputDrops?.()).toBe(false) + + // An unnegotiated frame is ignored rather than trusted. + pauseDropAndResume(stream, true) + + expect(onPausedOutputDropped).not.toHaveBeenCalled() + expect(data).toEqual([]) + expect(snapshotRequests).toHaveLength(1) + }) +}) diff --git a/src/renderer/src/runtime/remote-runtime-terminal-response-controller.ts b/src/renderer/src/runtime/remote-runtime-terminal-response-controller.ts index 8f2618430bb..6a32666c261 100644 --- a/src/renderer/src/runtime/remote-runtime-terminal-response-controller.ts +++ b/src/renderer/src/runtime/remote-runtime-terminal-response-controller.ts @@ -16,19 +16,30 @@ import { unwrapRuntimeRpcResult } from './runtime-rpc-result' type SubscribedCapabilities = { ackOutputSourceRanges: boolean outputPause: boolean + outputPauseDropNotice: boolean inputAck: boolean } // Why parse: the frame comes off the wire, so a malformed or legacy capabilities value must read as "unsupported". function parseSubscribedCapabilities(value: unknown): SubscribedCapabilities { if (typeof value !== 'object' || value === null) { - return { ackOutputSourceRanges: false, outputPause: false, inputAck: false } + return { + ackOutputSourceRanges: false, + outputPause: false, + outputPauseDropNotice: false, + inputAck: false + } } - const record: { ackOutputSourceRanges?: unknown; outputPause?: unknown; inputAck?: unknown } = - value + const record: { + ackOutputSourceRanges?: unknown + outputPause?: unknown + outputPauseDropNotice?: unknown + inputAck?: unknown + } = value return { ackOutputSourceRanges: record.ackOutputSourceRanges === 1, outputPause: record.outputPause === 1, + outputPauseDropNotice: record.outputPause === 1 && record.outputPauseDropNotice === 1, inputAck: record.inputAck === 1 } } @@ -75,6 +86,7 @@ export abstract class RemoteRuntimeTerminalResponseController extends RemoteRunt stream.streamGeneration = event.streamGeneration } stream.supportsOutputPause = capabilities.outputPause + stream.supportsPausedOutputDropNotice = capabilities.outputPauseDropNotice // Why require the ledger id: without it a runtime restart could not be told apart, and replay would run input twice. stream.inputLedgerId = capabilities.inputAck && diff --git a/src/renderer/src/runtime/runtime-terminal-stream.test.ts b/src/renderer/src/runtime/runtime-terminal-stream.test.ts index b0094dea0f2..3a3eb26e684 100644 --- a/src/renderer/src/runtime/runtime-terminal-stream.test.ts +++ b/src/renderer/src/runtime/runtime-terminal-stream.test.ts @@ -110,6 +110,7 @@ describe('remote runtime terminal data subscriptions', () => { ackOutputSourceRanges: 1, desktopViewportClaims: 1, outputPause: 1, + outputPauseDropNotice: 1, writeUnavailable: 1 }) diff --git a/src/shared/terminal-stream-protocol.ts b/src/shared/terminal-stream-protocol.ts index 0b74d7433bd..c91d937e08f 100644 --- a/src/shared/terminal-stream-protocol.ts +++ b/src/shared/terminal-stream-protocol.ts @@ -34,7 +34,9 @@ export enum TerminalStreamOpcode { // Negotiated per stream because older clients reject unknown opcodes. WriteUnavailable = 17, // Host->client; `seq` is the cumulative input sequence the host applied (see encodeTerminalInputAck). Sent only after `inputAck` negotiation. - InputAck = 18 + InputAck = 18, + // Host->client, empty payload; at most once per pause, when the host first discards output for a paused stream. Sent only after `outputPauseDropNotice` negotiation. + PausedOutputDropped = 19 } export type TerminalInputAckKind = 'applied' | 'delivery-unknown' | 'resend' @@ -165,6 +167,7 @@ function isTerminalStreamOpcode(value: number): value is TerminalStreamOpcode { value === TerminalStreamOpcode.OutputSpan || value === TerminalStreamOpcode.SetOutputPaused || value === TerminalStreamOpcode.WriteUnavailable || - value === TerminalStreamOpcode.InputAck + value === TerminalStreamOpcode.InputAck || + value === TerminalStreamOpcode.PausedOutputDropped ) } diff --git a/tests/e2e/cross-version-wire/cross-version-terminal-wire.unit.test.ts b/tests/e2e/cross-version-wire/cross-version-terminal-wire.unit.test.ts index 22efc31bd3c..52c31b48ff4 100644 --- a/tests/e2e/cross-version-wire/cross-version-terminal-wire.unit.test.ts +++ b/tests/e2e/cross-version-wire/cross-version-terminal-wire.unit.test.ts @@ -108,8 +108,9 @@ function expectWireCompatible(record: JourneyRecord): void { // The subscribe handshake still negotiates the optional output-pause opcode, // which is what keeps opcode 16 legal to send on this pairing. + // Drop notices (opcode 19) ride on pause and appear only when both peers know them. for (const event of record.subscribedEvents) { - expect(event.capabilities).toEqual({ outputPause: 1 }) + expect(event.capabilities).toMatchObject({ outputPause: 1 }) } // Input reached the process, before and after the reconnect. @@ -153,6 +154,9 @@ describe('cross-version remote terminal wire', () => { it('current client against current server completes the journey, and is the reference for a current host', () => { expectJourneyActuallyRan(currentReference) expectWireCompatible(currentReference) + for (const event of currentReference.subscribedEvents) { + expect(event.capabilities).toEqual({ outputPause: 1, outputPauseDropNotice: 1 }) + } expect(currentReference.snapshotStarts).toEqual([ expect.objectContaining({ alternateScreen: false, terminalOwner: 'shell' }), expect.objectContaining({ alternateScreen: false, terminalOwner: 'shell' }),