test(relay): prove desktop lifecycle and E2EE splice

Co-authored-by: Orca <help@stably.ai>

src/main/runtime/relay/mobile-relay-e2ee.integration.test.ts

@@ -0,0 +1,242 @@
+import { mkdtempSync, rmSync } from 'node:fs'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { afterEach, describe, expect, it, vi } from 'vitest'
+import nacl from 'tweetnacl'
+import WebSocketClient, { WebSocketServer, type RawData, type WebSocket } from 'ws'
+import { DeviceRegistry } from '../device-registry'
+import { MobileSocketWiring } from '../rpc/mobile-socket-wiring'
+import { CloudRelayTransport } from '../rpc/relay-transport'
+import { deriveRelayHostId } from './relay-http-client'
+import { SimulatedMobileE2EEV2Peer } from './simulated-mobile-e2ee-v2-peer'
+
+function deferred<T>() {
+  let resolve!: (value: T) => void
+  const promise = new Promise<T>((resolvePromise) => {
+    resolve = resolvePromise
+  })
+  return { promise, resolve }
+}
+
+function waitForOpen(socket: WebSocketClient): Promise<void> {
+  return new Promise((resolve, reject) => {
+    socket.once('open', resolve)
+    socket.once('error', reject)
+  })
+}
+
+function nextText(socket: WebSocketClient): Promise<string> {
+  return new Promise((resolve) => {
+    socket.once('message', (raw) => resolve(raw.toString()))
+  })
+}
+
+function forward(socket: WebSocket, raw: RawData, isBinary: boolean): void {
+  if (socket.readyState === socket.OPEN) {
+    socket.send(raw, { binary: isBinary })
+  }
+}
+
+describe('desktop relay E2EE integration', () => {
+  const servers: WebSocketServer[] = []
+  const transports: CloudRelayTransport[] = []
+  const userDataPaths: string[] = []
+
+  afterEach(async () => {
+    await Promise.all(transports.splice(0).map((transport) => transport.stop()))
+    await Promise.all(
+      servers.splice(0).map(
+        (server) =>
+          new Promise<void>((resolve) => {
+            for (const client of server.clients) {
+              client.terminate()
+            }
+            server.close(() => resolve())
+          })
+      )
+    )
+    for (const path of userDataPaths.splice(0)) {
+      rmSync(path, { recursive: true, force: true })
+    }
+  })
+
+  it('splices a simulated phone through CloudRelayTransport with real NaCl E2EE v2', async () => {
+    const relay = new WebSocketServer({ port: 0, perMessageDeflate: false })
+    servers.push(relay)
+    await new Promise<void>((resolve) => relay.once('listening', resolve))
+    const address = relay.address()
+    if (!address || typeof address === 'string') {
+      throw new Error('expected local relay TCP address')
+    }
+    let hostSocket: WebSocket | null = null
+    let phoneSocket: WebSocket | null = null
+    let phoneAuthorized = false
+    const maybeSplice = (): void => {
+      if (!hostSocket || !phoneSocket || !phoneAuthorized) {
+        return
+      }
+      const host = hostSocket
+      const phone = phoneSocket
+      host.on('message', (raw, isBinary) => forward(phone, raw, isBinary))
+      phone.on('message', (raw, isBinary) => forward(host, raw, isBinary))
+      phone.send(
+        JSON.stringify({
+          type: 'relay-hello',
+          ok: true,
+          credentialKind: 'invite',
+          leaseExpiresAt: Date.now() + 60_000
+        })
+      )
+    }
+    relay.on('connection', (socket, request) => {
+      expect(request.url).not.toContain('?')
+      if (request.url === '/v1/host/data/connection-1') {
+        socket.once('message', (raw) => {
+          expect(JSON.parse(raw.toString())).toEqual({
+            type: 'host-data-auth',
+            v: 1,
+            connTicket: 'A'.repeat(43),
+            generation: 1
+          })
+          hostSocket = socket
+          maybeSplice()
+        })
+        return
+      }
+      if (request.url?.startsWith('/v1/connect/')) {
+        socket.once('message', (raw) => {
+          expect(JSON.parse(raw.toString())).toEqual({
+            type: 'relay-auth',
+            v: 1,
+            mode: 'connect',
+            credential: 'B'.repeat(43)
+          })
+          phoneSocket = socket
+          phoneAuthorized = true
+          maybeSplice()
+        })
+      }
+    })
+
+    const userDataPath = mkdtempSync(join(tmpdir(), 'orca-relay-e2ee-'))
+    userDataPaths.push(userDataPath)
+    const registry = new DeviceRegistry(userDataPath)
+    const device = registry.addDevice('Phone', 'mobile')
+    const desktopKeys = nacl.box.keyPair()
+    const relayHostId = deriveRelayHostId(desktopKeys.publicKey)
+    const receivedText = deferred<string>()
+    const receivedBinary = deferred<Uint8Array>()
+    const phoneBinary = deferred<Uint8Array>()
+    const wiring = new MobileSocketWiring({
+      deviceRegistry: registry,
+      e2eeKeypair: {
+        publicKey: desktopKeys.publicKey,
+        secretKey: desktopKeys.secretKey,
+        publicKeyB64: Buffer.from(desktopKeys.publicKey).toString('base64')
+      },
+      onText: (socket, plaintext, reply, sendBinary) => {
+        expect(socket.transport).toEqual({
+          transport: 'relay',
+          relayHostId,
+          relayDeviceId: device.deviceId,
+          basisConnId: 'connection-1',
+          credentialKind: 'invite'
+        })
+        receivedText.resolve(plaintext)
+        reply(JSON.stringify({ id: 'rpc-1', ok: true, result: { path: 'relay' } }))
+        sendBinary(new Uint8Array([4, 5, 6]))
+      },
+      onBinary: (_socket, bytes) => receivedBinary.resolve(new Uint8Array(bytes)),
+      onClose: vi.fn()
+    })
+    const transport = new CloudRelayTransport({
+      cellUrl: `http://127.0.0.1:${address.port}`,
+      relayHostId,
+      generation: 1
+    })
+    transports.push(transport)
+    wiring.attachTransport(transport, (socket) => transport.metadataFor(socket))
+    await transport.start()
+    await transport.openConnection({
+      connId: 'connection-1',
+      connTicket: 'A'.repeat(43),
+      kind: 'invite',
+      relayDeviceId: device.deviceId,
+      attachDeadlineMs: 5_000
+    })
+
+    const phone = new WebSocketClient(`ws://127.0.0.1:${address.port}/v1/connect/${relayHostId}`, {
+      perMessageDeflate: false
+    })
+    await waitForOpen(phone)
+    const relayHello = nextText(phone)
+    phone.send(
+      JSON.stringify({
+        type: 'relay-auth',
+        v: 1,
+        mode: 'connect',
+        credential: 'B'.repeat(43)
+      })
+    )
+    await expect(relayHello).resolves.toMatch('"ok":true')
+
+    const authenticated = deferred<void>()
+    const phoneText = deferred<string>()
+    const phoneSession = new SimulatedMobileE2EEV2Peer(
+      nacl.box.keyPair(),
+      desktopKeys.publicKey,
+      relayHostId
+    )
+    let phoneState: 'awaiting-ready' | 'awaiting-authenticated' | 'ready' = 'awaiting-ready'
+    phone.on('message', (raw, isBinary) => {
+      if (phoneState === 'awaiting-ready') {
+        expect(isBinary).toBe(false)
+        expect(phoneSession.acceptReady(JSON.parse(raw.toString()))).toBe(true)
+        phoneState = 'awaiting-authenticated'
+        phone.send(
+          phoneSession.sealText(
+            JSON.stringify({
+              type: 'e2ee_auth',
+              v: 2,
+              transcriptHashB64: phoneSession.transcriptHashB64,
+              deviceToken: device.token
+            })
+          )
+        )
+        return
+      }
+      const plaintext = isBinary
+        ? phoneSession.openBinary(new Uint8Array(raw as Buffer))
+        : phoneSession.openText(raw.toString())
+      expect(plaintext).not.toBeNull()
+      if (phoneState === 'awaiting-authenticated') {
+        expect(JSON.parse(plaintext as string)).toEqual({
+          type: 'e2ee_authenticated',
+          v: 2,
+          transcriptHashB64: phoneSession.transcriptHashB64
+        })
+        phoneState = 'ready'
+        authenticated.resolve()
+      } else if (typeof plaintext === 'string') {
+        phoneText.resolve(plaintext)
+      } else {
+        phoneBinary.resolve(plaintext!)
+      }
+    })
+    phone.send(JSON.stringify(phoneSession.hello))
+    await authenticated.promise
+    phone.send(phoneSession.sealText(JSON.stringify({ id: 'rpc-1', method: 'status.get' })))
+    phone.send(phoneSession.sealBinary(new Uint8Array([1, 2, 3])))
+
+    await expect(receivedText.promise).resolves.toBe(
+      JSON.stringify({ id: 'rpc-1', method: 'status.get' })
+    )
+    await expect(receivedBinary.promise).resolves.toEqual(new Uint8Array([1, 2, 3]))
+    await expect(phoneText.promise).resolves.toBe(
+      JSON.stringify({ id: 'rpc-1', ok: true, result: { path: 'relay' } })
+    )
+    await expect(phoneBinary.promise).resolves.toEqual(new Uint8Array([4, 5, 6]))
+
+    phone.terminate()
+  }, 15_000)
+})

