fix(server): cap checkpoint diffs across all batches

apps/server/src/checkpointing/CheckpointDiffQuery.test.ts

@@ -1,13 +1,18 @@
 import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
+import * as NodeServices from "@effect/platform-node/NodeServices";
 import { assert, it, vi } from "@effect/vitest";
 import { CheckpointRef, CheckpointScopeId, RunId, ThreadId } from "@t3tools/contracts";
 import * as Effect from "effect/Effect";
+import * as FileSystem from "effect/FileSystem";
 import * as Layer from "effect/Layer";
+import * as Path from "effect/Path";
 
 import { checkpointRefForScopeOrdinal } from "../orchestration-v2/CheckpointService.ts";
 import { OrchestratorProjectionError } from "../orchestration-v2/Orchestrator.ts";
 import type { ProjectionCheckpointContext } from "../orchestration-v2/ProjectionStore.ts";
 import * as ThreadManagement from "../orchestration-v2/ThreadManagementService.ts";
+import * as VcsDriverRegistry from "../vcs/VcsDriverRegistry.ts";
+import * as VcsProcess from "../vcs/VcsProcess.ts";
 import * as CheckpointDiffQuery from "./CheckpointDiffQuery.ts";
 import * as CheckpointStore from "./CheckpointStore.ts";
 import {
@@ -83,7 +88,7 @@ it.effect("computes V2 run diffs from projected checkpoint scopes", () => {
       toTurnCount: 2,
       diff: "diff --git a/file b/file",
     });
-    assert.deepEqual(diffCheckpoints.mock.calls[0]?.[0], {
+    assert.containSubset(diffCheckpoints.mock.calls[0]?.[0], {
       cwd: "/repo",
       fromCheckpointRef: yield* checkpointRefForScopeOrdinal({
         scopeId: firstScopeId,
@@ -310,3 +315,157 @@ it.effect("preserves the typed missing-baseline-ref error contract", () => {
     );
   }).pipe(Effect.provide(layer));
 });
+
+const layerVcsProcess = VcsProcess.layer.pipe(Layer.provide(NodeServices.layer));
+const layerStore = CheckpointStore.layer.pipe(
+  Layer.provide(VcsDriverRegistry.layer.pipe(Layer.provide(layerVcsProcess))),
+  Layer.provide(NodeServices.layer),
+);
+const layerTest = Layer.merge(layerStore, layerVcsProcess).pipe(
+  Layer.provideMerge(NodeServices.layer),
+  Layer.provideMerge(NodeCrypto.layer),
+);
+
+it.effect("shares the turn and full-thread diff budget across real Git batches", () =>
+  Effect.gen(function* () {
+    const fs = yield* FileSystem.FileSystem;
+    const path = yield* Path.Path;
+    const process = yield* VcsProcess.VcsProcess;
+    const store = yield* CheckpointStore.CheckpointStore;
+    const cwd = yield* fs.makeTempDirectoryScoped({ prefix: "t3-diff-aggregate-" });
+    const git = (args: ReadonlyArray<string>, env?: NodeJS.ProcessEnv) =>
+      process.run({
+        operation: "CheckpointDiffQuery.test.git",
+        command: "git",
+        cwd,
+        args,
+        ...(env === undefined ? {} : { env }),
+      });
+    yield* git(["init"]);
+    yield* git(["config", "user.email", "test@test.com"]);
+    yield* git(["config", "user.name", "Test"]);
+    yield* fs.writeFileString(path.join(cwd, "README.md"), "initial\n");
+    yield* git(["add", "."]);
+    yield* git(["commit", "-m", "initial"]);
+
+    const threadId = ThreadId.make("thread:aggregate-runtime");
+    const runId = RunId.make("run:aggregate-runtime");
+    const scopeId = CheckpointScopeId.make("scope:aggregate-runtime");
+    const fromCheckpointRef = yield* checkpointRefForScopeOrdinal({
+      scopeId,
+      ordinalWithinScope: 0,
+    });
+    const toCheckpointRef = CheckpointRef.make("refs/t3/test/aggregate-after");
+    yield* store.captureCheckpoint({ cwd, checkpointRef: fromCheckpointRef });
+    // The old commit models files imported by a pull after the baseline was captured.
+    yield* fs.writeFileString(path.join(cwd, "upstream.ts"), "imported\n");
+    yield* git(["add", "."]);
+    yield* git(["commit", "-m", "imported"], {
+      ...globalThis.process.env,
+      GIT_COMMITTER_DATE: "2000-01-01T00:00:00Z",
+    });
+    const ownPaths = Array.from({ length: 1_600 }, (_, index) => `src/module-${index}/file.ts`);
+    yield* Effect.forEach(
+      ownPaths,
+      (file) =>
+        Effect.gen(function* () {
+          yield* fs.makeDirectory(path.dirname(path.join(cwd, file)), { recursive: true });
+          yield* fs.writeFileString(path.join(cwd, file), `${"x".repeat(12_000)}\n`);
+        }),
+      { concurrency: 16, discard: true },
+    );
+    yield* store.captureCheckpoint({ cwd, checkpointRef: toCheckpointRef });
+
+    const patchSizes: number[] = [];
+    const query = yield* CheckpointDiffQuery.make.pipe(
+      Effect.provide(
+        Layer.mock(ThreadManagement.ThreadManagementService)({
+          getCheckpointContext: () =>
+            Effect.succeed({
+              runs: [{ id: runId, ordinal: 1, status: "completed" }],
+              checkpointScopes: [{ id: scopeId, runId, kind: "root_run", cwd }],
+              checkpoints: [
+                {
+                  scopeId,
+                  runId,
+                  ordinalWithinScope: 1,
+                  appRunOrdinal: 1,
+                  status: "ready",
+                  ref: toCheckpointRef,
+                },
+              ],
+            }),
+        }),
+      ),
+      Effect.provideService(CheckpointStore.CheckpointStore, {
+        ...store,
+        diffCheckpoints: (input) =>
+          store.diffCheckpoints(input).pipe(
+            Effect.tap((patch) =>
+              Effect.sync(() => {
+                if (input.format !== "numstat") {
+                  patchSizes.push(Buffer.byteLength(patch));
+                }
+              }),
+            ),
+          ),
+      }),
+    );
+    const turn = yield* query.getTurnDiff({ threadId, fromTurnCount: 0, toTurnCount: 1 });
+    const turnPatchSizes = patchSizes.splice(0);
+    const fullThread = yield* query.getFullThreadDiff({ threadId, toTurnCount: 1 });
+    assert.isTrue(fullThread.diff === turn.diff);
+    assert.isAtMost(Buffer.byteLength(turn.diff), 10_000_000);
+    assert.isTrue(turn.diff.endsWith("\n\n[truncated]"));
+    assert.include(turn.diff, "src/module-0/file.ts");
+    assert.notInclude(turn.diff, "upstream.ts");
+    assert.lengthOf(turnPatchSizes, 2);
+    assert.deepEqual(patchSizes, turnPatchSizes);
+    assert.isBelow(turnPatchSizes[0]!, 10_000_000);
+    assert.strictEqual(
+      turnPatchSizes.reduce((sum, size) => sum + size, 0),
+      10_000_001,
+    );
+  }).pipe(Effect.provide(layerTest), Effect.scoped),
+);
+
+it.effect("truncates a storage-cut UTF-8 character without replacement text", () => {
+  const prefix = "x".repeat(9_999_985);
+  const layer = layerFor({
+    projection: Effect.succeed(makeProjection()),
+    diffCheckpoints: (input) =>
+      Effect.succeed(
+        Buffer.from(`${prefix}😀${"x".repeat(9)}😀tail`)
+          .subarray(0, input.maxOutputBytes)
+          .toString("utf8"),
+      ),
+  });
+  return Effect.gen(function* () {
+    const query = yield* CheckpointDiffQuery.CheckpointDiffQuery;
+    const result = yield* query.getTurnDiff({ threadId, fromTurnCount: 0, toTurnCount: 2 });
+    assert.isAtMost(Buffer.byteLength(result.diff), 10_000_000);
+    assert.isTrue(result.diff === `${prefix}\n\n[truncated]`);
+  }).pipe(Effect.provide(layer));
+});
+
+it.effect("preserves an exact-fit filtered patch across batches without a marker", () => {
+  const ownPaths = Array.from({ length: 1_000 }, (_, index) => `src/module-${index}/file.ts`);
+  const patches = ["é".repeat(3_000_000), "x".repeat(4_000_000)];
+  let patchReads = 0;
+  const layer = layerFor({
+    projection: Effect.succeed(makeProjection()),
+    authoredPaths: new Set(ownPaths),
+    diffCheckpoints: (input) =>
+      Effect.sync(() =>
+        input.format === "numstat"
+          ? [...ownPaths, "upstream.ts"].map((path) => `1\t0\t${path}\0`).join("")
+          : patches[patchReads++]!,
+      ),
+  });
+  return Effect.gen(function* () {
+    const query = yield* CheckpointDiffQuery.CheckpointDiffQuery;
+    const result = yield* query.getFullThreadDiff({ threadId, toTurnCount: 2 });
+    assert.isTrue(result.diff === patches.join(""));
+    assert.strictEqual(patchReads, 2);
+  }).pipe(Effect.provide(layer));
+});

apps/server/src/checkpointing/CheckpointDiffQuery.ts

@@ -12,6 +12,7 @@ import * as Effect from "effect/Effect";
 import * as Layer from "effect/Layer";
 import * as Schema from "effect/Schema";
 
+import { CHECKPOINT_DIFF_MAX_OUTPUT_BYTES } from "../vcs/VcsDriver.ts";
 import { checkpointRefForScopeOrdinal } from "../orchestration-v2/CheckpointService.ts";
 import * as ThreadManagement from "../orchestration-v2/ThreadManagementService.ts";
 import {
@@ -27,6 +28,7 @@ import { isGitImport, parseTurnDiffFilesFromNumstat, type TurnDiffFileSummary }
 
 // Windows limits a command line to 32,767 characters. Leave room for the rest of git's arguments.
 const MAX_PATHSPEC_CHARS = 24_000;
+const DIFF_TRUNCATED_MARKER = "\n\n[truncated]";
 
 function batchFilePaths(files: ReadonlyArray<TurnDiffFileSummary>) {
   const batches: string[][] = [];
@@ -219,19 +221,36 @@ export const make = Effect.gen(function* () {
             );
       const retainedFiles = files.filter((file) => !isGitImport(file, authoredPaths));
       // Select retained paths before generating a patch, so imported bulk cannot exhaust its output limit.
-      const diff =
-        retainedFiles.length === files.length
-          ? yield* checkpointStore
-              .diffCheckpoints(comparison)
-              .pipe(Effect.withSpan("checkpoint.turnDiff.diffCheckpoints"))
-          : (yield* Effect.forEach(
-              batchFilePaths(retainedFiles),
-              (filePaths) =>
-                checkpointStore
-                  .diffCheckpoints({ ...comparison, filePaths })
-                  .pipe(Effect.withSpan("checkpoint.turnDiff.diffCheckpoints")),
-              { concurrency: 4 },
-            )).join("");
+      const pathBatches =
+        retainedFiles.length === files.length ? [undefined] : batchFilePaths(retainedFiles);
+      let diff = "";
+      let remainingBytes = CHECKPOINT_DIFF_MAX_OUTPUT_BYTES;
+      for (const filePaths of pathBatches) {
+        const patch = yield* checkpointStore
+          .diffCheckpoints({
+            ...comparison,
+            ...(filePaths === undefined ? {} : { filePaths }),
+            // One extra byte distinguishes an exact fit from a truncated patch.
+            maxOutputBytes: remainingBytes + 1,
+          })
+          .pipe(Effect.withSpan("checkpoint.turnDiff.diffCheckpoints"));
+        const patchBytes = Buffer.byteLength(patch);
+        diff += patch;
+        if (patchBytes > remainingBytes) {
+          // Reserve marker space only on overflow, which may trim an earlier batch.
+          // Streaming decode drops an incomplete final UTF-8 character at the byte boundary.
+          diff =
+            new TextDecoder().decode(
+              Buffer.from(diff).subarray(
+                0,
+                CHECKPOINT_DIFF_MAX_OUTPUT_BYTES - Buffer.byteLength(DIFF_TRUNCATED_MARKER),
+              ),
+              { stream: true },
+            ) + DIFF_TRUNCATED_MARKER;
+          break;
+        }
+        remainingBytes -= patchBytes;
+      }
 
       const turnDiff = buildTurnDiffResult(input, diff);
       if (!isTurnDiffResult(turnDiff)) {

apps/server/src/checkpointing/CheckpointStore.ts

@@ -25,6 +25,8 @@ export interface DiffCheckpointsInput {
   readonly fallbackFromToHead?: boolean;
   readonly ignoreWhitespace: boolean;
   readonly format?: "patch" | "numstat";
+  /** Maximum bytes collected from the diff command. */
+  readonly maxOutputBytes?: number;
   /** Limits the diff to these exact paths. An empty list yields an empty diff. */
   readonly filePaths?: ReadonlyArray<string>;
 }

apps/server/src/vcs/GitVcsDriver.ts

@@ -421,7 +421,6 @@ const WORKSPACE_FILES_MAX_OUTPUT_BYTES = 16 * 1024 * 1024;
 const CHECKPOINT_RECOVERY_MAX_CANDIDATES = 64;
 const CHECKPOINT_RECOVERY_TIMEOUT = "5 seconds";
 const GIT_CHECK_IGNORE_MAX_STDIN_BYTES = 256 * 1024;
-const CHECKPOINT_DIFF_MAX_OUTPUT_BYTES = 10_000_000;
 const WORKSPACE_GIT_HARDENED_CONFIG_ARGS = [
   "-c",
   "core.fsmonitor=false",
@@ -1238,7 +1237,7 @@ export const makeVcsDriverShape = Effect.fn("makeGitVcsDriverShape")(function* (
             : []),
         ],
         allowNonZeroExit: true,
-        maxOutputBytes: CHECKPOINT_DIFF_MAX_OUTPUT_BYTES,
+        maxOutputBytes: input.maxOutputBytes ?? VcsDriver.CHECKPOINT_DIFF_MAX_OUTPUT_BYTES,
         outputMode: input.format === "numstat" ? "error" : "truncate",
       });
 
@@ -1280,7 +1279,7 @@ export const makeVcsDriverShape = Effect.fn("makeGitVcsDriverShape")(function* (
           operation,
           cwd: input.cwd,
           args: [...args, "--name-only", "-z", "--no-renames", "--no-ext-diff"],
-          maxOutputBytes: CHECKPOINT_DIFF_MAX_OUTPUT_BYTES,
+          maxOutputBytes: VcsDriver.CHECKPOINT_DIFF_MAX_OUTPUT_BYTES,
           outputMode: "error",
         }).pipe(Effect.map((result) => result.stdout.split("\0")));
       // Remerge diff lists a merge's paths only where the result differs from Git's

apps/server/src/vcs/VcsDriver.ts

@@ -25,13 +25,17 @@ export interface VcsRestoreCheckpointInput {
   readonly fallbackToHead?: boolean;
 }
 
+export const CHECKPOINT_DIFF_MAX_OUTPUT_BYTES = 10_000_000;
+
 export interface VcsDiffCheckpointsInput {
   readonly cwd: string;
   readonly fromCheckpointRef: CheckpointRef;
   readonly toCheckpointRef: CheckpointRef;
   readonly fallbackFromToHead?: boolean;
   readonly ignoreWhitespace: boolean;
   readonly format?: "patch" | "numstat";
+  /** Maximum bytes collected from the diff command. */
+  readonly maxOutputBytes?: number;
   /** An empty list yields an empty diff. */
   readonly filePaths?: ReadonlyArray<string>;
 }