From 5f0677ec0aa9f92250cc4e3625b489dcec4fbe68 Mon Sep 17 00:00:00 2001 From: Neil <4138956+nwparker@users.noreply.github.com> Date: Sat, 27 Jun 2026 23:32:55 -0700 Subject: [PATCH] fix(orchestration): scope coordinator runs/tasks/dispatches by workspace (#4389) Two orchestrators started in different worktrees of one Orca instance shared a single unscoped orchestration.db, so they blocked each other's start, stopped the wrong coordinator, counted/dispatched each other's tasks, and cross-attributed worker terminals in the agent-status UI. Add a nullable workspace_key column (schema v5 -> v6) to coordinator_runs, tasks, dispatch_contexts, and messages, and thread a per-coordinator workspace key through the writes and run-scoped reads. The start/stop guard now scopes getActiveCoordinatorRun by workspace and keys live coordinators in a Map instead of one module-level singleton, so a run in one workspace can't orphan or stop another's. Runtime helpers attribute a worker terminal to the coordinator that owns its dispatch's workspace. Existing rows keep NULL workspace_key (legacy/global) and every scoped read matches NULL, so single-orchestrator installs are unaffected. Scope resolution goes through the existing worktree-selector resolution, so it is provider-agnostic and works for SSH/remote runtimes. Fixes #4389 Co-authored-by: Orca --- src/cli/handlers/orchestration.ts | 9 +- src/cli/help.ts | 2 +- src/main/runtime/orca-runtime.test.ts | 4 +- src/main/runtime/orca-runtime.ts | 74 ++++++- src/main/runtime/orchestration/coordinator.ts | 57 ++++- .../orchestration/db-workspace-scope.test.ts | 110 ++++++++++ src/main/runtime/orchestration/db.ts | 201 +++++++++++++++--- .../runtime/orchestration/formatter.test.ts | 1 + src/main/runtime/orchestration/types.ts | 7 + .../rpc/methods/orchestration-gates.test.ts | 191 +++++++++++++++++ .../rpc/methods/orchestration-gates.ts | 61 ++++-- src/main/runtime/rpc/methods/orchestration.ts | 11 +- src/main/runtime/runtime-rpc.test.ts | 74 +++++++ 13 files changed, 733 insertions(+), 69 deletions(-) create mode 100644 src/main/runtime/orchestration/db-workspace-scope.test.ts create mode 100644 src/main/runtime/rpc/methods/orchestration-gates.test.ts diff --git a/src/cli/handlers/orchestration.ts b/src/cli/handlers/orchestration.ts index cdafcd66743..5bf47782b60 100644 --- a/src/cli/handlers/orchestration.ts +++ b/src/cli/handlers/orchestration.ts @@ -685,11 +685,16 @@ export const ORCHESTRATION_HANDLERS: Record = { printResult(result, json, (r) => `Run ${r.runId} started (${r.status})`) }, - 'orchestration run-stop': async ({ client, json }) => { + 'orchestration run-stop': async ({ flags, client, json }) => { const result = await client.call<{ runId: string stopped: boolean - }>('orchestration.runStop', {}) + }>('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: getOptionalStringFlag(flags, 'worktree') + }) printResult(result, json, (r) => `Run ${r.runId} stopped`) }, diff --git a/src/cli/help.ts b/src/cli/help.ts index 11d9fd2dc20..9a14d44d0e1 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 + orchestration run-stop Stop the active 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/main/runtime/orca-runtime.test.ts b/src/main/runtime/orca-runtime.test.ts index a63d25b9a04..b6db6769f26 100644 --- a/src/main/runtime/orca-runtime.test.ts +++ b/src/main/runtime/orca-runtime.test.ts @@ -978,6 +978,7 @@ class InMemoryOrchestrationMessages { priority?: MessagePriority threadId?: string payload?: string + workspaceKey?: string | null }): MessageRow { this.sequence += 1 const row: MessageRow = { @@ -994,7 +995,8 @@ class InMemoryOrchestrationMessages { sequence: this.sequence, created_at: '1970-01-01 00:00:00', delivered_at: null, - sender_pane_key: null + sender_pane_key: null, + workspace_key: msg.workspaceKey ?? null } this.messages.push(row) return row diff --git a/src/main/runtime/orca-runtime.ts b/src/main/runtime/orca-runtime.ts index fbb426994a2..3edb00d81e9 100644 --- a/src/main/runtime/orca-runtime.ts +++ b/src/main/runtime/orca-runtime.ts @@ -10976,7 +10976,10 @@ export class OrcaRuntimeService { // Why: create an escalation message so the coordinator is notified about // the unexpected exit on its next check cycle, even if the circuit breaker // hasn't tripped yet. - const run = this._orchestrationDb.getActiveCoordinatorRun() + // Why (#4389): resolve the run from THIS dispatch's workspace so the + // escalation goes to the coordinator that actually owns the worker, not the + // most-recent global run; legacy NULL dispatches fall back to global. + const run = this.resolveCoordinatorRunForDispatch(dispatch.workspace_key) if (run) { this._orchestrationDb.insertMessage({ from: handle, @@ -10984,6 +10987,7 @@ export class OrcaRuntimeService { subject: `Agent exited unexpectedly (code ${exitCode})`, type: 'escalation', priority: 'high', + workspaceKey: dispatch.workspace_key, payload: JSON.stringify({ taskId: dispatch.task_id, exitCode, @@ -10993,6 +10997,23 @@ export class OrcaRuntimeService { } } + // Why (#4389): a worker terminal is attributed to ITS coordinator by routing + // through the dispatch's workspace_key. A scoped dispatch resolves the run for + // that exact workspace (strict, no global fallback) so concurrent orchestrators + // never cross-attribute; a legacy NULL dispatch keeps the original global + // most-recent-run behavior. + private resolveCoordinatorRunForDispatch( + workspaceKey: string | null + ): ReturnType { + if (!this._orchestrationDb) { + return undefined + } + if (workspaceKey == null) { + return this._orchestrationDb.getActiveCoordinatorRun() + } + return this._orchestrationDb.getActiveCoordinatorRunForWorkspace(workspaceKey) + } + async listTerminals( worktreeSelector?: string, limit = DEFAULT_TERMINAL_LIST_LIMIT, @@ -21053,7 +21074,11 @@ export class OrcaRuntimeService { const activeDispatch = this._orchestrationDb?.getActiveDispatchForTerminal( input.callerTerminalHandle ) - const activeRun = this._orchestrationDb?.getActiveCoordinatorRun() + // Why (#4389): attribute the worker to the coordinator that owns its + // dispatch's workspace, not the most-recent global run. + const activeRun = activeDispatch + ? this.resolveCoordinatorRunForDispatch(activeDispatch.workspace_key) + : undefined if (activeDispatch) { candidates.push({ source: 'orchestration-context', @@ -21200,6 +21225,40 @@ export class OrcaRuntimeService { } } + // Why (#4389): resolves an orchestration worktree selector to the workspace + // key used to scope coordinator runs/tasks/dispatches. Goes through the same + // selector resolution as lineage so it is provider-agnostic and works for + // SSH/remote runtimes (no local-path assumptions). Returns null when there is + // no selector or it cannot be resolved, which falls back to legacy/global + // (unscoped) behavior for single-orchestrator installs. + async resolveWorkspaceKeyForSelector(selector?: string | null): Promise { + if (!selector) { + return null + } + try { + const worktree = await this.resolveWorktreeSelector(selector) + return worktreeWorkspaceKey(worktree.id) + } catch { + return null + } + } + + // Why (#4389): tasks created by a worker/coordinator terminal belong to that + // terminal's worktree; scoping them by it keeps each orchestrator's task DAG + // private. Returns null when the handle is absent/stale so task creation stays + // recoverable and legacy (global) when no workspace can be proven. + async resolveWorkspaceKeyForTerminalHandle(handle?: string | null): Promise { + if (!handle) { + return null + } + try { + const terminal = await this.showTerminal(handle) + return this.resolveWorkspaceKeyForSelector(`id:${terminal.worktreeId}`) + } catch { + return null + } + } + async hydrateInferredWorktreeLineage(): Promise { const store = this.store if ( @@ -22821,7 +22880,16 @@ export class OrcaRuntimeService { displayName: task.display_name }) : { taskTitle: '', displayName: '' } - const activeRun = dispatch.status === 'completed' ? undefined : db?.getActiveCoordinatorRun?.() + // Why (#4389): resolve the run from the dispatch's own workspace so a worker + // of coordinator B is never attributed to coordinator A's run/handle. A + // scoped dispatch uses the strict per-workspace lookup; a legacy NULL + // dispatch keeps the original global most-recent-run behavior. + const activeRun = + dispatch.status === 'completed' + ? undefined + : dispatch.workspace_key == null + ? db?.getActiveCoordinatorRun?.() + : db?.getActiveCoordinatorRunForWorkspace?.(dispatch.workspace_key) const parentTerminalHandle = task?.created_by_terminal_handle ?? (activeRun?.coordinator_handle && activeRun.coordinator_handle !== handle diff --git a/src/main/runtime/orchestration/coordinator.ts b/src/main/runtime/orchestration/coordinator.ts index d0e56b7e40a..d2d1bc000dd 100644 --- a/src/main/runtime/orchestration/coordinator.ts +++ b/src/main/runtime/orchestration/coordinator.ts @@ -57,6 +57,11 @@ export type CoordinatorOptions = { pollIntervalMs?: number maxConcurrent?: number worktree?: string + // Why (#4389): the workspace scope this coordinator owns, resolved from + // `worktree` by the caller. All run/task/dispatch reads are filtered by it so + // concurrent orchestrators in different worktrees can't see or mutate each + // other's state. Undefined falls back to legacy unscoped (global) behavior. + workspaceKey?: string | null onLog?: (msg: string) => void } @@ -79,9 +84,10 @@ export class Coordinator { private runtime: CoordinatorRuntime private state: CoordinatorState private stopped = false - private opts: Required> & { + private opts: Required> & { onLog: (msg: string) => void worktree?: string + workspaceKey?: string | null } constructor(db: OrchestrationDb, runtime: CoordinatorRuntime, options: CoordinatorOptions) { @@ -93,6 +99,7 @@ export class Coordinator { pollIntervalMs: options.pollIntervalMs ?? DEFAULT_POLL_MS, maxConcurrent: options.maxConcurrent ?? MAX_CONCURRENT_DEFAULT, worktree: options.worktree, + workspaceKey: options.workspaceKey ?? null, onLog: options.onLog ?? (() => {}) } this.state = { @@ -114,7 +121,8 @@ export class Coordinator { const run = this.db.createCoordinatorRun({ spec: this.opts.spec, coordinatorHandle: this.opts.coordinatorHandle, - pollIntervalMs: this.opts.pollIntervalMs + pollIntervalMs: this.opts.pollIntervalMs, + workspaceKey: this.opts.workspaceKey }) return this.executeLoop(run.id) } @@ -152,7 +160,7 @@ export class Coordinator { } // Why: an early stop leaves tasks incomplete, so the run counts as failed. - const tasks = this.db.listTasks() + const tasks = this.db.listTasks({ workspaceKey: this.opts.workspaceKey }) const allDone = tasks.every((t) => t.status === 'completed' || t.status === 'failed') const failedTasks = [ ...new Set([ @@ -185,7 +193,7 @@ export class Coordinator { // Why: decomposition isn't implemented yet — tasks must be pre-created before run(); AI-driven decomposition is a future phase. private async decompose(): Promise { this.state.phase = 'decomposing' - const existing = this.db.listTasks() + const existing = this.db.listTasks({ workspaceKey: this.opts.workspaceKey }) if (existing.length === 0) { throw new Error( 'No tasks found. Create tasks with orchestration.taskCreate before running the coordinator.' @@ -207,7 +215,7 @@ export class Coordinator { // Why: warn only, never auto-fail — a false positive (slow but correct worker) costs more than a false negative (hung worker holding a slot); see R6 of DESIGN_DOC_PREAMBLE_FIX.md. private warnStaleDispatches(): void { const thresholdIso = new Date(Date.now() - HUNG_THRESHOLD_MS).toISOString() - const stale = this.db.getStaleDispatches(thresholdIso) + const stale = this.db.getStaleDispatches(thresholdIso, this.opts.workspaceKey) for (const ctx of stale) { const minutes = Math.round(HUNG_THRESHOLD_MS / 60000) this.opts.onLog( @@ -217,7 +225,11 @@ export class Coordinator { } private processMessages(): void { - const messages = this.db.getUnreadMessages(this.opts.coordinatorHandle) + const messages = this.db.getUnreadMessages( + this.opts.coordinatorHandle, + undefined, + this.opts.workspaceKey + ) if (messages.length === 0) { return } @@ -332,7 +344,13 @@ export class Coordinator { const pendingGates = this.db.listGates({ status: 'pending' }) for (const gate of pendingGates) { const task = this.db.getTask(gate.task_id) - if (task && task.status !== 'blocked') { + // Why (#4389): decision_gates have no workspace_key, so derive scope from + // the gate's task and skip tasks owned by another workspace's coordinator + // — re-blocking them here would let one orchestrator stomp another's task. + if (!task || !this.ownsTask(task)) { + continue + } + if (task.status !== 'blocked') { // Why: gate exists but task isn't blocked — re-block to restore the invariant. this.db.updateTaskStatus(gate.task_id, 'blocked') } @@ -341,12 +359,15 @@ export class Coordinator { private async dispatchReadyTasks(): Promise { this.state.phase = 'dispatching' - const readyTasks = this.db.listTasks({ ready: true }) + const readyTasks = this.db.listTasks({ ready: true, workspaceKey: this.opts.workspaceKey }) if (readyTasks.length === 0) { return } - const dispatched = this.db.listTasks({ status: 'dispatched' }) + const dispatched = this.db.listTasks({ + status: 'dispatched', + workspaceKey: this.opts.workspaceKey + }) let slotsAvailable = this.opts.maxConcurrent - dispatched.length if (slotsAvailable <= 0) { return @@ -464,7 +485,10 @@ export class Coordinator { private async getAvailableTerminals(): Promise { try { const result = await this.runtime.listTerminals(this.opts.worktree) - const dispatched = this.db.listTasks({ status: 'dispatched' }) + const dispatched = this.db.listTasks({ + status: 'dispatched', + workspaceKey: this.opts.workspaceKey + }) const busyHandles = new Set() for (const task of dispatched) { @@ -490,7 +514,7 @@ export class Coordinator { } private checkConvergence(): boolean { - const tasks = this.db.listTasks() + const tasks = this.db.listTasks({ workspaceKey: this.opts.workspaceKey }) if (tasks.length === 0) { return true } @@ -515,6 +539,17 @@ export class Coordinator { return false } + // Why (#4389): true when this coordinator's workspace scope covers the task — + // either the coordinator is unscoped (legacy/global, owns everything) or the + // task is legacy NULL or carries this coordinator's exact workspace_key. Used + // where a row is read outside a workspace-scoped query (e.g. decision gates). + private ownsTask(task: TaskRow): boolean { + if (this.opts.workspaceKey == null) { + return true + } + return task.workspace_key == null || task.workspace_key === this.opts.workspaceKey + } + private sleep(ms: number): Promise { return new Promise((resolve) => { setTimeout(resolve, ms) diff --git a/src/main/runtime/orchestration/db-workspace-scope.test.ts b/src/main/runtime/orchestration/db-workspace-scope.test.ts new file mode 100644 index 00000000000..a9c794b028c --- /dev/null +++ b/src/main/runtime/orchestration/db-workspace-scope.test.ts @@ -0,0 +1,110 @@ +import { afterEach, describe, expect, it } from 'vitest' +import type Database from '../../sqlite/sync-database' +import { OrchestrationDb } from './db' + +// Why (#4389): two orchestrators in different worktrees of one Orca instance +// must not see or mutate each other's runs, tasks, or dispatches. +describe('OrchestrationDb workspace scoping', () => { + const KEY_A = 'worktree:wt_a' + const KEY_B = 'worktree:wt_b' + let db: OrchestrationDb | undefined + + afterEach(() => db?.close()) + + function createDb(): OrchestrationDb { + db = new OrchestrationDb(':memory:') + return db + } + + it('returns each workspace its own active run', () => { + const d = createDb() + const runA = d.createCoordinatorRun({ + spec: 'a', + coordinatorHandle: 'coord_a', + workspaceKey: KEY_A + }) + const runB = d.createCoordinatorRun({ + spec: 'b', + coordinatorHandle: 'coord_b', + workspaceKey: KEY_B + }) + + expect(d.getActiveCoordinatorRun(KEY_A)?.id).toBe(runA.id) + expect(d.getActiveCoordinatorRun(KEY_B)?.id).toBe(runB.id) + expect(d.getActiveCoordinatorRunForWorkspace(KEY_A)?.id).toBe(runA.id) + expect(d.getActiveCoordinatorRunForWorkspace(KEY_B)?.id).toBe(runB.id) + expect(d.getActiveCoordinatorRunForHandle('coord_a')?.id).toBe(runA.id) + expect(d.getActiveCoordinatorRunForHandle('coord_b')?.id).toBe(runB.id) + expect(d.getActiveCoordinatorRunForHandle('coord_missing')).toBeUndefined() + }) + + it('indexes active coordinator lookup by handle', () => { + const d = createDb() + const sqlite = (d as unknown as { db: Database.Database }).db + const indexes = sqlite + .prepare( + `SELECT name FROM sqlite_master WHERE type = 'index' AND name = 'idx_coordinator_runs_handle_status'` + ) + .all() + + expect(indexes).toHaveLength(1) + }) + + it('shares a legacy run without crossing scoped runs', () => { + const d = createDb() + const legacy = d.createCoordinatorRun({ spec: 'legacy', coordinatorHandle: 'coord' }) + expect(d.getActiveCoordinatorRun(KEY_A)?.id).toBe(legacy.id) + expect(d.getActiveCoordinatorRunForWorkspace(KEY_A)).toBeUndefined() + + const runA = d.createCoordinatorRun({ + spec: 'a', + coordinatorHandle: 'coord_a', + workspaceKey: KEY_A + }) + expect(d.getActiveCoordinatorRun(KEY_B)?.id).toBe(legacy.id) + expect(d.getActiveCoordinatorRunForWorkspace(KEY_B)).toBeUndefined() + expect(d.getActiveCoordinatorRunForWorkspace(KEY_A)?.id).toBe(runA.id) + }) + + it('lists scoped tasks without returning another workspace', () => { + const d = createDb() + const taskA = d.createTask({ spec: 'a-work', workspaceKey: KEY_A }) + const taskB = d.createTask({ spec: 'b-work', workspaceKey: KEY_B }) + const legacy = d.createTask({ spec: 'legacy-work' }) + + const aReady = d.listTasks({ ready: true, workspaceKey: KEY_A }).map((task) => task.id) + expect(aReady).toContain(taskA.id) + expect(aReady).toContain(legacy.id) + expect(aReady).not.toContain(taskB.id) + expect( + d + .listTasks({ workspaceKey: KEY_A }) + .map((task) => task.id) + .sort() + ).toEqual([taskA.id, legacy.id].sort()) + expect(d.listTasks()).toHaveLength(3) + }) + + it('does not promote dependencies owned by another workspace', () => { + const d = createDb() + const taskA = d.createTask({ spec: 'a', workspaceKey: KEY_A }) + const taskB = d.createTask({ spec: 'b', deps: [taskA.id], workspaceKey: KEY_B }) + + d.updateTaskStatus(taskA.id, 'completed') + + expect(d.getTask(taskB.id)?.status).toBe('pending') + }) + + it('does not surface another workspace stale dispatches', () => { + const d = createDb() + const taskA = d.createTask({ spec: 'a-work', workspaceKey: KEY_A }) + const taskB = d.createTask({ spec: 'b-work', workspaceKey: KEY_B }) + const ctxA = d.createDispatchContext(taskA.id, 'term_a') + const ctxB = d.createDispatchContext(taskB.id, 'term_b') + + const future = new Date(Date.now() + 60_000).toISOString() + const staleA = d.getStaleDispatches(future, KEY_A).map((context) => context.id) + expect(staleA).toContain(ctxA.id) + expect(staleA).not.toContain(ctxB.id) + }) +}) diff --git a/src/main/runtime/orchestration/db.ts b/src/main/runtime/orchestration/db.ts index 1c4b68f9d10..3a9b9b8ef35 100644 --- a/src/main/runtime/orchestration/db.ts +++ b/src/main/runtime/orchestration/db.ts @@ -83,8 +83,23 @@ function exposeMessageListTimestamps(messages: MessageRow[]): MessageRow[] { return messages.map(exposeMessageTimestamps) } -// Schema versions: v2 'heartbeat'+last_heartbeat_at, v3 delivered_at, v4 task-creator terminal, v5 task_title/display_name, v6 pane-identity columns. -const SCHEMA_VERSION = 6 +// Why (#4389): scope a query to one workspace without hiding pre-v6 rows. A +// provided key matches its own rows plus legacy NULL rows (single-orchestrator +// installs predate scoping); an omitted key adds no constraint, preserving the +// original global behavior for callers that have no worktree. The fragment +// carries a leading ` AND ` so callers can append it to an existing WHERE. +function workspaceScopeClause(workspaceKey?: string | null): { + clause: string + params: string[] +} { + if (workspaceKey == null) { + return { clause: '', params: [] } + } + return { clause: ' AND (workspace_key = ? OR workspace_key IS NULL)', params: [workspaceKey] } +} + +// Schema versions: v2 'heartbeat'+last_heartbeat_at, v3 delivered_at, v4 task-creator terminal, v5 task_title/display_name, v6 pane identity, v7 workspace ownership. +const SCHEMA_VERSION = 7 export class OrchestrationDb { private db: Database.Database @@ -119,7 +134,8 @@ export class OrchestrationDb { sequence INTEGER PRIMARY KEY AUTOINCREMENT, created_at TEXT NOT NULL DEFAULT (datetime('now')), delivered_at TEXT, - sender_pane_key TEXT + sender_pane_key TEXT, + workspace_key TEXT ); CREATE UNIQUE INDEX IF NOT EXISTS idx_messages_id ON messages(id); @@ -141,7 +157,8 @@ export class OrchestrationDb { deps TEXT NOT NULL DEFAULT '[]', result TEXT, created_at TEXT NOT NULL DEFAULT (datetime('now')), - completed_at TEXT + completed_at TEXT, + workspace_key TEXT ); CREATE INDEX IF NOT EXISTS idx_tasks_status ON tasks(status); @@ -159,7 +176,8 @@ export class OrchestrationDb { dispatched_at TEXT, completed_at TEXT, created_at TEXT NOT NULL DEFAULT (datetime('now')), - last_heartbeat_at TEXT + last_heartbeat_at TEXT, + workspace_key TEXT ); CREATE INDEX IF NOT EXISTS idx_dispatch_task ON dispatch_contexts(task_id); @@ -188,10 +206,12 @@ export class OrchestrationDb { coordinator_handle TEXT NOT NULL, poll_interval_ms INTEGER NOT NULL DEFAULT 2000, created_at TEXT NOT NULL DEFAULT (datetime('now')), - completed_at TEXT + completed_at TEXT, + workspace_key TEXT ); `) this.createUndeliveredInboxIndexIfPossible() + this.createWorkspaceScopeIndexesIfPossible() } // Why: CREATE TABLE IF NOT EXISTS won't alter existing DBs; migrate in a txn that bumps user_version only on success (atomic all-or-nothing). @@ -279,7 +299,27 @@ export class OrchestrationDb { this.db.exec(`ALTER TABLE messages ADD COLUMN sender_pane_key TEXT`) } } + // v6 → v7: add nullable workspace_key to the four orchestration tables so + // concurrent orchestrators in different worktrees scope their own runs, + // tasks, dispatches, and messages (#4389). Existing rows stay NULL, which + // every scoped read still matches (legacy/global), preserving behavior for + // single-orchestrator installs. + if (current < 7) { + if (!this.hasColumn('coordinator_runs', 'workspace_key')) { + this.db.exec(`ALTER TABLE coordinator_runs ADD COLUMN workspace_key TEXT`) + } + if (!this.hasColumn('tasks', 'workspace_key')) { + this.db.exec(`ALTER TABLE tasks ADD COLUMN workspace_key TEXT`) + } + if (!this.hasColumn('dispatch_contexts', 'workspace_key')) { + this.db.exec(`ALTER TABLE dispatch_contexts ADD COLUMN workspace_key TEXT`) + } + if (!this.hasColumn('messages', 'workspace_key')) { + this.db.exec(`ALTER TABLE messages ADD COLUMN workspace_key TEXT`) + } + } this.createUndeliveredInboxIndexIfPossible() + this.createWorkspaceScopeIndexesIfPossible() this.db.pragma(`user_version = ${SCHEMA_VERSION}`) this.db.exec('COMMIT') @@ -304,6 +344,34 @@ export class OrchestrationDb { `) } + // Why: the (workspace_key, status) indexes can only be created after the + // v6 → v7 ALTER adds the column to upgraded on-disk tables. createTables() + // runs before migrate(), so attaching these indexes there would fail against + // a pre-v7 schema; gating on the column keeps both the fresh-install and + // upgrade paths idempotent. + private createWorkspaceScopeIndexesIfPossible(): void { + if (this.hasColumn('coordinator_runs', 'workspace_key')) { + this.db.exec(` + CREATE INDEX IF NOT EXISTS idx_coordinator_runs_workspace_status + ON coordinator_runs(workspace_key, status); + CREATE INDEX IF NOT EXISTS idx_coordinator_runs_handle_status + ON coordinator_runs(coordinator_handle, status) + `) + } + if (this.hasColumn('tasks', 'workspace_key')) { + this.db.exec(` + CREATE INDEX IF NOT EXISTS idx_tasks_workspace_status + ON tasks(workspace_key, status) + `) + } + if (this.hasColumn('dispatch_contexts', 'workspace_key')) { + this.db.exec(` + CREATE INDEX IF NOT EXISTS idx_dispatch_workspace_status + ON dispatch_contexts(workspace_key, status) + `) + } + } + // Why: sqlite_master holds the table's CREATE SQL incl. the CHECK — cheapest reliable probe for whether it already allows 'heartbeat'. private messagesTypeCheckAllowsHeartbeat(): boolean { const row = this.db @@ -324,11 +392,12 @@ export class OrchestrationDb { threadId?: string payload?: string senderPaneKey?: string + workspaceKey?: string | null }): MessageRow { const id = generateId('msg') const stmt = this.db.prepare(` - INSERT INTO messages (id, from_handle, to_handle, subject, body, type, priority, thread_id, payload, sender_pane_key) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + INSERT INTO messages (id, from_handle, to_handle, subject, body, type, priority, thread_id, payload, sender_pane_key, workspace_key) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) `) stmt.run( id, @@ -340,28 +409,36 @@ export class OrchestrationDb { msg.priority ?? 'normal', msg.threadId ?? null, msg.payload ?? null, - msg.senderPaneKey ?? null + msg.senderPaneKey ?? null, + msg.workspaceKey ?? null ) return exposeMessageTimestamps( this.db.prepare('SELECT * FROM messages WHERE id = ?').get(id) as MessageRow ) } - getUnreadMessages(toHandle: string, types?: MessageType[]): MessageRow[] { + getUnreadMessages( + toHandle: string, + types?: MessageType[], + workspaceKey?: string | null + ): MessageRow[] { + const scope = workspaceScopeClause(workspaceKey) if (types && types.length > 0) { const placeholders = types.map(() => '?').join(',') return exposeMessageListTimestamps( this.db .prepare( - `SELECT * FROM messages WHERE to_handle = ? AND read = 0 AND type IN (${placeholders}) ORDER BY sequence` + `SELECT * FROM messages WHERE to_handle = ? AND read = 0 AND type IN (${placeholders})${scope.clause} ORDER BY sequence` ) - .all(toHandle, ...types) as MessageRow[] + .all(toHandle, ...types, ...scope.params) as MessageRow[] ) } return exposeMessageListTimestamps( this.db - .prepare('SELECT * FROM messages WHERE to_handle = ? AND read = 0 ORDER BY sequence') - .all(toHandle) as MessageRow[] + .prepare( + `SELECT * FROM messages WHERE to_handle = ? AND read = 0${scope.clause} ORDER BY sequence` + ) + .all(toHandle, ...scope.params) as MessageRow[] ) } @@ -509,6 +586,7 @@ export class OrchestrationDb { deps?: string[] parentId?: string createdByTerminalHandle?: string + workspaceKey?: string | null }): TaskRow { const id = generateId('task') const depsJson = JSON.stringify(task.deps ?? []) @@ -521,7 +599,7 @@ export class OrchestrationDb { }) this.db .prepare( - 'INSERT INTO tasks (id, parent_id, created_by_terminal_handle, task_title, display_name, spec, status, deps) VALUES (?, ?, ?, ?, ?, ?, ?, ?)' + 'INSERT INTO tasks (id, parent_id, created_by_terminal_handle, task_title, display_name, spec, status, deps, workspace_key) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)' ) .run( id, @@ -531,7 +609,8 @@ export class OrchestrationDb { display.displayName || null, task.spec, status, - depsJson + depsJson, + task.workspaceKey ?? null ) return this.db.prepare('SELECT * FROM tasks WHERE id = ?').get(id) as TaskRow } @@ -540,16 +619,30 @@ export class OrchestrationDb { return this.db.prepare('SELECT * FROM tasks WHERE id = ?').get(id) as TaskRow | undefined } - listTasks(filter?: { status?: TaskStatus; ready?: boolean }): TaskRow[] { + listTasks(filter?: { + status?: TaskStatus + ready?: boolean + workspaceKey?: string | null + }): TaskRow[] { + const scope = workspaceScopeClause(filter?.workspaceKey) if (filter?.ready) { return this.db - .prepare("SELECT * FROM tasks WHERE status = 'ready' ORDER BY created_at") - .all() as TaskRow[] + .prepare(`SELECT * FROM tasks WHERE status = 'ready'${scope.clause} ORDER BY created_at`) + .all(...scope.params) as TaskRow[] } if (filter?.status) { return this.db - .prepare('SELECT * FROM tasks WHERE status = ? ORDER BY created_at') - .all(filter.status) as TaskRow[] + .prepare(`SELECT * FROM tasks WHERE status = ?${scope.clause} ORDER BY created_at`) + .all(filter.status, ...scope.params) as TaskRow[] + } + if (scope.clause) { + // Why: the no-filter branch has no existing WHERE, so the scope fragment's + // leading ` AND ` would be a syntax error; emit a WHERE form instead. + return this.db + .prepare( + 'SELECT * FROM tasks WHERE (workspace_key = ? OR workspace_key IS NULL) ORDER BY created_at' + ) + .all(...scope.params) as TaskRow[] } return this.db.prepare('SELECT * FROM tasks ORDER BY created_at').all() as TaskRow[] } @@ -664,12 +757,22 @@ export class OrchestrationDb { const priorFailures = prior?.max_failures ?? 0 const id = generateId('ctx') + // Why (#4389): a dispatch inherits its task's workspace_key so the + // coordinator that owns the task is the only one whose stale-dispatch and + // idle-terminal scans see it; legacy tasks carry NULL and stay global. this.db .prepare( - `INSERT INTO dispatch_contexts (id, task_id, assignee_handle, assignee_pane_key, status, failure_count, dispatched_at) - VALUES (?, ?, ?, ?, 'dispatched', ?, datetime('now'))` + `INSERT INTO dispatch_contexts (id, task_id, assignee_handle, assignee_pane_key, status, failure_count, dispatched_at, workspace_key) + VALUES (?, ?, ?, ?, 'dispatched', ?, datetime('now'), ?)` + ) + .run( + id, + taskId, + assigneeHandle, + assigneePaneKey ?? null, + priorFailures, + task.workspace_key ?? null ) - .run(id, taskId, assigneeHandle, assigneePaneKey ?? null, priorFailures) this.db.prepare("UPDATE tasks SET status = 'dispatched' WHERE id = ?").run(taskId) @@ -771,16 +874,17 @@ export class OrchestrationDb { } // Why: dispatched_at grace skips workers still within their first heartbeat interval; julianday() vs raw-TEXT compare avoids misflagging space-format timestamps as stale (#8452). - getStaleDispatches(thresholdIso: string): DispatchContextRow[] { + getStaleDispatches(thresholdIso: string, workspaceKey?: string | null): DispatchContextRow[] { + const scope = workspaceScopeClause(workspaceKey) return this.db .prepare( `SELECT * FROM dispatch_contexts WHERE status = 'dispatched' AND dispatched_at IS NOT NULL AND julianday(dispatched_at) < julianday(?) - AND (last_heartbeat_at IS NULL OR julianday(last_heartbeat_at) < julianday(?))` + AND (last_heartbeat_at IS NULL OR julianday(last_heartbeat_at) < julianday(?))${scope.clause}` ) - .all(thresholdIso, thresholdIso) as DispatchContextRow[] + .all(thresholdIso, thresholdIso, ...scope.params) as DispatchContextRow[] } failDispatch(ctxId: string, error: string): DispatchContextRow | undefined { @@ -892,13 +996,20 @@ export class OrchestrationDb { spec: string coordinatorHandle: string pollIntervalMs?: number + workspaceKey?: string | null }): CoordinatorRun { const id = generateId('run') this.db .prepare( - "INSERT INTO coordinator_runs (id, spec, status, coordinator_handle, poll_interval_ms) VALUES (?, ?, 'running', ?, ?)" + "INSERT INTO coordinator_runs (id, spec, status, coordinator_handle, poll_interval_ms, workspace_key) VALUES (?, ?, 'running', ?, ?, ?)" + ) + .run( + id, + run.spec, + run.coordinatorHandle, + run.pollIntervalMs ?? 2000, + run.workspaceKey ?? null ) - .run(id, run.spec, run.coordinatorHandle, run.pollIntervalMs ?? 2000) return this.db.prepare('SELECT * FROM coordinator_runs WHERE id = ?').get(id) as CoordinatorRun } @@ -919,12 +1030,38 @@ export class OrchestrationDb { return this.getCoordinatorRun(id) } - getActiveCoordinatorRun(): CoordinatorRun | undefined { + // Why (#4389): an optional workspaceKey scopes the active-run lookup so the + // start/stop guard and worker attribution see only THIS workspace's run (plus + // legacy NULL runs). Called with no key it keeps the original global behavior + // for callers that have no worktree context. + getActiveCoordinatorRun(workspaceKey?: string | null): CoordinatorRun | undefined { + const scope = workspaceScopeClause(workspaceKey) return this.db .prepare( - "SELECT * FROM coordinator_runs WHERE status = 'running' ORDER BY created_at DESC LIMIT 1" + `SELECT * FROM coordinator_runs WHERE status = 'running'${scope.clause} ORDER BY created_at DESC LIMIT 1` ) - .get() as CoordinatorRun | undefined + .get(...scope.params) as CoordinatorRun | undefined + } + + // Why (#4389): strict per-workspace lookup (no legacy NULL fallback) so a + // worker terminal whose dispatch carries a workspace_key is attributed only + // to the coordinator that actually owns that workspace, never the most-recent + // global run. Callers with a known dispatch workspace use this; callers + // without one fall back to getActiveCoordinatorRun(). + getActiveCoordinatorRunForWorkspace(workspaceKey: string): CoordinatorRun | undefined { + return this.db + .prepare( + "SELECT * FROM coordinator_runs WHERE status = 'running' AND workspace_key = ? ORDER BY created_at DESC LIMIT 1" + ) + .get(workspaceKey) as CoordinatorRun | undefined + } + + getActiveCoordinatorRunForHandle(handle: string): CoordinatorRun | undefined { + return this.db + .prepare( + "SELECT * FROM coordinator_runs WHERE coordinator_handle = ? AND status = 'running' ORDER BY created_at DESC LIMIT 1" + ) + .get(handle) as CoordinatorRun | undefined } // ── Queries for Coordinator ── diff --git a/src/main/runtime/orchestration/formatter.test.ts b/src/main/runtime/orchestration/formatter.test.ts index ea8ae94bc19..5a7d938e5b6 100644 --- a/src/main/runtime/orchestration/formatter.test.ts +++ b/src/main/runtime/orchestration/formatter.test.ts @@ -18,6 +18,7 @@ function makeMessage(overrides: Partial = {}): MessageRow { created_at: '2026-01-01T00:00:00Z', delivered_at: null, sender_pane_key: null, + workspace_key: null, ...overrides } } diff --git a/src/main/runtime/orchestration/types.ts b/src/main/runtime/orchestration/types.ts index 58b86e41b08..e7421d1e5fa 100644 --- a/src/main/runtime/orchestration/types.ts +++ b/src/main/runtime/orchestration/types.ts @@ -18,6 +18,9 @@ export type GateStatus = 'pending' | 'resolved' | 'timeout' export type CoordinatorStatus = 'idle' | 'running' | 'completed' | 'failed' +// Why (#4389): `workspace_key` scopes a row to the worktree of the orchestrator +// that created it. NULL = legacy/global (pre-v6 rows or callers without a +// worktree), which every scoped read still matches. export type MessageRow = { id: string from_handle: string @@ -33,6 +36,7 @@ export type MessageRow = { created_at: string delivered_at: string | null sender_pane_key: string | null + workspace_key: string | null } export type TaskRow = { @@ -47,6 +51,7 @@ export type TaskRow = { result: string | null created_at: string completed_at: string | null + workspace_key: string | null } export type DispatchContextRow = { @@ -61,6 +66,7 @@ export type DispatchContextRow = { completed_at: string | null created_at: string last_heartbeat_at: string | null + workspace_key: string | null } export type DecisionGateRow = { @@ -82,4 +88,5 @@ export type CoordinatorRun = { poll_interval_ms: number created_at: string completed_at: string | null + workspace_key: string | null } diff --git a/src/main/runtime/rpc/methods/orchestration-gates.test.ts b/src/main/runtime/rpc/methods/orchestration-gates.test.ts new file mode 100644 index 00000000000..3a15ae14bb0 --- /dev/null +++ b/src/main/runtime/rpc/methods/orchestration-gates.test.ts @@ -0,0 +1,191 @@ +import { describe, expect, it, vi } from 'vitest' +import { RpcDispatcher } from '../dispatcher' +import type { RpcRequest, RpcResponse } from '../core' +import { ORCHESTRATION_GATE_METHODS } from './orchestration-gates' +import { OrchestrationDb } from '../../orchestration/db' +import type { OrcaRuntimeService } from '../../orca-runtime' + +// Why (#4389): the start/stop guard must scope by workspace so two orchestrators +// in different worktrees of one Orca instance neither block nor stop each other. +// A fake runtime maps each worktree selector to a stable workspace key and +// stubs the CoordinatorRuntime surface so the background loop can start without +// touching real terminals or git. + +function makeRequest(method: string, params: unknown = {}): RpcRequest { + return { id: `req_${method}`, authToken: 'tok', method, params } +} + +function expectOk(response: RpcResponse): Extract { + if (!response.ok) { + throw new Error(`expected ok response, got error: ${response.error.message}`) + } + return response +} + +function runId(response: RpcResponse): string { + return (expectOk(response).result as { runId: string }).runId +} + +// Maps `worktree:wt_a` style selectors straight through; the coordinator loop's +// terminal calls are no-ops so dispatchReadyTasks stays inert and harmless. +function makeRuntime(db: OrchestrationDb): OrcaRuntimeService { + return { + getRuntimeId: () => 'test-runtime', + getOrchestrationDb: () => db, + resolveWorkspaceKeyForSelector: vi.fn(async (selector?: string | null) => + selector ? selector : null + ), + // CoordinatorRuntime surface — no terminals available, so the loop just + // polls without dispatching anything. + listTerminals: vi.fn(async () => ({ terminals: [] })), + createTerminal: vi.fn(async () => { + throw new Error('no terminals in test') + }), + sendTerminal: vi.fn(async () => ({})), + waitForTerminal: vi.fn(async (handle: string) => ({ handle, condition: 'idle' })), + probeWorktreeDrift: vi.fn(async () => null) + } as unknown as OrcaRuntimeService +} + +// Why: orchestration.run fires the coordinator loop in the background. A short +// poll interval plus runStop for each started workspace lets every loop observe +// `stopped` and finish its final DB write before the test closes the DB, so no +// post-close write leaks as an unhandled rejection. +const SHORT_POLL_MS = 5 + +async function stopAndDrain( + dispatcher: RpcDispatcher, + worktrees: (string | undefined)[] +): Promise { + for (const worktree of worktrees) { + await dispatcher.dispatch(makeRequest('orchestration.runStop', worktree ? { worktree } : {})) + } + await new Promise((resolve) => setTimeout(resolve, SHORT_POLL_MS * 4)) +} + +describe('orchestration start/stop guard scoping (#4389)', () => { + it('does not reject a run in worktree B while a run is active in worktree A', async () => { + const db = new OrchestrationDb(':memory:') + const runtime = makeRuntime(db) + const dispatcher = new RpcDispatcher({ runtime, methods: ORCHESTRATION_GATE_METHODS }) + try { + // Each workspace owns its own ready task so the coordinator's decompose + // step finds work instead of throwing "No tasks found". + 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 + }) + ) + ) + + // The 2nd run in a different workspace must succeed, not throw + // "Coordinator already running". + const runBId = runId( + await dispatcher.dispatch( + makeRequest('orchestration.run', { + spec: 'b', + from: 'coord_b', + worktree: 'worktree:wt_b', + pollIntervalMs: SHORT_POLL_MS + }) + ) + ) + + expect(runAId).not.toBe(runBId) + expect(db.getActiveCoordinatorRun('worktree:wt_a')?.id).toBe(runAId) + expect(db.getActiveCoordinatorRun('worktree:wt_b')?.id).toBe(runBId) + + await stopAndDrain(dispatcher, ['worktree:wt_a', 'worktree:wt_b']) + } finally { + db.close() + } + }) + + it('rejects a second run in the SAME worktree', 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' }) + + expectOk( + await dispatcher.dispatch( + makeRequest('orchestration.run', { + spec: 'a', + from: 'coord_a', + worktree: 'worktree:wt_a', + pollIntervalMs: SHORT_POLL_MS + }) + ) + ) + const second = await dispatcher.dispatch( + makeRequest('orchestration.run', { + spec: 'a2', + from: 'coord_a2', + worktree: 'worktree:wt_a', + pollIntervalMs: SHORT_POLL_MS + }) + ) + expect(second.ok).toBe(false) + if (!second.ok) { + expect(second.error.message).toMatch(/Coordinator already running/) + } + + await stopAndDrain(dispatcher, ['worktree:wt_a']) + } finally { + db.close() + } + }) + + it('runStop for worktree B leaves worktree A running', 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 + }) + ) + ) + expectOk( + await dispatcher.dispatch( + makeRequest('orchestration.run', { + spec: 'b', + from: 'coord_b', + worktree: 'worktree:wt_b', + pollIntervalMs: SHORT_POLL_MS + }) + ) + ) + + const stopB = expectOk( + await dispatcher.dispatch( + makeRequest('orchestration.runStop', { worktree: 'worktree:wt_b' }) + ) + ) + expect((stopB.result as { stopped: boolean }).stopped).toBe(true) + + // A's run must still be the active run for its workspace after stopping B. + expect(db.getActiveCoordinatorRun('worktree:wt_a')?.id).toBe(runAId) + + await stopAndDrain(dispatcher, ['worktree:wt_a']) + } finally { + db.close() + } + }) +}) diff --git a/src/main/runtime/rpc/methods/orchestration-gates.ts b/src/main/runtime/rpc/methods/orchestration-gates.ts index ca16b9719f0..cbacc05a73d 100644 --- a/src/main/runtime/rpc/methods/orchestration-gates.ts +++ b/src/main/runtime/rpc/methods/orchestration-gates.ts @@ -4,10 +4,19 @@ import { OptionalFiniteNumber, OptionalString, requiredString } from '../schemas import type { GateStatus } from '../../orchestration/db' import { Coordinator } from '../../orchestration/coordinator' -// Why: the coordinator instance is stored at module scope so orchestration.runStop -// can signal it to halt. Only one coordinator can run at a time (enforced by -// the DB's active-run check), so a single reference suffices. -let activeCoordinator: Coordinator | null = null +// 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 +// pre-#4389 single-coordinator semantics are preserved. +const GLOBAL_COORDINATOR_KEY = '__global__' +const activeCoordinators = new Map() + +function coordinatorMapKey(workspaceKey: string | null): string { + return workspaceKey ?? GLOBAL_COORDINATOR_KEY +} const RunParams = z.object({ spec: requiredString('Missing --spec'), @@ -17,7 +26,9 @@ const RunParams = z.object({ worktree: OptionalString }) -const RunStopParams = z.object({}) +const RunStopParams = z.object({ + worktree: OptionalString +}) const GateCreateParams = z.object({ task: requiredString('Missing --task'), @@ -43,10 +54,15 @@ export const ORCHESTRATION_GATE_METHODS: RpcMethod[] = [ defineMethod({ name: 'orchestration.run', params: RunParams, - handler: (params, { runtime }) => { + handler: async (params, { runtime }) => { const db = runtime.getOrchestrationDb() - const existing = db.getActiveCoordinatorRun() + // 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) + const existing = db.getActiveCoordinatorRun(workspaceKey) if (existing) { throw new Error(`Coordinator already running: ${existing.id}`) } @@ -57,23 +73,29 @@ export const ORCHESTRATION_GATE_METHODS: RpcMethod[] = [ coordinatorHandle, pollIntervalMs: params.pollIntervalMs, maxConcurrent: params.maxConcurrent, - worktree: params.worktree + worktree: params.worktree, + workspaceKey }) - activeCoordinator = coordinator + const mapKey = coordinatorMapKey(workspaceKey) + activeCoordinators.set(mapKey, coordinator) const run = db.createCoordinatorRun({ spec: params.spec, coordinatorHandle, - pollIntervalMs: params.pollIntervalMs + pollIntervalMs: params.pollIntervalMs, + workspaceKey }) // Why: fire-and-forget — the coordinator loop runs in the event loop // background. Results are persisted to the DB; callers query via // orchestration.taskList or orchestration.runStatus. coordinator.runFromExistingRun(run.id).finally(() => { - if (activeCoordinator === coordinator) { - activeCoordinator = null + // Why (#4389): only clear this workspace's slot, and only if it still + // points at this coordinator — a newer run for the same workspace must + // not be evicted by an older one's completion. + if (activeCoordinators.get(mapKey) === coordinator) { + activeCoordinators.delete(mapKey) } }) @@ -84,16 +106,21 @@ export const ORCHESTRATION_GATE_METHODS: RpcMethod[] = [ defineMethod({ name: 'orchestration.runStop', params: RunStopParams, - handler: (_params, { runtime }) => { + handler: async (params, { runtime }) => { const db = runtime.getOrchestrationDb() - const run = db.getActiveCoordinatorRun() + // 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') } - if (activeCoordinator) { - activeCoordinator.stop() - activeCoordinator = null + const mapKey = coordinatorMapKey(workspaceKey) + const coordinator = activeCoordinators.get(mapKey) + if (coordinator) { + coordinator.stop() + activeCoordinators.delete(mapKey) } return { runId: run.id, stopped: true } diff --git a/src/main/runtime/rpc/methods/orchestration.ts b/src/main/runtime/rpc/methods/orchestration.ts index 7a731f91774..4ad0c973ebd 100644 --- a/src/main/runtime/rpc/methods/orchestration.ts +++ b/src/main/runtime/rpc/methods/orchestration.ts @@ -374,7 +374,7 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [ defineMethod({ name: 'orchestration.taskCreate', params: TaskCreateParams, - handler: (params, { runtime }) => { + handler: async (params, { runtime }) => { const db = runtime.getOrchestrationDb() let deps: string[] | undefined if (params.deps) { @@ -388,13 +388,20 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [ throw new Error('Invalid --deps: must be a JSON array of task IDs') } } + // Why (#4389): scope the task to the creating terminal's worktree so the + // orchestrator for that workspace is the only one whose loop picks it up. + // Null (no/stale handle) stays legacy/global, matching prior behavior. + const workspaceKey = await runtime.resolveWorkspaceKeyForTerminalHandle( + params.callerTerminalHandle + ) const task = db.createTask({ spec: params.spec, taskTitle: params.taskTitle, displayName: params.displayName, deps, parentId: params.parent, - createdByTerminalHandle: params.callerTerminalHandle + createdByTerminalHandle: params.callerTerminalHandle, + workspaceKey }) return { task } } diff --git a/src/main/runtime/runtime-rpc.test.ts b/src/main/runtime/runtime-rpc.test.ts index 7744bbe67eb..8d37fce054c 100644 --- a/src/main/runtime/runtime-rpc.test.ts +++ b/src/main/runtime/runtime-rpc.test.ts @@ -4082,5 +4082,79 @@ describe('OrcaRuntimeRpcServer', () => { realPrototype.exec = originalExec } }) + + // Why (#4389): v5 → v6 must add the nullable workspace_key column to the + // four orchestration tables, bump user_version to 6, and leave existing + // rows as NULL (legacy/global) so single-orchestrator installs are intact. + it('upgrades a v5 on-disk DB to v6 with workspace_key columns, preserving rows as NULL', () => { + const tmpPath = join(mkdtempSync(join(tmpdir(), 'orca-orch-mig-v6-')), 'orch.sqlite') + const seed = new Database(tmpPath) + // Minimal pre-v6 (v5) shapes — no workspace_key on any table. + seed.exec(` + CREATE TABLE coordinator_runs ( + id TEXT PRIMARY KEY, spec TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'idle', + coordinator_handle TEXT NOT NULL, + poll_interval_ms INTEGER NOT NULL DEFAULT 2000, + created_at TEXT NOT NULL DEFAULT (datetime('now')), completed_at TEXT + ); + CREATE TABLE tasks ( + id TEXT PRIMARY KEY, parent_id TEXT, created_by_terminal_handle TEXT, + task_title TEXT, display_name TEXT, spec TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'pending', deps TEXT NOT NULL DEFAULT '[]', + result TEXT, created_at TEXT NOT NULL DEFAULT (datetime('now')), completed_at TEXT + ); + CREATE TABLE dispatch_contexts ( + id TEXT PRIMARY KEY, task_id TEXT NOT NULL, assignee_handle TEXT, + status TEXT NOT NULL DEFAULT 'pending', failure_count INTEGER NOT NULL DEFAULT 0, + last_failure TEXT, dispatched_at TEXT, completed_at TEXT, + created_at TEXT NOT NULL DEFAULT (datetime('now')), last_heartbeat_at TEXT + ); + CREATE TABLE messages ( + id TEXT NOT NULL, from_handle TEXT NOT NULL, to_handle TEXT NOT NULL, + subject TEXT NOT NULL, body TEXT NOT NULL DEFAULT '', + type TEXT NOT NULL DEFAULT 'status', priority TEXT NOT NULL DEFAULT 'normal', + thread_id TEXT, payload TEXT, read INTEGER NOT NULL DEFAULT 0, + sequence INTEGER PRIMARY KEY AUTOINCREMENT, + created_at TEXT NOT NULL DEFAULT (datetime('now')), delivered_at TEXT + ); + INSERT INTO coordinator_runs (id, spec, status, coordinator_handle) + VALUES ('run_legacy', 'legacy', 'running', 'coord'); + INSERT INTO tasks (id, spec, status) VALUES ('task_legacy', 'legacy', 'ready'); + INSERT INTO dispatch_contexts (id, task_id, assignee_handle, status) + VALUES ('ctx_legacy', 'task_legacy', 'term_legacy', 'dispatched'); + INSERT INTO messages (id, from_handle, to_handle, subject) + VALUES ('msg_legacy', 'a', 'coord', 'hi'); + `) + seed.pragma('user_version = 5') + seed.close() + + const db = new OrchestrationDb(tmpPath) + try { + const probe = (db as unknown as { db: Database.Database }).db + const version = probe.pragma('user_version', { simple: true }) as number + expect(version).toBe(6) + + const hasColumn = (table: string): boolean => + (probe.pragma(`table_info(${table})`) as { name: string }[]).some( + (r) => r.name === 'workspace_key' + ) + expect(hasColumn('coordinator_runs')).toBe(true) + expect(hasColumn('tasks')).toBe(true) + expect(hasColumn('dispatch_contexts')).toBe(true) + expect(hasColumn('messages')).toBe(true) + + // Pre-existing rows survive the migration with NULL workspace_key, so a + // scoped read still treats them as legacy/global. + expect(db.getActiveCoordinatorRun('worktree:wt_a')?.id).toBe('run_legacy') + const legacyTask = db.getTask('task_legacy') + expect(legacyTask?.workspace_key).toBeNull() + expect(db.listTasks({ workspaceKey: 'worktree:wt_a' }).map((t) => t.id)).toContain( + 'task_legacy' + ) + } finally { + db.close() + } + }) }) })