src/main/runtime/relay/relay-auth-coordinator.test.ts

@@ -91,4 +91,111 @@ describe('RelayAuthCoordinator', () => {
     coordinator.reconcile()
     await expect(refreshAccessToken!()).resolves.toBeNull()
   })
+
+  it('invalidates pending ownership immediately while broker opening is paused', async () => {
+    const firstOpen = deferred<{ closeNow(): void }>()
+    const firstClose = vi.fn()
+    let firstIsCurrent: (() => boolean) | null = null
+    const openBroker = vi
+      .fn()
+      .mockImplementationOnce((input) => {
+        firstIsCurrent = input.isCurrent
+        return firstOpen.promise
+      })
+      .mockResolvedValueOnce({ closeNow: vi.fn() })
+    const coordinator = new RelayAuthCoordinator({
+      readContext: async () => context,
+      openBroker,
+      onStatus: vi.fn()
+    })
+
+    coordinator.reconcile()
+    await vi.waitFor(() => expect(firstIsCurrent).not.toBeNull())
+    coordinator.reconcile()
+    expect(firstIsCurrent!()).toBe(false)
+    await vi.waitFor(() => expect(openBroker).toHaveBeenCalledTimes(2))
+    firstOpen.resolve({ closeNow: firstClose })
+    await vi.waitFor(() => expect(firstClose).toHaveBeenCalledOnce())
+  })
+
+  it('rejects a token refresh whose session read crosses an auth mutation', async () => {
+    const refreshRead = deferred<RelayAuthContext | null>()
+    let readCount = 0
+    let current = context
+    let refreshAccessToken: (() => Promise<string | null>) | null = null
+    const coordinator = new RelayAuthCoordinator({
+      readContext: () => {
+        readCount += 1
+        return readCount === 2 ? refreshRead.promise : Promise.resolve(current)
+      },
+      openBroker: async (input) => {
+        refreshAccessToken = input.refreshAccessToken
+        return { closeNow: vi.fn() }
+      },
+      onStatus: vi.fn()
+    })
+    coordinator.reconcile()
+    await vi.waitFor(() => expect(refreshAccessToken).not.toBeNull())
+    const refreshing = refreshAccessToken!()
+    current = { ...context, identity: { ...context.identity, organizationId: 'org-2' } }
+    coordinator.reconcile()
+    refreshRead.resolve(context)
+
+    await expect(refreshing).resolves.toBeNull()
+    await vi.waitFor(() => expect(readCount).toBeGreaterThanOrEqual(3))
+  })
+
+  it('reconnects automatically after a signed-in process restart or relaunch fence', async () => {
+    const firstBroker = { closeNow: vi.fn() }
+    const firstCoordinator = new RelayAuthCoordinator({
+      readContext: async () => context,
+      openBroker: async () => firstBroker,
+      onStatus: vi.fn()
+    })
+    firstCoordinator.reconcile()
+    await vi.waitFor(() => expect(firstCoordinator.getActiveBroker()).toBe(firstBroker))
+    firstCoordinator.fenceAndCloseNow()
+    expect(firstBroker.closeNow).toHaveBeenCalledOnce()
+
+    const reopenedBroker = { closeNow: vi.fn() }
+    const reopenedCoordinator = new RelayAuthCoordinator({
+      // Why: normal quit/relaunch preserves the session store, so a fresh
+      // process reads the same entitled identity and opens without new login.
+      readContext: async () => context,
+      openBroker: async () => reopenedBroker,
+      onStatus: vi.fn()
+    })
+    reopenedCoordinator.reconcile()
+    await vi.waitFor(() => expect(reopenedCoordinator.getActiveBroker()).toBe(reopenedBroker))
+  })
+
+  it('closes and reopens for valid profile and organization identity switches', async () => {
+    let current = context
+    const brokers = Array.from({ length: 3 }, () => ({ closeNow: vi.fn() }))
+    const openBroker = vi
+      .fn()
+      .mockResolvedValueOnce(brokers[0])
+      .mockResolvedValueOnce(brokers[1])
+      .mockResolvedValueOnce(brokers[2])
+    const coordinator = new RelayAuthCoordinator({
+      readContext: async () => current,
+      openBroker,
+      onStatus: vi.fn()
+    })
+    coordinator.reconcile()
+    await vi.waitFor(() => expect(coordinator.getActiveBroker()).toBe(brokers[0]))
+
+    current = { ...context, identity: { ...context.identity, profileId: 'profile-2' } }
+    coordinator.reconcile()
+    await vi.waitFor(() => expect(coordinator.getActiveBroker()).toBe(brokers[1]))
+    expect(brokers[0]!.closeNow).toHaveBeenCalledOnce()
+
+    current = {
+      ...context,
+      identity: { ...context.identity, profileId: 'profile-2', organizationId: 'org-2' }
+    }
+    coordinator.reconcile()
+    await vi.waitFor(() => expect(coordinator.getActiveBroker()).toBe(brokers[2]))
+    expect(brokers[1]!.closeNow).toHaveBeenCalledOnce()
+  })
 })

