authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-09-04 08:05:37-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-04 08:05:37-07:00
logc163d502dd32b993c03d8c20a5fae55b159bd8dc
tree7aaea18d35524268a41de29d9e490ab0cf18e8b5
parent4ee2a9d046a0986451c703c9221bb32ba53540bb
signature Signed by PGP key B5690EEEBB952194

perf(server): avoid full patches for checkpoint summaries (#9694)

14 files changed, 376 insertions(+), 137 deletions(-)

apps/server/package.json-1
...@@ -28,7 +28,6 @@...@@ -28,7 +28,6 @@
28 "@effect/sql-sqlite-bun": "catalog:",28 "@effect/sql-sqlite-bun": "catalog:",
29 "@ff-labs/fff-node": "0.9.4",29 "@ff-labs/fff-node": "0.9.4",
30 "@opencode-ai/sdk": "^1.3.15",30 "@opencode-ai/sdk": "^1.3.15",
31 "@pierre/diffs": "catalog:",
32 "effect": "catalog:",31 "effect": "catalog:",
33 "msgpackr-extract": "3.0.4",32 "msgpackr-extract": "3.0.4",
34 "node-pty": "^1.1.0",33 "node-pty": "^1.1.0",
apps/server/src/checkpointing/CheckpointStore.test.ts+178
...@@ -4,6 +4,7 @@ import * as NodePath from "node:path";...@@ -4,6 +4,7 @@ import * as NodePath from "node:path";
4import * as NodeServices from "@effect/platform-node/NodeServices";4import * as NodeServices from "@effect/platform-node/NodeServices";
5import { it } from "@effect/vitest";5import { it } from "@effect/vitest";
6import { ThreadId, type VcsError } from "@t3tools/contracts";6import { ThreadId, type VcsError } from "@t3tools/contracts";
7import { HostProcessPlatform } from "@t3tools/shared/hostProcess";
7import * as Effect from "effect/Effect";8import * as Effect from "effect/Effect";
8import * as FileSystem from "effect/FileSystem";9import * as FileSystem from "effect/FileSystem";
9import * as Layer from "effect/Layer";10import * as Layer from "effect/Layer";
...@@ -12,6 +13,7 @@ import * as Scope from "effect/Scope";...@@ -12,6 +13,7 @@ import * as Scope from "effect/Scope";
12import { describe, expect } from "vite-plus/test";13import { describe, expect } from "vite-plus/test";
1314
14import { checkpointRefForThreadTurn } from "./Utils.ts";15import { checkpointRefForThreadTurn } from "./Utils.ts";
16import { parseTurnDiffFilesFromNumstat } from "./Diffs.ts";
15import * as CheckpointStore from "./CheckpointStore.ts";17import * as CheckpointStore from "./CheckpointStore.ts";
16import * as VcsDriverRegistry from "../vcs/VcsDriverRegistry.ts";18import * as VcsDriverRegistry from "../vcs/VcsDriverRegistry.ts";
17import * as VcsProcess from "../vcs/VcsProcess.ts";19import * as VcsProcess from "../vcs/VcsProcess.ts";
...@@ -250,6 +252,182 @@ it.layer(TestLayer)("CheckpointStore.layer", (it) => {...@@ -250,6 +252,182 @@ it.layer(TestLayer)("CheckpointStore.layer", (it) => {
250 expect(whitespaceIgnoredDiff).toContain("+ <div>");252 expect(whitespaceIgnoredDiff).toContain("+ <div>");
251 expect(whitespaceIgnoredDiff).not.toContain("- <h1>Title</h1>");253 expect(whitespaceIgnoredDiff).not.toContain("- <h1>Title</h1>");
252 expect(whitespaceIgnoredDiff).not.toContain("+ <h1>Title</h1>");254 expect(whitespaceIgnoredDiff).not.toContain("+ <h1>Title</h1>");
255
256 for (const ignoreWhitespace of [false, true]) {
257 const numstat = yield* checkpointStore.diffCheckpoints({
258 cwd: tmp,
259 fromCheckpointRef,
260 toCheckpointRef,
261 ignoreWhitespace,
262 format: "numstat",
263 });
264 expect(parseTurnDiffFilesFromNumstat(numstat)).toEqual([
265 {
266 path: "Component.tsx",
267 additions: ignoreWhitespace ? 4 : 6,
268 deletions: ignoreWhitespace ? 0 : 2,
269 },
270 ]);
271 }
272 }),
273 );
274 });
275
276 describe("checkpoint file summaries", () => {
277 it.effect("counts changes whose full patch exceeds the output limit", () =>
278 Effect.gen(function* () {
279 const tmp = yield* makeTmpDir();
280 yield* initRepoWithCommit(tmp);
281 const checkpointStore = yield* CheckpointStore.CheckpointStore;
282 const threadId = ThreadId.make("large-checkpoint-summary");
283 const fromCheckpointRef = checkpointRefForThreadTurn(threadId, 0);
284 const toCheckpointRef = checkpointRefForThreadTurn(threadId, 1);
285 const filePath = NodePath.join(tmp, "README.md");
286 const lineCount = 20_000;
287 yield* writeTextFile(filePath, `${"before".repeat(50)}\n`.repeat(lineCount));
288 yield* checkpointStore.captureCheckpoint({ cwd: tmp, checkpointRef: fromCheckpointRef });
289 yield* writeTextFile(filePath, `${"after".repeat(60)}\n`.repeat(lineCount));
290 yield* checkpointStore.captureCheckpoint({ cwd: tmp, checkpointRef: toCheckpointRef });
291
292 const numstat = yield* checkpointStore.diffCheckpoints({
293 cwd: tmp,
294 fromCheckpointRef,
295 toCheckpointRef,
296 ignoreWhitespace: false,
297 format: "numstat",
298 });
299
300 expect(parseTurnDiffFilesFromNumstat(numstat)).toEqual([
301 { path: "README.md", additions: lineCount, deletions: lineCount },
302 ]);
303 expect(numstat.length).toBeLessThan(100);
304 }),
305 );
306
307 it.effect("preserves file paths and turn ranges without changing the user index", () =>
308 Effect.gen(function* () {
309 const tmp = yield* makeTmpDir();
310 yield* initRepoWithCommit(tmp);
311 yield* git(tmp, ["config", "diff.renames", "copies"]);
312 const fileSystem = yield* FileSystem.FileSystem;
313 const checkpointStore = yield* CheckpointStore.CheckpointStore;
314 const threadId = ThreadId.make("checkpoint-summary-paths");
315 const baseline = checkpointRefForThreadTurn(threadId, 0);
316 const firstTurn = checkpointRefForThreadTurn(threadId, 1);
317 const secondTurn = checkpointRefForThreadTurn(threadId, 2);
318 const copiedText = Array.from({ length: 20 }, (_, index) => `copy line ${index}\n`).join(
319 "",
320 );
321 const platform = yield* HostProcessPlatform;
322 const renamedPath = platform === "win32" ? "renamed café.txt" : "renamed\tcafé\nname.txt";
323 const addedPath = platform === "win32" ? "new café.txt" : "new\tfile\n名.txt";
324 for (const [path, contents] of Object.entries({
325 "copy-source.txt": copiedText,
326 "deleted.txt": "delete me\n",
327 "rename-old.txt": "before\nkeep one\nkeep two\nkeep three\n",
328 "binary.bin": "\0before",
329 })) {
330 yield* writeTextFile(NodePath.join(tmp, path), contents);
331 }
332 yield* checkpointStore.captureCheckpoint({ cwd: tmp, checkpointRef: baseline });
333
334 yield* fileSystem.rename(
335 NodePath.join(tmp, "rename-old.txt"),
336 NodePath.join(tmp, renamedPath),
337 );
338 yield* fileSystem.remove(NodePath.join(tmp, "deleted.txt"));
339 for (const [path, contents] of Object.entries({
340 "copy-source.txt": `${copiedText}one more\n`,
341 "copied.txt": copiedText,
342 [renamedPath]: "after\nkeep one\nkeep two\nkeep three\n",
343 "binary.bin": "\0after",
344 "empty.txt": "",
345 [addedPath]: "first\nsecond\n",
346 })) {
347 yield* writeTextFile(NodePath.join(tmp, path), contents);
348 }
349 yield* checkpointStore.captureCheckpoint({ cwd: tmp, checkpointRef: firstTurn });
350 const userIndex = yield* fileSystem.readFile(NodePath.join(tmp, ".git/index"));
351 const input = {
352 cwd: tmp,
353 fromCheckpointRef: baseline,
354 toCheckpointRef: firstTurn,
355 ignoreWhitespace: false,
356 format: "numstat" as const,
357 };
358 const firstSummary = parseTurnDiffFilesFromNumstat(
359 yield* checkpointStore.diffCheckpoints(input),
360 );
361 const expectedFiles = [
362 { path: "binary.bin", additions: 0, deletions: 0 },
363 { path: "copied.txt", additions: 0, deletions: 0 },
364 { path: "copy-source.txt", additions: 1, deletions: 0 },
365 { path: "deleted.txt", additions: 0, deletions: 1 },
366 { path: "empty.txt", additions: 0, deletions: 0 },
367 { path: addedPath, additions: 2, deletions: 0 },
368 { path: renamedPath, additions: 1, deletions: 1 },
369 ].toSorted((left, right) => left.path.localeCompare(right.path));
370 expect(firstSummary).toEqual(expectedFiles);
371
372 yield* fileSystem.remove(NodePath.join(tmp, "empty.txt"));
373 yield* writeTextFile(NodePath.join(tmp, "copy-source.txt"), "replacement\n");
374 yield* checkpointStore.captureCheckpoint({ cwd: tmp, checkpointRef: secondTurn });
375 const secondSummary = parseTurnDiffFilesFromNumstat(
376 yield* checkpointStore.diffCheckpoints({
377 ...input,
378 fromCheckpointRef: firstTurn,
379 toCheckpointRef: secondTurn,
380 }),
381 );
382 expect(secondSummary).toEqual([
383 { path: "copy-source.txt", additions: 1, deletions: 21 },
384 { path: "empty.txt", additions: 0, deletions: 0 },
385 ]);
386
387 const inclusiveSummary = parseTurnDiffFilesFromNumstat(
388 yield* checkpointStore.diffCheckpoints({ ...input, toCheckpointRef: secondTurn }),
389 );
390 expect(inclusiveSummary).toEqual(
391 expectedFiles
392 .filter((file) => file.path !== "empty.txt")
393 .map((file) =>
394 file.path === "copy-source.txt" ? { ...file, additions: 1, deletions: 20 } : file,
395 ),
396 );
397 expect(
398 yield* checkpointStore.diffCheckpoints({ ...input, toCheckpointRef: baseline }),
399 ).toBe("");
400 expect(yield* fileSystem.readFile(NodePath.join(tmp, ".git/index"))).toEqual(userIndex);
401 }),
402 );
403
404 it.effect("uses HEAD for a missing baseline only when requested", () =>
405 Effect.gen(function* () {
406 const tmp = yield* makeTmpDir();
407 yield* initRepoWithCommit(tmp);
408 const checkpointStore = yield* CheckpointStore.CheckpointStore;
409 const threadId = ThreadId.make("checkpoint-summary-fallback");
410 const fromCheckpointRef = checkpointRefForThreadTurn(threadId, 0);
411 const toCheckpointRef = checkpointRefForThreadTurn(threadId, 1);
412 yield* writeTextFile(NodePath.join(tmp, "README.md"), "changed\n");
413 yield* checkpointStore.captureCheckpoint({ cwd: tmp, checkpointRef: toCheckpointRef });
414 const input = {
415 cwd: tmp,
416 fromCheckpointRef,
417 toCheckpointRef,
418 ignoreWhitespace: false,
419 format: "numstat" as const,
420 };
421
422 const error = yield* Effect.flip(checkpointStore.diffCheckpoints(input));
423 expect(error._tag).toBe("VcsProcessExitError");
424 const numstat = yield* checkpointStore.diffCheckpoints({
425 ...input,
426 fallbackFromToHead: true,
427 });
428 expect(parseTurnDiffFilesFromNumstat(numstat)).toEqual([
429 { path: "README.md", additions: 1, deletions: 1 },
430 ]);
253 }),431 }),
254 );432 );
255 });433 });
apps/server/src/checkpointing/CheckpointStore.ts+3-1
...@@ -39,6 +39,7 @@ export interface DiffCheckpointsInput {...@@ -39,6 +39,7 @@ export interface DiffCheckpointsInput {
39 readonly toCheckpointRef: CheckpointRef;39 readonly toCheckpointRef: CheckpointRef;
40 readonly fallbackFromToHead?: boolean;40 readonly fallbackFromToHead?: boolean;
41 readonly ignoreWhitespace: boolean;41 readonly ignoreWhitespace: boolean;
42 readonly format?: "patch" | "numstat";
42}43}
4344
44export interface DeleteCheckpointRefsInput {45export interface DeleteCheckpointRefsInput {
...@@ -77,8 +78,9 @@ export class CheckpointStore extends Context.Service<...@@ -77,8 +78,9 @@ export class CheckpointStore extends Context.Service<
77 ) => Effect.Effect<boolean, CheckpointStoreError>;78 ) => Effect.Effect<boolean, CheckpointStoreError>;
7879
79 /**80 /**
80 * Compute a patch diff between two checkpoint refs.81 * Compute a diff between two checkpoint refs. Defaults to a full patch.
81 *82 *
83 * Numstat output has NUL-delimited paths for file summaries.
82 * Can optionally treat a missing "from" ref as `HEAD`.84 * Can optionally treat a missing "from" ref as `HEAD`.
83 */85 */
84 readonly diffCheckpoints: (86 readonly diffCheckpoints: (
apps/server/src/checkpointing/Diffs.test.ts+34-48
...@@ -1,68 +1,54 @@...@@ -1,68 +1,54 @@
1import { describe, expect, it } from "vite-plus/test";1import { describe, expect, it } from "vite-plus/test";
22
3import { parseTurnDiffFilesFromUnifiedDiff } from "./Diffs.ts";3import { parseTurnDiffFilesFromNumstat } from "./Diffs.ts";
44
5describe("parseTurnDiffFilesFromUnifiedDiff", () => {5describe("parseTurnDiffFilesFromNumstat", () => {
6 it("returns empty list for empty diff", () => {6 it("returns an empty list when no files changed", () => {
7 expect(parseTurnDiffFilesFromUnifiedDiff("")).toEqual([]);7 expect(parseTurnDiffFilesFromNumstat("")).toEqual([]);
8 });8 });
99
10 it("parses per-file additions and deletions", () => {10 it("sorts files and preserves addition and deletion counts", () => {
11 const diff = [11 const numstat = ["0\t2\tsrc/b.ts", "2\t1\ta.txt", ""].join("\0");
12 "diff --git a/a.txt b/a.txt",12 expect(parseTurnDiffFilesFromNumstat(numstat)).toEqual([
13 "index 1111111..2222222 100644",
14 "--- a/a.txt",
15 "+++ b/a.txt",
16 "@@ -1,2 +1,3 @@",
17 " one",
18 "-two",
19 "+two updated",
20 "+three",
21 "diff --git a/src/b.ts b/src/b.ts",
22 "index 3333333..4444444 100644",
23 "--- a/src/b.ts",
24 "+++ b/src/b.ts",
25 "@@ -3,2 +3,0 @@",
26 "-old",
27 "-stale",
28 "",
29 ].join("\n");
30
31 expect(parseTurnDiffFilesFromUnifiedDiff(diff)).toEqual([
32 { path: "a.txt", additions: 2, deletions: 1 },13 { path: "a.txt", additions: 2, deletions: 1 },
33 { path: "src/b.ts", additions: 0, deletions: 2 },14 { path: "src/b.ts", additions: 0, deletions: 2 },
34 ]);15 ]);
35 });16 });
3617
37 it("parses rename-only diffs with zero line changes", () => {18 it("uses destination paths for renames and copies", () => {
38 const diff = [19 const numstat = [
39 "diff --git a/src/old.ts b/src/new.ts",20 "0\t0\t",
40 "similarity index 100%",21 "src/old.ts",
41 "rename from src/old.ts",22 "src/new.ts",
42 "rename to src/new.ts",23 "2\t1\t",
24 "src/source.ts",
25 "src/copied.ts",
26 "1\t0\tother.ts",
43 "",27 "",
44 ].join("\n");28 ].join("\0");
4529
46 expect(parseTurnDiffFilesFromUnifiedDiff(diff)).toEqual([30 expect(parseTurnDiffFilesFromNumstat(numstat)).toEqual([
31 { path: "other.ts", additions: 1, deletions: 0 },
32 { path: "src/copied.ts", additions: 2, deletions: 1 },
47 { path: "src/new.ts", additions: 0, deletions: 0 },33 { path: "src/new.ts", additions: 0, deletions: 0 },
48 ]);34 ]);
49 });35 });
5036
51 it("normalizes CRLF input before parsing", () => {37 it("keeps binary files and empty files with zero line changes", () => {
52 const diff = [38 const numstat = ["-\t-\timage.png", "0\t0\tempty.txt", ""].join("\0");
53 "diff --git a/a.txt b/a.txt",39 expect(parseTurnDiffFilesFromNumstat(numstat)).toEqual([
54 "index 1111111..2222222 100644",40 { path: "empty.txt", additions: 0, deletions: 0 },
55 "--- a/a.txt",41 { path: "image.png", additions: 0, deletions: 0 },
56 "+++ b/a.txt",42 ]);
57 "@@ -1 +1,2 @@",43 });
58 "-one",
59 "+one updated",
60 "+two",
61 "",
62 ].join("\r\n");
6344
64 expect(parseTurnDiffFilesFromUnifiedDiff(diff)).toEqual([45 it("preserves Unicode, tabs, line endings, and spaces in paths", () => {
65 { path: "a.txt", additions: 2, deletions: 1 },46 const path = " café\tline\r\nname.txt ";
47 const numstat = `3\t2\t\0old\tname\n.txt\0${path}\0`;
48
49 expect(parseTurnDiffFilesFromNumstat(numstat)).toEqual([{ path, additions: 3, deletions: 2 }]);
50 expect(parseTurnDiffFilesFromNumstat(`1\t0\t${path}\0`)).toEqual([
51 { path, additions: 1, deletions: 0 },
66 ]);52 ]);
67 });53 });
68});54});
apps/server/src/checkpointing/Diffs.ts+23-17
...@@ -1,27 +1,33 @@...@@ -1,27 +1,33 @@
1import { parsePatchFiles } from "@pierre/diffs/utils/parsePatchFiles";
2
3export interface TurnDiffFileSummary {1export interface TurnDiffFileSummary {
4 readonly path: string;2 readonly path: string;
5 readonly additions: number;3 readonly additions: number;
6 readonly deletions: number;4 readonly deletions: number;
7}5}
86
9export function parseTurnDiffFilesFromUnifiedDiff(7/** Reads Git's NUL-delimited numstat output without decoding display paths. */
10 diff: string,8export function parseTurnDiffFilesFromNumstat(numstat: string): ReadonlyArray<TurnDiffFileSummary> {
11): ReadonlyArray<TurnDiffFileSummary> {9 const records = numstat.split("\0");
12 const normalized = diff.replace(/\r\n/g, "\n").trim();10 const files: TurnDiffFileSummary[] = [];
13 if (normalized.length === 0) {11
14 return [];12 for (let index = 0; index < records.length; index += 1) {
15 }13 const record = records[index]!;
14 const counts = /^(\d+|-)\t(\d+|-)\t/.exec(record);
15 if (!counts) continue;
1616
17 const parsedPatches = parsePatchFiles(normalized);17 let path = record.slice(counts[0].length);
18 const files = parsedPatches.flatMap((patch) =>18 if (path.length === 0) {
19 patch.files.map((file) => ({19 // Renames and copies use two more records: the source and destination.
20 path: file.name,20 path = records[index + 2] ?? "";
21 additions: file.hunks.reduce((total, hunk) => total + hunk.additionLines, 0),21 index += 2;
22 deletions: file.hunks.reduce((total, hunk) => total + hunk.deletionLines, 0),22 }
23 })),23 if (path.length === 0) continue;
24 );24
25 files.push({
26 path,
27 additions: counts[1] === "-" ? 0 : Number(counts[1]),
28 deletions: counts[2] === "-" ? 0 : Number(counts[2]),
29 });
30 }
2531
26 return files.toSorted((left, right) => left.path.localeCompare(right.path));32 return files.toSorted((left, right) => left.path.localeCompare(right.path));
27}33}
apps/server/src/orchestration/Layers/CheckpointReactor.test.ts+101-61
...@@ -27,6 +27,7 @@ import * as Exit from "effect/Exit";...@@ -27,6 +27,7 @@ import * as Exit from "effect/Exit";
27import * as Layer from "effect/Layer";27import * as Layer from "effect/Layer";
28import * as ManagedRuntime from "effect/ManagedRuntime";28import * as ManagedRuntime from "effect/ManagedRuntime";
29import * as PubSub from "effect/PubSub";29import * as PubSub from "effect/PubSub";
30import * as Queue from "effect/Queue";
30import * as Scope from "effect/Scope";31import * as Scope from "effect/Scope";
31import * as Stream from "effect/Stream";32import * as Stream from "effect/Stream";
32import { it as effectIt } from "@effect/vitest";33import { it as effectIt } from "@effect/vitest";
...@@ -43,7 +44,8 @@ import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts";...@@ -43,7 +44,8 @@ import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts";
43import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts";44import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts";
44import * as ThreadBackgroundLiveness from "../ThreadBackgroundLiveness.ts";45import * as ThreadBackgroundLiveness from "../ThreadBackgroundLiveness.ts";
45import * as ThreadPlanProgress from "../ThreadPlanProgress.ts";46import * as ThreadPlanProgress from "../ThreadPlanProgress.ts";
46import { RuntimeReceiptBusLive } from "./RuntimeReceiptBus.ts";47import { RuntimeReceiptBusTest } from "./RuntimeReceiptBus.ts";
48import * as RuntimeReceiptBus from "../Services/RuntimeReceiptBus.ts";
47import { OrchestrationEventStoreLive } from "../../persistence/Layers/OrchestrationEventStore.ts";49import { OrchestrationEventStoreLive } from "../../persistence/Layers/OrchestrationEventStore.ts";
48import { OrchestrationCommandReceiptRepositoryLive } from "../../persistence/Layers/OrchestrationCommandReceipts.ts";50import { OrchestrationCommandReceiptRepositoryLive } from "../../persistence/Layers/OrchestrationCommandReceipts.ts";
49import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts";51import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts";
...@@ -262,7 +264,8 @@ describe("CheckpointReactor", () => {...@@ -262,7 +264,8 @@ describe("CheckpointReactor", () => {
262 | OrchestrationEngineService264 | OrchestrationEngineService
263 | CheckpointReactor265 | CheckpointReactor
264 | CheckpointStore.CheckpointStore266 | CheckpointStore.CheckpointStore
265 | ProjectionSnapshotQuery,267 | ProjectionSnapshotQuery
268 | RuntimeReceiptBus.RuntimeReceiptBus,
266 unknown269 unknown
267 > | null = null;270 > | null = null;
268 let scope: Scope.Closeable | null = null;271 let scope: Scope.Closeable | null = null;
...@@ -356,7 +359,7 @@ describe("CheckpointReactor", () => {...@@ -356,7 +359,7 @@ describe("CheckpointReactor", () => {
356 const layer = CheckpointReactorLive.pipe(359 const layer = CheckpointReactorLive.pipe(
357 Layer.provideMerge(orchestrationLayer),360 Layer.provideMerge(orchestrationLayer),
358 Layer.provideMerge(projectionSnapshotLayer),361 Layer.provideMerge(projectionSnapshotLayer),
359 Layer.provideMerge(RuntimeReceiptBusLive),362 Layer.provideMerge(RuntimeReceiptBusTest),
360 Layer.provideMerge(Layer.succeed(ProviderService, provider.service)),363 Layer.provideMerge(Layer.succeed(ProviderService, provider.service)),
361 Layer.provideMerge(Layer.mock(PullRequestService)({ refreshAfterTurn })),364 Layer.provideMerge(Layer.mock(PullRequestService)({ refreshAfterTurn })),
362 Layer.provideMerge(vcsStatusBroadcasterLayer),365 Layer.provideMerge(vcsStatusBroadcasterLayer),
...@@ -380,8 +383,21 @@ describe("CheckpointReactor", () => {...@@ -380,8 +383,21 @@ describe("CheckpointReactor", () => {
380 const checkpointStore = await runtime.runPromise(383 const checkpointStore = await runtime.runPromise(
381 Effect.service(CheckpointStore.CheckpointStore),384 Effect.service(CheckpointStore.CheckpointStore),
382 );385 );
383 scope = await Effect.runPromise(Scope.make("sequential"));386 const receiptBus = await runtime.runPromise(
384 await Effect.runPromise(reactor.start().pipe(Scope.provide(scope)));387 Effect.service(RuntimeReceiptBus.RuntimeReceiptBus),
388 );
389 const testScope = await Effect.runPromise(Scope.make("sequential"));
390 scope = testScope;
391 const receipts = await Effect.runPromise(
392 Effect.gen(function* () {
393 const receipts = yield* Queue.unbounded<RuntimeReceiptBus.OrchestrationRuntimeReceipt>();
394 yield* Stream.runForEach(receiptBus.streamEventsForTest, (receipt) =>
395 Queue.offer(receipts, receipt),
396 ).pipe(Effect.forkIn(testScope, { startImmediately: true }));
397 yield* reactor.start().pipe(Scope.provide(testScope));
398 return receipts;
399 }),
400 );
385 const drain = () => Effect.runPromise(reactor.drain);401 const drain = () => Effect.runPromise(reactor.drain);
386402
387 const createdAt = "2026-01-01T00:00:00.000Z";403 const createdAt = "2026-01-01T00:00:00.000Z";
...@@ -470,16 +486,19 @@ describe("CheckpointReactor", () => {...@@ -470,16 +486,19 @@ describe("CheckpointReactor", () => {
470 provider,486 provider,
471 cwd,487 cwd,
472 drain,488 drain,
489 nextReceipt: Queue.take(receipts),
473 pullRequestRefreshes,490 pullRequestRefreshes,
474 };491 };
475 }492 }
476493
477 it("captures pre-turn baseline on turn.started and post-turn checkpoint on turn.completed", async () => {494 effectIt.effect("captures baseline and large turn summaries before completion receipts", () =>
478 const harness = await createHarness({ seedFilesystemCheckpoints: false });495 Effect.gen(function* () {
479 const createdAt = "2026-01-01T00:00:00.000Z";496 const harness = yield* Effect.promise(() =>
497 createHarness({ seedFilesystemCheckpoints: false }),
498 );
499 const createdAt = "2026-01-01T00:00:00.000Z";
480500
481 await Effect.runPromise(501 yield* harness.engine.dispatch({
482 harness.engine.dispatch({
483 type: "thread.session.set",502 type: "thread.session.set",
484 commandId: CommandId.make("cmd-session-set-capture"),503 commandId: CommandId.make("cmd-session-set-capture"),
485 threadId: ThreadId.make("thread-1"),504 threadId: ThreadId.make("thread-1"),
...@@ -493,62 +512,83 @@ describe("CheckpointReactor", () => {...@@ -493,62 +512,83 @@ describe("CheckpointReactor", () => {
493 updatedAt: createdAt,512 updatedAt: createdAt,
494 },513 },
495 createdAt,514 createdAt,
496 }),515 });
497 );
498516
499 harness.provider.emit({517 harness.provider.emit({
500 type: "turn.started",518 type: "turn.started",
501 eventId: EventId.make("evt-turn-started-1"),519 eventId: EventId.make("evt-turn-started-1"),
502 provider: ProviderDriverKind.make("codex"),520 provider: ProviderDriverKind.make("codex"),
503521
504 createdAt: "2026-01-01T00:00:00.000Z",522 createdAt: "2026-01-01T00:00:00.000Z",
505 threadId: ThreadId.make("thread-1"),523 threadId: ThreadId.make("thread-1"),
506 turnId: asTurnId("turn-1"),524 turnId: asTurnId("turn-1"),
507 });525 });
508 await waitForGitRefExists(526 expect(yield* harness.nextReceipt).toMatchObject({
509 harness.cwd,527 type: "checkpoint.baseline.captured",
510 checkpointRefForThreadTurn(ThreadId.make("thread-1"), 0),528 checkpointTurnCount: 0,
511 );529 });
512530
513 NodeFS.writeFileSync(NodePath.join(harness.cwd, "README.md"), "v2\n", "utf8");531 NodeFS.writeFileSync(NodePath.join(harness.cwd, "README.md"), "v2\n", "utf8");
514 harness.provider.emit({532 const largeFileLineCount = 25_000;
515 type: "turn.completed",533 NodeFS.writeFileSync(
516 eventId: EventId.make("evt-turn-completed-1"),534 NodePath.join(harness.cwd, "large.txt"),
517 provider: ProviderDriverKind.make("codex"),535 `${"payload".repeat(64)}\n`.repeat(largeFileLineCount),
536 "utf8",
537 );
538 harness.provider.emit({
539 type: "turn.completed",
540 eventId: EventId.make("evt-turn-completed-1"),
541 provider: ProviderDriverKind.make("codex"),
518542
519 createdAt: "2026-01-01T00:00:00.000Z",543 createdAt: "2026-01-01T00:00:00.000Z",
520 threadId: ThreadId.make("thread-1"),544 threadId: ThreadId.make("thread-1"),
521 turnId: asTurnId("turn-1"),545 turnId: asTurnId("turn-1"),
522 payload: { state: "completed" },546 payload: { state: "completed" },
523 });547 });
524548
525 await waitForEvent(harness.engine, (event) => event.type === "thread.turn-diff-completed");549 expect(yield* harness.nextReceipt).toMatchObject({
526 const thread = await waitForThread(550 type: "checkpoint.diff.finalized",
527 harness.readModel,551 turnId: "turn-1",
528 (entry) => entry.latestTurn?.turnId === "turn-1" && entry.checkpoints.length === 1,552 checkpointTurnCount: 1,
529 );553 });
530 expect(thread.checkpoints[0]?.checkpointTurnCount).toBe(1);554 const thread = (yield* Effect.promise(harness.readModel)).threads.find(
531 expect(555 (entry) => entry.id === "thread-1",
532 gitRefExists(harness.cwd, checkpointRefForThreadTurn(ThreadId.make("thread-1"), 0)),556 );
533 ).toBe(true);557 expect(thread?.checkpoints[0]).toMatchObject({
534 expect(558 checkpointTurnCount: 1,
535 gitRefExists(harness.cwd, checkpointRefForThreadTurn(ThreadId.make("thread-1"), 1)),559 files: [
536 ).toBe(true);560 { path: "large.txt", kind: "modified", additions: largeFileLineCount, deletions: 0 },
537 expect(561 { path: "README.md", kind: "modified", additions: 1, deletions: 1 },
538 gitShowFileAtRef(562 ],
539 harness.cwd,563 });
540 checkpointRefForThreadTurn(ThreadId.make("thread-1"), 0),564 expect(yield* harness.nextReceipt).toMatchObject({
541 "README.md",565 type: "turn.processing.quiesced",
542 ),566 turnId: "turn-1",
543 ).toBe("v1\n");567 checkpointTurnCount: 1,
544 expect(568 });
545 gitShowFileAtRef(569 yield* Effect.promise(harness.drain);
546 harness.cwd,570 expect(
547 checkpointRefForThreadTurn(ThreadId.make("thread-1"), 1),571 gitRefExists(harness.cwd, checkpointRefForThreadTurn(ThreadId.make("thread-1"), 0)),
548 "README.md",572 ).toBe(true);
549 ),573 expect(
550 ).toBe("v2\n");574 gitRefExists(harness.cwd, checkpointRefForThreadTurn(ThreadId.make("thread-1"), 1)),
551 });575 ).toBe(true);
576 expect(
577 gitShowFileAtRef(
578 harness.cwd,
579 checkpointRefForThreadTurn(ThreadId.make("thread-1"), 0),
580 "README.md",
581 ),
582 ).toBe("v1\n");
583 expect(
584 gitShowFileAtRef(
585 harness.cwd,
586 checkpointRefForThreadTurn(ThreadId.make("thread-1"), 1),
587 "README.md",
588 ),
589 ).toBe("v2\n");
590 }),
591 );
552592
553 it("refreshes local git status state on turn completion using the session cwd", async () => {593 it("refreshes local git status state on turn completion using the session cwd", async () => {
554 const gitStatusRefreshCalls: string[] = [];594 const gitStatusRefreshCalls: string[] = [];
apps/server/src/orchestration/Layers/CheckpointReactor.ts+3-2
...@@ -21,7 +21,7 @@ import * as Stream from "effect/Stream";...@@ -21,7 +21,7 @@ import * as Stream from "effect/Stream";
21import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker";21import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker";
22import { isTemporaryWorktreeBranch } from "@t3tools/shared/git";22import { isTemporaryWorktreeBranch } from "@t3tools/shared/git";
2323
24import { parseTurnDiffFilesFromUnifiedDiff } from "../../checkpointing/Diffs.ts";24import { parseTurnDiffFilesFromNumstat } from "../../checkpointing/Diffs.ts";
25import {25import {
26 checkpointRefForThreadTurn,26 checkpointRefForThreadTurn,
27 resolveThreadWorkspaceCwd,27 resolveThreadWorkspaceCwd,
...@@ -270,10 +270,11 @@ const make = Effect.gen(function* () {...@@ -270,10 +270,11 @@ const make = Effect.gen(function* () {
270 toCheckpointRef: targetCheckpointRef,270 toCheckpointRef: targetCheckpointRef,
271 fallbackFromToHead: false,271 fallbackFromToHead: false,
272 ignoreWhitespace: false,272 ignoreWhitespace: false,
273 format: "numstat",
273 })274 })
274 .pipe(275 .pipe(
275 Effect.map((diff) =>276 Effect.map((diff) =>
276 parseTurnDiffFilesFromUnifiedDiff(diff).map((file) => ({277 parseTurnDiffFilesFromNumstat(diff).map((file) => ({
277 path: file.path,278 path: file.path,
278 kind: "modified" as const,279 kind: "modified" as const,
279 additions: file.additions,280 additions: file.additions,
apps/server/src/vcs/GitVcsDriver.test.ts+4
...@@ -68,6 +68,7 @@ runVcsDriverContractSuite<GitVcsDriver.GitVcsDriver, GitContractError>({...@@ -68,6 +68,7 @@ runVcsDriverContractSuite<GitVcsDriver.GitVcsDriver, GitContractError>({
68it.effect("GitVcsDriver forwards execute env to the VCS process", () => {68it.effect("GitVcsDriver forwards execute env to the VCS process", () => {
69 let observedEnv: NodeJS.ProcessEnv | undefined;69 let observedEnv: NodeJS.ProcessEnv | undefined;
70 let observedAppendTruncationMarker: boolean | undefined;70 let observedAppendTruncationMarker: boolean | undefined;
71 let observedOutputMode: VcsProcess.VcsProcessInput["outputMode"];
7172
72 return Effect.gen(function* () {73 return Effect.gen(function* () {
73 const driver = yield* GitVcsDriver.makeVcsDriverShape();74 const driver = yield* GitVcsDriver.makeVcsDriverShape();
...@@ -80,12 +81,14 @@ it.effect("GitVcsDriver forwards execute env to the VCS process", () => {...@@ -80,12 +81,14 @@ it.effect("GitVcsDriver forwards execute env to the VCS process", () => {
80 GIT_INDEX_FILE: "/tmp/t3-index",81 GIT_INDEX_FILE: "/tmp/t3-index",
81 },82 },
82 appendTruncationMarker: true,83 appendTruncationMarker: true,
84 outputMode: "error",
83 });85 });
8486
85 assert.deepStrictEqual(observedEnv, {87 assert.deepStrictEqual(observedEnv, {
86 GIT_INDEX_FILE: "/tmp/t3-index",88 GIT_INDEX_FILE: "/tmp/t3-index",
87 });89 });
88 assert.strictEqual(observedAppendTruncationMarker, true);90 assert.strictEqual(observedAppendTruncationMarker, true);
91 assert.strictEqual(observedOutputMode, "error");
89 }).pipe(92 }).pipe(
90 Effect.provide(93 Effect.provide(
91 Layer.mergeAll(94 Layer.mergeAll(
...@@ -95,6 +98,7 @@ it.effect("GitVcsDriver forwards execute env to the VCS process", () => {...@@ -95,6 +98,7 @@ it.effect("GitVcsDriver forwards execute env to the VCS process", () => {
95 Effect.sync(() => {98 Effect.sync(() => {
96 observedEnv = input.env;99 observedEnv = input.env;
97 observedAppendTruncationMarker = input.appendTruncationMarker;100 observedAppendTruncationMarker = input.appendTruncationMarker;
101 observedOutputMode = input.outputMode;
98 return {102 return {
99 exitCode: ChildProcessSpawner.ExitCode(0),103 exitCode: ChildProcessSpawner.ExitCode(0),
100 stdout: "",104 stdout: "",
apps/server/src/vcs/GitVcsDriver.ts+6-1
...@@ -431,6 +431,7 @@ const gitCommand = (...@@ -431,6 +431,7 @@ const gitCommand = (
431 readonly allowNonZeroExit?: boolean;431 readonly allowNonZeroExit?: boolean;
432 readonly timeoutMs?: number;432 readonly timeoutMs?: number;
433 readonly maxOutputBytes?: number;433 readonly maxOutputBytes?: number;
434 readonly outputMode?: VcsProcess.VcsProcessInput["outputMode"];
434 readonly appendTruncationMarker?: boolean;435 readonly appendTruncationMarker?: boolean;
435 },436 },
436) =>437) =>
...@@ -447,6 +448,7 @@ const gitCommand = (...@@ -447,6 +448,7 @@ const gitCommand = (
447 : {}),448 : {}),
448 ...(options?.timeoutMs !== undefined ? { timeoutMs: options.timeoutMs } : {}),449 ...(options?.timeoutMs !== undefined ? { timeoutMs: options.timeoutMs } : {}),
449 ...(options?.maxOutputBytes !== undefined ? { maxOutputBytes: options.maxOutputBytes } : {}),450 ...(options?.maxOutputBytes !== undefined ? { maxOutputBytes: options.maxOutputBytes } : {}),
451 ...(options?.outputMode !== undefined ? { outputMode: options.outputMode } : {}),
450 ...(options?.appendTruncationMarker !== undefined452 ...(options?.appendTruncationMarker !== undefined
451 ? { appendTruncationMarker: options.appendTruncationMarker }453 ? { appendTruncationMarker: options.appendTruncationMarker }
452 : {}),454 : {}),
...@@ -485,6 +487,7 @@ export const makeVcsDriverShape = Effect.fn("makeGitVcsDriverShape")(function* (...@@ -485,6 +487,7 @@ export const makeVcsDriverShape = Effect.fn("makeGitVcsDriverShape")(function* (
485 ...(input.allowNonZeroExit !== undefined ? { allowNonZeroExit: input.allowNonZeroExit } : {}),487 ...(input.allowNonZeroExit !== undefined ? { allowNonZeroExit: input.allowNonZeroExit } : {}),
486 ...(input.timeoutMs !== undefined ? { timeoutMs: input.timeoutMs } : {}),488 ...(input.timeoutMs !== undefined ? { timeoutMs: input.timeoutMs } : {}),
487 ...(input.maxOutputBytes !== undefined ? { maxOutputBytes: input.maxOutputBytes } : {}),489 ...(input.maxOutputBytes !== undefined ? { maxOutputBytes: input.maxOutputBytes } : {}),
490 ...(input.outputMode !== undefined ? { outputMode: input.outputMode } : {}),
488 ...(input.appendTruncationMarker !== undefined491 ...(input.appendTruncationMarker !== undefined
489 ? { appendTruncationMarker: input.appendTruncationMarker }492 ? { appendTruncationMarker: input.appendTruncationMarker }
490 : {}),493 : {}),
...@@ -838,6 +841,7 @@ export const makeVcsDriverShape = Effect.fn("makeGitVcsDriverShape")(function* (...@@ -838,6 +841,7 @@ export const makeVcsDriverShape = Effect.fn("makeGitVcsDriverShape")(function* (
838 "checkpoint.from_ref": input.fromCheckpointRef,841 "checkpoint.from_ref": input.fromCheckpointRef,
839 "checkpoint.to_ref": input.toCheckpointRef,842 "checkpoint.to_ref": input.toCheckpointRef,
840 "checkpoint.ignore_whitespace": input.ignoreWhitespace,843 "checkpoint.ignore_whitespace": input.ignoreWhitespace,
844 "checkpoint.format": input.format ?? "patch",
841 "checkpoint.fallback_from_to_head": input.fallbackFromToHead,845 "checkpoint.fallback_from_to_head": input.fallbackFromToHead,
842 });846 });
843847
...@@ -869,7 +873,7 @@ export const makeVcsDriverShape = Effect.fn("makeGitVcsDriverShape")(function* (...@@ -869,7 +873,7 @@ export const makeVcsDriverShape = Effect.fn("makeGitVcsDriverShape")(function* (
869 cwd: input.cwd,873 cwd: input.cwd,
870 args: [874 args: [
871 "diff",875 "diff",
872 "--patch",876 ...(input.format === "numstat" ? ["--numstat", "-z"] : ["--patch"]),
873 "--no-color",877 "--no-color",
874 "--no-ext-diff",878 "--no-ext-diff",
875 "--no-textconv",879 "--no-textconv",
...@@ -880,6 +884,7 @@ export const makeVcsDriverShape = Effect.fn("makeGitVcsDriverShape")(function* (...@@ -880,6 +884,7 @@ export const makeVcsDriverShape = Effect.fn("makeGitVcsDriverShape")(function* (
880 ],884 ],
881 allowNonZeroExit: true,885 allowNonZeroExit: true,
882 maxOutputBytes: CHECKPOINT_DIFF_MAX_OUTPUT_BYTES,886 maxOutputBytes: CHECKPOINT_DIFF_MAX_OUTPUT_BYTES,
887 outputMode: input.format === "numstat" ? "error" : "truncate",
883 });888 });
884889
885 if (result.exitCode !== 0) {890 if (result.exitCode !== 0) {
apps/server/src/vcs/VcsDriver.ts+1
...@@ -31,6 +31,7 @@ export interface VcsDiffCheckpointsInput {...@@ -31,6 +31,7 @@ export interface VcsDiffCheckpointsInput {
31 readonly toCheckpointRef: CheckpointRef;31 readonly toCheckpointRef: CheckpointRef;
32 readonly fallbackFromToHead?: boolean;32 readonly fallbackFromToHead?: boolean;
33 readonly ignoreWhitespace: boolean;33 readonly ignoreWhitespace: boolean;
34 readonly format?: "patch" | "numstat";
34}35}
3536
36export interface VcsDeleteCheckpointRefsInput {37export interface VcsDeleteCheckpointRefsInput {
apps/server/src/vcs/VcsProcess.test.ts+20-1
...@@ -1,5 +1,6 @@...@@ -1,5 +1,6 @@
1import * as NodeServices from "@effect/platform-node/NodeServices";1import * as NodeServices from "@effect/platform-node/NodeServices";
2import { describe, expect, it } from "@effect/vitest";2import { assert, describe, expect, it } from "@effect/vitest";
3import { HostProcessWorkingDirectory } from "@t3tools/shared/hostProcess";
3import * as Duration from "effect/Duration";4import * as Duration from "effect/Duration";
4import * as Deferred from "effect/Deferred";5import * as Deferred from "effect/Deferred";
5import * as Effect from "effect/Effect";6import * as Effect from "effect/Effect";
...@@ -359,6 +360,24 @@ describe("VcsProcess.run", () => {...@@ -359,6 +360,24 @@ describe("VcsProcess.run", () => {
359 }).pipe(provideLive),360 }).pipe(provideLive),
360 );361 );
361362
363 it.effect("fails with measured byte counts when output must not be truncated", () =>
364 Effect.gen(function* () {
365 const error = yield* run({
366 operation: "test.output-limit",
367 command: "node",
368 args: ["-e", "process.stdout.write('x'.repeat(2048))"],
369 cwd: yield* HostProcessWorkingDirectory,
370 maxOutputBytes: 128,
371 outputMode: "error",
372 }).pipe(Effect.flip);
373
374 assert(error._tag === "VcsProcessOutputLimitError");
375 expect(error.stream).toBe("stdout");
376 expect(error.maxBytes).toBe(128);
377 expect(error.observedBytes).toBeGreaterThan(error.maxBytes);
378 }).pipe(provideLive),
379 );
380
362 it.effect("fails with VcsProcessTimeoutError on timeout", () =>381 it.effect("fails with VcsProcessTimeoutError on timeout", () =>
363 Effect.gen(function* () {382 Effect.gen(function* () {
364 const errorFiber = yield* run({383 const errorFiber = yield* run({
apps/server/src/vcs/VcsProcess.ts+2-1
...@@ -29,6 +29,7 @@ export interface VcsProcessInput {...@@ -29,6 +29,7 @@ export interface VcsProcessInput {
29 readonly allowNonZeroExit?: boolean;29 readonly allowNonZeroExit?: boolean;
30 readonly timeoutMs?: number;30 readonly timeoutMs?: number;
31 readonly maxOutputBytes?: number;31 readonly maxOutputBytes?: number;
32 readonly outputMode?: ProcessRunner.ProcessRunInput["outputMode"];
32 readonly appendTruncationMarker?: boolean;33 readonly appendTruncationMarker?: boolean;
33}34}
3435
...@@ -125,7 +126,7 @@ export const make = Effect.gen(function* () {...@@ -125,7 +126,7 @@ export const make = Effect.gen(function* () {
125 ...(input.env !== undefined ? { env: input.env } : {}),126 ...(input.env !== undefined ? { env: input.env } : {}),
126 timeout: input.timeoutMs ?? DEFAULT_TIMEOUT_MS,127 timeout: input.timeoutMs ?? DEFAULT_TIMEOUT_MS,
127 maxOutputBytes: input.maxOutputBytes ?? DEFAULT_MAX_OUTPUT_BYTES,128 maxOutputBytes: input.maxOutputBytes ?? DEFAULT_MAX_OUTPUT_BYTES,
128 outputMode: "truncate",129 outputMode: input.outputMode ?? "truncate",
129 truncatedMarker: input.appendTruncationMarker ? OUTPUT_TRUNCATED_MARKER : "",130 truncatedMarker: input.appendTruncationMarker ? OUTPUT_TRUNCATED_MARKER : "",
130 timeoutBehavior: "error",131 timeoutBehavior: "error",
131 })132 })
docs/internals/glossary.md+1-1
...@@ -147,7 +147,7 @@ The starting checkpoint for diffing a thread timeline. This flow is surfaced thr...@@ -147,7 +147,7 @@ The starting checkpoint for diffing a thread timeline. This flow is surfaced thr
147147
148#### Checkpoint diff148#### Checkpoint diff
149149
150The patch difference between two checkpoints. Query logic lives in [CheckpointDiffQuery.ts][20], diff parsing lives in [Diffs.ts][23], and finalization is coordinated by [CheckpointReactor.ts][6].150The difference between two checkpoints. [CheckpointDiffQuery.ts][20] reads full patches on demand. [CheckpointReactor.ts][6] uses NUL-delimited Git numstat output for automatic file summaries, parsed by [Diffs.ts][23].
151151
152#### Turn diff152#### Turn diff
153153
pnpm-lock.yaml-3
...@@ -496,9 +496,6 @@ importers:...@@ -496,9 +496,6 @@ importers:
496 '@opencode-ai/sdk':496 '@opencode-ai/sdk':
497 specifier: ^1.3.15497 specifier: ^1.3.15
498 version: 1.15.13498 version: 1.15.13
499 '@pierre/diffs':
500 specifier: 'catalog:'
501 version: 1.3.0-beta.10(patch_hash=7ef7cb0cbabb17c15cdb137554068b36f15f6f5265e73fb51452aa3380db91aa)(@shikijs/themes@4.3.0)(react-dom@19.2.6(react@19.2.6))(react@19.2.6)
502 effect:499 effect:
503 specifier: 4.0.0-beta.103500 specifier: 4.0.0-beta.103
504 version: 4.0.0-beta.103(patch_hash=af36b7948b6f9c56623074662b51dade5699880c1a7c71245de73e13c3185fb6)501 version: 4.0.0-beta.103(patch_hash=af36b7948b6f9c56623074662b51dade5699880c1a7c71245de73e13c3185fb6)