orca/src/main/runtime/runtime-rpc.ts

/* eslint-disable max-lines -- Why: this file is the single security boundary for the bundled CLI — transport setup, auth-token enforcement, admission control, keepalive framing, and orphan-socket sweeping all co-locate deliberately so a reviewer can audit the boundary in one sitting. Splitting this across files would scatter the invariants without reducing complexity. */
// Why: this is the single security boundary for the bundled CLI. It owns
// auth-token enforcement, bootstrap-metadata publication, and transport
// orchestration so a running runtime is always discoverable via exactly
// one on-disk file. Method handling lives in `rpc/` and transport specifics
// live in `rpc/unix-socket-transport.ts` and `rpc/ws-transport.ts`.
import { randomBytes } from 'node:crypto'
import { readdirSync, rmSync } from 'node:fs'
import { join } from 'node:path'
import type { RuntimeMetadata, RuntimeTransportMetadata } from '../../shared/runtime-bootstrap'
import type { OrcaRuntimeService } from './orca-runtime'
import { writeRuntimeMetadata } from './runtime-metadata'
import { RpcDispatcher } from './rpc/dispatcher'
import type { RpcRequest, RpcResponse } from './rpc/core'
import { errorResponse } from './rpc/errors'
import type { RpcMessageContext, RpcTransport } from './rpc/transport'
import { UnixSocketTransport } from './rpc/unix-socket-transport'
import { WebSocketTransport } from './rpc/ws-transport'
import { readWsFallbackPort, writeWsFallbackPort } from './rpc/ws-fallback-port-store'
import type { WebSocket } from 'ws'
import { DeviceRegistry, type DeviceScope } from './device-registry'
import { loadOrCreateE2EEKeypair, type E2EEKeypair } from './e2ee-keypair'
import {
  MobileSocketWiring,
  type AuthenticatedMobileSocket,
  type MobileSocketTransportMetadata
} from './rpc/mobile-socket-wiring'
import type { PairingRelay } from '../../shared/mobile-relay-pairing-offer'
import type { MobilePairingConnectionMode } from '../../shared/mobile-pairing-connection-mode'
import {
  RelayRevokeOutbox,
  type RelayDeviceBinding,
  type RelayRevokeOutboxItem
} from './relay/relay-revoke-outbox'
import type {
  DeviceCredentialInstalled,
  PairingGetEndpointsParams,
  PairingGetEndpointsResult,
  PairingProvisionRelayParams
} from '../../shared/mobile-relay-credential-contract'
import { encodePairingOffer, PAIRING_OFFER_VERSION } from '../../shared/pairing'
import {
  decodeTerminalStreamFrame,
  type TerminalStreamFrame
} from '../../shared/terminal-stream-protocol'

const DEFAULT_WS_PORT = 6768

type OrcaRuntimeRpcServerOptions = {
  runtime: OrcaRuntimeService
  userDataPath: string
  pid?: number
  platform?: NodeJS.Platform
  enableWebSocket?: boolean
  wsPort?: number
  // Why: true when the caller set an explicit port (e.g. `orca serve --port`).
  // Distinguishes that pin from the DEFAULT_WS_PORT default so transport bind
  // order can prefer the pin over a stale STA-1511 fallback (issue #8535).
  preferPinnedWsPort?: boolean
  webClientRoot?: string
  // Why: test-only overrides for the two time-bound constants below.
  // Production callers must not pass these — defaults are set by the design
  // doc (§3.1) and changing them in production would weaken the admission
  // fence or flood the socket with keepalive frames.
  keepaliveIntervalMs?: number
  longPollCap?: number
}

type MobileRelayPairingProvider = {
  createPairingRelay(
    relayDeviceId: string
  ): Promise<{ relay: PairingRelay; binding: RelayDeviceBinding }>
  onDeviceRevokeQueued(item: RelayRevokeOutboxItem): void
  onDemandStateChanged?(): void
  getEndpoints(
    context: MobilePairingConnectionContext,
    params: PairingGetEndpointsParams
  ): Promise<PairingGetEndpointsResult>
  provisionRelay(
    context: MobilePairingConnectionContext,
    params: PairingProvisionRelayParams
  ): Promise<DeviceCredentialInstalled>
}

export type MobilePairingConnectionContext = Readonly<{
  deviceId: string
  connectionId: string
  transport: MobileSocketTransportMetadata
}>

// Why: after 10 s of a pending dispatch we emit a tiny `{"_keepalive":true}`
// frame every 10 s until the handler resolves. Each write resets both the
// server's own socket idle timer (30 s) and — once §3.1 ships on the client —
// the client's idle timer, because any byte counts as socket activity. This
// is the transport-layer fix for feedback #1: long-poll RPCs (i.e.
// orchestration.check --wait) can now run past the 30 s/60 s idle caps
// without either end tearing the socket down. See design doc §3.1.
const KEEPALIVE_INTERVAL_MS = 10_000

// Why: long-poll slot cap. With keepalives a `check --wait --timeout-ms
// 600000` can hold a connection for up to 10 minutes; unbounded that would
// saturate MAX_RUNTIME_RPC_CONNECTIONS (32) with 32 waiting coordinators
// and lock out normal short RPCs. Capping at half the connection budget
// leaves the other half for short traffic. On overflow the server responds
// immediately with `runtime_busy` (CLI exit 75) — fail fast, not silent
// queuing. See design doc §3.1 + §7 risk #2.
const LONG_POLL_CAP = 16

function resolvePairingEndpoint(rawEndpoint: string, address: string | null | undefined): string {
  const endpoint = new URL(rawEndpoint)
  const override = address?.trim()
  if (!override) {
    endpoint.hostname = '127.0.0.1'
    return formatWebSocketUrl(endpoint)
  }
  if (/^wss?:\/\//i.test(override)) {
    return formatWebSocketUrl(new URL(override))
  }
  const parsed = parsePairingAddressOverride(override)
  endpoint.hostname = parsed.host.includes(':')
    ? `[${parsed.host.replace(/^\[|\]$/g, '')}]`
    : parsed.host
  if (parsed.port) {
    endpoint.port = parsed.port
  }
  return formatWebSocketUrl(endpoint)
}

