fix(orchestration): preserve sibling dispatch authority

This commit is contained in:
Jinwoo-H
2026-08-16 23:20:49 -04:00
parent 2b3a6aa91d
commit fc6a1d97ff
4 changed files with 129 additions and 24 deletions
@@ -224,6 +224,38 @@ describe('Task/Dispatch lifecycle guards', () => {
}
)
it.each(['stop', 'abandon'] as const)(
'%s preserves a live worker sibling and lets it report',
(operation) => {
const database = createDatabase()
const task = database.createTask({ spec: `${operation} legacy worker split` })
const live = startWorker(database, task.id, `${operation}_live`)
sqliteFor(database).prepare("UPDATE tasks SET status = 'ready' WHERE id = ?").run(task.id)
const released = startWorker(database, task.id, `${operation}_released`)
if (operation === 'stop') {
expect(database.beginWorkerStop(released.dispatchId).disposition).toBe('stopping')
expect(database.settleWorkerStop(released.dispatchId).state).toBe('stopped')
} else {
expect(database.abandonWorkerDispatch(released.dispatchId).disposition).toBe('abandoned')
}
expect(database.getTask(task.id)?.status).toBe('dispatched')
expect(database.getDispatchContextById(live.dispatchId)?.status).toBe('dispatched')
expectCapability(database, live, true)
expect(
database.settleWorkerReport({
taskId: task.id,
dispatchId: live.dispatchId,
outcome: 'succeeded',
result: `${operation} sibling completed`
})
).toEqual({ action: 'settled', outcome: 'succeeded', duplicate: false })
expect(database.getTask(task.id)?.status).toBe('completed')
expectCapability(database, live, false)
}
)
it('rejects gate creation while a supervised worker remains active', () => {
const database = createDatabase()
const task = database.createTask({ spec: 'worker gate guard' })
+24 -8
View File
@@ -6414,7 +6414,7 @@ export class OrchestrationDb {
WHERE id = ?`
)
.run(dispatchId)
this.db.prepare("UPDATE tasks SET status = 'blocked' WHERE id = ?").run(dispatch.task_id)
this.blockTaskWithoutActiveSibling(dispatch.task_id, dispatchId)
this.closeQuestionsForDispatch(dispatchId)
this.db.exec('COMMIT')
return {
@@ -6592,7 +6592,7 @@ export class OrchestrationDb {
WHERE id = ?`
)
.run(dispatchId)
this.db.prepare("UPDATE tasks SET status = 'blocked' WHERE id = ?").run(dispatch.task_id)
this.blockTaskWithoutActiveSibling(dispatch.task_id, dispatchId)
this.closeQuestionsForDispatch(dispatchId)
this.db.exec('COMMIT')
return {
@@ -7316,6 +7316,26 @@ export class OrchestrationDb {
.get(taskId) as DispatchContextRow | undefined
}
private getActiveDispatchForTask(taskId: string): DispatchContextRow | undefined {
return this.db
.prepare(
"SELECT * FROM dispatch_contexts WHERE task_id = ? AND status IN ('pending', 'dispatched') ORDER BY rowid DESC LIMIT 1"
)
.get(taskId) as DispatchContextRow | undefined
}
private blockTaskWithoutActiveSibling(taskId: string, dispatchId: string): void {
this.db
.prepare(
`UPDATE tasks SET status = 'blocked'
WHERE id = ? AND NOT EXISTS (
SELECT 1 FROM dispatch_contexts
WHERE task_id = tasks.id AND id != ? AND status IN ('pending', 'dispatched')
)`
)
.run(taskId, dispatchId)
}
getDispatchContextById(dispatchId: string): DispatchContextRow | undefined {
return this.db.prepare('SELECT * FROM dispatch_contexts WHERE id = ?').get(dispatchId) as
| DispatchContextRow
@@ -7586,7 +7606,7 @@ export class OrchestrationDb {
reason: `inactive dispatch ${params.dispatchId}: it or task ${params.taskId} is already settled.`
}
}
const latest = this.getDispatchContext(params.taskId)
const latest = this.getActiveDispatchForTask(params.taskId)
if (latest?.id !== params.dispatchId) {
return {
action: 'rejected',
@@ -7670,11 +7690,7 @@ export class OrchestrationDb {
}
failActiveDispatchForTask(taskId: string, error: string): DispatchContextRow | undefined {
const active = this.db
.prepare(
"SELECT * FROM dispatch_contexts WHERE task_id = ? AND status IN ('pending', 'dispatched') ORDER BY rowid DESC LIMIT 1"
)
.get(taskId) as DispatchContextRow | undefined
const active = this.getActiveDispatchForTask(taskId)
return active ? this.failDispatch(active.id, error) : undefined
}
@@ -206,6 +206,50 @@ describe('orchestration RPC methods', () => {
expect(result).toMatchObject({ lifecycle: { action: 'completed' } })
})
it('fences a replacement process for a capability-less manual Dispatch', async () => {
setup()
const task = db.createTask({ spec: 'process-bound manual work' })
const dispatch = db.createDispatchContext(
task.id,
'term_worker',
'tab_worker:leaf_worker',
undefined,
'runtime_test:term_worker:1'
)
vi.mocked(runtime.getTerminalPaneKey).mockImplementation((handle) =>
handle === 'term_worker' ? 'tab_worker:leaf_worker' : coordinatorPaneKey
)
vi.mocked(runtime.getTerminalProcessIncarnation).mockReturnValue('runtime_test:term_worker:2')
const payload = JSON.stringify({
taskId: task.id,
dispatchId: dispatch.id,
outcome: 'succeeded'
})
const rejected = (await call('orchestration.send', {
from: 'term_worker',
subject: 'Done after replacement',
type: 'worker_done',
payload
})) as { lifecycle: { action: string; code: string } }
expect(rejected.lifecycle).toEqual({
action: 'rejected',
code: 'sender_not_assignee',
reason: `Dispatch ${dispatch.id} process incarnation is no longer current for its pane.`
})
expect(db.getTask(task.id)?.status).toBe('dispatched')
vi.mocked(runtime.getTerminalProcessIncarnation).mockReturnValue('runtime_test:term_worker:1')
const accepted = (await call('orchestration.send', {
from: 'term_worker',
subject: 'Done by assignee',
type: 'worker_done',
payload
})) as { lifecycle: { action: string } }
expect(accepted.lifecycle.action).toBe('completed')
})
it('rejects an identity-less lifecycle send resolved through the coordinator handle', async () => {
setup()
const task = db.createTask({ spec: 'work' })
+29 -16
View File
@@ -639,29 +639,42 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
const dispatch = routing.dispatchId
? db.getDispatchContextById(routing.dispatchId)
: undefined
if ((msg.type === 'worker_done' || msg.type === 'heartbeat') && dispatch?.capability_hash) {
const authority = db.verifyDispatchCapability({
dispatchId: dispatch.id,
capability: orchestrationCapability,
paneKey: senderPaneKey,
processIncarnation:
attestedCaller?.processIncarnation ??
runtime.getTerminalProcessIncarnation(from) ??
undefined
})
const lifecycleMessage = msg.type === 'worker_done' || msg.type === 'heartbeat'
if (
lifecycleMessage &&
dispatch &&
(dispatch.capability_hash || dispatch.process_incarnation)
) {
const processIncarnation =
attestedCaller?.processIncarnation ??
runtime.getTerminalProcessIncarnation(from) ??
undefined
const capabilityBacked = Boolean(dispatch.capability_hash)
const authority = capabilityBacked
? db.verifyDispatchCapability({
dispatchId: dispatch.id,
capability: orchestrationCapability,
paneKey: senderPaneKey,
processIncarnation
})
: {
valid: db.isDispatchProcessCurrent({
dispatchId: dispatch.id,
paneKey: senderPaneKey ?? null,
processIncarnation: processIncarnation ?? null
}),
reason: `Dispatch ${dispatch.id} process incarnation is no longer current for its pane.`
}
if (!authority.valid) {
const code = capabilityBacked ? 'dispatch_capability_invalid' : 'sender_not_assignee'
const rejection =
db.convertLifecycleMessageToRejection(
msg.id,
'dispatch_capability_invalid',
authority.reason
) ?? msg
db.convertLifecycleMessageToRejection(msg.id, code, authority.reason) ?? msg
runtime.notifyMessageArrived(rejection.to_handle, rejection.type)
return {
message: rejection,
lifecycle: {
action: 'rejected',
code: 'dispatch_capability_invalid',
code,
reason: authority.reason
}
}