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