1import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
2import { assert, it } from "@effect/vitest";
3import {
4 CheckpointScopeId,
5 NodeId,
6 ProviderThreadId,
7 RunId,
8 ThreadId,
9 type OrchestrationV2CheckpointScope,
10 VcsProcessExitError,
11 VcsProcessTimeoutError,
12 VcsUnsupportedOperationError,
13} from "@t3tools/contracts";
14import * as DateTime from "effect/DateTime";
15import * as Deferred from "effect/Deferred";
16import * as Effect from "effect/Effect";
17import * as Fiber from "effect/Fiber";
18import * as Layer from "effect/Layer";
19import * as Ref from "effect/Ref";
20import * as TestClock from "effect/testing/TestClock";
21
22import * as CheckpointStore from "../checkpointing/CheckpointStore.ts";
23import * as CheckpointService from "./CheckpointService.ts";
24import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
25
26it.effect(
27 "skips unsupported baseline and turn checkpoints without swallowing capture failures",
28 () =>
29 Effect.gen(function* () {
30 const scope: OrchestrationV2CheckpointScope = {
31 id: CheckpointScopeId.make("checkpoint-scope:capture-failure"),
32 threadId: ThreadId.make("thread:capture-failure"),
33 runId: RunId.make("run:capture-failure:1"),
34 nodeId: NodeId.make("node:capture-failure:1"),
35 parentScopeId: null,
36 providerThreadId: ProviderThreadId.make("provider-thread:capture-failure"),
37 kind: "root_run",
38 ordinalWithinParent: 0,
39 advancesAppRunCount: true,
40 cwd: "/repo/ignored",
41 createdAt: DateTime.makeUnsafe("2026-07-28T00:00:00.000Z"),
42 };
43 for (const captureError of [
44 new VcsProcessTimeoutError({
45 operation: "test.capture",
46 command: "git",
47 cwd: scope.cwd,
48 timeoutMs: 30000,
49 }),
50 new VcsUnsupportedOperationError({
51 operation: "test.capture",
52 kind: "git",
53 detail: "The workspace directory is ignored by its Git repository.",
54 }),
55 ]) {
56 const layerTest = CheckpointService.layer.pipe(
57 Layer.provide(
58 Layer.mergeAll(
59 IdAllocator.layer,
60 Layer.mock(CheckpointStore.CheckpointStore)({
61 isGitRepository: () => Effect.succeed(true),
62 hasCheckpointRef: () => Effect.succeed(false),
63 captureCheckpoint: () => Effect.fail(captureError),
64 }),
65 ),
66 ),
67 Layer.provideMerge(NodeCrypto.layer),
68 );
69 yield* Effect.gen(function* () {
70 const checkpoints = yield* CheckpointService.CheckpointServiceV2;
71 const baseline = yield* Effect.result(
72 checkpoints.captureBaseline({ scope, ordinalWithinScope: 0 }),
73 );
74 const checkpoint = yield* checkpoints.capture({
75 scope,
76 ordinalWithinScope: 1,
77 runId: scope.runId!,
78 nodeId: scope.nodeId!,
79 appRunOrdinal: 1,
80 capturedAt: scope.createdAt,
81 });
82
83 assert.deepEqual(
84 { baseline: baseline._tag, turn: checkpoint.status, files: checkpoint.files },
85 captureError._tag === "VcsUnsupportedOperationError"
86 ? { baseline: "Success", turn: "missing", files: [] }
87 : { baseline: "Failure", turn: "error", files: [] },
88 );
89 if (baseline._tag === "Failure") {
90 assert.strictEqual(baseline.failure.cause, captureError);
91 } else {
92 const materialized = yield* checkpoints.materializeBaselineCheckpoint({
93 scope,
94 ordinalWithinScope: 0,
95 nodeId: scope.nodeId,
96 capturedAt: scope.createdAt,
97 });
98 assert.strictEqual(materialized.status, "missing");
99 }
100 }).pipe(Effect.provide(layerTest));
101 }
102 }),
103);
104
105it.effect(
106 "bounds a slow baseline snapshot but waits for an in-flight capture of the same ref",
107 () =>
108 Effect.gen(function* () {
109 const scope: OrchestrationV2CheckpointScope = {
110 id: CheckpointScopeId.make("checkpoint-scope:slow-baseline"),
111 threadId: ThreadId.make("thread:slow-baseline"),
112 runId: RunId.make("run:slow-baseline:1"),
113 nodeId: NodeId.make("node:slow-baseline:1"),
114 parentScopeId: null,
115 providerThreadId: ProviderThreadId.make("provider-thread:slow-baseline"),
116 kind: "root_run",
117 ordinalWithinParent: 0,
118 advancesAppRunCount: true,
119 cwd: "/repo/slow",
120 createdAt: DateTime.makeUnsafe("2026-07-28T00:00:00.000Z"),
121 };
122 const refs = yield* Ref.make<ReadonlySet<string>>(new Set());
123 const snapshots = yield* Ref.make(0);
124 const firstSnapshotStarted = yield* Deferred.make<void>();
125 const layerTest = CheckpointService.layer.pipe(
126 Layer.provide(
127 Layer.mergeAll(
128 IdAllocator.layer,
129 Layer.mock(CheckpointStore.CheckpointStore)({
130 isGitRepository: () => Effect.succeed(true),
131 hasCheckpointRef: ({ checkpointRef }) =>
132 Ref.get(refs).pipe(Effect.map((current) => current.has(checkpointRef))),
133 captureCheckpoint: ({ checkpointRef }) =>
134 Effect.gen(function* () {
135 yield* Ref.update(snapshots, (count) => count + 1);
136 yield* Deferred.succeed(firstSnapshotStarted, undefined);
137 yield* Effect.sleep("10 seconds");
138 yield* Ref.update(refs, (current) => new Set([...current, checkpointRef]));
139 }),
140 }),
141 ),
142 ),
143 Layer.provideMerge(NodeCrypto.layer),
144 );
145
146 yield* Effect.gen(function* () {
147 const checkpoints = yield* CheckpointService.CheckpointServiceV2;
148 // A stopped run's post-turn capture writes the next run's baseline ref while holding the lock.
149 const postTurn = yield* checkpoints
150 .capture({
151 scope,
152 ordinalWithinScope: 1,
153 runId: scope.runId!,
154 nodeId: scope.nodeId!,
155 appRunOrdinal: 1,
156 capturedAt: scope.createdAt,
157 })
158 .pipe(Effect.forkChild);
159 yield* Deferred.await(firstSnapshotStarted);
160 const waitingBaseline = yield* checkpoints
161 .captureBaseline({ scope, ordinalWithinScope: 1 })
162 .pipe(Effect.result, Effect.forkChild);
163 yield* TestClock.adjust("10 seconds");
164 assert.strictEqual((yield* Fiber.join(postTurn)).status, "ready");
165 assert.strictEqual((yield* Fiber.join(waitingBaseline))._tag, "Success");
166
167 const slowBaseline = yield* checkpoints
168 .captureBaseline({ scope, ordinalWithinScope: 2 })
169 .pipe(Effect.result, Effect.forkChild);
170 yield* TestClock.adjust("5 seconds");
171 assert.strictEqual((yield* Fiber.join(slowBaseline))._tag, "Failure");
172 assert.strictEqual(yield* Ref.get(snapshots), 2);
173 assert.strictEqual((yield* Ref.get(refs)).size, 1);
174 }).pipe(Effect.provide(layerTest));
175 }),
176);
177
178it.effect("records a repository probe failure as an error checkpoint, not a missing one", () =>
179 Effect.gen(function* () {
180 const scope: OrchestrationV2CheckpointScope = {
181 id: CheckpointScopeId.make("checkpoint-scope:probe-failure"),
182 threadId: ThreadId.make("thread:probe-failure"),
183 runId: RunId.make("run:probe-failure:1"),
184 nodeId: NodeId.make("node:probe-failure:1"),
185 parentScopeId: null,
186 providerThreadId: ProviderThreadId.make("provider-thread:probe-failure"),
187 kind: "root_run",
188 ordinalWithinParent: 0,
189 advancesAppRunCount: true,
190 cwd: "/repo/bad-config",
191 createdAt: DateTime.makeUnsafe("2026-07-28T00:00:00.000Z"),
192 };
193 const layerTest = CheckpointService.layer.pipe(
194 Layer.provide(
195 Layer.mergeAll(
196 IdAllocator.layer,
197 Layer.mock(CheckpointStore.CheckpointStore)({
198 isGitRepository: () =>
199 Effect.fail(
200 new VcsProcessExitError({
201 operation: "GitVcsDriver.detectRepository",
202 command: "git",
203 cwd: scope.cwd,
204 exitCode: 128,
205 detail: "fatal: bad config line 1 in file .git/config",
206 }),
207 ),
208 }),
209 ),
210 ),
211 Layer.provideMerge(NodeCrypto.layer),
212 );
213
214 const statuses = yield* Effect.gen(function* () {
215 const checkpoints = yield* CheckpointService.CheckpointServiceV2;
216 const baseline = yield* checkpoints.materializeBaselineCheckpoint({
217 scope,
218 ordinalWithinScope: 0,
219 nodeId: scope.nodeId,
220 capturedAt: scope.createdAt,
221 });
222 const turn = yield* checkpoints.capture({
223 scope,
224 ordinalWithinScope: 1,
225 runId: scope.runId!,
226 nodeId: scope.nodeId!,
227 appRunOrdinal: 1,
228 capturedAt: scope.createdAt,
229 });
230 return { baseline: baseline.status, turn: turn.status };
231 }).pipe(Effect.provide(layerTest));
232
233 assert.deepEqual(statuses, { baseline: "error", turn: "error" });
234 }),
235);