feat(mobile): let the OTA page send extra top-level request fields (#27815)

* feat(mobile): let the OTA page send extra top-level request fields

A page request could only carry method, params and three fixed options, so any new
top-level wire field needed a new app binary. Requests and subscribes now carry an
optional verbatim `envelope`, gated by a new `rpcEnvelope` grant; nothing sets it yet.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* refactor(mobile): one envelope merge seam per transport

Review follow-up: the direct tracker and registry now hand frames to a single
sendFrame that adds the credential and envelope, like the relay; page refusals
use one fixed code per error class and an envelope-only options bag posts no options.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* fix(mobile): refuse a non-object envelope before treating it as empty

null threw synchronously and [] or a Date read as empty and was dropped; the schema
now runs first. The tracker tests settle their requests instead of leaving timers.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* refactor(mobile): type the direct socket's request as the wire request

The typecheck now proves every direct request carries the transport fields, and the
native-verb fence says why it drops an envelope. No behaviour change.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb
This commit is contained in:
Jinwoo Hong
2026-10-12 04:02:25 -04:00
committed by GitHub
parent 14ec470146
commit 7cb30b7f8f
31 changed files with 800 additions and 97 deletions
@@ -6,7 +6,7 @@ import {
BRIDGE_MAX_ROUTE_PATHNAME_CHARS,
BRIDGE_MAX_SUBSCRIPTIONS
} from './bridge/bridge-caps'
import { BRIDGE_FAULT_GRANT } from './bridge/bridge-envelope'
import { BRIDGE_FAULT_GRANT, BRIDGE_RPC_ENVELOPE_GRANT } from './bridge/bridge-envelope'
import { routeViewOf } from './page-route-policy'
describe('init and state', () => {
@@ -40,6 +40,7 @@ describe('init and state', () => {
// What the shell will do for the page, and what makes its `navigate` frame acceptable.
native: [
BRIDGE_FAULT_GRANT,
BRIDGE_RPC_ENVELOPE_GRANT,
'navigate',
'storage',
'externalLink',
@@ -144,7 +145,11 @@ describe('init and state', () => {
if (init.type !== 'init') {
throw new Error('expected an init frame')
}
expect(init.grants.native).toEqual([BRIDGE_FAULT_GRANT, ...view.routeGrants])
expect(init.grants.native).toEqual([
BRIDGE_FAULT_GRANT,
BRIDGE_RPC_ENVELOPE_GRANT,
...view.routeGrants
])
const pair = (init.pageRouteGrants ?? []).find((entry) => entry.pathname === '/h/[hostId]')
expect(pair?.grants).toEqual(view.routeGrants)
// The optional name is in both, so the case is the lane and not two equal required lists.
@@ -292,7 +292,12 @@ describe('grants scoped to the route the page was opened for', () => {
const bridge = harness({ routeGrants: navigationOnly })
bridge.host.receive(clientFrame({ type: 'ready' }))
const init = bridge.last()
expect(init.type === 'init' && init.grants.native).toEqual(['fault', 'navigate', 'storage'])
expect(init.type === 'init' && init.grants.native).toEqual([
'fault',
'rpcEnvelope',
'navigate',
'storage'
])
})
it('refuses a verb that route never asked for', async () => {
@@ -69,6 +69,7 @@ export class BridgeHostRequests {
* are the whole boundary, and a test that reads only `client.requests` sees only this half. */
private forward(message: RequestMessage): Promise<RpcResponse> {
const { client } = this.deps
// An envelope is wire fields for a desktop RPC; a native verb never reaches the wire.
if (isBridgeNativeMethod(message.method)) {
return this.deps.serveNative(message.id, message.method, message.params)
}
@@ -78,8 +79,12 @@ export class BridgeHostRequests {
message.params,
this.deps.readClientIdentity()
)
if (message.options !== undefined) {
return client.sendRequest(message.method, params, message.options)
const options =
message.envelope === undefined
? message.options
: { ...message.options, envelope: message.envelope }
if (options !== undefined) {
return client.sendRequest(message.method, params, options)
}
return 'params' in message
? client.sendRequest(message.method, params)
@@ -77,7 +77,8 @@ export function createBridgeHostStreamFrames(deps: {
id,
message.method,
substituteBridgePageClientIdentity(message.params, readClientIdentity()),
lane === 'serve'
lane === 'serve',
message.envelope
)
} catch (error) {
sendError(id, error)
@@ -16,6 +16,7 @@ import {
import type { BridgeBinaryEvent } from './bridge/bridge-screencast-binary'
import type { BrowserScreencastFrame } from '../transport/browser-screencast-protocol'
import type { RpcClient } from '../transport/rpc-client'
import type { RpcEnvelopeFields } from '../transport/rpc-envelope-fields'
/** Derived, so an arm added to the envelope's closed list is a compile error here rather than a
* reason this module never sends. */
@@ -116,7 +117,13 @@ export class BridgeHostSubscriptions {
* Asked for only then, because encoding one costs the shell a base64 pass over the whole image
* and a page with no listener would pay for frames it drops.
*/
start(id: string, method: string, params: unknown, wantsBinary = false): void {
start(
id: string,
method: string,
params: unknown,
wantsBinary = false,
envelope?: RpcEnvelopeFields
): void {
const record: OpenSubscription = {
unsubscribe: () => undefined,
seq: 0,
@@ -133,13 +140,18 @@ export class BridgeHostSubscriptions {
: null
}
this.open.set(id, record)
const onBinaryFrame = wantsBinary
? (frame: BrowserScreencastFrame) => this.deliverBinaryFrame(id, frame)
: undefined
let unsubscribe: () => void
try {
unsubscribe = this.options.client.subscribe(
method,
params,
(payload) => this.deliver(id, payload),
wantsBinary ? { onBinaryFrame: (frame) => this.deliverBinaryFrame(id, frame) } : undefined
onBinaryFrame === undefined && envelope === undefined
? undefined
: { onBinaryFrame, envelope }
)
} catch (error) {
this.open.delete(id)
@@ -1,5 +1,6 @@
import type { RpcClient, SendRequestOptions } from '../transport/rpc-client'
import type { BrowserScreencastFrame } from '../transport/browser-screencast-protocol'
import type { RpcEnvelopeFields } from '../transport/rpc-envelope-fields'
import type { ConnectionState, RpcResponse } from '../transport/types'
import { BRIDGE_PROTOCOL_VERSION } from './bridge/bridge-envelope'
@@ -17,6 +18,7 @@ export type OpenStream = {
emit: (payload: unknown) => void
/** Null when the host did not ask for binary, which is what says `wantsBinary` reached the client. */
emitBinary: ((frame: BrowserScreencastFrame) => void) | null
envelope: RpcEnvelopeFields | undefined
unsubscribes: number
}
@@ -54,6 +56,7 @@ export function createFakeRpcClient(getters: ClientGetters = {}): FakeRpcClient
params,
emit: onData,
emitBinary: options?.onBinaryFrame ?? null,
envelope: options?.envelope,
unsubscribes: 0
}
streams.push(stream)
+7 -2
View File
@@ -7,6 +7,7 @@ import { createBridgeHostStreamFrames } from './bridge-host-stream-frames'
import {
BRIDGE_FAULT_GRANT,
BRIDGE_PROTOCOL_VERSION,
BRIDGE_RPC_ENVELOPE_GRANT,
readBridgeClientMessage,
type BridgeClientMessage,
type BridgeConnectionSnapshot,
@@ -69,8 +70,12 @@ export type BridgeHost = {
*/
export function createBridgeHost(options: BridgeHostOptions): BridgeHost {
const { client, buildId, sessionId, pageRoutes, host } = options
// The protocol's own grant rides with every session; the rest is what this route asked for.
const granted: readonly string[] = [BRIDGE_FAULT_GRANT, ...options.routeGrants]
// The protocol's own grants ride with every session; the rest is what this route asked for.
const granted: readonly string[] = [
BRIDGE_FAULT_GRANT,
BRIDGE_RPC_ENVELOPE_GRANT,
...options.routeGrants
]
// Checked here for the reason the route is: a pair the page's reader would refuse takes the whole
// `init` with it, and a session that never gets one is worse than one that never starts.
const parsedRouteGrants =
@@ -86,3 +86,24 @@ export class BridgeClientNotNativeVerbError extends Error {
this.name = 'BridgeClientNotNativeVerbError'
}
}
/** Extra wire fields for a shell that granted no `rpcEnvelope`. Refused rather than dropped, because
* the call would otherwise run on the desktop without what its caller said it needs. */
export class BridgeRpcEnvelopeUngrantedError extends Error {
readonly code = 'bridge_rpc_envelope_ungranted'
constructor() {
super('this app version cannot send extra request fields; update the Orca app')
this.name = 'BridgeRpcEnvelopeUngrantedError'
}
}
/** Extra wire fields that are not a plain JSON object, which the shell's reader would drop. */
export class BridgeRpcEnvelopeInvalidError extends Error {
readonly code = 'bridge_rpc_envelope_invalid'
constructor() {
super('extra request fields must be a plain JSON object')
this.name = 'BridgeRpcEnvelopeInvalidError'
}
}
@@ -0,0 +1,59 @@
import type { SendRequestOptions } from '../../transport/rpc-client'
import {
RpcEnvelopeFieldsSchema,
type RpcEnvelopeFields
} from '../../transport/rpc-envelope-fields'
import {
BridgeRpcEnvelopeInvalidError,
BridgeRpcEnvelopeUngrantedError
} from './bridge-client-errors'
import { BRIDGE_RPC_ENVELOPE_GRANT } from './bridge-frame-fields'
type HasGrant = (name: string) => boolean
/**
* The `envelope` a page frame carries for a caller's extra wire fields; throws instead of sending.
*
* Absent and empty both mean no field, so a caller passing neither posts today's exact frame. The
* check is the shell reader's own schema: a frame that reader drops is answered by nothing, and the
* caller would wait for the life of the page.
*/
export function readBridgeRpcEnvelope(
fields: RpcEnvelopeFields | undefined,
hasGrant: HasGrant
): RpcEnvelopeFields | undefined {
if (fields === undefined) {
return undefined
}
// Checked before the empty shortcut, so `[]`, a `Date` or `null` is refused, not read as empty.
const parsed = RpcEnvelopeFieldsSchema.safeParse(fields)
if (!parsed.success) {
throw new BridgeRpcEnvelopeInvalidError()
}
if (Object.keys(parsed.data).length === 0) {
return undefined
}
if (!hasGrant(BRIDGE_RPC_ENVELOPE_GRANT)) {
throw new BridgeRpcEnvelopeUngrantedError()
}
return parsed.data
}
/**
* A request's `options` and `envelope` frame fields. A bag that held only the envelope crosses as
* no `options`, so the shell replays the arity it did before the envelope existed.
*/
export function readBridgeRequestFrameOptions(
requestOptions: SendRequestOptions | undefined,
hasGrant: HasGrant
): { options?: Omit<SendRequestOptions, 'envelope'>; envelope?: RpcEnvelopeFields } {
if (requestOptions === undefined) {
return {}
}
const { envelope: fields, ...options } = requestOptions
const envelope = readBridgeRpcEnvelope(fields, hasGrant)
return {
...(fields !== undefined && Object.keys(options).length === 0 ? {} : { options }),
...(envelope === undefined ? {} : { envelope })
}
}
@@ -1,4 +1,5 @@
import type { BrowserScreencastFrame } from '../../transport/browser-screencast-protocol'
import type { RpcEnvelopeFields } from '../../transport/rpc-envelope-fields'
import {
BRIDGE_PROTOCOL_VERSION,
type BridgeClientMessage,
@@ -72,7 +73,8 @@ export class BridgeClientSubscriptions {
method: string,
params: unknown,
onData: (result: unknown) => void,
onBinaryFrame?: (frame: BrowserScreencastFrame) => void
onBinaryFrame?: (frame: BrowserScreencastFrame) => void,
envelope?: RpcEnvelopeFields
): boolean {
this.streams.set(id, { onData, onBinaryFrame, lastSeq: 0, unackedFrames: 0, unackedBytes: 0 })
const sent = this.options.send({
@@ -83,7 +85,8 @@ export class BridgeClientSubscriptions {
params,
// Asked for only when there is something to hand the frames to, so a shell that pays to
// encode binary is one a listener is waiting on.
...(onBinaryFrame === undefined ? {} : { wantsBinary: true })
...(onBinaryFrame === undefined ? {} : { wantsBinary: true }),
...(envelope === undefined ? {} : { envelope })
})
if (sent) {
return true
@@ -1,6 +1,7 @@
import { z } from 'zod'
import { salvagedOptional } from '../../../../src/shared/zod-salvage'
import { NODE_PLATFORM_NAMES } from '../../transport/mobile-runtime-host-platform'
import { RpcEnvelopeFieldsSchema } from '../../transport/rpc-envelope-fields'
import { isRpcResponse } from '../../transport/rpc-response-shape'
import type { RpcResponse } from '../../transport/types'
import { BridgeErrorCaptureSchema } from './bridge-error-capture'
@@ -18,7 +19,8 @@ export {
BRIDGE_FOREGROUND_NUDGE_REASONS,
BRIDGE_ID_PATTERN,
BRIDGE_NAVIGATE_BACK_NOTIFY,
BRIDGE_PROTOCOL_VERSION
BRIDGE_PROTOCOL_VERSION,
BRIDGE_RPC_ENVELOPE_GRANT
} from './bridge-frame-fields'
import {
isPageStorageKey,
@@ -121,7 +123,7 @@ export const BridgeInitStorageSchema = z
)
.refine((entries) => Object.keys(entries).length <= PAGE_STORAGE_MAX_ENTRIES)
/** Pinned against `SendRequestOptions` in this module's test. */
/** Pinned against `SendRequestOptions` in this module's test; its `envelope` is a frame field. */
export const BridgeSendRequestOptionsSchema = z.object({
timeoutMs: z.number().int().positive().optional(),
budgetSpansConnect: z.boolean().optional(),
@@ -179,7 +181,9 @@ const BridgeClientMessageSchema = z.discriminatedUnion('type', [
// Absent stays absent: `sendRequest(method)` and `sendRequest(method, undefined)` are different
// calls to the recorder, so the host replays the arity the page used.
params: z.unknown().optional(),
options: BridgeSendRequestOptionsSchema.optional()
options: BridgeSendRequestOptionsSchema.optional(),
// Verbatim like a reply's `payload`, bounded by the frame caps `params` already has.
envelope: RpcEnvelopeFieldsSchema.optional()
}),
z.object({
v: versionSchema,
@@ -187,7 +191,8 @@ const BridgeClientMessageSchema = z.discriminatedUnion('type', [
id: idSchema,
method: methodSchema,
params: z.unknown(),
wantsBinary: z.boolean().optional()
wantsBinary: z.boolean().optional(),
envelope: RpcEnvelopeFieldsSchema.optional()
}),
z.object({
v: versionSchema,
@@ -71,3 +71,10 @@ export const BRIDGE_NAVIGATE_BACK_NOTIFY = 'navigate-back'
* to say so.
*/
export const BRIDGE_EXTERNAL_LINK_GRANT = 'externalLink'
/**
* The shell copies a `request` or `subscribe` frame's `envelope` onto the wire request, under the
* transport's own fields. A protocol grant like `fault`: a page asks first because an older shell
* drops the key and would send the request without the fields the caller asked for.
*/
export const BRIDGE_RPC_ENVELOPE_GRANT = 'rpcEnvelope'
@@ -3,6 +3,7 @@ import { MOBILE_WEB_SHELL_GRANTS } from '../page-route-policy'
import {
BRIDGE_FAULT_GRANT,
BRIDGE_PROTOCOL_VERSION,
BRIDGE_RPC_ENVELOPE_GRANT,
type BridgeConnectionSnapshot,
type BridgeHostMessage,
type BridgeInitHost,
@@ -27,6 +28,7 @@ import {
*/
export const BRIDGE_NATIVE_GRANTS: readonly string[] = [
BRIDGE_FAULT_GRANT,
BRIDGE_RPC_ENVELOPE_GRANT,
...MOBILE_WEB_SHELL_GRANTS
]
@@ -1,4 +1,3 @@
import type { BrowserScreencastFrame } from '../../transport/browser-screencast-protocol'
import type { RpcClient, SendRequestOptions } from '../../transport/rpc-client'
import type { ConnectionState, RpcResponse, RpcSuccess } from '../../transport/types'
import {
@@ -27,6 +26,7 @@ import { createBridgeClientNotifications } from './bridge-client-notifications'
import type { BridgeSafeAreaInsets } from './bridge-safe-area-insets'
import { createPageBackConsumers, type PageBackConsumer } from './page-back-consumers'
import { BridgeClientRequests } from './bridge-client-requests'
import { readBridgeRequestFrameOptions, readBridgeRpcEnvelope } from './bridge-client-rpc-envelope'
import { BridgeClientSubscriptions } from './bridge-client-subscriptions'
import { isBridgeNativeMethod, type BridgeNativeVerb } from './bridge-native-verbs'
import { BRIDGE_ROUTE_PARAM_CLEAR, type BridgeClearableRouteParam } from './bridge-route-update'
@@ -45,6 +45,8 @@ export {
BridgeClientNotReadyError,
BridgeReplyRefusedError,
BridgeRequestOversizedError,
BridgeRpcEnvelopeInvalidError,
BridgeRpcEnvelopeUngrantedError,
BridgeSendFailedError,
BridgeShellReplacedError
} from './bridge-client-errors'
@@ -205,6 +207,8 @@ export function createBridgeRpcClient(options: BridgeRpcClientOptions): BridgeRp
}
}
const hasGrant = (name: string) => shellSession.current()?.grants.native.includes(name) === true
/** For the members whose contract is a boolean: a frame that did not leave is a false, whichever
* of the two reasons it was. */
const posted = (frame: BridgeClientMessage): boolean => sendFrame(frame) === 'sent'
@@ -334,8 +338,10 @@ export function createBridgeRpcClient(options: BridgeRpcClientOptions): BridgeRp
)
}
const [method, params, requestOptions] = args
const id = nextId()
return new Promise<RpcResponse>((resolve, reject) => {
// A refused envelope throws here, which rejects before an id is minted or a frame is posted.
const frameOptions = readBridgeRequestFrameOptions(requestOptions, hasGrant)
const id = nextId()
requests.open(id, { resolve, reject })
const outcome = sendFrame({
v: BRIDGE_PROTOCOL_VERSION,
@@ -347,7 +353,7 @@ export function createBridgeRpcClient(options: BridgeRpcClientOptions): BridgeRp
// as `sendRequest(m)`; no call site passes one, and no wire that carries `undefined` exists
// to carry it. The spread is what states the intent for a carrier that would.
...(args.length > 1 ? { params } : {}),
...(requestOptions === undefined ? {} : { options: requestOptions })
...frameOptions
})
if (outcome !== 'sent') {
requests.abandon(id)
@@ -364,7 +370,7 @@ export function createBridgeRpcClient(options: BridgeRpcClientOptions): BridgeRp
method: string,
params: unknown,
onData: (result: unknown) => void,
subscribeOptions?: { onBinaryFrame?: (frame: BrowserScreencastFrame) => void }
subscribeOptions?: Parameters<RpcClient['subscribe']>[3]
): () => void {
requireSession()
if (closed) {
@@ -376,10 +382,12 @@ export function createBridgeRpcClient(options: BridgeRpcClientOptions): BridgeRp
if (subscriptions.size >= BRIDGE_MAX_SUBSCRIPTIONS) {
throw new BridgeClientCapExceededError(`over ${BRIDGE_MAX_SUBSCRIPTIONS} subscriptions`)
}
const { onBinaryFrame, envelope: fields } = subscribeOptions ?? {}
const envelope = readBridgeRpcEnvelope(fields, hasGrant)
const id = nextId()
// A frame that never left already told the listener and gave the slot back; the caller still
// gets a dispose, because it has no way to know which of the two it is holding.
if (!subscriptions.open(id, method, params, onData, subscribeOptions?.onBinaryFrame)) {
if (!subscriptions.open(id, method, params, onData, onBinaryFrame, envelope)) {
return () => undefined
}
let disposed = false
@@ -411,7 +419,7 @@ export function createBridgeRpcClient(options: BridgeRpcClientOptions): BridgeRp
send: posted,
requireSession,
isClosed: () => closed,
hasGrant: (name) => shellSession.current()?.grants.native.includes(name) === true
hasGrant
})
const unsubscribeFromMessages = options.onMessage(receive)
@@ -0,0 +1,227 @@
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import { createBridgeInitFrame } from './bridge-init-frame'
import {
BRIDGE_PROTOCOL_VERSION,
BRIDGE_RPC_ENVELOPE_GRANT,
readBridgeClientMessage,
readBridgeHostMessage
} from './bridge-envelope'
import {
BridgeRpcEnvelopeInvalidError,
BridgeRpcEnvelopeUngrantedError
} from './bridge-client-errors'
import { INIT, createPageClient, readError } from './bridge-page-client-test-harness'
import { createFakeBridgePortPair } from './bridge-port-pair-test-harness'
const ENVELOPE = { executionHost: 'runtime:a', nested: { list: [1, 'two', null, true] } }
function grantedPage() {
const page = createPageClient()
page.deliver({
...INIT,
grants: { ...INIT.grants, native: [BRIDGE_RPC_ENVELOPE_GRANT] }
})
return page
}
function frameKey(json: string): string {
return json.replace(/"id":"[^"]+"/, '"id":"_"')
}
beforeEach(() => {
vi.useFakeTimers()
})
afterEach(() => {
vi.useRealTimers()
})
describe('bridge rpc envelope: page side', () => {
it('posts the same bytes as before when there is no envelope or an empty one', () => {
const page = grantedPage()
void page.client.sendRequest('worktree.ps', { a: 1 }, { timeoutMs: 50 })
void page.client.sendRequest('worktree.ps', { a: 1 }, { timeoutMs: 50, envelope: {} })
page.client.subscribe('terminal.subscribe', { terminal: 't' }, () => undefined)
page.client.subscribe('terminal.subscribe', { terminal: 't' }, () => undefined, {
envelope: {}
})
const [, request, emptyRequest, subscribe, emptySubscribe] = page.sent.map(frameKey)
expect(request).toBe(
'{"v":1,"type":"request","id":"_","method":"worktree.ps","params":{"a":1},"options":{"timeoutMs":50}}'
)
expect(emptyRequest).toBe(request)
expect(subscribe).toBe(
'{"v":1,"type":"subscribe","id":"_","method":"terminal.subscribe","params":{"terminal":"t"}}'
)
expect(emptySubscribe).toBe(subscribe)
})
it('posts no options when the only option was the envelope, and keeps an empty bag', () => {
const page = grantedPage()
void page.client.sendRequest('worktree.ps', { a: 1 }, { envelope: {} })
void page.client.sendRequest('worktree.ps', { a: 1 }, { envelope: ENVELOPE })
void page.client.sendRequest('worktree.ps', { a: 1 }, {})
const [, emptyOnly, envelopeOnly, emptyBag] = page.frames()
expect(emptyOnly).not.toHaveProperty('options')
expect(emptyOnly).not.toHaveProperty('envelope')
expect(envelopeOnly).not.toHaveProperty('options')
expect(envelopeOnly).toMatchObject({ envelope: ENVELOPE })
expect(emptyBag).toMatchObject({ options: {} })
})
it('carries a granted envelope beside params, out of options', () => {
const page = grantedPage()
void page.client.sendRequest('worktree.ps', undefined, { timeoutMs: 50, envelope: ENVELOPE })
page.client.subscribe('terminal.subscribe', { terminal: 't' }, () => undefined, {
envelope: ENVELOPE
})
const [, request, subscribe] = page.frames()
expect(request).toMatchObject({ options: { timeoutMs: 50 }, envelope: ENVELOPE })
expect(request).not.toHaveProperty('options.envelope')
expect(subscribe).toMatchObject({ type: 'subscribe', envelope: ENVELOPE })
})
it('refuses an envelope the shell did not grant, and posts nothing', async () => {
const page = createPageClient()
page.start()
const posted = page.sent.length
const refusal = await page.client
.sendRequest('worktree.ps', {}, { envelope: ENVELOPE })
.catch(readError)
expect(refusal).toBeInstanceOf(BridgeRpcEnvelopeUngrantedError)
expect(refusal).toMatchObject({ code: 'bridge_rpc_envelope_ungranted' })
expect(() =>
page.client.subscribe('terminal.subscribe', {}, () => undefined, { envelope: ENVELOPE })
).toThrow(BridgeRpcEnvelopeUngrantedError)
expect(page.sent).toHaveLength(posted)
})
it('refuses an envelope the shell reader would drop, instead of leaving the call pending', async () => {
const page = grantedPage()
const posted = page.sent.length
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: a caller that widened the type, which the check exists for.
const notJson = { when: new Date(0), skip: undefined } as unknown as Record<string, null>
const refusal = await page.client
.sendRequest('worktree.ps', {}, { envelope: notJson })
.catch(readError)
expect(refusal).toBeInstanceOf(BridgeRpcEnvelopeInvalidError)
expect(refusal).toMatchObject({ code: 'bridge_rpc_envelope_invalid' })
expect(page.sent).toHaveLength(posted)
})
it.each([
['null', null],
['an array', []],
['a Date', new Date(0)]
])(
'refuses %s as an envelope with a rejection, never a throw or a silent drop',
async (_, value) => {
const page = grantedPage()
const posted = page.sent.length
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: a caller that widened the type, which the check exists for.
const envelope = value as unknown as Record<string, null>
let answer: Promise<unknown> | undefined
expect(() => {
answer = page.client.sendRequest('worktree.ps', {}, { envelope })
}).not.toThrow()
expect(await answer?.catch(readError)).toBeInstanceOf(BridgeRpcEnvelopeInvalidError)
expect(() =>
page.client.subscribe('terminal.subscribe', {}, () => undefined, { envelope })
).toThrow(BridgeRpcEnvelopeInvalidError)
expect(page.sent).toHaveLength(posted)
}
)
})
describe('bridge rpc envelope: shell side', () => {
it('grants every session the envelope', async () => {
const pair = createFakeBridgePortPair()
await pair.flush()
expect(pair.client.getShellSession()?.grants.native).toContain(BRIDGE_RPC_ENVELOPE_GRANT)
})
it('hands a request and a subscribe envelope to the native client options', async () => {
const pair = createFakeBridgePortPair()
await pair.flush()
void pair.client.sendRequest('worktree.ps', { a: 1 }, { timeoutMs: 50, envelope: ENVELOPE })
void pair.client.sendRequest('worktree.ps', undefined, { envelope: ENVELOPE })
pair.client.subscribe('terminal.subscribe', { terminal: 't' }, () => undefined, {
envelope: ENVELOPE
})
await pair.flush()
expect(pair.rpc.requests.map((request) => request.args)).toEqual([
['worktree.ps', { a: 1 }, { timeoutMs: 50, envelope: ENVELOPE }],
['worktree.ps', undefined, { envelope: ENVELOPE }]
])
expect(pair.rpc.streams.map((stream) => stream.envelope)).toEqual([ENVELOPE])
})
it('calls the native client exactly as before when no envelope crossed', async () => {
const pair = createFakeBridgePortPair()
await pair.flush()
void pair.client.sendRequest('worktree.ps')
void pair.client.sendRequest('worktree.ps', { a: 1 })
void pair.client.sendRequest('worktree.ps', { a: 1 }, { timeoutMs: 50 })
pair.client.subscribe('terminal.subscribe', { terminal: 't' }, () => undefined)
await pair.flush()
expect(pair.rpc.requests.map((request) => request.args)).toEqual([
['worktree.ps'],
['worktree.ps', { a: 1 }],
['worktree.ps', { a: 1 }, { timeoutMs: 50 }]
])
expect(pair.rpc.streams.map((stream) => stream.envelope)).toEqual([undefined])
})
it('refuses a frame whose envelope is not a plain JSON object', () => {
for (const envelope of [[1], 'x', null, 3]) {
const frame = { v: 1, type: 'request', id: 'a'.repeat(22), method: 'm', envelope }
expect(readBridgeClientMessage(JSON.stringify(frame))).toEqual({
ok: false,
refusal: 'unrecognised-message'
})
}
})
})
describe('bridge rpc envelope: mixed versions', () => {
it('an older page reading the new init ignores a grant name it does not know', () => {
const init = createBridgeInitFrame({
sessionId: 's',
buildId: 'b',
connection: INIT.connection,
route: { pathname: '/h/a' },
pageRoutes: [],
granted: [BRIDGE_RPC_ENVELOPE_GRANT, 'someFutureGrant'],
host: { id: 'a', name: 'A', endpoint: 'ws://a', lastConnected: 0 },
storage: {}
})
const read = readBridgeHostMessage(JSON.stringify(init))
expect(read.ok && read.message.type === 'init' && read.message.grants.native).toEqual([
BRIDGE_RPC_ENVELOPE_GRANT,
'someFutureGrant'
])
})
it('a shell drops a frame key it does not declare and still serves the frame', () => {
// The rule an older shell applies to `envelope`: unknown keys are stripped, never refused.
const frame = {
v: BRIDGE_PROTOCOL_VERSION,
type: 'request',
id: 'a'.repeat(22),
method: 'm',
params: {},
laterField: { executionHost: 'runtime:a' }
}
const read = readBridgeClientMessage(JSON.stringify(frame))
expect(read).toEqual({
ok: true,
message: {
v: BRIDGE_PROTOCOL_VERSION,
type: 'request',
id: 'a'.repeat(22),
method: 'm',
params: {}
}
})
})
})
@@ -63,9 +63,8 @@ function registryClient() {
let id = 0
const registry = new RpcClientStreamRegistry({
nextId: () => `rpc-${++id}`,
deviceToken: 'device-token',
getState: () => 'connected',
sendEncrypted: (request) => {
sendFrame: (request) => {
sent.push(readSentFrame(request))
return true
}
@@ -131,9 +131,8 @@ function directRig(): TransportRig {
let nextId = 0
const registry = new RpcClientStreamRegistry({
nextId: () => `rpc-${++nextId}`,
deviceToken: 'device-token',
getState: () => 'connected',
sendEncrypted: (request) => {
sendFrame: (request) => {
sent.push(readSentRequest(request))
return true
}
@@ -1,5 +1,6 @@
import type { ConnectionState, RpcResponse } from '../../transport/types'
import type { RpcClient } from '../../transport/rpc-client'
import { buildDirectRpcWireRequest } from '../../transport/rpc-envelope-fields'
import { RpcClientRequestTracker } from '../../transport/rpc-client-request-tracker'
import { RpcClientStreamRegistry } from '../../transport/rpc-client-stream-registry'
import { createStableLogicalRpcClient } from '../../transport/stable-logical-rpc-client'
@@ -81,16 +82,14 @@ export class ScriptedRpcTransport {
throw new Error('Scripted transport disconnected')
}
},
deviceToken: DEVICE_TOKEN,
sendEncrypted: (value) => {
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the physical client publishes the frame this transport just serialized.
const payload = value as { id: string; method: string; params: unknown }
sendFrame: (frame) => {
const payload = buildDirectRpcWireRequest(frame, DEVICE_TOKEN)
const name = this.wireNames.shift()
if (!name) {
throw new Error('Unbound physical request')
}
this.bindings.set(name, { id: payload.id, params: payload.params, completed: false })
this.publish(name, value)
this.publish(name, payload)
return true
}
})
@@ -159,11 +158,9 @@ export class ScriptedRpcTransport {
// routed through the session that published its subscribe.
const streams: RpcClientStreamRegistry = new RpcClientStreamRegistry({
nextId: () => this.nextFrameId(),
deviceToken: DEVICE_TOKEN,
getState: () => this.state,
sendEncrypted: (value) => {
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the stream registry publishes the frame it just built.
const payload = value as { id: string; method: string; params: unknown }
sendFrame: (frame) => {
const payload = buildDirectRpcWireRequest(frame, DEVICE_TOKEN)
const name = this.occurrence(payload.method)
// Only a subscribe opens a stream. The registry sends its unsubscribes through this same
// hook, and filing one under `openStreams` made a frame aimed at an unsubscribe name route
@@ -179,7 +176,7 @@ export class ScriptedRpcTransport {
// already-registered id under a fresh occurrence, so the latest payload is the one a
// teardown observation should name.
this.streamPayloads.set(payload.id, name)
this.publish(name, value)
this.publish(name, payload)
return true
}
})
+7 -8
View File
@@ -6,6 +6,7 @@ import {
RpcClientReconnectSchedule,
RPC_RECONNECT_ATTEMPT_LIMIT
} from './rpc-client-reconnect-schedule'
import { buildDirectRpcWireRequest, type RpcRequestFrame } from './rpc-envelope-fields'
import { RpcClientRequestTracker } from './rpc-client-request-tracker'
import { RpcClientSocketCloseController } from './rpc-client-socket-close-controller'
import { RpcClientSocketFactory } from './rpc-client-socket-factory'
@@ -59,16 +60,14 @@ export class DirectRpcClient implements RpcClient {
})
this.streams = new RpcClientStreamRegistry({
nextId: () => this.nextId(),
deviceToken,
getState: () => this.connectionState.get(),
sendEncrypted: (request) => this.sendEncrypted(request)
sendFrame: (frame) => this.sendFrame(frame)
})
this.requests = new RpcClientRequestTracker({
nextId: () => this.nextId(),
deviceToken,
getState: () => this.connectionState.get(),
waitForConnected: (timeoutMs) => this.waitForConnected(timeoutMs),
sendEncrypted: (request) => this.sendEncrypted(request)
sendFrame: (frame) => this.sendFrame(frame)
})
this.liveness = new RpcSessionLivenessWatchdog({
transport: 'direct',
@@ -285,9 +284,10 @@ export class DirectRpcClient implements RpcClient {
this.requests.rejectAll(reason, { deliveryUnknown: true })
}
private sendEncrypted(request: unknown): boolean {
// The one place a direct request gains its credential and its caller's envelope.
private sendFrame(frame: RpcRequestFrame): boolean {
if (this.socketSession) {
return this.socketSession.sendEncrypted(request)
return this.socketSession.sendEncrypted(buildDirectRpcWireRequest(frame, this.deviceToken))
}
console.log('[net] sendEncrypted FAILED — channel not ready', {
hasWs: false,
@@ -301,9 +301,8 @@ export class DirectRpcClient implements RpcClient {
if (this.getState() !== 'connected') {
return false
}
return this.sendEncrypted({
return this.sendFrame({
id: `${LIVENESS_REQUEST_ID_PREFIX}${this.nextId()}`,
deviceToken: this.deviceToken,
method: 'status.get'
})
}
@@ -367,4 +367,44 @@ describe('mobile relay RPC session', () => {
vi.useRealTimers()
}
})
it('sends requests and streams with no envelope byte for byte as before', async () => {
const { session } = await authenticateSession()
void session.sendRequest('status.get', { a: 1 })
session.subscribe('terminal.subscribe', { terminal: 't1' }, () => undefined)
await vi.waitFor(() => expect(fakes.sendText).toHaveBeenCalledTimes(2))
expect(
fakes.sendText.mock.calls.map((call) => String(call[0]).replace(/"id":"[^"]+"/, ''))
).toEqual([
'{,"method":"status.get","params":{"a":1},"deviceToken":"device-token"}',
'{,"method":"terminal.subscribe","params":{"terminal":"t1"},"deviceToken":"device-token"}'
])
})
it('merges an envelope under the transport fields on requests, subscribes and unsubscribes', async () => {
const { session } = await authenticateSession()
const envelope = {
executionHost: 'runtime:a',
id: 'x',
method: 'x',
params: 'x',
deviceToken: 'x'
}
void session.sendRequest('status.get', { a: 1 }, { envelope })
const dispose = session.subscribe('terminal.subscribe', { terminal: 't1' }, () => undefined, {
envelope
})
await vi.waitFor(() => expect(fakes.sendText).toHaveBeenCalledTimes(2))
dispose()
const frames = fakes.sendText.mock.calls.map((call) => JSON.parse(String(call[0])))
expect(frames).toEqual([
expect.objectContaining({ method: 'status.get', params: { a: 1 } }),
expect.objectContaining({ method: 'terminal.subscribe', params: { terminal: 't1' } }),
expect.objectContaining({ method: 'terminal.unsubscribe' })
])
for (const frame of frames) {
expect(frame).toMatchObject({ executionHost: 'runtime:a', deviceToken: 'device-token' })
expect(frame.id).not.toBe('x')
}
})
})
@@ -7,6 +7,11 @@ import { MobileRelayE2eeLink } from './mobile-relay-e2ee-link'
import { MobileRelayRpcStreams } from './mobile-relay-rpc-streams'
import { MobileE2EEAuthenticationError } from './mobile-e2ee-v2-physical-channel'
import { markRpcDeliveryUnknown } from './rpc-delivery-ambiguity'
import {
withRpcEnvelopeFields,
type RpcEnvelopeFields,
type RpcRequestFrame
} from './rpc-envelope-fields'
import { openRpcRequestBudget, resolvePostConnectRequestTimeout } from './rpc-request-budget'
import { isRpcResponse } from './rpc-response-shape'
import { RelayDialStageTracker, type RelayDialStageSource } from './relay-dial-stage'
@@ -102,7 +107,8 @@ export function connectMobileRelayRpcSession(args: {
async sendRequest(method, params, options) {
const budget = openRpcRequestBudget(options)
await waitForConnected(budget.timeoutMs)
return sendRpc(method, params, resolvePostConnectRequestTimeout(budget, requestTimeoutMs))
const timeoutMs = resolvePostConnectRequestTimeout(budget, requestTimeoutMs)
return sendRpc(method, params, timeoutMs, false, options?.envelope)
},
subscribe(method, params, listener, options) {
@@ -204,7 +210,8 @@ export function connectMobileRelayRpcSession(args: {
method: string,
params: unknown,
timeoutMs = requestTimeoutMs,
beforeConnected = false
beforeConnected = false,
envelope?: RpcEnvelopeFields
): Promise<RpcResponse> {
if (closed || (!beforeConnected && state !== 'connected')) {
return Promise.reject(new Error('relay session not connected'))
@@ -217,7 +224,7 @@ export function connectMobileRelayRpcSession(args: {
reject(markRpcDeliveryUnknown(new Error(`relay RPC timed out: ${method}`)))
}, timeoutMs)
pending.track(id, { resolve, reject, timer, written: false })
if (!sendFrame({ id, method, params })) {
if (!sendFrame({ id, method, params, envelope })) {
clearTimeout(timer)
pending.drop(id)
reject(new Error('relay E2EE channel not ready'))
@@ -227,8 +234,9 @@ export function connectMobileRelayRpcSession(args: {
})
}
function sendFrame(request: { id: string; method: string; params?: unknown }): boolean {
return link.sendText(JSON.stringify({ ...request, deviceToken: args.deviceToken }))
function sendFrame({ envelope, ...request }: RpcRequestFrame): boolean {
const wire = withRpcEnvelopeFields({ ...request, deviceToken: args.deviceToken }, envelope)
return link.sendText(JSON.stringify(wire))
}
function handleText(plaintext: string): void {
@@ -11,11 +11,13 @@ import {
import { buildReadyStreamUnsubscribe, isReadyIdStream } from './rpc-client-server-subscription'
import { isStreamingOpenerReply } from './rpc-acceptance-policies'
import type { RpcClient } from './rpc-client'
import type { RpcEnvelopeFields, RpcRequestFrame } from './rpc-envelope-fields'
import type { RpcResponse, RpcSuccess } from './types'
type StreamRecord = {
method: string
params: unknown
envelope?: RpcEnvelopeFields
listener: (result: unknown) => void
onBinaryFrame?: Parameters<RpcClient['subscribe']>[3] extends
| { onBinaryFrame?: infer Listener }
@@ -60,7 +62,8 @@ function buildParamsUnsubscribe(
type StreamManagerOptions = {
nextId: () => string
sendFrame: (request: { id: string; method: string; params?: unknown }) => boolean
// A stream's envelope rides its subscribe and unsubscribes alike, so routing reaches one host.
sendFrame: (frame: RpcRequestFrame) => boolean
waitForConnected: () => Promise<void>
}
@@ -68,7 +71,7 @@ export class MobileRelayRpcStreams {
private readonly streams = new Map<string, StreamRecord>()
private readonly cancelledSubscriptions = new Map<
string,
{ method: string; unsubscribe?: StreamUnsubscribe }
{ method: string; unsubscribe?: StreamUnsubscribe; envelope?: RpcEnvelopeFields }
>()
private readonly terminalListeners = new Map<number, (result: unknown) => void>()
private readonly terminalSnapshots = new Map<number, TerminalSnapshotState>()
@@ -87,6 +90,7 @@ export class MobileRelayRpcStreams {
const stream: StreamRecord = {
method,
params,
envelope: subscribeOptions?.envelope,
listener,
onBinaryFrame: subscribeOptions?.onBinaryFrame,
streamIds: new Set(),
@@ -99,7 +103,8 @@ export class MobileRelayRpcStreams {
.then(() => {
if (!stream.cancelled) {
stream.sendOrder = ++this.sendCount
if (!this.options.sendFrame({ id, method, params: stream.params })) {
const { envelope } = stream
if (!this.options.sendFrame({ id, method, params: stream.params, envelope })) {
this.fail(id, stream, 'Connection interrupted')
}
}
@@ -126,12 +131,20 @@ export class MobileRelayRpcStreams {
this.cancelledSubscriptions.delete(response.id)
} else if (result.type === 'snapshot' && cancelled.unsubscribe) {
this.cancelledSubscriptions.delete(response.id)
this.options.sendFrame({ id: this.options.nextId(), ...cancelled.unsubscribe })
this.options.sendFrame({
id: this.options.nextId(),
...cancelled.unsubscribe,
envelope: cancelled.envelope
})
} else if (typeof result.subscriptionId === 'string') {
this.cancelledSubscriptions.delete(response.id)
const unsubscribe = buildReadyStreamUnsubscribe(cancelled.method, result.subscriptionId)
if (unsubscribe) {
this.options.sendFrame({ id: this.options.nextId(), ...unsubscribe })
this.options.sendFrame({
id: this.options.nextId(),
...unsubscribe,
envelope: cancelled.envelope
})
}
}
}
@@ -208,7 +221,7 @@ export class MobileRelayRpcStreams {
const byParams = buildParamsUnsubscribe(stream.method, stream.params, id)
if (stream.method === 'terminal.subscribe') {
if (byParams) {
this.sendUnsubscribe(byParams, stream.sendOrder, id)
this.sendUnsubscribe(byParams, stream.sendOrder, stream.envelope, id)
}
} else {
const unsubscribe = stream.subscriptionId
@@ -216,12 +229,16 @@ export class MobileRelayRpcStreams {
: null
if (byParams && stream.method === 'session.tabs.subscribe' && !stream.receivedSnapshot) {
// The host registers cleanup only after resolving the initial snapshot.
this.cancelledSubscriptions.set(id, { method: stream.method, unsubscribe: byParams })
this.cancelledSubscriptions.set(id, {
method: stream.method,
unsubscribe: byParams,
envelope: stream.envelope
})
} else if (unsubscribe || byParams) {
this.sendUnsubscribe((unsubscribe ?? byParams)!, stream.sendOrder)
this.sendUnsubscribe((unsubscribe ?? byParams)!, stream.sendOrder, stream.envelope)
} else if (isReadyIdStream(stream.method)) {
// Keep only the cleanup route while the server assigns its subscription ID.
this.cancelledSubscriptions.set(id, { method: stream.method })
this.cancelledSubscriptions.set(id, { method: stream.method, envelope: stream.envelope })
}
}
}
@@ -232,6 +249,7 @@ export class MobileRelayRpcStreams {
private sendUnsubscribe(
unsubscribe: StreamUnsubscribe,
sendOrder: number,
envelope: RpcEnvelopeFields | undefined,
terminalRequestId?: string
): void {
if (this.hasNewerSlotOwner(unsubscribe, sendOrder)) {
@@ -241,7 +259,12 @@ export class MobileRelayRpcStreams {
const params = terminalRequestId
? { ...unsubscribe.params, requestId: terminalRequestId }
: unsubscribe.params
this.options.sendFrame({ id: this.options.nextId(), method: unsubscribe.method, params })
this.options.sendFrame({
id: this.options.nextId(),
method: unsubscribe.method,
params,
envelope
})
}
private hasNewerSlotOwner(unsubscribe: StreamUnsubscribe, sendOrder: number): boolean {
@@ -40,9 +40,8 @@ function directTransport() {
let id = 0
const registry = new RpcClientStreamRegistry({
nextId: () => `rpc-${++id}`,
deviceToken: 'device-token',
getState: () => 'connected',
sendEncrypted: (request) => {
sendFrame: (request) => {
sent.push(readSentFrame(request))
return true
}
@@ -1,6 +1,7 @@
import { markRpcDeliveryUnknown } from './rpc-delivery-ambiguity'
import { openRpcRequestBudget, resolvePostConnectRequestTimeout } from './rpc-request-budget'
import type { SendRequestOptions } from './rpc-client'
import type { RpcEnvelopeFields, RpcRequestFrame } from './rpc-envelope-fields'
import type { ConnectionState, RpcResponse } from './types'
const REQUEST_TIMEOUT_MS = 30_000
@@ -14,8 +15,7 @@ type RequestTrackerOptions = {
nextId: () => string
getState: () => ConnectionState
waitForConnected: (timeoutMs?: number) => Promise<void>
sendEncrypted: (request: unknown) => boolean
deviceToken: string
sendFrame: (frame: RpcRequestFrame) => boolean
}
export class RpcClientRequestTracker {
@@ -45,7 +45,8 @@ export class RpcClientRequestTracker {
return this.sendConnectedRequest(
method,
params,
resolvePostConnectRequestTimeout(budget, REQUEST_TIMEOUT_MS)
resolvePostConnectRequestTimeout(budget, REQUEST_TIMEOUT_MS),
requestOptions?.envelope
)
}
@@ -60,7 +61,8 @@ export class RpcClientRequestTracker {
private sendConnectedRequest(
method: string,
params: unknown,
timeoutMs: number
timeoutMs: number,
envelope?: RpcEnvelopeFields
): Promise<RpcResponse> {
return new Promise((resolve, reject) => {
const id = this.options.nextId()
@@ -83,14 +85,7 @@ export class RpcClientRequestTracker {
reject(error)
}
})
if (
!this.options.sendEncrypted({
id,
deviceToken: this.options.deviceToken,
method,
params
})
) {
if (!this.options.sendFrame({ id, method, params, envelope })) {
this.pending.delete(id)
clearTimeout(timeout)
reject(new Error('Connection interrupted'))
@@ -4,7 +4,7 @@ import { createDirectSocketFraming, type DirectSocketFraming } from './direct-so
import { isRpcResponse } from './rpc-response-shape'
import { isStaleRpcSocketEvent, logRpcSocketClose } from './rpc-socket-close-evidence'
import { describeSocketEvent, redactSocketEndpoint } from './socket-event-debug'
import type { ConnectionLogEmitter, ConnectionState, RpcResponse } from './types'
import type { ConnectionLogEmitter, ConnectionState, RpcRequest, RpcResponse } from './types'
const CONNECT_TIMEOUT_MS = 12_000
const HANDSHAKE_TIMEOUT_MS = 5_000
@@ -50,7 +50,7 @@ export class RpcClientSocketSession {
this.armConnectTimeout()
}
sendEncrypted(request: unknown): boolean {
sendEncrypted(request: RpcRequest): boolean {
if (this.socket.readyState === WebSocket.OPEN && this.framing) {
try {
if (this.framing.sendText(JSON.stringify(request))) {
@@ -15,9 +15,8 @@ function createRegistry(initialState: ConnectionState = 'connected') {
let id = 0
const registry = new RpcClientStreamRegistry({
nextId: () => `rpc-${++id}`,
deviceToken: 'device-token',
getState: () => state,
sendEncrypted: (request) => {
sendFrame: (request) => {
if (state !== 'connected') {
return false
}
@@ -15,17 +15,20 @@ import {
} from './rpc-subscription-result-shapes'
import { RpcClientTerminalStreamRouter } from './rpc-client-terminal-stream-router'
import * as sessionTabsStream from './rpc-client-session-tabs-stream'
import type { RpcEnvelopeFields, RpcRequestFrame } from './rpc-envelope-fields'
import type { ConnectionState, RpcResponse, RpcSuccess } from './types'
export type RpcStreamingListener = (result: unknown) => void
export type RpcStreamSubscribeOptions = {
onBinaryFrame?: (frame: BrowserScreencastFrame) => void
envelope?: RpcEnvelopeFields
}
type StreamRequest = {
method: string
params: unknown
envelope?: RpcEnvelopeFields
listener?: RpcStreamingListener
onBinaryFrame?: (frame: BrowserScreencastFrame) => void
subscriptionId?: string
@@ -36,9 +39,8 @@ type StreamRequest = {
type StreamRegistryOptions = {
nextId: () => string
deviceToken: string
getState: () => ConnectionState
sendEncrypted: (request: unknown) => boolean
sendFrame: (frame: RpcRequestFrame) => boolean
}
export class RpcClientStreamRegistry {
@@ -59,6 +61,7 @@ export class RpcClientStreamRegistry {
const stream: StreamRequest = {
method,
params,
envelope: subscribeOptions?.envelope,
listener,
onBinaryFrame: subscribeOptions?.onBinaryFrame
}
@@ -189,7 +192,7 @@ export class RpcClientStreamRegistry {
this.pendingBrowserRequestId !== response.id &&
this.activeBrowserRequestId !== response.id
) {
this.sendBrowserUnsubscribe(result.subscriptionId)
this.sendBrowserUnsubscribe(result.subscriptionId, stream.envelope)
this.remove(response.id)
return
}
@@ -224,7 +227,7 @@ export class RpcClientStreamRegistry {
}
const unsubscribe = buildRequestStreamUnsubscribe(stream?.method, stream?.params, id)
if (unsubscribe) {
this.sendRpc(unsubscribe.method, unsubscribe.params)
this.sendRpc(unsubscribe.method, unsubscribe.params, stream?.envelope)
}
this.remove(id)
}
@@ -256,30 +259,22 @@ export class RpcClientStreamRegistry {
}
const unsubscribe = buildReadyStreamUnsubscribe(stream.method, stream.subscriptionId)
if (unsubscribe) {
this.sendRpc(unsubscribe.method, unsubscribe.params)
this.sendRpc(unsubscribe.method, unsubscribe.params, stream.envelope)
}
}
private sendBrowserUnsubscribe(subscriptionId: string): void {
this.sendRpc('browser.screencast.unsubscribe', { subscriptionId })
private sendBrowserUnsubscribe(subscriptionId: string, envelope?: RpcEnvelopeFields): void {
this.sendRpc('browser.screencast.unsubscribe', { subscriptionId }, envelope)
}
private sendRpc(method: string, params: unknown): boolean {
return this.options.sendEncrypted({
id: this.options.nextId(),
deviceToken: this.options.deviceToken,
method,
params
})
// The stream's envelope rides its unsubscribes too: a routing field must reach the same host.
private sendRpc(method: string, params: unknown, envelope?: RpcEnvelopeFields): boolean {
return this.options.sendFrame({ id: this.options.nextId(), method, params, envelope })
}
private send(id: string, stream: StreamRequest): boolean {
return this.options.sendEncrypted({
id,
deviceToken: this.options.deviceToken,
method: stream.method,
params: stream.params
})
const { method, params, envelope } = stream
return this.options.sendFrame({ id, method, params, envelope })
}
private remove(id: string): void {
+3
View File
@@ -1,5 +1,6 @@
import type { BrowserScreencastFrame } from './browser-screencast-protocol'
import { DirectRpcClient } from './direct-rpc-client'
import type { RpcEnvelopeFields } from './rpc-envelope-fields'
import type { ConnectionLogSink, ConnectionState, ForegroundNudgeReason } from './types'
import type { UnvalidatedRpcRequestPort } from './unvalidated-rpc-request-port'
@@ -9,6 +10,8 @@ export type { SendRequestOptions } from './unvalidated-rpc-request-port'
type SubscribeOptions = {
onBinaryFrame?: (frame: BrowserScreencastFrame) => void
/** Extra top-level wire fields, sent with the subscribe and every resend and unsubscribe of it. */
envelope?: RpcEnvelopeFields
}
type StreamingListener = (result: unknown) => void
@@ -0,0 +1,223 @@
import { describe, expect, it } from 'vitest'
import { MobileRelayRpcStreams } from './mobile-relay-rpc-streams'
import { RpcClientRequestTracker } from './rpc-client-request-tracker'
import { RpcClientStreamRegistry } from './rpc-client-stream-registry'
import {
RPC_TRANSPORT_FIELDS,
buildDirectRpcWireRequest,
withRpcEnvelopeFields,
type RpcRequestFrame
} from './rpc-envelope-fields'
import type { RpcResponse } from './types'
const DEVICE_TOKEN = 'device-token'
const ENVELOPE = { executionHost: 'runtime:a' }
/** What the direct client writes to its socket, through the same builder it uses. */
function directWire() {
const sent: string[] = []
let counter = 0
return {
sent,
frames: () => sent.map((json) => JSON.parse(json)),
nextId: () => `rpc-${++counter}`,
sendFrame: (frame: RpcRequestFrame) => {
sent.push(JSON.stringify(buildDirectRpcWireRequest(frame, DEVICE_TOKEN)))
return true
}
}
}
function tracker(socket: ReturnType<typeof directWire>) {
return new RpcClientRequestTracker({
nextId: socket.nextId,
getState: () => 'connected',
waitForConnected: async () => undefined,
sendFrame: socket.sendFrame
})
}
function registry(socket: ReturnType<typeof directWire>) {
return new RpcClientStreamRegistry({
nextId: socket.nextId,
getState: () => 'connected',
sendFrame: socket.sendFrame
})
}
function relayStreams() {
const frames: RpcRequestFrame[] = []
let counter = 0
const streams = new MobileRelayRpcStreams({
nextId: () => `relay-${++counter}`,
sendFrame: (frame) => {
frames.push(frame)
return true
},
waitForConnected: async () => undefined
})
return { streams, frames }
}
function answered(id: string): RpcResponse {
return { id, ok: true, result: null, _meta: { runtimeId: 'runtime-1' } }
}
function streaming(id: string, result: unknown): RpcResponse {
return { id, ok: true, streaming: true, result, _meta: { runtimeId: 'runtime-1' } }
}
describe('withRpcEnvelopeFields', () => {
const request = { id: 'rpc-1', deviceToken: DEVICE_TOKEN, method: 'm', params: { a: 1 } }
it('hands back the request itself when there is nothing to merge', () => {
expect(withRpcEnvelopeFields(request, undefined)).toBe(request)
expect(withRpcEnvelopeFields(request, {})).toBe(request)
})
it('adds extra fields beside the transport ones', () => {
expect(withRpcEnvelopeFields(request, { executionHost: 'runtime:a', n: [1, null] })).toEqual({
...request,
executionHost: 'runtime:a',
n: [1, null]
})
})
it.each(RPC_TRANSPORT_FIELDS)('never lets the envelope set %s', (field) => {
const merged = withRpcEnvelopeFields(request, { [field]: 'forged' })
expect(merged).toEqual(request)
})
it('cannot fill in a transport field the request left out', () => {
const { params: _omitted, ...bare } = request
expect(JSON.stringify(withRpcEnvelopeFields(bare, { params: 'forged', x: 1 }))).toBe(
JSON.stringify({ x: 1, ...bare })
)
})
})
describe('direct transport envelope', () => {
it('sends a request with no envelope byte for byte as before', async () => {
const socket = directWire()
const requests = tracker(socket)
const answer = requests.sendRequest('status.get', { a: 1 }, { timeoutMs: 1000 })
await Promise.resolve()
expect(socket.sent).toEqual([
'{"id":"rpc-1","deviceToken":"device-token","method":"status.get","params":{"a":1}}'
])
requests.resolve(answered('rpc-1'))
await answer
})
it('writes a frame without params, like the liveness probe, as before', () => {
expect(
JSON.stringify(buildDirectRpcWireRequest({ id: 'p-1', method: 'status.get' }, DEVICE_TOKEN))
).toBe('{"id":"p-1","deviceToken":"device-token","method":"status.get"}')
})
it('merges a request envelope under every transport field', async () => {
const socket = directWire()
const forged = Object.fromEntries(RPC_TRANSPORT_FIELDS.map((field) => [field, 'forged']))
const requests = tracker(socket)
const answer = requests.sendRequest(
'status.get',
{ a: 1 },
{ envelope: { ...forged, ...ENVELOPE } }
)
await Promise.resolve()
expect(socket.frames()).toEqual([
{
id: 'rpc-1',
deviceToken: DEVICE_TOKEN,
method: 'status.get',
params: { a: 1 },
...ENVELOPE
}
])
requests.resolve(answered('rpc-1'))
await answer
})
it('sends a subscribe with no envelope byte for byte as before', () => {
const socket = directWire()
registry(socket).subscribe('terminal.subscribe', { terminal: 't1' }, () => undefined)
expect(socket.sent).toEqual([
'{"id":"rpc-1","deviceToken":"device-token","method":"terminal.subscribe","params":{"terminal":"t1"}}'
])
})
it('keeps a stream envelope on its subscribe, its replay and its unsubscribe', () => {
const socket = directWire()
const streams = registry(socket)
const dispose = streams.subscribe('terminal.subscribe', { terminal: 't1' }, () => undefined, {
envelope: { ...ENVELOPE, id: 'forged' }
})
streams.markForReplay()
streams.replayAfterAuthentication()
dispose()
expect(socket.frames().map((frame) => [frame.method, frame.executionHost, frame.id])).toEqual([
['terminal.subscribe', 'runtime:a', 'rpc-1'],
['terminal.subscribe', 'runtime:a', 'rpc-1'],
['terminal.unsubscribe', 'runtime:a', 'rpc-2']
])
})
it('keeps it on the unsubscribe a ready id sends after the stream was disposed', () => {
const socket = directWire()
const streams = registry(socket)
const dispose = streams.subscribe('runtime.clientEvents.subscribe', {}, () => undefined, {
envelope: ENVELOPE
})
dispose()
streams.handleResponse(streaming('rpc-1', { type: 'ready', subscriptionId: 'server-1' }))
expect(socket.frames().at(-1)).toMatchObject({
method: 'runtime.clientEvents.unsubscribe',
...ENVELOPE
})
})
it('keeps it on the unsubscribe for a browser ready that no longer owns the screencast', () => {
const socket = directWire()
const streams = registry(socket)
streams.subscribe('browser.screencast', { page: 'p' }, () => undefined, { envelope: ENVELOPE })
streams.markForReplay()
streams.handleResponse(streaming('rpc-1', { type: 'ready', subscriptionId: 'stale' }))
expect(socket.frames().at(-1)).toMatchObject({
method: 'browser.screencast.unsubscribe',
params: { subscriptionId: 'stale' },
...ENVELOPE
})
})
})
describe('relay transport envelope', () => {
it('keeps it on a session tabs unsubscribe deferred until the snapshot', async () => {
const { streams, frames } = relayStreams()
const dispose = streams.subscribe('session.tabs.subscribe', { worktree: 'w' }, () => {}, {
envelope: ENVELOPE
})
await Promise.resolve()
dispose()
expect(frames).toHaveLength(1)
streams.handleResponse(streaming(frames[0]!.id, { type: 'snapshot', tabs: [] }))
expect(frames.at(-1)).toMatchObject({
method: 'session.tabs.unsubscribe',
envelope: ENVELOPE
})
})
it('keeps it on a ready-id unsubscribe deferred until the ready', async () => {
const { streams, frames } = relayStreams()
const dispose = streams.subscribe('runtime.clientEvents.subscribe', {}, () => {}, {
envelope: ENVELOPE
})
await Promise.resolve()
dispose()
expect(frames).toHaveLength(1)
streams.handleResponse(streaming(frames[0]!.id, { type: 'ready', subscriptionId: 'server-1' }))
expect(frames.at(-1)).toMatchObject({
method: 'runtime.clientEvents.unsubscribe',
envelope: ENVELOPE
})
})
})
@@ -0,0 +1,53 @@
import { z } from 'zod'
import type { RpcRequest } from './types'
/**
* Extra top-level fields for one outgoing RPC request, beside `params`.
*
* What a request needs at the top level changes with the host, so this is a plain JSON object the
* physical sender copies verbatim rather than a list of known fields: a new one reaches the wire
* without a new app binary. The page's bridge carries it under the same name and this schema.
*/
export const RpcEnvelopeFieldsSchema = z.record(z.string(), z.json())
export type RpcEnvelopeFields = z.infer<typeof RpcEnvelopeFieldsSchema>
// Key-exhaustive over the wire request, so a field the transport starts sending is protected too.
const TRANSPORT_FIELDS: Record<keyof RpcRequest, true> = {
id: true,
deviceToken: true,
method: true,
params: true
}
/** Every field the transport owns; none can be set through `RpcEnvelopeFields`. */
export const RPC_TRANSPORT_FIELDS: readonly string[] = Object.keys(TRANSPORT_FIELDS)
/** What a caller-facing transport hands its physical sender, which adds the credential. */
export type RpcRequestFrame = Omit<RpcRequest, 'deviceToken'> & { envelope?: RpcEnvelopeFields }
/**
* The wire request with `fields` merged under it. Transport fields are dropped from `fields`
* whether or not this request carries them, so an absent `params` cannot be filled in either.
* Absent or empty fields hand back the request itself, which keeps today's bytes identical.
*/
export function withRpcEnvelopeFields(
request: RpcRequest,
fields: RpcEnvelopeFields | undefined
): RpcRequest {
if (fields === undefined || Object.keys(fields).length === 0) {
return request
}
const extra = Object.fromEntries(
Object.entries(fields).filter(([key]) => !Object.hasOwn(TRANSPORT_FIELDS, key))
)
return { ...extra, ...request }
}
/** The direct transport's wire request, in the key order its sockets and goldens have always had. */
export function buildDirectRpcWireRequest(
{ id, method, params, envelope }: RpcRequestFrame,
deviceToken: string
): RpcRequest {
return withRpcEnvelopeFields({ id, deviceToken, method, params }, envelope)
}
@@ -1,3 +1,4 @@
import type { RpcEnvelopeFields } from './rpc-envelope-fields'
import type { RpcResponse } from './types'
// The raw request port, kept in its own module so that reaching it is a visible act.
@@ -19,6 +20,8 @@ export type SendRequestOptions = {
budgetSpansConnect?: boolean
/** Reject instead of replaying the request after reconnect. */
failWhenDisconnected?: boolean
/** Extra top-level wire fields; the transport's own always win. */
envelope?: RpcEnvelopeFields
}
/** Unvalidated: an arbitrary method name in, an unread envelope out. */