fix(orchestration-cli): point dropped mutations at --retry-request (#14586)

* fix(orchestration-cli): guide dropped mutations to idempotent retry * test(orchestration-cli): preserve read-only drop message * fix(orchestration): harden mutation replay identity * fix(orchestration): preserve replay across remints * fix(orchestration): defer local mutation identity
2026-09-23 00:02:29 +00:00 · 2026-08-14 18:11:12 -07:00
parent 9cb04b5a35
commit 78d5920446
22 changed files with 637 additions and 83 deletions
@@ -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')
+  })
 })
@@ -78,6 +78,9 @@ export function printResult<TResult>(
 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) {
@@ -195,4 +195,47 @@ describe('orchestration gate commands carry caller identity', () => {
    expect(stderr).toContain('Pass --from <terminal-handle>')
    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<string, unknown> }
+    }
+    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' })
+      ])
+    })
+  })
 })
@@ -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<TResult>(
  options?: { timeoutMs?: number; orchestrationCapability?: string }
 ) {
  const requestId = getOptionalStringFlag(flags, 'retry-request')
-  if (!requestId) {
-    return options
+  const result = requestId
+    ? client.call<TResult>(method, params, { ...options, orchestrationRequestId: requestId })
+    : options
      ? client.call<TResult>(method, params, options)
      : client.call<TResult>(method, params)
-  }
-  return client.call<TResult>(method, params, {
-    ...options,
-    orchestrationRequestId: requestId
+  return result.catch((error) => {
+    throw orchestrationMutationRecoveryError(error)
  })
 }

@@ -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<string, unknown> | undefined {
+  return value !== null && typeof value === 'object'
+    ? (value as Record<string, unknown>)
+    : 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}.`, '')
+}
@@ -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<string, unknown> | 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<string, unknown>
+        socket.end()
+      })
+    })
+    servers.add(server)
+    await new Promise<void>((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')
@@ -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)
  })
@@ -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 <text> [--from <handle>] [--json]',
+    usage:
+      'orca orchestration run-create --objective <text> [--from <handle>] [--retry-request <id>] [--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 <run_id> [--from <handle>] [--takeover-legacy] [--json]',
+      'orca orchestration run-use --id <run_id> [--from <handle>] [--takeover-legacy] [--retry-request <id>] [--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 <text> [--to <run:id|dispatch:id|legacy_handle>] [--run <run_id>] [--from <handle>] [--body <text>] [--type <type>] [--priority <level>] [--thread-id <id>] [--payload <json>] [--task-id <id>] [--dispatch-id <id>] [--outcome <succeeded|failed>] [--files-modified <csv>] [--report-path <path>] [--phase <text>] [--json]',
+      'orca orchestration send --subject <text> [--to <run:id|dispatch:id|legacy_handle>] [--run <run_id>] [--from <handle>] [--body <text>] [--type <type>] [--priority <level>] [--thread-id <id>] [--payload <json>] [--task-id <id>] [--dispatch-id <id>] [--outcome <succeeded|failed>] [--files-modified <csv>] [--report-path <path>] [--phase <text>] [--retry-request <id>] [--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 <handle>] [--run <run_id>] [--ack <delivery_id>] [--unread | --peek | --all] [--types <type,...>] [--format] [--wait] [--timeout-ms <n>] [--json]\n' +
+      'orca orchestration check [--terminal <handle>] [--run <run_id>] [--ack <delivery_id>] [--unread | --peek | --all] [--types <type,...>] [--format] [--wait] [--timeout-ms <n>] [--retry-request <id>] [--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 <msg_id> --body <text> [--run <run_id>] [--from <handle>] [--json]',
+      'orca orchestration reply --id <msg_id> --body <text> [--run <run_id>] [--from <handle>] [--retry-request <id>] [--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 <text> [--task-title <text>] [--display-name <text>] [--deps <json_array>] [--parent <task_id>] [--run <run_id>] [--from <handle>] [--json]',
+      'orca orchestration task-create --spec <text> [--task-title <text>] [--display-name <text>] [--deps <json_array>] [--parent <task_id>] [--run <run_id>] [--from <handle>] [--retry-request <id>] [--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 <task_id> --status <status> [--result <json>] [--run <run_id>] [--from <handle>] [--json]',
+      'orca orchestration task-update --id <task_id> --status <status> [--result <json>] [--run <run_id>] [--from <handle>] [--retry-request <id>] [--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 <task_id> --to <handle> [--from <handle>] [--run <run_id>] [--inject] [--dry-run] [--return-preamble] [--json]',
+      'orca orchestration dispatch --task <task_id> --to <handle> [--from <handle>] [--run <run_id>] [--inject] [--dry-run] [--return-preamble] [--retry-request <id>] [--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 <text> | --resume <message_id>) [--to <run:id>] [--run <run_id>] [--options <csv>] [--timeout-ms <n>] [--from <handle>] [--json]',
+      'orca orchestration ask (--question <text> | --resume <message_id>) [--to <run:id>] [--run <run_id>] [--options <csv>] [--timeout-ms <n>] [--from <handle>] [--retry-request <id>] [--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 <task_id> --question <text> [--options <json_array>] [--from <handle>] [--json]',
+      'orca orchestration gate-create --task <task_id> --question <text> [--options <json_array>] [--from <handle>] [--retry-request <id>] [--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 <gate_id> --resolution <text> [--from <handle>] [--json]',
+      'orca orchestration gate-resolve --id <gate_id> --resolution <text> [--from <handle>] [--retry-request <id>] [--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 <id>] [--json]',
    allowedFlags: [...GLOBAL_FLAGS, 'all', 'tasks', 'messages', 'retry-request']
  }
 ]
@@ -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
@@ -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
@@ -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
    })
@@ -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')
  })
 })
@@ -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()
+  })
 })
@@ -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
@@ -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<RpcResponse> {
+  async dispatch(
+    request: RpcRequest,
+    options?: { signal?: AbortSignal; authenticatedCallerFingerprint?: string }
+  ): Promise<RpcResponse> {
    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))
+  )
+}
@@ -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<unknown>
+        const response = (await workerDispatcher.dispatch(
+          {
+            id: `remote_${method}`,
+            authToken: homeToken,
+            method,
+            params,
+            orchestrationContractVersion: envelope?.orchestrationContractVersion,
+            orchestrationRequestId: envelope?.orchestrationRequestId
+          },
+          { authenticatedCallerFingerprint: homeFingerprint }
+        )) as RuntimeRpcResponse<unknown>
        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
+    })
+  }
 })
@@ -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<string, unknown> = {}) {
    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<string, unknown>), 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:
@@ -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()
    }
  )
@@ -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<OrcaRuntimeService, OrchestrationMutationExecutor>()
@@ -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<string, unknown>
+  const result = { ...source }
+  for (const property of ['from', 'callerTerminalHandle'] as const) {
+    const handle = source[property]
+    if (typeof handle !== 'string') {
+      continue
+    }
+    const paneKey =
+      property === 'from' && typeof source.senderPaneKey === 'string'
+        ? source.senderPaneKey
+        : runtime.getTerminalPaneKey(handle)
+    if (paneKey) {
+      const leafId = parsePaneKey(paneKey)?.leafId
+      result[property] = leafId ? { paneLeafId: leafId } : { paneKey }
+    }
+  }
+  return result
 }

 function canonicalize(value: unknown): unknown {
@@ -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')
@@ -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<string, unknown>[] = []
+      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<string, unknown>),
+        () => {},
+        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<string, unknown>[] = []
+
+    try {
+      await server['handleWebSocketMessage'](
+        JSON.stringify(
+          withCurrentOrchestrationContract({
+            id: 'show-existing-remote',
+            method: 'orchestration.federationShow',
+            params: { dispatchId: 'ctx_existing_remote' }
+          })
+        ),
+        (response) => replies.push(JSON.parse(response) as Record<string, unknown>),
+        () => {},
+        undefined,
+        undefined,
+        device.token
+      )
+
+      expect(replies[0]).toMatchObject({
+        ok: true,
+        result: { dispatchId: 'ctx_existing_remote', attachment: { state: 'starting' } }
+      })
+    } finally {
+      db.close()
+      await server.stop()
+    }
+  })
+
  it('rejects unpaired terminal creates before runtime dispatch', async () => {
    const userDataPath = mkdtempSync(join(tmpdir(), 'orca-runtime-rpc-'))
    const createMobileSessionTerminal = vi.fn()
@@ -15,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,