fix(diff): preserve revision ordering and readable files

2026-11-06 16:34:36 +00:00 · 2026-11-06 06:50:25 +01:00
parent 870180cbe5
commit 8e1e377120
21 changed files with 467 additions and 282 deletions
@@ -165,3 +165,31 @@ test("moving a tag refreshes binary file metadata even though binary files have
    await subscription.unsubscribe();
  });
 }, 30000);
+
+test("legacy diffs retain successful bodies when an oversized sibling is removed", async () => {
+  await withCheckout(async ({ cwd, manager }) => {
+    await writeFile(join(cwd, "a.ts"), "const value = 1;\n");
+    await writeFile(join(cwd, "b.txt"), "x".repeat(1024 * 1024 + 1));
+    const updates: CheckoutDiffSnapshotPayload[] = [];
+    const subscription = await manager.subscribe(
+      { cwd, compare: { mode: "uncommitted" } },
+      (snapshot) => updates.push(snapshot),
+    );
+    expect(subscription.initial.files).toMatchObject([
+      { path: "a.ts", status: "ok" },
+      { path: "b.txt", status: "too_large" },
+    ]);
+    await rm(join(cwd, "b.txt"));
+    manager.scheduleRefreshForCwd(cwd);
+    await expect.poll(() => updates.at(-1)?.error).toBeNull();
+    expect(updates.at(-1)?.files[0]).toMatchObject({
+      path: "a.ts",
+      hunks: [
+        {
+          lines: expect.arrayContaining([expect.objectContaining({ content: "const value = 1;" })]),
+        },
+      ],
+    });
+    subscription.unsubscribe();
+  });
+});
@@ -89,8 +89,8 @@ export class CheckoutDiffManager {
        return;
      }
      let payload: CheckoutDiffSnapshotPayload;
