Fix orchestration CLI recovery, settled-Dispatch mail, and guide defects

Five reported orchestration CLI defects, verified individually before fixing.
Two were real code defects, one was a docs error, one was correct as-is, and
one was correct on both ends except for its recovery wording.

- Mail addressed to a settled `dispatch:<id>` was accepted and silently dropped.
  Local sends bypassed the settlement check the federated branch already had, so
  the caller was told success for a delivery no worker would ever read. Reject
  with `dispatch_inactive` and name the Run mailbox to use instead.

- A lost mutation response offered no read-only way to ask whether it took
  effect. `--retry-request` does dedupe correctly, but the recovery guidance
  emitted a query command only when the payload carried a dispatch id, which is
  exactly what a lost response lacks. Add read-only
  `orca orchestration request-show --request <id>` over the durable receipt
  ledger, and always emit a read-only step before the keyed retry.

- The bundled `orca-cli` guide documented `check --unread --inject`, a flag the
  parser rejects. Correct it to `--format` and add a ratchet that runs every
  orchestration invocation in the bundled guides through the real CLI parser.

- `check --json` is one stdout document and its keepalives are stderr-only; the
  reported `Extra data: line 2` came from merging the streams. Document the
  contract rather than changing the wire.

- A rejected lifecycle message is loud on both ends already, but the rejection
  never named the flag that supplies the missing capability. Name it.