function parsePairingAddressOverride(address: string): { host: string; port: string | null } {
  if (address.startsWith('[') || address.split(':').length === 2) {
    try {
      const parsed = new URL(`ws://${address}`)
      return { host: parsed.hostname.replace(/^\[|\]$/g, ''), port: parsed.port || null }
    } catch {
      return { host: address, port: null }
    }
  }
  return { host: address, port: null }
}

function formatWebSocketUrl(url: URL): string {
  const formatted = url.toString()
  return url.pathname === '/' && !url.search && !url.hash ? formatted.replace(/\/$/, '') : formatted
}

function createWebClientUrl(endpoint: string, pairingUrl: string): string {
  const url = new URL(endpoint)
  url.protocol = url.protocol === 'wss:' ? 'https:' : 'http:'
  url.pathname = webClientPathForEndpoint(url.pathname)
  url.search = ''
  // Why: pairing URLs include full runtime credentials. Keeping them in the
  // fragment avoids proxy logs and Referer headers while the web app loads.
  url.hash = `pairing=${encodeURIComponent(pairingUrl)}`
  return url.toString()
}

function webClientPathForEndpoint(pathname: string): string {
  if (!pathname || pathname === '/') {
    return '/web-index.html'
  }
  return `${pathname.replace(/\/$/, '')}/web-index.html`
}

const MOBILE_RPC_METHOD_ALLOWLIST = new Set([
  'accounts.list',
  'accounts.selectClaude',
  'accounts.selectCodex',
  'accounts.subscribe',
  'accounts.unsubscribe',
  'aiVault.listSessions',
  'browser.back',
  'browser.dialogAccept',
  'browser.dialogDismiss',
  'browser.forward',
  'browser.goto',
  'browser.keyboardInsertText',
  'browser.keypress',
  'browser.mouseDown',
  'browser.mouseClick',
  'browser.mouseMove',
  'browser.mouseUp',
  'browser.mouseWheel',
  'browser.reload',
  'browser.screencast',
  'browser.screencast.unsubscribe',
  'browser.tabCreate',
  'browser.viewport',
  'clipboard.abortImageUpload',
  'clipboard.appendImageUploadChunk',
  'clipboard.commitImageUpload',
  'clipboard.saveImageAsTempFile',
  'clipboard.startImageUpload',
  'diagnostics.memory',
  'files.browseServerDir',
  'files.createFile',
  'files.list',
  'files.open',
  'files.openDiff',
  'files.read',
  'files.readChunk',
  'files.readDir',
  'files.readPreview',
  'files.readTerminalArtifact',
  'files.readTerminalArtifactPreview',
  'files.resolveTerminalPath',
  'files.searchPaths',
  'files.writeTerminalArtifact',
  'folderWorkspace.list',
  'git.abortMerge',
  'git.abortRebase',
  'git.bulkStage',
  'git.bulkUnstage',
  'git.branchCompare',
  'git.branchDiff',
  'git.cancelGenerateCommitMessage',
  'git.cancelGeneratePullRequestFields',
  'git.checkout',
  'git.commit',
  'git.commitCompare',
  'git.commitDiff',
  'git.discard',
  'git.discoverCommitMessageModels',
  'git.diff',
  'git.fetch',
  'git.forkSync',
  'git.fastForward',
  'git.generateCommitMessage',
  'git.generatePullRequestFields',
  'git.history',
  'git.localBranches',
  'git.pull',
  'git.push',
  'git.rebaseFromBase',
  'git.stage',
  'git.status',
  'git.unstage',
  'git.upstreamStatus',
  'github.createIssue',
  'github.addIssueComment',
  'github.addPRReviewComment',
  'github.addPRReviewCommentReply',
  'github.countWorkItems',
  'github.listAssignableUsers',
  'github.listLabels',
  'github.listWorkItems',
  'github.mergePR',
  'github.setPRAutoMerge',
  'github.requestPRReviewers',
  'github.removePRReviewers',
  'github.project.listAccessible',
  'github.project.listAssignableUsersBySlug',
  'github.project.listIssueTypesBySlug',
  'github.project.listLabelsBySlug',
  'github.project.listViews',
  'github.project.resolveRef',
  'github.project.addIssueCommentBySlug',
  'github.project.updateIssueCommentBySlug',
  'github.project.deleteIssueCommentBySlug',
  'github.project.clearItemField',
  'github.project.updateIssueBySlug',
  'github.project.updateIssueTypeBySlug',
  'github.project.updateItemField',
  'github.project.updatePullRequestBySlug',
  'github.project.viewTable',
  'github.project.workItemDetailsBySlug',
  'github.prForBranch',
  'github.prFileContents',
  'github.prChecks',
  'github.prCheckDetails',
  'github.rerunPRChecks',
  'github.resolveReviewThread',
  'github.setPRFileViewed',
  'github.updateIssue',
  'github.updatePR',
  'github.updatePRTitle',
  'github.updatePRState',
  'github.repoSlug',
  'github.workItem',
  // Cross-repo GitHub work-item lookup: lets the mobile create-workspace Smart
  // picker resolve a pasted github.com URL that points at a different repo.
  'github.workItemByOwnerRepo',
  'github.workItemDetails',
  'gitlab.createIssue',
  'gitlab.addIssueComment',
  'gitlab.addMRComment',
  'gitlab.listWorkItems',
  // Mobile create-workspace Smart picker: resolve a pasted GitLab URL to an exact
  // issue/MR. (MR listing reuses gitlab.listWorkItems, which returns issues + MRs.)
  'gitlab.workItemByPath',
  'gitlab.mergeMR',
  'gitlab.resolveMRDiscussion',
  'gitlab.todos',
  'gitlab.updateIssue',
  'gitlab.updateMR',
  'gitlab.updateMRState',
  'gitlab.workItemDetails',
  'host.gitBash.isAvailable',
  'host.platform',
  'host.pwsh.isAvailable',
  'host.wsl.isAvailable',
  'host.wsl.listDistros',
  'hostedReview.create',
  'hostedReview.forBranch',
  'hostedReview.getCreationEligibility',
  'linear.getCustomView',
  'linear.getIssue',
  'linear.getProject',
  'linear.agentSearchIssues',
  'linear.issueContext',
  'linear.resolveCurrentIssue',
  'linear.addIssueComment',
  'linear.connect',
  'linear.createIssue',
  'linear.createProject',
  'linear.issueComments',
  'linear.listCustomViewIssues',
  'linear.listCustomViewProjects',
  'linear.listCustomViews',
  'linear.listIssues',
  'linear.listProjectIssues',
  'linear.listProjects',
  'linear.teamLabels',
  'linear.teamMembers',
  'linear.listTeams',
  'linear.searchIssues',
  'linear.selectWorkspace',
  'linear.status',
  'linear.teamStates',
  'linear.updateIssue',
  'markdown.readTab',
  'markdown.saveTab',
  'notifications.getMissedSince',
  'notifications.subscribe',
  'notifications.unsubscribe',
  'pairing.getEndpoints',
  'pairing.provisionRelay',
  'preflight.check',
  'preflight.detectAgents',
  'preflight.detectRemoteAgents',
  'projectGroup.list',
  'repo.baseRefDefault',
  'repo.gitAvailable',
  'repo.hooks',
  'repo.list',
  'repo.saveSparsePreset',
  'repo.searchRefs',
  'repo.sparsePresets',
  'repo.update',
  'runtime.clientEvents.subscribe',
  'runtime.clientEvents.unsubscribe',
  'session.tabs.activate',
  'session.tabs.close',
  'session.tabs.createTerminal',
  'session.tabs.list',
  'session.tabs.listAll',
  'session.tabs.move',
  'session.tabs.subscribe',
  'session.tabs.subscribeAll',
  'session.tabs.unsubscribe',
  'session.tabs.unsubscribeAll',
  'nativeChat.readSession',
  'nativeChat.subscribe',
  'nativeChat.unsubscribe',
  'settings.get',
  'settings.update',
  'ssh.connect',
  'ssh.getState',
  'ssh.listRemovedTargetLabels',
  'ssh.listTargets',
  'speech.dictation.cancel',
  'speech.dictation.chunk',
  'speech.dictation.finish',
  'speech.dictation.setup',
  'speech.dictation.start',
  'speech.models.delete',
  'speech.models.download',
  'speech.models.list',
  'stats.summary',
  'status.get',
  'agentTeams.prepareLaunch',
  'agentTeams.tmuxCompat',
  'terminal.clearBuffer',
  'terminal.close',
  'terminal.closeTab',
  'terminal.create',
  'terminal.focus',
  'terminal.agentStatus',
  'terminal.getAutoRestoreFit',
  'terminal.isRunningAgent',
  'terminal.list',
  'terminal.multiplex',
  'terminal.read',
  'terminal.rename',
  'terminal.send',
  'terminal.setAutoRestoreFit',
  'terminal.setDisplayMode',
  'terminal.subscribe',
  'terminal.unsubscribe',
  'terminal.updateViewport',
  'terminal.wait',
  'ui.get',
  'ui.recordFeatureInteraction',
  'ui.set',
  'worktree.activate',
  'worktree.create',
  'worktree.forceDeleteBranch',
  'worktree.prefetchCreateBase',
  'worktree.ps',
  'worktree.show',
  'worktree.resolveMrBase',
  'worktree.resolvePrBase',
  'worktree.rm',
  'worktree.set',
  'worktree.sleep'
])

