1import * as ScriptProgress from "@t3tools/provider-core/server/ScriptProgress";
2import * as SourceControlProviderRegistry from "../sourceControl/SourceControlProviderRegistry.ts";
3import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions";
4import * as NodeServices from "@effect/platform-node/NodeServices";
5import { assert, it } from "@effect/vitest";
6import {
7 CommandId,
8 EventId,
9 MessageId,
10 type ModelSelection,
11 NodeId,
12 type OrchestrationV2Run,
13 ProjectId,
14 ProviderDriverKind,
15 ProviderInstanceId,
16 ProviderThreadId,
17 RunId,
18 ThreadId,
19 TurnItemId,
20} from "@t3tools/contracts";
21import * as DateTime from "effect/DateTime";
22import * as Effect from "effect/Effect";
23import * as Layer from "effect/Layer";
24import * as Stream from "effect/Stream";
25import * as Tracer from "effect/Tracer";
26
27import * as CheckpointStore from "../checkpointing/CheckpointStore.ts";
28import * as ServerConfig from "../config.ts";
29import * as McpSessionRegistryTestkit from "../mcp/McpSessionRegistry.testkit.ts";
30import * as SqlitePersistence from "../persistence/Sqlite.ts";
31import * as ProjectEnrichmentService from "../project/ProjectEnrichmentService.ts";
32import * as ProjectService from "../project/ProjectService.ts";
33import type { ProviderInstance } from "@t3tools/provider-core/server/driver";
34import * as ProviderInstanceRegistry from "../provider/ProviderInstanceRegistry.ts";
35import * as ServerSettings from "../serverSettings.ts";
36import * as VcsDriverRegistry from "../vcs/VcsDriverRegistry.ts";
37import * as VcsProcess from "../vcs/VcsProcess.ts";
38import * as WorkspacePaths from "../workspace/WorkspacePaths.ts";
39import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
40import * as EventSink from "./EventSink.ts";
41import * as Orchestrator from "./Orchestrator.ts";
42import { continueRestartedRun } from "./RestartContinuation.ts";
43import * as RuntimeLayer from "./runtimeLayer.ts";
44import * as ProviderTurnStartServiceTestkit from "./ProviderTurnStartService.testkit.ts";
45import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
46import * as ProviderContinuationRequests from "@t3tools/provider-core/server/ProviderContinuationRequests";
47
48const layerPlatformTest = Layer.merge(
49 NodeServices.layer,
50 Layer.mock(SourceControlProviderRegistry.SourceControlProviderRegistry)({
51 resolveLink: () => Effect.die("unused title link"),
52 }),
53);
54
55const layerServerConfig = ServerConfig.layerTest(process.cwd(), {
56 prefix: "t3-orchestration-v2-delegated-completion-",
57});
58
59const modelSelection = {
60 instanceId: ProviderInstanceId.make("codex"),
61 model: "gpt-5.4",
62} satisfies ModelSelection;
63
64const layerVcsDriverRegistryTest = VcsDriverRegistry.layer.pipe(
65 Layer.provide(VcsProcess.layer),
66 Layer.provide(layerServerConfig),
67 Layer.provide(layerPlatformTest),
68);
69
70const layerCheckpointStoreTest = CheckpointStore.layer.pipe(
71 Layer.provide(layerVcsDriverRegistryTest),
72);
73
74const driver = ProviderDriverKind.make("codex");
75const orchestrationAdapter = {
76 instanceId: modelSelection.instanceId,
77 driver,
78 getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2),
79 planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }),
80 openSession: () => Effect.die("sessions are not used by delegated completion tests"),
81} as ProviderAdapter.ProviderAdapterV2["Service"];
82const providerInstance = {
83 instanceId: modelSelection.instanceId,
84 driverKind: driver,
85 continuationIdentity: {
86 driverKind: driver,
87 continuationKey: "codex:test",
88 },
89 displayName: "Codex test",
90 enabled: true,
91 // No supportedRuntimeModes: every runtime mode runs as stored.
92 snapshot: { getSnapshot: Effect.succeed({}) } as unknown as ProviderInstance["snapshot"],
93 orchestrationAdapter,
94 textGeneration: {} as ProviderInstance["textGeneration"],
95} satisfies ProviderInstance;
96
97const layerTestProviderInstanceRegistry = Layer.succeed(
98 ProviderInstanceRegistry.ProviderInstanceRegistry,
99 {
100 getInstance: (instanceId) =>
101 Effect.succeed(instanceId === providerInstance.instanceId ? providerInstance : undefined),
102 listInstances: Effect.succeed([providerInstance]),
103 listUnavailable: Effect.succeed([]),
104 streamChanges: Stream.empty,
105 subscribeChanges: Effect.never,
106 },
107);
108
109const layerTest = Layer.mergeAll(RuntimeLayer.layer, RuntimeLayer.layerEventSink).pipe(
110 Layer.provideMerge(RuntimeLayer.layerProjectService),
111 Layer.provide(
112 Layer.mock(WorkspacePaths.WorkspacePaths)({
113 normalizeWorkspaceRoot: (workspaceRoot) => Effect.succeed(workspaceRoot),
114 }),
115 ),
116 Layer.provide(ProviderTurnStartServiceTestkit.layer),
117 Layer.provide(
118 Layer.succeed(ProjectEnrichmentService.ProjectEnrichmentService, {
119 peek: () =>
120 Effect.succeed({
121 repositoryIdentity: null,
122 faviconPath: null,
123 repositoryIdentityResolved: false,
124 }),
125 request: () => Effect.void,
126 getAvailable: () =>
127 Effect.succeed({
128 repositoryIdentity: null,
129 faviconPath: null,
130 repositoryIdentityResolved: false,
131 }),
132 invalidate: () => Effect.void,
133 subscribeChanges: Effect.never,
134 }),
135 ),
136 Layer.provide(McpSessionRegistryTestkit.layer),
137 Layer.provide(Layer.merge(McpProviderSessions.layer, ScriptProgress.layerUnavailable)),
138 Layer.provide(SqlitePersistence.layerMemory),
139 Layer.provide(layerCheckpointStoreTest),
140 Layer.provide(layerServerConfig),
141 Layer.provide(ServerSettings.layerTest()),
142 Layer.provide(layerTestProviderInstanceRegistry),
143 Layer.provide(layerPlatformTest),
144);
145
146const seedParentWithTerminalTask = (input: {
147 readonly threadId: ThreadId;
148 readonly projectId: ProjectId;
149 readonly runId: RunId;
150 readonly rootNodeId: NodeId;
151 readonly taskId: NodeId;
152 readonly deliveryState: "delivered" | "claimed" | "acknowledged" | "disposed";
153 readonly completionWake?: "always" | "settled_only";
154 readonly deliveryTaskIds?: ReadonlyArray<NodeId>;
155 readonly now: DateTime.Utc;
156}) =>
157 Effect.gen(function* () {
158 const projects = yield* ProjectService.ProjectService;
159 const orchestrator = yield* Orchestrator.OrchestratorV2;
160 const eventSink = yield* EventSink.EventSinkV2;
161 const providerThreadId = ProviderThreadId.make(
162 `provider-thread:${String(input.threadId).replace("thread:", "")}`,
163 );
164
165 yield* projects.create({
166 commandId: CommandId.make(`command:seed-project:${input.threadId}`),
167 projectId: input.projectId,
168 title: "Delegated completion delivery",
169 workspaceRoot: `/workspace/${input.projectId}`,
170 });
171
172 yield* orchestrator.dispatch({
173 type: "thread.create",
174 createdBy: "user",
175 creationSource: "web",
176 commandId: CommandId.make(`command:seed-create:${input.threadId}`),
177 threadId: input.threadId,
178 projectId: input.projectId,
179 title: "Delegated completion delivery",
180 modelSelection,
181 runtimeMode: "full-access",
182 interactionMode: "default",
183 branch: null,
184 worktreePath: null,
185 });
186
187 yield* eventSink.write({
188 commandId: CommandId.make(`command:seed-projection:${input.threadId}`),
189 events: [
190 {
191 id: EventId.make(`event:seed-provider-thread:${input.threadId}`),
192 type: "provider-thread.updated",
193 threadId: input.threadId,
194 driver,
195 providerInstanceId: modelSelection.instanceId,
196 occurredAt: input.now,
197 payload: {
198 id: providerThreadId,
199 driver,
200 providerInstanceId: modelSelection.instanceId,
201 providerSessionId: null,
202 appThreadId: input.threadId,
203 ownerNodeId: input.rootNodeId,
204 nativeThreadRef: {
205 driver,
206 nativeId: `native:${input.threadId}`,
207 strength: "strong",
208 },
209 nativeConversationHeadRef: null,
210 status: "active",
211 firstRunOrdinal: 1,
212 lastRunOrdinal: 1,
213 handoffIds: [],
214 forkedFrom: null,
215 createdAt: input.now,
216 updatedAt: input.now,
217 },
218 },
219 {
220 id: EventId.make(`event:seed-run:${input.threadId}`),
221 type: "run.updated",
222 threadId: input.threadId,
223 runId: input.runId,
224 nodeId: input.rootNodeId,
225 providerInstanceId: modelSelection.instanceId,
226 occurredAt: input.now,
227 payload: {
228 id: input.runId,
229 threadId: input.threadId,
230 ordinal: 1,
231 providerInstanceId: modelSelection.instanceId,
232 modelSelection,
233 providerThreadId,
234 userMessageId: MessageId.make(`message:seed-user:${input.threadId}`),
235 rootNodeId: input.rootNodeId,
236 activeAttemptId: null,
237 status: "running",
238 requestedAt: input.now,
239 startedAt: input.now,
240 completedAt: null,
241 checkpointId: null,
242 contextHandoffId: null,
243 delegatedCompletion: {
244 disposition: "open",
245 nextGeneration: 2,
246 delivery:
247 input.deliveryTaskIds === undefined
248 ? null
249 : {
250 generation: 1,
251 messageId: MessageId.make(`message:delegated-delivery:${input.threadId}`),
252 taskIds: input.deliveryTaskIds,
253 },
254 },
255 },
256 },
257 {
258 id: EventId.make(`event:seed-task:${input.threadId}`),
259 type: "subagent.updated",
260 threadId: input.threadId,
261 runId: input.runId,
262 nodeId: input.taskId,
263 driver,
264 providerInstanceId: modelSelection.instanceId,
265 occurredAt: input.now,
266 payload: {
267 id: input.taskId,
268 threadId: input.threadId,
269 runId: input.runId,
270 parentNodeId: input.rootNodeId,
271 origin: "app_owned",
272 createdBy: "agent",
273 driver,
274 providerInstanceId: modelSelection.instanceId,
275 providerThreadId: null,
276 childThreadId: null,
277 nativeTaskRef: null,
278 prompt: "Inspect the delivered ownership edge.",
279 title: null,
280 model: null,
281 completionWake: input.completionWake ?? "settled_only",
282 completionDelivery: {
283 state: input.deliveryState,
284 observedByRunId: input.deliveryState === "acknowledged" ? input.runId : null,
285 },
286 status: "completed",
287 result: "child finished",
288 startedAt: input.now,
289 completedAt: input.now,
290 updatedAt: input.now,
291 },
292 },
293 ],
294 });
295 });
296
297it.layer(layerTest)("delegated completion delivery repairs", (it) => {
298 it.effect("acceptance batches pending siblings without acknowledging their results", () =>
299 Effect.gen(function* () {
300 const orchestrator = yield* Orchestrator.OrchestratorV2;
301 const sink = yield* EventSink.EventSinkV2;
302 const now = yield* DateTime.now;
303 const threadId = ThreadId.make("mailbox-batch");
304 const runId = RunId.make("mailbox-parent");
305 const taskId = NodeId.make("mailbox-first");
306 const messageId = MessageId.make(`message:delegated-delivery:${threadId}`);
307 yield* seedParentWithTerminalTask({
308 threadId,
309 runId,
310 projectId: ProjectId.make("mailbox-project"),
311 rootNodeId: NodeId.make("mailbox-root"),
312 taskId,
313 deliveryState: "claimed",
314 completionWake: "always",
315 deliveryTaskIds: [taskId],
316 now,
317 });
318 const projection = yield* orchestrator.getThreadProjection(threadId);
319 const task = projection.subagents[0]!;
320 const pendingIds = [NodeId.make("mailbox-second"), NodeId.make("mailbox-third")];
321 yield* sink.write({
322 events: [
323 {
324 id: EventId.make("mailbox-message"),
325 type: "message.updated",
326 threadId,
327 runId,
328 occurredAt: now,
329 payload: {
330 id: messageId,
331 threadId,
332 runId,
333 nodeId: task.parentNodeId,
334 role: "user",
335 text: "Background task finished",
336 attachments: [],
337 streaming: false,
338 createdBy: "agent",
339 creationSource: "server",
340 createdAt: now,
341 updatedAt: now,
342 delegatedCompletion: { parentRunId: runId, generation: 1, taskIds: [taskId] },
343 },
344 },
345 ...pendingIds.map((id) => ({
346 id: EventId.make(`event:${id}`),
347 type: "subagent.updated" as const,
348 threadId,
349 runId,
350 nodeId: id,
351 occurredAt: now,
352 payload: {
353 ...task,
354 id,
355 completionDelivery: { state: "pending" as const, observedByRunId: null },
356 },
357 })),
358 ],
359 });
360 yield* orchestrator.dispatch({
361 type: "notification.delivery.accept",
362 commandId: CommandId.make("accept-first"),
363 threadId,
364 messageId,
365 });
366 const accepted = yield* orchestrator.getThreadProjection(threadId);
367 assert.equal(
368 accepted.subagents.find((row) => row.id === taskId)?.completionDelivery?.state,
369 "delivered",
370 );
371 const cohort = accepted.runs.find((row) => row.id === runId)?.delegatedCompletion;
372 assert.deepEqual(cohort?.delivery?.taskIds, pendingIds);
373 assert.equal(cohort?.delivery?.generation, 2);
374 for (const id of pendingIds) {
375 assert.deepEqual(accepted.subagents.find((row) => row.id === id)?.completionDelivery, {
376 state: "claimed",
377 observedByRunId: null,
378 });
379 }
380 yield* orchestrator.dispatch({
381 type: "notification.delivery.accept",
382 commandId: CommandId.make("repeat-old-acceptance"),
383 threadId,
384 messageId,
385 });
386 const duplicate = yield* orchestrator.getThreadProjection(threadId);
387 assert.deepEqual(duplicate.runs.find((row) => row.id === runId)?.delegatedCompletion, cohort);
388 }),
389 );
390
391 it.effect("acceptance batches a settled_only sibling once its spawning run ended", () =>
392 Effect.gen(function* () {
393 const orchestrator = yield* Orchestrator.OrchestratorV2;
394 const sink = yield* EventSink.EventSinkV2;
395 const now = yield* DateTime.now;
396 const threadId = ThreadId.make("mailbox-mixed");
397 const runId = RunId.make("mailbox-mixed-parent");
398 const taskId = NodeId.make("mailbox-mixed-always");
399 const siblingId = NodeId.make("mailbox-mixed-settled-only");
400 const messageId = MessageId.make(`message:delegated-delivery:${threadId}`);
401 yield* seedParentWithTerminalTask({
402 threadId,
403 runId,
404 projectId: ProjectId.make("mailbox-mixed-project"),
405 rootNodeId: NodeId.make("mailbox-mixed-root"),
406 taskId,
407 deliveryState: "claimed",
408 completionWake: "always",
409 deliveryTaskIds: [taskId],
410 now,
411 });
412 const projection = yield* orchestrator.getThreadProjection(threadId);
413 const task = projection.subagents[0]!;
414 const spawningRun = projection.runs.find((row) => row.id === runId)!;
415 // A restart cut the spawning run; its continuation is the live run now.
416 yield* sink.write({
417 events: [
418 {
419 ...runEvent({ threadId, runId, ordinal: 1, status: "cancelled", now }),
420 payload: { ...spawningRun, status: "cancelled" as const, completedAt: now },
421 },
422 runEvent({
423 threadId,
424 runId: RunId.make("mailbox-mixed-continuation"),
425 ordinal: 2,
426 status: "running",
427 now,
428 }),
429 {
430 id: EventId.make("mailbox-mixed-message"),
431 type: "message.updated",
432 threadId,
433 runId,
434 occurredAt: now,
435 payload: {
436 id: messageId,
437 threadId,
438 runId,
439 nodeId: task.parentNodeId,
440 role: "user",
441 text: "Background task finished",
442 attachments: [],
443 streaming: false,
444 createdBy: "agent",
445 creationSource: "server",
446 createdAt: now,
447 updatedAt: now,
448 delegatedCompletion: { parentRunId: runId, generation: 1, taskIds: [taskId] },
449 },
450 },
451 {
452 id: EventId.make(`event:${siblingId}`),
453 type: "subagent.updated" as const,
454 threadId,
455 runId,
456 nodeId: siblingId,
457 occurredAt: now,
458 payload: {
459 ...task,
460 id: siblingId,
461 completionWake: "settled_only" as const,
462 completionDelivery: { state: "pending" as const, observedByRunId: null },
463 },
464 },
465 ],
466 });
467 yield* orchestrator.dispatch({
468 type: "notification.delivery.accept",
469 commandId: CommandId.make("accept-mixed"),
470 threadId,
471 messageId,
472 });
473 const accepted = yield* orchestrator.getThreadProjection(threadId);
474 const cohort = accepted.runs.find((row) => row.id === runId)?.delegatedCompletion;
475 assert.deepEqual(cohort?.delivery?.taskIds, [siblingId]);
476 assert.equal(
477 accepted.subagents.find((row) => row.id === siblingId)?.completionDelivery?.state,
478 "claimed",
479 );
480 }),
481 );
482
483 it.effect("builds completion text and metadata from the same live cohort", () =>
484 Effect.gen(function* () {
485 const orchestrator = yield* Orchestrator.OrchestratorV2;
486 const now = yield* DateTime.now;
487 const threadId = ThreadId.make("thread:delegated-delivery-live-cohort");
488 const projectId = ProjectId.make("project:delegated-delivery-live-cohort");
489 const runId = RunId.make("run:delegated-delivery-live-cohort");
490 const rootNodeId = NodeId.make("node:delegated-delivery-live-cohort-root");
491 const firstTaskId = NodeId.make("node:delegated-delivery-live-cohort-first");
492 const secondTaskId = NodeId.make("node:delegated-delivery-live-cohort-second");
493 const messageId = MessageId.make(`message:delegated-delivery:${threadId}`);
494
495 yield* seedParentWithTerminalTask({
496 threadId,
497 projectId,
498 runId,
499 rootNodeId,
500 taskId: firstTaskId,
501 deliveryState: "claimed",
502 completionWake: "always",
503 deliveryTaskIds: [firstTaskId, secondTaskId],
504 now,
505 });
506
507 yield* orchestrator.dispatch({
508 type: "message.dispatch",
509 commandId: CommandId.make("command:delegated-delivery-live-cohort"),
510 threadId,
511 messageId,
512 text: `Delegated task ${firstTaskId} reached a terminal state.`,
513 attachments: [],
514 dispatchMode: { type: "queue_after_active" },
515 createdBy: "agent",
516 creationSource: "server",
517 delegatedCompletion: {
518 parentRunId: runId,
519 generation: 1,
520 taskIds: [firstTaskId],
521 },
522 });
523
524 const projection = yield* orchestrator.getThreadProjection(threadId);
525 const message = projection.messages.find((candidate) => candidate.id === messageId);
526 assert.deepEqual(message?.delegatedCompletion?.taskIds, [firstTaskId, secondTaskId]);
527 assert.include(message?.text ?? "", String(firstTaskId));
528 assert.include(message?.text ?? "", String(secondTaskId));
529 assert.include(message?.text ?? "", "task_status");
530 }),
531 );
532
533 it.effect("does not re-offer when wake-policy upgrades after delivered ownership settled", () =>
534 Effect.gen(function* () {
535 const orchestrator = yield* Orchestrator.OrchestratorV2;
536 const now = yield* DateTime.now;
537 const threadId = ThreadId.make("thread:delegated-delivery-a1");
538 const projectId = ProjectId.make("project:delegated-delivery-a1");
539 const runId = RunId.make("run:delegated-delivery-a1");
540 const rootNodeId = NodeId.make("node:delegated-delivery-a1-root");
541 const taskId = NodeId.make("node:delegated-delivery-a1-task");
542
543 yield* seedParentWithTerminalTask({
544 threadId,
545 projectId,
546 runId,
547 rootNodeId,
548 taskId,
549 deliveryState: "delivered",
550 completionWake: "settled_only",
551 now,
552 });
553
554 const upgrade = yield* orchestrator.dispatch({
555 type: "delegated_task.wake-policy",
556 commandId: CommandId.make("command:delegated-delivery-a1:wake-policy"),
557 parentThreadId: threadId,
558 taskId,
559 completionWake: "always",
560 });
561
562 const projection = yield* orchestrator.getThreadProjection(threadId);
563 const task = projection.subagents.find((candidate) => candidate.id === taskId);
564 const parentRun = projection.runs.find((candidate) => candidate.id === runId);
565
566 assert.equal(task?.completionWake, "always");
567 assert.deepEqual(task?.completionDelivery, {
568 state: "delivered",
569 observedByRunId: null,
570 });
571 assert.deepEqual(parentRun?.delegatedCompletion, {
572 disposition: "open",
573 nextGeneration: 2,
574 delivery: null,
575 });
576 assert.isFalse(
577 upgrade.storedEvents.some(
578 (stored) =>
579 stored.event.type === "subagent.updated" &&
580 stored.event.payload.id === taskId &&
581 stored.event.payload.completionDelivery?.state === "claimed",
582 ),
583 );
584 assert.isFalse(
585 upgrade.storedEvents.some(
586 (stored) =>
587 stored.event.type === "run.updated" &&
588 stored.event.payload.id === runId &&
589 stored.event.payload.delegatedCompletion?.delivery !== null &&
590 stored.event.payload.delegatedCompletion?.delivery !== undefined,
591 ),
592 );
593 }),
594 );
595
596 it.effect(
597 "treats repeated acknowledge and dispose with distinct command IDs as successful no-ops",
598 () =>
599 Effect.gen(function* () {
600 const orchestrator = yield* Orchestrator.OrchestratorV2;
601 const now = yield* DateTime.now;
602 const threadId = ThreadId.make("thread:delegated-delivery-a2");
603 const projectId = ProjectId.make("project:delegated-delivery-a2");
604 const runId = RunId.make("run:delegated-delivery-a2");
605 const rootNodeId = NodeId.make("node:delegated-delivery-a2-root");
606 const taskId = NodeId.make("node:delegated-delivery-a2-task");
607
608 yield* seedParentWithTerminalTask({
609 threadId,
610 projectId,
611 runId,
612 rootNodeId,
613 taskId,
614 deliveryState: "delivered",
615 completionWake: "always",
616 now,
617 });
618
619 // Distinct command IDs mirror task_status vs t3_thread_read racing after
620 // their shared read preflight saw delivered ownership.
621 const firstAck = yield* orchestrator.dispatch({
622 type: "delegated_task.completion-delivery.acknowledge",
623 commandId: CommandId.make("command:delegated-delivery-a2:ack-task-status"),
624 parentThreadId: threadId,
625 taskId,
626 observedByRunId: runId,
627 });
628 const secondAck = yield* orchestrator.dispatch({
629 type: "delegated_task.completion-delivery.acknowledge",
630 commandId: CommandId.make("command:delegated-delivery-a2:ack-thread-read"),
631 parentThreadId: threadId,
632 taskId,
633 observedByRunId: runId,
634 });
635
636 const firstAckTask = firstAck.storedEvents.find(
637 (stored) =>
638 stored.event.type === "subagent.updated" && stored.event.payload.id === taskId,
639 );
640 const secondAckTask = secondAck.storedEvents.find(
641 (stored) =>
642 stored.event.type === "subagent.updated" && stored.event.payload.id === taskId,
643 );
644 assert.isDefined(firstAckTask);
645 assert.isDefined(secondAckTask);
646 if (
647 firstAckTask?.event.type !== "subagent.updated" ||
648 secondAckTask?.event.type !== "subagent.updated"
649 ) {
650 return yield* Effect.die(new Error("Acknowledge events missing."));
651 }
652 assert.equal(firstAckTask.event.payload.completionDelivery?.state, "acknowledged");
653 assert.equal(secondAckTask.event.payload.completionDelivery?.state, "acknowledged");
654 // Idempotent replay keeps the first observation's ownership and timestamp.
655 assert.deepEqual(
656 secondAckTask.event.payload.completionDelivery,
657 firstAckTask.event.payload.completionDelivery,
658 );
659 assert.deepEqual(
660 secondAckTask.event.payload.updatedAt,
661 firstAckTask.event.payload.updatedAt,
662 );
663 assert.equal(secondAck.storedEvents.length, 1);
664
665 const afterAck = yield* orchestrator.getThreadProjection(threadId);
666 assert.deepEqual(
667 afterAck.subagents.find((candidate) => candidate.id === taskId)?.completionDelivery,
668 {
669 state: "acknowledged",
670 observedByRunId: runId,
671 },
672 );
673
674 const firstDispose = yield* orchestrator.dispatch({
675 type: "delegated_task.completion-delivery.dispose",
676 commandId: CommandId.make("command:delegated-delivery-a2:dispose-task-status"),
677 parentThreadId: threadId,
678 taskId,
679 });
680 const secondDispose = yield* orchestrator.dispatch({
681 type: "delegated_task.completion-delivery.dispose",
682 commandId: CommandId.make("command:delegated-delivery-a2:dispose-thread-read"),
683 parentThreadId: threadId,
684 taskId,
685 });
686
687 const firstDisposeTask = firstDispose.storedEvents.find(
688 (stored) =>
689 stored.event.type === "subagent.updated" && stored.event.payload.id === taskId,
690 );
691 const secondDisposeTask = secondDispose.storedEvents.find(
692 (stored) =>
693 stored.event.type === "subagent.updated" && stored.event.payload.id === taskId,
694 );
695 assert.isDefined(firstDisposeTask);
696 assert.isDefined(secondDisposeTask);
697 if (
698 firstDisposeTask?.event.type !== "subagent.updated" ||
699 secondDisposeTask?.event.type !== "subagent.updated"
700 ) {
701 return yield* Effect.die(new Error("Dispose events missing."));
702 }
703 assert.equal(firstDisposeTask.event.payload.completionDelivery?.state, "disposed");
704 assert.equal(secondDisposeTask.event.payload.completionDelivery?.state, "disposed");
705 assert.deepEqual(
706 secondDisposeTask.event.payload.completionDelivery,
707 firstDisposeTask.event.payload.completionDelivery,
708 );
709 assert.equal(secondDispose.storedEvents.length, 1);
710
711 const afterDispose = yield* orchestrator.getThreadProjection(threadId);
712 assert.deepEqual(
713 afterDispose.subagents.find((candidate) => candidate.id === taskId)?.completionDelivery,
714 {
715 state: "disposed",
716 observedByRunId: null,
717 },
718 );
719
720 const acknowledgeAfterDispose = yield* orchestrator.dispatch({
721 type: "delegated_task.completion-delivery.acknowledge",
722 commandId: CommandId.make("command:delegated-delivery-a2:ack-after-dispose"),
723 parentThreadId: threadId,
724 taskId,
725 observedByRunId: runId,
726 });
727 const acknowledgedTask = acknowledgeAfterDispose.storedEvents.find(
728 (stored) => stored.event.type === "subagent.updated",
729 );
730 if (acknowledgedTask?.event.type !== "subagent.updated") {
731 return yield* Effect.die(new Error("Acknowledge-after-dispose event missing."));
732 }
733 assert.deepEqual(acknowledgedTask.event.payload.completionDelivery, {
734 state: "disposed",
735 observedByRunId: null,
736 });
737 assert.equal(acknowledgeAfterDispose.storedEvents.length, 1);
738
739 const afterStaleAcknowledge = yield* orchestrator.getThreadProjection(threadId);
740 assert.deepEqual(
741 afterStaleAcknowledge.subagents.find((candidate) => candidate.id === taskId)
742 ?.completionDelivery,
743 {
744 state: "disposed",
745 observedByRunId: null,
746 },
747 );
748 }),
749 );
750});
751
752// Runtime reconciliation writes restart cancellations under this command
753// prefix, which the live terminal-run listener skips.
754// Collects the SQL text of every statement run under the returned tracer.
755const traceSqlStatements = () => {
756 const statements: Array<string> = [];
757 const tracer = Tracer.make({
758 span(options) {
759 const span = new Tracer.NativeSpan(options);
760 const end = span.end.bind(span);
761 span.end = (endTime, exit) => {
762 end(endTime, exit);
763 const query = span.attributes.get("db.query.text");
764 if (typeof query === "string") statements.push(query);
765 };
766 return span;
767 },
768 });
769 return { statements, tracer };
770};
771
772const reconcileCommandId = (name: string) => CommandId.make(`command:runtime-reconcile:${name}`);
773
774const parentProviderThreadId = (threadId: ThreadId) =>
775 ProviderThreadId.make(`provider-thread:${String(threadId).replace("thread:", "")}`);
776
777const runEvent = (input: {
778 readonly threadId: ThreadId;
779 readonly runId: RunId;
780 readonly ordinal: number;
781 readonly status: OrchestrationV2Run["status"];
782 readonly now: DateTime.Utc;
783 readonly providerThreadId?: ProviderThreadId;
784 readonly userMessageId?: MessageId;
785 readonly delegatedCompletion?: OrchestrationV2Run["delegatedCompletion"];
786}) => ({
787 id: EventId.make(`event:${input.runId}:${input.status}`),
788 type: "run.updated" as const,
789 threadId: input.threadId,
790 runId: input.runId,
791 providerInstanceId: modelSelection.instanceId,
792 occurredAt: input.now,
793 payload: {
794 id: input.runId,
795 threadId: input.threadId,
796 ordinal: input.ordinal,
797 providerInstanceId: modelSelection.instanceId,
798 modelSelection,
799 providerThreadId: input.providerThreadId ?? null,
800 userMessageId: input.userMessageId ?? MessageId.make(`message:${input.runId}`),
801 rootNodeId: null,
802 activeAttemptId: null,
803 status: input.status,
804 requestedAt: input.now,
805 startedAt: input.now,
806 completedAt: input.status === "running" ? null : input.now,
807 checkpointId: null,
808 contextHandoffId: null,
809 ...(input.delegatedCompletion === undefined
810 ? {}
811 : { delegatedCompletion: input.delegatedCompletion }),
812 },
813});
814
815/** A running app-owned task whose child thread's first run the restart cancelled. */
816const seedRestartCancelledChild = (input: {
817 readonly parentThreadId: ThreadId;
818 readonly projectId: ProjectId;
819 readonly parentRunId: RunId;
820 readonly rootNodeId: NodeId;
821 readonly name: string;
822 readonly completionWake: "always" | "settled_only";
823 readonly continuationPending: boolean;
824 /** "completed" seeds a settled turn whose background work the restart cancelled. */
825 readonly runStatus?: "cancelled" | "completed";
826 /** Background work the restart cancelled on the child's run. */
827 readonly lostWork?: OrchestrationV2Run["restartCancelledBackgroundWork"];
828 readonly now: DateTime.Utc;
829}) =>
830 Effect.gen(function* () {
831 const eventSink = yield* EventSink.EventSinkV2;
832 const taskId = NodeId.make(`node:${input.name}`);
833 const childThreadId = ThreadId.make(`thread:${input.name}`);
834 const childRunId = RunId.make(`run:${input.name}:1`);
835 yield* eventSink.write({
836 commandId: CommandId.make(`command:seed-child:${input.name}`),
837 events: [
838 {
839 id: EventId.make(`event:${input.name}:thread`),
840 type: "thread.created",
841 threadId: childThreadId,
842 occurredAt: input.now,
843 payload: {
844 createdBy: "agent",
845 creationSource: "server",
846 id: childThreadId,
847 projectId: input.projectId,
848 title: input.name,
849 providerInstanceId: modelSelection.instanceId,
850 modelSelection,
851 runtimeMode: "full-access",
852 interactionMode: "default",
853 branch: null,
854 worktreePath: null,
855 activeProviderThreadId: null,
856 lineage: {
857 parentThreadId: input.parentThreadId,
858 relationshipToParent: "subagent",
859 rootThreadId: input.parentThreadId,
860 },
861 forkedFrom: { type: "node", nodeId: taskId },
862 createdAt: input.now,
863 updatedAt: input.now,
864 archivedAt: null,
865 settledOverride: null,
866 settledAt: null,
867 lastVisitedAt: null,
868 deletedAt: null,
869 },
870 },
871 {
872 id: EventId.make(`event:${input.name}:task`),
873 type: "subagent.updated",
874 threadId: input.parentThreadId,
875 runId: input.parentRunId,
876 nodeId: taskId,
877 driver,
878 providerInstanceId: modelSelection.instanceId,
879 occurredAt: input.now,
880 payload: {
881 id: taskId,
882 threadId: input.parentThreadId,
883 runId: input.parentRunId,
884 parentNodeId: input.rootNodeId,
885 origin: "app_owned",
886 createdBy: "agent",
887 driver,
888 providerInstanceId: modelSelection.instanceId,
889 providerThreadId: null,
890 childThreadId,
891 nativeTaskRef: null,
892 prompt: `Run ${input.name}.`,
893 title: null,
894 model: null,
895 completionWake: input.completionWake,
896 status: "running",
897 result: null,
898 startedAt: input.now,
899 completedAt: null,
900 updatedAt: input.now,
901 },
902 },
903 ],
904 });
905 const childRun = runEvent({
906 threadId: childThreadId,
907 runId: childRunId,
908 ordinal: 1,
909 status: input.runStatus ?? "cancelled",
910 now: input.now,
911 });
912 yield* eventSink.writeWithEffects({
913 commandId: reconcileCommandId(input.name),
914 events: [
915 input.lostWork === undefined
916 ? childRun
917 : {
918 ...childRun,
919 payload: { ...childRun.payload, restartCancelledBackgroundWork: input.lostWork },
920 },
921 ],
922 effects: input.continuationPending
923 ? [
924 {
925 id: `effect:restart-continuation:${childRunId}`,
926 commandId: reconcileCommandId(input.name),
927 threadId: childThreadId,
928 request: { type: "provider-runtime.continue", sourceRunId: childRunId },
929 },
930 ]
931 : [],
932 });
933 return { taskId, childThreadId, childRunId };
934 });
935
936it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))(
937 "subagent follow-ups",
938 (it) => {
939 const outcomes = [
940 "completed",
941 "failed",
942 "interrupted",
943 "restarted",
944 "stopped",
945 "overlapping",
946 ] as const;
947 it.effect.each(outcomes)(
948 "delivers a %s follow-up independently of the original task",
949 (outcome) =>
950 Effect.gen(function* () {
951 const orchestrator = yield* Orchestrator.OrchestratorV2;
952 const eventSink = yield* EventSink.EventSinkV2;
953 const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
954 const now = yield* DateTime.now;
955 const parentThreadId = ThreadId.make(`thread:followup-${outcome}`);
956 const projectId = ProjectId.make(`project:followup-${outcome}`);
957 const parentRunId = RunId.make(`run:followup-${outcome}`);
958 const rootNodeId = NodeId.make(`node:followup-${outcome}`);
959 yield* seedParentWithTerminalTask({
960 threadId: parentThreadId,
961 projectId,
962 runId: parentRunId,
963 rootNodeId,
964 taskId: NodeId.make(`node:seed-${outcome}`),
965 deliveryState: "acknowledged",
966 now,
967 });
968 const child = yield* seedRestartCancelledChild({
969 parentThreadId,
970 projectId,
971 parentRunId,
972 rootNodeId,
973 name: `followup-child-${outcome}`,
974 completionWake: "always",
975 continuationPending: false,
976 runStatus: "completed",
977 now,
978 });
979 yield* orchestrator.recoverDelegatedTasks;
980 let firstWake = yield* requests.take;
981 while (firstWake.threadId !== parentThreadId) firstWake = yield* requests.take;
982 assert.equal(firstWake.delegatedCompletion?.parentRunId, parentRunId);
983 yield* orchestrator.dispatch({
984 type: "delegated_task.completion-delivery.acknowledge",
985 commandId: CommandId.make(`command:followup-ack-${outcome}`),
986 parentThreadId,
987 taskId: child.taskId,
988 observedByRunId: parentRunId,
989 });
990 const original = (yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find(
991 (task) => task.id === child.taskId,
992 )!;
993 const parentRun = (yield* orchestrator.getThreadProjection(parentThreadId)).runs.find(
994 (run) => run.id === parentRunId,
995 )!;
996 yield* eventSink.write({
997 commandId: reconcileCommandId(`followup-parent-finished-${outcome}`),
998 events: [
999 {
1000 id: EventId.make(`event:followup-parent-finished-${outcome}`),
1001 type: "run.updated",
1002 threadId: parentThreadId,
1003 runId: parentRunId,
1004 occurredAt: now,
1005 payload: { ...parentRun, status: "completed", completedAt: now },
1006 },
1007 ],
1008 });
1009 const messageId = MessageId.make(`message:followup-${outcome}`);
1010 yield* orchestrator.dispatch({
1011 type: "message.dispatch",
1012 createdBy: outcome === "failed" ? "user" : "agent",
1013 creationSource: outcome === "failed" ? "web" : "mcp",
1014 commandId: CommandId.make(`command:followup-send-${outcome}`),
1015 threadId: child.childThreadId,
1016 senderThreadId: parentThreadId,
1017 messageId,
1018 text: "Check the same work again.",
1019 attachments: [],
1020 dispatchMode: { type: "start_immediately" },
1021 });
1022 const childProjection = yield* orchestrator.getThreadProjection(child.childThreadId);
1023 let run = childProjection.runs.find((row) => row.userMessageId === messageId)!;
1024 assert.isDefined(run.delegatedTaskId);
1025 const workingParent = yield* orchestrator.getShellSnapshot();
1026 const backgroundTask = workingParent.threads
1027 .find((thread) => thread.id === parentThreadId)
1028 ?.pendingBackgroundTasks?.find((task) => task.kind === "subagent");
1029 assert.equal(backgroundTask?.childThreadId, child.childThreadId);
1030 if (outcome === "overlapping") {
1031 yield* orchestrator.dispatch({
1032 type: "message.dispatch",
1033 createdBy: "agent",
1034 creationSource: "mcp",
1035 commandId: CommandId.make("command:overlapping-followup"),
1036 threadId: child.childThreadId,
1037 senderThreadId: parentThreadId,
1038 messageId: MessageId.make("message:overlapping-followup"),
1039 text: "Keep working on another follow-up.",
1040 attachments: [],
1041 dispatchMode: { type: "queue_after_active" },
1042 });
1043 }
1044 if (outcome === "restarted") {
1045 const cut = { ...run, status: "cancelled" as const, startedAt: now, completedAt: now };
1046 yield* eventSink.writeWithEffects({
1047 commandId: reconcileCommandId("followup-restart-cut"),
1048 events: [
1049 {
1050 id: EventId.make("event:followup-restart-cut"),
1051 type: "run.updated",
1052 threadId: child.childThreadId,
1053 runId: run.id,
1054 occurredAt: now,
1055 payload: cut,
1056 },
1057 ],
1058 effects: [
1059 {
1060 id: `effect:restart-continuation:${run.id}`,
1061 commandId: reconcileCommandId("followup-restart-cut"),
1062 threadId: child.childThreadId,
1063 request: { type: "provider-runtime.continue", sourceRunId: run.id },
1064 },
1065 ],
1066 });
1067 yield* orchestrator.recoverDelegatedTasks;
1068 assert.isNull(
1069 (yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find(
1070 (task) => task.id === run.delegatedTaskId,
1071 )!.result,
1072 );
1073 const continuationMessageId = MessageId.make("message:followup-restart-continuation");
1074 yield* orchestrator.dispatch({
1075 type: "message.dispatch",
1076 createdBy: "agent",
1077 creationSource: "server",
1078 commandId: CommandId.make("command:followup-restart-continuation"),
1079 threadId: child.childThreadId,
1080 restartContinuationOfRunId: run.id,
1081 messageId: continuationMessageId,
1082 text: "Continue where you left off.",
1083 attachments: [],
1084 dispatchMode: { type: "start_immediately" },
1085 });
1086 const continuation = (yield* orchestrator.getThreadProjection(
1087 child.childThreadId,
1088 )).runs.find((row) => row.userMessageId === continuationMessageId)!;
1089 assert.equal(continuation.delegatedTaskId, run.delegatedTaskId);
1090 run = continuation;
1091 }
1092 if (outcome === "stopped") {
1093 yield* orchestrator.dispatch({
1094 type: "thread.stop",
1095 threadId: parentThreadId,
1096 commandId: CommandId.make(`command:followup-stop-${outcome}`),
1097 });
1098 }
1099 // Reconciliation skips the live terminal listener; recovery must finish this recorded task.
1100 yield* eventSink.write({
1101 commandId: reconcileCommandId(`followup-finish-${outcome}`),
1102 events: [
1103 {
1104 id: EventId.make(`event:followup-finish-${outcome}`),
1105 type: "run.updated",
1106 threadId: child.childThreadId,
1107 runId: run.id,
1108 occurredAt: now,
1109 payload: {
1110 ...run,
1111 status:
1112 outcome === "stopped"
1113 ? "interrupted"
1114 : outcome === "restarted" || outcome === "overlapping"
1115 ? "completed"
1116 : outcome,
1117 startedAt: now,
1118 completedAt: now,
1119 },
1120 },
1121 ],
1122 });
1123 yield* orchestrator.recoverDelegatedTasks;
1124 const finished = yield* orchestrator.getThreadProjection(parentThreadId);
1125 assert.deepEqual(
1126 finished.subagents.find((task) => task.id === child.taskId),
1127 original,
1128 );
1129 const followup = finished.subagents.find((task) => task.id === run.delegatedTaskId)!;
1130 if (outcome === "overlapping") {
1131 const later = finished.subagents.find(
1132 (task) =>
1133 task.childThreadId === child.childThreadId &&
1134 task.id !== child.taskId &&
1135 task.id !== followup.id,
1136 )!;
1137 assert.equal(later.status, "pending");
1138 assert.isNull(later.result);
1139 }
1140 assert.equal(
1141 followup.status,
1142 outcome === "stopped"
1143 ? "interrupted"
1144 : outcome === "restarted" || outcome === "overlapping"
1145 ? "completed"
1146 : outcome,
1147 );
1148 assert.equal(
1149 followup.completionDelivery?.state,
1150 outcome === "stopped" ? "disposed" : "claimed",
1151 );
1152 if (outcome !== "stopped") {
1153 let wake = yield* requests.take;
1154 while (
1155 wake.threadId !== parentThreadId ||
1156 wake.delegatedCompletion?.messageId === firstWake.delegatedCompletion?.messageId
1157 ) {
1158 wake = yield* requests.take;
1159 }
1160 assert.equal(wake.delegatedCompletion?.parentRunId, parentRunId);
1161 assert.notEqual(
1162 wake.delegatedCompletion?.messageId,
1163 firstWake.delegatedCompletion?.messageId,
1164 );
1165 }
1166 assert.equal(
1167 finished.contextTransfers.filter(
1168 (transfer) =>
1169 transfer.type === "subagent_result" &&
1170 transfer.sourceThreadId === child.childThreadId,
1171 ).length,
1172 2,
1173 );
1174 yield* orchestrator.recoverDelegatedTasks;
1175 const recovered = yield* orchestrator.getThreadProjection(parentThreadId);
1176 assert.equal(recovered.contextTransfers.length, finished.contextTransfers.length);
1177 if (outcome === "completed") {
1178 const automaticMessageId = MessageId.make("message:followup-automatic-wake");
1179 yield* orchestrator.dispatch({
1180 type: "message.dispatch",
1181 createdBy: "agent",
1182 creationSource: "server",
1183 commandId: CommandId.make("command:followup-automatic-wake"),
1184 threadId: child.childThreadId,
1185 messageId: automaticMessageId,
1186 text: "A monitor reported new output.",
1187 attachments: [],
1188 notification: {
1189 source: { kind: "monitor" },
1190 outcome: "updated",
1191 summary: "Monitor updated",
1192 },
1193 dispatchMode: { type: "queue_after_active" },
1194 });
1195 const automatic = (yield* orchestrator.getThreadProjection(
1196 child.childThreadId,
1197 )).runs.find((run) => run.userMessageId === automaticMessageId)!;
1198 assert.isUndefined(automatic.delegatedTaskId);
1199 yield* eventSink.write({
1200 commandId: reconcileCommandId("followup-automatic-finished"),
1201 events: [
1202 {
1203 id: EventId.make("event:followup-automatic-finished"),
1204 type: "run.updated",
1205 threadId: child.childThreadId,
1206 runId: automatic.id,
1207 occurredAt: now,
1208 payload: { ...automatic, status: "completed", startedAt: now, completedAt: now },
1209 },
1210 ],
1211 });
1212 yield* orchestrator.recoverDelegatedTasks;
1213 const afterAutomatic = yield* orchestrator.getThreadProjection(parentThreadId);
1214 assert.deepEqual(afterAutomatic.subagents, recovered.subagents);
1215 assert.equal(afterAutomatic.contextTransfers.length, recovered.contextTransfers.length);
1216 const control = yield* seedRestartCancelledChild({
1217 parentThreadId,
1218 projectId,
1219 parentRunId,
1220 rootNodeId,
1221 name: "followup-control-child",
1222 completionWake: "always",
1223 continuationPending: false,
1224 runStatus: "completed",
1225 now,
1226 });
1227 yield* orchestrator.recoverDelegatedTasks;
1228 const afterControl = yield* orchestrator.getThreadProjection(parentThreadId);
1229 assert.equal(
1230 afterControl.subagents.find((task) => task.id === control.taskId)?.completionDelivery
1231 ?.state,
1232 "claimed",
1233 );
1234 assert.equal(
1235 afterControl.contextTransfers.filter(
1236 (transfer) =>
1237 transfer.type === "subagent_result" &&
1238 transfer.sourceThreadId === control.childThreadId,
1239 ).length,
1240 1,
1241 );
1242 }
1243 }),
1244 );
1245 },
1246);
1247
1248it.layer(layerTest)("delegated tasks across a server restart", (it) => {
1249 it.effect("holds a restart-cancelled child for its continuation's result", () =>
1250 Effect.gen(function* () {
1251 const orchestrator = yield* Orchestrator.OrchestratorV2;
1252 const eventSink = yield* EventSink.EventSinkV2;
1253 const now = yield* DateTime.now;
1254 const threadId = ThreadId.make("thread:restart-parent");
1255 const projectId = ProjectId.make("project:restart-parent");
1256 const runId = RunId.make("run:restart-parent");
1257 const rootNodeId = NodeId.make("node:restart-parent-root");
1258 yield* seedParentWithTerminalTask({
1259 threadId,
1260 projectId,
1261 runId,
1262 rootNodeId,
1263 taskId: NodeId.make("node:restart-parent-settled"),
1264 deliveryState: "delivered",
1265 now,
1266 });
1267 const child = (
1268 name: string,
1269 continuationPending: boolean,
1270 runStatus?: "cancelled" | "completed",
1271 ) =>
1272 seedRestartCancelledChild({
1273 parentThreadId: threadId,
1274 projectId,
1275 parentRunId: runId,
1276 rootNodeId,
1277 name,
1278 completionWake: "always",
1279 continuationPending,
1280 ...(runStatus === undefined ? {} : { runStatus }),
1281 now,
1282 });
1283 const resumed = yield* child("restart-resumed-child", true);
1284 const stopped = yield* child("restart-stopped-child", false);
1285 // Settled with only background work left: its interim reply is not the result.
1286 const backgrounded = yield* child("restart-backgrounded-child", true, "completed");
1287 // A second restart cut the first continuation before it started.
1288 const recut = yield* child("restart-recut-child", false);
1289 const recutContinuationId = RunId.make("run:restart-recut-child:2");
1290 const recutCommandId = CommandId.make("command:restart-recut-child:reconcile");
1291 const recutRun = runEvent({
1292 threadId: recut.childThreadId,
1293 runId: recutContinuationId,
1294 ordinal: 2,
1295 status: "cancelled",
1296 now,
1297 });
1298 yield* eventSink.writeWithEffects({
1299 commandId: recutCommandId,
1300 events: [
1301 {
1302 ...recutRun,
1303 payload: {
1304 ...recutRun.payload,
1305 startedAt: null,
1306 restartContinuationOfRunId: recut.childRunId,
1307 },
1308 },
1309 ],
1310 effects: [
1311 {
1312 id: `effect:restart-continuation:${recutContinuationId}`,
1313 commandId: recutCommandId,
1314 threadId: recut.childThreadId,
1315 request: { type: "provider-runtime.continue", sourceRunId: recutContinuationId },
1316 },
1317 ],
1318 });
1319
1320 yield* orchestrator.recoverDelegatedTasks;
1321
1322 const recovered = yield* orchestrator.getThreadProjection(threadId);
1323 const task = (id: NodeId) => recovered.subagents.find((row) => row.id === id);
1324 assert.equal(task(stopped.taskId)?.status, "cancelled");
1325 assert.equal(task(stopped.taskId)?.completionDelivery?.state, "claimed");
1326 assert.equal(task(resumed.taskId)?.status, "running");
1327 assert.isNull(task(resumed.taskId)?.result ?? null);
1328 assert.equal(task(backgrounded.taskId)?.status, "running");
1329 assert.isNull(task(backgrounded.taskId)?.result ?? null);
1330 assert.equal(task(recut.taskId)?.status, "running");
1331 assert.isTrue(yield* orchestrator.delegatedTaskResultPending(recut.childThreadId));
1332 assert.isTrue(yield* orchestrator.delegatedTaskResultPending(resumed.childThreadId));
1333 assert.isFalse(yield* orchestrator.delegatedTaskResultPending(stopped.childThreadId));
1334 // A replayed first continuation settling must not release the second one's hold.
1335 yield* orchestrator.recoverDelegatedTask(recut.childThreadId, recut.childRunId);
1336 const replayed = yield* orchestrator.getThreadProjection(threadId);
1337 assert.equal(replayed.subagents.find((row) => row.id === recut.taskId)?.status, "running");
1338 // A caller that read the cancelled run before the child resumed sees it as pending.
1339 yield* eventSink.write({
1340 commandId: CommandId.make("command:restart-stopped-child:resumed"),
1341 events: [
1342 runEvent({
1343 threadId: stopped.childThreadId,
1344 runId: RunId.make("run:restart-stopped-child:2"),
1345 ordinal: 2,
1346 status: "running",
1347 now,
1348 }),
1349 ],
1350 });
1351 assert.isTrue(yield* orchestrator.delegatedTaskResultPending(stopped.childThreadId));
1352 assert.isFalse(
1353 recovered.contextTransfers.some(
1354 (transfer) => transfer.sourceThreadId === resumed.childThreadId,
1355 ),
1356 );
1357
1358 // The continuation's own run finishing settles the task with its result.
1359 const afterSequence = yield* eventSink.latestSequence();
1360 const continuationRunId = RunId.make("run:restart-resumed-child:2");
1361 yield* eventSink.write({
1362 commandId: CommandId.make("command:restart-resumed-child:completed"),
1363 events: [
1364 {
1365 id: EventId.make("event:restart-resumed-child:result"),
1366 type: "message.updated",
1367 threadId: resumed.childThreadId,
1368 runId: continuationRunId,
1369 occurredAt: now,
1370 payload: {
1371 id: MessageId.make("message:restart-resumed-child:result"),
1372 threadId: resumed.childThreadId,
1373 runId: continuationRunId,
1374 nodeId: null,
1375 role: "assistant",
1376 text: "Finished after the restart.",
1377 attachments: [],
1378 streaming: false,
1379 createdBy: "agent",
1380 creationSource: "server",
1381 createdAt: now,
1382 updatedAt: now,
1383 },
1384 },
1385 runEvent({
1386 threadId: resumed.childThreadId,
1387 runId: continuationRunId,
1388 ordinal: 2,
1389 status: "completed",
1390 now,
1391 }),
1392 ],
1393 });
1394 const settled = yield* eventSink
1395 .stream({ afterSequence, eventType: "subagent.updated" })
1396 .pipe(
1397 Stream.filter(
1398 (stored) =>
1399 stored.event.type === "subagent.updated" &&
1400 stored.event.payload.id === resumed.taskId,
1401 ),
1402 Stream.take(1),
1403 Stream.runHead,
1404 );
1405 assert.isTrue(settled._tag === "Some");
1406 const finished = yield* orchestrator.getThreadProjection(threadId);
1407 const finishedTask = finished.subagents.find((row) => row.id === resumed.taskId);
1408 assert.equal(finishedTask?.status, "completed");
1409 assert.equal(finishedTask?.result, "Finished after the restart.");
1410 }),
1411 );
1412
1413 it.effect("settles a restart-cancelled child whose continuation declines to start", () =>
1414 Effect.gen(function* () {
1415 const orchestrator = yield* Orchestrator.OrchestratorV2;
1416 const now = yield* DateTime.now;
1417 const threadId = ThreadId.make("thread:restart-declined-parent");
1418 const projectId = ProjectId.make("project:restart-declined-parent");
1419 const runId = RunId.make("run:restart-declined-parent");
1420 const rootNodeId = NodeId.make("node:restart-declined-parent-root");
1421 yield* seedParentWithTerminalTask({
1422 threadId,
1423 projectId,
1424 runId,
1425 rootNodeId,
1426 taskId: NodeId.make("node:restart-declined-parent-settled"),
1427 deliveryState: "delivered",
1428 now,
1429 });
1430 const child = yield* seedRestartCancelledChild({
1431 parentThreadId: threadId,
1432 projectId,
1433 parentRunId: runId,
1434 rootNodeId,
1435 name: "restart-declined-child",
1436 completionWake: "always",
1437 continuationPending: true,
1438 now,
1439 });
1440 yield* orchestrator.recoverDelegatedTasks;
1441 const held = yield* orchestrator.getThreadProjection(threadId);
1442 assert.equal(held.subagents.find((row) => row.id === child.taskId)?.status, "running");
1443
1444 yield* continueRestartedRun({
1445 threadId: child.childThreadId,
1446 sourceRunId: child.childRunId,
1447 }).pipe(
1448 Effect.provide(ServerSettings.layerTest({ continueThreadsAfterServerUpdate: false })),
1449 );
1450
1451 const settled = yield* orchestrator.getThreadProjection(threadId);
1452 const task = settled.subagents.find((row) => row.id === child.taskId);
1453 assert.equal(task?.status, "cancelled");
1454 assert.equal(task?.completionDelivery?.state, "claimed");
1455 }),
1456 );
1457
1458 it.effect("reports a child whose monitor the restart cut as interrupted, not completed", () =>
1459 Effect.gen(function* () {
1460 const orchestrator = yield* Orchestrator.OrchestratorV2;
1461 const now = yield* DateTime.now;
1462 const threadId = ThreadId.make("thread:restart-monitor-parent");
1463 const projectId = ProjectId.make("project:restart-monitor-parent");
1464 const runId = RunId.make("run:restart-monitor-parent");
1465 const rootNodeId = NodeId.make("node:restart-monitor-parent-root");
1466 yield* seedParentWithTerminalTask({
1467 threadId,
1468 projectId,
1469 runId,
1470 rootNodeId,
1471 taskId: NodeId.make("node:restart-monitor-parent-settled"),
1472 deliveryState: "delivered",
1473 now,
1474 });
1475 // The child ended its turn to wait on a gate monitor; the restart cut it.
1476 const child = yield* seedRestartCancelledChild({
1477 parentThreadId: threadId,
1478 projectId,
1479 parentRunId: runId,
1480 rootNodeId,
1481 name: "restart-monitor-child",
1482 completionWake: "always",
1483 continuationPending: true,
1484 runStatus: "completed",
1485 lostWork: [{ kind: "monitor", label: "Gate attempt 9" }],
1486 now,
1487 });
1488 yield* orchestrator.recoverDelegatedTasks;
1489 const held = yield* orchestrator.getThreadProjection(threadId);
1490 assert.equal(held.subagents.find((row) => row.id === child.taskId)?.status, "running");
1491
1492 // Continuation is off, so the cut run is the outcome. It still goes to the parent.
1493 yield* continueRestartedRun({
1494 threadId: child.childThreadId,
1495 sourceRunId: child.childRunId,
1496 }).pipe(
1497 Effect.provide(ServerSettings.layerTest({ continueThreadsAfterServerUpdate: false })),
1498 );
1499
1500 const settled = yield* orchestrator.getThreadProjection(threadId);
1501 const task = settled.subagents.find((row) => row.id === child.taskId);
1502 assert.equal(task?.status, "interrupted");
1503 assert.equal(task?.result, "Child task ended with status interrupted.");
1504 assert.equal(task?.completionDelivery?.state, "claimed");
1505 }),
1506 );
1507
1508 // A running provider turn does not prove the provider consumed the delivery
1509 // (Claude marks the turn running before it reads the prompt), so a cut
1510 // delivery is offered again even when a continuation resumes that turn.
1511 it.effect("re-offers a cut delivery even when a continuation resumes its turn", () =>
1512 Effect.gen(function* () {
1513 const orchestrator = yield* Orchestrator.OrchestratorV2;
1514 const eventSink = yield* EventSink.EventSinkV2;
1515 const now = yield* DateTime.now;
1516 const seed = (name: string, continuationPending: boolean) =>
1517 Effect.gen(function* () {
1518 const threadId = ThreadId.make(`thread:${name}`);
1519 const runId = RunId.make(`run:${name}`);
1520 const taskId = NodeId.make(`node:${name}-task`);
1521 const messageId = MessageId.make(`message:delegated-delivery:${threadId}`);
1522 const deliveryRunId = RunId.make(`run:${name}:delivery`);
1523 yield* seedParentWithTerminalTask({
1524 threadId,
1525 projectId: ProjectId.make(`project:${name}`),
1526 runId,
1527 rootNodeId: NodeId.make(`node:${name}-root`),
1528 taskId,
1529 deliveryState: "claimed",
1530 completionWake: "always",
1531 deliveryTaskIds: [taskId],
1532 now,
1533 });
1534 yield* eventSink.writeWithEffects({
1535 commandId: reconcileCommandId(name),
1536 events: [
1537 {
1538 id: EventId.make(`event:${name}:delivery-message`),
1539 type: "message.updated",
1540 threadId,
1541 runId: deliveryRunId,
1542 occurredAt: now,
1543 payload: {
1544 id: messageId,
1545 threadId,
1546 runId: deliveryRunId,
1547 nodeId: null,
1548 role: "user",
1549 text: `Delegated task ${taskId} reached a terminal state.`,
1550 attachments: [],
1551 streaming: false,
1552 createdBy: "agent",
1553 creationSource: "server",
1554 createdAt: now,
1555 updatedAt: now,
1556 delegatedCompletion: { parentRunId: runId, generation: 1, taskIds: [taskId] },
1557 },
1558 },
1559 runEvent({
1560 threadId,
1561 runId: deliveryRunId,
1562 ordinal: 2,
1563 status: "cancelled",
1564 now,
1565 providerThreadId: parentProviderThreadId(threadId),
1566 userMessageId: messageId,
1567 }),
1568 ],
1569 effects: continuationPending
1570 ? [
1571 {
1572 id: `effect:restart-continuation:${deliveryRunId}`,
1573 commandId: reconcileCommandId(name),
1574 threadId,
1575 request: { type: "provider-runtime.continue", sourceRunId: deliveryRunId },
1576 },
1577 ]
1578 : [],
1579 });
1580 return { threadId, runId, taskId };
1581 });
1582 const cut = yield* seed("delivery-cut", false);
1583 const resumed = yield* seed("delivery-resumed", true);
1584
1585 yield* orchestrator.recoverDelegatedTasks;
1586
1587 for (const seeded of [cut, resumed]) {
1588 const projection = yield* orchestrator.getThreadProjection(seeded.threadId);
1589 assert.deepEqual(
1590 projection.subagents.find((row) => row.id === seeded.taskId)?.completionDelivery?.state,
1591 "claimed",
1592 );
1593 const delivery = projection.runs.find((row) => row.id === seeded.runId)?.delegatedCompletion
1594 ?.delivery;
1595 assert.equal(delivery?.generation, 2);
1596 assert.deepEqual(delivery?.taskIds, [seeded.taskId]);
1597 }
1598 }),
1599 );
1600
1601 it.effect("settles only the current delivery of a parent with long delivery history", () =>
1602 Effect.gen(function* () {
1603 const orchestrator = yield* Orchestrator.OrchestratorV2;
1604 const eventSink = yield* EventSink.EventSinkV2;
1605 const now = yield* DateTime.now;
1606 const threadId = ThreadId.make("thread:delivery-history");
1607 const runId = RunId.make("run:delivery-history");
1608 const taskId = NodeId.make("node:delivery-history-task");
1609 const currentMessageId = MessageId.make(`message:delegated-delivery:${threadId}`);
1610 const currentRunId = RunId.make("run:delivery-history:current");
1611 const HISTORY = 40;
1612 yield* seedParentWithTerminalTask({
1613 threadId,
1614 projectId: ProjectId.make("project:delivery-history"),
1615 runId,
1616 rootNodeId: NodeId.make("node:delivery-history-root"),
1617 taskId,
1618 deliveryState: "claimed",
1619 completionWake: "always",
1620 deliveryTaskIds: [taskId],
1621 now,
1622 });
1623 const deliveryMessage = (id: MessageId, deliveryRunId: RunId) => ({
1624 id: EventId.make(`event:${deliveryRunId}:message`),
1625 type: "message.updated" as const,
1626 threadId,
1627 runId: deliveryRunId,
1628 occurredAt: now,
1629 payload: {
1630 id,
1631 threadId,
1632 runId: deliveryRunId,
1633 nodeId: null,
1634 role: "user" as const,
1635 text: `Delegated task ${taskId} reached a terminal state.`,
1636 attachments: [],
1637 streaming: false,
1638 createdBy: "agent" as const,
1639 creationSource: "server" as const,
1640 createdAt: now,
1641 updatedAt: now,
1642 delegatedCompletion: { parentRunId: runId, generation: 1, taskIds: [taskId] },
1643 },
1644 });
1645 // Earlier deliveries that already settled. Each one still owns a
1646 // delegated message and a terminal run.
1647 const history = Array.from({ length: HISTORY }, (_, index) => {
1648 const historyRunId = RunId.make(`run:delivery-history:${index}`);
1649 const messageId = MessageId.make(`message:delivery-history:${index}`);
1650 return [
1651 deliveryMessage(messageId, historyRunId),
1652 runEvent({
1653 threadId,
1654 runId: historyRunId,
1655 ordinal: index + 2,
1656 status: "completed",
1657 now,
1658 providerThreadId: parentProviderThreadId(threadId),
1659 userMessageId: messageId,
1660 }),
1661 ];
1662 }).flat();
1663 yield* eventSink.write({
1664 commandId: reconcileCommandId("delivery-history"),
1665 events: [
1666 ...history,
1667 deliveryMessage(currentMessageId, currentRunId),
1668 runEvent({
1669 threadId,
1670 runId: currentRunId,
1671 ordinal: HISTORY + 2,
1672 status: "completed",
1673 now,
1674 providerThreadId: parentProviderThreadId(threadId),
1675 userMessageId: currentMessageId,
1676 }),
1677 ],
1678 });
1679
1680 const { statements, tracer } = traceSqlStatements();
1681 yield* orchestrator.recoverDelegatedTasks.pipe(Effect.withTracer(tracer));
1682
1683 const projection = yield* orchestrator.getThreadProjection(threadId);
1684 assert.equal(
1685 projection.subagents.find((row) => row.id === taskId)?.completionDelivery?.state,
1686 "delivered",
1687 );
1688 assert.isNull(projection.runs.find((row) => row.id === runId)?.delegatedCompletion?.delivery);
1689 // Recovery reads the delivery message once, for the current delivery
1690 // run only. History must not cost one thread reload per past delivery.
1691 assert.lengthOf(
1692 statements.filter((statement) =>
1693 statement.includes("FROM orchestration_v2_projection_messages AS message"),
1694 ),
1695 1,
1696 );
1697
1698 // A second recovery finds nothing open on this thread and changes nothing.
1699 yield* orchestrator.recoverDelegatedTasks;
1700 const again = yield* orchestrator.getThreadProjection(threadId);
1701 assert.deepEqual(again.runs, projection.runs);
1702 assert.deepEqual(again.subagents, projection.subagents);
1703 }),
1704 );
1705
1706 it.effect("settles a cancelled child whose held wakes wait behind it", () =>
1707 Effect.gen(function* () {
1708 const orchestrator = yield* Orchestrator.OrchestratorV2;
1709 const eventSink = yield* EventSink.EventSinkV2;
1710 const now = yield* DateTime.now;
1711 const threadId = ThreadId.make("thread:held-wake-parent");
1712 const projectId = ProjectId.make("project:held-wake-parent");
1713 const runId = RunId.make("run:held-wake-parent");
1714 const rootNodeId = NodeId.make("node:held-wake-parent-root");
1715 yield* seedParentWithTerminalTask({
1716 threadId,
1717 projectId,
1718 runId,
1719 rootNodeId,
1720 taskId: NodeId.make("node:held-wake-parent-settled"),
1721 deliveryState: "delivered",
1722 now,
1723 });
1724 const child = yield* seedRestartCancelledChild({
1725 parentThreadId: threadId,
1726 projectId,
1727 parentRunId: runId,
1728 rootNodeId,
1729 name: "held-wake-child",
1730 completionWake: "always",
1731 continuationPending: false,
1732 now,
1733 });
1734 // Pull request watch wakes queued while the child ran; the restart held them.
1735 const held = runEvent({
1736 threadId: child.childThreadId,
1737 runId: RunId.make("run:held-wake-child:2"),
1738 ordinal: 2,
1739 status: "queued",
1740 now,
1741 });
1742 yield* eventSink.write({
1743 commandId: CommandId.make("command:held-wake-child:held"),
1744 events: [
1745 {
1746 ...held,
1747 payload: { ...held.payload, startedAt: null, completedAt: null, queueHeld: true },
1748 },
1749 ],
1750 });
1751
1752 yield* orchestrator.recoverDelegatedTasks;
1753
1754 const recovered = yield* orchestrator.getThreadProjection(threadId);
1755 const task = recovered.subagents.find((row) => row.id === child.taskId);
1756 assert.equal(task?.status, "cancelled");
1757 assert.isNotNull(task?.result ?? null);
1758 assert.isTrue(
1759 recovered.contextTransfers.some(
1760 (transfer) =>
1761 transfer.type === "subagent_result" && transfer.sourceThreadId === child.childThreadId,
1762 ),
1763 );
1764 assert.isFalse(yield* orchestrator.delegatedTaskResultPending(child.childThreadId));
1765 // The held wake stays held for the user to resume or discard.
1766 const childProjection = yield* orchestrator.getThreadProjection(child.childThreadId);
1767 assert.deepEqual(
1768 childProjection.runs.map((run) => [run.status, run.queueHeld ?? false]),
1769 [
1770 ["cancelled", false],
1771 ["queued", true],
1772 ],
1773 );
1774 }),
1775 );
1776
1777 it.effect("settles a child whose provider failure holds its queued wakes", () =>
1778 Effect.gen(function* () {
1779 const orchestrator = yield* Orchestrator.OrchestratorV2;
1780 const eventSink = yield* EventSink.EventSinkV2;
1781 const now = yield* DateTime.now;
1782 const threadId = ThreadId.make("thread:failed-hold-parent");
1783 const projectId = ProjectId.make("project:failed-hold-parent");
1784 const runId = RunId.make("run:failed-hold-parent");
1785 const rootNodeId = NodeId.make("node:failed-hold-parent-root");
1786 yield* seedParentWithTerminalTask({
1787 threadId,
1788 projectId,
1789 runId,
1790 rootNodeId,
1791 taskId: NodeId.make("node:failed-hold-parent-settled"),
1792 deliveryState: "delivered",
1793 now,
1794 });
1795 const child = yield* seedRestartCancelledChild({
1796 parentThreadId: threadId,
1797 projectId,
1798 parentRunId: runId,
1799 rootNodeId,
1800 name: "failed-hold-child",
1801 completionWake: "always",
1802 continuationPending: false,
1803 now,
1804 });
1805 // The child is resumed, and a wake queues behind its running turn.
1806 const activeRun = runEvent({
1807 threadId: child.childThreadId,
1808 runId: RunId.make("run:failed-hold-child:2"),
1809 ordinal: 2,
1810 status: "running",
1811 now,
1812 });
1813 const queued = runEvent({
1814 threadId: child.childThreadId,
1815 runId: RunId.make("run:failed-hold-child:3"),
1816 ordinal: 3,
1817 status: "queued",
1818 now,
1819 });
1820 yield* eventSink.write({
1821 commandId: CommandId.make("command:failed-hold-child:resumed"),
1822 events: [
1823 activeRun,
1824 { ...queued, payload: { ...queued.payload, startedAt: null, completedAt: null } },
1825 ],
1826 });
1827 const afterSequence = yield* eventSink.latestSequence();
1828 const errorItemId = TurnItemId.make("turn-item:failed-hold-child:error");
1829 yield* eventSink.write({
1830 commandId: CommandId.make("command:failed-hold-child:failed"),
1831 events: [
1832 {
1833 id: EventId.make("event:failed-hold-child:error"),
1834 type: "turn-item.updated",
1835 threadId: child.childThreadId,
1836 runId: activeRun.runId,
1837 providerInstanceId: modelSelection.instanceId,
1838 occurredAt: now,
1839 payload: {
1840 id: errorItemId,
1841 type: "error",
1842 threadId: child.childThreadId,
1843 runId: activeRun.runId,
1844 nodeId: null,
1845 providerThreadId: null,
1846 providerTurnId: null,
1847 nativeItemRef: null,
1848 parentItemId: null,
1849 ordinal: 1,
1850 status: "failed",
1851 title: "Provider failure",
1852 startedAt: now,
1853 completedAt: now,
1854 updatedAt: now,
1855 failure: {
1856 class: "provider_error",
1857 message: "Provider failed.",
1858 code: "provider_failed",
1859 retryable: null,
1860 },
1861 },
1862 },
1863 {
1864 ...activeRun,
1865 id: EventId.make("event:failed-hold-child:failed"),
1866 payload: { ...activeRun.payload, status: "failed", completedAt: now },
1867 },
1868 ],
1869 });
1870
1871 // The failure holds the queue, which leaves the failure as the task's result.
1872 const settled = yield* eventSink
1873 .stream({ afterSequence, eventType: "subagent.updated" })
1874 .pipe(
1875 Stream.filter(
1876 (stored) =>
1877 stored.event.type === "subagent.updated" && stored.event.payload.id === child.taskId,
1878 ),
1879 Stream.take(1),
1880 Stream.runHead,
1881 );
1882 assert.isTrue(settled._tag === "Some");
1883 const recovered = yield* orchestrator.getThreadProjection(threadId);
1884 assert.equal(recovered.subagents.find((row) => row.id === child.taskId)?.status, "failed");
1885 const childProjection = yield* orchestrator.getThreadProjection(child.childThreadId);
1886 assert.deepEqual(
1887 childProjection.runs.map((run) => [run.status, run.queueHeld ?? false]),
1888 [
1889 ["cancelled", false],
1890 ["failed", false],
1891 ["queued", true],
1892 ],
1893 );
1894 }),
1895 );
1896
1897 it.effect("wakes the parent for a settled_only task once its spawning turn ended", () =>
1898 Effect.gen(function* () {
1899 const orchestrator = yield* Orchestrator.OrchestratorV2;
1900 const eventSink = yield* EventSink.EventSinkV2;
1901 const now = yield* DateTime.now;
1902 const threadId = ThreadId.make("thread:settled-only-restart");
1903 const projectId = ProjectId.make("project:settled-only-restart");
1904 const runId = RunId.make("run:settled-only-restart");
1905 const rootNodeId = NodeId.make("node:settled-only-restart-root");
1906 yield* seedParentWithTerminalTask({
1907 threadId,
1908 projectId,
1909 runId,
1910 rootNodeId,
1911 taskId: NodeId.make("node:settled-only-restart-settled"),
1912 deliveryState: "delivered",
1913 now,
1914 });
1915 // The restart cut the turn blocked in delegate_task(wait); its
1916 // continuation is a new, live run.
1917 yield* eventSink.write({
1918 commandId: reconcileCommandId("settled-only-restart"),
1919 events: [
1920 runEvent({
1921 threadId,
1922 runId,
1923 ordinal: 1,
1924 status: "cancelled",
1925 now,
1926 providerThreadId: parentProviderThreadId(threadId),
1927 userMessageId: MessageId.make(`message:seed-user:${threadId}`),
1928 delegatedCompletion: { disposition: "open", nextGeneration: 2, delivery: null },
1929 }),
1930 runEvent({
1931 threadId,
1932 runId: RunId.make("run:settled-only-restart:continuation"),
1933 ordinal: 2,
1934 status: "running",
1935 now,
1936 providerThreadId: parentProviderThreadId(threadId),
1937 }),
1938 ],
1939 });
1940 const child = yield* seedRestartCancelledChild({
1941 parentThreadId: threadId,
1942 projectId,
1943 parentRunId: runId,
1944 rootNodeId,
1945 name: "settled-only-restart-child",
1946 completionWake: "settled_only",
1947 continuationPending: false,
1948 now,
1949 });
1950
1951 yield* orchestrator.recoverDelegatedTasks;
1952
1953 const projection = yield* orchestrator.getThreadProjection(threadId);
1954 assert.equal(
1955 projection.subagents.find((row) => row.id === child.taskId)?.completionDelivery?.state,
1956 "claimed",
1957 );
1958 assert.deepEqual(
1959 projection.runs.find((row) => row.id === runId)?.delegatedCompletion?.delivery?.taskIds,
1960 [child.taskId],
1961 );
1962 }),
1963 );
1964});