1import * as NodeURL from "node:url";
2import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
3import * as NodeServices from "@effect/platform-node/NodeServices";
4import { describe, expect, it } from "@effect/vitest";
5import {
6 CommandId,
7 EnvironmentId,
8 isProviderNativeSubagentThread,
9 MessageId,
10 type ModelSelection,
11 OrchestratorMcpDelegateTaskResult,
12 OrchestratorMcpThreadInterruptResult,
13 OrchestratorMcpThreadSendResult,
14 OrchestratorMcpThreadReadResult,
15 ProjectId,
16 ProviderDriverKind,
17 ProviderInstanceId,
18 type ProviderOptionDescriptor,
19 type ServerProvider,
20 ThreadId,
21} from "@t3tools/contracts";
22import * as Effect from "effect/Effect";
23import * as FileSystem from "effect/FileSystem";
24import * as Layer from "effect/Layer";
25import * as Option from "effect/Option";
26import * as Schema from "effect/Schema";
27import * as Stream from "effect/Stream";
28import * as Tracer from "effect/Tracer";
29import { McpSchema, McpServer } from "effect/ai";
30
31import { CodexOrchestratorReplayHarness } from "../orchestration-v2/Adapters/CodexAdapterV2.testkit.ts";
32import * as CommandReceiptStore from "../orchestration-v2/CommandReceiptStore.ts";
33import * as Orchestrator from "../orchestration-v2/Orchestrator.ts";
34import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts";
35import * as GitVcsDriver from "../vcs/GitVcsDriver.ts";
36import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts";
37import * as ServerSettings from "../serverSettings.ts";
38import * as SqlitePersistence from "../persistence/Sqlite.ts";
39import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions";
40import * as SecretResolver from "@t3tools/secret-provider-core/server/SecretResolver";
41import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace";
42import * as ProviderReplayHarness from "../orchestration-v2/testkit/ProviderReplayHarness.ts";
43import {
44 decodeProviderReplayNdjson,
45 materializeReplayTranscriptWorkspace,
46} from "@t3tools/provider-testing/replayTranscript";
47import * as ProviderRegistryMock from "../provider/testUtils/providerRegistryMock.ts";
48import * as ProjectService from "../project/ProjectService.ts";
49import * as ScheduledTaskService from "../scheduledTasks/ScheduledTaskService.ts";
50import * as SecretRequests from "../secrets/SecretRequests.ts";
51import * as RemoteDelegation from "../peer/RemoteDelegation.ts";
52import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
53import * as McpHttpServer from "./McpHttpServer.ts";
54import * as McpInvocationContext from "./McpInvocationContext.ts";
55import * as PeerForwarding from "../peer/PeerForwarding.ts";
56import * as PeerLinkRequests from "../peer/PeerLinkRequests.ts";
57import * as PeerLinks from "../peer/PeerLinks.ts";
58
59const layerOrchestratorToolkit = McpHttpServer.layerOrchestratorToolkit.pipe(
60 Layer.provide(Layer.mock(ThreadLaunchService.ThreadLaunchService)({})),
61 Layer.provide(Layer.mock(GitVcsDriver.GitVcsDriver)({})),
62);
63
64/** A task delegated here always has a local child thread. */
65const localChild = (childThreadId: ThreadId | null): ThreadId => {
66 if (childThreadId === null) throw new Error("expected a local delegated child");
67 return childThreadId;
68};
69
70// Effect returns a declared tool failure as `isError` with its encoded payload
71// as JSON text, never as `structuredContent`.
72
73const parentThreadId = ThreadId.make("thread:mcp-orchestrator-parent");
74const projectId = ProjectId.make("project:mcp-orchestrator");
75const codexInstanceId = ProviderInstanceId.make("codex");
76const codexModel = "gpt-5.4";
77const parentPrompt = "Keep this parent turn active while orchestration tools are tested.";
78const delegatedPrompt = "Inspect the delegated API boundary and return the result.";
79const delegatedResult = "Delegated API boundary inspected.";
80const cancellationPrompt = "Remain active until the parent cancels this delegated task.";
81const queuedFollowupPrompt = "Complete the queued follow-up and return the final result.";
82const queuedFollowupResult = "Queued delegated follow-up completed.";
83const decodeThreadRead = Schema.decodeUnknownEffect(OrchestratorMcpThreadReadResult);
84const longFollowupResult = "Long delegated follow-up line. ".repeat(1_500);
85
86const decodeDelegateTaskResult = Schema.decodeUnknownEffect(OrchestratorMcpDelegateTaskResult);
87const decodeThreadInterruptResult = Schema.decodeUnknownEffect(
88 OrchestratorMcpThreadInterruptResult,
89);
90const decodeThreadSendResult = Schema.decodeUnknownEffect(OrchestratorMcpThreadSendResult);
91
92const codexSelection = {
93 instanceId: codexInstanceId,
94 model: codexModel,
95} satisfies ModelSelection;
96
97const delegatedTaskStatusTranscriptFile = new URL(
98 "../orchestration-v2/testkit/fixtures/delegated_task_status/codex_transcript.ndjson",
99 import.meta.url,
100);
101
102const readDelegatedTaskStatusTranscript = Effect.fn("readDelegatedTaskStatusTranscript")(
103 function* (followupResult: string) {
104 const fs = yield* FileSystem.FileSystem;
105 const text = yield* fs.readFileString(NodeURL.fileURLToPath(delegatedTaskStatusTranscriptFile));
106 const transcript = yield* decodeProviderReplayNdjson(
107 text.replaceAll(queuedFollowupResult, followupResult),
108 );
109 // The recording ends when the recorder closes the app-server while the parent
110 // turn is still open. This test keeps acting as that parent afterwards, and
111 // replaying the exit would kill the shared process and fail the parent run.
112 return {
113 ...transcript,
114 entries: transcript.entries.filter((entry) => entry.type !== "runtime_exit"),
115 };
116 },
117 Effect.provide(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)),
118);
119
120function makeProviderSnapshot(input: {
121 readonly instanceId: ProviderInstanceId;
122 readonly driver: ProviderDriverKind;
123 readonly model: string;
124 readonly optionDescriptors?: ReadonlyArray<ProviderOptionDescriptor>;
125}): ServerProvider {
126 return {
127 instanceId: input.instanceId,
128 driver: input.driver,
129 enabled: true,
130 installed: true,
131 version: "test",
132 status: "ready",
133 auth: { status: "authenticated" },
134 checkedAt: "2026-06-17T00:00:00.000Z",
135 models: [
136 {
137 slug: input.model,
138 name: input.model,
139 isCustom: false,
140 capabilities:
141 input.optionDescriptors === undefined
142 ? null
143 : { optionDescriptors: input.optionDescriptors },
144 },
145 ],
146 slashCommands: [],
147 skills: [],
148 };
149}
150
151const client = McpSchema.McpServerClient.of({
152 clientId: 1,
153 protocolVersion: "2025-06-18",
154 clientCapabilities: {},
155 clientInfo: { name: "orchestrator-mcp-test", version: "1.0.0" },
156 initializePayload: {
157 protocolVersion: "2025-06-18",
158 capabilities: {},
159 clientInfo: { name: "orchestrator-mcp-test", version: "1.0.0" },
160 },
161 getClient: Effect.die("unused"),
162});
163
164const layerMemorySecretStore = Layer.sync(ServerSecretStore.ServerSecretStore, () => {
165 const stored = new Map<string, Uint8Array>();
166 return ServerSecretStore.ServerSecretStore.of({
167 get: (name) => Effect.succeed(Option.fromNullishOr(stored.get(name))),
168 set: (name, value) => Effect.sync(() => void stored.set(name, value)),
169 create: (name, value) => Effect.sync(() => void stored.set(name, value)),
170 getOrCreateRandom: (name, bytes) =>
171 Effect.sync(() => {
172 const existing = stored.get(name);
173 if (existing) return existing;
174 const value = new Uint8Array(bytes).fill(7);
175 stored.set(name, value);
176 return value;
177 }),
178 remove: (name) => Effect.sync(() => void stored.delete(name)),
179 });
180});
181
182const layerUnusedScheduledTaskStub = Layer.succeed(
183 ScheduledTaskService.ScheduledTaskService,
184 ScheduledTaskService.ScheduledTaskService.of({
185 list: () => Effect.succeed({ tasks: [] }),
186 subscribeList: () => Stream.succeed({ tasks: [] }),
187 upsert: () => Effect.die("ScheduledTaskService.upsert is unused in this test"),
188 setEnabled: () => Effect.die("ScheduledTaskService.setEnabled is unused in this test"),
189 delete: () => Effect.die("ScheduledTaskService.delete is unused in this test"),
190 deleteForThread: () => Effect.die("unused in this test"),
191 moveThreadTasks: () => Effect.die("unused in this test"),
192 runNow: () => Effect.die("ScheduledTaskService.runNow is unused in this test"),
193 rotateWebhookToken: () => Effect.die("unused in this test"),
194 listWebhookDeliveries: () => Effect.die("unused in this test"),
195 getWebhookDelivery: () => Effect.die("unused in this test"),
196 triggerWebhook: () => Effect.die("unused in this test"),
197 }),
198);
199
200describe("orchestrator MCP toolkit", () => {
201 it.live.each([
202 { label: "distinct", result: queuedFollowupResult, latestSummary: queuedFollowupResult },
203 { label: "identical", result: delegatedResult, latestSummary: null },
204 { label: "emoji", result: "😀", latestSummary: "😀" },
205 { label: "long", result: longFollowupResult, latestSummary: longFollowupResult },
206 ])("reports $label child results", ({ result, latestSummary }) =>
207 Effect.gen(function* () {
208 const cwd = yield* checkpointWorkspace("delegated-task-status");
209 const rawTranscript = yield* readDelegatedTaskStatusTranscript(result);
210 const transcript = yield* CodexOrchestratorReplayHarness.decodeTranscript(
211 materializeReplayTranscriptWorkspace(rawTranscript, cwd),
212 );
213 const layerOrchestrator = ProviderReplayHarness.layerProviderReplay(
214 {
215 name: "delegated-task-status/codex",
216 transcript,
217 commands: [],
218 runtimePolicyOverride: { cwd },
219 },
220 CodexOrchestratorReplayHarness,
221 );
222 // SQL spans carry no bound parameters, so this records which thread each tool read.
223 const threadRecordReads: Array<{
224 readonly threadId: ThreadId;
225 readonly fields: ReadonlyArray<string>;
226 }> = [];
227 const layerOrchestration = Layer.merge(
228 layerOrchestrator,
229 Layer.effect(
230 ThreadManagementService.ThreadManagementService,
231 Effect.gen(function* () {
232 const service = yield* ThreadManagementService.ThreadManagementService;
233 return ThreadManagementService.ThreadManagementService.of({
234 ...service,
235 getThreadRecords: (threadId, fields, filter) =>
236 Effect.sync(() => threadRecordReads.push({ threadId, fields })).pipe(
237 Effect.andThen(service.getThreadRecords(threadId, fields, filter)),
238 ),
239 });
240 }),
241 ).pipe(
242 Layer.provide(
243 ThreadManagementService.layer.pipe(
244 Layer.provide(layerOrchestrator),
245 Layer.provide(ServerSettings.layerTest()),
246 Layer.provide(
247 CommandReceiptStore.layer.pipe(Layer.provide(SqlitePersistence.layerMemory)),
248 ),
249 ),
250 ),
251 ),
252 );
253 const layerProviderRegistry = ProviderRegistryMock.layer([
254 makeProviderSnapshot({
255 instanceId: codexInstanceId,
256 driver: ProviderDriverKind.make("codex"),
257 model: codexModel,
258 }),
259 ]);
260 const layerTest = layerOrchestratorToolkit.pipe(
261 Layer.provideMerge(McpHttpServer.layerInProcessMcpServer),
262 Layer.provideMerge(layerOrchestration),
263 Layer.provide(
264 CodexOrchestratorReplayHarness.makeProviderAdapterRegistryLayer(transcript).pipe(
265 Layer.provide(McpProviderSessions.layer),
266 ),
267 ),
268 Layer.provide(layerProviderRegistry),
269 Layer.provide(ServerSettings.layerTest()),
270 Layer.provide(layerUnusedScheduledTaskStub),
271 Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})),
272 Layer.provide(Layer.mock(PeerLinkRequests.PeerLinkRequests)({})),
273 Layer.provide(Layer.mock(PeerLinks.PeerLinks)({})),
274 Layer.provide(Layer.mock(ProjectService.ProjectService)({})),
275 Layer.provideMerge(
276 SecretRequests.layer.pipe(
277 Layer.provide(layerMemorySecretStore),
278 Layer.provide(Layer.mock(SecretResolver.SecretResolver)({})),
279 Layer.provide(layerOrchestration),
280 ),
281 ),
282 Layer.provide(Layer.mock(RemoteDelegation.RemoteDelegation)({})),
283 Layer.provide(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)),
284 );
285
286 yield* Effect.gen(function* () {
287 const orchestrator = yield* Orchestrator.OrchestratorV2;
288 const server = yield* McpServer.McpServer;
289 const parentCreate = yield* orchestrator.dispatch({
290 type: "thread.create",
291 createdBy: "user",
292 creationSource: "web",
293 commandId: CommandId.make("command:mcp-replay-parent:create"),
294 threadId: parentThreadId,
295 projectId,
296 title: "MCP replay parent",
297 modelSelection: codexSelection,
298 runtimeMode: "full-access",
299 interactionMode: "default",
300 branch: null,
301 worktreePath: cwd,
302 });
303 yield* orchestrator.dispatch({
304 type: "message.dispatch",
305 createdBy: "user",
306 creationSource: "web",
307 commandId: CommandId.make("command:mcp-replay-parent:start"),
308 threadId: parentThreadId,
309 messageId: MessageId.make("message:mcp-replay-parent:start"),
310 text: parentPrompt,
311 attachments: [],
312 modelSelection: codexSelection,
313 dispatchMode: { type: "start_immediately" },
314 });
315 yield* orchestrator
316 .streamStoredEventsFrom({
317 threadId: parentThreadId,
318 afterSequence: parentCreate.sequence,
319 })
320 .pipe(
321 Stream.filter(
322 (stored) =>
323 stored.event.type === "provider-turn.updated" &&
324 stored.event.payload.status === "running",
325 ),
326 Stream.runHead,
327 Effect.flatMap(
328 Option.match({
329 onNone: () => Effect.die("Parent provider turn did not start."),
330 onSome: () => Effect.void,
331 }),
332 ),
333 );
334
335 const invocation: McpInvocationContext.McpInvocationScope = {
336 environmentId: EnvironmentId.make("environment:mcp-replay"),
337 requestNamespace: "mcp-provider-session-replay-parent",
338 thread: {
339 threadId: parentThreadId,
340 providerSessionId: "mcp-provider-session-replay-parent",
341 providerInstanceId: codexInstanceId,
342 },
343 client: undefined,
344 capabilities: new Set(["orchestration"]),
345 issuedAt: 1,
346 };
347 const invoke = (name: string, args: Record<string, unknown>, scope = invocation) =>
348 server
349 .callTool({ name, arguments: args })
350 .pipe(
351 Effect.provideService(McpInvocationContext.McpInvocationContext, scope),
352 Effect.provideService(McpSchema.McpServerClient, client),
353 );
354
355 const delegationStartSequence = yield* orchestrator.getThreadEventSequence(parentThreadId);
356 const delegatedCall = yield* invoke("delegate_task", {
357 task: delegatedPrompt,
358 target: {
359 providerInstanceId: codexInstanceId,
360 model: codexModel,
361 },
362 mode: "async",
363 clientRequestId: "delegate-codex-replay-1",
364 });
365 const delegatedStart = yield* decodeDelegateTaskResult(
366 delegatedCall.structuredContent,
367 ).pipe(Effect.orDie);
368 expect(delegatedStart.childRunId).not.toBeNull();
369 yield* orchestrator
370 .streamStoredEventsFrom({
371 threadId: parentThreadId,
372 afterSequence: delegationStartSequence,
373 })
374 .pipe(
375 Stream.filter(
376 (stored) =>
377 stored.event.type === "context-transfer.created" &&
378 stored.event.payload.type === "subagent_result" &&
379 stored.event.payload.sourceThreadId === delegatedStart.childThreadId &&
380 stored.event.payload.sourcePoint.runId === delegatedStart.childRunId,
381 ),
382 Stream.runHead,
383 Effect.flatMap(
384 Option.match({
385 onNone: () => Effect.die("Delegated result transfer was not created."),
386 onSome: () => Effect.void,
387 }),
388 ),
389 );
390 const delegatedStatusCall = yield* invoke("task_status", {
391 taskId: delegatedStart.taskId,
392 });
393 const delegated = yield* decodeDelegateTaskResult(
394 delegatedStatusCall.structuredContent,
395 ).pipe(Effect.orDie);
396 expect(delegated).toMatchObject({
397 status: "completed",
398 summary: delegatedResult,
399 providerInstanceId: codexInstanceId,
400 hasPendingChildRuns: false,
401 latestTerminalRunId: delegated.childRunId,
402 latestTerminalStatus: "completed",
403 latestTerminalSummary: null,
404 latestTerminalResultContextTransferId: null,
405 });
406 expect(delegated.resultContextTransferId).not.toBeNull();
407
408 const delegatedChild = yield* orchestrator.getThreadProjection(
409 localChild(delegated.childThreadId),
410 );
411 expect(delegatedChild.thread.lineage.relationshipToParent).toBe("subagent");
412 expect(isProviderNativeSubagentThread(delegatedChild.thread)).toBe(false);
413 const followupStartSequence = yield* orchestrator.getThreadEventSequence(
414 localChild(delegated.childThreadId),
415 );
416 const runningFollowupCall = yield* invoke("t3_thread_send", {
417 threadId: localChild(delegated.childThreadId),
418 message: cancellationPrompt,
419 clientRequestId: "delegated-child-replay-running-1",
420 });
421 const runningFollowup = yield* decodeThreadSendResult(
422 runningFollowupCall.structuredContent,
423 ).pipe(Effect.orDie);
424 yield* orchestrator
425 .streamStoredEventsFrom({
426 threadId: localChild(delegated.childThreadId),
427 afterSequence: followupStartSequence,
428 })
429 .pipe(
430 Stream.filter(
431 (stored) =>
432 stored.event.type === "provider-turn.updated" &&
433 stored.event.payload.status === "running",
434 ),
435 Stream.runHead,
436 Effect.flatMap(
437 Option.match({
438 onNone: () => Effect.die("Follow-up provider turn did not start."),
439 onSome: () => Effect.void,
440 }),
441 ),
442 );
443
444 const queuedFollowupCall = yield* invoke("t3_thread_send", {
445 threadId: localChild(delegated.childThreadId),
446 message: queuedFollowupPrompt,
447 mode: "queue",
448 clientRequestId: "delegated-child-replay-queued-1",
449 });
450 const queuedFollowup = yield* decodeThreadSendResult(
451 queuedFollowupCall.structuredContent,
452 ).pipe(Effect.orDie);
453 expect(queuedFollowup).toMatchObject({
454 status: "queued",
455 delivery: "queued",
456 });
457
458 const pendingProjection = yield* orchestrator.getThreadProjection(
459 localChild(delegated.childThreadId),
460 );
461 const followupTasks = (yield* orchestrator.getThreadProjection(
462 parentThreadId,
463 )).subagents.filter(
464 (task) => task.childThreadId === delegated.childThreadId && task.id !== delegated.taskId,
465 );
466 expect(followupTasks).toHaveLength(2);
467 expect(followupTasks.every((task) => task.result === null)).toBe(true);
468 expect(followupTasks.map((task) => task.id)).toEqual([
469 runningFollowup.taskId,
470 queuedFollowup.taskId,
471 ]);
472 expect(pendingProjection.runs.find((run) => run.id === delegated.childRunId)?.status).toBe(
473 "completed",
474 );
475 expect(pendingProjection.runs.find((run) => run.id === runningFollowup.runId)?.status).toBe(
476 "running",
477 );
478 expect(pendingProjection.runs.find((run) => run.id === queuedFollowup.runId)?.status).toBe(
479 "queued",
480 );
481 expect(
482 (yield* orchestrator.getThreadProjection(parentThreadId)).runs.some(
483 (run) => run.status === "running",
484 ),
485 ).toBe(true);
486
487 const pendingStatusCall = yield* invoke("task_status", {
488 taskId: delegated.taskId,
489 });
490 const pendingStatus = yield* decodeDelegateTaskResult(
491 pendingStatusCall.structuredContent,
492 ).pipe(Effect.orDie);
493 expect(pendingStatus).toMatchObject({
494 childRunId: delegated.childRunId,
495 status: "completed",
496 summary: delegatedResult,
497 resultContextTransferId: delegated.resultContextTransferId,
498 hasPendingChildRuns: true,
499 latestTerminalRunId: delegated.childRunId,
500 latestTerminalStatus: "completed",
501 latestTerminalSummary: null,
502 latestTerminalResultContextTransferId: null,
503 });
504
505 const finalSequence = yield* orchestrator.getThreadEventSequence(
506 localChild(delegated.childThreadId),
507 );
508 const interruptCall = yield* invoke("t3_thread_interrupt", {
509 threadId: localChild(delegated.childThreadId),
510 runId: runningFollowup.runId,
511 reason: "Allow the queued replay follow-up to run.",
512 clientRequestId: "interrupt-delegated-child-replay-1",
513 });
514 const interrupt = yield* decodeThreadInterruptResult(interruptCall.structuredContent).pipe(
515 Effect.orDie,
516 );
517 expect(interrupt).toMatchObject({
518 runId: runningFollowup.runId,
519 status: "interrupt_requested",
520 });
521 yield* orchestrator
522 .streamStoredEventsFrom({
523 threadId: localChild(delegated.childThreadId),
524 afterSequence: finalSequence,
525 })
526 .pipe(
527 Stream.filter(
528 (stored) =>
529 stored.event.type === "run.updated" &&
530 stored.event.payload.id === queuedFollowup.runId &&
531 stored.event.payload.status === "completed",
532 ),
533 Stream.runHead,
534 Effect.flatMap(
535 Option.match({
536 onNone: () => Effect.die("Queued follow-up did not complete."),
537 onSome: () => Effect.void,
538 }),
539 ),
540 );
541
542 const finalProjection = yield* orchestrator.getThreadProjection(
543 localChild(delegated.childThreadId),
544 );
545 expect(finalProjection.runs.find((run) => run.id === runningFollowup.runId)?.status).toBe(
546 "interrupted",
547 );
548 expect(finalProjection.runs.find((run) => run.id === queuedFollowup.runId)?.status).toBe(
549 "completed",
550 );
551 expect(
552 finalProjection.turnItems.find(
553 (item) => item.runId === runningFollowup.runId && item.type === "run_interrupt_result",
554 ),
555 ).toMatchObject({
556 message: "Stopped by an agent: Allow the queued replay follow-up to run.",
557 });
558 yield* orchestrator
559 .streamStoredEventsFrom({
560 threadId: parentThreadId,
561 afterSequence: finalSequence,
562 })
563 .pipe(
564 Stream.filter(
565 (stored) =>
566 stored.event.type === "subagent.updated" &&
567 stored.event.payload.id === followupTasks[1]!.id &&
568 stored.event.payload.status === "completed",
569 ),
570 Stream.runHead,
571 Effect.flatMap(
572 Option.match({
573 onNone: () => Effect.die("Follow-up task did not settle."),
574 onSome: () => Effect.void,
575 }),
576 ),
577 );
578 const finishedParent = yield* orchestrator.getThreadProjection(parentThreadId);
579 expect(
580 finishedParent.subagents.find((task) => task.id === followupTasks[0]!.id)?.status,
581 ).toBe("interrupted");
582 expect(
583 finishedParent.subagents.find((task) => task.id === followupTasks[1]!.id),
584 ).toMatchObject({
585 status: "completed",
586 result,
587 });
588 expect(
589 finishedParent.contextTransfers.filter(
590 (transfer) =>
591 transfer.type === "subagent_result" &&
592 transfer.sourceThreadId === delegated.childThreadId,
593 ),
594 ).toHaveLength(3);
595 expect(
596 finishedParent.subagents.find((task) => task.id === followupTasks[1]!.id)
597 ?.completionDelivery?.state,
598 ).toBe("claimed");
599 const finalStatusCall = yield* invoke("task_status", {
600 taskId: delegated.taskId,
601 });
602 const finalStatus = yield* decodeDelegateTaskResult(finalStatusCall.structuredContent).pipe(
603 Effect.orDie,
604 );
605 expect(finalStatus).toMatchObject({
606 childRunId: delegated.childRunId,
607 status: "completed",
608 summary: delegatedResult,
609 resultContextTransferId: delegated.resultContextTransferId,
610 hasPendingChildRuns: false,
611 latestTerminalRunId: queuedFollowup.runId,
612 latestTerminalStatus: "completed",
613 latestTerminalSummary: latestSummary,
614 });
615 const followupTransfer = finishedParent.contextTransfers.find(
616 (transfer) =>
617 transfer.type === "subagent_result" &&
618 transfer.sourcePoint.runId === queuedFollowup.runId,
619 );
620 expect(followupTransfer).toBeDefined();
621 expect(finalStatus.latestTerminalResultContextTransferId).toBe(followupTransfer?.id);
622 expect(finalStatus.latestTerminalResultContextTransferId).not.toBe(
623 delegated.resultContextTransferId,
624 );
625 const transferReads: Array<string> = [];
626 const messageReads: Array<string> = [];
627 const runReads: Array<string> = [];
628 const projectionTableReads: Array<string> = [];
629 const acknowledgementReads: Array<string> = [];
630 const tracer = Tracer.make({
631 span(options) {
632 const span = new Tracer.NativeSpan(options);
633 const end = span.end.bind(span);
634 span.end = (endTime, exit) => {
635 end(endTime, exit);
636 const query = span.attributes.get("db.query.text");
637 if (typeof query !== "string") return;
638 let parent = Option.getOrUndefined(span.parent);
639 while (parent?._tag === "Span") {
640 if (
641 parent.name === "Orchestrator.dispatchDelegatedTaskCompletionDeliveryResolution"
642 ) {
643 acknowledgementReads.push(query);
644 break;
645 }
646 parent = Option.getOrUndefined(parent.parent);
647 }
648 for (const [, table] of query.matchAll(/FROM orchestration_v2_projection_(\w+)/gi)) {
649 projectionTableReads.push(table!);
650 }
651 if (/SELECT[\s\S]*FROM orchestration_v2_projection_context_transfers/i.test(query)) {
652 transferReads.push(query);
653 }
654 if (/SELECT\s+payload_json\s+FROM orchestration_v2_projection_runs/i.test(query)) {
655 runReads.push(query);
656 }
657 if (
658 /SELECT\s+payload_json\s+FROM orchestration_v2_projection_messages/i.test(query)
659 ) {
660 messageReads.push(query);
661 }
662 };
663 return span;
664 },
665 });
666 const wait = yield* invoke(
667 "t3_thread_wait",
668 { threadId: delegated.childThreadId, runId: queuedFollowup.runId },
669 { ...invocation, thread: undefined },
670 ).pipe(Effect.withTracer(tracer));
671 expect(wait.isError).not.toBe(true);
672 const expectedWait = {
673 threadId: delegated.childThreadId,
674 runId: queuedFollowup.runId,
675 status: "completed",
676 timedOut: false,
677 };
678 expect(wait.structuredContent).toEqual(expectedWait);
679 expect(messageReads).toEqual([]);
680 expect(transferReads).toHaveLength(0);
681 const settledWait = yield* invoke(
682 "t3_thread_wait",
683 {
684 threadId: delegated.childThreadId,
685 runId: queuedFollowup.runId,
686 untilDelegatedWorkSettled: true,
687 },
688 { ...invocation, thread: undefined },
689 ).pipe(Effect.withTracer(tracer));
690 expect(settledWait.isError).not.toBe(true);
691 expect(settledWait.structuredContent).toMatchObject({
692 ...expectedWait,
693 resultRun: { runId: queuedFollowup.runId, status: "completed" },
694 });
695 expect(messageReads.length).toBeGreaterThan(0);
696 expect(
697 messageReads.filter((query) => query.includes("$.notification.source.kind")),
698 ).toHaveLength(1);
699 // The orchestrator's restart-continuation guard still reads its own records.
700 expect(
701 messageReads.filter((query) => !query.includes("$.notification.source.kind")),
702 ).toHaveLength(1);
703 const parentlessRead = yield* invoke(
704 "t3_thread_read",
705 {
706 threadId: delegated.childThreadId,
707 limit: 100,
708 maxCharsPerItem: result === "😀" ? 1 : 50_000,
709 },
710 { ...invocation, thread: undefined },
711 ).pipe(Effect.withTracer(tracer));
712 expect(parentlessRead.isError).not.toBe(true);
713 expect(transferReads).toHaveLength(0);
714 const afterParentlessRead = yield* orchestrator.getThreadProjection(parentThreadId);
715 expect(
716 afterParentlessRead.subagents.find((task) => task.id === queuedFollowup.taskId)
717 ?.completionDelivery?.state,
718 ).toBe("claimed");
719
720 const read = yield* invoke("t3_thread_read", {
721 threadId: delegated.childThreadId,
722 limit: 100,
723 ...(result === "😀" ? { maxCharsPerItem: 1 } : {}),
724 }).pipe(Effect.withTracer(tracer));
725 expect(read.isError).not.toBe(true);
726 expect(transferReads.length).toBeGreaterThan(0);
727 const firstPage = yield* decodeThreadRead(read.structuredContent);
728 const finalItem = firstPage.items.find((item) => item.text === result.slice(0, 500));
729 expect(finalItem).toBeDefined();
730 if (result.length > 500) {
731 yield* invoke("t3_thread_read", {
732 threadId: delegated.childThreadId,
733 itemId: finalItem!.itemId,
734 textOffset: result.length,
735 });
736 } else {
737 expect(read.structuredContent).toEqual(parentlessRead.structuredContent);
738 }
739 let text = finalItem!.text ?? "";
740 let nextTextOffset = finalItem?.nextTextOffset ?? null;
741 while (nextTextOffset !== null) {
742 const beforeContinuation = yield* orchestrator.getThreadProjection(parentThreadId);
743 expect(
744 beforeContinuation.subagents.find((task) => task.id === queuedFollowup.taskId)
745 ?.completionDelivery?.state,
746 ).toBe("claimed");
747 const continuation = yield* invoke("t3_thread_read", {
748 threadId: delegated.childThreadId,
749 itemId: finalItem!.itemId,
750 textOffset: nextTextOffset,
751 maxCharsPerItem: 50_000,
752 }).pipe(Effect.withTracer(tracer));
753 const nextPage = yield* decodeThreadRead(continuation.structuredContent);
754 text += nextPage.items[0]?.text ?? "";
755 nextTextOffset = nextPage.items[0]?.nextTextOffset ?? null;
756 }
757 expect(text).toBe(result);
758 // The acknowledgement reads the task and its spawning run by id, not the parent's history.
759 expect(acknowledgementReads.length).toBeGreaterThan(0);
760 expect(
761 acknowledgementReads.filter(
762 (query) =>
763 /FROM orchestration_v2_projection_(nodes|run_attempts)\b/.test(query) ||
764 (/FROM orchestration_v2_projection_runs\b/.test(query) &&
765 !query.includes("AND run_id IN")) ||
766 (/FROM orchestration_v2_projection_subagents\b/.test(query) &&
767 !query.includes("AND subagent_id IN")),
768 ),
769 ).toEqual([]);
770 const afterRead = yield* orchestrator.getThreadProjection(parentThreadId);
771 expect(
772 afterRead.subagents.find((task) => task.id === queuedFollowup.taskId)?.completionDelivery
773 ?.state,
774 ).toBe("acknowledged");
775
776 const messageReadsBeforeCapabilities = messageReads.length;
777 const tableReadsBeforeCapabilities = projectionTableReads.length;
778 const capabilities = yield* invoke("orchestrator_capabilities", {}).pipe(
779 Effect.withTracer(tracer),
780 );
781 expect(capabilities.isError).not.toBe(true);
782 expect(messageReads).toHaveLength(messageReadsBeforeCapabilities);
783 // The caller check reads only the thread row, never its history.
784 expect(new Set(projectionTableReads.slice(tableReadsBeforeCapabilities))).toEqual(
785 new Set(["threads"]),
786 );
787 const runReadsBeforeStatus = runReads.length;
788 const transferReadsBeforeStatus = transferReads.length;
789 const threadRecordReadsBeforeStatus = threadRecordReads.length;
790 const statusAgain = yield* invoke("task_status", { taskId: delegated.taskId }).pipe(
791 Effect.withTracer(tracer),
792 );
793 expect(statusAgain.isError).not.toBe(true);
794 // A task wait reads result transfers and runs from the child; the parent's stay unread.
795 expect(runReads).toHaveLength(runReadsBeforeStatus + 1);
796 expect(transferReads).toHaveLength(transferReadsBeforeStatus + 1);
797 expect(
798 threadRecordReads
799 .slice(threadRecordReadsBeforeStatus)
800 .filter(({ fields }) => fields.includes("runs") || fields.includes("contextTransfers"))
801 .map(({ threadId }) => threadId),
802 ).toEqual([delegated.childThreadId]);
803 }).pipe(Effect.provide(layerTest));
804 }),
805 );
806});