perf(orchestration): route federated reads over shared control (#13814)

src/main/ipc/runtime-environment-federated-read-routing.test.ts

@@ -0,0 +1,173 @@
+import { mkdtempSync, rmSync } from 'node:fs'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import { encodePairingOffer } from '../../shared/pairing'
+import { REMOTE_RUNTIME_SHARED_CONTROL_CAPABILITY } from '../../shared/protocol-version'
+import { addEnvironmentFromPairingCode } from '../../shared/runtime-environment-store'
+
+const { sendRemoteRuntimeRequestMock, sendRemoteRuntimeSharedControlRequestMock } = vi.hoisted(
+  () => ({
+    sendRemoteRuntimeRequestMock: vi.fn(),
+    sendRemoteRuntimeSharedControlRequestMock: vi.fn()
+  })
+)
+
+vi.mock('../../shared/remote-runtime-client', () => ({
+  sendRemoteRuntimeRequest: sendRemoteRuntimeRequestMock
+}))
+
+vi.mock('./runtime-environment-request-connections', () => ({
+  sendRemoteRuntimeConnectionRequest: vi.fn(),
+  sendRemoteRuntimeSharedControlRequest: sendRemoteRuntimeSharedControlRequestMock,
+  reconnectRemoteRuntimeSharedControlConnection: vi.fn()
+}))
+
+import {
+  callRuntimeEnvironment,
+  resetSharedControlSupport
+} from './runtime-environment-transport-routing'
+
+describe('federated read RPC transport routing', () => {
+  let userDataPath: string
+  let environmentId: string
+
+  beforeEach(() => {
+    userDataPath = mkdtempSync(join(tmpdir(), 'orca-federated-read-routing-'))
+    environmentId = addEnvironmentFromPairingCode(userDataPath, {
+      name: 'worker',
+      pairingCode: encodePairingOffer({
+        v: 2,
+        endpoint: 'ws://127.0.0.1:6768',
+        deviceToken: 'device-token',
+        publicKeyB64: Buffer.from(new Uint8Array(32).fill(1)).toString('base64')
+      })
+    }).id
+    resetSharedControlSupport()
+    sendRemoteRuntimeRequestMock.mockReset()
+    sendRemoteRuntimeSharedControlRequestMock.mockReset()
+  })
+
+  afterEach(() => {
+    rmSync(userDataPath, { recursive: true, force: true })
+  })
+
+  it('routes an enveloped federation read over shared control with its authority intact', async () => {
+    const envelope = {
+      orchestrationCapability: 'capability',
+      orchestrationContractVersion: 1,
+      orchestrationRequestId: 'request-1',
+      compatibilityInvocationId: 'compatibility-1',
+      orchestrationCompatibilityEvidence: {
+        terminalHandle: 'term-1',
+        paneKey: 'pane-1',
+        launchToken: 'launch-1'
+      }
+    }
+    sendRemoteRuntimeRequestMock.mockResolvedValue({
+      id: 'status',
+      ok: true,
+      result: { capabilities: [REMOTE_RUNTIME_SHARED_CONTROL_CAPABILITY] },
+      _meta: { runtimeId: 'runtime-worker' }
+    })
+    sendRemoteRuntimeSharedControlRequestMock.mockResolvedValue({
+      id: 'pull',
+      ok: true,
+      result: { runtimeEpoch: 'runtime-worker', items: [] },
+      _meta: { runtimeId: 'runtime-worker' }
+    })
+
+    await expect(
+      callRuntimeEnvironment(
+        userDataPath,
+        environmentId,
+        'orchestration.federationPull',
+        { dispatchId: 'dispatch-1', afterSequence: 0, limit: 50 },
+        15_000,
+        undefined,
+        envelope
+      )
+    ).resolves.toMatchObject({ ok: true })
+
+    expect(sendRemoteRuntimeRequestMock.mock.calls.map((call) => call[1])).toEqual(['status.get'])
+    expect(sendRemoteRuntimeSharedControlRequestMock).toHaveBeenCalledWith(
+      environmentId,
+      expect.any(Object),
+      'orchestration.federationPull',
+      { dispatchId: 'dispatch-1', afterSequence: 0, limit: 50 },
+      15_000,
+      envelope
+    )
+  })
+
+  it('keeps enveloped federation mutations on the one-shot transport', async () => {
+    const envelope = { orchestrationContractVersion: 1, orchestrationRequestId: 'ack-1' }
+    sendRemoteRuntimeRequestMock.mockResolvedValue({
+      id: 'ack',
+      ok: true,
+      result: { acknowledgedThrough: 4 },
+      _meta: { runtimeId: 'runtime-worker' }
+    })
+
+    await callRuntimeEnvironment(
+      userDataPath,
+      environmentId,
+      'orchestration.federationAck',
+      { dispatchId: 'dispatch-1', throughSequence: 4 },
+      15_000,
+      undefined,
+      envelope
+    )
+
+    expect(sendRemoteRuntimeRequestMock).toHaveBeenCalledWith(
+      expect.any(Object),
+      'orchestration.federationAck',
+      { dispatchId: 'dispatch-1', throughSequence: 4 },
+      15_000,
+      envelope
+    )
+    expect(sendRemoteRuntimeSharedControlRequestMock).not.toHaveBeenCalled()
+  })
+
+  it('falls back to one-shot with the envelope when shared control is unavailable', async () => {
+    const envelope = { orchestrationContractVersion: 1 }
+    sendRemoteRuntimeRequestMock.mockImplementation(async (_pairing, method) =>
+      method === 'status.get'
+        ? {
+            id: 'status',
+            ok: true,
+            result: { capabilities: [] },
+            _meta: { runtimeId: 'runtime-worker' }
+          }
+        : {
+            id: 'pull',
+            ok: true,
+            result: { runtimeEpoch: 'runtime-worker', items: [] },
+            _meta: { runtimeId: 'runtime-worker' }
+          }
+    )
+
+    await callRuntimeEnvironment(
+      userDataPath,
+      environmentId,
+      'orchestration.federationPull',
+      { dispatchId: 'dispatch-1', afterSequence: 0, limit: 50 },
+      15_000,
+      undefined,
+      envelope
+    )
+
+    expect(sendRemoteRuntimeRequestMock.mock.calls.map((call) => call[1])).toEqual([
+      'status.get',
+      'orchestration.federationPull'
+    ])
+    expect(sendRemoteRuntimeRequestMock).toHaveBeenLastCalledWith(
+      expect.any(Object),
+      'orchestration.federationPull',
+      { dispatchId: 'dispatch-1', afterSequence: 0, limit: 50 },
+      15_000,
+      envelope
+    )
+    expect(sendRemoteRuntimeSharedControlRequestMock).not.toHaveBeenCalled()
+  })
+})

src/main/ipc/runtime-environment-federated-read-transport.bench.test.ts

@@ -0,0 +1,116 @@
+import { readFileSync } from 'node:fs'
+import { performance } from 'node:perf_hooks'
+import { join } from 'node:path'
+import { describe, expect, it } from 'vitest'
+import { sendRemoteRuntimeRequest } from '../../shared/remote-runtime-client'
+import { RemoteRuntimeSharedControlConnection } from '../../shared/remote-runtime-shared-control-connection'
+import { RuntimeEnvironmentStoreSchema } from '../../shared/runtime-environments'
+
+const runLiveBenchmark = process.env.ORCA_FEDERATED_READ_BENCH === '1'
+
+it('interpolates even-sized benchmark samples', () => expect(percentile([1, 3], 0.5)).toBe(2))
+
+describe.runIf(runLiveBenchmark)('federated read RPC transport benchmark', () => {
+  it('compares one-shot and shared-control latency on one saved runtime', async () => {
+    const userDataPath = process.env.ORCA_RUNTIME_USER_DATA_PATH
+    const environmentName = process.env.ORCA_RUNTIME_ENVIRONMENT
+    if (!userDataPath || !environmentName) {
+      throw new Error('Set ORCA_RUNTIME_USER_DATA_PATH and ORCA_RUNTIME_ENVIRONMENT.')
+    }
+    const store = RuntimeEnvironmentStoreSchema.parse(
+      JSON.parse(readFileSync(join(userDataPath, 'orca-environments.json'), 'utf8'))
+    )
+    const environment = store.environments.find((entry) => entry.name === environmentName)
+    if (!environment) {
+      throw new Error(`Unknown runtime environment: ${environmentName}`)
+    }
+    const endpoint =
+      environment.endpoints.find((entry) => entry.id === environment.preferredEndpointId) ??
+      environment.endpoints[0]
+    if (!endpoint) {
+      throw new Error(`Runtime environment ${environmentName} has no endpoint.`)
+    }
+    const pairing = {
+      v: 2 as const,
+      endpoint: endpoint.endpoint,
+      deviceToken: endpoint.deviceToken,
+      publicKeyB64: endpoint.publicKeyB64
+    }
+    const params = {
+      dispatchId: 'sta3880_benchmark_missing',
+      afterSequence: 0,
+      limit: 50
+    }
+    const oneShot = await measureRequests(() =>
+      sendRemoteRuntimeRequest(pairing, 'orchestration.federationPull', params, 15_000, {
+        orchestrationContractVersion: 1
+      })
+    )
+    const shared = new RemoteRuntimeSharedControlConnection(pairing, {
+      environmentId: environment.id
+    })
+    const sharedControl = await measureRequests(() =>
+      shared.request('orchestration.federationPull', params, 15_000, {
+        orchestrationContractVersion: 1
+      })
+    )
+    const diagnostics = shared.getDiagnostics()
+    shared.close()
+
+    expect(diagnostics).toMatchObject({ state: 'ready', pendingRequestCount: 0 })
+    process.stdout.write(`${JSON.stringify({ oneShot, sharedControl }, null, 2)}\n`)
+  }, 120_000)
+})
+
+async function measureRequests(
+  request: () => Promise<{ ok: boolean; error?: { code: string } }>,
+  count = 30
+): Promise<{
+  count: number
+  meanMs: number
+  medianMs: number
+  p95Ms: number
+  minMs: number
+  maxMs: number
+  responseCodes: string[]
+}> {
+  for (let index = 0; index < 3; index++) {
+    await request()
+  }
+  const durations: number[] = []
+  const responseCodes = new Set<string>()
+  for (let index = 0; index < count; index++) {
+    const startedAt = performance.now()
+    const response = await request()
+    durations.push(performance.now() - startedAt)
+    if (!response.ok && response.error) {
+      responseCodes.add(response.error.code)
+    }
+  }
+  durations.sort((left, right) => left - right)
+  return {
+    count,
+    meanMs: round(durations.reduce((sum, value) => sum + value, 0) / count),
+    medianMs: round(percentile(durations, 0.5)),
+    p95Ms: round(percentile(durations, 0.95)),
+    minMs: round(durations[0] ?? 0),
+    maxMs: round(durations.at(-1) ?? 0),
+    responseCodes: [...responseCodes]
+  }
+}
+
+function percentile(sorted: number[], fraction: number): number {
+  if (sorted.length === 0) {
+    return 0
+  }
+  const rank = (sorted.length - 1) * fraction
+  const lowerIndex = Math.floor(rank)
+  const upperIndex = Math.ceil(rank)
+  const lower = sorted[lowerIndex] ?? 0
+  const upper = sorted[upperIndex] ?? lower
+  return lower + (upper - lower) * (rank - lowerIndex)
+}
+
+function round(value: number): number {
+  return Math.round(value * 100) / 100
+}

src/main/ipc/runtime-environment-request-connections.ts

@@ -1,5 +1,8 @@
 import type { PairingOffer } from '../../shared/pairing'
-import type { RuntimeRpcResponse } from '../../shared/runtime-rpc-envelope'
+import type {
+  RuntimeOrchestrationEnvelope,
+  RuntimeRpcResponse
+} from '../../shared/runtime-rpc-envelope'
 import { RemoteRuntimeRequestConnection } from '../../shared/remote-runtime-request-connection'
 import { RemoteRuntimeSharedControlConnection } from '../../shared/remote-runtime-shared-control-connection'
 import type {
@@ -52,9 +55,15 @@ export function sendRemoteRuntimeSharedControlRequest<TResult>(
   pairing: PairingOffer,
   method: string,
   params: unknown,
-  timeoutMs: number
+  timeoutMs: number,
+  envelope?: RuntimeOrchestrationEnvelope
 ): Promise<RuntimeRpcResponse<TResult>> {
-  return getSharedControlConnection(environmentId, pairing).request(method, params, timeoutMs)
+  return getSharedControlConnection(environmentId, pairing).request(
+    method,
+    params,
+    timeoutMs,
+    envelope
+  )
 }
 
 export function subscribeRemoteRuntimeSharedControlRequest<TResult>(

src/main/ipc/runtime-environment-transport-routing.ts

@@ -1,5 +1,6 @@
 import { getPreferredPairingOffer } from '../../shared/runtime-environments'
 import { resolveEnvironment, markEnvironmentUsed } from '../../shared/runtime-environment-store'
+import { isOrchestrationMutation } from '../../shared/orchestration-rpc-contract'
 import type {
   RuntimeOrchestrationEnvelope,
   RuntimeRpcResponse
@@ -108,7 +109,8 @@ export async function callRuntimeEnvironment(
       const pairing = getPreferredPairingOffer(currentEnvironment)
       endpoint = pairing.endpoint
       const effectiveTimeoutMs = timeoutMs ?? DEFAULT_REMOTE_RUNTIME_TIMEOUT_MS
-      if (envelope) {
+      const sharedControlEnvelope = shouldUseSharedControlEnvelope(method, params, envelope)
+      if (envelope && !sharedControlEnvelope) {
         const response = await sendRemoteRuntimeRequest(
           pairing,
           method,
@@ -135,19 +137,36 @@ export async function callRuntimeEnvironment(
         !shouldUseOneShotRequest(method) &&
         (await supportsSharedControl(userDataPath, currentEnvironment, pairing, effectiveTimeoutMs))
       ) {
-        const response = await sendRemoteRuntimeSharedControlRequest(
-          currentEnvironment.id,
-          pairing,
-          method,
-          params,
-          effectiveTimeoutMs
-        )
+        const response = sharedControlEnvelope
+          ? await sendRemoteRuntimeSharedControlRequest(
+              currentEnvironment.id,
+              pairing,
+              method,
+              params,
+              effectiveTimeoutMs,
+              sharedControlEnvelope
+            )
+          : await sendRemoteRuntimeSharedControlRequest(
+              currentEnvironment.id,
+              pairing,
+              method,
+              params,
+              effectiveTimeoutMs
+            )
         markEnvironmentUsedFromResponse(userDataPath, currentEnvironment.id, response)
         return response
       }
       // Why: startup/control-plane RPCs use the proven one-shot path so repo
       // hydration cannot be coupled to a stale terminal-control connection.
-      const response = await sendRemoteRuntimeRequest(pairing, method, params, effectiveTimeoutMs)
+      const response = sharedControlEnvelope
+        ? await sendRemoteRuntimeRequest(
+            pairing,
+            method,
+            params,
+            effectiveTimeoutMs,
+            sharedControlEnvelope
+          )
+        : await sendRemoteRuntimeRequest(pairing, method, params, effectiveTimeoutMs)
       markEnvironmentUsedFromResponse(userDataPath, currentEnvironment.id, response)
       return response
     })
@@ -253,6 +272,16 @@ function shouldUseCachedRequestConnection(method: string): boolean {
   return method === 'terminal.send' || method === 'terminal.updateViewport'
 }
 
+function shouldUseSharedControlEnvelope(
+  method: string,
+  params: unknown,
+  envelope: RuntimeOrchestrationEnvelope | undefined
+): RuntimeOrchestrationEnvelope | undefined {
+  return envelope && method.startsWith('orchestration.') && !isOrchestrationMutation(method, params)
+    ? envelope
+    : undefined
+}
+
 function shouldUseOneShotRequest(method: string): boolean {
   // Why: snapshot recovery must remain available while a retained shared-control stream is reconnecting after a HUB restart.
   return method === 'session.tabs.list' || method === 'session.tabs.listAll'

src/shared/remote-runtime-client.ts

@@ -144,16 +144,12 @@ async function sendRemoteRuntimeRequestOnSocket<TResult>(
   })
   const pendingRequest = {
     preparedRequest: prepareRemoteRuntimeRequest(new Map(), () =>
-      serializeRemoteRuntimePayload({
-        id: requestId,
+      serializeRemoteRuntimeRpcRequest({
+        requestId,
         deviceToken: pairing.deviceToken,
         method,
         params,
-        orchestrationCapability: envelope?.orchestrationCapability,
-        orchestrationContractVersion: envelope?.orchestrationContractVersion,
-        orchestrationRequestId: envelope?.orchestrationRequestId,
-        compatibilityInvocationId: envelope?.compatibilityInvocationId,
-        orchestrationCompatibilityEvidence: envelope?.orchestrationCompatibilityEvidence
+        envelope
       })
     )
   }

src/shared/remote-runtime-memory-limits.ts

@@ -3,6 +3,7 @@ import {
   stringifyJsonWithinByteLimit
 } from './node-bounded-json-stringify'
 import { RemoteRuntimeClientError } from './remote-runtime-client-error'
+import type { RuntimeOrchestrationEnvelope } from './runtime-rpc-envelope'
 
 export const REMOTE_RUNTIME_MAX_OUTBOUND_JSON_BYTES = 4 * 1024 * 1024
 export const REMOTE_RUNTIME_MAX_WEBSOCKET_FRAME_BYTES = 8 * 1024 * 1024 + 64
@@ -63,12 +64,18 @@ export function serializeRemoteRuntimeRpcRequest(args: {
   deviceToken: string
   method: string
   params: unknown
+  envelope?: RuntimeOrchestrationEnvelope
 }): string {
   return serializeRemoteRuntimePayload({
     id: args.requestId,
     deviceToken: args.deviceToken,
     method: args.method,
-    params: args.params
+    params: args.params,
+    orchestrationCapability: args.envelope?.orchestrationCapability,
+    orchestrationContractVersion: args.envelope?.orchestrationContractVersion,
+    orchestrationRequestId: args.envelope?.orchestrationRequestId,
+    compatibilityInvocationId: args.envelope?.compatibilityInvocationId,
+    orchestrationCompatibilityEvidence: args.envelope?.orchestrationCompatibilityEvidence
   })
 }
 

src/shared/remote-runtime-shared-control-connection.test.ts

@@ -78,6 +78,37 @@ describe('RemoteRuntimeSharedControlConnection', () => {
     connection.close()
   })
 
+  it('preserves orchestration authority fields on shared-control requests', async () => {
+    const server = await createServer()
+    const connection = new RemoteRuntimeSharedControlConnection(server.pairing)
+    const envelope = {
+      orchestrationCapability: 'capability',
+      orchestrationContractVersion: 1,
+      orchestrationRequestId: 'request-1',
+      compatibilityInvocationId: 'compatibility-1',
+      orchestrationCompatibilityEvidence: {
+        terminalHandle: 'term-1',
+        paneKey: 'pane-1',
+        launchToken: 'launch-1'
+      },
+      id: 'forged-id',
+      deviceToken: 'forged-token',
+      method: 'orchestration.federationAck',
+      params: { dispatchId: 'forged-dispatch' }
+    }
+
+    await connection.request('orchestration.federationPull', {}, 1000, envelope)
+
+    expect(server.requests).toContainEqual({
+      ...envelope,
+      id: expect.any(String),
+      deviceToken: 'device-token',
+      method: 'orchestration.federationPull',
+      params: {}
+    })
+    connection.close()
+  })
+
   it('does not expose a binary sender on the shared control protocol surface', () => {
     expect('sendSharedControlEncryptedBinary' in sharedControlProtocol).toBe(false)
   })

src/shared/remote-runtime-shared-control-connection.ts

@@ -1,6 +1,5 @@
 import WebSocket from 'ws'
 import type { PairingOffer } from './pairing'
-import type { RuntimeRpcResponse } from './runtime-rpc-envelope'
 import type { RemoteRuntimeClientError } from './remote-runtime-client-error'
 import { remoteRuntimeUnavailableError } from './remote-runtime-request-frames'
 import { openSharedControlSocket } from './remote-runtime-shared-control-open'
@@ -66,14 +65,16 @@ export class RemoteRuntimeSharedControlConnection {
   request<TResult>(
     method: string,
     params: unknown,
-    timeoutMs: number
-  ): Promise<RuntimeRpcResponse<TResult>> {
-    return requestSharedControl({
+    timeoutMs: number,
+    envelope?: Parameters<typeof requestSharedControl>[0]['envelope']
+  ): ReturnType<typeof requestSharedControl<TResult>> {
+    return requestSharedControl<TResult>({
       pendingRequests: this.pendingRequests,
       deviceToken: this.pairing.deviceToken,
       method,
       params,
       timeoutMs,
+      envelope,
       ensureReady: () => this.ensureReadyWithTimeout(timeoutMs),
       send: (requestId) => this.sendRequest(requestId),
       retireRequestId: (requestId) => this.retiredRequestIds.retire(requestId)

src/shared/remote-runtime-shared-control-requests.ts

@@ -6,7 +6,7 @@ import {
   type RemoteRuntimePreparedRequest
 } from './remote-runtime-prepared-request-admission'
 import { remoteRuntimeTimeoutError } from './remote-runtime-request-frames'
-import type { RuntimeRpcResponse } from './runtime-rpc-envelope'
+import type { RuntimeOrchestrationEnvelope, RuntimeRpcResponse } from './runtime-rpc-envelope'
 import { toRemoteRuntimeClientError } from './remote-runtime-shared-control-protocol'
 import { rejectSharedControlPendingRequest } from './remote-runtime-shared-control-state'
 import type { SharedControlPendingRequest } from './remote-runtime-shared-control-types'
@@ -19,6 +19,7 @@ export function requestSharedControl<TResult>(args: {
   method: string
   params: unknown
   timeoutMs: number
+  envelope?: RuntimeOrchestrationEnvelope
   ensureReady: () => Promise<void>
   send: (requestId: string) => void
   retireRequestId?: (requestId: string) => void
@@ -35,7 +36,8 @@ export function requestSharedControl<TResult>(args: {
         requestId,
         deviceToken: args.deviceToken,
         method: args.method,
-        params: args.params
+        params: args.params,
+        envelope: args.envelope
       })
     )
   } catch (error) {