fix(mobile): stop phones losing terminal output sent as output spans (#33405)

* fix(mobile): stop losing phone terminal output sent as output spans

Hosts sent stripped/transformed terminal output to terminal.subscribe as
OutputSpan (opcode 15), which no phone has ever decoded. The phone dropped
it silently, e.g. the reset after a killed full-screen program, leaving
the phone stuck on the alternate screen.

- Host: terminal.subscribe sends spans as plain Output text; only the
  desktop multiplex (which decodes spans) keeps OutputSpan.
- Mobile: decode OutputSpan for older hosts, use the shared protocol
  table, and end the stream with a logged error on any frame it can't read.

* fix(mobile): throttle unreadable-frame resets and log once per stream

* fix(mobile): reset a throttled stream after the window; forget released stream ids

* fix(mobile): defer throttled unreadable-frame resets instead of dropping them
This commit is contained in:
Neil
2026-10-24 22:56:11 -07:00
committed by GitHub
parent 3d24e61726
commit 4ff84bb835
17 changed files with 272 additions and 103 deletions
@@ -3,7 +3,10 @@ import {
BrowserScreencastOpcode,
encodeBrowserScreencastFrame
} from '../../../src/shared/browser-screencast-protocol'
import { encodeTerminalStreamFrame, TerminalStreamOpcode } from './terminal-stream-protocol'
import {
encodeTerminalStreamFrame,
TerminalStreamOpcode
} from '../../../src/shared/terminal-stream-protocol'
import { scopeRpcClientToExecutionHost } from './execution-host-scoped-rpc-client'
import { isRpcDeliveryUnknown } from './rpc-delivery-ambiguity'
@@ -3,6 +3,7 @@ import {
handleTerminalBinaryFrame,
type TerminalSnapshotState
} from './rpc-client-terminal-binary-frame'
import { TerminalUnreadableFrameResets } from './terminal-unreadable-frame-resets'
import {
buildStreamUnsubscribe,
buildTerminalUnsubscribeParams,
@@ -80,6 +81,7 @@ export class MobileRelayRpcStreams {
>()
private readonly terminalListeners = new Map<number, (result: unknown) => void>()
private readonly terminalSnapshots = new Map<number, TerminalSnapshotState>()
private readonly unreadableTerminalFrames = new TerminalUnreadableFrameResets()
private activeBrowserStream: StreamRecord | null = null
private sendCount = 0
@@ -201,6 +203,7 @@ export class MobileRelayRpcStreams {
}
handleTerminalBinaryFrame(bytes, {
terminalSnapshots: this.terminalSnapshots,
unreadableFrames: this.unreadableTerminalFrames,
getListener: (streamId) => this.terminalListeners.get(streamId)
})
}
@@ -213,6 +216,7 @@ export class MobileRelayRpcStreams {
this.cancelledSubscriptions.clear()
this.terminalListeners.clear()
this.terminalSnapshots.clear()
this.unreadableTerminalFrames.clear()
this.activeBrowserStream = null
}
@@ -301,6 +305,7 @@ export class MobileRelayRpcStreams {
for (const streamId of stream.streamIds) {
this.terminalListeners.delete(streamId)
this.terminalSnapshots.delete(streamId)
this.unreadableTerminalFrames.forget(streamId)
}
if (this.activeBrowserStream === stream) {
this.activeBrowserStream = null
@@ -1,6 +1,9 @@
import { describe, expect, it } from 'vitest'
import { RpcClientStreamRegistry } from './rpc-client-stream-registry'
import { encodeTerminalStreamFrame, TerminalStreamOpcode } from './terminal-stream-protocol'
import {
encodeTerminalStreamFrame,
TerminalStreamOpcode
} from '../../../src/shared/terminal-stream-protocol'
import type { ConnectionState, RpcResponse } from './types'
type SentRequest = {
@@ -1,6 +1,11 @@
import { describe, expect, it, vi } from 'vitest'
import { handleTerminalBinaryFrame } from './rpc-client-terminal-binary-frame'
import { encodeTerminalStreamFrame, TerminalStreamOpcode } from './terminal-stream-protocol'
import { TerminalUnreadableFrameResets } from './terminal-unreadable-frame-resets'
import { RpcClientTerminalStreamRouter } from './rpc-client-terminal-stream-router'
import {
encodeTerminalStreamFrame,
TerminalStreamOpcode
} from '../../../src/shared/terminal-stream-protocol'
function encodeFrame(opcode: TerminalStreamOpcode, streamId: number, payload: unknown): Uint8Array {
const body =
@@ -23,10 +28,98 @@ describe('handleTerminalBinaryFrame', () => {
encodeFrame(TerminalStreamOpcode.Metadata, 42, { cwd: '/repo/src' }),
{
terminalSnapshots: new Map(),
unreadableFrames: new TerminalUnreadableFrameResets(),
getListener: (streamId) => (streamId === 42 ? listener : undefined)
}
)
expect(listener).toHaveBeenCalledWith({ type: 'metadata', streamId: 42, cwd: '/repo/src' })
})
it('writes a transformed output span from a host that predates text-only subscribe', () => {
const listener = vi.fn()
handleTerminalBinaryFrame(
encodeFrame(TerminalStreamOpcode.OutputSpan, 7, {
data: 'after the stripped query',
rawLength: 31,
transformed: true
}),
{
terminalSnapshots: new Map(),
unreadableFrames: new TerminalUnreadableFrameResets(),
getListener: () => listener
}
)
expect(listener).toHaveBeenCalledWith({
type: 'data',
streamId: 7,
chunk: 'after the stripped query'
})
})
it('ends a stream on an unreadable frame, deferring resets inside the interval', () => {
vi.useFakeTimers()
const warn = vi.spyOn(console, 'warn').mockImplementation(() => {})
try {
const listener = vi.fn()
const resets = new TerminalUnreadableFrameResets()
const options = {
terminalSnapshots: new Map(),
unreadableFrames: resets,
getListener: () => listener
}
handleTerminalBinaryFrame(unreadableFrame(9), options)
handleTerminalBinaryFrame(unreadableFrame(9), options)
// Another pane hits one inside the interval: deferred, never dropped.
handleTerminalBinaryFrame(unreadableFrame(10), options)
handleTerminalBinaryFrame(unreadableFrame(11), options)
resets.forget(11)
expect(listener.mock.calls.map(([event]) => event)).toEqual([
expect.objectContaining({ type: 'error', streamId: 9 })
])
vi.advanceTimersByTime(30_000)
expect(listener.mock.calls.map(([event]) => event)).toEqual([
expect.objectContaining({ type: 'error', streamId: 9 }),
expect.objectContaining({ type: 'error', streamId: 10 })
])
expect(warn).toHaveBeenCalledTimes(3)
} finally {
warn.mockRestore()
vi.useRealTimers()
}
})
it('lets a released stream id that the host reuses be reset again', () => {
vi.useFakeTimers()
const warn = vi.spyOn(console, 'warn').mockImplementation(() => {})
try {
const router = new RpcClientTerminalStreamRouter()
const first = vi.fn()
router.register('request-1', 5, first)
router.handle(unreadableFrame(5))
router.reset('request-1')
vi.advanceTimersByTime(30_000)
const second = vi.fn()
router.register('request-2', 5, second)
router.handle(unreadableFrame(5))
expect(first).toHaveBeenCalledWith(expect.objectContaining({ type: 'error', streamId: 5 }))
expect(second).toHaveBeenCalledWith(expect.objectContaining({ type: 'error', streamId: 5 }))
} finally {
warn.mockRestore()
vi.useRealTimers()
}
})
})
function unreadableFrame(streamId: number): Uint8Array {
const frame = encodeFrame(TerminalStreamOpcode.Output, streamId, 'text')
// An opcode no build knows yet, as a newer host could send.
frame[2] = 250
return frame
}
@@ -2,8 +2,10 @@ import {
TerminalStreamOpcode,
decodeTerminalStreamFrame,
decodeTerminalStreamJson,
decodeTerminalStreamText
} from './terminal-stream-protocol'
decodeTerminalStreamText,
readTerminalStreamFrameId
} from '../../../src/shared/terminal-stream-protocol'
import type { TerminalUnreadableFrameResets } from './terminal-unreadable-frame-resets'
export type TerminalSnapshotState = {
streamId: number
@@ -15,6 +17,7 @@ type StreamingListener = (result: unknown) => void
type TerminalBinaryFrameOptions = {
terminalSnapshots: Map<number, TerminalSnapshotState>
unreadableFrames: TerminalUnreadableFrameResets
getListener: (streamId: number) => StreamingListener | undefined
}
@@ -24,6 +27,11 @@ export function handleTerminalBinaryFrame(
): void {
const frame = decodeTerminalStreamFrame(bytes)
if (!frame) {
const streamId = readTerminalStreamFrameId(bytes)
if (streamId !== null) {
// A terminal frame this build can't read (unknown opcode) may carry output.
reportUnreadableTerminalFrame(options, streamId, bytes[2] ?? -1)
}
return
}
const listener = options.getListener(frame.streamId)
@@ -38,6 +46,16 @@ export function handleTerminalBinaryFrame(
})
return
}
if (frame.opcode === TerminalStreamOpcode.OutputSpan) {
// Hosts before text-only terminal.subscribe send stripped-query output as a span.
const span = decodeTerminalStreamJson<{ data?: unknown }>(frame.payload)
if (typeof span?.data !== 'string') {
reportUnreadableTerminalFrame(options, frame.streamId, frame.opcode)
return
}
listener({ type: 'data', streamId: frame.streamId, chunk: span.data })
return
}
if (frame.opcode === TerminalStreamOpcode.SnapshotStart) {
const meta = decodeTerminalStreamJson<Record<string, unknown>>(frame.payload)
if (!meta) {
@@ -103,5 +121,31 @@ export function handleTerminalBinaryFrame(
streamId: frame.streamId,
message: decodeTerminalStreamText(frame.payload)
})
return
}
reportUnreadableTerminalFrame(options, frame.streamId, frame.opcode)
}
// Why an error event: ending the stream makes the next subscription redraw from a fresh snapshot.
function reportUnreadableTerminalFrame(
options: TerminalBinaryFrameOptions,
streamId: number,
opcode: number
): void {
if (!options.getListener(streamId)) {
return
}
options.unreadableFrames.report(streamId, opcode, () => {
// Re-read: a deferred reset must reach only the stream that is still live.
const listener = options.getListener(streamId)
if (!listener) {
return
}
options.terminalSnapshots.delete(streamId)
listener({
type: 'error',
streamId,
message: `Unsupported terminal stream frame (opcode ${opcode}).`
})
})
}
@@ -1,6 +1,9 @@
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import { connect } from './rpc-client'
import { encodeTerminalStreamFrame, TerminalStreamOpcode } from './terminal-stream-protocol'
import {
encodeTerminalStreamFrame,
TerminalStreamOpcode
} from '../../../src/shared/terminal-stream-protocol'
vi.mock('./direct-e2ee-framing-policy', () => import('./legacy-direct-framing-test-policy'))
@@ -2,6 +2,7 @@ import {
handleTerminalBinaryFrame,
type TerminalSnapshotState
} from './rpc-client-terminal-binary-frame'
import { TerminalUnreadableFrameResets } from './terminal-unreadable-frame-resets'
type TerminalListener = (result: unknown) => void
@@ -9,6 +10,7 @@ export class RpcClientTerminalStreamRouter {
private readonly listeners = new Map<number, TerminalListener>()
private readonly idsByRequest = new Map<string, Set<number>>()
private readonly snapshots = new Map<number, TerminalSnapshotState>()
private readonly unreadableFrames = new TerminalUnreadableFrameResets()
register(requestId: string, streamId: number, listener: TerminalListener): void {
const ids = this.idsByRequest.get(requestId) ?? new Set<number>()
@@ -25,6 +27,7 @@ export class RpcClientTerminalStreamRouter {
for (const streamId of streamIds) {
this.listeners.delete(streamId)
this.snapshots.delete(streamId)
this.unreadableFrames.forget(streamId)
}
this.idsByRequest.delete(requestId)
}
@@ -32,6 +35,7 @@ export class RpcClientTerminalStreamRouter {
handle(bytes: Uint8Array): void {
handleTerminalBinaryFrame(bytes, {
terminalSnapshots: this.snapshots,
unreadableFrames: this.unreadableFrames,
getListener: (streamId) => this.listeners.get(streamId)
})
}
+4 -1
View File
@@ -1,6 +1,9 @@
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import { connect } from './rpc-client'
import { encodeTerminalStreamFrame, TerminalStreamOpcode } from './terminal-stream-protocol'
import {
encodeTerminalStreamFrame,
TerminalStreamOpcode
} from '../../../src/shared/terminal-stream-protocol'
vi.mock('./direct-e2ee-framing-policy', () => import('./legacy-direct-framing-test-policy'))
@@ -1,6 +1,9 @@
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import { connect } from './rpc-client'
import { encodeTerminalStreamFrame, TerminalStreamOpcode } from './terminal-stream-protocol'
import {
encodeTerminalStreamFrame,
TerminalStreamOpcode
} from '../../../src/shared/terminal-stream-protocol'
vi.mock('./direct-e2ee-framing-policy', () => import('./legacy-direct-framing-test-policy'))
@@ -1,81 +0,0 @@
const TERMINAL_STREAM_KIND = 0x74
const TERMINAL_STREAM_VERSION = 1
const HEADER_BYTES = 16
export enum TerminalStreamOpcode {
Output = 1,
SnapshotStart = 2,
SnapshotChunk = 3,
SnapshotEnd = 4,
Resized = 5,
Error = 6,
Metadata = 12
}
export type TerminalStreamFrame = {
opcode: TerminalStreamOpcode
streamId: number
seq: number
payload: Uint8Array
}
export function encodeTerminalStreamFrame(frame: TerminalStreamFrame): Uint8Array {
const out = new Uint8Array(HEADER_BYTES + frame.payload.length)
const view = new DataView(out.buffer, out.byteOffset, out.byteLength)
view.setUint8(0, TERMINAL_STREAM_KIND)
view.setUint8(1, TERMINAL_STREAM_VERSION)
view.setUint8(2, frame.opcode)
view.setUint8(3, 0)
view.setUint32(4, frame.streamId, true)
const seq = Math.max(0, Math.floor(frame.seq))
view.setUint32(8, Math.floor(seq / 0x100000000), true)
view.setUint32(12, seq >>> 0, true)
out.set(frame.payload, HEADER_BYTES)
return out
}
export function decodeTerminalStreamFrame(bytes: Uint8Array): TerminalStreamFrame | null {
if (bytes.length < HEADER_BYTES) {
return null
}
const view = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength)
if (view.getUint8(0) !== TERMINAL_STREAM_KIND || view.getUint8(1) !== TERMINAL_STREAM_VERSION) {
return null
}
const opcode = view.getUint8(2)
if (!isTerminalStreamOpcode(opcode)) {
return null
}
const high = view.getUint32(8, true)
const low = view.getUint32(12, true)
return {
opcode,
streamId: view.getUint32(4, true),
seq: high * 0x100000000 + low,
payload: bytes.slice(HEADER_BYTES)
}
}
export function decodeTerminalStreamJson<T>(payload: Uint8Array): T | null {
try {
return JSON.parse(new TextDecoder().decode(payload)) as T
} catch {
return null
}
}
export function decodeTerminalStreamText(payload: Uint8Array): string {
return new TextDecoder().decode(payload)
}
function isTerminalStreamOpcode(value: number): value is TerminalStreamOpcode {
return (
value === TerminalStreamOpcode.Output ||
value === TerminalStreamOpcode.SnapshotStart ||
value === TerminalStreamOpcode.SnapshotChunk ||
value === TerminalStreamOpcode.SnapshotEnd ||
value === TerminalStreamOpcode.Resized ||
value === TerminalStreamOpcode.Error ||
value === TerminalStreamOpcode.Metadata
)
}
@@ -0,0 +1,63 @@
// Why: a host that always sends an unreadable frame must not turn every resubscribe into another reset.
const RESET_INTERVAL_MS = 30_000
/**
* Paces the stream resets that recover output from terminal frames this build can't read.
* At most one batch of resets per interval; a stream that arrives inside the interval is
* reset when it ends, not dropped. Owners call `forget` wherever they release a stream id.
*/
export class TerminalUnreadableFrameResets {
private lastResetAt: number | null = null
private readonly handledStreamIds = new Set<number>()
private readonly deferred = new Map<number, () => void>()
private timer: ReturnType<typeof setTimeout> | null = null
report(streamId: number, opcode: number, reset: () => void): void {
if (this.handledStreamIds.has(streamId) || this.deferred.has(streamId)) {
return
}
const now = Date.now()
const waitMs = this.lastResetAt === null ? 0 : this.lastResetAt + RESET_INTERVAL_MS - now
console.warn('[terminal-stream] unreadable frame', { streamId, opcode, waitMs })
if (waitMs <= 0) {
this.lastResetAt = now
this.handledStreamIds.add(streamId)
reset()
return
}
this.deferred.set(streamId, reset)
this.timer ??= setTimeout(() => this.flushDeferred(), waitMs)
}
forget(streamId: number): void {
this.handledStreamIds.delete(streamId)
this.deferred.delete(streamId)
if (this.deferred.size === 0) {
this.cancelTimer()
}
}
clear(): void {
this.handledStreamIds.clear()
this.deferred.clear()
this.cancelTimer()
}
private flushDeferred(): void {
this.timer = null
this.lastResetAt = Date.now()
const resets = [...this.deferred]
this.deferred.clear()
for (const [streamId, reset] of resets) {
this.handledStreamIds.add(streamId)
reset()
}
}
private cancelTimer(): void {
if (this.timer !== null) {
clearTimeout(this.timer)
this.timer = null
}
}
}
@@ -69,7 +69,7 @@ export async function initializeMultiplexStream(
meta.seq
)
}
for (const chunk of iterateTerminalOutputFrameChunks(data, meta)) {
for (const chunk of iterateTerminalOutputFrameChunks(data, meta, 'span')) {
state.queueOrSendOutput(stream, chunk)
}
}),
@@ -166,7 +166,7 @@ export function* iterateTerminalStreamTextPayloads(
if (!data) {
return
}
for (const chunk of iterateTerminalOutputFrameChunks(data)) {
for (const chunk of iterateTerminalOutputFrameChunks(data, undefined, 'text')) {
yield chunk.bytes
}
}
@@ -93,7 +93,8 @@ export async function runTerminalBinarySubscription(args: TerminalSubscriptionAr
meta.seq
)
}
for (const chunk of iterateTerminalOutputFrameChunks(data, meta)) {
// Why text: phones before this release can't decode OutputSpan, so a span here is output they drop.
for (const chunk of iterateTerminalOutputFrameChunks(data, meta, 'text')) {
sendFrame(chunk.opcode ?? TerminalStreamOpcode.Output, chunk.bytes, chunk.seq)
}
})
@@ -18,13 +18,15 @@ describe('terminal output frame chunks', () => {
const seq = 9_007_199_254_740_994
const meta = seqPreservingMeta(data, seq)
const frames = [...iterateTerminalOutputFrameChunks(data, meta)]
const frames = [...iterateTerminalOutputFrameChunks(data, meta, 'span')]
expect(frames.at(-1)?.seq).toBe(9_007_199_254_740_992)
})
it('keeps every emitted frame within the wire cap and reassembles to the input', () => {
const data = `${'a'.repeat(200 * 1024)}${SURROGATE_PAIR.repeat(4096)}${LONE_HIGH}`
const frames = [...iterateTerminalOutputFrameChunks(data, seqPreservingMeta(data, 999_999))]
const frames = [
...iterateTerminalOutputFrameChunks(data, seqPreservingMeta(data, 999_999), 'span')
]
expect(frames.length).toBeGreaterThan(4)
for (const frame of frames) {
expect(frame.bytes.byteLength).toBeLessThanOrEqual(TERMINAL_STREAM_CHUNK_BYTES)
@@ -42,8 +44,22 @@ describe('terminal output frame chunks', () => {
// 3-byte code points: 16384 code units = 49152 bytes = exactly the cap.
const exact = '\u20ac'.repeat(TERMINAL_STREAM_CHUNK_BYTES / 3)
expect(Buffer.byteLength(exact, 'utf8')).toBe(TERMINAL_STREAM_CHUNK_BYTES)
expect([...iterateTerminalOutputFrameChunks(exact)]).toHaveLength(1)
expect([...iterateTerminalOutputFrameChunks(exact, undefined, 'span')]).toHaveLength(1)
const overByOne = `${exact}a`
expect([...iterateTerminalOutputFrameChunks(overByOne)].length).toBeGreaterThan(1)
expect(
[...iterateTerminalOutputFrameChunks(overByOne, undefined, 'span')].length
).toBeGreaterThan(1)
})
it('sends a transformed span as capped text frames with its seq only on the last', () => {
const data = 'x'.repeat(TERMINAL_STREAM_CHUNK_BYTES * 2 + 1)
const meta = { seq: 500_000, rawLength: data.length + 8, transformed: true }
const frames = [...iterateTerminalOutputFrameChunks(data, meta, 'text')]
expect(frames.map((frame) => frame.opcode)).toEqual([undefined, undefined, undefined])
expect(frames.map((frame) => frame.seq)).toEqual([undefined, undefined, 500_000])
expect(frames.map((frame) => new TextDecoder().decode(frame.bytes)).join('')).toBe(data)
expect([...iterateTerminalOutputFrameChunks('', meta, 'text')]).toEqual([])
})
})
@@ -20,6 +20,9 @@ export type TerminalOutputFrameChunk = {
opcode?: TerminalStreamOpcode
}
/** `span` only for subscribers that decode OutputSpan (the multiplex); `text` is the opcode-1 encoding every client reads. */
export type TerminalOutputFrameEncoding = 'span' | 'text'
export const TERMINAL_STREAM_BYTE_PROBE_CODE_UNITS = 8 * 1024
const MAX_UTF8_BYTES_PER_CODE_UNIT = 3
@@ -49,10 +52,12 @@ export function exceedsTerminalStreamChunkBytes(data: string): boolean {
export function* iterateTerminalOutputFrameChunks(
data: string,
meta?: TerminalOutputMeta
meta: TerminalOutputMeta | undefined,
encoding: TerminalOutputFrameEncoding
): Generator<TerminalOutputFrameChunk> {
const rawLength = meta?.rawLength ?? data.length
if (meta?.transformed || rawLength !== data.length) {
const isSpan = meta?.transformed === true || rawLength !== data.length
if (isSpan && encoding === 'span') {
yield {
opcode: TerminalStreamOpcode.OutputSpan,
bytes: encodeTerminalStreamJson({ data, rawLength, transformed: true }),
@@ -61,6 +66,10 @@ export function* iterateTerminalOutputFrameChunks(
}
return
}
if (isSpan && !data) {
// Why: a fully stripped span has no text, and a text frame has no field for its raw length.
return
}
if (!exceedsTerminalStreamChunkBytes(data)) {
yield {
bytes: encodeTerminalStreamText(data),
@@ -69,9 +78,8 @@ export function* iterateTerminalOutputFrameChunks(
}
return
}
const canPreserveChunkSeq = typeof meta?.seq === 'number' && rawLength === data.length
// Unreachable past the OutputSpan branch above (rawLength === data.length there), but kept
// as defence in depth: it is the only shape that can carry a seq the chunk offsets can't map.
const canPreserveChunkSeq = typeof meta?.seq === 'number' && !isSpan
// A span sent as text: chunk offsets can't map back to its raw sequence.
const shouldDelayFinalSeq = !canPreserveChunkSeq && typeof meta?.seq === 'number'
const startSeq = canPreserveChunkSeq ? meta.seq! - rawLength : undefined
let chunkStart = 0
@@ -98,10 +98,11 @@ describe('mobile subscribe residual metadata', () => {
...(transformedRawLength === undefined
? []
: [
// Phones never decoded OutputSpan, so a transformed span reaches them as text.
{
opcode: TerminalStreamOpcode.OutputSpan,
opcode: TerminalStreamOpcode.Output,
seq: 15 + transformedRawLength,
payload: { data: 'PAINT', rawLength: transformedRawLength, transformed: true }
payload: 'PAINT'
}
])
])