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([
466 "live",
467 "no-match",
468 "no-match-clear",
469 "no-session",
470 "unsupported-runtime",
471] as const)(
472 "marks background work awaited on the thread's active provider thread with state %s",
473 (state) =>
474 Effect.gen(function* () {
475 const threadId = ThreadId.make("thread:background-wait");
476 const providerThreadId = ProviderThreadId.make("provider-thread:background-wait");
477 const providerSessionId = ProviderSessionId.make("provider-session:background-wait");
478 const driver = ProviderDriverKind.make("claudeAgent");
479 const now = yield* DateTime.now;
480 const providerThread: OrchestrationV2ProviderThread = {
481 id: providerThreadId,
482 driver,
483 providerInstanceId: ProviderInstanceId.make("claudeAgent"),
484 providerSessionId,
485 appThreadId: threadId,
486 ownerNodeId: null,
487 nativeThreadRef: { driver, nativeId: "native-thread", strength: "strong" },
488 nativeConversationHeadRef: null,
489 status: "idle",
490 firstRunOrdinal: 1,
491 lastRunOrdinal: 1,
492 handoffIds: [],
493 forkedFrom: null,
494 pendingBackgroundTasks: [{ taskId: "tests", kind: "command" }],
495 contextUsage: null,
496 nativeMetadata: null,
497 createdAt: now,
498 updatedAt: now,
499 };
500 const calls: Array<{ providerThreadId: ProviderThreadId; awaited: boolean }> = [];
501 const runtime = {
502 driver,
503 providerSessionId,
504 ...(state === "unsupported-runtime"
505 ? {}
506 : {
507 awaitBackgroundTasks: (input: {
508 readonly providerThread: OrchestrationV2ProviderThread;
509 readonly taskIds: ReadonlyArray<string>;
510 readonly awaited: boolean;
511 }) =>
512 Effect.sync(() => {
513 calls.push({
514 providerThreadId: input.providerThread.id,
515 awaited: input.awaited,
516 });
517 return { matched: input.taskIds.filter((taskId) => taskId === "tests") };
518 }),
519 }),
520 } as unknown as ProviderAdapter.ProviderAdapterV2SessionRuntime;
521 const layerTest = ThreadManagementService.layer.pipe(
522 Layer.provide(
523 Layer.mergeAll(
524 Layer.mock(Orchestrator.OrchestratorV2)({
525 getThreadRecords: () =>
526 Effect.succeed({
527 thread: {
528 id: threadId,
529 deletedAt: null,
530 activeProviderThreadId: providerThreadId,
531 },
532 providerThreads: [providerThread],
533 } as unknown as OrchestrationV2ThreadProjection),
534 }),
535 Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({
536 get: () =>
537 Effect.succeed(state === "no-session" ? Option.none() : Option.some(runtime)),
538 }),
539 ),
540 ),
541 Layer.provide(ServerSettings.layerTest()),
542 Layer.provide(layerReceiptsTest),
543 );
544 const result = yield* Effect.gen(function* () {
545 const service = yield* ThreadManagementService.ThreadManagementService;
546 return yield* service
547 .awaitBackgroundTasks({
548 threadId,
549 taskIds: state.startsWith("no-match") ? ["finished"] : ["tests", "finished"],
550 awaited: state !== "no-match-clear",
551 })
552 .pipe(Effect.result);
553 }).pipe(Effect.provide(Layer.merge(layerTest, ThreadCommandExecutor.layer)));
554
555 if (state === "live") {
556 expect(result).toMatchObject({
557 success: { matched: ["tests"], unmatched: ["finished"] },
558 });
559 expect(calls).toEqual([{ providerThreadId, awaited: true }]);
560 } else if (state === "no-match") {
561 expect(result).toMatchObject({
562 failure: { _tag: "ThreadManagementBackgroundWaitError", reason: "unmatched" },
563 });
564 expect(calls).toEqual([{ providerThreadId, awaited: true }]);
565 } else if (state === "no-match-clear") {
566 expect(result).toMatchObject({ success: { matched: [], unmatched: ["finished"] } });
567 expect(calls).toEqual([{ providerThreadId, awaited: false }]);
568 } else {
569 expect(result).toMatchObject({
570 failure: {
571 _tag: "ThreadManagementBackgroundWaitError",
572 reason: state === "no-session" ? "unavailable" : "unsupported",
573 },
574 });
575 expect(calls).toEqual([]);
576 }
577 }),
578);
579
580it.effect.each([
581 { status: "completed" as const, followUp: "none", settles: true },
582 { status: "failed" as const, followUp: "none", settles: false },
583 { status: "interrupted" as const, followUp: "none", settles: false },
584 { status: "completed" as const, followUp: "notification", settles: true },
585 { status: "completed" as const, followUp: "replayed-notification", settles: true },
586 { status: "completed" as const, followUp: "delegated", settles: true },
587 { status: "completed" as const, followUp: "failed-notification", settles: false },
588 { status: "completed" as const, followUp: "held-notification", settles: true },
589 { status: "completed" as const, followUp: "user", settles: false },
590 { status: "completed" as const, followUp: "steered-notification", settles: false },
591])("settleAfterRun handles $status with $followUp follow-up", ({ status, followUp, settles }) =>
592 Effect.gen(function* () {
593 const projectId = ProjectId.make("project:thread-management:settle-after-run");
594 const threadId = ThreadId.make("thread:thread-management:settle-after-run");
595 const runId = RunId.make("run:thread-management:settle-after-run");
596 const followUpId = RunId.make(`${runId}:follow-up`);
597 const subscribed = yield* Deferred.make<void>();
598 const events = yield* Queue.unbounded<OrchestrationV2StoredEvent>();
599 const replayRejected = yield* Deferred.make<void>();
600 let sequence = 0;
601 let sequenceReads = 0;
602 let rejectedCommandId: CommandId | undefined;
603 const runs = [{ id: runId, ordinal: 1, status, userMessageId: MessageId.make("original") }];
604 const nextRun = {
605 id: followUpId,
606 ordinal: 2,
607 status: "running" as OrchestrationV2Run["status"],
608 userMessageId: MessageId.make("follow-up"),
609 };
610 const dispatched: Array<string> = [];
611 // The notification starts after the completed run was observed, before settlement commits.
612 let raced = false;
613 const layerTest = ThreadManagementService.layer.pipe(
614 Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})),
615 Layer.provide(
616 Layer.mock(Orchestrator.OrchestratorV2)({
617 getThreadEventSequence: () =>
618 Effect.sync(() => {
619 if (followUp === "replayed-notification" && sequenceReads++ === 2) sequence = 1;
620 return sequence;
621 }),
622 restartContinuationPending: () => Effect.succeed(false),
623 streamStoredEventsFrom: () =>
624 Stream.fromEffect(Deferred.succeed(subscribed, undefined)).pipe(
625 Stream.drain,
626 Stream.concat(Stream.fromQueue(events)),
627 ),
628 getThreadRecords: (_threadId, _fields, filter) =>
629 Effect.sync(
630 () =>
631 ({
632 thread: { id: threadId, projectId, deletedAt: null },
633 runs: raced ? [...runs, nextRun] : runs,
634 messages: [
635 { id: MessageId.make("original"), runId, role: "user" as const },
636 {
637 id: nextRun.userMessageId,
638 runId: followUpId,
639 role: "user" as const,
640 ...(followUp.endsWith("notification") ? { notification: {} } : {}),
641 ...(followUp === "delegated" ? { delegatedCompletion: {} } : {}),
642 },
643 { id: MessageId.make("reply"), runId: followUpId, role: "assistant" as const },
644 ...(followUp === "steered-notification"
645 ? [
646 {
647 id: MessageId.make("user-steer"),
648 runId: followUpId,
649 role: "user" as const,
650 },
651 ]
652 : []),
653 ].filter(
654 (message) =>
655 (filter?.messageIds === undefined ||
656 filter.messageIds.includes(message.id)) &&
657 (filter?.messageRunIds === undefined ||
658 filter.messageRunIds.includes(message.runId)) &&
659 (filter?.messageRoles === undefined ||
660 filter.messageRoles.includes(message.role)),
661 ),
662 }) as unknown as OrchestrationV2ThreadProjection,
663 ),
664 dispatch: (command) =>
665 Effect.gen(function* () {
666 if (followUp !== "none" && !raced) {
667 raced = true;
668 rejectedCommandId = command.commandId;
669 return yield* new Orchestrator.OrchestratorDispatchError({
670 commandId: command.commandId,
671 commandType: command.type,
672 cause: "Thread is still running",
673 });
674 }
675 if (followUp === "replayed-notification" && command.commandId === rejectedCommandId) {
676 yield* Deferred.succeed(replayRejected, undefined);
677 return yield* new Orchestrator.OrchestratorCommandPreviouslyRejectedError({
678 commandId: command.commandId,
679 commandType: command.type,
680 detail: "Thread is still running",
681 });
682 }
683 dispatched.push(command.type);
684 return { sequence: 1, storedEvents: [] } as never;
685 }),
686 }),
687 ),
688 Layer.provide(ServerSettings.layerTest()),
689 Layer.provide(layerReceiptsTest),
690 );
691 const service = yield* ThreadManagementService.ThreadManagementService.pipe(
692 Effect.provide(layerTest),
693 );
694 const fiber = yield* service
695 .settleAfterRun({ projectId, threadId, runId })
696 .pipe(Effect.forkChild);
697 if (followUp !== "none") {
698 yield* Deferred.await(subscribed);
699 expect(dispatched).toEqual([]);
700 nextRun.status =
701 followUp === "failed-notification"
702 ? "failed"
703 : followUp === "held-notification"
704 ? "queued"
705 : "completed";
706 yield* Queue.offer(events, {
707 sequence: 1,
708 event: { type: "run.updated", threadId, payload: nextRun },
709 } as unknown as OrchestrationV2StoredEvent);
710 }
711 if (followUp === "replayed-notification") {
712 yield* Deferred.await(replayRejected);
713 expect(dispatched).toEqual([]);
714 sequence = 2;
715 yield* Queue.offer(events, {
716 sequence,
717 event: { type: "run.updated", threadId, payload: nextRun },
718 } as unknown as OrchestrationV2StoredEvent);
719 }
720 yield* Fiber.join(fiber);
721 expect(dispatched).toEqual(settles ? ["thread.settle"] : []);
722 }),
723);
724
725it.effect.each([
726 {
727 name: "kills the thread's own agent and ends its turn",
728 others: [] as ReadonlyArray<string>,
729 killed: true,
730 expected: { outcome: "killed", sharedWithThreadIds: [] },
731 teardown: ["release"],
732 },
733 {
734 name: "keeps a process another working thread shares and only ends this turn",
735 others: ["thread:thread-management:force-stop:other"],
736 killed: false,
737 expected: {
738 outcome: "shared",
739 sharedWithThreadIds: ["thread:thread-management:force-stop:other"],
740 },
741 teardown: ["detach"],
742 },
743 {
744 name: "still settles a finished run's background work in a shared process",
745 others: ["thread:thread-management:force-stop:other"],
746 killed: false,
747 settledRun: true,
748 expected: {
749 outcome: "shared",
750 sharedWithThreadIds: ["thread:thread-management:force-stop:other"],
751 },
752 teardown: ["detach"],
753 },
754 {
755 name: "detaches when a sibling starts a turn just before the session is released",
756 others: [] as ReadonlyArray<string>,
757 killed: false,
758 siblingBusy: true,
759 expected: { outcome: "shared", sharedWithThreadIds: [] },
760 teardown: ["release", "detach"],
761 },
762 {
763 name: "reports a teardown that has not finished in time as still releasing",
764 others: [] as ReadonlyArray<string>,
765 killed: false,
766 releaseHangs: true,
767 expected: { outcome: "releasing", sharedWithThreadIds: [] },
768 teardown: ["release"],
769 },
770])("Force stop $name", ({ others, killed, expected, teardown, ...row }) =>
771 Effect.gen(function* () {
772 const threadId = ThreadId.make("thread:thread-management:force-stop");
773 const providerSessionId = ProviderSessionId.make("provider-session:force-stop");
774 const providerThreadId = ProviderThreadId.make("provider-thread:force-stop");
775 const runId = RunId.make("run:thread-management:force-stop");
776 const dispatched: Array<OrchestrationV2Command | { readonly type: string }> = [];
777 const teardowns: Array<string> = [];
778 const tornDown = yield* Deferred.make<void>();
779 const layerTest = ThreadManagementService.layer.pipe(
780 Layer.provide(
781 Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({
782 isLive: () => Effect.succeed(true),
783 otherWorkingThreads: () => Effect.succeed(others.map((id) => ThreadId.make(id))),
784 releaseUnlessBusy: () =>
785 Effect.sync(() => teardowns.push("release")).pipe(
786 Effect.andThen(
787 "releaseHangs" in row ? Effect.never : Effect.succeed(!("siblingBusy" in row)),
788 ),
789 ),
790 detach: () =>
791 Effect.sync(() => teardowns.push("detach")).pipe(
792 Effect.andThen(Deferred.succeed(tornDown, undefined)),
793 Effect.asVoid,
794 ),
795 }),
796 ),
797 Layer.provide(
798 Layer.mock(Orchestrator.OrchestratorV2)({
799 getThreadRecords: () =>
800 Effect.succeed({
801 thread: { id: threadId, deletedAt: null },
802 runs: [
803 {
804 id: runId,
805 ordinal: 1,
806 status: "settledRun" in row ? "completed" : "running",
807 activeAttemptId: "attempt",
808 },
809 // A queued follow-up neither supersedes the request nor becomes the target.
810 { id: RunId.make("run:queued-follow-up"), ordinal: 2, status: "queued" },
811 ],
812 providerThreads: [{ id: providerThreadId, appThreadId: threadId, providerSessionId }],
813 providerTurns: [
814 {
815 id: "provider-turn:force-stop",
816 providerThreadId,
817 runAttemptId: "attempt",
818 status: "settledRun" in row ? "completed" : "running",
819 },
820 ],
821 } as unknown as OrchestrationV2ThreadProjection),
822 dispatch: (command) =>
823 Effect.sync(() => {
824 dispatched.push(command);
825 return { sequence: 1, storedEvents: [] } as never;
826 }),
827 }),
828 ),
829 Layer.provide(
830 Layer.succeed(AgentScope, {
831 ...AgentScope.defaultValue(),
832 kill: () => Effect.succeed(killed),
833 }),
834 ),
835 Layer.provide(ServerSettings.layerTest()),
836 Layer.provide(layerReceiptsTest),
837 );
838 yield* Effect.gen(function* () {
839 const service = yield* ThreadManagementService.ThreadManagementService;
840 const stopping = yield* service.forceStopThread({ threadId, runId }).pipe(Effect.forkChild);
841 yield* TestClock.adjust("10 seconds");
842 expect(yield* Fiber.join(stopping)).toEqual(expected);
843 expect(dispatched).toMatchObject([
844 {
845 type: "thread.background-work.settle",
846 providerThreadId,
847 reason: "Force stopped by user",
848 },
849 ]);
850 if (teardown.at(-1) === "detach") yield* Deferred.await(tornDown);
851 expect(teardowns).toEqual(teardown);
852 }).pipe(Effect.provide(layerTest));
853 }),
854);
855
856it.effect("Force stop refuses once a newer run has started, before killing anything", () =>
857 Effect.gen(function* () {
858 const threadId = ThreadId.make("thread:thread-management:force-stop-superseded");
859 let killed = false;
860 const layerTest = ThreadManagementService.layer.pipe(
861 Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})),
862 Layer.provide(
863 Layer.mock(Orchestrator.OrchestratorV2)({
864 getThreadRecords: () =>
865 Effect.succeed({
866 thread: { id: threadId, deletedAt: null },
867 runs: [
868 { id: RunId.make("run:stopped"), ordinal: 1, status: "running" },
869 { id: RunId.make("run:newer"), ordinal: 2, status: "running" },
870 ],
871 providerThreads: [],
872 providerTurns: [],
873 } as unknown as OrchestrationV2ThreadProjection),
874 }),
875 ),
876 Layer.provide(
877 Layer.succeed(AgentScope, {
878 ...AgentScope.defaultValue(),
879 kill: () => Effect.sync(() => (killed = true)),
880 }),
881 ),
882 Layer.provide(ServerSettings.layerTest()),
883 Layer.provide(layerReceiptsTest),
884 );
885 const error = yield* Effect.gen(function* () {
886 const service = yield* ThreadManagementService.ThreadManagementService;
887 return yield* Effect.flip(
888 service.forceStopThread({ threadId, runId: RunId.make("run:stopped") }),
889 );
890 }).pipe(Effect.provide(layerTest));
891 expect(error.reason).toBe("superseded");
892 expect(killed).toBe(false);
893 }),
894);
895
896it.effect(
897 "admits two transcript exports at a time and frees a slot on failure or interruption",
898 () =>
899 Effect.gen(function* () {
900 const started = yield* Queue.unbounded<string>();
901 const gates = new Map(
902 ["a", "b", "c", "d", "e", "f"].map((id) => [
903 id,
904 Deferred.makeUnsafe<void, Orchestrator.OrchestratorV2Error>(),
905 ]),
906 );
907 const layerTest = ThreadManagementService.layer.pipe(
908 Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})),
909 Layer.provide(
910 Layer.mock(Orchestrator.OrchestratorV2)({
911 getTranscriptByteFloor: () => Effect.succeed(0),
912 getThreadSnapshot: (threadId) =>
913 Queue.offer(started, threadId).pipe(
914 Effect.andThen(Deferred.await(gates.get(threadId)!)),
915 Effect.as({
916 schemaVersion: 1,
917 snapshotSequence: 1,
918 projection: {
919 thread: { id: threadId, title: threadId },
920 updatedAt: DateTime.makeUnsafe(0),
921 visibleTurnItems: [],
922 } as unknown as OrchestrationV2ThreadProjection,
923 }),
924 ),
925 }),
926 ),
927 Layer.provide(ServerSettings.layerTest()),
928 Layer.provide(layerReceiptsTest),
929 );
930
931 yield* Effect.gen(function* () {
932 const threads = yield* ThreadManagementService.ThreadManagementService;
933 const exportThread = (id: string) =>
934 threads.getThreadTranscript(ThreadId.make(id)).pipe(
935 Effect.provideService(ServerEnvironment.ServerEnvironmentIdentity, {
936 getEnvironmentId: Effect.succeed(EnvironmentId.make("environment:export")),
937 }),
938 Effect.forkChild({ startImmediately: true }),
939 );
940 const a = yield* exportThread("a");
941 const b = yield* exportThread("b");
942 const c = yield* exportThread("c");
943 const d = yield* exportThread("d");
944 const e = yield* exportThread("e");
945 const f = yield* exportThread("f");
946 expect(yield* Queue.takeAll(started)).toEqual(["a", "b"]);
947
948 yield* Fiber.interrupt(c);
949 yield* Deferred.fail(
950 gates.get("a")!,
951 new Orchestrator.OrchestratorProjectionError({ threadId: ThreadId.make("a"), cause: "" }),
952 );
953 expect(yield* Queue.take(started)).toBe("d");
954 yield* Fiber.interrupt(b);
955 expect(yield* Queue.take(started)).toBe("e");
956 yield* Deferred.succeed(gates.get("d")!, undefined);
957 expect(yield* Queue.take(started)).toBe("f");
958 yield* Deferred.succeed(gates.get("e")!, undefined);
959 yield* Deferred.succeed(gates.get("f")!, undefined);
960
961 expect((yield* Fiber.await(a))._tag).toBe("Failure");
962 expect((yield* Fiber.join(d)).threadId).toBe("d");
963 expect((yield* Fiber.join(e)).threadId).toBe("e");
964 expect((yield* Fiber.join(f)).threadId).toBe("f");
965 expect(yield* Queue.size(started)).toBe(0);
966 }).pipe(Effect.provide(layerTest));
967 }),
968);