1import { expect, it } from "@effect/vitest";
2import {
3 CommandId,
4 EnvironmentId,
5 MessageId,
6 type OrchestrationV2Command,
7 type OrchestrationV2Run,
8 type OrchestrationV2ProviderThread,
9 ProviderThreadId,
10 ProviderDriverKind,
11 ProviderSessionId,
12 type OrchestrationV2StoredEvent,
13 type OrchestrationV2ThreadProjection,
14 ProjectId,
15 ProviderInstanceId,
16 RunId,
17 ThreadId,
18} from "@t3tools/contracts";
19import * as Deferred from "effect/Deferred";
20import * as DateTime from "effect/DateTime";
21import * as Duration from "effect/Duration";
22import * as Option from "effect/Option";
23import * as Effect from "effect/Effect";
24import * as Fiber from "effect/Fiber";
25import * as Layer from "effect/Layer";
26import * as Queue from "effect/Queue";
27import * as Stream from "effect/Stream";
28import * as TestClock from "effect/testing/TestClock";
29
30import { AgentScope } from "@t3tools/shared/AgentScope";
31
32import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
33import * as ServerSettings from "../serverSettings.ts";
34import * as SqlitePersistence from "../persistence/Sqlite.ts";
35import * as CommandReceiptStore from "./CommandReceiptStore.ts";
36import * as Orchestrator from "./Orchestrator.ts";
37import * as ProviderSessionManager from "./ProviderSessionManager.ts";
38import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
39import * as ThreadManagementService from "./ThreadManagementService.ts";
40import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts";
41
42it("stamps authoritative provenance on commands that create threads or messages", () => {
43 const command: OrchestrationV2Command = {
44 type: "thread.create",
45 createdBy: "agent",
46 creationSource: "mcp",
47 commandId: CommandId.make("command:thread-management:create"),
48 threadId: ThreadId.make("thread:thread-management:create"),
49 projectId: ProjectId.make("project:thread-management"),
50 title: "Thread management",
51 modelSelection: {
52 instanceId: ProviderInstanceId.make("codex"),
53 model: "gpt-5-codex",
54 },
55 runtimeMode: "full-access",
56 interactionMode: "default",
57 branch: null,
58 worktreePath: null,
59 // A client cannot claim a linked environment's origin, or a parent there.
60 linkOrigin: { sessionId: "forged", label: "Somewhere else" },
61 delegatedFrom: {
62 environmentId: EnvironmentId.make("environment:elsewhere"),
63 threadId: ThreadId.make("thread:elsewhere"),
64 title: "Somewhere else",
65 },
66 };
67
68 const stamped = ThreadManagementService.withCreationProvenance(command, {
69 createdBy: "user",
70 creationSource: "web",
71 });
72 expect(stamped).toMatchObject({
73 createdBy: "user",
74 creationSource: "web",
75 });
76 expect(stamped).not.toHaveProperty("linkOrigin");
77 expect(stamped).not.toHaveProperty("delegatedFrom");
78});
79
80const layerReceiptsTest = CommandReceiptStore.layer.pipe(
81 Layer.provide(SqlitePersistence.layerMemory),
82);
83
84const layerDispatchChokepointTest =
85 (settingsLayer: ReturnType<typeof ServerSettings.layerTest>) => (dispatched: Array<string>) =>
86 ThreadManagementService.layer.pipe(
87 Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})),
88 Layer.provide(
89 Layer.mock(Orchestrator.OrchestratorV2)({
90 getThreadRecords: () =>
91 Effect.succeed({
92 thread: {
93 id: ThreadId.make("thread:thread-management:chokepoint"),
94 projectId: ProjectId.make("project:thread-management:chokepoint"),
95 deletedAt: null,
96 modelSelection: {
97 instanceId: ProviderInstanceId.make("codex_work"),
98 model: "test",
99 },
100 },
101 } as unknown as OrchestrationV2ThreadProjection),
102 dispatch: (command) =>
103 Effect.sync(() => {
104 dispatched.push(command.type);
105 return { sequence: 1, storedEvents: [] } as never;
106 }),
107 }),
108 ),
109 Layer.provide(settingsLayer),
110 Layer.provideMerge(layerReceiptsTest),
111 );
112
113it.effect(
114 "the dispatch chokepoint refuses a continuation message.dispatch on a project-disabled instance before reaching the orchestrator",
115 () => {
116 const projectId = ProjectId.make("project:thread-management:chokepoint");
117 const threadId = ThreadId.make("thread:thread-management:chokepoint");
118 const instanceId = ProviderInstanceId.make("codex_work");
119 const commandId = CommandId.make("command:thread-management:chokepoint-disabled");
120 const dispatched: Array<string> = [];
121 const layerTest = layerDispatchChokepointTest(
122 ServerSettings.layerTest({
123 providerInstances: { [instanceId]: { driver: "codex", enabled: true } },
124 projectSettingsOverrides: {
125 [projectId]: { providerInstances: { [instanceId]: { enabled: false } } },
126 },
127 }),
128 )(dispatched);
129
130 return Effect.gen(function* () {
131 const threads = yield* ThreadManagementService.ThreadManagementService;
132 const error = yield* Effect.flip(
133 threads.dispatch({
134 type: "message.dispatch",
135 commandId,
136 threadId,
137 messageId: MessageId.make("message:thread-management:chokepoint-disabled"),
138 text: "Background task completed.",
139 attachments: [],
140 dispatchMode: { type: "queue_after_active" },
141 createdBy: "agent",
142 creationSource: "server",
143 }),
144 );
145
146 expect(error).toBeInstanceOf(
147 ThreadManagementService.ThreadManagementProviderInstanceDisabledError,
148 );
149 expect(error).toMatchObject({ projectId, instanceId });
150 expect(dispatched).toEqual([]);
151 }).pipe(Effect.provide(layerTest));
152 },
153);
154
155it.effect(
156 "a replay of an already-accepted command id dispatches despite the instance now being disabled",
157 () => {
158 const projectId = ProjectId.make("project:thread-management:chokepoint");
159 const threadId = ThreadId.make("thread:thread-management:chokepoint");
160 const instanceId = ProviderInstanceId.make("codex_work");
161 const commandId = CommandId.make("command:thread-management:chokepoint-replay");
162 const dispatched: Array<string> = [];
163 const layerTest = layerDispatchChokepointTest(
164 ServerSettings.layerTest({
165 providerInstances: { [instanceId]: { driver: "codex", enabled: true } },
166 projectSettingsOverrides: {
167 [projectId]: { providerInstances: { [instanceId]: { enabled: false } } },
168 },
169 }),
170 )(dispatched);
171
172 return Effect.gen(function* () {
173 const receipts = yield* CommandReceiptStore.CommandReceiptStoreV2;
174 yield* receipts.insertIfAbsent({
175 commandId,
176 threadId,
177 commandType: "message.dispatch",
178 acceptedAt: yield* DateTime.now,
179 resultSequence: 1,
180 status: "accepted",
181 error: null,
182 });
183
184 const threads = yield* ThreadManagementService.ThreadManagementService;
185 yield* threads.dispatch({
186 type: "message.dispatch",
187 commandId,
188 threadId,
189 messageId: MessageId.make("message:thread-management:chokepoint-replay"),
190 text: "Background task completed.",
191 attachments: [],
192 dispatchMode: { type: "queue_after_active" },
193 createdBy: "agent",
194 creationSource: "server",
195 });
196
197 expect(dispatched).toEqual(["message.dispatch"]);
198 }).pipe(Effect.provide(layerTest));
199 },
200);
201
202it.effect("waitForThread follows a restart continuation instead of the cut run", () =>
203 Effect.gen(function* () {
204 const projectId = ProjectId.make("project:thread-management:wait-restart");
205 const threadId = ThreadId.make("thread:thread-management:wait-restart");
206 const runId = RunId.make("run:thread-management:wait-restart");
207 const continuationId = RunId.make("run:thread-management:wait-restart:2");
208 const subscribed = yield* Deferred.make<void>();
209 const events = yield* Queue.unbounded<OrchestrationV2StoredEvent>();
210 let runs: Array<{
211 id: RunId;
212 status: OrchestrationV2Run["status"];
213 restartContinuationOfRunId?: RunId;
214 }> = [{ id: runId, status: "cancelled" }];
215 let pending = true;
216 const stored = (sequence: number, event: object) =>
217 ({ sequence, event: { threadId, ...event } }) as unknown as OrchestrationV2StoredEvent;
218 const layerTest = ThreadManagementService.layer.pipe(
219 Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})),
220 Layer.provide(
221 Layer.mock(Orchestrator.OrchestratorV2)({
222 getThreadEventSequence: () => Effect.succeed(0),
223 restartContinuationPending: (run) => Effect.succeed(pending && run.id === runId),
224 awaitRestartContinuation: () => Effect.never,
225 streamStoredEventsFrom: (input) =>
226 input?.eventType === "run.updated"
227 ? Stream.fromEffect(Deferred.succeed(subscribed, undefined)).pipe(
228 Stream.drain,
229 Stream.concat(Stream.fromQueue(events)),
230 )
231 : Stream.never,
232 getThreadRecords: () =>
233 Effect.sync(
234 () =>
235 ({
236 thread: { id: threadId, projectId, deletedAt: null },
237 runs,
238 }) as unknown as OrchestrationV2ThreadProjection,
239 ),
240 }),
241 ),
242 Layer.provide(ServerSettings.layerTest()),
243 Layer.provide(layerReceiptsTest),
244 );
245 const service = yield* ThreadManagementService.ThreadManagementService.pipe(
246 Effect.provide(layerTest),
247 );
248 const fiber = yield* service
249 .waitForThread({ projectId, threadId, runId, timeoutMs: 60 * 60 * 1_000 })
250 .pipe(Effect.forkChild);
251 yield* Deferred.await(subscribed);
252 const continuation = { id: continuationId, restartContinuationOfRunId: runId };
253 runs = [...runs, { ...continuation, status: "running" }];
254 pending = false;
255 yield* Queue.offer(
256 events,
257 stored(1, { type: "run.updated", payload: { ...continuation, status: "running" } }),
258 );
259 runs = [runs[0]!, { ...continuation, status: "completed" }];
260 yield* Queue.offer(
261 events,
262 stored(2, { type: "run.updated", payload: { ...continuation, status: "completed" } }),
263 );
264 const result = yield* Fiber.join(fiber);
265
266 expect(result).toMatchObject({
267 timedOut: false,
268 run: { id: continuationId, status: "completed" },
269 });
270 }),
271);
272
273it.effect("waitForThread can wait past an ended run for the thread's delegated work", () =>
274 Effect.gen(function* () {
275 const projectId = ProjectId.make("project:thread-management:wait-delegated");
276 const threadId = ThreadId.make("thread:thread-management:wait-delegated");
277 const runId = RunId.make("run:thread-management:wait-delegated");
278 const wakeRunId = RunId.make("run:thread-management:wait-delegated:wake");
279 const subscribed = yield* Deferred.make<void>();
280 const events = yield* Queue.unbounded<OrchestrationV2StoredEvent>();
281 const at = (seconds: number) => DateTime.makeUnsafe(seconds * 1_000);
282 const run = (id: RunId, ordinal: number, status: OrchestrationV2Run["status"], end: number) =>
283 ({
284 id,
285 ordinal,
286 status,
287 startedAt: at(end - 1),
288 completedAt: status === "running" ? null : at(end),
289 }) as unknown as OrchestrationV2Run;
290 // The run ended, but a task it delegated still runs and will wake the thread.
291 let runs = [run(runId, 1, "completed", 1)];
292 let subagents = [{ status: "running" }];
293 const stored = (sequence: number, event: object) =>
294 ({ sequence, event: { threadId, ...event } }) as unknown as OrchestrationV2StoredEvent;
295 const layerTest = ThreadManagementService.layer.pipe(
296 Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})),
297 Layer.provide(
298 Layer.mock(Orchestrator.OrchestratorV2)({
299 getThreadEventSequence: () => Effect.succeed(0),
300 restartContinuationPending: () => Effect.succeed(false),
301 delegatedTaskResultPending: () => Effect.succeed(false),
302 streamStoredEventsFrom: (input) =>
303 input?.eventType === "subagent.updated"
304 ? Stream.fromEffect(Deferred.succeed(subscribed, undefined)).pipe(
305 Stream.drain,
306 Stream.concat(Stream.fromQueue(events)),
307 )
308 : Stream.never,
309 getThreadRecords: () =>
310 Effect.sync(
311 () =>
312 ({
313 thread: { id: threadId, projectId, deletedAt: null },
314 runs,
315 messages: [],
316 subagents,
317 providerThreads: [],
318 }) as unknown as OrchestrationV2ThreadProjection,
319 ),
320 }),
321 ),
322 Layer.provide(ServerSettings.layerTest()),
323 Layer.provide(layerReceiptsTest),
324 );
325 const service = yield* ThreadManagementService.ThreadManagementService.pipe(
326 Effect.provide(layerTest),
327 );
328 // Without the option, the ended run is the answer, as before.
329 expect(
330 yield* service.waitForThread({ projectId, threadId, runId, timeoutMs: 60 * 60 * 1_000 }),
331 ).toMatchObject({ timedOut: false, run: { id: runId } });
332 expect(
333 yield* service.waitForThread({ projectId, threadId, runId, timeoutMs: 60 * 60 * 1_000 }),
334 ).not.toHaveProperty("resultRun");
335
336 const fiber = yield* service
337 .waitForThread({
338 projectId,
339 threadId,
340 runId,
341 timeoutMs: 60 * 60 * 1_000,
342 untilDelegatedWorkSettled: true,
343 })
344 .pipe(Effect.forkChild);
345 // A wait that returns early fails below instead of hanging here.
346 yield* Effect.raceFirst(Deferred.await(subscribed), Fiber.await(fiber));
347 // The task finishes; its delivery wakes the thread, whose next run answers.
348 subagents = [{ status: "completed" }];
349 runs = [...runs, run(wakeRunId, 2, "completed", 3)];
350 yield* Queue.offer(events, stored(1, { type: "subagent.updated", payload: {} }));
351 const result = yield* Fiber.join(fiber);
352
353 expect(result).toMatchObject({
354 timedOut: false,
355 run: { id: runId, status: "completed" },
356 resultRun: { id: wakeRunId, status: "completed" },
357 });
358 }),
359);
360
361it.effect("waitForThread settles delegated work on a thread rolled back past its only run", () =>
362 Effect.gen(function* () {
363 const projectId = ProjectId.make("project:thread-management:wait-rolled-back");
364 const threadId = ThreadId.make("thread:thread-management:wait-rolled-back");
365 const runId = RunId.make("run:thread-management:wait-rolled-back");
366 const rolledBack = {
367 id: runId,
368 ordinal: 1,
369 status: "rolled_back",
370 startedAt: DateTime.makeUnsafe(0),
371 completedAt: DateTime.makeUnsafe(1_000),
372 } as unknown as OrchestrationV2Run;
373 const layerTest = ThreadManagementService.layer.pipe(
374 Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})),
375 Layer.provide(
376 Layer.mock(Orchestrator.OrchestratorV2)({
377 getThreadEventSequence: () => Effect.succeed(0),
378 restartContinuationPending: () => Effect.succeed(false),
379 delegatedTaskResultPending: () => Effect.succeed(false),
380 streamStoredEventsFrom: () => Stream.never,
381 getThreadRecords: () =>
382 Effect.succeed({
383 thread: { id: threadId, projectId, deletedAt: null },
384 runs: [rolledBack],
385 messages: [],
386 subagents: [],
387 providerThreads: [],
388 } as unknown as OrchestrationV2ThreadProjection),
389 }),
390 ),
391 Layer.provide(ServerSettings.layerTest()),
392 Layer.provide(layerReceiptsTest),
393 );
394 const service = yield* ThreadManagementService.ThreadManagementService.pipe(
395 Effect.provide(layerTest),
396 );
397 const fiber = yield* service
398 .waitForThread({
399 projectId,
400 threadId,
401 runId,
402 timeoutMs: 60_000,
403 untilDelegatedWorkSettled: true,
404 })
405 .pipe(Effect.forkChild);
406 // Nothing will ever run there again, so a wait still open runs out here.
407 yield* TestClock.adjust(Duration.millis(60_000));
408 const result = yield* Fiber.join(fiber);
409
410 expect(result).toMatchObject({
411 timedOut: false,
412 run: { id: runId, status: "rolled_back" },
413 resultRun: { id: runId, status: "rolled_back" },
414 });
415 }),
416);
417
418it.effect("waitForThread returns the cut run once its continuation is declined", () =>
419 Effect.gen(function* () {
420 const projectId = ProjectId.make("project:thread-management:wait-declined");
421 const threadId = ThreadId.make("thread:thread-management:wait-declined");
422 const runId = RunId.make("run:thread-management:wait-declined");
423 const subscribed = yield* Deferred.make<void>();
424 const declined = yield* Deferred.make<void>();
425 let pending = true;
426 const layerTest = ThreadManagementService.layer.pipe(
427 Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})),
428 Layer.provide(
429 Layer.mock(Orchestrator.OrchestratorV2)({
430 getThreadEventSequence: () => Effect.succeed(0),
431 restartContinuationPending: () => Effect.succeed(pending),
432 awaitRestartContinuation: () => Deferred.await(declined),
433 streamStoredEventsFrom: (input) =>
434 input?.eventType === "run.updated"
435 ? Stream.fromEffect(Deferred.succeed(subscribed, undefined)).pipe(
436 Stream.drain,
437 Stream.concat(Stream.never),
438 )
439 : Stream.never,
440 getThreadRecords: () =>
441 Effect.succeed({
442 thread: { id: threadId, projectId, deletedAt: null },
443 runs: [{ id: runId, status: "cancelled" }],
444 } as unknown as OrchestrationV2ThreadProjection),
445 }),
446 ),
447 Layer.provide(ServerSettings.layerTest()),
448 Layer.provide(layerReceiptsTest),
449 );
450 const service = yield* ThreadManagementService.ThreadManagementService.pipe(
451 Effect.provide(layerTest),
452 );
453 const fiber = yield* service
454 .waitForThread({ projectId, threadId, runId, timeoutMs: 60 * 60 * 1_000 })
455 .pipe(Effect.forkChild);
456 yield* Deferred.await(subscribed);
457 pending = false;
458 yield* Deferred.succeed(declined, undefined);
459 const result = yield* Fiber.join(fiber);
460
461 expect(result).toMatchObject({ timedOut: false, run: { id: runId, status: "cancelled" } });
462 }),
463);
464
465it.effect.each(["live", "no-session", "unsupported-runtime"] as const)(
466 "marks background work awaited on the thread's active provider thread with state %s",
467 (state) =>
468 Effect.gen(function* () {
469 const threadId = ThreadId.make("thread:background-wait");
470 const providerThreadId = ProviderThreadId.make("provider-thread:background-wait");
471 const providerSessionId = ProviderSessionId.make("provider-session:background-wait");
472 const driver = ProviderDriverKind.make("claudeAgent");
473 const now = yield* DateTime.now;
474 const providerThread: OrchestrationV2ProviderThread = {
475 id: providerThreadId,
476 driver,
477 providerInstanceId: ProviderInstanceId.make("claudeAgent"),
478 providerSessionId,
479 appThreadId: threadId,
480 ownerNodeId: null,
481 nativeThreadRef: { driver, nativeId: "native-thread", strength: "strong" },
482 nativeConversationHeadRef: null,
483 status: "idle",
484 firstRunOrdinal: 1,
485 lastRunOrdinal: 1,
486 handoffIds: [],
487 forkedFrom: null,
488 pendingBackgroundTasks: [{ taskId: "tests", kind: "command" }],
489 contextUsage: null,
490 nativeMetadata: null,
491 createdAt: now,
492 updatedAt: now,
493 };
494 const calls: Array<{ providerThreadId: ProviderThreadId; awaited: boolean }> = [];
495 const runtime = {
496 driver,
497 providerSessionId,
498 ...(state === "unsupported-runtime"
499 ? {}
500 : {
501 awaitBackgroundTasks: (input: {
502 readonly providerThread: OrchestrationV2ProviderThread;
503 readonly taskIds: ReadonlyArray<string>;
504 readonly awaited: boolean;
505 }) =>
506 Effect.sync(() => {
507 calls.push({
508 providerThreadId: input.providerThread.id,
509 awaited: input.awaited,
510 });
511 return { matched: input.taskIds.filter((taskId) => taskId === "tests") };
512 }),
513 }),
514 } as unknown as ProviderAdapter.ProviderAdapterV2SessionRuntime;
515 const layerTest = ThreadManagementService.layer.pipe(
516 Layer.provide(
517 Layer.mergeAll(
518 Layer.mock(Orchestrator.OrchestratorV2)({
519 getThreadRecords: () =>
520 Effect.succeed({
521 thread: {
522 id: threadId,
523 deletedAt: null,
524 activeProviderThreadId: providerThreadId,
525 },
526 providerThreads: [providerThread],
527 } as unknown as OrchestrationV2ThreadProjection),
528 }),
529 Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({
530 get: () =>
531 Effect.succeed(state === "no-session" ? Option.none() : Option.some(runtime)),
532 }),
533 ),
534 ),
535 Layer.provide(ServerSettings.layerTest()),
536 Layer.provide(layerReceiptsTest),
537 );
538 const result = yield* Effect.gen(function* () {
539 const service = yield* ThreadManagementService.ThreadManagementService;
540 return yield* service
541 .awaitBackgroundTasks({ threadId, taskIds: ["tests", "finished"], awaited: true })
542 .pipe(Effect.result);
543 }).pipe(Effect.provide(Layer.merge(layerTest, ThreadCommandExecutor.layer)));
544
545 if (state === "live") {
546 expect(result).toMatchObject({
547 success: { matched: ["tests"], unmatched: ["finished"] },
548 });
549 expect(calls).toEqual([{ providerThreadId, awaited: true }]);
550 } else {
551 expect(result).toMatchObject({
552 failure: {
553 _tag: "ThreadManagementBackgroundWaitError",
554 reason: state === "no-session" ? "unavailable" : "unsupported",
555 },
556 });
557 expect(calls).toEqual([]);
558 }
559 }),
560);
561
562it.effect.each([
563 { status: "completed" as const, followUp: "none", settles: true },
564 { status: "failed" as const, followUp: "none", settles: false },
565 { status: "interrupted" as const, followUp: "none", settles: false },
566 { status: "completed" as const, followUp: "notification", settles: true },
567 { status: "completed" as const, followUp: "replayed-notification", settles: true },
568 { status: "completed" as const, followUp: "delegated", settles: true },
569 { status: "completed" as const, followUp: "failed-notification", settles: false },
570 { status: "completed" as const, followUp: "held-notification", settles: true },
571 { status: "completed" as const, followUp: "user", settles: false },
572 { status: "completed" as const, followUp: "steered-notification", settles: false },
573])("settleAfterRun handles $status with $followUp follow-up", ({ status, followUp, settles }) =>
574 Effect.gen(function* () {
575 const projectId = ProjectId.make("project:thread-management:settle-after-run");
576 const threadId = ThreadId.make("thread:thread-management:settle-after-run");
577 const runId = RunId.make("run:thread-management:settle-after-run");
578 const followUpId = RunId.make(`${runId}:follow-up`);
579 const subscribed = yield* Deferred.make<void>();
580 const events = yield* Queue.unbounded<OrchestrationV2StoredEvent>();
581 const replayRejected = yield* Deferred.make<void>();
582 let sequence = 0;
583 let sequenceReads = 0;
584 let rejectedCommandId: CommandId | undefined;
585 const runs = [{ id: runId, ordinal: 1, status, userMessageId: MessageId.make("original") }];
586 const nextRun = {
587 id: followUpId,
588 ordinal: 2,
589 status: "running" as OrchestrationV2Run["status"],
590 userMessageId: MessageId.make("follow-up"),
591 };
592 const dispatched: Array<string> = [];
593 // The notification starts after the completed run was observed, before settlement commits.
594 let raced = false;
595 const layerTest = ThreadManagementService.layer.pipe(
596 Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})),
597 Layer.provide(
598 Layer.mock(Orchestrator.OrchestratorV2)({
599 getThreadEventSequence: () =>
600 Effect.sync(() => {
601 if (followUp === "replayed-notification" && sequenceReads++ === 2) sequence = 1;
602 return sequence;
603 }),
604 restartContinuationPending: () => Effect.succeed(false),
605 streamStoredEventsFrom: () =>
606 Stream.fromEffect(Deferred.succeed(subscribed, undefined)).pipe(
607 Stream.drain,
608 Stream.concat(Stream.fromQueue(events)),
609 ),
610 getThreadRecords: (_threadId, _fields, filter) =>
611 Effect.sync(
612 () =>
613 ({
614 thread: { id: threadId, projectId, deletedAt: null },
615 runs: raced ? [...runs, nextRun] : runs,
616 messages: [
617 { id: MessageId.make("original"), runId, role: "user" as const },
618 {
619 id: nextRun.userMessageId,
620 runId: followUpId,
621 role: "user" as const,
622 ...(followUp.endsWith("notification") ? { notification: {} } : {}),
623 ...(followUp === "delegated" ? { delegatedCompletion: {} } : {}),
624 },
625 { id: MessageId.make("reply"), runId: followUpId, role: "assistant" as const },
626 ...(followUp === "steered-notification"
627 ? [
628 {
629 id: MessageId.make("user-steer"),
630 runId: followUpId,
631 role: "user" as const,
632 },
633 ]
634 : []),
635 ].filter(
636 (message) =>
637 (filter?.messageIds === undefined ||
638 filter.messageIds.includes(message.id)) &&
639 (filter?.messageRunIds === undefined ||
640 filter.messageRunIds.includes(message.runId)) &&
641 (filter?.messageRoles === undefined ||
642 filter.messageRoles.includes(message.role)),
643 ),
644 }) as unknown as OrchestrationV2ThreadProjection,
645 ),
646 dispatch: (command) =>
647 Effect.gen(function* () {
648 if (followUp !== "none" && !raced) {
649 raced = true;
650 rejectedCommandId = command.commandId;
651 return yield* new Orchestrator.OrchestratorDispatchError({
652 commandId: command.commandId,
653 commandType: command.type,
654 cause: "Thread is still running",
655 });
656 }
657 if (followUp === "replayed-notification" && command.commandId === rejectedCommandId) {
658 yield* Deferred.succeed(replayRejected, undefined);
659 return yield* new Orchestrator.OrchestratorCommandPreviouslyRejectedError({
660 commandId: command.commandId,
661 commandType: command.type,
662 detail: "Thread is still running",
663 });
664 }
665 dispatched.push(command.type);
666 return { sequence: 1, storedEvents: [] } as never;
667 }),
668 }),
669 ),
670 Layer.provide(ServerSettings.layerTest()),
671 Layer.provide(layerReceiptsTest),
672 );
673 const service = yield* ThreadManagementService.ThreadManagementService.pipe(
674 Effect.provide(layerTest),
675 );
676 const fiber = yield* service
677 .settleAfterRun({ projectId, threadId, runId })
678 .pipe(Effect.forkChild);
679 if (followUp !== "none") {
680 yield* Deferred.await(subscribed);
681 expect(dispatched).toEqual([]);
682 nextRun.status =
683 followUp === "failed-notification"
684 ? "failed"
685 : followUp === "held-notification"
686 ? "queued"
687 : "completed";
688 yield* Queue.offer(events, {
689 sequence: 1,
690 event: { type: "run.updated", threadId, payload: nextRun },
691 } as unknown as OrchestrationV2StoredEvent);
692 }
693 if (followUp === "replayed-notification") {
694 yield* Deferred.await(replayRejected);
695 expect(dispatched).toEqual([]);
696 sequence = 2;
697 yield* Queue.offer(events, {
698 sequence,
699 event: { type: "run.updated", threadId, payload: nextRun },
700 } as unknown as OrchestrationV2StoredEvent);
701 }
702 yield* Fiber.join(fiber);
703 expect(dispatched).toEqual(settles ? ["thread.settle"] : []);
704 }),
705);
706
707it.effect.each([
708 {
709 name: "kills the thread's own agent and ends its turn",
710 others: [] as ReadonlyArray<string>,
711 killed: true,
712 expected: { outcome: "killed", sharedWithThreadIds: [] },
713 teardown: ["release"],
714 },
715 {
716 name: "keeps a process another working thread shares and only ends this turn",
717 others: ["thread:thread-management:force-stop:other"],
718 killed: false,
719 expected: {
720 outcome: "shared",
721 sharedWithThreadIds: ["thread:thread-management:force-stop:other"],
722 },
723 teardown: ["detach"],
724 },
725 {
726 name: "still settles a finished run's background work in a shared process",
727 others: ["thread:thread-management:force-stop:other"],
728 killed: false,
729 settledRun: true,
730 expected: {
731 outcome: "shared",
732 sharedWithThreadIds: ["thread:thread-management:force-stop:other"],
733 },
734 teardown: ["detach"],
735 },
736 {
737 name: "detaches when a sibling starts a turn just before the session is released",
738 others: [] as ReadonlyArray<string>,
739 killed: false,
740 siblingBusy: true,
741 expected: { outcome: "shared", sharedWithThreadIds: [] },
742 teardown: ["release", "detach"],
743 },
744 {
745 name: "reports a teardown that has not finished in time as still releasing",
746 others: [] as ReadonlyArray<string>,
747 killed: false,
748 releaseHangs: true,
749 expected: { outcome: "releasing", sharedWithThreadIds: [] },
750 teardown: ["release"],
751 },
752])("Force stop $name", ({ others, killed, expected, teardown, ...row }) =>
753 Effect.gen(function* () {
754 const threadId = ThreadId.make("thread:thread-management:force-stop");
755 const providerSessionId = ProviderSessionId.make("provider-session:force-stop");
756 const providerThreadId = ProviderThreadId.make("provider-thread:force-stop");
757 const runId = RunId.make("run:thread-management:force-stop");
758 const dispatched: Array<OrchestrationV2Command | { readonly type: string }> = [];
759 const teardowns: Array<string> = [];
760 const tornDown = yield* Deferred.make<void>();
761 const layerTest = ThreadManagementService.layer.pipe(
762 Layer.provide(
763 Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({
764 isLive: () => Effect.succeed(true),
765 otherWorkingThreads: () => Effect.succeed(others.map((id) => ThreadId.make(id))),
766 releaseUnlessBusy: () =>
767 Effect.sync(() => teardowns.push("release")).pipe(
768 Effect.andThen(
769 "releaseHangs" in row ? Effect.never : Effect.succeed(!("siblingBusy" in row)),
770 ),
771 ),
772 detach: () =>
773 Effect.sync(() => teardowns.push("detach")).pipe(
774 Effect.andThen(Deferred.succeed(tornDown, undefined)),
775 Effect.asVoid,
776 ),
777 }),
778 ),
779 Layer.provide(
780 Layer.mock(Orchestrator.OrchestratorV2)({
781 getThreadRecords: () =>
782 Effect.succeed({
783 thread: { id: threadId, deletedAt: null },
784 runs: [
785 {
786 id: runId,
787 ordinal: 1,
788 status: "settledRun" in row ? "completed" : "running",
789 activeAttemptId: "attempt",
790 },
791 // A queued follow-up neither supersedes the request nor becomes the target.
792 { id: RunId.make("run:queued-follow-up"), ordinal: 2, status: "queued" },
793 ],
794 providerThreads: [{ id: providerThreadId, appThreadId: threadId, providerSessionId }],
795 providerTurns: [
796 {
797 id: "provider-turn:force-stop",
798 providerThreadId,
799 runAttemptId: "attempt",
800 status: "settledRun" in row ? "completed" : "running",
801 },
802 ],
803 } as unknown as OrchestrationV2ThreadProjection),
804 dispatch: (command) =>
805 Effect.sync(() => {
806 dispatched.push(command);
807 return { sequence: 1, storedEvents: [] } as never;
808 }),
809 }),
810 ),
811 Layer.provide(
812 Layer.succeed(AgentScope, {
813 ...AgentScope.defaultValue(),
814 kill: () => Effect.succeed(killed),
815 }),
816 ),
817 Layer.provide(ServerSettings.layerTest()),
818 Layer.provide(layerReceiptsTest),
819 );
820 yield* Effect.gen(function* () {
821 const service = yield* ThreadManagementService.ThreadManagementService;
822 const stopping = yield* service.forceStopThread({ threadId, runId }).pipe(Effect.forkChild);
823 yield* TestClock.adjust("10 seconds");
824 expect(yield* Fiber.join(stopping)).toEqual(expected);
825 expect(dispatched).toMatchObject([
826 {
827 type: "thread.background-work.settle",
828 providerThreadId,
829 reason: "Force stopped by user",
830 },
831 ]);
832 if (teardown.at(-1) === "detach") yield* Deferred.await(tornDown);
833 expect(teardowns).toEqual(teardown);
834 }).pipe(Effect.provide(layerTest));
835 }),
836);
837
838it.effect("Force stop refuses once a newer run has started, before killing anything", () =>
839 Effect.gen(function* () {
840 const threadId = ThreadId.make("thread:thread-management:force-stop-superseded");
841 let killed = false;
842 const layerTest = ThreadManagementService.layer.pipe(
843 Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})),
844 Layer.provide(
845 Layer.mock(Orchestrator.OrchestratorV2)({
846 getThreadRecords: () =>
847 Effect.succeed({
848 thread: { id: threadId, deletedAt: null },
849 runs: [
850 { id: RunId.make("run:stopped"), ordinal: 1, status: "running" },
851 { id: RunId.make("run:newer"), ordinal: 2, status: "running" },
852 ],
853 providerThreads: [],
854 providerTurns: [],
855 } as unknown as OrchestrationV2ThreadProjection),
856 }),
857 ),
858 Layer.provide(
859 Layer.succeed(AgentScope, {
860 ...AgentScope.defaultValue(),
861 kill: () => Effect.sync(() => (killed = true)),
862 }),
863 ),
864 Layer.provide(ServerSettings.layerTest()),
865 Layer.provide(layerReceiptsTest),
866 );
867 const error = yield* Effect.gen(function* () {
868 const service = yield* ThreadManagementService.ThreadManagementService;
869 return yield* Effect.flip(
870 service.forceStopThread({ threadId, runId: RunId.make("run:stopped") }),
871 );
872 }).pipe(Effect.provide(layerTest));
873 expect(error.reason).toBe("superseded");
874 expect(killed).toBe(false);
875 }),
876);
877
878it.effect(
879 "admits two transcript exports at a time and frees a slot on failure or interruption",
880 () =>
881 Effect.gen(function* () {
882 const started = yield* Queue.unbounded<string>();
883 const gates = new Map(
884 ["a", "b", "c", "d", "e", "f"].map((id) => [
885 id,
886 Deferred.makeUnsafe<void, Orchestrator.OrchestratorV2Error>(),
887 ]),
888 );
889 const layerTest = ThreadManagementService.layer.pipe(
890 Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})),
891 Layer.provide(
892 Layer.mock(Orchestrator.OrchestratorV2)({
893 getTranscriptByteFloor: () => Effect.succeed(0),
894 getThreadSnapshot: (threadId) =>
895 Queue.offer(started, threadId).pipe(
896 Effect.andThen(Deferred.await(gates.get(threadId)!)),
897 Effect.as({
898 schemaVersion: 1,
899 snapshotSequence: 1,
900 projection: {
901 thread: { id: threadId, title: threadId },
902 updatedAt: DateTime.makeUnsafe(0),
903 visibleTurnItems: [],
904 } as unknown as OrchestrationV2ThreadProjection,
905 }),
906 ),
907 }),
908 ),
909 Layer.provide(ServerSettings.layerTest()),
910 Layer.provide(layerReceiptsTest),
911 );
912
913 yield* Effect.gen(function* () {
914 const threads = yield* ThreadManagementService.ThreadManagementService;
915 const exportThread = (id: string) =>
916 threads.getThreadTranscript(ThreadId.make(id)).pipe(
917 Effect.provideService(ServerEnvironment.ServerEnvironmentIdentity, {
918 getEnvironmentId: Effect.succeed(EnvironmentId.make("environment:export")),
919 }),
920 Effect.forkChild({ startImmediately: true }),
921 );
922 const a = yield* exportThread("a");
923 const b = yield* exportThread("b");
924 const c = yield* exportThread("c");
925 const d = yield* exportThread("d");
926 const e = yield* exportThread("e");
927 const f = yield* exportThread("f");
928 expect(yield* Queue.takeAll(started)).toEqual(["a", "b"]);
929
930 yield* Fiber.interrupt(c);
931 yield* Deferred.fail(
932 gates.get("a")!,
933 new Orchestrator.OrchestratorProjectionError({ threadId: ThreadId.make("a"), cause: "" }),
934 );
935 expect(yield* Queue.take(started)).toBe("d");
936 yield* Fiber.interrupt(b);
937 expect(yield* Queue.take(started)).toBe("e");
938 yield* Deferred.succeed(gates.get("d")!, undefined);
939 expect(yield* Queue.take(started)).toBe("f");
940 yield* Deferred.succeed(gates.get("e")!, undefined);
941 yield* Deferred.succeed(gates.get("f")!, undefined);
942
943 expect((yield* Fiber.await(a))._tag).toBe("Failure");
944 expect((yield* Fiber.join(d)).threadId).toBe("d");
945 expect((yield* Fiber.join(e)).threadId).toBe("e");
946 expect((yield* Fiber.join(f)).threadId).toBe("f");
947 expect(yield* Queue.size(started)).toBe(0);
948 }).pipe(Effect.provide(layerTest));
949 }),
950);