1import { expect, it } from "@effect/vitest";
2import {
3 CommandId,
4 MessageId,
5 NodeId,
6 type OrchestrationV2Command,
7 type OrchestrationV2Run,
8 type OrchestrationV2Subagent,
9 type OrchestrationV2ProviderThread,
10 ProviderThreadId,
11 ProviderDriverKind,
12 ProviderSessionId,
13 type OrchestrationV2StoredEvent,
14 type OrchestrationV2ThreadProjection,
15 type OrchestrationV2ThreadShell,
16 ProjectId,
17 ProviderInstanceId,
18 RunId,
19 ThreadId,
20} from "@t3tools/contracts";
21import * as Deferred from "effect/Deferred";
22import * as DateTime from "effect/DateTime";
23import * as Option from "effect/Option";
24import * as Duration from "effect/Duration";
25import * as Effect from "effect/Effect";
26import * as Fiber from "effect/Fiber";
27import * as Layer from "effect/Layer";
28import * as Queue from "effect/Queue";
29import * as Ref from "effect/Ref";
30import * as Stream from "effect/Stream";
31import * as TestClock from "effect/testing/TestClock";
32
33import * as LegacyV1ThreadImporter from "./legacy/LegacyV1ThreadImporter.ts";
34import * as ServerSettings from "../serverSettings.ts";
35import * as SqlitePersistence from "../persistence/Sqlite.ts";
36import * as CommandReceiptStore from "./CommandReceiptStore.ts";
37import * as Orchestrator from "./Orchestrator.ts";
38import * as ProviderSessionManager from "./ProviderSessionManager.ts";
39import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
40import * as ThreadManagementService from "./ThreadManagementService.ts";
41import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts";
42import { DispatchModeLimit, type DispatchModeRefusal } from "./DispatchModeLimit.ts";
43
44it("stamps authoritative provenance on commands that create threads or messages", () => {
45 const command: OrchestrationV2Command = {
46 type: "thread.create",
47 createdBy: "agent",
48 creationSource: "mcp",
49 commandId: CommandId.make("command:thread-management:create"),
50 threadId: ThreadId.make("thread:thread-management:create"),
51 projectId: ProjectId.make("project:thread-management"),
52 title: "Thread management",
53 modelSelection: {
54 instanceId: ProviderInstanceId.make("codex"),
55 model: "gpt-5-codex",
56 },
57 runtimeMode: "full-access",
58 interactionMode: "default",
59 branch: null,
60 worktreePath: null,
61 };
62
63 expect(
64 ThreadManagementService.withCreationProvenance(command, {
65 createdBy: "user",
66 creationSource: "web",
67 }),
68 ).toMatchObject({
69 createdBy: "user",
70 creationSource: "web",
71 });
72});
73
74it("leaves commands that do not create durable authored content unchanged", () => {
75 const command: OrchestrationV2Command = {
76 type: "run.interrupt",
77 commandId: CommandId.make("command:thread-management:interrupt"),
78 threadId: ThreadId.make("thread:thread-management:interrupt"),
79 runId: RunId.make("run:thread-management:interrupt"),
80 };
81
82 expect(
83 ThreadManagementService.withCreationProvenance(command, {
84 createdBy: "user",
85 creationSource: "web",
86 }),
87 ).toBe(command);
88});
89
90it("identifies every existing thread that must be hydrated before dispatch", () => {
91 const sourceThreadId = ThreadId.make("thread:thread-management:source");
92 const targetThreadId = ThreadId.make("thread:thread-management:target");
93 const parentThreadId = ThreadId.make("thread:thread-management:parent");
94
95 expect(
96 ThreadManagementService.existingThreadIdsForCommand({
97 type: "thread.create",
98 createdBy: "user",
99 creationSource: "web",
100 commandId: CommandId.make("command:thread-management:create"),
101 threadId: targetThreadId,
102 projectId: ProjectId.make("project:thread-management"),
103 title: "Created thread",
104 modelSelection: {
105 instanceId: ProviderInstanceId.make("codex"),
106 model: "gpt-5-codex",
107 },
108 runtimeMode: "full-access",
109 interactionMode: "default",
110 branch: null,
111 worktreePath: null,
112 }),
113 ).toEqual([]);
114
115 expect(
116 ThreadManagementService.existingThreadIdsForCommand({
117 type: "thread.archive",
118 commandId: CommandId.make("command:thread-management:archive"),
119 threadId: targetThreadId,
120 }),
121 ).toEqual([targetThreadId]);
122
123 // Read-state commands skip transcript hydration entirely: they fire on
124 // every activity bump while a thread is open and never touch messages.
125 expect(
126 ThreadManagementService.existingThreadIdsForCommand({
127 type: "thread.visit",
128 commandId: CommandId.make("command:thread-management:visit"),
129 threadId: targetThreadId,
130 visitedAt: "2026-07-30T00:00:00.000Z",
131 }),
132 ).toEqual([]);
133
134 expect(
135 ThreadManagementService.existingThreadIdsForCommand({
136 type: "thread.mark-unread",
137 commandId: CommandId.make("command:thread-management:mark-unread"),
138 threadId: targetThreadId,
139 }),
140 ).toEqual([]);
141
142 expect(
143 ThreadManagementService.existingThreadIdsForCommand({
144 type: "thread.fork",
145 createdBy: "user",
146 creationSource: "web",
147 commandId: CommandId.make("command:thread-management:fork"),
148 sourceThreadId,
149 targetThreadId,
150 sourcePoint: {
151 type: "run",
152 runId: RunId.make("run:thread-management:source"),
153 },
154 }),
155 ).toEqual([sourceThreadId]);
156
157 expect(
158 ThreadManagementService.existingThreadIdsForCommand({
159 type: "thread.merge_back",
160 createdBy: "user",
161 creationSource: "web",
162 commandId: CommandId.make("command:thread-management:merge"),
163 sourceThreadId,
164 targetThreadId,
165 sourcePoint: {
166 type: "run",
167 runId: RunId.make("run:thread-management:source"),
168 },
169 }),
170 ).toEqual([sourceThreadId, targetThreadId]);
171
172 expect(
173 ThreadManagementService.existingThreadIdsForCommand({
174 type: "delegated_task.request",
175 createdBy: "agent",
176 creationSource: "provider",
177 commandId: CommandId.make("command:thread-management:delegate"),
178 parentThreadId,
179 parentRunId: RunId.make("run:thread-management:parent"),
180 parentNodeId: NodeId.make("node:thread-management:parent"),
181 task: "Inspect the migration",
182 modelSelection: {
183 instanceId: ProviderInstanceId.make("codex"),
184 model: "gpt-5-codex",
185 },
186 runtimeMode: "full-access",
187 interactionMode: "default",
188 }),
189 ).toEqual([parentThreadId]);
190
191 expect(
192 ThreadManagementService.existingThreadIdsForCommand({
193 type: "delegated_task.wake-policy",
194 commandId: CommandId.make("command:thread-management:wake-policy"),
195 parentThreadId,
196 taskId: NodeId.make("node:thread-management:delegated"),
197 completionWake: "always",
198 }),
199 ).toEqual([parentThreadId]);
200
201 expect(
202 ThreadManagementService.existingThreadIdsForCommand({
203 type: "thread.created.record",
204 commandId: CommandId.make("command:thread-management:record"),
205 parentThreadId,
206 parentRunId: RunId.make("run:thread-management:parent"),
207 parentNodeId: NodeId.make("node:thread-management:parent"),
208 targetThreadId,
209 targetRunId: null,
210 }),
211 ).toEqual([parentThreadId, targetThreadId]);
212});
213
214it("derives thread management messages from structural error attributes", () => {
215 const projectId = ProjectId.make("project:thread-management:errors");
216 const threadId = ThreadId.make("thread:thread-management:errors");
217 const runId = RunId.make("run:thread-management:errors");
218 const messageId = MessageId.make("message:thread-management:errors");
219 const infrastructureCause = new Error("private sqlite detail");
220
221 const threadNotFound = new ThreadManagementService.ThreadManagementThreadNotFoundError({
222 projectId,
223 threadId,
224 });
225 expect(threadNotFound).toMatchObject({ projectId, threadId });
226 expect(threadNotFound.message).toBe(`Thread ${threadId} was not found in project ${projectId}.`);
227
228 const runNotFound = new ThreadManagementService.ThreadManagementRunNotFoundError({
229 threadId,
230 runId,
231 });
232 expect(runNotFound).toMatchObject({ threadId, runId });
233 expect(runNotFound.message).toBe(`Run ${runId} does not belong to thread ${threadId}.`);
234
235 const archived = new ThreadManagementService.ThreadManagementThreadArchivedError({
236 threadId,
237 });
238 expect(archived).toMatchObject({ threadId });
239 expect(archived.message).toBe(`Thread ${threadId} is archived and cannot receive messages.`);
240
241 const notSteerable = new ThreadManagementService.ThreadManagementNoSteerableRunError({
242 threadId,
243 mode: "restart",
244 });
245 expect(notSteerable).toMatchObject({
246 threadId,
247 mode: "restart",
248 });
249 expect(notSteerable.message).toBe(
250 `Thread ${threadId} has no running turn that can be restarted.`,
251 );
252
253 const notInterruptible = new ThreadManagementService.ThreadManagementThreadNotInterruptibleError({
254 threadId,
255 runId,
256 });
257 expect(notInterruptible).toMatchObject({ threadId, runId });
258 expect(notInterruptible.message).toBe(`Run ${runId} is not currently interruptible.`);
259
260 const listFailure = new ThreadManagementService.ThreadManagementProjectThreadsListError({
261 projectId,
262 cause: infrastructureCause,
263 });
264 expect(listFailure).toMatchObject({ projectId, cause: infrastructureCause });
265 expect(listFailure.message).toBe(`Unable to list threads in project ${projectId}.`);
266 expect(listFailure.message).not.toContain(infrastructureCause.message);
267
268 const durableProjectionFailure =
269 new ThreadManagementService.ThreadManagementDurableRunProjectionError({
270 threadId,
271 messageId,
272 });
273 expect(durableProjectionFailure).toMatchObject({ threadId, messageId });
274 expect(durableProjectionFailure.message).toBe(
275 `Message ${messageId} was accepted on thread ${threadId} without a durable run projection.`,
276 );
277});
278
279const layerReceiptsTest = CommandReceiptStore.layer.pipe(
280 Layer.provide(SqlitePersistence.layerMemory),
281);
282
283const layerThreadManagementTest = (settingsLayer: ReturnType<typeof ServerSettings.layerTest>) =>
284 ThreadManagementService.layer.pipe(
285 Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})),
286 Layer.provide(Layer.mock(Orchestrator.OrchestratorV2)({})),
287 Layer.provide(settingsLayer),
288 Layer.provide(layerReceiptsTest),
289 );
290
291it.effect("refuses a new turn when the project effectively disables the instance", () => {
292 const projectId = ProjectId.make("project:thread-management:gate-disabled");
293 const instanceId = ProviderInstanceId.make("codex_work");
294 const commandId = CommandId.make("command:thread-management:gate-disabled");
295 const layerTest = layerThreadManagementTest(
296 ServerSettings.layerTest({
297 providerInstances: { [instanceId]: { driver: "codex", enabled: true } },
298 projectSettingsOverrides: {
299 [projectId]: { providerInstances: { [instanceId]: { enabled: false } } },
300 },
301 }),
302 );
303
304 return Effect.gen(function* () {
305 const threads = yield* ThreadManagementService.ThreadManagementService;
306 const error = yield* Effect.flip(
307 threads.assertProviderInstanceEnabled({ projectId, commandId, instanceId }),
308 );
309
310 expect(error).toBeInstanceOf(
311 ThreadManagementService.ThreadManagementProviderInstanceDisabledError,
312 );
313 expect(error).toMatchObject({ projectId, instanceId });
314 expect(error.message).toBe(
315 `Provider instance "${instanceId}" is disabled for project ${projectId}.`,
316 );
317 }).pipe(Effect.provide(layerTest));
318});
319
320it.effect("allows a new turn when the project enables an instance the machine disabled", () => {
321 const projectId = ProjectId.make("project:thread-management:gate-enabled");
322 const instanceId = ProviderInstanceId.make("codex_work");
323 const commandId = CommandId.make("command:thread-management:gate-enabled");
324 const layerTest = layerThreadManagementTest(
325 ServerSettings.layerTest({
326 providerInstances: { [instanceId]: { driver: "codex", enabled: false } },
327 projectSettingsOverrides: {
328 [projectId]: { providerInstances: { [instanceId]: { enabled: true } } },
329 },
330 }),
331 );
332
333 return Effect.gen(function* () {
334 const threads = yield* ThreadManagementService.ThreadManagementService;
335 yield* threads.assertProviderInstanceEnabled({ projectId, commandId, instanceId });
336 }).pipe(Effect.provide(layerTest));
337});
338
339it.effect("inherits the machine value when the project has no override", () => {
340 const projectId = ProjectId.make("project:thread-management:gate-inherit");
341 const instanceId = ProviderInstanceId.make("codex_work");
342 const commandId = CommandId.make("command:thread-management:gate-inherit");
343 const layerTest = layerThreadManagementTest(
344 ServerSettings.layerTest({
345 providerInstances: { [instanceId]: { driver: "codex", enabled: false } },
346 }),
347 );
348
349 return Effect.gen(function* () {
350 const threads = yield* ThreadManagementService.ThreadManagementService;
351 const error = yield* Effect.flip(
352 threads.assertProviderInstanceEnabled({ projectId, commandId, instanceId }),
353 );
354
355 expect(error).toBeInstanceOf(
356 ThreadManagementService.ThreadManagementProviderInstanceDisabledError,
357 );
358 }).pipe(Effect.provide(layerTest));
359});
360
361const layerDispatchChokepointTest =
362 (settingsLayer: ReturnType<typeof ServerSettings.layerTest>) => (dispatched: Array<string>) =>
363 ThreadManagementService.layer.pipe(
364 Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})),
365 Layer.provide(
366 Layer.mock(Orchestrator.OrchestratorV2)({
367 getThreadRecords: () =>
368 Effect.succeed({
369 thread: {
370 id: ThreadId.make("thread:thread-management:chokepoint"),
371 projectId: ProjectId.make("project:thread-management:chokepoint"),
372 deletedAt: null,
373 modelSelection: {
374 instanceId: ProviderInstanceId.make("codex_work"),
375 model: "test",
376 },
377 },
378 } as unknown as OrchestrationV2ThreadProjection),
379 dispatch: (command) =>
380 Effect.sync(() => {
381 dispatched.push(command.type);
382 return { sequence: 1, storedEvents: [] } as never;
383 }),
384 }),
385 ),
386 Layer.provide(settingsLayer),
387 Layer.provideMerge(layerReceiptsTest),
388 );
389
390it.effect(
391 "the dispatch chokepoint refuses a continuation message.dispatch on a project-disabled instance before reaching the orchestrator",
392 () => {
393 const projectId = ProjectId.make("project:thread-management:chokepoint");
394 const threadId = ThreadId.make("thread:thread-management:chokepoint");
395 const instanceId = ProviderInstanceId.make("codex_work");
396 const commandId = CommandId.make("command:thread-management:chokepoint-disabled");
397 const dispatched: Array<string> = [];
398 const layerTest = layerDispatchChokepointTest(
399 ServerSettings.layerTest({
400 providerInstances: { [instanceId]: { driver: "codex", enabled: true } },
401 projectSettingsOverrides: {
402 [projectId]: { providerInstances: { [instanceId]: { enabled: false } } },
403 },
404 }),
405 )(dispatched);
406
407 return Effect.gen(function* () {
408 const threads = yield* ThreadManagementService.ThreadManagementService;
409 const error = yield* Effect.flip(
410 threads.dispatch({
411 type: "message.dispatch",
412 commandId,
413 threadId,
414 messageId: MessageId.make("message:thread-management:chokepoint-disabled"),
415 text: "Background task completed.",
416 attachments: [],
417 dispatchMode: { type: "queue_after_active" },
418 createdBy: "agent",
419 creationSource: "server",
420 }),
421 );
422
423 expect(error).toBeInstanceOf(
424 ThreadManagementService.ThreadManagementProviderInstanceDisabledError,
425 );
426 expect(error).toMatchObject({ projectId, instanceId });
427 expect(dispatched).toEqual([]);
428 }).pipe(Effect.provide(layerTest));
429 },
430);
431
432it.effect(
433 "a replay of an already-accepted command id dispatches despite the instance now being disabled",
434 () => {
435 const projectId = ProjectId.make("project:thread-management:chokepoint");
436 const threadId = ThreadId.make("thread:thread-management:chokepoint");
437 const instanceId = ProviderInstanceId.make("codex_work");
438 const commandId = CommandId.make("command:thread-management:chokepoint-replay");
439 const dispatched: Array<string> = [];
440 const layerTest = layerDispatchChokepointTest(
441 ServerSettings.layerTest({
442 providerInstances: { [instanceId]: { driver: "codex", enabled: true } },
443 projectSettingsOverrides: {
444 [projectId]: { providerInstances: { [instanceId]: { enabled: false } } },
445 },
446 }),
447 )(dispatched);
448
449 return Effect.gen(function* () {
450 const receipts = yield* CommandReceiptStore.CommandReceiptStoreV2;
451 yield* receipts.insertIfAbsent({
452 commandId,
453 threadId,
454 commandType: "message.dispatch",
455 acceptedAt: yield* DateTime.now,
456 resultSequence: 1,
457 status: "accepted",
458 error: null,
459 });
460
461 const threads = yield* ThreadManagementService.ThreadManagementService;
462 yield* threads.dispatch({
463 type: "message.dispatch",
464 commandId,
465 threadId,
466 messageId: MessageId.make("message:thread-management:chokepoint-replay"),
467 text: "Background task completed.",
468 attachments: [],
469 dispatchMode: { type: "queue_after_active" },
470 createdBy: "agent",
471 creationSource: "server",
472 });
473
474 expect(dispatched).toEqual(["message.dispatch"]);
475 }).pipe(Effect.provide(layerTest));
476 },
477);
478
479it.effect("classifies projection infrastructure failures separately from a missing thread", () => {
480 const projectId = ProjectId.make("project:thread-management:projection-failure");
481 const threadId = ThreadId.make("thread:thread-management:projection-failure");
482 const infrastructureCause = new Error("sqlite read failed");
483 const projectionError = new Orchestrator.OrchestratorProjectionError({
484 threadId,
485 cause: infrastructureCause,
486 });
487 const layerTest = ThreadManagementService.layer.pipe(
488 Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})),
489 Layer.provide(
490 Layer.mock(Orchestrator.OrchestratorV2)({
491 getThreadProjection: () => Effect.fail(projectionError),
492 }),
493 ),
494 Layer.provide(ServerSettings.layerTest()),
495 Layer.provide(layerReceiptsTest),
496 );
497
498 return Effect.gen(function* () {
499 const service = yield* ThreadManagementService.ThreadManagementService;
500 const error = yield* Effect.flip(service.getProjectThread({ projectId, threadId }));
501
502 expect(error).toBeInstanceOf(ThreadManagementService.ThreadManagementProjectionLoadError);
503 expect(error).toMatchObject({
504 projectId,
505 threadId,
506 cause: projectionError,
507 });
508 expect(error.message).toBe(`Unable to load thread ${threadId} in project ${projectId}.`);
509 }).pipe(Effect.provide(layerTest));
510});
511
512it.effect("uses thread-not-found only after a projection loads outside the project", () => {
513 const projectId = ProjectId.make("project:thread-management:requested");
514 const otherProjectId = ProjectId.make("project:thread-management:other");
515 const threadId = ThreadId.make("thread:thread-management:wrong-project");
516 const projection = {
517 thread: {
518 id: threadId,
519 projectId: otherProjectId,
520 deletedAt: null,
521 },
522 } as OrchestrationV2ThreadProjection;
523 const layerTest = ThreadManagementService.layer.pipe(
524 Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})),
525 Layer.provide(
526 Layer.mock(Orchestrator.OrchestratorV2)({
527 getThreadProjection: () => Effect.succeed(projection),
528 }),
529 ),
530 Layer.provide(ServerSettings.layerTest()),
531 Layer.provide(layerReceiptsTest),
532 );
533
534 return Effect.gen(function* () {
535 const service = yield* ThreadManagementService.ThreadManagementService;
536 const error = yield* Effect.flip(service.getProjectThread({ projectId, threadId }));
537
538 expect(error).toBeInstanceOf(ThreadManagementService.ThreadManagementThreadNotFoundError);
539 expect(error).toMatchObject({ projectId, threadId });
540 expect("cause" in error).toBe(false);
541 }).pipe(Effect.provide(layerTest));
542});
543
544it.effect("preserves failed legacy materialization when reading checkpoint context", () => {
545 const threadId = ThreadId.make("thread:thread-management:checkpoint-import-failure");
546 const importError = new LegacyV1ThreadImporter.LegacyV1ThreadImportError({
547 threadId,
548 operation: "hydrate transcript for",
549 cause: new Error("checkpoint import failed"),
550 });
551 const layerTest = ThreadManagementService.layerWithLegacyImporter.pipe(
552 Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})),
553 Layer.provide(
554 Layer.mergeAll(
555 Layer.mock(Orchestrator.OrchestratorV2)({
556 getCheckpointContext: () =>
557 Effect.succeed({ runs: [], checkpointScopes: [], checkpoints: [] }),
558 }),
559 Layer.mock(LegacyV1ThreadImporter.LegacyV1ThreadImporter)({
560 ensureTranscript: () => Effect.fail(importError),
561 }),
562 ),
563 ),
564 Layer.provide(ServerSettings.layerTest()),
565 Layer.provide(layerReceiptsTest),
566 );
567 return Effect.gen(function* () {
568 const service = yield* ThreadManagementService.ThreadManagementService;
569 const error = yield* service.getCheckpointContext(threadId).pipe(Effect.flip);
570 expect(error).toBeInstanceOf(Orchestrator.OrchestratorProjectionError);
571 expect(error).toMatchObject({ threadId, cause: importError });
572 }).pipe(Effect.provide(layerTest));
573});
574
575it.effect.each([
576 { finalStatus: "completed" as const, timedOut: false },
577 { finalStatus: "failed" as const, timedOut: false },
578 { finalStatus: "cancelled" as const, timedOut: false },
579 { finalStatus: "interrupted" as const, timedOut: false },
580 { finalStatus: "rolled_back" as const, timedOut: false },
581 { finalStatus: "running" as const, timedOut: true },
582 { finalStatus: "missing" as const },
583])("waitForThread timeout final read when selected run is $finalStatus", (scenario) =>
584 Effect.gen(function* () {
585 const projectId = ProjectId.make("project:thread-management:wait-timeout");
586 const threadId = ThreadId.make("thread:thread-management:wait-timeout");
587 const runId = RunId.make("run:thread-management:wait-timeout");
588 const subscribed = yield* Deferred.make<void>();
589 let reads = 0;
590 const projection = (status: OrchestrationV2Run["status"] | "missing") =>
591 ({
592 thread: { id: threadId, projectId, deletedAt: null },
593 runs: status === "missing" ? [] : [{ id: runId, status }],
594 }) as unknown as OrchestrationV2ThreadProjection;
595 const layerTest = ThreadManagementService.layer.pipe(
596 Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})),
597 Layer.provide(
598 Layer.mock(Orchestrator.OrchestratorV2)({
599 getThreadEventSequence: () => Effect.succeed(0),
600 restartContinuationPending: () => Effect.succeed(false),
601 // No run update arrives, so the timeout path runs while a final
602 // projection read can still observe a terminal run.
603 streamStoredEventsFrom: () =>
604 Stream.fromEffect(Deferred.succeed(subscribed, undefined)).pipe(
605 Stream.drain,
606 Stream.concat(Stream.never),
607 ),
608 getThreadRecords: () =>
609 Effect.sync(() => {
610 reads += 1;
611 return projection(reads === 1 ? "running" : scenario.finalStatus);
612 }),
613 }),
614 ),
615 Layer.provide(ServerSettings.layerTest()),
616 Layer.provide(layerReceiptsTest),
617 );
618 const service = yield* ThreadManagementService.ThreadManagementService.pipe(
619 Effect.provide(layerTest),
620 );
621 const fiber = yield* service
622 .waitForThread({
623 projectId,
624 threadId,
625 runId,
626 timeoutMs: 1,
627 })
628 .pipe(Effect.result, Effect.forkChild);
629 yield* Deferred.await(subscribed);
630 yield* TestClock.adjust(Duration.millis(1));
631 const result = yield* Fiber.join(fiber);
632
633 if (scenario.finalStatus === "missing") {
634 expect(result._tag).toBe("Failure");
635 expect(result).toMatchObject({
636 failure: expect.any(ThreadManagementService.ThreadManagementRunNotFoundError),
637 });
638 expect(result).toMatchObject({
639 failure: { threadId, runId },
640 });
641 } else {
642 expect(result._tag).toBe("Success");
643 expect(result).toMatchObject({
644 success: {
645 threadId,
646 timedOut: scenario.timedOut,
647 run: { id: runId, status: scenario.finalStatus },
648 },
649 });
650 }
651 }),
652);
653
654it.effect("waitForThread reads the run again only when the run updates", () =>
655 Effect.gen(function* () {
656 const projectId = ProjectId.make("project:thread-management:wait-event");
657 const threadId = ThreadId.make("thread:thread-management:wait-event");
658 const runId = RunId.make("run:thread-management:wait-event");
659 const subscribed = yield* Deferred.make<void>();
660 const events = yield* Queue.unbounded<OrchestrationV2StoredEvent>();
661 let status: OrchestrationV2Run["status"] = "running";
662 let reads = 0;
663 const stored = (sequence: number, event: object) =>
664 ({ sequence, event: { threadId, ...event } }) as unknown as OrchestrationV2StoredEvent;
665 const layerTest = ThreadManagementService.layer.pipe(
666 Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})),
667 Layer.provide(
668 Layer.mock(Orchestrator.OrchestratorV2)({
669 getThreadEventSequence: () => Effect.succeed(0),
670 restartContinuationPending: () => Effect.succeed(false),
671 // Only the run.updated stream carries events in this test.
672 streamStoredEventsFrom: (input) =>
673 input?.eventType === "run.updated"
674 ? Stream.fromEffect(Deferred.succeed(subscribed, undefined)).pipe(
675 Stream.drain,
676 Stream.concat(Stream.fromQueue(events)),
677 )
678 : Stream.never,
679 getThreadRecords: () =>
680 Effect.sync(() => {
681 reads += 1;
682 return {
683 thread: { id: threadId, projectId, deletedAt: null },
684 runs: [{ id: runId, status }],
685 } as unknown as OrchestrationV2ThreadProjection;
686 }),
687 }),
688 ),
689 Layer.provide(ServerSettings.layerTest()),
690 Layer.provide(layerReceiptsTest),
691 );
692 const service = yield* ThreadManagementService.ThreadManagementService.pipe(
693 Effect.provide(layerTest),
694 );
695 const fiber = yield* service
696 .waitForThread({ projectId, threadId, runId, timeoutMs: 60 * 60 * 1_000 })
697 .pipe(Effect.forkChild);
698 yield* Deferred.await(subscribed);
699 status = "completed";
700 yield* Queue.offer(events, stored(1, { type: "run.updated", payload: { id: "other-run" } }));
701 yield* Queue.offer(events, stored(2, { type: "run.updated", payload: { id: runId, status } }));
702 const result = yield* Fiber.join(fiber);
703
704 expect(result).toMatchObject({ timedOut: false, run: { id: runId, status: "completed" } });
705 // The first read plus one for this run's update. The other run caused none.
706 expect(reads).toBe(2);
707 }),
708);
709
710it.effect("waitForThread follows a restart continuation instead of the cut run", () =>
711 Effect.gen(function* () {
712 const projectId = ProjectId.make("project:thread-management:wait-restart");
713 const threadId = ThreadId.make("thread:thread-management:wait-restart");
714 const runId = RunId.make("run:thread-management:wait-restart");
715 const continuationId = RunId.make("run:thread-management:wait-restart:2");
716 const subscribed = yield* Deferred.make<void>();
717 const events = yield* Queue.unbounded<OrchestrationV2StoredEvent>();
718 // The restart cut the run; its continuation effect has not started yet.
719 let runs: Array<{
720 id: RunId;
721 status: OrchestrationV2Run["status"];
722 restartContinuationOfRunId?: RunId;
723 }> = [{ id: runId, status: "cancelled" }];
724 let pending = true;
725 const stored = (sequence: number, event: object) =>
726 ({ sequence, event: { threadId, ...event } }) as unknown as OrchestrationV2StoredEvent;
727 const layerTest = ThreadManagementService.layer.pipe(
728 Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})),
729 Layer.provide(
730 Layer.mock(Orchestrator.OrchestratorV2)({
731 getThreadEventSequence: () => Effect.succeed(0),
732 restartContinuationPending: (run) => Effect.succeed(pending && run.id === runId),
733 awaitRestartContinuation: () => Effect.never,
734 streamStoredEventsFrom: (input) =>
735 input?.eventType === "run.updated"
736 ? Stream.fromEffect(Deferred.succeed(subscribed, undefined)).pipe(
737 Stream.drain,
738 Stream.concat(Stream.fromQueue(events)),
739 )
740 : Stream.never,
741 getThreadRecords: () =>
742 Effect.sync(
743 () =>
744 ({
745 thread: { id: threadId, projectId, deletedAt: null },
746 runs,
747 }) as unknown as OrchestrationV2ThreadProjection,
748 ),
749 }),
750 ),
751 Layer.provide(ServerSettings.layerTest()),
752 Layer.provide(layerReceiptsTest),
753 );
754 const service = yield* ThreadManagementService.ThreadManagementService.pipe(
755 Effect.provide(layerTest),
756 );
757 const fiber = yield* service
758 .waitForThread({ projectId, threadId, runId, timeoutMs: 60 * 60 * 1_000 })
759 .pipe(Effect.forkChild);
760 yield* Deferred.await(subscribed);
761 const continuation = { id: continuationId, restartContinuationOfRunId: runId };
762 runs = [...runs, { ...continuation, status: "running" }];
763 pending = false;
764 yield* Queue.offer(
765 events,
766 stored(1, { type: "run.updated", payload: { ...continuation, status: "running" } }),
767 );
768 runs = [runs[0]!, { ...continuation, status: "completed" }];
769 yield* Queue.offer(
770 events,
771 stored(2, { type: "run.updated", payload: { ...continuation, status: "completed" } }),
772 );
773 const result = yield* Fiber.join(fiber);
774
775 expect(result).toMatchObject({
776 timedOut: false,
777 run: { id: continuationId, status: "completed" },
778 });
779 }),
780);
781
782it.effect("waitForThread returns the cut run once its continuation is declined", () =>
783 Effect.gen(function* () {
784 const projectId = ProjectId.make("project:thread-management:wait-declined");
785 const threadId = ThreadId.make("thread:thread-management:wait-declined");
786 const runId = RunId.make("run:thread-management:wait-declined");
787 const subscribed = yield* Deferred.make<void>();
788 const declined = yield* Deferred.make<void>();
789 let pending = true;
790 const layerTest = ThreadManagementService.layer.pipe(
791 Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})),
792 Layer.provide(
793 Layer.mock(Orchestrator.OrchestratorV2)({
794 getThreadEventSequence: () => Effect.succeed(0),
795 restartContinuationPending: () => Effect.succeed(pending),
796 awaitRestartContinuation: () => Deferred.await(declined),
797 // Declining records no thread event.
798 streamStoredEventsFrom: (input) =>
799 input?.eventType === "run.updated"
800 ? Stream.fromEffect(Deferred.succeed(subscribed, undefined)).pipe(
801 Stream.drain,
802 Stream.concat(Stream.never),
803 )
804 : Stream.never,
805 getThreadRecords: () =>
806 Effect.succeed({
807 thread: { id: threadId, projectId, deletedAt: null },
808 runs: [{ id: runId, status: "cancelled" }],
809 } as unknown as OrchestrationV2ThreadProjection),
810 }),
811 ),
812 Layer.provide(ServerSettings.layerTest()),
813 Layer.provide(layerReceiptsTest),
814 );
815 const service = yield* ThreadManagementService.ThreadManagementService.pipe(
816 Effect.provide(layerTest),
817 );
818 const fiber = yield* service
819 .waitForThread({ projectId, threadId, runId, timeoutMs: 60 * 60 * 1_000 })
820 .pipe(Effect.forkChild);
821 yield* Deferred.await(subscribed);
822 pending = false;
823 yield* Deferred.succeed(declined, undefined);
824 const result = yield* Fiber.join(fiber);
825
826 expect(result).toMatchObject({ timedOut: false, run: { id: runId, status: "cancelled" } });
827 }),
828);
829
830it.effect.each([
831 "running",
832 "missing",
833 "completed",
834 "unsupported",
835 "missing-native-id",
836 "wrong-thread",
837 "no-session",
838 "unsupported-runtime",
839 "provider-failed",
840 "provider-timeout",
841 "within-limit",
842 "runtime-raised",
843 "interaction-raised",
844] as const)("stops a persisted workflow with state %s", (state) =>
845 Effect.gen(function* () {
846 const threadId = ThreadId.make("thread:workflow-stop");
847 const subagentId = NodeId.make("node:workflow-stop");
848 const runId = RunId.make("run:workflow-stop");
849 const providerThreadId = ProviderThreadId.make("provider-thread:workflow-stop");
850 const providerSessionId = ProviderSessionId.make("provider-session:workflow-stop");
851 const driver = ProviderDriverKind.make("claudeAgent");
852 const instanceId = ProviderInstanceId.make("claudeAgent");
853 const now = yield* DateTime.now;
854 let targetModes: Pick<OrchestrationV2ThreadShell, "runtimeMode" | "interactionMode"> = {
855 runtimeMode: "approval-required",
856 interactionMode: "plan",
857 };
858 const task: OrchestrationV2Subagent = {
859 id: subagentId,
860 threadId,
861 runId,
862 parentNodeId: NodeId.make("node:workflow-parent"),
863 origin: "provider_native",
864 createdBy: "agent",
865 driver,
866 providerInstanceId: instanceId,
867 providerThreadId: null,
868 childThreadId: null,
869 nativeTaskRef:
870 state === "missing-native-id"
871 ? null
872 : {
873 driver,
874 nativeId: "native-workflow-id",
875 strength: "strong",
876 },
877 workflow: { phases: [], agents: [] },
878 prompt: "Run the workflow",
879 title: null,
880 model: null,
881 status: state === "completed" ? "completed" : "running",
882 result: null,
883 startedAt: now,
884 completedAt: null,
885 updatedAt: now,
886 ...(state === "unsupported" ? { driver: ProviderDriverKind.make("codex") } : {}),
887 };
888 const providerThread: OrchestrationV2ProviderThread = {
889 id: providerThreadId,
890 driver,
891 providerInstanceId: instanceId,
892 providerSessionId,
893 appThreadId: state === "wrong-thread" ? ThreadId.make("other-thread") : threadId,
894 ownerNodeId: null,
895 nativeThreadRef: { driver, nativeId: "native-thread", strength: "strong" },
896 nativeConversationHeadRef: null,
897 status: "idle",
898 firstRunOrdinal: 1,
899 lastRunOrdinal: 1,
900 handoffIds: [],
901 forkedFrom: null,
902 pendingBackgroundTasks: [],
903 contextUsage: null,
904 nativeMetadata: null,
905 createdAt: now,
906 updatedAt: now,
907 };
908 const calls: Array<{ taskId: string; providerThread: OrchestrationV2ProviderThread }> = [];
909 const stopStarted = yield* Deferred.make<void>();
910 const runtime = {
911 driver,
912 providerSessionId,
913 instanceId,
914 ...(state === "unsupported-runtime"
915 ? {}
916 : {
917 stopTask: (input: (typeof calls)[number]) =>
918 state === "provider-failed"
919 ? Effect.fail(
920 new ProviderAdapter.ProviderAdapterProtocolError({
921 driver,
922 detail: "private provider detail",
923 }),
924 )
925 : state === "provider-timeout"
926 ? Deferred.succeed(stopStarted, undefined).pipe(Effect.andThen(Effect.never))
927 : Effect.sync(() => {
928 calls.push(input);
929 }),
930 }),
931 } as unknown as ProviderAdapter.ProviderAdapterV2SessionRuntime;
932 const layerTest = ThreadManagementService.layer.pipe(
933 Layer.provide(
934 Layer.mergeAll(
935 Layer.mock(Orchestrator.OrchestratorV2)({
936 getThreadRecords: () =>
937 Effect.sync(
938 () =>
939 ({
940 thread: { id: threadId, deletedAt: null, ...targetModes },
941 subagents: state === "missing" ? [] : [task],
942 runs: [{ id: runId, providerThreadId }],
943 providerThreads: [providerThread],
944 }) as unknown as OrchestrationV2ThreadProjection,
945 ),
946 }),
947 Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({
948 get: (id) => {
949 expect(id).toBe(providerSessionId);
950 return Effect.succeed(state === "no-session" ? Option.none() : Option.some(runtime));
951 },
952 }),
953 ),
954 ),
955 Layer.provide(ServerSettings.layerTest()),
956 Layer.provide(layerReceiptsTest),
957 );
958 yield* Effect.gen(function* () {
959 const service = yield* ThreadManagementService.ThreadManagementService;
960 const refused = yield* Ref.make<DispatchModeRefusal | undefined>(undefined);
961 const stop = service
962 .stopWorkflow({ threadId, subagentId })
963 .pipe(
964 Effect.provideService(
965 DispatchModeLimit,
966 state === "within-limit" || state === "runtime-raised" || state === "interaction-raised"
967 ? { runtimeMode: "approval-required", interactionMode: "plan", refused }
968 : undefined,
969 ),
970 Effect.result,
971 );
972 const executor = yield* ThreadCommandExecutor.ThreadCommandExecutor;
973 const result =
974 state === "runtime-raised" || state === "interaction-raised"
975 ? yield* Effect.gen(function* () {
976 const started = yield* Deferred.make<void>();
977 const pending = yield* executor.withLock(
978 threadId,
979 Effect.gen(function* () {
980 const fiber = yield* Deferred.succeed(started, undefined).pipe(
981 Effect.andThen(stop),
982 Effect.forkChild,
983 );
984 yield* Deferred.await(started);
985 targetModes =
986 state === "runtime-raised"
987 ? { ...targetModes, runtimeMode: "full-access" }
988 : { ...targetModes, interactionMode: "default" };
989 return fiber;
990 }),
991 );
992 return yield* Fiber.join(pending);
993 })
994 : state === "provider-timeout"
995 ? yield* Effect.gen(function* () {
996 const finished = yield* Ref.make(false);
997 const pending = yield* stop.pipe(
998 Effect.tap(() => Ref.set(finished, true)),
999 Effect.forkChild,
1000 );
1001 yield* Deferred.await(stopStarted);
1002 yield* TestClock.adjust("15 seconds");
1003 expect(yield* Ref.get(finished)).toBe(true);
1004 const result = yield* Fiber.join(pending);
1005 expect(yield* executor.withLock(threadId, Effect.succeed("lock-released"))).toBe(
1006 "lock-released",
1007 );
1008 return result;
1009 })
1010 : yield* stop;
1011 if (state === "running" || state === "within-limit") {
1012 expect(result).toMatchObject({ _tag: "Success" });
1013 expect(calls).toEqual([{ taskId: "native-workflow-id", providerThread }]);
1014 expect(task.status).toBe("running");
1015 } else {
1016 expect(result).toMatchObject({
1017 _tag: "Failure",
1018 failure: {
1019 _tag: "OrchestrationV2StopWorkflowError",
1020 threadId,
1021 subagentId,
1022 reason:
1023 state === "missing" || state === "completed"
1024 ? "not-running"
1025 : state === "unsupported" || state === "unsupported-runtime"
1026 ? "unsupported"
1027 : state === "provider-failed" || state === "provider-timeout"
1028 ? "stop-failed"
1029 : "unavailable",
1030 },
1031 });
1032 expect(calls).toEqual([]);
1033 if (state === "provider-timeout") {
1034 expect(result).toMatchObject({ failure: { cause: { _tag: "TimeoutError" } } });
1035 }
1036 if (state === "runtime-raised" || state === "interaction-raised") {
1037 expect(yield* Ref.get(refused)).toEqual({
1038 threadId,
1039 ...targetModes,
1040 mode: state === "runtime-raised" ? "runtime" : "interaction",
1041 });
1042 }
1043 }
1044 }).pipe(Effect.provide(Layer.merge(layerTest, ThreadCommandExecutor.layer)));
1045 }),
1046);
1047
1048it.effect.each([
1049 { status: "completed" as const, settles: true },
1050 { status: "failed" as const, settles: false },
1051 { status: "interrupted" as const, settles: false },
1052])("settleAfterRun settles only when the run $status", ({ status, settles }) =>
1053 Effect.gen(function* () {
1054 const projectId = ProjectId.make("project:thread-management:settle-after-run");
1055 const threadId = ThreadId.make("thread:thread-management:settle-after-run");
1056 const runId = RunId.make("run:thread-management:settle-after-run");
1057 const dispatched: Array<string> = [];
1058 const layerTest = ThreadManagementService.layer.pipe(
1059 Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})),
1060 Layer.provide(
1061 Layer.mock(Orchestrator.OrchestratorV2)({
1062 getThreadEventSequence: () => Effect.succeed(0),
1063 restartContinuationPending: () => Effect.succeed(false),
1064 getThreadRecords: () =>
1065 Effect.succeed({
1066 thread: { id: threadId, projectId, deletedAt: null },
1067 runs: [{ id: runId, status }],
1068 } as unknown as OrchestrationV2ThreadProjection),
1069 dispatch: (command) =>
1070 Effect.sync(() => {
1071 dispatched.push(command.type);
1072 return { sequence: 1, storedEvents: [] } as never;
1073 }),
1074 }),
1075 ),
1076 Layer.provide(ServerSettings.layerTest()),
1077 Layer.provide(layerReceiptsTest),
1078 );
1079 const service = yield* ThreadManagementService.ThreadManagementService.pipe(
1080 Effect.provide(layerTest),
1081 );
1082
1083 yield* service.settleAfterRun({ projectId, threadId, runId });
1084
1085 expect(dispatched).toEqual(settles ? ["thread.settle"] : []);
1086 }),
1087);