src/main/runtime/relay/relay-session-broker-contract.ts

@@ -0,0 +1,31 @@
+import type WebSocket from 'ws'
+import type { OrcaCloudAuthConfig } from '../../orca-profiles/profile-cloud-auth-config'
+import type { MobileRelayStatus } from '../../../shared/mobile-relay-status'
+import type { E2EEKeypair } from '../e2ee-keypair'
+import type { MobileSocketWiring } from '../rpc/mobile-socket-wiring'
+
+export type RelayBrokerStatus = MobileRelayStatus
+
+export type RelayIdentity = {
+  userId: string
+  profileId: string
+  organizationId: string
+}
+
+export type RelaySessionBrokerOptions = {
+  authConfig: OrcaCloudAuthConfig
+  accessToken: string
+  identity: RelayIdentity
+  keypair: E2EEKeypair
+  appVersion: string
+  mobileSocketWiring: MobileSocketWiring
+  isCurrent: () => boolean
+  refreshAccessToken: () => Promise<string | null>
+  onStatus: (status: RelayBrokerStatus) => void
+  onResolveDirector: () => void
+  fetch?: typeof globalThis.fetch
+  createControlSocket?: (url: string, relayJwt: string) => WebSocket
+  createDataSocket?: (url: string) => WebSocket
+  random?: () => number
+  now?: () => number
+}

