From b0d7804d23ac44e1a6130838d4e2eede1a5ed91a Mon Sep 17 00:00:00 2001 From: Neil <4138956+nwparker@users.noreply.github.com> Date: Mon, 20 Jul 2026 19:51:00 -0700 Subject: [PATCH] fix(orchestration): stamp fallback-resolved ids onto null-id worker_done A null-id worker_done resolved via the sole-active-dispatch fallback left its payload id-less, so a later re-read of the same message re-resolved against current sender state. Once the pane received a fresh dispatch B, the stale message wrongly completed B (potentially releasing B's dependents early). Persist the resolved taskId/dispatchId onto the message payload before completing the task, reusing the existing payload-update mechanism. A re-read now takes the stable id-bearing path and re-targets the original dispatch A instead of B. Idempotent and recoverable on partial failure. Co-authored-by: Orca --- src/main/runtime/orchestration/db.ts | 27 ++++++++++++++ .../lifecycle-reconciliation.test.ts | 35 +++++++++++++++++++ .../orchestration/lifecycle-reconciliation.ts | 5 ++- 3 files changed, 66 insertions(+), 1 deletion(-) diff --git a/src/main/runtime/orchestration/db.ts b/src/main/runtime/orchestration/db.ts index 3a64150060c..616584e71dc 100644 --- a/src/main/runtime/orchestration/db.ts +++ b/src/main/runtime/orchestration/db.ts @@ -61,6 +61,23 @@ function addLifecycleRejectionMarker(payload: string | null, reason: string): st }) } +function stampResolvedLifecycleIds( + payload: string | null, + taskId: string, + dispatchId: string +): string { + let parsed: Record = {} + try { + const value: unknown = payload ? JSON.parse(payload) : {} + if (value && typeof value === 'object' && !Array.isArray(value)) { + parsed = value as Record + } + } catch { + // Fallback resolution only reaches this path with object/empty payloads. + } + return JSON.stringify({ ...parsed, taskId, dispatchId }) +} + const SQLITE_UTC_TIMESTAMP_RE = /^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}(?:\.\d+)?$/ function exposeUtcTimestamp(timestamp: string | null): string | null { @@ -385,6 +402,16 @@ export class OrchestrationDb { return this.getMessageById(messageId) } + // Why: stamp fallback-resolved ids onto a null-id worker_done so a later re-read takes the stable id path instead of re-resolving against a newer dispatch (#7429). + stampResolvedWorkerDoneIds(messageId: string, taskId: string, dispatchId: string): void { + const message = this.getMessageById(messageId) + if (!message || message.type !== 'worker_done') { + return + } + const payload = stampResolvedLifecycleIds(message.payload, taskId, dispatchId) + this.db.prepare('UPDATE messages SET payload = ? WHERE id = ?').run(payload, messageId) + } + // Why: delivered_at IS NULL filter — push-on-idle delivers each row at most once; read (set only by check) wouldn't prevent replay. getUndeliveredUnreadMessages(toHandle: string, types?: MessageType[]): MessageRow[] { if (types && types.length > 0) { diff --git a/src/main/runtime/orchestration/lifecycle-reconciliation.test.ts b/src/main/runtime/orchestration/lifecycle-reconciliation.test.ts index 5f13eb7075a..bca66b30d14 100644 --- a/src/main/runtime/orchestration/lifecycle-reconciliation.test.ts +++ b/src/main/runtime/orchestration/lifecycle-reconciliation.test.ts @@ -432,6 +432,41 @@ describe('lifecycle reconciliation', () => { expect(db.getDispatchContextById(dispatch.id)?.status).toBe('completed') }) + it('does not let a re-read of a fallback-resolved worker_done complete a later dispatch to the same pane', () => { + db = new OrchestrationDb(':memory:') + const taskA = db.createTask({ spec: 'work A' }) + const dispatchA = db.createDispatchContext(taskA.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}` + }) + + // First reconcile completes A via the sole-active-dispatch fallback. + expect(reconcileLifecycleMessage(db, done)).toMatchObject({ + action: 'completed', + taskId: taskA.id, + dispatchId: dispatchA.id + }) + + // The same pane now owns a fresh dispatch B. + const taskB = db.createTask({ spec: 'work B' }) + const dispatchB = db.createDispatchContext(taskB.id, 'term_worker', `tab_w:${LEAF_A}`) + + // Re-reading the ORIGINAL message (as the coordinator does) must re-target A, not B. + const reread = db.getMessageById(done.id) + expect(reread && reconcileLifecycleMessage(db, reread)).toMatchObject({ + action: 'completed', + taskId: taskA.id, + dispatchId: dispatchA.id + }) + expect(db.getTask(taskB.id)?.status).toBe('dispatched') + expect(db.getDispatchContextById(dispatchB.id)?.status).toBe('dispatched') + }) + it('ignores a null-id worker_done sent from a foreign pane and handle', () => { db = new OrchestrationDb(':memory:') const task = db.createTask({ spec: 'work' }) diff --git a/src/main/runtime/orchestration/lifecycle-reconciliation.ts b/src/main/runtime/orchestration/lifecycle-reconciliation.ts index a4efd7cf448..c6b2ce1dbdd 100644 --- a/src/main/runtime/orchestration/lifecycle-reconciliation.ts +++ b/src/main/runtime/orchestration/lifecycle-reconciliation.ts @@ -252,7 +252,10 @@ function resolveWorkerDoneTarget( 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 } + const resolved = { taskId: taskId || fallback.task_id, dispatchId: dispatchId || fallback.id } + // Why: persist before completion so a re-read of this same message uses the recorded ids; else the sole-dispatch fallback would re-resolve onto a later dispatch B and wrongly complete it (#7429). + db.stampResolvedWorkerDoneIds(msg.id, resolved.taskId, resolved.dispatchId) + return resolved } onLog(