// Why: a long-poll request is one whose handler blocks waiting for an external
// event. This function is the single place that classifies it — the long-poll
// counter, abort wiring, keepalives, and runtime_busy admission check all
// share this decision. See §3.1.
function isLongPollRequest(request: RpcRequest): boolean {
  if (request.method === 'terminal.wait') {
    return true
  }
  if (request.method === 'orchestration.check') {
    const params = request.params as { wait?: unknown } | undefined
    return params?.wait === true
  }
  return false
}

// Why: stamp the authenticated connection's scope onto the status.get success
// envelope. status.get has no per-connection context inside the dispatcher, so
// the scope is added here at the transport boundary where the device is known.
// Failures fall back to the untouched reply rather than dropping the response.
function injectDeviceScope(response: string, scope: DeviceScope): string {
  try {
    const parsed = JSON.parse(response) as RpcResponse
    if (parsed.ok !== true || typeof parsed.result !== 'object' || parsed.result === null) {
      return response
    }
    ;(parsed.result as Record<string, unknown>).deviceScope = scope
    return JSON.stringify(parsed)
  } catch {
    return response
  }
}

export class OrcaRuntimeRpcServer {
  private readonly runtime: OrcaRuntimeService
  private readonly dispatcher: RpcDispatcher
  private readonly userDataPath: string
  private readonly pid: number
  private readonly platform: NodeJS.Platform
  private readonly enableWebSocket: boolean
  private readonly wsPort: number
  private readonly preferPinnedWsPort: boolean
  private readonly webClientRoot: string | undefined
  private readonly authToken = randomBytes(24).toString('hex')
  private readonly keepaliveIntervalMs: number
  private readonly longPollCap: number
  private readonly relayRevokeOutbox: RelayRevokeOutbox
  private deviceRegistry: DeviceRegistry | null = null
  private e2eeKeypair: E2EEKeypair | null = null
  private tlsFingerprint: string | null = null
  private activeTransports: RpcTransport[] = []
  private transports: RuntimeTransportMetadata[] = []
  private mobileSocketWiring: MobileSocketWiring | null = null
  private mobileRelayPairingProvider: MobileRelayPairingProvider | null = null
  private readonly binaryStreamHandlers = new Map<
    string,
    Map<number, (frame: TerminalStreamFrame) => void>
  >()
  private readonly wsDispatchAbortStates = new Map<
    WebSocket,
    { controllers: Set<AbortController>; abortOnClose: () => void }
  >()
  // Why: separate from Node's server.maxConnections because we need to count
  // only long-running dispatches, not every in-flight short RPC. See §3.1 +
  // §7 risk #2.
  private activeLongPolls = 0

