From 6b147f073703f3efb6a9c156496eb4f284beb6b3 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 8 Nov 2026 03:09:17 +0000 Subject: [PATCH 1/3] test(server): reproduce queued completion wake cancel disposing running siblings Reproduces #32806: cancelling a queued delegated-completion wake with queued-run.cancel disposes the whole cohort, so a still-running sibling the wake did not carry never wakes the parent when it finishes. Co-Authored-By: Claude Opus 5.5 --- .../DelegatedCompletionDelivery.test.ts | 165 ++++++++++++++++++ 1 file changed, 165 insertions(+) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index db3a2759e3ac5..0b200b1e7d41f 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -1623,6 +1623,171 @@ it.layer(layerContinuationWorkerTest)("interrupted completion wakes", (it) => { ); }), ); + + it.effect("cancelling a queued completion wake keeps its other siblings", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const sink = yield* EventSink.EventSinkV2; + const now = yield* DateTime.now; + const key = "sibling-queued-cancel"; + const threadId = ThreadId.make(`${key}-parent`); + const originRunId = RunId.make(`${key}-origin`); + const originRootId = NodeId.make(`${key}-origin-root`); + const incomingTaskId = NodeId.make(`${key}-incoming`); + yield* seedParentWithTerminalTask({ + threadId, + projectId: ProjectId.make(`${key}-project`), + runId: originRunId, + rootNodeId: originRootId, + taskId: incomingTaskId, + deliveryState: "claimed", + deliveryTaskIds: [incomingTaskId], + completionWake: "always", + now, + }); + yield* sink.write({ + events: [ + { + id: EventId.make(`event:${key}-origin-root`), + type: "node.updated", + threadId, + runId: originRunId, + nodeId: originRootId, + occurredAt: now, + payload: { + id: originRootId, + threadId, + runId: originRunId, + parentNodeId: null, + rootNodeId: originRootId, + kind: "root_turn", + status: "running", + countsForRun: true, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + runtimeRequestId: null, + checkpointScopeId: null, + startedAt: now, + completedAt: null, + }, + }, + ], + }); + yield* orchestrator.dispatch({ + type: "delegated_task.request", + createdBy: "agent", + creationSource: "mcp", + commandId: CommandId.make(`${key}-sibling`), + parentThreadId: threadId, + parentRunId: originRunId, + parentNodeId: originRootId, + task: "Keep working", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + completionWake: "always", + }); + const seeded = yield* orchestrator.getThreadProjection(threadId); + const origin = seeded.runs.find((run) => run.id === originRunId)!; + const sibling = seeded.subagents.find((task) => task.id !== incomingTaskId)!; + // The coordinator finished its spawning turn but is busy with another run, so the wake queues. + yield* sink.write({ + events: [ + { + ...runEvent({ threadId, runId: originRunId, ordinal: 1, status: "completed", now }), + payload: { ...origin, status: "completed", completedAt: now }, + }, + runEvent({ + threadId, + runId: RunId.make(`${key}-busy`), + ordinal: 2, + status: "running", + now, + }), + ], + }); + const delivery = origin.delegatedCompletion!.delivery!; + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "agent", + creationSource: "server", + commandId: CommandId.make(`${key}-wake`), + threadId, + messageId: delivery.messageId, + text: "Delegated task completed", + attachments: [], + modelSelection, + dispatchMode: { type: "queue_after_active" }, + delegatedCompletion: { + parentRunId: originRunId, + generation: delivery.generation, + taskIds: delivery.taskIds, + }, + }); + const wake = (yield* orchestrator.getThreadProjection(threadId)).runs.find( + (run) => run.userMessageId === delivery.messageId, + )!; + assert.equal(wake.status, "queued"); + + const cancelled = yield* orchestrator.dispatch({ + type: "queued-run.cancel", + commandId: CommandId.make(`${key}-cancel`), + threadId, + runId: wake.id, + }); + const cancelDelivery = new Map( + cancelled.storedEvents.flatMap(({ event }) => + event.type === "subagent.updated" + ? [[event.payload.id, event.payload.completionDelivery?.state] as const] + : [], + ), + ); + assert.equal(cancelDelivery.get(incomingTaskId), "disposed"); + assert.notEqual( + cancelDelivery.get(sibling.id), + "disposed", + "cancelling the wake must not dispose a still-running sibling it did not carry", + ); + const afterCancel = yield* orchestrator.getThreadProjection(threadId); + assert.equal( + afterCancel.runs.find((run) => run.id === originRunId)?.delegatedCompletion?.disposition, + "open", + ); + + const childRun = (yield* orchestrator.getThreadProjection(sibling.childThreadId!)).runs[0]!; + const afterSequence = yield* sink.latestSequence(); + yield* sink.write({ + events: [ + { + ...runEvent({ + threadId: sibling.childThreadId!, + runId: childRun.id, + ordinal: childRun.ordinal, + status: "completed", + now, + }), + payload: { ...childRun, status: "completed", completedAt: now }, + }, + ], + }); + yield* sink.stream({ afterSequence, eventType: "subagent.updated" }).pipe( + Stream.filter( + (stored) => + stored.event.type === "subagent.updated" && + stored.event.payload.id === sibling.id && + stored.event.payload.completionDelivery?.state === "claimed", + ), + Stream.take(1), + Stream.runDrain, + ); + const successor = (yield* orchestrator.getThreadProjection(threadId)).runs.find( + (run) => run.id === originRunId, + )?.delegatedCompletion?.delivery; + assert.deepEqual(successor?.taskIds, [sibling.id]); + assert.notEqual(successor?.messageId, delivery.messageId); + }), + ); }); it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( From dc3203aa06cb215ebdc28d87c427a5bb448ace46 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 8 Nov 2026 03:15:16 +0000 Subject: [PATCH 2/3] fix(server): cancelling a queued completion wake keeps its siblings' wakes Removing a queued delegated-completion wake disposed every outstanding task of the owning parent run, so a sibling that was still running never woke the parent when it finished. The queued-cancel path now disposes only the tasks the wake's message carries and cancels the requested run itself. When siblings are still outstanding the cohort stays open, and the cancelled wake's terminal update clears its delivery or reserves a successor for siblings that already finished. With no sibling outstanding the cohort is disposed as before. Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/Orchestrator.ts | 64 ++----------------- 1 file changed, 6 insertions(+), 58 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 5d3c4391b1fe8..22d467f3e49e3 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -2794,8 +2794,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio readonly disposition: "stopped" | "disposed"; readonly now: DateTime.Utc; readonly cancelQueuedDelivery?: boolean; - // An interrupted wake disposes only the results it carried. The cohort stays open, and its - // delivery stays the wake's, so the wake's terminal reserves any sibling's result. + // An interrupted or cancelled wake disposes only the results it carried. The cohort stays + // open, and its delivery stays the wake's, so the wake's terminal reserves any sibling's result. readonly interruptedWakeTaskIds?: ReadonlyArray; }) { const parentRun = input.projection.runs.find((candidate) => candidate.id === input.parentRunId); @@ -10071,8 +10071,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const queuedMessage = projection.messages.find( (candidate) => candidate.id === queuedRun.userMessageId, ); + const now = yield* DateTime.now; if (queuedMessage?.delegatedCompletion !== undefined) { - const now = yield* DateTime.now; yield* disposeDelegatedCompletionCohort({ command, events, @@ -10080,64 +10080,12 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio parentRunId: queuedMessage.delegatedCompletion.parentRunId, disposition: "disposed", now, + cancelQueuedDelivery: false, + interruptedWakeTaskIds: queuedMessage.delegatedCompletion.taskIds, }); - return; } - const queuedRootNode = - queuedRun.rootNodeId === null - ? undefined - : projection.nodes.find((candidate) => candidate.id === queuedRun.rootNodeId); - const queuedAttempt = - queuedRun.activeAttemptId === null - ? undefined - : projection.attempts.find((candidate) => candidate.id === queuedRun.activeAttemptId); - - const now = yield* DateTime.now; + yield* emitQueuedRunCancellation({ command, events, projection, run: queuedRun, now }); const emitEvent = emit(events, command); - yield* emitEvent({ - type: "run.updated", - threadId: command.threadId, - runId: queuedRun.id, - ...(queuedRun.rootNodeId === null ? {} : { nodeId: queuedRun.rootNodeId }), - providerInstanceId: queuedRun.providerInstanceId, - occurredAt: now, - payload: { - ...queuedRun, - status: "cancelled", - queuePosition: null, - completedAt: now, - }, - }); - if (queuedAttempt !== undefined && queuedRootNode !== undefined) { - yield* emitEvent({ - type: "run-attempt.updated", - threadId: command.threadId, - runId: queuedRun.id, - nodeId: queuedRootNode.id, - providerInstanceId: queuedRun.providerInstanceId, - occurredAt: now, - payload: { - ...queuedAttempt, - status: "cancelled", - completedAt: now, - }, - }); - } - if (queuedRootNode !== undefined) { - yield* emitEvent({ - type: "node.updated", - threadId: command.threadId, - runId: queuedRun.id, - nodeId: queuedRootNode.id, - providerInstanceId: queuedRun.providerInstanceId, - occurredAt: now, - payload: { - ...queuedRootNode, - status: "cancelled", - completedAt: now, - }, - }); - } const guest = queuedMessage?.context?.guest; if (guest !== undefined) { const thread = guestShareAfterCancelledSend(projection.thread, guest, now); From 51709dc458c26a0d8aacf894b3c8e33a66242693 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 8 Nov 2026 03:30:19 +0000 Subject: [PATCH 3/3] fix(server): settling still disposes a cancelled wake's siblings thread.settle cancels queued automatic runs through dispatchQueuedRunCancel, which now disposes only the results a cancelled completion wake carried. A sibling still running at settle time then woke the parent with its late result and unsettled the thread. Settle now asks for the wake's whole cohort to be disposed, as before; queued-run.cancel keeps the message-scoped disposal. Co-Authored-By: Claude Opus 5.5 --- .../DelegatedCompletionDelivery.test.ts | 339 ++++++++++++------ .../src/orchestration-v2/Orchestrator.ts | 7 +- 2 files changed, 228 insertions(+), 118 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index 0b200b1e7d41f..f3f35ec27af02 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -1380,6 +1380,145 @@ const seedRestartCancelledChild = (input: { return { taskId, childThreadId, childRunId }; }); +// Parent spawned two always-wake tasks: the incoming one's completion wake is queued behind a busy +// run while the sibling's child run is still going. +const queueCompletionWakeBesideRunningSibling = (key: string) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const sink = yield* EventSink.EventSinkV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make(`${key}-parent`); + const originRunId = RunId.make(`${key}-origin`); + const originRootId = NodeId.make(`${key}-origin-root`); + const incomingTaskId = NodeId.make(`${key}-incoming`); + yield* seedParentWithTerminalTask({ + threadId, + projectId: ProjectId.make(`${key}-project`), + runId: originRunId, + rootNodeId: originRootId, + taskId: incomingTaskId, + deliveryState: "claimed", + deliveryTaskIds: [incomingTaskId], + completionWake: "always", + now, + }); + yield* sink.write({ + events: [ + { + id: EventId.make(`event:${key}-origin-root`), + type: "node.updated", + threadId, + runId: originRunId, + nodeId: originRootId, + occurredAt: now, + payload: { + id: originRootId, + threadId, + runId: originRunId, + parentNodeId: null, + rootNodeId: originRootId, + kind: "root_turn", + status: "running", + countsForRun: true, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + runtimeRequestId: null, + checkpointScopeId: null, + startedAt: now, + completedAt: null, + }, + }, + ], + }); + yield* orchestrator.dispatch({ + type: "delegated_task.request", + createdBy: "agent", + creationSource: "mcp", + commandId: CommandId.make(`${key}-sibling`), + parentThreadId: threadId, + parentRunId: originRunId, + parentNodeId: originRootId, + task: "Keep working", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + completionWake: "always", + }); + const seeded = yield* orchestrator.getThreadProjection(threadId); + const origin = seeded.runs.find((run) => run.id === originRunId)!; + const sibling = seeded.subagents.find((task) => task.id !== incomingTaskId)!; + // The coordinator finished its spawning turn but is busy with another run, so the wake queues. + yield* sink.write({ + events: [ + { + ...runEvent({ threadId, runId: originRunId, ordinal: 1, status: "completed", now }), + payload: { ...origin, status: "completed", completedAt: now }, + }, + runEvent({ + threadId, + runId: RunId.make(`${key}-busy`), + ordinal: 2, + status: "running", + now, + }), + ], + }); + const delivery = origin.delegatedCompletion!.delivery!; + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "agent", + creationSource: "server", + commandId: CommandId.make(`${key}-wake`), + threadId, + messageId: delivery.messageId, + text: "Delegated task completed", + attachments: [], + modelSelection, + dispatchMode: { type: "queue_after_active" }, + delegatedCompletion: { + parentRunId: originRunId, + generation: delivery.generation, + taskIds: delivery.taskIds, + }, + }); + const wake = (yield* orchestrator.getThreadProjection(threadId)).runs.find( + (run) => run.userMessageId === delivery.messageId, + )!; + assert.equal(wake.status, "queued"); + const completeSibling = Effect.gen(function* () { + const childRun = (yield* orchestrator.getThreadProjection(sibling.childThreadId!)).runs[0]!; + const afterSequence = yield* sink.latestSequence(); + yield* sink.write({ + events: [ + { + ...runEvent({ + threadId: sibling.childThreadId!, + runId: childRun.id, + ordinal: childRun.ordinal, + status: "completed", + now, + }), + payload: { ...childRun, status: "completed", completedAt: now }, + }, + ], + }); + return afterSequence; + }); + return { + orchestrator, + sink, + now, + threadId, + originRunId, + incomingTaskId, + sibling, + delivery, + wake, + completeSibling, + }; + }); + it.layer(layerContinuationWorkerTest)("interrupted completion wakes", (it) => { it.effect.each([ { name: "an interrupt keeps", holdQueue: false, siblingFinishesFirst: false }, @@ -1626,113 +1765,20 @@ it.layer(layerContinuationWorkerTest)("interrupted completion wakes", (it) => { it.effect("cancelling a queued completion wake keeps its other siblings", () => Effect.gen(function* () { - const orchestrator = yield* Orchestrator.OrchestratorV2; - const sink = yield* EventSink.EventSinkV2; - const now = yield* DateTime.now; - const key = "sibling-queued-cancel"; - const threadId = ThreadId.make(`${key}-parent`); - const originRunId = RunId.make(`${key}-origin`); - const originRootId = NodeId.make(`${key}-origin-root`); - const incomingTaskId = NodeId.make(`${key}-incoming`); - yield* seedParentWithTerminalTask({ + const { + orchestrator, + sink, threadId, - projectId: ProjectId.make(`${key}-project`), - runId: originRunId, - rootNodeId: originRootId, - taskId: incomingTaskId, - deliveryState: "claimed", - deliveryTaskIds: [incomingTaskId], - completionWake: "always", - now, - }); - yield* sink.write({ - events: [ - { - id: EventId.make(`event:${key}-origin-root`), - type: "node.updated", - threadId, - runId: originRunId, - nodeId: originRootId, - occurredAt: now, - payload: { - id: originRootId, - threadId, - runId: originRunId, - parentNodeId: null, - rootNodeId: originRootId, - kind: "root_turn", - status: "running", - countsForRun: true, - providerThreadId: null, - providerTurnId: null, - nativeItemRef: null, - runtimeRequestId: null, - checkpointScopeId: null, - startedAt: now, - completedAt: null, - }, - }, - ], - }); - yield* orchestrator.dispatch({ - type: "delegated_task.request", - createdBy: "agent", - creationSource: "mcp", - commandId: CommandId.make(`${key}-sibling`), - parentThreadId: threadId, - parentRunId: originRunId, - parentNodeId: originRootId, - task: "Keep working", - modelSelection, - runtimeMode: "full-access", - interactionMode: "default", - completionWake: "always", - }); - const seeded = yield* orchestrator.getThreadProjection(threadId); - const origin = seeded.runs.find((run) => run.id === originRunId)!; - const sibling = seeded.subagents.find((task) => task.id !== incomingTaskId)!; - // The coordinator finished its spawning turn but is busy with another run, so the wake queues. - yield* sink.write({ - events: [ - { - ...runEvent({ threadId, runId: originRunId, ordinal: 1, status: "completed", now }), - payload: { ...origin, status: "completed", completedAt: now }, - }, - runEvent({ - threadId, - runId: RunId.make(`${key}-busy`), - ordinal: 2, - status: "running", - now, - }), - ], - }); - const delivery = origin.delegatedCompletion!.delivery!; - yield* orchestrator.dispatch({ - type: "message.dispatch", - createdBy: "agent", - creationSource: "server", - commandId: CommandId.make(`${key}-wake`), - threadId, - messageId: delivery.messageId, - text: "Delegated task completed", - attachments: [], - modelSelection, - dispatchMode: { type: "queue_after_active" }, - delegatedCompletion: { - parentRunId: originRunId, - generation: delivery.generation, - taskIds: delivery.taskIds, - }, - }); - const wake = (yield* orchestrator.getThreadProjection(threadId)).runs.find( - (run) => run.userMessageId === delivery.messageId, - )!; - assert.equal(wake.status, "queued"); - + originRunId, + incomingTaskId, + sibling, + delivery, + wake, + completeSibling, + } = yield* queueCompletionWakeBesideRunningSibling("sibling-queued-cancel"); const cancelled = yield* orchestrator.dispatch({ type: "queued-run.cancel", - commandId: CommandId.make(`${key}-cancel`), + commandId: CommandId.make("sibling-queued-cancel-cancel"), threadId, runId: wake.id, }); @@ -1755,37 +1801,96 @@ it.layer(layerContinuationWorkerTest)("interrupted completion wakes", (it) => { "open", ); - const childRun = (yield* orchestrator.getThreadProjection(sibling.childThreadId!)).runs[0]!; - const afterSequence = yield* sink.latestSequence(); + const afterSequence = yield* completeSibling; + yield* sink.stream({ afterSequence, eventType: "subagent.updated" }).pipe( + Stream.filter( + (stored) => + stored.event.type === "subagent.updated" && + stored.event.payload.id === sibling.id && + stored.event.payload.completionDelivery?.state === "claimed", + ), + Stream.take(1), + Stream.runDrain, + ); + const successor = (yield* orchestrator.getThreadProjection(threadId)).runs.find( + (run) => run.id === originRunId, + )?.delegatedCompletion?.delivery; + assert.deepEqual(successor?.taskIds, [sibling.id]); + assert.notEqual(successor?.messageId, delivery.messageId); + }), + ); + + it.effect("settling with a queued completion wake disposes its running siblings", () => + Effect.gen(function* () { + const { orchestrator, sink, now, threadId, originRunId, sibling, wake, completeSibling } = + yield* queueCompletionWakeBesideRunningSibling("sibling-settle"); + const busy = (yield* orchestrator.getThreadProjection(threadId)).runs.find( + (run) => run.status === "running", + )!; yield* sink.write({ events: [ { ...runEvent({ - threadId: sibling.childThreadId!, - runId: childRun.id, - ordinal: childRun.ordinal, - status: "completed", + threadId, + runId: busy.id, + ordinal: busy.ordinal, + status: "cancelled", now, }), - payload: { ...childRun, status: "completed", completedAt: now }, + payload: { ...busy, status: "cancelled", completedAt: now }, + }, + { + ...runEvent({ threadId, runId: wake.id, ordinal: wake.ordinal, status: "queued", now }), + payload: { ...wake, queueHeld: true }, }, ], }); + + const settled = yield* orchestrator.dispatch({ + type: "thread.settle", + commandId: CommandId.make("sibling-settle-settle"), + threadId, + }); + assert.deepEqual( + settled.storedEvents.flatMap(({ event }) => + event.type === "run.updated" && event.payload.id === wake.id + ? [event.payload.status] + : [], + ), + ["cancelled"], + ); + const afterSettle = yield* orchestrator.getThreadProjection(threadId); + assert.equal( + afterSettle.subagents.find((task) => task.id === sibling.id)?.completionDelivery?.state, + "disposed", + ); + + const afterSequence = yield* completeSibling; yield* sink.stream({ afterSequence, eventType: "subagent.updated" }).pipe( Stream.filter( (stored) => stored.event.type === "subagent.updated" && stored.event.payload.id === sibling.id && - stored.event.payload.completionDelivery?.state === "claimed", + stored.event.payload.status === "completed", ), Stream.take(1), Stream.runDrain, ); - const successor = (yield* orchestrator.getThreadProjection(threadId)).runs.find( - (run) => run.id === originRunId, - )?.delegatedCompletion?.delivery; - assert.deepEqual(successor?.taskIds, [sibling.id]); - assert.notEqual(successor?.messageId, delivery.messageId); + const afterLateResult = yield* orchestrator.getThreadProjection(threadId); + assert.equal( + afterLateResult.subagents.find((task) => task.id === sibling.id)?.completionDelivery?.state, + "disposed", + ); + assert.deepEqual( + afterLateResult.runs.map((run) => run.id), + afterSettle.runs.map((run) => run.id), + ); + assert.equal( + afterLateResult.runs.find((run) => run.id === originRunId)?.delegatedCompletion + ?.disposition, + "disposed", + ); + assert.equal(afterLateResult.thread.settledOverride, "settled"); }), ); }); diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 22d467f3e49e3..5f3211647d03a 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -3534,6 +3534,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio runId: run.id, }, events, + "cohort", ); } } @@ -10054,6 +10055,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const dispatchQueuedRunCancel = Effect.fn("Orchestrator.dispatchQueuedRunCancel")(function* ( command: Extract, events: Ref.Ref>, + // Settling disposes a cancelled wake's whole cohort, so a sibling's late result cannot unsettle it. + disposes: "wake" | "cohort" = "wake", ) { const projection = yield* loadProjectionForCommand( command, @@ -10081,7 +10084,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio disposition: "disposed", now, cancelQueuedDelivery: false, - interruptedWakeTaskIds: queuedMessage.delegatedCompletion.taskIds, + ...(disposes === "wake" + ? { interruptedWakeTaskIds: queuedMessage.delegatedCompletion.taskIds } + : {}), }); } yield* emitQueuedRunCancellation({ command, events, projection, run: queuedRun, now });