fix(server): cap checkpoint diffs across all batches

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

@@ -0,0 +1,122 @@
+import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
+import * as NodeServices from "@effect/platform-node/NodeServices";
+import { assert, it } 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 * 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";
+
+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("bounds a filtered query through real checkpoint storage and Git", () =>
+  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_000 }, (_, 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(24_000)}\n`);
+        }),
+      { concurrency: 16, discard: true },
+    );
+    yield* store.captureCheckpoint({ cwd, checkpointRef: toCheckpointRef });
+
+    let patchReads = 0;
+    let collectedBytes = 0;
+    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, 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") {
+                  patchReads += 1;
+                  collectedBytes += Buffer.byteLength(patch);
+                }
+              }),
+            ),
+          ),
+      }),
+    );
+    const result = yield* query.getFullThreadDiff({ threadId, toTurnCount: 1 });
+    yield* Effect.logInfo(
+      JSON.stringify({
+        samples: 1,
+        patchReads,
+        collectedBytes,
+        responseBytes: Buffer.byteLength(result.diff),
+      }),
+    );
+    assert.isAtMost(Buffer.byteLength(result.diff), 10_000_000);
+    assert.isTrue(result.diff.endsWith("\n\n[truncated]"));
+    assert.notInclude(result.diff, "upstream.ts");
+    assert.strictEqual(patchReads, 1);
+    assert.isAtMost(collectedBytes, 9_999_988);
+  }).pipe(Effect.provide(layerTest), Effect.scoped),
+);

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

@@ -91,6 +91,7 @@ it.effect("computes V2 run diffs from projected checkpoint scopes", () => {
       toCheckpointRef: secondRef,
       fallbackFromToHead: false,
       ignoreWhitespace: true,
+      maxOutputBytes: 9_999_988,
     });
   }).pipe(Effect.provide(layer));
 });
@@ -112,7 +113,10 @@ it.effect("leaves Git imports out of large turn diffs", () => {
 
   return Effect.gen(function* () {
     const query = yield* CheckpointDiffQuery.CheckpointDiffQuery;
-    yield* query.getFullThreadDiff({ threadId, toTurnCount: 2 });
+    const result = yield* query.getFullThreadDiff({ threadId, toTurnCount: 2 });
+    for (const path of ownPaths) assert.include(result.diff, path);
+    assert.notInclude(result.diff, "upstream.ts");
+    assert.notInclude(result.diff, "[truncated]");
 
     const patchPaths = diffCheckpoints.mock.calls
       .map(([call]) => call)
@@ -123,6 +127,113 @@ it.effect("leaves Git imports out of large turn diffs", () => {
   }).pipe(Effect.provide(layer));
 });
 
+it.effect.each(["x".repeat(10_000_000), "é".repeat(3_000_000)])(
+  "caps filtered diffs across batches (case %#)",
+  (firstPatch) => {
+    const ownPaths = Array.from({ length: 2_000 }, (_, index) => `src/module-${index}/file.ts`);
+    const allowances: number[] = [];
+    const diffCheckpoints = (input: CheckpointStore.DiffCheckpointsInput) => {
+      if (input.format === "numstat") {
+        return Effect.succeed(
+          [...ownPaths, "upstream.ts"].map((path) => `1\t0\t${path}\0`).join(""),
+        );
+      }
+      allowances.push(input.maxOutputBytes ?? 10_000_000);
+      const patch = allowances.length === 1 ? firstPatch : "x".repeat(10_000_000);
+      return Effect.succeed(patch);
+    };
+    const layer = layerFor({
+      projection: Effect.succeed(makeProjection()),
+      authoredPaths: new Set(ownPaths),
+      diffCheckpoints,
+    });
+    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.endsWith("\n\n[truncated]"));
+      assert.deepEqual(
+        allowances,
+        Buffer.byteLength(firstPatch) >= 9_999_987 ? [9_999_988] : [9_999_988, 3_999_988],
+      );
+    }).pipe(Effect.provide(layer));
+  },
+);
+
+it.effect("caps unfiltered diffs without splitting a UTF-8 character", () => {
+  const layer = layerFor({
+    projection: Effect.succeed(makeProjection()),
+    diffCheckpoints: () => Effect.succeed("😀".repeat(2_500_000)),
+  });
+  return Effect.gen(function* () {
+    const query = yield* CheckpointDiffQuery.CheckpointDiffQuery;
+    const result = yield* query.getFullThreadDiff({ threadId, toTurnCount: 2 });
+    assert.isAtMost(Buffer.byteLength(result.diff), 10_000_000);
+    assert.isTrue(result.diff.endsWith("\n\n[truncated]"));
+    assert.notInclude(result.diff, "�");
+  }).pipe(Effect.provide(layer));
+});
+
+it.effect.each([9_999_984, 9_999_985])(
+  "drops a UTF-8 character cut by storage at byte %i",
+  (prefixBytes) => {
+    const layer = layerFor({
+      projection: Effect.succeed(makeProjection()),
+      diffCheckpoints: (input) =>
+        Effect.succeed(
+          Buffer.from(`${"x".repeat(prefixBytes)}😀`)
+            .subarray(0, input.maxOutputBytes)
+            .toString("utf8"),
+        ),
+    });
+    return Effect.gen(function* () {
+      const query = yield* CheckpointDiffQuery.CheckpointDiffQuery;
+      const result = yield* query.getFullThreadDiff({ threadId, toTurnCount: 2 });
+      assert.isAtMost(Buffer.byteLength(result.diff), 10_000_000);
+      assert.isTrue(result.diff.endsWith("\n\n[truncated]"));
+      assert.notInclude(result.diff, "�");
+    }).pipe(Effect.provide(layer));
+  },
+);
+
+it.effect("does not mark an exact-fit complete patch as truncated", () => {
+  const layer = layerFor({
+    projection: Effect.succeed(makeProjection()),
+    diffCheckpoints: () => Effect.succeed("x".repeat(9_999_987)),
+  });
+  return Effect.gen(function* () {
+    const query = yield* CheckpointDiffQuery.CheckpointDiffQuery;
+    const result = yield* query.getFullThreadDiff({ threadId, toTurnCount: 2 });
+    assert.strictEqual(Buffer.byteLength(result.diff), 9_999_987);
+    assert.notInclude(result.diff, "[truncated]");
+  }).pipe(Effect.provide(layer));
+});
+
+it.effect("stops before the next batch when a patch exactly exhausts the budget", () => {
+  const ownPaths = Array.from({ length: 1_000 }, (_, index) => `src/module-${index}/file.ts`);
+  let patchReads = 0;
+  const layer = layerFor({
+    projection: Effect.succeed(makeProjection()),
+    authoredPaths: new Set(ownPaths),
+    diffCheckpoints: (input) => {
+      if (input.format === "numstat") {
+        return Effect.succeed(
+          [...ownPaths, "upstream.ts"].map((path) => `1\t0\t${path}\0`).join(""),
+        );
+      }
+      patchReads += 1;
+      return Effect.succeed("x".repeat(9_999_987));
+    },
+  });
+  return Effect.gen(function* () {
+    const query = yield* CheckpointDiffQuery.CheckpointDiffQuery;
+    const result = yield* query.getFullThreadDiff({ threadId, toTurnCount: 2 });
+    assert.strictEqual(Buffer.byteLength(result.diff), 10_000_000);
+    assert.isTrue(result.diff.endsWith("\n\n[truncated]"));
+    assert.strictEqual(patchReads, 1);
+  }).pipe(Effect.provide(layer));
+});
+
 it.effect("preserves the typed missing-thread error contract", () => {
   const layer = layerFor({
     projection: Effect.fail(new OrchestratorProjectionError({ threadId })),

apps/server/src/checkpointing/CheckpointDiffQuery.ts

@@ -20,6 +20,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 {
@@ -35,6 +36,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]";
 
 /** Splits files into pathspec lists that each fit on one git command line. Renames keep both paths together. */
 function batchFilePaths(files: ReadonlyArray<TurnDiffFileSummary>) {
@@ -228,19 +230,40 @@ 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);
+      const patches: string[] = [];
+      // Reserve the marker once for the entire query, including an unfiltered diff.
+      let remainingBytes =
+        CHECKPOINT_DIFF_MAX_OUTPUT_BYTES - Buffer.byteLength(DIFF_TRUNCATED_MARKER);
+      for (const filePaths of pathBatches) {
+        if (remainingBytes === 0) {
+          patches.push(DIFF_TRUNCATED_MARKER);
+          break;
+        }
+        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);
+        if (patchBytes > remainingBytes) {
+          // Streaming decode drops an incomplete final UTF-8 character at the byte boundary.
+          patches.push(
+            new TextDecoder().decode(Buffer.from(patch).subarray(0, remainingBytes), {
+              stream: true,
+            }),
+            DIFF_TRUNCATED_MARKER,
+          );
+          break;
+        }
+        patches.push(patch);
+        remainingBytes -= patchBytes;
+      }
+      const diff = patches.join("");
 
       const turnDiff = buildTurnDiffResult(input, diff);
       if (!isTurnDiffResult(turnDiff)) {

apps/server/src/checkpointing/CheckpointStore.ts

@@ -40,6 +40,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

@@ -424,7 +424,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",
@@ -1217,7 +1216,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",
       });
 
@@ -1260,7 +1259,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;
   /** Limits the diff to these exact paths. An empty list yields an empty diff. */
   readonly filePaths?: ReadonlyArray<string>;
 }