diff --git a/mobile/src/transport/mobile-relay-rpc-session.test.ts b/mobile/src/transport/mobile-relay-rpc-session.test.ts index 377d0cb317a..9c83bf38de9 100644 --- a/mobile/src/transport/mobile-relay-rpc-session.test.ts +++ b/mobile/src/transport/mobile-relay-rpc-session.test.ts @@ -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' diff --git a/mobile/src/transport/mobile-relay-rpc-streams.ts b/mobile/src/transport/mobile-relay-rpc-streams.ts index b431a57b210..235955b8452 100644 --- a/mobile/src/transport/mobile-relay-rpc-streams.ts +++ b/mobile/src/transport/mobile-relay-rpc-streams.ts @@ -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 void>() private readonly terminalSnapshots = new Map() + 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 diff --git a/mobile/src/transport/rpc-client-stream-registry.test.ts b/mobile/src/transport/rpc-client-stream-registry.test.ts index bf39326b9f4..67164b887af 100644 --- a/mobile/src/transport/rpc-client-stream-registry.test.ts +++ b/mobile/src/transport/rpc-client-stream-registry.test.ts @@ -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 = { diff --git a/mobile/src/transport/rpc-client-terminal-binary-frame.test.ts b/mobile/src/transport/rpc-client-terminal-binary-frame.test.ts index 973522ec400..0e1b95cedd9 100644 --- a/mobile/src/transport/rpc-client-terminal-binary-frame.test.ts +++ b/mobile/src/transport/rpc-client-terminal-binary-frame.test.ts @@ -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 +} diff --git a/mobile/src/transport/rpc-client-terminal-binary-frame.ts b/mobile/src/transport/rpc-client-terminal-binary-frame.ts index 8e252d7136f..2a70df0aa04 100644 --- a/mobile/src/transport/rpc-client-terminal-binary-frame.ts +++ b/mobile/src/transport/rpc-client-terminal-binary-frame.ts @@ -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 + 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>(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}).` + }) + }) } diff --git a/mobile/src/transport/rpc-client-terminal-reconnect.test.ts b/mobile/src/transport/rpc-client-terminal-reconnect.test.ts index 57b3183b809..58b93cf6835 100644 --- a/mobile/src/transport/rpc-client-terminal-reconnect.test.ts +++ b/mobile/src/transport/rpc-client-terminal-reconnect.test.ts @@ -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')) diff --git a/mobile/src/transport/rpc-client-terminal-stream-router.ts b/mobile/src/transport/rpc-client-terminal-stream-router.ts index 0ac4d26c3fc..293555a5e82 100644 --- a/mobile/src/transport/rpc-client-terminal-stream-router.ts +++ b/mobile/src/transport/rpc-client-terminal-stream-router.ts @@ -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() private readonly idsByRequest = new Map>() private readonly snapshots = new Map() + private readonly unreadableFrames = new TerminalUnreadableFrameResets() register(requestId: string, streamId: number, listener: TerminalListener): void { const ids = this.idsByRequest.get(requestId) ?? new Set() @@ -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) }) } diff --git a/mobile/src/transport/rpc-client.test.ts b/mobile/src/transport/rpc-client.test.ts index 1adc32dadbc..5c9d4c04f4b 100644 --- a/mobile/src/transport/rpc-client.test.ts +++ b/mobile/src/transport/rpc-client.test.ts @@ -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')) diff --git a/mobile/src/transport/rpc-session-liveness-integration.test.ts b/mobile/src/transport/rpc-session-liveness-integration.test.ts index 675ad54da77..aaccc8b71fe 100644 --- a/mobile/src/transport/rpc-session-liveness-integration.test.ts +++ b/mobile/src/transport/rpc-session-liveness-integration.test.ts @@ -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')) diff --git a/mobile/src/transport/terminal-stream-protocol.ts b/mobile/src/transport/terminal-stream-protocol.ts deleted file mode 100644 index 384d66996f1..00000000000 --- a/mobile/src/transport/terminal-stream-protocol.ts +++ /dev/null @@ -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(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 - ) -} diff --git a/mobile/src/transport/terminal-unreadable-frame-resets.ts b/mobile/src/transport/terminal-unreadable-frame-resets.ts new file mode 100644 index 00000000000..64476c883ba --- /dev/null +++ b/mobile/src/transport/terminal-unreadable-frame-resets.ts @@ -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() + private readonly deferred = new Map void>() + private timer: ReturnType | 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 + } + } +} diff --git a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-stream-initialization.ts b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-stream-initialization.ts index a9a30d872ed..307b6987a50 100644 --- a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-stream-initialization.ts +++ b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-stream-initialization.ts @@ -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) } }), diff --git a/src/main/runtime/rpc/methods/terminal/terminal-stream-replay.ts b/src/main/runtime/rpc/methods/terminal/terminal-stream-replay.ts index 769f8cf4083..d864cf1c8c9 100644 --- a/src/main/runtime/rpc/methods/terminal/terminal-stream-replay.ts +++ b/src/main/runtime/rpc/methods/terminal/terminal-stream-replay.ts @@ -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 } } diff --git a/src/main/runtime/rpc/methods/terminal/terminal-subscribe-binary.ts b/src/main/runtime/rpc/methods/terminal/terminal-subscribe-binary.ts index 59158b348c3..cac0c8f5035 100644 --- a/src/main/runtime/rpc/methods/terminal/terminal-subscribe-binary.ts +++ b/src/main/runtime/rpc/methods/terminal/terminal-subscribe-binary.ts @@ -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) } }) diff --git a/src/main/runtime/rpc/terminal-output-frame-chunks.test.ts b/src/main/runtime/rpc/terminal-output-frame-chunks.test.ts index 106ace50781..2e7a7f59bc4 100644 --- a/src/main/runtime/rpc/terminal-output-frame-chunks.test.ts +++ b/src/main/runtime/rpc/terminal-output-frame-chunks.test.ts @@ -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([]) }) }) diff --git a/src/main/runtime/rpc/terminal-output-frame-chunks.ts b/src/main/runtime/rpc/terminal-output-frame-chunks.ts index 22a7cf07471..f79def728f6 100644 --- a/src/main/runtime/rpc/terminal-output-frame-chunks.ts +++ b/src/main/runtime/rpc/terminal-output-frame-chunks.ts @@ -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 { 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 diff --git a/src/main/runtime/rpc/terminal-subscribe-residual-metadata.test.ts b/src/main/runtime/rpc/terminal-subscribe-residual-metadata.test.ts index 033f2c9a835..319cd8b5b3b 100644 --- a/src/main/runtime/rpc/terminal-subscribe-residual-metadata.test.ts +++ b/src/main/runtime/rpc/terminal-subscribe-residual-metadata.test.ts @@ -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' } ]) ])