  constructor({
    runtime,
    userDataPath,
    pid = process.pid,
    platform = process.platform,
    enableWebSocket = false,
    wsPort = DEFAULT_WS_PORT,
    preferPinnedWsPort = false,
    webClientRoot,
    keepaliveIntervalMs = KEEPALIVE_INTERVAL_MS,
    longPollCap = LONG_POLL_CAP
  }: OrcaRuntimeRpcServerOptions) {
    this.runtime = runtime
    this.dispatcher = new RpcDispatcher({ runtime })
    this.userDataPath = userDataPath
    this.pid = pid
    this.platform = platform
    this.enableWebSocket = enableWebSocket
    this.wsPort = wsPort
    this.preferPinnedWsPort = preferPinnedWsPort
    this.webClientRoot = webClientRoot
    this.keepaliveIntervalMs = keepaliveIntervalMs
    this.longPollCap = longPollCap
    this.relayRevokeOutbox = new RelayRevokeOutbox(userDataPath)
  }

  getDeviceRegistry(): DeviceRegistry | null {
    return this.deviceRegistry
  }

  getTlsFingerprint(): string | null {
    return this.tlsFingerprint
  }

  getE2EEPublicKey(): string | null {
    return this.e2eeKeypair?.publicKeyB64 ?? null
  }

  getE2EEKeypair(): E2EEKeypair | null {
    return this.e2eeKeypair
  }

  getMobileSocketWiring(): MobileSocketWiring | null {
    return this.mobileSocketWiring
  }

  getRelayRevokeOutbox(): RelayRevokeOutbox {
    return this.relayRevokeOutbox
  }

  setMobileRelayBinding(deviceId: string, binding: RelayDeviceBinding): boolean {
    const current = this.deviceRegistry?.getDevice(deviceId)
    if (
      current?.scope !== 'mobile' ||
      this.deviceRegistry?.getMobilePairingConnectionMode(deviceId) === 'local-only'
    ) {
      return false
    }
    if (
      current.relayBinding &&
      (current.relayBinding.relayHostId !== binding.relayHostId ||
        current.relayBinding.ownerIdentityKey !== binding.ownerIdentityKey)
    ) {
      // Why: switching the account/host that owns this local pairing cannot
      // strand the old cloud credential family even if that account is offline.
      this.queueRelayDeviceRevoke(current.relayBinding)
    }
    const updated = this.deviceRegistry?.setRelayBinding(deviceId, binding) ?? false
    if (updated) {
      this.mobileRelayPairingProvider?.onDemandStateChanged?.()
    }
    return updated
  }

  setMobileRelayPairingProvider(provider: MobileRelayPairingProvider | null): void {
    this.mobileRelayPairingProvider = provider
  }

  async revokeMobileDevice(deviceId: string): Promise<boolean> {
    const device = this.deviceRegistry?.getDevice(deviceId)
    if (device?.scope !== 'mobile') {
      return false
    }
    if (device.relayBinding) {
      this.queueRelayDeviceRevoke(device.relayBinding)
    }
    if (!this.deviceRegistry?.removeDevice(deviceId)) {
      return false
    }
    this.mobileRelayPairingProvider?.onDemandStateChanged?.()
    this.mobileSocketWiring?.terminateDeviceConnections(device.token)
    return true
  }

  revokeRuntimeAccess(deviceId: string): boolean {
    const device = this.deviceRegistry?.getDevice(deviceId)
    if (device?.scope !== 'runtime' || !this.deviceRegistry?.removeDevice(deviceId)) {
      return false
    }
    this.mobileSocketWiring?.terminateDeviceConnections(device.token)
    return true
  }

  getWebSocketEndpoint(): string | null {
    const ws = this.transports.find((t) => t.kind === 'websocket')
    return ws?.endpoint ?? null
  }

  createPairingOffer(args: {
    address?: string | null
    name?: string
    rotate?: boolean
    scope?: DeviceScope
  }):
    | { available: false }
    | {
        available: true
        pairingUrl: string
        endpoint: string
        deviceId: string
        webClientUrl: string | null
      } {
    const rawEndpoint = this.getWebSocketEndpoint()
    const publicKeyB64 = this.getE2EEPublicKey()
    if (!rawEndpoint || !this.deviceRegistry || !publicKeyB64) {
      return { available: false }
    }

    const endpoint = resolvePairingEndpoint(rawEndpoint, args.address)
    const deviceName = args.name ?? `CLI ${new Date().toLocaleDateString()}`
    const scope = args.scope ?? 'runtime'
    const device = args.rotate
      ? this.deviceRegistry.rotatePendingDevice(deviceName, scope)
      : this.deviceRegistry.getOrCreatePendingDevice(deviceName, scope)
    const pairingUrl = encodePairingOffer({
      v: PAIRING_OFFER_VERSION,
      endpoint,
      deviceToken: device.token,
      publicKeyB64,
      scope
    })
    return {
      available: true,
      pairingUrl,
      endpoint,
      deviceId: device.deviceId,
      webClientUrl:
        this.webClientRoot && scope === 'runtime' ? createWebClientUrl(endpoint, pairingUrl) : null
    }
  }

