diff --git a/src/main/runtime/orchestration/db.ts b/src/main/runtime/orchestration/db.ts index 1c4b68f9d10..3a64150060c 100644 --- a/src/main/runtime/orchestration/db.ts +++ b/src/main/runtime/orchestration/db.ts @@ -694,6 +694,28 @@ export class OrchestrationDb { return this.findActiveDispatchForAssignee(handle) } + // Why: attributing an id-less lifecycle message is only safe when the sender owns + // exactly one active dispatch (#7429) — more than one match must not be guessed. + getSoleActiveDispatchForSender( + assigneeHandle: string, + assigneePaneKey?: string | null + ): DispatchContextRow | undefined { + const matches = ( + this.db + .prepare("SELECT * FROM dispatch_contexts WHERE status IN ('pending', 'dispatched')") + .all() as DispatchContextRow[] + ).filter( + (row) => + row.assignee_handle === assigneeHandle || + Boolean( + assigneePaneKey && + row.assignee_pane_key && + isEquivalentPaneKey(row.assignee_pane_key, assigneePaneKey) + ) + ) + return matches.length === 1 ? matches[0] : undefined + } + private findActiveDispatchForAssignee( assigneeHandle: string, assigneePaneKey?: string diff --git a/src/main/runtime/orchestration/lifecycle-reconciliation.test.ts b/src/main/runtime/orchestration/lifecycle-reconciliation.test.ts index 8fed046d421..5f13eb7075a 100644 --- a/src/main/runtime/orchestration/lifecycle-reconciliation.test.ts +++ b/src/main/runtime/orchestration/lifecycle-reconciliation.test.ts @@ -409,4 +409,87 @@ describe('lifecycle reconciliation', () => { expect(db.getMessageById(lateHeartbeat.id)).toMatchObject({ read: 1 }) expect(db.getMessageById(lateHeartbeat.id)?.delivered_at).not.toBeNull() }) + + it('completes a null-id worker_done sent from the pane that owns the sole dispatch', () => { + db = new OrchestrationDb(':memory:') + const task = db.createTask({ spec: 'work' }) + const dispatch = db.createDispatchContext(task.id, 'term_worker', `tab_w:${LEAF_A}`) + const done = db.insertMessage({ + from: 'term_worker', + to: 'term_coordinator', + subject: 'Done', + type: 'worker_done', + payload: JSON.stringify({}), + senderPaneKey: `tab_w:${LEAF_A}` + }) + + expect(reconcileLifecycleMessage(db, done)).toEqual({ + action: 'completed', + taskId: task.id, + dispatchId: dispatch.id + }) + expect(db.getTask(task.id)?.status).toBe('completed') + expect(db.getDispatchContextById(dispatch.id)?.status).toBe('completed') + }) + + it('ignores a null-id worker_done sent from a foreign pane and handle', () => { + db = new OrchestrationDb(':memory:') + const task = db.createTask({ spec: 'work' }) + const dispatch = db.createDispatchContext(task.id, 'term_owner', `tab_w:${LEAF_A}`) + const logs: string[] = [] + const done = db.insertMessage({ + from: 'term_stranger', + to: 'term_coordinator', + subject: 'Done', + type: 'worker_done', + payload: JSON.stringify({}), + senderPaneKey: `tab_w:${LEAF_B}` + }) + + expect(reconcileLifecycleMessage(db, done, (line) => logs.push(line))).toEqual({ + action: 'ignored' + }) + expect(db.getTask(task.id)?.status).toBe('dispatched') + expect(db.getDispatchContextById(dispatch.id)?.status).toBe('dispatched') + expect(logs.some((line) => line.includes('without taskId'))).toBe(true) + }) + + it('ignores a null-id worker_done when the sender has no active dispatch', () => { + db = new OrchestrationDb(':memory:') + const task = db.createTask({ spec: 'work' }) + const dispatch = db.createDispatchContext(task.id, 'term_worker', `tab_w:${LEAF_A}`) + db.updateTaskStatus(task.id, 'completed') + const done = db.insertMessage({ + from: 'term_worker', + to: 'term_coordinator', + subject: 'Done again', + type: 'worker_done', + payload: JSON.stringify({}), + senderPaneKey: `tab_w:${LEAF_A}` + }) + + expect(reconcileLifecycleMessage(db, done)).toEqual({ action: 'ignored' }) + expect(db.getDispatchContextById(dispatch.id)?.status).toBe('completed') + }) + + it('ignores a null-id worker_done when the sender matches two active dispatches', () => { + db = new OrchestrationDb(':memory:') + const ownTask = db.createTask({ spec: 'work' }) + const ownDispatch = db.createDispatchContext(ownTask.id, 'term_worker', `tab_w:${LEAF_A}`) + const paneTask = db.createTask({ spec: 'other work' }) + const paneDispatch = db.createDispatchContext(paneTask.id, 'term_other', `tab_w:${LEAF_B}`) + // Sender matches one dispatch by handle and another by pane, so neither can be attributed. + const done = db.insertMessage({ + from: 'term_worker', + to: 'term_coordinator', + subject: 'Done', + type: 'worker_done', + payload: JSON.stringify({}), + senderPaneKey: `tab_w:${LEAF_B}` + }) + + expect(reconcileLifecycleMessage(db, done)).toEqual({ action: 'ignored' }) + expect(db.getDispatchContextById(ownDispatch.id)?.status).toBe('dispatched') + expect(db.getDispatchContextById(paneDispatch.id)?.status).toBe('dispatched') + }) }) diff --git a/src/main/runtime/orchestration/lifecycle-reconciliation.ts b/src/main/runtime/orchestration/lifecycle-reconciliation.ts index 96917babaf6..a4efd7cf448 100644 --- a/src/main/runtime/orchestration/lifecycle-reconciliation.ts +++ b/src/main/runtime/orchestration/lifecycle-reconciliation.ts @@ -170,17 +170,11 @@ function reconcileWorkerDoneMessage( return persistedRejection } - const taskId = payload.taskId - if (typeof taskId !== 'string' || taskId.length === 0) { - onLog(`Warning: worker_done without taskId from ${msg.from_handle}`) - return { action: 'ignored' } - } - - const dispatchId = payload.dispatchId - if (typeof dispatchId !== 'string' || dispatchId.length === 0) { - onLog(`Warning: worker_done without dispatchId from ${msg.from_handle}`) + const target = resolveWorkerDoneTarget(db, msg, payload, onLog) + if (!target) { return { action: 'ignored' } } + const { taskId, dispatchId } = target const task = db.getTask(taskId) if (!task) { @@ -239,6 +233,36 @@ function reconcileWorkerDoneMessage( return { action: 'completed', taskId, dispatchId } } +// Why: a worker_done from the pane that owns the sole in-flight dispatch is still +// authoritative (#7429); the payload ids are a convenience, not the authority. +function resolveWorkerDoneTarget( + db: OrchestrationDb, + msg: MessageRow, + payload: Record, + onLog: LogFn +): { taskId: string; dispatchId: string } | undefined { + const payloadTaskId = payload.taskId + const payloadDispatchId = payload.dispatchId + const taskId = typeof payloadTaskId === 'string' && payloadTaskId.length > 0 ? payloadTaskId : '' + const dispatchId = + typeof payloadDispatchId === 'string' && payloadDispatchId.length > 0 ? payloadDispatchId : '' + if (taskId && dispatchId) { + return { taskId, dispatchId } + } + + const fallback = db.getSoleActiveDispatchForSender(msg.from_handle, msg.sender_pane_key) + if (fallback && hasLifecycleAuthority(fallback, msg)) { + return { taskId: taskId || fallback.task_id, dispatchId: dispatchId || fallback.id } + } + + onLog( + taskId + ? `Warning: worker_done without dispatchId from ${msg.from_handle}` + : `Warning: worker_done without taskId from ${msg.from_handle}` + ) + return undefined +} + function buildLifecycleAuthorityRejectionReason( dispatchId: string, dispatch: { assignee_handle: string | null; assignee_pane_key: string | null }, diff --git a/src/main/runtime/rpc/methods/orchestration.test.ts b/src/main/runtime/rpc/methods/orchestration.test.ts index 95339cb3190..8fa360618d6 100644 --- a/src/main/runtime/rpc/methods/orchestration.test.ts +++ b/src/main/runtime/rpc/methods/orchestration.test.ts @@ -748,11 +748,26 @@ describe('orchestration RPC methods', () => { expect(db.getUnreadMessages('term_coord', ['worker_done'])).toHaveLength(1) }) - it('does not complete worker_done missing taskId or dispatchId', async () => { + it('completes worker_done missing taskId or dispatchId from the assigned terminal', async () => { setup() const { task, dispatch } = createDispatchedTask() insertWorkerDone({ dispatchId: dispatch.id }) - insertWorkerDone({ taskId: task.id }) + + const result = (await call('orchestration.check', { + terminal: 'term_coord', + types: 'worker_done' + })) as { count: number } + + expect(result.count).toBe(1) + expect(db.getTask(task.id)?.status).toBe('completed') + expect(db.getDispatchContextById(dispatch.id)?.status).toBe('completed') + }) + + it('does not complete worker_done missing taskId or dispatchId from another terminal', async () => { + setup() + const { task, dispatch } = createDispatchedTask() + insertWorkerDone({ from: 'term_other', dispatchId: dispatch.id }) + insertWorkerDone({ from: 'term_other', taskId: task.id }) const result = (await call('orchestration.check', { terminal: 'term_coord',