mirror of
https://github.com/stablyai/orca.git
synced 2026-10-22 08:03:37 +00:00
fix(orchestration): preserve replay across remints
This commit is contained in:
@@ -6,12 +6,12 @@ import { OrchestrationDb } from '../../orchestration/db'
|
||||
import type { OrchestrationEnvironmentTransport } from '../../orchestration/environment-transport'
|
||||
import type { RpcRequest } from '../core'
|
||||
import { RpcDispatcher } from '../dispatcher'
|
||||
import { fingerprintAuthenticatedPairingDevice } from '../orchestration-mutation-executor'
|
||||
import { fingerprintAuthenticatedPairingCredential } from '../orchestration-mutation-executor'
|
||||
import { ORCHESTRATION_METHODS } from './orchestration'
|
||||
|
||||
describe('orchestration federation control mail', () => {
|
||||
const homeToken = 'run-home-device-token'
|
||||
const homeFingerprint = fingerprintAuthenticatedPairingDevice('run-home-device')
|
||||
const homeFingerprint = fingerprintAuthenticatedPairingCredential(homeToken)
|
||||
const workerToken = 'worker-local-token'
|
||||
const workerPeerFingerprint = 'worker-peer'
|
||||
const coordinatorPaneKey = 'tab_coord:aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa'
|
||||
|
||||
@@ -627,6 +627,11 @@ describe('orchestration new-worktree workers', () => {
|
||||
db = new OrchestrationDb(join(dir, 'orchestration.db'))
|
||||
const restartedRuntime = new OrcaRuntimeService()
|
||||
restartedRuntime.setOrchestrationDb(db)
|
||||
vi.spyOn(restartedRuntime, 'getTerminalPaneKey').mockImplementation((handle) =>
|
||||
handle === 'term_coord_reminted'
|
||||
? `tab_coord_reminted:${coordinatorPaneKey.split(':')[1]}`
|
||||
: null
|
||||
)
|
||||
const recreateWorktree = vi
|
||||
.spyOn(restartedRuntime, 'createManagedWorktree')
|
||||
.mockRejectedValue(new Error('replay recreated the worktree'))
|
||||
@@ -640,7 +645,8 @@ describe('orchestration new-worktree workers', () => {
|
||||
const replay = await restartedDispatcher.dispatch({
|
||||
...request,
|
||||
id: 'rpc_worker_start_retry',
|
||||
authToken: 'caller-token-after-restart'
|
||||
authToken: 'caller-token-after-restart',
|
||||
params: { ...(request.params as Record<string, unknown>), from: 'term_coord_reminted' }
|
||||
})
|
||||
|
||||
expect(first).toMatchObject({
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import { createHash } from 'node:crypto'
|
||||
import { isOrchestrationMutation } from '../../../shared/orchestration-rpc-contract'
|
||||
import { parsePaneKey } from '../../../shared/stable-pane-id'
|
||||
import type { OrcaRuntimeService } from '../orca-runtime'
|
||||
import { OrchestrationError } from '../orchestration/orchestration-error'
|
||||
import type { RpcRequest } from './core'
|
||||
@@ -32,7 +33,14 @@ export class OrchestrationMutationExecutor {
|
||||
const callerFingerprint =
|
||||
callerFingerprintOverride ?? this.getLocalAuthenticatedCallerFingerprint()
|
||||
const payloadHash = createHash('sha256')
|
||||
.update(JSON.stringify(canonicalize({ method: request.method, params })))
|
||||
.update(
|
||||
JSON.stringify(
|
||||
canonicalize({
|
||||
method: request.method,
|
||||
params: replayStableCallerParams(this.runtime, params)
|
||||
})
|
||||
)
|
||||
)
|
||||
.digest('hex')
|
||||
const key = `${callerFingerprint}:${requestId}`
|
||||
const db = this.runtime.getOrchestrationDb()
|
||||
@@ -130,8 +138,31 @@ export function getOrchestrationMutationExecutor(
|
||||
return executor
|
||||
}
|
||||
|
||||
export function fingerprintAuthenticatedPairingDevice(deviceId: string): string {
|
||||
return createHash('sha256').update(`paired_device\0${deviceId}`).digest('hex')
|
||||
export function fingerprintAuthenticatedPairingCredential(token: string): string {
|
||||
return createHash('sha256').update(token).digest('hex')
|
||||
}
|
||||
|
||||
function replayStableCallerParams(runtime: OrcaRuntimeService, params: unknown): unknown {
|
||||
if (!params || typeof params !== 'object' || Array.isArray(params)) {
|
||||
return params
|
||||
}
|
||||
const source = params as Record<string, unknown>
|
||||
const result = { ...source }
|
||||
for (const property of ['from', 'callerTerminalHandle'] as const) {
|
||||
const handle = source[property]
|
||||
if (typeof handle !== 'string') {
|
||||
continue
|
||||
}
|
||||
const paneKey =
|
||||
property === 'from' && typeof source.senderPaneKey === 'string'
|
||||
? source.senderPaneKey
|
||||
: runtime.getTerminalPaneKey(handle)
|
||||
if (paneKey) {
|
||||
const leafId = parsePaneKey(paneKey)?.leafId
|
||||
result[property] = leafId ? { paneLeafId: leafId } : { paneKey }
|
||||
}
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
function canonicalize(value: unknown): unknown {
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
/* eslint-disable max-lines -- Why: this integration-style RPC test keeps the request/response contract together so regressions in the external CLI surface are easier to spot. */
|
||||
import { existsSync, mkdirSync, mkdtempSync } from 'node:fs'
|
||||
import { createHash } from 'node:crypto'
|
||||
import { rm, writeFile } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
@@ -3541,6 +3542,56 @@ describe('OrcaRuntimeRpcServer', () => {
|
||||
}
|
||||
})
|
||||
|
||||
it('keeps authenticated paired callers attached to existing federated workers', async () => {
|
||||
const userDataPath = mkdtempSync(join(tmpdir(), 'orca-runtime-rpc-'))
|
||||
const runtime = new OrcaRuntimeService()
|
||||
const db = new OrchestrationDb(':memory:')
|
||||
runtime.setOrchestrationDb(db)
|
||||
const server = new OrcaRuntimeRpcServer({ runtime, userDataPath, enableWebSocket: false })
|
||||
server['deviceRegistry'] = new DeviceRegistry(userDataPath)
|
||||
const device = server['deviceRegistry']!.addDevice('existing-cli', 'runtime')
|
||||
const existingFingerprint = createHash('sha256').update(device.token).digest('hex')
|
||||
db.createRemoteDispatchAttachment({
|
||||
dispatchId: 'ctx_existing_remote',
|
||||
taskId: 'task_existing_remote',
|
||||
homePeerFingerprint: existingFingerprint,
|
||||
protocolVersion: 1,
|
||||
runtimeEpoch: 'runtime_before_upgrade',
|
||||
mutationReceipt: {
|
||||
callerFingerprint: existingFingerprint,
|
||||
requestId: 'request_existing_remote',
|
||||
method: 'orchestration.federationAttachStart',
|
||||
payloadHash: 'hash_existing_remote'
|
||||
}
|
||||
})
|
||||
const replies: Record<string, unknown>[] = []
|
||||
|
||||
try {
|
||||
await server['handleWebSocketMessage'](
|
||||
JSON.stringify(
|
||||
withCurrentOrchestrationContract({
|
||||
id: 'show-existing-remote',
|
||||
method: 'orchestration.federationShow',
|
||||
params: { dispatchId: 'ctx_existing_remote' }
|
||||
})
|
||||
),
|
||||
(response) => replies.push(JSON.parse(response) as Record<string, unknown>),
|
||||
() => {},
|
||||
undefined,
|
||||
undefined,
|
||||
device.token
|
||||
)
|
||||
|
||||
expect(replies[0]).toMatchObject({
|
||||
ok: true,
|
||||
result: { dispatchId: 'ctx_existing_remote', attachment: { state: 'starting' } }
|
||||
})
|
||||
} finally {
|
||||
db.close()
|
||||
await server.stop()
|
||||
}
|
||||
})
|
||||
|
||||
it('rejects unpaired terminal creates before runtime dispatch', async () => {
|
||||
const userDataPath = mkdtempSync(join(tmpdir(), 'orca-runtime-rpc-'))
|
||||
const createMobileSessionTerminal = vi.fn()
|
||||
|
||||
@@ -15,7 +15,7 @@ import {
|
||||
import { RpcDispatcher } from './rpc/dispatcher'
|
||||
import type { RpcRequest, RpcResponse } from './rpc/core'
|
||||
import { errorResponse } from './rpc/errors'
|
||||
import { fingerprintAuthenticatedPairingDevice } from './rpc/orchestration-mutation-executor'
|
||||
import { fingerprintAuthenticatedPairingCredential } from './rpc/orchestration-mutation-executor'
|
||||
import type { RpcMessageContext, RpcTransport } from './rpc/transport'
|
||||
import { UnixSocketTransport } from './rpc/unix-socket-transport'
|
||||
import { WebSocketTransport } from './rpc/ws-transport'
|
||||
@@ -1712,7 +1712,8 @@ export class OrcaRuntimeRpcServer {
|
||||
: undefined
|
||||
try {
|
||||
await this.dispatcher.dispatchStreaming(request, replyForRequest, {
|
||||
authenticatedCallerFingerprint: fingerprintAuthenticatedPairingDevice(device.deviceId),
|
||||
// Why: the validated credential preserves existing federation ownership without trusting request fields.
|
||||
authenticatedCallerFingerprint: fingerprintAuthenticatedPairingCredential(token),
|
||||
connectionId,
|
||||
clientId: token,
|
||||
pairedDeviceId: device.deviceId,
|
||||
|
||||
Reference in New Issue
Block a user