  async createMobilePairingOffer(args: {
    address?: string | null
    connectionMode?: MobilePairingConnectionMode
    name?: string
    rotate?: boolean
  }): Promise<ReturnType<OrcaRuntimeRpcServer['createPairingOffer']>> {
    // Why: the renderer is outside the trust boundary; only the explicit
    // local-only value may suppress Relay provisioning.
    const connectionMode = args.connectionMode === 'local-only' ? 'local-only' : 'automatic'
    const pending = this.deviceRegistry?.getPendingDevice('mobile')
    const switchingPendingToLocal =
      connectionMode === 'local-only' && pending?.relayBinding !== undefined
    if (args.rotate || switchingPendingToLocal) {
      if (pending?.relayBinding) {
        // Why: the durable cloud revoke is recorded before rotating the local
        // token, so a previously displayed relay invite cannot outlive the QR.
        this.queueRelayDeviceRevoke(pending.relayBinding)
      }
    }
    const direct = this.createPairingOffer({
      ...args,
      rotate: args.rotate || switchingPendingToLocal,
      scope: 'mobile'
    })
    if (!direct.available) {
      return direct
    }
    this.deviceRegistry?.setMobilePairingConnectionMode(direct.deviceId, connectionMode)
    if (connectionMode === 'local-only' || !this.mobileRelayPairingProvider) {
      return direct
    }
    const device = this.deviceRegistry?.getDevice(direct.deviceId)
    const publicKeyB64 = this.getE2EEPublicKey()
    if (!device || !publicKeyB64) {
      return direct
    }
    try {
      const relayPairing = await this.mobileRelayPairingProvider.createPairingRelay(device.deviceId)
      if (!this.deviceRegistry?.setRelayBinding(device.deviceId, relayPairing.binding)) {
        return direct
      }
      this.mobileRelayPairingProvider.onDemandStateChanged?.()
      return {
        ...direct,
        pairingUrl: encodePairingOffer({
          v: PAIRING_OFFER_VERSION,
          endpoint: direct.endpoint,
          deviceToken: device.token,
          publicKeyB64,
          scope: 'mobile',
          relay: relayPairing.relay
        })
      }
    } catch {
      // Why: relay is additive. A transient auth/director/control outage must
      // still yield the valid LAN/Tailscale pairing offer.
      return direct
    }
  }

  private queueRelayDeviceRevoke(binding: RelayDeviceBinding): void {
    const item = this.relayRevokeOutbox.enqueue(binding)
    this.mobileRelayPairingProvider?.onDeviceRevokeQueued(item)
  }

  private registerBinaryStreamHandler(
    connectionId: string | undefined,
    streamId: number,
    handler: (frame: TerminalStreamFrame) => void
  ): () => void {
    if (!connectionId || !Number.isInteger(streamId) || streamId < 0) {
      return () => {}
    }
    let handlers = this.binaryStreamHandlers.get(connectionId)
    if (!handlers) {
      handlers = new Map()
      this.binaryStreamHandlers.set(connectionId, handlers)
    }
    handlers.set(streamId, handler)
    return () => {
      const current = this.binaryStreamHandlers.get(connectionId)
      if (!current || current.get(streamId) !== handler) {
        return
      }
      current.delete(streamId)
      if (current.size === 0) {
        this.binaryStreamHandlers.delete(connectionId)
      }
    }
  }

  private handleWebSocketBinaryMessage(bytes: Uint8Array<ArrayBufferLike>, ws: WebSocket): void {
    const connectionId = this.mobileSocketWiring?.getConnectionId(ws)
    if (!connectionId) {
      return
    }
    const frame = decodeTerminalStreamFrame(bytes)
    if (!frame) {
      return
    }
    this.binaryStreamHandlers.get(connectionId)?.get(frame.streamId)?.(frame)
  }

  private registerWebSocketDispatchAbort(ws: WebSocket): {
    signal: AbortSignal
    dispose: () => void
  } {
    const abortController = new AbortController()
    if (ws.readyState !== ws.OPEN) {
      abortController.abort()
      return { signal: abortController.signal, dispose: () => {} }
    }

    let state = this.wsDispatchAbortStates.get(ws)
    if (!state) {
      state = {
        controllers: new Set(),
        abortOnClose: () => this.abortWebSocketDispatches(ws)
      }
      this.wsDispatchAbortStates.set(ws, state)
      // Why: many streaming RPCs can share one WebSocket. A single socket-level
      // abort fan-out avoids MaxListenersExceededWarning while preserving cleanup.
      ws.on('close', state.abortOnClose)
      ws.on('error', state.abortOnClose)
    }
    state.controllers.add(abortController)

    return {
      signal: abortController.signal,
      dispose: () => {
        const current = this.wsDispatchAbortStates.get(ws)
        if (!current) {
          return
        }
        current.controllers.delete(abortController)
        if (current.controllers.size > 0) {
          return
        }
        this.wsDispatchAbortStates.delete(ws)
        ws.off('close', current.abortOnClose)
        ws.off('error', current.abortOnClose)
      }
    }
  }

  private abortWebSocketDispatches(ws: WebSocket): void {
    const state = this.wsDispatchAbortStates.get(ws)
    if (!state) {
      return
    }
    this.wsDispatchAbortStates.delete(ws)
    ws.off('close', state.abortOnClose)
    ws.off('error', state.abortOnClose)
    for (const controller of state.controllers) {
      controller.abort()
    }
    state.controllers.clear()
  }