src/main/runtime/relay/relay-session-broker.test.ts

@@ -0,0 +1,117 @@
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+import nacl from 'tweetnacl'
+import type { OrcaCloudAuthConfig } from '../../orca-profiles/profile-cloud-auth-config'
+import type { RelayHostHelloAckMessage } from './relay-control-protocol'
+import type * as RelayHttpClientModule from './relay-http-client'
+
+const fakes = vi.hoisted(() => ({
+  controls: [] as {
+    connect: ReturnType<typeof vi.fn>
+    closeNow: ReturnType<typeof vi.fn>
+  }[],
+  transports: [] as {
+    start: ReturnType<typeof vi.fn>
+    stop: ReturnType<typeof vi.fn>
+    setGeneration: ReturnType<typeof vi.fn>
+    metadataFor: ReturnType<typeof vi.fn>
+  }[],
+  controlConnect: vi.fn(),
+  exchange: vi.fn(),
+  assign: vi.fn()
+}))
+
+vi.mock('./relay-http-client', async (importOriginal) => ({
+  ...(await importOriginal<typeof RelayHttpClientModule>()),
+  exchangeRelayAuthorization: fakes.exchange,
+  requestRelayAssignment: fakes.assign
+}))
+
+vi.mock('./relay-control-client', () => ({
+  RelayControlClient: class {
+    connect = fakes.controlConnect
+    closeNow = vi.fn()
+
+    constructor() {
+      fakes.controls.push(this)
+    }
+  }
+}))
+
+vi.mock('../rpc/relay-transport', () => ({
+  CloudRelayTransport: class {
+    start = vi.fn().mockResolvedValue(undefined)
+    stop = vi.fn().mockResolvedValue(undefined)
+    setGeneration = vi.fn()
+    metadataFor = vi.fn()
+
+    constructor() {
+      fakes.transports.push(this)
+    }
+  }
+}))
+
+import { RelaySessionBroker, StaleRelayBrokerError } from './relay-session-broker'
+
+function deferred<T>() {
+  let resolve!: (value: T) => void
+  const promise = new Promise<T>((resolvePromise) => {
+    resolve = resolvePromise
+  })
+  return { promise, resolve }
+}
+
+describe('RelaySessionBroker lifecycle ownership', () => {
+  beforeEach(() => {
+    fakes.controls.length = 0
+    fakes.transports.length = 0
+    fakes.controlConnect.mockReset()
+    fakes.exchange.mockReset().mockResolvedValue({ relayToken: 'relay-jwt', expiresAt: 60_000 })
+    fakes.assign.mockReset().mockResolvedValue({
+      cellUrl: 'https://relay.example.test',
+      assignmentEpoch: 1,
+      leaseExpiresAt: 60_000
+    })
+  })
+
+  it('closes partially opened resources without publishing stale state', async () => {
+    const controlAck = deferred<RelayHostHelloAckMessage>()
+    fakes.controlConnect.mockReturnValue(controlAck.promise)
+    let current = true
+    const statuses: string[] = []
+    const keypair = nacl.box.keyPair()
+    const connecting = RelaySessionBroker.connect({
+      authConfig: {
+        relayTokenEndpoint: 'https://auth.example.test/v1/relay-token',
+        relayDirectorUrl: 'https://relay.example.test'
+      } as OrcaCloudAuthConfig,
+      accessToken: 'access-token',
+      identity: { userId: 'user-1', profileId: 'profile-1', organizationId: 'org-1' },
+      keypair: {
+        ...keypair,
+        publicKeyB64: Buffer.from(keypair.publicKey).toString('base64')
+      },
+      appVersion: '1.0.0',
+      mobileSocketWiring: { attachTransport: vi.fn() } as never,
+      isCurrent: () => current,
+      refreshAccessToken: async () => null,
+      onStatus: (status) => statuses.push(status),
+      onResolveDirector: vi.fn()
+    })
+    await vi.waitFor(() => expect(fakes.controls).toHaveLength(1))
+    current = false
+    controlAck.resolve({
+      type: 'host-hello-ack',
+      v: 1,
+      generation: 1,
+      controlResumeSecret: 'A'.repeat(43),
+      leaseExpiresAt: 60_000,
+      activeConnIds: [],
+      pendingConns: []
+    })
+
+    await expect(connecting).rejects.toBeInstanceOf(StaleRelayBrokerError)
+    expect(fakes.controls[0]!.closeNow).toHaveBeenCalledOnce()
+    expect(fakes.transports[0]!.stop).toHaveBeenCalledOnce()
+    expect(statuses).toEqual(['connecting'])
+  })
+})