-      if (frame.error || frame.diffTooLarge) {
-        const message = frame.error ? frame.error.message : "Diff too large to display";
+      if (frame.error) {
+        const message = frame.error.message;
        payload = {
          cwd,
          files: [],
@@ -100,14 +100,17 @@ export class CheckoutDiffManager {
        files.clear();
      } else {
        const next = new Map<string, ParsedDiffFile>();
+        const omitted = new Set(frame.omitted);
        for (const entry of pending.entries) {
          const previous = pending.baseRevision ? files.get(entry.path) : undefined;
          const body = bodies.get(entry.path) ?? null;
-          const file = materializeDiffFile(entry, body, false);
-          if (!entry.body && previous && entry.content === "text") file.hunks = previous.hunks;
+          const file = materializeDiffFile(entry, body, omitted.has(entry.path));
+          if (!entry.body && previous && entry.content === "text" && !omitted.has(entry.path))
+            file.hunks = previous.hunks;
          next.set(entry.path, file);
        }
        files = next;
+        // Stream omissions remain file placeholders; the JSON budget rejects a legacy snapshot.
        payload = legacySnapshot(cwd, [...next.values()]);
      }
      pending = null;
@@ -178,13 +181,17 @@ function legacySnapshot(cwd: string, files: ParsedDiffFile[]): CheckoutDiffSnaps
  for (let index = 0; index < files.length; index++) {
    bytes += Buffer.byteLength(JSON.stringify(files[index])) + Number(index > 0);
    if (bytes > CHECKOUT_DIFF_MAX_STRUCTURED_BYTES) {
-      return {
-        cwd,
-        files: [],
-        diffTooLarge: true,
-        error: toCheckoutError(new Error("Diff too large to display")),
-      };
+      return legacyTooLarge(cwd);
    }
  }
  return { cwd, files, error: null };
 }
+
+function legacyTooLarge(cwd: string): CheckoutDiffSnapshotPayload {
+  return {
+    cwd,
+    files: [],
+    diffTooLarge: true,
+    error: toCheckoutError(new Error("Diff too large to display")),
+  };
+}
@@ -295,11 +295,9 @@ test("large source diffs stay renderable for reads, subscriptions, and commit fi
 }, 60_000);

 test("streams whole-file batches and answers a same-socket ping before the revision ends", async () => {
-  const { WebSocket } = await import("ws");
-  const { decodeDiffFrame, DiffOpcode } = await import("@getpaseo/protocol/binary-frames/index");
  const ctx = await createDaemonTestContext();
  const cwd = mkdtempSync(join(tmpdir(), "diff-stream-wire-"));
-  let socket: InstanceType<typeof WebSocket> | null = null;
+  let probe: Awaited<ReturnType<typeof openDiffWireProbe>> | null = null;
  try {
    git(cwd, "init", "-b", "main");
    git(cwd, "config", "user.name", "Test");
@@ -310,7 +308,7 @@ test("streams whole-file batches and answers a same-socket ping before the revis
    for (let file = 0; file < 252; file++) {
      const content =
        Array.from(
-          { length: 90 },
+          { length: 80 },
          (_, line) =>
            `export const fn${line} = (param: string) => ({ key: param, value: ${file} });`,
        ).join("\n") + "\n";
@@ -322,82 +320,8 @@ test("streams whole-file batches and answers a same-socket ping before the revis
    const legacy = await ctx.client.getCheckoutDiff(cwd, { mode: "uncommitted" });
    const legacyJsonBytes = Buffer.byteLength(JSON.stringify(legacy.files));
    expect(legacyJsonBytes).toBeGreaterThanOrEqual(10 * 1024 * 1024);
-    const activeSocket = new WebSocket(`ws://127.0.0.1:${ctx.daemon.port}/ws`);
-    socket = activeSocket;
-    await new Promise<void>((resolve, reject) => {
-      activeSocket.once("open", resolve);
-      activeSocket.once("error", reject);
-    });
-    let encodedBytes = 0;
-    let endReceived = false;
-    let pingSent = false;
-    let pongDuringRevision = false;
-    let bodyFrames = 0;
-    let entryCount = 0;
-    const done = new Promise<void>((resolve, reject) => {
-      activeSocket.on("message", (data, binary) => {
-        try {
-          if (binary) {
-            if (!Buffer.isBuffer(data)) throw new Error("Expected buffer");
-            encodedBytes += data.byteLength;
-            const frame = decodeDiffFrame(new Uint8Array(data));
-            if (frame.opcode === DiffOpcode.Begin) entryCount = frame.entries.length;
-            if (frame.opcode === DiffOpcode.Bodies) {
-              bodyFrames++;
-              if (!pingSent) {
-                pingSent = true;
-                activeSocket.send(
-                  JSON.stringify({
-                    type: "session",
-                    message: { type: "ping", requestId: "mid-revision", clientSentAt: 1 },
-                  }),
-                );
-              }
-            }
-            if (frame.opcode === DiffOpcode.End) {
-              expect(frame.error).toBeNull();
-              endReceived = true;
-              resolve();
-            }
-          } else {
-            const envelope = JSON.parse(data.toString());
-            if (
-              envelope.message?.type === "pong" &&
-              envelope.message.payload.requestId === "mid-revision"
-            )
-              pongDuringRevision = !endReceived;
-            if (envelope.message?.payload?.status === "server_info")
-              activeSocket.send(
-                JSON.stringify({
-                  type: "session",
-                  message: {
-                    type: "diff.subscribe.request",
-                    requestId: "stream-diff",
-                    cwd,
-                    base: { kind: "rev", rev: "HEAD" },
-                    head: { kind: "worktree" },
-                  },
-                }),
-              );
-            if (envelope.message?.type === "diff.subscribe.response")
-              expect(envelope.message.payload.error).toBeNull();
-          }
-        } catch (error) {
-          reject(error);
-        }
-      });
-      activeSocket.once("error", reject);
-    });
-    activeSocket.send(
-      JSON.stringify({
-        type: "hello",
-        clientId: "diff-wire-test",
-        clientType: "browser",
-        protocolVersion: 1,
-        capabilities: { owned_subscriptions: true },
-      }),
-    );
-    await done;
+    probe = await openDiffWireProbe(ctx.daemon.port, cwd);
+    const { encodedBytes, bodyFrames, pongDuringRevision, entryCount } = await probe.done;
    expect(entryCount).toBe(252);
    expect(bodyFrames).toBeGreaterThan(1);
    expect(pongDuringRevision).toBe(true);
@@ -411,12 +335,98 @@ test("streams whole-file batches and answers a same-socket ping before the revis
      JSON.stringify({ legacyJsonBytes, encodedBytes, bodyFrames, pongDuringRevision }, null, 2),
    );
  } finally {
-    socket?.terminate();
+    probe?.socket.terminate();
    await ctx.cleanup();
    rmSync(cwd, { recursive: true, force: true });
  }
 }, 60_000);

+interface DiffWireResult {
+  encodedBytes: number;
+  bodyFrames: number;
+  pongDuringRevision: boolean;
+  entryCount: number;
+}
+async function openDiffWireProbe(port: number, cwd: string) {
+  const { WebSocket } = await import("ws");
+  const { decodeDiffFrame, DiffOpcode } = await import("@getpaseo/protocol/binary-frames/index");
+  const activeSocket = new WebSocket(`ws://127.0.0.1:${port}/ws`);
+  await new Promise<void>((resolve, reject) => {
+    activeSocket.once("open", resolve);
+    activeSocket.once("error", reject);
+  });
+  let encodedBytes = 0;
+  let endReceived = false;
+  let pingSent = false;
+  let pongDuringRevision = false;
+  let bodyFrames = 0;
+  let entryCount = 0;
+  const done = new Promise<DiffWireResult>((resolve, reject) => {
+    activeSocket.on("message", (data, binary) => {
+      try {
+        if (binary) {
+          if (!Buffer.isBuffer(data)) throw new Error("Expected buffer");
+          encodedBytes += data.byteLength;
+          const frame = decodeDiffFrame(new Uint8Array(data));
+          if (frame.opcode === DiffOpcode.Begin) entryCount = frame.entries.length;
+          if (frame.opcode === DiffOpcode.Bodies) {
+            bodyFrames++;
+            if (!pingSent) {
+              pingSent = true;
+              activeSocket.send(
+                JSON.stringify({
+                  type: "session",
+                  message: { type: "ping", requestId: "mid-revision", clientSentAt: 1 },
+                }),
+              );
+            }
+          }
+          if (frame.opcode === DiffOpcode.End) {
+            expect(frame.error).toBeNull();
+            endReceived = true;
+            resolve({ encodedBytes, bodyFrames, pongDuringRevision, entryCount });
+          }
+        } else {
+          const envelope = JSON.parse(data.toString());
+          if (
+            envelope.message?.type === "pong" &&
+            envelope.message.payload.requestId === "mid-revision"
+          )
+            pongDuringRevision = !endReceived;
+          if (envelope.message?.payload?.status === "server_info")
+            activeSocket.send(
+              JSON.stringify({
+                type: "session",
+                message: {
+                  type: "diff.subscribe.request",
+                  requestId: "stream-diff",
+                  cwd,
+                  base: { kind: "rev", rev: "HEAD" },
+                  head: { kind: "worktree" },
+                },
+              }),
+            );
+          if (envelope.message?.type === "diff.subscribe.response")
+            expect(envelope.message.payload.error).toBeNull();
+        }
+      } catch (error) {
+        reject(error);
+      }
+    });
+    activeSocket.once("error", reject);
+  });
+  activeSocket.send(
+    JSON.stringify({
+      type: "hello",
+      clientId: "diff-wire-test",
+      clientType: "browser",
+      protocolVersion: 1,
+      capabilities: { owned_subscriptions: true },
+    }),
+  );
+  return { socket: activeSocket, done };
+}
+
 test("SDK diff journey publishes a tag tree, commits a one-file refresh and shares two sockets", async () => {
  const ctx = await createDaemonTestContext();
  const cwd = mkdtempSync(join(tmpdir(), "sdk-stream-diff-"));
@@ -486,3 +496,28 @@ test("SDK diff journey publishes a tag tree, commits a one-file refresh and shar
    rmSync(cwd, { recursive: true, force: true });
  }
 }, 45000);
+
+test("commit-file RPC keeps the existing read ceiling for patches larger than one MiB", async () => {
+  const ctx = await createDaemonTestContext();
+  const cwd = mkdtempSync(join(tmpdir(), "commit-file-budget-"));
+  try {
+    git(cwd, "init", "-b", "main");
+    git(cwd, "config", "user.name", "Test");
+    git(cwd, "config", "user.email", "test@example.com");
+    writeFileSync(join(cwd, "large.txt"), "base\n");
+    git(cwd, "add", ".");
+    git(cwd, "-c", "commit.gpgsign=false", "commit", "-m", "base");
+    writeFileSync(join(cwd, "large.txt"), ("x".repeat(100) + "\n").repeat(11000));
+    git(cwd, "add", ".");
+    git(cwd, "-c", "commit.gpgsign=false", "commit", "-m", "large change");
+    const sha = git(cwd, "rev-parse", "HEAD").trim();
+    const { getCommitFileDiff } = await import("../../utils/checkout-git.js");
+    const expected = await getCommitFileDiff({ cwd, sha, path: "large.txt" });
+    const actual = await ctx.client.getCommitFileDiff(cwd, sha, "large.txt");
+    expect(expected?.hunks.length).toBeGreaterThan(0);
+    expect(actual.file?.hunks).toEqual(expected?.hunks);
+  } finally {
+    await ctx.cleanup();
+    rmSync(cwd, { recursive: true, force: true });
+  }
+}, 30000);
@@ -29,8 +29,8 @@ export class DiffDependencyObserver {
      });
      const roots = new Set(
        paths.stdout
-          .trim()
-          .split("\n")
+          .split(/\r?\n/)
+          .filter((path) => path.length > 0)
          .map((path) => resolve(input.cwd, path)),
      );
      for (const root of roots) {
@@ -240,3 +240,54 @@ async function readBody(comparison: Comparison) {
  if (record.kind !== "ready") throw new Error("Expected a readable file body");
  return record;
 }
+
+test.each(["worktree", "index", "rev"] as const)(
+  "%s whitespace filtering omits spacing-only edits and retains meaningful edits",
+  async (headKind) => {
+    const cwd = await mkdtemp(join(tmpdir(), "diff-whitespace-"));
+    const git = (...args: string[]) => execFileSync("git", args, { cwd, encoding: "utf8" });
+    try {
+      git("init", "-b", "main");
+      git("config", "user.email", "test@example.com");
+      git("config", "user.name", "Test");
+      await writeFile(join(cwd, "a.ts"), "const x = 1;\n");
+      await writeFile(join(cwd, "b.txt"), "before\n");
+      git("add", ".");
+      git("-c", "commit.gpgsign=false", "commit", "-m", "base");
+      const base = git("rev-parse", "HEAD").trim();
+      await writeFile(join(cwd, "a.ts"), "const  x = 1;\n");
+      await writeFile(join(cwd, "b.txt"), "after\n");
+      if (headKind !== "worktree") git("add", ".");
+      if (headKind === "rev") git("-c", "commit.gpgsign=false", "commit", "-m", "edit");
+      const head = headKind === "rev" ? { kind: "rev" as const, rev: "HEAD" } : { kind: headKind };
+      const comparison = await listComparison({
+        cwd,
+        base: { kind: "rev", rev: base },
+        head,
+        ignoreWhitespace: true,
+        highlight: false,
+      });
+      expect(comparison.files.map((file) => file.entry.path)).toEqual(["b.txt"]);
+      await comparison.close();
+    } finally {
+      await rm(cwd, { recursive: true, force: true });
+    }
+  },
+);
+
+test.skipIf(process.platform === "win32")(
+  "comparison roots preserve trailing folder spaces",
+  async () => {
+    const root = await mkdtemp(join(tmpdir(), "diff-space-"));
+    const cwd = join(root, "repo ");
+    const { mkdir } = await import("node:fs/promises");
+    await mkdir(cwd);
+    try {
+      execFileSync("git", ["init", "-b", "main"], { cwd });
+      const { worktreeRoot } = await import("./git.js");
+      expect(await worktreeRoot(cwd)).toBe(cwd);
+    } finally {
+      await rm(root, { recursive: true, force: true });
+    }
+  },
+);
@@ -1,3 +1,4 @@
+import { parseGitRevParsePath } from "../../utils/git-rev-parse-path.js";
 import { HighlightSources } from "./sources.js";
 import {
  prepareAllChangesIndex,
@@ -33,7 +34,10 @@ export interface Comparison extends Pick<
  files: ListedFile[];
  metrics: { highlightCalls: number };
  close(): Promise<void>;
-  read(file: ListedFile, budget?: number): Promise<DiffFileRead>;
+  read(
+    file: ListedFile,
+    budget?: { remainingBytes?: number; maxPatchBytes?: number },
+  ): Promise<DiffFileRead>;
 }
 export class DiffReadError extends Error {
  constructor(
@@ -60,7 +64,9 @@ export async function worktreeRoot(cwd: string): Promise<string> {
    acceptExitCodes: [0, 128],
  });
  if (result.exitCode !== 0) throw new DiffReadError("NOT_GIT_REPO", "Not a git repository");
-  return result.stdout.trim();
+  const root = parseGitRevParsePath(result.stdout);
+  if (root === null) throw new DiffReadError("NOT_GIT_REPO", "Git returned no working tree root");
+  return resolve(root);
 }
 async function resolveEndpoint(
  cwd: string,
@@ -192,6 +198,8 @@ async function readComparisonList(
      path = fields[++index];
    }
    const counts = stats.get(path);
+    // With -w, a raw metadata entry need not have a numstat record.
+    if (!counts && compare.ignoreWhitespace) continue;
    if (!counts) throw new Error(`Missing diff statistics for ${path}`);
    let change: DiffEntry["change"] = "modified";
    if (code === "A") change = "added";
@@ -234,10 +242,12 @@ async function readComparisonList(
    close: async () => {
      await privateIndex?.cleanup();
    },
-    async read(file, budget = PER_FILE_DIFF_MAX_BYTES) {
+    async read(file, budget = {}) {
+      const maxPatchBytes = budget.maxPatchBytes ?? PER_FILE_DIFF_MAX_BYTES;
+      const remainingBytes = budget.remainingBytes ?? maxPatchBytes;
      const paths = [file.entry.path];
      if (file.entry.oldPath !== null) paths.unshift(file.entry.oldPath);
-      if (file.source.kind === "untracked" && file.source.size > PER_FILE_DIFF_MAX_BYTES) {
+      if (file.source.kind === "untracked" && file.source.size > maxPatchBytes) {
        return { kind: "too_large" };
      }
      const command =
@@ -247,10 +257,10 @@ async function readComparisonList(
      const result = await runGitCommand(command, {
        ...options,
        acceptExitCodes: [0, 1],
-        maxOutputBytes: PER_FILE_DIFF_MAX_BYTES + 1,
+        maxOutputBytes: maxPatchBytes + 1,
      });
      const bytes = Buffer.byteLength(result.stdout);
-      if (result.truncated || bytes > budget || bytes > PER_FILE_DIFF_MAX_BYTES)
+      if (result.truncated || bytes > remainingBytes || bytes > maxPatchBytes)
        return { kind: "too_large" };
      const parsed = parseDiff(result.stdout)[0];
      if (!parsed) return { kind: "ready", body: { path: file.entry.path, hunks: [] }, bytes };
@@ -1,5 +1,5 @@
 import { TOTAL_DIFF_MAX_BYTES } from "../../utils/checkout-git.js";
-import { runGitCommand } from "../../utils/run-git-command.js";
+import { DEFAULT_MAX_OUTPUT_BYTES, runGitCommand } from "../../utils/run-git-command.js";
 import { expandTilde } from "../../utils/path.js";
 import { packBodies } from "./batches.js";
 import { setImmediate } from "node:timers/promises";
@@ -35,6 +35,7 @@ interface RecordCache {
 interface Listener {
  emit(frame: DiffFrame): Promise<void>;
  active: boolean;
+  delivery: Promise<void>;
 }
 interface Target {
  key: string;
@@ -99,7 +100,7 @@ export class DiffSubscriptions {
      input.signal?.throwIfAborted();
      const file = comparison.files.find((candidate) => candidate.entry.path === input.path);
      if (!file || file.entry.content === "binary") return null;
-      const record = await comparison.read(file);
+      const record = await comparison.read(file, { maxPatchBytes: DEFAULT_MAX_OUTPUT_BYTES });
      input.signal?.throwIfAborted();
      return materializeDiffFile(
        file.entry,
@@ -182,16 +183,20 @@ export class DiffSubscriptions {
    }
    if (target.closed) throw new Error("Diff subscription cancelled");
    const current = target;
-    const listener: Listener = { emit, active: false };
+    const listener: Listener = { emit, active: false, delivery: Promise.resolve() };
    current.listeners.add(listener);
    let released = false;
    return {
      start: async () => {
-        if (current.refresh) await current.refresh;
+        while (current.refresh) await current.refresh;
        if (current.closed || released) return;
        listener.active = true;
        if (!current.committed) await this.refresh(current);
-        else await this.replay(current, listener);
+        else {
+          const replay = this.replay(current, listener);
+          listener.delivery = replay.catch(() => {});
+          await replay;
+        }
      },
      release: async () => {
        if (released) return;
@@ -225,7 +230,11 @@ export class DiffSubscriptions {
    await Promise.all(
      [...target.listeners]
        .filter((listener) => listener.active)
-        .map((listener) => listener.emit(frame)),
+        .map((listener) => {
+          const sent = listener.delivery.then(() => listener.emit(frame));
+          listener.delivery = sent.catch(() => {});
+          return sent;
+        }),
    );
    await setImmediate();
  }
@@ -371,7 +380,9 @@ export class DiffSubscriptions {
      if (entry.content === "binary") continue;
      let record = target.cache.get(entry.path);
      if (entry.body || !record) {
-        const read = await comparison.read(file, TOTAL_DIFF_MAX_BYTES - state.total);
+        const read = await comparison.read(file, {
+          remainingBytes: TOTAL_DIFF_MAX_BYTES - state.total,
+        });
        if (read.kind === "too_large") {
          state.omitted.push(entry.path);
          continue;
@@ -105,7 +105,11 @@ test("shares computation, highlights only the changed file, recovers refs and re
    expect(subscriptions.getMetrics().metadataObservers).toBe(0);
    const missing: DiffFrame[] = [];
    const recovery = await subscriptions.subscribe(
-      { ...compare, base: { kind: "rev", rev: "refs/tags/later" } },
+      {
+        ...compare,
+        base: { kind: "rev", rev: "refs/tags/later" },
+        head: { kind: "rev", rev: "HEAD" },
+      },
      async (frame) => {
        missing.push(frame);
      },
@@ -291,3 +295,67 @@ test("moving, deleting and recreating a branch recovers, and staging changes the
    await rm(cwd, { recursive: true, force: true });
  }
 }, 30000);
+
+test("a cached replay completes before a live refresh can depend on it", async () => {
+  const cwd = await mkdtemp(join(tmpdir(), "diff-replay-order-"));
+  const git = (...args: string[]) => execFileSync("git", args, { cwd, encoding: "utf8" });
+  const subscriptions = new DiffSubscriptions({
+    async requestWorkingTreeWatch() {
+      return { repoRoot: cwd, unsubscribe: async () => {} };
+    },
+  });
+  let resume!: () => void;
+  let began!: () => void;
+  const paused = new Promise<void>((resolve) => {
+    resume = resolve;
+  });
+  const replayBegan = new Promise<void>((resolve) => {
+    began = resolve;
+  });
+  const frames: DiffFrame[] = [];
+  try {
+    git("init", "-b", "main");
+    git("config", "user.email", "test@example.com");
+    git("config", "user.name", "Test");
+    await writeFile(join(cwd, "a.txt"), "base\n");
+    git("add", ".");
+    git("-c", "commit.gpgsign=false", "commit", "-m", "base");
+    await writeFile(join(cwd, "a.txt"), "first\n");
+    const compare = {
+      cwd,
+      base: { kind: "rev", rev: "HEAD" },
+      head: { kind: "worktree" },
+      highlight: false,
+    } as const;
+    const first = await subscriptions.subscribe(compare, async () => {});
+    await first.start();
+    const second = await subscriptions.subscribe(compare, async (frame) => {
+      frames.push(frame);
+      if (frames.length === 1) {
+        began();
+        await paused;
+      }
+    });
+    const replay = second.start();
+    await replayBegan;
+    await writeFile(join(cwd, "a.txt"), "second\n");
+    subscriptions.invalidateForCwd(cwd);
+    await new Promise((resolve) => setTimeout(resolve, 250));
+    resume();
+    await replay;
+    await expect
+      .poll(() => frames.filter((frame) => frame.opcode === DiffOpcode.End).length)
+      .toBe(2);
+    let committed = 0;
+    for (const frame of frames) {
+      if (frame.opcode === DiffOpcode.Begin) expect([0, committed]).toContain(frame.baseRevision);
+      if (frame.opcode === DiffOpcode.End) committed = frame.revision;
+    }
+    await second.release();
+    await first.release();
+  } finally {
+    resume();
+    await subscriptions.close();
+    await rm(cwd, { recursive: true, force: true });
+  }
+});
@@ -1,3 +1,4 @@
+import { DiffSubscriptions } from "./diff/index.js";
 import { createWorkspaceRuntimeStore } from "./workspace-runtime-store/index.js";
 import { createAgentMessageDeliveryStub } from "./test-utils/session-stubs.js";
 import {
@@ -105,6 +106,13 @@ describe("snapshot mutation ownership boundary", () => {

    const session = asInternals<SessionInternals>(
      new Session({
+        checkoutDiffManager: createStub<SessionOptions["checkoutDiffManager"]>({
+          streams: new DiffSubscriptions({
+            async requestWorkingTreeWatch() {
+              throw new Error("Unexpected diff observation");
+            },
+          }),
+        }),
        workspaceRuntimeStore: createWorkspaceRuntimeStore(),
        messageReceipts: createMessageReceiptsStub(),
        agentMessages: createAgentMessageDeliveryStub(),
@@ -11,7 +11,6 @@ import {
  type CheckoutSnapshotFacts,
  type CheckoutStatusGit,
 } from "../utils/checkout-git.js";
-import { CheckoutDiffManager } from "./checkout-diff-manager.js";
 import { createFileObserver } from "./file-observer/index.js";
 import { WorkspaceGitServiceImpl } from "./workspace-git-service.js";
 import type { FileChange, FileObserver, SubscribeToFileChanges } from "./file-observer/index.js";
@@ -250,24 +249,15 @@ test("recursive observation updates tracked state and prunes ignored storms", as
  });
  const workspaceRuntimeStore = createWorkspaceRuntimeStore();
  service.onSnapshotUpdated((snapshot) => workspaceRuntimeStore.set(snapshot));
-  const diffManager = new CheckoutDiffManager({
-    workspaceRuntimeStore: workspaceRuntimeStore,
-    logger: createLogger(),
-    paseoHome: path.join(tempDir, "paseo-home"),
-    workspaceGitService: service,
-  });
+
  const summaryListener = vi.fn();
  const diffListener = vi.fn();
  const summarySubscription = service.registerWorkspace({ cwd: repoDir }, summaryListener);
-  const diffSubscription = await diffManager.subscribe(
-    { cwd: repoDir, compare: { mode: "uncommitted" } },
-    diffListener,
-  );
+  const diffSubscription = await service.requestWorkingTreeWatch(repoDir, diffListener);

  cleanup.push(async () => {
    diffSubscription.unsubscribe();
    summarySubscription.unsubscribe();
-    diffManager.dispose();
    await service.dispose();
    expect(activeWatcherCount).toBe(0);
    rmSync(tempDir, { recursive: true, force: true });
@@ -312,6 +302,7 @@ test("recursive observation updates tracked state and prunes ignored storms", as
  getCheckoutShortstat.mockClear();
  getCheckoutWorktreeState.mockClear();
  getCheckoutDiff.mockClear();
+  diffListener.mockClear();
  runGitCommand.mockClear();
  deliveredEvents.length = 0;

@@ -344,18 +335,7 @@ test("recursive observation updates tracked state and prunes ignored storms", as
          }),
        }),
      );
-      expect(diffListener).toHaveBeenLastCalledWith({
-        cwd: repoDir,
-        files: [
-          {
-            path: "src/tracked.txt",
-            additions: 2,
-            deletions: 0,
-            status: "modified",
-          },
-        ],
-        error: null,
-      });
+      expect(diffListener).toHaveBeenCalled();
      expect(service.getMetrics().workspaceRefreshQueuedCount).toBe(0);
    },
    { timeout: 5_000 },
@@ -371,11 +351,7 @@ test("recursive observation updates tracked state and prunes ignored storms", as
          }),
        }),
      );
-      expect(diffListener).toHaveBeenLastCalledWith(
-        expect.objectContaining({
-          files: [expect.objectContaining({ additions: 3 })],
-        }),
-      );
+      expect(diffListener).toHaveBeenCalled();
      expect(service.getMetrics()).toMatchObject({
        workspaceRefreshInFlightCount: 0,
        workspaceRefreshQueuedCount: 0,
@@ -409,18 +385,7 @@ test("recursive observation updates tracked state and prunes ignored storms", as
          }),
        }),
      );
-      expect(diffListener).toHaveBeenLastCalledWith({
-        cwd: repoDir,
-        files: [
-          {
-            path: "build/tracked.txt",
-            additions: 2,
-            deletions: 0,
-            status: "modified",
-          },
-        ],
-        error: null,
-      });
+      expect(diffListener).toHaveBeenCalled();
      expect(service.getMetrics()).toMatchObject({
        workspaceRefreshInFlightCount: 0,
        workspaceRefreshQueuedCount: 0,
@@ -434,6 +399,7 @@ test("recursive observation updates tracked state and prunes ignored storms", as
  getCheckoutShortstat.mockClear();
  getCheckoutWorktreeState.mockClear();
  getCheckoutDiff.mockClear();
+  diffListener.mockClear();
  runGitCommand.mockClear();
  summaryListener.mockClear();
  diffListener.mockClear();
@@ -449,18 +415,7 @@ test("recursive observation updates tracked state and prunes ignored storms", as
          }),
        }),
      );
-      expect(diffListener).toHaveBeenLastCalledWith({
-        cwd: repoDir,
-        files: [
-          {
-            path: "build/tracked.txt",
-            additions: 4,
-            deletions: 0,
-            status: "modified",
-          },
-        ],
-        error: null,
-      });
+      expect(diffListener).toHaveBeenCalled();
      expect(service.getMetrics()).toMatchObject({
        workspaceRefreshInFlightCount: 0,
        workspaceRefreshQueuedCount: 0,
@@ -474,6 +429,7 @@ test("recursive observation updates tracked state and prunes ignored storms", as
  getCheckoutShortstat.mockClear();
  getCheckoutWorktreeState.mockClear();
  getCheckoutDiff.mockClear();
+  diffListener.mockClear();
  runGitCommand.mockClear();

  for (let index = 0; index < 100; index += 1) {
@@ -4,7 +4,6 @@ import { pathToFileURL } from "node:url";
 import type pino from "pino";
 import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
 import type { CheckoutSnapshotFacts, CheckoutStatusGit } from "../utils/checkout-git.js";
-import { CheckoutDiffManager } from "./checkout-diff-manager.js";
 import type { FileObserver } from "./file-observer/index.js";
 import { WorkspaceGitServiceImpl } from "./workspace-git-service.js";

@@ -428,41 +427,30 @@ describe("WorkspaceGitService checkout observation", () => {
    service.dispose();
  });

-  test("a tracked edit refreshes summary and active uncommitted diff without structural or forge work", async () => {
+  test("a tracked edit refreshes summary and notifies worktree consumers without structural or forge work", async () => {
    const watcher = createWatcherHarness();
    const getCheckoutSnapshotFacts = vi.fn(async (cwd: string) => createCheckoutFacts(cwd));
    let diffStat = { additions: 1, deletions: 0 };
-    let diffFile = { path: "tracked.txt", additions: 1, deletions: 0, status: "modified" };
    const getCheckoutStatus = vi.fn(async (cwd: string) =>
      createCheckoutStatus(cwd, { isDirty: true }),
    );
    const getCheckoutShortstat = vi.fn(async () => diffStat);
    const getPullRequestStatus = vi.fn();
-    const getCheckoutDiff = vi.fn(async () => ({ diff: "", structured: [diffFile] }));
    const service = createService(watcher, {
      getCheckoutSnapshotFacts,
      getCheckoutStatus,
      getCheckoutShortstat,
      getPullRequestStatus,
-      getCheckoutDiff,
    });
    const workspaceRuntimeStore = createWorkspaceRuntimeStore();
    service.onSnapshotUpdated((snapshot) => workspaceRuntimeStore.set(snapshot));
-    const diffManager = new CheckoutDiffManager({
-      workspaceRuntimeStore: workspaceRuntimeStore,
-      logger: createLogger(),
-      paseoHome: "/tmp/paseo-home",
-      workspaceGitService: service,
-    });
+
    const summaryListener = vi.fn();
    const diffListener = vi.fn();

    await service.getSnapshot(REPO_CWD);
    const summarySubscription = service.registerWorkspace({ cwd: REPO_CWD }, summaryListener);
-    const diffSubscription = await diffManager.subscribe(
-      { cwd: REPO_CWD, compare: { mode: "uncommitted" } },
-      diffListener,
-    );
+    const diffSubscription = await service.requestWorkingTreeWatch(REPO_CWD, diffListener);
    await vi.waitFor(() => {
      expect(watcher.subscribe).toHaveBeenCalledWith(
        REPO_CWD,
@@ -475,7 +463,6 @@ describe("WorkspaceGitService checkout observation", () => {
    });

    diffStat = { additions: 4, deletions: 2 };
-    diffFile = { path: "tracked.txt", additions: 4, deletions: 2, status: "modified" };
    watcher.records[0]?.callback(null, [
      { path: path.join(REPO_CWD, "tracked.txt"), type: "update" },
    ]);
@@ -489,11 +476,7 @@ describe("WorkspaceGitService checkout observation", () => {
          }),
        }),
      );
-      expect(diffListener).toHaveBeenLastCalledWith({
-        cwd: REPO_CWD,
-        files: [{ path: "tracked.txt", additions: 4, deletions: 2, status: "modified" }],
-        error: null,
-      });
+      expect(diffListener).toHaveBeenCalledTimes(1);
    });
    await flushPromises();

@@ -502,7 +485,6 @@ describe("WorkspaceGitService checkout observation", () => {

    diffSubscription.unsubscribe();
    summarySubscription.unsubscribe();
-    diffManager.dispose();
    service.dispose();
  });

@@ -1231,7 +1213,7 @@ describe("WorkspaceGitService checkout observation", () => {
    service.dispose();
  });

-  test("metadata-only changes refresh worktree summary and active uncommitted diff", async () => {
+  test("metadata-only changes refresh worktree summary and invalidate the uncommitted projection", async () => {
    const watcher = createWatcherHarness();
    let isDirty = true;
    let diffStat = { additions: 3, deletions: 1 };
@@ -1246,19 +1228,10 @@ describe("WorkspaceGitService checkout observation", () => {
    });
    const workspaceRuntimeStore = createWorkspaceRuntimeStore();
    service.onSnapshotUpdated((snapshot) => workspaceRuntimeStore.set(snapshot));
-    const diffManager = new CheckoutDiffManager({
-      workspaceRuntimeStore: workspaceRuntimeStore,
-      logger: createLogger(),
-      paseoHome: "/tmp/paseo-home",
-      workspaceGitService: service,
-    });
+
    const summaryListener = vi.fn();
-    const diffListener = vi.fn();
    const summarySubscription = service.registerWorkspace({ cwd: REPO_CWD }, summaryListener);
-    const diffSubscription = await diffManager.subscribe(
-      { cwd: REPO_CWD, compare: { mode: "uncommitted" } },
-      diffListener,
-    );
+    await service.getCheckoutDiff(REPO_CWD, { mode: "uncommitted" });

    await vi.waitFor(() => {
      expect(service.getMetrics().workspaceObservationSetupInFlightCount).toBe(0);
@@ -1281,16 +1254,13 @@ describe("WorkspaceGitService checkout observation", () => {
          }),
        }),
      );
-      expect(diffListener).toHaveBeenLastCalledWith({
-        cwd: REPO_CWD,
-        files: [],
-        error: null,
-      });
    });

-    diffSubscription.unsubscribe();
+    expect(await service.getCheckoutDiff(REPO_CWD, { mode: "uncommitted" })).toMatchObject({
+      structured: [],
+    });
+    expect(getCheckoutDiff).toHaveBeenCalledTimes(2);
    summarySubscription.unsubscribe();
-    diffManager.dispose();
    service.dispose();
  });

@@ -1304,20 +1274,14 @@ describe("WorkspaceGitService checkout observation", () => {
    const service = createService(watcher, { getCheckoutDiff, getCheckoutWorktreeState });
    const workspaceRuntimeStore = createWorkspaceRuntimeStore();
    service.onSnapshotUpdated((snapshot) => workspaceRuntimeStore.set(snapshot));
-    const diffManager = new CheckoutDiffManager({
-      workspaceRuntimeStore: workspaceRuntimeStore,
-      logger: createLogger(),
-      paseoHome: "/tmp/paseo-home",
-      workspaceGitService: service,
-    });
-    const diffSubscription = await diffManager.subscribe(
-      { cwd: REPO_CWD, compare: { mode: "base", baseRef: "main" } },
-      vi.fn(),
-    );
+
+    const subscription = service.registerWorkspace({ cwd: REPO_CWD }, () => {});
+    await service.getCheckoutDiff(REPO_CWD, { mode: "base", baseRef: "main" });

    await vi.waitFor(() => {
      expect(service.getMetrics().workspaceObservationSetupInFlightCount).toBe(0);
    });
+    await service.getCheckoutDiff(REPO_CWD, { mode: "base", baseRef: "main" });
    expect(getCheckoutDiff).toHaveBeenCalledTimes(1);

    watcher.records
@@ -1325,6 +1289,7 @@ describe("WorkspaceGitService checkout observation", () => {
      ?.callback(null, [{ path: path.join(REPO_CWD, "tracked.txt"), type: "update" }]);
    await vi.advanceTimersByTimeAsync(1_000);
    await flushPromises();
+    await service.getCheckoutDiff(REPO_CWD, { mode: "base", baseRef: "main" });
    expect(getCheckoutDiff).toHaveBeenCalledTimes(1);

    watcher.records
@@ -1332,12 +1297,10 @@ describe("WorkspaceGitService checkout observation", () => {
      ?.callback(null, [{ path: path.join(GIT_DIR, "packed-refs"), type: "update" }]);
    await vi.advanceTimersByTimeAsync(1_000);
    await vi.advanceTimersByTimeAsync(150);
-    await vi.waitFor(() => {
-      expect(getCheckoutDiff).toHaveBeenCalledTimes(2);
-    });
+    await service.getCheckoutDiff(REPO_CWD, { mode: "base", baseRef: "main" });
+    expect(getCheckoutDiff).toHaveBeenCalledTimes(2);

-    diffSubscription.unsubscribe();
-    diffManager.dispose();
+    subscription.unsubscribe();
    service.dispose();
  });

@@ -2759,39 +2722,23 @@ describe("WorkspaceGitService checkout observation", () => {
    service.dispose();
  });

-  test("a working-tree watcher event refreshes summary and diff subscribers", async () => {
+  test("a working-tree watcher event refreshes summary and notifies worktree consumers", async () => {
    const watcher = createWatcherHarness();
    let isDirty = false;
    let diffStat = { additions: 0, deletions: 0 };
-    let diffFiles: Array<{
-      path: string;
-      additions: number;
-      deletions: number;
-      status: string;
-    }> = [];
    const getCheckoutStatus = vi.fn(async (cwd: string) => createCheckoutStatus(cwd, { isDirty }));
    const getCheckoutShortstat = vi.fn(async () => diffStat);
-    const getCheckoutDiff = vi.fn(async () => ({ diff: "", structured: diffFiles }));
    const service = createService(watcher, {
      getCheckoutStatus,
      getCheckoutShortstat,
-      getCheckoutDiff,
    });
    const workspaceRuntimeStore = createWorkspaceRuntimeStore();
    service.onSnapshotUpdated((snapshot) => workspaceRuntimeStore.set(snapshot));
-    const diffManager = new CheckoutDiffManager({
-      workspaceRuntimeStore: workspaceRuntimeStore,
-      logger: createLogger(),
-      paseoHome: "/tmp/paseo-home",
-      workspaceGitService: service,
-    });
+
    const listener = vi.fn();
    const diffListener = vi.fn();
    const subscription = service.registerWorkspace({ cwd: REPO_CWD }, listener);
-    const diffSubscription = await diffManager.subscribe(
-      { cwd: REPO_CWD, compare: { mode: "uncommitted" } },
-      diffListener,
-    );
+    const diffSubscription = await service.requestWorkingTreeWatch(REPO_CWD, diffListener);

    await vi.waitFor(() => {
      expect(workspaceRuntimeStore.get(REPO_CWD)).not.toBeNull();
@@ -2799,7 +2746,6 @@ describe("WorkspaceGitService checkout observation", () => {
    });
    isDirty = true;
    diffStat = { additions: 7, deletions: 3 };
-    diffFiles = [{ path: "tracked.txt", additions: 7, deletions: 3, status: "modified" }];

    getWatcherRecordsForDirectory(watcher, REPO_CWD)[0]?.callback(null, [
      { path: path.join(REPO_CWD, "tracked.txt"), type: "update" },
@@ -2814,17 +2760,12 @@ describe("WorkspaceGitService checkout observation", () => {
          }),
        }),
      );
-      expect(diffListener).toHaveBeenLastCalledWith({
-        cwd: REPO_CWD,
-        files: [{ path: "tracked.txt", additions: 7, deletions: 3, status: "modified" }],
-        error: null,
-      });
+      expect(diffListener).toHaveBeenCalledTimes(1);
    });
    expect(getCheckoutStatus).toHaveBeenCalledTimes(2);

    diffSubscription.unsubscribe();
    subscription.unsubscribe();
-    diffManager.dispose();
    service.dispose();
  });
 });
@@ -428,6 +428,8 @@ it("compares tag-based worktrees in status, history and legacy diff modes", asyn
 it("announces a usable comparison base while checked out on the base branch", async () => {
  const cwd = makeTempDir();
  git(["init", "-b", "main"], cwd);
+  git(["config", "user.name", "Test"], cwd);
+  git(["config", "user.email", "test@example.com"], cwd);
  writeFileSync(join(cwd, "a.txt"), "before\n");
  git(["add", "."], cwd);
  commit(cwd, "base");
@@ -1744,12 +1744,12 @@ async function resolveCommitComparisonRef(
  name: string,
  context?: CheckoutContext,
 ): Promise<string> {
-  await runGitCommand(["rev-parse", "--verify", `${name}^{commit}`], {
+  await getRunGitCommand(context)(["rev-parse", "--verify", `${name}^{commit}`], {
    cwd,
    envOverlay: READ_ONLY_GIT_ENV,
    logger: context?.logger,
  });
-  const result = await runGitCommand(["rev-parse", "--symbolic-full-name", name], {
+  const result = await getRunGitCommand(context)(["rev-parse", "--symbolic-full-name", name], {
    cwd,
    envOverlay: READ_ONLY_GIT_ENV,
    logger: context?.logger,
@@ -20,7 +20,7 @@ import {
 } from "./git-process-scheduler.js";

 const DEFAULT_TIMEOUT_MS = 30_000;
-const DEFAULT_MAX_OUTPUT_BYTES = 20 * 1024 * 1024; // 20MB
+export const DEFAULT_MAX_OUTPUT_BYTES = 20 * 1024 * 1024; // 20MB
 const DEFAULT_STDERR_LIMIT = 2048;

 let gitProcessScheduler = new GitProcessScheduler(resolveGitProcessPolicy({ env: process.env }));