1import { assert, describe, it } from "@effect/vitest";
2import {
3 MessageId,
4 ProjectId,
5 ProviderDriverKind,
6 ProviderInstanceId,
7 ProviderThreadId,
8 RunId,
9 ThreadId,
10 type OrchestrationV2ThreadProjection,
11 type OrchestrationV2Notification,
12} from "@t3tools/contracts";
13import * as Deferred from "effect/Deferred";
14import * as DateTime from "effect/DateTime";
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 ProviderContinuationService from "./ProviderContinuationService.ts";
27import * as ThreadManagementService from "./ThreadManagementService.ts";
28
29const threadId = ThreadId.make("thread-provider-continuation");
30const providerThreadId = ProviderThreadId.make("provider-thread-continuation");
31const driver = ProviderDriverKind.make("continuation-test");
32const parentRunId = RunId.make("run-provider-continuation-parent");
33const delegatedMessageId = MessageId.make("message-provider-continuation-delegated");
34const projection = {
35 thread: { archivedAt: null, deletedAt: null },
36 messages: [],
37} as unknown as OrchestrationV2ThreadProjection;
38
39const request = (
40 dispatchIfCurrent?: ProviderContinuationRequests.ProviderContinuationRequest["dispatchIfCurrent"],
41 detail: string | null = null,
42): ProviderContinuationRequests.ProviderContinuationRequest => ({
43 threadId,
44 providerThreadId,
45 driver,
46 detail,
47 ...(dispatchIfCurrent === undefined ? {} : { dispatchIfCurrent }),
48});
49
50const delegatedProjection = (disposition: "open" | "stopped" | "disposed" = "open") =>
51 ({
52 ...projection,
53 runs: [
54 {
55 id: parentRunId,
56 delegatedCompletion: {
57 disposition,
58 nextGeneration: 2,
59 delivery: {
60 generation: 1,
61 messageId: delegatedMessageId,
62 taskIds: ["node-provider-continuation-first", "node-provider-continuation-second"],
63 },
64 },
65 },
66 ],
67 }) as unknown as OrchestrationV2ThreadProjection;
68
69const makeGuard = Effect.fnUntraced(function* (completed?: Deferred.Deferred<void>) {
70 const generation = yield* Ref.make(0);
71 const permit = yield* Semaphore.make(1);
72 const capture = Effect.gen(function* () {
73 const captured = yield* Ref.updateAndGet(generation, (value) => value + 1);
74 return <A, E, R>(effect: Effect.Effect<A, E, R>) =>
75 permit
76 .withPermit(
77 Effect.gen(function* () {
78 if ((yield* Ref.get(generation)) !== captured) return Option.none();
79 return Option.some(yield* effect);
80 }),
81 )
82 .pipe(
83 completed === undefined
84 ? (effect) => effect
85 : Effect.ensuring(Deferred.succeed(completed, undefined)),
86 );
87 });
88 return {
89 capture,
90 invalidate: permit.withPermit(Ref.update(generation, (value) => value + 1)),
91 };
92});
93
94function layerTest(input: {
95 readonly dispatched: Queue.Queue<unknown>;
96 readonly getThreadRecords: () => Effect.Effect<OrchestrationV2ThreadProjection>;
97}) {
98 const layerThreads = Layer.mock(ThreadManagementService.ThreadManagementService)({
99 getThreadRecords: input.getThreadRecords,
100 dispatch: (command) => Queue.offer(input.dispatched, command).pipe(Effect.as({} as never)),
101 });
102 const layerWorker = ProviderContinuationService.layer.pipe(
103 Layer.provide(
104 Layer.mergeAll(IdAllocator.layer, ProviderContinuationRequests.layer, layerThreads),
105 ),
106 );
107 return Layer.merge(ProviderContinuationRequests.layer, layerWorker);
108}
109
110describe("ProviderContinuationService", () => {
111 it.effect("recovers an unaccepted persisted steer using the same delivery identity", () =>
112 Effect.gen(function* () {
113 const dispatched = yield* Queue.unbounded<unknown>();
114 const original = delegatedProjection();
115 const interruptedRunId = RunId.make("interrupted-mailbox-run");
116 const recovered = {
117 ...original,
118 runs: [
119 ...original.runs,
120 {
121 id: interruptedRunId,
122 status: "interrupted",
123 userMessageId: MessageId.make("original-user-input"),
124 },
125 ],
126 messages: [
127 {
128 id: delegatedMessageId,
129 runId: interruptedRunId,
130 delegatedCompletion: { parentRunId, generation: 1, taskIds: [] },
131 },
132 ],
133 } as unknown as OrchestrationV2ThreadProjection;
134 yield* Effect.gen(function* () {
135 const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
136 yield* requests.offer({
137 threadId,
138 providerThreadId,
139 driver,
140 detail: null,
141 delivery: "message_text",
142 delegatedCompletion: { parentRunId, generation: 1, messageId: delegatedMessageId },
143 });
144 const command = (yield* Queue.take(dispatched)) as {
145 messageId: MessageId;
146 delegatedCompletion: { generation: number; taskIds: readonly string[] };
147 };
148 assert.equal(command.messageId, delegatedMessageId);
149 assert.equal(command.delegatedCompletion.generation, 1);
150 assert.equal(command.delegatedCompletion.taskIds.length, 2);
151 }).pipe(
152 Effect.provide(
153 layerTest({ dispatched, getThreadRecords: () => Effect.succeed(recovered) }),
154 ),
155 Effect.scoped,
156 );
157 }),
158 );
159
160 it.effect("marks an adapter-buffered wake with the provider creation source", () => {
161 return Effect.gen(function* () {
162 const dispatched = yield* Queue.unbounded<unknown>();
163 yield* Effect.gen(function* () {
164 const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
165 yield* requests.offer(request());
166 const command = (yield* Queue.take(dispatched)) as {
167 readonly creationSource: string;
168 readonly notification: OrchestrationV2Notification;
169 };
170 // ClaudeAdapterV2 keys on this to attach buffered CLI output.
171 assert.equal(command.creationSource, "provider");
172 assert.deepEqual(command.notification, {
173 source: { kind: "background_task" },
174 outcome: "updated",
175 summary: "Background activity updated",
176 });
177 }).pipe(
178 Effect.provide(
179 layerTest({ dispatched, getThreadRecords: () => Effect.succeed(projection) }),
180 ),
181 Effect.scoped,
182 );
183 });
184 });
185
186 it.effect("dispatches a wake under the message its adapter named", () => {
187 return Effect.gen(function* () {
188 const dispatched = yield* Queue.unbounded<unknown>();
189 const messageId = MessageId.make("message-provider-continuation-adapter-named");
190 yield* Effect.gen(function* () {
191 const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
192 yield* requests.offer({ ...request(), messageId });
193 const command = (yield* Queue.take(dispatched)) as { readonly messageId: MessageId };
194 // The adapter matches the continuation run's message to the wake it holds.
195 assert.equal(command.messageId, messageId);
196 }).pipe(
197 Effect.provide(
198 layerTest({ dispatched, getThreadRecords: () => Effect.succeed(projection) }),
199 ),
200 Effect.scoped,
201 );
202 });
203 });
204
205 it.effect("delivers a message_text wake as a real prompt, not a buffered wake", () => {
206 return Effect.gen(function* () {
207 const dispatched = yield* Queue.unbounded<unknown>();
208 yield* Effect.gen(function* () {
209 const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
210 yield* requests.offer({
211 threadId,
212 providerThreadId,
213 driver,
214 detail: "Delegated task completed.",
215 notification: {
216 source: { kind: "monitor" },
217 outcome: "updated",
218 summary: "Monitor updated",
219 },
220 delivery: "message_text",
221 });
222 const command = (yield* Queue.take(dispatched)) as {
223 readonly creationSource: string;
224 readonly text: string;
225 readonly notification: OrchestrationV2Notification;
226 };
227 // An app-owned child buffers nothing in the adapter, so this text is
228 // the whole wake. Marking it "provider" would make ClaudeAdapterV2
229 // drop it and settle the turn having prompted nothing.
230 assert.notEqual(command.creationSource, "provider");
231 assert.equal(command.creationSource, "server");
232 assert.equal(command.text, "Delegated task completed.");
233 assert.deepEqual(command.notification, {
234 source: { kind: "monitor" },
235 outcome: "updated",
236 summary: "Monitor updated",
237 });
238 }).pipe(
239 Effect.provide(
240 layerTest({ dispatched, getThreadRecords: () => Effect.succeed(projection) }),
241 ),
242 Effect.scoped,
243 );
244 });
245 });
246
247 it.effect("dispatches a current delegated completion as one server-owned queued message", () => {
248 return Effect.gen(function* () {
249 const dispatched = yield* Queue.unbounded<unknown>();
250 yield* Effect.gen(function* () {
251 const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
252 yield* requests.offer({
253 threadId,
254 providerThreadId,
255 driver,
256 detail: null,
257 delivery: "message_text",
258 delegatedCompletion: {
259 parentRunId,
260 generation: 1,
261 messageId: delegatedMessageId,
262 },
263 });
264 const command = (yield* Queue.take(dispatched)) as {
265 readonly createdBy: string;
266 readonly creationSource: string;
267 readonly delegatedCompletion: {
268 readonly generation: number;
269 readonly parentRunId: RunId;
270 readonly taskIds: ReadonlyArray<string>;
271 };
272 readonly dispatchMode: { readonly type: string };
273 readonly messageId: MessageId;
274 readonly text: string;
275 };
276 assert.equal(command.createdBy, "agent");
277 assert.equal(command.creationSource, "server");
278 assert.deepEqual(command.dispatchMode, { type: "queue_after_active" });
279 assert.equal(command.messageId, delegatedMessageId);
280 assert.equal(command.delegatedCompletion.parentRunId, parentRunId);
281 assert.equal(command.delegatedCompletion.generation, 1);
282 assert.deepEqual(command.delegatedCompletion.taskIds, [
283 "node-provider-continuation-first",
284 "node-provider-continuation-second",
285 ]);
286 assert.include(command.text, "task_status");
287 }).pipe(
288 Effect.provide(
289 layerTest({
290 dispatched,
291 getThreadRecords: () => Effect.succeed(delegatedProjection()),
292 }),
293 ),
294 Effect.scoped,
295 );
296 });
297 });
298
299 it.effect("retries a delegated completion after a transient dispatch failure", () => {
300 return Effect.gen(function* () {
301 const attempts = yield* Ref.make(0);
302 const dispatched = yield* Queue.unbounded<unknown>();
303 const layerThreads = Layer.mock(ThreadManagementService.ThreadManagementService)({
304 getThreadRecords: () => Effect.succeed(delegatedProjection()),
305 dispatch: (command) =>
306 Ref.getAndUpdate(attempts, (count) => count + 1).pipe(
307 Effect.flatMap((attempt) =>
308 attempt === 0
309 ? Effect.fail(new Error("simulated transient dispatch failure") as never)
310 : Queue.offer(dispatched, command).pipe(Effect.as({} as never)),
311 ),
312 ),
313 });
314 const layerWorker = ProviderContinuationService.layer.pipe(
315 Layer.provide(
316 Layer.mergeAll(IdAllocator.layer, ProviderContinuationRequests.layer, layerThreads),
317 ),
318 );
319
320 yield* Effect.gen(function* () {
321 const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
322 yield* requests.offer({
323 threadId,
324 providerThreadId,
325 driver,
326 detail: null,
327 delivery: "message_text",
328 delegatedCompletion: {
329 parentRunId,
330 generation: 1,
331 messageId: delegatedMessageId,
332 },
333 });
334 yield* Effect.yieldNow;
335 assert.equal(yield* Ref.get(attempts), 1);
336 assert.isTrue(Option.isNone(yield* Queue.poll(dispatched)));
337
338 yield* TestClock.adjust("100 millis");
339 const command = (yield* Queue.take(dispatched)) as {
340 readonly messageId: MessageId;
341 };
342 assert.equal(command.messageId, delegatedMessageId);
343 assert.equal(yield* Ref.get(attempts), 2);
344 }).pipe(
345 Effect.provide(Layer.merge(ProviderContinuationRequests.layer, layerWorker)),
346 Effect.scoped,
347 );
348 });
349 });
350
351 it.effect("backs off repeated delegated completion dispatch failures", () => {
352 return Effect.gen(function* () {
353 const attempts = yield* Ref.make(0);
354 const disposition = yield* Ref.make<"open" | "disposed">("open");
355 const layerThreads = Layer.mock(ThreadManagementService.ThreadManagementService)({
356 getThreadRecords: () =>
357 Ref.get(disposition).pipe(Effect.map((state) => delegatedProjection(state))),
358 dispatch: () =>
359 Ref.update(attempts, (count) => count + 1).pipe(
360 Effect.andThen(Effect.fail(new Error("simulated persistent failure") as never)),
361 ),
362 });
363 const layerWorker = ProviderContinuationService.layer.pipe(
364 Layer.provide(
365 Layer.mergeAll(IdAllocator.layer, ProviderContinuationRequests.layer, layerThreads),
366 ),
367 );
368
369 yield* Effect.gen(function* () {
370 const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
371 yield* requests.offer({
372 threadId,
373 providerThreadId,
374 driver,
375 detail: null,
376 delivery: "message_text",
377 delegatedCompletion: {
378 parentRunId,
379 generation: 1,
380 messageId: delegatedMessageId,
381 },
382 });
383 yield* Effect.yieldNow;
384 assert.equal(yield* Ref.get(attempts), 1);
385
386 yield* TestClock.adjust("100 millis");
387 yield* Effect.yieldNow;
388 assert.equal(yield* Ref.get(attempts), 2);
389 yield* TestClock.adjust("199 millis");
390 yield* Effect.yieldNow;
391 assert.equal(yield* Ref.get(attempts), 2);
392 yield* TestClock.adjust("1 millis");
393 yield* Effect.yieldNow;
394 assert.equal(yield* Ref.get(attempts), 3);
395
396 yield* Ref.set(disposition, "disposed");
397 yield* TestClock.adjust("400 millis");
398 }).pipe(
399 Effect.provide(Layer.merge(ProviderContinuationRequests.layer, layerWorker)),
400 Effect.scoped,
401 );
402 });
403 });
404
405 it.effect("resets delegated completion retry backoff after archive or deletion", () => {
406 return Effect.gen(function* () {
407 for (const barrier of ["archivedAt", "deletedAt"] as const) {
408 const attempts = yield* Ref.make(0);
409 const blockedProjectionReads = yield* Ref.make(0);
410 const blockedRequestDropped = yield* Deferred.make<void>();
411 const state = yield* Ref.make<"open" | "blocked" | "disposed">("open");
412 const layerThreads = Layer.mock(ThreadManagementService.ThreadManagementService)({
413 getThreadRecords: () =>
414 Effect.gen(function* () {
415 const currentState = yield* Ref.get(state);
416 if (currentState === "blocked") {
417 const reads = yield* Ref.updateAndGet(blockedProjectionReads, (count) => count + 1);
418 if (reads === 2) {
419 yield* Deferred.succeed(blockedRequestDropped, undefined);
420 }
421 }
422 return {
423 ...delegatedProjection(currentState === "disposed" ? "disposed" : "open"),
424 thread: {
425 ...projection.thread,
426 [barrier]:
427 currentState === "blocked"
428 ? DateTime.makeUnsafe("2026-08-05T00:00:00.000Z")
429 : null,
430 },
431 };
432 }),
433 dispatch: () =>
434 Ref.update(attempts, (count) => count + 1).pipe(
435 Effect.andThen(Effect.fail(new Error("simulated persistent failure") as never)),
436 ),
437 });
438 const layerWorker = ProviderContinuationService.layer.pipe(
439 Layer.provide(
440 Layer.mergeAll(IdAllocator.layer, ProviderContinuationRequests.layer, layerThreads),
441 ),
442 );
443 const completionRequest = {
444 threadId,
445 providerThreadId,
446 driver,
447 detail: null,
448 delivery: "message_text" as const,
449 delegatedCompletion: {
450 parentRunId,
451 generation: 1,
452 messageId: delegatedMessageId,
453 },
454 };
455
456 yield* Effect.gen(function* () {
457 const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
458 yield* requests.offer(completionRequest);
459 yield* Effect.yieldNow;
460 assert.equal(yield* Ref.get(attempts), 1);
461
462 yield* Ref.set(state, "blocked");
463 yield* TestClock.adjust("100 millis");
464 yield* Deferred.await(blockedRequestDropped);
465 yield* Effect.yieldNow;
466 assert.equal(yield* Ref.get(attempts), 1);
467
468 yield* Ref.set(state, "open");
469 yield* requests.offer(completionRequest);
470 yield* Effect.yieldNow;
471 assert.equal(yield* Ref.get(attempts), 2);
472 yield* TestClock.adjust("99 millis");
473 yield* Effect.yieldNow;
474 assert.equal(yield* Ref.get(attempts), 2);
475 yield* TestClock.adjust("1 millis");
476 yield* Effect.yieldNow;
477 assert.equal(yield* Ref.get(attempts), 3);
478
479 yield* Ref.set(state, "disposed");
480 yield* TestClock.adjust("200 millis");
481 }).pipe(
482 Effect.provide(Layer.merge(ProviderContinuationRequests.layer, layerWorker)),
483 Effect.scoped,
484 );
485 }
486 });
487 });
488
489 it.effect("drops a delegated completion retry after its delivery closes", () => {
490 return Effect.gen(function* () {
491 const attempts = yield* Ref.make(0);
492 const disposition = yield* Ref.make<"open" | "disposed">("open");
493 const layerThreads = Layer.mock(ThreadManagementService.ThreadManagementService)({
494 getThreadRecords: () =>
495 Ref.get(disposition).pipe(Effect.map((state) => delegatedProjection(state))),
496 dispatch: () =>
497 Ref.update(attempts, (count) => count + 1).pipe(
498 Effect.andThen(Effect.fail(new Error("simulated dispatch failure") as never)),
499 ),
500 });
501 const layerWorker = ProviderContinuationService.layer.pipe(
502 Layer.provide(
503 Layer.mergeAll(IdAllocator.layer, ProviderContinuationRequests.layer, layerThreads),
504 ),
505 );
506
507 yield* Effect.gen(function* () {
508 const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
509 yield* requests.offer({
510 threadId,
511 providerThreadId,
512 driver,
513 detail: null,
514 delivery: "message_text",
515 delegatedCompletion: {
516 parentRunId,
517 generation: 1,
518 messageId: delegatedMessageId,
519 },
520 });
521 yield* Effect.yieldNow;
522 assert.equal(yield* Ref.get(attempts), 1);
523
524 yield* Ref.set(disposition, "disposed");
525 yield* TestClock.adjust("100 millis");
526 yield* Effect.yieldNow;
527 assert.equal(yield* Ref.get(attempts), 1);
528 }).pipe(
529 Effect.provide(Layer.merge(ProviderContinuationRequests.layer, layerWorker)),
530 Effect.scoped,
531 );
532 });
533 });
534
535 it.effect(
536 "a project-disabled instance refuses a delegated-completion wake exactly once, acknowledges it, and never retries",
537 () => {
538 return Effect.gen(function* () {
539 const attempts = yield* Ref.make(0);
540 const cleared = yield* Ref.make(false);
541 const acknowledged = yield* Queue.unbounded<unknown>();
542 const layerThreads = Layer.mock(ThreadManagementService.ThreadManagementService)({
543 getThreadRecords: () => Effect.succeed(delegatedProjection()),
544 dispatch: (command) => {
545 if ((command as { type: string }).type === "message.dispatch") {
546 return Ref.update(attempts, (count) => count + 1).pipe(
547 Effect.andThen(
548 Effect.fail(
549 new ThreadManagementService.ThreadManagementProviderInstanceDisabledError({
550 projectId: ProjectId.make("project-provider-continuation-disabled"),
551 instanceId: ProviderInstanceId.make(
552 "instance-provider-continuation-disabled",
553 ),
554 }) as never,
555 ),
556 ),
557 );
558 }
559 return Queue.offer(acknowledged, command).pipe(Effect.as({} as never));
560 },
561 });
562 const layerWorker = ProviderContinuationService.layer.pipe(
563 Layer.provide(
564 Layer.mergeAll(IdAllocator.layer, ProviderContinuationRequests.layer, layerThreads),
565 ),
566 );
567
568 yield* Effect.gen(function* () {
569 const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
570 yield* requests.offer({
571 threadId,
572 providerThreadId,
573 driver,
574 detail: null,
575 delivery: "message_text",
576 clearIfCurrent: () => Ref.set(cleared, true),
577 delegatedCompletion: {
578 parentRunId,
579 generation: 1,
580 messageId: delegatedMessageId,
581 },
582 });
583 yield* Effect.yieldNow;
584 assert.equal(yield* Ref.get(attempts), 1);
585 assert.isTrue(yield* Ref.get(cleared));
586
587 // The refused delivery is acknowledged administratively, task by
588 // task, so the cohort's delivery is consumed instead of left open.
589 const acknowledgements = (yield* Queue.takeAll(acknowledged)) as ReadonlyArray<{
590 type: string;
591 taskId: string;
592 observedByRunId: string | null;
593 }>;
594 assert.deepEqual(
595 acknowledgements.map((command) => command.taskId),
596 ["node-provider-continuation-first", "node-provider-continuation-second"],
597 );
598 for (const command of acknowledgements) {
599 assert.equal(command.type, "delegated_task.completion-delivery.acknowledge");
600 assert.isNull(command.observedByRunId);
601 }
602
603 // A refusal is terminal: no backoff timer is pending behind it,
604 // so no further dispatch attempt can ever land.
605 yield* TestClock.adjust("1 hour");
606 yield* Effect.yieldNow;
607 assert.equal(yield* Ref.get(attempts), 1);
608 assert.equal(yield* Queue.size(acknowledged), 0);
609 }).pipe(
610 Effect.provide(Layer.merge(ProviderContinuationRequests.layer, layerWorker)),
611 Effect.scoped,
612 );
613 });
614 },
615 );
616
617 it.effect(
618 "clears the adapter's sticky continuation flag when a non-delegated wake is refused",
619 () => {
620 return Effect.gen(function* () {
621 const requested = yield* Ref.make(true);
622 const layerThreads = Layer.mock(ThreadManagementService.ThreadManagementService)({
623 getThreadRecords: () => Effect.succeed(projection),
624 dispatch: () =>
625 Effect.fail(
626 new ThreadManagementService.ThreadManagementProviderInstanceDisabledError({
627 projectId: ProjectId.make("project-provider-continuation-disabled"),
628 instanceId: ProviderInstanceId.make("instance-provider-continuation-disabled"),
629 }) as never,
630 ),
631 });
632 const layerWorker = ProviderContinuationService.layer.pipe(
633 Layer.provide(
634 Layer.mergeAll(IdAllocator.layer, ProviderContinuationRequests.layer, layerThreads),
635 ),
636 );
637
638 // Mirrors the ACP adapter's own dispatchIfCurrent: on a failed effect
639 // it clears its sticky "continuation requested" flag before
640 // re-raising. If this service pre-recovered the failure before
641 // handing the effect here, this flag would stay stuck set and
642 // hasPendingBackgroundWork would never clear.
643 const dispatchIfCurrent = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
644 Effect.gen(function* () {
645 const exit = yield* Effect.exit(effect);
646 if (Exit.isFailure(exit)) {
647 yield* Ref.set(requested, false);
648 return yield* Effect.failCause(exit.cause);
649 }
650 return Option.some(exit.value);
651 });
652
653 yield* Effect.gen(function* () {
654 const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
655 yield* requests.offer(request(dispatchIfCurrent));
656 yield* Effect.yieldNow;
657 yield* Effect.yieldNow;
658 assert.isFalse(
659 yield* Ref.get(requested),
660 "a disabled instance must still clear the adapter's sticky offer flag",
661 );
662 }).pipe(
663 Effect.provide(Layer.merge(ProviderContinuationRequests.layer, layerWorker)),
664 Effect.scoped,
665 );
666 });
667 },
668 );
669
670 it.effect(
671 "clears a sticky continuation offer for an adapter whose dispatchIfCurrent does not self-clear on failure",
672 () => {
673 return Effect.gen(function* () {
674 const wakeHeld = yield* Ref.make(true);
675 const layerThreads = Layer.mock(ThreadManagementService.ThreadManagementService)({
676 getThreadRecords: () => Effect.succeed(projection),
677 dispatch: () =>
678 Effect.fail(
679 new ThreadManagementService.ThreadManagementProviderInstanceDisabledError({
680 projectId: ProjectId.make("project-provider-continuation-disabled"),
681 instanceId: ProviderInstanceId.make("instance-provider-continuation-disabled"),
682 }) as never,
683 ),
684 });
685 const layerWorker = ProviderContinuationService.layer.pipe(
686 Layer.provide(
687 Layer.mergeAll(IdAllocator.layer, ProviderContinuationRequests.layer, layerThreads),
688 ),
689 );
690
691 // Mirrors the Muse adapter: dispatchIfCurrent forwards the dispatch
692 // effect without any failure handling of its own (unlike ACP or Pi),
693 // so only the explicit clearIfCurrent callback can clear the sticky
694 // "wake held" flag. Without it, a disabled instance leaves the flag
695 // set forever and no later background turn is ever offered.
696 const dispatchIfCurrent = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
697 Effect.asSome(effect);
698
699 yield* Effect.gen(function* () {
700 const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
701 yield* requests.offer({
702 ...request(dispatchIfCurrent),
703 clearIfCurrent: () => Ref.set(wakeHeld, false),
704 });
705 yield* Effect.yieldNow;
706 yield* Effect.yieldNow;
707 assert.isFalse(
708 yield* Ref.get(wakeHeld),
709 "a disabled instance must clear the offer via the explicit callback " +
710 "when the adapter itself does not clear on dispatch failure",
711 );
712 }).pipe(
713 Effect.provide(Layer.merge(ProviderContinuationRequests.layer, layerWorker)),
714 Effect.scoped,
715 );
716 });
717 },
718 );
719
720 it.effect("keeps a Grok delegated completion queued instead of restarting active work", () => {
721 return Effect.gen(function* () {
722 const dispatched = yield* Queue.unbounded<unknown>();
723 yield* Effect.gen(function* () {
724 const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
725 yield* requests.offer({
726 threadId,
727 providerThreadId,
728 driver: ProviderDriverKind.make("grok"),
729 detail: null,
730 delivery: "message_text",
731 delegatedCompletion: {
732 parentRunId,
733 generation: 1,
734 messageId: delegatedMessageId,
735 },
736 });
737 const command = (yield* Queue.take(dispatched)) as {
738 readonly dispatchMode: { readonly type: string };
739 };
740 assert.deepEqual(command.dispatchMode, { type: "queue_after_active" });
741 }).pipe(
742 Effect.provide(
743 layerTest({
744 dispatched,
745 getThreadRecords: () => Effect.succeed(delegatedProjection()),
746 }),
747 ),
748 Effect.scoped,
749 );
750 });
751 });
752
753 it.effect(
754 "drops stopped and disposed delegated completions instead of reviving them after recovery",
755 () => {
756 return Effect.gen(function* () {
757 for (const disposition of ["stopped", "disposed"] as const) {
758 const dispatched = yield* Queue.unbounded<unknown>();
759 yield* Effect.gen(function* () {
760 const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
761 yield* requests.offer({
762 threadId,
763 providerThreadId,
764 driver,
765 detail: null,
766 delivery: "message_text",
767 delegatedCompletion: {
768 parentRunId,
769 generation: 1,
770 messageId: delegatedMessageId,
771 },
772 });
773 yield* Effect.yieldNow;
774 yield* Effect.yieldNow;
775 assert.isTrue(Option.isNone(yield* Queue.poll(dispatched)));
776 }).pipe(
777 Effect.provide(
778 layerTest({
779 dispatched,
780 getThreadRecords: () => Effect.succeed(delegatedProjection(disposition)),
781 }),
782 ),
783 Effect.scoped,
784 );
785 }
786 });
787 },
788 );
789
790 it.effect("does not redispatch a persisted delegated completion message during recovery", () => {
791 return Effect.gen(function* () {
792 const dispatched = yield* Queue.unbounded<unknown>();
793 yield* Effect.gen(function* () {
794 const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
795 yield* requests.offer({
796 threadId,
797 providerThreadId,
798 driver,
799 detail: null,
800 delivery: "message_text",
801 delegatedCompletion: {
802 parentRunId,
803 generation: 1,
804 messageId: delegatedMessageId,
805 },
806 });
807 yield* Effect.yieldNow;
808 yield* Effect.yieldNow;
809 assert.isTrue(Option.isNone(yield* Queue.poll(dispatched)));
810 }).pipe(
811 Effect.provide(
812 layerTest({
813 dispatched,
814 getThreadRecords: () =>
815 Effect.succeed({
816 ...delegatedProjection(),
817 messages: [{ id: delegatedMessageId }],
818 } as unknown as OrchestrationV2ThreadProjection),
819 }),
820 ),
821 Effect.scoped,
822 );
823 });
824 });
825
826 it.effect("drops archived and deleted delegated completions during recovery", () => {
827 return Effect.gen(function* () {
828 for (const barrier of ["archivedAt", "deletedAt"] as const) {
829 const dispatched = yield* Queue.unbounded<unknown>();
830 yield* Effect.gen(function* () {
831 const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
832 yield* requests.offer({
833 threadId,
834 providerThreadId,
835 driver,
836 detail: null,
837 delivery: "message_text",
838 delegatedCompletion: {
839 parentRunId,
840 generation: 1,
841 messageId: delegatedMessageId,
842 },
843 });
844 yield* Effect.yieldNow;
845 yield* Effect.yieldNow;
846 assert.isTrue(Option.isNone(yield* Queue.poll(dispatched)));
847 }).pipe(
848 Effect.provide(
849 layerTest({
850 dispatched,
851 getThreadRecords: () =>
852 Effect.succeed({
853 ...delegatedProjection(),
854 thread: {
855 ...projection.thread,
856 [barrier]: DateTime.makeUnsafe("2026-08-03T00:00:00.000Z"),
857 },
858 } as unknown as OrchestrationV2ThreadProjection),
859 }),
860 ),
861 Effect.scoped,
862 );
863 }
864 });
865 });
866
867 it.effect("dispatches a current request exactly once", () => {
868 return Effect.gen(function* () {
869 const dispatched = yield* Queue.unbounded<unknown>();
870 yield* Effect.gen(function* () {
871 const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
872 yield* requests.offer(request());
873 yield* Queue.take(dispatched);
874 yield* Effect.yieldNow;
875 assert.isTrue(Option.isNone(yield* Queue.poll(dispatched)));
876 }).pipe(
877 Effect.provide(
878 layerTest({ dispatched, getThreadRecords: () => Effect.succeed(projection) }),
879 ),
880 Effect.scoped,
881 );
882 });
883 });
884
885 it.effect("queues a continuation behind an active user run", () => {
886 return Effect.gen(function* () {
887 const dispatched = yield* Queue.unbounded<unknown>();
888 yield* Effect.gen(function* () {
889 const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
890 yield* requests.offer(request());
891 const command = yield* Queue.take(dispatched);
892 assert.deepEqual((command as { readonly dispatchMode?: unknown }).dispatchMode, {
893 type: "queue_after_active",
894 });
895 }).pipe(
896 Effect.provide(
897 layerTest({ dispatched, getThreadRecords: () => Effect.succeed(projection) }),
898 ),
899 Effect.scoped,
900 );
901 });
902 });
903
904 it.effect("drops a request invalidated before dispatch", () => {
905 return Effect.gen(function* () {
906 const dispatched = yield* Queue.unbounded<unknown>();
907 yield* Effect.gen(function* () {
908 const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
909 const guard = yield* makeGuard();
910 const dispatchIfCurrent = yield* guard.capture;
911 yield* guard.invalidate;
912 yield* requests.offer(request(dispatchIfCurrent));
913 yield* Effect.yieldNow;
914 yield* Effect.yieldNow;
915 assert.isTrue(Option.isNone(yield* Queue.poll(dispatched)));
916 }).pipe(
917 Effect.provide(
918 layerTest({ dispatched, getThreadRecords: () => Effect.succeed(projection) }),
919 ),
920 Effect.scoped,
921 );
922 });
923 });
924
925 it.effect("drops a request invalidated while projection is blocked", () => {
926 return Effect.gen(function* () {
927 const dispatched = yield* Queue.unbounded<unknown>();
928 const projectionEntered = yield* Deferred.make<void>();
929 const releaseProjection = yield* Deferred.make<void>();
930 const guardCompleted = yield* Deferred.make<void>();
931 yield* Effect.gen(function* () {
932 const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
933 const guard = yield* makeGuard(guardCompleted);
934 const dispatchIfCurrent = yield* guard.capture;
935 yield* requests.offer(request(dispatchIfCurrent));
936 yield* Deferred.await(projectionEntered);
937 yield* guard.invalidate;
938 yield* Deferred.succeed(releaseProjection, undefined);
939 yield* Deferred.await(guardCompleted);
940 assert.isTrue(Option.isNone(yield* Queue.poll(dispatched)));
941 }).pipe(
942 Effect.provide(
943 layerTest({
944 dispatched,
945 getThreadRecords: () =>
946 Deferred.succeed(projectionEntered, undefined).pipe(
947 Effect.andThen(Deferred.await(releaseProjection)),
948 Effect.as(projection),
949 ),
950 }),
951 ),
952 Effect.scoped,
953 );
954 });
955 });
956
957 it.effect("does not revive an old request when a later generation is current", () => {
958 return Effect.gen(function* () {
959 const dispatched = yield* Queue.unbounded<unknown>();
960 const firstProjectionEntered = yield* Deferred.make<void>();
961 const releaseFirstProjection = yield* Deferred.make<void>();
962 let projectionCalls = 0;
963 yield* Effect.gen(function* () {
964 const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
965 const guard = yield* makeGuard();
966 const first = yield* guard.capture;
967 yield* requests.offer(request(first, "A"));
968 yield* Deferred.await(firstProjectionEntered);
969 const second = yield* guard.capture;
970 yield* requests.offer(request(second, "B"));
971 yield* Deferred.succeed(releaseFirstProjection, undefined);
972 const command = yield* Queue.take(dispatched);
973 assert.equal((command as { readonly text?: unknown }).text, "B");
974 yield* Effect.yieldNow;
975 assert.isTrue(Option.isNone(yield* Queue.poll(dispatched)));
976 }).pipe(
977 Effect.provide(
978 layerTest({
979 dispatched,
980 getThreadRecords: () => {
981 projectionCalls += 1;
982 return projectionCalls === 1
983 ? Deferred.succeed(firstProjectionEntered, undefined).pipe(
984 Effect.andThen(Deferred.await(releaseFirstProjection)),
985 Effect.as(projection),
986 )
987 : Effect.succeed(projection);
988 },
989 }),
990 ),
991 Effect.scoped,
992 );
993 });
994 });
995
996 it.effect("clears a sticky continuation offer when its thread is archived", () => {
997 return Effect.gen(function* () {
998 const dispatched = yield* Queue.unbounded<unknown>();
999 const cleared = yield* Ref.make(false);
1000 yield* Effect.gen(function* () {
1001 const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
1002 yield* requests.offer({
1003 ...request(),
1004 clearIfCurrent: () => Ref.set(cleared, true),
1005 });
1006 for (let attempt = 0; attempt < 20 && !(yield* Ref.get(cleared)); attempt += 1) {
1007 yield* Effect.yieldNow;
1008 }
1009 assert.isTrue(yield* Ref.get(cleared));
1010 assert.isTrue(Option.isNone(yield* Queue.poll(dispatched)));
1011 }).pipe(
1012 Effect.provide(
1013 layerTest({
1014 dispatched,
1015 getThreadRecords: () =>
1016 Effect.succeed({
1017 ...projection,
1018 thread: {
1019 ...projection.thread,
1020 archivedAt: DateTime.makeUnsafe("2026-07-20T00:00:00.000Z"),
1021 },
1022 }),
1023 }),
1024 ),
1025 Effect.scoped,
1026 );
1027 });
1028 });
1029});