feat(mobile): relay a paired phone's server-workspace requests through the desktop

A phone mirrors its desktop, including workspaces on the desktop's configured servers. The desktop
now forwards a request that names `executionHost: runtime:<env>` to that server over one socket per
(phone, server), signed in as the phone's delegated device; replies pass through untouched except
stream ids. Hidden until the phone client ships: the relay capability is not advertised yet.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb
This commit is contained in:
Jinwoo-H
2026-10-08 00:14:50 -04:00
parent d0e729ca01
commit 07272c0259
24 changed files with 2024 additions and 0 deletions
@@ -0,0 +1,40 @@
import { app } from 'electron'
import { listEnvironments } from '../../shared/runtime-environment-store'
import { getPreferredPairingOffer } from '../../shared/runtime-environments'
import type { MobileDesktopRelayHosts } from '../runtime/mobile-desktop-relay/mobile-desktop-relay-hosts'
import { resolveManagedRuntimeEnvironment } from './runtime-environment-managed-tunnel'
import { callRuntimeEnvironment } from './runtime-environment-transport-routing'
const retirementListeners = new Set<(environmentId: string) => void>()
/** Called wherever the desktop's own transport to a server is invalidated. */
export function retireMobileDesktopRelayEnvironment(environmentId: string): void {
for (const listener of retirementListeners) {
listener(environmentId)
}
}
/** The desktop's configured servers, addressed only by exact environment id. */
export function createRuntimeEnvironmentMobileRelayHosts(): MobileDesktopRelayHosts {
// Why userData: configured servers live where the runtime-environment handlers keep them.
const userDataPath = app.getPath('userData')
return {
resolve: async (environmentId) => {
if (!listEnvironments(userDataPath).some((entry) => entry.id === environmentId)) {
return null
}
const environment = await resolveManagedRuntimeEnvironment(userDataPath, environmentId)
return {
environmentId,
fence: `${environment.pairingRevision ?? environment.createdAt}\0${environment.runtimeId}`,
pairing: getPreferredPairingOffer(environment)
}
},
call: (host, method, params) =>
callRuntimeEnvironment(userDataPath, host.environmentId, method, params),
onEnvironmentRetired: (listener) => {
retirementListeners.add(listener)
return () => retirementListeners.delete(listener)
}
}
}
+2
View File
@@ -29,6 +29,7 @@ import { registerRuntimeSshAccessHandlers } from './runtime-ssh-access-handlers'
import { retirePairedRuntimeBrowserClientHostEnvironment } from '../browser/paired-runtime-browser-client-host-runtime'
import { registerRuntimeEnvironmentBrowserClientHostHandler } from './runtime-environment-browser-client-host-handler'
import { advanceRuntimeEnvironmentCapabilityIncarnation } from './runtime-environment-capability-evidence'
import { retireMobileDesktopRelayEnvironment } from './runtime-environment-mobile-relay-hosts'
import { watchRuntimeEnvironmentPreference } from './runtime-environment-preference'
const remoteRuntimeSubscriptions = new Map<string, RetainedRemoteRuntimeSubscription>()
@@ -71,6 +72,7 @@ export function invalidateRuntimeEnvironmentTransport(environmentId: string): Pr
advanceRuntimeEnvironmentTransportGeneration(environmentId)
closeRemoteRuntimeRequestConnection(environmentId)
closeSubscriptionsForEnvironment(environmentId)
retireMobileDesktopRelayEnvironment(environmentId)
return retirePairedRuntimeBrowserClientHostEnvironment(
environmentId,
new Error('Runtime environment transport was invalidated')
@@ -0,0 +1,66 @@
import { z } from 'zod'
import {
DELEGATED_MOBILE_DEVICE_SYNC_METHOD,
DELEGATED_MOBILE_DEVICE_SYNC_MAX_PHONES,
DELEGATED_MOBILE_DEVICE_SYNC_RUNTIME_CAPABILITY,
type DelegatedMobileDeviceSyncParams
} from '../../../shared/delegated-mobile-device-contract'
import { RemoteRuntimeClientError } from '../../../shared/remote-runtime-client-error'
import type { MobileDesktopRelayHost, MobileDesktopRelayHosts } from './mobile-desktop-relay-hosts'
const SyncResultSchema = z.object({
devices: z.array(z.object({ phoneKey: z.string(), deviceId: z.string(), token: z.string() }))
})
const StatusCapabilitiesSchema = z.object({ capabilities: z.array(z.string()).optional() })
export type DelegatedPhoneGrant = { deviceId: string; token: string }
/** One sync's outcome for a host; held in memory only, so the next sync is the only recovery. */
export type DelegatedPhoneGrants =
| { kind: 'update-needed'; fence: string }
| {
kind: 'ready'
fence: string
grants: ReadonlyMap<string, DelegatedPhoneGrant>
// Phones whose host token was refused (revoked on the host); not retried until the next sync.
refused: Set<string>
}
export async function syncDelegatedPhoneGrants(
hosts: MobileDesktopRelayHosts,
host: MobileDesktopRelayHost,
phones: DelegatedMobileDeviceSyncParams['phones']
): Promise<DelegatedPhoneGrants> {
const status = await hosts.call(host, 'status.get', undefined)
if (!status.ok) {
throw new RemoteRuntimeClientError('remote_runtime_unavailable', status.error.message)
}
const capabilities = StatusCapabilitiesSchema.safeParse(status.result).data?.capabilities ?? []
if (!capabilities.includes(DELEGATED_MOBILE_DEVICE_SYNC_RUNTIME_CAPABILITY)) {
return { kind: 'update-needed', fence: host.fence }
}
const response = await hosts.call(host, DELEGATED_MOBILE_DEVICE_SYNC_METHOD, {
phones: phones.slice(0, DELEGATED_MOBILE_DEVICE_SYNC_MAX_PHONES)
})
if (!response.ok) {
throw new RemoteRuntimeClientError('remote_runtime_unavailable', response.error.message)
}
const result = SyncResultSchema.safeParse(response.result)
if (!result.success) {
throw new RemoteRuntimeClientError(
'invalid_runtime_response',
'Remote Orca runtime returned an invalid delegated device sync.'
)
}
return {
kind: 'ready',
fence: host.fence,
grants: new Map(
result.data.devices.map((device) => [
device.phoneKey,
{ deviceId: device.deviceId, token: device.token }
])
),
refused: new Set()
}
}
@@ -0,0 +1,25 @@
import type { PairingOffer } from '../../../shared/pairing'
import type { RuntimeRpcResponse } from '../../../shared/runtime-rpc-envelope'
/** A configured server the desktop can relay to. */
export type MobileDesktopRelayHost = {
environmentId: string
// Pairing revision and runtime identity; delegated grants from another fence are never reused.
fence: string
// The desktop's own runtime-scope pairing; the relay swaps in a phone's delegated token.
pairing: PairingOffer
}
/** The desktop's configured servers, as the relay sees them. */
export type MobileDesktopRelayHosts = {
/** Null when the id names no configured server; a phone-supplied endpoint is never used. */
resolve: (environmentId: string) => Promise<MobileDesktopRelayHost | null>
/** A call as the desktop itself, over its own connection to the server. */
call: (
host: MobileDesktopRelayHost,
method: string,
params: unknown
) => Promise<RuntimeRpcResponse<unknown>>
/** Fires when a server is removed, re-paired or disconnected; returns an unsubscribe. */
onEnvironmentRetired: (listener: (environmentId: string) => void) => () => void
}
@@ -0,0 +1,445 @@
import { mkdtempSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { afterEach, describe, expect, it, vi } from 'vitest'
import WebSocket from 'ws'
import { z } from 'zod'
import {
decrypt,
decryptBytes,
deriveSharedKey,
encrypt,
generateKeyPair,
publicKeyFromBase64,
publicKeyToBase64
} from '../../../shared/e2ee-crypto'
import { parsePairingCode, type PairingOffer } from '../../../shared/pairing'
import {
DELEGATED_MOBILE_DEVICE_SYNC_RUNTIME_CAPABILITY,
DelegatedMobileDeviceSyncParamsSchema
} from '../../../shared/delegated-mobile-device-contract'
import { decodeTerminalStreamFrame } from '../../../shared/terminal-stream-protocol'
import type { DeviceEntry } from '../device-registry'
import { OrcaRuntimeService } from '../orca-runtime'
import { readRuntimeMetadata } from '../runtime-metadata'
import { OrcaRuntimeRpcServer } from '../runtime-rpc'
import { sendRequest } from '../runtime-rpc-test-harness'
import { makeStore } from '../runtime-rpc-worktree-store-fixtures'
import type { MobileDesktopRelayHosts } from './mobile-desktop-relay-hosts'
import type * as PassthroughSocketModule from '../../../shared/remote-runtime-passthrough-socket'
const passthroughOpens = vi.hoisted(() => {
const hostReplies: string[] = []
return { count: 0, hostReplies }
})
vi.mock('../../../shared/remote-runtime-passthrough-socket', async (importOriginal) => {
const original = await importOriginal<typeof PassthroughSocketModule>()
return {
...original,
openRemoteRuntimePassthroughSocket: (
...[pairing, capabilities, callbacks, options]: Parameters<
typeof original.openRemoteRuntimePassthroughSocket
>
) => {
passthroughOpens.count += 1
return original.openRemoteRuntimePassthroughSocket(
pairing,
capabilities,
{
...callbacks,
onText: (plaintext) => {
passthroughOpens.hostReplies.push(plaintext)
callbacks.onText(plaintext)
}
},
options
)
}
}
})
vi.mock('../../git/worktree', () => {
const worktrees = [
{
path: '/tmp/worktree-a',
head: 'abc',
branch: 'feature/foo',
isBare: false,
isMainWorktree: false
}
]
return {
listWorktrees: vi.fn().mockResolvedValue(worktrees),
listWorktreesStrict: vi.fn().mockResolvedValue(worktrees)
}
})
/** Exposes the per-terminal phone seats the runtime keeps behind a protected map. */
class SeatObservableRuntime extends OrcaRuntimeService {
seatClientIds(ptyId: string): string[] {
return [...(this.mobileSubscribers.get(ptyId)?.keys() ?? [])]
}
}
const WORKTREE = 'id:repo-1::/tmp/worktree-a'
const TARGET = 'runtime:env-1'
type Frame = Record<string, unknown>
describe('mobile desktop relay: phone -> desktop -> server', () => {
const cleanups: (() => Promise<void> | void)[] = []
afterEach(async () => {
for (const cleanup of cleanups.splice(0).toReversed()) {
await cleanup()
}
})
async function startRuntime(spawnedOnThisHost: string[] = []) {
const writes: string[] = []
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the fixture store carries the surface these RPCs read, as in the sibling streaming tests.
const runtime = new SeatObservableRuntime(makeStore() as never)
runtime.setPtyController({
spawn: vi.fn().mockImplementation(async () => {
spawnedOnThisHost.push('pty-shared')
return { id: 'pty-shared' }
}),
write: (_ptyId, data) => {
writes.push(data)
return true
},
kill: () => true,
getForegroundProcess: async () => null,
resize: () => true,
getSize: () => ({ cols: 80, rows: 24 })
})
const userDataPath = mkdtempSync(join(tmpdir(), 'orca-relay-'))
const server = new OrcaRuntimeRpcServer({
runtime,
userDataPath,
enableWebSocket: true,
wsPort: 0
})
await server.start()
cleanups.push(() => server.stop())
return { server, runtime, writes, userDataPath }
}
function pair(
server: OrcaRuntimeRpcServer,
scope: 'mobile' | 'runtime',
name: string
): PairingOffer {
const offer = server.createPairingOffer({ address: '127.0.0.1', name, scope })
if (!offer.available) {
throw new Error('pairing unavailable')
}
return parsePairingCode(offer.pairingUrl)!
}
/** A server, and a desktop whose only configured server is it, as `env-1`. */
async function startTopology() {
const host = await startRuntime()
const desktopSpawns: string[] = []
const desktop = await startRuntime(desktopSpawns)
const desktopOnHost = pair(host.server, 'runtime', 'MacBook')
const registry = host.server.getDeviceRegistry()!
const children = new Map<string, DeviceEntry>()
const syncedNames: string[][] = []
const state = { capable: true, retire: (_id: string) => {} }
const ok = (result: unknown) => ({
id: 'x',
ok: true as const,
result,
_meta: { runtimeId: 'h' }
})
const hosts: MobileDesktopRelayHosts = {
resolve: async (environmentId) =>
environmentId === 'env-1'
? { environmentId, fence: 'pairing-1', pairing: desktopOnHost }
: null,
call: async (_host, method, params) => {
if (method === 'status.get') {
return ok({
capabilities: state.capable ? [DELEGATED_MOBILE_DEVICE_SYNC_RUNTIME_CAPABILITY] : []
})
}
// Stand-in for the server's sync (lane S1a): mint or return ordinary phone devices on the host.
const { phones } = DelegatedMobileDeviceSyncParamsSchema.parse(params)
syncedNames.push(phones.map((phone) => phone.name))
for (const phone of phones) {
const existing = children.get(phone.phoneKey)
if (!existing || !registry.getDevice(existing.deviceId)) {
const device = registry.addDevice(phone.name, 'mobile')
registry.updateLastSeen(device.deviceId)
children.set(phone.phoneKey, device)
}
}
return ok({
devices: phones.map((phone) => {
const device = children.get(phone.phoneKey)!
return { phoneKey: phone.phoneKey, deviceId: device.deviceId, token: device.token }
})
})
},
onEnvironmentRetired: (listener) => {
state.retire = listener
return () => {}
}
}
desktop.server.setMobileDesktopRelayHosts(hosts)
const metadata = readRuntimeMetadata(host.userDataPath)!
const created = await sendRequest(metadata.transports[0]!.endpoint, {
id: 'create',
authToken: metadata.authToken,
method: 'terminal.create',
params: {
worktree: WORKTREE,
command: 'bash',
tabId: 'tab-1',
leafId: '11111111-1111-4111-8111-111111111111',
activate: true
}
})
const handle = z.object({ terminal: z.object({ handle: z.string() }) }).parse(created.result)
.terminal.handle
return { host, desktop, desktopSpawns, children, syncedNames, state, handle }
}
async function connectPhone(desktop: OrcaRuntimeRpcServer, name: string) {
const pairing = pair(desktop, 'mobile', name)
const keys = generateKeyPair()
const sharedKey = deriveSharedKey(keys.secretKey, publicKeyFromBase64(pairing.publicKeyB64))
const ws = new WebSocket(pairing.endpoint)
const frames: Frame[] = []
const binaries: Uint8Array[] = []
let authenticated = false
const ready = new Promise<void>((resolve) => {
ws.on('message', (data: Buffer, isBinary) => {
if (isBinary) {
binaries.push(decryptBytes(new Uint8Array(data), sharedKey)!)
return
}
const text = data.toString()
if (text.includes('e2ee_ready')) {
ws.send(
encrypt(
JSON.stringify({ type: 'e2ee_auth', deviceToken: pairing.deviceToken }),
sharedKey
)
)
return
}
const frame = z.record(z.string(), z.unknown()).parse(JSON.parse(decrypt(text, sharedKey)!))
if (!authenticated) {
authenticated = frame.type === 'e2ee_authenticated'
resolve()
return
}
frames.push(frame)
})
})
ws.once('open', () =>
ws.send(
JSON.stringify({ type: 'e2ee_hello', publicKeyB64: publicKeyToBase64(keys.publicKey) })
)
)
await ready
cleanups.push(() => ws.close())
const token = pairing.deviceToken
return {
token,
frames,
binaries,
close: () => ws.close(),
send(id: string, method: string, params: unknown, executionHost: string | null = TARGET) {
ws.send(
encrypt(
JSON.stringify({
id,
deviceToken: token,
method,
params,
executionHost: executionHost ?? undefined
}),
sharedKey
)
)
},
next: (id: string, predicate: (frame: Frame) => boolean = () => true) =>
vi.waitFor(() => {
const frame = frames.find((candidate) => candidate.id === id && predicate(candidate))
expect(frame).toBeDefined()
return frame!
})
}
}
type Phone = Awaited<ReturnType<typeof connectPhone>>
async function subscribe(phone: Phone, handle: string, id = 'sub') {
phone.send(id, 'terminal.subscribe', {
terminal: handle,
client: { id: phone.token, type: 'mobile' },
viewport: { cols: 40, rows: 20 },
capabilities: { terminalBinaryStream: 1 }
})
const subscribed = await phone.next(
id,
(frame) =>
z.object({ result: z.object({ type: z.literal('subscribed') }) }).safeParse(frame).success
)
return z.object({ result: z.object({ streamId: z.number() }) }).parse(subscribed).result
.streamId
}
const isOk = (frame: Frame) => frame.ok === true
const errorCode = (frame: Frame) =>
z.object({ error: z.object({ code: z.string() }) }).parse(frame).error.code
const sendAccepted = (frame: Frame) =>
z.object({ result: z.object({ send: z.object({ accepted: z.boolean() }) }) }).parse(frame)
.result.send.accepted
it('relays a terminal subscribe, its binary stream, input and a query reply as the phone delegated device', async () => {
const { host, desktop, children, syncedNames, handle } = await startTopology()
const phone = await connectPhone(desktop.server, 'iPhone')
const streamId = await subscribe(phone, handle)
// The snapshot arrives as binary frames under the id the phone was given.
await vi.waitFor(() =>
expect(
phone.binaries.some((bytes) => decodeTerminalStreamFrame(bytes)?.streamId === streamId)
).toBe(true)
)
expect(syncedNames).toEqual([[`iPhone via ${desktop.runtime.readMachineName()}`]])
const delegated = [...children.values()][0]!
// The host seats the phone's delegated device, never the desktop's token for it.
expect(host.runtime.seatClientIds('pty-shared')).toEqual([delegated.token])
phone.send('in', 'terminal.send', {
terminal: handle,
text: 'ls',
client: { id: phone.token, type: 'mobile' }
})
expect(isOk(await phone.next('in'))).toBe(true)
expect(host.writes).toContain('ls')
// terminal.send cross-checks client.id against the socket's device: only a swapped token passes.
phone.send('query', 'terminal.send', {
terminal: handle,
text: '\x1b[0n',
enter: false,
inputKind: 'query-reply',
client: { id: phone.token, type: 'mobile' }
})
expect(sendAccepted(await phone.next('query'))).toBe(true)
expect(host.writes).toContain('\x1b[0n')
})
it('gives two phones on one server terminal their own seats, and cleans up the one that disconnects', async () => {
const { host, desktop, children, handle } = await startTopology()
const phoneA = await connectPhone(desktop.server, 'Phone A')
const phoneB = await connectPhone(desktop.server, 'Phone B')
const streamA = await subscribe(phoneA, handle)
const streamB = await subscribe(phoneB, handle)
expect(streamA).not.toBe(streamB)
const tokens = [...children.values()].map((device) => device.token)
expect(new Set(tokens).size).toBe(2)
expect(host.runtime.seatClientIds('pty-shared').sort()).toEqual([...tokens].sort())
phoneA.close()
await vi.waitFor(() => expect(host.runtime.seatClientIds('pty-shared')).toHaveLength(1))
expect(host.runtime.seatClientIds('pty-shared')).not.toContain(tokens[0])
})
it('applies a phone capability update on the desktop and on its open server socket, answering once', async () => {
const { desktop, handle } = await startTopology()
const phone = await connectPhone(desktop.server, 'iPhone')
await subscribe(phone, handle)
phone.send(
'caps',
'runtime.clientCapabilities.update',
{ clientCapabilities: ['relay.test.v1'] },
null
)
expect(await phone.next('caps')).toMatchObject({
ok: true,
result: { clientCapabilities: ['relay.test.v1'] }
})
// The server applied it to the relayed phone's socket; its answer stays with the relay.
await vi.waitFor(() =>
expect(
passthroughOpens.hostReplies.some((frame) =>
frame.includes('"clientCapabilities":["relay.test.v1"]')
)
).toBe(true)
)
await new Promise((resolve) => setTimeout(resolve, 50))
expect(
phone.frames.filter((frame) => frame.id === 'caps' || String(frame.id).startsWith('relay-'))
).toHaveLength(1)
})
it('stops relaying a phone the server revoked until the next sync', async () => {
const { host, desktop, children, state, handle } = await startTopology()
const phone = await connectPhone(desktop.server, 'iPhone')
await subscribe(phone, handle)
const revoked = [...children.values()][0]!
await host.server.revokeMobileDevice(revoked.deviceId)
// The open stream ends as unavailable, never as an unpaired phone.
expect(errorCode(await phone.next('sub', (frame) => frame.ok === false))).toBe(
'remote_runtime_unavailable'
)
const opensBefore = passthroughOpens.count
phone.send('after-1', 'terminal.list', {})
expect(errorCode(await phone.next('after-1'))).toBe('remote_runtime_unavailable')
phone.send('after-2', 'terminal.list', {})
expect(errorCode(await phone.next('after-2'))).toBe('remote_runtime_unavailable')
// One refused attempt, then no retries of the dead token.
expect(passthroughOpens.count - opensBefore).toBe(1)
state.retire('env-1')
phone.send('after-sync', 'terminal.list', {})
expect(isOk(await phone.next('after-sync'))).toBe(true)
})
it('refuses desktop-owned methods and unknown servers without running anything on the desktop', async () => {
const { desktop, desktopSpawns } = await startTopology()
const phone = await connectPhone(desktop.server, 'iPhone')
phone.send('settings', 'settings.get', {})
expect(errorCode(await phone.next('settings'))).toBe('forbidden')
phone.send('multiplex', 'terminal.multiplex', {})
expect(errorCode(await phone.next('multiplex'))).toBe('forbidden')
const create = {
worktree: WORKTREE,
command: 'bash',
tabId: 'tab-2',
leafId: '22222222-2222-4222-8222-222222222222'
}
phone.send('unknown', 'terminal.create', create, 'runtime:env-unknown')
expect(errorCode(await phone.next('unknown'))).toBe('remote_runtime_unavailable')
phone.send('garbled', 'terminal.create', create, 'runtime:')
expect(errorCode(await phone.next('garbled'))).toBe('invalid_argument')
expect(desktopSpawns).toEqual([])
// No target is today's local call.
phone.send('local', 'terminal.list', {}, null)
expect(isOk(await phone.next('local'))).toBe(true)
})
it('reports a server without delegated devices as update-needed and relays nothing to it', async () => {
const { desktop, state } = await startTopology()
state.capable = false
const phone = await connectPhone(desktop.server, 'iPhone')
const opensBefore = passthroughOpens.count
phone.send('list', 'terminal.list', {})
expect(errorCode(await phone.next('list'))).toBe('remote_runtime_unavailable')
await expect(desktop.server.getMobileDesktopRelayHostState('env-1')).resolves.toBe(
'update-needed'
)
await expect(desktop.server.getMobileDesktopRelayHostState('env-unknown')).resolves.toBe(
'unavailable'
)
expect(passthroughOpens.count).toBe(opensBefore)
state.capable = true
state.retire('env-1')
await expect(desktop.server.getMobileDesktopRelayHostState('env-1')).resolves.toBe('ready')
})
})
@@ -0,0 +1,140 @@
import { describe, expect, it, vi } from 'vitest'
import { DELEGATED_MOBILE_DEVICE_SYNC_RUNTIME_CAPABILITY } from '../../../shared/delegated-mobile-device-contract'
import { RemoteRuntimeClientError } from '../../../shared/remote-runtime-client-error'
import type { RemoteRuntimePassthroughCallbacks } from '../../../shared/remote-runtime-passthrough-socket'
import { MobileDesktopRelay, type RelayedPhone } from './mobile-desktop-relay'
import type { MobileDesktopRelayHosts } from './mobile-desktop-relay-hosts'
type FakeSocket = {
sent: string[]
capabilities: readonly string[]
callbacks: RemoteRuntimePassthroughCallbacks
closed: boolean
}
const sockets = vi.hoisted(() => {
const opened: FakeSocket[] = []
return opened
})
vi.mock('../../../shared/remote-runtime-passthrough-socket', () => ({
openRemoteRuntimePassthroughSocket: async (
_pairing: unknown,
capabilities: readonly string[],
callbacks: RemoteRuntimePassthroughCallbacks
) => {
const socket: FakeSocket = { sent: [], capabilities, callbacks, closed: false }
sockets.push(socket)
return {
send: (frame: string) => {
socket.sent.push(frame)
return true
},
close: () => {
socket.closed = true
}
}
}
}))
const ok = (result: unknown) => ({ id: 'x', ok: true as const, result, _meta: { runtimeId: 'h' } })
function relayWithPhone() {
const hosts: MobileDesktopRelayHosts = {
resolve: async (environmentId) => ({
environmentId,
fence: 'f',
pairing: { v: 2, endpoint: 'ws://127.0.0.1:1', deviceToken: 'desktop', publicKeyB64: 'k' }
}),
call: async (_host, method) =>
method === 'status.get'
? ok({ capabilities: [DELEGATED_MOBILE_DEVICE_SYNC_RUNTIME_CAPABILITY] })
: ok({ devices: [{ phoneKey: 'phone-1', deviceId: 'child', token: 'host-token' }] }),
onEnvironmentRetired: () => () => {}
}
const relay = new MobileDesktopRelay({
hosts,
listPhones: () => [{ phoneKey: 'phone-1', name: 'iPhone via Mac' }],
allocateStreamId: () => 1
})
const replies: string[] = []
let capabilities: readonly string[] = ['cap.a']
const phone: RelayedPhone = {
connectionId: 'conn-1',
deviceId: 'phone-1',
deviceToken: 'desktop-token',
clientCapabilities: () => capabilities,
reply: (frame) => replies.push(frame),
sendBinary: () => true
}
return { relay, phone, replies, setCapabilities: (next: string[]) => (capabilities = next) }
}
describe('MobileDesktopRelay', () => {
it('signs in with exactly the phone capabilities and mirrors its updates without a second answer', async () => {
sockets.length = 0
const { relay, phone, replies, setCapabilities } = relayWithPhone()
relay.forward(
phone,
'env-1',
{ id: 'r1', method: 'terminal.list' },
'{"id":"r1","method":"terminal.list"}'
)
relay.forward(
phone,
'env-2',
{ id: 'r2', method: 'terminal.list' },
'{"id":"r2","method":"terminal.list"}'
)
await vi.waitFor(() => expect(sockets.map((socket) => socket.sent.length)).toEqual([1, 1]))
expect(sockets[0]!.capabilities).toEqual(['cap.a'])
setCapabilities(['cap.b'])
const update = JSON.stringify({
id: 'caps',
deviceToken: 'desktop-token',
method: 'runtime.clientCapabilities.update',
params: { clientCapabilities: ['cap.b'] }
})
relay.forwardClientCapabilities('conn-1', update)
relay.forwardClientCapabilities('other-connection', update)
await vi.waitFor(() => expect(sockets.map((socket) => socket.sent.length)).toEqual([2, 2]))
for (const socket of sockets) {
const mirrored = JSON.parse(socket.sent[1]!)
expect(mirrored).toMatchObject({
method: 'runtime.clientCapabilities.update',
params: { clientCapabilities: ['cap.b'] }
})
expect(mirrored.id).not.toBe('caps')
expect(mirrored.deviceToken).toBeUndefined()
// The host's answer to the mirror is the relay's, not the phone's.
socket.callbacks.onText(JSON.stringify({ id: mirrored.id, ok: true, result: {}, _meta: {} }))
}
expect(replies).toEqual([])
})
it('answers every request still open when the server socket drops, then closes on phone disconnect', async () => {
sockets.length = 0
const { relay, phone, replies } = relayWithPhone()
relay.forward(phone, 'env-1', { id: 'a', method: 'terminal.list' }, '{"id":"a"}')
relay.forward(phone, 'env-1', { id: 'b', method: 'terminal.list' }, '{"id":"b"}')
await vi.waitFor(() => expect(sockets[0]?.sent).toHaveLength(2))
// One socket per (phone connection, server) carries both requests.
expect(sockets).toHaveLength(1)
sockets[0]!.callbacks.onText('{"id":"a","ok":true,"result":{},"_meta":{"runtimeId":"h"}}')
sockets[0]!.callbacks.onClose(
new RemoteRuntimeClientError('remote_runtime_unavailable', 'gone')
)
expect(replies.map((frame) => JSON.parse(frame))).toEqual([
{ id: 'a', ok: true, result: {}, _meta: { runtimeId: 'h' } },
expect.objectContaining({
id: 'b',
ok: false,
error: expect.objectContaining({ code: 'remote_runtime_unavailable' })
})
])
relay.forward(phone, 'env-1', { id: 'c', method: 'terminal.list' }, '{"id":"c"}')
await vi.waitFor(() => expect(sockets).toHaveLength(2))
relay.closePhoneConnection('conn-1')
expect(sockets[1]!.closed).toBe(true)
})
})
@@ -0,0 +1,287 @@
import { randomUUID } from 'node:crypto'
import type { MobileDesktopRelayHostState } from '../../../shared/mobile-desktop-relay-contract'
import type { DelegatedMobileDeviceSyncParams } from '../../../shared/delegated-mobile-device-contract'
import type { RuntimeCapability } from '../../../shared/protocol-version'
import { RemoteRuntimeClientError } from '../../../shared/remote-runtime-client-error'
import {
openRemoteRuntimePassthroughSocket,
type RemoteRuntimePassthroughSocket
} from '../../../shared/remote-runtime-passthrough-socket'
import { MOBILE_RPC_METHOD_ROUTES } from '../runtime-rpc/runtime-rpc-mobile-method-routing'
import { syncDelegatedPhoneGrants, type DelegatedPhoneGrants } from './delegated-phone-grants'
import type { MobileDesktopRelayHost, MobileDesktopRelayHosts } from './mobile-desktop-relay-hosts'
import { RELAYED_STREAM_CARRIERS } from './relayed-stream-carriers'
import { RelayedPhoneReplies, rewriteRelayedPhoneRequest } from './relayed-phone-frames'
const RELAY_CONNECT_TIMEOUT_MS = 15_000
/** The phone connection a relayed request arrived on. */
export type RelayedPhone = {
connectionId: string
deviceId: string
deviceToken: string
clientCapabilities: () => readonly RuntimeCapability[]
reply: (frame: string) => void
sendBinary: (bytes: Uint8Array<ArrayBufferLike>) => boolean | void
}
type OpenedRelayLink = { socket: RemoteRuntimePassthroughSocket; hostToken: string }
class RelayLink {
socket?: RemoteRuntimePassthroughSocket
readonly replies: RelayedPhoneReplies
readonly opened: Promise<OpenedRelayLink>
constructor(
readonly environmentId: string,
readonly phone: RelayedPhone,
allocateStreamId: () => number,
open: (link: RelayLink) => Promise<OpenedRelayLink>
) {
this.replies = new RelayedPhoneReplies(allocateStreamId)
this.opened = open(this)
}
}
/**
* Relays a paired phone's requests for a server's workspace to that server, over one socket per
* (phone connection, server) signed in as the phone's delegated device. It never reads a reply
* beyond its id and stream ids; see docs/reference/mobile-desktop-relay.md.
*/
export class MobileDesktopRelay {
private readonly links = new Map<string, RelayLink>()
private readonly grants = new Map<string, Promise<DelegatedPhoneGrants>>()
private readonly stopRetirementWatch: () => void
constructor(
private readonly options: {
hosts: MobileDesktopRelayHosts
listPhones: () => DelegatedMobileDeviceSyncParams['phones']
allocateStreamId: () => number
}
) {
this.stopRetirementWatch = options.hosts.onEnvironmentRetired((environmentId) =>
this.retireEnvironment(environmentId)
)
}
dispose(): void {
this.stopRetirementWatch()
for (const link of this.links.values()) {
this.endLink(link, unavailable('The paired desktop stopped relaying.'))
}
this.grants.clear()
}
forward(
phone: RelayedPhone,
environmentId: string,
request: { id: string; method: string },
frame: string
): void {
if (
MOBILE_RPC_METHOD_ROUTES.get(request.method) !== 'execution-host' ||
RELAYED_STREAM_CARRIERS.get(request.method) === 'phone-binary-frames'
) {
phone.reply(
failure(request.id, 'forbidden', `Method '${request.method}' runs on the paired desktop`)
)
return
}
const link = this.linkFor(phone, environmentId)
link.replies.noteForwarded(request.id)
link.opened.then(
({ socket, hostToken }) => {
if (!socket.send(rewriteRelayedPhoneRequest(frame, phone.deviceToken, hostToken))) {
this.endLink(link, unavailable('The server connection could not take the request.'))
}
},
() => {
// The open failure already answered every request queued on this link.
}
)
}
/** Mirrors a phone's capability update to every server socket that phone has open. */
forwardClientCapabilities(connectionId: string, frame: string): void {
for (const link of this.links.values()) {
if (link.phone.connectionId !== connectionId) {
continue
}
// Why a fresh id: the phone already has the desktop's answer to its own id.
const id = `relay-capabilities:${randomUUID()}`
link.replies.noteForwarded(id, { swallowReplies: true })
link.opened.then(
({ socket, hostToken }) =>
socket.send(rewriteRelayedPhoneRequest(frame, link.phone.deviceToken, hostToken, id)),
() => {}
)
}
}
closePhoneConnection(connectionId: string): void {
for (const link of this.links.values()) {
if (link.phone.connectionId === connectionId) {
this.endLink(link, null)
}
}
}
/** The server was removed, re-paired or disconnected: drop its sockets and grants. */
retireEnvironment(environmentId: string): void {
this.grants.delete(environmentId)
for (const link of this.links.values()) {
if (link.environmentId === environmentId) {
this.endLink(link, unavailable('The server connection was reset.'))
}
}
}
/** The desktop's paired phones changed: re-sync every server already holding grants. */
phonesChanged(): void {
for (const environmentId of this.grants.keys()) {
this.grants.delete(environmentId)
void this.hostState(environmentId)
}
}
async hostState(environmentId: string): Promise<MobileDesktopRelayHostState> {
try {
const host = await this.options.hosts.resolve(environmentId)
return host ? (await this.grantsFor(host, null)).kind : 'unavailable'
} catch {
return 'unavailable'
}
}
private linkFor(phone: RelayedPhone, environmentId: string): RelayLink {
const key = linkKey(phone.connectionId, environmentId)
const existing = this.links.get(key)
if (existing) {
return existing
}
const link = new RelayLink(environmentId, phone, this.options.allocateStreamId, (opening) =>
this.open(opening)
)
link.opened.catch((error: unknown) => {
this.endLink(
link,
error instanceof RemoteRuntimeClientError ? error : unavailable(String(error))
)
})
this.links.set(key, link)
return link
}
private async open(link: RelayLink): Promise<OpenedRelayLink> {
const { phone } = link
const host = await this.options.hosts.resolve(link.environmentId)
if (!host) {
throw unavailable('The server is not configured on this computer.')
}
const grants = await this.grantsFor(host, phone.deviceId)
if (grants.kind === 'update-needed') {
throw unavailable('The server needs an Orca update to open its workspaces from a phone.')
}
const grant = grants.grants.get(phone.deviceId)
if (!grant || grants.refused.has(phone.deviceId)) {
throw unavailable('The server does not accept this phone.')
}
try {
const socket = await openRemoteRuntimePassthroughSocket(
{
...host.pairing,
deviceToken: grant.token,
pairedDeviceId: grant.deviceId,
scope: 'mobile'
},
phone.clientCapabilities(),
{
onText: (plaintext) => {
const frame = link.replies.text(plaintext)
if (frame !== null) {
phone.reply(frame)
}
},
onBinary: (bytes) => {
const frame = link.replies.binary(bytes)
if (frame) {
phone.sendBinary(frame)
}
},
onClose: (error) => this.endLink(link, error)
},
{ timeoutMs: RELAY_CONNECT_TIMEOUT_MS }
)
if (this.links.get(linkKey(phone.connectionId, link.environmentId)) !== link) {
socket.close()
throw unavailable('The relayed request was cancelled.')
}
link.socket = socket
return { socket, hostToken: grant.token }
} catch (error) {
if (error instanceof RemoteRuntimeClientError && error.code === 'unauthorized') {
// Why: retrying a revoked token would only raise the host's "unpaired device" prompt.
grants.refused.add(phone.deviceId)
throw unavailable('The server revoked this phone.')
}
throw error
}
}
private grantsFor(
host: MobileDesktopRelayHost,
phoneKey: string | null
): Promise<DelegatedPhoneGrants> {
const cached = this.grants.get(host.environmentId)
const sync = (): Promise<DelegatedPhoneGrants> => {
const pending = syncDelegatedPhoneGrants(this.options.hosts, host, this.options.listPhones())
this.grants.set(host.environmentId, pending)
pending.catch(() => {
// Why: an unreachable server is not an answer; the next request syncs again.
if (this.grants.get(host.environmentId) === pending) {
this.grants.delete(host.environmentId)
}
})
return pending
}
if (!cached) {
return sync()
}
return cached.then((grants) =>
grants.fence !== host.fence ||
(phoneKey !== null && grants.kind === 'ready' && !grants.grants.has(phoneKey))
? this.grants.get(host.environmentId) === cached
? sync()
: this.grantsFor(host, phoneKey)
: grants
)
}
private endLink(link: RelayLink, error: RemoteRuntimeClientError | null): void {
const key = linkKey(link.phone.connectionId, link.environmentId)
if (this.links.get(key) !== link) {
return
}
this.links.delete(key)
link.socket?.close()
const openRequestIds = link.replies.takeOpenRequestIds()
if (error) {
for (const id of openRequestIds) {
link.phone.reply(failure(id, 'remote_runtime_unavailable', error.message))
}
}
}
}
function linkKey(connectionId: string, environmentId: string): string {
return `${connectionId}\0${environmentId}`
}
function unavailable(message: string): RemoteRuntimeClientError {
return new RemoteRuntimeClientError('remote_runtime_unavailable', message)
}
function failure(id: string, code: string, message: string): string {
return JSON.stringify({ id, ok: false, error: { code, message }, _meta: { runtimeId: null } })
}
@@ -0,0 +1,120 @@
import { describe, expect, it } from 'vitest'
import {
encodeTerminalStreamFrame,
decodeTerminalStreamFrame,
TerminalStreamOpcode
} from '../../../shared/terminal-stream-protocol'
import {
BrowserScreencastOpcode,
encodeBrowserScreencastFrame
} from '../../../shared/browser-screencast-protocol'
import { RelayedPhoneReplies, rewriteRelayedPhoneRequest } from './relayed-phone-frames'
const DESKTOP_TOKEN = 'a'.repeat(48)
const HOST_TOKEN = 'b'.repeat(48)
describe('rewriteRelayedPhoneRequest', () => {
it('swaps the desktop token everywhere and strips the envelope routing fields', () => {
const frame = JSON.stringify({
id: 'r1',
deviceToken: DESKTOP_TOKEN,
executionHost: 'runtime:env-1',
method: 'terminal.unsubscribe',
params: {
subscriptionId: `terminal-1:${DESKTOP_TOKEN}`,
client: { id: DESKTOP_TOKEN, type: 'mobile' },
nested: [{ mobileClient: { id: DESKTOP_TOKEN } }]
}
})
const rewritten = rewriteRelayedPhoneRequest(frame, DESKTOP_TOKEN, HOST_TOKEN)
expect(frame).toContain(DESKTOP_TOKEN)
expect(rewritten).not.toContain(DESKTOP_TOKEN)
expect(JSON.parse(rewritten)).toEqual({
id: 'r1',
method: 'terminal.unsubscribe',
params: {
subscriptionId: `terminal-1:${HOST_TOKEN}`,
client: { id: HOST_TOKEN, type: 'mobile' },
nested: [{ mobileClient: { id: HOST_TOKEN } }]
}
})
})
it('can re-id a frame the relay sends on its own behalf', () => {
const frame = JSON.stringify({ id: 'phone-id', method: 'm', params: {} })
expect(
JSON.parse(rewriteRelayedPhoneRequest(frame, DESKTOP_TOKEN, HOST_TOKEN, 'relay-id'))
).toEqual({ id: 'relay-id', method: 'm', params: {} })
})
})
describe('RelayedPhoneReplies', () => {
function replies(firstDesktopId = 100) {
let next = firstDesktopId
return new RelayedPhoneReplies(() => next++)
}
const subscribed = (id: string, streamId: number) =>
JSON.stringify({
id,
ok: true,
streaming: true,
result: { type: 'subscribed', streamId },
_meta: { runtimeId: 'host' }
})
const output = (streamId: number) =>
encodeTerminalStreamFrame({
opcode: TerminalStreamOpcode.Output,
streamId,
seq: 7,
payload: new TextEncoder().encode('hi')
})
it('renumbers host stream ids in results and terminal frame headers into the desktop space', () => {
const relay = replies()
relay.noteForwarded('s1')
relay.noteForwarded('s2')
// Both hosts allocate from 1; the phone must see distinct ids.
expect(JSON.parse(relay.text(subscribed('s1', 1))!).result.streamId).toBe(100)
expect(JSON.parse(relay.text(subscribed('s2', 2))!).result.streamId).toBe(101)
const frame = decodeTerminalStreamFrame(relay.binary(output(2))!)
expect(frame).toMatchObject({ streamId: 101, seq: 7, opcode: TerminalStreamOpcode.Output })
expect(new TextDecoder().decode(frame!.payload)).toBe('hi')
})
it('drops terminal frames for streams the phone was never told of and passes other binary untouched', () => {
const relay = replies()
expect(relay.binary(output(9))).toBeNull()
const screencast = encodeBrowserScreencastFrame({
opcode: BrowserScreencastOpcode.Frame,
seq: 1,
format: 'jpeg',
metadata: { deviceWidth: 1 },
image: new Uint8Array([1, 2, 3])
})
expect(relay.binary(screencast)).toBe(screencast)
})
it('passes non-stream replies through byte for byte and forgets a stream once it ends', () => {
const relay = replies()
relay.noteForwarded('s1')
relay.noteForwarded('q')
const plain = '{"id":"q","ok":true,"result":{"x":1},"_meta":{"runtimeId":"h"}}'
expect(relay.text(plain)).toBe(plain)
expect(relay.text('{"_keepalive":true}')).toBe('{"_keepalive":true}')
relay.text(subscribed('s1', 1))
relay.text(
JSON.stringify({ id: 's1', ok: true, streaming: true, result: { type: 'end' }, _meta: {} })
)
expect(relay.binary(output(1))).toBeNull()
expect(relay.takeOpenRequestIds()).toEqual([])
})
it('reports unfinished requests, and swallows replies to requests the relay sent itself', () => {
const relay = replies()
relay.noteForwarded('open')
relay.noteForwarded('mine', { swallowReplies: true })
expect(relay.text('{"id":"mine","ok":true,"result":{},"_meta":{"runtimeId":"h"}}')).toBeNull()
relay.noteForwarded('mine-2', { swallowReplies: true })
expect(relay.takeOpenRequestIds()).toEqual(['open'])
})
})
@@ -0,0 +1,119 @@
import {
readTerminalStreamFrameId,
writeTerminalStreamFrameId
} from '../../../shared/terminal-stream-protocol'
const STRIPPED_ENVELOPE_FIELDS = new Set(['deviceToken', 'executionHost'])
// Why: the phone stamps its desktop token into params (client.id, mobileClient.id, `${terminal}:${token}`
// subscription ids). A field list would rot silently, so every occurrence in the text is swapped.
export function rewriteRelayedPhoneRequest(
frame: string,
phoneDesktopToken: string,
phoneHostToken: string,
requestId?: string
): string {
const parsed: unknown = JSON.parse(frame.split(phoneDesktopToken).join(phoneHostToken))
if (!isRecord(parsed)) {
throw new Error('relayed_phone_frame_not_an_object')
}
// Why: the host binds identity to the authenticated socket; the routing target is the desktop's only.
const forwarded = Object.fromEntries(
Object.entries(parsed).filter(([field]) => !STRIPPED_ENVELOPE_FIELDS.has(field))
)
return JSON.stringify(requestId === undefined ? forwarded : { ...forwarded, id: requestId })
}
/**
* The host's replies on one relayed socket, passed through except for stream ids. Why renumber:
* every host allocates terminal stream ids from 1, and the phone routes binary frames by id
* across everything its desktop socket carries.
*/
export class RelayedPhoneReplies {
private readonly openRequests = new Map<string, number[]>()
private readonly swallowedRequests = new Set<string>()
private readonly desktopStreamIds = new Map<number, number>()
constructor(private readonly allocateStreamId: () => number) {}
noteForwarded(requestId: string, options: { swallowReplies?: boolean } = {}): void {
this.openRequests.set(requestId, [])
if (options.swallowReplies) {
this.swallowedRequests.add(requestId)
}
}
/** Request ids still waiting on the host; the caller fails them when the socket ends. */
takeOpenRequestIds(): string[] {
const ids = [...this.openRequests.keys()].filter((id) => !this.swallowedRequests.has(id))
this.openRequests.clear()
this.swallowedRequests.clear()
this.desktopStreamIds.clear()
return ids
}
/** The frame to hand the phone, or null to drop it. */
text(plaintext: string): string | null {
let parsed: unknown
try {
parsed = JSON.parse(plaintext)
} catch {
return plaintext
}
if (!isRecord(parsed) || typeof parsed.id !== 'string') {
return plaintext
}
const streamIds = this.openRequests.get(parsed.id)
const swallowed = this.swallowedRequests.has(parsed.id)
const result = parsed.result
let renumbered = false
if (streamIds && isRecord(result) && typeof result.streamId === 'number') {
result.streamId = this.desktopStreamId(result.streamId, streamIds)
renumbered = true
}
if (streamIds && (parsed.streaming !== true || (isRecord(result) && result.type === 'end'))) {
this.finish(parsed.id, streamIds)
}
if (swallowed) {
return null
}
return renumbered ? JSON.stringify(parsed) : plaintext
}
/** The frame to hand the phone, or null to drop a terminal frame for a stream it was never told of. */
binary(bytes: Uint8Array<ArrayBufferLike>): Uint8Array<ArrayBufferLike> | null {
const hostStreamId = readTerminalStreamFrameId(bytes)
if (hostStreamId === null) {
return bytes
}
const desktopStreamId = this.desktopStreamIds.get(hostStreamId)
if (desktopStreamId === undefined) {
return null
}
writeTerminalStreamFrameId(bytes, desktopStreamId)
return bytes
}
private desktopStreamId(hostStreamId: number, requestStreamIds: number[]): number {
const known = this.desktopStreamIds.get(hostStreamId)
if (known !== undefined) {
return known
}
const allocated = this.allocateStreamId()
this.desktopStreamIds.set(hostStreamId, allocated)
requestStreamIds.push(hostStreamId)
return allocated
}
private finish(requestId: string, hostStreamIds: number[]): void {
this.openRequests.delete(requestId)
this.swallowedRequests.delete(requestId)
for (const hostStreamId of hostStreamIds) {
this.desktopStreamIds.delete(hostStreamId)
}
}
}
function isRecord(value: unknown): value is Record<string, unknown> {
return typeof value === 'object' && value !== null && !Array.isArray(value)
}
@@ -0,0 +1,24 @@
import { describe, expect, it } from 'vitest'
import { ALL_RPC_METHODS } from '../rpc/methods'
import { MOBILE_RPC_METHOD_ALLOWLIST } from '../runtime-rpc/runtime-rpc-mobile-method-allowlist'
import { MOBILE_RPC_METHOD_ROUTES } from '../runtime-rpc/runtime-rpc-mobile-method-routing'
import { RELAYED_STREAM_CARRIERS } from './relayed-stream-carriers'
describe('relayed stream carrier census', () => {
const relayableStreams: string[] = ALL_RPC_METHODS.filter(
(method) =>
'stream' in method &&
MOBILE_RPC_METHOD_ALLOWLIST.has(method.name) &&
MOBILE_RPC_METHOD_ROUTES.get(method.name) === 'execution-host'
).map((method) => method.name)
it('classifies how every relayable stream names itself', () => {
expect(relayableStreams.filter((name) => !RELAYED_STREAM_CARRIERS.has(name))).toEqual([])
})
it('classifies nothing that is not a relayable stream', () => {
expect(
[...RELAYED_STREAM_CARRIERS.keys()].filter((name) => !relayableStreams.includes(name))
).toEqual([])
})
})
@@ -0,0 +1,19 @@
/**
* How each streaming method a phone can relay names its stream. `terminal-stream-id`: the host
* allocates ids in `result.streamId` and terminal binary headers, renumbered by RelayedPhoneReplies.
* `none`: replies carry no host-allocated id (the phone holds one screencast at a time).
* `phone-binary-frames`: the client drives the stream with binary frames, which the relay never
* forwards, so the call is refused rather than opened dead. The census test fails on a new stream.
*/
export type RelayedStreamCarrier = 'terminal-stream-id' | 'none' | 'phone-binary-frames'
export const RELAYED_STREAM_CARRIERS: ReadonlyMap<string, RelayedStreamCarrier> = new Map([
['terminal.subscribe', 'terminal-stream-id'],
['terminal.multiplex', 'phone-binary-frames'],
['browser.screencast', 'none'],
['agentSession.subscribe', 'none'],
['agentSession.subscribeStatus', 'none'],
['nativeChat.subscribe', 'none'],
['session.tabs.subscribe', 'none'],
['session.tabs.subscribeAll', 'none']
])
@@ -249,6 +249,7 @@ export class RuntimeRpcLifecycle extends RuntimeRpcWebSocketDispatch {
return
}
this.abortWebSocketDispatches(socket.ws)
this.mobileDesktopRelay?.closePhoneConnection(socket.connectionId)
// Why: subscriptions and binary streams are socket-scoped, but disconnect state is device-scoped across transports.
this.runtime.cleanupSubscriptionsForConnection(socket.connectionId)
this.runtime.cancelMobileDictationForConnection(socket.connectionId)
@@ -0,0 +1,23 @@
import { describe, expect, it } from 'vitest'
import { MOBILE_RPC_METHOD_ALLOWLIST } from './runtime-rpc-mobile-method-allowlist'
import {
MOBILE_RPC_METHOD_ROUTE_ENTRY_COUNT,
MOBILE_RPC_METHOD_ROUTES
} from './runtime-rpc-mobile-method-routing'
describe('mobile RPC method routing census', () => {
it('tags every allowlisted method exactly once', () => {
const untagged = [...MOBILE_RPC_METHOD_ALLOWLIST].filter(
(method) => !MOBILE_RPC_METHOD_ROUTES.has(method)
)
expect(untagged).toEqual([])
expect(MOBILE_RPC_METHOD_ROUTE_ENTRY_COUNT).toBe(MOBILE_RPC_METHOD_ROUTES.size)
})
it('tags nothing outside the allowlist', () => {
const stale = [...MOBILE_RPC_METHOD_ROUTES.keys()].filter(
(method) => !MOBILE_RPC_METHOD_ALLOWLIST.has(method)
)
expect(stale).toEqual([])
})
})
@@ -0,0 +1,227 @@
import { PAIRED_DESKTOP_METHODS } from './runtime-rpc-mobile-paired-desktop-methods'
// Why: a relaying Mac sends `execution-host` calls to the host named by the call's target (itself when
// local) and keeps `paired-desktop` calls; every allowlisted method must pick one (census test).
export type MobileRpcMethodRoute = 'execution-host' | 'paired-desktop'
const EXECUTION_HOST_METHODS = [
'agent.launch',
'agent.launchReplay',
'agentTeams.prepareLaunch',
'agentTeams.tmuxCompat',
'browser.back',
'browser.dialogAccept',
'browser.dialogDismiss',
'browser.forward',
'browser.goto',
'browser.keyboardInsertText',
'browser.keypress',
'browser.mouseClick',
'browser.mouseDown',
'browser.mouseMove',
'browser.mouseUp',
'browser.mouseWheel',
'browser.reload',
'browser.screencast',
'browser.screencast.unsubscribe',
'browser.tabCreate',
'browser.viewport',
// Pasted images must land on the host whose terminal reads the path.
'clipboard.abortImageUpload',
'clipboard.appendImageUploadChunk',
'clipboard.commitImageUpload',
'clipboard.saveImageAsTempFile',
'clipboard.startImageUpload',
'files.browseServerDir',
'files.createFile',
'files.list',
'files.open',
'files.openDiff',
'files.read',
'files.readChunk',
'files.readDir',
'files.readDocPreview',
'files.readPreview',
'files.readTerminalArtifact',
'files.readTerminalArtifactPreview',
'files.resolveTerminalPath',
'files.searchPaths',
'files.stat',
'files.writeTerminalArtifact',
'folderWorkspace.list',
'git.abortMerge',
'git.abortRebase',
'git.branchCompare',
'git.branchDiff',
'git.bulkStage',
'git.bulkUnstage',
'git.cancelGenerateCommitMessage',
'git.cancelGeneratePullRequestFields',
'git.checkout',
'git.commit',
'git.commitCompare',
'git.commitDiff',
'git.diff',
'git.discard',
'git.discoverCommitMessageModels',
'git.fastForward',
'git.fetch',
'git.forkSync',
'git.generateCommitMessage',
'git.generatePullRequestFields',
'git.history',
'git.localBranches',
'git.pull',
'git.push',
'git.rebaseFromBase',
'git.stage',
'git.status',
'git.unstage',
'git.upstreamStatus',
// Repo-scoped provider calls run where the desktop runs them: on the repo's host, with its login.
'github.addIssueComment',
'github.addPRReviewComment',
'github.addPRReviewCommentReply',
'github.countWorkItems',
'github.createIssue',
'github.listAssignableUsers',
'github.listLabels',
'github.listWorkItems',
'github.mergePR',
'github.prCheckDetails',
'github.prChecks',
'github.prFileContents',
'github.prForBranch',
'github.removePRReviewers',
'github.repoSlug',
'github.requestPRReviewers',
'github.rerunPRChecks',
'github.resolveReviewThread',
'github.setPRAutoMerge',
'github.setPRFileViewed',
'github.updateIssue',
'github.updatePR',
'github.updatePRState',
'github.updatePRTitle',
'github.workItem',
'github.workItemByOwnerRepo',
'github.workItemDetails',
'gitlab.addIssueComment',
'gitlab.addMRComment',
'gitlab.createIssue',
'gitlab.listWorkItems',
'gitlab.mergeMR',
'gitlab.resolveMRDiscussion',
'gitlab.todos',
'gitlab.updateIssue',
'gitlab.updateMR',
'gitlab.updateMRState',
'gitlab.workItemByPath',
'gitlab.workItemDetails',
// Shell and agent availability are facts about the machine that spawns the process.
'host.gitBash.isAvailable',
'host.platform',
'host.pwsh.isAvailable',
'host.wsl.isAvailable',
'host.wsl.listDistros',
'preflight.check',
'preflight.detectAgents',
'preflight.detectRemoteAgents',
'hostedReview.create',
'hostedReview.createStacked',
'hostedReview.forBranch',
'hostedReview.getCreationEligibility',
'linear.resolveCurrentIssue',
'markdown.readTab',
'markdown.saveTab',
'nativeChat.readSession',
'nativeChat.subscribe',
'nativeChat.unsubscribe',
'orchestration.workerTerminalUserInput',
'projectGroup.list',
'repo.baseRefDefault',
'repo.gitAvailable',
'repo.hooks',
'repo.list',
'repo.saveSparsePreset',
'repo.searchRefs',
'repo.sparsePresets',
'repo.update',
'session.tabs.activate',
'session.tabs.close',
'session.tabs.closeLifecycle',
'session.tabs.createTerminal',
'session.tabs.list',
'session.tabs.listAll',
'session.tabs.move',
'session.tabs.subscribe',
'session.tabs.subscribeAll',
'session.tabs.unsubscribe',
'session.tabs.unsubscribeAll',
'agentSession.cancel',
'agentSession.close',
'agentSession.commands',
'agentSession.conversationCommand',
'agentSession.create',
'agentSession.createSupport',
'agentSession.ensure',
'agentSession.handoffStatus',
'agentSession.history',
'agentSession.hold',
'agentSession.modelCatalog',
'agentSession.options',
'agentSession.queuedMessageDelete',
'agentSession.queuedMessageSend',
'agentSession.queuedMessagesResume',
'agentSession.readVisual',
'agentSession.release',
'agentSession.respondToApproval',
'agentSession.respondToQuestion',
'agentSession.reveal',
'agentSession.send',
'agentSession.setOption',
'agentSession.subscribe',
'agentSession.subscribeStatus',
'agentSession.unsubscribe',
'terminal.adoptOrphans',
'terminal.agentStatus',
'terminal.clearBuffer',
'terminal.close',
'terminal.closeAll',
'terminal.closeTab',
'terminal.create',
'terminal.createAgentSession',
'terminal.ensureAgentSession',
'terminal.focus',
'terminal.isRunningAgent',
'terminal.list',
'terminal.multiplex',
'terminal.read',
'terminal.rename',
'terminal.send',
'terminal.setDisplayMode',
'terminal.subscribe',
'terminal.unsubscribe',
'terminal.updateViewport',
'terminal.wait',
'worktree.activate',
'worktree.create',
'worktree.forceDeleteBranch',
'worktree.listRetiredNames',
'worktree.prefetchCreateBase',
'worktree.ps',
'worktree.resolveMrBase',
'worktree.resolvePrBase',
'worktree.rm',
'worktree.set',
'worktree.show',
'worktree.sleep'
] as const
export const MOBILE_RPC_METHOD_ROUTES: ReadonlyMap<string, MobileRpcMethodRoute> = new Map([
...PAIRED_DESKTOP_METHODS.map((method) => [method, 'paired-desktop'] as const),
...EXECUTION_HOST_METHODS.map((method) => [method, 'execution-host'] as const)
])
export const MOBILE_RPC_METHOD_ROUTE_ENTRY_COUNT =
PAIRED_DESKTOP_METHODS.length + EXECUTION_HOST_METHODS.length
@@ -0,0 +1,103 @@
/** Allowlisted phone methods the relaying Mac always answers itself; never sent with a target. */
export const PAIRED_DESKTOP_METHODS = [
// Device and pairing concerns of the phone's own pairing.
'mobileWeb.bundle.chunk',
'mobileWeb.bundle.manifest',
'mobileWeb.bundle.range',
'notifications.getMissedSince',
'notifications.registerPush',
'notifications.subscribe',
'notifications.testPush',
'notifications.unregisterPush',
'notifications.unsubscribe',
'pairing.getEndpoints',
'pairing.provisionRelay',
'speech.dictation.cancel',
'speech.dictation.chunk',
'speech.dictation.finish',
'speech.dictation.setup',
'speech.dictation.start',
'speech.models.delete',
'speech.models.download',
'speech.models.list',
// The Mac also forwards it to every open server connection of that phone.
'runtime.clientCapabilities.update',
// The Mac republishes host changes it already observes.
'runtime.clientEvents.subscribe',
'runtime.clientEvents.unsubscribe',
'status.get',
'diagnostics.memory',
'stats.summary',
'ui.get',
'ui.recordFeatureInteraction',
'ui.set',
// The desktop's accounts pane, Linear and project pickers target its active runtime, not a workspace's host.
'accounts.consumeCodexResetCredit',
'accounts.list',
'accounts.selectClaude',
'accounts.selectCodex',
'accounts.selectCodexForTarget',
'accounts.subscribe',
'accounts.unsubscribe',
'linear.addIssueComment',
'linear.agentSearchIssues',
'linear.connect',
'linear.createIssue',
'linear.createProject',
'linear.getCustomView',
'linear.getIssue',
'linear.getProject',
'linear.issueComments',
'linear.issueContext',
'linear.listCustomViewIssues',
'linear.listCustomViewProjects',
'linear.listCustomViews',
'linear.listIssues',
'linear.listProjectIssues',
'linear.listProjects',
'linear.listTeams',
'linear.mcpListIssues',
'linear.searchIssues',
'linear.selectWorkspace',
'linear.status',
'linear.teamLabels',
'linear.teamMembers',
'linear.teamStates',
'linear.updateIssue',
'github.project.addIssueCommentBySlug',
'github.project.clearItemField',
'github.project.deleteIssueCommentBySlug',
'github.project.listAccessible',
'github.project.listAssignableUsersBySlug',
'github.project.listIssueTypesBySlug',
'github.project.listLabelsBySlug',
'github.project.listViews',
'github.project.resolveRef',
'github.project.updateIssueBySlug',
'github.project.updateIssueCommentBySlug',
'github.project.updateIssueTypeBySlug',
'github.project.updateItemField',
'github.project.updatePullRequestBySlug',
'github.project.viewTable',
'github.project.workItemDetailsBySlug',
// Desktop main federates session history across hosts itself (executionHostId / scope params).
'aiVault.listSessions',
'aiVault.prepareSessionResume',
'aiVault.resolveSessionTitles',
'aiVault.searchSessions',
'aiVault.searchStatus',
// User preferences live on the Mac; the phone's settings screens mirror the desktop's.
'settings.get',
'settings.getTerminalQuickCommands',
'settings.mutateNativeChatSessionOptions',
'settings.update',
'settings.updateTerminalQuickCommands',
'terminal.getAutoRestoreFit',
'terminal.setAutoRestoreFit',
// SSH targets are configured on the Mac.
'ssh.connect',
'ssh.getState',
'ssh.listRemovedTargetLabels',
'ssh.listTargetSummaries',
'ssh.listTargets'
] as const
@@ -11,6 +11,10 @@ import { encodePairingOffer, PAIRING_OFFER_VERSION } from '../../../shared/pairi
import type { RuntimePairingReach } from '../../../shared/runtime-pairing-reach'
import { resolveAdvertisedPairingEndpoint } from '../pairing-endpoint'
import { RuntimeRpcNetworkExposure } from './runtime-rpc-network-exposure'
import { MobileDesktopRelay } from '../mobile-desktop-relay/mobile-desktop-relay'
import type { MobileDesktopRelayHosts } from '../mobile-desktop-relay/mobile-desktop-relay-hosts'
import type { MobileDesktopRelayHostState } from '../../../shared/mobile-desktop-relay-contract'
import { allocateTerminalSubscriptionStreamId } from '../rpc/methods/terminal/terminal-subscription-stream-id'
import {
createWebClientUrl,
DEVICE_REGISTRY_UNAVAILABLE_GUIDANCE,
@@ -85,6 +89,33 @@ export class RuntimeRpcPairing extends RuntimeRpcNetworkExposure {
this.mobileRelayPairingProvider = provider
}
setMobileDesktopRelayHosts(hosts: MobileDesktopRelayHosts | null): void {
this.mobileDesktopRelay?.dispose()
this.mobileDesktopRelay = hosts
? new MobileDesktopRelay({
hosts,
listPhones: () => this.listRelayedPhones(),
// Why: one allocator with local streams, so relayed ids never collide on the phone's socket.
allocateStreamId: allocateTerminalSubscriptionStreamId
})
: null
}
/** Whether a paired phone can open this server's workspaces through this desktop. */
getMobileDesktopRelayHostState(environmentId: string): Promise<MobileDesktopRelayHostState> {
return this.mobileDesktopRelay?.hostState(environmentId) ?? Promise.resolve('unavailable')
}
private listRelayedPhones(): { phoneKey: string; name: string }[] {
const computerName = this.runtime.readMachineName()
return (this.deviceRegistry?.listDevices() ?? [])
.filter((device) => device.scope === 'mobile' && device.lastSeenAt > 0)
.map((device) => ({
phoneKey: device.deviceId,
name: `${device.name} via ${computerName}`.slice(0, 128)
}))
}
async revokeMobileDevice(deviceId: string): Promise<boolean> {
const device = this.deviceRegistry?.getDevice(deviceId)
if (device?.scope !== 'mobile') {
@@ -104,6 +135,7 @@ export class RuntimeRpcPairing extends RuntimeRpcNetworkExposure {
this.mobileRelayPairingProvider?.onDemandStateChanged?.()
this.runtime.forgetClientNavigationState(deviceId)
this.mobileSocketWiring?.terminateDeviceConnections(device.token)
this.mobileDesktopRelay?.phonesChanged()
return true
}
@@ -9,6 +9,7 @@ import type { DeviceRegistry } from '../device-registry'
import type { E2EEKeypair } from '../e2ee-keypair'
import type { UnpairedDeviceAuthThrottle } from '../rpc/unpaired-device-auth-throttle'
import type { MobileSocketWiring } from '../rpc/mobile-socket-wiring'
import type { MobileDesktopRelay } from '../mobile-desktop-relay/mobile-desktop-relay'
import { RelayRevokeOutbox } from '../relay/relay-revoke-outbox'
import { PushUnregisterOutbox } from '../push/push-unregister-outbox'
import { RuntimeBinaryMessageRouter } from '../runtime-binary-message-router'
@@ -70,6 +71,8 @@ export class RuntimeRpcState {
// transports under the SAME wiring (see ensureMobileSocketWiring) instead of orphaning relay sockets.
protected detachWebSocketWiring: (() => void) | null = null
protected mobileRelayPairingProvider: MobileRelayPairingProvider | null = null
// Why: only a desktop with configured servers relays; headless serve leaves this unset.
protected mobileDesktopRelay: MobileDesktopRelay | null = null
protected mobileRelayPairingOfferQueue: Promise<void> = Promise.resolve()
protected mobileRelayPairingOfferInFlight: {
generation: number
@@ -11,6 +11,7 @@ import type { DeviceScope } from '../device-registry'
import { RuntimeRpcRequestAdmission } from './runtime-rpc-request-admission'
import { classifyRuntimeLongPoll } from './runtime-rpc-long-poll'
import { MOBILE_RPC_METHOD_ALLOWLIST } from './runtime-rpc-mobile-method-allowlist'
import { parseExecutionHostId } from '../../../shared/execution-host'
// Why: status.get has no per-connection context in the dispatcher, so stamp the scope here at the transport boundary.
function injectDeviceScope(response: string, scope: DeviceScope): string {
@@ -27,6 +28,58 @@ function injectDeviceScope(response: string, scope: DeviceScope): string {
}
export class RuntimeRpcWebSocketDispatch extends RuntimeRpcRequestAdmission {
// Why: a phone names a server's workspace with `executionHost`; only configured servers are reached,
// and a target the desktop cannot reach is an error, never a silent local run.
private relayMobileRequest(
request: RpcRequest,
rawMessage: string,
reply: (response: string) => void,
sendBinary: (response: Uint8Array<ArrayBufferLike>) => boolean | void,
socket: AuthenticatedMobileSocket
): boolean {
if (request.method === 'runtime.clientCapabilities.update') {
this.mobileDesktopRelay?.forwardClientCapabilities(socket.connectionId, rawMessage)
}
const executionHost: unknown = Reflect.get(request, 'executionHost')
if (executionHost === undefined) {
return false
}
const target = parseExecutionHostId(typeof executionHost === 'string' ? executionHost : null)
if (target?.kind === 'local' || target?.kind === 'ssh') {
return false
}
if (!target) {
reply(
JSON.stringify(this.buildError(request.id, 'invalid_argument', 'Invalid executionHost'))
)
} else if (!this.mobileDesktopRelay) {
reply(
JSON.stringify(
this.buildError(
request.id,
'remote_runtime_unavailable',
'This computer does not relay to servers'
)
)
)
} else {
this.mobileDesktopRelay.forward(
{
connectionId: socket.connectionId,
deviceId: socket.device.deviceId,
deviceToken: socket.device.deviceToken,
clientCapabilities: () => socket.clientCapabilities,
reply,
sendBinary
},
target.environmentId,
request,
rawMessage
)
}
return true
}
// Why: WebSocket dispatch is streaming (multiple responses) and auths via per-device tokens, not the shared token.
protected async handleWebSocketMessage(
rawMessage: string,
@@ -86,6 +139,14 @@ export class RuntimeRpcWebSocketDispatch extends RuntimeRpcRequestAdmission {
return
}
if (
device.scope === 'mobile' &&
authenticatedSocket &&
this.relayMobileRequest(request, rawMessage, reply, sendBinary, authenticatedSocket)
) {
return
}
// Why: bind deviceToken to this socket so ws.on('close') knows which mobile client disconnected.
if (wsTransport && ws) {
wsTransport.setClientId(ws, token)
@@ -27,9 +27,13 @@ vi.mock('../runtime/runtime-rpc', () => ({
OrcaRuntimeRpcServer: class {
start = vi.fn(async () => {})
setOnUnpairedDeviceAuthFailure = vi.fn()
setMobileDesktopRelayHosts = vi.fn()
}
}))
vi.mock('../ipc/mobile', () => ({ registerMobileHandlers: vi.fn() }))
vi.mock('../ipc/runtime-environment-mobile-relay-hosts', () => ({
createRuntimeEnvironmentMobileRelayHosts: vi.fn()
}))
vi.mock('../ipc/pty', () => ({
getLocalPtyProvider: vi.fn(),
registerHeadlessPtyRuntime: vi.fn()
@@ -8,6 +8,7 @@ import {
} from '../persistence'
import { OrcaRuntimeRpcServer } from '../runtime/runtime-rpc'
import { registerMobileHandlers } from '../ipc/mobile'
import { createRuntimeEnvironmentMobileRelayHosts } from '../ipc/runtime-environment-mobile-relay-hosts'
import { getLocalPtyProvider, registerHeadlessPtyRuntime } from '../ipc/pty'
import { LocalPtyProvider } from '../providers/local-pty-provider'
import { publishHeadlessRuntimeGraph } from '../runtime/headless-runtime-graph'
@@ -95,6 +96,7 @@ function installRuntimeRpc(
webClientRoot: getBundledWebClientRoot()
})
state.runtimeRpc = runtimeRpc
runtimeRpc.setMobileDesktopRelayHosts(createRuntimeEnvironmentMobileRelayHosts())
registerMobileHandlers(runtimeRpc, {
getRelayStatus: getDesktopRelayStatus,
consumePendingUnpairedDeviceAuthFailure: (webContentsId) => {
@@ -0,0 +1,33 @@
import { z } from 'zod'
// Why: a paired desktop relays each of its phones to this host as an ordinary phone device it owns.
export const DELEGATED_MOBILE_DEVICE_SYNC_METHOD = 'pairing.delegatedMobileDevice.sync'
export const DELEGATED_MOBILE_DEVICE_SYNC_MAX_PHONES = 32
const DelegatedPhoneSchema = z
.object({
phoneKey: z.string().min(1).max(128),
name: z.string().min(1).max(128)
})
.strict()
export const DelegatedMobileDeviceSyncParamsSchema = z
.object({
phones: z
.array(DelegatedPhoneSchema)
.max(DELEGATED_MOBILE_DEVICE_SYNC_MAX_PHONES)
.refine((phones) => new Set(phones.map((phone) => phone.phoneKey)).size === phones.length, {
message: 'duplicate phoneKey'
})
})
.strict()
export type DelegatedMobileDeviceSyncParams = z.infer<typeof DelegatedMobileDeviceSyncParamsSchema>
export type DelegatedMobileDeviceSyncResult = {
devices: { phoneKey: string; deviceId: string; token: string }[]
}
// Why: the paired desktop relays a phone to this host only when the host can mint its delegated device.
export const DELEGATED_MOBILE_DEVICE_SYNC_RUNTIME_CAPABILITY =
'pairing.delegated-mobile-device.v1' as const
@@ -0,0 +1,7 @@
// Why: a phone may put `executionHost` on a request only after its desktop advertises this; an older
// desktop ignores the field and would run the call on itself. Not in RUNTIME_CAPABILITIES yet: the
// phone's host-scoped client ships first (S2), and adding it there is the one line that turns relay on.
export const MOBILE_DESKTOP_RELAY_RUNTIME_CAPABILITY = 'mobile.desktop-relay.v1' as const
/** A paired desktop's relay readiness for one of its servers. */
export type MobileDesktopRelayHostState = 'ready' | 'update-needed' | 'unavailable'
@@ -0,0 +1,224 @@
import WebSocket from 'ws'
import type { PairingOffer } from './pairing'
import type { RuntimeCapability } from './protocol-version'
import {
decrypt,
decryptBytes,
deriveSharedKey,
encrypt,
generateKeyPair,
publicKeyFromBase64,
publicKeyToBase64
} from './e2ee-crypto'
import {
classifyRemoteRuntimeReadyFrame,
formatRemoteRuntimeCloseMessage,
parseRemoteRuntimeAuthenticatedFrame
} from './remote-runtime-client-handshake'
import { RemoteRuntimeClientError } from './remote-runtime-client-error'
import {
remoteRuntimeConnectFailureMessage,
remoteRuntimeConnectOptions
} from './remote-runtime-connect-bound'
import {
REMOTE_RUNTIME_MAX_WEBSOCKET_FRAME_BYTES,
serializeRemoteRuntimePayload
} from './remote-runtime-memory-limits'
import { closeRemoteRuntimeSocket } from './remote-runtime-socket-close'
import {
startRemoteRuntimeSocketLiveness,
type RemoteRuntimeSocketLivenessOptions
} from './remote-runtime-socket-liveness'
import { RemoteRuntimeSubscriptionOutbound } from './remote-runtime-subscription-outbound'
export type RemoteRuntimePassthroughSocket = {
/** Queues one plaintext RPC frame; false when the socket can no longer carry it. */
send: (plaintext: string) => boolean
close: () => void
}
export type RemoteRuntimePassthroughCallbacks = {
onText: (plaintext: string) => void
onBinary: (bytes: Uint8Array<ArrayBufferLike>) => void
/** Called once when an open socket ends for any reason other than `close()`. */
onClose: (error: RemoteRuntimeClientError) => void
}
/**
* An authenticated E2EE socket that carries any number of requests and streams and hands every
* decrypted frame back untouched. Why not the subscription transport: it owns one request id per
* socket and fails on any other, while a relay forwards whatever its client multiplexes.
*/
export function openRemoteRuntimePassthroughSocket(
pairing: PairingOffer,
clientCapabilities: readonly RuntimeCapability[],
callbacks: RemoteRuntimePassthroughCallbacks,
options: RemoteRuntimeSocketLivenessOptions & { timeoutMs: number }
): Promise<RemoteRuntimePassthroughSocket> {
return new Promise((resolve, reject) => {
const keyPair = generateKeyPair()
const sharedKey = deriveSharedKey(keyPair.secretKey, publicKeyFromBase64(pairing.publicKeyB64))
let state: 'awaiting_ready' | 'awaiting_authenticated' | 'ready' | 'ended' = 'awaiting_ready'
let ws: WebSocket
const outbound = new RemoteRuntimeSubscriptionOutbound({ fail: (error) => end(error) })
const handle: RemoteRuntimePassthroughSocket = {
send: (plaintext) =>
state === 'ready' && ws.readyState === WebSocket.OPEN
? outbound.enqueueRequest(ws, encrypt(plaintext, sharedKey))
: false,
close: () => end(null)
}
const timeout = setTimeout(
() =>
end(
new RemoteRuntimeClientError(
'runtime_timeout',
'Timed out connecting to the remote Orca runtime.'
)
),
options.timeoutMs
)
const liveness = startRemoteRuntimeSocketLiveness({
ping: () => {
if (ws.readyState === WebSocket.OPEN) {
ws.ping()
}
},
onDead: () =>
end(
new RemoteRuntimeClientError(
'remote_runtime_unavailable',
'Remote Orca runtime stopped responding; the connection was reset.'
)
),
options
})
function end(error: RemoteRuntimeClientError | null): void {
if (state === 'ended') {
return
}
const wasReady = state === 'ready'
state = 'ended'
clearTimeout(timeout)
liveness.stop()
outbound.releaseQueues()
outbound.retainSocketMemoryUntilClose(ws)
ws.off('message', onMessage)
closeRemoteRuntimeSocket(ws)
if (!wasReady) {
reject(
error ??
new RemoteRuntimeClientError('remote_runtime_unavailable', 'Remote runtime closed.')
)
} else if (error) {
callbacks.onClose(error)
}
}
function protocolError(message: string): RemoteRuntimeClientError {
return new RemoteRuntimeClientError('invalid_runtime_response', message)
}
function onMessage(data: WebSocket.RawData, isBinary: boolean): void {
liveness.noteActivity()
if (isBinary) {
// Why Buffer only: a socket left at the default binaryType delivers nothing else.
const bytes =
state === 'ready' && Buffer.isBuffer(data)
? decryptBytes(new Uint8Array(data), sharedKey)
: null
if (!bytes) {
end(protocolError('Remote Orca runtime returned an unreadable binary frame.'))
return
}
callbacks.onBinary(bytes)
return
}
const frame = data.toString()
if (state === 'awaiting_ready') {
if (classifyRemoteRuntimeReadyFrame(frame) !== 'ready') {
end(protocolError('Remote Orca runtime returned an invalid E2EE handshake frame.'))
return
}
state = 'awaiting_authenticated'
ws.send(
encrypt(
serializeRemoteRuntimePayload({
type: 'e2ee_auth',
deviceToken: pairing.deviceToken,
clientCapabilities
}),
sharedKey
)
)
return
}
const plaintext = decrypt(frame, sharedKey)
if (plaintext === null) {
end(protocolError('Remote Orca runtime returned an undecryptable frame.'))
return
}
if (state === 'ready') {
callbacks.onText(plaintext)
return
}
const authenticated = parseRemoteRuntimeAuthenticatedFrame(plaintext)
if (authenticated.kind !== 'authenticated') {
end(
authenticated.kind === 'rejected' && authenticated.unauthorized
? new RemoteRuntimeClientError(
'unauthorized',
'Remote Orca runtime rejected the pairing token.'
)
: protocolError('Remote Orca runtime returned an invalid E2EE auth frame.')
)
return
}
state = 'ready'
clearTimeout(timeout)
resolve(handle)
}
try {
ws = new WebSocket(
pairing.endpoint,
remoteRuntimeConnectOptions({ maxPayload: REMOTE_RUNTIME_MAX_WEBSOCKET_FRAME_BYTES })
)
} catch (error) {
clearTimeout(timeout)
liveness.stop()
reject(
new RemoteRuntimeClientError(
'invalid_argument',
`Invalid remote endpoint: ${String(error)}`
)
)
return
}
ws.once('open', () =>
ws.send(
JSON.stringify({ type: 'e2ee_hello', publicKeyB64: publicKeyToBase64(keyPair.publicKey) })
)
)
ws.on('error', (error) =>
end(
new RemoteRuntimeClientError(
'remote_runtime_unavailable',
remoteRuntimeConnectFailureMessage(error, pairing.endpoint)
)
)
)
ws.on('close', (code, reason) =>
end(
new RemoteRuntimeClientError(
'remote_runtime_unavailable',
formatRemoteRuntimeCloseMessage(code, reason)
)
)
)
ws.on('message', onMessage)
ws.on('pong', () => liveness.noteActivity())
ws.on('ping', () => liveness.noteActivity())
})
}
+17
View File
@@ -79,6 +79,23 @@ export function decodeTerminalStreamFrame(bytes: Uint8Array): TerminalStreamFram
}
}
/** The stream id in a terminal frame's header, or null when the bytes are not a terminal frame. */
export function readTerminalStreamFrameId(bytes: Uint8Array): number | null {
if (
bytes.length < HEADER_BYTES ||
bytes[0] !== TERMINAL_STREAM_KIND ||
bytes[1] !== TERMINAL_STREAM_VERSION
) {
return null
}
return new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength).getUint32(4, true)
}
/** Overwrites the stream id of a frame `readTerminalStreamFrameId` accepted. */
export function writeTerminalStreamFrameId(bytes: Uint8Array, streamId: number): void {
new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength).setUint32(4, streamId, true)
}
export function encodeTerminalStreamJson(value: unknown): Uint8Array {
return new TextEncoder().encode(JSON.stringify(value))
}