fix(ssh): switching back to a quiet remote terminal no longer refetches its snapshot (#29253)

* fix(ssh): switching back to a quiet remote terminal no longer refetches its snapshot

Co-Authored-By: Claude <noreply@anthropic.com>

* test(cross-version-wire): drop notice negotiates only between current peers

Co-Authored-By: Claude <noreply@anthropic.com>

---------

Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
Neil
2026-10-15 19:39:59 -07:00
committed by GitHub
co-authored by Claude
parent 82029e6908
commit 7306273d89
26 changed files with 546 additions and 19 deletions
@@ -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()
})
@@ -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 (
@@ -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 } : {})
}
}),
@@ -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<TerminalMultiplexConnection, 'sendFrame'>,
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)
}
@@ -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
}
@@ -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) {
@@ -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
@@ -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<string, 1>) {
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<void> => {
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()
})
})
@@ -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)
@@ -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<string, unknown>[] = []
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<void> => {
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()
})
})
}
@@ -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 = {
@@ -111,6 +111,7 @@ export function connectPanePty(
session.syncHiddenRendererPtyDelivery = () => {}
session.releaseHiddenRendererPtyDelivery = () => {}
session.handleRemoteOutputPauseChanged = () => {}
session.handleRemotePausedOutputDropped = () => {}
session.handleRendererOwnedAgentStatus = () => {}
session.remoteOutputGatedPtyId = null
session.remoteOutputFactConsumerPtyId = null
@@ -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 => {
@@ -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 => {
@@ -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
}
@@ -77,6 +77,7 @@ describe('createRemoteRuntimePtyTransport', () => {
ackOutputSourceRanges: 1,
desktopViewportClaims: 1,
outputPause: 1,
outputPauseDropNotice: 1,
writeUnavailable: 1,
inputAck: 1
})
@@ -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?.()
})
@@ -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
},
@@ -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
}
@@ -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 } : {})
@@ -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<RemoteRuntimeSnapshotImage | null>
@@ -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
@@ -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<ArrayBufferLike>) => 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<string, unknown> | 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<ArrayBufferLike>) => {
const frame = decodeTerminalStreamFrame(bytes)
if (frame?.opcode === TerminalStreamOpcode.Subscribe) {
const payload = decodeTerminalStreamJson<{
streamId: number
capabilities?: Record<string, unknown>
}>(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)
})
})
@@ -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 &&
@@ -110,6 +110,7 @@ describe('remote runtime terminal data subscriptions', () => {
ackOutputSourceRanges: 1,
desktopViewportClaims: 1,
outputPause: 1,
outputPauseDropNotice: 1,
writeUnavailable: 1
})
+5 -2
View File
@@ -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
)
}
@@ -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' }),