1import {
2 EnvironmentId,
3 NodeId,
4 type OrchestrationV2ThreadProjection,
5 type OrchestrationV2ThreadShell,
6 ProjectId,
7 ProviderDriverKind,
8 ProviderInstanceId,
9 RunId,
10 ScheduledTaskId,
11 ThreadId,
12} from "@t3tools/contracts";
13import * as DateTime from "effect/DateTime";
14import * as Effect from "effect/Effect";
15import * as Layer from "effect/Layer";
16import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
17import { expect, it } from "vite-plus/test";
18
19import * as ProviderAdapterRegistry from "../orchestration-v2/ProviderAdapterRegistry.ts";
20import * as ProviderRegistry from "../provider/ProviderRegistry.ts";
21import * as ProjectService from "../project/ProjectService.ts";
22import * as ScheduledTaskService from "../scheduledTasks/ScheduledTaskService.ts";
23import * as ServerSettings from "../serverSettings.ts";
24import * as SecretRequests from "../secrets/SecretRequests.ts";
25import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts";
26import * as GitVcsDriver from "../vcs/GitVcsDriver.ts";
27import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts";
28import type * as McpInvocationContext from "./McpInvocationContext.ts";
29import * as OrchestratorMcpService from "./OrchestratorMcpService.ts";
30
31const layerOrchestratorService = OrchestratorMcpService.layer.pipe(
32 Layer.provide(Layer.mock(ThreadLaunchService.ThreadLaunchService)({})),
33 Layer.provide(Layer.mock(GitVcsDriver.GitVcsDriver)({})),
34);
35
36const environmentId = EnvironmentId.make("environment-mcp-orchestrator-detail");
37const projectId = ProjectId.make("project-mcp-orchestrator-detail");
38const parentThreadId = ThreadId.make("thread-mcp-orchestrator-parent");
39const childThreadId = ThreadId.make("thread-mcp-orchestrator-child");
40const activeRunId = RunId.make("run-mcp-active");
41const cancelledRunId = RunId.make("run-mcp-cancelled");
42const childRunId = RunId.make("run-mcp-child");
43const taskId = NodeId.make("node-mcp-task-1");
44const now = DateTime.makeUnsafe("2026-08-04T12:00:00.000Z");
45const codexDriver = ProviderDriverKind.make("codex");
46// Distinct from driver kind so a regression that re-derives from driver fails.
47const customCodexInstanceId = ProviderInstanceId.make("codex-custom-workspace");
48const parentInstanceId = ProviderInstanceId.make("codex");
49
50const makeScope = (): McpInvocationContext.McpInvocationScope => ({
51 environmentId,
52 requestNamespace: "provider-session-mcp-orchestrator-detail",
53 thread: {
54 threadId: parentThreadId,
55 providerSessionId: "provider-session-mcp-orchestrator-detail",
56 providerInstanceId: parentInstanceId,
57 },
58 client: undefined,
59 capabilities: new Set(["orchestration"]),
60 issuedAt: 1,
61});
62
63function baseThread(input: {
64 readonly threadId: ThreadId;
65 readonly title: string;
66 readonly instanceId: ProviderInstanceId;
67 readonly model: string;
68}) {
69 return {
70 id: input.threadId,
71 projectId,
72 title: input.title,
73 createdBy: "user" as const,
74 creationSource: "mcp" as const,
75 modelSelection: {
76 instanceId: input.instanceId,
77 model: input.model,
78 },
79 runtimeMode: "full-access" as const,
80 interactionMode: "default" as const,
81 branch: null,
82 worktreePath: null,
83 lineage: {
84 parentThreadId: null,
85 relationshipToParent: null,
86 rootThreadId: input.threadId,
87 },
88 archivedAt: null,
89 deletedAt: null,
90 providerInstanceId: input.instanceId,
91 createdAt: now,
92 updatedAt: now,
93 };
94}
95
96function makeRun(input: {
97 readonly id: RunId;
98 readonly ordinal: number;
99 readonly status: "running" | "waiting" | "cancelled" | "queued" | "completed";
100 readonly instanceId?: ProviderInstanceId;
101}) {
102 return {
103 id: input.id,
104 ordinal: input.ordinal,
105 status: input.status,
106 modelSelection: {
107 instanceId: input.instanceId ?? parentInstanceId,
108 model: "gpt-5.4",
109 },
110 providerInstanceId: input.instanceId ?? parentInstanceId,
111 requestedAt: now,
112 startedAt: input.status === "cancelled" || input.status === "queued" ? null : now,
113 completedAt: input.status === "cancelled" || input.status === "completed" ? now : null,
114 };
115}
116
117it("readThread prefers activity-run status over a newer cancelled queued run", async () => {
118 const projection = {
119 thread: baseThread({
120 threadId: parentThreadId,
121 title: "Parent",
122 instanceId: parentInstanceId,
123 model: "gpt-5.4",
124 }),
125 runs: [
126 makeRun({ id: activeRunId, ordinal: 1, status: "running" }),
127 makeRun({ id: cancelledRunId, ordinal: 2, status: "cancelled" }),
128 ],
129 visibleTurnItems: [],
130 runtimeRequests: [],
131 messages: [],
132 contextTransfers: [],
133 subagents: [],
134 updatedAt: now,
135 } as unknown as OrchestrationV2ThreadProjection;
136
137 const layer = layerOrchestratorService.pipe(
138 Layer.provide(
139 Layer.mergeAll(
140 Layer.mock(ThreadManagementService.ThreadManagementService)({
141 getTimelinePage: () => Effect.succeed({ items: [], totalItems: 0, hasMore: false }),
142 getThreadRecords: (threadId) =>
143 threadId === parentThreadId
144 ? Effect.succeed(projection)
145 : Effect.die(`unexpected thread ${threadId}`),
146 getThreadShell: (threadId) =>
147 Effect.succeed(
148 threadId === parentThreadId
149 ? (projection.thread as unknown as OrchestrationV2ThreadShell)
150 : null,
151 ),
152 getProjectThreadRecords: (input) =>
153 input.threadId === parentThreadId
154 ? Effect.succeed(projection)
155 : Effect.die(`unexpected thread ${input.threadId}`),
156 } satisfies Partial<ThreadManagementService.ThreadManagementService["Service"]>),
157 ServerSettings.layerTest(),
158 Layer.mock(ProviderRegistry.ProviderRegistry)({
159 getProviders: Effect.succeed([]),
160 } satisfies Partial<ProviderRegistry.ProviderRegistry["Service"]>),
161 Layer.mock(ProjectService.ProjectService)({}),
162 Layer.mock(SecretRequests.SecretRequests)({}),
163 Layer.mock(ScheduledTaskService.ScheduledTaskService)({
164 list: () => Effect.succeed({ tasks: [] }),
165 } satisfies Partial<ScheduledTaskService.ScheduledTaskService["Service"]>),
166 Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({
167 list: () => Effect.succeed([]),
168 } satisfies Partial<ProviderAdapterRegistry.ProviderAdapterRegistryV2["Service"]>),
169 NodeCrypto.layer,
170 ),
171 ),
172 );
173
174 await Effect.gen(function* () {
175 const service = yield* OrchestratorMcpService.OrchestratorMcpService;
176 const result = yield* service.readThread(makeScope(), { threadId: parentThreadId });
177 expect(result.thread.status).toBe("running");
178 expect(result.thread.latestRunId).toBe(cancelledRunId);
179 expect(result.thread.activeRunId).toBe(activeRunId);
180 }).pipe(Effect.provide(layer), Effect.runPromise);
181});
182
183it("readThread prefers waiting activity status over a newer cancelled queued run", async () => {
184 const projection = {
185 thread: baseThread({
186 threadId: parentThreadId,
187 title: "Parent waiting",
188 instanceId: parentInstanceId,
189 model: "gpt-5.4",
190 }),
191 runs: [
192 makeRun({ id: activeRunId, ordinal: 1, status: "waiting" }),
193 makeRun({ id: cancelledRunId, ordinal: 2, status: "cancelled" }),
194 ],
195 visibleTurnItems: [],
196 runtimeRequests: [],
197 messages: [],
198 contextTransfers: [],
199 subagents: [],
200 updatedAt: now,
201 } as unknown as OrchestrationV2ThreadProjection;
202
203 const layer = layerOrchestratorService.pipe(
204 Layer.provide(
205 Layer.mergeAll(
206 Layer.mock(ThreadManagementService.ThreadManagementService)({
207 getTimelinePage: () => Effect.succeed({ items: [], totalItems: 0, hasMore: false }),
208 getThreadRecords: (threadId) =>
209 threadId === parentThreadId
210 ? Effect.succeed(projection)
211 : Effect.die(`unexpected thread ${threadId}`),
212 getThreadShell: (threadId) =>
213 Effect.succeed(
214 threadId === parentThreadId
215 ? (projection.thread as unknown as OrchestrationV2ThreadShell)
216 : null,
217 ),
218 getProjectThreadRecords: (input) =>
219 input.threadId === parentThreadId
220 ? Effect.succeed(projection)
221 : Effect.die(`unexpected thread ${input.threadId}`),
222 } satisfies Partial<ThreadManagementService.ThreadManagementService["Service"]>),
223 ServerSettings.layerTest(),
224 Layer.mock(ProviderRegistry.ProviderRegistry)({
225 getProviders: Effect.succeed([]),
226 } satisfies Partial<ProviderRegistry.ProviderRegistry["Service"]>),
227 Layer.mock(ProjectService.ProjectService)({}),
228 Layer.mock(SecretRequests.SecretRequests)({}),
229 Layer.mock(ScheduledTaskService.ScheduledTaskService)({
230 list: () => Effect.succeed({ tasks: [] }),
231 } satisfies Partial<ScheduledTaskService.ScheduledTaskService["Service"]>),
232 Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({
233 list: () => Effect.succeed([]),
234 } satisfies Partial<ProviderAdapterRegistry.ProviderAdapterRegistryV2["Service"]>),
235 NodeCrypto.layer,
236 ),
237 ),
238 );
239
240 await Effect.gen(function* () {
241 const service = yield* OrchestratorMcpService.OrchestratorMcpService;
242 const result = yield* service.readThread(makeScope(), { threadId: parentThreadId });
243 expect(result.thread.status).toBe("waiting");
244 expect(result.thread.activeRunId).toBe(activeRunId);
245 }).pipe(Effect.provide(layer), Effect.runPromise);
246});
247
248it("taskStatus returns task.providerInstanceId rather than the driver kind", async () => {
249 const parentProjection = {
250 thread: baseThread({
251 threadId: parentThreadId,
252 title: "Parent",
253 instanceId: parentInstanceId,
254 model: "gpt-5.4",
255 }),
256 runs: [makeRun({ id: activeRunId, ordinal: 1, status: "running" })],
257 visibleTurnItems: [],
258 runtimeRequests: [],
259 messages: [],
260 contextTransfers: [],
261 subagents: [
262 {
263 id: taskId,
264 threadId: parentThreadId,
265 runId: activeRunId,
266 parentNodeId: NodeId.make("node-parent"),
267 origin: "app_owned",
268 createdBy: "agent",
269 driver: codexDriver,
270 providerInstanceId: customCodexInstanceId,
271 providerThreadId: null,
272 childThreadId,
273 nativeTaskRef: null,
274 prompt: "Inspect the custom instance.",
275 title: null,
276 model: "gpt-5.4",
277 status: "running",
278 result: null,
279 startedAt: now,
280 completedAt: null,
281 updatedAt: now,
282 },
283 ],
284 updatedAt: now,
285 } as unknown as OrchestrationV2ThreadProjection;
286
287 const childProjection = {
288 thread: {
289 ...baseThread({
290 threadId: childThreadId,
291 title: "Child",
292 instanceId: customCodexInstanceId,
293 model: "gpt-5.4",
294 }),
295 lineage: {
296 parentThreadId,
297 relationshipToParent: "subagent",
298 rootThreadId: parentThreadId,
299 },
300 createdBy: "agent",
301 },
302 runs: [
303 makeRun({
304 id: childRunId,
305 ordinal: 1,
306 status: "running",
307 instanceId: customCodexInstanceId,
308 }),
309 ],
310 visibleTurnItems: [],
311 runtimeRequests: [],
312 messages: [],
313 contextTransfers: [
314 {
315 type: "subagent_spawn",
316 sourceThreadId: parentThreadId,
317 targetThreadId: childThreadId,
318 targetRunId: childRunId,
319 },
320 ],
321 subagents: [],
322 providerThreads: [],
323 updatedAt: now,
324 } as unknown as OrchestrationV2ThreadProjection;
325
326 const layer = layerOrchestratorService.pipe(
327 Layer.provide(
328 Layer.mergeAll(
329 Layer.mock(ThreadManagementService.ThreadManagementService)({
330 getTimelinePage: () => Effect.succeed({ items: [], totalItems: 0, hasMore: false }),
331 getThreadRecords: (threadId) => {
332 if (threadId === parentThreadId) return Effect.succeed(parentProjection);
333 if (threadId === childThreadId) return Effect.succeed(childProjection);
334 return Effect.die(`unexpected thread ${threadId}`);
335 },
336 } satisfies Partial<ThreadManagementService.ThreadManagementService["Service"]>),
337 ServerSettings.layerTest(),
338 Layer.mock(ProviderRegistry.ProviderRegistry)({
339 getProviders: Effect.succeed([]),
340 } satisfies Partial<ProviderRegistry.ProviderRegistry["Service"]>),
341 Layer.mock(ProjectService.ProjectService)({}),
342 Layer.mock(SecretRequests.SecretRequests)({}),
343 Layer.mock(ScheduledTaskService.ScheduledTaskService)({
344 list: () => Effect.succeed({ tasks: [] }),
345 } satisfies Partial<ScheduledTaskService.ScheduledTaskService["Service"]>),
346 Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({
347 list: () => Effect.succeed([]),
348 } satisfies Partial<ProviderAdapterRegistry.ProviderAdapterRegistryV2["Service"]>),
349 NodeCrypto.layer,
350 ),
351 ),
352 );
353
354 await Effect.gen(function* () {
355 const service = yield* OrchestratorMcpService.OrchestratorMcpService;
356 const result = yield* service.taskStatus(makeScope(), taskId);
357 expect(result.providerInstanceId).toBe(customCodexInstanceId);
358 expect(result.providerInstanceId).not.toBe(ProviderInstanceId.make(String(codexDriver)));
359 expect(result.status).toBe("running");
360 expect(result.taskId).toBe(taskId);
361 expect(result.childThreadId).toBe(childThreadId);
362 }).pipe(Effect.provide(layer), Effect.runPromise);
363});
364
365it("readThread and sendToThread reach threads in other projects", async () => {
366 let parentRuns: ReadonlyArray<unknown> = [
367 makeRun({ id: RunId.make("run-parent-live"), ordinal: 1, status: "running" }),
368 ];
369 const foreignProjectId = ProjectId.make("project-mcp-orchestrator-foreign");
370 const foreignThreadId = ThreadId.make("thread-mcp-orchestrator-foreign");
371 const parentProjection = {
372 thread: baseThread({
373 threadId: parentThreadId,
374 title: "Parent",
375 instanceId: parentInstanceId,
376 model: "gpt-5.4",
377 }),
378 get runs() {
379 return parentRuns;
380 },
381 visibleTurnItems: [],
382 runtimeRequests: [],
383 messages: [],
384 contextTransfers: [],
385 subagents: [],
386 updatedAt: now,
387 } as unknown as OrchestrationV2ThreadProjection;
388 const foreignProjection = (threadId: ThreadId) =>
389 ({
390 thread: {
391 ...baseThread({
392 threadId,
393 title: "Foreign",
394 instanceId: parentInstanceId,
395 model: "gpt-5.4",
396 }),
397 projectId: foreignProjectId,
398 },
399 runs: [],
400 visibleTurnItems: [
401 {
402 position: 0,
403 visibility: "inherited",
404 sourceThreadId: threadId,
405 sourceItemId: "item-1",
406 item: {
407 id: "item-1",
408 type: "assistant_message",
409 messageId: "assistant-1",
410 status: "completed",
411 title: null,
412 text: "Foreign thread said hello",
413 createdAt: now,
414 updatedAt: now,
415 },
416 },
417 ],
418 runtimeRequests: [],
419 messages: [],
420 contextTransfers: [],
421 subagents: [],
422 updatedAt: now,
423 }) as unknown as OrchestrationV2ThreadProjection;
424
425 const layer = layerOrchestratorService.pipe(
426 Layer.provide(
427 Layer.mergeAll(
428 Layer.mock(ThreadManagementService.ThreadManagementService)({
429 getThreadRecords: (threadId) => {
430 if (threadId === parentThreadId) return Effect.succeed(parentProjection);
431 if (threadId === foreignThreadId) return Effect.succeed(foreignProjection(threadId));
432 return Effect.die(`unexpected thread ${threadId}`);
433 },
434 getThreadShell: (threadId) =>
435 Effect.succeed(
436 threadId === foreignThreadId
437 ? (foreignProjection(threadId).thread as unknown as OrchestrationV2ThreadShell)
438 : null,
439 ),
440 getTimelinePage: (threadId) =>
441 Effect.succeed({
442 items: foreignProjection(threadId).visibleTurnItems,
443 totalItems: 1,
444 hasMore: false,
445 }),
446 getProjectThreadRecords: (input) =>
447 input.threadId === foreignThreadId && input.projectId === foreignProjectId
448 ? Effect.succeed(foreignProjection(input.threadId))
449 : Effect.fail(
450 new ThreadManagementService.ThreadManagementThreadNotFoundError({
451 projectId: input.projectId,
452 threadId: input.threadId,
453 }),
454 ),
455 sendToThread: () =>
456 Effect.succeed({
457 run: { id: "run-foreign", status: "queued" },
458 delivery: "started",
459 } as unknown as ThreadManagementService.ThreadManagementSendResult),
460 } satisfies Partial<ThreadManagementService.ThreadManagementService["Service"]>),
461 ServerSettings.layerTest(),
462 Layer.mock(ProviderRegistry.ProviderRegistry)({
463 getProviders: Effect.succeed([]),
464 } satisfies Partial<ProviderRegistry.ProviderRegistry["Service"]>),
465 Layer.mock(ProjectService.ProjectService)({}),
466 Layer.mock(SecretRequests.SecretRequests)({}),
467 Layer.mock(ScheduledTaskService.ScheduledTaskService)({
468 list: () =>
469 Effect.succeed({
470 tasks: [
471 {
472 id: "task-foreign",
473 projectId: foreignProjectId,
474 runtimeMode: "approval-required",
475 interactionMode: "default",
476 } as never,
477 ],
478 }),
479 } satisfies Partial<ScheduledTaskService.ScheduledTaskService["Service"]>),
480 Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({
481 list: () => Effect.succeed([]),
482 } satisfies Partial<ProviderAdapterRegistry.ProviderAdapterRegistryV2["Service"]>),
483 NodeCrypto.layer,
484 ),
485 ),
486 );
487
488 await Effect.gen(function* () {
489 const service = yield* OrchestratorMcpService.OrchestratorMcpService;
490 const foreign = yield* service.readThread(makeScope(), { threadId: foreignThreadId });
491 expect(foreign.thread.threadId).toBe(foreignThreadId);
492 expect(foreign.thread.projectId).toBe(foreignProjectId);
493 expect(foreign.items.map((item) => item.text)).toEqual(["Foreign thread said hello"]);
494
495 const sent = yield* service.sendToThread(makeScope(), {
496 threadId: foreignThreadId,
497 message: "hi",
498 });
499 expect(sent.threadId).toBe(foreignThreadId);
500
501 // Once the caller's run ends, it can still read other threads but no longer write to them.
502 parentRuns = [];
503 yield* service.readThread(makeScope(), { threadId: foreignThreadId });
504 const stale = yield* service
505 .sendToThread(makeScope(), { threadId: foreignThreadId, message: "hi again" })
506 .pipe(Effect.flip);
507 expect(stale.code).toBe("parent_not_active");
508 const staleInterrupt = yield* service
509 .interruptThread(makeScope(), { threadId: foreignThreadId })
510 .pipe(Effect.flip);
511 expect(staleInterrupt.code).toBe("parent_not_active");
512 // Nor create, change or remove scheduled work in another project.
513 const staleSchedule = yield* service
514 .scheduleTask(makeScope(), {
515 projectId: foreignProjectId,
516 prompt: "check in later",
517 schedule: { type: "interval", everyMs: 3_600_000 },
518 })
519 .pipe(Effect.flip);
520 expect(staleSchedule.code).toBe("parent_not_active");
521 const staleUpdate = yield* service
522 .updateScheduledTask(makeScope(), {
523 scheduledTaskId: ScheduledTaskId.make("task-foreign"),
524 enabled: false,
525 })
526 .pipe(Effect.flip);
527 expect(staleUpdate.code).toBe("parent_not_active");
528 const staleDelete = yield* service
529 .deleteScheduledTask(makeScope(), { scheduledTaskId: ScheduledTaskId.make("task-foreign") })
530 .pipe(Effect.flip);
531 expect(staleDelete.code).toBe("parent_not_active");
532 }).pipe(Effect.provide(layer), Effect.runPromise);
533});