diff --git a/src/cli/format-recovery.test.ts b/src/cli/format-recovery.test.ts index 9646c32d85b..fea52395cd8 100644 --- a/src/cli/format-recovery.test.ts +++ b/src/cli/format-recovery.test.ts @@ -57,4 +57,17 @@ describe('CLI error recovery', () => { expect(output).toContain('Fix the command flags or RPC params') }) + + it('does not replace mutation recovery with generic runtime startup advice', () => { + const error = new RuntimeClientError( + 'runtime_unavailable', + 'Re-issue the same command with --retry-request mutation_1.', + { orchestrationRequestId: 'mutation_1' } + ) + + const output = formatCliError(error) + + expect(output).toContain('--retry-request mutation_1') + expect(output).not.toContain('orca open') + }) }) diff --git a/src/cli/format.ts b/src/cli/format.ts index 2bc8de3aa64..c76f13bd944 100644 --- a/src/cli/format.ts +++ b/src/cli/format.ts @@ -78,6 +78,9 @@ export function printResult( export function formatCliError(error: unknown, context: CliErrorContext = {}): string { const message = error instanceof Error ? error.message : String(error) if (error instanceof RuntimeClientError && error.code === 'runtime_unavailable') { + if (hasOrchestrationRequestId(error.data)) { + return message + } return `${message}\nOrca is not running. Run 'orca open' first.` } // Why: error-specific recovery must win over the generic computer fallback. @@ -105,6 +108,14 @@ export function formatCliError(error: unknown, context: CliErrorContext = {}): s return message } +function hasOrchestrationRequestId(data: unknown): boolean { + return ( + data !== null && + typeof data === 'object' && + typeof (data as { orchestrationRequestId?: unknown }).orchestrationRequestId === 'string' + ) +} + export function reportCliError(error: unknown, json: boolean, context: CliErrorContext = {}): void { if (json) { if (error instanceof RuntimeRpcFailureError) { diff --git a/src/cli/handlers/orchestration-gate-cli.test.ts b/src/cli/handlers/orchestration-gate-cli.test.ts index e7b16e345a6..0bcdddbc98e 100644 --- a/src/cli/handlers/orchestration-gate-cli.test.ts +++ b/src/cli/handlers/orchestration-gate-cli.test.ts @@ -195,4 +195,47 @@ describe('orchestration gate commands carry caller identity', () => { expect(stderr).toContain('Pass --from ') expect(callMock).not.toHaveBeenCalledWith('orchestration.gateCreate', expect.anything()) }) + + it('reports idempotent recovery when a mutation connection drops', async () => { + process.env.ORCA_TERMINAL_HANDLE = 'term_coord' + callMock + .mockResolvedValueOnce(okFixture('req_show', { terminal: { handle: 'term_coord' } })) + .mockRejectedValueOnce( + new RuntimeClientError( + 'runtime_unavailable', + 'The Orca runtime closed the connection before responding. Restart Orca and try again. Orchestration mutation request ID: mutation_1.', + { + orchestrationRequestId: 'mutation_1', + failedStage: 'dispatch_input', + residualResources: [ + { kind: 'worktree', id: 'repo::child' }, + { kind: 'terminal', id: 'term_worker' } + ] + } + ) + ) + + await main( + ['orchestration', 'gate-create', '--task', 'task_1', '--question', 'ship?', '--json'], + '/tmp/repo' + ) + + expect(process.exitCode).toBe(1) + const output = JSON.parse(String(logSpy.mock.calls[0]?.[0])) as { + error: { message: string; data: Record } + } + expect(output.error.message).toContain('--retry-request mutation_1') + expect(output.error.message).toContain('may already have taken effect') + expect(output.error.message).toContain('Failed stage: dispatch_input') + expect(output.error.message).toMatch(/Residual resources:.*repo::child.*term_worker/) + expect(output.error.message).not.toMatch(/restart Orca/i) + expect(output.error.data).toMatchObject({ + orchestrationRequestId: 'mutation_1', + failedStage: 'dispatch_input', + residualResources: expect.arrayContaining([ + expect.objectContaining({ kind: 'worktree', id: 'repo::child' }), + expect.objectContaining({ kind: 'terminal', id: 'term_worker' }) + ]) + }) + }) }) diff --git a/src/cli/handlers/orchestration.ts b/src/cli/handlers/orchestration.ts index 337f026f9bf..5fe16b22181 100644 --- a/src/cli/handlers/orchestration.ts +++ b/src/cli/handlers/orchestration.ts @@ -32,6 +32,7 @@ import { type LegacyCompatibilityResult, type OrchestrationMessageSummary as MessageSummary } from '../../shared/orchestration-check-output' +import { orchestrationMutationRecoveryError } from '../orchestration-mutation-recovery' // Why: 15 s is well under Claude Code's ~2 min Bash-tool silence budget while keeping log volume low. See design doc §3.4. const DEFAULT_KEEPALIVE_INTERVAL_MS = 15_000 @@ -397,14 +398,13 @@ function callMutation( options?: { timeoutMs?: number; orchestrationCapability?: string } ) { const requestId = getOptionalStringFlag(flags, 'retry-request') - if (!requestId) { - return options + const result = requestId + ? client.call(method, params, { ...options, orchestrationRequestId: requestId }) + : options ? client.call(method, params, options) : client.call(method, params) - } - return client.call(method, params, { - ...options, - orchestrationRequestId: requestId + return result.catch((error) => { + throw orchestrationMutationRecoveryError(error) }) } diff --git a/src/cli/orchestration-mutation-recovery.ts b/src/cli/orchestration-mutation-recovery.ts new file mode 100644 index 00000000000..8e4a03f911d --- /dev/null +++ b/src/cli/orchestration-mutation-recovery.ts @@ -0,0 +1,44 @@ +import { RuntimeClientError } from './runtime-client' + +export function orchestrationMutationRecoveryError(error: unknown): unknown { + if (!(error instanceof RuntimeClientError) || !isUnknownMutationOutcomeCode(error.code)) { + return error + } + const data = objectRecord(error.data) + const requestId = data?.orchestrationRequestId + if (typeof requestId !== 'string' || requestId.length === 0) { + return error + } + const message = [ + stripUnsafeRetryAdvice(error.message, requestId), + '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) + ? `Residual resources: ${JSON.stringify(data.residualResources)}.` + : undefined + ].filter((line): line is string => line !== undefined) + return new RuntimeClientError(error.code, message.join('\n'), error.data) +} + +function isUnknownMutationOutcomeCode(code: string): boolean { + return [ + 'runtime_unavailable', + 'remote_runtime_unavailable', + 'runtime_timeout', + 'invalid_runtime_response' + ].includes(code) +} + +function objectRecord(value: unknown): Record | undefined { + return value !== null && typeof value === 'object' + ? (value as Record) + : undefined +} + +function stripUnsafeRetryAdvice(message: string, requestId: string): string { + return message + .replace(' Restart Orca and try again.', '') + .replace(' Retry the command.', '') + .replace(` Orchestration mutation request ID: ${requestId}.`, '') +} diff --git a/src/cli/runtime-client.test.ts b/src/cli/runtime-client.test.ts index bb0af016108..06c1af65e4f 100644 --- a/src/cli/runtime-client.test.ts +++ b/src/cli/runtime-client.test.ts @@ -4,7 +4,7 @@ import { join } from 'node:path' import { createServer, type Socket } from 'node:net' import { afterEach, describe, expect, it, vi } from 'vitest' import { ORCHESTRATION_CONTRACT_RUNTIME_CAPABILITY } from '../shared/protocol-version' -import { RuntimeClient, RuntimeRpcFailureError } from './runtime-client' +import { RuntimeClient, RuntimeClientError, RuntimeRpcFailureError } from './runtime-client' import { launchOrcaApp } from './runtime/launch' vi.mock('./runtime/launch', () => ({ @@ -415,6 +415,36 @@ describe.skipIf(process.platform === 'win32')('RuntimeClient', () => { }) }) + it('preserves a dropped read-only orchestration failure exactly', async () => { + const userDataPath = mkdtempSync(join(tmpdir(), 'orca-runtime-client-')) + const endpoint = join(userDataPath, 'runtime.sock') + let request: Record | undefined + const server = createServer((socket) => { + sockets.add(socket) + socket.once('close', () => sockets.delete(socket)) + socket.once('data', (data) => { + request = JSON.parse(String(data).trim()) as Record + socket.end() + }) + }) + servers.add(server) + await new Promise((resolve) => server.listen(endpoint, resolve)) + writeMetadata(userDataPath, endpoint) + const client = new RuntimeClient(userDataPath, 100) + + const failure = await client + .call('orchestration.workerShow', { dispatch: 'ctx_1' }) + .catch((error: unknown) => error) + + expect(failure).toBeInstanceOf(RuntimeClientError) + expect((failure as RuntimeClientError).message).toBe( + 'The Orca runtime closed the connection before responding. Restart Orca and try again.' + ) + expect((failure as RuntimeClientError).data).toBeUndefined() + expect(request).toMatchObject({ method: 'orchestration.workerShow' }) + expect(request).not.toHaveProperty('orchestrationRequestId') + }) + it('allows a per-call timeout override for long runtime requests', async () => { const userDataPath = mkdtempSync(join(tmpdir(), 'orca-runtime-client-')) const endpoint = join(userDataPath, 'runtime.sock') diff --git a/src/cli/runtime/transport.test.ts b/src/cli/runtime/transport.test.ts index 8de35f619e5..4d713c842a4 100644 --- a/src/cli/runtime/transport.test.ts +++ b/src/cli/runtime/transport.test.ts @@ -132,7 +132,9 @@ describe.skipIf(process.platform === 'win32')('runtime transport', () => { // pre-fix behavior would hang the full duration and trip vitest's own limit. const start = Date.now() await expect(sendRequest(metadata, 'status.get', undefined, 60000)).rejects.toMatchObject({ - code: 'runtime_unavailable' + code: 'runtime_unavailable', + message: + 'The Orca runtime closed the connection before responding. Restart Orca and try again.' }) expect(Date.now() - start).toBeLessThan(5000) }) 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 efad7c761c3..2a1f166f6ba 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-computer-errors.test.ts b/src/main/runtime/rpc/dispatcher-computer-errors.test.ts index 862973e3913..bbc37e57ab1 100644 --- a/src/main/runtime/rpc/dispatcher-computer-errors.test.ts +++ b/src/main/runtime/rpc/dispatcher-computer-errors.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, it } from 'vitest' +import { describe, expect, it, vi } from 'vitest' import { z } from 'zod' import { RpcDispatcher } from './dispatcher' import { defineMethod, InvalidArgumentError, type RpcRequest } from './core' @@ -37,6 +37,20 @@ const METHODS = [ handler: () => { throw new InvalidArgumentError('Async validation rejected payload') } + }), + defineMethod({ + name: 'orchestration.inspectCaller', + params: z.object({}), + handler: (_params, { authenticatedCallerFingerprint }) => ({ + authenticatedCallerFingerprint + }) + }), + defineMethod({ + name: 'orchestration.federationInspectCaller', + params: z.object({}), + handler: (_params, { authenticatedCallerFingerprint }) => ({ + authenticatedCallerFingerprint + }) }) ] @@ -109,12 +123,16 @@ describe('RpcDispatcher computer-use validation errors', () => { }) }) - it('maps async validation errors to invalid_argument without shadowing Zod formatting', async () => { + it('maps async validation errors over streaming transport without shadowing Zod formatting', async () => { + const messages: string[] = [] const dispatcher = new RpcDispatcher({ runtime: makeRuntime(), methods: METHODS }) - const response = await dispatcher.dispatch(makeRequest('orchestration.invalidArgument', {})) + await dispatcher.dispatchStreaming( + makeRequest('orchestration.invalidArgument', {}), + (message) => messages.push(message) + ) - expect(response).toMatchObject({ + expect(JSON.parse(messages[0]!)).toMatchObject({ ok: false, error: { code: 'invalid_argument', @@ -122,4 +140,35 @@ describe('RpcDispatcher computer-use validation errors', () => { } }) }) + + it('forwards a paired caller fingerprint without requiring local orchestration state', async () => { + const dispatcher = new RpcDispatcher({ runtime: makeRuntime(), methods: METHODS }) + + const response = await dispatcher.dispatch(makeRequest('orchestration.inspectCaller', {}), { + authenticatedCallerFingerprint: 'paired-caller' + }) + + expect(response).toMatchObject({ + ok: true, + result: { authenticatedCallerFingerprint: 'paired-caller' } + }) + }) + + it('provides local identity to read-only federation authorization', async () => { + const getOrCreateLocalMutationCallerFingerprint = vi.fn(() => 'local-caller') + const runtime = Object.assign(makeRuntime(), { + getOrchestrationDb: () => ({ getOrCreateLocalMutationCallerFingerprint }) + }) + const dispatcher = new RpcDispatcher({ runtime, methods: METHODS }) + + const response = await dispatcher.dispatch( + makeRequest('orchestration.federationInspectCaller', {}) + ) + + expect(response).toMatchObject({ + ok: true, + result: { authenticatedCallerFingerprint: 'local-caller' } + }) + expect(getOrCreateLocalMutationCallerFingerprint).toHaveBeenCalledOnce() + }) }) 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..7e7ef60f6fb 100644 --- a/src/main/runtime/rpc/dispatcher.ts +++ b/src/main/runtime/rpc/dispatcher.ts @@ -10,12 +10,12 @@ import { } from './core' import type { FeatureInteractionId } from '../../../shared/feature-interactions' +import { isOrchestrationMutation } from '../../../shared/orchestration-rpc-contract' import { errorResponse, successResponse } from './errors' 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 +41,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 +92,11 @@ export class RpcDispatcher { request, compatibility.legacyCoordinatorAuthority ) + const authenticatedCallerFingerprint = + options?.authenticatedCallerFingerprint ?? + (needsLocalCallerFingerprint(request, effectiveParams) + ? this.orchestrationMutations.getLocalAuthenticatedCallerFingerprint() + : undefined) const invoke = (mutation?: DurableMutationInvocation) => { const legacyCoordinatorRunId = legacyCoordinator?.revalidate() return method.handler(effectiveParams, { @@ -97,7 +105,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 +120,7 @@ export class RpcDispatcher { request, effectiveParams, invoke, - legacyCoordinator?.mutationCallerFingerprint + legacyCoordinator?.mutationCallerFingerprint ?? authenticatedCallerFingerprint ) recordRuntimeFeatureInteraction( this.runtime, @@ -177,6 +185,11 @@ export class RpcDispatcher { request, compatibility.legacyCoordinatorAuthority ) + const authenticatedCallerFingerprint = + options?.authenticatedCallerFingerprint ?? + (needsLocalCallerFingerprint(request, effectiveParams) + ? this.orchestrationMutations.getLocalAuthenticatedCallerFingerprint() + : undefined) const invoke = (mutation?: DurableMutationInvocation) => { const legacyCoordinatorRunId = legacyCoordinator?.revalidate() return method.handler(effectiveParams, { @@ -190,7 +203,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 +221,7 @@ export class RpcDispatcher { request, effectiveParams, invoke, - legacyCoordinator?.mutationCallerFingerprint + legacyCoordinator?.mutationCallerFingerprint ?? authenticatedCallerFingerprint ) recordRuntimeFeatureInteraction( this.runtime, @@ -291,3 +304,10 @@ export class RpcDispatcher { return { runtimeId: this.runtime.getRuntimeId() } } } + +function needsLocalCallerFingerprint(request: RpcRequest, params: unknown): boolean { + return ( + request.method.startsWith('orchestration.federation') || + (!!request.orchestrationRequestId && isOrchestrationMutation(request.method, params)) + ) +} 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..43bc0bc96ff 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 { fingerprintAuthenticatedPairingCredential } from '../orchestration-mutation-executor' import { ORCHESTRATION_METHODS } from './orchestration' describe('orchestration federation control mail', () => { const homeToken = 'run-home-device-token' + const homeFingerprint = fingerprintAuthenticatedPairingCredential(homeToken) 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 b83ab5b087b..2410a74c10f 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,6 +582,101 @@ describe('orchestration new-worktree workers', () => { }) }) + 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') + ) + const task = db.createTask({ spec: 'recover dispatch input', runId }) + const dispatcher = new RpcDispatcher({ runtime, methods: ORCHESTRATION_METHODS }) + const request: RpcRequest = { + id: 'rpc_worker_start', + authToken: 'caller-token', + orchestrationContractVersion: ORCHESTRATION_CONTRACT_VERSION, + orchestrationRequestId: 'worker_start_request', + method: 'orchestration.workerStart', + params: { + task: task.id, + from: 'term_coord', + worktree: 'new-child', + name: 'recover-input-worker', + agent: 'codex' + } + } + + const first = await dispatcher.dispatch(request) + 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) + 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')) + 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', + params: { ...(request.params as Record), from: 'term_coord_reminted' } + }) + + expect(first).toMatchObject({ + ok: true, + result: { + state: 'failed', + failedStage: 'dispatch_input', + residualResources: expect.arrayContaining([ + expect.objectContaining({ kind: 'worktree', id: 'repo::created' }), + expect.objectContaining({ kind: 'terminal', id: 'term_worker' }) + ]), + mutation: { requestId: 'worker_start_request', replayed: false } + } + }) + 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 () => { mockCreatedWorktree({ hookFound: false }) let finishWait: 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..fde60ff2b60 100644 --- a/src/main/runtime/rpc/orchestration-mutation-executor.ts +++ b/src/main/runtime/rpc/orchestration-mutation-executor.ts @@ -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' @@ -29,9 +30,17 @@ 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 }))) + .update( + JSON.stringify( + canonicalize({ + method: request.method, + params: replayStableCallerParams(this.runtime, params) + }) + ) + ) .digest('hex') const key = `${callerFingerprint}:${requestId}` const db = this.runtime.getOrchestrationDb() @@ -109,6 +118,10 @@ export class OrchestrationMutationExecutor { this.inFlight.delete(key) } } + + getLocalAuthenticatedCallerFingerprint(): string { + return this.runtime.getOrchestrationDb().getOrCreateLocalMutationCallerFingerprint() + } } const executorsByRuntime = new WeakMap() @@ -125,12 +138,31 @@ 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 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 + 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 { 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..a8159384b16 100644 --- a/src/main/runtime/runtime-rpc.test.ts +++ b/src/main/runtime/runtime-rpc.test.ts @@ -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' @@ -3483,6 +3484,114 @@ 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('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[] = [] + + 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), + () => {}, + 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() diff --git a/src/main/runtime/runtime-rpc.ts b/src/main/runtime/runtime-rpc.ts index 326801c6fa9..14c69ceb960 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 { 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' @@ -1711,6 +1712,8 @@ export class OrcaRuntimeRpcServer { : undefined try { await this.dispatcher.dispatchStreaming(request, replyForRequest, { + // Why: the validated credential preserves existing federation ownership without trusting request fields. + authenticatedCallerFingerprint: fingerprintAuthenticatedPairingCredential(token), connectionId, clientId: token, pairedDeviceId: device.deviceId,