Open History to see the last commit
1import { assert, describe, it } from "@effect/vitest";
2import {
3 CommandId,
4 MessageId,
5 ProjectId,
6 ProviderDriverKind,
7 ProviderInstanceId,
8 ProviderThreadId,
9 RunId,
10 ThreadId,
11 type OrchestrationV2ThreadProjection,
12 type OrchestrationV2Notification,
13} from "@t3tools/contracts";
14import * as Deferred from "effect/Deferred";
15import * as Effect from "effect/Effect";
16import * as Exit from "effect/Exit";
17import * as Layer from "effect/Layer";
18import * as Option from "effect/Option";
19import * as Queue from "effect/Queue";
20import * as Ref from "effect/Ref";
21import * as Semaphore from "effect/Semaphore";
22import * as TestClock from "effect/testing/TestClock";
23
24import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
25import * as ProviderContinuationRequests from "@t3tools/provider-core/server/ProviderContinuationRequests";
26import * as Orchestrator from "./Orchestrator.ts";
27import * as ProviderContinuationService from "./ProviderContinuationService.ts";
28import * as ThreadManagementService from "./ThreadManagementService.ts";
29
30const threadId = ThreadId.make("thread-provider-continuation");
31const providerThreadId = ProviderThreadId.make("provider-thread-continuation");
32const driver = ProviderDriverKind.make("continuation-test");
33const parentRunId = RunId.make("run-provider-continuation-parent");
34const delegatedMessageId = MessageId.make("message-provider-continuation-delegated");
35const projection = {
36 thread: { archivedAt: null, deletedAt: null },
37 messages: [],
38} as unknown as OrchestrationV2ThreadProjection;
39
40const request = (
41 dispatchIfCurrent?: ProviderContinuationRequests.ProviderContinuationRequest["dispatchIfCurrent"],
42 detail: string | null = null,
43): ProviderContinuationRequests.ProviderContinuationRequest => ({
44 threadId,
45 providerThreadId,
46 driver,
47 detail,
48 ...(dispatchIfCurrent === undefined ? {} : { dispatchIfCurrent }),
49});
50
51const delegatedProjection = (disposition: "open" | "stopped" | "disposed" = "open") =>
52 ({
53 ...projection,
54 runs: [
55 {
56 id: parentRunId,
57 delegatedCompletion: {
58 disposition,
59 nextGeneration: 2,
60 delivery: {
61 generation: 1,
62 messageId: delegatedMessageId,
63 taskIds: ["node-provider-continuation-first", "node-provider-continuation-second"],
64 },
65 },
66 },
67 ],
68 }) as unknown as OrchestrationV2ThreadProjection;
69
70const makeGuard = Effect.fnUntraced(function* (completed?: Deferred.Deferred<void>) {
71 const generation = yield* Ref.make(0);
72 const permit = yield* Semaphore.make(1);
73 const capture = Effect.gen(function* () {
74 const captured = yield* Ref.updateAndGet(generation, (value) => value + 1);
75 return <A, E, R>(effect: Effect.Effect<A, E, R>) =>
76 permit
77 .withPermit(
78 Effect.gen(function* () {
79 if ((yield* Ref.get(generation)) !== captured) return Option.none();
80 return Option.some(yield* effect);
81 }),
82 )
83 .pipe(
84 completed === undefined
85 ? (effect) => effect
86 : Effect.ensuring(Deferred.succeed(completed, undefined)),
87 );
88 });
89 return {
90 capture,
91 invalidate: permit.withPermit(Ref.update(generation, (value) => value + 1)),
92 };
93});
94
95function layerTest(input: {
96 readonly dispatched: Queue.Queue<unknown>;
97 readonly getThreadRecords: () => Effect.Effect<OrchestrationV2ThreadProjection>;
98}) {
99 const layerThreads = Layer.mock(ThreadManagementService.ThreadManagementService)({
100 getThreadRecords: input.getThreadRecords,
101 dispatch: (command) => Queue.offer(input.dispatched, command).pipe(Effect.as({} as never)),
102 });
103 const layerWorker = ProviderContinuationService.layer.pipe(
104 Layer.provide(
105 Layer.mergeAll(IdAllocator.layer, ProviderContinuationRequests.layer, layerThreads),
106 ),
107 );
108 return Layer.merge(ProviderContinuationRequests.layer, layerWorker);
109}
110
111describe("ProviderContinuationService", () => {
112 it.effect("recovers an unaccepted persisted steer using the same delivery identity", () =>
113 Effect.gen(function* () {
114 const dispatched = yield* Queue.unbounded<unknown>();
115 const original = delegatedProjection();
116 const interruptedRunId = RunId.make("interrupted-mailbox-run");
117 const recovered = {
118 ...original,
119 runs: [
120 ...original.runs,
121 {
122 id: interruptedRunId,
123 status: "interrupted",
124 userMessageId: MessageId.make("original-user-input"),
125 },
126 ],
127 messages: [
128 {
129 id: delegatedMessageId,
130 runId: interruptedRunId,
131 delegatedCompletion: { parentRunId, generation: 1, taskIds: [] },
132 },
133 ],
134 } as unknown as OrchestrationV2ThreadProjection;
135 yield* Effect.gen(function* () {
136 const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
137 yield* requests.offer({
138 threadId,
139 providerThreadId,
140 driver,
141 detail: null,
142 delivery: "message_text",
143 delegatedCompletion: { parentRunId, generation: 1, messageId: delegatedMessageId },
144 });
145 const command = (yield* Queue.take(dispatched)) as {
146 messageId: MessageId;
147 delegatedCompletion: { generation: number; taskIds: readonly string[] };
148 };
149 assert.equal(command.messageId, delegatedMessageId);
150 assert.equal(command.delegatedCompletion.generation, 1);
151 assert.equal(command.delegatedCompletion.taskIds.length, 2);
152 }).pipe(
153 Effect.provide(
154 layerTest({ dispatched, getThreadRecords: () => Effect.succeed(recovered) }),
155 ),
156 Effect.scoped,
157 );
158 }),
159 );
160
161 it.effect("delivers a message_text wake as a real prompt, not a buffered wake", () => {
162 return Effect.gen(function* () {
163 const dispatched = yield* Queue.unbounded<unknown>();
164 yield* Effect.gen(function* () {
165 const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
166 yield* requests.offer({
167 threadId,
168 providerThreadId,
169 driver,
170 detail: "Delegated task completed.",
171 notification: {
172 source: { kind: "monitor" },
173 outcome: "updated",
174 summary: "Monitor updated",
175 },
176 delivery: "message_text",
177 });
178 const command = (yield* Queue.take(dispatched)) as {
179 readonly creationSource: string;
180 readonly text: string;
181 readonly notification: OrchestrationV2Notification;
182 };
183 assert.notEqual(command.creationSource, "provider");
184 assert.equal(command.creationSource, "server");
185 assert.equal(command.text, "Delegated task completed.");
186 assert.deepEqual(command.notification, {
187 source: { kind: "monitor" },
188 outcome: "updated",
189 summary: "Monitor updated",
190 });
191 }).pipe(
192 Effect.provide(
193 layerTest({ dispatched, getThreadRecords: () => Effect.succeed(projection) }),
194 ),
195 Effect.scoped,
196 );
197 });
198 });
199
200 it.effect("dispatches a current delegated completion as one server-owned queued message", () => {
201 return Effect.gen(function* () {
202 const dispatched = yield* Queue.unbounded<unknown>();
203 yield* Effect.gen(function* () {
204 const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
205 yield* requests.offer({
206 threadId,
207 providerThreadId,
208 driver,
209 detail: null,
210 delivery: "message_text",
211 delegatedCompletion: {
212 parentRunId,
213 generation: 1,
214 messageId: delegatedMessageId,
215 },
216 });
217 const command = (yield* Queue.take(dispatched)) as {
218 readonly createdBy: string;
219 readonly creationSource: string;
220 readonly delegatedCompletion: {
221 readonly generation: number;
222 readonly parentRunId: RunId;
223 readonly taskIds: ReadonlyArray<string>;
224 };
225 readonly dispatchMode: { readonly type: string };
226 readonly messageId: MessageId;
227 readonly text: string;
228 };
229 assert.equal(command.createdBy, "agent");
230 assert.equal(command.creationSource, "server");
231 assert.deepEqual(command.dispatchMode, { type: "queue_after_active" });
232 assert.equal(command.messageId, delegatedMessageId);
233 assert.equal(command.delegatedCompletion.parentRunId, parentRunId);
234 assert.equal(command.delegatedCompletion.generation, 1);
235 assert.deepEqual(command.delegatedCompletion.taskIds, [
236 "node-provider-continuation-first",
237 "node-provider-continuation-second",
238 ]);
239 assert.include(command.text, "task_status");
240 }).pipe(
241 Effect.provide(
242 layerTest({
243 dispatched,
244 getThreadRecords: () => Effect.succeed(delegatedProjection()),
245 }),
246 ),
247 Effect.scoped,
248 );
249 });
250 });
251
252 it.effect(
253 "a project-disabled instance refuses a delegated-completion wake exactly once, disposes it, and never retries",
254 () => {
255 return Effect.gen(function* () {
256 const attempts = yield* Ref.make(0);
257 const cleared = yield* Ref.make(false);
258 const disposed = yield* Queue.unbounded<unknown>();
259 const layerThreads = Layer.mock(ThreadManagementService.ThreadManagementService)({
260 getThreadRecords: () => Effect.succeed(delegatedProjection()),
261 dispatch: (command) => {
262 if ((command as { type: string }).type === "message.dispatch") {
263 return Ref.update(attempts, (count) => count + 1).pipe(
264 Effect.andThen(
265 Effect.fail(
266 new ThreadManagementService.ThreadManagementProviderInstanceDisabledError({
267 projectId: ProjectId.make("project-provider-continuation-disabled"),
268 instanceId: ProviderInstanceId.make(
269 "instance-provider-continuation-disabled",
270 ),
271 }) as never,
272 ),
273 ),
274 );
275 }
276 return Queue.offer(disposed, command).pipe(Effect.as({} as never));
277 },
278 });
279 const layerWorker = ProviderContinuationService.layer.pipe(
280 Layer.provide(
281 Layer.mergeAll(IdAllocator.layer, ProviderContinuationRequests.layer, layerThreads),
282 ),
283 );
284
285 yield* Effect.gen(function* () {
286 const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
287 yield* requests.offer({
288 threadId,
289 providerThreadId,
290 driver,
291 detail: null,
292 delivery: "message_text",
293 clearIfCurrent: () => Ref.set(cleared, true),
294 delegatedCompletion: {
295 parentRunId,
296 generation: 1,
297 messageId: delegatedMessageId,
298 },
299 });
300 yield* Effect.yieldNow;
301 assert.equal(yield* Ref.get(attempts), 1);
302 assert.isTrue(yield* Ref.get(cleared));
303
304 const dispositions = (yield* Queue.takeAll(disposed)) as ReadonlyArray<{
305 type: string;
306 taskId: string;
307 }>;
308 assert.deepEqual(
309 dispositions.map((command) => command.taskId),
310 ["node-provider-continuation-first", "node-provider-continuation-second"],
311 );
312 for (const command of dispositions) {
313 assert.equal(command.type, "delegated_task.completion-delivery.dispose");
314 }
315
316 yield* TestClock.adjust("1 hour");
317 yield* Effect.yieldNow;
318 assert.equal(yield* Ref.get(attempts), 1);
319 assert.equal(yield* Queue.size(disposed), 0);
320 }).pipe(
321 Effect.provide(Layer.merge(ProviderContinuationRequests.layer, layerWorker)),
322 Effect.scoped,
323 );
324 });
325 },
326 );
327
328 it.effect(
329 "clears the adapter's sticky continuation flag when a non-delegated wake is refused",
330 () => {
331 return Effect.gen(function* () {
332 const requested = yield* Ref.make(true);
333 const layerThreads = Layer.mock(ThreadManagementService.ThreadManagementService)({
334 getThreadRecords: () => Effect.succeed(projection),
335 dispatch: () =>
336 Effect.fail(
337 new ThreadManagementService.ThreadManagementProviderInstanceDisabledError({
338 projectId: ProjectId.make("project-provider-continuation-disabled"),
339 instanceId: ProviderInstanceId.make("instance-provider-continuation-disabled"),
340 }) as never,
341 ),
342 });
343 const layerWorker = ProviderContinuationService.layer.pipe(
344 Layer.provide(
345 Layer.mergeAll(IdAllocator.layer, ProviderContinuationRequests.layer, layerThreads),
346 ),
347 );
348
349 // ACP clears its sticky continuation flag on raw dispatch failure; pre-recovery would prevent that cleanup.
350 const dispatchIfCurrent = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
351 Effect.gen(function* () {
352 const exit = yield* Effect.exit(effect);
353 if (Exit.isFailure(exit)) {
354 yield* Ref.set(requested, false);
355 return yield* Effect.failCause(exit.cause);
356 }
357 return Option.some(exit.value);
358 });
359
360 yield* Effect.gen(function* () {
361 const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
362 yield* requests.offer(request(dispatchIfCurrent));
363 yield* Effect.yieldNow;
364 yield* Effect.yieldNow;
365 assert.isFalse(
366 yield* Ref.get(requested),
367 "a disabled instance must still clear the adapter's sticky offer flag",
368 );
369 }).pipe(
370 Effect.provide(Layer.merge(ProviderContinuationRequests.layer, layerWorker)),
371 Effect.scoped,
372 );
373 });
374 },
375 );
376
377 it.effect(
378 "clears a sticky continuation offer for an adapter whose dispatchIfCurrent does not self-clear on failure",
379 () => {
380 return Effect.gen(function* () {
381 const wakeHeld = yield* Ref.make(true);
382 const layerThreads = Layer.mock(ThreadManagementService.ThreadManagementService)({
383 getThreadRecords: () => Effect.succeed(projection),
384 dispatch: () =>
385 Effect.fail(
386 new ThreadManagementService.ThreadManagementProviderInstanceDisabledError({
387 projectId: ProjectId.make("project-provider-continuation-disabled"),
388 instanceId: ProviderInstanceId.make("instance-provider-continuation-disabled"),
389 }) as never,
390 ),
391 });
392 const layerWorker = ProviderContinuationService.layer.pipe(
393 Layer.provide(
394 Layer.mergeAll(IdAllocator.layer, ProviderContinuationRequests.layer, layerThreads),
395 ),
396 );
397
398 // Muse forwards dispatch failures unchanged; clearIfCurrent must release its sticky wake flag.
399 const dispatchIfCurrent = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
400 Effect.asSome(effect);
401
402 yield* Effect.gen(function* () {
403 const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
404 yield* requests.offer({
405 ...request(dispatchIfCurrent),
406 clearIfCurrent: () => Ref.set(wakeHeld, false),
407 });
408 yield* Effect.yieldNow;
409 yield* Effect.yieldNow;
410 assert.isFalse(
411 yield* Ref.get(wakeHeld),
412 "a disabled instance must clear the offer via the explicit callback " +
413 "when the adapter itself does not clear on dispatch failure",
414 );
415 }).pipe(
416 Effect.provide(Layer.merge(ProviderContinuationRequests.layer, layerWorker)),
417 Effect.scoped,
418 );
419 });
420 },
421 );
422
423 it.effect(
424 "clears a sticky continuation offer when a non-delegated wake dispatch fails generically",
425 () => {
426 return Effect.gen(function* () {
427 const cleared = yield* Deferred.make<void>();
428 const layerThreads = Layer.mock(ThreadManagementService.ThreadManagementService)({
429 getThreadRecords: () => Effect.succeed(projection),
430 dispatch: (command) =>
431 Effect.fail(
432 new Orchestrator.OrchestratorDispatchError({
433 commandId: CommandId.make(String(command.commandId)),
434 commandType: command.type,
435 cause: new Error("SQLITE_FULL: database or disk is full"),
436 }),
437 ),
438 });
439 const layerWorker = ProviderContinuationService.layer.pipe(
440 Layer.provide(
441 Layer.mergeAll(IdAllocator.layer, ProviderContinuationRequests.layer, layerThreads),
442 ),
443 );
444
445 yield* Effect.gen(function* () {
446 const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
447 yield* requests.offer({
448 ...request(),
449 clearIfCurrent: () => Deferred.succeed(cleared, undefined),
450 });
451 yield* Deferred.await(cleared);
452 }).pipe(
453 Effect.provide(Layer.merge(ProviderContinuationRequests.layer, layerWorker)),
454 Effect.scoped,
455 );
456 });
457 },
458 );
459
460 it.effect(
461 "drops stopped and disposed delegated completions instead of reviving them after recovery",
462 () => {
463 return Effect.gen(function* () {
464 for (const disposition of ["stopped", "disposed"] as const) {
465 const dispatched = yield* Queue.unbounded<unknown>();
466 yield* Effect.gen(function* () {
467 const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
468 yield* requests.offer({
469 threadId,
470 providerThreadId,
471 driver,
472 detail: null,
473 delivery: "message_text",
474 delegatedCompletion: {
475 parentRunId,
476 generation: 1,
477 messageId: delegatedMessageId,
478 },
479 });
480 yield* Effect.yieldNow;
481 yield* Effect.yieldNow;
482 assert.isTrue(Option.isNone(yield* Queue.poll(dispatched)));
483 }).pipe(
484 Effect.provide(
485 layerTest({
486 dispatched,
487 getThreadRecords: () => Effect.succeed(delegatedProjection(disposition)),
488 }),
489 ),
490 Effect.scoped,
491 );
492 }
493 });
494 },
495 );
496
497 it.effect("does not redispatch a persisted delegated completion message during recovery", () => {
498 return Effect.gen(function* () {
499 const dispatched = yield* Queue.unbounded<unknown>();
500 yield* Effect.gen(function* () {
501 const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
502 yield* requests.offer({
503 threadId,
504 providerThreadId,
505 driver,
506 detail: null,
507 delivery: "message_text",
508 delegatedCompletion: {
509 parentRunId,
510 generation: 1,
511 messageId: delegatedMessageId,
512 },
513 });
514 yield* Effect.yieldNow;
515 yield* Effect.yieldNow;
516 assert.isTrue(Option.isNone(yield* Queue.poll(dispatched)));
517 }).pipe(
518 Effect.provide(
519 layerTest({
520 dispatched,
521 getThreadRecords: () =>
522 Effect.succeed({
523 ...delegatedProjection(),
524 messages: [{ id: delegatedMessageId }],
525 } as unknown as OrchestrationV2ThreadProjection),
526 }),
527 ),
528 Effect.scoped,
529 );
530 });
531 });
532
533 it.effect("dispatches a current request exactly once", () => {
534 return Effect.gen(function* () {
535 const dispatched = yield* Queue.unbounded<unknown>();
536 yield* Effect.gen(function* () {
537 const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
538 yield* requests.offer(request());
539 yield* Queue.take(dispatched);
540 yield* Effect.yieldNow;
541 assert.isTrue(Option.isNone(yield* Queue.poll(dispatched)));
542 }).pipe(
543 Effect.provide(
544 layerTest({ dispatched, getThreadRecords: () => Effect.succeed(projection) }),
545 ),
546 Effect.scoped,
547 );
548 });
549 });
550
551 it.effect("drops a request invalidated while projection is blocked", () => {
552 return Effect.gen(function* () {
553 const dispatched = yield* Queue.unbounded<unknown>();
554 const projectionEntered = yield* Deferred.make<void>();
555 const releaseProjection = yield* Deferred.make<void>();
556 const guardCompleted = yield* Deferred.make<void>();
557 yield* Effect.gen(function* () {
558 const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
559 const guard = yield* makeGuard(guardCompleted);
560 const dispatchIfCurrent = yield* guard.capture;
561 yield* requests.offer(request(dispatchIfCurrent));
562 yield* Deferred.await(projectionEntered);
563 yield* guard.invalidate;
564 yield* Deferred.succeed(releaseProjection, undefined);
565 yield* Deferred.await(guardCompleted);
566 assert.isTrue(Option.isNone(yield* Queue.poll(dispatched)));
567 }).pipe(
568 Effect.provide(
569 layerTest({
570 dispatched,
571 getThreadRecords: () =>
572 Deferred.succeed(projectionEntered, undefined).pipe(
573 Effect.andThen(Deferred.await(releaseProjection)),
574 Effect.as(projection),
575 ),
576 }),
577 ),
578 Effect.scoped,
579 );
580 });
581 });
582
583 it.effect("does not revive an old request when a later generation is current", () => {
584 return Effect.gen(function* () {
585 const dispatched = yield* Queue.unbounded<unknown>();
586 const firstProjectionEntered = yield* Deferred.make<void>();
587 const releaseFirstProjection = yield* Deferred.make<void>();
588 let projectionCalls = 0;
589 yield* Effect.gen(function* () {
590 const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
591 const guard = yield* makeGuard();
592 const first = yield* guard.capture;
593 yield* requests.offer(request(first, "A"));
594 yield* Deferred.await(firstProjectionEntered);
595 const second = yield* guard.capture;
596 yield* requests.offer(request(second, "B"));
597 yield* Deferred.succeed(releaseFirstProjection, undefined);
598 const command = yield* Queue.take(dispatched);
599 assert.equal((command as { readonly text?: unknown }).text, "B");
600 yield* Effect.yieldNow;
601 assert.isTrue(Option.isNone(yield* Queue.poll(dispatched)));
602 }).pipe(
603 Effect.provide(
604 layerTest({
605 dispatched,
606 getThreadRecords: () => {
607 projectionCalls += 1;
608 return projectionCalls === 1
609 ? Deferred.succeed(firstProjectionEntered, undefined).pipe(
610 Effect.andThen(Deferred.await(releaseFirstProjection)),
611 Effect.as(projection),
612 )
613 : Effect.succeed(projection);
614 },
615 }),
616 ),
617 Effect.scoped,
618 );
619 });
620 });
621});