fix(server): stage only changed checkpoint paths

apps/server/src/processRunner.test.ts

@@ -246,6 +246,24 @@ describe("runProcess", () => {
     }),
   );
 
+  it.effect("preserves raw stdout bytes when requested", () =>
+    Effect.gen(function* () {
+      const stdoutBytes = new Uint8Array([0x66, 0x80, 0x6f]);
+      const spawner = makeSpawner(() =>
+        Effect.succeed(makeHandle({ stdout: Stream.make(stdoutBytes) })),
+      );
+
+      const result = yield* runWith(spawner)({
+        command: "fake",
+        args: ["raw-stdout"],
+        captureStdoutBytes: true,
+      });
+
+      expect(result.stdoutInvalidUtf8).toBe(true);
+      expect(Array.from(result.stdoutBytes ?? [])).toEqual(Array.from(stdoutBytes));
+    }),
+  );
+
   it.effect("fails fast on output limit before timeout for long-running output", () =>
     Effect.gen(function* () {
       const textChunk = "x".repeat(64);
@@ -319,6 +337,32 @@ describe("runProcess", () => {
     }),
   );
 
+  it.effect("writes raw stdin bytes without UTF-8 encoding", () =>
+    Effect.gen(function* () {
+      const stdinBytes = new Uint8Array([0x66, 0x80, 0x6f]);
+      const receivedBytes: number[] = [];
+      const spawner = makeSpawner(() =>
+        Effect.succeed(
+          makeHandle({
+            stdin: Sink.forEach((chunk: Uint8Array) =>
+              Effect.sync(() => {
+                receivedBytes.push(...chunk);
+              }),
+            ),
+          }),
+        ),
+      );
+
+      yield* runWith(spawner)({
+        command: "fake",
+        args: ["raw-stdin"],
+        stdin: stdinBytes,
+      });
+
+      expect(receivedBytes).toEqual(Array.from(stdinBytes));
+    }),
+  );
+
   it.effect("returns output for non-zero exit codes", () =>
     Effect.gen(function* () {
       const spawner = makeSpawner(() => Effect.succeed(makeHandle({ stderr: "boom", code: 2 })));

apps/server/src/processRunner.ts

@@ -24,7 +24,8 @@ export interface ProcessRunInput {
   readonly spawnCwd?: string | undefined;
   readonly timeout?: Duration.Input | undefined;
   readonly env?: NodeJS.ProcessEnv | undefined;
-  readonly stdin?: string | undefined;
+  readonly stdin?: string | Uint8Array | undefined;
+  readonly captureStdoutBytes?: boolean | undefined;
   readonly maxOutputBytes?: number | undefined;
   readonly outputMode?: "error" | "truncate" | undefined;
   readonly truncatedMarker?: string | undefined;
@@ -44,6 +45,7 @@ export interface ProcessRunOutput {
   readonly stderrTruncated: boolean;
   readonly stdoutInvalidUtf8: boolean;
   readonly stderrInvalidUtf8: boolean;
+  readonly stdoutBytes?: Uint8Array;
 }
 
 const ProcessInvocationFields = {
@@ -182,6 +184,7 @@ const collectText = Effect.fn("processRunner.collectText")(function* (input: {
   readonly maxOutputBytes: number;
   readonly outputMode: "error" | "truncate";
   readonly truncatedMarker: string;
+  readonly preserveBytes: boolean;
 }) {
   const stream = input.stream.pipe(
     Stream.mapError(
@@ -202,6 +205,7 @@ const collectText = Effect.fn("processRunner.collectText")(function* (input: {
       stream,
       maxBytes: input.maxOutputBytes,
       truncatedMarker: input.truncatedMarker,
+      preserveBytes: input.preserveBytes,
     });
   }
 
@@ -239,13 +243,15 @@ const collectText = Effect.fn("processRunner.collectText")(function* (input: {
         });
       },
     ),
-    Effect.map(
-      (state): CollectedUint8StreamText => ({
-        ...decodeUtf8(Buffer.concat(state.chunks, state.bytes)),
+    Effect.map((state): CollectedUint8StreamText => {
+      const rawBytes = Buffer.concat(state.chunks, state.bytes);
+      return {
+        ...decodeUtf8(rawBytes),
         bytes: state.bytes,
         truncated: false,
-      }),
-    ),
+        ...(input.preserveBytes ? { rawBytes } : {}),
+      };
+    }),
   );
 });
 
@@ -335,15 +341,18 @@ const runProcessCore = Effect.fn("processRunner.runProcessCore")(function* (
   const writeStdin =
     stdin === undefined
       ? Effect.void
-      : Stream.run(Stream.encodeText(Stream.make(stdin)), child.stdin).pipe(
+      : Stream.run(
+          typeof stdin === "string" ? Stream.encodeText(Stream.make(stdin)) : Stream.make(stdin),
+          child.stdin,
+        ).pipe(
           Effect.mapError(
             (cause) =>
               new ProcessStdinError({
                 command: input.command,
                 argumentCount: input.args.length,
                 cwd: input.cwd,
                 spawnCwd: input.spawnCwd,
-                stdinBytes: Buffer.byteLength(stdin),
+                stdinBytes: typeof stdin === "string" ? Buffer.byteLength(stdin) : stdin.byteLength,
                 cause,
               }),
           ),
@@ -361,6 +370,7 @@ const runProcessCore = Effect.fn("processRunner.runProcessCore")(function* (
         maxOutputBytes,
         outputMode,
         truncatedMarker,
+        preserveBytes: input.captureStdoutBytes === true,
       }),
       collectText({
         command: input.command,
@@ -372,6 +382,7 @@ const runProcessCore = Effect.fn("processRunner.runProcessCore")(function* (
         maxOutputBytes,
         outputMode,
         truncatedMarker,
+        preserveBytes: false,
       }),
       writeStdin,
     ],
@@ -401,6 +412,7 @@ const runProcessCore = Effect.fn("processRunner.runProcessCore")(function* (
     stderrTruncated: stderr.truncated,
     stdoutInvalidUtf8: stdout.invalidUtf8,
     stderrInvalidUtf8: stderr.invalidUtf8,
+    ...(stdout.rawBytes !== undefined ? { stdoutBytes: stdout.rawBytes } : {}),
   } satisfies ProcessRunOutput;
 });
 

apps/server/src/stream/collectUint8StreamText.ts

@@ -7,6 +7,7 @@ export interface CollectedUint8StreamText {
   readonly truncated: boolean;
   readonly bytes: number;
   readonly invalidUtf8: boolean;
+  readonly rawBytes?: Uint8Array;
 }
 
 export const decodeUtf8 = (
@@ -26,6 +27,7 @@ export const collectUint8StreamText = <E>(input: {
   readonly stream: Stream.Stream<Uint8Array, E>;
   readonly maxBytes?: number | undefined;
   readonly truncatedMarker?: string | null | undefined;
+  readonly preserveBytes?: boolean | undefined;
 }): Effect.Effect<CollectedUint8StreamText, E> => {
   const maxBytes = input.maxBytes ?? Number.POSITIVE_INFINITY;
   const truncatedMarker = input.truncatedMarker ?? "";
@@ -68,7 +70,8 @@ export const collectUint8StreamText = <E>(input: {
       },
     ),
     Effect.map((state): CollectedUint8StreamText => {
-      const decoded = decodeUtf8(Buffer.concat(state.chunks, state.bytes));
+      const rawBytes = Buffer.concat(state.chunks, state.bytes);
+      const decoded = decodeUtf8(rawBytes);
       return {
         text:
           state.truncated && truncatedMarker.length > 0
@@ -77,6 +80,7 @@ export const collectUint8StreamText = <E>(input: {
         bytes: state.bytes,
         truncated: state.truncated,
         invalidUtf8: decoded.invalidUtf8,
+        ...(input.preserveBytes ? { rawBytes } : {}),
       };
     }),
   );

apps/server/src/vcs/GitVcsDriver.test.ts

@@ -7,7 +7,7 @@ import * as PlatformError from "effect/PlatformError";
 import { ChildProcessSpawner } from "effect/unstable/process";
 import { assert, it } from "@effect/vitest";
 
-import { GitCommandError } from "@t3tools/contracts";
+import { CheckpointRef, GitCommandError } from "@t3tools/contracts";
 import * as ServerConfig from "../config.ts";
 import * as GitVcsDriver from "./GitVcsDriver.ts";
 import * as VcsProcess from "./VcsProcess.ts";
@@ -108,3 +108,163 @@ it.effect("GitVcsDriver forwards execute env to the VCS process", () => {
     ),
   );
 });
+
+it.effect("GitVcsDriver stages only changed paths when capturing a checkpoint", () => {
+  const commands: Array<VcsProcess.VcsProcessInput> = [];
+  const changedPaths = ["changed.ts", "deleted.ts", "dir/new\nfile.ts", ":literal.ts"];
+  const invalidUtf8Path = new Uint8Array([0xff, ...new TextEncoder().encode(".txt")]);
+  const statusBytes = Buffer.concat([
+    Buffer.from(
+      ` M ${changedPaths[0]}\0 D ${changedPaths[1]}\0?? ${changedPaths[2]}\0?? ${changedPaths[3]}\0?? `,
+    ),
+    invalidUtf8Path,
+    Buffer.from([0]),
+  ]);
+  const expectedPathspec = Buffer.concat([
+    Buffer.from(`${changedPaths.join("\0")}\0`),
+    invalidUtf8Path,
+    Buffer.from([0]),
+  ]);
+
+  return Effect.gen(function* () {
+    const driver = yield* GitVcsDriver.makeVcsDriverShape();
+    assert.isDefined(driver.checkpoints);
+
+    yield* driver.checkpoints.captureCheckpoint({
+      cwd: process.cwd(),
+      checkpointRef: CheckpointRef.make("refs/t3/checkpoints/test/turn/1"),
+    });
+
+    const statusCommand = commands.find((command) => command.args[2] === "status");
+    assert.isDefined(statusCommand);
+    assert.deepStrictEqual(statusCommand.args.slice(2), [
+      "status",
+      "--porcelain=v1",
+      "-z",
+      "--untracked-files=all",
+      "--no-renames",
+      "--",
+      ".",
+    ]);
+    assert.strictEqual(statusCommand.env?.GIT_OPTIONAL_LOCKS, "0");
+    assert.strictEqual(statusCommand.captureStdoutBytes, true);
+
+    const addCommand = commands.find((command) => command.args[3] === "add");
+    assert.isDefined(addCommand);
+    assert.deepStrictEqual(addCommand.args.slice(2), [
+      "--literal-pathspecs",
+      "add",
+      "-A",
+      "--pathspec-from-file=-",
+      "--pathspec-file-nul",
+    ]);
+    assert.instanceOf(addCommand.stdin, Uint8Array);
+    assert.deepStrictEqual(addCommand.stdin, expectedPathspec);
+  }).pipe(
+    Effect.provide(
+      Layer.mergeAll(
+        NodeServices.layer,
+        Layer.mock(VcsProcess.VcsProcess)({
+          run: (input) =>
+            Effect.sync(() => {
+              commands.push(input);
+              const args = input.args.slice(2);
+              const stdout =
+                args[0] === "status"
+                  ? new TextDecoder().decode(statusBytes)
+                  : args[0] === "rev-parse" && args[1] === "--git-common-dir"
+                    ? ".git\n"
+                    : args[0] === "rev-parse" && args[1] === "--show-toplevel"
+                      ? `${process.cwd()}\n`
+                      : args[0] === "write-tree"
+                        ? "tree-oid\n"
+                        : args[0] === "commit-tree"
+                          ? "commit-oid\n"
+                          : "";
+              return {
+                exitCode: ChildProcessSpawner.ExitCode(0),
+                stdout,
+                stderr: "",
+                stdoutTruncated: false,
+                stderrTruncated: false,
+                ...(args[0] === "status" ? { stdoutBytes: statusBytes } : {}),
+              };
+            }),
+        }),
+      ),
+    ),
+  );
+});
+
+it.effect("GitVcsDriver captures scoped working tree changes without changing the real index", () =>
+  Effect.scoped(
+    Effect.gen(function* () {
+      const fileSystem = yield* FileSystem.FileSystem;
+      const path = yield* Path.Path;
+      const driver = yield* GitVcsDriver.makeVcsDriverShape();
+      const cwd = yield* fileSystem.makeTempDirectoryScoped({
+        prefix: "t3-git-checkpoint-paths-",
+      });
+      const scopeCwd = path.join(cwd, "scope");
+      const checkpointRef = CheckpointRef.make("refs/t3/checkpoints/test/turn/1");
+      assert.isDefined(driver.checkpoints);
+
+      const git = (gitCwd: string, args: ReadonlyArray<string>, allowNonZeroExit = false) =>
+        driver.execute({
+          operation: "GitVcsDriver.test.checkpointPaths",
+          cwd: gitCwd,
+          args,
+          allowNonZeroExit,
+          timeoutMs: 10_000,
+        });
+      const write = (relativePath: string, contents: string) =>
+        fileSystem.writeFileString(path.join(cwd, relativePath), contents);
+
+      yield* git(cwd, ["init"]);
+      yield* git(cwd, ["config", "user.email", "test@test.com"]);
+      yield* git(cwd, ["config", "user.name", "Test"]);
+      yield* fileSystem.makeDirectory(scopeCwd);
+      yield* write("scope/changed.txt", "original changed\n");
+      yield* write("scope/deleted.txt", "original deleted\n");
+      yield* write("scope/staged.txt", "original staged\n");
+      yield* write("outside.txt", "original outside\n");
+      yield* git(cwd, ["add", "."]);
+      yield* git(cwd, ["commit", "-m", "initial"]);
+
+      yield* write("scope/staged.txt", "staged version\n");
+      yield* git(cwd, ["add", "scope/staged.txt"]);
+      const indexTreeBefore = (yield* git(cwd, ["write-tree"])).stdout.trim();
+
+      yield* write("scope/changed.txt", "working changed\n");
+      yield* write("scope/staged.txt", "working version\n");
+      yield* write("scope/new file [1].txt", "new file\n");
+      yield* write("outside.txt", "working outside\n");
+      yield* fileSystem.remove(path.join(scopeCwd, "deleted.txt"));
+
+      yield* driver.checkpoints.captureCheckpoint({ cwd: scopeCwd, checkpointRef });
+
+      const indexTreeAfter = (yield* git(cwd, ["write-tree"])).stdout.trim();
+      assert.strictEqual(indexTreeAfter, indexTreeBefore);
+      assert.strictEqual(
+        (yield* git(cwd, ["show", `${checkpointRef}:scope/changed.txt`])).stdout,
+        "working changed\n",
+      );
+      assert.strictEqual(
+        (yield* git(cwd, ["show", `${checkpointRef}:scope/staged.txt`])).stdout,
+        "working version\n",
+      );
+      assert.strictEqual(
+        (yield* git(cwd, ["show", `${checkpointRef}:scope/new file [1].txt`])).stdout,
+        "new file\n",
+      );
+      assert.strictEqual(
+        (yield* git(cwd, ["show", `${checkpointRef}:outside.txt`])).stdout,
+        "original outside\n",
+      );
+      assert.notStrictEqual(
+        (yield* git(cwd, ["cat-file", "-e", `${checkpointRef}:scope/deleted.txt`], true)).exitCode,
+        0,
+      );
+    }).pipe(Effect.provide(GitContractLayer)),
+  ),
+);