mirror of
https://github.com/stablyai/orca.git
synced 2026-10-19 00:03:20 +00:00
fix(orchestration): harden workspace isolation after rebase
This commit is contained in:
@@ -53,7 +53,7 @@ Do not use orchestration merely because the user says "hand off", "handoff", "ha
|
||||
|
||||
## Ownership
|
||||
|
||||
Orchestration messages and tasks are runtime-global. Lifecycle authority comes from the payload `taskId` + `dispatchId` of the active dispatch, verified against the dispatched pane. Terminal handles are routing metadata — a pane can receive a new handle after restart — so never accept or reject lifecycle provenance by comparing handles. Send `worker_done` and `heartbeat` from the worker's own terminal; the runtime ignores them when sent from a different pane.
|
||||
Orchestration state is persisted by the runtime. Rows with a `workspace_key` are owned by that workspace; NULL rows are legacy/global and remain visible to scoped coordinators for compatibility. Lifecycle authority comes from the payload `taskId` + `dispatchId` of the active dispatch, verified against the dispatched pane. Terminal handles are routing metadata — a pane can receive a new handle after restart — so never accept or reject lifecycle provenance by comparing handles. Send `worker_done` and `heartbeat` from the worker's own terminal; the runtime ignores them when sent from a different pane.
|
||||
|
||||
Classify inherited context before sending lifecycle messages:
|
||||
|
||||
@@ -130,11 +130,11 @@ Dispatch rules:
|
||||
orca orchestration gate-create --task <task_id> --question <text> [--options <json_array>] [--json]
|
||||
orca orchestration gate-resolve --id <gate_id> --resolution <text> [--json]
|
||||
orca orchestration gate-list [--task <task_id>] [--status <status>] [--json]
|
||||
orca orchestration run --spec <text> [--from <handle>] [--poll-interval-ms <n>] [--max-concurrent <n>] [--worktree <selector>] [--json]
|
||||
orca orchestration run-stop [--json]
|
||||
orca orchestration run --spec <text> [--from <handle>] [--poll-interval-ms <n>] [--max-concurrent <n>] [--worktree <selector|all>] [--json]
|
||||
orca orchestration run-stop [--worktree <selector|all>] [--json]
|
||||
```
|
||||
|
||||
`run` returns immediately with a run ID. Query progress with `task-list`. Use `ask` for worker-to-coordinator questions; it creates a `decision_gate` message that the coordinator answers with `reply`. Use `gate-create` only for coordinator-managed task DAG decisions, not for answering a worker's `ask`.
|
||||
`run` returns immediately with a run ID. Query progress with `task-list`. Pass `--worktree <selector>` to isolate a coordinator to one workspace; omit it or pass `--worktree all` for the global coordinator, which cannot run alongside scoped coordinators. `run-stop --worktree <selector>` stops that workspace exactly, while `--worktree all` stops the global coordinator; without a selector, stop succeeds only when exactly one coordinator is active. Use `ask` for worker-to-coordinator questions; it creates a `decision_gate` message that the coordinator answers with `reply`. Use `gate-create` only for coordinator-managed task DAG decisions, not for answering a worker's `ask`.
|
||||
|
||||
Recovery only: `orca orchestration reset --tasks|--messages|--all --json` clears runtime-global orchestration state. Do not run it during active coordination unless explicitly abandoning that state.
|
||||
|
||||
|
||||
@@ -53,7 +53,7 @@ Do not use orchestration merely because the user says "hand off", "handoff", "ha
|
||||
|
||||
## Ownership
|
||||
|
||||
Orchestration messages and tasks are runtime-global. Lifecycle authority comes from the payload `taskId` + `dispatchId` of the active dispatch, verified against the dispatched pane. Terminal handles are routing metadata — a pane can receive a new handle after restart — so never accept or reject lifecycle provenance by comparing handles. Send `worker_done` and `heartbeat` from the worker's own terminal; the runtime ignores them when sent from a different pane.
|
||||
Orchestration state is persisted by the runtime. Rows with a `workspace_key` are owned by that workspace; NULL rows are legacy/global and remain visible to scoped coordinators for compatibility. Lifecycle authority comes from the payload `taskId` + `dispatchId` of the active dispatch, verified against the dispatched pane. Terminal handles are routing metadata — a pane can receive a new handle after restart — so never accept or reject lifecycle provenance by comparing handles. Send `worker_done` and `heartbeat` from the worker's own terminal; the runtime ignores them when sent from a different pane.
|
||||
|
||||
Classify inherited context before sending lifecycle messages:
|
||||
|
||||
@@ -134,7 +134,7 @@ orca orchestration run --spec <text> [--from <handle>] [--poll-interval-ms <n>]
|
||||
orca orchestration run-stop [--worktree <selector|all>] [--json]
|
||||
```
|
||||
|
||||
`run` returns immediately with a run ID. `--worktree <selector>` scopes ownership to that workspace; `--worktree all` (or no `--worktree`) is the explicit global coordinator and cannot safely run alongside scoped coordinators because it sees all tasks. `run-stop` with no `--worktree` is only for the legacy single-active-run case; pass `--worktree <selector>` or `--worktree all` when more than one coordinator may be active. Query progress with `task-list`. Use `ask` for worker-to-coordinator questions; it creates a `decision_gate` message that the coordinator answers with `reply`. Use `gate-create` only for coordinator-managed task DAG decisions, not for answering a worker's `ask`.
|
||||
`run` returns immediately with a run ID. Query progress with `task-list`. Pass `--worktree <selector>` to isolate a coordinator to one workspace; omit it or pass `--worktree all` for the global coordinator, which cannot run alongside scoped coordinators. `run-stop --worktree <selector>` stops that workspace exactly, while `--worktree all` stops the global coordinator; without a selector, stop succeeds only when exactly one coordinator is active. Use `ask` for worker-to-coordinator questions; it creates a `decision_gate` message that the coordinator answers with `reply`. Use `gate-create` only for coordinator-managed task DAG decisions, not for answering a worker's `ask`.
|
||||
|
||||
Recovery only: `orca orchestration reset --tasks|--messages|--all --json` clears runtime-global orchestration state. Do not run it during active coordination unless explicitly abandoning that state.
|
||||
|
||||
|
||||
File diff suppressed because one or more lines are too long
@@ -21242,18 +21242,19 @@ export class OrcaRuntimeService {
|
||||
|
||||
// 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.
|
||||
// private. Only an omitted handle is legacy/global: an explicit stale handle
|
||||
// must fail closed instead of silently creating a globally visible task.
|
||||
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 {
|
||||
const terminal = await this.showTerminal(handle)
|
||||
// Why: the live floating terminal intentionally has no managed worktree;
|
||||
// its orchestration tasks belong to the legacy/global scope.
|
||||
if (terminal.worktreeId === FLOATING_TERMINAL_WORKTREE_ID) {
|
||||
return null
|
||||
}
|
||||
return this.resolveWorkspaceKeyForSelector(`id:${terminal.worktreeId}`)
|
||||
}
|
||||
|
||||
async hydrateInferredWorktreeLineage(): Promise<void> {
|
||||
|
||||
@@ -0,0 +1,80 @@
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import { Coordinator, type CoordinatorRuntime } from './coordinator'
|
||||
import { OrchestrationDb } from './db'
|
||||
|
||||
function createRuntimeWithoutWorkers(): CoordinatorRuntime {
|
||||
return {
|
||||
async sendTerminalAgentPrompt() {
|
||||
return {}
|
||||
},
|
||||
async listTerminals() {
|
||||
return { terminals: [] }
|
||||
},
|
||||
async createTerminal() {
|
||||
throw new Error('no terminals in test')
|
||||
},
|
||||
async waitForTerminal(handle: string) {
|
||||
return { handle, condition: 'idle' }
|
||||
},
|
||||
async probeWorktreeDrift() {
|
||||
return null
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
describe('Coordinator workspace-scoped message payloads', () => {
|
||||
let db: OrchestrationDb
|
||||
|
||||
afterEach(() => db?.close())
|
||||
|
||||
it('does not mutate foreign tasks named by escalation or decision-gate payloads', async () => {
|
||||
db = new OrchestrationDb(':memory:')
|
||||
const ownWorkspace = 'worktree:wt_b'
|
||||
const foreignWorkspace = 'worktree:wt_a'
|
||||
db.createTask({ spec: 'keep coordinator alive', workspaceKey: ownWorkspace })
|
||||
|
||||
const escalationTask = db.createTask({
|
||||
spec: 'foreign dispatched task',
|
||||
workspaceKey: foreignWorkspace
|
||||
})
|
||||
const foreignDispatch = db.createDispatchContext(escalationTask.id, 'term_foreign')
|
||||
const gateTask = db.createTask({ spec: 'foreign gate task', workspaceKey: foreignWorkspace })
|
||||
|
||||
const escalation = db.insertMessage({
|
||||
from: 'term_b_worker',
|
||||
to: 'coord_b',
|
||||
subject: 'malicious foreign escalation',
|
||||
type: 'escalation',
|
||||
payload: JSON.stringify({ taskId: escalationTask.id }),
|
||||
workspaceKey: ownWorkspace
|
||||
})
|
||||
const decisionGate = db.insertMessage({
|
||||
from: 'term_b_worker',
|
||||
to: 'coord_b',
|
||||
subject: 'malicious foreign gate',
|
||||
type: 'decision_gate',
|
||||
payload: JSON.stringify({ taskId: gateTask.id, question: 'Block the foreign task?' }),
|
||||
workspaceKey: ownWorkspace
|
||||
})
|
||||
|
||||
const coordinator = new Coordinator(db, createRuntimeWithoutWorkers(), {
|
||||
spec: 'workspace B coordinator',
|
||||
coordinatorHandle: 'coord_b',
|
||||
pollIntervalMs: 5,
|
||||
workspaceKey: ownWorkspace
|
||||
})
|
||||
const run = coordinator.run()
|
||||
|
||||
await vi.waitFor(() => {
|
||||
expect(db.getMessageById(escalation.id)?.read).toBe(1)
|
||||
expect(db.getMessageById(decisionGate.id)?.read).toBe(1)
|
||||
})
|
||||
coordinator.stop()
|
||||
await run
|
||||
|
||||
expect(db.getDispatchContextById(foreignDispatch.id)?.status).toBe('dispatched')
|
||||
expect(db.getTask(escalationTask.id)?.status).toBe('dispatched')
|
||||
expect(db.getTask(gateTask.id)?.status).toBe('ready')
|
||||
expect(db.listGates({ taskId: gateTask.id })).toHaveLength(0)
|
||||
})
|
||||
})
|
||||
@@ -289,7 +289,9 @@ export class Coordinator {
|
||||
}
|
||||
|
||||
const task = this.db.getTask(taskId)
|
||||
if (!task || task.status === 'completed' || task.status === 'failed') {
|
||||
// Why (#4389): message payloads are routing input, not ownership proof;
|
||||
// never let one coordinator fail a task owned by another workspace.
|
||||
if (!task || !this.ownsTask(task) || task.status === 'completed' || task.status === 'failed') {
|
||||
return
|
||||
}
|
||||
|
||||
@@ -326,6 +328,13 @@ export class Coordinator {
|
||||
return
|
||||
}
|
||||
|
||||
const task = this.db.getTask(payload.taskId)
|
||||
// Why (#4389): a foreign task ID in a visible message must not let this
|
||||
// coordinator create a gate that blocks another workspace's task.
|
||||
if (!task || !this.ownsTask(task)) {
|
||||
return
|
||||
}
|
||||
|
||||
this.db.createGate({
|
||||
taskId: payload.taskId,
|
||||
question: payload.question,
|
||||
@@ -341,7 +350,7 @@ export class Coordinator {
|
||||
|
||||
private processDecisionGates(): void {
|
||||
// Why: the coordinator never auto-resolves gates (humans do, via orchestration.gateResolve) — that would defeat them as approval checkpoints.
|
||||
const pendingGates = this.db.listGates({ status: 'pending' })
|
||||
const pendingGates = this.db.listPendingGatesForWorkspace(this.opts.workspaceKey)
|
||||
for (const gate of pendingGates) {
|
||||
const task = this.db.getTask(gate.task_id)
|
||||
// Why (#4389): decision_gates have no workspace_key, so derive scope from
|
||||
|
||||
@@ -107,4 +107,28 @@ describe('OrchestrationDb workspace scoping', () => {
|
||||
expect(staleA).toContain(ctxA.id)
|
||||
expect(staleA).not.toContain(ctxB.id)
|
||||
})
|
||||
|
||||
it('lists only pending gates owned by the coordinator 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 gateA = d.createGate({ taskId: taskA.id, question: 'A?' })
|
||||
d.createGate({ taskId: taskB.id, question: 'B?' })
|
||||
|
||||
expect(d.listPendingGatesForWorkspace(KEY_A).map((gate) => gate.id)).toEqual([gateA.id])
|
||||
})
|
||||
|
||||
it('keeps terminals with a foreign active dispatch globally unavailable', () => {
|
||||
const d = createDb()
|
||||
d.insertMessage({ from: 'coord_a', to: 'term_a', subject: 'known terminal' })
|
||||
d.insertMessage({ from: 'coord_b', to: 'term_b', subject: 'known terminal' })
|
||||
const taskA = d.createTask({ spec: 'a-work', workspaceKey: KEY_A })
|
||||
const taskB = d.createTask({ spec: 'b-work', workspaceKey: KEY_B })
|
||||
d.createDispatchContext(taskA.id, 'term_a')
|
||||
d.createDispatchContext(taskB.id, 'term_b')
|
||||
|
||||
const idle = d.getIdleTerminals()
|
||||
expect(idle).not.toContain('term_a')
|
||||
expect(idle).not.toContain('term_b')
|
||||
})
|
||||
})
|
||||
|
||||
@@ -83,7 +83,7 @@ function exposeMessageListTimestamps(messages: MessageRow[]): MessageRow[] {
|
||||
return messages.map(exposeMessageTimestamps)
|
||||
}
|
||||
|
||||
// Why (#4389): scope a query to one workspace without hiding pre-v6 rows. A
|
||||
// Why (#4389): scope a query to one workspace without hiding pre-v7 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
|
||||
@@ -625,14 +625,21 @@ export class OrchestrationDb {
|
||||
workspaceKey?: string | null
|
||||
}): TaskRow[] {
|
||||
const scope = workspaceScopeClause(filter?.workspaceKey)
|
||||
// Why (#4389): without the scoped index hint SQLite prefers idx_tasks_status
|
||||
// and examines every workspace's rows on each coordinator poll.
|
||||
const scopedIndex = filter?.workspaceKey == null ? '' : ' INDEXED BY idx_tasks_workspace_status'
|
||||
if (filter?.ready) {
|
||||
return this.db
|
||||
.prepare(`SELECT * FROM tasks WHERE status = 'ready'${scope.clause} ORDER BY created_at`)
|
||||
.prepare(
|
||||
`SELECT * FROM tasks${scopedIndex} 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 = ?${scope.clause} ORDER BY created_at`)
|
||||
.prepare(
|
||||
`SELECT * FROM tasks${scopedIndex} WHERE status = ?${scope.clause} ORDER BY created_at`
|
||||
)
|
||||
.all(filter.status, ...scope.params) as TaskRow[]
|
||||
}
|
||||
if (scope.clause) {
|
||||
@@ -640,7 +647,7 @@ export class OrchestrationDb {
|
||||
// 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'
|
||||
`SELECT * FROM tasks${scopedIndex} WHERE (workspace_key = ? OR workspace_key IS NULL) ORDER BY created_at`
|
||||
)
|
||||
.all(...scope.params) as TaskRow[]
|
||||
}
|
||||
@@ -705,9 +712,13 @@ export class OrchestrationDb {
|
||||
|
||||
// Why: runs in the status-update transaction, so a completed task never leaves its ready children unpromoted.
|
||||
private promoteReadyTasks(completedTaskId: string): void {
|
||||
const completedTask = this.getTask(completedTaskId)
|
||||
const scope = workspaceScopeClause(completedTask?.workspace_key)
|
||||
const scopedIndex =
|
||||
completedTask?.workspace_key == null ? '' : ' INDEXED BY idx_tasks_workspace_status'
|
||||
const candidates = this.db
|
||||
.prepare("SELECT * FROM tasks WHERE status = 'pending'")
|
||||
.all() as TaskRow[]
|
||||
.prepare(`SELECT * FROM tasks${scopedIndex} WHERE status = 'pending'${scope.clause}`)
|
||||
.all(...scope.params) as TaskRow[]
|
||||
|
||||
for (const task of candidates) {
|
||||
const deps: string[] = JSON.parse(task.deps)
|
||||
@@ -876,9 +887,10 @@ 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, workspaceKey?: string | null): DispatchContextRow[] {
|
||||
const scope = workspaceScopeClause(workspaceKey)
|
||||
const scopedIndex = workspaceKey == null ? '' : ' INDEXED BY idx_dispatch_workspace_status'
|
||||
return this.db
|
||||
.prepare(
|
||||
`SELECT * FROM dispatch_contexts
|
||||
`SELECT * FROM dispatch_contexts${scopedIndex}
|
||||
WHERE status = 'dispatched'
|
||||
AND dispatched_at IS NOT NULL
|
||||
AND julianday(dispatched_at) < julianday(?)
|
||||
@@ -984,6 +996,23 @@ export class OrchestrationDb {
|
||||
.all() as DecisionGateRow[]
|
||||
}
|
||||
|
||||
listPendingGatesForWorkspace(workspaceKey?: string | null): DecisionGateRow[] {
|
||||
if (workspaceKey == null) {
|
||||
return this.listGates({ status: 'pending' })
|
||||
}
|
||||
// Why (#4389): decision gates inherit ownership from their task. Filtering
|
||||
// in SQL avoids every concurrent coordinator scanning every workspace's gates.
|
||||
return this.db
|
||||
.prepare(
|
||||
`SELECT g.* FROM decision_gates g
|
||||
INNER JOIN tasks t ON t.id = g.task_id
|
||||
WHERE g.status = 'pending'
|
||||
AND (t.workspace_key = ? OR t.workspace_key IS NULL)
|
||||
ORDER BY g.created_at`
|
||||
)
|
||||
.all(workspaceKey) as DecisionGateRow[]
|
||||
}
|
||||
|
||||
getGate(id: string): DecisionGateRow | undefined {
|
||||
return this.db.prepare('SELECT * FROM decision_gates WHERE id = ?').get(id) as
|
||||
| DecisionGateRow
|
||||
|
||||
@@ -19,7 +19,7 @@ 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
|
||||
// that created it. NULL = legacy/global (pre-v7 rows or callers without a
|
||||
// worktree), which every scoped read still matches.
|
||||
export type MessageRow = {
|
||||
id: string
|
||||
|
||||
@@ -0,0 +1,105 @@
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import type { RpcContext } from '../core'
|
||||
import { OrcaRuntimeService } from '../../orca-runtime'
|
||||
import { OrchestrationDb } from '../../orchestration/db'
|
||||
import { FLOATING_TERMINAL_WORKTREE_ID } from '../../../../shared/constants'
|
||||
import { ORCHESTRATION_METHODS } from './orchestration'
|
||||
|
||||
function findMethod(name: string) {
|
||||
const method = ORCHESTRATION_METHODS.find((candidate) => candidate.name === name)
|
||||
if (!method) {
|
||||
throw new Error(`Method not found: ${name}`)
|
||||
}
|
||||
return method
|
||||
}
|
||||
|
||||
async function call(ctx: RpcContext, name: string, params: Record<string, unknown>) {
|
||||
const method = findMethod(name)
|
||||
const parsed = method.params ? method.params.parse(params) : undefined
|
||||
return method.handler(parsed, ctx)
|
||||
}
|
||||
|
||||
describe('orchestration RPC workspace ownership', () => {
|
||||
let db: OrchestrationDb
|
||||
|
||||
afterEach(() => db?.close())
|
||||
|
||||
it('rejects task creation from an explicit stale terminal without inserting a global task', async () => {
|
||||
db = new OrchestrationDb(':memory:')
|
||||
const runtime = new OrcaRuntimeService()
|
||||
runtime.setOrchestrationDb(db)
|
||||
|
||||
await expect(
|
||||
call({ runtime }, 'orchestration.taskCreate', {
|
||||
spec: 'must not become global',
|
||||
callerTerminalHandle: 'term_stale'
|
||||
})
|
||||
).rejects.toThrow()
|
||||
|
||||
expect(db.listTasks()).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('creates a global task for a live floating-terminal caller', async () => {
|
||||
db = new OrchestrationDb(':memory:')
|
||||
const runtime = new OrcaRuntimeService()
|
||||
runtime.setOrchestrationDb(db)
|
||||
vi.spyOn(runtime, 'showTerminal').mockResolvedValue({
|
||||
handle: 'term_floating',
|
||||
worktreeId: FLOATING_TERMINAL_WORKTREE_ID
|
||||
} as Awaited<ReturnType<OrcaRuntimeService['showTerminal']>>)
|
||||
|
||||
const result = (await call({ runtime }, 'orchestration.taskCreate', {
|
||||
spec: 'global floating-terminal task',
|
||||
callerTerminalHandle: 'term_floating'
|
||||
})) as { task: { workspace_key: string | null } }
|
||||
|
||||
expect(result.task.workspace_key).toBeNull()
|
||||
expect(db.listTasks()).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('scopes each group status copy to its recipient before sender fallback', async () => {
|
||||
db = new OrchestrationDb(':memory:')
|
||||
const runtime = {
|
||||
getOrchestrationDb: () => db,
|
||||
getTerminalPaneKey: () => null,
|
||||
listTerminals: vi.fn(async () => ({
|
||||
terminals: [
|
||||
{ handle: 'term_sender', worktreeId: 'wt_sender' },
|
||||
{ handle: 'term_coord', worktreeId: 'wt_coord' },
|
||||
{ handle: 'term_worker', worktreeId: 'wt_worker' },
|
||||
{ handle: 'term_fallback', worktreeId: 'wt_fallback' }
|
||||
]
|
||||
})),
|
||||
getAgentStatusForHandle: () => null,
|
||||
deliverPendingMessagesForHandle: vi.fn(),
|
||||
notifyMessageArrived: vi.fn()
|
||||
} as unknown as OrcaRuntimeService
|
||||
const ctx = { runtime }
|
||||
|
||||
db.createCoordinatorRun({
|
||||
spec: 'coordinate',
|
||||
coordinatorHandle: 'term_coord',
|
||||
workspaceKey: 'worktree:wt_coord'
|
||||
})
|
||||
const workerTask = db.createTask({ spec: 'worker', workspaceKey: 'worktree:wt_worker' })
|
||||
db.createDispatchContext(workerTask.id, 'term_worker')
|
||||
const senderTask = db.createTask({ spec: 'sender', workspaceKey: 'worktree:wt_sender' })
|
||||
db.createDispatchContext(senderTask.id, 'term_sender')
|
||||
|
||||
const result = (await call(ctx, 'orchestration.send', {
|
||||
from: 'term_sender',
|
||||
to: '@all',
|
||||
subject: 'status update',
|
||||
type: 'status'
|
||||
})) as { messages: { to_handle: string; workspace_key: string | null }[] }
|
||||
const scopes = Object.fromEntries(
|
||||
result.messages.map((message) => [message.to_handle, message.workspace_key])
|
||||
)
|
||||
|
||||
expect(scopes).toEqual({
|
||||
term_coord: 'worktree:wt_coord',
|
||||
term_worker: 'worktree:wt_worker',
|
||||
term_fallback: 'worktree:wt_sender'
|
||||
})
|
||||
})
|
||||
})
|
||||
@@ -237,6 +237,22 @@ describe('orchestration RPC methods', () => {
|
||||
expect(notify).toHaveBeenCalledWith('term_coord', 'heartbeat')
|
||||
})
|
||||
|
||||
it('scopes worker messages to the sender dispatch workspace', async () => {
|
||||
setup()
|
||||
vi.spyOn(runtime, 'deliverPendingMessagesForHandle').mockImplementation(() => {})
|
||||
const task = db.createTask({ spec: 'work', workspaceKey: 'worktree:wt_a' })
|
||||
db.createDispatchContext(task.id, 'term_worker')
|
||||
|
||||
const result = (await call('orchestration.send', {
|
||||
from: 'term_worker',
|
||||
to: 'shared_coordinator',
|
||||
subject: 'needs input',
|
||||
type: 'decision_gate'
|
||||
})) as { message: { workspace_key: string | null } }
|
||||
|
||||
expect(result.message.workspace_key).toBe('worktree:wt_a')
|
||||
})
|
||||
|
||||
it('rejects missing --to', () => {
|
||||
const method = findMethod('orchestration.send')
|
||||
expect(() => method.params!.parse({ subject: 'hi' })).toThrow()
|
||||
@@ -1160,6 +1176,7 @@ describe('orchestration RPC methods', () => {
|
||||
|
||||
it('records the caller terminal handle when creating a task', async () => {
|
||||
setup()
|
||||
vi.spyOn(runtime, 'resolveWorkspaceKeyForTerminalHandle').mockResolvedValue(null)
|
||||
const result = (await call('orchestration.taskCreate', {
|
||||
spec: 'spawn related workspace',
|
||||
callerTerminalHandle: 'term_creator'
|
||||
@@ -1168,6 +1185,37 @@ describe('orchestration RPC methods', () => {
|
||||
expect(db.getTask(result.task.id)?.created_by_terminal_handle).toBe('term_creator')
|
||||
})
|
||||
|
||||
it('inherits a child task workspace from its parent', async () => {
|
||||
setup()
|
||||
const parent = db.createTask({ spec: 'parent', workspaceKey: 'worktree:wt_a' })
|
||||
const resolveCallerWorkspace = vi
|
||||
.spyOn(runtime, 'resolveWorkspaceKeyForTerminalHandle')
|
||||
.mockResolvedValue('worktree:wt_b')
|
||||
|
||||
const result = (await call('orchestration.taskCreate', {
|
||||
spec: 'child',
|
||||
parent: parent.id,
|
||||
callerTerminalHandle: 'term_b'
|
||||
})) as { task: { workspace_key: string | null } }
|
||||
|
||||
expect(result.task.workspace_key).toBe('worktree:wt_a')
|
||||
expect(resolveCallerWorkspace).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('rejects dependencies owned by another workspace', async () => {
|
||||
setup()
|
||||
const dependency = db.createTask({ spec: 'dependency', workspaceKey: 'worktree:wt_a' })
|
||||
vi.spyOn(runtime, 'resolveWorkspaceKeyForTerminalHandle').mockResolvedValue('worktree:wt_b')
|
||||
|
||||
await expect(
|
||||
call('orchestration.taskCreate', {
|
||||
spec: 'dependent',
|
||||
deps: JSON.stringify([dependency.id]),
|
||||
callerTerminalHandle: 'term_b'
|
||||
})
|
||||
).rejects.toThrow('belongs to a different workspace')
|
||||
})
|
||||
|
||||
it('rejects invalid deps JSON', async () => {
|
||||
setup()
|
||||
await expect(
|
||||
|
||||
@@ -2,7 +2,12 @@
|
||||
import { z } from 'zod'
|
||||
import { defineMethod, type RpcMethod } from '../core'
|
||||
import { OptionalFiniteNumber, OptionalString, OptionalBoolean, requiredString } from '../schemas'
|
||||
import type { MessageType, MessagePriority, TaskStatus } from '../../orchestration/db'
|
||||
import type {
|
||||
MessageType,
|
||||
MessagePriority,
|
||||
OrchestrationDb,
|
||||
TaskStatus
|
||||
} from '../../orchestration/db'
|
||||
import { buildDispatchPreamble } from '../../orchestration/preamble'
|
||||
import { formatMessageBanner } from '../../orchestration/formatter'
|
||||
import { isGroupAddress, resolveGroupAddress } from '../../orchestration/groups'
|
||||
@@ -30,6 +35,27 @@ const TASK_STATUSES: TaskStatus[] = [
|
||||
'blocked'
|
||||
]
|
||||
|
||||
function resolveMessageWorkspaceKey(
|
||||
db: OrchestrationDb,
|
||||
fromHandle: string,
|
||||
toHandle: string,
|
||||
recipientFirst = false
|
||||
): string | null {
|
||||
const senderDispatch = db.getActiveDispatchForTerminal(fromHandle)
|
||||
if (!recipientFirst && senderDispatch) {
|
||||
return senderDispatch.workspace_key
|
||||
}
|
||||
const recipientRun = db.getActiveCoordinatorRunForHandle(toHandle)
|
||||
if (recipientRun) {
|
||||
return recipientRun.workspace_key
|
||||
}
|
||||
const recipientDispatch = db.getActiveDispatchForTerminal(toHandle)
|
||||
if (recipientDispatch) {
|
||||
return recipientDispatch.workspace_key
|
||||
}
|
||||
return senderDispatch?.workspace_key ?? null
|
||||
}
|
||||
|
||||
function getLifecycleGroupRecipientError(type: 'worker_done' | 'heartbeat'): string {
|
||||
return `${type} messages must be sent to a concrete coordinator terminal handle, not a group address.`
|
||||
}
|
||||
@@ -212,7 +238,8 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
|
||||
priority: params.priority as MessagePriority,
|
||||
threadId: params.threadId,
|
||||
payload: params.payload,
|
||||
senderPaneKey
|
||||
senderPaneKey,
|
||||
workspaceKey: resolveMessageWorkspaceKey(db, from, params.to)
|
||||
})
|
||||
// Why: reconcile releases the dispatch lock before waking recipients, else a woken coordinator re-dispatches while the lock is still held.
|
||||
if (msg.type === 'worker_done' || msg.type === 'heartbeat') {
|
||||
@@ -254,7 +281,10 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
|
||||
priority: params.priority as MessagePriority,
|
||||
threadId,
|
||||
payload: params.payload,
|
||||
senderPaneKey
|
||||
senderPaneKey,
|
||||
// Why (#4389): one group send can cross workspaces, so each persisted
|
||||
// copy follows its recipient instead of sharing the sender's scope.
|
||||
workspaceKey: resolveMessageWorkspaceKey(db, from, handle, true)
|
||||
})
|
||||
)
|
||||
for (const message of messages) {
|
||||
@@ -350,7 +380,8 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
|
||||
to: original.from_handle,
|
||||
subject: `Re: ${original.subject}`,
|
||||
body: params.body,
|
||||
threadId: original.thread_id ?? original.id
|
||||
threadId: original.thread_id ?? original.id,
|
||||
workspaceKey: original.workspace_key
|
||||
})
|
||||
|
||||
runtime.notifyMessageArrived(original.from_handle, reply.type)
|
||||
@@ -388,12 +419,22 @@ 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 parentTask = params.parent ? db.getTask(params.parent) : undefined
|
||||
// Why (#4389): child ownership follows the parent DAG, not whichever
|
||||
// terminal happened to create it; roots still follow the caller workspace.
|
||||
const workspaceKey = parentTask
|
||||
? parentTask.workspace_key
|
||||
: await runtime.resolveWorkspaceKeyForTerminalHandle(params.callerTerminalHandle)
|
||||
for (const depId of deps ?? []) {
|
||||
const dependency = db.getTask(depId)
|
||||
if (
|
||||
workspaceKey != null &&
|
||||
dependency?.workspace_key != null &&
|
||||
dependency.workspace_key !== workspaceKey
|
||||
) {
|
||||
throw new Error(`Dependency ${depId} belongs to a different workspace`)
|
||||
}
|
||||
}
|
||||
const task = db.createTask({
|
||||
spec: params.spec,
|
||||
taskTitle: params.taskTitle,
|
||||
@@ -588,7 +629,8 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
|
||||
subject: 'Question',
|
||||
body: params.question,
|
||||
type: 'decision_gate',
|
||||
payload
|
||||
payload,
|
||||
workspaceKey: resolveMessageWorkspaceKey(db, from, params.to)
|
||||
})
|
||||
runtime.deliverPendingMessagesForHandle(params.to)
|
||||
runtime.notifyMessageArrived(params.to, outbound.type)
|
||||
|
||||
@@ -4083,13 +4083,13 @@ describe('OrcaRuntimeRpcServer', () => {
|
||||
}
|
||||
})
|
||||
|
||||
// Why (#4389): v5 → v6 must add the nullable workspace_key column to the
|
||||
// four orchestration tables, bump user_version to 6, and leave existing
|
||||
// Why (#4389): v6 → v7 must add the nullable workspace_key column to the
|
||||
// four orchestration tables, bump user_version to 7, 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')
|
||||
it('upgrades a v6 on-disk DB to v7 with workspace_key columns, preserving rows as NULL', () => {
|
||||
const tmpPath = join(mkdtempSync(join(tmpdir(), 'orca-orch-mig-v7-')), 'orch.sqlite')
|
||||
const seed = new Database(tmpPath)
|
||||
// Minimal pre-v6 (v5) shapes — no workspace_key on any table.
|
||||
// Minimal v6 shapes include pane authority but no workspace_key.
|
||||
seed.exec(`
|
||||
CREATE TABLE coordinator_runs (
|
||||
id TEXT PRIMARY KEY, spec TEXT NOT NULL,
|
||||
@@ -4106,6 +4106,7 @@ describe('OrcaRuntimeRpcServer', () => {
|
||||
);
|
||||
CREATE TABLE dispatch_contexts (
|
||||
id TEXT PRIMARY KEY, task_id TEXT NOT NULL, assignee_handle TEXT,
|
||||
assignee_pane_key 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
|
||||
@@ -4116,7 +4117,8 @@ describe('OrcaRuntimeRpcServer', () => {
|
||||
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
|
||||
created_at TEXT NOT NULL DEFAULT (datetime('now')), delivered_at TEXT,
|
||||
sender_pane_key TEXT
|
||||
);
|
||||
INSERT INTO coordinator_runs (id, spec, status, coordinator_handle)
|
||||
VALUES ('run_legacy', 'legacy', 'running', 'coord');
|
||||
@@ -4126,14 +4128,14 @@ describe('OrcaRuntimeRpcServer', () => {
|
||||
INSERT INTO messages (id, from_handle, to_handle, subject)
|
||||
VALUES ('msg_legacy', 'a', 'coord', 'hi');
|
||||
`)
|
||||
seed.pragma('user_version = 5')
|
||||
seed.pragma('user_version = 6')
|
||||
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)
|
||||
expect(version).toBe(7)
|
||||
|
||||
const hasColumn = (table: string): boolean =>
|
||||
(probe.pragma(`table_info(${table})`) as { name: string }[]).some(
|
||||
|
||||
Reference in New Issue
Block a user