fix(remote): resync remote terminal on dropped output frames

The multiplex output path silently drops Output frames when the
websocket buffer exceeds MAX_BINARY_BUFFERED_AMOUNT (encryptedBinaryReply
returns false). The client never noticed the gap, so a congested link
(fast PTY over slow/SSH transport) permanently corrupted the remote
terminal with missing/garbled output and no error.

Output frame seq is a byte high-water, so a drop is self-describing on
the wire. The client now tracks the expected next offset and, on a gap,
discards the corrupt tail and pulls a fresh authoritative snapshot
(self-healing, no user-facing error). Server-side, seq-less Output
chunks now carry the sentinel 0 instead of the control-frame cursor so
they don't poison the client's gap detector, and sendFrame propagates
the send/drop result end-to-end.

Adds an end-to-end repro test driving the real client multiplexer
through a fake backpressured transport.

Co-authored-by: Orca <help@stably.ai>
This commit is contained in:
Jinwoo-H
2026-07-10 04:24:29 -04:00
co-authored by Orca
parent 67447fe64f
commit 1f768d3dbe
8 changed files with 554 additions and 98 deletions
+1 -1
View File
@@ -102,7 +102,7 @@ export class RpcDispatcher {
signal?: AbortSignal
clientId?: string
clientKind?: 'mobile' | 'runtime'
sendBinary?: (bytes: Uint8Array<ArrayBufferLike>) => void
sendBinary?: (bytes: Uint8Array<ArrayBufferLike>) => boolean | void
registerBinaryStreamHandler?: (
streamId: number,
handler: (frame: TerminalStreamFrame) => void
+11 -9
View File
@@ -1221,18 +1221,20 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
opcode: TerminalStreamOpcode,
payload: Uint8Array<ArrayBufferLike> = new Uint8Array(),
seq?: number
): void => {
): boolean => {
if (closed) {
return
return false
}
sendBinary(
encodeTerminalStreamFrame({
opcode,
streamId,
seq: typeof seq === 'number' ? seq : cursor++,
payload
})
// Why: Output `seq` is a byte high-water the client uses for frame-drop
// gap detection, so a seq-less Output chunk must carry the sentinel 0
// (== "no seq") rather than the cursor value that orders control frames;
// a cursor value would poison the client's expected-seq tracker.
const resolvedSeq =
typeof seq === 'number' ? seq : opcode === TerminalStreamOpcode.Output ? 0 : cursor++
const sent = sendBinary(
encodeTerminalStreamFrame({ opcode, streamId, seq: resolvedSeq, payload })
)
return sent !== false
}
const sendStreamError = (streamId: number, message: string): void => {
sendFrame(streamId, TerminalStreamOpcode.Error, encodeTerminalStreamText(message))
@@ -70,14 +70,19 @@ describe('terminal.multiplex pending-escape-tail threading (#7329)', () => {
waitForTerminal: vi.fn(() => new Promise<RuntimeTerminalWait>(() => {})),
updateDesktopViewport: vi.fn().mockResolvedValue(true)
})
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
const dispatcher = new RpcDispatcher({
runtime,
methods: TERMINAL_METHODS
})
const dispatchPromise = dispatcher.dispatchStreaming(
makeRequest('terminal.multiplex', {}),
(msg) => messages.push(msg),
{
connectionId: 'conn-1',
sendBinary: (bytes) => binaryFrames.push(bytes),
sendBinary: (bytes) => {
binaryFrames.push(bytes)
},
registerBinaryStreamHandler: (streamId, handler) => {
handlers.set(streamId, handler)
return () => handlers.delete(streamId)
+172 -45
View File
@@ -77,14 +77,19 @@ describe('terminal multiplex RPC', () => {
sendTerminal: vi.fn().mockResolvedValue({ accepted: true }),
updateDesktopViewport: vi.fn().mockResolvedValue(true)
})
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
const dispatcher = new RpcDispatcher({
runtime,
methods: TERMINAL_METHODS
})
const dispatchPromise = dispatcher.dispatchStreaming(
makeRequest('terminal.multiplex', {}),
(msg) => messages.push(msg),
{
connectionId: 'conn-1',
sendBinary: (bytes) => binaryFrames.push(bytes),
sendBinary: (bytes) => {
binaryFrames.push(bytes)
},
registerBinaryStreamHandler: (streamId, handler) => {
handlers.set(streamId, handler)
return () => handlers.delete(streamId)
@@ -199,7 +204,10 @@ describe('terminal multiplex RPC', () => {
opcode: TerminalStreamOpcode.SnapshotRequest,
streamId: 5,
seq: 4,
payload: encodeTerminalStreamJson({ requestId: 7, scrollbackRows: 5000 })
payload: encodeTerminalStreamJson({
requestId: 7,
scrollbackRows: 5000
})
})
)!
)
@@ -291,14 +299,19 @@ describe('terminal multiplex RPC', () => {
sendTerminal: vi.fn().mockResolvedValue({ accepted: true }),
updateMobileViewport: vi.fn().mockResolvedValue({ updated: true, applied: true })
})
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
const dispatcher = new RpcDispatcher({
runtime,
methods: TERMINAL_METHODS
})
const dispatchPromise = dispatcher.dispatchStreaming(
makeRequest('terminal.multiplex', {}),
(msg) => messages.push(msg),
{
connectionId: 'conn-stale-multiplex-resize',
sendBinary: (bytes) => binaryFrames.push(bytes),
sendBinary: (bytes) => {
binaryFrames.push(bytes)
},
registerBinaryStreamHandler: (streamId, handler) => {
handlers.set(streamId, handler)
return () => handlers.delete(streamId)
@@ -327,8 +340,20 @@ describe('terminal multiplex RPC', () => {
await vi.waitFor(() => expect(resizeListener).toBeDefined())
binaryFrames.splice(0)
resizeListener?.({ cols: 90, rows: 24, displayMode: 'auto', reason: 'apply-layout', seq: 2 })
resizeListener?.({ cols: 100, rows: 24, displayMode: 'auto', reason: 'apply-layout', seq: 3 })
resizeListener?.({
cols: 90,
rows: 24,
displayMode: 'auto',
reason: 'apply-layout',
seq: 2
})
resizeListener?.({
cols: 100,
rows: 24,
displayMode: 'auto',
reason: 'apply-layout',
seq: 3
})
await vi.waitFor(() => expect(restreamResolves).toHaveLength(2))
restreamResolves[1]?.({ data: 'newer', cols: 100, rows: 24 })
@@ -402,14 +427,19 @@ describe('terminal multiplex RPC', () => {
sendTerminal: vi.fn().mockResolvedValue({ accepted: true }),
updateDesktopViewport: vi.fn().mockResolvedValue(true)
})
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
const dispatcher = new RpcDispatcher({
runtime,
methods: TERMINAL_METHODS
})
const dispatchPromise = dispatcher.dispatchStreaming(
makeRequest('terminal.multiplex', {}),
(msg) => messages.push(msg),
{
connectionId: 'conn-multibyte-output-batch',
sendBinary: (bytes) => binaryFrames.push(bytes),
sendBinary: (bytes) => {
binaryFrames.push(bytes)
},
registerBinaryStreamHandler: (streamId, handler) => {
handlers.set(streamId, handler)
return () => handlers.delete(streamId)
@@ -478,9 +508,11 @@ describe('terminal multiplex RPC', () => {
const cleanups = new Map<string, () => void>()
const runtime = stubRuntime({
resolveLiveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }),
readTerminal: vi
.fn()
.mockResolvedValue({ tail: ['line 120'], truncated: false, limited: true }),
readTerminal: vi.fn().mockResolvedValue({
tail: ['line 120'],
truncated: false,
limited: true
}),
serializeTerminalBuffer: vi.fn().mockResolvedValue(null),
getTerminalSize: vi.fn().mockReturnValue({ cols: 120, rows: 40 }),
getMobileDisplayMode: vi.fn().mockReturnValue('auto'),
@@ -502,14 +534,19 @@ describe('terminal multiplex RPC', () => {
waitForTerminal: vi.fn(() => new Promise<RuntimeTerminalWait>(() => {})),
sendTerminal: vi.fn().mockResolvedValue({ accepted: true })
})
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
const dispatcher = new RpcDispatcher({
runtime,
methods: TERMINAL_METHODS
})
const dispatchPromise = dispatcher.dispatchStreaming(
makeRequest('terminal.multiplex', {}),
(msg) => messages.push(msg),
{
connectionId: 'conn-multiplex-limited',
sendBinary: (bytes) => binaryFrames.push(bytes),
sendBinary: (bytes) => {
binaryFrames.push(bytes)
},
registerBinaryStreamHandler: (streamId, handler) => {
handlers.set(streamId, handler)
return () => handlers.delete(streamId)
@@ -578,8 +615,16 @@ describe('terminal multiplex RPC', () => {
serializeTerminalBuffer: vi
.fn()
.mockResolvedValueOnce({ data: 'initial', cols: 120, rows: 40 })
.mockResolvedValueOnce({ data: 'x'.repeat(2 * 1024 * 1024 + 1), cols: 120, rows: 40 })
.mockResolvedValueOnce({ data: 'budgeted snapshot', cols: 120, rows: 40 }),
.mockResolvedValueOnce({
data: 'x'.repeat(2 * 1024 * 1024 + 1),
cols: 120,
rows: 40
})
.mockResolvedValueOnce({
data: 'budgeted snapshot',
cols: 120,
rows: 40
}),
getTerminalSize: vi.fn().mockReturnValue({ cols: 120, rows: 40 }),
getMobileDisplayMode: vi.fn().mockReturnValue('auto'),
getLayout: vi.fn().mockReturnValue({ seq: 1 }),
@@ -596,14 +641,19 @@ describe('terminal multiplex RPC', () => {
sendTerminal: vi.fn().mockResolvedValue({ accepted: true }),
updateDesktopViewport: vi.fn().mockResolvedValue(true)
})
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
const dispatcher = new RpcDispatcher({
runtime,
methods: TERMINAL_METHODS
})
const dispatchPromise = dispatcher.dispatchStreaming(
makeRequest('terminal.multiplex', {}),
(msg) => messages.push(msg),
{
connectionId: 'conn-budgeted-request',
sendBinary: (bytes) => binaryFrames.push(bytes),
sendBinary: (bytes) => {
binaryFrames.push(bytes)
},
registerBinaryStreamHandler: (streamId, handler) => {
handlers.set(streamId, handler)
return () => handlers.delete(streamId)
@@ -640,7 +690,10 @@ describe('terminal multiplex RPC', () => {
opcode: TerminalStreamOpcode.SnapshotRequest,
streamId: 14,
seq: 2,
payload: encodeTerminalStreamJson({ requestId: 55, scrollbackRows: 5000 })
payload: encodeTerminalStreamJson({
requestId: 55,
scrollbackRows: 5000
})
})
)!
)
@@ -704,7 +757,10 @@ describe('terminal multiplex RPC', () => {
sendTerminal: vi.fn().mockResolvedValue({ accepted: true }),
updateDesktopViewport: vi.fn().mockResolvedValue(true)
})
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
const dispatcher = new RpcDispatcher({
runtime,
methods: TERMINAL_METHODS
})
const dispatchPromise = dispatcher.dispatchStreaming(
makeRequest('terminal.multiplex', {}),
@@ -805,7 +861,10 @@ describe('terminal multiplex RPC', () => {
sendTerminal: vi.fn().mockResolvedValue({ accepted: true }),
updateDesktopViewport: vi.fn().mockResolvedValue(true)
})
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
const dispatcher = new RpcDispatcher({
runtime,
methods: TERMINAL_METHODS
})
const dispatchPromise = dispatcher.dispatchStreaming(
makeRequest('terminal.multiplex', {}),
@@ -892,7 +951,10 @@ describe('terminal multiplex RPC', () => {
sendTerminal: vi.fn().mockResolvedValue({ accepted: true }),
updateDesktopViewport: vi.fn().mockResolvedValue(true)
})
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
const dispatcher = new RpcDispatcher({
runtime,
methods: TERMINAL_METHODS
})
const dispatchPromise = dispatcher.dispatchStreaming(
makeRequest('terminal.subscribe', {
@@ -981,7 +1043,10 @@ describe('terminal multiplex RPC', () => {
sendTerminal: vi.fn().mockResolvedValue({ accepted: true }),
updateDesktopViewport: vi.fn().mockResolvedValue(true)
})
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
const dispatcher = new RpcDispatcher({
runtime,
methods: TERMINAL_METHODS
})
const dispatchPromise = dispatcher.dispatchStreaming(
makeRequest('terminal.subscribe', {
@@ -992,7 +1057,9 @@ describe('terminal multiplex RPC', () => {
(msg) => messages.push(msg),
{
connectionId: 'conn-subscribe-output-chunking',
sendBinary: (bytes) => binaryFrames.push(bytes),
sendBinary: (bytes) => {
binaryFrames.push(bytes)
},
registerBinaryStreamHandler: (streamId, handler) => {
handlers.set(streamId, handler)
return () => handlers.delete(streamId)
@@ -1067,7 +1134,10 @@ describe('terminal multiplex RPC', () => {
sendTerminal: vi.fn().mockResolvedValue({ accepted: true }),
updateDesktopViewport: vi.fn().mockResolvedValue(true)
})
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
const dispatcher = new RpcDispatcher({
runtime,
methods: TERMINAL_METHODS
})
const dispatchPromise = dispatcher.dispatchStreaming(
makeRequest('terminal.subscribe', {
@@ -1078,13 +1148,18 @@ describe('terminal multiplex RPC', () => {
(msg) => messages.push(msg),
{
connectionId: 'conn-buffered-output-on-subscribe',
sendBinary: (bytes) => binaryFrames.push(bytes),
sendBinary: (bytes) => {
binaryFrames.push(bytes)
},
registerBinaryStreamHandler: vi.fn(() => vi.fn())
}
)
await vi.waitFor(() => expect(dataListenerRef.current).toBeDefined())
dataListenerRef.current?.('starting shell\r\n', { seq: 16, rawLength: 16 })
dataListenerRef.current?.('starting shell\r\n', {
seq: 16,
rawLength: 16
})
resolveSnapshot({ data: '', cols: 120, rows: 40 })
await vi.waitFor(() =>
expect(messages.some((msg) => JSON.parse(msg).result?.type === 'subscribed')).toBe(true)
@@ -1125,7 +1200,12 @@ describe('terminal multiplex RPC', () => {
readTerminal: vi.fn().mockResolvedValue({ tail: [], truncated: false }),
serializeTerminalBuffer: vi.fn(
() =>
new Promise<{ data: string; cols: number; rows: number; seq: number }>((resolve) => {
new Promise<{
data: string
cols: number
rows: number
seq: number
}>((resolve) => {
resolveSnapshot = resolve
})
),
@@ -1151,7 +1231,10 @@ describe('terminal multiplex RPC', () => {
sendTerminal: vi.fn().mockResolvedValue({ accepted: true }),
updateDesktopViewport: vi.fn().mockResolvedValue(true)
})
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
const dispatcher = new RpcDispatcher({
runtime,
methods: TERMINAL_METHODS
})
const dispatchPromise = dispatcher.dispatchStreaming(
makeRequest('terminal.subscribe', {
@@ -1162,7 +1245,9 @@ describe('terminal multiplex RPC', () => {
(msg) => messages.push(msg),
{
connectionId: 'conn-buffered-output-covered-by-snapshot',
sendBinary: (bytes) => binaryFrames.push(bytes),
sendBinary: (bytes) => {
binaryFrames.push(bytes)
},
registerBinaryStreamHandler: vi.fn(() => vi.fn())
}
)
@@ -1222,7 +1307,12 @@ describe('terminal multiplex RPC', () => {
readTerminal: vi.fn().mockResolvedValue({ tail: [], truncated: false }),
serializeTerminalBuffer: vi.fn(
() =>
new Promise<{ data: string; cols: number; rows: number; seq: number }>((resolve) => {
new Promise<{
data: string
cols: number
rows: number
seq: number
}>((resolve) => {
resolveSnapshot = resolve
})
),
@@ -1248,7 +1338,10 @@ describe('terminal multiplex RPC', () => {
sendTerminal: vi.fn().mockResolvedValue({ accepted: true }),
updateDesktopViewport: vi.fn().mockResolvedValue(true)
})
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
const dispatcher = new RpcDispatcher({
runtime,
methods: TERMINAL_METHODS
})
const dispatchPromise = dispatcher.dispatchStreaming(
makeRequest('terminal.subscribe', {
@@ -1259,7 +1352,9 @@ describe('terminal multiplex RPC', () => {
(msg) => messages.push(msg),
{
connectionId: 'conn-buffered-output-partially-covered-by-snapshot',
sendBinary: (bytes) => binaryFrames.push(bytes),
sendBinary: (bytes) => {
binaryFrames.push(bytes)
},
registerBinaryStreamHandler: vi.fn(() => vi.fn())
}
)
@@ -1319,7 +1414,10 @@ describe('terminal multiplex RPC', () => {
cleanups.set(id, cleanup)
})
})
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
const dispatcher = new RpcDispatcher({
runtime,
methods: TERMINAL_METHODS
})
const dispatchPromise = dispatcher.dispatchStreaming(
makeRequest('terminal.multiplex', {}),
@@ -1327,7 +1425,9 @@ describe('terminal multiplex RPC', () => {
{
signal: controller.signal,
connectionId: 'conn-phone-multiplex',
sendBinary: (bytes) => binaryFrames.push(bytes),
sendBinary: (bytes) => {
binaryFrames.push(bytes)
},
registerBinaryStreamHandler: (streamId, handler) => {
handlers.set(streamId, handler)
return () => handlers.delete(streamId)
@@ -1393,14 +1493,19 @@ describe('terminal multiplex RPC', () => {
cleanups.set(id, cleanup)
})
})
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
const dispatcher = new RpcDispatcher({
runtime,
methods: TERMINAL_METHODS
})
const dispatchPromise = dispatcher.dispatchStreaming(
makeRequest('terminal.multiplex', {}),
(msg) => messages.push(msg),
{
connectionId: 'conn-stale-handle',
sendBinary: (bytes) => binaryFrames.push(bytes),
sendBinary: (bytes) => {
binaryFrames.push(bytes)
},
registerBinaryStreamHandler: (streamId, handler) => {
handlers.set(streamId, handler)
return () => handlers.delete(streamId)
@@ -1484,14 +1589,19 @@ describe('terminal multiplex RPC', () => {
sendTerminal: vi.fn().mockResolvedValue({ accepted: true }),
updateDesktopViewport: vi.fn().mockResolvedValue(true)
})
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
const dispatcher = new RpcDispatcher({
runtime,
methods: TERMINAL_METHODS
})
const dispatchPromise = dispatcher.dispatchStreaming(
makeRequest('terminal.multiplex', {}),
(msg) => messages.push(msg),
{
connectionId: 'conn-buffered',
sendBinary: (bytes) => binaryFrames.push(bytes),
sendBinary: (bytes) => {
binaryFrames.push(bytes)
},
registerBinaryStreamHandler: (streamId, handler) => {
handlers.set(streamId, handler)
return () => handlers.delete(streamId)
@@ -1593,14 +1703,19 @@ describe('terminal multiplex RPC', () => {
sendTerminal: vi.fn().mockResolvedValue({ accepted: true }),
updateDesktopViewport: vi.fn().mockResolvedValue(true)
})
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
const dispatcher = new RpcDispatcher({
runtime,
methods: TERMINAL_METHODS
})
const dispatchPromise = dispatcher.dispatchStreaming(
makeRequest('terminal.multiplex', {}),
(msg) => messages.push(msg),
{
connectionId: 'conn-buffered-multibyte',
sendBinary: (bytes) => binaryFrames.push(bytes),
sendBinary: (bytes) => {
binaryFrames.push(bytes)
},
registerBinaryStreamHandler: (streamId, handler) => {
handlers.set(streamId, handler)
return () => handlers.delete(streamId)
@@ -1690,7 +1805,11 @@ describe('terminal multiplex RPC', () => {
resolveRequestedSnapshot = resolve
})
)
.mockResolvedValueOnce({ data: 'retry snapshot', cols: 120, rows: 40 }),
.mockResolvedValueOnce({
data: 'retry snapshot',
cols: 120,
rows: 40
}),
getTerminalSize: vi.fn().mockReturnValue({ cols: 120, rows: 40 }),
getMobileDisplayMode: vi.fn().mockReturnValue('auto'),
getLayout: vi.fn().mockReturnValue({ seq: 1 }),
@@ -1710,14 +1829,19 @@ describe('terminal multiplex RPC', () => {
sendTerminal: vi.fn().mockResolvedValue({ accepted: true }),
updateDesktopViewport: vi.fn().mockResolvedValue(true)
})
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
const dispatcher = new RpcDispatcher({
runtime,
methods: TERMINAL_METHODS
})
const dispatchPromise = dispatcher.dispatchStreaming(
makeRequest('terminal.multiplex', {}),
(msg) => messages.push(msg),
{
connectionId: 'conn-request-overflow',
sendBinary: (bytes) => binaryFrames.push(bytes),
sendBinary: (bytes) => {
binaryFrames.push(bytes)
},
registerBinaryStreamHandler: (streamId, handler) => {
handlers.set(streamId, handler)
return () => handlers.delete(streamId)
@@ -1754,7 +1878,10 @@ describe('terminal multiplex RPC', () => {
opcode: TerminalStreamOpcode.SnapshotRequest,
streamId: 12,
seq: 2,
payload: encodeTerminalStreamJson({ requestId: 44, scrollbackRows: 5000 })
payload: encodeTerminalStreamJson({
requestId: 44,
scrollbackRows: 5000
})
})
)!
)
@@ -52,7 +52,10 @@ describe('terminal output batching', () => {
}),
waitForTerminal: vi.fn(() => new Promise<RuntimeTerminalWait>(() => {}))
})
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
const dispatcher = new RpcDispatcher({
runtime,
methods: TERMINAL_METHODS
})
const dispatchPromise = dispatcher.dispatchStreaming(
makeRequest('terminal.subscribe', {
@@ -78,7 +81,9 @@ describe('terminal output batching', () => {
.map((msg) => JSON.parse(msg))
.filter((message) => message.result?.type === 'data')
expect(dataMessages).toHaveLength(1)
expect(dataMessages[0]).toMatchObject({ result: { type: 'data', chunk: 'ab' } })
expect(dataMessages[0]).toMatchObject({
result: { type: 'data', chunk: 'ab' }
})
runtime.cleanupSubscription('terminal-1:desktop-1')
await dispatchPromise
@@ -123,7 +128,10 @@ describe('terminal output batching', () => {
sendTerminal: vi.fn().mockResolvedValue({ accepted: true }),
updateMobileViewport: vi.fn().mockResolvedValue(false)
})
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
const dispatcher = new RpcDispatcher({
runtime,
methods: TERMINAL_METHODS
})
const dispatchPromise = dispatcher.dispatchStreaming(
makeRequest('terminal.subscribe', {
@@ -134,7 +142,9 @@ describe('terminal output batching', () => {
(msg) => messages.push(msg),
{
connectionId: 'conn-1',
sendBinary: (bytes) => binaryFrames.push(bytes)
sendBinary: (bytes) => {
binaryFrames.push(bytes)
}
}
)
@@ -144,7 +154,10 @@ describe('terminal output batching', () => {
const subscribed = messages
.map((msg) => JSON.parse(msg))
.find((msg) => msg.result?.type === 'subscribed')
expect(subscribed?.result).toMatchObject({ type: 'subscribed', streamId: expect.any(Number) })
expect(subscribed?.result).toMatchObject({
type: 'subscribed',
streamId: expect.any(Number)
})
await vi.waitFor(() => expect(dataListenerRef.current).toBeDefined())
const emitData = dataListenerRef.current
@@ -206,7 +219,10 @@ describe('terminal output batching', () => {
sendTerminal: vi.fn().mockResolvedValue({ accepted: true }),
updateMobileViewport: vi.fn().mockResolvedValue(false)
})
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
const dispatcher = new RpcDispatcher({
runtime,
methods: TERMINAL_METHODS
})
const dispatchPromise = dispatcher.dispatchStreaming(
makeRequest('terminal.subscribe', {
@@ -286,7 +302,10 @@ describe('terminal output batching', () => {
sendTerminal: vi.fn().mockResolvedValue({ accepted: true }),
updateMobileViewport: vi.fn().mockResolvedValue(false)
})
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
const dispatcher = new RpcDispatcher({
runtime,
methods: TERMINAL_METHODS
})
const messages: string[] = []
const dispatchPromise = dispatcher.dispatchStreaming(
@@ -40,7 +40,10 @@ describe('terminal subscribe buffering', () => {
),
readTerminal: vi.fn().mockResolvedValue({ tail: [], truncated: false })
})
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
const dispatcher = new RpcDispatcher({
runtime,
methods: TERMINAL_METHODS
})
const dispatchPromise = dispatcher.dispatchStreaming(
makeRequest('terminal.subscribe', {
@@ -77,11 +80,16 @@ describe('terminal subscribe buffering', () => {
const messages: string[] = []
const runtime = stubRuntime({
resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: null }),
readTerminal: vi
.fn()
.mockResolvedValue({ tail: ['line 120'], truncated: false, limited: true })
readTerminal: vi.fn().mockResolvedValue({
tail: ['line 120'],
truncated: false,
limited: true
})
})
const dispatcher = new RpcDispatcher({
runtime,
methods: TERMINAL_METHODS
})
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
await dispatcher.dispatchStreaming(
makeRequest('terminal.subscribe', { terminal: 'terminal-1' }),
@@ -104,9 +112,11 @@ describe('terminal subscribe buffering', () => {
const cleanups = new Map<string, () => void>()
const runtime = stubRuntime({
resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }),
readTerminal: vi
.fn()
.mockResolvedValue({ tail: ['line 120'], truncated: false, limited: true }),
readTerminal: vi.fn().mockResolvedValue({
tail: ['line 120'],
truncated: false,
limited: true
}),
serializeTerminalBuffer: vi.fn().mockResolvedValue(null),
getTerminalSize: vi.fn().mockReturnValue({ cols: 80, rows: 24 }),
getMobileDisplayMode: vi.fn().mockReturnValue('auto'),
@@ -122,7 +132,10 @@ describe('terminal subscribe buffering', () => {
}),
waitForTerminal: vi.fn(() => new Promise<RuntimeTerminalWait>(() => {}))
})
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
const dispatcher = new RpcDispatcher({
runtime,
methods: TERMINAL_METHODS
})
const dispatchPromise = dispatcher.dispatchStreaming(
makeRequest('terminal.subscribe', {
@@ -172,7 +185,10 @@ describe('terminal subscribe buffering', () => {
cleanupSubscription: vi.fn(),
waitForTerminal: vi.fn(() => new Promise<RuntimeTerminalWait>(() => {}))
})
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
const dispatcher = new RpcDispatcher({
runtime,
methods: TERMINAL_METHODS
})
const dispatchPromise = dispatcher.dispatchStreaming(
makeRequest('terminal.subscribe', {
@@ -209,9 +225,11 @@ describe('terminal subscribe buffering', () => {
const cleanups = new Map<string, () => void>()
const runtime = stubRuntime({
resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }),
readTerminal: vi
.fn()
.mockResolvedValue({ tail: ['line 120'], truncated: false, limited: true }),
readTerminal: vi.fn().mockResolvedValue({
tail: ['line 120'],
truncated: false,
limited: true
}),
serializeTerminalBuffer: vi.fn().mockResolvedValue(null),
getTerminalSize: vi.fn().mockReturnValue({ cols: 80, rows: 24 }),
getMobileDisplayMode: vi.fn().mockReturnValue('auto'),
@@ -230,7 +248,10 @@ describe('terminal subscribe buffering', () => {
sendTerminal: vi.fn().mockResolvedValue({ accepted: true }),
updateMobileViewport: vi.fn().mockResolvedValue(false)
})
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
const dispatcher = new RpcDispatcher({
runtime,
methods: TERMINAL_METHODS
})
const dispatchPromise = dispatcher.dispatchStreaming(
makeRequest('terminal.subscribe', {
@@ -241,7 +262,9 @@ describe('terminal subscribe buffering', () => {
(msg) => messages.push(msg),
{
connectionId: 'conn-binary-limited',
sendBinary: (bytes) => binaryFrames.push(bytes)
sendBinary: (bytes) => {
binaryFrames.push(bytes)
}
}
)
@@ -273,9 +296,11 @@ describe('terminal subscribe buffering', () => {
const cleanups = new Map<string, () => void>()
const runtime = stubRuntime({
resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }),
readTerminal: vi
.fn()
.mockResolvedValue({ tail: ['line 120'], truncated: false, limited: true }),
readTerminal: vi.fn().mockResolvedValue({
tail: ['line 120'],
truncated: false,
limited: true
}),
serializeTerminalBuffer: vi.fn().mockResolvedValue({
data: 'serialized snapshot\r\n',
cols: 100,
@@ -298,7 +323,10 @@ describe('terminal subscribe buffering', () => {
sendTerminal: vi.fn().mockResolvedValue({ accepted: true }),
updateMobileViewport: vi.fn().mockResolvedValue(false)
})
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
const dispatcher = new RpcDispatcher({
runtime,
methods: TERMINAL_METHODS
})
const dispatchPromise = dispatcher.dispatchStreaming(
makeRequest('terminal.subscribe', {
@@ -309,7 +337,9 @@ describe('terminal subscribe buffering', () => {
(msg) => messages.push(msg),
{
connectionId: 'conn-binary-serialized-limited',
sendBinary: (bytes) => binaryFrames.push(bytes)
sendBinary: (bytes) => {
binaryFrames.push(bytes)
}
}
)
@@ -375,7 +405,10 @@ describe('terminal subscribe buffering', () => {
sendTerminal: vi.fn().mockResolvedValue({ accepted: true }),
updateMobileViewport: vi.fn().mockResolvedValue(false)
})
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
const dispatcher = new RpcDispatcher({
runtime,
methods: TERMINAL_METHODS
})
const dispatchPromise = dispatcher.dispatchStreaming(
makeRequest('terminal.subscribe', {
@@ -386,7 +419,9 @@ describe('terminal subscribe buffering', () => {
(msg) => messages.push(msg),
{
connectionId: 'conn-buffered',
sendBinary: (bytes) => binaryFrames.push(bytes)
sendBinary: (bytes) => {
binaryFrames.push(bytes)
}
}
)
@@ -446,7 +481,12 @@ describe('terminal subscribe buffering', () => {
data: string
cols: number
rows: number
oscLinks?: { row: number; startCol: number; endCol: number; uri: string }[]
oscLinks?: {
row: number
startCol: number
endCol: number
uri: string
}[]
}) => void)[] = []
const serializeTerminalBuffer = vi
.fn()
@@ -457,7 +497,12 @@ describe('terminal subscribe buffering', () => {
data: string
cols: number
rows: number
oscLinks?: { row: number; startCol: number; endCol: number; uri: string }[]
oscLinks?: {
row: number
startCol: number
endCol: number
uri: string
}[]
}>((resolve) => {
restreamResolves.push(resolve)
})
@@ -490,7 +535,10 @@ describe('terminal subscribe buffering', () => {
sendTerminal: vi.fn().mockResolvedValue({ accepted: true }),
updateMobileViewport: vi.fn().mockResolvedValue({ updated: true, applied: true })
})
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
const dispatcher = new RpcDispatcher({
runtime,
methods: TERMINAL_METHODS
})
const dispatchPromise = dispatcher.dispatchStreaming(
makeRequest('terminal.subscribe', {
@@ -501,19 +549,38 @@ describe('terminal subscribe buffering', () => {
(msg) => messages.push(msg),
{
connectionId: 'conn-stale-resize',
sendBinary: (bytes) => binaryFrames.push(bytes)
sendBinary: (bytes) => {
binaryFrames.push(bytes)
}
}
)
await vi.waitFor(() => expect(resizeListener).toBeDefined())
binaryFrames.splice(0)
resizeListener?.({ cols: 90, rows: 24, displayMode: 'auto', reason: 'apply-layout', seq: 2 })
resizeListener?.({ cols: 100, rows: 24, displayMode: 'auto', reason: 'apply-layout', seq: 3 })
resizeListener?.({
cols: 90,
rows: 24,
displayMode: 'auto',
reason: 'apply-layout',
seq: 2
})
resizeListener?.({
cols: 100,
rows: 24,
displayMode: 'auto',
reason: 'apply-layout',
seq: 3
})
await vi.waitFor(() => expect(restreamResolves).toHaveLength(2))
const newerOscLinks = [{ row: 0, startCol: 4, endCol: 9, uri: 'https://example.com' }]
restreamResolves[1]?.({ data: 'newer', cols: 100, rows: 24, oscLinks: newerOscLinks })
restreamResolves[1]?.({
data: 'newer',
cols: 100,
rows: 24,
oscLinks: newerOscLinks
})
await vi.waitFor(() =>
expect(
binaryFrames.some((frame) => {
@@ -0,0 +1,174 @@
import { afterEach, 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'
// Why: reproduces the silent frame-drop corruption. The server multiplex path
// drops Output frames when the websocket buffer is over its cap
// (encryptedBinaryReply returns false); the wire `seq` is a byte high-water, so
// a drop leaves a detectable gap. This harness drives the real client
// multiplexer through the same subscribe transport the app uses and forces a
// drop, asserting the client resyncs instead of rendering a corrupt tail.
type SubscribeCallbacks = {
onResponse: (response: unknown) => void
onBinary?: (bytes: Uint8Array<ArrayBufferLike>) => void
onError?: (error: { message: string }) => void
onClose?: () => void
}
/**
* Minimal server that mimics src/main/runtime/rpc/methods/terminal.ts's multiplex
* output path: Output frames carry a monotonic byte high-water `seq`, and a
* SnapshotRequest is answered with an initial-style snapshot (no requestId).
*/
class FakeMultiplexServer {
private cursorBytes = 0
private streamId = 0
dropNextOutput = false
droppedFrames = 0
private snapshotData = 'INITIAL'
constructor(
private readonly toClient: (bytes: Uint8Array<ArrayBufferLike>) => void,
private readonly onServerSideDrop?: () => void
) {}
/** Client -> server frames arrive here (Subscribe / SnapshotRequest / Input). */
receive(bytes: Uint8Array<ArrayBufferLike>): void {
const frame = decodeTerminalStreamFrame(bytes)
if (!frame) {
return
}
if (frame.opcode === TerminalStreamOpcode.Subscribe) {
const payload = decodeTerminalStreamJson<{ streamId: number }>(frame.payload)
this.streamId = payload?.streamId ?? 0
this.sendSnapshot()
return
}
if (frame.opcode === TerminalStreamOpcode.SnapshotRequest) {
// Resync request: the server serializes the *current* buffer, so recovery
// includes everything the client missed.
this.snapshotData = 'RECOVERED'
this.sendSnapshot()
}
}
private send(opcode: TerminalStreamOpcode, payload: Uint8Array, seq: number): void {
this.toClient(encodeTerminalStreamFrame({ opcode, streamId: this.streamId, seq, payload }))
}
private sendSnapshot(): void {
this.send(
TerminalStreamOpcode.SnapshotStart,
encodeTerminalStreamJson({ cols: 80, rows: 24, seq: this.cursorBytes }),
0
)
this.send(TerminalStreamOpcode.SnapshotChunk, encodeTerminalStreamText(this.snapshotData), 0)
this.send(TerminalStreamOpcode.SnapshotEnd, new Uint8Array(), 0)
}
/** Emit an Output chunk, honoring simulated websocket backpressure. */
output(text: string): void {
const startSeq = this.cursorBytes
this.cursorBytes += text.length
if (this.dropNextOutput) {
// encryptedBinaryReply returned false: frame is NOT sent. The byte
// high-water still advances (server keeps producing), so the next frame's
// seq jumps past what the client last saw.
this.dropNextOutput = false
this.droppedFrames += 1
this.onServerSideDrop?.()
return
}
void startSeq
this.send(TerminalStreamOpcode.Output, encodeTerminalStreamText(text), this.cursorBytes)
}
}
describe('remote terminal frame-drop resync', () => {
const unsubscribe = vi.fn()
let server: FakeMultiplexServer
beforeEach(() => {
vi.clearAllMocks()
resetRemoteRuntimeTerminalMultiplexersForTests()
const subscribe = vi.fn(async (_args: unknown, callbacks: SubscribeCallbacks) => {
server = new FakeMultiplexServer((bytes) => callbacks.onBinary?.(bytes))
queueMicrotask(() => callbacks.onResponse({ ok: true, result: { type: 'ready' } }))
return {
unsubscribe,
sendBinary: (bytes: Uint8Array<ArrayBufferLike>) => server.receive(bytes)
}
})
vi.stubGlobal('window', {
api: { runtimeEnvironments: { subscribe } }
})
})
afterEach(() => {
vi.unstubAllGlobals()
})
async function subscribeClient(): Promise<{ data: string[]; snapshots: string[] }> {
const data: string[] = []
const snapshots: string[] = []
const multiplexer = getRemoteRuntimeTerminalMultiplexer('env-1')
await multiplexer.subscribeTerminal({
terminal: 'terminal-1',
client: { id: 'desktop-1', type: 'desktop' },
callbacks: {
onData: (chunk) => data.push(chunk),
onSnapshot: (chunk) => snapshots.push(chunk)
}
})
// Let the initial snapshot round-trip settle.
await Promise.resolve()
await Promise.resolve()
return { data, snapshots }
}
it('detects a dropped Output frame via the seq gap and resyncs', async () => {
const { data, snapshots } = await subscribeClient()
expect(snapshots).toEqual(['INITIAL'])
server.output('aaa')
server.dropNextOutput = true
server.output('bbb') // dropped under backpressure — never reaches the client
server.output('ccc') // seq jumps past 'bbb', exposing the gap
// Flush the client's resync SnapshotRequest -> server snapshot round-trip.
await Promise.resolve()
await Promise.resolve()
// The corrupt tail ('ccc', which followed a gap) is NOT rendered as live data.
expect(data).toEqual(['aaa'])
expect(server.droppedFrames).toBe(1)
// Instead, a fresh authoritative snapshot recovers the terminal.
expect(snapshots).toEqual(['INITIAL', 'RECOVERED'])
})
it('passes contiguous output straight through without resyncing', async () => {
const { data, snapshots } = await subscribeClient()
server.output('one')
server.output('two')
server.output('three')
await Promise.resolve()
await Promise.resolve()
expect(data).toEqual(['one', 'two', 'three'])
expect(snapshots).toEqual(['INITIAL'])
})
})
@@ -77,6 +77,12 @@ type RemoteRuntimeMultiplexedTerminalState = {
snapshotInfo: RemoteRuntimeSnapshotInfo | null
initialSnapshotReceived: boolean
pendingSnapshotRequest: RemoteRuntimeSnapshotRequest | null
// Why: Output frames carry a byte high-water `seq`; a jump past the expected
// next offset means the server dropped frames under websocket backpressure.
// Track it so a gap triggers a self-healing snapshot resync instead of
// silently rendering corrupt/missing output (frame-drop resync).
expectedSeq: number | undefined
resyncInFlight: boolean
}
type RemoteRuntimeSnapshotInfo = {
@@ -149,7 +155,9 @@ class RemoteRuntimeTerminalMultiplexer {
snapshotTarget: 'initial',
snapshotInfo: null,
initialSnapshotReceived: false,
pendingSnapshotRequest: null
pendingSnapshotRequest: null,
expectedSeq: undefined,
resyncInFlight: false
}
this.streams.set(streamId, state)
@@ -334,10 +342,20 @@ class RemoteRuntimeTerminalMultiplexer {
}
if (frame.opcode === TerminalStreamOpcode.Output) {
const data = decodeTerminalStreamText(frame.payload)
stream.callbacks.onData(data, {
seq: typeof frame.seq === 'number' && frame.seq > 0 ? frame.seq : undefined,
rawLength: data.length
})
// Why: a resync snapshot is authoritative; drop live output that arrives
// while it is in flight so the corrupt post-gap tail is never rendered.
if (stream.resyncInFlight) {
return
}
const seq = typeof frame.seq === 'number' && frame.seq > 0 ? frame.seq : undefined
if (this.detectOutputGap(stream, seq, data.length)) {
this.requestResyncSnapshot(stream)
return
}
if (typeof seq === 'number') {
stream.expectedSeq = seq
}
stream.callbacks.onData(data, { seq, rawLength: data.length })
return
}
if (frame.opcode === TerminalStreamOpcode.SnapshotStart) {
@@ -400,7 +418,12 @@ class RemoteRuntimeTerminalMultiplexer {
clearPendingSnapshotRequest(stream)
}
clearSnapshot(stream)
// Why: the snapshot is the new authoritative byte high-water; align the
// gap detector to it and re-open the live path (used by both the initial
// snapshot and a frame-drop resync, which reuses the 'initial' target).
if (target === 'initial') {
stream.expectedSeq = typeof info?.seq === 'number' ? info.seq : undefined
stream.resyncInFlight = false
stream.initialSnapshotReceived = true
stream.callbacks.onSubscribed?.()
}
@@ -408,6 +431,8 @@ class RemoteRuntimeTerminalMultiplexer {
}
if (frame.opcode === TerminalStreamOpcode.Error) {
clearSnapshot(stream)
// Why: a failed resync must re-open the live path or output stalls forever.
stream.resyncInFlight = false
const pendingSnapshotRequest = stream.pendingSnapshotRequest
if (pendingSnapshotRequest) {
clearPendingSnapshotRequest(stream)
@@ -418,6 +443,43 @@ class RemoteRuntimeTerminalMultiplexer {
}
}
// Why: Output `seq` is the byte high-water at the end of a chunk, so a chunk
// that begins after the last high-water (startSeq > expectedSeq) means the
// server dropped intervening frames under backpressure. Only flag a gap when
// both offsets are known, and never on the first seq (nothing to compare to).
private detectOutputGap(
stream: RemoteRuntimeMultiplexedTerminalState,
seq: number | undefined,
rawLength: number
): boolean {
if (typeof seq !== 'number' || typeof stream.expectedSeq !== 'number') {
return false
}
const startSeq = seq - rawLength
return startSeq > stream.expectedSeq
}
// Why: on a detected gap, discard the corrupt tail and pull a fresh
// authoritative snapshot. The request carries no requestId so the server
// reply renders through the initial-snapshot path (full reset), self-healing
// without surfacing an error to the user.
private requestResyncSnapshot(stream: RemoteRuntimeMultiplexedTerminalState): void {
if (stream.resyncInFlight) {
return
}
stream.resyncInFlight = true
stream.expectedSeq = undefined
const sent = this.sendFrame(
stream.streamId,
TerminalStreamOpcode.SnapshotRequest,
encodeTerminalStreamJson({ scrollbackRows: undefined })
)
if (!sent) {
// Transport is down; the reconnect path re-subscribes from scratch.
stream.resyncInFlight = false
}
}
private requestSnapshot(
stream: RemoteRuntimeMultiplexedTerminalState,
opts?: { scrollbackRows?: number }