From 7ab3eed9c284ea8bcacce7cf71f4286bf92d9694 Mon Sep 17 00:00:00 2001 From: PR Batch Tester Date: Sat, 31 Oct 2026 17:07:47 +0000 Subject: [PATCH 1/2] fix(server): preserve sibling results when interrupting a completion wake Limit interrupted wake disposal to its own completion task IDs. Keep undelivered siblings eligible, including tasks that finish during the wake, and reserve the successor through the existing delivery finalization. Explicit Stop still disposes the entire cohort. Co-authored-by: Claude Opus 5.5 --- .../DelegatedCompletionDelivery.test.ts | 191 ++++++++++++++++++ .../src/orchestration-v2/Orchestrator.ts | 41 +++- 2 files changed, 225 insertions(+), 7 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index 3f0f9cf0342ee..634f15f15e379 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -458,6 +458,197 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => { }), ); + it.effect.each([ + { name: "an interrupt keeps", holdQueue: false, siblingFinishesFirst: false }, + { name: "Stop disposes", holdQueue: true, siblingFinishesFirst: false }, + { name: "an interrupt reserves", holdQueue: false, siblingFinishesFirst: true }, + ])("$name a completion wake's other siblings", ({ holdQueue, siblingFinishesFirst }) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const sink = yield* EventSink.EventSinkV2; + const now = yield* DateTime.now; + const key = holdQueue + ? "sibling-stop" + : siblingFinishesFirst + ? "sibling-finished-interrupt" + : "sibling-interrupt"; + const threadId = ThreadId.make(`${key}-parent`); + const originRunId = RunId.make(`${key}-origin`); + const incomingTaskId = NodeId.make(`${key}-incoming`); + yield* seedParentWithTerminalTask({ + threadId, + projectId: ProjectId.make(`${key}-project`), + runId: originRunId, + rootNodeId: NodeId.make(`${key}-origin-root`), + taskId: incomingTaskId, + deliveryState: "claimed", + deliveryTaskIds: [incomingTaskId], + completionWake: "always", + now, + }); + const originRootId = NodeId.make(`${key}-origin-root`); + 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)!; + yield* sink.write({ + events: [ + { + ...runEvent({ threadId, runId: originRunId, ordinal: 1, status: "completed", now }), + payload: { ...origin, status: "completed", completedAt: 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, + }, + }); + yield* orchestrator.resumeQueuedRuns; + const wake = (yield* orchestrator.getThreadProjection(threadId)).runs.find( + (run) => run.userMessageId === delivery.messageId, + )!; + const childRun = (yield* orchestrator.getThreadProjection(sibling.childThreadId!)).runs[0]!; + const completeSibling = Effect.gen(function* () { + 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 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.status === "completed", + ), + Stream.take(1), + Stream.runDrain, + Effect.as(afterSequence), + ); + }); + if (siblingFinishesFirst) { + yield* completeSibling; + assert.equal( + (yield* orchestrator.getThreadProjection(threadId)).subagents.find( + (task) => task.id === sibling.id, + )?.completionDelivery?.state, + "pending", + ); + } + const beforeInterrupt = yield* sink.latestSequence(); + yield* orchestrator.dispatch({ + type: "run.interrupt", + commandId: CommandId.make(`${key}-interrupt`), + threadId, + runId: wake.id, + ...(holdQueue ? { holdQueue } : {}), + }); + const stopped = yield* orchestrator.getThreadProjection(threadId); + const taskDelivery = (taskId: NodeId) => + stopped.subagents.find((task) => task.id === taskId)?.completionDelivery?.state; + assert.equal(taskDelivery(incomingTaskId), "disposed"); + assert.equal( + stopped.runs.find((run) => run.id === originRunId)?.delegatedCompletion?.disposition, + holdQueue ? "stopped" : "open", + ); + assert.equal( + taskDelivery(sibling.id), + holdQueue ? "disposed" : siblingFinishesFirst ? "pending" : undefined, + ); + if (holdQueue) return; + + const afterSequence = siblingFinishesFirst ? beforeInterrupt : yield* completeSibling; + // The wake's terminal, or the sibling's own completion, reserves the successor; the + // generations below show it happened once. + 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 woken = yield* orchestrator.getThreadProjection(threadId); + const cohort = woken.runs.find((run) => run.id === originRunId)?.delegatedCompletion; + assert.equal(cohort?.disposition, "open"); + assert.equal(cohort?.nextGeneration, delivery.generation + 2); + assert.equal(cohort?.delivery?.generation, delivery.generation + 1); + assert.deepEqual(cohort?.delivery?.taskIds, [sibling.id]); + assert.equal( + woken.subagents.find((task) => task.id === sibling.id)?.completionDelivery?.state, + "claimed", + ); + }), + ); + it.effect("acceptance batches pending siblings without acknowledging their results", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index b4bdbd705c290..18df0c47f11bc 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -2696,6 +2696,9 @@ 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. + readonly interruptedWakeTaskIds?: ReadonlyArray; }) { const parentRun = input.projection.runs.find((candidate) => candidate.id === input.parentRunId); if (parentRun === undefined) { @@ -2709,12 +2712,28 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio return; } const emitEvent = emit(input.events, input.command); - const nextDisposition = cohort?.disposition === "disposed" ? "disposed" : input.disposition; - const nextCohort = { - disposition: nextDisposition, - nextGeneration: cohort?.nextGeneration ?? 1, - delivery: null, - } as const; + const wakeTaskIds = input.interruptedWakeTaskIds; + const keptTaskIds = new Set( + wakeTaskIds !== undefined && cohort?.disposition === "open" + ? tasks + .filter( + (task) => + !wakeTaskIds.includes(task.id) && + task.completionDelivery?.state !== "acknowledged" && + task.completionDelivery?.state !== "delivered" && + task.completionDelivery?.state !== "disposed", + ) + .map((task) => task.id) + : [], + ); + const nextCohort = + keptTaskIds.size > 0 + ? cohort! + : ({ + disposition: cohort?.disposition === "disposed" ? "disposed" : input.disposition, + nextGeneration: cohort?.nextGeneration ?? 1, + delivery: null, + } as const); yield* emitEvent({ type: "run.updated", threadId: parentRun.threadId, @@ -2729,6 +2748,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); for (const task of tasks) { if ( + keptTaskIds.has(task.id) || task.completionDelivery?.state === "acknowledged" || task.completionDelivery?.state === "delivered" || task.completionDelivery?.state === "disposed" @@ -2754,7 +2774,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); } const deliveryRun = completionDeliveryRun(input.projection, cohort?.delivery ?? null); - if (input.cancelQueuedDelivery !== false && deliveryRun?.status === "queued") { + if ( + keptTaskIds.size === 0 && + input.cancelQueuedDelivery !== false && + deliveryRun?.status === "queued" + ) { yield* emitQueuedRunCancellation({ command: input.command, events: input.events, @@ -11380,6 +11404,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio parentRunId, disposition: "stopped", now, + ...(parentRunId !== run.id && completionMessage?.delegatedCompletion !== undefined + ? { interruptedWakeTaskIds: completionMessage.delegatedCompletion.taskIds } + : {}), }); } }); From 945a79c333009a5c02a4d369eb7014f8d6752145 Mon Sep 17 00:00:00 2001 From: PR Batch Tester Date: Sat, 31 Oct 2026 17:22:05 +0000 Subject: [PATCH 2/2] test(server): follow an interrupted wake's sibling to its one successor run The interrupt table now reads the interrupt's own stored events instead of a later projection, so the wake's terminal claiming the sibling cannot race it. It runs with the real ProviderContinuationService and follows the claimed sibling through the continuation dispatch to one started successor run, then checks that two recovery passes do not wake the parent again. Co-Authored-By: Claude Opus 5.5 --- .../DelegatedCompletionDelivery.test.ts | 444 ++++++++++-------- 1 file changed, 253 insertions(+), 191 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index 634f15f15e379..956de5eca8399 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -49,6 +49,8 @@ import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts"; import * as ProviderTurnStartServiceTestkit from "./ProviderTurnStartService.testkit.ts"; import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; import * as ProviderContinuationRequests from "@t3tools/provider-core/server/ProviderContinuationRequests"; +import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; +import * as ProviderContinuationService from "./ProviderContinuationService.ts"; const layerPlatformTest = Layer.merge( Layer.mergeAll(NodeServices.layer, NodeCrypto.layer), @@ -152,6 +154,12 @@ const layerTest = Layer.mergeAll( Layer.provide(layerPlatformTest), ); +// ProviderContinuationService consumes the orchestrator's continuation queue, as in production. +const layerContinuationWorkerTest = ProviderContinuationService.layer.pipe( + Layer.provide(Layer.merge(ProviderContinuationRequests.layer, IdAllocator.layer)), + Layer.provideMerge(layerTest), +); + const seedParentWithTerminalTask = (input: { readonly threadId: ThreadId; readonly projectId: ProjectId; @@ -458,197 +466,6 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => { }), ); - it.effect.each([ - { name: "an interrupt keeps", holdQueue: false, siblingFinishesFirst: false }, - { name: "Stop disposes", holdQueue: true, siblingFinishesFirst: false }, - { name: "an interrupt reserves", holdQueue: false, siblingFinishesFirst: true }, - ])("$name a completion wake's other siblings", ({ holdQueue, siblingFinishesFirst }) => - Effect.gen(function* () { - const orchestrator = yield* Orchestrator.OrchestratorV2; - const sink = yield* EventSink.EventSinkV2; - const now = yield* DateTime.now; - const key = holdQueue - ? "sibling-stop" - : siblingFinishesFirst - ? "sibling-finished-interrupt" - : "sibling-interrupt"; - const threadId = ThreadId.make(`${key}-parent`); - const originRunId = RunId.make(`${key}-origin`); - const incomingTaskId = NodeId.make(`${key}-incoming`); - yield* seedParentWithTerminalTask({ - threadId, - projectId: ProjectId.make(`${key}-project`), - runId: originRunId, - rootNodeId: NodeId.make(`${key}-origin-root`), - taskId: incomingTaskId, - deliveryState: "claimed", - deliveryTaskIds: [incomingTaskId], - completionWake: "always", - now, - }); - const originRootId = NodeId.make(`${key}-origin-root`); - 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)!; - yield* sink.write({ - events: [ - { - ...runEvent({ threadId, runId: originRunId, ordinal: 1, status: "completed", now }), - payload: { ...origin, status: "completed", completedAt: 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, - }, - }); - yield* orchestrator.resumeQueuedRuns; - const wake = (yield* orchestrator.getThreadProjection(threadId)).runs.find( - (run) => run.userMessageId === delivery.messageId, - )!; - const childRun = (yield* orchestrator.getThreadProjection(sibling.childThreadId!)).runs[0]!; - const completeSibling = Effect.gen(function* () { - 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 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.status === "completed", - ), - Stream.take(1), - Stream.runDrain, - Effect.as(afterSequence), - ); - }); - if (siblingFinishesFirst) { - yield* completeSibling; - assert.equal( - (yield* orchestrator.getThreadProjection(threadId)).subagents.find( - (task) => task.id === sibling.id, - )?.completionDelivery?.state, - "pending", - ); - } - const beforeInterrupt = yield* sink.latestSequence(); - yield* orchestrator.dispatch({ - type: "run.interrupt", - commandId: CommandId.make(`${key}-interrupt`), - threadId, - runId: wake.id, - ...(holdQueue ? { holdQueue } : {}), - }); - const stopped = yield* orchestrator.getThreadProjection(threadId); - const taskDelivery = (taskId: NodeId) => - stopped.subagents.find((task) => task.id === taskId)?.completionDelivery?.state; - assert.equal(taskDelivery(incomingTaskId), "disposed"); - assert.equal( - stopped.runs.find((run) => run.id === originRunId)?.delegatedCompletion?.disposition, - holdQueue ? "stopped" : "open", - ); - assert.equal( - taskDelivery(sibling.id), - holdQueue ? "disposed" : siblingFinishesFirst ? "pending" : undefined, - ); - if (holdQueue) return; - - const afterSequence = siblingFinishesFirst ? beforeInterrupt : yield* completeSibling; - // The wake's terminal, or the sibling's own completion, reserves the successor; the - // generations below show it happened once. - 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 woken = yield* orchestrator.getThreadProjection(threadId); - const cohort = woken.runs.find((run) => run.id === originRunId)?.delegatedCompletion; - assert.equal(cohort?.disposition, "open"); - assert.equal(cohort?.nextGeneration, delivery.generation + 2); - assert.equal(cohort?.delivery?.generation, delivery.generation + 1); - assert.deepEqual(cohort?.delivery?.taskIds, [sibling.id]); - assert.equal( - woken.subagents.find((task) => task.id === sibling.id)?.completionDelivery?.state, - "claimed", - ); - }), - ); - it.effect("acceptance batches pending siblings without acknowledging their results", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; @@ -1561,6 +1378,251 @@ const seedRestartCancelledChild = (input: { return { taskId, childThreadId, childRunId }; }); +it.layer(layerContinuationWorkerTest)("interrupted completion wakes", (it) => { + it.effect.each([ + { name: "an interrupt keeps", holdQueue: false, siblingFinishesFirst: false }, + { name: "Stop disposes", holdQueue: true, siblingFinishesFirst: false }, + { name: "an interrupt reserves", holdQueue: false, siblingFinishesFirst: true }, + ])("$name a completion wake's other siblings", ({ holdQueue, siblingFinishesFirst }) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const sink = yield* EventSink.EventSinkV2; + const now = yield* DateTime.now; + const key = holdQueue + ? "sibling-stop" + : siblingFinishesFirst + ? "sibling-finished-interrupt" + : "sibling-interrupt"; + const threadId = ThreadId.make(`${key}-parent`); + const originRunId = RunId.make(`${key}-origin`); + const incomingTaskId = NodeId.make(`${key}-incoming`); + yield* seedParentWithTerminalTask({ + threadId, + projectId: ProjectId.make(`${key}-project`), + runId: originRunId, + rootNodeId: NodeId.make(`${key}-origin-root`), + taskId: incomingTaskId, + deliveryState: "claimed", + deliveryTaskIds: [incomingTaskId], + completionWake: "always", + now, + }); + const originRootId = NodeId.make(`${key}-origin-root`); + 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)!; + yield* sink.write({ + events: [ + { + ...runEvent({ threadId, runId: originRunId, ordinal: 1, status: "completed", now }), + payload: { ...origin, status: "completed", completedAt: 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, + }, + }); + yield* orchestrator.resumeQueuedRuns; + const wake = (yield* orchestrator.getThreadProjection(threadId)).runs.find( + (run) => run.userMessageId === delivery.messageId, + )!; + const childRun = (yield* orchestrator.getThreadProjection(sibling.childThreadId!)).runs[0]!; + const completeSibling = Effect.gen(function* () { + 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 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.status === "completed", + ), + Stream.take(1), + Stream.runDrain, + Effect.as(afterSequence), + ); + }); + if (siblingFinishesFirst) { + yield* completeSibling; + assert.equal( + (yield* orchestrator.getThreadProjection(threadId)).subagents.find( + (task) => task.id === sibling.id, + )?.completionDelivery?.state, + "pending", + ); + } + const beforeInterrupt = yield* sink.latestSequence(); + const interrupted = yield* orchestrator.dispatch({ + type: "run.interrupt", + commandId: CommandId.make(`${key}-interrupt`), + threadId, + runId: wake.id, + ...(holdQueue ? { holdQueue } : {}), + }); + // The wake's terminal may claim the sibling right after commit, so read what the interrupt + // itself wrote rather than a later projection. + const interruptDelivery = new Map( + interrupted.storedEvents.flatMap(({ event }) => + event.type === "subagent.updated" + ? [[event.payload.id, event.payload.completionDelivery?.state] as const] + : [], + ), + ); + const interruptCohort = interrupted.storedEvents.flatMap(({ event }) => + event.type === "run.updated" && event.payload.id === originRunId + ? [event.payload.delegatedCompletion] + : [], + ); + assert.equal(interruptDelivery.get(incomingTaskId), "disposed"); + assert.equal(interruptCohort.at(-1)?.disposition, holdQueue ? "stopped" : "open"); + if (holdQueue) { + assert.equal(interruptDelivery.get(sibling.id), "disposed"); + return; + } + assert.isFalse(interruptDelivery.has(sibling.id)); + + const afterSequence = siblingFinishesFirst ? beforeInterrupt : yield* completeSibling; + // The wake's terminal, or the sibling's own completion, reserves the successor. + 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 woken = yield* orchestrator.getThreadProjection(threadId); + const successor = woken.runs.find((run) => run.id === originRunId)?.delegatedCompletion + ?.delivery; + assert.equal(successor?.generation, delivery.generation + 1); + assert.deepEqual(successor?.taskIds, [sibling.id]); + assert.notEqual(successor?.messageId, delivery.messageId); + + // ProviderContinuationService consumes the claim's offer and dispatches the successor. + yield* sink.stream({ afterSequence, eventType: "message.updated" }).pipe( + Stream.filter( + (stored) => + stored.event.type === "message.updated" && + stored.event.payload.id === successor!.messageId, + ), + Stream.take(1), + Stream.runDrain, + ); + // Repeated recovery re-offers every open delivery; none may wake the parent twice. + yield* orchestrator.resumeQueuedRuns; + yield* orchestrator.recoverDelegatedTasks; + yield* orchestrator.recoverDelegatedTasks; + yield* orchestrator.resumeQueuedRuns; + const delivered = yield* orchestrator.getThreadProjection(threadId); + const cohort = delivered.runs.find((run) => run.id === originRunId)?.delegatedCompletion; + assert.equal(cohort?.disposition, "open"); + assert.equal(cohort?.nextGeneration, delivery.generation + 2); + assert.deepEqual(cohort?.delivery, successor); + assert.deepEqual( + delivered.messages + .filter((message) => message.delegatedCompletion?.parentRunId === originRunId) + .map((message) => [message.id, message.delegatedCompletion]), + [ + [ + delivery.messageId, + { + parentRunId: originRunId, + generation: delivery.generation, + taskIds: [incomingTaskId], + }, + ], + [ + successor!.messageId, + { parentRunId: originRunId, generation: successor!.generation, taskIds: [sibling.id] }, + ], + ], + ); + const successorRuns = delivered.runs.filter( + (run) => run.userMessageId === successor!.messageId, + ); + assert.equal(successorRuns.length, 1); + assert.equal(successorRuns[0]?.status, "starting"); + assert.equal( + delivered.subagents.find((task) => task.id === sibling.id)?.completionDelivery?.state, + "claimed", + ); + }), + ); +}); + it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( "subagent follow-ups", (it) => {