fix(orchestration): make coordinator scoping explicit

This commit is contained in:
Neil
2026-07-20 13:29:16 -07:00
committed by Brennan Benson
parent 1f546ba362
commit e8d9267116
8 changed files with 312 additions and 36 deletions
+3 -3
View File
@@ -130,11 +130,11 @@ Dispatch rules:
orca orchestration gate-create --task <task_id> --question <text> [--options <json_array>] [--json]
orca orchestration gate-resolve --id <gate_id> --resolution <text> [--json]
orca orchestration gate-list [--task <task_id>] [--status <status>] [--json]
orca orchestration run --spec <text> [--from <handle>] [--poll-interval-ms <n>] [--max-concurrent <n>] [--worktree <selector>] [--json]
orca orchestration run-stop [--json]
orca orchestration run --spec <text> [--from <handle>] [--poll-interval-ms <n>] [--max-concurrent <n>] [--worktree <selector|all>] [--json]
orca orchestration run-stop [--worktree <selector|all>] [--json]
```
`run` returns immediately with a run ID. Query progress with `task-list`. Use `ask` for worker-to-coordinator questions; it creates a `decision_gate` message that the coordinator answers with `reply`. Use `gate-create` only for coordinator-managed task DAG decisions, not for answering a worker's `ask`.
`run` returns immediately with a run ID. `--worktree <selector>` scopes ownership to that workspace; `--worktree all` (or no `--worktree`) is the explicit global coordinator and cannot safely run alongside scoped coordinators because it sees all tasks. `run-stop` with no `--worktree` is only for the legacy single-active-run case; pass `--worktree <selector>` or `--worktree all` when more than one coordinator may be active. Query progress with `task-list`. Use `ask` for worker-to-coordinator questions; it creates a `decision_gate` message that the coordinator answers with `reply`. Use `gate-create` only for coordinator-managed task DAG decisions, not for answering a worker's `ask`.
Recovery only: `orca orchestration reset --tasks|--messages|--all --json` clears runtime-global orchestration state. Do not run it during active coordination unless explicitly abandoning that state.
+18 -6
View File
@@ -312,6 +312,17 @@ function throwNoActiveSenderTerminal(): never {
)
}
async function getOptionalOrchestrationWorktreeSelector(
flags: Map<string, string | boolean>,
cwd: string,
client: Parameters<CommandHandler>[0]['client']
): Promise<string | undefined> {
if (getOptionalStringFlag(flags, 'worktree') === 'all') {
return 'all'
}
return await getOptionalWorktreeSelector(flags, 'worktree', cwd, client)
}
function isDevCliInvocation(): boolean {
return process.env.ORCA_USER_DATA_PATH?.includes('orca-dev') ?? false
}
@@ -679,8 +690,9 @@ export const ORCHESTRATION_HANDLERS: Record<string, CommandHandler> = {
},
'orchestration run': async ({ flags, client, cwd, json }) => {
const worktree = await getOptionalWorktreeSelector(flags, 'worktree', cwd, client)
const from = await resolveCoordinatorTerminalHandle(flags, cwd, client, worktree)
const worktree = await getOptionalOrchestrationWorktreeSelector(flags, cwd, client)
const terminalWorktree = worktree === 'all' ? undefined : worktree
const from = await resolveCoordinatorTerminalHandle(flags, cwd, client, terminalWorktree)
const result = await client.call<{
runId: string
status: string
@@ -699,10 +711,10 @@ export const ORCHESTRATION_HANDLERS: Record<string, CommandHandler> = {
runId: string
stopped: boolean
}>('orchestration.runStop', {
// Why (#4389): target the coordinator for a specific worktree so stopping
// one workspace's run does not tear down another's. Omitting it targets
// the legacy global (unscoped) run.
worktree: await getOptionalWorktreeSelector(flags, 'worktree', cwd, client)
// Why (#4389): pass through an explicit worktree or `all` so the runtime
// can stop an exact coordinator. Omitting it keeps the legacy single-run
// convenience, but the runtime rejects ambiguity when several runs exist.
worktree: await getOptionalOrchestrationWorktreeSelector(flags, cwd, client)
})
printResult(result, json, (r) => `Run ${r.runId} stopped`)
},
+1 -1
View File
@@ -95,7 +95,7 @@ Orchestration:
orchestration dispatch Dispatch a task to a terminal
orchestration dispatch-show Show dispatch context for a task
orchestration run Start the coordinator loop
orchestration run-stop Stop the active coordinator run [--worktree <selector>]
orchestration run-stop Stop one coordinator run [--worktree <selector|all>]
orchestration gate-create Create a decision gate blocking a task
orchestration gate-resolve Resolve a pending decision gate
orchestration gate-list List decision gates
+35
View File
@@ -3573,6 +3573,26 @@ describe('orca cli worktree awareness', () => {
})
})
it('treats --worktree all as explicit global orchestration run scope', async () => {
process.env.ORCA_TERMINAL_HANDLE = 'term_coord'
queueFixtures(callMock, okFixture('req_run', { runId: 'run_1', status: 'running' }))
vi.spyOn(console, 'log').mockImplementation(() => {})
await main(
['orchestration', 'run', '--spec', 'ship it', '--worktree', 'all', '--json'],
'/tmp/repo/feature/src'
)
expect(callMock).toHaveBeenCalledTimes(1)
expect(callMock).toHaveBeenCalledWith('orchestration.run', {
spec: 'ship it',
from: 'term_coord',
pollIntervalMs: undefined,
maxConcurrent: undefined,
worktree: 'all'
})
})
it('accepts and resolves --worktree for orchestration run-stop', async () => {
queueFixtures(
callMock,
@@ -3591,6 +3611,21 @@ describe('orca cli worktree awareness', () => {
})
})
it('passes --worktree all through for explicit global orchestration run-stop', async () => {
queueFixtures(callMock, okFixture('req_run_stop', { runId: 'run_1', stopped: true }))
vi.spyOn(console, 'log').mockImplementation(() => {})
await main(
['orchestration', 'run-stop', '--worktree', 'all', '--json'],
'/tmp/repo/feature/src'
)
expect(callMock).toHaveBeenCalledTimes(1)
expect(callMock).toHaveBeenCalledWith('orchestration.runStop', {
worktree: 'all'
})
})
it('passes dev mode to injected orchestration dispatches', async () => {
process.env.ORCA_TERMINAL_HANDLE = 'term_sender'
process.env.ORCA_USER_DATA_PATH = '/tmp/orca-dev'
+9 -3
View File
@@ -117,7 +117,7 @@ export const ORCHESTRATION_COMMAND_SPECS: CommandSpec[] = [
path: ['orchestration', 'run'],
summary: 'Start the coordinator loop',
usage:
'orca orchestration run --spec <text> [--from <handle>] [--poll-interval-ms <n>] [--max-concurrent <n>] [--worktree <selector>] [--json]',
'orca orchestration run --spec <text> [--from <handle>] [--poll-interval-ms <n>] [--max-concurrent <n>] [--worktree <selector|all>] [--json]',
allowedFlags: [
...GLOBAL_FLAGS,
'spec',
@@ -125,13 +125,19 @@ export const ORCHESTRATION_COMMAND_SPECS: CommandSpec[] = [
'poll-interval-ms',
'max-concurrent',
'worktree'
],
notes: [
'Use --worktree <selector> for an isolated coordinator. Omit --worktree or pass --worktree all for the global coordinator, which cannot run alongside scoped coordinators.'
]
},
{
path: ['orchestration', 'run-stop'],
summary: 'Stop the active coordinator run',
usage: 'orca orchestration run-stop [--worktree <selector>] [--json]',
allowedFlags: [...GLOBAL_FLAGS, 'worktree']
usage: 'orca orchestration run-stop [--worktree <selector|all>] [--json]',
allowedFlags: [...GLOBAL_FLAGS, 'worktree'],
notes: [
'Use --worktree all for the global coordinator. With no --worktree, run-stop succeeds only when exactly one coordinator is active.'
]
},
{
path: ['orchestration', 'gate-create'],
+14
View File
@@ -1064,6 +1064,20 @@ export class OrchestrationDb {
.get(handle) as CoordinatorRun | undefined
}
getActiveGlobalCoordinatorRun(): CoordinatorRun | undefined {
return this.db
.prepare(
"SELECT * FROM coordinator_runs WHERE status = 'running' AND workspace_key IS NULL ORDER BY created_at DESC LIMIT 1"
)
.get() as CoordinatorRun | undefined
}
listActiveCoordinatorRuns(): CoordinatorRun[] {
return this.db
.prepare("SELECT * FROM coordinator_runs WHERE status = 'running' ORDER BY created_at")
.all() as CoordinatorRun[]
}
// ── Queries for Coordinator ──
getIdleTerminals(excludeHandles: string[] = []): string[] {
@@ -22,6 +22,14 @@ function expectOk(response: RpcResponse): Extract<RpcResponse, { ok: true }> {
return response
}
function expectError(response: RpcResponse, message: RegExp): Extract<RpcResponse, { ok: false }> {
if (response.ok) {
throw new Error(`expected error response, got ok: ${JSON.stringify(response.result)}`)
}
expect(response.error.message).toMatch(message)
return response
}
function runId(response: RpcResponse): string {
return (expectOk(response).result as { runId: string }).runId
}
@@ -108,6 +116,46 @@ describe('orchestration start/stop guard scoping (#4389)', () => {
}
})
it('treats --worktree all as the explicit global coordinator scope', async () => {
const db = new OrchestrationDb(':memory:')
const runtime = makeRuntime(db)
const dispatcher = new RpcDispatcher({ runtime, methods: ORCHESTRATION_GATE_METHODS })
try {
db.createTask({ spec: 'global-work' })
const globalRunId = runId(
await dispatcher.dispatch(
makeRequest('orchestration.run', {
spec: 'global',
from: 'coord_global',
worktree: 'all',
pollIntervalMs: SHORT_POLL_MS
})
)
)
expect(db.getCoordinatorRun(globalRunId)?.workspace_key).toBeNull()
expect(db.getActiveGlobalCoordinatorRun()?.id).toBe(globalRunId)
db.createTask({ spec: 'a-work', workspaceKey: 'worktree:wt_a' })
expectError(
await dispatcher.dispatch(
makeRequest('orchestration.run', {
spec: 'scoped',
from: 'coord_a',
worktree: 'worktree:wt_a',
pollIntervalMs: SHORT_POLL_MS
})
),
/Coordinator already running/
)
await stopAndDrain(dispatcher, ['all'])
} finally {
db.close()
}
})
it('does not fall back to the global coordinator when an explicit worktree selector fails', async () => {
const db = new OrchestrationDb(':memory:')
const runtime = makeRuntime(db)
@@ -156,10 +204,7 @@ describe('orchestration start/stop guard scoping (#4389)', () => {
pollIntervalMs: SHORT_POLL_MS
})
)
expect(second.ok).toBe(false)
if (!second.ok) {
expect(second.error.message).toMatch(/Coordinator already running/)
}
expectError(second, /Coordinator already running/)
await stopAndDrain(dispatcher, ['worktree:wt_a'])
} finally {
@@ -211,4 +256,132 @@ describe('orchestration start/stop guard scoping (#4389)', () => {
db.close()
}
})
it('runStop --worktree all stops the global coordinator exactly', async () => {
const db = new OrchestrationDb(':memory:')
const runtime = makeRuntime(db)
const dispatcher = new RpcDispatcher({ runtime, methods: ORCHESTRATION_GATE_METHODS })
try {
db.createTask({ spec: 'global-work' })
const globalRunId = runId(
await dispatcher.dispatch(
makeRequest('orchestration.run', {
spec: 'global',
from: 'coord_global',
worktree: 'all',
pollIntervalMs: SHORT_POLL_MS
})
)
)
const stopGlobal = expectOk(
await dispatcher.dispatch(makeRequest('orchestration.runStop', { worktree: 'all' }))
)
expect((stopGlobal.result as { runId: string }).runId).toBe(globalRunId)
await new Promise((resolve) => setTimeout(resolve, SHORT_POLL_MS * 4))
expect(db.getCoordinatorRun(globalRunId)?.status).toBe('failed')
} finally {
db.close()
}
})
it('runStop with a scoped worktree does not match the global coordinator', async () => {
const db = new OrchestrationDb(':memory:')
const runtime = makeRuntime(db)
const dispatcher = new RpcDispatcher({ runtime, methods: ORCHESTRATION_GATE_METHODS })
try {
db.createTask({ spec: 'global-work' })
const globalRunId = runId(
await dispatcher.dispatch(
makeRequest('orchestration.run', {
spec: 'global',
from: 'coord_global',
worktree: 'all',
pollIntervalMs: SHORT_POLL_MS
})
)
)
expectError(
await dispatcher.dispatch(
makeRequest('orchestration.runStop', { worktree: 'worktree:wt_a' })
),
/No active coordinator run for requested worktree/
)
expect(db.getActiveGlobalCoordinatorRun()?.id).toBe(globalRunId)
await stopAndDrain(dispatcher, ['all'])
} finally {
db.close()
}
})
it('runStop with no worktree stops the only active scoped coordinator', async () => {
const db = new OrchestrationDb(':memory:')
const runtime = makeRuntime(db)
const dispatcher = new RpcDispatcher({ runtime, methods: ORCHESTRATION_GATE_METHODS })
try {
db.createTask({ spec: 'a-work', workspaceKey: 'worktree:wt_a' })
const runAId = runId(
await dispatcher.dispatch(
makeRequest('orchestration.run', {
spec: 'a',
from: 'coord_a',
worktree: 'worktree:wt_a',
pollIntervalMs: SHORT_POLL_MS
})
)
)
const stopOnlyRun = expectOk(await dispatcher.dispatch(makeRequest('orchestration.runStop')))
expect((stopOnlyRun.result as { runId: string }).runId).toBe(runAId)
await new Promise((resolve) => setTimeout(resolve, SHORT_POLL_MS * 4))
expect(db.getCoordinatorRun(runAId)?.status).toBe('failed')
} finally {
db.close()
}
})
it('runStop with no worktree rejects multiple active scoped coordinators', async () => {
const db = new OrchestrationDb(':memory:')
const runtime = makeRuntime(db)
const dispatcher = new RpcDispatcher({ runtime, methods: ORCHESTRATION_GATE_METHODS })
try {
db.createTask({ spec: 'a-work', workspaceKey: 'worktree:wt_a' })
db.createTask({ spec: 'b-work', workspaceKey: 'worktree:wt_b' })
const runAId = runId(
await dispatcher.dispatch(
makeRequest('orchestration.run', {
spec: 'a',
from: 'coord_a',
worktree: 'worktree:wt_a',
pollIntervalMs: SHORT_POLL_MS
})
)
)
const runBId = runId(
await dispatcher.dispatch(
makeRequest('orchestration.run', {
spec: 'b',
from: 'coord_b',
worktree: 'worktree:wt_b',
pollIntervalMs: SHORT_POLL_MS
})
)
)
expectError(
await dispatcher.dispatch(makeRequest('orchestration.runStop')),
/Multiple active coordinator runs/
)
expect(db.getActiveCoordinatorRunForWorkspace('worktree:wt_a')?.id).toBe(runAId)
expect(db.getActiveCoordinatorRunForWorkspace('worktree:wt_b')?.id).toBe(runBId)
await stopAndDrain(dispatcher, ['worktree:wt_a', 'worktree:wt_b'])
} finally {
db.close()
}
})
})
@@ -1,15 +1,15 @@
import { z } from 'zod'
import { defineMethod, type RpcMethod } from '../core'
import { OptionalFiniteNumber, OptionalString, requiredString } from '../schemas'
import type { GateStatus } from '../../orchestration/db'
import type { CoordinatorRun, GateStatus } from '../../orchestration/db'
import { Coordinator } from '../../orchestration/coordinator'
// Why (#4389): live coordinators are keyed by workspace scope so concurrent
// orchestrators in different worktrees of one Orca instance don't orphan or
// stop each other. A 2nd run in another workspace adds its own entry instead of
// overwriting a single module-level reference, and runStop halts only the
// coordinator for the requested workspace. The unscoped/global run (no worktree
// selector, or one that can't be resolved) uses a fixed sentinel key so its
// coordinator for the requested workspace. The unscoped/global run (no
// worktree selector or `--worktree all`) uses a fixed sentinel key so its
// pre-#4389 single-coordinator semantics are preserved.
const GLOBAL_COORDINATOR_KEY = '__global__'
const activeCoordinators = new Map<string, Coordinator>()
@@ -18,6 +18,19 @@ function coordinatorMapKey(workspaceKey: string | null): string {
return workspaceKey ?? GLOBAL_COORDINATOR_KEY
}
function normalizeOrchestrationWorktreeSelector(worktree?: string | null): string | undefined {
return worktree === 'all' ? undefined : (worktree ?? undefined)
}
function stopCoordinatorForRun(run: CoordinatorRun): void {
const mapKey = coordinatorMapKey(run.workspace_key)
const coordinator = activeCoordinators.get(mapKey)
if (coordinator) {
coordinator.stop()
activeCoordinators.delete(mapKey)
}
}
const RunParams = z.object({
spec: requiredString('Missing --spec'),
from: OptionalString,
@@ -56,12 +69,14 @@ export const ORCHESTRATION_GATE_METHODS: RpcMethod[] = [
params: RunParams,
handler: async (params, { runtime }) => {
const db = runtime.getOrchestrationDb()
const worktree = normalizeOrchestrationWorktreeSelector(params.worktree)
// Why (#4389): scope the active-run guard to this worktree so a 2nd
// orchestrator in a different workspace is not rejected by another
// workspace's running coordinator. A null key (no/unresolvable worktree)
// keeps the original global single-run guard.
const workspaceKey = await runtime.resolveWorkspaceKeyForSelector(params.worktree)
// workspace's running coordinator. A null key (no worktree or
// `--worktree all`) is the explicit global coordinator; because it sees
// every task, it still conflicts with any running scoped coordinator.
const workspaceKey = await runtime.resolveWorkspaceKeyForSelector(worktree)
const existing = db.getActiveCoordinatorRun(workspaceKey)
if (existing) {
throw new Error(`Coordinator already running: ${existing.id}`)
@@ -73,7 +88,7 @@ export const ORCHESTRATION_GATE_METHODS: RpcMethod[] = [
coordinatorHandle,
pollIntervalMs: params.pollIntervalMs,
maxConcurrent: params.maxConcurrent,
worktree: params.worktree,
worktree,
workspaceKey
})
@@ -108,20 +123,41 @@ export const ORCHESTRATION_GATE_METHODS: RpcMethod[] = [
params: RunStopParams,
handler: async (params, { runtime }) => {
const db = runtime.getOrchestrationDb()
// Why (#4389): stop only the coordinator for the requested workspace so a
// stop in one worktree does not tear down another worktree's orchestrator.
const workspaceKey = await runtime.resolveWorkspaceKeyForSelector(params.worktree)
const run = db.getActiveCoordinatorRun(workspaceKey)
if (!run) {
throw new Error('No active coordinator run')
const worktree = normalizeOrchestrationWorktreeSelector(params.worktree)
let run: CoordinatorRun | undefined
if (params.worktree === 'all') {
run = db.getActiveGlobalCoordinatorRun()
if (!run) {
throw new Error('No active global coordinator run')
}
} else if (worktree) {
// Why (#4389): explicit scoped stops must be exact. Scoped read helpers
// intentionally include legacy NULL rows for task visibility, but using
// that fallback here could report a global run as stopped while leaving
// the live global coordinator running.
const workspaceKey = await runtime.resolveWorkspaceKeyForSelector(worktree)
run =
workspaceKey === null
? db.getActiveGlobalCoordinatorRun()
: db.getActiveCoordinatorRunForWorkspace(workspaceKey)
if (!run) {
throw new Error('No active coordinator run for requested worktree')
}
} else {
const activeRuns = db.listActiveCoordinatorRuns()
if (activeRuns.length === 0) {
throw new Error('No active coordinator run')
}
if (activeRuns.length > 1) {
throw new Error(
'Multiple active coordinator runs; pass --worktree <selector> or --worktree all'
)
}
run = activeRuns[0]
}
const mapKey = coordinatorMapKey(workspaceKey)
const coordinator = activeCoordinators.get(mapKey)
if (coordinator) {
coordinator.stop()
activeCoordinators.delete(mapKey)
}
stopCoordinatorForRun(run)
return { runId: run.id, stopped: true }
}