  async start(): Promise<void> {
    if (this.activeTransports.length > 0) {
      return
    }

    // Why: processes killed by SIGKILL / OOM-kill / forced-shutdown skip
    // stop() and leave behind `o-<pid>-*.sock` files in userData. Sweeping
    // dead-pid sockets at startup keeps the directory from accumulating
    // orphans over the app's lifetime. Named-pipe transports on Windows do
    // not leave filesystem entries in userData, so the sweep is a no-op
    // there.
    if (this.platform !== 'win32') {
      sweepOrphanedRuntimeSockets(this.userDataPath, this.pid)
    }

    const transportMeta = createRuntimeTransportMetadata(
      this.userDataPath,
      this.pid,
      this.platform,
      this.runtime.getRuntimeId()
    )

    const socketTransport = new UnixSocketTransport({
      endpoint: transportMeta.endpoint,
      kind: transportMeta.kind as 'unix' | 'named-pipe',
      keepaliveIntervalMs: this.keepaliveIntervalMs
    })

    // Why: Unix socket transport uses the shared runtime auth token. This is
    // the existing security model for CLI connections — the token lives in a
    // 0o600-permissioned file on disk.
    // Why: the `.catch` guarantees `reply()` always fires even if
    // `handleMessage` (or `JSON.stringify` on a pathological response) throws.
    // Without it, a throw would leave the client waiting for a terminal frame
    // that never arrives AND leak the dispatch's AbortController in the
    // transport's in-flight set until the 30 s socket idle timer closes the
    // connection.
    socketTransport.onMessage((msg, reply, context) => {
      void this.handleMessage(msg, context)
        .then((response) => {
          reply(JSON.stringify(response))
        })
        .catch((error) => {
          const message = error instanceof Error ? error.message : String(error)
          // Why: best-effort id recovery so the client can correlate the
          // error frame to its pending request. A malformed message would
          // have been caught by handleMessage and returned an envelope
          // instead of throwing, so in practice the id is always present.
          let id = 'unknown'
          try {
            const parsed = JSON.parse(msg) as { id?: unknown }
            if (typeof parsed.id === 'string' && parsed.id.length > 0) {
              id = parsed.id
            }
          } catch {
            // ignore — fall through with id='unknown'
          }
          reply(JSON.stringify(this.buildError(id, 'internal_error', message)))
        })
    })

    await socketTransport.start()

    const activeTransports: RpcTransport[] = [socketTransport]
    const transportsMeta: RuntimeTransportMetadata[] = [transportMeta]

    // Why: WebSocket transport is opt-in and starts alongside the Unix socket.
    // It uses per-device tokens and E2EE (application-layer encryption via
    // tweetnacl) rather than TLS, since React Native can't pin self-signed certs.
    if (this.enableWebSocket) {
      try {
        this.deviceRegistry = new DeviceRegistry(this.userDataPath)
        this.e2eeKeypair = loadOrCreateE2EEKeypair(this.userDataPath)

        const wsTransport = new WebSocketTransport({
          host: '0.0.0.0',
          port: this.wsPort,
          staticRoot: this.webClientRoot,
          // Why: keep the fallback port stable across restarts so paired
          // devices' stored endpoints stay valid (STA-1511) — the transport
          // binds a persisted fallback before the preferred port unless the
          // caller explicitly pinned a port (serve --port). wsPort 0 means
          // the caller wants a random port (E2E) — don't pin it.
          ...(this.wsPort !== 0 ? { fallbackPort: readWsFallbackPort(this.userDataPath) } : {}),
          ...(this.preferPinnedWsPort ? { preferPinnedPort: true } : {})
        })
        const mobileSocketWiring = new MobileSocketWiring({
          deviceRegistry: this.deviceRegistry,
          e2eeKeypair: this.e2eeKeypair,
          onText: (socket, plaintext, reply, sendBinary) => {
            void this.handleWebSocketMessage(
              plaintext,
              reply,
              sendBinary,
              undefined,
              socket.ws,
              socket.device.deviceToken,
              socket
            )
          },
          onBinary: (socket, bytes) => this.handleWebSocketBinaryMessage(bytes, socket.ws),
          onReady: () => {
            // Why: first authenticated mobile/remote client (direct WS and
            // cloud relay both attach here) starts path-candidate tracking.
            // Activation is a local-host concern: candidate buffers live on the
            // buffer-owning host's runtime, so a remote runtime proxy may
            // legitimately lack this method (its own server activates it).
            this.runtime.activateRecentPtyPathCandidateTracking?.()
            this.mobileRelayPairingProvider?.onDemandStateChanged?.()
          },
          onClose: (socket, hasOtherConnections) => {
            if (!socket) {
              return
            }
            this.abortWebSocketDispatches(socket.ws)
            // Why: subscriptions and binary streams are socket-scoped, while
            // client disconnect state is device-scoped across both transports.
            this.runtime.cleanupSubscriptionsForConnection(socket.connectionId)
            this.runtime.cancelMobileDictationForConnection(socket.connectionId)
            this.binaryStreamHandlers.delete(socket.connectionId)
            if (!hasOtherConnections) {
              this.runtime.onClientDisconnected(socket.device.deviceToken)
            }
          }
        })
        mobileSocketWiring.attachTransport(wsTransport)
        this.mobileSocketWiring = mobileSocketWiring

        await wsTransport.start()
        if (this.wsPort !== 0 && wsTransport.resolvedPort !== this.wsPort) {
          writeWsFallbackPort(this.userDataPath, wsTransport.resolvedPort)
        }
        activeTransports.push(wsTransport)
        transportsMeta.push({
          kind: 'websocket',
          endpoint: `ws://0.0.0.0:${wsTransport.resolvedPort}`
        })
      } catch (error) {
        // Why: WebSocket transport is supplementary — the runtime must still
        // function if it fails to start (e.g., port in use). Log and continue
        // with Unix socket only.
        console.error('[runtime] Failed to start WebSocket transport:', error)
        this.mobileSocketWiring = null
      }
    }

    // Why: publish the transport into in-memory state before writing metadata
    // so the bootstrap file always contains the real endpoint/token pair. The
    // CLI only discovers the runtime through that file.
    this.activeTransports = activeTransports
    this.transports = transportsMeta

    try {
      this.writeMetadata()
    } catch (error) {
      // Why: a runtime that cannot publish bootstrap metadata is invisible to
      // the `orca` CLI. Close all transports immediately instead of leaving
      // behind a live but undiscoverable control plane.
      this.activeTransports = []
      this.transports = []
      await Promise.all(activeTransports.map((t) => t.stop().catch(() => {}))).catch(() => {})
      throw error
    }
  }

