From 80691e1ec44bae9d82c0ce020783d570f12ea720 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Fri, 14 Aug 2026 16:33:14 -0700 Subject: [PATCH] fix(orchestration): harden mutation replay identity --- src/cli/orchestration-mutation-recovery.ts | 2 +- src/cli/specs/orchestration.ts | 26 +++---- src/main/runtime/orchestration/db.ts | 46 ++++++++++++- ...ederation-acknowledgment-migration.test.ts | 2 +- .../mutation-receipt-capacity.test.ts | 2 +- ...hestration-db-retention-pagination.test.ts | 8 +-- .../runtime/rpc/dispatcher-stream-options.ts | 1 + src/main/runtime/rpc/dispatcher.ts | 24 +++++-- ...hestration-federation-control-mail.test.ts | 47 ++++++------- ...orchestration-workers-new-worktree.test.ts | 58 ++++++++++++++-- .../rpc/orchestration-contract-fence.test.ts | 3 +- .../rpc/orchestration-mutation-executor.ts | 15 ++-- .../rpc/orchestration-mutation-ledger.test.ts | 68 ++++++++++++++++--- src/main/runtime/runtime-rpc.test.ts | 58 ++++++++++++++++ src/main/runtime/runtime-rpc.ts | 2 + 15 files changed, 289 insertions(+), 73 deletions(-) diff --git a/src/cli/orchestration-mutation-recovery.ts b/src/cli/orchestration-mutation-recovery.ts index 28c905ceec6..8e4a03f911d 100644 --- a/src/cli/orchestration-mutation-recovery.ts +++ b/src/cli/orchestration-mutation-recovery.ts @@ -11,7 +11,7 @@ export function orchestrationMutationRecoveryError(error: unknown): unknown { } const message = [ stripUnsafeRetryAdvice(error.message, requestId), - 'The orchestration mutation may already have taken effect; a worktree, terminal, or dispatch may already exist.', + 'The orchestration mutation may already have taken effect; do not assume it failed.', `Re-issue the same command with --retry-request ${requestId} to recover idempotently. Do not retry this mutation without --retry-request.`, typeof data?.failedStage === 'string' ? `Failed stage: ${data.failedStage}.` : undefined, Array.isArray(data?.residualResources) diff --git a/src/cli/specs/orchestration.ts b/src/cli/specs/orchestration.ts index 5161960e0fc..f266e93b10a 100644 --- a/src/cli/specs/orchestration.ts +++ b/src/cli/specs/orchestration.ts @@ -6,7 +6,8 @@ export const ORCHESTRATION_COMMAND_SPECS: CommandSpec[] = [ { path: ['orchestration', 'run-create'], summary: 'Create and bind a lightweight orchestration Run', - usage: 'orca orchestration run-create --objective [--from ] [--json]', + usage: + 'orca orchestration run-create --objective [--from ] [--retry-request ] [--json]', allowedFlags: [...GLOBAL_FLAGS, 'objective', 'from', 'retry-request'], notes: [ 'A Run is a namespace and home inbox. It never schedules or places workers.', @@ -17,7 +18,7 @@ export const ORCHESTRATION_COMMAND_SPECS: CommandSpec[] = [ path: ['orchestration', 'run-use'], summary: 'Bind this coordinator terminal to an existing Run', usage: - 'orca orchestration run-use --id [--from ] [--takeover-legacy] [--json]', + 'orca orchestration run-use --id [--from ] [--takeover-legacy] [--retry-request ] [--json]', allowedFlags: [...GLOBAL_FLAGS, 'id', 'from', 'takeover-legacy', 'retry-request'], notes: [ '--takeover-legacy must run in the live coordinator agent terminal it binds; it preserves existing worker assignments.' @@ -45,7 +46,7 @@ export const ORCHESTRATION_COMMAND_SPECS: CommandSpec[] = [ path: ['orchestration', 'send'], summary: 'Send an inter-agent message', usage: - 'orca orchestration send --subject [--to ] [--run ] [--from ] [--body ] [--type ] [--priority ] [--thread-id ] [--payload ] [--task-id ] [--dispatch-id ] [--outcome ] [--files-modified ] [--report-path ] [--phase ] [--json]', + 'orca orchestration send --subject [--to ] [--run ] [--from ] [--body ] [--type ] [--priority ] [--thread-id ] [--payload ] [--task-id ] [--dispatch-id ] [--outcome ] [--files-modified ] [--report-path ] [--phase ] [--retry-request ] [--json]', allowedFlags: [ ...GLOBAL_FLAGS, 'to', @@ -80,7 +81,7 @@ export const ORCHESTRATION_COMMAND_SPECS: CommandSpec[] = [ path: ['orchestration', 'check'], summary: 'Check messages for a terminal', usage: - 'orca orchestration check [--terminal ] [--run ] [--ack ] [--unread | --peek | --all] [--types ] [--format] [--wait] [--timeout-ms ] [--json]\n' + + 'orca orchestration check [--terminal ] [--run ] [--ack ] [--unread | --peek | --all] [--types ] [--format] [--wait] [--timeout-ms ] [--retry-request ] [--json]\n' + " default: return the bound Run's oldest unacknowledged FIFO batch.\n" + ' --ack: acknowledge the prior whole batch before checking/waiting.\n' + ' --peek: return only unread messages without marking them read.\n' + @@ -114,7 +115,7 @@ export const ORCHESTRATION_COMMAND_SPECS: CommandSpec[] = [ path: ['orchestration', 'reply'], summary: 'Reply to a message', usage: - 'orca orchestration reply --id --body [--run ] [--from ] [--json]', + 'orca orchestration reply --id --body [--run ] [--from ] [--retry-request ] [--json]', allowedFlags: [...GLOBAL_FLAGS, 'id', 'body', 'run', 'from', 'retry-request'] }, { @@ -127,7 +128,7 @@ export const ORCHESTRATION_COMMAND_SPECS: CommandSpec[] = [ path: ['orchestration', 'task-create'], summary: 'Create an orchestration task', usage: - 'orca orchestration task-create --spec [--task-title ] [--display-name ] [--deps ] [--parent ] [--run ] [--from ] [--json]', + 'orca orchestration task-create --spec [--task-title ] [--display-name ] [--deps ] [--parent ] [--run ] [--from ] [--retry-request ] [--json]', allowedFlags: [ ...GLOBAL_FLAGS, 'spec', @@ -152,7 +153,7 @@ export const ORCHESTRATION_COMMAND_SPECS: CommandSpec[] = [ path: ['orchestration', 'task-update'], summary: 'Update a task status', usage: - 'orca orchestration task-update --id --status [--result ] [--run ] [--from ] [--json]', + 'orca orchestration task-update --id --status [--result ] [--run ] [--from ] [--retry-request ] [--json]', allowedFlags: [...GLOBAL_FLAGS, 'id', 'status', 'result', 'run', 'from', 'retry-request'], notes: ['Valid --status values: pending, ready, dispatched, completed, failed, blocked.'] }, @@ -161,7 +162,7 @@ export const ORCHESTRATION_COMMAND_SPECS: CommandSpec[] = [ path: ['orchestration', 'dispatch'], summary: 'Dispatch a task to a terminal', usage: - 'orca orchestration dispatch --task --to [--from ] [--run ] [--inject] [--dry-run] [--return-preamble] [--json]', + 'orca orchestration dispatch --task --to [--from ] [--run ] [--inject] [--dry-run] [--return-preamble] [--retry-request ] [--json]', allowedFlags: [ ...GLOBAL_FLAGS, 'task', @@ -185,7 +186,7 @@ export const ORCHESTRATION_COMMAND_SPECS: CommandSpec[] = [ path: ['orchestration', 'ask'], summary: 'Ask the coordinator a question and block until answered', usage: - 'orca orchestration ask (--question | --resume ) [--to ] [--run ] [--options ] [--timeout-ms ] [--from ] [--json]', + 'orca orchestration ask (--question | --resume ) [--to ] [--run ] [--options ] [--timeout-ms ] [--from ] [--retry-request ] [--json]', allowedFlags: [ ...GLOBAL_FLAGS, 'to', @@ -236,14 +237,14 @@ export const ORCHESTRATION_COMMAND_SPECS: CommandSpec[] = [ path: ['orchestration', 'gate-create'], summary: 'Create a decision gate blocking a task', usage: - 'orca orchestration gate-create --task --question [--options ] [--from ] [--json]', + 'orca orchestration gate-create --task --question [--options ] [--from ] [--retry-request ] [--json]', allowedFlags: [...GLOBAL_FLAGS, 'task', 'question', 'options', 'from', 'retry-request'] }, { path: ['orchestration', 'gate-resolve'], summary: 'Resolve a pending decision gate', usage: - 'orca orchestration gate-resolve --id --resolution [--from ] [--json]', + 'orca orchestration gate-resolve --id --resolution [--from ] [--retry-request ] [--json]', allowedFlags: [...GLOBAL_FLAGS, 'id', 'resolution', 'from', 'retry-request'] }, { @@ -257,7 +258,8 @@ export const ORCHESTRATION_COMMAND_SPECS: CommandSpec[] = [ { path: ['orchestration', 'reset'], summary: 'Reset one explicit orchestration state scope', - usage: 'orca orchestration reset (--all | --tasks | --messages) [--json]', + usage: + 'orca orchestration reset (--all | --tasks | --messages) [--retry-request ] [--json]', allowedFlags: [...GLOBAL_FLAGS, 'all', 'tasks', 'messages', 'retry-request'] } ] diff --git a/src/main/runtime/orchestration/db.ts b/src/main/runtime/orchestration/db.ts index 0613be520fa..79cc9402e6f 100644 --- a/src/main/runtime/orchestration/db.ts +++ b/src/main/runtime/orchestration/db.ts @@ -297,8 +297,8 @@ type RunListCursor = { id: string } -// Schema versions: v2 'heartbeat'+last_heartbeat_at, v3 delivered_at, v4 task-creator terminal, v5 task_title/display_name, v6 pane identity, v7 lightweight Runs, v8 crash-safe Run deliveries, v9 durable question threads, v10 Dispatch capabilities, v11 durable mutation receipts, v12 composed worker state, v18 post-v6 version-skew repair, v19 adopted legacy Runs and compatibility receipts, v20 legacy question backfill, v21 legacy scheduler-loss provenance, v22 dispatch assignee lookup, v23 worker terminal resource ownership, v24 creator-incarnation authority, v25 active Dispatch handle lookup, v26 indexed mutation receipt capacity, v27 durable federation acknowledgments. -const SCHEMA_VERSION = 27 +// Schema versions: v2 'heartbeat'+last_heartbeat_at, v3 delivered_at, v4 task-creator terminal, v5 task_title/display_name, v6 pane identity, v7 lightweight Runs, v8 crash-safe Run deliveries, v9 durable question threads, v10 Dispatch capabilities, v11 durable mutation receipts, v12 composed worker state, v18 post-v6 version-skew repair, v19 adopted legacy Runs and compatibility receipts, v20 legacy question backfill, v21 legacy scheduler-loss provenance, v22 dispatch assignee lookup, v23 worker terminal resource ownership, v24 creator-incarnation authority, v25 active Dispatch handle lookup, v26 indexed mutation receipt capacity, v27 durable federation acknowledgments, v28 durable local mutation caller identity. +const SCHEMA_VERSION = 28 function hardenOrchestrationDatabaseFiles(dbPath: (string & {}) | ':memory:'): void { if (dbPath === ':memory:' || process.platform === 'win32') { @@ -322,6 +322,7 @@ export class OrchestrationDb { // emptiness so the non-orchestration majority short-circuits the whole // per-terminal fan-out. Only createDispatchContext flips this false→true. private hasAnyDispatchContextsCache: boolean | undefined + private localMutationCallerFingerprint: string | undefined constructor(dbPath: (string & {}) | ':memory:') { this.db = new Database(dbPath) @@ -405,6 +406,11 @@ export class OrchestrationDb { PRIMARY KEY (caller_fingerprint, request_id) ); + CREATE TABLE IF NOT EXISTS mutation_caller_identities ( + transport TEXT PRIMARY KEY, + caller_fingerprint TEXT NOT NULL UNIQUE + ); + CREATE TABLE IF NOT EXISTS worker_dispatches ( dispatch_id TEXT PRIMARY KEY, runtime_epoch TEXT, @@ -1012,6 +1018,14 @@ export class OrchestrationDb { 'ALTER TABLE federated_dispatches ADD COLUMN to_home_acknowledged_sequence INTEGER NOT NULL DEFAULT 0' ) } + if (current < 28) { + this.db.exec(` + CREATE TABLE IF NOT EXISTS mutation_caller_identities ( + transport TEXT PRIMARY KEY, + caller_fingerprint TEXT NOT NULL UNIQUE + ); + `) + } this.db.exec(` CREATE INDEX IF NOT EXISTS idx_dispatch_assignee_pane_leaf ON dispatch_contexts(${DISPATCH_PANE_KEY_MATCH_SUFFIX_SQL}) @@ -1427,6 +1441,34 @@ export class OrchestrationDb { // ── Durable mutation receipts ── + getOrCreateLocalMutationCallerFingerprint(): string { + if (this.localMutationCallerFingerprint) { + return this.localMutationCallerFingerprint + } + const transport = 'local_authenticated_transport' + const existing = this.db + .prepare('SELECT caller_fingerprint FROM mutation_caller_identities WHERE transport = ?') + .get(transport) as { caller_fingerprint: string } | undefined + if (existing) { + this.localMutationCallerFingerprint = existing.caller_fingerprint + return this.localMutationCallerFingerprint + } + this.db + .prepare( + `INSERT OR IGNORE INTO mutation_caller_identities (transport, caller_fingerprint) + VALUES (?, ?)` + ) + .run(transport, randomBytes(32).toString('hex')) + const created = this.db + .prepare('SELECT caller_fingerprint FROM mutation_caller_identities WHERE transport = ?') + .get(transport) as { caller_fingerprint: string } | undefined + if (!created) { + throw new Error('Failed to create the local orchestration mutation caller identity.') + } + this.localMutationCallerFingerprint = created.caller_fingerprint + return this.localMutationCallerFingerprint + } + beginMutationReceipt(params: { callerFingerprint: string requestId: string diff --git a/src/main/runtime/orchestration/federation-acknowledgment-migration.test.ts b/src/main/runtime/orchestration/federation-acknowledgment-migration.test.ts index 80959d3df48..059bac2d8c0 100644 --- a/src/main/runtime/orchestration/federation-acknowledgment-migration.test.ts +++ b/src/main/runtime/orchestration/federation-acknowledgment-migration.test.ts @@ -41,7 +41,7 @@ describe('federation acknowledgment migration', () => { db = new OrchestrationDb(dbPath) const sqlite = (db as unknown as { db: Database.Database }).db - expect(sqlite.pragma('user_version', { simple: true })).toBe(27) + expect(sqlite.pragma('user_version', { simple: true })).toBe(28) expect(db.getFederatedDispatch('ctx_migrated')).toMatchObject({ to_home_imported_sequence: 2, to_home_acknowledged_sequence: 0 diff --git a/src/main/runtime/orchestration/mutation-receipt-capacity.test.ts b/src/main/runtime/orchestration/mutation-receipt-capacity.test.ts index 327caf5f922..547117c1ed8 100644 --- a/src/main/runtime/orchestration/mutation-receipt-capacity.test.ts +++ b/src/main/runtime/orchestration/mutation-receipt-capacity.test.ts @@ -103,7 +103,7 @@ describe('mutation receipt capacity schema', () => { db = new OrchestrationDb(dbPath) const sqlite = sqliteFor(db) - expect(sqlite.pragma('user_version', { simple: true })).toBe(27) + expect(sqlite.pragma('user_version', { simple: true })).toBe(28) expect(sqlite.prepare('SELECT receipt_count FROM mutation_receipt_ledger').get()).toEqual({ receipt_count: 20 }) diff --git a/src/main/runtime/orchestration/orchestration-db-retention-pagination.test.ts b/src/main/runtime/orchestration/orchestration-db-retention-pagination.test.ts index 4b8095da873..9a3afab1d81 100644 --- a/src/main/runtime/orchestration/orchestration-db-retention-pagination.test.ts +++ b/src/main/runtime/orchestration/orchestration-db-retention-pagination.test.ts @@ -240,7 +240,7 @@ describe('OrchestrationDb dispatch assignee index migration', () => { db = new OrchestrationDb(dbPath) const sqlite = sqliteFor(db) - expect(sqlite.pragma('user_version', { simple: true })).toBe(27) + expect(sqlite.pragma('user_version', { simple: true })).toBe(28) expect(db.getDispatchContextById(dispatch.id)).toMatchObject({ assignee_handle: 'term_worker' }) expect(db.getTask(task.id)).toMatchObject({ created_by_pane_key: null, @@ -277,7 +277,7 @@ describe('OrchestrationDb dispatch assignee index migration', () => { db.close() db = new OrchestrationDb(dbPath) - expect(sqliteFor(db).pragma('user_version', { simple: true })).toBe(27) + expect(sqliteFor(db).pragma('user_version', { simple: true })).toBe(28) expect(db.getDispatchContextById(dispatch.id)).toBeDefined() }) @@ -309,7 +309,7 @@ describe('OrchestrationDb dispatch assignee index migration', () => { db = new OrchestrationDb(dbPath) const sqlite = sqliteFor(db) - expect(sqlite.pragma('user_version', { simple: true })).toBe(27) + expect(sqlite.pragma('user_version', { simple: true })).toBe(28) expect(db.getTask(task.id)).toMatchObject({ created_by_pane_key: 'tab_creator:leaf_creator', created_by_process_incarnation: 'pty_creator:incarnation-a', @@ -328,7 +328,7 @@ describe('OrchestrationDb dispatch assignee index migration', () => { db.close() db = new OrchestrationDb(dbPath) - expect(sqliteFor(db).pragma('user_version', { simple: true })).toBe(27) + expect(sqliteFor(db).pragma('user_version', { simple: true })).toBe(28) expect(db.getTask(task.id)?.created_by_process_incarnation).toBe('pty_creator:incarnation-a') }) }) diff --git a/src/main/runtime/rpc/dispatcher-stream-options.ts b/src/main/runtime/rpc/dispatcher-stream-options.ts index 0e492c7ad9a..19342993f47 100644 --- a/src/main/runtime/rpc/dispatcher-stream-options.ts +++ b/src/main/runtime/rpc/dispatcher-stream-options.ts @@ -3,6 +3,7 @@ import type { TerminalStreamFrame } from '../../../shared/terminal-stream-protoc import type { PairingRpcContext } from './core' export type RpcDispatchStreamingOptions = { + authenticatedCallerFingerprint?: string connectionId?: string signal?: AbortSignal clientId?: string diff --git a/src/main/runtime/rpc/dispatcher.ts b/src/main/runtime/rpc/dispatcher.ts index be11b2b0c64..b86b30f7c52 100644 --- a/src/main/runtime/rpc/dispatcher.ts +++ b/src/main/runtime/rpc/dispatcher.ts @@ -15,7 +15,6 @@ import { ALL_RPC_METHODS } from './methods' import { emulatorProbe, emulatorProbeError } from '../../emulator/emulator-probe' import type { OrcaRuntimeService } from '../orca-runtime' import { - authenticatedCallerFingerprint, getOrchestrationMutationExecutor, type OrchestrationMutationExecutor, type DurableMutationInvocation @@ -41,7 +40,10 @@ export class RpcDispatcher { this.legacyOrchestration = new OrchestrationLegacyCompatibility(runtime) } - async dispatch(request: RpcRequest, options?: { signal?: AbortSignal }): Promise { + async dispatch( + request: RpcRequest, + options?: { signal?: AbortSignal; authenticatedCallerFingerprint?: string } + ): Promise { const meta = this.meta() const method = this.registry.get(request.method) if (!method) { @@ -89,6 +91,11 @@ export class RpcDispatcher { request, compatibility.legacyCoordinatorAuthority ) + const authenticatedCallerFingerprint = + options?.authenticatedCallerFingerprint ?? + (request.method.startsWith('orchestration.') + ? this.orchestrationMutations.getLocalAuthenticatedCallerFingerprint() + : undefined) const invoke = (mutation?: DurableMutationInvocation) => { const legacyCoordinatorRunId = legacyCoordinator?.revalidate() return method.handler(effectiveParams, { @@ -97,7 +104,7 @@ export class RpcDispatcher { requestId: request.id, orchestrationCapability: request.orchestrationCapability, authenticatedCallerFingerprint: - mutation?.identity.callerFingerprint ?? authenticatedCallerFingerprint(request), + mutation?.identity.callerFingerprint ?? authenticatedCallerFingerprint, recordMutationReceipt: mutation?.recordReceipt, orchestrationMutation: mutation?.identity, legacyCoordinatorRunId, @@ -112,7 +119,7 @@ export class RpcDispatcher { request, effectiveParams, invoke, - legacyCoordinator?.mutationCallerFingerprint + legacyCoordinator?.mutationCallerFingerprint ?? authenticatedCallerFingerprint ) recordRuntimeFeatureInteraction( this.runtime, @@ -177,6 +184,11 @@ export class RpcDispatcher { request, compatibility.legacyCoordinatorAuthority ) + const authenticatedCallerFingerprint = + options?.authenticatedCallerFingerprint ?? + (request.method.startsWith('orchestration.') + ? this.orchestrationMutations.getLocalAuthenticatedCallerFingerprint() + : undefined) const invoke = (mutation?: DurableMutationInvocation) => { const legacyCoordinatorRunId = legacyCoordinator?.revalidate() return method.handler(effectiveParams, { @@ -190,7 +202,7 @@ export class RpcDispatcher { clientCapabilities: options?.clientCapabilities, orchestrationCapability: request.orchestrationCapability, authenticatedCallerFingerprint: - mutation?.identity.callerFingerprint ?? authenticatedCallerFingerprint(request), + mutation?.identity.callerFingerprint ?? authenticatedCallerFingerprint, recordMutationReceipt: mutation?.recordReceipt, orchestrationMutation: mutation?.identity, pairing: options?.pairing, @@ -208,7 +220,7 @@ export class RpcDispatcher { request, effectiveParams, invoke, - legacyCoordinator?.mutationCallerFingerprint + legacyCoordinator?.mutationCallerFingerprint ?? authenticatedCallerFingerprint ) recordRuntimeFeatureInteraction( this.runtime, diff --git a/src/main/runtime/rpc/methods/orchestration-federation-control-mail.test.ts b/src/main/runtime/rpc/methods/orchestration-federation-control-mail.test.ts index b061b7dcb2d..a2505ad806e 100644 --- a/src/main/runtime/rpc/methods/orchestration-federation-control-mail.test.ts +++ b/src/main/runtime/rpc/methods/orchestration-federation-control-mail.test.ts @@ -6,11 +6,12 @@ import { OrchestrationDb } from '../../orchestration/db' import type { OrchestrationEnvironmentTransport } from '../../orchestration/environment-transport' import type { RpcRequest } from '../core' import { RpcDispatcher } from '../dispatcher' -import { authenticatedCallerFingerprint } from '../orchestration-mutation-executor' +import { fingerprintAuthenticatedPairingDevice } 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 workerToken = 'worker-local-token' const workerPeerFingerprint = 'worker-peer' const coordinatorPaneKey = 'tab_coord:aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa' @@ -55,14 +56,17 @@ describe('orchestration federation control mail', () => { _meta: { runtimeId: workerRuntime.getRuntimeId() } } } - const response = (await workerDispatcher.dispatch({ - id: `remote_${method}`, - authToken: homeToken, - method, - params, - orchestrationContractVersion: envelope?.orchestrationContractVersion, - orchestrationRequestId: envelope?.orchestrationRequestId - })) as RuntimeRpcResponse + const response = (await workerDispatcher.dispatch( + { + id: `remote_${method}`, + authToken: homeToken, + method, + params, + orchestrationContractVersion: envelope?.orchestrationContractVersion, + orchestrationRequestId: envelope?.orchestrationRequestId + }, + { authenticatedCallerFingerprint: homeFingerprint } + )) as RuntimeRpcResponse return response } } @@ -99,11 +103,6 @@ describe('orchestration federation control mail', () => { dispatchId = started.dispatch.id homeDb.markWorkerDispatchReady(dispatchId) - const homeFingerprint = authenticatedCallerFingerprint({ - id: 'home', - authToken: homeToken, - method: 'orchestration.federationImport' - }) workerDb.createRemoteDispatchAttachment({ dispatchId, taskId: task.id, @@ -182,9 +181,7 @@ describe('orchestration federation control mail', () => { const waiting = workerDispatcher.dispatch(checkRequest('wait-for-control', true)) await Promise.resolve() - const imported = await workerDispatcher.dispatch( - importRequest('import-control', 1, 'relay-control') - ) + const imported = await dispatchImport(importRequest('import-control', 1, 'relay-control')) expect(imported).toMatchObject({ ok: true, @@ -201,8 +198,8 @@ describe('orchestration federation control mail', () => { }) it('accepts a repeated import after a lost acknowledgment without duplicating mail', async () => { - const first = await workerDispatcher.dispatch(importRequest('first-import', 1, 'relay-control')) - const repeated = await workerDispatcher.dispatch( + const first = await dispatchImport(importRequest('first-import', 1, 'relay-control')) + const repeated = await dispatchImport( importRequest('repeated-import', 1, 'different-message-id') ) @@ -258,7 +255,7 @@ describe('orchestration federation control mail', () => { expect(workerDb.getUnreadMessages(`dispatch:${dispatchId}`)).toHaveLength(0) expect(homeDb.listPendingFederationRelay(dispatchId, 'to_worker')).toHaveLength(1) await expect( - workerDispatcher.dispatch(importRequest('late-direct-import', 1, 'late-control')) + dispatchImport(importRequest('late-direct-import', 1, 'late-control')) ).resolves.toMatchObject({ ok: false, error: { code: 'dispatch_inactive' } @@ -276,9 +273,7 @@ describe('orchestration federation control mail', () => { const statusWaiter = workerDispatcher.dispatch(checkRequest('wait-status', true, 30, 'status')) await Promise.resolve() - await workerDispatcher.dispatch( - importRequest('import-escalation', 1, 'relay-escalation', 'escalation') - ) + await dispatchImport(importRequest('import-escalation', 1, 'relay-escalation', 'escalation')) await expect(escalationWaiter).resolves.toMatchObject({ ok: true, @@ -342,4 +337,10 @@ describe('orchestration federation control mail', () => { } } } + + function dispatchImport(request: RpcRequest) { + return workerDispatcher.dispatch(request, { + authenticatedCallerFingerprint: homeFingerprint + }) + } }) diff --git a/src/main/runtime/rpc/methods/orchestration-workers-new-worktree.test.ts b/src/main/runtime/rpc/methods/orchestration-workers-new-worktree.test.ts index 24c6f9defb4..3c6a6b01bad 100644 --- a/src/main/runtime/rpc/methods/orchestration-workers-new-worktree.test.ts +++ b/src/main/runtime/rpc/methods/orchestration-workers-new-worktree.test.ts @@ -1,4 +1,6 @@ -import { createHash } from 'node:crypto' +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { ORCHESTRATION_CONTRACT_VERSION } from '../../../../shared/protocol-version' import { OrcaRuntimeService } from '../../orca-runtime' @@ -13,6 +15,7 @@ describe('orchestration new-worktree workers', () => { let db: OrchestrationDb let runtime: OrcaRuntimeService let runId: string + const paths: string[] = [] beforeEach(() => { db = new OrchestrationDb(':memory:') @@ -71,7 +74,12 @@ describe('orchestration new-worktree workers', () => { }) }) - afterEach(() => db.close()) + afterEach(() => { + db.close() + for (const path of paths.splice(0)) { + rmSync(path, { recursive: true, force: true }) + } + }) async function startWorker(overrides: Record = {}) { const task = db.createTask({ spec: 'new-worktree task', runId }) @@ -542,7 +550,7 @@ describe('orchestration new-worktree workers', () => { const pending = dispatcher.dispatch(request) await vi.waitFor(() => expect(db.getDispatchContext(task.id)).toBeDefined()) const acceptedDispatch = db.getDispatchContext(task.id)! - const callerFingerprint = createHash('sha256').update('caller-token').digest('hex') + const callerFingerprint = db.getOrCreateLocalMutationCallerFingerprint() const receipt = db.getMutationReceipt(callerFingerprint, 'worker_start_request') expect(receipt).toMatchObject({ @@ -574,7 +582,17 @@ describe('orchestration new-worktree workers', () => { }) }) - it('replays a dispatch-input failure without creating another worker', async () => { + it('replays a dispatch-input failure after restart without creating another worker', async () => { + const dir = mkdtempSync(join(tmpdir(), 'orca-worker-start-replay-')) + paths.push(dir) + db.close() + db = new OrchestrationDb(join(dir, 'orchestration.db')) + runtime.setOrchestrationDb(db) + runId = db.createRun({ + objective: 'Recover dispatch input', + coordinatorHandle: 'term_coord', + coordinatorPaneKey + }).id mockCreatedWorktree({ hookFound: false }) vi.mocked(runtime.sendTerminalAgentPrompt).mockRejectedValueOnce( new Error('connection closed before dispatch input was accepted') @@ -597,7 +615,33 @@ describe('orchestration new-worktree workers', () => { } const first = await dispatcher.dispatch(request) - const replay = await dispatcher.dispatch({ ...request, id: 'rpc_worker_start_retry' }) + if (!first.ok) { + throw new Error(`Initial worker start failed: ${first.error.code}`) + } + const firstReceipt = first.result as { + dispatchId: string + residualResources: unknown[] + } + db.close() + + db = new OrchestrationDb(join(dir, 'orchestration.db')) + const restartedRuntime = new OrcaRuntimeService() + restartedRuntime.setOrchestrationDb(db) + const recreateWorktree = vi + .spyOn(restartedRuntime, 'createManagedWorktree') + .mockRejectedValue(new Error('replay recreated the worktree')) + const reinjectPrompt = vi + .spyOn(restartedRuntime, 'sendTerminalAgentPrompt') + .mockRejectedValue(new Error('replay reinjected the prompt')) + const restartedDispatcher = new RpcDispatcher({ + runtime: restartedRuntime, + methods: ORCHESTRATION_METHODS + }) + const replay = await restartedDispatcher.dispatch({ + ...request, + id: 'rpc_worker_start_retry', + authToken: 'caller-token-after-restart' + }) expect(first).toMatchObject({ ok: true, @@ -614,13 +658,17 @@ describe('orchestration new-worktree workers', () => { expect(replay).toMatchObject({ ok: true, result: { + dispatchId: firstReceipt.dispatchId, state: 'failed', failedStage: 'dispatch_input', + residualResources: firstReceipt.residualResources, mutation: { requestId: 'worker_start_request', replayed: true } } }) expect(runtime.createManagedWorktree).toHaveBeenCalledOnce() expect(runtime.sendTerminalAgentPrompt).toHaveBeenCalledOnce() + expect(recreateWorktree).not.toHaveBeenCalled() + expect(reinjectPrompt).not.toHaveBeenCalled() }) it('persists pre-effect, post-effect, and post-input stages in order', async () => { diff --git a/src/main/runtime/rpc/orchestration-contract-fence.test.ts b/src/main/runtime/rpc/orchestration-contract-fence.test.ts index a6618542d9c..17aee4b0b95 100644 --- a/src/main/runtime/rpc/orchestration-contract-fence.test.ts +++ b/src/main/runtime/rpc/orchestration-contract-fence.test.ts @@ -1,4 +1,3 @@ -import { createHash } from 'node:crypto' import { z } from 'zod' import { afterEach, describe, expect, it, vi } from 'vitest' import { ORCHESTRATION_CONTRACT_VERSION } from '../../../shared/protocol-version' @@ -73,7 +72,7 @@ describe('orchestration contract fence', () => { } }) expect(effect).not.toHaveBeenCalled() - const callerFingerprint = createHash('sha256').update('caller-token').digest('hex') + const callerFingerprint = database.getOrCreateLocalMutationCallerFingerprint() expect(database.getMutationReceipt(callerFingerprint, 'mutation_1')).toBeUndefined() } ) diff --git a/src/main/runtime/rpc/orchestration-mutation-executor.ts b/src/main/runtime/rpc/orchestration-mutation-executor.ts index 701970eb58c..79fd14a28ca 100644 --- a/src/main/runtime/rpc/orchestration-mutation-executor.ts +++ b/src/main/runtime/rpc/orchestration-mutation-executor.ts @@ -29,7 +29,8 @@ export class OrchestrationMutationExecutor { if (!requestId || !isOrchestrationMutation(request.method, params)) { return await invoke() } - const callerFingerprint = callerFingerprintOverride ?? authenticatedCallerFingerprint(request) + const callerFingerprint = + callerFingerprintOverride ?? this.getLocalAuthenticatedCallerFingerprint() const payloadHash = createHash('sha256') .update(JSON.stringify(canonicalize({ method: request.method, params }))) .digest('hex') @@ -109,6 +110,10 @@ export class OrchestrationMutationExecutor { this.inFlight.delete(key) } } + + getLocalAuthenticatedCallerFingerprint(): string { + return this.runtime.getOrchestrationDb().getOrCreateLocalMutationCallerFingerprint() + } } const executorsByRuntime = new WeakMap() @@ -125,12 +130,8 @@ export function getOrchestrationMutationExecutor( return executor } -export function authenticatedCallerFingerprint(request: RpcRequest): string { - const callerToken = - request.authToken || - (request as RpcRequest & { deviceToken?: string }).deviceToken || - 'authenticated_transport' - return createHash('sha256').update(callerToken).digest('hex') +export function fingerprintAuthenticatedPairingDevice(deviceId: string): string { + return createHash('sha256').update(`paired_device\0${deviceId}`).digest('hex') } function canonicalize(value: unknown): unknown { diff --git a/src/main/runtime/rpc/orchestration-mutation-ledger.test.ts b/src/main/runtime/rpc/orchestration-mutation-ledger.test.ts index a9cbfbc5ab7..bc54f1643ed 100644 --- a/src/main/runtime/rpc/orchestration-mutation-ledger.test.ts +++ b/src/main/runtime/rpc/orchestration-mutation-ledger.test.ts @@ -93,7 +93,7 @@ describe('durable orchestration mutation ledger', () => { db.close() }) - it('applies the same ledger on authenticated WebSocket dispatch', async () => { + it('keys WebSocket replay to the authenticated device across reconnects', async () => { const { db, dispatcher, effect } = createHarness() const replies: string[] = [] const firstRequest = request({ @@ -102,10 +102,23 @@ describe('durable orchestration mutation ledger', () => { subject: 'remote' }) as RpcRequest & { deviceToken?: string } firstRequest.authToken = '' - firstRequest.deviceToken = 'paired-device' - await dispatcher.dispatchStreaming(firstRequest, (reply) => replies.push(reply)) - const replayRequest = { ...firstRequest, id: 'rpc_2' } - await dispatcher.dispatchStreaming(replayRequest, (reply) => replies.push(reply)) + firstRequest.deviceToken = 'untrusted-request-value-a' + await dispatcher.dispatchStreaming(firstRequest, (reply) => replies.push(reply), { + authenticatedCallerFingerprint: 'paired-device-a' + }) + const replayRequest = { + ...firstRequest, + id: 'rpc_2', + deviceToken: 'untrusted-request-value-b' + } + await dispatcher.dispatchStreaming(replayRequest, (reply) => replies.push(reply), { + authenticatedCallerFingerprint: 'paired-device-a' + }) + await dispatcher.dispatchStreaming( + { ...replayRequest, id: 'rpc_3' }, + (reply) => replies.push(reply), + { authenticatedCallerFingerprint: 'paired-device-b' } + ) expect(JSON.parse(replies[0] ?? '{}')).toMatchObject({ ok: true, @@ -115,7 +128,12 @@ describe('durable orchestration mutation ledger', () => { ok: true, result: { mutation: { replayed: true } } }) - expect(effect).toHaveBeenCalledTimes(1) + expect(JSON.parse(replies[2] ?? '{}')).toMatchObject({ + ok: true, + result: { mutation: { replayed: false } } + }) + expect(effect).toHaveBeenCalledTimes(2) + expect(db.getInbox(10)).toHaveLength(2) db.close() }) @@ -178,10 +196,42 @@ describe('durable orchestration mutation ledger', () => { second.db.close() }) + it('replays a local mutation after runtime authentication rotates', async () => { + const dir = mkdtempSync(join(tmpdir(), 'orca-mutation-ledger-')) + paths.push(dir) + const dbPath = join(dir, 'orchestration.db') + const firstRuntime = createHarness(dbPath) + const first = await firstRuntime.dispatcher.dispatch( + request({ + rpcId: 'rpc_1', + mutationId: 'mutation_restart', + subject: 'once', + authToken: 'before-restart' + }) + ) + firstRuntime.db.close() + + const restartedRuntime = createHarness(dbPath) + const replay = await restartedRuntime.dispatcher.dispatch( + request({ + rpcId: 'rpc_2', + mutationId: 'mutation_restart', + subject: 'once', + authToken: 'after-restart' + }) + ) + + expect(first).toMatchObject({ ok: true, result: { mutation: { replayed: false } } }) + expect(replay).toMatchObject({ ok: true, result: { mutation: { replayed: true } } }) + expect(firstRuntime.effect).toHaveBeenCalledOnce() + expect(restartedRuntime.effect).not.toHaveBeenCalled() + restartedRuntime.db.close() + }) + it('returns unknown for a pending receipt left by a previous process', async () => { const { db, dispatcher } = createHarness() db.beginMutationReceipt({ - callerFingerprint: createHash('sha256').update('caller-token').digest('hex'), + callerFingerprint: db.getOrCreateLocalMutationCallerFingerprint(), requestId: 'mutation_1', method: 'orchestration.send', payloadHash: createHash('sha256') @@ -201,7 +251,7 @@ describe('durable orchestration mutation ledger', () => { const runtime = new OrcaRuntimeService() runtime.setOrchestrationDb(db) const params = { dispatch: 'ctx_release' } - const callerFingerprint = createHash('sha256').update('caller-token').digest('hex') + const callerFingerprint = db.getOrCreateLocalMutationCallerFingerprint() const payloadHash = createHash('sha256') .update(JSON.stringify({ method: 'orchestration.workerRelease', params })) .digest('hex') @@ -249,7 +299,7 @@ describe('durable orchestration mutation ledger', () => { const runtime = new OrcaRuntimeService() runtime.setOrchestrationDb(db) const params = { from: 'term_coord', task: db.createTask({ spec: 'restart' }).id } - const callerFingerprint = createHash('sha256').update('caller-token').digest('hex') + const callerFingerprint = db.getOrCreateLocalMutationCallerFingerprint() const payloadHash = createHash('sha256') .update(JSON.stringify({ method: 'orchestration.workerStart', params })) .digest('hex') diff --git a/src/main/runtime/runtime-rpc.test.ts b/src/main/runtime/runtime-rpc.test.ts index 9fde32745e5..53160e879df 100644 --- a/src/main/runtime/runtime-rpc.test.ts +++ b/src/main/runtime/runtime-rpc.test.ts @@ -3483,6 +3483,64 @@ describe('OrcaRuntimeRpcServer', () => { ) }) + it('isolates mutation replay by the authenticated paired device across reconnects', 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 firstDevice = server['deviceRegistry']!.addDevice('first-cli', 'runtime') + const secondDevice = server['deviceRegistry']!.addDevice('second-cli', 'runtime') + + const resetMessages = async (id: string, authenticatedToken: string) => { + const replies: Record[] = [] + await server['handleWebSocketMessage']( + JSON.stringify( + withCurrentOrchestrationContract({ + id, + method: 'orchestration.reset', + orchestrationRequestId: 'paired-reset-request', + params: { messages: true } + }) + ), + (response) => replies.push(JSON.parse(response) as Record), + () => {}, + undefined, + undefined, + authenticatedToken + ) + return replies[0] + } + + try { + db.insertMessage({ from: 'worker', to: 'coordinator', subject: 'before reset' }) + const first = await resetMessages('reset-first', firstDevice.token) + db.insertMessage({ from: 'worker', to: 'coordinator', subject: 'after reset' }) + const replay = await resetMessages('reset-replay', firstDevice.token) + + expect(first).toMatchObject({ + ok: true, + result: { reset: 'messages', mutation: { replayed: false } } + }) + expect(replay).toMatchObject({ + ok: true, + result: { reset: 'messages', mutation: { replayed: true } } + }) + expect(db.getInbox()).toEqual([expect.objectContaining({ subject: 'after reset' })]) + + const isolated = await resetMessages('reset-second-device', secondDevice.token) + expect(isolated).toMatchObject({ + ok: true, + result: { reset: 'messages', mutation: { replayed: false } } + }) + expect(db.getInbox()).toEqual([]) + } 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() diff --git a/src/main/runtime/runtime-rpc.ts b/src/main/runtime/runtime-rpc.ts index 326801c6fa9..543029a18ae 100644 --- a/src/main/runtime/runtime-rpc.ts +++ b/src/main/runtime/runtime-rpc.ts @@ -15,6 +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 type { RpcMessageContext, RpcTransport } from './rpc/transport' import { UnixSocketTransport } from './rpc/unix-socket-transport' import { WebSocketTransport } from './rpc/ws-transport' @@ -1711,6 +1712,7 @@ export class OrcaRuntimeRpcServer { : undefined try { await this.dispatcher.dispatchStreaming(request, replyForRequest, { + authenticatedCallerFingerprint: fingerprintAuthenticatedPairingDevice(device.deviceId), connectionId, clientId: token, pairedDeviceId: device.deviceId,