1import { assert, it } from "@effect/vitest";
2import {
3 CommandId,
4 NodeId,
5 ProviderSessionId,
6 ProviderThreadId,
7 ProviderTurnId,
8 RunAttemptId,
9 RunId,
10 ThreadId,
11} from "@t3tools/contracts";
12import * as Cause from "effect/Cause";
13import * as DateTime from "effect/DateTime";
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 TestClock from "effect/testing/TestClock";
22
23import * as CheckpointRollbackService from "./CheckpointRollbackService.ts";
24import * as EffectOutbox from "./EffectOutbox.ts";
25import * as EffectWorker from "./EffectWorker.ts";
26import * as RunFinalizationService from "./RunFinalizationService.ts";
27import * as ProviderSessionManager from "./ProviderSessionManager.ts";
28import * as ProviderTurnControlService from "./ProviderTurnControlService.ts";
29import * as ProviderTurnStartService from "./ProviderTurnStartService.ts";
30import * as RuntimeRequestService from "./RuntimeRequestService.ts";
31import * as ThreadTitleRegenerationService from "./ThreadTitleRegenerationService.ts";
32import * as ThreadManagementService from "./ThreadManagementService.ts";
33import * as ServerSettings from "../serverSettings.ts";
34
35const threadId = ThreadId.make("thread:effect-worker-restart");
36const oldSessionId = ProviderSessionId.make("provider-session:effect-worker-restart:old");
37const replacementSessionId = ProviderSessionId.make(
38 "provider-session:effect-worker-restart:replacement",
39);
40const providerThreadId = ProviderThreadId.make("provider-thread:effect-worker-restart");
41const providerTurnId = ProviderTurnId.make("provider-turn:effect-worker-restart");
42const attemptId = RunAttemptId.make("run-attempt:effect-worker-restart");
43const runId = RunId.make("run:effect-worker-restart");
44
45function restartEffect(
46 now: DateTime.Utc,
47 sessionTransition: NonNullable<
48 Extract<
49 EffectOutbox.OrchestrationEffectV2["request"],
50 { readonly type: "provider-turn.restart" }
51 >["sessionTransition"]
52 >,
53): EffectOutbox.OrchestrationEffectV2 {
54 const timestamp = DateTime.formatIso(now);
55 return {
56 id: `effect:restart:${sessionTransition.type}`,
57 commandId: CommandId.make(`command:restart:${sessionTransition.type}`),
58 threadId,
59 request: {
60 type: "provider-turn.restart",
61 providerSessionId: oldSessionId,
62 providerThreadId,
63 providerTurnId,
64 interruptedAttemptId: attemptId,
65 runId,
66 sessionTransition,
67 },
68 status: "running",
69 attemptCount: 1,
70 availableAt: timestamp,
71 leaseOwner: "test-worker",
72 leaseExpiresAt: timestamp,
73 createdAt: timestamp,
74 updatedAt: timestamp,
75 completedAt: null,
76 lastError: null,
77 };
78}
79
80function layerExecutorFor(input: {
81 readonly events: Ref.Ref<ReadonlyArray<string>>;
82 readonly failFirstStart?: Ref.Ref<boolean>;
83 readonly threads?: Partial<ThreadManagementService.ThreadManagementService["Service"]>;
84 readonly continueAfterRestart?: boolean;
85 readonly interrupt?: ProviderTurnControlService.ProviderTurnControlServiceV2Shape["interrupt"];
86}) {
87 const record = (event: string) => Ref.update(input.events, (events) => [...events, event]);
88 const layerDependencies = Layer.mergeAll(
89 Layer.succeed(
90 ProviderTurnControlService.ProviderTurnControlServiceV2,
91 ProviderTurnControlService.ProviderTurnControlServiceV2.of({
92 interrupt: input.interrupt ?? (() => Effect.void),
93 steer: () => Effect.void,
94 interruptAndAwaitTerminal: (request) =>
95 record(
96 request.replacementProviderSessionId === undefined
97 ? "interrupt"
98 : `interrupt:${request.replacementProviderSessionId}`,
99 ),
100 }),
101 ),
102 Layer.succeed(
103 ProviderSessionManager.ProviderSessionManagerV2,
104 ProviderSessionManager.ProviderSessionManagerV2.of({
105 shutdown: Effect.void,
106 open: () => Effect.die("unused open"),
107 get: () => Effect.succeed(Option.none()),
108 isLive: () => Effect.succeed(false),
109 close: () => Effect.void,
110 closeInstance: () => Effect.void,
111 release: () => record("release"),
112 detach: () => record("detach"),
113 }),
114 ),
115 Layer.succeed(
116 ProviderTurnStartService.ProviderTurnStartServiceV2,
117 ProviderTurnStartService.ProviderTurnStartServiceV2.of({
118 start: () =>
119 Effect.gen(function* () {
120 yield* record("start");
121 if (
122 input.failFirstStart !== undefined &&
123 (yield* Ref.getAndSet(input.failFirstStart, false))
124 ) {
125 return yield* new ProviderTurnStartService.ProviderTurnStartError({
126 runId,
127 cause: "simulated first start failure",
128 });
129 }
130 }),
131 }),
132 ),
133 Layer.succeed(
134 RunFinalizationService.RunFinalizationService,
135 RunFinalizationService.RunFinalizationService.of({ finalize: () => Effect.void }),
136 ),
137 Layer.succeed(
138 CheckpointRollbackService.CheckpointRollbackServiceV2,
139 CheckpointRollbackService.CheckpointRollbackServiceV2.of({ execute: () => Effect.void }),
140 ),
141 Layer.succeed(
142 RuntimeRequestService.RuntimeRequestServiceV2,
143 RuntimeRequestService.RuntimeRequestServiceV2.of({ respond: () => Effect.void }),
144 ),
145 Layer.succeed(
146 ThreadTitleRegenerationService.ThreadTitleRegenerationService,
147 ThreadTitleRegenerationService.ThreadTitleRegenerationService.of({
148 execute: () => Effect.void,
149 }),
150 ),
151 );
152 return EffectWorker.layerExecutor.pipe(
153 Layer.provide(
154 Layer.mergeAll(
155 layerDependencies,
156 Layer.mock(ThreadManagementService.ThreadManagementService)(input.threads ?? {}),
157 ServerSettings.layerTest(
158 input.continueAfterRestart === true ? { continueThreadsAfterServerUpdate: true } : {},
159 ),
160 ),
161 ),
162 );
163}
164
165it("does not retry pure interrupt races where the turn is already gone", () => {
166 assert.isTrue(
167 EffectWorker.isNonRetryableProviderTurnControlFailure(
168 "provider-turn.interrupt",
169 "ProviderAdapterInterruptError: ... ACP provider turn provider-turn:x is not active",
170 ),
171 );
172 assert.isTrue(
173 EffectWorker.isNonRetryableProviderTurnControlFailure(
174 "provider-turn.interrupt",
175 "Provider session provider-session:x is not active.",
176 ),
177 );
178 // Restart is compound (interrupt + detach + start). Do not swallow start failures.
179 assert.isFalse(
180 EffectWorker.isNonRetryableProviderTurnControlFailure(
181 "provider-turn.restart",
182 "Provider session provider-session:x is not active.",
183 ),
184 );
185 assert.isFalse(
186 EffectWorker.isNonRetryableProviderTurnControlFailure(
187 "provider-turn.start",
188 "Provider session provider-session:x is not active.",
189 ),
190 );
191 assert.isFalse(
192 EffectWorker.isNonRetryableProviderTurnControlFailure(
193 "provider-turn.interrupt",
194 "ACP hard teardown failed unexpectedly; the session is poisoned",
195 ),
196 );
197});
198
199it.effect("settles a stopped run when its adapter has already lost the native turn", () =>
200 Effect.gen(function* () {
201 const now = yield* DateTime.now;
202 const events = yield* Ref.make<ReadonlyArray<string>>([]);
203 const layer = layerExecutorFor({
204 events,
205 interrupt: () =>
206 new ProviderTurnControlService.ProviderTurnControlError({
207 threadId,
208 operation: "interrupt",
209 providerTurnId,
210 cause: "Provider turn is not active.",
211 }),
212 threads: {
213 dispatch: (command) =>
214 Ref.update(events, (current) => [...current, command.type]).pipe(
215 Effect.as({ sequence: 1, storedEvents: [] }),
216 ),
217 },
218 });
219 yield* Effect.gen(function* () {
220 const executor = yield* EffectWorker.OrchestrationEffectExecutorV2;
221 yield* executor.execute({
222 ...restartEffect(now, { type: "detach" }),
223 request: {
224 type: "provider-turn.interrupt",
225 providerSessionId: oldSessionId,
226 providerThreadId,
227 providerTurnId,
228 },
229 });
230 }).pipe(Effect.provide(layer));
231 assert.deepEqual(yield* Ref.get(events), ["thread.background-work.settle"]);
232 }),
233);
234
235it.effect("requeues a claim when a pre-execution worker check fails", () =>
236 Effect.gen(function* () {
237 const now = DateTime.formatIso(yield* DateTime.now);
238 const effectId = "effect:worker-pre-execution-failure";
239 const workerId = "worker-pre-execution-failure";
240 const claimedEffect: EffectOutbox.OrchestrationEffectV2 = {
241 id: effectId,
242 commandId: CommandId.make("command:worker-pre-execution-failure"),
243 threadId: ThreadId.make("thread:worker-pre-execution-failure"),
244 request: { type: "terminal.cleanup" },
245 status: "running",
246 attemptCount: 1,
247 availableAt: now,
248 leaseOwner: workerId,
249 leaseExpiresAt: now,
250 createdAt: now,
251 updatedAt: now,
252 completedAt: null,
253 lastError: null,
254 };
255 const retries = yield* Ref.make<
256 ReadonlyArray<{
257 readonly effectId: string;
258 readonly workerId: string;
259 readonly error: string;
260 readonly delayMs: number;
261 }>
262 >([]);
263 const executionCount = yield* Ref.make(0);
264 const layerOutbox = Layer.mock(EffectOutbox.EffectOutboxV2)({
265 claimNext: () => Effect.succeed(Option.some(claimedEffect)),
266 get: () =>
267 Effect.fail(
268 new EffectOutbox.EffectOutboxError({
269 operation: "get",
270 effectId,
271 cause: "simulated cancellation-state read failure",
272 }),
273 ),
274 retry: (input) =>
275 Ref.update(retries, (existing) => [...existing, input]).pipe(Effect.as(true)),
276 });
277 const layerExecutor = Layer.succeed(
278 EffectWorker.OrchestrationEffectExecutorV2,
279 EffectWorker.OrchestrationEffectExecutorV2.of({
280 execute: () => Ref.update(executionCount, (count) => count + 1),
281 }),
282 );
283 const layerWorker = EffectWorker.layerWithOptions({ workerId }).pipe(
284 Layer.provide(Layer.merge(layerOutbox, layerExecutor)),
285 );
286
287 const exit = yield* EffectWorker.OrchestrationEffectWorkerV2.pipe(
288 Effect.flatMap((worker) => worker.runOnce),
289 Effect.provide(layerWorker),
290 Effect.exit,
291 );
292
293 assert.isTrue(Exit.isFailure(exit));
294 if (Exit.isFailure(exit)) {
295 assert.include(Cause.pretty(exit.cause), "simulated cancellation-state read failure");
296 }
297 assert.equal(yield* Ref.get(executionCount), 0);
298 const retry = (yield* Ref.get(retries))[0];
299 assert.isDefined(retry);
300 assert.equal(retry.effectId, effectId);
301 assert.equal(retry.workerId, workerId);
302 assert.equal(retry.delayMs, 0);
303 assert.include(retry.error, "simulated cancellation-state read failure");
304 }),
305);
306
307it.effect("arms cancellation before the durable pre-execution check", () =>
308 Effect.gen(function* () {
309 const now = DateTime.formatIso(yield* DateTime.now);
310 const effectId = "effect:worker-cancellation-registration-race";
311 const workerId = "worker-cancellation-registration-race";
312 const claimedEffect: EffectOutbox.OrchestrationEffectV2 = {
313 id: effectId,
314 commandId: CommandId.make("command:worker-cancellation-registration-race"),
315 threadId: ThreadId.make("thread:worker-cancellation-registration-race"),
316 request: { type: "terminal.cleanup" },
317 status: "running",
318 attemptCount: 1,
319 availableAt: now,
320 leaseOwner: workerId,
321 leaseExpiresAt: now,
322 createdAt: now,
323 updatedAt: now,
324 completedAt: null,
325 lastError: null,
326 };
327 const signal = yield* Deferred.make<void>();
328 let cancellationArmed = false;
329 const executionCount = yield* Ref.make(0);
330 const settlementCount = yield* Ref.make(0);
331 const layerOutbox = Layer.mock(EffectOutbox.EffectOutboxV2)({
332 claimNext: () => Effect.succeed(Option.some(claimedEffect)),
333 awaitCancellation: () => {
334 cancellationArmed = true;
335 return Deferred.await(signal);
336 },
337 get: () =>
338 Effect.gen(function* () {
339 // Model a cancellation commit immediately after this durable read
340 // took its snapshot. Its process-local signal is only delivered when
341 // the worker registered the waiter before starting the read.
342 if (cancellationArmed) {
343 yield* Deferred.succeed(signal, undefined);
344 }
345 return Option.some(claimedEffect);
346 }),
347 clearCancellation: () => Effect.void,
348 succeed: () => Ref.update(settlementCount, (count) => count + 1).pipe(Effect.as(true)),
349 });
350 const layerExecutor = Layer.succeed(
351 EffectWorker.OrchestrationEffectExecutorV2,
352 EffectWorker.OrchestrationEffectExecutorV2.of({
353 execute: () =>
354 Effect.yieldNow.pipe(Effect.andThen(Ref.update(executionCount, (count) => count + 1))),
355 }),
356 );
357 const layerWorker = EffectWorker.layerWithOptions({ workerId }).pipe(
358 Layer.provide(Layer.merge(layerOutbox, layerExecutor)),
359 );
360
361 const exit = yield* EffectWorker.OrchestrationEffectWorkerV2.pipe(
362 Effect.flatMap((worker) => worker.runOnce),
363 Effect.provide(layerWorker),
364 Effect.exit,
365 );
366
367 if (Exit.isFailure(exit)) {
368 assert.fail(Cause.pretty(exit.cause));
369 }
370 assert.equal(yield* Ref.get(executionCount), 0);
371 assert.equal(yield* Ref.get(settlementCount), 0);
372 }),
373);
374
375it.effect("terminalizes a process-bound claim when success settlement fails", () =>
376 Effect.gen(function* () {
377 const now = DateTime.formatIso(yield* DateTime.now);
378 const effectId = "effect:worker-process-bound-settlement-failure";
379 const workerId = "worker-process-bound-settlement-failure";
380 const claimedEffect: EffectOutbox.OrchestrationEffectV2 = {
381 id: effectId,
382 commandId: CommandId.make("command:worker-process-bound-settlement-failure"),
383 threadId: ThreadId.make("thread:worker-process-bound-settlement-failure"),
384 request: {
385 type: "provider-turn.start",
386 runId: RunId.make("run:worker-process-bound-settlement-failure"),
387 },
388 status: "running",
389 attemptCount: 1,
390 availableAt: now,
391 leaseOwner: workerId,
392 leaseExpiresAt: now,
393 createdAt: now,
394 updatedAt: now,
395 completedAt: null,
396 lastError: null,
397 };
398 const retries = yield* Ref.make(0);
399 const terminalErrors = yield* Ref.make<ReadonlyArray<string>>([]);
400 const executionCount = yield* Ref.make(0);
401 const layerOutbox = Layer.mock(EffectOutbox.EffectOutboxV2)({
402 claimNext: () => Effect.succeed(Option.some(claimedEffect)),
403 get: () => Effect.succeed(Option.some(claimedEffect)),
404 awaitCancellation: () => Effect.never,
405 clearCancellation: () => Effect.void,
406 succeed: () =>
407 Effect.fail(
408 new EffectOutbox.EffectOutboxError({
409 operation: "succeed",
410 effectId,
411 cause: "simulated success settlement failure",
412 }),
413 ),
414 retry: () => Ref.update(retries, (count) => count + 1).pipe(Effect.as(true)),
415 fail: ({ error }) =>
416 Ref.update(terminalErrors, (existing) => [...existing, error]).pipe(Effect.as(true)),
417 });
418 const layerExecutor = Layer.succeed(
419 EffectWorker.OrchestrationEffectExecutorV2,
420 EffectWorker.OrchestrationEffectExecutorV2.of({
421 execute: () => Ref.update(executionCount, (count) => count + 1),
422 }),
423 );
424 const layerWorker = EffectWorker.layerWithOptions({ workerId }).pipe(
425 Layer.provide(Layer.merge(layerOutbox, layerExecutor)),
426 );
427
428 const exit = yield* EffectWorker.OrchestrationEffectWorkerV2.pipe(
429 Effect.flatMap((worker) => worker.runOnce),
430 Effect.provide(layerWorker),
431 Effect.exit,
432 );
433
434 assert.isTrue(Exit.isFailure(exit));
435 assert.equal(yield* Ref.get(executionCount), 1);
436 assert.equal(yield* Ref.get(retries), 0);
437 const terminalError = (yield* Ref.get(terminalErrors))[0];
438 assert.isDefined(terminalError);
439 assert.include(terminalError, "after execution started");
440 assert.include(terminalError, "simulated success settlement failure");
441 }),
442);
443
444it.effect("requeues a replay-safe claim when success settlement fails", () =>
445 Effect.gen(function* () {
446 const now = DateTime.formatIso(yield* DateTime.now);
447 const effectId = "effect:worker-replay-safe-settlement-failure";
448 const workerId = "worker-replay-safe-settlement-failure";
449 const claimedEffect: EffectOutbox.OrchestrationEffectV2 = {
450 id: effectId,
451 commandId: CommandId.make("command:worker-replay-safe-settlement-failure"),
452 threadId: ThreadId.make("thread:worker-replay-safe-settlement-failure"),
453 request: { type: "terminal.cleanup" },
454 status: "running",
455 attemptCount: 1,
456 availableAt: now,
457 leaseOwner: workerId,
458 leaseExpiresAt: now,
459 createdAt: now,
460 updatedAt: now,
461 completedAt: null,
462 lastError: null,
463 };
464 const retries = yield* Ref.make(0);
465 const terminalizations = yield* Ref.make(0);
466 const layerOutbox = Layer.mock(EffectOutbox.EffectOutboxV2)({
467 claimNext: () => Effect.succeed(Option.some(claimedEffect)),
468 get: () => Effect.succeed(Option.some(claimedEffect)),
469 awaitCancellation: () => Effect.never,
470 clearCancellation: () => Effect.void,
471 succeed: () =>
472 Effect.fail(
473 new EffectOutbox.EffectOutboxError({
474 operation: "succeed",
475 effectId,
476 cause: "simulated replay-safe settlement failure",
477 }),
478 ),
479 retry: () => Ref.update(retries, (count) => count + 1).pipe(Effect.as(true)),
480 fail: () => Ref.update(terminalizations, (count) => count + 1).pipe(Effect.as(true)),
481 });
482 const layerExecutor = Layer.succeed(
483 EffectWorker.OrchestrationEffectExecutorV2,
484 EffectWorker.OrchestrationEffectExecutorV2.of({ execute: () => Effect.void }),
485 );
486 const layerWorker = EffectWorker.layerWithOptions({ workerId }).pipe(
487 Layer.provide(Layer.merge(layerOutbox, layerExecutor)),
488 );
489
490 const exit = yield* EffectWorker.OrchestrationEffectWorkerV2.pipe(
491 Effect.flatMap((worker) => worker.runOnce),
492 Effect.provide(layerWorker),
493 Effect.exit,
494 );
495
496 assert.isTrue(Exit.isFailure(exit));
497 assert.equal(yield* Ref.get(retries), 1);
498 assert.equal(yield* Ref.get(terminalizations), 0);
499 }),
500);
501
502it.effect("keeps a process-bound executor failure retryable when retry settlement fails", () =>
503 Effect.gen(function* () {
504 const now = DateTime.formatIso(yield* DateTime.now);
505 const effectId = "effect:worker-process-bound-retry-settlement-failure";
506 const workerId = "worker-process-bound-retry-settlement-failure";
507 const claimedEffect: EffectOutbox.OrchestrationEffectV2 = {
508 id: effectId,
509 commandId: CommandId.make("command:worker-process-bound-retry-settlement-failure"),
510 threadId: ThreadId.make("thread:worker-process-bound-retry-settlement-failure"),
511 request: {
512 type: "provider-turn.start",
513 runId: RunId.make("run:worker-process-bound-retry-settlement-failure"),
514 },
515 status: "running",
516 attemptCount: 1,
517 availableAt: now,
518 leaseOwner: workerId,
519 leaseExpiresAt: now,
520 createdAt: now,
521 updatedAt: now,
522 completedAt: null,
523 lastError: null,
524 };
525 const retryAttempts = yield* Ref.make(0);
526 const terminalizations = yield* Ref.make(0);
527 const layerOutbox = Layer.mock(EffectOutbox.EffectOutboxV2)({
528 claimNext: () => Effect.succeed(Option.some(claimedEffect)),
529 get: () => Effect.succeed(Option.some(claimedEffect)),
530 awaitCancellation: () => Effect.never,
531 clearCancellation: () => Effect.void,
532 retry: () =>
533 Ref.updateAndGet(retryAttempts, (count) => count + 1).pipe(
534 Effect.flatMap((attempt) =>
535 attempt === 1
536 ? Effect.fail(
537 new EffectOutbox.EffectOutboxError({
538 operation: "retry",
539 effectId,
540 cause: "simulated retry settlement failure",
541 }),
542 )
543 : Effect.succeed(true),
544 ),
545 ),
546 fail: () => Ref.update(terminalizations, (count) => count + 1).pipe(Effect.as(true)),
547 });
548 const layerExecutor = Layer.succeed(
549 EffectWorker.OrchestrationEffectExecutorV2,
550 EffectWorker.OrchestrationEffectExecutorV2.of({
551 execute: () =>
552 Effect.fail(
553 new EffectWorker.OrchestrationEffectExecutionError({
554 effectId,
555 effectType: claimedEffect.request.type,
556 cause: "simulated provider execution failure",
557 }),
558 ),
559 }),
560 );
561 const layerWorker = EffectWorker.layerWithOptions({ workerId }).pipe(
562 Layer.provide(Layer.merge(layerOutbox, layerExecutor)),
563 );
564
565 const exit = yield* EffectWorker.OrchestrationEffectWorkerV2.pipe(
566 Effect.flatMap((worker) => worker.runOnce),
567 Effect.provide(layerWorker),
568 Effect.exit,
569 );
570
571 assert.isTrue(Exit.isFailure(exit));
572 assert.equal(yield* Ref.get(retryAttempts), 2);
573 assert.equal(yield* Ref.get(terminalizations), 0);
574 }),
575);
576
577it.effect("keeps a max-attempt replay-safe failure terminal when fail settlement fails", () =>
578 Effect.gen(function* () {
579 const now = DateTime.formatIso(yield* DateTime.now);
580 const effectId = "effect:worker-replay-safe-terminal-settlement-failure";
581 const workerId = "worker-replay-safe-terminal-settlement-failure";
582 const claimedEffect: EffectOutbox.OrchestrationEffectV2 = {
583 id: effectId,
584 commandId: CommandId.make("command:worker-replay-safe-terminal-settlement-failure"),
585 threadId: ThreadId.make("thread:worker-replay-safe-terminal-settlement-failure"),
586 request: { type: "terminal.cleanup" },
587 status: "running",
588 attemptCount: 5,
589 availableAt: now,
590 leaseOwner: workerId,
591 leaseExpiresAt: now,
592 createdAt: now,
593 updatedAt: now,
594 completedAt: null,
595 lastError: null,
596 };
597 const failAttempts = yield* Ref.make(0);
598 const retries = yield* Ref.make(0);
599 const layerOutbox = Layer.mock(EffectOutbox.EffectOutboxV2)({
600 claimNext: () => Effect.succeed(Option.some(claimedEffect)),
601 get: () => Effect.succeed(Option.some(claimedEffect)),
602 awaitCancellation: () => Effect.never,
603 clearCancellation: () => Effect.void,
604 fail: () =>
605 Ref.updateAndGet(failAttempts, (count) => count + 1).pipe(
606 Effect.flatMap((attempt) =>
607 attempt === 1
608 ? Effect.fail(
609 new EffectOutbox.EffectOutboxError({
610 operation: "fail",
611 effectId,
612 cause: "simulated terminal settlement failure",
613 }),
614 )
615 : Effect.succeed(true),
616 ),
617 ),
618 retry: () => Ref.update(retries, (count) => count + 1).pipe(Effect.as(true)),
619 });
620 const layerExecutor = Layer.succeed(
621 EffectWorker.OrchestrationEffectExecutorV2,
622 EffectWorker.OrchestrationEffectExecutorV2.of({
623 execute: () =>
624 Effect.fail(
625 new EffectWorker.OrchestrationEffectExecutionError({
626 effectId,
627 effectType: claimedEffect.request.type,
628 cause: "simulated terminal cleanup failure",
629 }),
630 ),
631 }),
632 );
633 const layerWorker = EffectWorker.layerWithOptions({ workerId, maxAttempts: 5 }).pipe(
634 Layer.provide(Layer.merge(layerOutbox, layerExecutor)),
635 );
636
637 const exit = yield* EffectWorker.OrchestrationEffectWorkerV2.pipe(
638 Effect.flatMap((worker) => worker.runOnce),
639 Effect.provide(layerWorker),
640 Effect.exit,
641 );
642
643 assert.isTrue(Exit.isFailure(exit));
644 assert.equal(yield* Ref.get(failAttempts), 2);
645 assert.equal(yield* Ref.get(retries), 0);
646 }),
647);
648
649it.effect("uses durable deadlines, notifications, and a slow liveness poll", () =>
650 Effect.gen(function* () {
651 const attempts = yield* Ref.make(0);
652 const available = yield* Queue.unbounded<void>();
653 const now = yield* DateTime.now;
654 const nextClaimableAt = yield* Ref.make<Option.Option<DateTime.Utc>>(
655 Option.some(DateTime.add(now, { milliseconds: 100 })),
656 );
657 const worker = EffectWorker.OrchestrationEffectWorkerV2.of({
658 awaitWork: Queue.take(available),
659 runRecoveryOnce: Effect.succeed(false),
660 runOnce: Effect.gen(function* () {
661 const count = yield* Ref.updateAndGet(attempts, (current) => current + 1);
662 if (count === 2) {
663 yield* Ref.set(nextClaimableAt, Option.some(DateTime.add(now, { milliseconds: 5_000 })));
664 }
665 if (count === 3) yield* Ref.set(nextClaimableAt, Option.none());
666 return false;
667 }),
668 nextClaimableAt: Ref.get(nextClaimableAt),
669 drain: () => Effect.succeed(0),
670 });
671 const awaitAttempts = Effect.fnUntraced(function* (expected: number) {
672 while ((yield* Ref.get(attempts)) < expected) {
673 yield* Effect.yieldNow;
674 }
675 });
676
677 yield* EffectWorker.runDaemonWithOptions({
678 concurrency: 1,
679 livenessPollIntervalMs: 1_000,
680 }).pipe(
681 Effect.provideService(EffectWorker.OrchestrationEffectWorkerV2, worker),
682 Effect.forkScoped,
683 );
684
685 yield* awaitAttempts(1);
686 yield* TestClock.adjust("99 millis");
687 assert.equal(yield* Ref.get(attempts), 1);
688
689 yield* TestClock.adjust("1 millis");
690 yield* awaitAttempts(2);
691 yield* TestClock.adjust("999 millis");
692 assert.equal(yield* Ref.get(attempts), 2);
693
694 yield* Queue.offer(available, undefined);
695 yield* awaitAttempts(3);
696 yield* TestClock.adjust("999 millis");
697 assert.equal(yield* Ref.get(attempts), 3);
698
699 yield* TestClock.adjust("1 millis");
700 yield* awaitAttempts(4);
701 }).pipe(Effect.provide(TestClock.layer())),
702);
703
704it.effect("does not hot-loop when a claim fails", () =>
705 Effect.gen(function* () {
706 const attempts = yield* Ref.make(0);
707 const now = yield* DateTime.now;
708 const worker = EffectWorker.OrchestrationEffectWorkerV2.of({
709 awaitWork: Effect.never,
710 runRecoveryOnce: Effect.succeed(false),
711 runOnce: Ref.update(attempts, (count) => count + 1).pipe(
712 Effect.andThen(
713 new EffectWorker.OrchestrationEffectWorkerError({
714 operation: "claim",
715 cause: "simulated database failure",
716 }),
717 ),
718 ),
719 nextClaimableAt: Effect.succeed(Option.some(now)),
720 drain: () => Effect.succeed(0),
721 });
722
723 yield* EffectWorker.runDaemonWithOptions({
724 concurrency: 1,
725 livenessPollIntervalMs: 1_000,
726 }).pipe(
727 Effect.provideService(EffectWorker.OrchestrationEffectWorkerV2, worker),
728 Effect.forkScoped,
729 );
730
731 while ((yield* Ref.get(attempts)) < 1) yield* Effect.yieldNow;
732 yield* TestClock.adjust("999 millis");
733 assert.equal(yield* Ref.get(attempts), 1);
734 yield* TestClock.adjust("1 millis");
735 while ((yield* Ref.get(attempts)) < 2) yield* Effect.yieldNow;
736 }).pipe(Effect.provide(TestClock.layer())),
737);
738
739it.effect("backs off briefly when a due deadline loses a claim race", () =>
740 Effect.gen(function* () {
741 const attempts = yield* Ref.make(0);
742 const now = yield* DateTime.now;
743 const worker = EffectWorker.OrchestrationEffectWorkerV2.of({
744 awaitWork: Effect.never,
745 runRecoveryOnce: Effect.succeed(false),
746 runOnce: Ref.update(attempts, (count) => count + 1).pipe(Effect.as(false)),
747 nextClaimableAt: Effect.succeed(Option.some(now)),
748 drain: () => Effect.succeed(0),
749 });
750
751 yield* EffectWorker.runDaemonWithOptions({
752 concurrency: 1,
753 livenessPollIntervalMs: 1_000,
754 }).pipe(
755 Effect.provideService(EffectWorker.OrchestrationEffectWorkerV2, worker),
756 Effect.forkScoped,
757 );
758
759 while ((yield* Ref.get(attempts)) < 1) yield* Effect.yieldNow;
760 yield* TestClock.adjust("24 millis");
761 assert.equal(yield* Ref.get(attempts), 1);
762 yield* TestClock.adjust("1 millis");
763 while ((yield* Ref.get(attempts)) < 2) yield* Effect.yieldNow;
764 }).pipe(Effect.provide(TestClock.layer())),
765);
766
767it.effect("safely retries after replacement cleanup succeeds and start fails", () =>
768 Effect.gen(function* () {
769 const now = yield* DateTime.now;
770 const events = yield* Ref.make<ReadonlyArray<string>>([]);
771 const failFirstStart = yield* Ref.make(true);
772 const effect = restartEffect(now, {
773 type: "replace",
774 replacementProviderSessionId: replacementSessionId,
775 });
776 const layer = layerExecutorFor({ events, failFirstStart });
777
778 const first = yield* Effect.gen(function* () {
779 const executor = yield* EffectWorker.OrchestrationEffectExecutorV2;
780 return yield* Effect.exit(executor.execute(effect));
781 }).pipe(Effect.provide(layer));
782 assert.isTrue(Exit.isFailure(first));
783
784 yield* Effect.gen(function* () {
785 const executor = yield* EffectWorker.OrchestrationEffectExecutorV2;
786 yield* executor.execute(effect);
787 }).pipe(Effect.provide(layer));
788
789 assert.deepEqual(yield* Ref.get(events), [
790 `interrupt:${replacementSessionId}`,
791 "detach",
792 "start",
793 `interrupt:${replacementSessionId}`,
794 "detach",
795 "start",
796 ]);
797 }),
798);
799
800it.effect("settles a delegated child once its restart continuation fails for good", () =>
801 Effect.gen(function* () {
802 const timestamp = DateTime.formatIso(yield* DateTime.now);
803 const events = yield* Ref.make<ReadonlyArray<string>>([]);
804 const recovered = yield* Ref.make<ReadonlyArray<ThreadId>>([]);
805 const layer = layerExecutorFor({
806 events,
807 continueAfterRestart: true,
808 threads: {
809 getThreadRecords: () => Effect.fail(new Error("provider instance removed") as never),
810 recoverDelegatedTask: (childThreadId) =>
811 Ref.update(recovered, (ids) => [...ids, childThreadId]),
812 },
813 });
814 const effect: EffectOutbox.OrchestrationEffectV2 = {
815 id: `effect:restart-continuation:${runId}`,
816 commandId: CommandId.make("command:restart-continuation-failure"),
817 threadId,
818 request: { type: "provider-runtime.continue", sourceRunId: runId },
819 status: "running",
820 attemptCount: 1,
821 availableAt: timestamp,
822 leaseOwner: "test-worker",
823 leaseExpiresAt: timestamp,
824 createdAt: timestamp,
825 updatedAt: timestamp,
826 completedAt: null,
827 lastError: null,
828 };
829 yield* Effect.gen(function* () {
830 const executor = yield* EffectWorker.OrchestrationEffectExecutorV2;
831 assert.isTrue(
832 Exit.isFailure(yield* Effect.exit(executor.execute(effect, { willRetry: true }))),
833 );
834 assert.deepEqual(yield* Ref.get(recovered), []);
835 assert.isTrue(
836 Exit.isFailure(yield* Effect.exit(executor.execute(effect, { willRetry: false }))),
837 );
838 assert.deepEqual(yield* Ref.get(recovered), [threadId]);
839 }).pipe(Effect.provide(layer));
840 }),
841);
842
843it.effect.each(["confirmed", "unconfirmed", "failed"] as const)(
844 "a native subagent stop only repairs %s provider evidence",
845 (outcome) =>
846 Effect.gen(function* () {
847 const now = yield* DateTime.now;
848 const events = yield* Ref.make<ReadonlyArray<string>>([]);
849 const layer = layerExecutorFor({
850 events,
851 interrupt: (input) => {
852 assert.equal(input.subagent?.nativeTaskId, "native-task");
853 return Ref.update(events, (current) => [...current, "stop-child"]).pipe(
854 Effect.andThen(
855 outcome === "failed"
856 ? new ProviderTurnControlService.ProviderTurnControlError({
857 threadId,
858 providerTurnId,
859 operation: "interrupt",
860 cause: "SDK stop failed",
861 })
862 : Effect.succeed(
863 outcome === "confirmed"
864 ? {
865 status: "cancelled" as const,
866 completedAt: now,
867 result: "Stopped by Claude",
868 }
869 : undefined,
870 ),
871 ),
872 );
873 },
874 threads: {
875 dispatch: (command) => {
876 assert.equal(command.type, "thread.background-work.settle");
877 if (command.type === "thread.background-work.settle") {
878 assert.equal(command.subagent?.id, "subagent");
879 assert.deepEqual(command.subagent?.outcome, {
880 status: "cancelled",
881 completedAt: now,
882 result: "Stopped by Claude",
883 });
884 }
885 return Ref.update(events, (current) => [...current, "settle-child"]).pipe(
886 Effect.as({ sequence: 1, storedEvents: [] }),
887 );
888 },
889 },
890 });
891 const execution = yield* Effect.gen(function* () {
892 const executor = yield* EffectWorker.OrchestrationEffectExecutorV2;
893 yield* executor.execute({
894 ...restartEffect(now, { type: "detach" }),
895 request: {
896 type: "provider-turn.interrupt",
897 providerSessionId: oldSessionId,
898 providerThreadId,
899 providerTurnId,
900 subagent: { id: NodeId.make("subagent"), nativeTaskId: "native-task" },
901 },
902 });
903 }).pipe(Effect.provide(layer), Effect.exit);
904 assert.equal(Exit.isFailure(execution), outcome === "failed");
905 assert.deepEqual(
906 yield* Ref.get(events),
907 outcome === "confirmed" ? ["stop-child", "settle-child"] : ["stop-child"],
908 );
909 }),
910);