  async stop(): Promise<void> {
    const transports = this.activeTransports
    this.activeTransports = []
    this.transports = []
    this.mobileSocketWiring = null
    if (transports.length === 0) {
      return
    }
    await Promise.all(transports.map((t) => t.stop()))
    // Why: we intentionally leave the last metadata file behind instead of
    // deleting it on shutdown. Shared userData paths can briefly host multiple
    // Orca processes during restarts, updates, or development, and stale
    // metadata is safer than letting one process erase another live runtime's
    // bootstrap file.
  }

  // Why: Unix socket messages use one-shot dispatch (single response per
  // request) and the shared runtime auth token from the 0o600 metadata file.
  // The transport layer owns socket lifecycle, keepalive writes, and the
  // per-connection abort signal — this method just parses, auths, and
  // dispatches. See design doc §3.1.
  private async handleMessage(
    rawMessage: string,
    context?: RpcMessageContext
  ): Promise<RpcResponse> {
    // Why: empty messages are sent by the Unix socket transport layer when a
    // client exceeds the max message size. The transport closes the connection
    // after this response.
    if (!rawMessage) {
      return this.buildError('unknown', 'request_too_large', 'RPC request exceeds the maximum size')
    }

    const parsed = this.parseAndAuth(rawMessage)
    if ('error' in parsed) {
      return parsed.error
    }
    const request = parsed.request

    // Why: long-poll admission fence. Short RPCs bypass the counter entirely
    // — it only guards handlers that can block for minutes. See §7 risk #2.
    const longPoll = isLongPollRequest(request)
    if (longPoll && this.activeLongPolls >= this.longPollCap) {
      return this.buildError(
        request.id,
        'runtime_busy',
        'long-poll capacity reached; retry with backoff'
      )
    }
    if (longPoll) {
      this.activeLongPolls += 1
      // Why: arm the keepalive timer only for long-polls. Short RPCs never
      // touch it so the `setInterval` is never created. See §3.1.
      context?.startKeepalive()
    }

    try {
      return await this.dispatcher.dispatch(request, {
        signal: longPoll ? context?.signal : undefined
      })
    } finally {
      if (longPoll) {
        this.activeLongPolls = Math.max(0, this.activeLongPolls - 1)
      }
    }
  }

  private parseAndAuth(rawMessage: string): { request: RpcRequest } | { error: RpcResponse } {
    let request: RpcRequest
    try {
      request = JSON.parse(rawMessage) as RpcRequest
    } catch {
      return { error: this.buildError('unknown', 'bad_request', 'Invalid JSON request') }
    }

    if (typeof request.id !== 'string' || request.id.length === 0) {
      return { error: this.buildError('unknown', 'bad_request', 'Missing request id') }
    }
    if (typeof request.method !== 'string' || request.method.length === 0) {
      return { error: this.buildError(request.id, 'bad_request', 'Missing RPC method') }
    }
    if (typeof request.authToken !== 'string' || request.authToken.length === 0) {
      return { error: this.buildError(request.id, 'unauthorized', 'Missing auth token') }
    }
    if (request.authToken !== this.authToken) {
      return { error: this.buildError(request.id, 'unauthorized', 'Invalid auth token') }
    }

    return { request }
  }

