From 1d8eacd38aa2b7319cf3a5a1e2ea38f8166bce27 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Fri, 6 Nov 2026 05:52:32 +0100 Subject: [PATCH] feat(server): stream shared diff revisions with paced file bodies --- packages/protocol/src/binary-frames/diff.ts | 9 +- packages/server/package.json | 3 +- .../scripts/measure-diff-subscriptions.ts | 111 ++++ .../src/server/checkout-diff-manager.test.ts | 603 ++++-------------- .../src/server/checkout-diff-manager.ts | 467 ++++---------- .../checkout-diff-content.e2e.test.ts | 123 ++++ packages/server/src/server/diff/batches.ts | 30 + .../server/src/server/diff/dependencies.ts | 76 +++ packages/server/src/server/diff/git.test.ts | 242 +++++++ packages/server/src/server/diff/git.ts | 339 ++++++++++ packages/server/src/server/diff/index.ts | 428 +++++++++++++ .../server/src/server/diff/session.test.ts | 96 +++ packages/server/src/server/diff/session.ts | 86 +++ packages/server/src/server/diff/sources.ts | 61 ++ .../src/server/diff/subscriptions.test.ts | 293 +++++++++ .../server/git-metadata-event-rules.test.ts | 2 +- .../src/server/git-metadata-event-rules.ts | 8 + packages/server/src/server/session.ts | 7 + .../session/checkout/checkout-session.test.ts | 1 + .../session/checkout/checkout-session.ts | 9 +- .../server/src/server/websocket-server.ts | 3 +- packages/server/src/utils/checkout-git.ts | 94 +-- .../server/src/utils/git-file-contents.ts | 16 +- 23 files changed, 2250 insertions(+), 857 deletions(-) create mode 100644 packages/server/scripts/measure-diff-subscriptions.ts create mode 100644 packages/server/src/server/diff/batches.ts create mode 100644 packages/server/src/server/diff/dependencies.ts create mode 100644 packages/server/src/server/diff/git.test.ts create mode 100644 packages/server/src/server/diff/git.ts create mode 100644 packages/server/src/server/diff/index.ts create mode 100644 packages/server/src/server/diff/session.test.ts create mode 100644 packages/server/src/server/diff/session.ts create mode 100644 packages/server/src/server/diff/sources.ts create mode 100644 packages/server/src/server/diff/subscriptions.test.ts diff --git a/packages/protocol/src/binary-frames/diff.ts b/packages/protocol/src/binary-frames/diff.ts index c3e6fb5f7e..4e2457e29e 100644 --- a/packages/protocol/src/binary-frames/diff.ts +++ b/packages/protocol/src/binary-frames/diff.ts @@ -92,6 +92,9 @@ export interface DiffEnd { export type DiffFrame = DiffBegin | DiffBodies | DiffEnd; export class InvalidDiffFrameError extends Error {} +const textEncoder = new TextEncoder(); +const textDecoder = new TextDecoder("utf-8", { fatal: true }); + class Writer { private bytes: number[] = []; byte(value: number): void { @@ -109,7 +112,7 @@ class Writer { } while (value); } str(value: string): void { - const bytes = new TextEncoder().encode(value); + const bytes = textEncoder.encode(value); this.uint(bytes.length); for (const byte of bytes) this.byte(byte); } @@ -147,9 +150,7 @@ class Reader { const start = this.offset; this.offset += size; try { - return new TextDecoder("utf-8", { fatal: true }).decode( - this.bytes.subarray(start, this.offset), - ); + return textDecoder.decode(this.bytes.subarray(start, this.offset)); } catch { throw new InvalidDiffFrameError("Invalid UTF-8"); } diff --git a/packages/server/package.json b/packages/server/package.json index 471e558b77..43aafa8044 100644 --- a/packages/server/package.json +++ b/packages/server/package.json @@ -102,7 +102,8 @@ "test:e2e:real": "npm run test:integration:real", "test:e2e:local": "npm run test:integration:local", "test:e2e:relay:live": "cross-env RUN_LIVE_RELAY_E2E=1 vitest run src/server/daemon-e2e/live-relay.real.e2e.test.ts --maxWorkers=1", - "test:e2e:ui": "vitest --ui e2e.test.ts" + "test:e2e:ui": "vitest --ui e2e.test.ts", + "measure:diff-subscriptions": "tsx scripts/measure-diff-subscriptions.ts" }, "dependencies": { "@agentclientprotocol/sdk": "^0.17.1", diff --git a/packages/server/scripts/measure-diff-subscriptions.ts b/packages/server/scripts/measure-diff-subscriptions.ts new file mode 100644 index 0000000000..e8e058708e --- /dev/null +++ b/packages/server/scripts/measure-diff-subscriptions.ts @@ -0,0 +1,111 @@ +import { performance } from "node:perf_hooks"; +import { mkdir, writeFile } from "node:fs/promises"; +import { resolve } from "node:path"; +import { + encodeDiffFrame, + DiffOpcode, + type DiffFrame, +} from "@getpaseo/protocol/binary-frames/index"; +import { DiffSubscriptions } from "../src/server/diff/index.js"; +import { createFileObserver } from "../src/server/file-observer/index.js"; +import { runGitCommand } from "../src/utils/run-git-command.js"; +import { parseAndHighlightDiff } from "../src/server/utils/diff-highlighter.js"; +import { validateWSOutboundMessage } from "@getpaseo/protocol/validation/ws-outbound"; + +const cwd = resolve(import.meta.dirname, "../../.."); +const artifacts = resolve(cwd, ".artifacts/diff-subscription"); +await mkdir(artifacts, { recursive: true }); +const pinnedHead = "3ed57e9dd"; +const observer = createFileObserver(); +const engine = new DiffSubscriptions({ + async requestWorkingTreeWatch(root, changed) { + const watch = await observer.subscribe(root, () => changed(), { + ignore: [resolve(root, ".git")], + }); + return { + repoRoot: root, + unsubscribe: async () => { + await watch.unsubscribe(); + }, + }; + }, +}); +try { + for (const count of [1, 30]) { + const [base, head] = await Promise.all([ + runGitCommand(["rev-parse", `${pinnedHead}~${count}`], { cwd }), + runGitCommand(["rev-parse", pinnedHead], { cwd }), + ]); + const started = performance.now(); + const frames: DiffFrame[] = []; + const encoded: Uint8Array[] = []; + const subscription = await engine.subscribe( + { + cwd, + base: { kind: "rev", rev: base.stdout.trim() }, + head: { kind: "rev", rev: head.stdout.trim() }, + }, + async (frame) => { + frames.push(frame); + encoded.push(encodeDiffFrame(frame)); + }, + ); + await subscription.start(); + const coldMs = performance.now() - started; + await subscription.release(); + const raw = await runGitCommand(["diff", base.stdout.trim(), head.stdout.trim()], { + cwd, + maxOutputBytes: 50 * 1024 * 1024, + }); + const contentAt = async (sha: string, path: string) => { + const result = await runGitCommand(["show", `${sha}:${path}`], { + cwd, + acceptExitCodes: [0, 128], + }); + return result.exitCode === 0 ? result.stdout : null; + }; + const legacyFiles = await parseAndHighlightDiff(raw.stdout, cwd, { + getOldFileContent: (file) => contentAt(base.stdout.trim(), file.path), + getNewFileContent: (file) => contentAt(head.stdout.trim(), file.path), + }); + const legacy = JSON.stringify({ + type: "session", + message: { + type: "checkout_diff_update", + payload: { cwd, subscriptionId: "benchmark", files: legacyFiles, error: null }, + }, + }); + const parseStart = performance.now(); + const validation = validateWSOutboundMessage(JSON.parse(legacy)); + if (!validation.success) throw validation.error; + const jsonParseValidationMs = performance.now() - parseStart; + const begin = frames.find((frame) => frame.opcode === DiffOpcode.Begin); + const end = frames.findLast((frame) => frame.opcode === DiffOpcode.End); + if (!begin || begin.opcode !== DiffOpcode.Begin || !end || end.opcode !== DiffOpcode.End) + throw new Error("Revision was not delimited"); + if (end.error) throw new Error(end.error.message); + const metrics = { + pinnedHead: head.stdout.trim(), + count, + files: begin.entries.length, + rawPatchBytes: Buffer.byteLength(raw.stdout), + encodedBytes: encoded.reduce((sum, bytes) => sum + bytes.length, 0), + legacyJsonBytes: Buffer.byteLength(legacy), + coldMs, + jsonParseValidationMs, + omitted: end.omitted, + }; + console.log(JSON.stringify(metrics)); + await writeFile( + resolve(artifacts, `corpus-${count}.json`), + JSON.stringify({ + metrics, + legacy, + frames: encoded.map((bytes) => Buffer.from(bytes).toString("base64")), + }), + ); + } +} finally { + await engine.close(); + await observer.close(); +} diff --git a/packages/server/src/server/checkout-diff-manager.test.ts b/packages/server/src/server/checkout-diff-manager.test.ts index d55171831b..540af69677 100644 --- a/packages/server/src/server/checkout-diff-manager.test.ts +++ b/packages/server/src/server/checkout-diff-manager.test.ts @@ -1,488 +1,167 @@ +import { execFileSync } from "node:child_process"; +import { mkdtemp, writeFile, rm, mkdir } from "node:fs/promises"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import pino from "pino"; +import { expect, test } from "vitest"; +import { CheckoutDiffManager, type CheckoutDiffSnapshotPayload } from "./checkout-diff-manager.js"; import { createWorkspaceRuntimeStore } from "./workspace-runtime-store/index.js"; -import { afterEach, beforeEach, describe, expect, test, vi } from "vitest"; +import { createFileObserver } from "./file-observer/index.js"; -const { toCheckoutErrorMock } = vi.hoisted(() => ({ - toCheckoutErrorMock: vi.fn((error: unknown) => ({ - message: error instanceof Error ? error.message : String(error), - })), -})); - -vi.mock("./checkout-git-utils.js", () => ({ - toCheckoutError: toCheckoutErrorMock, -})); - -import type pino from "pino"; -import { CheckoutDiffManager } from "./checkout-diff-manager.js"; -import type { WorkspaceGitRuntimeSnapshot, WorkspaceGitService } from "./workspace-git-service.js"; - -interface Deferred { - promise: Promise; - resolve(value: T): void; -} - -function createDeferred(): Deferred { - let resolve!: (value: T) => void; - const promise = new Promise((resolvePromise) => { - resolve = resolvePromise; - }); - return { promise, resolve }; -} - -function createWorkspaceSnapshot( - overrides?: Partial, -): WorkspaceGitRuntimeSnapshot { - return { - cwd: "/tmp/repo", - git: { - isGit: true, - repoRoot: "/tmp/repo", - mainRepoRoot: null, - currentBranch: "feature", - remoteUrl: "https://github.com/acme/repo.git", - isPaseoOwnedWorktree: false, - isDirty: false, - baseRef: "main", - aheadBehind: { ahead: 1, behind: 0 }, - aheadOfOrigin: 1, - behindOfOrigin: 0, - hasRemote: true, - diffStat: { additions: 1, deletions: 0 }, - ...overrides, - }, - forge: { - featuresEnabled: false, - authState: "no_remote", - pullRequest: null, - error: null, - }, - }; -} - -function createPendingManager() { - const watches: Array<{ +async function withCheckout( + run: (input: { cwd: string; - onChange: () => void; - unsubscribeCalls: number; - resolve(): void; - }> = []; - const workspaceGitService = { - getCheckoutDiff: async () => ({ diff: "", structured: [] }), - getSnapshot: async () => createWorkspaceSnapshot(), - - registerWorkspace: () => ({ unsubscribe: () => {} }), - requestWorkingTreeWatch: (cwd: string, onChange: () => void) => { - const pending = createDeferred<{ repoRoot: string | null; unsubscribe: () => void }>(); - const watch = { - cwd, - onChange, - unsubscribeCalls: 0, - resolve: () => { - pending.resolve({ - repoRoot: "/tmp/repo", - unsubscribe: () => { - watch.unsubscribeCalls += 1; - }, - }); - }, - }; - watches.push(watch); - return pending.promise; - }, - }; - const logger = { child: () => logger, warn: () => {} }; + manager: CheckoutDiffManager; + git(...args: string[]): string; + }) => Promise, +): Promise { + const cwd = await mkdtemp(join(tmpdir(), "legacy-diff-adapter-")); + const observer = createFileObserver(); + const git = (...args: string[]) => execFileSync("git", args, { cwd, encoding: "utf8" }).trim(); const manager = new CheckoutDiffManager({ + logger: pino({ level: "silent" }), + paseoHome: cwd, workspaceRuntimeStore: createWorkspaceRuntimeStore(), - logger: logger as unknown as pino.Logger, - paseoHome: "/tmp/paseo-test", - workspaceGitService, + workspaceGitService: { + async requestWorkingTreeWatch(root, onChange) { + const subscription = await observer.subscribe(root, () => onChange(), { + ignore: [join(root, ".git")], + }); + return { + repoRoot: root, + unsubscribe: async () => { + await subscription.unsubscribe(); + }, + }; + }, + }, }); - return { manager, watches }; + try { + git("init", "-b", "main"); + git("config", "user.email", "test@example.com"); + git("config", "user.name", "Test"); + await writeFile(join(cwd, "README"), "base\n"); + git("add", "."); + git("-c", "commit.gpgsign=false", "commit", "-m", "base"); + await run({ cwd, manager, git }); + } finally { + await manager.dispose(); + await observer.close(); + await rm(cwd, { recursive: true, force: true }); + } } -describe("CheckoutDiffManager", () => { - beforeEach(() => { - vi.useFakeTimers(); - toCheckoutErrorMock.mockClear(); - }); - - afterEach(() => { - vi.useRealTimers(); - }); - - function createManager(options?: { - repoRoot?: string | null; - getCheckoutDiffImplementation?: ReturnType; - }) { - const unsubscribe = vi.fn(); - const workspaceUnsubscribe = vi.fn(); - let onChange: (() => void) | null = null; - let onWorkspaceSnapshot: ((snapshot: WorkspaceGitRuntimeSnapshot) => void) | null = null; - const mockRequestWorkingTreeWatch = vi.fn(async (_cwd: string, listener: () => void) => { - onChange = listener; - return { - repoRoot: options?.repoRoot === undefined ? "/tmp/repo" : options.repoRoot, - unsubscribe, - }; - }); - - const workspaceGitService = { - subscribe: vi.fn(), - - registerWorkspace: vi.fn( - (_params: { cwd: string }, listener: (snapshot: WorkspaceGitRuntimeSnapshot) => void) => { - onWorkspaceSnapshot = listener; - return { unsubscribe: workspaceUnsubscribe }; - }, +test("legacy callers share one live comparison, receive edits, and release it", async () => { + await withCheckout(async ({ cwd, manager }) => { + await mkdir(join(cwd, "sub")); + await writeFile(join(cwd, "x.ts"), "const value = 1;\n"); + const first: CheckoutDiffSnapshotPayload[] = []; + const second: CheckoutDiffSnapshotPayload[] = []; + const [a, b] = await Promise.all([ + manager.subscribe({ cwd: join(cwd, "sub"), compare: { mode: "uncommitted" } }, (snapshot) => + first.push(snapshot), ), - getSnapshot: vi.fn(async () => createWorkspaceSnapshot()), - getCheckoutDiff: - options?.getCheckoutDiffImplementation ?? vi.fn(async () => ({ diff: "", structured: [] })), - refresh: vi.fn(), - scheduleRefreshForCwd: vi.fn(), - requestWorkingTreeWatch: mockRequestWorkingTreeWatch, - dispose: vi.fn(), - }; - - const logger = { - child: () => logger, - warn: vi.fn(), - }; - - const manager = new CheckoutDiffManager({ - workspaceRuntimeStore: createWorkspaceRuntimeStore(), - logger: logger as unknown as pino.Logger, - paseoHome: "/tmp/paseo-test", - workspaceGitService: workspaceGitService as unknown as WorkspaceGitService, + manager.subscribe({ cwd, compare: { mode: "uncommitted" } }, (snapshot) => + second.push(snapshot), + ), + ]); + expect(a.initial.cwd).toBe(join(cwd, "sub")); + expect(a.initial.files).toEqual(b.initial.files); + expect(a.initial.files.map((file) => file.path)).toEqual(["x.ts"]); + expect(manager.getMetrics()).toMatchObject({ + checkoutDiffTargetCount: 1, + checkoutDiffSubscriptionCount: 2, }); + await writeFile(join(cwd, "x.ts"), "const value = 2;\n"); + await expect.poll(() => first.length, { timeout: 10000 }).toBeGreaterThan(0); + expect(first.at(-1)?.files).toEqual(second.at(-1)?.files); + await a.unsubscribe(); + await b.unsubscribe(); + expect(manager.getMetrics().checkoutDiffTargetCount).toBe(0); + }); +}, 30_000); - return { - manager, - workspaceGitService, - mockRequestWorkingTreeWatch, - unsubscribe, - getOnChange: () => onChange, - getOnWorkspaceSnapshot: () => onWorkspaceSnapshot, - workspaceUnsubscribe, - }; - } - - test("subscribe requests a working tree watch with the correct cwd", async () => { - const { manager, mockRequestWorkingTreeWatch } = createManager(); - - await manager.subscribe( - { - cwd: "/tmp/repo/packages/server", - compare: { mode: "uncommitted" }, - }, - () => {}, +test("Committed ignores working-tree edits while All and Uncommitted include them", async () => { + await withCheckout(async ({ cwd, manager, git }) => { + git("checkout", "-b", "feature"); + await writeFile(join(cwd, "x.ts"), "const value = 1;\n"); + git("add", "."); + git("-c", "commit.gpgsign=false", "commit", "-m", "feature"); + await writeFile(join(cwd, "x.ts"), "const value = 2;\n"); + const committed = await manager.read({ cwd, compare: { mode: "base", baseRef: "main" } }); + const all = await manager.read({ cwd, compare: { mode: "all", baseRef: "main" } }); + const uncommitted = await manager.read({ cwd, compare: { mode: "uncommitted" } }); + expect(committed.files[0].hunks[0].lines).toContainEqual( + expect.objectContaining({ type: "add", content: "const value = 1;" }), ); - - expect(mockRequestWorkingTreeWatch).toHaveBeenCalledWith( - "/tmp/repo/packages/server", - expect.any(Function), + expect(all.files[0].hunks[0].lines).toContainEqual( + expect.objectContaining({ type: "add", content: "const value = 2;" }), + ); + expect(uncommitted.files[0].hunks[0].lines).toContainEqual( + expect.objectContaining({ type: "remove", content: "const value = 1;" }), ); }); +}, 30_000); - test("unsubscribe calls the working tree watch unsubscribe", async () => { - const { manager, unsubscribe } = createManager(); - - const subscription = await manager.subscribe( - { - cwd: "/tmp/repo/packages/server", - compare: { mode: "uncommitted" }, - }, - () => {}, - ); - - subscription.unsubscribe(); - - expect(unsubscribe).toHaveBeenCalledTimes(1); - }); - - test("cancels a subscription while its working tree watch is still opening", async () => { - const { manager, watches } = createPendingManager(); +test("cancelling a legacy subscriber during a shared revision does not revive it", async () => { + await withCheckout(async ({ cwd, manager }) => { + await writeFile(join(cwd, "x.ts"), "export const value = 1;\n"); const abort = new AbortController(); - - const pendingSubscription = manager.subscribe( - { - cwd: "/tmp/repo/packages/server", - compare: { mode: "uncommitted" }, - signal: abort.signal, + let cancelled = false; + const modern = await manager.streams.subscribe( + { cwd, base: { kind: "rev", rev: "HEAD" }, head: { kind: "worktree" } }, + async (frame) => { + if (frame.opcode !== 0x40 || cancelled) return; + await waitForSubscriberCount(manager, 2); + cancelled = true; + abort.abort(); }, - () => {}, ); - abort.abort(); - watches[0].resolve(); - await pendingSubscription; - - expect(watches[0].unsubscribeCalls).toBe(1); - expect(manager.getMetrics()).toEqual({ - checkoutDiffTargetCount: 0, - checkoutDiffSubscriptionCount: 0, - checkoutDiffWatcherCount: 0, - checkoutDiffFallbackRefreshTargetCount: 0, - }); - }); - - test("shares one opening target between concurrent subscriptions", async () => { - const { manager, watches } = createPendingManager(); - - const firstSubscription = manager.subscribe( - { cwd: "/tmp/repo/packages/server", compare: { mode: "uncommitted" } }, - () => {}, - ); - const secondSubscription = manager.subscribe( - { cwd: "/tmp/repo/packages/server", compare: { mode: "uncommitted" } }, - () => {}, - ); - - expect(watches).toHaveLength(1); - watches[0].resolve(); - const [first, second] = await Promise.all([firstSubscription, secondSubscription]); - expect(manager.getMetrics().checkoutDiffSubscriptionCount).toBe(2); - - first.unsubscribe(); - expect(watches[0].unsubscribeCalls).toBe(0); - second.unsubscribe(); - expect(watches[0].unsubscribeCalls).toBe(1); - }); - - test("diffCwd uses repoRoot from the working tree watch result", async () => { - const { manager, workspaceGitService } = createManager({ repoRoot: "/tmp/repo" }); - - await manager.subscribe( - { - cwd: "/tmp/repo/packages/server", - compare: { mode: "uncommitted" }, + const legacy = manager.subscribe( + { cwd, compare: { mode: "uncommitted" }, signal: abort.signal }, + () => { + throw new Error("Cancelled subscriber received an update"); }, - () => {}, ); - - expect(workspaceGitService.getCheckoutDiff).toHaveBeenCalledWith( - "/tmp/repo", - expect.objectContaining({ mode: "uncommitted", includeStructured: true }), - undefined, - ); - }); - - test("diff refresh is triggered when the working tree watch callback fires", async () => { - const getCheckoutDiff = vi - .fn() - .mockResolvedValueOnce({ - diff: "", - structured: [{ path: "a.ts", additions: 1, deletions: 0, status: "modified" }], - }) - .mockResolvedValueOnce({ - diff: "", - structured: [{ path: "b.ts", additions: 2, deletions: 0, status: "modified" }], + try { + await modern.start(); + const result = await legacy; + expect(result.initial).toMatchObject({ + files: [], + error: { message: "Diff subscription cancelled" }, }); - - const { manager, getOnChange } = createManager({ - getCheckoutDiffImplementation: getCheckoutDiff, - }); - const listener = vi.fn(); - - await manager.subscribe( - { - cwd: "/tmp/repo/packages/server", - compare: { mode: "uncommitted" }, - }, - listener, - ); - - const onChange = getOnChange(); - expect(onChange).toBeTypeOf("function"); - - onChange?.(); - await vi.advanceTimersByTimeAsync(150); - - expect(listener).toHaveBeenCalledTimes(1); - expect(listener).toHaveBeenCalledWith({ - cwd: "/tmp/repo/packages/server", - files: [{ path: "b.ts", additions: 2, deletions: 0, status: "modified" }], - error: null, - }); - }); - - test("watch-triggered refresh forces a cache bypass on getCheckoutDiff", async () => { - const getCheckoutDiff = vi - .fn() - .mockResolvedValueOnce({ - diff: "", - structured: [{ path: "a.ts", additions: 1, deletions: 0, status: "modified" }], - }) - .mockResolvedValueOnce({ - diff: "", - structured: [{ path: "b.ts", additions: 2, deletions: 0, status: "modified" }], - }); - - const { manager, getOnChange } = createManager({ - getCheckoutDiffImplementation: getCheckoutDiff, - }); - - await manager.subscribe( - { - cwd: "/tmp/repo/packages/server", - compare: { mode: "uncommitted" }, - }, - vi.fn(), - ); - - expect(getCheckoutDiff).toHaveBeenNthCalledWith( - 1, - "/tmp/repo", - expect.objectContaining({ mode: "uncommitted" }), - undefined, - ); - - const onChange = getOnChange(); - onChange?.(); - await vi.advanceTimersByTimeAsync(150); - - expect(getCheckoutDiff).toHaveBeenCalledTimes(2); - const watchFiredCall = getCheckoutDiff.mock.calls[1]; - expect(watchFiredCall[2]).toEqual({ - force: true, - reason: expect.stringContaining("working-tree"), - }); - }); - - test("an edit during an in-flight diff refresh produces one final follow-up", async () => { - const inFlightDiff = createDeferred<{ - diff: string; - structured: Array<{ - path: string; - additions: number; - deletions: number; - status: "modified"; - }>; - }>(); - const getCheckoutDiff = vi - .fn() - .mockResolvedValueOnce({ - diff: "", - structured: [{ path: "tracked.ts", additions: 1, deletions: 0, status: "modified" }], - }) - .mockImplementationOnce(() => inFlightDiff.promise) - .mockResolvedValue({ - diff: "", - structured: [{ path: "tracked.ts", additions: 100, deletions: 25, status: "modified" }], - }); - const { manager, getOnChange } = createManager({ - getCheckoutDiffImplementation: getCheckoutDiff, - }); - const listener = vi.fn(); - - await manager.subscribe({ cwd: "/tmp/repo", compare: { mode: "uncommitted" } }, listener); - getOnChange()?.(); - await vi.advanceTimersByTimeAsync(150); - expect(getCheckoutDiff).toHaveBeenCalledTimes(2); - - for (let event = 0; event < 100; event += 1) { - getOnChange()?.(); + await result.unsubscribe(); + } finally { + await modern.release(); } - await vi.advanceTimersByTimeAsync(150); - expect(getCheckoutDiff).toHaveBeenCalledTimes(2); - - inFlightDiff.resolve({ - diff: "", - structured: [{ path: "tracked.ts", additions: 2, deletions: 1, status: "modified" }], - }); - await vi.waitFor(() => { - expect(getCheckoutDiff).toHaveBeenCalledTimes(3); - expect(listener).toHaveBeenLastCalledWith({ - cwd: "/tmp/repo", - files: [{ path: "tracked.ts", additions: 100, deletions: 25, status: "modified" }], - error: null, - }); - }); }); +}, 30_000); - test("base diff subscriptions ignore ordinary working tree edits", async () => { - const getCheckoutDiff = vi.fn(async () => ({ - diff: "", - structured: [{ path: "committed.ts", additions: 1, deletions: 0, status: "modified" }], - })); - const { manager, getOnChange, mockRequestWorkingTreeWatch } = createManager({ - getCheckoutDiffImplementation: getCheckoutDiff, +async function waitForSubscriberCount(manager: CheckoutDiffManager, count: number): Promise { + await expect + .poll(() => manager.getMetrics().checkoutDiffSubscriptionCount, { timeout: 10_000 }) + .toBe(count); +} + +test("moving a tag refreshes binary file metadata even though binary files have no bodies", async () => { + await withCheckout(async ({ cwd, manager, git }) => { + git("tag", "v1"); + await writeFile(join(cwd, "image.bin"), Buffer.from([0, 1])); + git("add", "."); + git("-c", "commit.gpgsign=false", "commit", "-m", "binary"); + await writeFile(join(cwd, "image.bin"), Buffer.from([0, 2])); + const updates: CheckoutDiffSnapshotPayload[] = []; + const subscription = await manager.subscribe( + { cwd, compare: { mode: "all", baseRef: "v1" } }, + (snapshot) => updates.push(snapshot), + ); + expect(subscription.initial.files[0]).toMatchObject({ + path: "image.bin", + isNew: true, + status: "binary", }); - - await manager.subscribe( - { cwd: "/tmp/repo", compare: { mode: "base", baseRef: "main" } }, - vi.fn(), - ); - getOnChange()?.(); - await vi.advanceTimersByTimeAsync(150); - - expect(getCheckoutDiff).toHaveBeenCalledTimes(1); - expect(mockRequestWorkingTreeWatch).not.toHaveBeenCalled(); + git("update-ref", "refs/tags/v1", "HEAD"); + await expect.poll(() => updates.at(-1)?.files[0]?.isNew, { timeout: 10000 }).toBe(false); + await subscription.unsubscribe(); }); - - test("base diff subscriptions ignore worktree-only workspace snapshot updates", async () => { - const getCheckoutDiff = vi.fn(async () => ({ diff: "", structured: [] })); - const { manager, getOnWorkspaceSnapshot } = createManager({ - getCheckoutDiffImplementation: getCheckoutDiff, - }); - - await manager.subscribe( - { cwd: "/tmp/repo", compare: { mode: "base", baseRef: "main" } }, - vi.fn(), - ); - - getOnWorkspaceSnapshot()?.( - createWorkspaceSnapshot({ - isDirty: true, - diffStat: { additions: 5, deletions: 2 }, - }), - ); - await vi.advanceTimersByTimeAsync(150); - - expect(getCheckoutDiff).toHaveBeenCalledTimes(1); - }); - - test("base diff subscriptions refresh for structural workspace changes", async () => { - const getCheckoutDiff = vi.fn(async () => ({ diff: "", structured: [] })); - const { manager, getOnWorkspaceSnapshot, workspaceGitService } = createManager({ - getCheckoutDiffImplementation: getCheckoutDiff, - }); - - await manager.subscribe( - { cwd: "/tmp/repo", compare: { mode: "base", baseRef: "main" } }, - vi.fn(), - ); - - expect(workspaceGitService.registerWorkspace).toHaveBeenCalledTimes(1); - getOnWorkspaceSnapshot()?.( - createWorkspaceSnapshot({ - currentBranch: "feature-2", - isDirty: true, - diffStat: { additions: 5, deletions: 2 }, - }), - ); - await vi.advanceTimersByTimeAsync(150); - expect(getCheckoutDiff).toHaveBeenCalledTimes(2); - expect(getCheckoutDiff.mock.calls[1]?.[2]).toBeUndefined(); - - getOnWorkspaceSnapshot()?.(createWorkspaceSnapshot({ aheadBehind: { ahead: 2, behind: 0 } })); - await vi.advanceTimersByTimeAsync(150); - expect(getCheckoutDiff).toHaveBeenCalledTimes(3); - }); - - test("falls back to cwd when the working tree watch returns no repo root", async () => { - const { manager, workspaceGitService } = createManager({ repoRoot: null }); - - await manager.subscribe( - { - cwd: "/tmp/plain", - compare: { mode: "uncommitted" }, - }, - () => {}, - ); - - expect(workspaceGitService.getCheckoutDiff).toHaveBeenCalledWith( - "/tmp/plain", - expect.objectContaining({ mode: "uncommitted", includeStructured: true }), - undefined, - ); - }); -}); +}, 30000); diff --git a/packages/server/src/server/checkout-diff-manager.ts b/packages/server/src/server/checkout-diff-manager.ts index 8293ce965f..655ec91e9f 100644 --- a/packages/server/src/server/checkout-diff-manager.ts +++ b/packages/server/src/server/checkout-diff-manager.ts @@ -1,400 +1,171 @@ -import { type WorkspaceRuntimeStore } from "./workspace-runtime-store/index.js"; import type pino from "pino"; import type { SubscribeCheckoutDiffRequest, SessionOutboundMessage } from "./messages.js"; -import type { WorkspaceGitRuntimeSnapshot, WorkspaceGitService } from "./workspace-git-service.js"; +import type { WorkspaceGitService } from "./workspace-git-service.js"; +import type { WorkspaceRuntimeStore } from "./workspace-runtime-store/index.js"; +import { DiffSubscriptions, type CommitFileRead } from "./diff/index.js"; +import { + DiffOpcode, + materializeDiffFile, + type DiffFrame, + type DiffBegin, + type DiffBody, +} from "@getpaseo/protocol/binary-frames/index"; +import type { ParsedDiffFile } from "@getpaseo/protocol/messages"; +import { resolveCheckoutDiffBase } from "../utils/checkout-git.js"; import { expandTilde } from "../utils/path.js"; import { toCheckoutError } from "./checkout-git-utils.js"; -const CHECKOUT_DIFF_WATCH_DEBOUNCE_MS = 150; - -type CheckoutDiffWorkspace = Pick< - WorkspaceGitService, - "getCheckoutDiff" | "getSnapshot" | "registerWorkspace" | "requestWorkingTreeWatch" ->; - export type CheckoutDiffCompareInput = SubscribeCheckoutDiffRequest["compare"]; - export type CheckoutDiffSnapshotPayload = Omit< Extract["payload"], "subscriptionId" >; - export interface CheckoutDiffMetrics { checkoutDiffTargetCount: number; checkoutDiffSubscriptionCount: number; checkoutDiffWatcherCount: number; checkoutDiffFallbackRefreshTargetCount: number; } - -interface CheckoutDiffWatchTarget { - key: string; - cwd: string; - diffCwd: string; - compare: CheckoutDiffCompareInput; - listeners: Set<(snapshot: CheckoutDiffSnapshotPayload) => void>; - workingTreeWatchUnsubscribe: (() => void) | null; - workspaceGitUnsubscribe: (() => void) | null; - latestWorkspaceStructureFingerprint: string | null; - latestWorkspaceWorktreeFingerprint: string | null; - latestWorkspaceForgeFingerprint: string | null; - debounceTimer: NodeJS.Timeout | null; - pendingDebounceForce: boolean; - refreshPromise: Promise | null; - refreshQueued: boolean; - refreshQueuedForce: boolean; - latestPayload: CheckoutDiffSnapshotPayload | null; - latestFingerprint: string | null; - openPromise: Promise | null; -} - export interface CheckoutDiffSubscriptionRequest { cwd: string; compare: CheckoutDiffCompareInput; signal?: AbortSignal; } - export interface CheckoutDiffSubscription { initial: CheckoutDiffSnapshotPayload; - unsubscribe: () => void; + unsubscribe(): void; } +// COMPAT(diffSubscription): added in v0.15.0, remove after 2027-05-06. +// Old clients receive the same JSON snapshot while the shared engine owns reads and observation. export class CheckoutDiffManager { - private readonly workspaceGitService: CheckoutDiffWorkspace; - private readonly workspaceRuntimeStore: WorkspaceRuntimeStore; - private readonly targets = new Map(); - + readonly streams: DiffSubscriptions; + private readonly paseoHome: string; constructor(options: { logger: pino.Logger; paseoHome: string; - workspaceGitService: CheckoutDiffWorkspace; + workspaceGitService: Pick; workspaceRuntimeStore: WorkspaceRuntimeStore; }) { - this.workspaceGitService = options.workspaceGitService; - this.workspaceRuntimeStore = options.workspaceRuntimeStore; + this.streams = new DiffSubscriptions(options.workspaceGitService); + this.paseoHome = options.paseoHome; + } + readCommitFile(input: CommitFileRead) { + return this.streams.readCommitFile(input); } - async read( params: Omit, ): Promise { - return this.computeCheckoutDiffSnapshot(params.cwd, this.normalizeCompare(params.compare)); + const subscription = await this.subscribe(params, () => {}); + await subscription.unsubscribe(); + return subscription.initial; } - async subscribe( params: CheckoutDiffSubscriptionRequest, listener: (snapshot: CheckoutDiffSnapshotPayload) => void, ): Promise { - const cwd = params.cwd; - const compare = this.normalizeCompare(params.compare); - const target = this.ensureTarget(cwd, compare); - target.listeners.add(listener); - target.openPromise ??= this.openTarget(target); - - let isSubscribed = true; - const unsubscribe = () => { - if (!isSubscribed) { + const cwd = expandTilde(params.cwd); + let resolveInitial!: (snapshot: CheckoutDiffSnapshotPayload) => void; + const initial = new Promise((resolve) => { + resolveInitial = resolve; + }); + let delivered = false; + let pending: DiffBegin | null = null; + let files = new Map(); + let bodies = new Map(); + const receive = async (frame: DiffFrame) => { + if (params.signal?.aborted) return; + if (frame.opcode === DiffOpcode.Begin) { + pending = frame; + bodies = new Map(); return; } - isSubscribed = false; - params.signal?.removeEventListener("abort", unsubscribe); - this.removeListener(target, listener); - }; - params.signal?.addEventListener("abort", unsubscribe, { once: true }); - if (params.signal?.aborted) { - unsubscribe(); - } - - try { - await target.openPromise; - const initial = - target.latestPayload ?? - (await this.computeCheckoutDiffSnapshot(target.cwd, target.compare, { - diffCwd: target.diffCwd, - })); - target.latestPayload = initial; - target.latestFingerprint = JSON.stringify(initial); - return { initial, unsubscribe }; - } catch (error) { - unsubscribe(); - throw error; - } - } - - scheduleRefreshForCwd(cwd: string): void { - const resolvedCwd = expandTilde(cwd); - for (const target of this.targets.values()) { - if (target.cwd !== resolvedCwd && target.diffCwd !== resolvedCwd) { - continue; + if (!pending || frame.revision !== pending.revision) return; + if (frame.opcode === DiffOpcode.Bodies) { + for (const body of frame.files) bodies.set(body.path, body); + return; } - this.scheduleTargetRefresh(target); + let payload: CheckoutDiffSnapshotPayload; + if (frame.error || frame.diffTooLarge) { + const message = frame.error ? frame.error.message : "Diff too large to display"; + payload = { + cwd, + files: [], + error: toCheckoutError(new Error(message)), + diffTooLarge: frame.diffTooLarge, + }; + files.clear(); + } else { + const next = new Map(); + 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; + next.set(entry.path, file); + } + files = next; + payload = { cwd, files: [...next.values()], error: null }; + } + pending = null; + if (delivered) listener(payload); + else { + delivered = true; + resolveInitial(payload); + } + }; + try { + const baseRef = await resolveCheckoutDiffBase(cwd, params.compare, { + paseoHome: this.paseoHome, + }); + if (!baseRef) return { initial: { cwd, files: [], error: null }, unsubscribe() {} }; + const subscription = await this.streams.subscribe( + { + cwd, + base: { kind: "rev", rev: baseRef }, + head: + params.compare.mode === "base" ? { kind: "rev", rev: "HEAD" } : { kind: "worktree" }, + mergeBase: params.compare.mode !== "uncommitted", + ignoreWhitespace: params.compare.ignoreWhitespace, + }, + receive, + ); + const unsubscribe = async () => { + if (!delivered) { + delivered = true; + resolveInitial({ + cwd, + files: [], + error: toCheckoutError(new Error("Diff subscription cancelled")), + }); + } + params.signal?.removeEventListener("abort", unsubscribe); + await subscription.release(); + }; + params.signal?.addEventListener("abort", unsubscribe, { once: true }); + if (params.signal?.aborted) { + await subscription.release(); + throw new Error("Diff subscription cancelled"); + } + await subscription.start(); + return { initial: await initial, unsubscribe }; + } catch (error) { + return { initial: { cwd, files: [], error: toCheckoutError(error) }, unsubscribe() {} }; } } - + scheduleRefreshForCwd(cwd: string): void { + this.streams.invalidateForCwd(expandTilde(cwd)); + } getMetrics(): CheckoutDiffMetrics { - let checkoutDiffSubscriptionCount = 0; - - for (const target of this.targets.values()) { - checkoutDiffSubscriptionCount += target.listeners.size; - } - + const metrics = this.streams.getMetrics(); return { - checkoutDiffTargetCount: this.targets.size, - checkoutDiffSubscriptionCount, - checkoutDiffWatcherCount: 0, + checkoutDiffTargetCount: metrics.targets, + checkoutDiffSubscriptionCount: metrics.subscriptions, + checkoutDiffWatcherCount: metrics.metadataObservers, checkoutDiffFallbackRefreshTargetCount: 0, }; } - - dispose(): void { - for (const target of this.targets.values()) { - this.closeTarget(target); - } - this.targets.clear(); - } - - private normalizeCompare(compare: CheckoutDiffCompareInput): CheckoutDiffCompareInput { - const ignoreWhitespace = compare.ignoreWhitespace === true; - if (compare.mode === "uncommitted") { - return { mode: "uncommitted", ignoreWhitespace }; - } - const trimmedBaseRef = compare.baseRef?.trim(); - return trimmedBaseRef - ? { mode: compare.mode, baseRef: trimmedBaseRef, ignoreWhitespace } - : { mode: compare.mode, ignoreWhitespace }; - } - - private buildTargetKey(cwd: string, compare: CheckoutDiffCompareInput): string { - return JSON.stringify([ - cwd, - compare.mode, - compare.mode !== "uncommitted" ? (compare.baseRef ?? "") : "", - compare.ignoreWhitespace === true, - ]); - } - - private closeTarget(target: CheckoutDiffWatchTarget): void { - if (target.debounceTimer) { - clearTimeout(target.debounceTimer); - target.debounceTimer = null; - } - target.workingTreeWatchUnsubscribe?.(); - target.workingTreeWatchUnsubscribe = null; - target.workspaceGitUnsubscribe?.(); - target.workspaceGitUnsubscribe = null; - target.listeners.clear(); - } - - private rememberWorkspaceSnapshot( - target: CheckoutDiffWatchTarget, - snapshot: WorkspaceGitRuntimeSnapshot, - ): void { - const structureFingerprint = JSON.stringify({ - isGit: snapshot.git.isGit, - repoRoot: snapshot.git.repoRoot, - mainRepoRoot: snapshot.git.mainRepoRoot, - currentBranch: snapshot.git.currentBranch, - remoteUrl: snapshot.git.remoteUrl, - isPaseoOwnedWorktree: snapshot.git.isPaseoOwnedWorktree, - baseRef: snapshot.git.baseRef, - baseRefExact: snapshot.git.baseRefExact, - aheadBehind: snapshot.git.aheadBehind, - aheadOfOrigin: snapshot.git.aheadOfOrigin, - behindOfOrigin: snapshot.git.behindOfOrigin, - hasRemote: snapshot.git.hasRemote, - }); - const worktreeFingerprint = JSON.stringify({ - isDirty: snapshot.git.isDirty, - diffStat: snapshot.git.diffStat, - }); - const forgeFingerprint = JSON.stringify(snapshot.forge); - const previousStructureFingerprint = target.latestWorkspaceStructureFingerprint; - const previousWorktreeFingerprint = target.latestWorkspaceWorktreeFingerprint; - const previousForgeFingerprint = target.latestWorkspaceForgeFingerprint; - target.latestWorkspaceStructureFingerprint = structureFingerprint; - target.latestWorkspaceWorktreeFingerprint = worktreeFingerprint; - target.latestWorkspaceForgeFingerprint = forgeFingerprint; - - if (previousStructureFingerprint === null) { - return; - } - const structureChanged = structureFingerprint !== previousStructureFingerprint; - const worktreeChanged = worktreeFingerprint !== previousWorktreeFingerprint; - const forgeChanged = forgeFingerprint !== previousForgeFingerprint; - if (structureChanged || (!worktreeChanged && !forgeChanged)) { - this.scheduleTargetRefresh(target, false); - } - } - - private removeListener( - target: CheckoutDiffWatchTarget, - listener: (snapshot: CheckoutDiffSnapshotPayload) => void, - ): void { - target.listeners.delete(listener); - if (target.listeners.size > 0) { - return; - } - this.closeTarget(target); - if (this.targets.get(target.key) === target) { - this.targets.delete(target.key); - } - } - - private scheduleTargetRefresh(target: CheckoutDiffWatchTarget, force = true): void { - target.pendingDebounceForce ||= force; - if (target.debounceTimer) { - clearTimeout(target.debounceTimer); - } - target.debounceTimer = setTimeout(() => { - target.debounceTimer = null; - const pendingForce = target.pendingDebounceForce; - target.pendingDebounceForce = false; - void this.refreshTarget(target, pendingForce); - }, CHECKOUT_DIFF_WATCH_DEBOUNCE_MS); - } - - private async computeCheckoutDiffSnapshot( - cwd: string, - compare: CheckoutDiffCompareInput, - options?: { diffCwd?: string; force?: boolean; reason?: string }, - ): Promise { - const diffCwd = options?.diffCwd ?? cwd; - try { - const diffResult = await this.workspaceGitService.getCheckoutDiff( - diffCwd, - { - mode: compare.mode, - baseRef: compare.baseRef, - ignoreWhitespace: compare.ignoreWhitespace, - includeStructured: true, - }, - options?.force - ? { force: true, reason: options.reason ?? "checkout-diff-refresh" } - : undefined, - ); - if (diffResult.diffTooLarge) { - return { - cwd, - files: [], - diffTooLarge: true, - error: toCheckoutError(new Error("Diff too large to display")), - }; - } - const files = [...(diffResult.structured ?? [])]; - files.sort((a, b) => { - if (a.path === b.path) return 0; - return a.path < b.path ? -1 : 1; - }); - return { - cwd, - files, - error: null, - }; - } catch (error) { - return { - cwd, - files: [], - error: toCheckoutError(error), - }; - } - } - - private async refreshTarget(target: CheckoutDiffWatchTarget, force: boolean): Promise { - if (target.refreshPromise) { - target.refreshQueued = true; - target.refreshQueuedForce ||= force; - return; - } - - target.refreshPromise = (async () => { - let currentForce = force; - do { - target.refreshQueued = false; - target.refreshQueuedForce = false; - const snapshot = await this.computeCheckoutDiffSnapshot(target.cwd, target.compare, { - diffCwd: target.diffCwd, - force: currentForce, - ...(currentForce ? { reason: "working-tree-watch" } : {}), - }); - target.latestPayload = snapshot; - const fingerprint = JSON.stringify(snapshot); - if (fingerprint !== target.latestFingerprint) { - target.latestFingerprint = fingerprint; - for (const listener of target.listeners) { - listener(snapshot); - } - } - currentForce = target.refreshQueuedForce; - } while (target.refreshQueued); - })(); - - try { - await target.refreshPromise; - } finally { - target.refreshPromise = null; - } - } - - private ensureTarget(cwd: string, compare: CheckoutDiffCompareInput): CheckoutDiffWatchTarget { - const targetKey = this.buildTargetKey(cwd, compare); - const existing = this.targets.get(targetKey); - if (existing) { - return existing; - } - - const target: CheckoutDiffWatchTarget = { - key: targetKey, - cwd, - diffCwd: cwd, - compare, - listeners: new Set(), - workingTreeWatchUnsubscribe: null, - workspaceGitUnsubscribe: null, - latestWorkspaceStructureFingerprint: null, - latestWorkspaceWorktreeFingerprint: null, - latestWorkspaceForgeFingerprint: null, - debounceTimer: null, - pendingDebounceForce: false, - refreshPromise: null, - refreshQueued: false, - refreshQueuedForce: false, - latestPayload: null, - latestFingerprint: null, - openPromise: null, - }; - this.targets.set(targetKey, target); - return target; - } - - private async openTarget(target: CheckoutDiffWatchTarget): Promise { - if (target.compare.mode !== "uncommitted") { - const snapshot = - this.workspaceRuntimeStore.get(target.cwd) ?? - (await this.workspaceGitService.getSnapshot(target.cwd, { includeForge: false })); - target.diffCwd = snapshot.git.repoRoot ?? target.cwd; - if (this.targets.get(target.key) !== target || target.listeners.size === 0) { - return; - } - this.rememberWorkspaceSnapshot(target, snapshot); - const workspaceSubscription = this.workspaceGitService.registerWorkspace( - { cwd: target.cwd }, - (nextSnapshot) => this.rememberWorkspaceSnapshot(target, nextSnapshot), - ); - if (this.targets.get(target.key) !== target || target.listeners.size === 0) { - workspaceSubscription.unsubscribe(); - return; - } - target.workspaceGitUnsubscribe = workspaceSubscription.unsubscribe; - if (target.compare.mode === "base") return; - } - - const { repoRoot, unsubscribe } = await this.workspaceGitService.requestWorkingTreeWatch( - target.cwd, - () => this.scheduleTargetRefresh(target), - ); - target.diffCwd = repoRoot ?? target.cwd; - if (this.targets.get(target.key) !== target || target.listeners.size === 0) { - unsubscribe(); - return; - } - target.workingTreeWatchUnsubscribe = unsubscribe; + async dispose(): Promise { + await this.streams.close(); } } diff --git a/packages/server/src/server/daemon-e2e/checkout-diff-content.e2e.test.ts b/packages/server/src/server/daemon-e2e/checkout-diff-content.e2e.test.ts index 56e7cdefb8..c1be3b873c 100644 --- a/packages/server/src/server/daemon-e2e/checkout-diff-content.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/checkout-diff-content.e2e.test.ts @@ -293,3 +293,126 @@ test("large source diffs stay renderable for reads, subscriptions, and commit fi await rm(cwd, { recursive: true, force: true, maxRetries: 10, retryDelay: 100 }); } }, 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 | null = null; + try { + git(cwd, "init", "-b", "main"); + git(cwd, "config", "user.name", "Test"); + git(cwd, "config", "user.email", "test@example.com"); + writeFileSync(join(cwd, "README"), "base\n"); + git(cwd, "add", "."); + git(cwd, "-c", "commit.gpgsign=false", "commit", "-m", "base"); + for (let file = 0; file < 252; file++) { + const content = + Array.from( + { length: 90 }, + (_, line) => + `export const fn${line} = (param: string) => ({ key: param, value: ${file} });`, + ).join("\n") + "\n"; + writeFileSync(join(cwd, `${file}.ts`), content); + } + expect(ctx.client.getLastServerInfoMessage()).toMatchObject({ + features: { diffSubscription: true }, + }); + 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((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((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; + expect(entryCount).toBe(252); + expect(bodyFrames).toBeGreaterThan(1); + expect(pongDuringRevision).toBe(true); + const artifactDirectory = join( + import.meta.dirname, + "../../../../../.artifacts/diff-subscription", + ); + mkdirSync(artifactDirectory, { recursive: true }); + writeFileSync( + join(artifactDirectory, "wire-metrics.json"), + JSON.stringify({ legacyJsonBytes, encodedBytes, bodyFrames, pongDuringRevision }, null, 2), + ); + } finally { + socket?.terminate(); + await ctx.cleanup(); + rmSync(cwd, { recursive: true, force: true }); + } +}, 60_000); diff --git a/packages/server/src/server/diff/batches.ts b/packages/server/src/server/diff/batches.ts new file mode 100644 index 0000000000..25628eed57 --- /dev/null +++ b/packages/server/src/server/diff/batches.ts @@ -0,0 +1,30 @@ +import { encodeDiffFrame, DiffOpcode, type DiffBody } from "@getpaseo/protocol/binary-frames/index"; + +export async function* packBodies( + files: AsyncIterable | Iterable, + revision: number, +): AsyncGenerator { + let batch: DiffBody[] = []; + let size = 0; + for await (const file of files) { + const bytes = encodeDiffFrame({ + opcode: DiffOpcode.Bodies, + slot: 0, + revision, + files: [file], + }).length; + if (batch.length && size + bytes > 256 * 1024) { + yield batch; + batch = []; + size = 0; + } + batch.push(file); + size += bytes; + if (size >= 256 * 1024) { + yield batch; + batch = []; + size = 0; + } + } + if (batch.length) yield batch; +} diff --git a/packages/server/src/server/diff/dependencies.ts b/packages/server/src/server/diff/dependencies.ts new file mode 100644 index 0000000000..bbe060c37d --- /dev/null +++ b/packages/server/src/server/diff/dependencies.ts @@ -0,0 +1,76 @@ +import { classifyGitMetadataPath } from "../git-metadata-event-rules.js"; +import { resolve } from "node:path"; +import { runGitCommand } from "../../utils/run-git-command.js"; +import type { WorkspaceGitService } from "../workspace-git-service.js"; +import { createFileObserver, type FileObserverSubscription } from "../file-observer/index.js"; + +export interface DiffDependencies { + cwd: string; + refs: string[]; + index: boolean; + worktree: boolean; +} +// Own metadata observations here; comparison managers declare inputs rather than watching paths. +export class DiffDependencyObserver { + private observer = createFileObserver(); + constructor(private workspaceGit: Pick) {} + async declareDependencies( + input: DiffDependencies, + changed: () => void, + ): Promise<() => Promise> { + if (input.refs.length === 0 && !input.index && !input.worktree) return async () => {}; + let polling: ReturnType | null = null; + const subscriptions: FileObserverSubscription[] = []; + let workingTree: Awaited> | null = + null; + try { + const paths = await runGitCommand(["rev-parse", "--absolute-git-dir", "--git-common-dir"], { + cwd: input.cwd, + }); + const roots = new Set( + paths.stdout + .trim() + .split("\n") + .map((path) => resolve(input.cwd, path)), + ); + for (const root of roots) { + const subscription = await this.observer.subscribe( + root, + (error, events) => { + if (error) { + changed(); + polling ??= setInterval(changed, 5000); + return; + } + const affectsInput = events.some((event) => { + const path = event.path.slice(root.length + 1).replaceAll("\\", "/"); + if (path === "index") return input.index; + return classifyGitMetadataPath("common", path).kind !== "ignore"; + }); + if (affectsInput) changed(); + }, + { ignore: [resolve(root, "objects"), resolve(root, "logs"), resolve(root, "hooks")] }, + ); + subscriptions.push(subscription); + } + if (input.worktree) + workingTree = await this.workspaceGit.requestWorkingTreeWatch(input.cwd, changed); + return async () => { + if (polling) clearInterval(polling); + await workingTree?.unsubscribe(); + await Promise.all(subscriptions.map((subscription) => subscription.unsubscribe())); + }; + } catch (error) { + if (polling) clearInterval(polling); + await workingTree?.unsubscribe(); + await Promise.all(subscriptions.map((subscription) => subscription.unsubscribe())); + throw error; + } + } + getObservationCount(): number { + return this.observer.getDiagnostics().activeObservationCount; + } + async close(): Promise { + await this.observer.close(); + } +} diff --git a/packages/server/src/server/diff/git.test.ts b/packages/server/src/server/diff/git.test.ts new file mode 100644 index 0000000000..3abae1ccc5 --- /dev/null +++ b/packages/server/src/server/diff/git.test.ts @@ -0,0 +1,242 @@ +import { execFileSync } from "node:child_process"; +import { mkdtemp, writeFile, rm, readFile, lstat } from "node:fs/promises"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import { expect, test } from "vitest"; +import { listComparison, type Comparison } from "./git.js"; + +test("working tree comparisons preserve staged deletions whose file remains on disk", async () => { + const cwd = await mkdtemp(join(tmpdir(), "diff-comparison-")); + const git = (...args: string[]) => execFileSync("git", args, { cwd, encoding: "utf8" }).trim(); + try { + git("init", "-b", "main"); + git("config", "user.email", "test@example.com"); + git("config", "user.name", "Test"); + await writeFile(join(cwd, "x.ts"), "const value = 1;\n"); + git("add", "."); + git("-c", "commit.gpgsign=false", "commit", "-m", "base"); + git("rm", "--cached", "x.ts"); + await writeFile(join(cwd, "x.ts"), "const value = 2;\n"); + const comparison = await listComparison({ + cwd, + base: { kind: "rev", rev: "HEAD" }, + head: { kind: "worktree" }, + highlight: false, + }); + expect(comparison.files.map((file) => file.entry)).toEqual([ + { + path: "x.ts", + oldPath: null, + change: "modified", + content: "text", + additions: 1, + deletions: 1, + body: true, + }, + ]); + const body = await readBody(comparison); + expect(body.body.hunks[0].lines).toEqual([ + { type: "header", content: "@@ -1 +1 @@" }, + { type: "remove", content: "const value = 1;" }, + { type: "add", content: "const value = 2;" }, + ]); + expect(comparison.metrics.highlightCalls).toBe(0); + await comparison.close(); + expect(git("diff", "--cached", "--name-status")).toBe("D\tx.ts"); + } finally { + await rm(cwd, { recursive: true, force: true }); + } +}); + +test("supports empty trees, unborn HEAD, staged and unstaged comparisons", async () => { + const cwd = await mkdtemp(join(tmpdir(), "diff-endpoints-")); + const git = (...args: string[]) => execFileSync("git", args, { cwd, encoding: "utf8" }).trim(); + try { + git("init", "-b", "main"); + git("config", "user.email", "test@example.com"); + git("config", "user.name", "Test"); + await writeFile(join(cwd, "x.ts"), "const value = 1;\n"); + const unborn = await listComparison({ + cwd, + base: { kind: "rev", rev: "HEAD" }, + head: { kind: "worktree" }, + }); + expect(unborn.files.map((file) => file.entry.path)).toEqual(["x.ts"]); + await unborn.close(); + git("add", "."); + git("-c", "commit.gpgsign=false", "commit", "-m", "base"); + const empty = await listComparison({ + cwd, + base: { kind: "empty" }, + head: { kind: "rev", rev: "HEAD" }, + }); + expect(empty.files[0].entry.change).toBe("added"); + await empty.close(); + await writeFile(join(cwd, "x.ts"), "const value = 2;\n"); + git("add", "."); + const indexPath = join(cwd, ".git", "index"); + const originalIndex = await readFile(indexPath); + const originalMtime = (await lstat(indexPath, { bigint: true })).mtimeNs; + const staged = await listComparison({ + cwd, + base: { kind: "rev", rev: "HEAD" }, + head: { kind: "index" }, + highlight: false, + }); + expect(await readFile(indexPath)).toEqual(originalIndex); + expect((await lstat(indexPath, { bigint: true })).mtimeNs).toBe(originalMtime); + const stagedSha = staged.headSha; + expect(staged.files[0].entry).toMatchObject({ additions: 1, deletions: 1 }); + await staged.close(); + await writeFile(join(cwd, "x.ts"), "const value = 3;\n"); + const unstaged = await listComparison({ + cwd, + base: { kind: "index" }, + head: { kind: "worktree" }, + highlight: false, + }); + expect(unstaged.baseSha).toBe(stagedSha); + const record = await readBody(unstaged); + expect(record.body.hunks[0].lines).toContainEqual({ + type: "remove", + content: "const value = 2;", + }); + await unstaged.close(); + } finally { + await rm(cwd, { recursive: true, force: true }); + } +}); + +test("lists oversized untracked files without treating their truncated contents as additions", async () => { + const cwd = await mkdtemp(join(tmpdir(), "diff-large-file-")); + const git = (...args: string[]) => execFileSync("git", args, { cwd, encoding: "utf8" }).trim(); + try { + git("init", "-b", "main"); + git("config", "user.email", "test@example.com"); + git("config", "user.name", "Test"); + await writeFile(join(cwd, "README"), "base\n"); + git("add", "."); + git("-c", "commit.gpgsign=false", "commit", "-m", "base"); + await writeFile(join(cwd, "huge.ts"), Buffer.alloc(2 * 1024 * 1024, 120)); + const comparison = await listComparison({ + cwd, + base: { kind: "rev", rev: "HEAD" }, + head: { kind: "worktree" }, + }); + expect(comparison.files[0].entry).toMatchObject({ + path: "huge.ts", + additions: 0, + deletions: 0, + content: "text", + }); + expect(await comparison.read(comparison.files[0])).toEqual({ kind: "too_large" }); + await comparison.close(); + } finally { + await rm(cwd, { recursive: true, force: true }); + } +}); + +test("preserves highlighting context outside the emitted hunks", async () => { + const cwd = await mkdtemp(join(tmpdir(), "diff-highlight-context-")); + const git = (...args: string[]) => execFileSync("git", args, { cwd, encoding: "utf8" }).trim(); + try { + git("init", "-b", "main"); + git("config", "user.email", "test@example.com"); + git("config", "user.name", "Test"); + const lines = [ + "/*", + ...Array.from({ length: 30 }, (_, index) => `comment ${index}`), + "*/", + "export const value = 1;", + ]; + await writeFile(join(cwd, "x.ts"), lines.join("\n") + "\n"); + git("add", "."); + git("-c", "commit.gpgsign=false", "commit", "-m", "base"); + lines[15] = "new comment"; + await writeFile(join(cwd, "x.ts"), lines.join("\n") + "\n"); + const comparison = await listComparison({ + cwd, + base: { kind: "rev", rev: "HEAD" }, + head: { kind: "worktree" }, + }); + const record = await readBody(comparison); + expect(record.body.hunks[0].lines.find((line) => line.type === "add")).toMatchObject({ + content: "new comment", + runs: [{ style: "comment", length: 11 }], + }); + await comparison.close(); + } finally { + await rm(cwd, { recursive: true, force: true }); + } +}); + +test("reads patches independently of repository color and prefix presentation settings", async () => { + const cwd = await mkdtemp(join(tmpdir(), "diff-presentation-")); + const git = (...args: string[]) => execFileSync("git", args, { cwd, encoding: "utf8" }).trim(); + try { + git("init", "-b", "main"); + git("config", "user.email", "test@example.com"); + git("config", "user.name", "Test"); + await writeFile(join(cwd, "x.ts"), "before\n"); + git("add", "."); + git("-c", "commit.gpgsign=false", "commit", "-m", "base"); + git("config", "color.ui", "always"); + git("config", "diff.noprefix", "true"); + await writeFile(join(cwd, "x.ts"), "after\n"); + const comparison = await listComparison({ + cwd, + base: { kind: "rev", rev: "HEAD" }, + head: { kind: "worktree" }, + highlight: false, + }); + const record = await readBody(comparison); + expect(record.body.hunks).toHaveLength(1); + expect(record.body.hunks[0].lines).toContainEqual({ type: "add", content: "after" }); + await comparison.close(); + } finally { + await rm(cwd, { recursive: true, force: true }); + } +}); + +test("merge-base index comparisons resolve against HEAD while reporting both endpoint SHAs", async () => { + const cwd = await mkdtemp(join(tmpdir(), "diff-index-merge-base-")); + const git = (...args: string[]) => execFileSync("git", args, { cwd, encoding: "utf8" }).trim(); + try { + git("init", "-b", "main"); + git("config", "user.email", "test@example.com"); + git("config", "user.name", "Test"); + await writeFile(join(cwd, "README"), "base\n"); + git("add", "."); + git("-c", "commit.gpgsign=false", "commit", "-m", "base"); + const common = git("rev-parse", "HEAD"); + git("checkout", "-b", "side"); + await writeFile(join(cwd, "side.txt"), "side\n"); + git("add", "."); + git("-c", "commit.gpgsign=false", "commit", "-m", "side"); + const side = git("rev-parse", "HEAD"); + git("checkout", "main"); + await writeFile(join(cwd, "main.txt"), "main\n"); + git("add", "."); + git("-c", "commit.gpgsign=false", "commit", "-m", "main"); + await writeFile(join(cwd, "staged.txt"), "staged\n"); + git("add", "staged.txt"); + const comparison = await listComparison({ + cwd, + base: { kind: "rev", rev: "side" }, + head: { kind: "index" }, + mergeBase: true, + }); + expect(comparison.baseSha).toBe(side); + expect(comparison.mergeBaseSha).toBe(common); + expect(comparison.files.map((file) => file.entry.path)).toEqual(["main.txt", "staged.txt"]); + await comparison.close(); + } finally { + await rm(cwd, { recursive: true, force: true }); + } +}); + +async function readBody(comparison: Comparison) { + const record = await comparison.read(comparison.files[0]); + if (record.kind !== "ready") throw new Error("Expected a readable file body"); + return record; +} diff --git a/packages/server/src/server/diff/git.ts b/packages/server/src/server/diff/git.ts new file mode 100644 index 0000000000..80b1e2ea0a --- /dev/null +++ b/packages/server/src/server/diff/git.ts @@ -0,0 +1,339 @@ +import { HighlightSources } from "./sources.js"; +import { + prepareAllChangesIndex, + PER_FILE_DIFF_MAX_BYTES, + GIT_PATCH_FORMAT_ARGS, + inspectUntrackedFile, + snapshotComparisonIndex, +} from "../../utils/checkout-git.js"; +import { lstat, readFile, readlink } from "node:fs/promises"; +import { resolve } from "node:path"; +import type { DiffEndpoint, DiffSubscribeRequest } from "@getpaseo/protocol/messages"; +import type { DiffEntry, DiffBody } from "@getpaseo/protocol/binary-frames/index"; +import { runGitCommand, type GitCommandOptions } from "../../utils/run-git-command.js"; +import { parseDiff, highlightDiffWithFileContent } from "../utils/diff-highlighter.js"; + +export type Compare = Omit; +export interface ListedFile { + entry: DiffEntry; + identity: string; + source: { kind: "tracked" } | { kind: "untracked"; size: number }; +} +interface ResolvedComparison { + baseSha: string; + headSha: string; + mergeBaseSha: string; + comparisonBaseSha: string; +} +export type DiffFileRead = { kind: "ready"; body: DiffBody; bytes: number } | { kind: "too_large" }; +export interface Comparison extends Pick< + ResolvedComparison, + "baseSha" | "headSha" | "mergeBaseSha" +> { + files: ListedFile[]; + metrics: { highlightCalls: number }; + close(): Promise; + read(file: ListedFile, budget?: number): Promise; +} +export class DiffReadError extends Error { + constructor( + readonly code: "INVALID_COMPARE" | "NOT_GIT_REPO" | "REF_NOT_FOUND", + message: string, + ) { + super(message); + } +} +const EMPTY = "4b825dc642cb6eb9a060e54bf8d69288fbee4904"; +export function validateCompare(compare: Compare): void { + if (compare.head.kind === "empty") + throw new DiffReadError("INVALID_COMPARE", "The head cannot be empty"); + if (compare.base.kind === "worktree") + throw new DiffReadError("INVALID_COMPARE", "The working tree cannot be the base"); + if (compare.base.kind === "index" && compare.head.kind !== "worktree") + throw new DiffReadError("INVALID_COMPARE", "An index base requires a working tree head"); + if (compare.mergeBase && compare.base.kind !== "rev") + throw new DiffReadError("INVALID_COMPARE", "Merge base requires a revision base"); +} +export async function worktreeRoot(cwd: string): Promise { + const result = await runGitCommand(["rev-parse", "--show-toplevel"], { + cwd, + acceptExitCodes: [0, 128], + }); + if (result.exitCode !== 0) throw new DiffReadError("NOT_GIT_REPO", "Not a git repository"); + return result.stdout.trim(); +} +async function resolveEndpoint( + cwd: string, + endpoint: DiffEndpoint, + indexFile: string | null, +): Promise { + if (endpoint.kind === "empty") return EMPTY; + if (endpoint.kind === "index") { + if (indexFile === null) throw new Error("Index comparisons require an index snapshot"); + const result = await runGitCommand(["write-tree"], { + cwd, + envOverlay: { GIT_INDEX_FILE: indexFile }, + }); + return result.stdout.trim(); + } + const ref = endpoint.kind === "rev" ? endpoint.rev : "HEAD"; + const result = await runGitCommand( + ["rev-parse", "--verify", "--end-of-options", `${ref}^{commit}`], + { + cwd, + acceptExitCodes: [0, 128], + }, + ); + if (result.exitCode === 0) return result.stdout.trim(); + if (ref === "HEAD") { + const unborn = await runGitCommand(["rev-parse", "--verify", "HEAD"], { + cwd, + acceptExitCodes: [0, 128], + }); + if (unborn.exitCode !== 0) return EMPTY; + } + throw new DiffReadError("REF_NOT_FOUND", `Revision not found: ${ref}`); +} +function parseStats( + text: string, +): Map { + const fields = text.split("\0"); + const stats = new Map(); + for (let index = 0; index < fields.length - 1; index++) { + const line = fields[index]; + const tab = line.indexOf("\t"); + const second = line.indexOf("\t", tab + 1); + const add = line.slice(0, tab); + const del = line.slice(tab + 1, second); + let path = line.slice(second + 1); + if (!path) { + index++; + path = fields[++index]; + } + stats.set(path, { + additions: Number(add) || 0, + deletions: Number(del) || 0, + binary: add === "-", + }); + } + return stats; +} +export async function listComparison(compare: Compare): Promise { + validateCompare(compare); + const cwd = compare.cwd; + let privateIndex: Awaited> = null; + try { + if (compare.base.kind === "index" || compare.head.kind === "index") + privateIndex = await snapshotComparisonIndex(cwd); + const indexFile = privateIndex ? privateIndex.indexFile : null; + const baseSha = await resolveEndpoint(cwd, compare.base, indexFile); + const headSha = await resolveEndpoint(cwd, compare.head, indexFile); + let mergeBaseSha = ""; + let comparisonBaseSha = baseSha; + if (compare.mergeBase) { + const headCommit = + compare.head.kind === "index" + ? await resolveEndpoint(cwd, { kind: "rev", rev: "HEAD" }, indexFile) + : headSha; + if (headCommit === EMPTY) comparisonBaseSha = EMPTY; + else if (baseSha !== EMPTY) { + const result = await runGitCommand(["merge-base", baseSha, headCommit], { + cwd, + acceptExitCodes: [0, 1], + }); + mergeBaseSha = result.stdout.trim(); + if (mergeBaseSha) comparisonBaseSha = mergeBaseSha; + } + } + if (compare.head.kind === "worktree" && compare.base.kind !== "index") + privateIndex = await prepareAllChangesIndex(cwd, comparisonBaseSha); + return await readComparisonList( + compare, + { baseSha, headSha, mergeBaseSha, comparisonBaseSha }, + privateIndex, + ); + } catch (error) { + await privateIndex?.cleanup(); + throw error; + } +} + +async function readComparisonList( + compare: Compare, + resolved: ResolvedComparison, + privateIndex: Awaited>, +): Promise { + const { baseSha, headSha, mergeBaseSha, comparisonBaseSha } = resolved; + const cwd = compare.cwd; + const args = diffArguments(compare, comparisonBaseSha, headSha); + const options: GitCommandOptions = { + cwd, + envOverlay: { + GIT_OPTIONAL_LOCKS: "0", + ...(privateIndex ? { GIT_INDEX_FILE: privateIndex.indexFile } : {}), + }, + }; + const [raw, stat] = await Promise.all([ + runGitCommand([...args, "--raw", "--no-abbrev", "-z"], options), + runGitCommand([...args, "--numstat", "-z"], options), + ]); + if (raw.truncated || stat.truncated) + throw new Error("Diff entry list exceeded the git output budget"); + const stats = parseStats(stat.stdout); + const fields = raw.stdout.split("\0"); + const files: ListedFile[] = []; + for (let index = 0; index < fields.length - 1; index++) { + const header = fields[index].split(" "); + const code = header[4]?.[0]; + let path = fields[++index]; + let oldPath: string | null = null; + if (code === "R" || code === "C") { + oldPath = path; + path = fields[++index]; + } + const counts = stats.get(path); + if (!counts) throw new Error(`Missing diff statistics for ${path}`); + let change: DiffEntry["change"] = "modified"; + if (code === "A") change = "added"; + else if (code === "D") change = "deleted"; + else if (oldPath !== null) change = "renamed"; + let identity = header.join(" "); + if (compare.head.kind === "worktree" && code !== "D") { + const value = await lstat(resolve(cwd, path), { bigint: true }); + identity += `:${value.size}:${value.mtimeNs}`; + } + files.push({ + entry: { + path, + oldPath, + change, + content: counts.binary ? "binary" : "text", + additions: counts.additions, + deletions: counts.deletions, + body: !counts.binary, + }, + identity, + source: { kind: "tracked" }, + }); + } + if (compare.head.kind === "worktree") files.push(...(await listUntrackedFiles(options))); + files.sort((a, b) => a.entry.path.localeCompare(b.entry.path)); + const metrics = { highlightCalls: 0 }; + const sources = new HighlightSources({ + cwd, + baseSha: comparisonBaseSha, + headSha: compare.head.kind === "worktree" ? null : headSha, + files, + }); + return { + baseSha, + headSha, + mergeBaseSha, + files, + metrics, + close: async () => { + await privateIndex?.cleanup(); + }, + async read(file, budget = PER_FILE_DIFF_MAX_BYTES) { + 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) { + return { kind: "too_large" }; + } + const command = + file.source.kind === "untracked" + ? ["diff", "--no-index", ...GIT_PATCH_FORMAT_ARGS, "--", "/dev/null", file.entry.path] + : [...args, "--patch", "--", ...paths.map((path) => `:(literal)${path}`)]; + const result = await runGitCommand(command, { + ...options, + acceptExitCodes: [0, 1], + maxOutputBytes: PER_FILE_DIFF_MAX_BYTES + 1, + }); + const bytes = Buffer.byteLength(result.stdout); + if (result.truncated || bytes > budget || bytes > PER_FILE_DIFF_MAX_BYTES) + return { kind: "too_large" }; + const parsed = parseDiff(result.stdout)[0]; + if (!parsed) return { kind: "ready", body: { path: file.entry.path, hunks: [] }, bytes }; + let highlighted = parsed; + if (compare.highlight !== false) { + metrics.highlightCalls++; + const context = await sources.read(file); + highlighted = await highlightDiffWithFileContent(parsed, cwd, context); + } + const hunks = highlighted.hunks.map((hunk) => ({ + oldStart: hunk.oldStart, + oldCount: hunk.oldCount, + newStart: hunk.newStart, + newCount: hunk.newCount, + lines: hunk.lines.map((line) => { + if (!line.tokens) return { type: line.type, content: line.content }; + return { + type: line.type, + content: line.content, + runs: line.tokens + .filter((token) => token.text.length > 0) + .map((token) => ({ style: token.style, length: token.text.length })), + }; + }), + })); + return { kind: "ready", body: { path: file.entry.path, hunks }, bytes }; + }, + }; +} + +function diffArguments(compare: Compare, baseSha: string, headSha: string): string[] { + const revisions: string[] = []; + if (compare.base.kind !== "index") revisions.push(baseSha); + if (compare.head.kind === "rev") revisions.push(headSha); + if (compare.head.kind === "index") revisions.unshift("--cached"); + const args = ["diff", ...GIT_PATCH_FORMAT_ARGS, "--no-textconv", "-M", ...revisions]; + if (compare.ignoreWhitespace) args.push("-w"); + return args; +} + +async function readUntrackedText( + path: string, + symlink: boolean, + omitted: boolean, +): Promise { + if (symlink) return Buffer.from(await readlink(path)); + if (omitted) return Buffer.alloc(0); + return readFile(path); +} + +async function listUntrackedFiles(options: GitCommandOptions): Promise { + const cwd = options.cwd; + const files: ListedFile[] = []; + const untracked = await runGitCommand( + ["ls-files", "--others", "--exclude-standard", "-z"], + options, + ); + for (const path of untracked.stdout.split("\0").filter(Boolean)) { + const value = await lstat(resolve(cwd, path), { bigint: true }); + const inspected = await inspectUntrackedFile(cwd, path); + const binary = inspected.stat?.isBinary === true; + const content = await readUntrackedText( + resolve(cwd, path), + value.isSymbolicLink(), + binary || inspected.truncated, + ); + const additions = + binary || content.length === 0 + ? 0 + : content.toString("utf8").split("\n").length - Number(content.at(-1) === 10); + files.push({ + entry: { + path, + oldPath: null, + change: "added", + content: binary ? "binary" : "text", + additions, + deletions: 0, + body: !binary, + }, + identity: `${value.size}:${value.mtimeNs}`, + source: { kind: "untracked", size: Number(value.size) }, + }); + } + return files; +} diff --git a/packages/server/src/server/diff/index.ts b/packages/server/src/server/diff/index.ts new file mode 100644 index 0000000000..bb176c2e86 --- /dev/null +++ b/packages/server/src/server/diff/index.ts @@ -0,0 +1,428 @@ +import { TOTAL_DIFF_MAX_BYTES } from "../../utils/checkout-git.js"; +import { runGitCommand } from "../../utils/run-git-command.js"; +import { expandTilde } from "../../utils/path.js"; +import { packBodies } from "./batches.js"; +import { setImmediate } from "node:timers/promises"; +import { + DiffOpcode, + materializeDiffFile, + type DiffFrame, + type DiffEntry, + type DiffBody, +} from "@getpaseo/protocol/binary-frames/index"; +import type { + DiffError, + DiffEndpoint, + ParsedDiffFile, + CheckoutCommitFileDiffRequest, +} from "@getpaseo/protocol/messages"; +import type { WorkspaceGitService } from "../workspace-git-service.js"; +import { + listComparison, + validateCompare, + worktreeRoot, + DiffReadError, + type Compare, + type Comparison, +} from "./git.js"; +import { DiffDependencyObserver } from "./dependencies.js"; + +interface RecordCache { + identity: string; + body: DiffBody; + bytes: number; +} +interface Listener { + emit(frame: DiffFrame): Promise; + active: boolean; +} +interface Target { + key: string; + compare: Compare; + listeners: Set; + revision: number; + committed: number; + cache: Map; + entries: DiffEntry[]; + resolved: Pick; + error: DiffError | null; + omitted: string[]; + refresh: Promise | null; + dirty: boolean; + release: (() => Promise) | null; + opening: Promise | null; + closed: boolean; + timer: ReturnType | null; +} +export interface CommitFileRead extends Pick< + CheckoutCommitFileDiffRequest, + "cwd" | "sha" | "path" +> { + signal?: AbortSignal; +} +export interface DiffSubscription { + start(): Promise; + release(): Promise; +} +export class DiffSubscriptions { + private targets = new Map(); + private dependencies: DiffDependencyObserver; + private computations = 0; + private highlightCalls = 0; + constructor(workspaceGit: Pick) { + this.dependencies = new DiffDependencyObserver(workspaceGit); + } + getMetrics() { + return { + targets: this.targets.size, + subscriptions: [...this.targets.values()].reduce( + (sum, target) => sum + target.listeners.size, + 0, + ), + computations: this.computations, + highlightCalls: this.highlightCalls, + metadataObservers: this.dependencies.getObservationCount(), + }; + } + async readCommitFile(input: CommitFileRead): Promise { + input.signal?.throwIfAborted(); + const cwd = await worktreeRoot(expandTilde(input.cwd)); + const result = await runGitCommand( + ["rev-list", "--parents", "--max-count=1", "--end-of-options", input.sha], + { cwd }, + ); + const [head, parent] = result.stdout.trim().split(" "); + if (!head) throw new DiffReadError("REF_NOT_FOUND", `Revision not found: ${input.sha}`); + const base: DiffEndpoint = parent ? { kind: "rev", rev: parent } : { kind: "empty" }; + const comparison = await listComparison({ cwd, base, head: { kind: "rev", rev: head } }); + try { + 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); + input.signal?.throwIfAborted(); + return materializeDiffFile( + file.entry, + record.kind === "ready" ? record.body : null, + record.kind === "too_large", + ); + } finally { + await comparison.close(); + } + } + async subscribe(compare: Compare, emit: Listener["emit"]): Promise { + validateCompare(compare); + compare = { + ...compare, + cwd: await worktreeRoot(expandTilde(compare.cwd)), + mergeBase: compare.mergeBase === true, + ignoreWhitespace: compare.ignoreWhitespace === true, + highlight: compare.highlight !== false, + }; + const key = JSON.stringify([ + compare.cwd, + endpointKey(compare.base), + endpointKey(compare.head), + compare.mergeBase, + compare.ignoreWhitespace, + compare.highlight, + ]); + let target = this.targets.get(key); + if (!target) { + target = { + key, + compare, + listeners: new Set(), + revision: 0, + committed: 0, + cache: new Map(), + entries: [], + resolved: { baseSha: "", headSha: "", mergeBaseSha: "" }, + error: null, + omitted: [], + refresh: null, + dirty: false, + release: null, + opening: null, + closed: false, + timer: null, + }; + this.targets.set(key, target); + const captured = target; + try { + target.opening = this.dependencies + .declareDependencies( + { + cwd: compare.cwd, + refs: [compare.base, compare.head].flatMap((endpoint) => + endpoint.kind === "rev" && !/^[a-f0-9]{40,64}$/.test(endpoint.rev) + ? [endpoint.rev] + : [], + ), + index: compare.base.kind === "index" || compare.head.kind !== "rev", + worktree: compare.head.kind === "worktree", + }, + () => this.invalidate(captured), + ) + .then((release) => { + captured.release = release; + return undefined; + }); + } catch (error) { + this.targets.delete(key); + throw error; + } + } + try { + await target.opening; + } catch (error) { + target.closed = true; + this.targets.delete(key); + throw error; + } + if (target.closed) throw new Error("Diff subscription cancelled"); + const current = target; + const listener: Listener = { emit, active: false }; + current.listeners.add(listener); + let released = false; + return { + start: async () => { + if (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); + }, + release: async () => { + if (released) return; + released = true; + listener.active = false; + current.listeners.delete(listener); + if (current.listeners.size) return; + current.closed = true; + if (current.timer) clearTimeout(current.timer); + this.targets.delete(key); + await current.release?.(); + await current.refresh; + }, + }; + } + invalidateForCwd(cwd: string): void { + for (const target of this.targets.values()) + if (target.compare.cwd === cwd) this.invalidate(target); + } + private invalidate(target: Target): void { + if (target.closed) return; + target.dirty = true; + if (target.listeners.size === 0) return; + if (target.timer) clearTimeout(target.timer); + target.timer = setTimeout(() => { + target.timer = null; + void this.refresh(target); + }, 150); + } + private async emit(target: Target, frame: DiffFrame): Promise { + await Promise.all( + [...target.listeners] + .filter((listener) => listener.active) + .map((listener) => listener.emit(frame)), + ); + await setImmediate(); + } + private async replay(target: Target, listener: Listener): Promise { + const entries = target.entries.map((entry) => ({ ...entry, body: entry.content !== "binary" })); + await listener.emit({ + opcode: DiffOpcode.Begin, + slot: 0, + revision: target.committed, + baseRevision: 0, + ...target.resolved, + entries, + }); + const bodies: DiffBody[] = []; + for (const entry of entries) { + const cached = target.cache.get(entry.path); + if (entry.body && cached) bodies.push(cached.body); + } + for await (const files of packBodies(bodies, target.committed)) { + await listener.emit({ + opcode: DiffOpcode.Bodies, + slot: 0, + revision: target.committed, + files, + }); + await setImmediate(); + } + await listener.emit({ + opcode: DiffOpcode.End, + slot: 0, + revision: target.committed, + diffTooLarge: target.omitted.length > 0, + omitted: target.omitted, + error: target.error, + }); + } + private async refresh(target: Target): Promise { + if (target.refresh || target.closed) return target.refresh ?? undefined; + target.refresh = this.run(target).finally(() => { + target.refresh = null; + }); + return target.refresh; + } + private async run(target: Target): Promise { + do { + target.dirty = false; + const revision = ++target.revision; + let comparison: Comparison | null = null; + let began = false; + try { + this.computations++; + comparison = await listComparison(target.compare); + const resolved = { + baseSha: comparison.baseSha, + headSha: comparison.headSha, + mergeBaseSha: comparison.mergeBaseSha, + }; + const tooManyFiles = comparison.files.length > 10000; + const entries = comparison.files.map((file) => ({ + ...file.entry, + body: tooManyFiles + ? file.entry.content === "text" + : needsBody(target, file.entry, file.identity), + })); + await this.emit(target, { + opcode: DiffOpcode.Begin, + slot: 0, + revision, + baseRevision: target.error ? 0 : target.committed, + ...resolved, + entries, + }); + began = true; + const cache = new Map(); + const omitted: string[] = []; + const state = { cache, omitted, total: 0 }; + const records = this.readBodies(target, comparison, entries, state); + for await (const files of packBodies(records, revision)) { + if (isInterrupted(target)) break; + await this.emit(target, { opcode: DiffOpcode.Bodies, slot: 0, revision, files }); + } + if (isInterrupted(target)) continue; + await this.emit(target, { + opcode: DiffOpcode.End, + slot: 0, + revision, + omitted, + diffTooLarge: omitted.length > 0, + error: null, + }); + target.committed = revision; + target.resolved = resolved; + target.entries = entries; + target.cache = cache; + target.omitted = omitted; + target.error = null; + } catch (error) { + const failure = diffError(error); + if (!began) + await this.emit(target, { + opcode: DiffOpcode.Begin, + slot: 0, + revision, + baseRevision: 0, + baseSha: "", + headSha: "", + mergeBaseSha: "", + entries: [], + }); + await this.emit(target, { + opcode: DiffOpcode.End, + slot: 0, + revision, + diffTooLarge: false, + omitted: [], + error: failure, + }); + target.error = failure; + target.committed = revision; + target.entries = []; + target.cache.clear(); + target.omitted = []; + } finally { + if (comparison) this.highlightCalls += comparison.metrics.highlightCalls; + await comparison?.close(); + } + } while (target.dirty && !target.closed); + } + private async *readBodies( + target: Target, + comparison: Comparison, + entries: DiffEntry[], + state: { cache: Map; omitted: string[]; total: number }, + ): AsyncGenerator { + if (entries.length > 10000) { + for (const entry of entries) if (entry.content === "text") state.omitted.push(entry.path); + return; + } + for (let index = 0; index < comparison.files.length; index++) { + if (isInterrupted(target)) return; + const file = comparison.files[index]; + const entry = entries[index]; + 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); + if (read.kind === "too_large") { + state.omitted.push(entry.path); + continue; + } + record = { identity: file.identity, body: read.body, bytes: read.bytes }; + } + if (exceedsBudget(state.total, record.bytes)) { + state.omitted.push(entry.path); + continue; + } + state.total += record.bytes; + state.cache.set(entry.path, record); + if (entry.body) yield record.body; + } + } + async close(): Promise { + for (const target of this.targets.values()) { + target.closed = true; + if (target.timer) clearTimeout(target.timer); + await target.opening; + await target.release?.(); + await target.refresh; + } + this.targets.clear(); + await this.dependencies.close(); + } +} + +function isInterrupted(target: Target): boolean { + return target.closed || target.dirty; +} +function exceedsBudget(total: number, bytes: number): boolean { + return total + bytes > TOTAL_DIFF_MAX_BYTES; +} +function needsBody(target: Target, entry: DiffEntry, identity: string): boolean { + return ( + entry.content === "text" && + (target.error !== null || target.cache.get(entry.path)?.identity !== identity) + ); +} + +function diffError(error: unknown): DiffError { + return { + code: error instanceof DiffReadError ? error.code : "UNKNOWN", + message: error instanceof Error ? error.message : String(error), + }; +} + +export { DiffSession } from "./session.js"; + +function endpointKey(endpoint: DiffEndpoint): string[] { + if (endpoint.kind === "rev") return [endpoint.kind, endpoint.rev]; + return [endpoint.kind]; +} diff --git a/packages/server/src/server/diff/session.test.ts b/packages/server/src/server/diff/session.test.ts new file mode 100644 index 0000000000..3ffc0a40e5 --- /dev/null +++ b/packages/server/src/server/diff/session.test.ts @@ -0,0 +1,96 @@ +import { execFileSync } from "node:child_process"; +import { mkdtemp, writeFile, rm } from "node:fs/promises"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import pino from "pino"; +import { expect, test } from "vitest"; +import { + decodeDiffFrame, + DiffOpcode, + type DiffFrame, +} from "@getpaseo/protocol/binary-frames/index"; +import { DiffSession, DiffSubscriptions } from "./index.js"; +import { createFileObserver } from "../file-observer/index.js"; +import { SessionDelivery } from "../session/owned-subscriptions/index.js"; + +test("a failed physical send releases that subscriber while its sibling completes the revision", async () => { + const cwd = await mkdtemp(join(tmpdir(), "diff-send-failure-")); + const git = (...args: string[]) => execFileSync("git", args, { cwd, encoding: "utf8" }).trim(); + const observer = createFileObserver(); + const subscriptions = new DiffSubscriptions({ + async requestWorkingTreeWatch(root, changed) { + const watch = await observer.subscribe(root, () => changed(), { + ignore: [join(root, ".git")], + }); + return { + repoRoot: root, + unsubscribe: async () => { + await watch.unsubscribe(); + }, + }; + }, + }); + const first = {}; + const second = {}; + let failFirst = false; + const firstFrames: DiffFrame[] = []; + const secondFrames: DiffFrame[] = []; + const delivery = new SessionDelivery( + () => {}, + async (source, bytes) => { + if (source === first && failFirst) throw new Error("Physical send failed"); + const frame = decodeDiffFrame(bytes); + const received = source === first ? firstFrames : secondFrames; + received.push(frame); + }, + ); + delivery.attach(first, true); + delivery.attach(second, true); + const session = new DiffSession(subscriptions, pino({ level: "silent" })); + try { + git("init", "-b", "main"); + git("config", "user.email", "test@example.com"); + git("config", "user.name", "Test"); + await writeFile(join(cwd, "README"), "base\n"); + git("add", "."); + git("-c", "commit.gpgsign=false", "commit", "-m", "base"); + await writeFile(join(cwd, "x.ts"), "export const value = 1;\n"); + for (const [source, requestId] of [ + [first, "first"], + [second, "second"], + ] as const) { + const message = { + type: "diff.subscribe.request" as const, + requestId, + cwd, + base: { kind: "rev" as const, rev: "HEAD" }, + head: { kind: "worktree" as const }, + }; + await delivery.request(source, message, () => session.subscribe(message, delivery, () => {})); + const received = source === first ? firstFrames : secondFrames; + await waitForEnd(received); + } + secondFrames.length = 0; + failFirst = true; + await writeFile(join(cwd, "x.ts"), "export const value = 2;\n"); + await waitForEnd(secondFrames); + const begins = secondFrames + .filter((frame) => frame.opcode === DiffOpcode.Begin) + .map((frame) => frame.revision); + expect(new Set(begins).size).toBe(begins.length); + expect(secondFrames.at(-1)).toMatchObject({ opcode: DiffOpcode.End, error: null }); + expect(delivery.subscriptionIds(first, "diffs")).toEqual([]); + expect(delivery.subscriptionIds(second, "diffs")).toHaveLength(1); + } finally { + await delivery.close(); + await subscriptions.close(); + await observer.close(); + await rm(cwd, { recursive: true, force: true }); + } +}, 30000); + +async function waitForEnd(frames: DiffFrame[]): Promise { + await expect + .poll(() => frames.some((frame) => frame.opcode === DiffOpcode.End), { timeout: 10000 }) + .toBe(true); +} diff --git a/packages/server/src/server/diff/session.ts b/packages/server/src/server/diff/session.ts new file mode 100644 index 0000000000..a965b2fa91 --- /dev/null +++ b/packages/server/src/server/diff/session.ts @@ -0,0 +1,86 @@ +import type { Logger } from "pino"; +import { encodeDiffFrame } from "@getpaseo/protocol/binary-frames/index"; +import type { + DiffSubscribeRequest, + SessionOutboundMessage, + DiffError, +} from "@getpaseo/protocol/messages"; +import type { SessionDelivery, OwnedSubscription } from "../session/owned-subscriptions/index.js"; +import { DiffReadError } from "./git.js"; +import type { DiffSubscriptions, DiffSubscription } from "./index.js"; + +export class DiffSession { + private slots = new Set(); + constructor( + private subscriptions: DiffSubscriptions, + private logger: Pick, + ) {} + async subscribe( + msg: DiffSubscribeRequest, + ownership: SessionDelivery, + emit: (message: SessionOutboundMessage) => void, + ): Promise { + const slot = Array.from({ length: 256 }, (_, value) => value).find( + (value) => !this.slots.has(value), + ); + if (slot === undefined) { + this.reject(msg, emit, { + code: "TOO_MANY_SUBSCRIPTIONS", + message: "Too many diff subscriptions", + }); + return; + } + this.slots.add(slot); + let subscription: DiffSubscription | null = null; + const owner = ownership.begin("diffs", undefined, async () => { + this.slots.delete(slot); + await subscription?.release(); + }); + try { + subscription = await this.subscriptions.subscribe(msg, async (frame) => { + if (owner.signal.aborted) return; + const bytes = encodeDiffFrame({ ...frame, slot }); + try { + await owner.emitBinaryAndWait(bytes); + } catch (error) { + void this.stopFailedSource(owner, error); + } + }); + if (owner.signal.aborted) { + await subscription.release(); + return; + } + emit({ + type: "diff.subscribe.response", + payload: { requestId: msg.requestId, subscriptionId: owner.responseId, slot, error: null }, + }); + // Start outside the request context: the owned subscription carries delivery authority. + void subscription.start().catch((error) => this.stopFailedSource(owner, error)); + } catch (error) { + await owner.release(); + const code = error instanceof DiffReadError ? error.code : "UNKNOWN"; + this.reject(msg, emit, { + code, + message: error instanceof Error ? error.message : String(error), + }); + } + } + private async stopFailedSource(owner: OwnedSubscription, error: unknown): Promise { + if (!owner.signal.aborted) this.logger.debug({ err: error }, "Diff frame delivery failed"); + try { + await owner.release(); + } catch (failure) { + this.logger.error({ err: failure }, "Failed to release diff subscription"); + } + } + private reject( + msg: DiffSubscribeRequest, + emit: (message: SessionOutboundMessage) => void, + error: DiffError, + ): void { + emit({ + type: "diff.subscribe.response", + payload: { requestId: msg.requestId, subscriptionId: "", slot: 0, error }, + }); + } +} diff --git a/packages/server/src/server/diff/sources.ts b/packages/server/src/server/diff/sources.ts new file mode 100644 index 0000000000..01aacf4c82 --- /dev/null +++ b/packages/server/src/server/diff/sources.ts @@ -0,0 +1,61 @@ +import { readGitFileContents, readGitFileContentAtRef } from "../../utils/git-file-contents.js"; +import type { ListedFile } from "./git.js"; + +interface SourceOptions { + cwd: string; + baseSha: string; + headSha: string | null; + files: ListedFile[]; +} +interface HighlightContext { + oldFileContent: string | null; + newFileContent: string | null | undefined; +} + +// Keep one bounded page of source text; bodies are still highlighted only when requested. +export class HighlightSources { + private indices: Map; + private page = -1; + private pending: Promise | null> | null = null; + constructor(private options: SourceOptions) { + this.indices = new Map(options.files.map((file, index) => [file.entry.path, index])); + } + async read(file: ListedFile): Promise { + const index = this.indices.get(file.entry.path); + if (index === undefined) throw new Error("Unknown comparison file"); + const page = Math.floor(index / 64) * 64; + if (this.pending === null || this.page !== page) { + this.page = page; + this.pending = this.load(page); + } + const contents = await this.pending; + const oldFileContent = + file.entry.change === "added" + ? null + : await this.content(contents, this.options.baseSha, file.entry.oldPath ?? file.entry.path); + let newFileContent: string | null | undefined; + if (this.options.headSha !== null) + newFileContent = await this.content(contents, this.options.headSha, file.entry.path); + return { oldFileContent, newFileContent }; + } + private async load(start: number): Promise | null> { + const specs: string[] = []; + for (const file of this.options.files.slice(start, start + 64)) { + if (file.entry.content === "binary") continue; + if (file.entry.change !== "added") + specs.push(`${this.options.baseSha}:${file.entry.oldPath ?? file.entry.path}`); + if (this.options.headSha !== null) specs.push(`${this.options.headSha}:${file.entry.path}`); + } + return readGitFileContents(this.options.cwd, specs); + } + private async content( + contents: Map | null, + ref: string, + path: string, + ): Promise { + if (contents === null) return readGitFileContentAtRef(this.options.cwd, ref, path); + const content = contents.get(`${ref}:${path}`); + if (content === undefined) throw new Error("Missing comparison source"); + return content; + } +} diff --git a/packages/server/src/server/diff/subscriptions.test.ts b/packages/server/src/server/diff/subscriptions.test.ts new file mode 100644 index 0000000000..4a4b62e36c --- /dev/null +++ b/packages/server/src/server/diff/subscriptions.test.ts @@ -0,0 +1,293 @@ +import { execFileSync } from "node:child_process"; +import { mkdtemp, writeFile, rm, mkdir } from "node:fs/promises"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import { expect, test } from "vitest"; +import { DiffOpcode, type DiffFrame } from "@getpaseo/protocol/binary-frames/index"; +import { DiffSubscriptions } from "./index.js"; +import { createFileObserver } from "../file-observer/index.js"; + +test("shares computation, highlights only the changed file, recovers refs and releases observations", async () => { + const cwd = await mkdtemp(join(tmpdir(), "diff-subscriptions-")); + const git = (...args: string[]) => execFileSync("git", args, { cwd, encoding: "utf8" }).trim(); + const observer = createFileObserver(); + const subscriptions = new DiffSubscriptions({ + async requestWorkingTreeWatch(root, onChange) { + const subscription = await observer.subscribe(root, () => onChange(), { + ignore: [join(root, ".git")], + }); + return { + repoRoot: root, + unsubscribe: () => { + void subscription.unsubscribe(); + }, + }; + }, + }); + const first: DiffFrame[] = []; + const second: DiffFrame[] = []; + try { + git("init", "-b", "main"); + git("config", "user.email", "test@example.com"); + git("config", "user.name", "Test"); + await writeFile(join(cwd, "README"), "base\n"); + git("add", "."); + git("-c", "commit.gpgsign=false", "commit", "-m", "base"); + for (let index = 0; index < 252; index++) + await writeFile(join(cwd, `${index}.ts`), `const value = ${index};\n`); + const compare = { + cwd, + base: { kind: "rev" as const, rev: "HEAD" }, + head: { kind: "worktree" as const }, + highlight: true, + }; + const coldStarted = performance.now(); + let lastEndAt = coldStarted; + const a = await subscriptions.subscribe(compare, async (frame) => { + first.push(frame); + if (frame.opcode === DiffOpcode.End) lastEndAt = performance.now(); + }); + await a.start(); + const coldMs = lastEndAt - coldStarted; + const before = subscriptions.getMetrics().computations; + const previousHighlightCalls = subscriptions.getMetrics().highlightCalls; + const b = await subscriptions.subscribe(compare, async (frame) => { + second.push(frame); + }); + await b.start(); + expect(subscriptions.getMetrics()).toEqual({ + targets: 1, + subscriptions: 2, + computations: before, + highlightCalls: previousHighlightCalls, + metadataObservers: 1, + }); + first.length = 0; + second.length = 0; + const refreshStarted = performance.now(); + await writeFile(join(cwd, "9.ts"), "const value = 999;\n"); + await expect + .poll(() => first.filter((frame) => frame.opcode === DiffOpcode.End).length) + .toBeGreaterThan(0); + const begin = first.find((frame) => frame.opcode === DiffOpcode.Begin); + if (!begin || begin.opcode !== DiffOpcode.Begin) throw new Error("Missing begin"); + expect(begin.entries.filter((entry) => entry.body).map((entry) => entry.path)).toEqual([ + "9.ts", + ]); + const bodies = first.filter((frame) => frame.opcode === DiffOpcode.Bodies); + expect(bodies).toHaveLength(1); + expect(second).toEqual(first); + await expect + .poll(() => subscriptions.getMetrics().highlightCalls, { timeout: 10000 }) + .toBe(previousHighlightCalls + 1); + const artifactDirectory = join( + import.meta.dirname, + "../../../../../.artifacts/diff-subscription", + ); + await mkdir(artifactDirectory, { recursive: true }); + await writeFile( + join(artifactDirectory, "refresh-metrics.json"), + JSON.stringify( + { + files: 252, + coldMs, + oneFileRefreshMs: lastEndAt - refreshStarted, + changedBodies: bodies.length, + highlightCalls: subscriptions.getMetrics().highlightCalls - previousHighlightCalls, + }, + null, + 2, + ), + ); + await a.release(); + await b.release(); + expect(subscriptions.getMetrics().targets).toBe(0); + expect(subscriptions.getMetrics().metadataObservers).toBe(0); + const missing: DiffFrame[] = []; + const recovery = await subscriptions.subscribe( + { ...compare, base: { kind: "rev", rev: "refs/tags/later" } }, + async (frame) => { + missing.push(frame); + }, + ); + await recovery.start(); + expect(missing.at(-1)).toMatchObject({ + opcode: DiffOpcode.End, + error: { code: "REF_NOT_FOUND" }, + }); + missing.length = 0; + git("update-ref", "refs/tags/later", "HEAD"); + await expect + .poll( + () => missing.some((frame) => frame.opcode === DiffOpcode.End && frame.error === null), + { timeout: 10_000 }, + ) + .toBe(true); + await recovery.release(); + } finally { + await subscriptions.close(); + await observer.close(); + await rm(cwd, { recursive: true, force: true }); + } +}, 30_000); + +test("supersedes an in-flight revision at a batch boundary and ends the replacement", async () => { + const cwd = await mkdtemp(join(tmpdir(), "diff-supersede-")); + const git = (...args: string[]) => execFileSync("git", args, { cwd, encoding: "utf8" }).trim(); + const observer = createFileObserver(); + const subscriptions = new DiffSubscriptions({ + async requestWorkingTreeWatch(root, onChange) { + const watch = await observer.subscribe(root, () => onChange(), { + ignore: [join(root, ".git")], + }); + return { + repoRoot: root, + unsubscribe: async () => { + await watch.unsubscribe(); + }, + }; + }, + }); + const frames: DiffFrame[] = []; + let interrupted = false; + try { + git("init", "-b", "main"); + git("config", "user.email", "test@example.com"); + git("config", "user.name", "Test"); + await writeFile(join(cwd, "README"), "base\n"); + git("add", "."); + git("-c", "commit.gpgsign=false", "commit", "-m", "base"); + const content = + Array.from( + { length: 100 }, + (_, index) => `export const fn${index} = (value: string) => ({ value, count: ${index} });`, + ).join("\n") + "\n"; + for (let index = 0; index < 80; index++) await writeFile(join(cwd, `${index}.ts`), content); + const subscription = await subscriptions.subscribe( + { cwd, base: { kind: "rev", rev: "HEAD" }, head: { kind: "worktree" } }, + async (frame) => { + frames.push(frame); + if (frame.opcode === DiffOpcode.Bodies && !interrupted) { + interrupted = true; + await writeFile(join(cwd, "0.ts"), content + "export const added = true;\n"); + subscriptions.invalidateForCwd(cwd); + } + }, + ); + await subscription.start(); + const begins = frames.filter((frame) => frame.opcode === DiffOpcode.Begin); + expect(interrupted).toBe(true); + expect(begins.length).toBeGreaterThan(1); + for (let index = 0; index < begins.length - 1; index++) { + const revision = begins[index].revision; + const ended = frames.some( + (frame) => frame.opcode === DiffOpcode.End && frame.revision === revision, + ); + const superseded = begins[index + 1].revision > revision; + expect(ended || superseded).toBe(true); + } + expect(frames.at(-1)).toMatchObject({ + opcode: DiffOpcode.End, + revision: begins.at(-1)?.revision, + error: null, + }); + await subscription.release(); + } finally { + await subscriptions.close(); + await observer.close(); + await rm(cwd, { recursive: true, force: true }); + } +}, 30_000); + +test("moving, deleting and recreating a branch recovers, and staging changes the resolved index SHA", async () => { + const cwd = await mkdtemp(join(tmpdir(), "diff-ref-index-")); + const git = (...args: string[]) => execFileSync("git", args, { cwd, encoding: "utf8" }).trim(); + const observer = createFileObserver(); + const subscriptions = new DiffSubscriptions({ + async requestWorkingTreeWatch(root, changed) { + const watch = await observer.subscribe(root, () => changed(), { + ignore: [join(root, ".git")], + }); + return { + repoRoot: root, + unsubscribe: async () => { + await watch.unsubscribe(); + }, + }; + }, + }); + const frames: DiffFrame[] = []; + try { + git("init", "-b", "main"); + git("config", "user.email", "test@example.com"); + git("config", "user.name", "Test"); + await writeFile(join(cwd, "README"), "base\n"); + git("add", "."); + git("-c", "commit.gpgsign=false", "commit", "-m", "base"); + git("branch", "compare"); + await writeFile(join(cwd, "x.ts"), "export const value = 1;\n"); + git("add", "."); + git("-c", "commit.gpgsign=false", "commit", "-m", "feature"); + const head = git("rev-parse", "HEAD"); + const branch = await subscriptions.subscribe( + { cwd, base: { kind: "rev", rev: "refs/heads/compare" }, head: { kind: "rev", rev: "HEAD" } }, + async (frame) => { + frames.push(frame); + }, + ); + await branch.start(); + frames.length = 0; + git("update-ref", "refs/heads/compare", head); + await expect + .poll(() => frames.some((frame) => frame.opcode === DiffOpcode.End), { timeout: 10000 }) + .toBe(true); + expect(frames.find((frame) => frame.opcode === DiffOpcode.Begin)).toMatchObject({ + baseSha: head, + entries: [], + }); + frames.length = 0; + git("update-ref", "-d", "refs/heads/compare"); + await expect + .poll( + () => + frames.some( + (frame) => frame.opcode === DiffOpcode.End && frame.error?.code === "REF_NOT_FOUND", + ), + { timeout: 10000 }, + ) + .toBe(true); + frames.length = 0; + git("update-ref", "refs/heads/compare", head); + await expect + .poll(() => frames.some((frame) => frame.opcode === DiffOpcode.End && frame.error === null), { + timeout: 10000, + }) + .toBe(true); + await branch.release(); + frames.length = 0; + const staged = await subscriptions.subscribe( + { cwd, base: { kind: "rev", rev: "HEAD" }, head: { kind: "index" } }, + async (frame) => { + frames.push(frame); + }, + ); + await staged.start(); + const old = frames.find((frame) => frame.opcode === DiffOpcode.Begin); + if (!old || old.opcode !== DiffOpcode.Begin) throw new Error("Missing initial index revision"); + frames.length = 0; + await writeFile(join(cwd, "staged.ts"), "export const staged = true;\n"); + git("add", "staged.ts"); + await expect + .poll(() => frames.some((frame) => frame.opcode === DiffOpcode.End), { timeout: 10000 }) + .toBe(true); + const changed = frames.find((frame) => frame.opcode === DiffOpcode.Begin); + if (!changed || changed.opcode !== DiffOpcode.Begin) throw new Error("Missing staged revision"); + expect(changed.headSha).not.toBe(old.headSha); + expect(changed.entries.map((entry) => entry.path)).toEqual(["staged.ts"]); + await staged.release(); + } finally { + await subscriptions.close(); + await observer.close(); + await rm(cwd, { recursive: true, force: true }); + } +}, 30000); diff --git a/packages/server/src/server/git-metadata-event-rules.test.ts b/packages/server/src/server/git-metadata-event-rules.test.ts index 4456573624..cee7b9c4b0 100644 --- a/packages/server/src/server/git-metadata-event-rules.test.ts +++ b/packages/server/src/server/git-metadata-event-rules.test.ts @@ -39,7 +39,7 @@ describe("Git metadata event rules", () => { ["common", "hooks/pre-commit", { kind: "ignore" }], ["common", "logs/refs/heads/main", { kind: "ignore" }], ["common", "objects/ab/cdef", { kind: "ignore" }], - ["common", "refs/tags/v1", { kind: "ignore" }], + ["common", "refs/tags/v1", { kind: "all", refreshBase: true }], ["common", "FETCH_HEAD", { kind: "ignore" }], ["common", "packed-refs.lock", { kind: "ignore" }], ] as const)("classifies %s metadata path %s", (scope, path, expected) => { diff --git a/packages/server/src/server/git-metadata-event-rules.ts b/packages/server/src/server/git-metadata-event-rules.ts index ff82747ba0..3a4fde345c 100644 --- a/packages/server/src/server/git-metadata-event-rules.ts +++ b/packages/server/src/server/git-metadata-event-rules.ts @@ -73,6 +73,14 @@ export const GIT_METADATA_EVENT_RULES = [ path: "refs/heads/", route: "local-ref", }, + { + id: "tag", + scope: "common", + match: "prefix", + path: "refs/tags/", + route: "all", + refreshBase: true, + }, { id: "remote-branch", scope: "common", diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 3e041b9b0d..9002285792 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -1,3 +1,4 @@ +import { DiffSession } from "./diff/index.js"; import { type WorkspaceRuntimeStore } from "./workspace-runtime-store/index.js"; import type { AgentMessageDelivery } from "./agent/agent-messages/index.js"; import { AgentDirectory, type BackgroundVisibility } from "./agent/directory.js"; @@ -841,6 +842,7 @@ export class Session { private readonly workspaceDirectory: WorkspaceDirectory; private readonly voiceSessions: VoiceSessions; private readonly checkoutSession: CheckoutSession; + private readonly diffSession: DiffSession; private readonly scheduleSession: ScheduleSession; private readonly providerCatalogSession: ProviderCatalogSession; private readonly usageSession: UsageSession; @@ -1004,6 +1006,7 @@ export class Session { await this.workspaceProvisioning.ensureWorkspaceRecordUnarchived(workspace); }, }); + this.diffSession = new DiffSession(checkoutDiffManager.streams, this.sessionLogger); this.checkoutSession = new CheckoutSession({ host: { emit: (msg) => this.emit(msg), @@ -3128,6 +3131,10 @@ export class Session { // eslint-disable-next-line complexity private dispatchCheckoutMessage(msg: SessionInboundMessage): Promise | undefined { switch (msg.type) { + case "diff.subscribe.request": + return this.diffSession.subscribe(msg, this.delivery, (reply) => this.emit(reply)); + case "diff.unsubscribe.request": + return this.delivery.release(msg.subscriptionId); case "checkout_status_request": return this.checkoutSession.handleStatusRequest(msg); case "checkout.commits.list.request": diff --git a/packages/server/src/server/session/checkout/checkout-session.test.ts b/packages/server/src/server/session/checkout/checkout-session.test.ts index aaa79e3da1..2a68b87de1 100644 --- a/packages/server/src/server/session/checkout/checkout-session.test.ts +++ b/packages/server/src/server/session/checkout/checkout-session.test.ts @@ -50,6 +50,7 @@ function createFakeDiffSubscriber(initial: CheckoutDiffSnapshotPayload) { const subscriptions: FakeDiffSubscription[] = []; const refreshedCwds: string[] = []; const subscriber: CheckoutDiffSubscriber = { + readCommitFile: async () => null, read: async (params) => ({ ...initial, cwd: params.cwd }), subscribe: async (params, listener) => { let isSubscribed = true; diff --git a/packages/server/src/server/session/checkout/checkout-session.ts b/packages/server/src/server/session/checkout/checkout-session.ts index 4deb28a56b..60fda36b01 100644 --- a/packages/server/src/server/session/checkout/checkout-session.ts +++ b/packages/server/src/server/session/checkout/checkout-session.ts @@ -51,7 +51,6 @@ import { pullCurrentBranch, pushCurrentBranch, listCheckoutCommits, - getCommitFileDiff, } from "../../../utils/checkout-git.js"; import { runGitCommand } from "../../../utils/run-git-command.js"; import { expandTilde } from "../../../utils/path.js"; @@ -109,6 +108,7 @@ function toLegacyGithubSearchItems(items: ForgeSearchResultItem[]): LegacyGithub * real CheckoutDiffManager satisfies this structurally; tests supply a fake. */ export interface CheckoutDiffSubscriber { + readCommitFile: import("../../diff/index.js").DiffSubscriptions["readCommitFile"]; read( params: Omit, ): Promise; @@ -301,7 +301,12 @@ export class CheckoutSession { if (path.length === 0 || isAbsolute(path) || path.split(/[\\/]/).includes("..")) { throw new Error(`Invalid path: ${path}`); } - const file = await getCommitFileDiff({ cwd: expandTilde(cwd), sha, path, signal }); + const file = await this.checkoutDiffManager.readCommitFile({ + cwd: expandTilde(cwd), + sha, + path, + signal, + }); this.host.emit({ type: "checkout.commits.file_diff.response", payload: { cwd, sha, path, file, error: null, requestId }, diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index 7c301556c3..381891eaad 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -1223,7 +1223,7 @@ export class VoiceAssistantWebSocketServer { await Promise.all(cleanupPromises); this.providerSnapshotManager.destroy(); - this.checkoutDiffManager.dispose(); + await this.checkoutDiffManager.dispose(); await this.workspaceGitService.dispose(); this.pendingConnections.clear(); this.sessions.clear(); @@ -2020,6 +2020,7 @@ export class VoiceAssistantWebSocketServer { usageSetting: true, usageSources: true, ownedSubscriptions: true, + diffSubscription: true, binaryChunkingV1: true, agentRequestReceipts: true, workspaceRequestReceipts: true, diff --git a/packages/server/src/utils/checkout-git.ts b/packages/server/src/utils/checkout-git.ts index 422ca2b0ee..50d312a757 100644 --- a/packages/server/src/utils/checkout-git.ts +++ b/packages/server/src/utils/checkout-git.ts @@ -40,7 +40,7 @@ import { } from "../services/forge-cli-command.js"; import { parseGitRevParsePath, resolveGitRevParsePath } from "./git-rev-parse-path.js"; import { runGitCommand, type RunGitCommand } from "./run-git-command.js"; -import { readGitFileContents } from "./git-file-contents.js"; +import { readGitFileContents, readGitFileContentAtRef } from "./git-file-contents.js"; import { unquoteGitPath } from "./git-quoted-path.js"; import { isPaseoOwnedWorktreeCwd, resolvePaseoWorktreesBaseRoot } from "./worktree.js"; import { @@ -613,22 +613,6 @@ async function listCheckoutFileChanges( return Array.from(byPath.values()); } -async function readGitFileContentAtRef( - cwd: string, - ref: string, - path: string, -): Promise { - try { - const { stdout } = await runGitCommand(["show", `${ref}:${path}`], { - cwd, - envOverlay: READ_ONLY_GIT_ENV, - }); - return stdout; - } catch { - return null; - } -} - async function tryResolveMergeBase(cwd: string, baseRef: string): Promise { try { const { stdout } = await runGitCommand(["merge-base", baseRef, "HEAD"], { @@ -669,7 +653,12 @@ function normalizeNumstatPath(pathField: string): string { } // Parsed patches need a stable wire format regardless of terminal presentation settings. -const GIT_PATCH_FORMAT_ARGS = ["--no-ext-diff", "--no-color", "--src-prefix=a/", "--dst-prefix=b/"]; +export const GIT_PATCH_FORMAT_ARGS = [ + "--no-ext-diff", + "--no-color", + "--src-prefix=a/", + "--dst-prefix=b/", +] as const; function buildGitDiffArgs(args: { ignoreWhitespace?: boolean; extra: string[] }): string[] { return [ @@ -2255,8 +2244,8 @@ export async function getCheckoutSnapshotFacts( }; } -const PER_FILE_DIFF_MAX_BYTES = 1024 * 1024; // 1MB -const TOTAL_DIFF_MAX_BYTES = 2 * 1024 * 1024; // 2MB +export const PER_FILE_DIFF_MAX_BYTES = 1024 * 1024; // 1MB +export const TOTAL_DIFF_MAX_BYTES = 2 * 1024 * 1024; // 2MB const CHECKOUT_DIFF_FRAME_HEADROOM_BYTES = 1024 * 1024; // Temporary until diffs load lazily per file. The Paseo relay's 32 MiB frame limit is // binding: string frames are encrypted and base64-encoded. Reserve 1 MiB plaintext for @@ -2317,7 +2306,7 @@ async function isLikelyBinaryFile(absolutePath: string): Promise { } } -async function inspectUntrackedFile( +export async function inspectUntrackedFile( cwd: string, relativePath: string, ): Promise<{ stat: FileStat; truncated: boolean }> { @@ -2822,6 +2811,7 @@ async function readCommitRenames(cwd: string, sha: string): Promise { + if (compare.mode === "uncommitted") return "HEAD"; const { storedBaseRef, resolvedBaseRef } = await resolveBaseRefForCwd(cwd, context); const baseRef = resolveOperationBaseRef({ storedBaseRef, @@ -3597,17 +3602,12 @@ async function resolveCheckoutDiffRefs( if (!baseRef) { return null; } - const bestBaseRef = await resolveBestComparisonBaseRef(cwd, baseRef); - return { - baseRef: (await tryResolveMergeBase(cwd, bestBaseRef)) ?? bestBaseRef, - targetRef: compare.mode === "base" ? "HEAD" : undefined, - includeUntracked: compare.mode === "all", - }; + return resolveBestComparisonBaseRef(cwd, baseRef); } // Git's working-tree diff ignores files absent from the index, even when they // exist in the comparison tree. Restore those entries only in a private index. -async function prepareAllChangesIndex(cwd: string, baseRef: string) { +export async function prepareAllChangesIndex(cwd: string, baseRef: string) { const untracked = await runGitCommand(["ls-files", "-z", "--others", "--exclude-standard"], { cwd, envOverlay: READ_ONLY_GIT_ENV, @@ -3630,21 +3630,41 @@ async function prepareAllChangesIndex(cwd: string, baseRef: string) { entries += result.stdout; } if (!entries) return null; - const directory = await mkdtemp(resolve(tmpdir(), "paseo-all-changes-")); + const snapshot = await snapshotComparisonIndex(cwd); + try { + await runGitCommand(["update-index", "-z", "--index-info"], { + cwd, + input: entries, + envOverlay: { ...READ_ONLY_GIT_ENV, GIT_INDEX_FILE: snapshot.indexFile }, + }); + return snapshot; + } catch (error) { + await snapshot.cleanup(); + throw error; + } +} + +// write-tree rewrites its index. A private snapshot prevents an index read from +// invalidating the observation that requested it. +export async function snapshotComparisonIndex(cwd: string) { + const directory = await mkdtemp(resolve(tmpdir(), "paseo-comparison-index-")); const indexFile = resolve(directory, "index"); try { const index = await runGitCommand(["rev-parse", "--git-path", "index"], { cwd, envOverlay: READ_ONLY_GIT_ENV, }); - const indexPath = resolveGitRevParsePath(cwd, index.stdout); - if (!indexPath) throw new Error("Git did not return an index path"); - await copyFile(indexPath, indexFile); - await runGitCommand(["update-index", "-z", "--index-info"], { - cwd, - input: entries, - envOverlay: { ...READ_ONLY_GIT_ENV, GIT_INDEX_FILE: indexFile }, - }); + const source = resolveGitRevParsePath(cwd, index.stdout); + if (!source) throw new Error("Git did not return an index path"); + try { + await copyFile(source, indexFile); + } catch (error) { + if (!(error instanceof Error) || !("code" in error) || error.code !== "ENOENT") throw error; + await runGitCommand(["read-tree", "--empty"], { + cwd, + envOverlay: { ...READ_ONLY_GIT_ENV, GIT_INDEX_FILE: indexFile }, + }); + } return { indexFile, cleanup: () => rm(directory, { recursive: true, force: true }) }; } catch (error) { await rm(directory, { recursive: true, force: true }); diff --git a/packages/server/src/utils/git-file-contents.ts b/packages/server/src/utils/git-file-contents.ts index e42d3c6c83..04a314f17c 100644 --- a/packages/server/src/utils/git-file-contents.ts +++ b/packages/server/src/utils/git-file-contents.ts @@ -1,4 +1,4 @@ -import { runGitCommandBytes } from "./run-git-command.js"; +import { runGitCommandBytes, runGitCommand } from "./run-git-command.js"; // Match the existing git-show ceiling. Oversized batches use bounded per-file reads. const MAX_CONTENT_BYTES = 20 * 1024 * 1024; @@ -46,3 +46,17 @@ export async function readGitFileContents( if (offset !== result.stdout.length) throw new Error("Unexpected git object output"); return contents; } + +export async function readGitFileContentAtRef( + cwd: string, + ref: string, + path: string, +): Promise { + const result = await runGitCommand(["show", `${ref}:${path}`], { + cwd, + envOverlay: { GIT_OPTIONAL_LOCKS: "0", LC_ALL: "C" }, + acceptExitCodes: [0, 128], + }); + if (result.exitCode !== 0 || result.truncated) return null; + return result.stdout; +}