diff --git a/src/main/runtime/orchestration/db-task-dispatch-lifecycle-guards.test.ts b/src/main/runtime/orchestration/db-task-dispatch-lifecycle-guards.test.ts index f57bc0e55c6..3514f25a2ae 100644 --- a/src/main/runtime/orchestration/db-task-dispatch-lifecycle-guards.test.ts +++ b/src/main/runtime/orchestration/db-task-dispatch-lifecycle-guards.test.ts @@ -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' }) diff --git a/src/main/runtime/orchestration/db.ts b/src/main/runtime/orchestration/db.ts index 1f0f7528d4a..a89fc393d50 100644 --- a/src/main/runtime/orchestration/db.ts +++ b/src/main/runtime/orchestration/db.ts @@ -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 } diff --git a/src/main/runtime/rpc/methods/orchestration-send.test.ts b/src/main/runtime/rpc/methods/orchestration-send.test.ts index 0462249bf32..afcac1d13ae 100644 --- a/src/main/runtime/rpc/methods/orchestration-send.test.ts +++ b/src/main/runtime/rpc/methods/orchestration-send.test.ts @@ -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' }) diff --git a/src/main/runtime/rpc/methods/orchestration.ts b/src/main/runtime/rpc/methods/orchestration.ts index f52ee65da19..1b876fe1347 100644 --- a/src/main/runtime/rpc/methods/orchestration.ts +++ b/src/main/runtime/rpc/methods/orchestration.ts @@ -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 } }