  // Why: WebSocket messages go through streaming dispatch which can emit
  // multiple responses. Auth uses per-device tokens from the device registry.
  private async handleWebSocketMessage(
    rawMessage: string,
    reply: (response: string) => void,
    sendBinary: (response: Uint8Array<ArrayBufferLike>) => boolean | void,
    wsTransport?: WebSocketTransport,
    ws?: WebSocket,
    authenticatedDeviceToken?: string | null,
    authenticatedSocket?: AuthenticatedMobileSocket
  ): Promise<void> {
    let request: RpcRequest
    try {
      request = JSON.parse(rawMessage) as RpcRequest
    } catch {
      reply(JSON.stringify(this.buildError('unknown', 'bad_request', 'Invalid JSON request')))
      return
    }

    if (typeof request.id !== 'string' || request.id.length === 0) {
      reply(JSON.stringify(this.buildError('unknown', 'bad_request', 'Missing request id')))
      return
    }
    if (typeof request.method !== 'string' || request.method.length === 0) {
      reply(JSON.stringify(this.buildError(request.id, 'bad_request', 'Missing RPC method')))
      return
    }

    const requestToken =
      typeof (request as Record<string, unknown>).deviceToken === 'string'
        ? ((request as Record<string, unknown>).deviceToken as string)
        : null
    if (authenticatedDeviceToken && requestToken && requestToken !== authenticatedDeviceToken) {
      reply(JSON.stringify(this.buildError(request.id, 'unauthorized', 'Device token mismatch')))
      return
    }
    // Why: E2EE already authenticated the WebSocket channel. Use that bound
    // identity for authorization instead of trusting a repeated request field.
    const token = authenticatedDeviceToken ?? requestToken
    if (!token) {
      reply(JSON.stringify(this.buildError(request.id, 'unauthorized', 'Missing device token')))
      return
    }
    const device = this.deviceRegistry?.validateToken(token)
    if (!device) {
      reply(JSON.stringify(this.buildError(request.id, 'unauthorized', 'Invalid device token')))
      return
    }
    if (device.scope === 'mobile' && !MOBILE_RPC_METHOD_ALLOWLIST.has(request.method)) {
      reply(
        JSON.stringify(
          this.buildError(
            request.id,
            'forbidden',
            `Method '${request.method}' is not available to mobile clients`
          )
        )
      )
      return
    }

    // Why: associate the deviceToken with this WebSocket so ws.on('close')
    // can notify the runtime which mobile client disconnected.
    if (wsTransport && ws) {
      wsTransport.setClientId(ws, token)
    }

    const longPoll = isLongPollRequest(request)
    if (longPoll && this.activeLongPolls >= this.longPollCap) {
      reply(
        JSON.stringify(
          this.buildError(
            request.id,
            'runtime_busy',
            'long-poll capacity reached; retry with backoff'
          )
        )
      )
      return
    }

    const abortRegistration = ws ? this.registerWebSocketDispatchAbort(ws) : null
    if (longPoll) {
      this.activeLongPolls += 1
    }

    // Why: older/saved WebSocket pairings may not carry scope metadata, so
    // stamp the authenticated scope onto the one method that probes the runtime.
    const replyForRequest =
      request.method === 'status.get'
        ? (response: string): void => reply(injectDeviceScope(response, device.scope))
        : reply

    const connectionId = ws ? this.mobileSocketWiring?.getConnectionId(ws) : undefined
    const pairingProvider = this.mobileRelayPairingProvider
    const pairingContext =
      pairingProvider && authenticatedSocket
        ? {
            getEndpoints: (params: PairingGetEndpointsParams) =>
              pairingProvider.getEndpoints(
                {
                  deviceId: authenticatedSocket.device.deviceId,
                  connectionId: authenticatedSocket.connectionId,
                  transport: authenticatedSocket.transport
                },
                params
              ),
            provisionRelay: (params: PairingProvisionRelayParams) =>
              pairingProvider.provisionRelay(
                {
                  deviceId: authenticatedSocket.device.deviceId,
                  connectionId: authenticatedSocket.connectionId,
                  transport: authenticatedSocket.transport
                },
                params
              )
          }
        : undefined
    try {
      await this.dispatcher.dispatchStreaming(request, replyForRequest, {
        connectionId,
        clientId: token,
        // Why: gates the mobile-only payload diet (native-chat char clipping) so
        // full-screen web/desktop runtime clients aren't truncated.
        clientKind: device.scope,
        pairing: pairingContext,
        signal: abortRegistration?.signal,
        sendBinary,
        registerBinaryStreamHandler: (streamId, handler) =>
          this.registerBinaryStreamHandler(connectionId, streamId, handler)
      })
    } finally {
      abortRegistration?.dispose()
      if (longPoll) {
        this.activeLongPolls = Math.max(0, this.activeLongPolls - 1)
      }
    }
  }

  private buildError(id: string, code: string, message: string): RpcResponse {
    return errorResponse(id, { runtimeId: this.runtime.getRuntimeId() }, code, message)
  }

  private writeMetadata(): void {
    const metadata: RuntimeMetadata = {
      runtimeId: this.runtime.getRuntimeId(),
      pid: this.pid,
      transports: this.transports,
      authToken: this.authToken,
      startedAt: this.runtime.getStartedAt()
    }
    writeRuntimeMetadata(this.userDataPath, metadata)
  }
}

/**
 * Why: the regex MUST stay in lockstep with createRuntimeTransportMetadata()
 * below, which emits `o-${pid}-${endpointSuffix}.sock` where endpointSuffix
 * is `[A-Za-z0-9_-]{1,4}` (derived from a sanitised runtimeId prefix, or
 * `'rt'` as the fallback). The invariant is covered by a unit test so any
 * future change to the transport-name shape trips CI.
 */
export const RUNTIME_SOCKET_NAME_REGEX = /^o-(\d+)-[A-Za-z0-9_-]+\.sock$/

export function sweepOrphanedRuntimeSockets(userDataPath: string, ownPid: number): void {
  let entries: string[]
  try {
    entries = readdirSync(userDataPath)
  } catch {
    // Why: first-launch userData may not exist yet; the cold-start path
    // below will create it. Nothing to sweep in that case.
    return
  }
  for (const entry of entries) {
    const match = RUNTIME_SOCKET_NAME_REGEX.exec(entry)
    if (!match) {
      continue
    }
    const pid = Number(match[1])
    if (!Number.isFinite(pid)) {
      continue
    }
    // Why: never touch the current process's socket. start() already
    // rmSync's it if it exists, but belt-and-braces — a bug in the own-pid
    // path here would rmSync a socket we're about to bind to.
    if (pid === ownPid) {
      continue
    }
    try {
      // Why: signal 0 is the POSIX liveness probe — it delivers no signal
      // but returns success iff the pid resolves AND the caller has
      // permission to signal it. ESRCH = no such process; EPERM = pid
      // exists but owned by another user, which is extremely unusual on a
      // desktop app's userData dir but we conservatively leave those
      // sockets alone.
      process.kill(pid, 0)
    } catch (error) {
      if ((error as NodeJS.ErrnoException).code === 'ESRCH') {
        try {
          rmSync(join(userDataPath, entry), { force: true })
        } catch {
          // Why: best-effort sweep — a permission error on unlink is fine
          // to ignore; the socket will be cleaned by a later start() or
          // by the OS on reboot.
        }
      }
    }
  }
}

export function createRuntimeTransportMetadata(
  userDataPath: string,
  pid: number,
  platform: NodeJS.Platform,
  runtimeId = 'runtime'
): RuntimeTransportMetadata {
  const endpointSuffix = runtimeId.replace(/[^a-zA-Z0-9_-]/g, '').slice(0, 4) || 'rt'
  if (platform === 'win32') {
    return {
      kind: 'named-pipe',
      // Why: Windows named pipes do not get the same chmod hardening path as
      // Unix sockets, so include a per-runtime suffix to avoid exposing a
      // stable, guessable control endpoint name across launches.
      endpoint: `\\\\.\\pipe\\orca-${pid}-${endpointSuffix}`
    }
  }
  return {
    kind: 'unix',
    endpoint: join(userDataPath, `o-${pid}-${endpointSuffix}.sock`)
  }
}