src/main/runtime/relay/relay-session-broker.ts

@@ -1,5 +1,3 @@
-import type WebSocket from 'ws'
-import type { OrcaCloudAuthConfig } from '../../orca-profiles/profile-cloud-auth-config'
 import type { PairingRelay } from '../../../shared/mobile-relay-pairing-offer'
 import type {
   DeviceCredentialInstalled,
@@ -8,8 +6,6 @@ import type {
   MobileRelayEndpoint,
   PairingProvisionRelayParams
 } from '../../../shared/mobile-relay-credential-contract'
-import type { E2EEKeypair } from '../e2ee-keypair'
-import type { MobileSocketWiring } from '../rpc/mobile-socket-wiring'
 import { CloudRelayTransport } from '../rpc/relay-transport'
 import { RelayControlClient } from './relay-control-client'
 import type { DeviceCredentialInstallAuthorization } from './relay-control-requests'
@@ -20,32 +16,9 @@ import {
   type RelayAuthorization,
   type RelayAssignment
 } from './relay-http-client'
+import type { RelayBrokerStatus, RelaySessionBrokerOptions } from './relay-session-broker-contract'
 
-export type RelayBrokerStatus = 'connecting' | 'registered' | 'draining' | 'offline'
-
-type RelayIdentity = {
-  userId: string
-  profileId: string
-  organizationId: string
-}
-
-type RelaySessionBrokerOptions = {
-  authConfig: OrcaCloudAuthConfig
-  accessToken: string
-  identity: RelayIdentity
-  keypair: E2EEKeypair
-  appVersion: string
-  mobileSocketWiring: MobileSocketWiring
-  isCurrent: () => boolean
-  refreshAccessToken: () => Promise<string | null>
-  onStatus: (status: RelayBrokerStatus) => void
-  onResolveDirector: () => void
-  fetch?: typeof globalThis.fetch
-  createControlSocket?: (url: string, relayJwt: string) => WebSocket
-  createDataSocket?: (url: string) => WebSocket
-  random?: () => number
-  now?: () => number
-}
+export type { RelayBrokerStatus } from './relay-session-broker-contract'
 
 export class StaleRelayBrokerError extends Error {
   constructor() {
@@ -185,6 +158,7 @@ export class RelaySessionBroker {
     if (this.closed) {
       return
     }
+    const publishOffline = this.options.isCurrent()
     this.closed = true
     if (this.refreshTimer) {
       clearTimeout(this.refreshTimer)
@@ -194,11 +168,13 @@ export class RelaySessionBroker {
     this.control = null
     void this.transport?.stop()
     this.transport = null
-    this.options.onStatus('offline')
+    if (publishOffline) {
+      this.options.onStatus('offline')
+    }
   }
 
   private async open(accessToken: string): Promise<void> {
-    this.options.onStatus('connecting')
+    this.publishStatus('connecting')
     const authorization = await exchangeRelayAuthorization({
       endpoint: this.options.authConfig.relayTokenEndpoint,
       accessToken,
@@ -219,6 +195,9 @@ export class RelaySessionBroker {
       generation: 0,
       createSocket: this.options.createDataSocket
     })
+    // Why: stale-epoch cleanup must own partially opened resources before the
+    // next await, even though the broker is not externally active yet.
+    this.transport = transport
     this.options.mobileSocketWiring.attachTransport(transport, (ws) => transport.metadataFor(ws))
     await transport.start()
     this.assertCurrent()
@@ -231,21 +210,27 @@ export class RelaySessionBroker {
       keypair: this.options.keypair,
       appVersion: this.options.appVersion,
       onConnectionOpen: (message) => {
-        void transport.openConnection(message).catch(() => {})
+        if (this.isCurrent()) {
+          void transport.openConnection(message).catch(() => {})
+        }
       },
       onDrain: () => {
-        this.options.onStatus('draining')
+        if (!this.isCurrent()) {
+          return
+        }
+        this.publishStatus('draining')
         // Why: the data plane never chooses recovery targets; every drain
         // returns to the configured stable director.
         this.options.onResolveDirector()
       },
       onClose: () => {
-        if (!this.closed) {
-          this.options.onStatus('offline')
+        if (this.isCurrent()) {
+          this.publishStatus('offline')
         }
       },
       createSocket: this.options.createControlSocket
     })
+    this.control = control
     const ack = await control.connect()
     this.assertCurrent()
     if (ack.generation <= 0) {
@@ -254,9 +239,7 @@ export class RelaySessionBroker {
     transport.setGeneration(ack.generation)
     this.authorization = authorization
     this.assignment = assignment
-    this.transport = transport
-    this.control = control
-    this.options.onStatus('registered')
+    this.publishStatus('registered')
     this.scheduleRefresh()
   }
 
@@ -307,8 +290,18 @@ export class RelaySessionBroker {
   }
 
   private assertCurrent(): void {
-    if (this.closed || !this.options.isCurrent()) {
+    if (!this.isCurrent()) {
       throw new StaleRelayBrokerError()
     }
   }
+
+  private isCurrent(): boolean {
+    return !this.closed && this.options.isCurrent()
+  }
+
+  private publishStatus(status: RelayBrokerStatus): void {
+    if (this.isCurrent()) {
+      this.options.onStatus(status)
+    }
+  }
 }

src/main/runtime/relay/simulated-mobile-e2ee-v2-peer.ts

@@ -0,0 +1,116 @@
+import nacl from 'tweetnacl'
+import {
+  encodeMobileE2EEV2Transcript,
+  validateMobileE2EEV2Handshake,
+  type MobileE2EEPayloadKind,
+  type MobileE2EEV2Hello
+} from '../../../shared/mobile-e2ee-v2-contract'
+import {
+  openMobileE2EEV2Frame,
+  sealMobileE2EEV2Frame
+} from '../../../shared/mobile-e2ee-v2-framing'
+import { deriveSharedKey } from '../rpc/e2ee-crypto'
+import { deriveMobileE2EEV2KeySchedule } from '../rpc/mobile-e2ee-v2-key-schedule'
+
+// Why: the relay integration test needs an independent mobile-side wire peer
+// without importing Expo modules into the desktop Node TypeScript project.
+export class SimulatedMobileE2EEV2Peer {
+  readonly hello: MobileE2EEV2Hello
+  private inboundCounter = 0n
+  private outboundCounter = 0n
+  private schedule: ReturnType<typeof deriveMobileE2EEV2KeySchedule> | null = null
+
+  constructor(
+    private readonly clientKeys: nacl.BoxKeyPair,
+    private readonly desktopPublicKey: Uint8Array,
+    relayHostId: string,
+    clientNonce = nacl.randomBytes(32)
+  ) {
+    this.hello = {
+      type: 'e2ee_hello',
+      v: 2,
+      clientPublicKeyB64: Buffer.from(clientKeys.publicKey).toString('base64'),
+      clientNonceB64: Buffer.from(clientNonce).toString('base64'),
+      capabilities: { framing: [2], payloadKinds: ['text', 'binary'] },
+      context: {
+        protocol: 'orca-mobile-e2ee',
+        initiator: 'mobile',
+        responder: 'desktop',
+        transport: 'relay',
+        relayHostId
+      }
+    }
+  }
+
+  acceptReady(value: unknown): boolean {
+    const handshake = validateMobileE2EEV2Handshake(this.hello, value)
+    if (!handshake || !nacl.verify(handshake.desktopPublicKey, this.desktopPublicKey)) {
+      return false
+    }
+    this.schedule = deriveMobileE2EEV2KeySchedule({
+      sharedSecret: deriveSharedKey(this.clientKeys.secretKey, this.desktopPublicKey),
+      transcript: encodeMobileE2EEV2Transcript(handshake),
+      clientNonce: handshake.clientNonce,
+      desktopNonce: handshake.desktopNonce
+    })
+    return true
+  }
+
+  get transcriptHashB64(): string {
+    return Buffer.from(this.requireSchedule().transcriptHash).toString('base64')
+  }
+
+  sealText(plaintext: string): string {
+    return Buffer.from(this.seal(new TextEncoder().encode(plaintext), 'text')).toString('base64')
+  }
+
+  sealBinary(plaintext: Uint8Array): Uint8Array {
+    return this.seal(plaintext, 'binary')
+  }
+
+  openText(frameB64: string): string | null {
+    const plaintext = this.open(Buffer.from(frameB64, 'base64'), 'text')
+    return plaintext ? new TextDecoder().decode(plaintext) : null
+  }
+
+  openBinary(frame: Uint8Array): Uint8Array | null {
+    return this.open(frame, 'binary')
+  }
+
+  private seal(payload: Uint8Array, payloadKind: MobileE2EEPayloadKind): Uint8Array {
+    const schedule = this.requireSchedule()
+    const frame = sealMobileE2EEV2Frame({
+      payload,
+      key: schedule.mobileToDesktopKey,
+      sessionId: schedule.sessionId,
+      direction: 'mobile-to-desktop',
+      payloadKind,
+      counter: this.outboundCounter
+    })
+    this.outboundCounter++
+    return frame
+  }
+
+  private open(frame: Uint8Array, payloadKind: MobileE2EEPayloadKind): Uint8Array | null {
+    const schedule = this.requireSchedule()
+    const plaintext = openMobileE2EEV2Frame({
+      frame,
+      key: schedule.desktopToMobileKey,
+      sessionId: schedule.sessionId,
+      direction: 'desktop-to-mobile',
+      payloadKind,
+      expectedCounter: this.inboundCounter
+    })
+    if (plaintext) {
+      this.inboundCounter++
+    }
+    return plaintext
+  }
+
+  private requireSchedule(): ReturnType<typeof deriveMobileE2EEV2KeySchedule> {
+    if (!this.schedule) {
+      throw new Error('Simulated mobile peer has not accepted E2EE ready')
+    }
+    return this.schedule
+  }
+}