This commit is contained in:
Brennan Benson
2026-08-27 18:23:48 -07:00
parent 39ac32f279
commit a9ef923053
20 changed files with 658 additions and 19 deletions
+1 -1
View File
@@ -199,7 +199,7 @@ Terminal rules:
- `terminal list --json` omits `visualLayouts` to keep the common agent payload bounded. Add `--include-visual-layouts` only when tab and pane topology is required.
- Use `terminal read` before `terminal send` unless the next input is obvious.
- Use `terminal send` only for direct terminal input or one-off prompts where no task state, inbox, or reply tracking is needed.
- For structured coordination, invoke the `orchestration` skill; it uses `orca orchestration ...` commands for messages, handoffs, task DAGs, dispatches, inbox/reply flows, and coordinator loops. A receiving agent can run `orca orchestration check --unread --inject` to render its unread mail in agent-readable form; this checks the caller's inbox and does not remotely deliver input to another terminal.
- For structured coordination, invoke the `orchestration` skill; it uses `orca orchestration ...` commands for messages, handoffs, task DAGs, dispatches, inbox/reply flows, and coordinator loops. A receiving agent can run `orca orchestration check --unread --format` to render its unread mail in agent-readable form; this checks the caller's inbox and does not remotely deliver input to another terminal.
- Use `terminal create --worktree active --command "<agent>"` for a fresh agent in the current worktree. Use `worktree create --agent <agent>` only for a separate checkout (agent in the first terminal — do not also `terminal create` the same agent).
- Use `terminal wait --for tui-idle` for agent CLIs such as Claude Code, Gemini, Codex, OMP, Pi, and Grok; always pass `--timeout-ms`.
- Terminal handles are runtime-scoped. Use `startupTerminal.handle` as the sole agent handle when `worktree create --agent` returns it; if Orca restarts, omits the handle, or returns `terminal_handle_stale`, reacquire with `terminal list` and continue with the replacement only.
+2
View File
@@ -143,6 +143,7 @@ Rules:
- `terminal list --json` omits `visualLayouts` because handle recovery does not need topology. Add `--include-visual-layouts` only for explicit tab and pane inspection.
- `orca orchestration check --peek --format --json` returns locally formatted unread mail without consuming it; it never writes to terminal input or remotely wakes another terminal. Use `orchestration dispatch --inject` to deliver a tracked task, or `terminal send` when an existing agent needs a free-form prompt.
- While supervising workers manually, use `check --wait --types worker_done,escalation,question --timeout-ms <n>` instead of sleep/poll loops. Process the whole Delivery, reply to `question` messages with `orca orchestration reply --id <msg_id> --body <answer> --json`, then acknowledge and keep waiting.
- `check --json` prints exactly one JSON document on stdout. While `--wait` blocks it also prints keepalive lines (`{"_keepalive":true,...}`) to stderr so you can tell the process is alive; those are never on stdout. Do not merge the streams before a parser — `check --wait --json 2>&1 | <parser>` fails with "Extra data: line 2". Pipe stdout only, or add `2>/dev/null`.
- Treat a `check --wait` timeout or `{count:0}` as a checkpoint, not a worker failure. Long coding tasks routinely run 15-60 minutes; keep using rolling waits unless you receive `worker_done`/`escalation`, the terminal exits or disappears, or the user explicitly asks you to stop.
- Heartbeats and visible terminal activity mean the worker is alive, not done. Do not stop, close, kill, or restart a worker just because it has not produced a completion message yet.
- Use `ask` when a worker needs a blocking answer from the coordinator; it defaults to the active Dispatch's Run. Timeout or disconnect leaves the question pending, so resume by its original message ID instead of asking again.
@@ -290,6 +291,7 @@ orca orchestration reply --id <message_id> --body "<answer>" --json
Recovery is conditional, never a fixed destructive sequence:
- The response was lost and named no Dispatch: run `orca orchestration request-show --request <request_id> --json` first. It is read-only. `completed` means the mutation already took effect and `pending` means Orca restarted mid-mutation; in both cases replay the original command with `--retry-request <request_id>`, which returns the recorded outcome instead of starting a second operation. `absent` means this runtime holds no receipt under your caller identity and is not proof that nothing happened.
- `worker-show --dispatch <id>` says `ready`: keep waiting or read bounded output.
- It proves `failed` or `stopped`: start a replacement with `worker-start --task <task> --retry-of <id>` plus an explicit `--on`/`--worktree` and `--agent`/`--terminal` choice. Retry does not silently inherit placement.
- It remains `outcome_unknown`: either `worker-stop --dispatch <id>` and inspect again, or explicitly `worker-abandon --dispatch <id>` while accepting that resources may still be live. Abandon performs no remote, process, or filesystem action.
File diff suppressed because one or more lines are too long
+1
View File
@@ -128,6 +128,7 @@ export const HANDLER_GROUPS: readonly HandlerGroup[] = [
'orchestration dispatch-show',
'orchestration coordinator-start',
'orchestration coordinator-stop',
'orchestration request-show',
'orchestration gate-create',
'orchestration gate-resolve',
'orchestration gate-list',
@@ -0,0 +1,68 @@
// `request-show` is the recovery path a lost mutation response sends you down, so it must
// stay read-only, render the honest reading, and name a version gap instead of a raw RPC error.
import { beforeEach, describe, expect, it, vi } from 'vitest'
const callMock = vi.fn()
vi.mock('../format', () => ({ printResult: vi.fn() }))
vi.mock('../selectors', () => ({ getTerminalHandle: vi.fn() }))
import { ORCHESTRATION_HANDLERS } from './orchestration'
import { printResult } from '../format'
import { RuntimeClientError } from '../runtime-client'
async function runRequestShow(): Promise<void> {
await ORCHESTRATION_HANDLERS['orchestration request-show']({
flags: new Map<string, string | boolean>([['request', 'request_1']]),
client: { call: callMock },
cwd: '/tmp/repo',
json: false
} as never)
}
describe('orchestration request-show', () => {
beforeEach(() => {
callMock.mockReset()
vi.mocked(printResult).mockReset()
})
it('asks the runtime without sending a mutation request id', async () => {
callMock.mockResolvedValue({ requestId: 'request_1', state: 'absent', interpretation: 'none' })
await runRequestShow()
expect(callMock).toHaveBeenCalledWith('orchestration.requestShow', { request: 'request_1' })
})
it('renders the state and the honest reading of it', async () => {
const result = {
requestId: 'request_1',
state: 'completed',
method: 'orchestration.workerStart',
interpretation: 'Request request_1 already took effect.'
}
callMock.mockResolvedValue(result)
await runRequestShow()
const [value, , render] = vi.mocked(printResult).mock.calls[0] as [
unknown,
boolean,
(value: unknown) => string
]
expect(render(value)).toBe(
'request_1 [completed] orchestration.workerStart\nRequest request_1 already took effect.'
)
})
it('names the version gap when the server predates the command', async () => {
callMock.mockRejectedValue(
new RuntimeClientError('method_not_found', 'Unknown method: orchestration.requestShow')
)
await expect(runRequestShow()).rejects.toMatchObject({
code: 'incompatible_runtime',
message: expect.stringContaining('Update Orca on the server')
})
})
})
+2
View File
@@ -7,6 +7,7 @@ import {
import { ORCHESTRATION_GATE_HANDLERS } from './orchestration/gate-handlers'
import { ORCHESTRATION_INBOX_HANDLERS } from './orchestration/message-inbox-handlers'
import { ORCHESTRATION_QUESTION_HANDLER } from './orchestration/question-handler'
import { ORCHESTRATION_REQUEST_SHOW_HANDLER } from './orchestration/mutation-request-show-handler'
import { ORCHESTRATION_RESET_HANDLER } from './orchestration/reset-handler'
import { ORCHESTRATION_RUN_HANDLERS } from './orchestration/run-handlers'
import { ORCHESTRATION_SEND_HANDLER } from './orchestration/message-send-handler'
@@ -27,6 +28,7 @@ export const ORCHESTRATION_HANDLERS: Record<string, CommandHandler> = {
...ORCHESTRATION_DISPATCH_HANDLER,
...ORCHESTRATION_QUESTION_HANDLER,
...ORCHESTRATION_DISPATCH_INSPECTION_HANDLERS,
...ORCHESTRATION_REQUEST_SHOW_HANDLER,
...ORCHESTRATION_GATE_HANDLERS,
...ORCHESTRATION_RESET_HANDLER
}
@@ -0,0 +1,30 @@
import type { CommandHandler } from '../../dispatch'
import { printResult } from '../../format'
import { getRequiredStringFlag } from '../../flags'
import { RuntimeClientError } from '../../runtime-client'
import type { OrchestrationMutationRequestShowResult } from '../../../shared/orchestration-mutation-request'
export const ORCHESTRATION_REQUEST_SHOW_HANDLER: Record<string, CommandHandler> = {
'orchestration request-show': async ({ flags, client, json }) => {
const request = getRequiredStringFlag(flags, 'request')
const result = await client
.call<OrchestrationMutationRequestShowResult>('orchestration.requestShow', { request })
.catch((error: unknown) => {
// Why: an Orca server older than request-show answers method_not_found, which reads
// as a bug rather than a version gap on the very path a lost response sends you down.
if (error instanceof RuntimeClientError && error.code === 'method_not_found') {
throw new RuntimeClientError(
'incompatible_runtime',
'This Orca server cannot look up orchestration mutation requests yet. Update Orca on the server, or inspect the Dispatch directly with orchestration worker-show.'
)
}
throw error
})
printResult(
result,
json,
(value) =>
`${value.requestId} [${value.state}]${value.method ? ` ${value.method}` : ''}\n${value.interpretation}`
)
}
}
@@ -45,7 +45,7 @@ describe('orchestration mutation recovery', () => {
)
expect((result.data as { nextSteps?: string[] }).nextSteps).toEqual([
'Run orca orchestration worker-show --dispatch dispatch_1 --json before retrying.',
'Run orca orchestration worker-start --task task_1 --retry-request request_1.'
'Run orca orchestration worker-start --task task_1 --retry-request request_1. --retry-request replays the recorded outcome instead of starting a second operation.'
])
})
@@ -68,6 +68,37 @@ describe('orchestration mutation recovery', () => {
expect(result.message).not.toContain('worker death')
})
it('offers a read-only request lookup when the response carried no dispatch', () => {
const result = orchestrationMutationRecoveryError(
new RuntimeClientError('runtime_unavailable', 'runtime unavailable', {
orchestrationRequestId: 'request_4',
originalCommand: ['orca', 'orchestration', 'worker-start', '--task', 'task_4']
})
) as RuntimeClientError
expect(result.data).toMatchObject({
recovery: {
queryCommand: ['orca', 'orchestration', 'request-show', '--request', 'request_4', '--json']
}
})
expect((result.data as { nextSteps?: string[] }).nextSteps?.[0]).toBe(
'Run orca orchestration request-show --request request_4 --json before retrying.'
)
})
it('says the keyed retry replays instead of starting a second operation', () => {
const result = orchestrationMutationRecoveryError(
new RuntimeClientError('runtime_timeout', 'request timed out', {
orchestrationRequestId: 'request_5',
originalCommand: ['orca', 'orchestration', 'worker-start', '--task', 'task_5']
})
) as RuntimeClientError
expect((result.data as { nextSteps?: string[] }).nextSteps?.[1]).toContain(
'replays the recorded outcome instead of starting a second operation'
)
})
it('renders the exact executable and safely quotes original arguments', () => {
const result = orchestrationMutationRecoveryError(
new RuntimeClientError('runtime_timeout', 'request timed out', {
@@ -87,7 +118,7 @@ describe('orchestration mutation recovery', () => {
expect((result.data as { nextSteps?: string[] }).nextSteps).toEqual([
'Run orca-dev orchestration worker-show --dispatch dispatch_3 --json before retrying.',
"Run orca-dev orchestration worker-start --task 'task 3' --comment 'literal $(do-not-run)' --retry-request request_3."
"Run orca-dev orchestration worker-start --task 'task 3' --comment 'literal $(do-not-run)' --retry-request request_3. --retry-request replays the recorded outcome instead of starting a second operation."
])
expect(result.message).toContain("'literal $(do-not-run)'")
})
+6 -6
View File
@@ -28,24 +28,24 @@ export function orchestrationMutationRecoveryError(error: unknown): unknown {
? [...originalCommand, '--retry-request', requestId]
: undefined
const executable = originalCommand?.[0] ?? resolveOrchestrationCliExecutable()
// Why: a lost response usually carries no dispatch id, and the old code then emitted
// no read-only step at all — leaving retry as the only way to learn what happened.
const queryCommand = dispatchId
? [executable, 'orchestration', 'worker-show', '--dispatch', dispatchId, '--json']
: undefined
: [executable, 'orchestration', 'request-show', '--request', requestId, '--json']
const recovery = {
orchestrationRequestId: requestId,
...(dispatchId ? { dispatchId } : {}),
...(queryCommand ? { queryCommand } : {}),
queryCommand,
...(retryCommand ? { retryCommand } : {}),
recoveryBlocked: !retryCommand,
disposition: 'outcome_unknown',
workerDeathInferred: false
}
const retryStep = retryCommand
? `Run ${renderCommand(retryCommand)}.`
? `Run ${renderCommand(retryCommand)}. --retry-request replays the recorded outcome instead of starting a second operation.`
: 'Recovery is blocked until the exact original command is available; no retry command was emitted.'
const nextSteps = queryCommand
? [`Run ${renderCommand(queryCommand)} before retrying.`, retryStep]
: [retryStep]
const nextSteps = [`Run ${renderCommand(queryCommand)} before retrying.`, retryStep]
const message = [
stripUnsafeRetryAdvice(error.message, requestId),
'The orchestration mutation may already have taken effect; do not assume it failed.',
+82
View File
@@ -0,0 +1,82 @@
import { describe, expect, it } from 'vitest'
import { BUNDLED_SKILL_GUIDES } from '../bundled-skill-guides'
import { validateCommandAndFlags, type CommandSpec } from '../args'
import { COMMAND_SPECS } from './index'
type GuideInvocation = {
guide: string
command: string
flag: string
snippet: string
}
// Why: guides write examples as `orca ...`, `orca-dev ...`, or the `ORCA` placeholder.
const CLI_INVOCATION = /(?:^|[\s`(])(?:orca|orca-dev|orca-ide|ORCA)\s+([^\n`]*)/g
// Longest path first so `orchestration worker-start` never resolves as `orchestration worker`.
const SPECS_BY_DEPTH: CommandSpec[] = [...COMMAND_SPECS].sort(
(a, b) => b.path.length - a.path.length
)
function findSpecForInvocation(invocation: string): CommandSpec | undefined {
return SPECS_BY_DEPTH.find((spec) => {
const prefix = `${spec.path.join(' ')} `
return invocation === spec.path.join(' ') || invocation.startsWith(prefix)
})
}
function collectGuideInvocations(): GuideInvocation[] {
const found: GuideInvocation[] = []
for (const guide of BUNDLED_SKILL_GUIDES) {
for (const match of guide.fullMarkdown.matchAll(CLI_INVOCATION)) {
const invocation = match[1].trim()
const spec = findSpecForInvocation(invocation)
if (!spec) {
continue
}
// Why: a quoted flag value belongs to the nested program (`--command 'codex --model ...'`), not to orca.
const orcaArgs = invocation.replace(/'[^']*'|"[^"]*"/g, ' ')
const flags = [...orcaArgs.matchAll(/(?:^|[\s[(])--([a-z][a-z0-9-]*)/g)].map(
(flag) => flag[1]
)
for (const flag of flags) {
found.push({
guide: guide.name,
command: spec.path.join(' '),
flag,
snippet: invocation.slice(0, 120)
})
}
}
}
return found
}
describe('bundled skill guides', () => {
const invocations = collectGuideInvocations()
it('scans the shipped guides for orchestration examples', () => {
// Why: the flag ratchet below is vacuous if the scanner stops matching guide prose.
expect(
invocations.filter((invocation) => invocation.command.startsWith('orchestration ')).length
).toBeGreaterThan(20)
})
// Why: the guides ship inside the binary, so an example the parser rejects is a shipped defect.
it('only documents flags the CLI parser accepts', () => {
const rejected = invocations.filter((invocation) => {
try {
validateCommandAndFlags(COMMAND_SPECS, {
commandPath: invocation.command.split(' '),
flags: new Map([[invocation.flag, 'placeholder']])
})
return false
} catch {
return true
}
})
expect(
rejected.map((entry) => `${entry.guide}: ${entry.command} --${entry.flag} (${entry.snippet})`)
).toEqual([])
})
})
+11
View File
@@ -175,6 +175,17 @@ export const ORCHESTRATION_COMMAND_SPECS: CommandSpec[] = [
'retry-request'
]
},
{
path: ['orchestration', 'request-show'],
summary: 'Ask whether one orchestration mutation request already took effect',
usage: 'orca orchestration request-show --request <request_id> [--json]',
allowedFlags: [...GLOBAL_FLAGS, 'request'],
notes: [
'Read-only: it never starts, retries, or settles anything, so it is safe to run after any lost response.',
'completed means the mutation landed and --retry-request replays the recorded outcome instead of starting a second one. pending means Orca restarted mid-mutation; replay with --retry-request.',
'absent means this runtime holds no receipt for that request under your caller identity: it never arrived, it failed before recording anything, or the receipt was pruned. Absent is not proof that nothing happened.'
]
},
{
path: ['orchestration', 'dispatch-show'],
summary: 'Show dispatch context for a task',
@@ -0,0 +1,60 @@
import { afterEach, beforeEach, describe, expect, it } from 'vitest'
import { OrchestrationDb } from '../orchestration-db'
import { createRootDispatch } from '../root-dispatch-test-fixture'
describe('verifyDispatchCapability', () => {
let db: OrchestrationDb
beforeEach(() => {
db = new OrchestrationDb(':memory:')
})
afterEach(() => {
db.close()
})
function mintedWorker() {
const run = db.createRun({
objective: 'capability guidance',
coordinatorHandle: 'term_coord',
coordinatorPaneKey: 'tab_coord:aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa'
})
const task = db.createTask({ spec: 'work', runId: run.id })
const paneKey = 'tab_worker:bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb'
const dispatch = createRootDispatch(db, task.id, 'term_worker', paneKey)
const capability = db.mintDispatchCapability({
dispatchId: dispatch.id,
paneKey,
processIncarnation: 'worker:1'
})
return { dispatchId: dispatch.id, paneKey, capability }
}
// Why: the caller that hits this is an agent reading only the error string, so it must name the flag.
it('names the flag that supplies a missing capability', () => {
const worker = mintedWorker()
const result = db.verifyDispatchCapability({
dispatchId: worker.dispatchId,
capability: undefined,
paneKey: worker.paneKey,
processIncarnation: 'worker:1'
})
expect(result).toMatchObject({ valid: false })
expect(result.valid === false && result.reason).toContain('--dispatch-capability')
})
it('still accepts the minted capability', () => {
const worker = mintedWorker()
expect(
db.verifyDispatchCapability({
dispatchId: worker.dispatchId,
capability: worker.capability,
paneKey: worker.paneKey,
processIncarnation: 'worker:1'
})
).toEqual({ valid: true })
})
})
@@ -56,7 +56,12 @@ export function verifyDispatchCapability(
return { valid: false, reason: `Dispatch ${params.dispatchId} capability is revoked.` }
}
if (!params.capability) {
return { valid: false, reason: 'The Dispatch capability is missing.' }
// Why: a worker that omits the flag needs the flag name, not just the diagnosis.
return {
valid: false,
reason:
'The Dispatch capability is missing. Pass --dispatch-capability <token> from your dispatch preamble.'
}
}
const expected = Buffer.from(dispatch.capability_hash, 'hex')
const observed = Buffer.from(hashDispatchCapability(params.capability), 'hex')
@@ -0,0 +1,61 @@
import {
describeMutationRequestState,
type OrchestrationMutationRequestShowResult
} from '../../../../shared/orchestration-mutation-request'
import { defineMethod, type RpcMethod } from '../core'
import { requiredString } from '../schemas'
import { z } from 'zod'
const RequestShowParams = z.object({ request: requiredString('Missing --request') })
export const ORCHESTRATION_MUTATION_REQUEST_METHODS: RpcMethod[] = [
defineMethod({
name: 'orchestration.requestShow',
params: RequestShowParams,
// Why: recovery needs a way to ask whether a mutation landed without mutating
// again; this reads the durable receipt and never writes one, so it must stay
// out of ORCHESTRATION_MUTATION_METHODS.
handler: (
params,
{ runtime, authenticatedCallerFingerprint }
): OrchestrationMutationRequestShowResult => {
const db = runtime.getOrchestrationDb()
// Why: receipts are keyed by caller identity; a paired client brings its own
// fingerprint, a local caller shares the one its mutations were recorded under.
const callerFingerprint =
authenticatedCallerFingerprint ?? db.getOrCreateLocalMutationCallerFingerprint()
const row = db.getMutationReceipt(callerFingerprint, params.request)
if (!row) {
return {
requestId: params.request,
state: 'absent',
interpretation: describeMutationRequestState({
requestId: params.request,
state: 'absent'
})
}
}
return {
requestId: params.request,
state: row.state,
method: row.method,
createdAt: row.created_at,
updatedAt: row.updated_at,
...(row.receipt ? { receipt: parseReceipt(row.receipt) } : {}),
interpretation: describeMutationRequestState({
requestId: params.request,
state: row.state,
method: row.method
})
}
}
})
]
function parseReceipt(receipt: string): unknown {
try {
return JSON.parse(receipt)
} catch {
return undefined
}
}
@@ -1,7 +1,28 @@
import type { LegacyAdoptedMailboxOwner, OrchestrationDb } from '../../orchestration/db'
import type { DispatchContextRow } from '../../orchestration/types'
import { OrchestrationError } from '../../orchestration/orchestration-error'
import type { DispatchContextRow, DispatchStatus } from '../../orchestration/types'
import type { OrcaRuntimeService } from '../../orca-runtime'
const ACTIVE_DISPATCH_STATUSES: readonly DispatchStatus[] = ['pending', 'dispatched']
/**
* Reject mail addressed to a Dispatch mailbox nobody will ever read again.
*
* Why: a settled Dispatch has no active reader and `check` never reroutes its
* mailbox, so accepting the message reports success for a delivery that cannot
* happen. Federated targets keep their own liveness check.
*/
export function assertDispatchMailboxDeliverable(db: OrchestrationDb, dispatchId: string): void {
const dispatch = db.getDispatchContextById(dispatchId)
if (!dispatch || ACTIVE_DISPATCH_STATUSES.includes(dispatch.status)) {
return
}
throw new OrchestrationError(
'dispatch_inactive',
`Dispatch ${dispatchId} is ${dispatch.status}; its worker will never read that mailbox. Send to run:${dispatch.run_id} instead, or start a new Dispatch for follow-up work.`
)
}
export type SendRecipientWarning = {
code:
| 'legacy_terminal_recipient'
@@ -26,7 +26,7 @@ describe('orchestration RPC methods', () => {
it('registers all expected methods', () => {
const registry = buildRegistry(ORCHESTRATION_METHODS)
expect(registry.size).toBe(38)
expect(registry.size).toBe(39)
expect(registry.has('orchestration.workerRelease')).toBe(true)
expect(registry.has('orchestration.workerRetain')).toBe(true)
expect(registry.has('orchestration.workerList')).toBe(true)
@@ -64,6 +64,7 @@ describe('orchestration RPC methods', () => {
expect(registry.has('orchestration.gateCreate')).toBe(true)
expect(registry.has('orchestration.gateResolve')).toBe(true)
expect(registry.has('orchestration.gateList')).toBe(true)
expect(registry.has('orchestration.requestShow')).toBe(true)
expect(registry.has('orchestration.reset')).toBe(true)
})
@@ -0,0 +1,92 @@
import { afterEach, describe, expect, it } from 'vitest'
import type { RpcContext } from '../core'
import type { OrchestrationDb } from '../../orchestration/db'
import { createRootDispatch } from '../../orchestration/db/root-dispatch-test-fixture'
import { createOrchestrationRpcHarness } from './orchestration-rpc-test-harness'
describe('orchestration.send to a settled Dispatch mailbox', () => {
const h = createOrchestrationRpcHarness()
let db: OrchestrationDb
let ctx: RpcContext
afterEach(() => {
h.cleanup()
})
function setup(): void {
;({ db, ctx } = h.setup())
}
async function call(name: string, params: Record<string, unknown>) {
return h.call(name, params, ctx)
}
it.each([
['completed', (settledDb: OrchestrationDb, id: string) => settledDb.completeDispatch(id)],
[
'failed',
(settledDb: OrchestrationDb, id: string) =>
settledDb.failDispatch(id, 'worker terminal closed')
]
])('rejects mail to a %s Dispatch instead of reporting success', async (_status, settle) => {
setup()
const task = db.createTask({ spec: 'worker that already reported' })
const dispatch = createRootDispatch(db, task.id, 'term_worker')
settle(db, dispatch.id)
await expect(
call('orchestration.send', {
from: 'term_coord',
to: `dispatch:${dispatch.id}`,
subject: 'One more thing'
})
).rejects.toMatchObject({ code: 'dispatch_inactive' })
})
it('names the Run mailbox that is still reachable', async () => {
setup()
const task = db.createTask({ spec: 'worker that already reported' })
const dispatch = createRootDispatch(db, task.id, 'term_worker')
db.completeDispatch(dispatch.id)
await expect(
call('orchestration.send', {
from: 'term_coord',
to: `dispatch:${dispatch.id}`,
subject: 'One more thing'
})
).rejects.toThrow(new RegExp(`run:${dispatch.run_id}`))
})
it('does not write an undeliverable message row', async () => {
setup()
const task = db.createTask({ spec: 'worker that already reported' })
const dispatch = createRootDispatch(db, task.id, 'term_worker')
db.completeDispatch(dispatch.id)
await call('orchestration.send', {
from: 'term_coord',
to: `dispatch:${dispatch.id}`,
subject: 'One more thing'
}).catch(() => undefined)
const stranded = db.db
.prepare('SELECT COUNT(*) AS count FROM messages WHERE to_handle = ?')
.get(`dispatch:${dispatch.id}`) as { count: number }
expect(stranded.count).toBe(0)
})
it('still delivers to an active Dispatch mailbox', async () => {
setup()
const task = db.createTask({ spec: 'worker still running' })
const dispatch = createRootDispatch(db, task.id, 'term_worker')
const result = (await call('orchestration.send', {
from: 'term_coord',
to: `dispatch:${dispatch.id}`,
subject: 'Pause after this step'
})) as { message: { to_handle: string } }
expect(result.message.to_handle).toBe(`dispatch:${dispatch.id}`)
})
})
+11 -5
View File
@@ -25,6 +25,7 @@ import {
import { clampOrchestrationAskTimeoutMs } from '../../../../shared/orchestration-ask-timeout'
import { ORCHESTRATION_GATE_METHODS } from './orchestration-gates'
import {
assertDispatchMailboxDeliverable,
resolveBareOrchestrationRecipient,
type SendRecipientWarning
} from './orchestration-recipient-routing'
@@ -33,6 +34,7 @@ import { resolveRunScope } from './orchestration-run-scope'
import { ORCHESTRATION_RUN_METHODS } from './orchestration-runs'
import { ORCHESTRATION_WORKER_METHODS } from './orchestration-worker-methods'
import { ORCHESTRATION_FEDERATION_METHODS } from './orchestration-federation-methods'
import { ORCHESTRATION_MUTATION_REQUEST_METHODS } from './orchestration-mutation-request-show'
import { OrchestrationError } from '../../orchestration/orchestration-error'
import type { OrcaRuntimeService } from '../../orca-runtime'
import type { RunRow } from '../../orchestration/types'
@@ -440,6 +442,7 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
...ORCHESTRATION_RUN_METHODS,
...ORCHESTRATION_WORKER_METHODS,
...ORCHESTRATION_FEDERATION_METHODS,
...ORCHESTRATION_MUTATION_REQUEST_METHODS,
defineMethod({
name: 'orchestration.send',
params: SendParams,
@@ -625,15 +628,18 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
} => (sendWarnings.length > 0 ? { ...receipt, warnings: sendWarnings } : receipt)
if (!isGroupAddress(to)) {
const federatedDispatchId = to.startsWith('dispatch:')
const addressedDispatchId = to.startsWith('dispatch:')
? to.slice('dispatch:'.length)
: undefined
const federatedTarget =
federatedDispatchId && to === `dispatch:${federatedDispatchId}`
? db.getFederatedDispatch(federatedDispatchId)
addressedDispatchId && to === `dispatch:${addressedDispatchId}`
? db.getFederatedDispatch(addressedDispatchId)
: undefined
if (federatedTarget && federatedDispatchId) {
const dispatchId = federatedDispatchId
if (addressedDispatchId && !federatedTarget) {
assertDispatchMailboxDeliverable(db, addressedDispatchId)
}
if (federatedTarget && addressedDispatchId) {
const dispatchId = addressedDispatchId
if (
federatedTarget.protocol_version <
ORCHESTRATION_FEDERATION_CONTROL_MAIL_PROTOCOL_VERSION
@@ -0,0 +1,133 @@
// A lost mutation response must be answerable without mutating again: these cover
// `orchestration.requestShow` reading the same durable receipt --retry-request replays.
import { describe, expect, it, vi } from 'vitest'
import { z } from 'zod'
import { ORCHESTRATION_CONTRACT_VERSION } from '../../../shared/protocol-version'
import { OrcaRuntimeService } from '../orca-runtime'
import { OrchestrationDb } from '../orchestration/db'
import { defineMethod, type RpcRequest } from './core'
import { RpcDispatcher } from './dispatcher'
import { ORCHESTRATION_METHODS } from './methods/orchestration'
// Why: filter the shipped array rather than importing the module, so the test also
// fails if the method exists but was never registered with the orchestration surface.
const REQUEST_SHOW_METHODS = ORCHESTRATION_METHODS.filter(
(method) => method.name === 'orchestration.requestShow'
)
const Params = z.object({ subject: z.string() })
function createHarness() {
const db = new OrchestrationDb(':memory:')
const runtime = new OrcaRuntimeService()
runtime.setOrchestrationDb(db)
const effect = vi.fn((subject: string) =>
db.insertMessage({ from: 'caller', to: 'recipient', subject })
)
const dispatcher = new RpcDispatcher({
runtime,
methods: [
defineMethod({
name: 'orchestration.send',
params: Params,
handler: ({ subject }) => ({ message: effect(subject) })
}),
...REQUEST_SHOW_METHODS
]
})
return { db, runtime, dispatcher, effect }
}
function sendRequest(mutationId: string): RpcRequest {
return {
id: `rpc_${mutationId}`,
authToken: 'caller-token',
method: 'orchestration.send',
params: { subject: 'hello' },
orchestrationContractVersion: ORCHESTRATION_CONTRACT_VERSION,
orchestrationRequestId: mutationId
}
}
function showRequest(requestId: string): RpcRequest {
return {
id: `rpc_show_${requestId}`,
authToken: 'caller-token',
method: 'orchestration.requestShow',
params: { request: requestId },
orchestrationContractVersion: ORCHESTRATION_CONTRACT_VERSION
}
}
describe('orchestration.requestShow', () => {
it('reports a recorded mutation as completed without repeating its effect', async () => {
const { dispatcher, effect } = createHarness()
await dispatcher.dispatch(sendRequest('mutation_completed'))
const response = await dispatcher.dispatch(showRequest('mutation_completed'))
expect(response).toMatchObject({
ok: true,
result: { requestId: 'mutation_completed', state: 'completed', method: 'orchestration.send' }
})
expect(effect).toHaveBeenCalledOnce()
})
it('returns the stored receipt so the caller can read the original outcome', async () => {
const { dispatcher } = createHarness()
const first = await dispatcher.dispatch(sendRequest('mutation_receipt'))
const response = await dispatcher.dispatch(showRequest('mutation_receipt'))
const result = (response as { result: { receipt: { message: { id: string } } } }).result
const sent = (first as { result: { message: { id: string } } }).result
expect(result.receipt.message.id).toBe(sent.message.id)
})
it('reports an interrupted mutation as pending and names the keyed replay', async () => {
const { db, dispatcher } = createHarness()
db.beginMutationReceipt({
callerFingerprint: db.getOrCreateLocalMutationCallerFingerprint(),
requestId: 'mutation_pending',
method: 'orchestration.workerStart',
payloadHash: 'hash'
})
const response = await dispatcher.dispatch(showRequest('mutation_pending'))
const result = (response as { result: { state: string; interpretation: string } }).result
expect(result.state).toBe('pending')
expect(result.interpretation).toContain('--retry-request mutation_pending')
})
it('reports an unknown request as absent without claiming nothing happened', async () => {
const { dispatcher } = createHarness()
const response = await dispatcher.dispatch(showRequest('mutation_missing'))
const result = (response as { result: { state: string; interpretation: string } }).result
expect(result.state).toBe('absent')
expect(result.interpretation).toContain('not proof that nothing happened')
})
it('scopes receipts to the caller identity that recorded them', async () => {
const { dispatcher } = createHarness()
await dispatcher.dispatch(sendRequest('mutation_scoped'))
const response = await dispatcher.dispatch(showRequest('mutation_scoped'), {
authenticatedCallerFingerprint: 'some-other-paired-device'
})
expect((response as { result: { state: string } }).result.state).toBe('absent')
})
it('never records a receipt of its own', async () => {
const { db, dispatcher } = createHarness()
const response = await dispatcher.dispatch(showRequest('mutation_readonly'))
expect(response).toMatchObject({ ok: true })
expect(
db.getMutationReceipt(db.getOrCreateLocalMutationCallerFingerprint(), 'mutation_readonly')
).toBeUndefined()
})
})
@@ -0,0 +1,33 @@
// Shared shape for `orchestration request-show`, the read-only answer to
// "did my mutation take effect?" when the response was lost in transit.
export const MUTATION_REQUEST_STATES = ['completed', 'pending', 'absent'] as const
export type OrchestrationMutationRequestState = (typeof MUTATION_REQUEST_STATES)[number]
export type OrchestrationMutationRequestShowResult = {
requestId: string
state: OrchestrationMutationRequestState
method?: string
createdAt?: string
updatedAt?: string
receipt?: unknown
// Why: `absent` is genuinely ambiguous, so the honest reading ships with the row
// instead of being re-derived (and softened) by every caller.
interpretation: string
}
export function describeMutationRequestState(params: {
requestId: string
state: OrchestrationMutationRequestState
method?: string
}): string {
const { requestId, state, method } = params
if (state === 'completed') {
return `Request ${requestId} already took effect${method ? ` (${method})` : ''}. Replaying it with --retry-request ${requestId} returns the recorded outcome and starts nothing new.`
}
if (state === 'pending') {
return `Request ${requestId} was accepted${method ? ` by ${method}` : ''} but its outcome was never recorded, so Orca restarted mid-mutation. Replay the original command with --retry-request ${requestId}; do not issue a fresh mutation.`
}
return `No receipt for request ${requestId} under this caller identity. It never reached this Orca runtime, failed before recording anything, or its receipt was pruned. Absent is not proof that nothing happened.`
}