1/**
2 * CheckpointDiffQuery - Query interface for computed checkpoint diffs.
3 *
4 * Provides read-only diff operations across checkpoint snapshots used by
5 * orchestration APIs.
6 *
7 * @module CheckpointDiffQuery
8 */
9import {
10 OrchestrationGetTurnDiffResult,
11 type OrchestrationGetFullThreadDiffInput,
12 type OrchestrationGetFullThreadDiffResult,
13 type OrchestrationGetTurnDiffInput,
14 type OrchestrationGetTurnDiffResult as OrchestrationGetTurnDiffResultType,
15 type ThreadId,
16} from "@t3tools/contracts";
17import * as Context from "effect/Context";
18import * as Crypto from "effect/Crypto";
19import * as Effect from "effect/Effect";
20import * as Layer from "effect/Layer";
21import * as Schema from "effect/Schema";
22
23import { checkpointRefForScopeOrdinal } from "../orchestration-v2/CheckpointService.ts";
24import * as ThreadManagement from "../orchestration-v2/ThreadManagementService.ts";
25import {
26 CheckpointDiffResultInvalidError,
27 CheckpointRefUnavailableError,
28 CheckpointThreadNotFoundError,
29 CheckpointTurnRangeUnavailableError,
30 CheckpointWorkspacePathMissingError,
31 type CheckpointServiceError,
32} from "./Errors.ts";
33import * as CheckpointStore from "./CheckpointStore.ts";
34import { isGitImport, parseTurnDiffFilesFromNumstat, type TurnDiffFileSummary } from "./Diffs.ts";
35
36// Windows limits a command line to 32,767 characters. Leave room for the rest of git's arguments.
37const MAX_PATHSPEC_CHARS = 24_000;
38
39/** Splits files into pathspec lists that each fit on one git command line. Renames keep both paths together. */
40function batchFilePaths(files: ReadonlyArray<TurnDiffFileSummary>) {
41 const batches: string[][] = [];
42 let batch: string[] = [];
43 let length = 0;
44 for (const file of files) {
45 const paths = file.previousPath === undefined ? [file.path] : [file.previousPath, file.path];
46 // Each pathspec also carries a `:(top,literal)` prefix and a separator.
47 const size = paths.reduce((total, path) => total + path.length + 16, 0);
48 if (batch.length > 0 && length + size > MAX_PATHSPEC_CHARS) {
49 batches.push(batch);
50 batch = [];
51 length = 0;
52 }
53 batch.push(...paths);
54 length += size;
55 }
56 if (batch.length > 0) batches.push(batch);
57 return batches;
58}
59
60/** Service tag for checkpoint diff queries. */
61export class CheckpointDiffQuery extends Context.Service<
62 CheckpointDiffQuery,
63 {
64 /**
65 * Read the patch diff for a single turn checkpoint transition.
66 *
67 * Verifies checkpoint availability in both projection state and filesystem.
68 */
69 readonly getTurnDiff: (
70 input: OrchestrationGetTurnDiffInput,
71 ) => Effect.Effect<OrchestrationGetTurnDiffResultType, CheckpointServiceError>;
72
73 /**
74 * Read the full patch diff across a thread range of checkpoints.
75 *
76 * Uses turn-diff semantics with `fromTurnCount = 0`.
77 */
78 readonly getFullThreadDiff: (
79 input: OrchestrationGetFullThreadDiffInput,
80 ) => Effect.Effect<OrchestrationGetFullThreadDiffResult, CheckpointServiceError>;
81 }
82>()("t3/checkpointing/CheckpointDiffQuery") {}
83
84const isTurnDiffResult = Schema.is(OrchestrationGetTurnDiffResult);
85
86function buildTurnDiffResult(
87 input: {
88 readonly threadId: ThreadId;
89 readonly fromTurnCount: number;
90 readonly toTurnCount: number;
91 },
92 diff: string,
93): OrchestrationGetTurnDiffResultType {
94 return {
95 threadId: input.threadId,
96 fromTurnCount: input.fromTurnCount,
97 toTurnCount: input.toTurnCount,
98 diff,
99 };
100}
101
102/** @public Service construction is part of the canonical Effect module API. */
103export const make = Effect.gen(function* () {
104 const threads = yield* ThreadManagement.ThreadManagementService;
105 const checkpointStore = yield* CheckpointStore.CheckpointStore;
106 const crypto = yield* Crypto.Crypto;
107
108 const getTurnDiff: CheckpointDiffQuery["Service"]["getTurnDiff"] = Effect.fn("getTurnDiff")(
109 function* (input) {
110 const operation = "CheckpointDiffQuery.getTurnDiff";
111 const ignoreWhitespace = input.ignoreWhitespace ?? true;
112 yield* Effect.annotateCurrentSpan({
113 "checkpoint.thread_id": input.threadId,
114 "checkpoint.from_turn_count": input.fromTurnCount,
115 "checkpoint.to_turn_count": input.toTurnCount,
116 "checkpoint.ignore_whitespace": ignoreWhitespace,
117 });
118
119 if (input.fromTurnCount === input.toTurnCount) {
120 const emptyDiff = buildTurnDiffResult(input, "");
121 if (!isTurnDiffResult(emptyDiff)) {
122 return yield* new CheckpointDiffResultInvalidError({
123 operation,
124 threadId: input.threadId,
125 });
126 }
127 return emptyDiff;
128 }
129
130 const projection = yield* threads.getCheckpointContext(input.threadId).pipe(
131 Effect.mapError(
132 () =>
133 new CheckpointThreadNotFoundError({
134 operation,
135 threadId: input.threadId,
136 }),
137 ),
138 Effect.withSpan("checkpoint.turnDiff.lookupContext"),
139 );
140 const completedRunIds = new Set(
141 projection.runs.filter((run) => run.status === "completed").map((run) => run.id),
142 );
143 const readyCheckpoints = projection.checkpoints.filter(
144 (checkpoint) =>
145 checkpoint.status === "ready" &&
146 checkpoint.appRunOrdinal !== null &&
147 checkpoint.runId !== null &&
148 completedRunIds.has(checkpoint.runId),
149 );
150 const maxTurnCount = readyCheckpoints.reduce(
151 (max, checkpoint) => Math.max(max, checkpoint.appRunOrdinal ?? 0),
152 0,
153 );
154 if (input.toTurnCount > maxTurnCount) {
155 return yield* new CheckpointTurnRangeUnavailableError({
156 operation,
157 threadId: input.threadId,
158 requestedTurnCount: input.toTurnCount,
159 availableTurnCount: maxTurnCount,
160 });
161 }
162
163 const toCheckpoint = readyCheckpoints.find(
164 (checkpoint) => checkpoint.appRunOrdinal === input.toTurnCount,
165 );
166 if (toCheckpoint === undefined) {
167 return yield* new CheckpointRefUnavailableError({
168 operation,
169 threadId: input.threadId,
170 turnCount: input.toTurnCount,
171 checkpoint: "to",
172 });
173 }
174
175 const toScope = projection.checkpointScopes.find(
176 (scope) => scope.id === toCheckpoint.scopeId,
177 );
178 if (toScope === undefined) {
179 return yield* new CheckpointWorkspacePathMissingError({
180 operation,
181 threadId: input.threadId,
182 });
183 }
184
185 // The root scope is shared by every run in this thread. Its runId
186 // tracks the latest owner, while ordinal zero stays the baseline.
187 const firstScope =
188 input.fromTurnCount === 0
189 ? projection.checkpointScopes.find((scope) => scope.kind === "root_run")
190 : undefined;
191 const fromCheckpointRef =
192 input.fromTurnCount === 0
193 ? firstScope === undefined
194 ? undefined
195 : yield* checkpointRefForScopeOrdinal({
196 scopeId: firstScope.id,
197 ordinalWithinScope: 0,
198 }).pipe(Effect.provideService(Crypto.Crypto, crypto))
199 : readyCheckpoints.find((checkpoint) => checkpoint.appRunOrdinal === input.fromTurnCount)
200 ?.ref;
201 if (fromCheckpointRef === undefined) {
202 return yield* new CheckpointRefUnavailableError({
203 operation,
204 threadId: input.threadId,
205 turnCount: input.fromTurnCount,
206 checkpoint: "from",
207 });
208 }
209
210 const comparison = {
211 cwd: toScope.cwd,
212 fromCheckpointRef,
213 toCheckpointRef: toCheckpoint.ref,
214 fallbackFromToHead: false,
215 ignoreWhitespace,
216 };
217 // Leave out files that a pull, merge, or rebase brought in, matching the turn's file summary.
218 // Filtering is optional: when it cannot run, the complete diff still loads.
219 const authoredPaths = yield* checkpointStore
220 .listAuthoredPaths(comparison)
221 .pipe(Effect.orElseSucceed(() => null));
222 const files =
223 authoredPaths === null
224 ? []
225 : yield* checkpointStore.diffCheckpoints({ ...comparison, format: "numstat" }).pipe(
226 Effect.map(parseTurnDiffFilesFromNumstat),
227 Effect.orElseSucceed(() => []),
228 );
229 const retainedFiles = files.filter((file) => !isGitImport(file, authoredPaths));
230 // Select retained paths before generating a patch, so imported bulk cannot exhaust its output limit.
231 const diff =
232 retainedFiles.length === files.length
233 ? yield* checkpointStore
234 .diffCheckpoints(comparison)
235 .pipe(Effect.withSpan("checkpoint.turnDiff.diffCheckpoints"))
236 : (yield* Effect.forEach(
237 batchFilePaths(retainedFiles),
238 (filePaths) =>
239 checkpointStore
240 .diffCheckpoints({ ...comparison, filePaths })
241 .pipe(Effect.withSpan("checkpoint.turnDiff.diffCheckpoints")),
242 { concurrency: 4 },
243 )).join("");
244
245 const turnDiff = buildTurnDiffResult(input, diff);
246 if (!isTurnDiffResult(turnDiff)) {
247 return yield* new CheckpointDiffResultInvalidError({
248 operation,
249 threadId: input.threadId,
250 });
251 }
252
253 return turnDiff;
254 },
255 );
256
257 const getFullThreadDiff: CheckpointDiffQuery["Service"]["getFullThreadDiff"] = Effect.fn(
258 "CheckpointDiffQuery.getFullThreadDiff",
259 )(function* (input) {
260 const operation = "CheckpointDiffQuery.getFullThreadDiff";
261 const ignoreWhitespace = input.ignoreWhitespace ?? true;
262 yield* Effect.annotateCurrentSpan({
263 "checkpoint.thread_id": input.threadId,
264 "checkpoint.from_turn_count": 0,
265 "checkpoint.to_turn_count": input.toTurnCount,
266 "checkpoint.ignore_whitespace": ignoreWhitespace,
267 "checkpoint.diff_kind": "full-thread",
268 });
269
270 if (input.toTurnCount === 0) {
271 const emptyDiff = buildTurnDiffResult(
272 {
273 threadId: input.threadId,
274 fromTurnCount: 0,
275 toTurnCount: 0,
276 },
277 "",
278 );
279 if (!isTurnDiffResult(emptyDiff)) {
280 return yield* new CheckpointDiffResultInvalidError({
281 operation,
282 threadId: input.threadId,
283 });
284 }
285 return emptyDiff satisfies OrchestrationGetFullThreadDiffResult;
286 }
287
288 const turnDiff = yield* getTurnDiff({
289 threadId: input.threadId,
290 fromTurnCount: 0,
291 toTurnCount: input.toTurnCount,
292 ignoreWhitespace,
293 });
294 if (!isTurnDiffResult(turnDiff)) {
295 return yield* new CheckpointDiffResultInvalidError({
296 operation,
297 threadId: input.threadId,
298 });
299 }
300
301 return turnDiff satisfies OrchestrationGetFullThreadDiffResult;
302 });
303
304 return CheckpointDiffQuery.of({
305 getTurnDiff,
306 getFullThreadDiff,
307 });
308});
309
310export const layer = Layer.effect(CheckpointDiffQuery, make);