mirror of
https://github.com/stablyai/orca.git
synced 2026-10-20 16:03:20 +00:00
fix(runtime): preserve compatibility on one remote auth
This commit is contained in:
+17
-16
@@ -11,13 +11,12 @@ import { getDefaultUserDataPath, readMetadata } from './metadata'
|
||||
import { getCliStatus, resolveDesktopWindowStatus } from './status'
|
||||
import { sendRequest } from './transport'
|
||||
import { RuntimeClientError, RuntimeRpcFailureError, type RuntimeRpcSuccess } from './types'
|
||||
import type { sendWebSocketRequest } from './websocket-transport'
|
||||
import { resolveEnvironmentPairingOffer } from './environments'
|
||||
import { RemoteRuntimeCompatGate } from './remote-runtime-compat-gate'
|
||||
import { markEnvironmentUsed, resolveEnvironmentPairingOffer } from './environments'
|
||||
import {
|
||||
ORCHESTRATION_CONTRACT_RUNTIME_CAPABILITY,
|
||||
ORCHESTRATION_CONTRACT_VERSION
|
||||
} from '../../shared/protocol-version'
|
||||
import { RemoteRuntimeCompatGate } from './remote-runtime-compat-gate'
|
||||
import { createOrchestrationCompatibilityEnvelope } from './orchestration-compatibility-envelope'
|
||||
import { getTimeoutMsParam, isWaitingCheck } from './runtime-request-timeout'
|
||||
|
||||
@@ -32,8 +31,8 @@ const LONG_POLL_CLIENT_GRACE_MS = 10_000
|
||||
// Why: ws + tweetnacl + the remote-runtime frame stack only matter once a
|
||||
// request actually goes over a pairing offer, which local CLI calls never do.
|
||||
// Both call sites already await this, so deferring the load changes no ordering.
|
||||
async function loadSendWebSocketRequest(): Promise<typeof sendWebSocketRequest> {
|
||||
return (await import('./websocket-transport.js')).sendWebSocketRequest
|
||||
async function loadWebSocketTransport() {
|
||||
return await import('./websocket-transport.js')
|
||||
}
|
||||
|
||||
export class RuntimeClient {
|
||||
@@ -96,26 +95,28 @@ export class RuntimeClient {
|
||||
...compatibilityEnvelope
|
||||
}
|
||||
if (this.remotePairing) {
|
||||
const sendWebSocketRequest = await loadSendWebSocketRequest()
|
||||
if (method !== 'status.get') {
|
||||
await this.remoteCompat.ensure(this.remotePairing, effectiveTimeoutMs, sendWebSocketRequest)
|
||||
}
|
||||
const transport = await loadWebSocketTransport()
|
||||
let response
|
||||
try {
|
||||
response = await sendWebSocketRequest<TResult>(
|
||||
this.remotePairing,
|
||||
response = await this.remoteCompat.send<TResult>({
|
||||
transport,
|
||||
pairing: this.remotePairing,
|
||||
method,
|
||||
params,
|
||||
effectiveTimeoutMs,
|
||||
timeoutMs: effectiveTimeoutMs,
|
||||
envelope
|
||||
)
|
||||
})
|
||||
} catch (error) {
|
||||
throw attachMutationRecovery(error, orchestrationRequestId)
|
||||
}
|
||||
if (response.ok === false) {
|
||||
throw new RuntimeRpcFailureError(response)
|
||||
}
|
||||
this.remoteCompat.noteRespondingRuntimeId(response._meta.runtimeId)
|
||||
if (this.environmentSelector) {
|
||||
markEnvironmentUsed(this.userDataPath, this.environmentSelector, {
|
||||
runtimeId: response._meta.runtimeId
|
||||
})
|
||||
}
|
||||
return response
|
||||
}
|
||||
const metadata = readMetadata(this.userDataPath)
|
||||
@@ -152,7 +153,7 @@ export class RuntimeClient {
|
||||
async getCliStatus(): Promise<RuntimeRpcSuccess<CliStatusResult>> {
|
||||
if (this.remotePairing) {
|
||||
const response = await this.call<RuntimeStatus>('status.get')
|
||||
this.remoteCompat.noteVerifiedStatus(response.result, response._meta.runtimeId)
|
||||
this.remoteCompat.noteVerifiedStatus(response.result)
|
||||
const graphState = response.result.graphStatus
|
||||
return {
|
||||
id: response.id,
|
||||
@@ -200,7 +201,7 @@ export class RuntimeClient {
|
||||
private async checkOrchestrationContractCompatibility(timeoutMs: number): Promise<void> {
|
||||
const response = await this.call<RuntimeStatus>('status.get', undefined, { timeoutMs })
|
||||
if (this.remotePairing) {
|
||||
this.remoteCompat.noteVerifiedStatus(response.result, response._meta.runtimeId)
|
||||
this.remoteCompat.noteVerifiedStatus(response.result)
|
||||
}
|
||||
if (!response.result.capabilities?.includes(ORCHESTRATION_CONTRACT_RUNTIME_CAPABILITY)) {
|
||||
throw new RuntimeClientError(
|
||||
|
||||
@@ -1,10 +1,8 @@
|
||||
import {
|
||||
addEnvironmentFromPairingCode as addEnvironmentFromPairingCodeInStore,
|
||||
getEnvironmentRuntimeCompat as getEnvironmentRuntimeCompatInStore,
|
||||
getEnvironmentStorePath,
|
||||
listEnvironments,
|
||||
markEnvironmentUsed as markEnvironmentUsedInStore,
|
||||
recordEnvironmentRuntimeCompat as recordEnvironmentRuntimeCompatInStore,
|
||||
removeEnvironment as removeEnvironmentFromStore,
|
||||
resolveEnvironment as resolveEnvironmentFromStore,
|
||||
resolveEnvironmentPairingOffer as resolveEnvironmentPairingOfferFromStore,
|
||||
@@ -13,8 +11,7 @@ import {
|
||||
} from '../../shared/runtime-environment-store'
|
||||
import type {
|
||||
KnownRuntimeEnvironment,
|
||||
PublicKnownRuntimeEnvironment,
|
||||
RuntimeEnvironmentCompat
|
||||
PublicKnownRuntimeEnvironment
|
||||
} from '../../shared/runtime-environments'
|
||||
import type { PairingOffer } from '../../shared/pairing'
|
||||
import { RuntimeClientError } from './types'
|
||||
@@ -62,21 +59,6 @@ export function markEnvironmentUsed(
|
||||
translateStoreError(() => markEnvironmentUsedInStore(userDataPath, selector, args))
|
||||
}
|
||||
|
||||
export function getEnvironmentRuntimeCompat(
|
||||
userDataPath: string,
|
||||
selector: string
|
||||
): RuntimeEnvironmentCompat | null {
|
||||
return translateStoreError(() => getEnvironmentRuntimeCompatInStore(userDataPath, selector))
|
||||
}
|
||||
|
||||
export function recordEnvironmentRuntimeCompat(
|
||||
userDataPath: string,
|
||||
selector: string,
|
||||
compat: RuntimeEnvironmentCompat
|
||||
): void {
|
||||
translateStoreError(() => recordEnvironmentRuntimeCompatInStore(userDataPath, selector, compat))
|
||||
}
|
||||
|
||||
function translateStoreError<TResult>(fn: () => TResult): TResult {
|
||||
try {
|
||||
return fn()
|
||||
|
||||
@@ -1,173 +0,0 @@
|
||||
// Why: every remote CLI command used to open two full WebSocket connections — one for the compat
|
||||
// preflight, one for the real call — and each connection pays a complete E2EE authentication.
|
||||
// sendRemoteRuntimeRequest constructs exactly one WebSocket per call, so counting calls counts sockets.
|
||||
import { mkdtempSync, readFileSync, writeFileSync } from 'node:fs'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import {
|
||||
MIN_COMPATIBLE_RUNTIME_SERVER_VERSION,
|
||||
RUNTIME_PROTOCOL_VERSION
|
||||
} from '../../shared/protocol-version'
|
||||
import { encodePairingOffer, PAIRING_OFFER_VERSION } from '../../shared/pairing'
|
||||
import type * as RemoteRuntimeClientModule from '../../shared/remote-runtime-client'
|
||||
import type { RuntimeEnvironmentStore } from '../../shared/runtime-environments'
|
||||
import { RuntimeClient } from './client'
|
||||
|
||||
const mocks = vi.hoisted(() => ({
|
||||
sendRemoteRuntimeRequest: vi.fn(),
|
||||
runtimeId: { current: 'runtime-1' }
|
||||
}))
|
||||
|
||||
vi.mock('../../shared/remote-runtime-client', async (importOriginal) => ({
|
||||
...(await importOriginal<RemoteRuntimeClientModule>()),
|
||||
sendRemoteRuntimeRequest: mocks.sendRemoteRuntimeRequest
|
||||
}))
|
||||
|
||||
const ENVIRONMENT_NAME = 'dev-box'
|
||||
|
||||
function writeEnvironmentStore(userDataPath: string, runtimeCompat?: unknown): void {
|
||||
const store = {
|
||||
version: 1,
|
||||
environments: [
|
||||
{
|
||||
id: 'env-1',
|
||||
name: ENVIRONMENT_NAME,
|
||||
createdAt: 1,
|
||||
updatedAt: 1,
|
||||
lastUsedAt: null,
|
||||
runtimeId: null,
|
||||
...(runtimeCompat ? { runtimeCompat } : {}),
|
||||
endpoints: [
|
||||
{
|
||||
id: 'ws-env-1',
|
||||
kind: 'websocket',
|
||||
label: 'WebSocket',
|
||||
endpoint: 'ws://127.0.0.1:65000/',
|
||||
deviceToken: 'device-token',
|
||||
publicKeyB64: 'cHVibGljLWtleQ=='
|
||||
}
|
||||
],
|
||||
preferredEndpointId: 'ws-env-1'
|
||||
}
|
||||
]
|
||||
}
|
||||
writeFileSync(join(userDataPath, 'orca-environments.json'), JSON.stringify(store), 'utf8')
|
||||
}
|
||||
|
||||
function readStoredCompat(userDataPath: string): unknown {
|
||||
const store = JSON.parse(
|
||||
readFileSync(join(userDataPath, 'orca-environments.json'), 'utf8')
|
||||
) as RuntimeEnvironmentStore
|
||||
return store.environments[0]?.runtimeCompat ?? null
|
||||
}
|
||||
|
||||
function verifiedCompat(runtimeId = 'runtime-1'): Record<string, unknown> {
|
||||
return {
|
||||
runtimeId,
|
||||
appVersion: '1.4.165',
|
||||
clientProtocolVersion: RUNTIME_PROTOCOL_VERSION,
|
||||
minCompatibleServerProtocolVersion: MIN_COMPATIBLE_RUNTIME_SERVER_VERSION
|
||||
}
|
||||
}
|
||||
|
||||
describe('remote CLI compat preflight', () => {
|
||||
let userDataPath = ''
|
||||
|
||||
beforeEach(() => {
|
||||
userDataPath = mkdtempSync(join(tmpdir(), 'orca-remote-compat-'))
|
||||
mocks.runtimeId.current = 'runtime-1'
|
||||
mocks.sendRemoteRuntimeRequest.mockReset()
|
||||
mocks.sendRemoteRuntimeRequest.mockImplementation(
|
||||
async (_pairing: unknown, method: string) => ({
|
||||
id: 'rpc-1',
|
||||
ok: true,
|
||||
result:
|
||||
method === 'status.get'
|
||||
? {
|
||||
runtimeId: mocks.runtimeId.current,
|
||||
appVersion: '1.4.165',
|
||||
graphStatus: 'ready',
|
||||
runtimeProtocolVersion: RUNTIME_PROTOCOL_VERSION,
|
||||
minCompatibleRuntimeClientVersion: MIN_COMPATIBLE_RUNTIME_SERVER_VERSION
|
||||
}
|
||||
: { terminals: [] },
|
||||
_meta: { runtimeId: mocks.runtimeId.current }
|
||||
})
|
||||
)
|
||||
})
|
||||
|
||||
function client(): RuntimeClient {
|
||||
return new RuntimeClient(userDataPath, 1_000, null, ENVIRONMENT_NAME)
|
||||
}
|
||||
|
||||
it('opens two connections when the environment has no verified runtime', async () => {
|
||||
writeEnvironmentStore(userDataPath)
|
||||
|
||||
await client().call('terminal.list')
|
||||
|
||||
expect(mocks.sendRemoteRuntimeRequest).toHaveBeenCalledTimes(2)
|
||||
expect(mocks.sendRemoteRuntimeRequest.mock.calls.map(([, method]) => method)).toEqual([
|
||||
'status.get',
|
||||
'terminal.list'
|
||||
])
|
||||
expect(readStoredCompat(userDataPath)).toEqual(verifiedCompat())
|
||||
})
|
||||
|
||||
it('opens one connection once the runtime is recorded as compatible', async () => {
|
||||
writeEnvironmentStore(userDataPath, verifiedCompat())
|
||||
|
||||
await client().call('terminal.list')
|
||||
|
||||
expect(mocks.sendRemoteRuntimeRequest).toHaveBeenCalledTimes(1)
|
||||
expect(mocks.sendRemoteRuntimeRequest.mock.calls.map(([, method]) => method)).toEqual([
|
||||
'terminal.list'
|
||||
])
|
||||
})
|
||||
|
||||
it('re-runs the preflight after the runtime restarts under a trusted record', async () => {
|
||||
writeEnvironmentStore(userDataPath, verifiedCompat())
|
||||
mocks.runtimeId.current = 'runtime-2'
|
||||
|
||||
await client().call('terminal.list')
|
||||
expect(readStoredCompat(userDataPath)).toBeNull()
|
||||
|
||||
mocks.sendRemoteRuntimeRequest.mockClear()
|
||||
await client().call('terminal.list')
|
||||
|
||||
expect(mocks.sendRemoteRuntimeRequest.mock.calls.map(([, method]) => method)).toEqual([
|
||||
'status.get',
|
||||
'terminal.list'
|
||||
])
|
||||
})
|
||||
|
||||
it('re-runs the preflight when the record predates this client protocol version', async () => {
|
||||
writeEnvironmentStore(userDataPath, {
|
||||
...verifiedCompat(),
|
||||
clientProtocolVersion: RUNTIME_PROTOCOL_VERSION - 1
|
||||
})
|
||||
|
||||
await client().call('terminal.list')
|
||||
|
||||
expect(mocks.sendRemoteRuntimeRequest.mock.calls.map(([, method]) => method)).toEqual([
|
||||
'status.get',
|
||||
'terminal.list'
|
||||
])
|
||||
})
|
||||
|
||||
it('keeps the preflight for a raw pairing code, which has no record to trust', async () => {
|
||||
const pairingCode = encodePairingOffer({
|
||||
v: PAIRING_OFFER_VERSION,
|
||||
endpoint: 'ws://127.0.0.1:65000/',
|
||||
deviceToken: 'device-token',
|
||||
publicKeyB64: 'cHVibGljLWtleQ=='
|
||||
})
|
||||
|
||||
await new RuntimeClient(userDataPath, 1_000, pairingCode, null).call('terminal.list')
|
||||
|
||||
expect(mocks.sendRemoteRuntimeRequest.mock.calls.map(([, method]) => method)).toEqual([
|
||||
'status.get',
|
||||
'terminal.list'
|
||||
])
|
||||
})
|
||||
})
|
||||
@@ -1,83 +1,69 @@
|
||||
import type { RuntimeStatus } from '../../shared/runtime-types'
|
||||
import type { PairingOffer } from '../../shared/pairing'
|
||||
import { describeRuntimeCompatBlock, evaluateRuntimeCompat } from '../../shared/protocol-compat'
|
||||
import {
|
||||
MIN_COMPATIBLE_RUNTIME_SERVER_VERSION,
|
||||
RUNTIME_PROTOCOL_VERSION
|
||||
} from '../../shared/protocol-version'
|
||||
import {
|
||||
getEnvironmentRuntimeCompat,
|
||||
markEnvironmentUsed,
|
||||
recordEnvironmentRuntimeCompat
|
||||
} from './environments'
|
||||
import type { sendWebSocketRequest } from './websocket-transport'
|
||||
import { RuntimeClientError, RuntimeRpcFailureError } from './types'
|
||||
import type { RuntimeOrchestrationEnvelope } from '../../shared/runtime-rpc-envelope'
|
||||
import type { RuntimeStatus } from '../../shared/runtime-types'
|
||||
import { markEnvironmentUsed } from './environments'
|
||||
import { RuntimeClientError, RuntimeRpcFailureError, type RuntimeRpcResponse } from './types'
|
||||
import type {
|
||||
sendWebSocketRequest,
|
||||
sendWebSocketRequestWithStatusPreflight
|
||||
} from './websocket-transport'
|
||||
|
||||
type WebSocketTransport = {
|
||||
sendWebSocketRequest: typeof sendWebSocketRequest
|
||||
sendWebSocketRequestWithStatusPreflight: typeof sendWebSocketRequestWithStatusPreflight
|
||||
}
|
||||
|
||||
/**
|
||||
* Keeps a remote runtime's protocol-compatibility verdict for one CLI process.
|
||||
*
|
||||
* Verifying it used to cost a second full WebSocket connection (and a second E2EE authentication) on
|
||||
* every command. A verdict is now saved against the runtime's `runtimeId`, which is minted per runtime
|
||||
* launch, so a saved verdict proves the same verified process is still answering; a restarted or upgraded
|
||||
* runtime answers under a different id, which retires the verdict for the next command.
|
||||
*/
|
||||
export class RemoteRuntimeCompatGate {
|
||||
private checked = false
|
||||
private trustedRuntimeId: string | null = null
|
||||
|
||||
constructor(
|
||||
private readonly userDataPath: string,
|
||||
private readonly environmentSelector: string | null
|
||||
) {}
|
||||
|
||||
/** Verifies compatibility before a remote call, reusing a saved verdict instead of reconnecting. */
|
||||
async ensure(
|
||||
pairing: PairingOffer,
|
||||
timeoutMs: number,
|
||||
send: typeof sendWebSocketRequest
|
||||
): Promise<void> {
|
||||
if (this.checked || this.adoptSavedVerdict()) {
|
||||
return
|
||||
send<TResult>(args: {
|
||||
transport: WebSocketTransport
|
||||
pairing: PairingOffer
|
||||
method: string
|
||||
params: unknown
|
||||
timeoutMs: number
|
||||
envelope?: RuntimeOrchestrationEnvelope
|
||||
}): Promise<RuntimeRpcResponse<TResult>> {
|
||||
if (this.checked || args.method === 'status.get') {
|
||||
return args.transport.sendWebSocketRequest<TResult>(
|
||||
args.pairing,
|
||||
args.method,
|
||||
args.params,
|
||||
args.timeoutMs,
|
||||
args.envelope
|
||||
)
|
||||
}
|
||||
const response = await send<RuntimeStatus>(pairing, 'status.get', undefined, timeoutMs)
|
||||
if (response.ok === false) {
|
||||
throw new RuntimeRpcFailureError(response)
|
||||
}
|
||||
this.noteVerifiedStatus(response.result, response._meta.runtimeId)
|
||||
return args.transport.sendWebSocketRequestWithStatusPreflight<TResult>(
|
||||
args.pairing,
|
||||
args.method,
|
||||
args.params,
|
||||
args.timeoutMs,
|
||||
(response) => {
|
||||
if (response.ok === false) {
|
||||
throw new RuntimeRpcFailureError(response)
|
||||
}
|
||||
this.noteVerifiedStatus(response.result)
|
||||
if (this.environmentSelector) {
|
||||
markEnvironmentUsed(this.userDataPath, this.environmentSelector, {
|
||||
runtimeId: response._meta.runtimeId
|
||||
})
|
||||
}
|
||||
},
|
||||
args.envelope
|
||||
)
|
||||
}
|
||||
|
||||
/** Records a status the caller obtained anyway, so the next command can skip the preflight. */
|
||||
noteVerifiedStatus(status: RuntimeStatus, runtimeId: string): void {
|
||||
this.assertCompatible(status)
|
||||
this.checked = true
|
||||
this.trustedRuntimeId = runtimeId
|
||||
if (!this.environmentSelector) {
|
||||
return
|
||||
}
|
||||
try {
|
||||
recordEnvironmentRuntimeCompat(this.userDataPath, this.environmentSelector, {
|
||||
runtimeId,
|
||||
...(status.appVersion ? { appVersion: status.appVersion } : {}),
|
||||
clientProtocolVersion: RUNTIME_PROTOCOL_VERSION,
|
||||
minCompatibleServerProtocolVersion: MIN_COMPATIBLE_RUNTIME_SERVER_VERSION
|
||||
})
|
||||
} catch {
|
||||
// Why: the verdict is an optimization; failing to cache it must not fail the command.
|
||||
}
|
||||
}
|
||||
|
||||
/** Reacts to whichever runtime actually answered, retiring a verdict that belonged to another one. */
|
||||
noteRespondingRuntimeId(runtimeId: string): void {
|
||||
if (this.trustedRuntimeId && this.trustedRuntimeId !== runtimeId) {
|
||||
this.trustedRuntimeId = null
|
||||
this.checked = false
|
||||
}
|
||||
if (this.environmentSelector) {
|
||||
markEnvironmentUsed(this.userDataPath, this.environmentSelector, { runtimeId })
|
||||
}
|
||||
}
|
||||
|
||||
assertCompatible(status: RuntimeStatus): void {
|
||||
noteVerifiedStatus(status: RuntimeStatus): void {
|
||||
const verdict = evaluateRuntimeCompat({
|
||||
clientProtocolVersion: RUNTIME_PROTOCOL_VERSION,
|
||||
minCompatibleServerProtocolVersion: MIN_COMPATIBLE_RUNTIME_SERVER_VERSION,
|
||||
@@ -88,28 +74,6 @@ export class RemoteRuntimeCompatGate {
|
||||
if (verdict.kind === 'blocked') {
|
||||
throw new RuntimeClientError('incompatible_runtime', describeRuntimeCompatBlock(verdict))
|
||||
}
|
||||
}
|
||||
|
||||
private adoptSavedVerdict(): boolean {
|
||||
if (!this.environmentSelector) {
|
||||
return false
|
||||
}
|
||||
try {
|
||||
const compat = getEnvironmentRuntimeCompat(this.userDataPath, this.environmentSelector)
|
||||
// Why: a client update can change either bound, so a verdict from an older CLI is not reusable.
|
||||
if (
|
||||
!compat ||
|
||||
compat.clientProtocolVersion !== RUNTIME_PROTOCOL_VERSION ||
|
||||
compat.minCompatibleServerProtocolVersion !== MIN_COMPATIBLE_RUNTIME_SERVER_VERSION
|
||||
) {
|
||||
return false
|
||||
}
|
||||
this.trustedRuntimeId = compat.runtimeId
|
||||
this.checked = true
|
||||
return true
|
||||
} catch {
|
||||
// Why: an unreadable record must fall back to the preflight, never skip it.
|
||||
return false
|
||||
}
|
||||
this.checked = true
|
||||
}
|
||||
}
|
||||
|
||||
@@ -31,6 +31,8 @@ type TestRuntime = {
|
||||
publicKeyB64: string
|
||||
deviceToken: string
|
||||
authFrames: Record<string, unknown>[]
|
||||
requestMethods: string[]
|
||||
connectionCount: () => number
|
||||
close: () => Promise<void>
|
||||
}
|
||||
|
||||
@@ -168,6 +170,30 @@ describe('CLI remote WebSocket transport', () => {
|
||||
code: 'incompatible_runtime',
|
||||
message: expect.stringContaining('server is too old')
|
||||
})
|
||||
expect(runtime.connectionCount()).toBe(1)
|
||||
expect(runtime.authFrames).toHaveLength(1)
|
||||
expect(runtime.requestMethods).toEqual(['status.get'])
|
||||
})
|
||||
|
||||
it('preflights and dispatches through one authenticated connection', async () => {
|
||||
const runtime = await startTestRuntime('runtime-single-auth')
|
||||
servers.push(runtime)
|
||||
const client = new RuntimeClient(
|
||||
'/tmp/unused',
|
||||
5_000,
|
||||
encodePairingOffer({
|
||||
v: 2,
|
||||
endpoint: runtime.endpoint,
|
||||
deviceToken: runtime.deviceToken,
|
||||
publicKeyB64: runtime.publicKeyB64
|
||||
})
|
||||
)
|
||||
|
||||
await expect(client.call('repo.list')).rejects.toMatchObject({ code: 'method_not_found' })
|
||||
|
||||
expect(runtime.connectionCount()).toBe(1)
|
||||
expect(runtime.authFrames).toHaveLength(1)
|
||||
expect(runtime.requestMethods).toEqual(['status.get', 'repo.list'])
|
||||
})
|
||||
|
||||
it('blocks orchestration mutations when a remote runtime lacks the contract capability', async () => {
|
||||
@@ -214,8 +240,11 @@ async function startTestRuntime(
|
||||
const httpServer = createServer()
|
||||
const wss = new WebSocketServer({ server: httpServer })
|
||||
const authFrames: Record<string, unknown>[] = []
|
||||
const requestMethods: string[] = []
|
||||
let connectionCount = 0
|
||||
|
||||
wss.on('connection', (ws) => {
|
||||
connectionCount += 1
|
||||
let sharedKey: Uint8Array | null = null
|
||||
let authenticated = false
|
||||
|
||||
@@ -254,6 +283,7 @@ async function startTestRuntime(
|
||||
}
|
||||
|
||||
const request = JSON.parse(plaintext) as { id: string; method: string }
|
||||
requestMethods.push(request.method)
|
||||
const response =
|
||||
request.method === 'status.get'
|
||||
? {
|
||||
@@ -299,6 +329,8 @@ async function startTestRuntime(
|
||||
publicKeyB64: publicKeyToBase64(serverKeyPair.publicKey),
|
||||
deviceToken,
|
||||
authFrames,
|
||||
requestMethods,
|
||||
connectionCount: () => connectionCount,
|
||||
close: async () => {
|
||||
await new Promise<void>((resolve) => {
|
||||
wss.close(() => resolve())
|
||||
|
||||
@@ -2,8 +2,10 @@ import type { PairingOffer } from '../../shared/pairing'
|
||||
import type { RuntimeOrchestrationEnvelope } from '../../shared/runtime-rpc-envelope'
|
||||
import {
|
||||
RemoteRuntimeClientError,
|
||||
sendRemoteRuntimeRequest
|
||||
sendRemoteRuntimeRequest,
|
||||
sendRemoteRuntimeRequestWithStatusPreflight
|
||||
} from '../../shared/remote-runtime-client'
|
||||
import type { RuntimeStatus } from '../../shared/runtime-types'
|
||||
import { RuntimeClientError, type RuntimeRpcResponse } from './types'
|
||||
|
||||
export async function sendWebSocketRequest<TResult>(
|
||||
@@ -22,3 +24,28 @@ export async function sendWebSocketRequest<TResult>(
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
export async function sendWebSocketRequestWithStatusPreflight<TResult>(
|
||||
pairing: PairingOffer,
|
||||
method: string,
|
||||
params: unknown,
|
||||
timeoutMs: number,
|
||||
validateStatus: (response: RuntimeRpcResponse<RuntimeStatus>) => void,
|
||||
envelope?: RuntimeOrchestrationEnvelope
|
||||
): Promise<RuntimeRpcResponse<TResult>> {
|
||||
try {
|
||||
return await sendRemoteRuntimeRequestWithStatusPreflight<TResult>(
|
||||
pairing,
|
||||
method,
|
||||
params,
|
||||
timeoutMs,
|
||||
validateStatus,
|
||||
envelope
|
||||
)
|
||||
} catch (error) {
|
||||
if (error instanceof RemoteRuntimeClientError) {
|
||||
throw new RuntimeClientError(error.code, error.message)
|
||||
}
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
@@ -300,5 +300,7 @@ export class DeviceRegistry {
|
||||
|
||||
private save(devices: DeviceEntry[]): void {
|
||||
writeSecureJsonFile(this.registryPath, devices)
|
||||
// Why: every registry save includes the latest in-memory timestamps, so a later timer would rewrite it.
|
||||
this.cancelPendingLastSeenFlush()
|
||||
}
|
||||
}
|
||||
|
||||
@@ -191,4 +191,19 @@ describe('mobile auth critical path', () => {
|
||||
vi.advanceTimersByTime(250)
|
||||
expect(timeline.filter((entry) => entry === 'acl-spawn')).toHaveLength(2)
|
||||
})
|
||||
|
||||
it('cancels the deferred rewrite when another registry save persists the timestamp', () => {
|
||||
vi.useFakeTimers()
|
||||
const registry = new DeviceRegistry(userDataPath)
|
||||
const device = registry.addDevice('Phone', 'runtime')
|
||||
registry.updateLastSeen(device.deviceId)
|
||||
timeline.length = 0
|
||||
|
||||
registry.updateLastSeenDeferred(device.deviceId)
|
||||
registry.addDevice('Other client', 'runtime')
|
||||
expect(timeline.filter((entry) => entry === 'acl-spawn')).toHaveLength(2)
|
||||
|
||||
vi.advanceTimersByTime(250)
|
||||
expect(timeline.filter((entry) => entry === 'acl-spawn')).toHaveLength(2)
|
||||
})
|
||||
})
|
||||
|
||||
@@ -438,6 +438,37 @@ describe('OrcaRuntimeRpcServer', () => {
|
||||
expect(readRuntimeMetadata(userDataPath)).toMatchObject({ runtimeId: 'rt_second_instance' })
|
||||
})
|
||||
|
||||
it('flushes a lastSeen refresh scheduled while transports stop', async () => {
|
||||
const server = new OrcaRuntimeRpcServer({
|
||||
runtime: new OrcaRuntimeService(),
|
||||
userDataPath: mkdtempSync(join(tmpdir(), 'orca-runtime-rpc-')),
|
||||
enableWebSocket: false
|
||||
})
|
||||
let pending = false
|
||||
const timeline: string[] = []
|
||||
server['deviceRegistry'] = {
|
||||
flushPendingLastSeen: vi.fn(() => {
|
||||
timeline.push(pending ? 'flush-pending' : 'flush-empty')
|
||||
pending = false
|
||||
})
|
||||
} as unknown as DeviceRegistry
|
||||
server['activeTransports'] = [
|
||||
{
|
||||
start: vi.fn(async () => {}),
|
||||
stop: vi.fn(async () => {
|
||||
timeline.push('transport-stop')
|
||||
pending = true
|
||||
throw new Error('transport stop failed')
|
||||
})
|
||||
}
|
||||
]
|
||||
|
||||
await expect(server.stop()).rejects.toThrow('transport stop failed')
|
||||
|
||||
expect(timeline).toEqual(['transport-stop', 'flush-pending'])
|
||||
expect(pending).toBe(false)
|
||||
})
|
||||
|
||||
it('creates a pairing offer for the active WebSocket transport', async () => {
|
||||
const userDataPath = mkdtempSync(join(tmpdir(), 'orca-runtime-rpc-'))
|
||||
const runtime = new OrcaRuntimeService()
|
||||
|
||||
@@ -1497,13 +1497,14 @@ export class OrcaRuntimeRpcServer {
|
||||
this.transports = []
|
||||
this.metadataOwnershipWatch?.stop()
|
||||
this.metadataOwnershipWatch = null
|
||||
this.deviceRegistry?.flushPendingLastSeen()
|
||||
this.mobileSocketWiring = null
|
||||
this.detachWebSocketWiring = null
|
||||
if (transports.length === 0) {
|
||||
return
|
||||
try {
|
||||
await Promise.all(transports.map((t) => t.stop()))
|
||||
} finally {
|
||||
// Why: before-quit fences relay input; direct auth can still refresh lastSeen while these transports close.
|
||||
this.deviceRegistry?.flushPendingLastSeen()
|
||||
}
|
||||
await Promise.all(transports.map((t) => t.stop()))
|
||||
// Why: leave the metadata file on shutdown — shared userData may host another live runtime whose bootstrap file we'd erase.
|
||||
}
|
||||
|
||||
|
||||
@@ -21,6 +21,7 @@ import {
|
||||
type RuntimeOrchestrationEnvelope,
|
||||
type RuntimeRpcResponse
|
||||
} from './runtime-rpc-envelope'
|
||||
import type { RuntimeStatus } from './runtime-types'
|
||||
import { SESSION_TAB_CLOSE_INTENT_RUNTIME_CAPABILITY } from './protocol-version'
|
||||
// Re-export so existing value importers of `RemoteRuntimeClientError` are
|
||||
// unaffected; the class lives in a ws-free module so type-only consumers
|
||||
@@ -79,12 +80,41 @@ export type RemoteRuntimeSubscriptionCallbacks<TResult = unknown> = {
|
||||
onClose?: () => void
|
||||
}
|
||||
|
||||
export async function sendRemoteRuntimeRequest<TResult>(
|
||||
export function sendRemoteRuntimeRequest<TResult>(
|
||||
pairing: PairingOffer,
|
||||
method: string,
|
||||
params: unknown,
|
||||
timeoutMs: number,
|
||||
envelope?: RuntimeOrchestrationEnvelope
|
||||
): Promise<RuntimeRpcResponse<TResult>> {
|
||||
return sendRemoteRuntimeRequestOnSocket(pairing, method, params, timeoutMs, envelope)
|
||||
}
|
||||
|
||||
export function sendRemoteRuntimeRequestWithStatusPreflight<TResult>(
|
||||
pairing: PairingOffer,
|
||||
method: string,
|
||||
params: unknown,
|
||||
timeoutMs: number,
|
||||
validateStatus: (response: RuntimeRpcResponse<RuntimeStatus>) => void,
|
||||
envelope?: RuntimeOrchestrationEnvelope
|
||||
): Promise<RuntimeRpcResponse<TResult>> {
|
||||
return sendRemoteRuntimeRequestOnSocket(
|
||||
pairing,
|
||||
method,
|
||||
params,
|
||||
timeoutMs,
|
||||
envelope,
|
||||
validateStatus
|
||||
)
|
||||
}
|
||||
|
||||
async function sendRemoteRuntimeRequestOnSocket<TResult>(
|
||||
pairing: PairingOffer,
|
||||
method: string,
|
||||
params: unknown,
|
||||
timeoutMs: number,
|
||||
envelope?: RuntimeOrchestrationEnvelope,
|
||||
validateStatus?: (response: RuntimeRpcResponse<RuntimeStatus>) => void
|
||||
): Promise<RuntimeRpcResponse<TResult>> {
|
||||
if (!isSafeTimerDelayMs(timeoutMs)) {
|
||||
throw new RemoteRuntimeClientError(
|
||||
@@ -93,6 +123,14 @@ export async function sendRemoteRuntimeRequest<TResult>(
|
||||
)
|
||||
}
|
||||
const requestId = randomUUID()
|
||||
const statusRequestId = validateStatus ? randomUUID() : null
|
||||
const serializedStatusRequest = statusRequestId
|
||||
? serializeRemoteRuntimePayload({
|
||||
id: statusRequestId,
|
||||
deviceToken: pairing.deviceToken,
|
||||
method: 'status.get'
|
||||
})
|
||||
: null
|
||||
const serializedAuth = serializeRemoteRuntimePayload({
|
||||
type: 'e2ee_auth',
|
||||
deviceToken: pairing.deviceToken,
|
||||
@@ -114,6 +152,8 @@ export async function sendRemoteRuntimeRequest<TResult>(
|
||||
)
|
||||
}
|
||||
let serializedRequest = takeRemoteRuntimePreparedRequest(pendingRequest)
|
||||
let awaitingRequestId = statusRequestId ?? requestId
|
||||
let awaitingStatus = statusRequestId !== null
|
||||
return await new Promise<RuntimeRpcResponse<TResult>>((resolve, reject) => {
|
||||
const keyPair = generateKeyPair()
|
||||
const serverPublicKey = publicKeyFromBase64(pairing.publicKeyB64)
|
||||
@@ -359,6 +399,14 @@ export async function sendRemoteRuntimeRequest<TResult>(
|
||||
return
|
||||
}
|
||||
state = 'ready'
|
||||
if (serializedStatusRequest) {
|
||||
ws?.send(encrypt(serializedStatusRequest, sharedKey))
|
||||
return
|
||||
}
|
||||
sendRequestedRpc()
|
||||
}
|
||||
|
||||
function sendRequestedRpc(): void {
|
||||
const request = serializedRequest
|
||||
serializedRequest = null
|
||||
if (request === null) {
|
||||
@@ -405,8 +453,7 @@ export async function sendRemoteRuntimeRequest<TResult>(
|
||||
})
|
||||
return
|
||||
}
|
||||
const response = parsed.data as RuntimeRpcResponse<TResult>
|
||||
if (response.id !== requestId) {
|
||||
if (parsed.data.id !== awaitingRequestId) {
|
||||
finish({
|
||||
ok: false,
|
||||
error: new RemoteRuntimeClientError(
|
||||
@@ -417,6 +464,26 @@ export async function sendRemoteRuntimeRequest<TResult>(
|
||||
})
|
||||
return
|
||||
}
|
||||
if (awaitingStatus && validateStatus) {
|
||||
try {
|
||||
validateStatus(parsed.data as RuntimeRpcResponse<RuntimeStatus>)
|
||||
} catch (error) {
|
||||
finish({
|
||||
ok: false,
|
||||
error:
|
||||
error instanceof Error
|
||||
? error
|
||||
: new RemoteRuntimeClientError('runtime_error', String(error))
|
||||
})
|
||||
return
|
||||
}
|
||||
awaitingStatus = false
|
||||
awaitingRequestId = requestId
|
||||
refreshTimeout()
|
||||
sendRequestedRpc()
|
||||
return
|
||||
}
|
||||
const response = parsed.data as RuntimeRpcResponse<TResult>
|
||||
finish({ ok: true, response })
|
||||
}
|
||||
}).finally(() => releaseRemoteRuntimePreparedRequest(pendingRequest))
|
||||
|
||||
@@ -6,12 +6,10 @@ import { encodePairingOffer } from './pairing'
|
||||
import {
|
||||
RuntimeEnvironmentStoreError,
|
||||
addEnvironmentFromPairingCode,
|
||||
getEnvironmentRuntimeCompat,
|
||||
getEnvironmentStorePath,
|
||||
listEnvironments,
|
||||
MAX_RUNTIME_ENVIRONMENT_STORE_FILE_BYTES,
|
||||
markEnvironmentUsed,
|
||||
recordEnvironmentRuntimeCompat,
|
||||
updateEnvironmentFromPairingCode
|
||||
} from './runtime-environment-store'
|
||||
|
||||
@@ -153,32 +151,6 @@ describe('runtime environment store', () => {
|
||||
})
|
||||
})
|
||||
|
||||
it('retires a recorded compat verdict as soon as another runtime answers', () => {
|
||||
const userDataPath = mkdtempSync(join(tmpdir(), 'orca-runtime-env-store-'))
|
||||
tempDirs.push(userDataPath)
|
||||
const env = addEnvironmentFromPairingCode(userDataPath, {
|
||||
name: 'dev box',
|
||||
pairingCode: pairingCode()
|
||||
})
|
||||
const compat = {
|
||||
runtimeId: 'runtime-1',
|
||||
appVersion: '1.4.165',
|
||||
clientProtocolVersion: 7,
|
||||
minCompatibleServerProtocolVersion: 3
|
||||
}
|
||||
|
||||
recordEnvironmentRuntimeCompat(userDataPath, env.id, compat, 1_000)
|
||||
expect(getEnvironmentRuntimeCompat(userDataPath, env.id)).toEqual(compat)
|
||||
|
||||
// Same runtime inside the throttle window keeps the verdict.
|
||||
markEnvironmentUsed(userDataPath, env.id, { runtimeId: 'runtime-1', now: 2_000 })
|
||||
expect(getEnvironmentRuntimeCompat(userDataPath, env.id)).toEqual(compat)
|
||||
|
||||
// A restarted runtime mints a new runtimeId, so the verdict no longer applies.
|
||||
markEnvironmentUsed(userDataPath, env.id, { runtimeId: 'runtime-2', now: 3_000 })
|
||||
expect(getEnvironmentRuntimeCompat(userDataPath, env.id)).toBeNull()
|
||||
})
|
||||
|
||||
it('rejects an oversized sparse environment store before parsing it', () => {
|
||||
const userDataPath = mkdtempSync(join(tmpdir(), 'orca-runtime-env-store-bound-'))
|
||||
tempDirs.push(userDataPath)
|
||||
|
||||
@@ -13,7 +13,6 @@ import {
|
||||
KnownRuntimeEnvironmentSchema,
|
||||
RuntimeEnvironmentStoreSchema,
|
||||
type KnownRuntimeEnvironment,
|
||||
type RuntimeEnvironmentCompat,
|
||||
type RuntimeEnvironmentSource,
|
||||
type RuntimeEnvironmentStore
|
||||
} from './runtime-environments'
|
||||
@@ -183,71 +182,26 @@ export function markEnvironmentUsed(
|
||||
const environment = resolveEnvironmentFromStore(store, selector)
|
||||
const now = args.now ?? Date.now()
|
||||
const runtimeIdChanged = args.runtimeId != null && args.runtimeId !== environment.runtimeId
|
||||
// Why: the runtime that answered is not the one we verified, so its compat verdict no longer applies.
|
||||
const compatIsStale =
|
||||
environment.runtimeCompat != null &&
|
||||
args.runtimeId != null &&
|
||||
environment.runtimeCompat.runtimeId !== args.runtimeId
|
||||
const lastUsedIsFresh =
|
||||
environment.lastUsedAt != null &&
|
||||
now >= environment.lastUsedAt &&
|
||||
now - environment.lastUsedAt < LAST_USED_PERSIST_INTERVAL_MS
|
||||
if (!runtimeIdChanged && !compatIsStale && lastUsedIsFresh) {
|
||||
if (!runtimeIdChanged && lastUsedIsFresh) {
|
||||
return
|
||||
}
|
||||
const next = store.environments.map((entry) => {
|
||||
if (entry.id !== environment.id) {
|
||||
return entry
|
||||
}
|
||||
const { runtimeCompat, ...rest } = entry
|
||||
return {
|
||||
...rest,
|
||||
...(compatIsStale || !runtimeCompat ? {} : { runtimeCompat }),
|
||||
runtimeId: args.runtimeId ?? entry.runtimeId,
|
||||
lastUsedAt: now,
|
||||
updatedAt: now
|
||||
}
|
||||
})
|
||||
const next = store.environments.map((entry) =>
|
||||
entry.id === environment.id
|
||||
? {
|
||||
...entry,
|
||||
runtimeId: args.runtimeId ?? entry.runtimeId,
|
||||
lastUsedAt: now,
|
||||
updatedAt: now
|
||||
}
|
||||
: entry
|
||||
)
|
||||
writeEnvironmentStore(userDataPath, { version: 1, environments: next })
|
||||
}
|
||||
|
||||
/** The compat verdict recorded for this environment, or null when nothing has been verified yet. */
|
||||
export function getEnvironmentRuntimeCompat(
|
||||
userDataPath: string,
|
||||
selector: string
|
||||
): RuntimeEnvironmentCompat | null {
|
||||
const environment = resolveEnvironmentFromStore(readEnvironmentStore(userDataPath), selector)
|
||||
return environment.runtimeCompat ?? null
|
||||
}
|
||||
|
||||
/** Records a passed compat check so later commands can skip their preflight connection. */
|
||||
export function recordEnvironmentRuntimeCompat(
|
||||
userDataPath: string,
|
||||
selector: string,
|
||||
compat: RuntimeEnvironmentCompat,
|
||||
now = Date.now()
|
||||
): void {
|
||||
const store = readEnvironmentStore(userDataPath)
|
||||
const environment = resolveEnvironmentFromStore(store, selector)
|
||||
const current = environment.runtimeCompat
|
||||
if (
|
||||
current?.runtimeId === compat.runtimeId &&
|
||||
current.appVersion === compat.appVersion &&
|
||||
current.clientProtocolVersion === compat.clientProtocolVersion &&
|
||||
current.minCompatibleServerProtocolVersion === compat.minCompatibleServerProtocolVersion
|
||||
) {
|
||||
return
|
||||
}
|
||||
writeEnvironmentStore(userDataPath, {
|
||||
version: 1,
|
||||
environments: store.environments.map((entry) =>
|
||||
entry.id === environment.id
|
||||
? { ...entry, runtimeCompat: compat, runtimeId: compat.runtimeId, updatedAt: now }
|
||||
: entry
|
||||
)
|
||||
})
|
||||
}
|
||||
|
||||
function resolveEnvironmentFromStore(
|
||||
store: RuntimeEnvironmentStore,
|
||||
selector: string
|
||||
|
||||
@@ -20,18 +20,6 @@ export type PublicRuntimeAccessEndpoint = z.infer<typeof PublicRuntimeAccessEndp
|
||||
export const RuntimeEnvironmentSourceSchema = z.enum(['manual', 'ephemeral-vm'])
|
||||
export type RuntimeEnvironmentSource = z.infer<typeof RuntimeEnvironmentSourceSchema>
|
||||
|
||||
// Why: proof that this exact runtime launch already passed the protocol-compat check, so the CLI can skip
|
||||
// its preflight connection. runtimeId is minted per runtime launch, so any restart or upgrade retires it;
|
||||
// the client versions are recorded too so a CLI update re-checks.
|
||||
export const RuntimeEnvironmentCompatSchema = z.object({
|
||||
runtimeId: z.string().min(1),
|
||||
appVersion: z.string().min(1).optional(),
|
||||
clientProtocolVersion: z.number().finite(),
|
||||
minCompatibleServerProtocolVersion: z.number().finite()
|
||||
})
|
||||
|
||||
export type RuntimeEnvironmentCompat = z.infer<typeof RuntimeEnvironmentCompatSchema>
|
||||
|
||||
export const KnownRuntimeEnvironmentSchema = z.object({
|
||||
id: z.string().min(1),
|
||||
name: z.string().min(1),
|
||||
@@ -40,7 +28,6 @@ export const KnownRuntimeEnvironmentSchema = z.object({
|
||||
pairingRevision: z.number().finite().optional(),
|
||||
lastUsedAt: z.number().finite().nullable(),
|
||||
runtimeId: z.string().min(1).nullable(),
|
||||
runtimeCompat: RuntimeEnvironmentCompatSchema.optional(),
|
||||
source: RuntimeEnvironmentSourceSchema.optional(),
|
||||
connectionDependency: z.literal('ssh-tunnel').optional(),
|
||||
endpoints: z.array(RuntimeAccessEndpointSchema).min(1),
|
||||
|
||||
Reference in New Issue
Block a user