1import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
2import { assert, it } from "@effect/vitest";
3import {
4 CheckpointId,
5 CheckpointRef,
6 CheckpointScopeId,
7 EventId,
8 MessageId,
9 NodeId,
10 type OrchestrationV2DomainEvent,
11 type OrchestrationV2Run,
12 ProjectId,
13 ProviderDriverKind,
14 ProviderInstanceId,
15 ProviderThreadId,
16 RunId,
17 ThreadId,
18} from "@t3tools/contracts";
19import * as DateTime from "effect/DateTime";
20import * as Effect from "effect/Effect";
21import * as Layer from "effect/Layer";
22import * as Ref from "effect/Ref";
23import * as SqlClient from "effect/sql/SqlClient";
24
25import * as SqlitePersistence from "../persistence/Sqlite.ts";
26import * as CheckpointStore from "../checkpointing/CheckpointStore.ts";
27import { VcsProcessTimeoutError } from "@t3tools/contracts";
28import * as CheckpointService from "./CheckpointService.ts";
29import * as CheckpointCaptureService from "./CheckpointCaptureService.ts";
30import * as EventSink from "./EventSink.ts";
31import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
32import * as ProjectionStore from "./ProjectionStore.ts";
33
34const layerProjectionStoreTest = Layer.mergeAll(
35 ProjectionStore.layer.pipe(Layer.provideMerge(SqlitePersistence.layerMemory)),
36 SqlitePersistence.layerMemory,
37);
38
39const threadId = ThreadId.make("thread:checkpoint-capture-delegated");
40const projectId = ProjectId.make("project:checkpoint-capture-delegated");
41const runId = RunId.make("run:checkpoint-capture-delegated");
42const scopeId = CheckpointScopeId.make("scope:checkpoint-capture-delegated");
43const rootNodeId = NodeId.make("node:checkpoint-capture-delegated-root");
44const taskId = NodeId.make("node:checkpoint-capture-task");
45const deliveryMessageId = MessageId.make("message:checkpoint-capture-delivery");
46const providerThreadId = ProviderThreadId.make("provider-thread:checkpoint-capture-delegated");
47const providerInstanceId = ProviderInstanceId.make("codex");
48const driver = ProviderDriverKind.make("codex");
49const modelSelection = {
50 instanceId: providerInstanceId,
51 model: "gpt-5.4",
52} as const;
53
54it.layer(layerProjectionStoreTest)("CheckpointCaptureServiceV2", (it) => {
55 it.effect.each([false, true])(
56 "captures without decoding history or losing newer delegated completion, ref lookup fails=%s",
57 (refLookupFails) =>
58 Effect.gen(function* () {
59 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
60 const now = yield* DateTime.now;
61 const later = DateTime.add(now, { seconds: 1 });
62
63 const staleDelegatedCompletion = {
64 disposition: "open" as const,
65 nextGeneration: 1,
66 delivery: null,
67 };
68 const newerCohort = {
69 disposition: "open" as const,
70 nextGeneration: 2,
71 delivery: {
72 generation: 1,
73 messageId: deliveryMessageId,
74 taskIds: [taskId],
75 },
76 };
77 const staleRun: OrchestrationV2Run = {
78 id: runId,
79 threadId,
80 ordinal: 1,
81 providerInstanceId,
82 modelSelection,
83 providerThreadId,
84 userMessageId: MessageId.make("message:checkpoint-capture-user"),
85 rootNodeId,
86 activeAttemptId: null,
87 status: "waiting",
88 requestedAt: now,
89 startedAt: now,
90 completedAt: null,
91 checkpointId: null,
92 contextHandoffId: null,
93 // Snapshot taken before a concurrent cohort advanced during capture work.
94 delegatedCompletion: staleDelegatedCompletion,
95 };
96 const rootNode = {
97 id: rootNodeId,
98 threadId,
99 runId,
100 parentNodeId: null,
101 rootNodeId,
102 kind: "root_turn" as const,
103 status: "waiting" as const,
104 countsForRun: true,
105 providerThreadId,
106 providerTurnId: null,
107 nativeItemRef: null,
108 runtimeRequestId: null,
109 checkpointScopeId: scopeId,
110 startedAt: now,
111 completedAt: null,
112 };
113 const scope = {
114 id: scopeId,
115 threadId,
116 runId,
117 nodeId: rootNodeId,
118 parentScopeId: null,
119 providerThreadId,
120 kind: "root_run" as const,
121 ordinalWithinParent: 0,
122 advancesAppRunCount: true,
123 cwd: "/repo",
124 createdAt: now,
125 };
126 const providerThread = {
127 id: providerThreadId,
128 driver,
129 providerInstanceId,
130 providerSessionId: null,
131 appThreadId: threadId,
132 ownerNodeId: rootNodeId,
133 nativeThreadRef: null,
134 nativeConversationHeadRef: null,
135 status: "idle" as const,
136 firstRunOrdinal: 1,
137 lastRunOrdinal: 1,
138 handoffIds: [],
139 forkedFrom: null,
140 createdAt: now,
141 updatedAt: now,
142 };
143 const readyBaseline = {
144 id: CheckpointId.make("checkpoint:baseline-0"),
145 threadId,
146 scopeId,
147 runId: null,
148 nodeId: rootNodeId,
149 parentCheckpointId: null,
150 ordinalWithinScope: 0,
151 appRunOrdinal: null,
152 ref: CheckpointRef.make("checkpoint-ref:baseline-0"),
153 status: "ready" as const,
154 files: [],
155 capturedAt: now,
156 };
157 const captured = {
158 id: CheckpointId.make("checkpoint:captured-1"),
159 threadId,
160 scopeId,
161 runId,
162 nodeId: rootNodeId,
163 parentCheckpointId: readyBaseline.id,
164 ordinalWithinScope: 1,
165 appRunOrdinal: 1,
166 ref: CheckpointRef.make("checkpoint-ref:captured-1"),
167 status: "ready" as const,
168 files: [],
169 capturedAt: now,
170 };
171
172 yield* projectionStore.apply({
173 id: EventId.make("event:checkpoint-capture:thread"),
174 type: "thread.created",
175 threadId,
176 occurredAt: now,
177 payload: {
178 createdBy: "user",
179 creationSource: "web",
180 id: threadId,
181 projectId,
182 title: "Checkpoint capture delegated completion",
183 providerInstanceId,
184 modelSelection,
185 runtimeMode: "full-access",
186 interactionMode: "default",
187 branch: null,
188 worktreePath: null,
189 activeProviderThreadId: null,
190 lineage: {
191 parentThreadId: null,
192 relationshipToParent: null,
193 rootThreadId: threadId,
194 },
195 forkedFrom: null,
196 createdAt: now,
197 updatedAt: now,
198 archivedAt: null,
199 settledOverride: null,
200 settledAt: null,
201 lastVisitedAt: null,
202 deletedAt: null,
203 },
204 });
205 yield* projectionStore.apply({
206 id: EventId.make("event:checkpoint-capture:run-stale"),
207 type: "run.updated",
208 threadId,
209 runId,
210 nodeId: rootNodeId,
211 providerInstanceId,
212 occurredAt: now,
213 payload: staleRun,
214 });
215 yield* projectionStore.apply({
216 id: EventId.make("event:checkpoint-capture:node"),
217 type: "node.updated",
218 threadId,
219 runId,
220 nodeId: rootNodeId,
221 providerInstanceId,
222 occurredAt: now,
223 payload: rootNode,
224 });
225 yield* projectionStore.apply({
226 id: EventId.make("event:checkpoint-capture:provider-thread"),
227 type: "provider-thread.updated",
228 threadId,
229 nodeId: rootNodeId,
230 driver,
231 providerInstanceId,
232 occurredAt: now,
233 payload: providerThread,
234 });
235 yield* projectionStore.apply({
236 id: EventId.make("event:checkpoint-capture:scope"),
237 type: "checkpoint-scope.created",
238 threadId,
239 runId,
240 nodeId: rootNodeId,
241 occurredAt: now,
242 payload: scope,
243 });
244 yield* projectionStore.apply({
245 id: EventId.make("event:checkpoint-capture:baseline"),
246 type: "checkpoint.captured",
247 threadId,
248 nodeId: rootNodeId,
249 driver,
250 providerInstanceId,
251 occurredAt: now,
252 payload: readyBaseline,
253 });
254 const sql = yield* SqlClient.SqlClient;
255 yield* sql`INSERT OR REPLACE INTO orchestration_v2_projection_turn_items
256 (turn_item_id, thread_id, run_id, node_id, provider_thread_id, provider_turn_id,
257 parent_item_id, ordinal, type, status, updated_at, payload_json)
258 VALUES ('obsolete-history', ${threadId}, ${runId}, ${rootNodeId}, NULL, NULL,
259 NULL, 1, 'assistant_message', 'completed', ${DateTime.formatIso(now)}, '{"obsolete":true}')`;
260 yield* sql`UPDATE orchestration_v2_projection_checkpoints
261 SET payload_json = json_set(payload_json, '$.files', 'obsolete file summary')
262 WHERE checkpoint_id = ${readyBaseline.id}`;
263 assert.equal(
264 (yield* Effect.exit(projectionStore.getThreadProjection(threadId)))._tag,
265 "Failure",
266 );
267
268 const committed = yield* Ref.make<ReadonlyArray<OrchestrationV2DomainEvent>>([]);
269 const layerCapture = CheckpointCaptureService.layer.pipe(
270 Layer.provide(
271 Layer.mergeAll(
272 IdAllocator.layer,
273 refLookupFails
274 ? CheckpointService.layer.pipe(
275 Layer.provide(
276 Layer.mergeAll(
277 IdAllocator.layer,
278 NodeCrypto.layer,
279 Layer.mock(CheckpointStore.CheckpointStore)({
280 listAuthoredPaths: () => Effect.succeed(null),
281 isGitRepository: () => Effect.succeed(true),
282 captureCheckpoint: () => Effect.void,
283 hasCheckpointRef: () =>
284 Effect.fail(
285 new VcsProcessTimeoutError({
286 operation: "test.hasCheckpointRef",
287 command: "git",
288 cwd: "/repo",
289 timeoutMs: 30000,
290 }),
291 ),
292 }),
293 ),
294 ),
295 )
296 : Layer.mock(CheckpointService.CheckpointServiceV2)({
297 materializeBaselineCheckpoint: () =>
298 Effect.die(
299 "baseline materialization must be skipped when ordinal 0 is ready",
300 ),
301 capture: () => Effect.succeed(captured),
302 }),
303 Layer.mock(EventSink.EventSinkV2)({
304 commitCommand: (input) =>
305 Ref.set(committed, input.events).pipe(
306 Effect.as({
307 commandId: input.commandId,
308 committed: true,
309 sequence: 1,
310 events: input.events,
311 effects: [],
312 } as never),
313 ),
314 }),
315 ),
316 ),
317 );
318
319 yield* Effect.gen(function* () {
320 const service = yield* CheckpointCaptureService.CheckpointCaptureServiceV2;
321 const incomplete = yield* service
322 .execute({ threadId, runId, scopeId: CheckpointScopeId.make("missing-scope") })
323 .pipe(Effect.flip);
324 assert.instanceOf(incomplete, CheckpointCaptureService.CheckpointCaptureExecutionError);
325 // Capture reads the waiting run while the projection still holds the stale cohort.
326 yield* service.execute({ threadId, runId, scopeId });
327
328 const events = yield* Ref.get(committed);
329 const runUpdated = events.find((event) => event.type === "run.updated");
330 assert.isDefined(runUpdated);
331 if (runUpdated?.type !== "run.updated") {
332 return;
333 }
334 assert.equal(runUpdated.payload.status, "completed");
335 const capturedEvent = events.find((event) => event.type === "checkpoint.captured");
336 assert.equal(
337 runUpdated.payload.checkpointId,
338 refLookupFails ? capturedEvent?.payload.id : captured.id,
339 );
340 if (refLookupFails && capturedEvent?.type === "checkpoint.captured") {
341 assert.equal(capturedEvent.payload.status, "ready");
342 assert.deepEqual(capturedEvent.payload.files, []);
343 }
344 assert.isUndefined(
345 runUpdated.payload.delegatedCompletion,
346 "checkpoint capture must omit delegatedCompletion so ProjectionStore can keep a newer cohort",
347 );
348
349 // Race: a newer cohort lands on the projection after capture read the stale
350 // waiting run and before the capture command's run.updated is applied.
351 yield* projectionStore.apply({
352 id: EventId.make("event:checkpoint-capture:newer-cohort"),
353 type: "run.updated",
354 threadId,
355 runId,
356 nodeId: rootNodeId,
357 providerInstanceId,
358 occurredAt: later,
359 payload: {
360 ...staleRun,
361 delegatedCompletion: newerCohort,
362 },
363 });
364
365 // Apply the real capture-emitted run.updated through ProjectionStore.
366 yield* projectionStore.apply(runUpdated);
367
368 const projectedRun = (yield* projectionStore.getCheckpointCaptureContext(threadId, {
369 runId,
370 scopeId,
371 })).run;
372 assert.isDefined(projectedRun);
373 assert.equal(projectedRun?.status, "completed");
374 assert.equal(projectedRun?.checkpointId, runUpdated.payload.checkpointId);
375 assert.deepEqual(projectedRun?.delegatedCompletion, newerCohort);
376 assert.equal(projectedRun?.delegatedCompletion?.delivery?.messageId, deliveryMessageId);
377 assert.deepEqual(projectedRun?.delegatedCompletion?.delivery?.taskIds, [taskId]);
378 // The persisted completion is the at-least-once capture receipt.
379 yield* Ref.set(committed, []);
380 yield* service.execute({ threadId, runId, scopeId });
381 assert.deepEqual(yield* Ref.get(committed), []);
382 }).pipe(Effect.provide(layerCapture));
383 }),
384 );
385
386 // Capture and rollback share the thread's effect lane, so a rollback can only
387 // commit before a capture runs: while it waits out a retry, or when a restart
388 // requeues it behind a pending rollback. It must not revive the run.
389 it.effect("does not capture a stopped run that a rollback already discarded", () =>
390 Effect.gen(function* () {
391 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
392 const now = yield* DateTime.now;
393 const stoppedThreadId = ThreadId.make("thread:checkpoint-capture-rolled-back");
394 const stoppedRunId = RunId.make("run:checkpoint-capture-rolled-back");
395 yield* projectionStore.apply({
396 id: EventId.make("event:checkpoint-capture-rolled-back:thread"),
397 type: "thread.created",
398 threadId: stoppedThreadId,
399 occurredAt: now,
400 payload: {
401 createdBy: "user",
402 creationSource: "web",
403 id: stoppedThreadId,
404 projectId,
405 title: "Checkpoint capture after rollback",
406 providerInstanceId,
407 modelSelection,
408 runtimeMode: "full-access",
409 interactionMode: "default",
410 branch: null,
411 worktreePath: null,
412 activeProviderThreadId: null,
413 lineage: {
414 parentThreadId: null,
415 relationshipToParent: null,
416 rootThreadId: stoppedThreadId,
417 },
418 forkedFrom: null,
419 createdAt: now,
420 updatedAt: now,
421 archivedAt: null,
422 settledOverride: null,
423 settledAt: null,
424 lastVisitedAt: null,
425 deletedAt: null,
426 },
427 });
428 const stoppedRun: OrchestrationV2Run = {
429 id: stoppedRunId,
430 threadId: stoppedThreadId,
431 ordinal: 2,
432 providerInstanceId,
433 modelSelection,
434 providerThreadId,
435 userMessageId: MessageId.make("message:checkpoint-capture-rolled-back"),
436 rootNodeId,
437 activeAttemptId: null,
438 status: "interrupted",
439 requestedAt: now,
440 startedAt: now,
441 completedAt: now,
442 checkpointId: null,
443 contextHandoffId: null,
444 };
445 // The rollback's write, which landed before the capture ran.
446 yield* projectionStore.apply({
447 id: EventId.make("event:checkpoint-capture-rolled-back:run"),
448 type: "run.updated",
449 threadId: stoppedThreadId,
450 runId: stoppedRunId,
451 providerInstanceId,
452 occurredAt: now,
453 payload: { ...stoppedRun, status: "rolled_back" },
454 });
455
456 const committed = yield* Ref.make<ReadonlyArray<OrchestrationV2DomainEvent>>([]);
457 const layerCapture = CheckpointCaptureService.layer.pipe(
458 Layer.provide(
459 Layer.mergeAll(
460 IdAllocator.layer,
461 Layer.mock(CheckpointService.CheckpointServiceV2)({
462 materializeBaselineCheckpoint: () => Effect.die("a discarded run has no baseline"),
463 capture: () => Effect.die("a discarded run must not be captured"),
464 }),
465 Layer.mock(EventSink.EventSinkV2)({
466 commitCommand: (input) =>
467 Ref.set(committed, input.events).pipe(Effect.as({ committed: true } as never)),
468 }),
469 ),
470 ),
471 );
472
473 // Settles without retrying, and commits nothing that would revive the run.
474 yield* CheckpointCaptureService.CheckpointCaptureServiceV2.pipe(
475 Effect.flatMap((service) =>
476 service.execute({ threadId: stoppedThreadId, runId: stoppedRunId, scopeId }),
477 ),
478 Effect.provide(layerCapture),
479 );
480
481 assert.deepEqual(yield* Ref.get(committed), []);
482 const projected = yield* projectionStore.getCheckpointCaptureContext(stoppedThreadId, {
483 runId: stoppedRunId,
484 scopeId,
485 });
486 assert.equal(projected.run?.status, "rolled_back");
487 assert.isNull(projected.run?.checkpointId ?? null);
488 }),
489 );
490
491 // A cancelled run's checkpoint is the rollback point for the message after
492 // it, so capture records it without reporting the run as completed.
493 it.effect("records the checkpoint of a cancelled run and keeps it cancelled", () =>
494 Effect.gen(function* () {
495 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
496 const now = yield* DateTime.now;
497 const cancelledAt = DateTime.add(now, { seconds: 1 });
498 const cancelledThreadId = ThreadId.make("thread:checkpoint-capture-cancelled");
499 const cancelledRunId = RunId.make("run:checkpoint-capture-cancelled");
500 const cancelledScopeId = CheckpointScopeId.make("scope:checkpoint-capture-cancelled");
501 const cancelledRootNodeId = NodeId.make("node:checkpoint-capture-cancelled-root");
502 const cancelledProviderThreadId = ProviderThreadId.make(
503 "provider-thread:checkpoint-capture-cancelled",
504 );
505 yield* projectionStore.apply({
506 id: EventId.make("event:checkpoint-capture-cancelled:thread"),
507 type: "thread.created",
508 threadId: cancelledThreadId,
509 occurredAt: now,
510 payload: {
511 createdBy: "user",
512 creationSource: "web",
513 id: cancelledThreadId,
514 projectId,
515 title: "Checkpoint capture after cancel",
516 providerInstanceId,
517 modelSelection,
518 runtimeMode: "full-access",
519 interactionMode: "default",
520 branch: null,
521 worktreePath: null,
522 activeProviderThreadId: null,
523 lineage: {
524 parentThreadId: null,
525 relationshipToParent: null,
526 rootThreadId: cancelledThreadId,
527 },
528 forkedFrom: null,
529 createdAt: now,
530 updatedAt: now,
531 archivedAt: null,
532 settledOverride: null,
533 settledAt: null,
534 lastVisitedAt: null,
535 deletedAt: null,
536 },
537 });
538 const runningRun: OrchestrationV2Run = {
539 id: cancelledRunId,
540 threadId: cancelledThreadId,
541 ordinal: 1,
542 providerInstanceId,
543 modelSelection,
544 providerThreadId: cancelledProviderThreadId,
545 userMessageId: MessageId.make("message:checkpoint-capture-cancelled"),
546 rootNodeId: cancelledRootNodeId,
547 activeAttemptId: null,
548 status: "running",
549 requestedAt: now,
550 startedAt: now,
551 completedAt: null,
552 checkpointId: null,
553 contextHandoffId: null,
554 };
555 const runningRootNode = {
556 id: cancelledRootNodeId,
557 threadId: cancelledThreadId,
558 runId: cancelledRunId,
559 parentNodeId: null,
560 rootNodeId: cancelledRootNodeId,
561 kind: "root_turn" as const,
562 status: "running" as const,
563 countsForRun: true,
564 providerThreadId: cancelledProviderThreadId,
565 providerTurnId: null,
566 nativeItemRef: null,
567 runtimeRequestId: null,
568 checkpointScopeId: cancelledScopeId,
569 startedAt: now,
570 completedAt: null,
571 };
572 const scope = {
573 id: cancelledScopeId,
574 threadId: cancelledThreadId,
575 runId: cancelledRunId,
576 nodeId: cancelledRootNodeId,
577 parentScopeId: null,
578 providerThreadId: cancelledProviderThreadId,
579 kind: "root_run" as const,
580 ordinalWithinParent: 0,
581 advancesAppRunCount: true,
582 cwd: "/repo",
583 createdAt: now,
584 };
585 yield* projectionStore.apply({
586 id: EventId.make("event:checkpoint-capture-cancelled:provider-thread"),
587 type: "provider-thread.updated",
588 threadId: cancelledThreadId,
589 nodeId: cancelledRootNodeId,
590 driver,
591 providerInstanceId,
592 occurredAt: now,
593 payload: {
594 id: cancelledProviderThreadId,
595 driver,
596 providerInstanceId,
597 providerSessionId: null,
598 appThreadId: cancelledThreadId,
599 ownerNodeId: cancelledRootNodeId,
600 nativeThreadRef: null,
601 nativeConversationHeadRef: null,
602 status: "idle",
603 firstRunOrdinal: 1,
604 lastRunOrdinal: 1,
605 handoffIds: [],
606 forkedFrom: null,
607 createdAt: now,
608 updatedAt: now,
609 },
610 });
611 yield* projectionStore.apply({
612 id: EventId.make("event:checkpoint-capture-cancelled:scope"),
613 type: "checkpoint-scope.created",
614 threadId: cancelledThreadId,
615 runId: cancelledRunId,
616 nodeId: cancelledRootNodeId,
617 occurredAt: now,
618 payload: scope,
619 });
620 yield* projectionStore.apply({
621 id: EventId.make("event:checkpoint-capture-cancelled:baseline"),
622 type: "checkpoint.captured",
623 threadId: cancelledThreadId,
624 nodeId: cancelledRootNodeId,
625 driver,
626 providerInstanceId,
627 occurredAt: now,
628 payload: {
629 id: CheckpointId.make("checkpoint:cancelled-baseline-0"),
630 threadId: cancelledThreadId,
631 scopeId: cancelledScopeId,
632 runId: null,
633 nodeId: cancelledRootNodeId,
634 parentCheckpointId: null,
635 ordinalWithinScope: 0,
636 appRunOrdinal: null,
637 ref: CheckpointRef.make("checkpoint-ref:cancelled-baseline-0"),
638 status: "ready",
639 files: [],
640 capturedAt: now,
641 },
642 });
643 // The turn runs, then finalizes as cancelled the way RunExecutionService
644 // writes it, which also enqueues this capture.
645 for (const [status, completedAt] of [
646 ["running", null],
647 ["cancelled", cancelledAt],
648 ] as const) {
649 yield* projectionStore.apply({
650 id: EventId.make(`event:checkpoint-capture-cancelled:run-${status}`),
651 type: "run.updated",
652 threadId: cancelledThreadId,
653 runId: cancelledRunId,
654 nodeId: cancelledRootNodeId,
655 providerInstanceId,
656 occurredAt: completedAt ?? now,
657 payload: { ...runningRun, status, completedAt },
658 });
659 yield* projectionStore.apply({
660 id: EventId.make(`event:checkpoint-capture-cancelled:node-${status}`),
661 type: "node.updated",
662 threadId: cancelledThreadId,
663 runId: cancelledRunId,
664 nodeId: cancelledRootNodeId,
665 providerInstanceId,
666 occurredAt: completedAt ?? now,
667 payload: { ...runningRootNode, status, completedAt },
668 });
669 }
670
671 const captured = {
672 id: CheckpointId.make("checkpoint:cancelled-captured-1"),
673 threadId: cancelledThreadId,
674 scopeId: cancelledScopeId,
675 runId: cancelledRunId,
676 nodeId: cancelledRootNodeId,
677 parentCheckpointId: CheckpointId.make("checkpoint:cancelled-baseline-0"),
678 ordinalWithinScope: 1,
679 appRunOrdinal: 1,
680 ref: CheckpointRef.make("checkpoint-ref:cancelled-captured-1"),
681 status: "ready" as const,
682 files: [],
683 capturedAt: cancelledAt,
684 };
685 const commits = yield* Ref.make(0);
686 const layerCapture = CheckpointCaptureService.layer.pipe(
687 Layer.provide(
688 Layer.mergeAll(
689 IdAllocator.layer,
690 Layer.mock(CheckpointService.CheckpointServiceV2)({
691 materializeBaselineCheckpoint: () =>
692 Effect.die("baseline materialization must be skipped when ordinal 0 is ready"),
693 capture: () => Effect.succeed(captured),
694 }),
695 // Commit straight into the projection so the test reads what a
696 // client would see after the capture lands.
697 Layer.mock(EventSink.EventSinkV2)({
698 commitCommand: (input) =>
699 Effect.forEach(input.events, (event) => projectionStore.apply(event)).pipe(
700 Effect.andThen(Ref.update(commits, (count) => count + 1)),
701 Effect.as({ committed: true } as never),
702 Effect.orDie,
703 ),
704 }),
705 ),
706 ),
707 );
708
709 yield* Effect.gen(function* () {
710 const service = yield* CheckpointCaptureService.CheckpointCaptureServiceV2;
711 yield* service.execute({
712 threadId: cancelledThreadId,
713 runId: cancelledRunId,
714 scopeId: cancelledScopeId,
715 });
716 // A redelivered effect finds the recorded checkpoint and commits nothing.
717 yield* service.execute({
718 threadId: cancelledThreadId,
719 runId: cancelledRunId,
720 scopeId: cancelledScopeId,
721 });
722 }).pipe(Effect.provide(layerCapture));
723
724 assert.equal(yield* Ref.get(commits), 1);
725 const projected = yield* projectionStore.getCheckpointCaptureContext(cancelledThreadId, {
726 runId: cancelledRunId,
727 scopeId: cancelledScopeId,
728 });
729 assert.equal(projected.run?.status, "cancelled");
730 assert.equal(projected.run?.checkpointId, captured.id);
731 const completedAt = projected.run?.completedAt;
732 assert.equal(
733 completedAt ? DateTime.formatIso(completedAt) : completedAt,
734 DateTime.formatIso(cancelledAt),
735 );
736 assert.equal(projected.rootNode?.status, "cancelled");
737 assert.deepEqual([...projected.readyCheckpointOrdinals].toSorted(), [0, 1]);
738 }),
739 );
740});