diff --git a/src/main/runtime/rpc/methods/terminal.ts b/src/main/runtime/rpc/methods/terminal.ts index 05f94f2e631..837a73de446 100644 --- a/src/main/runtime/rpc/methods/terminal.ts +++ b/src/main/runtime/rpc/methods/terminal.ts @@ -135,6 +135,7 @@ type TerminalMultiplexStream = { ackInFlightBytes: number ackWindowBytes: number supportsOutputPause: boolean + supportsOutputSpan: boolean supportsWriteUnavailable: boolean outputPaused: boolean supportsDesktopViewportClaims: boolean @@ -570,7 +571,11 @@ function* iterateTerminalStreamTextPayloads(data: string): Generator): FrameShape[ function expectEquivalent(data: string, meta: TerminalOutputMeta | undefined, label: string): void { const legacy = describeFrames(legacyIterateTerminalOutputFrameChunks(data, meta)) - const next = describeFrames(iterateTerminalOutputFrameChunks(data, meta)) + const next = describeFrames(iterateTerminalOutputFrameChunks(data, meta, { supportsOutputSpan: true })) expect(next, label).toEqual(legacy) } @@ -321,7 +321,7 @@ describe('iterateTerminalOutputFrameChunks equivalence with the pre-optimization const meta = seqPreservingMeta(data, seq) expectEquivalent(data, meta, 'unsafe seq rounding') - const frames = [...iterateTerminalOutputFrameChunks(data, meta)] + const frames = [...iterateTerminalOutputFrameChunks(data, meta, { supportsOutputSpan: true })] expect(frames.at(-1)?.seq).toBe(9_007_199_254_740_992) }) @@ -498,7 +498,7 @@ describe('iterateTerminalOutputFrameChunks equivalence with the pre-optimization 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), { supportsOutputSpan: true })] expect(frames.length).toBeGreaterThan(4) for (const frame of frames) { expect(frame.bytes.byteLength).toBeLessThanOrEqual(TERMINAL_STREAM_CHUNK_BYTES) @@ -516,10 +516,10 @@ describe('iterateTerminalOutputFrameChunks equivalence with the pre-optimization // 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, { supportsOutputSpan: true })]).toHaveLength(1) expect([...legacyIterateTerminalOutputFrameChunks(exact)]).toHaveLength(1) const overByOne = `${exact}a` - expect([...iterateTerminalOutputFrameChunks(overByOne)].length).toBeGreaterThan(1) + expect([...iterateTerminalOutputFrameChunks(overByOne, undefined, { supportsOutputSpan: true })].length).toBeGreaterThan(1) expect([...legacyIterateTerminalOutputFrameChunks(overByOne)].length).toBeGreaterThan(1) }) diff --git a/src/main/runtime/rpc/terminal-output-frame-chunks.ts b/src/main/runtime/rpc/terminal-output-frame-chunks.ts index 91bbc506680..c6686a4a7ac 100644 --- a/src/main/runtime/rpc/terminal-output-frame-chunks.ts +++ b/src/main/runtime/rpc/terminal-output-frame-chunks.ts @@ -50,12 +50,19 @@ export function exceedsTerminalStreamChunkBytes(data: string): boolean { return false } +export type TerminalOutputFrameChunkOptions = { + // Required, not defaulted: the caller owns the peer's negotiated capability, and a + // default would silently re-introduce ungated spans at the next call site. + supportsOutputSpan: boolean +} + export function* iterateTerminalOutputFrameChunks( data: string, - meta?: TerminalOutputMeta + meta: TerminalOutputMeta | undefined, + options: TerminalOutputFrameChunkOptions ): Generator { const rawLength = meta?.rawLength ?? data.length - if (meta?.transformed || rawLength !== data.length) { + if (options.supportsOutputSpan && (meta?.transformed || rawLength !== data.length)) { yield { opcode: TerminalStreamOpcode.OutputSpan, bytes: encodeTerminalStreamJson({ data, rawLength, transformed: true }), @@ -75,8 +82,9 @@ 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. + // This is the downgrade path for a peer without `outputSpan`: it carries the raw + // high-water mark on the final frame only, because a transformed run's seq cannot + // map back to UTF-16 chunk offsets. Matches what hosts sent before OutputSpan existed. 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-output-frame-source-ranges.test.ts b/src/main/runtime/rpc/terminal-output-frame-source-ranges.test.ts index 0a27718f71c..3b0b054429f 100644 --- a/src/main/runtime/rpc/terminal-output-frame-source-ranges.test.ts +++ b/src/main/runtime/rpc/terminal-output-frame-source-ranges.test.ts @@ -52,7 +52,7 @@ describe('terminal output frame source ranges', () => { } }) ] - }) + }, { supportsOutputSpan: true }) ) expect(frames).toHaveLength(2) @@ -89,7 +89,7 @@ describe('terminal output frame source ranges', () => { rawLength: 9, transformed: true, sourceRanges: [sourceRange] - }) + }, { supportsOutputSpan: true }) expect(frame?.sourceRanges).toEqual([sourceRange]) }) diff --git a/src/main/runtime/rpc/terminal-subscribe-output-span-capability.test.ts b/src/main/runtime/rpc/terminal-subscribe-output-span-capability.test.ts new file mode 100644 index 00000000000..09a366f8351 --- /dev/null +++ b/src/main/runtime/rpc/terminal-subscribe-output-span-capability.test.ts @@ -0,0 +1,334 @@ +import { readFileSync } from 'node:fs' +import { join } from 'node:path' +import { describe, expect, it, vi } from 'vitest' +import type { RuntimeTerminalWait } from '../../../shared/runtime-types' +import { + TerminalStreamOpcode, + decodeTerminalStreamFrame, + decodeTerminalStreamText, + encodeTerminalStreamFrame, + encodeTerminalStreamJson +} from '../../../shared/terminal-stream-protocol' +import type { OrcaRuntimeService } from '../orca-runtime' +import type { RpcRequest } from './core' +import { RpcDispatcher } from './dispatcher' +import { TERMINAL_METHODS } from './methods/terminal' + +type TerminalDataMeta = { + seq?: number + rawLength?: number + transformed?: boolean + cwd?: string +} + +type LegacyStreamHarness = { + binaryFrames: Uint8Array[] + emit: (data: string, meta?: TerminalDataMeta) => void + dispatchPromise: Promise + cleanup: () => void +} + +// Why parsed from mobile's source instead of hard-coded: mobile vendors its own +// copy of the opcode table (separate pnpm workspace, so it cannot be imported). +// Reading the real file is what makes this test fail when the two drift. +function mobileDecodableOpcodes(): Set { + const source = readFileSync( + join(process.cwd(), 'mobile/src/transport/terminal-stream-protocol.ts'), + 'utf8' + ) + const body = /enum TerminalStreamOpcode \{([\s\S]*?)\}/.exec(source)?.[1] + if (!body) { + throw new Error('mobile vendored TerminalStreamOpcode enum not found') + } + const opcodes = new Set() + for (const match of body.matchAll(/^\s*\w+\s*=\s*(\d+)/gm)) { + opcodes.add(Number(match[1])) + } + if (opcodes.size === 0) { + throw new Error('mobile vendored TerminalStreamOpcode enum parsed empty') + } + return opcodes +} + +async function subscribeLegacyBinaryStream( + client: { id: string; type: 'mobile' | 'desktop' }, + capabilities: Record +): Promise { + const binaryFrames: Uint8Array[] = [] + const messages: string[] = [] + const cleanups = new Map void>() + let onData: ((data: string, meta?: TerminalDataMeta) => void) | undefined + const runtime = { + getRuntimeId: () => 'test-runtime', + resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-span' }), + hasHeadlessTerminalState: vi.fn(() => true), + getRendererTerminalSerializerGeneration: vi.fn(() => 1), + getPtyOutputSequence: vi.fn(() => 0), + handleMobileSubscribe: vi.fn().mockResolvedValue(true), + handleMobileUnsubscribe: vi.fn(), + updateMobileViewport: vi.fn().mockResolvedValue(true), + updateDesktopViewport: vi.fn().mockResolvedValue(true), + subscribeToTerminalData: vi.fn((_ptyId: string, listener: typeof onData) => { + onData = listener + return vi.fn() + }), + registerRemoteTerminalViewSubscriber: vi.fn(() => vi.fn()), + unregisterRemoteDesktopViewer: vi.fn(), + readTerminal: vi.fn().mockResolvedValue({ tail: [], truncated: false }), + serializeTerminalBuffer: vi.fn().mockResolvedValue(null), + getTerminalSize: vi.fn().mockReturnValue({ cols: 80, rows: 24 }), + getMobileDisplayMode: vi.fn().mockReturnValue('auto'), + getLayout: vi.fn().mockReturnValue({ seq: 1 }), + getDriver: vi.fn().mockReturnValue({ kind: 'idle' }), + isTerminalAlternateScreen: vi.fn().mockReturnValue(false), + subscribeToTerminalResize: vi.fn().mockReturnValue(vi.fn()), + subscribeToFitOverrideChanges: vi.fn().mockReturnValue(vi.fn()), + registerSubscriptionCleanup: vi.fn((id: string, cleanup: () => void) => { + cleanups.set(id, cleanup) + }), + cleanupSubscription: vi.fn((id: string) => { + cleanups.get(id)?.() + cleanups.delete(id) + }), + waitForTerminal: vi.fn(() => new Promise(() => {})) + } as unknown as OrcaRuntimeService + const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + const request: RpcRequest = { + id: `req-${client.id}`, + authToken: 'tok', + method: 'terminal.subscribe', + params: { terminal: 'terminal-span', client, capabilities } + } + const dispatchPromise = dispatcher.dispatchStreaming( + request, + (message) => messages.push(message), + { + connectionId: `conn-${client.id}`, + sendBinary: (bytes) => { + binaryFrames.push(bytes) + }, + registerBinaryStreamHandler: vi.fn(() => vi.fn()) + } + ) + await vi.waitFor(() => + expect(messages.some((message) => JSON.parse(message).result?.type === 'subscribed')).toBe(true) + ) + await vi.waitFor(() => expect(onData).toBeTypeOf('function')) + binaryFrames.splice(0) + return { + binaryFrames, + dispatchPromise, + emit: (data, meta) => onData?.(data, meta), + cleanup: () => runtime.cleanupSubscription(`terminal-span:${client.id}`) + } +} + +function outputTextIn(frames: readonly Uint8Array[]): string { + return frames + .map((bytes) => decodeTerminalStreamFrame(bytes)) + .filter((frame) => frame?.opcode === TerminalStreamOpcode.Output) + .map((frame) => decodeTerminalStreamText(frame!.payload)) + .join('') +} + +// A transformed emission (SSH relay / OSC-stripping): display text is shorter +// than the raw byte run it came from, which is what routes to OutputSpan. +const TRANSFORMED_META: TerminalDataMeta = { seq: 40, rawLength: 26, transformed: true } + +describe('terminal.subscribe OutputSpan capability gate', () => { + it('sends a legacy mobile client only opcodes its vendored decoder knows', async () => { + const legacy = await subscribeLegacyBinaryStream( + { id: 'mobile-span', type: 'mobile' }, + { terminalBinaryStream: 1 } + ) + + legacy.emit('visible output', TRANSFORMED_META) + await vi.waitFor(() => expect(legacy.binaryFrames.length).toBeGreaterThan(0)) + + const decodable = mobileDecodableOpcodes() + const sent = legacy.binaryFrames.map((bytes) => decodeTerminalStreamFrame(bytes)!.opcode) + expect(sent.filter((opcode) => !decodable.has(opcode))).toEqual([]) + + legacy.cleanup() + await legacy.dispatchPromise + }) + + it('preserves the transformed text for a legacy mobile client', async () => { + const legacy = await subscribeLegacyBinaryStream( + { id: 'mobile-text', type: 'mobile' }, + { terminalBinaryStream: 1 } + ) + + legacy.emit('visible output', TRANSFORMED_META) + await vi.waitFor(() => expect(legacy.binaryFrames.length).toBeGreaterThan(0)) + + expect(outputTextIn(legacy.binaryFrames)).toBe('visible output') + + legacy.cleanup() + await legacy.dispatchPromise + }) + + it('still sends OutputSpan to a client that negotiated the capability', async () => { + const capable = await subscribeLegacyBinaryStream( + { id: 'desktop-span', type: 'desktop' }, + { terminalBinaryStream: 1, outputSpan: 1 } + ) + + capable.emit('visible output', TRANSFORMED_META) + await vi.waitFor(() => expect(capable.binaryFrames.length).toBeGreaterThan(0)) + + const spans = capable.binaryFrames + .map((bytes) => decodeTerminalStreamFrame(bytes)) + .filter((frame) => frame?.opcode === TerminalStreamOpcode.OutputSpan) + expect(spans).toHaveLength(1) + expect(JSON.parse(decodeTerminalStreamText(spans[0]!.payload))).toMatchObject({ + data: 'visible output', + rawLength: 26, + transformed: true + }) + + capable.cleanup() + await capable.dispatchPromise + }) +}) + +async function multiplexStream(capabilities: Record): Promise<{ + binaryFrames: Uint8Array[] + emit: (data: string, meta?: TerminalDataMeta) => void + cleanup: () => void + dispatchPromise: Promise +}> { + const binaryFrames: Uint8Array[] = [] + const cleanups = new Map void>() + const handlers = new Map< + number, + (frame: NonNullable>) => void + >() + let onData: ((data: string, meta?: TerminalDataMeta) => void) | undefined + const runtime = { + getRuntimeId: () => 'test-runtime', + // The multiplex subscribe path resolves handles via resolveLiveLeafForHandle. + resolveLiveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-mux-span' }), + resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-mux-span' }), + requestRendererTerminalTabMount: vi.fn().mockReturnValue(true), + updateRemoteDesktopViewer: vi.fn().mockResolvedValue(true), + unregisterRemoteDesktopViewers: vi.fn().mockResolvedValue(true), + isPtyResizeDrivenRemotely: vi.fn().mockReturnValue(false), + getRemoteDesktopFitHold: vi.fn().mockReturnValue({ mode: 'desktop-fit', cols: 120, rows: 40 }), + isRemoteDesktopViewerOwner: vi.fn().mockReturnValue(false), + readTerminal: vi.fn().mockResolvedValue({ tail: [], truncated: false }), + serializeAuthoritativeTerminalBuffer: vi + .fn() + .mockResolvedValue({ data: 'snap', cols: 120, rows: 40 }), + serializeTerminalBuffer: vi.fn().mockResolvedValue({ data: 'snap', cols: 120, rows: 40 }), + getTerminalSize: vi.fn().mockReturnValue({ cols: 120, rows: 40 }), + getMobileDisplayMode: vi.fn().mockReturnValue('auto'), + getLayout: vi.fn().mockReturnValue({ seq: 1 }), + getDriver: vi.fn().mockReturnValue({ kind: 'idle' }), + getTerminalFitOverride: vi.fn().mockReturnValue(null), + getPtyOutputSequence: vi.fn(() => 0), + subscribeToTerminalData: vi.fn((_ptyId: string, listener: typeof onData) => { + onData = listener + return vi.fn() + }), + registerRemoteTerminalViewSubscriber: vi.fn(() => vi.fn()), + unregisterRemoteDesktopViewer: vi.fn(), + subscribeToTerminalResize: vi.fn().mockReturnValue(vi.fn()), + subscribeToFitOverrideChanges: vi.fn().mockReturnValue(vi.fn()), + subscribeToDriverChanges: vi.fn().mockReturnValue(vi.fn()), + updateDesktopViewport: vi.fn().mockResolvedValue(true), + registerSubscriptionCleanup: vi.fn((id: string, cleanup: () => void) => { + cleanups.set(id, cleanup) + }), + cleanupSubscription: vi.fn((id: string) => cleanups.get(id)?.()), + waitForTerminal: vi.fn(() => new Promise(() => {})) + } as unknown as OrcaRuntimeService + const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + const request: RpcRequest = { + id: 'req-mux-span', + authToken: 'tok', + method: 'terminal.multiplex', + params: {} + } + const dispatchPromise = dispatcher.dispatchStreaming(request, vi.fn(), { + connectionId: 'conn-mux-span', + sendBinary: (bytes) => { + binaryFrames.push(bytes) + }, + registerBinaryStreamHandler: (streamId, handler) => { + handlers.set(streamId, handler) + return () => handlers.delete(streamId) + } + }) + await vi.waitFor(() => expect(handlers.has(0)).toBe(true)) + handlers.get(0)!( + decodeTerminalStreamFrame( + encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.Subscribe, + streamId: 0, + seq: 1, + payload: encodeTerminalStreamJson({ + streamId: 7, + terminal: 'terminal-span', + client: { id: 'desktop-mux', type: 'desktop' }, + capabilities, + viewport: { cols: 120, rows: 40 } + }) + }) + )! + ) + await vi.waitFor(() => expect(onData).toBeTypeOf('function')) + binaryFrames.splice(0) + return { + binaryFrames, + emit: (data, meta) => onData?.(data, meta), + cleanup: () => runtime.cleanupSubscription('terminal-multiplex:conn-mux-span'), + dispatchPromise + } +} + +describe('terminal.multiplex OutputSpan capability gate', () => { + it('downgrades to Output for a desktop build that predates the capability', async () => { + const old = await multiplexStream({ ackOutput: 1, desktopViewportClaims: 1 }) + + old.emit('visible output', TRANSFORMED_META) + await vi.waitFor(() => expect(outputTextIn(old.binaryFrames)).toBe('visible output')) + + const opcodes = old.binaryFrames.map((bytes) => decodeTerminalStreamFrame(bytes)!.opcode) + expect(opcodes).not.toContain(TerminalStreamOpcode.OutputSpan) + + old.cleanup() + await old.dispatchPromise + }) + + it('keeps sending OutputSpan to a desktop build that negotiates it', async () => { + const current = await multiplexStream({ + ackOutput: 1, + desktopViewportClaims: 1, + outputSpan: 1 + }) + + current.emit('visible output', TRANSFORMED_META) + await vi.waitFor(() => + expect( + current.binaryFrames.some( + (bytes) => decodeTerminalStreamFrame(bytes)?.opcode === TerminalStreamOpcode.OutputSpan + ) + ).toBe(true) + ) + + current.cleanup() + await current.dispatchPromise + }) + + it('has the shipping desktop multiplexer declare the capability', () => { + // Guards the other half of the gate: if the renderer stops advertising + // outputSpan, the host silently downgrades every desktop stream. + const source = readFileSync( + join(process.cwd(), 'src/renderer/src/runtime/remote-runtime-terminal-multiplexer.ts'), + 'utf8' + ) + const capabilities = /capabilities: \{([\s\S]*?)\}/.exec(source)?.[1] + expect(capabilities).toContain('outputSpan: 1') + }) +}) diff --git a/src/renderer/src/runtime/remote-runtime-terminal-multiplexer.ts b/src/renderer/src/runtime/remote-runtime-terminal-multiplexer.ts index e153c1c75cb..8850638195b 100644 --- a/src/renderer/src/runtime/remote-runtime-terminal-multiplexer.ts +++ b/src/renderer/src/runtime/remote-runtime-terminal-multiplexer.ts @@ -507,6 +507,7 @@ class RemoteRuntimeTerminalMultiplexer { ackOutput: 1, ackOutputSourceRanges: 1, outputPause: 1, + outputSpan: 1, writeUnavailable: 1, ...(args.client.type === 'desktop' ? { desktopViewportClaims: 1 } : {}) } diff --git a/src/renderer/src/runtime/runtime-terminal-stream.test.ts b/src/renderer/src/runtime/runtime-terminal-stream.test.ts index 39cd8ce829a..83b241985a2 100644 --- a/src/renderer/src/runtime/runtime-terminal-stream.test.ts +++ b/src/renderer/src/runtime/runtime-terminal-stream.test.ts @@ -107,6 +107,7 @@ describe('remote runtime terminal data subscriptions', () => { ackOutputSourceRanges?: 1 desktopViewportClaims?: 1 outputPause?: 1 + outputSpan?: 1 writeUnavailable?: 1 } }>(subscribeFrame.payload) @@ -116,6 +117,7 @@ describe('remote runtime terminal data subscriptions', () => { ackOutputSourceRanges: 1, desktopViewportClaims: 1, outputPause: 1, + outputSpan: 1, writeUnavailable: 1 }) diff --git a/src/shared/terminal-stream-protocol.ts b/src/shared/terminal-stream-protocol.ts index 4641fb101da..5d4f52a6245 100644 --- a/src/shared/terminal-stream-protocol.ts +++ b/src/shared/terminal-stream-protocol.ts @@ -28,6 +28,9 @@ export enum TerminalStreamOpcode { // Why 14: Ack already occupies 13 on current clients; older runtimes ignore // this opcode and still receive the compatibility Resize frame behind it. ClaimViewport = 14, + // Negotiated per stream: mobile's vendored decoder stops at 12 and drops this + // frame silently, so an ungated span is lost output. Senders downgrade to + // Output when the peer did not confirm `outputSpan`. OutputSpan = 15, // Negotiated per stream; older hosts reject unknown opcodes, so clients send only after capability confirmation. SetOutputPaused = 16,