fix(orchestration): close final invariant review gaps

src/main/runtime/orchestration/__snapshots__/preamble.test.ts.snap

@@ -67,7 +67,7 @@ Slack, GitHub comments, or any other channel to reach a human during the run.
   orca orchestration send --from term_WORKER \\
     --type escalation --subject "Blocked: <reason>" \\
     --body "<details>" \\
-    --task-id task_SNAP
+    --task-id task_SNAP --dispatch-id ctx_SNAP
 
   # Check for messages from the coordinator:
   orca orchestration check --terminal term_WORKER

src/main/runtime/orchestration/coordinator-decision-gates.test.ts

@@ -66,4 +66,26 @@ describe('coordinator decision-gate authority', () => {
     expect(db.getDispatchContextById(victim.id)?.status).toBe('dispatched')
     expect(logs.at(-1)).toContain('Rejected decision gate from term_attacker')
   })
+
+  it('accepts the canonical sender of an imported federated Dispatch', () => {
+    db = new OrchestrationDb(':memory:')
+    const task = db.createTask({ spec: 'remote gate target' })
+    const dispatch = db.createDispatchContext(task.id, 'remote-worker')
+
+    openDecisionGateFromMessage(
+      db,
+      db.insertMessage({
+        from: `dispatch:${dispatch.id}`,
+        to: 'term_coordinator',
+        subject: 'Remote approval required',
+        type: 'decision_gate',
+        payload: JSON.stringify({ taskId: task.id, question: 'Proceed remotely?' })
+      }),
+      () => {}
+    )
+
+    expect(db.listGates({ taskId: task.id })).toHaveLength(1)
+    expect(db.getTask(task.id)?.status).toBe('blocked')
+    expect(db.getDispatchContextById(dispatch.id)?.status).toBe('completed')
+  })
 })

src/main/runtime/orchestration/coordinator-escalation-triage.test.ts

@@ -0,0 +1,61 @@
+import { afterEach, describe, expect, it } from 'vitest'
+import { applyEscalationToDispatch } from './coordinator-escalation-triage'
+import { OrchestrationDb } from './db'
+
+describe('coordinator escalation authority', () => {
+  let db: OrchestrationDb
+
+  afterEach(() => {
+    db?.close()
+  })
+
+  it('rejects an escalation targeting another active Dispatch', () => {
+    db = new OrchestrationDb(':memory:')
+    const attackerTask = db.createTask({ spec: 'attacker assignment' })
+    db.createDispatchContext(attackerTask.id, 'term_attacker', 'tab_attacker:leaf_attacker')
+    const victimTask = db.createTask({ spec: 'victim assignment' })
+    const victim = db.createDispatchContext(victimTask.id, 'term_victim')
+    const logs: string[] = []
+
+    applyEscalationToDispatch(
+      db,
+      db.insertMessage({
+        from: 'term_attacker',
+        to: 'term_coordinator',
+        subject: 'Fail the victim',
+        type: 'escalation',
+        senderPaneKey: 'tab_attacker:leaf_attacker',
+        payload: JSON.stringify({ taskId: victimTask.id })
+      }),
+      (message) => logs.push(message)
+    )
+
+    expect(db.getTask(victimTask.id)?.status).toBe('dispatched')
+    expect(db.getDispatchContextById(victim.id)?.status).toBe('dispatched')
+    expect(logs.at(-1)).toContain('Rejected escalation from term_attacker')
+  })
+
+  it('accepts the canonical sender of an imported federated Dispatch', () => {
+    db = new OrchestrationDb(':memory:')
+    const task = db.createTask({ spec: 'remote escalation target' })
+    const dispatch = db.createDispatchContext(task.id, 'remote-worker')
+
+    applyEscalationToDispatch(
+      db,
+      db.insertMessage({
+        from: `dispatch:${dispatch.id}`,
+        to: 'term_coordinator',
+        subject: 'Remote worker failed',
+        type: 'escalation',
+        payload: JSON.stringify({ taskId: task.id })
+      }),
+      () => {}
+    )
+
+    expect(db.getTask(task.id)?.status).toBe('ready')
+    expect(db.getDispatchContextById(dispatch.id)).toMatchObject({
+      status: 'failed',
+      failure_count: 1
+    })
+  })
+})

