mirror of
https://github.com/stablyai/orca.git
synced 2026-10-19 16:03:26 +00:00
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 <help@stably.ai>
This commit is contained in:
Neil
committed by
Brennan Benson
Brennan Benson
co-authored by
Orca
Orca
parent
f8b430f725
commit
5f0677ec0a
@@ -685,11 +685,16 @@ export const ORCHESTRATION_HANDLERS: Record<string, CommandHandler> = {
|
||||
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`)
|
||||
},
|
||||
|
||||
|
||||
+1
-1
@@ -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 <selector>]
|
||||
orchestration gate-create Create a decision gate blocking a task
|
||||
orchestration gate-resolve Resolve a pending decision gate
|
||||
orchestration gate-list List decision gates
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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<OrchestrationDb['getActiveCoordinatorRun']> {
|
||||
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<string | null> {
|
||||
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<string | null> {
|
||||
if (!handle) {
|
||||
return null
|
||||
}
|
||||
try {
|
||||
const terminal = await this.showTerminal(handle)
|
||||
return this.resolveWorkspaceKeyForSelector(`id:${terminal.worktreeId}`)
|
||||
} catch {
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
async hydrateInferredWorktreeLineage(): Promise<void> {
|
||||
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
|
||||
|
||||
@@ -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<Omit<CoordinatorOptions, 'onLog' | 'worktree'>> & {
|
||||
private opts: Required<Omit<CoordinatorOptions, 'onLog' | 'worktree' | 'workspaceKey'>> & {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<string[]> {
|
||||
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<string>()
|
||||
|
||||
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<void> {
|
||||
return new Promise((resolve) => {
|
||||
setTimeout(resolve, ms)
|
||||
|
||||
@@ -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)
|
||||
})
|
||||
})
|
||||
@@ -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 ──
|
||||
|
||||
@@ -18,6 +18,7 @@ function makeMessage(overrides: Partial<MessageRow> = {}): MessageRow {
|
||||
created_at: '2026-01-01T00:00:00Z',
|
||||
delivered_at: null,
|
||||
sender_pane_key: null,
|
||||
workspace_key: null,
|
||||
...overrides
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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<RpcResponse, { ok: true }> {
|
||||
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<void> {
|
||||
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()
|
||||
}
|
||||
})
|
||||
})
|
||||
@@ -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<string, Coordinator>()
|
||||
|
||||
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 }
|
||||
|
||||
@@ -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 }
|
||||
}
|
||||
|
||||
@@ -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()
|
||||
}
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user