From 7e28f70b835776787e1be278eaead25c808cbd60 Mon Sep 17 00:00:00 2001 From: Jinwoo-H Date: Mon, 17 Aug 2026 00:29:12 -0400 Subject: [PATCH] fix(orchestration): close final invariant review gaps --- .../__snapshots__/preamble.test.ts.snap | 2 +- .../coordinator-decision-gates.test.ts | 22 +++++ .../coordinator-escalation-triage.test.ts | 61 ++++++++++++ .../coordinator-escalation-triage.ts | 11 +++ .../runtime/orchestration/coordinator.test.ts | 3 +- .../db-task-dispatch-lifecycle-guards.test.ts | 34 +++++++ src/main/runtime/orchestration/db.ts | 35 ++++++- .../runtime/orchestration/preamble.test.ts | 1 + src/main/runtime/orchestration/preamble.ts | 2 +- .../rpc/methods/orchestration-check.test.ts | 4 +- ...hestration-send-dispatch-authority.test.ts | 92 +++++++++++++++++++ src/main/runtime/rpc/methods/orchestration.ts | 89 ++++++++++++++---- 12 files changed, 328 insertions(+), 28 deletions(-) create mode 100644 src/main/runtime/orchestration/coordinator-escalation-triage.test.ts create mode 100644 src/main/runtime/rpc/methods/orchestration-send-dispatch-authority.test.ts diff --git a/src/main/runtime/orchestration/__snapshots__/preamble.test.ts.snap b/src/main/runtime/orchestration/__snapshots__/preamble.test.ts.snap index d51c640c72d..d92228517aa 100644 --- a/src/main/runtime/orchestration/__snapshots__/preamble.test.ts.snap +++ b/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: " \\ --body "
" \\ - --task-id task_SNAP + --task-id task_SNAP --dispatch-id ctx_SNAP # Check for messages from the coordinator: orca orchestration check --terminal term_WORKER diff --git a/src/main/runtime/orchestration/coordinator-decision-gates.test.ts b/src/main/runtime/orchestration/coordinator-decision-gates.test.ts index 6e3bd3f8df1..74326daa079 100644 --- a/src/main/runtime/orchestration/coordinator-decision-gates.test.ts +++ b/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') + }) }) diff --git a/src/main/runtime/orchestration/coordinator-escalation-triage.test.ts b/src/main/runtime/orchestration/coordinator-escalation-triage.test.ts new file mode 100644 index 00000000000..33172e3a699 --- /dev/null +++ b/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 + }) + }) +}) diff --git a/src/main/runtime/orchestration/coordinator-escalation-triage.ts b/src/main/runtime/orchestration/coordinator-escalation-triage.ts index 9b687449719..5cbfad7f2c9 100644 --- a/src/main/runtime/orchestration/coordinator-escalation-triage.ts +++ b/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)) { diff --git a/src/main/runtime/orchestration/coordinator.test.ts b/src/main/runtime/orchestration/coordinator.test.ts index 1ace48cb1c3..5bcbaf28ad4 100644 --- a/src/main/runtime/orchestration/coordinator.test.ts +++ b/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', diff --git a/src/main/runtime/orchestration/db-task-dispatch-lifecycle-guards.test.ts b/src/main/runtime/orchestration/db-task-dispatch-lifecycle-guards.test.ts index b4839b67aeb..02f2abba521 100644 --- a/src/main/runtime/orchestration/db-task-dispatch-lifecycle-guards.test.ts +++ b/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' }) diff --git a/src/main/runtime/orchestration/db.ts b/src/main/runtime/orchestration/db.ts index e4a9cbc2be0..173681e3e47 100644 --- a/src/main/runtime/orchestration/db.ts +++ b/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', diff --git a/src/main/runtime/orchestration/preamble.test.ts b/src/main/runtime/orchestration/preamble.test.ts index 657ab49a00b..3cedb9fa242 100644 --- a/src/main/runtime/orchestration/preamble.test.ts +++ b/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') }) diff --git a/src/main/runtime/orchestration/preamble.ts b/src/main/runtime/orchestration/preamble.ts index 27d36d7336d..d98f33763d2 100644 --- a/src/main/runtime/orchestration/preamble.ts +++ b/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: " \\ --body "
" \\ - --task-id ${params.taskId} + --task-id ${params.taskId} --dispatch-id ${params.dispatchId} # Check for messages from the coordinator: ${cli} orchestration check --terminal ${params.workerHandle} diff --git a/src/main/runtime/rpc/methods/orchestration-check.test.ts b/src/main/runtime/rpc/methods/orchestration-check.test.ts index 9738c374afe..ce45a2cfec4 100644 --- a/src/main/runtime/rpc/methods/orchestration-check.test.ts +++ b/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 () => { diff --git a/src/main/runtime/rpc/methods/orchestration-send-dispatch-authority.test.ts b/src/main/runtime/rpc/methods/orchestration-send-dispatch-authority.test.ts new file mode 100644 index 00000000000..484e29ad100 --- /dev/null +++ b/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) { + 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') + }) +}) diff --git a/src/main/runtime/rpc/methods/orchestration.ts b/src/main/runtime/rpc/methods/orchestration.ts index 6280921bc18..16a5099c6c8 100644 --- a/src/main/runtime/rpc/methods/orchestration.ts +++ b/src/main/runtime/rpc/methods/orchestration.ts @@ -95,6 +95,22 @@ function parseRemoteWorkerPayload(payload: string | undefined): Record