diff --git a/skills/orchestration/SKILL.md b/skills/orchestration/SKILL.md index c7d15250e00..034bb5d9c9e 100644 --- a/skills/orchestration/SKILL.md +++ b/skills/orchestration/SKILL.md @@ -130,11 +130,11 @@ Dispatch rules: orca orchestration gate-create --task --question [--options ] [--json] orca orchestration gate-resolve --id --resolution [--json] orca orchestration gate-list [--task ] [--status ] [--json] -orca orchestration run --spec [--from ] [--poll-interval-ms ] [--max-concurrent ] [--worktree ] [--json] -orca orchestration run-stop [--json] +orca orchestration run --spec [--from ] [--poll-interval-ms ] [--max-concurrent ] [--worktree ] [--json] +orca orchestration run-stop [--worktree ] [--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 ` 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 ` 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. diff --git a/src/cli/handlers/orchestration.ts b/src/cli/handlers/orchestration.ts index 0308e94c53b..a113191b7bf 100644 --- a/src/cli/handlers/orchestration.ts +++ b/src/cli/handlers/orchestration.ts @@ -312,6 +312,17 @@ function throwNoActiveSenderTerminal(): never { ) } +async function getOptionalOrchestrationWorktreeSelector( + flags: Map, + cwd: string, + client: Parameters[0]['client'] +): Promise { + 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 = { }, '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 = { 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`) }, diff --git a/src/cli/help.ts b/src/cli/help.ts index 9a14d44d0e1..7608cf14d66 100644 --- a/src/cli/help.ts +++ b/src/cli/help.ts @@ -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 ] + orchestration run-stop Stop one coordinator run [--worktree ] orchestration gate-create Create a decision gate blocking a task orchestration gate-resolve Resolve a pending decision gate orchestration gate-list List decision gates diff --git a/src/cli/index.test.ts b/src/cli/index.test.ts index be7d057e0bb..1d33f2b342a 100644 --- a/src/cli/index.test.ts +++ b/src/cli/index.test.ts @@ -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' diff --git a/src/cli/specs/orchestration.ts b/src/cli/specs/orchestration.ts index a92ad06758e..098bc4d99ee 100644 --- a/src/cli/specs/orchestration.ts +++ b/src/cli/specs/orchestration.ts @@ -117,7 +117,7 @@ export const ORCHESTRATION_COMMAND_SPECS: CommandSpec[] = [ path: ['orchestration', 'run'], summary: 'Start the coordinator loop', usage: - 'orca orchestration run --spec [--from ] [--poll-interval-ms ] [--max-concurrent ] [--worktree ] [--json]', + 'orca orchestration run --spec [--from ] [--poll-interval-ms ] [--max-concurrent ] [--worktree ] [--json]', allowedFlags: [ ...GLOBAL_FLAGS, 'spec', @@ -125,13 +125,19 @@ export const ORCHESTRATION_COMMAND_SPECS: CommandSpec[] = [ 'poll-interval-ms', 'max-concurrent', 'worktree' + ], + notes: [ + 'Use --worktree 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 ] [--json]', - allowedFlags: [...GLOBAL_FLAGS, 'worktree'] + usage: 'orca orchestration run-stop [--worktree ] [--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'], diff --git a/src/main/runtime/orchestration/db.ts b/src/main/runtime/orchestration/db.ts index 3a9b9b8ef35..1f7e6e96a55 100644 --- a/src/main/runtime/orchestration/db.ts +++ b/src/main/runtime/orchestration/db.ts @@ -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[] { diff --git a/src/main/runtime/rpc/methods/orchestration-gates.test.ts b/src/main/runtime/rpc/methods/orchestration-gates.test.ts index c41cc438520..62f27c4daff 100644 --- a/src/main/runtime/rpc/methods/orchestration-gates.test.ts +++ b/src/main/runtime/rpc/methods/orchestration-gates.test.ts @@ -22,6 +22,14 @@ function expectOk(response: RpcResponse): Extract { return response } +function expectError(response: RpcResponse, message: RegExp): Extract { + 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() + } + }) }) diff --git a/src/main/runtime/rpc/methods/orchestration-gates.ts b/src/main/runtime/rpc/methods/orchestration-gates.ts index cbacc05a73d..276328859ae 100644 --- a/src/main/runtime/rpc/methods/orchestration-gates.ts +++ b/src/main/runtime/rpc/methods/orchestration-gates.ts @@ -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() @@ -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 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 } }