1import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
2import * as NodeServices from "@effect/platform-node/NodeServices";
3import { assert, it, vi } from "@effect/vitest";
4import { CheckpointRef, CheckpointScopeId, RunId, ThreadId } from "@t3tools/contracts";
5import * as Effect from "effect/Effect";
6import * as FileSystem from "effect/FileSystem";
7import * as Layer from "effect/Layer";
8import * as Path from "effect/Path";
9
10import { checkpointRefForScopeOrdinal } from "../orchestration-v2/CheckpointService.ts";
11import { OrchestratorProjectionError } from "../orchestration-v2/Orchestrator.ts";
12import type { ProjectionCheckpointContext } from "../orchestration-v2/ProjectionStore.ts";
13import * as ThreadManagement from "../orchestration-v2/ThreadManagementService.ts";
14import * as VcsDriverRegistry from "../vcs/VcsDriverRegistry.ts";
15import * as VcsProcess from "../vcs/VcsProcess.ts";
16import * as CheckpointDiffQuery from "./CheckpointDiffQuery.ts";
17import * as CheckpointStore from "./CheckpointStore.ts";
18import { CheckpointRefUnavailableError, CheckpointTurnRangeUnavailableError } from "./Errors.ts";
19
20const threadId = ThreadId.make("thread:checkpoint-diff-v2");
21const firstRunId = RunId.make("run:checkpoint-diff-v2:1");
22const secondRunId = RunId.make("run:checkpoint-diff-v2:2");
23const firstScopeId = CheckpointScopeId.make("scope:checkpoint-diff-v2:1");
24const secondScopeId = CheckpointScopeId.make("scope:checkpoint-diff-v2:2");
25const secondRef = CheckpointRef.make("refs/t3/test/second");
26
27function makeProjection(): ProjectionCheckpointContext {
28 return {
29 runs: [
30 { id: firstRunId, ordinal: 1, status: "completed" },
31 { id: secondRunId, ordinal: 2, status: "completed" },
32 ],
33 checkpointScopes: [
34 { id: firstScopeId, runId: firstRunId, kind: "root_run", cwd: "/repo" },
35 { id: secondScopeId, runId: secondRunId, kind: "root_run", cwd: "/repo" },
36 ],
37 checkpoints: [
38 {
39 scopeId: secondScopeId,
40 runId: secondRunId,
41 ordinalWithinScope: 1,
42 appRunOrdinal: 2,
43 status: "ready",
44 ref: secondRef,
45 },
46 ],
47 };
48}
49
50function layerFor(input: {
51 readonly projection: Effect.Effect<ProjectionCheckpointContext, OrchestratorProjectionError>;
52 readonly diffCheckpoints?: CheckpointStore.CheckpointStore["Service"]["diffCheckpoints"];
53 readonly authoredPaths?: ReadonlySet<string>;
54}) {
55 return CheckpointDiffQuery.layer.pipe(
56 Layer.provide(
57 Layer.mergeAll(
58 Layer.mock(ThreadManagement.ThreadManagementService)({
59 getCheckpointContext: () => input.projection,
60 }),
61 Layer.mock(CheckpointStore.CheckpointStore)({
62 diffCheckpoints: input.diffCheckpoints ?? (() => Effect.succeed("diff")),
63 listAuthoredPaths: () => Effect.succeed(input.authoredPaths ?? null),
64 }),
65 ),
66 ),
67 Layer.provideMerge(NodeCrypto.layer),
68 );
69}
70
71it.effect("excludes ready checkpoints from rolled-back runs", () => {
72 const projection = makeProjection();
73 const layer = layerFor({
74 projection: Effect.succeed({
75 ...projection,
76 runs: projection.runs.map((run) =>
77 run.id === secondRunId ? { ...run, status: "rolled_back" as const } : run,
78 ),
79 checkpoints: [
80 {
81 ...projection.checkpoints[0]!,
82 scopeId: firstScopeId,
83 runId: firstRunId,
84 appRunOrdinal: 1,
85 ref: CheckpointRef.make("refs/t3/test/first"),
86 },
87 ...projection.checkpoints,
88 ],
89 }),
90 });
91
92 return Effect.gen(function* () {
93 const query = yield* CheckpointDiffQuery.CheckpointDiffQuery;
94 const error = yield* query
95 .getTurnDiff({ threadId, fromTurnCount: 0, toTurnCount: 2 })
96 .pipe(Effect.flip);
97
98 assert.instanceOf(error, CheckpointTurnRangeUnavailableError);
99 assert.deepEqual(
100 {
101 requestedTurnCount: error.requestedTurnCount,
102 availableTurnCount: error.availableTurnCount,
103 },
104 { requestedTurnCount: 2, availableTurnCount: 1 },
105 );
106 }).pipe(Effect.provide(layer));
107});
108
109function makeProjectionAfterStoppedRun(stoppedRunCheckpoint: boolean) {
110 const thirdRunId = RunId.make("run:checkpoint-diff-v2:3");
111 const checkpoint = (
112 ordinalWithinScope: number,
113 runId: RunId | null,
114 ): ProjectionCheckpointContext["checkpoints"][number] => ({
115 scopeId: firstScopeId,
116 runId,
117 ordinalWithinScope,
118 appRunOrdinal: runId === null ? null : ordinalWithinScope,
119 status: "ready",
120 ref: CheckpointRef.make(`refs/t3/test/ordinal/${ordinalWithinScope}`),
121 });
122 return {
123 runs: [
124 { id: firstRunId, ordinal: 1, status: "completed" },
125 { id: secondRunId, ordinal: 2, status: "interrupted" },
126 { id: thirdRunId, ordinal: 3, status: "completed" },
127 ],
128 checkpointScopes: [
129 { id: firstScopeId, runId: thirdRunId, kind: "root_run", cwd: "/repo" },
130 { id: secondScopeId, runId: secondRunId, kind: "tool", cwd: "/repo" },
131 ],
132 checkpoints: [
133 {
134 ...checkpoint(2, null),
135 scopeId: secondScopeId,
136 ref: CheckpointRef.make("refs/t3/test/tool/2"),
137 },
138 checkpoint(0, null),
139 checkpoint(1, firstRunId),
140 checkpoint(2, stoppedRunCheckpoint ? secondRunId : null),
141 checkpoint(3, thirdRunId),
142 ],
143 } satisfies ProjectionCheckpointContext;
144}
145
146it.effect.each([false, true])(
147 "diffs the turn after a stopped run from its parent checkpoint (stopped run captured: %s)",
148 (stoppedRunCheckpoint) => {
149 const diffCheckpoints = vi.fn((_input: CheckpointStore.DiffCheckpointsInput) =>
150 Effect.succeed("diff --git a/file b/file"),
151 );
152 const layer = layerFor({
153 projection: Effect.succeed(makeProjectionAfterStoppedRun(stoppedRunCheckpoint)),
154 diffCheckpoints,
155 });
156
157 return Effect.gen(function* () {
158 const query = yield* CheckpointDiffQuery.CheckpointDiffQuery;
159 const result = yield* query.getTurnDiff({ threadId, fromTurnCount: 2, toTurnCount: 3 });
160
161 assert.strictEqual(result.diff, "diff --git a/file b/file");
162 yield* query.getTurnDiff({ threadId, fromTurnCount: 1, toTurnCount: 3 });
163 yield* query.getFullThreadDiff({ threadId, toTurnCount: 3 });
164 assert.deepEqual(
165 diffCheckpoints.mock.calls.map(([input]) => ({
166 from: input.fromCheckpointRef,
167 to: input.toCheckpointRef,
168 })),
169 [
170 { from: "refs/t3/test/ordinal/2", to: "refs/t3/test/ordinal/3" },
171 { from: "refs/t3/test/ordinal/1", to: "refs/t3/test/ordinal/3" },
172 {
173 from: yield* checkpointRefForScopeOrdinal({
174 scopeId: firstScopeId,
175 ordinalWithinScope: 0,
176 }),
177 to: "refs/t3/test/ordinal/3",
178 },
179 ],
180 );
181 const stoppedTurnError = yield* query
182 .getTurnDiff({ threadId, fromTurnCount: 1, toTurnCount: 2 })
183 .pipe(Effect.flip);
184 assert.instanceOf(stoppedTurnError, CheckpointRefUnavailableError);
185 assert.deepEqual(
186 { checkpoint: stoppedTurnError.checkpoint, turnCount: stoppedTurnError.turnCount },
187 { checkpoint: "to", turnCount: 2 },
188 );
189 }).pipe(Effect.provide(layer));
190 },
191);
192
193const layerVcsProcess = VcsProcess.layer.pipe(Layer.provide(NodeServices.layer));
194const layerStore = CheckpointStore.layer.pipe(
195 Layer.provide(VcsDriverRegistry.layer.pipe(Layer.provide(layerVcsProcess))),
196 Layer.provide(NodeServices.layer),
197);
198const layerTest = Layer.merge(layerStore, layerVcsProcess).pipe(
199 Layer.provideMerge(NodeServices.layer),
200 Layer.provideMerge(NodeCrypto.layer),
201);
202
203it.effect("shares the turn and full-thread diff budget across real Git batches", () =>
204 Effect.gen(function* () {
205 const fs = yield* FileSystem.FileSystem;
206 const path = yield* Path.Path;
207 const process = yield* VcsProcess.VcsProcess;
208 const store = yield* CheckpointStore.CheckpointStore;
209 const cwd = yield* fs.makeTempDirectoryScoped({ prefix: "t3-diff-aggregate-" });
210 const git = (args: ReadonlyArray<string>, env?: NodeJS.ProcessEnv) =>
211 process.run({
212 operation: "CheckpointDiffQuery.test.git",
213 command: "git",
214 cwd,
215 args,
216 ...(env === undefined ? {} : { env }),
217 });
218 yield* git(["init"]);
219 yield* git(["config", "user.email", "test@test.com"]);
220 yield* git(["config", "user.name", "Test"]);
221 yield* fs.writeFileString(path.join(cwd, "README.md"), "initial\n");
222 yield* git(["add", "."]);
223 yield* git(["commit", "-m", "initial"]);
224
225 const threadId = ThreadId.make("thread:aggregate-runtime");
226 const runId = RunId.make("run:aggregate-runtime");
227 const scopeId = CheckpointScopeId.make("scope:aggregate-runtime");
228 const fromCheckpointRef = yield* checkpointRefForScopeOrdinal({
229 scopeId,
230 ordinalWithinScope: 0,
231 });
232 const toCheckpointRef = CheckpointRef.make("refs/t3/test/aggregate-after");
233 yield* store.captureCheckpoint({ cwd, checkpointRef: fromCheckpointRef });
234 // The old commit models files imported by a pull after the baseline was captured.
235 yield* fs.writeFileString(path.join(cwd, "upstream.ts"), "imported\n");
236 yield* git(["add", "."]);
237 yield* git(["commit", "-m", "imported"], {
238 ...globalThis.process.env,
239 GIT_COMMITTER_DATE: "2000-01-01T00:00:00Z",
240 });
241 const ownPaths = Array.from({ length: 1_600 }, (_, index) => `src/module-${index}/file.ts`);
242 yield* Effect.forEach(
243 ownPaths,
244 (file) =>
245 Effect.gen(function* () {
246 yield* fs.makeDirectory(path.dirname(path.join(cwd, file)), { recursive: true });
247 yield* fs.writeFileString(path.join(cwd, file), `${"x".repeat(12_000)}\n`);
248 }),
249 { concurrency: 16, discard: true },
250 );
251 yield* store.captureCheckpoint({ cwd, checkpointRef: toCheckpointRef });
252
253 const patchSizes: number[] = [];
254 const query = yield* CheckpointDiffQuery.make.pipe(
255 Effect.provide(
256 Layer.mock(ThreadManagement.ThreadManagementService)({
257 getCheckpointContext: () =>
258 Effect.succeed({
259 runs: [{ id: runId, ordinal: 1, status: "completed" }],
260 checkpointScopes: [{ id: scopeId, runId, kind: "root_run", cwd }],
261 checkpoints: [
262 {
263 scopeId,
264 runId,
265 ordinalWithinScope: 1,
266 appRunOrdinal: 1,
267 status: "ready",
268 ref: toCheckpointRef,
269 },
270 ],
271 }),
272 }),
273 ),
274 Effect.provideService(CheckpointStore.CheckpointStore, {
275 ...store,
276 diffCheckpoints: (input) =>
277 store.diffCheckpoints(input).pipe(
278 Effect.tap((patch) =>
279 Effect.sync(() => {
280 if (input.format !== "numstat") {
281 patchSizes.push(Buffer.byteLength(patch));
282 }
283 }),
284 ),
285 ),
286 }),
287 );
288 const turn = yield* query.getTurnDiff({ threadId, fromTurnCount: 0, toTurnCount: 1 });
289 const turnPatchSizes = patchSizes.splice(0);
290 const fullThread = yield* query.getFullThreadDiff({ threadId, toTurnCount: 1 });
291 assert.isTrue(fullThread.diff === turn.diff);
292 assert.isAtMost(Buffer.byteLength(turn.diff), 10_000_000);
293 assert.isTrue(turn.diff.endsWith("\n\n[truncated]"));
294 assert.include(turn.diff, "src/module-0/file.ts");
295 assert.notInclude(turn.diff, "upstream.ts");
296 assert.lengthOf(turnPatchSizes, 2);
297 assert.deepEqual(patchSizes, turnPatchSizes);
298 assert.isBelow(turnPatchSizes[0]!, 10_000_000);
299 assert.strictEqual(
300 turnPatchSizes.reduce((sum, size) => sum + size, 0),
301 10_000_001,
302 );
303 }).pipe(Effect.provide(layerTest), Effect.scoped),
304);
305
306it.effect("truncates a storage-cut UTF-8 character without replacement text", () => {
307 const prefix = "x".repeat(9_999_985);
308 const layer = layerFor({
309 projection: Effect.succeed(makeProjection()),
310 diffCheckpoints: (input) =>
311 Effect.succeed(
312 Buffer.from(`${prefix}😀${"x".repeat(9)}😀tail`)
313 .subarray(0, input.maxOutputBytes)
314 .toString("utf8"),
315 ),
316 });
317 return Effect.gen(function* () {
318 const query = yield* CheckpointDiffQuery.CheckpointDiffQuery;
319 const result = yield* query.getTurnDiff({ threadId, fromTurnCount: 0, toTurnCount: 2 });
320 assert.isAtMost(Buffer.byteLength(result.diff), 10_000_000);
321 assert.isTrue(result.diff === `${prefix}\n\n[truncated]`);
322 }).pipe(Effect.provide(layer));
323});
324
325it.effect("preserves an exact-fit filtered patch across batches without a marker", () => {
326 const ownPaths = Array.from({ length: 1_000 }, (_, index) => `src/module-${index}/file.ts`);
327 const patches = ["é".repeat(3_000_000), "x".repeat(4_000_000)];
328 let patchReads = 0;
329 const layer = layerFor({
330 projection: Effect.succeed(makeProjection()),
331 authoredPaths: new Set(ownPaths),
332 diffCheckpoints: (input) =>
333 Effect.sync(() =>
334 input.format === "numstat"
335 ? [...ownPaths, "upstream.ts"].map((path) => `1\t0\t${path}\0`).join("")
336 : patches[patchReads++]!,
337 ),
338 });
339 return Effect.gen(function* () {
340 const query = yield* CheckpointDiffQuery.CheckpointDiffQuery;
341 const result = yield* query.getFullThreadDiff({ threadId, toTurnCount: 2 });
342 assert.isTrue(result.diff === patches.join(""));
343 assert.strictEqual(patchReads, 2);
344 }).pipe(Effect.provide(layer));
345});