src/main/runtime/orchestration/coordinator-escalation-triage.ts

@@ -31,6 +31,17 @@ export function applyEscalationToDispatch(
   if (!dispatch) {
     return null
   }
+  if (
+    !db.isDispatchMessageSender({
+      dispatchId: dispatch.id,
+      handle: msg.from_handle,
+      paneKey: msg.sender_pane_key,
+      allowCanonicalDispatchHandle: true
+    })
+  ) {
+    onLog(`Rejected escalation from ${msg.from_handle}: it does not own Task ${taskId}`)
+    return null
+  }
 
   const worker = db.getWorkerDispatch(dispatch.id)
   if (worker && !['failed', 'succeeded', 'stopped', 'abandoned'].includes(worker.state)) {

src/main/runtime/orchestration/coordinator.test.ts

@@ -335,8 +335,9 @@ describe('Coordinator', () => {
       await new Promise((r) => {
         setTimeout(r, 100)
       })
+      const dispatch = db.getDispatchContext(task.id)
       db.insertMessage({
-        from: `term_${i === 0 ? 'a' : 'b'}`,
+        from: dispatch?.assignee_handle ?? 'missing-worker',
         to: 'coord',
         subject: `Failed attempt ${i + 1}`,
         type: 'escalation',

src/main/runtime/orchestration/db-task-dispatch-lifecycle-guards.test.ts

@@ -318,6 +318,40 @@ describe('Task/Dispatch lifecycle guards', () => {
     ).toEqual({ action: 'settled', outcome: 'succeeded', duplicate: false })
   })
 
+  it.each(['federated-reconcile', 'missing-terminal'] as const)(
+    'restores a live sibling after an uncertain worker start fails through %s',
+    (recovery) => {
+      const database = createDatabase()
+      const task = database.createTask({ spec: `${recovery} uncertain sibling` })
+      const live = startWorker(database, task.id, `${recovery}_live`)
+      sqliteFor(database).prepare("UPDATE tasks SET status = 'ready' WHERE id = ?").run(task.id)
+      const uncertain = database.createStartingWorkerDispatch({ taskId: task.id, startOptions: {} })
+      database.markWorkerStartUnknown(uncertain.dispatch.id, 'agent_readiness', 'outcome unknown')
+
+      if (recovery === 'federated-reconcile') {
+        database.reconcileFederatedWorkerStart({
+          dispatchId: uncertain.dispatch.id,
+          state: 'failed',
+          stage: 'start_failed',
+          lastError: 'worker did not start'
+        })
+      } else {
+        database.reconcileMissingWorkerTerminal(uncertain.dispatch.id, 'worker terminal missing')
+      }
+
+      expect(database.getTask(task.id)?.status).toBe('dispatched')
+      expect(database.getDispatchContextById(live.dispatchId)?.status).toBe('dispatched')
+      expect(
+        database.settleWorkerReport({
+          taskId: task.id,
+          dispatchId: live.dispatchId,
+          outcome: 'succeeded',
+          result: 'live sibling completed'
+        })
+      ).toEqual({ action: 'settled', outcome: 'succeeded', duplicate: false })
+    }
+  )
+
   it('rejects gate creation while a supervised worker remains active', () => {
     const database = createDatabase()
     const task = database.createTask({ spec: 'worker gate guard' })

src/main/runtime/orchestration/db.ts

@@ -5258,6 +5258,7 @@ export class OrchestrationDb {
              WHERE id = ? AND status IN ('pending', 'dispatched')`
           )
           .run(reason, params.dispatchId)
+        this.reconcileTaskAfterDispatchInterruption(dispatch.task_id, params.dispatchId)
         this.db
           .prepare(
             `UPDATE tasks SET status = 'failed', completed_at = datetime('now')
@@ -5328,6 +5329,7 @@ export class OrchestrationDb {
           .run(dispatchStatus, failureCount, reason, dispatchId)
         if (!stopWasPending) {
           const taskStatus: TaskStatus = dispatchStatus === 'circuit_broken' ? 'failed' : 'ready'
+          this.reconcileTaskAfterDispatchInterruption(dispatch.task_id, dispatchId)
           this.db
             .prepare(
               `UPDATE tasks
@@ -7478,6 +7480,29 @@ export class OrchestrationDb {
     return this.findActiveDispatchForAssignee(handle, paneKey)
   }
 
+  isDispatchMessageSender(params: {
+    dispatchId: string
+    handle: string
+    paneKey?: string | null
+    allowCanonicalDispatchHandle?: boolean
+  }): boolean {
+    const dispatch = this.getDispatchContextById(params.dispatchId)
+    if (!dispatch || !['pending', 'dispatched'].includes(dispatch.status)) {
+      return false
+    }
+    if (params.allowCanonicalDispatchHandle && params.handle === `dispatch:${dispatch.id}`) {
+      return true
+    }
+    if (
+      params.paneKey &&
+      dispatch.assignee_pane_key &&
+      isEquivalentPaneKey(dispatch.assignee_pane_key, params.paneKey)
+    ) {
+      return true
+    }
+    return params.handle === dispatch.assignee_handle && !params.paneKey
+  }
+
   private findActiveDispatchForAssignee(
     assigneeHandle: string,
     assigneePaneKey?: string
@@ -7818,10 +7843,12 @@ export class OrchestrationDb {
       if (
         gate.requester &&
         (!active ||
-          active.assignee_handle !== gate.requester.handle ||
-          (gate.requester.paneKey &&
-            active.assignee_pane_key &&
-            !isEquivalentPaneKey(active.assignee_pane_key, gate.requester.paneKey)))
+          !this.isDispatchMessageSender({
+            dispatchId: active.id,
+            handle: gate.requester.handle,
+            paneKey: gate.requester.paneKey,
+            allowCanonicalDispatchHandle: true
+          }))
       ) {
         throw new OrchestrationError(
           'consumer_fenced',

src/main/runtime/orchestration/preamble.test.ts

@@ -113,6 +113,7 @@ describe('buildDispatchPreamble', () => {
 
     expect(result).toMatch(/orchestration ask --from term_worker/)
     expect(result).toMatch(/orchestration send --from term_worker \\\n    --type escalation/)
+    expect(result).toContain('--task-id task_abc123 --dispatch-id ctx_def456')
     expect(result).toContain('orchestration check --terminal term_worker')
   })
 

src/main/runtime/orchestration/preamble.ts

@@ -123,7 +123,7 @@ Slack, GitHub comments, or any other channel to reach a human during the run.
   ${cli} orchestration send --from ${params.workerHandle}${capabilityFlag} \\
     --type escalation --subject "Blocked: <reason>" \\
     --body "<details>" \\
-    --task-id ${params.taskId}
+    --task-id ${params.taskId} --dispatch-id ${params.dispatchId}
 
   # Check for messages from the coordinator:
   ${cli} orchestration check --terminal ${params.workerHandle}

src/main/runtime/rpc/methods/orchestration-check.test.ts

@@ -686,13 +686,13 @@ describe('orchestration RPC methods', () => {
         from: 'worker',
         to: 'coord',
         subject: 'needs attention',
-        type: 'escalation',
+        type: 'question',
         run: activeRunId
       })
 
       const result = await waitPromise
       expect(result.count).toBe(1)
-      expect(result.messages[0].type).toBe('escalation')
+      expect(result.messages[0].type).toBe('question')
     })
 
     it('does not mark existing messages read when the check starts aborted', async () => {

src/main/runtime/rpc/methods/orchestration-send-dispatch-authority.test.ts

@@ -0,0 +1,92 @@
+import { afterEach, describe, expect, it, vi } from 'vitest'
+import type { RpcContext } from '../core'
+import type { OrchestrationDb } from '../../orchestration/db'
+import type { OrcaRuntimeService } from '../../orca-runtime'
+import { createOrchestrationRpcHarness } from './orchestration-rpc-test-harness'
+
+describe('orchestration.send Dispatch authority', () => {
+  const harness = createOrchestrationRpcHarness()
+  let db: OrchestrationDb
+  let runtime: OrcaRuntimeService
+  let ctx: RpcContext
+
+  function setup(): void {
+    ;({ db, runtime, ctx } = harness.setup())
+  }
+
+  async function send(params: Record<string, unknown>) {
+    return harness.call('orchestration.send', params, ctx)
+  }
+
+  afterEach(() => {
+    harness.cleanup()
+  })
+
+  it.each([false, true])(
+    'rejects cross-Task escalation with legacy authority=%s',
+    async (legacyAuthority) => {
+      setup()
+      const attackerTask = db.createTask({ spec: 'attacker assignment' })
+      const attacker = db.createDispatchContext(
+        attackerTask.id,
+        'term_attacker',
+        'tab_attacker:leaf_attacker',
+        undefined,
+        legacyAuthority ? undefined : 'runtime_test:term_attacker:1'
+      )
+      const victimTask = db.createTask({ spec: 'victim assignment' })
+      const victim = db.createDispatchContext(victimTask.id, 'term_victim')
+      vi.mocked(runtime.getTerminalPaneKey).mockImplementation((handle) =>
+        handle === 'term_attacker' ? 'tab_attacker:leaf_attacker' : harness.coordinatorPaneKey
+      )
+      if (!legacyAuthority) {
+        ctx = {
+          runtime,
+          orchestrationCapability: db.mintDispatchCapability({
+            dispatchId: attacker.id,
+            paneKey: 'tab_attacker:leaf_attacker',
+            processIncarnation: 'runtime_test:term_attacker:1'
+          })
+        }
+      }
+
+      const result = (await send({
+        from: 'term_attacker',
+        type: 'escalation',
+        subject: 'Fail the victim',
+        payload: JSON.stringify({ taskId: victimTask.id })
+      })) as { lifecycle: { action: string; code: string }; message: { type: string } }
+
+      expect(result.lifecycle).toMatchObject({
+        action: 'rejected',
+        code: 'task_dispatch_mismatch'
+      })
+      expect(result.message.type).toBe('status')
+      expect(db.getTask(attackerTask.id)?.status).toBe('dispatched')
+      expect(db.getTask(victimTask.id)?.status).toBe('dispatched')
+      expect(db.getDispatchContextById(attacker.id)?.status).toBe('dispatched')
+      expect(db.getDispatchContextById(victim.id)?.status).toBe('dispatched')
+    }
+  )
+
+  it('rejects a caller-spoofed canonical Dispatch sender', async () => {
+    setup()
+    const task = db.createTask({ spec: 'legacy victim assignment' })
+    const dispatch = db.createDispatchContext(task.id, 'term_victim')
+
+    const result = (await send({
+      from: `dispatch:${dispatch.id}`,
+      type: 'escalation',
+      subject: 'Spoof imported federation mail',
+      payload: JSON.stringify({ taskId: task.id, dispatchId: dispatch.id })
+    })) as { lifecycle: { action: string; code: string }; message: { type: string } }
+
+    expect(result.lifecycle).toMatchObject({
+      action: 'rejected',
+      code: 'sender_not_assignee'
+    })
+    expect(result.message.type).toBe('status')
+    expect(db.getTask(task.id)?.status).toBe('dispatched')
+    expect(db.getDispatchContextById(dispatch.id)?.status).toBe('dispatched')
+  })
+})

src/main/runtime/rpc/methods/orchestration.ts

@@ -95,6 +95,22 @@ function parseRemoteWorkerPayload(payload: string | undefined): Record<string, u
   }
 }
 
+function parseMessageTaskId(payload: string | undefined): string | undefined {
+  if (!payload) {
+    return undefined
+  }
+  try {
+    const parsed: unknown = JSON.parse(payload)
+    return parsed && typeof parsed === 'object' && !Array.isArray(parsed)
+      ? typeof (parsed as { taskId?: unknown }).taskId === 'string'
+        ? (parsed as { taskId: string }).taskId
+        : undefined
+      : undefined
+  } catch {
+    return undefined
+  }
+}
+
 function isWorkerReportOutcome(value: unknown): value is 'succeeded' | 'failed' {
   return value === 'succeeded' || value === 'failed'
 }
@@ -649,33 +665,68 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
           ? db.getDispatchContextById(routing.dispatchId)
           : undefined
         const dispatchMutationMessage = isDispatchMutationMessageType(msg.type)
-        if (
-          dispatchMutationMessage &&
-          dispatch &&
-          (dispatch.capability_hash || dispatch.process_incarnation)
-        ) {
+        if (dispatchMutationMessage) {
           const processIncarnation =
             attestedCaller?.processIncarnation ??
             runtime.getTerminalProcessIncarnation(from) ??
             undefined
-          const capabilityBacked = Boolean(dispatch.capability_hash)
-          const authority = capabilityBacked
-            ? db.verifyDispatchCapability({
+          const taskId = parseMessageTaskId(params.payload)
+          const capabilityBacked = Boolean(dispatch?.capability_hash)
+          const coordinatorMutation = msg.type === 'escalation' || msg.type === 'decision_gate'
+          let authority: {
+            valid: boolean
+            code: 'sender_not_assignee' | 'task_dispatch_mismatch' | 'dispatch_capability_invalid'
+            reason: string
+          }
+          if (!dispatch) {
+            authority = {
+              valid: !coordinatorMutation,
+              code: 'sender_not_assignee',
+              reason: 'No active Dispatch belongs to this message sender.'
+            }
+          } else if (coordinatorMutation && taskId && taskId !== dispatch.task_id) {
+            authority = {
+              valid: false,
+              code: 'task_dispatch_mismatch',
+              reason: `Task ${taskId} does not belong to Dispatch ${dispatch.id}.`
+            }
+          } else if (capabilityBacked) {
+            const capabilityAuthority = db.verifyDispatchCapability({
+              dispatchId: dispatch.id,
+              capability: orchestrationCapability,
+              paneKey: senderPaneKey,
+              processIncarnation
+            })
+            authority = {
+              valid: capabilityAuthority.valid,
+              code: 'dispatch_capability_invalid',
+              reason: capabilityAuthority.valid ? '' : capabilityAuthority.reason
+            }
+          } else if (dispatch.process_incarnation) {
+            authority = {
+              valid: db.isDispatchProcessCurrent({
                 dispatchId: dispatch.id,
-                capability: orchestrationCapability,
-                paneKey: senderPaneKey,
-                processIncarnation
-              })
-            : {
-                valid: db.isDispatchProcessCurrent({
+                paneKey: senderPaneKey ?? null,
+                processIncarnation: processIncarnation ?? null
+              }),
+              code: 'sender_not_assignee',
+              reason: `Dispatch ${dispatch.id} process incarnation is no longer current for its pane.`
+            }
+          } else {
+            authority = {
+              valid:
+                !coordinatorMutation ||
+                db.isDispatchMessageSender({
                   dispatchId: dispatch.id,
-                  paneKey: senderPaneKey ?? null,
-                  processIncarnation: processIncarnation ?? null
+                  handle: from,
+                  paneKey: senderPaneKey
                 }),
-                reason: `Dispatch ${dispatch.id} process incarnation is no longer current for its pane.`
-              }
+              code: 'sender_not_assignee',
+              reason: `Terminal ${from} does not own Dispatch ${dispatch.id}.`
+            }
+          }
           if (!authority.valid) {
-            const code = capabilityBacked ? 'dispatch_capability_invalid' : 'sender_not_assignee'
+            const code = authority.code
             const rejection =
               db.convertLifecycleMessageToRejection(msg.id, code, authority.reason) ?? msg
             runtime.notifyMessageArrived(rejection.to_handle, rejection.type)