1import { assert, it } from "@effect/vitest";
2import {
3 MessageId,
4 NodeId,
5 ProjectId,
6 ProviderDriverKind,
7 ProviderInstanceId,
8 ProviderSessionId,
9 ProviderThreadId,
10 ProviderTurnId,
11 RunAttemptId,
12 RunId,
13 RuntimeRequestId,
14 ThreadId,
15 type OrchestrationV2Command,
16 type OrchestrationV2ThreadProjection,
17} from "@t3tools/contracts";
18import * as DateTime from "effect/DateTime";
19import * as Effect from "effect/Effect";
20import * as Deferred from "effect/Deferred";
21import * as Fiber from "effect/Fiber";
22import * as Layer from "effect/Layer";
23import * as ServerSettings from "../serverSettings.ts";
24import { restartContinuationRun, continueRestartedRun } from "./RestartContinuation.ts";
25import { continuationStartRetryDelayMs } from "./RestartBackgroundNote.ts";
26import * as ThreadManagementService from "./ThreadManagementService.ts";
27import * as ProviderRuntimeRecovery from "./ProviderRuntimeRecoveryService.ts";
28import * as ProjectionStore from "./ProjectionStore.ts";
29import * as EventSink from "./EventSink.ts";
30import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
31import * as EffectWorker from "./EffectWorker.ts";
32import * as EffectOutbox from "./EffectOutbox.ts";
33
34const threadId = ThreadId.make("thread:restart");
35const runId = RunId.make("run:restart");
36const instanceId = ProviderInstanceId.make("codex");
37const driver = ProviderDriverKind.make("codex");
38const providerThreadId = ProviderThreadId.make("provider-thread:restart");
39const sessionId = ProviderSessionId.make("session:restart");
40const attemptId = RunAttemptId.make("attempt:restart");
41const scratchProjectId = ProjectId.make("project:scratch");
42const now = DateTime.makeUnsafe("2026-10-03T12:00:00.000Z");
43
44function makeProjection() {
45 return {
46 thread: {
47 id: threadId,
48 projectId: ProjectId.make("restart-project"),
49 providerInstanceId: instanceId,
50 archivedAt: null,
51 deletedAt: null,
52 lineage: { parentThreadId: null, relationshipToParent: null },
53 forkedFrom: null,
54 },
55 runs: [
56 {
57 id: runId,
58 ordinal: 1,
59 providerInstanceId: instanceId,
60 modelSelection: { instanceId, model: "gpt-6" },
61 providerThreadId,
62 activeAttemptId: attemptId,
63 status: "running",
64 requestedAt: now,
65 completedAt: null,
66 },
67 ],
68 providerThreads: [
69 {
70 id: providerThreadId,
71 appThreadId: threadId,
72 ownerNodeId: null,
73 driver,
74 providerInstanceId: instanceId,
75 providerSessionId: sessionId,
76 nativeThreadRef: { driver, nativeId: "native-thread", strength: "strong" },
77 status: "active",
78 },
79 ],
80 providerSessions: [
81 { id: sessionId, driver, providerInstanceId: instanceId, status: "running" },
82 ],
83 providerTurns: [
84 {
85 id: ProviderTurnId.make("turn:restart"),
86 providerThreadId,
87 runAttemptId: attemptId,
88 status: "running",
89 },
90 ],
91 runtimeRequests: [],
92 attempts: [],
93 nodes: [],
94 subagents: [],
95 messages: [],
96 turnItems: [],
97 } as unknown as OrchestrationV2ThreadProjection;
98}
99
100it("requires matching saved native state for an unfinished root run", () => {
101 const projection = makeProjection();
102 assert.equal(restartContinuationRun(projection, now)?.id, runId);
103 for (const invalid of [
104 { ...projection, thread: { ...projection.thread, archivedAt: {} } },
105 { ...projection, thread: { ...projection.thread, deletedAt: {} } },
106 {
107 ...projection,
108 thread: { ...projection.thread, providerInstanceId: ProviderInstanceId.make("other") },
109 },
110 {
111 ...projection,
112 providerThreads: [{ ...projection.providerThreads[0]!, nativeThreadRef: null }],
113 },
114 {
115 ...projection,
116 providerSessions: [
117 {
118 ...projection.providerSessions[0]!,
119 providerInstanceId: ProviderInstanceId.make("other"),
120 },
121 ],
122 },
123 { ...projection, providerTurns: [] },
124 ...[
125 "queued",
126 "preparing",
127 "starting",
128 "waiting",
129 "completed",
130 "cancelled",
131 "failed",
132 "interrupted",
133 ].map((status) => ({ ...projection, runs: [{ ...projection.runs[0]!, status }] })),
134 ])
135 assert.isUndefined(restartContinuationRun(invalid as OrchestrationV2ThreadProjection, now));
136});
137
138it("continues a live turn whose session the adapter never marked running", () => {
139 const projection = makeProjection();
140 const withSessionStatus = (status: string) =>
141 ({
142 ...projection,
143 providerSessions: [{ ...projection.providerSessions[0]!, status }],
144 }) as OrchestrationV2ThreadProjection;
145 for (const status of ["starting", "ready", "running", "waiting"])
146 assert.equal(restartContinuationRun(withSessionStatus(status), now)?.id, runId, status);
147 for (const status of ["stopped", "error"])
148 assert.isUndefined(restartContinuationRun(withSessionStatus(status), now), status);
149});
150
151it("recovers an admitted continuation after another crash before provider start", () => {
152 const projection = makeProjection();
153 const starting = {
154 ...projection,
155 runs: [
156 {
157 ...projection.runs[0]!,
158 status: "starting" as const,
159 restartContinuationOfRunId: RunId.make("run:previous-crash"),
160 },
161 ],
162 providerThreads: [{ ...projection.providerThreads[0]!, status: "idle" as const }],
163 providerSessions: [{ ...projection.providerSessions[0]!, status: "stopped" as const }],
164 providerTurns: [],
165 };
166 assert.equal(restartContinuationRun(starting, now)?.id, runId);
167});
168
169it("does not continue settled root runs with restart-cancelled background work", () => {
170 const projection = makeProjection();
171 const settled = {
172 ...projection,
173 runs: [
174 {
175 ...projection.runs[0]!,
176 status: "completed" as const,
177 restartCancelledBackgroundWork: [{ kind: "shell" as const, label: "sleep 25" }],
178 },
179 ],
180 providerThreads: [{ ...projection.providerThreads[0]!, status: "idle" as const }],
181 providerSessions: [{ ...projection.providerSessions[0]!, status: "stopped" as const }],
182 providerTurns: [],
183 };
184 for (const projectId of [scratchProjectId, projection.thread.projectId])
185 for (const status of ["completed", "waiting"] as const)
186 assert.isUndefined(
187 restartContinuationRun(
188 {
189 ...settled,
190 thread: { ...settled.thread, projectId },
191 runs: [{ ...settled.runs[0]!, status }],
192 },
193 now,
194 ),
195 );
196});
197
198it.effect.each(["completed", "waiting", "cancelled"] as const)(
199 "wakes a %s run whose awaited monitor the restart cancelled, with the note as its prompt",
200 (status) =>
201 Effect.gen(function* () {
202 const base = makeProjection();
203 const work = [
204 { kind: "monitor" as const, label: "Watch CI run 42", awaited: true as const },
205 { kind: "shell" as const, label: "vp run dev" },
206 ];
207 const projection = {
208 ...base,
209 runs: [{ ...base.runs[0]!, status, restartCancelledBackgroundWork: work }],
210 providerTurns: [{ ...base.providerTurns[0]!, status: "completed" }],
211 } as unknown as OrchestrationV2ThreadProjection;
212 const commands: Parameters<
213 ThreadManagementService.ThreadManagementService["Service"]["dispatch"]
214 >[0][] = [];
215 yield* continueRestartedRun({ threadId, sourceRunId: runId }).pipe(
216 Effect.provide(
217 Layer.merge(
218 Layer.mock(ThreadManagementService.ThreadManagementService)({
219 getThreadRecords: () => Effect.succeed(projection),
220 recoverDelegatedTask: () => Effect.void,
221 dispatch: (command) => {
222 commands.push(command);
223 return Effect.succeed({} as never);
224 },
225 }),
226 ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true }),
227 ),
228 ),
229 );
230 assert.lengthOf(commands, 1);
231 const command = commands[0]!;
232 assert.equal(command.type, "message.dispatch");
233 if (command.type !== "message.dispatch") return;
234 assert.equal(command.restartContinuationOfRunId, runId);
235 assert.include(command.text, "- monitor: Watch CI run 42");
236 assert.include(command.text, "- shell: vp run dev");
237 }),
238);
239
240it.effect.each([
241 [false, undefined],
242 [true, undefined],
243 [false, true],
244 [true, false],
245] as const)(
246 "atomically records restart intent with cancellation when opt-in is %s and project override is %s",
247 ([enabled, projectOverride]) =>
248 Effect.gen(function* () {
249 let committed: Parameters<EventSink.EventSinkV2["Service"]["commitCommand"]>[0] | undefined;
250 const recovery = yield* ProviderRuntimeRecovery.make.pipe(
251 Effect.provide(
252 Layer.mergeAll(
253 ServerSettings.layerTest({
254 continueThreadsAfterServerUpdate: enabled,
255 projectSettingsOverrides:
256 projectOverride === undefined
257 ? {}
258 : {
259 [ProjectId.make("restart-project")]: {
260 continueThreadsAfterServerUpdate: projectOverride,
261 },
262 },
263 }),
264 Layer.mock(ProjectionStore.ProjectionStoreV2)({
265 getRecoveryThreadIds: () => Effect.succeed([threadId]),
266 getRuntimeRecoveryProjection: () => Effect.succeed(makeProjection()),
267 }),
268 Layer.mock(EventSink.EventSinkV2)({
269 commitCommand: (input) => {
270 committed = input;
271 return Effect.succeed({ committed: true, cancelledEffectCount: 1 } as never);
272 },
273 }),
274 IdAllocator.layer,
275 Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({
276 runRecoveryOnce: Effect.succeed(false),
277 }),
278 Layer.mock(EffectOutbox.EffectOutboxV2)({
279 reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }),
280 }),
281 ),
282 ),
283 );
284 yield* recovery.reconcile("startup");
285 assert.isDefined(committed);
286 assert.isTrue(
287 committed!.events.some(
288 (event) => event.type === "run.updated" && event.payload.status === "cancelled",
289 ),
290 );
291 assert.lengthOf(committed!.effects, (projectOverride ?? enabled) ? 1 : 0);
292 if (projectOverride ?? enabled)
293 assert.deepEqual(committed!.effects[0]?.request, {
294 type: "provider-runtime.continue",
295 sourceRunId: runId,
296 });
297 }),
298);
299
300it.effect("lets a cut blocking delegate_task wake its parent like an async task", () =>
301 Effect.gen(function* () {
302 const base = makeProjection();
303 const task = (id: string, completionWake: "settled_only" | "always", status: string) => ({
304 id: NodeId.make(id),
305 threadId,
306 runId,
307 origin: "app_owned",
308 childThreadId: ThreadId.make(`thread:${id}`),
309 driver,
310 providerInstanceId: instanceId,
311 completionWake,
312 status,
313 });
314 const projection = {
315 ...base,
316 subagents: [
317 task("task:waited", "settled_only", "running"),
318 task("task:async", "always", "running"),
319 task("task:finished", "settled_only", "completed"),
320 ],
321 } as unknown as OrchestrationV2ThreadProjection;
322 let committed: Parameters<EventSink.EventSinkV2["Service"]["commitCommand"]>[0] | undefined;
323 const recovery = yield* ProviderRuntimeRecovery.make.pipe(
324 Effect.provide(
325 Layer.mergeAll(
326 ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true }),
327 Layer.mock(ProjectionStore.ProjectionStoreV2)({
328 getRecoveryThreadIds: () => Effect.succeed([threadId]),
329 getRuntimeRecoveryProjection: () => Effect.succeed(projection),
330 }),
331 Layer.mock(EventSink.EventSinkV2)({
332 commitCommand: (input) => {
333 committed = input;
334 return Effect.succeed({ committed: true, cancelledEffectCount: 0 } as never);
335 },
336 }),
337 IdAllocator.layer,
338 Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({}),
339 Layer.mock(EffectOutbox.EffectOutboxV2)({
340 reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }),
341 }),
342 ),
343 ),
344 );
345 yield* recovery.reconcile("startup");
346 const upgraded = (committed?.events ?? []).flatMap((event) =>
347 event.type === "subagent.updated" ? [event.payload] : [],
348 );
349 assert.deepEqual(
350 upgraded.map((row) => ({ id: row.id, completionWake: row.completionWake })),
351 [{ id: NodeId.make("task:waited"), completionWake: "always" }],
352 );
353 }),
354);
355
356it.effect.each(["running", "waiting"])(
357 "wakes a %s run to restart the native subagent it waited on",
358 (status) =>
359 Effect.gen(function* () {
360 const base = makeProjection();
361 const item = (id: string, extra: Record<string, unknown>) => ({
362 id,
363 threadId,
364 runId,
365 nodeId: null,
366 status: "running",
367 ...extra,
368 });
369 const projection = {
370 ...base,
371 runs: [{ ...base.runs[0]!, status }],
372 providerTurns: [
373 { ...base.providerTurns[0]!, status: status === "waiting" ? "completed" : "running" },
374 ],
375 turnItems: [
376 item("item:native", {
377 type: "subagent",
378 origin: "provider_native",
379 providerThreadId: ProviderThreadId.make("provider-thread:child"),
380 nativeItemRef: { driver, nativeId: "agent-1" },
381 title: "Sleeper",
382 childThreadId: ThreadId.make("thread:child"),
383 }),
384 item("item:shell", { type: "command_execution", providerThreadId, input: "ls" }),
385 ],
386 } as unknown as OrchestrationV2ThreadProjection;
387 let committed: Parameters<EventSink.EventSinkV2["Service"]["commitCommand"]>[0] | undefined;
388 const recovery = yield* ProviderRuntimeRecovery.make.pipe(
389 Effect.provide(
390 Layer.mergeAll(
391 ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true }),
392 Layer.mock(ProjectionStore.ProjectionStoreV2)({
393 getRecoveryThreadIds: () => Effect.succeed([threadId]),
394 getRuntimeRecoveryProjection: () => Effect.succeed(projection),
395 }),
396 Layer.mock(EventSink.EventSinkV2)({
397 commitCommand: (input) => {
398 committed = input;
399 return Effect.succeed({ committed: true, cancelledEffectCount: 0 } as never);
400 },
401 }),
402 IdAllocator.layer,
403 Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({}),
404 Layer.mock(EffectOutbox.EffectOutboxV2)({
405 listByCommandId: () => Effect.succeed([]),
406 reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }),
407 }),
408 ),
409 ),
410 );
411 yield* recovery.reconcile("startup");
412 const work = (committed?.events ?? []).flatMap((event) =>
413 event.type === "run.background-work-cancelled" ? [event.payload] : [],
414 );
415 assert.deepEqual(work, [
416 {
417 runId,
418 restartCancelledBackgroundWork: [
419 { kind: "subagent", label: "Sleeper (id agent-1)", id: "item:native" },
420 ...(status === "waiting"
421 ? [{ kind: "shell" as const, label: "ls", id: "item:shell" }]
422 : []),
423 ],
424 },
425 ]);
426 assert.deepEqual(
427 committed?.effects.map((effect) => effect.request),
428 [{ type: "provider-runtime.continue", sourceRunId: runId }],
429 );
430 }),
431);
432
433it.effect("does not duplicate delivery and yields to newer user work or opt-out", () =>
434 Effect.gen(function* () {
435 let projection = makeProjection();
436 projection = { ...projection, runs: [{ ...projection.runs[0]!, status: "cancelled" }] };
437 const commands: Parameters<
438 ThreadManagementService.ThreadManagementService["Service"]["dispatch"]
439 >[0][] = [];
440 const layerThreads = Layer.mock(ThreadManagementService.ThreadManagementService)({
441 getThreadRecords: () => Effect.succeed(projection),
442 recoverDelegatedTask: () => Effect.void,
443 dispatch: (command) => {
444 commands.push(command);
445 if (command.type === "message.dispatch")
446 projection = { ...projection, messages: [{ id: command.messageId } as never] };
447 return Effect.succeed({} as never);
448 },
449 });
450 const layerEnabled = Layer.merge(
451 layerThreads,
452 ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true }),
453 );
454 yield* continueRestartedRun({ threadId, sourceRunId: runId }).pipe(
455 Effect.provide(layerEnabled),
456 );
457 yield* continueRestartedRun({ threadId, sourceRunId: runId }).pipe(
458 Effect.provide(layerEnabled),
459 );
460 assert.lengthOf(commands, 1);
461 assert.match(String(commands[0]!.commandId), /run:restart$/);
462 if (commands[0]!.type === "message.dispatch")
463 assert.equal(commands[0]!.restartContinuationOfRunId, runId);
464 projection = {
465 ...projection,
466 messages: [],
467 runs: [
468 ...projection.runs,
469 {
470 ...projection.runs[0]!,
471 id: RunId.make("run:user-newer"),
472 ordinal: 2,
473 status: "completed",
474 },
475 ],
476 };
477 yield* continueRestartedRun({ threadId, sourceRunId: runId }).pipe(
478 Effect.provide(layerEnabled),
479 );
480 assert.lengthOf(commands, 1);
481 projection = { ...projection, runs: [projection.runs[0]!] };
482 yield* continueRestartedRun({ threadId, sourceRunId: runId }).pipe(
483 Effect.provide(
484 Layer.merge(
485 layerThreads,
486 ServerSettings.layerTest({ continueThreadsAfterServerUpdate: false }),
487 ),
488 ),
489 );
490 assert.lengthOf(commands, 1);
491 }),
492);
493
494it.effect("prepares no continuation for background work another provider thread launched", () =>
495 Effect.gen(function* () {
496 const base = makeProjection();
497 const claudeThreadId = ProviderThreadId.make("provider-thread:claude");
498 const projection = {
499 ...base,
500 runs: [{ ...base.runs[0]!, status: "completed" }],
501 providerTurns: [{ ...base.providerTurns[0]!, status: "completed" }],
502 turnItems: [
503 {
504 id: "turn-item:claude-subagent",
505 runId: RunId.make("run:claude"),
506 providerThreadId: claudeThreadId,
507 type: "subagent",
508 status: "running",
509 },
510 ],
511 } as unknown as OrchestrationV2ThreadProjection;
512 const writes: Parameters<EventSink.EventSinkV2["Service"]["writeWithEffects"]>[0][] = [];
513 const recovery = yield* ProviderRuntimeRecovery.make.pipe(
514 Effect.provide(
515 Layer.mergeAll(
516 ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true }),
517 Layer.mock(ProjectionStore.ProjectionStoreV2)({
518 getRecoveryThreadIds: () => Effect.succeed([threadId]),
519 getRuntimeRecoveryProjection: () => Effect.succeed(projection),
520 }),
521 Layer.mock(EventSink.EventSinkV2)({
522 writeWithEffects: (input) =>
523 Effect.sync(() => {
524 writes.push(input);
525 return [];
526 }),
527 }),
528 IdAllocator.layer,
529 Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({}),
530 Layer.mock(EffectOutbox.EffectOutboxV2)({}),
531 ),
532 ),
533 );
534 yield* recovery.prepareForShutdown;
535 assert.lengthOf(writes, 0);
536 const ownWork = {
537 ...projection,
538 turnItems: [{ ...projection.turnItems[0]!, providerThreadId }],
539 } as OrchestrationV2ThreadProjection;
540 const ownRecovery = yield* ProviderRuntimeRecovery.make.pipe(
541 Effect.provide(
542 Layer.mergeAll(
543 ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true }),
544 Layer.mock(ProjectionStore.ProjectionStoreV2)({
545 getRecoveryThreadIds: () => Effect.succeed([threadId]),
546 getRuntimeRecoveryProjection: () => Effect.succeed(ownWork),
547 }),
548 Layer.mock(EventSink.EventSinkV2)({
549 writeWithEffects: (input) =>
550 Effect.sync(() => {
551 writes.push(input);
552 return [];
553 }),
554 }),
555 IdAllocator.layer,
556 Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({}),
557 Layer.mock(EffectOutbox.EffectOutboxV2)({}),
558 ),
559 ),
560 );
561 yield* ownRecovery.prepareForShutdown;
562 assert.lengthOf(writes, 1);
563 assert.deepEqual(writes[0]!.effects[0]?.request, {
564 type: "provider-runtime.continue",
565 sourceRunId: runId,
566 });
567 }),
568);
569
570const backgroundSubagent = {
571 turnItem: {
572 id: "turn-item:background-subagent",
573 runId,
574 nodeId: null,
575 providerThreadId,
576 providerInstanceId: instanceId,
577 type: "subagent",
578 subagentId: "subagent:background",
579 title: "Background reviewer",
580 status: "running",
581 },
582 work: {
583 kind: "subagent" as const,
584 label: "Background reviewer",
585 id: "turn-item:background-subagent",
586 },
587};
588const devServer = {
589 turnItem: {
590 id: "turn-item:dev-server",
591 runId,
592 nodeId: null,
593 providerThreadId,
594 providerInstanceId: instanceId,
595 type: "command_execution",
596 input: "vp run dev",
597 status: "running",
598 },
599 work: { kind: "shell" as const, label: "vp run dev", id: "turn-item:dev-server" },
600};
601
602it.effect.each([
603 ["completed", scratchProjectId, backgroundSubagent, true],
604 ["waiting", ProjectId.make("restart-project"), backgroundSubagent, true],
605 ["completed", ProjectId.make("restart-project"), devServer, false],
606 ["waiting", scratchProjectId, devServer, false],
607] as const)(
608 "cleans up background work of a %s run in project %s and wakes it only for awaited work",
609 ([status, projectId, background, wakes]) =>
610 Effect.gen(function* () {
611 const base = makeProjection();
612 const projection = {
613 ...base,
614 thread: { ...base.thread, projectId },
615 runs: [{ ...base.runs[0]!, status }],
616 providerTurns: [{ ...base.providerTurns[0]!, status: "completed" }],
617 turnItems: [background.turnItem],
618 subagents:
619 background === backgroundSubagent
620 ? [
621 {
622 id: "subagent:background",
623 runId,
624 origin: "provider_native",
625 driver,
626 providerInstanceId: instanceId,
627 status: "running",
628 },
629 ]
630 : [],
631 } as unknown as OrchestrationV2ThreadProjection;
632 for (const trigger of ["startup", "shutdown"] as const) {
633 const commits: Parameters<EventSink.EventSinkV2["Service"]["commitCommand"]>[0][] = [];
634 const writes: Parameters<EventSink.EventSinkV2["Service"]["writeWithEffects"]>[0][] = [];
635 const recovery = yield* ProviderRuntimeRecovery.make.pipe(
636 Effect.provide(
637 Layer.mergeAll(
638 ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true }),
639 Layer.mock(ProjectionStore.ProjectionStoreV2)({
640 getRecoveryThreadIds: () => Effect.succeed([threadId]),
641 getRuntimeRecoveryProjection: () => Effect.succeed(projection),
642 }),
643 Layer.mock(EventSink.EventSinkV2)({
644 writeWithEffects: (input) =>
645 Effect.sync(() => {
646 writes.push(input);
647 return [];
648 }),
649 commitCommand: (input) =>
650 Effect.sync(() => {
651 commits.push(input);
652 return { committed: true, cancelledEffectCount: 0 } as never;
653 }),
654 }),
655 IdAllocator.layer,
656 Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({
657 runRecoveryOnce: Effect.succeed(false),
658 }),
659 Layer.mock(EffectOutbox.EffectOutboxV2)({
660 listByCommandId: () => Effect.succeed([]),
661 reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }),
662 }),
663 ),
664 ),
665 );
666 if (trigger === "shutdown") yield* recovery.prepareForShutdown;
667 yield* recovery.reconcile(trigger);
668 const recorded = [...writes.flatMap((write) => write.effects), ...commits[0]!.effects];
669 assert.deepEqual(
670 recorded.map((effect) => effect.request),
671 wakes ? [{ type: "provider-runtime.continue", sourceRunId: runId }] : [],
672 );
673 assert.lengthOf(commits, 1);
674 const events = commits[0]!.events;
675 assert.isTrue(
676 events.some(
677 (event) => event.type === "turn-item.updated" && event.payload.status === "cancelled",
678 ),
679 );
680 const note = events.find((event) => event.type === "run.background-work-cancelled");
681 assert.equal(note?.runId, runId);
682 assert.deepEqual(note?.payload.restartCancelledBackgroundWork, [background.work]);
683 if (status === "completed")
684 assert.isFalse(events.some((event) => event.type === "run.updated"));
685 }
686 }),
687);
688
689it("does not wake a settled run for work left by a session that already stopped", () => {
690 const base = makeProjection();
691 const projection = {
692 ...base,
693 runs: [{ ...base.runs[0]!, status: "completed" }],
694 providerThreads: [{ ...base.providerThreads[0]!, status: "idle" }],
695 turnItems: [backgroundSubagent.turnItem],
696 } as unknown as OrchestrationV2ThreadProjection;
697 assert.equal(restartContinuationRun(projection, now)?.id, runId);
698 for (const status of ["stopped", "error"])
699 assert.isUndefined(
700 restartContinuationRun(
701 {
702 ...projection,
703 providerSessions: [{ ...projection.providerSessions[0]!, status }],
704 } as OrchestrationV2ThreadProjection,
705 now,
706 ),
707 status,
708 );
709});
710
711it.effect("wakes a run whose subagent's own shell the restart cut, past a wake it cut first", () =>
712 Effect.gen(function* () {
713 const base = makeProjection();
714 const cutWake = {
715 ...base.runs[0]!,
716 id: RunId.make("run:wake"),
717 ordinal: 2,
718 status: "starting",
719 startedAt: null,
720 };
721 const withShell = (parentTaskId?: string) =>
722 ({
723 ...base,
724 runs: [{ ...base.runs[0]!, status: "waiting", startedAt: now }, cutWake],
725 providerThreads: [
726 {
727 ...base.providerThreads[0]!,
728 status: "idle",
729 pendingBackgroundTasks: [
730 {
731 kind: "command",
732 taskId: "shell:1",
733 description: "sleep 150",
734 ...(parentTaskId === undefined ? {} : { parentTaskId }),
735 },
736 ],
737 },
738 ],
739 providerTurns: [{ ...base.providerTurns[0]!, status: "completed" }],
740 }) as unknown as OrchestrationV2ThreadProjection;
741 assert.equal(restartContinuationRun(withShell("agent:1"), now)?.id, runId);
742 assert.isUndefined(restartContinuationRun(withShell(), now));
743
744 const texts = yield* continuationTexts({
745 ...base,
746 runs: [
747 {
748 ...base.runs[0]!,
749 status: "completed",
750 startedAt: now,
751 completedAt: now,
752 restartCancelledBackgroundWork: [
753 { kind: "subagent", label: "sleep 150 (subagent agent:1)", id: "agent:1" },
754 ],
755 },
756 { ...cutWake, status: "cancelled", completedAt: now },
757 ],
758 providerTurns: [{ ...base.providerTurns[0]!, status: "completed" }],
759 } as unknown as OrchestrationV2ThreadProjection);
760 assert.lengthOf(texts, 1);
761 assert.include(texts[0], "- subagent: sleep 150 (subagent agent:1)");
762 const stopped = yield* continuationTexts({
763 ...base,
764 runs: [
765 {
766 ...base.runs[0]!,
767 status: "completed",
768 startedAt: now,
769 completedAt: now,
770 restartCancelledBackgroundWork: [
771 { kind: "subagent", label: "sleep 150 (subagent agent:1)", id: "agent:1" },
772 ],
773 },
774 { ...cutWake, status: "interrupted", completedAt: now },
775 ],
776 providerTurns: [{ ...base.providerTurns[0]!, status: "completed" }],
777 } as unknown as OrchestrationV2ThreadProjection);
778 assert.lengthOf(stopped, 0);
779 }),
780);
781
782it.effect("does not cancel or resume a run that completes while shutdown intent commits", () =>
783 Effect.gen(function* () {
784 let projection = makeProjection();
785 const commits: Parameters<EventSink.EventSinkV2["Service"]["commitCommand"]>[0][] = [];
786 const recovery = yield* ProviderRuntimeRecovery.make.pipe(
787 Effect.provide(
788 Layer.mergeAll(
789 ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true }),
790 Layer.mock(ProjectionStore.ProjectionStoreV2)({
791 getRecoveryThreadIds: () => Effect.succeed([threadId]),
792 getRuntimeRecoveryProjection: () => Effect.sync(() => projection),
793 }),
794 Layer.mock(EventSink.EventSinkV2)({
795 writeWithEffects: (input) =>
796 Effect.sync(() => {
797 assert.lengthOf(input.events, 0);
798 assert.equal(input.effects[0]?.request.type, "provider-runtime.continue");
799 projection = {
800 ...projection,
801 runs: [{ ...projection.runs[0]!, status: "completed" }],
802 };
803 return [];
804 }),
805 commitCommand: (input) =>
806 Effect.sync(() => {
807 commits.push(input);
808 return { committed: true, cancelledEffectCount: 0 } as never;
809 }),
810 }),
811 IdAllocator.layer,
812 Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({
813 runRecoveryOnce: Effect.succeed(false),
814 }),
815 Layer.mock(EffectOutbox.EffectOutboxV2)({
816 reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }),
817 }),
818 ),
819 ),
820 );
821 yield* recovery.prepareForShutdown;
822 yield* recovery.reconcile("shutdown");
823 assert.isFalse(
824 commits.some((commit) => commit.events.some((event) => event.type === "run.updated")),
825 );
826 let dispatched = false;
827 yield* continueRestartedRun({ threadId, sourceRunId: runId }).pipe(
828 Effect.provide(
829 Layer.mergeAll(
830 ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true }),
831 Layer.mock(ThreadManagementService.ThreadManagementService)({
832 getThreadRecords: () => Effect.succeed(projection),
833 recoverDelegatedTask: () => Effect.void,
834 dispatch: () =>
835 Effect.sync(() => {
836 dispatched = true;
837 return {} as never;
838 }),
839 }),
840 ),
841 ),
842 );
843 assert.isFalse(dispatched);
844 }),
845);
846
847const continuationDispatches = (
848 projection: OrchestrationV2ThreadProjection,
849 sourceRunId: RunId = runId,
850) =>
851 Effect.gen(function* () {
852 const dispatches: Array<Extract<OrchestrationV2Command, { type: "message.dispatch" }>> = [];
853 yield* continueRestartedRun({ threadId, sourceRunId }).pipe(
854 Effect.provide(
855 Layer.merge(
856 Layer.mock(ThreadManagementService.ThreadManagementService)({
857 getThreadRecords: () => Effect.succeed(projection),
858 recoverDelegatedTask: () => Effect.void,
859 dispatch: (command) => {
860 if (command.type === "message.dispatch") dispatches.push(command);
861 return Effect.succeed({} as never);
862 },
863 }),
864 ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true }),
865 ),
866 ),
867 );
868 return dispatches;
869 });
870
871const continuationTexts = (projection: OrchestrationV2ThreadProjection) =>
872 continuationDispatches(projection).pipe(
873 Effect.map((dispatches) => dispatches.map((command) => command.text)),
874 );
875
876const cutMidTurn = (extra: Record<string, unknown> = {}) => {
877 const base = makeProjection();
878 return {
879 ...base,
880 runs: [
881 {
882 ...base.runs[0]!,
883 status: "cancelled",
884 completedAt: now,
885 userMessageId: MessageId.make("message:user"),
886 ...extra,
887 },
888 ],
889 providerTurns: [{ ...base.providerTurns[0]!, status: "cancelled" }],
890 } as unknown as OrchestrationV2ThreadProjection;
891};
892
893const queuedFollowUp = {
894 id: RunId.make("run:queued"),
895 ordinal: 2,
896 providerInstanceId: instanceId,
897 providerThreadId,
898 status: "queued",
899 queueHeld: true,
900};
901
902it.effect("continues a cut run past queued follow-ups, which stay held", () =>
903 Effect.gen(function* () {
904 const live = makeProjection();
905 assert.equal(
906 restartContinuationRun(
907 {
908 ...live,
909 runs: [...live.runs, queuedFollowUp],
910 } as unknown as OrchestrationV2ThreadProjection,
911 now,
912 )?.id,
913 runId,
914 );
915 const cut = cutMidTurn();
916 const texts = yield* continuationTexts({
917 ...cut,
918 runs: [...cut.runs, queuedFollowUp],
919 } as unknown as OrchestrationV2ThreadProjection);
920 assert.deepEqual(texts, [
921 "Continue where you left off. Your last turn was cut short, so a command or other action it started may have finished or may still be running. Before you repeat one that changes state, confirm it did not take effect and is not still running; if you cannot tell, say so instead of repeating it.",
922 ]);
923 }),
924);
925
926it.effect("continues a resumed queued run that ran after an earlier continuation", () =>
927 Effect.gen(function* () {
928 const finishedContinuation = {
929 id: RunId.make("run:earlier-continuation"),
930 ordinal: 3,
931 providerInstanceId: instanceId,
932 providerThreadId,
933 status: "completed",
934 completedAt: DateTime.makeUnsafe("2026-10-03T10:00:00.000Z"),
935 };
936 const live = makeProjection();
937 assert.equal(
938 restartContinuationRun(
939 {
940 ...live,
941 runs: [...live.runs, finishedContinuation],
942 } as unknown as OrchestrationV2ThreadProjection,
943 now,
944 )?.id,
945 runId,
946 );
947 const cut = cutMidTurn({ completedAt: DateTime.makeUnsafe("2026-10-03T10:05:00.000Z") });
948 const texts = yield* continuationTexts({
949 ...cut,
950 runs: [...cut.runs, finishedContinuation],
951 } as unknown as OrchestrationV2ThreadProjection);
952 assert.deepEqual(texts, [
953 "Continue where you left off. Your last turn was cut short, so a command or other action it started may have finished or may still be running. Before you repeat one that changes state, confirm it did not take effect and is not still running; if you cannot tell, say so instead of repeating it.",
954 ]);
955 }),
956);
957
958it("backs off twice from a continuation the provider never took, then lets it fail", () => {
959 const base = makeProjection();
960 const failedStart = (id: string, of: RunId) => ({
961 ...base.runs[0]!,
962 id: RunId.make(id),
963 activeAttemptId: RunAttemptId.make(`attempt:${id}`),
964 status: "failed" as const,
965 restartContinuationOfRunId: of,
966 });
967 const first = failedStart("run:retry-1", runId);
968 const second = failedStart("run:retry-2", first.id);
969 const third = failedStart("run:retry-3", second.id);
970 const projection = (failureClass = "provider_error", providerTurns = base.providerTurns) =>
971 ({
972 runs: [{ ...base.runs[0]!, status: "cancelled" }, first, second, third],
973 providerTurns,
974 turnItems: [first, second, third].map((run) => ({
975 id: `error:${run.id}`,
976 type: "error",
977 status: "failed",
978 runId: run.id,
979 nodeId: run.rootNodeId,
980 ordinal: 1,
981 updatedAt: now,
982 failure: { class: failureClass, message: "Failed.", code: null, retryable: null },
983 })),
984 }) as unknown as Parameters<typeof continuationStartRetryDelayMs>[1];
985 assert.deepEqual(
986 [first, second, third].map((run) => continuationStartRetryDelayMs(run, projection())),
987 [60_000, 180_000, undefined],
988 );
989 for (const failureClass of ["usage_limit", "permission_error", "validation_error"])
990 assert.isUndefined(continuationStartRetryDelayMs(first, projection(failureClass)));
991 // Once the provider took the prompt, starting again would repeat the turn.
992 assert.isUndefined(
993 continuationStartRetryDelayMs(
994 first,
995 projection("provider_error", [
996 { ...base.providerTurns[0]!, runAttemptId: first.activeAttemptId, status: "failed" },
997 ] as never),
998 ),
999 );
1000});
1001
1002it.effect("continues again after a continuation fails before its prompt", () =>
1003 Effect.gen(function* () {
1004 const cut = cutMidTurn();
1005 const failedStart = {
1006 ...cut.runs[0]!,
1007 id: RunId.make("run:failed-start"),
1008 ordinal: 2,
1009 activeAttemptId: RunAttemptId.make("attempt:failed-start"),
1010 status: "failed",
1011 restartContinuationOfRunId: runId,
1012 };
1013 const dispatches = yield* continuationDispatches(
1014 { ...cut, runs: [...cut.runs, failedStart] } as unknown as OrchestrationV2ThreadProjection,
1015 failedStart.id,
1016 );
1017 assert.deepEqual(
1018 dispatches.map((command) => command.restartContinuationOfRunId),
1019 [failedStart.id],
1020 );
1021 }),
1022);
1023
1024it.effect.each([
1025 ["running", null, 0],
1026 ["completed", "Done.", 1],
1027] as const)(
1028 "a parent's Stop on the task's run blocks the child only while the task is %s",
1029 ([status, result, expected]) =>
1030 Effect.gen(function* () {
1031 const parentThreadId = ThreadId.make("thread:parent");
1032 const parentRunId = RunId.make("run:parent");
1033 const taskId = NodeId.make("node:task");
1034 const cut = cutMidTurn();
1035 const child = {
1036 ...cut,
1037 thread: {
1038 ...cut.thread,
1039 lineage: { parentThreadId, relationshipToParent: "subagent" },
1040 forkedFrom: { type: "node", nodeId: taskId },
1041 },
1042 } as unknown as OrchestrationV2ThreadProjection;
1043 const parent = {
1044 thread: { id: parentThreadId },
1045 subagents: [{ id: taskId, runId: parentRunId, origin: "app_owned", status, result }],
1046 turnItems: [{ id: "turn-item:stop", runId: parentRunId, type: "run_interrupt_request" }],
1047 };
1048 let dispatched = 0;
1049 yield* continueRestartedRun({ threadId, sourceRunId: runId }).pipe(
1050 Effect.provide(
1051 Layer.merge(
1052 Layer.mock(ThreadManagementService.ThreadManagementService)({
1053 getThreadRecords: (id) =>
1054 Effect.succeed((id === parentThreadId ? parent : child) as never),
1055 recoverDelegatedTask: () => Effect.void,
1056 dispatch: () =>
1057 Effect.sync(() => {
1058 dispatched += 1;
1059 return {} as never;
1060 }),
1061 }),
1062 ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true }),
1063 ),
1064 ),
1065 );
1066 assert.equal(dispatched, expected);
1067 }),
1068);
1069
1070it.effect("does not continue a run the user asked to stop before the restart", () =>
1071 Effect.gen(function* () {
1072 const texts = yield* continuationTexts({
1073 ...cutMidTurn(),
1074 turnItems: [{ id: "turn-item:interrupt", runId, type: "run_interrupt_request" }],
1075 } as unknown as OrchestrationV2ThreadProjection);
1076 assert.deepEqual(texts, []);
1077 }),
1078);
1079
1080it.effect("does not continue a cut /compact or /logout turn", () =>
1081 Effect.gen(function* () {
1082 for (const text of ["/compact", " /LOGOUT "]) {
1083 const texts = yield* continuationTexts({
1084 ...cutMidTurn(),
1085 messages: [{ id: MessageId.make("message:user"), text, attachments: [] }],
1086 } as unknown as OrchestrationV2ThreadProjection);
1087 assert.deepEqual(texts, [], text);
1088 }
1089 const texts = yield* continuationTexts({
1090 ...cutMidTurn(),
1091 messages: [{ id: MessageId.make("message:user"), text: "/compact later", attachments: [] }],
1092 } as unknown as OrchestrationV2ThreadProjection);
1093 assert.lengthOf(texts, 1);
1094 }),
1095);
1096
1097it.effect(
1098 "carries the note forward when its continuation was cut before reaching the provider",
1099 () =>
1100 Effect.gen(function* () {
1101 const base = makeProjection();
1102 const original = {
1103 ...base.runs[0]!,
1104 id: RunId.make("run:original"),
1105 status: "completed",
1106 activeAttemptId: RunAttemptId.make("attempt:original"),
1107 restartCancelledBackgroundWork: [{ kind: "shell", label: "sleep 25 && echo DONE" }],
1108 };
1109 const projection = {
1110 ...base,
1111 runs: [
1112 original,
1113 {
1114 ...base.runs[0]!,
1115 ordinal: 2,
1116 status: "cancelled",
1117 restartContinuationOfRunId: original.id,
1118 },
1119 ],
1120 providerTurns: [
1121 {
1122 ...base.providerTurns[0]!,
1123 runAttemptId: original.activeAttemptId,
1124 status: "completed",
1125 },
1126 ],
1127 } as unknown as OrchestrationV2ThreadProjection;
1128 const texts = yield* continuationTexts(projection);
1129 assert.lengthOf(texts, 1);
1130 assert.include(texts[0]!, "sleep 25 && echo DONE");
1131 assert.notInclude(texts[0]!, "Continue where");
1132 }),
1133);
1134
1135it.effect("prepares later threads' continuations when one thread fails", () =>
1136 Effect.gen(function* () {
1137 const brokenThreadId = ThreadId.make("thread:broken");
1138 const writes: Parameters<EventSink.EventSinkV2["Service"]["writeWithEffects"]>[0][] = [];
1139 const recovery = yield* ProviderRuntimeRecovery.make.pipe(
1140 Effect.provide(
1141 Layer.mergeAll(
1142 ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true }),
1143 Layer.mock(ProjectionStore.ProjectionStoreV2)({
1144 getRecoveryThreadIds: () => Effect.succeed([brokenThreadId, threadId]),
1145 getRuntimeRecoveryProjection: (id) =>
1146 id === brokenThreadId
1147 ? Effect.fail(
1148 new ProjectionStore.ProjectionStoreThreadNotFoundError({ threadId: id }),
1149 )
1150 : Effect.succeed(makeProjection()),
1151 }),
1152 Layer.mock(EventSink.EventSinkV2)({
1153 writeWithEffects: (input) =>
1154 Effect.sync(() => {
1155 writes.push(input);
1156 return [];
1157 }),
1158 }),
1159 IdAllocator.layer,
1160 Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({}),
1161 Layer.mock(EffectOutbox.EffectOutboxV2)({}),
1162 ),
1163 ),
1164 );
1165 yield* recovery.prepareForShutdown;
1166 assert.deepEqual(
1167 writes.map((write) => write.effects[0]?.request),
1168 [{ type: "provider-runtime.continue", sourceRunId: runId }],
1169 );
1170 }),
1171);
1172
1173it.effect.each([
1174 ["user_input", "pending", "live", false],
1175 ["user_input", "expired", "not_resumable", false],
1176 ["user_input", "cancelled", "not_resumable", false],
1177 ["command", "pending", "live", false],
1178 ["command", "expired", "not_resumable", false],
1179 ["user_input", "resolved", "live", true],
1180 ["command", "cancelled", "live", true],
1181 ["user_input", "pending", "message", true],
1182 ["dynamic_tool_call", "pending", "live", true],
1183 ["auth_refresh", "pending", "live", true],
1184] as const)(
1185 "a %s request that is %s with %s capability allows continuation: %s",
1186 ([kind, status, capability, shouldContinue]) =>
1187 Effect.gen(function* () {
1188 const base = makeProjection();
1189 const runtimeRequests = [
1190 {
1191 id: RuntimeRequestId.make("request:restart"),
1192 nodeId: NodeId.make("node:restart-question"),
1193 providerTurnId: base.providerTurns[0]!.id,
1194 kind,
1195 status,
1196 responseCapability:
1197 capability === "live"
1198 ? { type: capability, providerSessionId: sessionId }
1199 : capability === "not_resumable"
1200 ? { type: capability, reason: "The server restarted." }
1201 : { type: capability },
1202 },
1203 ];
1204 const live = { ...base, runtimeRequests } as unknown as OrchestrationV2ThreadProjection;
1205 assert.equal(restartContinuationRun(live, now) !== undefined, shouldContinue);
1206 const cut = {
1207 ...cutMidTurn(),
1208 runtimeRequests,
1209 } as unknown as OrchestrationV2ThreadProjection;
1210 assert.lengthOf(yield* continuationTexts(cut), shouldContinue ? 1 : 0);
1211 }),
1212);
1213
1214it.effect("does not wake a snoozed or settled thread, but continues once a snooze elapsed", () =>
1215 Effect.gen(function* () {
1216 const live = makeProjection();
1217 for (const held of [
1218 { snoozedUntil: DateTime.add(now, { days: 1 }) },
1219 { settledOverride: "settled" as const },
1220 ]) {
1221 const thread = { ...live.thread, ...held };
1222 assert.isUndefined(restartContinuationRun({ ...live, thread }, now));
1223 assert.deepEqual(yield* continuationTexts({ ...cutMidTurn(), thread }), []);
1224 }
1225 const thread = { ...live.thread, snoozedUntil: DateTime.subtract(now, { minutes: 1 }) };
1226 assert.equal(restartContinuationRun({ ...live, thread }, now)?.id, runId);
1227 }),
1228);
1229
1230it.effect("does not continue a run waiting on a secret the restart closed", () =>
1231 Effect.gen(function* () {
1232 const completedAt = DateTime.makeUnsafe("2026-10-03T10:05:00.000Z");
1233 const secret = (secretStatus: string, at: DateTime.Utc | null) =>
1234 ({ type: "secret_request", runId, secretStatus, completedAt: at }) as never;
1235 assert.isUndefined(
1236 restartContinuationRun({ ...makeProjection(), turnItems: [secret("pending", null)] }, now),
1237 );
1238 const cut = cutMidTurn({ completedAt });
1239 assert.deepEqual(
1240 yield* continuationTexts({ ...cut, turnItems: [secret("cancelled", completedAt)] }),
1241 [],
1242 );
1243 assert.lengthOf(
1244 yield* continuationTexts({
1245 ...cut,
1246 turnItems: [secret("cancelled", DateTime.subtract(completedAt, { seconds: 1 }))],
1247 }),
1248 1,
1249 );
1250 }),
1251);
1252
1253it.effect.each(["completed", "failed", "interrupted"] as const)(
1254 "does not continue when the provider turn %s before the app run settled",
1255 (status) =>
1256 Effect.gen(function* () {
1257 const cut = cutMidTurn();
1258 assert.deepEqual(
1259 yield* continuationTexts({
1260 ...cut,
1261 providerTurns: [{ ...cut.providerTurns[0]!, status }],
1262 }),
1263 [],
1264 );
1265 }),
1266);
1267
1268it.effect("prepares independent continuations together and waits for durable writes", () =>
1269 Effect.gen(function* () {
1270 const bothReading = yield* Deferred.make<void>();
1271 const bothWriting = yield* Deferred.make<void>();
1272 const committed = yield* Deferred.make<void>();
1273 let reads = 0;
1274 let writes = 0;
1275 const recovery = yield* ProviderRuntimeRecovery.make.pipe(
1276 Effect.provide(
1277 Layer.mergeAll(
1278 ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true }),
1279 Layer.mock(ProjectionStore.ProjectionStoreV2)({
1280 getRecoveryThreadIds: () => Effect.succeed([threadId, ThreadId.make("thread:second")]),
1281 getRuntimeRecoveryProjection: () =>
1282 Effect.gen(function* () {
1283 if (++reads === 2) yield* Deferred.succeed(bothReading, undefined);
1284 yield* Deferred.await(bothReading);
1285 return makeProjection();
1286 }),
1287 }),
1288 Layer.mock(EventSink.EventSinkV2)({
1289 writeWithEffects: () =>
1290 Effect.gen(function* () {
1291 if (++writes === 2) yield* Deferred.succeed(bothWriting, undefined);
1292 yield* Deferred.await(committed);
1293 return [];
1294 }),
1295 }),
1296 IdAllocator.layer,
1297 Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({}),
1298 Layer.mock(EffectOutbox.EffectOutboxV2)({}),
1299 ),
1300 ),
1301 );
1302 const fiber = yield* recovery.prepareForShutdown.pipe(Effect.forkChild);
1303 yield* Deferred.await(bothWriting);
1304 assert.equal(writes, 2);
1305 assert.isUndefined(fiber.pollUnsafe());
1306 yield* Deferred.succeed(committed, undefined);
1307 yield* Fiber.join(fiber);
1308 }),
1309);
1310
1311it.effect("skips restart discovery when no project enables continuation", () =>
1312 Effect.gen(function* () {
1313 const recovery = yield* ProviderRuntimeRecovery.make.pipe(
1314 Effect.provide(
1315 Layer.mergeAll(
1316 ServerSettings.layerTest({ continueThreadsAfterServerUpdate: false }),
1317 Layer.mock(ProjectionStore.ProjectionStoreV2)({}),
1318 Layer.mock(EventSink.EventSinkV2)({}),
1319 IdAllocator.layer,
1320 Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({}),
1321 Layer.mock(EffectOutbox.EffectOutboxV2)({}),
1322 ),
1323 ),
1324 );
1325 yield* recovery.prepareForShutdown;
1326 yield* recovery.prepareForShutdown;
1327 }),
1328);