mirror of
https://github.com/getpaseo/paseo.git
synced 2026-11-06 08:24:46 +00:00
feat(server): stream shared diff revisions with paced file bodies
This commit is contained in:
Mohamed Boudra
@@ -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");
|
||||
}
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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();
|
||||
}
|
||||
@@ -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<T> {
|
||||
promise: Promise<T>;
|
||||
resolve(value: T): void;
|
||||
}
|
||||
|
||||
function createDeferred<T>(): Deferred<T> {
|
||||
let resolve!: (value: T) => void;
|
||||
const promise = new Promise<T>((resolvePromise) => {
|
||||
resolve = resolvePromise;
|
||||
});
|
||||
return { promise, resolve };
|
||||
}
|
||||
|
||||
function createWorkspaceSnapshot(
|
||||
overrides?: Partial<WorkspaceGitRuntimeSnapshot["git"]>,
|
||||
): 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<void>,
|
||||
): Promise<void> {
|
||||
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<typeof vi.fn>;
|
||||
}) {
|
||||
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<void> {
|
||||
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);
|
||||
|
||||
@@ -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<SessionOutboundMessage, { type: "checkout_diff_update" }>["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<void> | null;
|
||||
refreshQueued: boolean;
|
||||
refreshQueuedForce: boolean;
|
||||
latestPayload: CheckoutDiffSnapshotPayload | null;
|
||||
latestFingerprint: string | null;
|
||||
openPromise: Promise<void> | 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<string, CheckoutDiffWatchTarget>();
|
||||
|
||||
readonly streams: DiffSubscriptions;
|
||||
private readonly paseoHome: string;
|
||||
constructor(options: {
|
||||
logger: pino.Logger;
|
||||
paseoHome: string;
|
||||
workspaceGitService: CheckoutDiffWorkspace;
|
||||
workspaceGitService: Pick<WorkspaceGitService, "requestWorkingTreeWatch">;
|
||||
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<CheckoutDiffSubscriptionRequest, "signal">,
|
||||
): Promise<CheckoutDiffSnapshotPayload> {
|
||||
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<CheckoutDiffSubscription> {
|
||||
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<CheckoutDiffSnapshotPayload>((resolve) => {
|
||||
resolveInitial = resolve;
|
||||
});
|
||||
let delivered = false;
|
||||
let pending: DiffBegin | null = null;
|
||||
let files = new Map<string, ParsedDiffFile>();
|
||||
let bodies = new Map<string, DiffBody>();
|
||||
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<string, ParsedDiffFile>();
|
||||
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<CheckoutDiffSnapshotPayload> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
await this.streams.close();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<typeof WebSocket> | 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<void>((resolve, reject) => {
|
||||
activeSocket.once("open", resolve);
|
||||
activeSocket.once("error", reject);
|
||||
});
|
||||
let encodedBytes = 0;
|
||||
let endReceived = false;
|
||||
let pingSent = false;
|
||||
let pongDuringRevision = false;
|
||||
let bodyFrames = 0;
|
||||
let entryCount = 0;
|
||||
const done = new Promise<void>((resolve, reject) => {
|
||||
activeSocket.on("message", (data, binary) => {
|
||||
try {
|
||||
if (binary) {
|
||||
if (!Buffer.isBuffer(data)) throw new Error("Expected buffer");
|
||||
encodedBytes += data.byteLength;
|
||||
const frame = decodeDiffFrame(new Uint8Array(data));
|
||||
if (frame.opcode === DiffOpcode.Begin) entryCount = frame.entries.length;
|
||||
if (frame.opcode === DiffOpcode.Bodies) {
|
||||
bodyFrames++;
|
||||
if (!pingSent) {
|
||||
pingSent = true;
|
||||
activeSocket.send(
|
||||
JSON.stringify({
|
||||
type: "session",
|
||||
message: { type: "ping", requestId: "mid-revision", clientSentAt: 1 },
|
||||
}),
|
||||
);
|
||||
}
|
||||
}
|
||||
if (frame.opcode === DiffOpcode.End) {
|
||||
expect(frame.error).toBeNull();
|
||||
endReceived = true;
|
||||
resolve();
|
||||
}
|
||||
} else {
|
||||
const envelope = JSON.parse(data.toString());
|
||||
if (
|
||||
envelope.message?.type === "pong" &&
|
||||
envelope.message.payload.requestId === "mid-revision"
|
||||
)
|
||||
pongDuringRevision = !endReceived;
|
||||
if (envelope.message?.payload?.status === "server_info")
|
||||
activeSocket.send(
|
||||
JSON.stringify({
|
||||
type: "session",
|
||||
message: {
|
||||
type: "diff.subscribe.request",
|
||||
requestId: "stream-diff",
|
||||
cwd,
|
||||
base: { kind: "rev", rev: "HEAD" },
|
||||
head: { kind: "worktree" },
|
||||
},
|
||||
}),
|
||||
);
|
||||
if (envelope.message?.type === "diff.subscribe.response")
|
||||
expect(envelope.message.payload.error).toBeNull();
|
||||
}
|
||||
} catch (error) {
|
||||
reject(error);
|
||||
}
|
||||
});
|
||||
activeSocket.once("error", reject);
|
||||
});
|
||||
activeSocket.send(
|
||||
JSON.stringify({
|
||||
type: "hello",
|
||||
clientId: "diff-wire-test",
|
||||
clientType: "browser",
|
||||
protocolVersion: 1,
|
||||
capabilities: { owned_subscriptions: true },
|
||||
}),
|
||||
);
|
||||
await done;
|
||||
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);
|
||||
|
||||
@@ -0,0 +1,30 @@
|
||||
import { encodeDiffFrame, DiffOpcode, type DiffBody } from "@getpaseo/protocol/binary-frames/index";
|
||||
|
||||
export async function* packBodies(
|
||||
files: AsyncIterable<DiffBody> | Iterable<DiffBody>,
|
||||
revision: number,
|
||||
): AsyncGenerator<DiffBody[]> {
|
||||
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;
|
||||
}
|
||||
@@ -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<WorkspaceGitService, "requestWorkingTreeWatch">) {}
|
||||
async declareDependencies(
|
||||
input: DiffDependencies,
|
||||
changed: () => void,
|
||||
): Promise<() => Promise<void>> {
|
||||
if (input.refs.length === 0 && !input.index && !input.worktree) return async () => {};
|
||||
let polling: ReturnType<typeof setInterval> | null = null;
|
||||
const subscriptions: FileObserverSubscription[] = [];
|
||||
let workingTree: Awaited<ReturnType<WorkspaceGitService["requestWorkingTreeWatch"]>> | 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<void> {
|
||||
await this.observer.close();
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
@@ -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<DiffSubscribeRequest, "type" | "requestId">;
|
||||
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<void>;
|
||||
read(file: ListedFile, budget?: number): Promise<DiffFileRead>;
|
||||
}
|
||||
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<string> {
|
||||
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<string> {
|
||||
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<string, { additions: number; deletions: number; binary: boolean }> {
|
||||
const fields = text.split("\0");
|
||||
const stats = new Map<string, { additions: number; deletions: number; binary: boolean }>();
|
||||
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<Comparison> {
|
||||
validateCompare(compare);
|
||||
const cwd = compare.cwd;
|
||||
let privateIndex: Awaited<ReturnType<typeof prepareAllChangesIndex>> = 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<ReturnType<typeof prepareAllChangesIndex>>,
|
||||
): Promise<Comparison> {
|
||||
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<Buffer> {
|
||||
if (symlink) return Buffer.from(await readlink(path));
|
||||
if (omitted) return Buffer.alloc(0);
|
||||
return readFile(path);
|
||||
}
|
||||
|
||||
async function listUntrackedFiles(options: GitCommandOptions): Promise<ListedFile[]> {
|
||||
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;
|
||||
}
|
||||
@@ -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<void>;
|
||||
active: boolean;
|
||||
}
|
||||
interface Target {
|
||||
key: string;
|
||||
compare: Compare;
|
||||
listeners: Set<Listener>;
|
||||
revision: number;
|
||||
committed: number;
|
||||
cache: Map<string, RecordCache>;
|
||||
entries: DiffEntry[];
|
||||
resolved: Pick<Comparison, "baseSha" | "headSha" | "mergeBaseSha">;
|
||||
error: DiffError | null;
|
||||
omitted: string[];
|
||||
refresh: Promise<void> | null;
|
||||
dirty: boolean;
|
||||
release: (() => Promise<void>) | null;
|
||||
opening: Promise<void> | null;
|
||||
closed: boolean;
|
||||
timer: ReturnType<typeof setTimeout> | null;
|
||||
}
|
||||
export interface CommitFileRead extends Pick<
|
||||
CheckoutCommitFileDiffRequest,
|
||||
"cwd" | "sha" | "path"
|
||||
> {
|
||||
signal?: AbortSignal;
|
||||
}
|
||||
export interface DiffSubscription {
|
||||
start(): Promise<void>;
|
||||
release(): Promise<void>;
|
||||
}
|
||||
export class DiffSubscriptions {
|
||||
private targets = new Map<string, Target>();
|
||||
private dependencies: DiffDependencyObserver;
|
||||
private computations = 0;
|
||||
private highlightCalls = 0;
|
||||
constructor(workspaceGit: Pick<WorkspaceGitService, "requestWorkingTreeWatch">) {
|
||||
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<ParsedDiffFile | null> {
|
||||
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<DiffSubscription> {
|
||||
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<void> {
|
||||
await Promise.all(
|
||||
[...target.listeners]
|
||||
.filter((listener) => listener.active)
|
||||
.map((listener) => listener.emit(frame)),
|
||||
);
|
||||
await setImmediate();
|
||||
}
|
||||
private async replay(target: Target, listener: Listener): Promise<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<string, RecordCache>();
|
||||
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<string, RecordCache>; omitted: string[]; total: number },
|
||||
): AsyncGenerator<DiffBody> {
|
||||
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<void> {
|
||||
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];
|
||||
}
|
||||
@@ -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<void> {
|
||||
await expect
|
||||
.poll(() => frames.some((frame) => frame.opcode === DiffOpcode.End), { timeout: 10000 })
|
||||
.toBe(true);
|
||||
}
|
||||
@@ -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<number>();
|
||||
constructor(
|
||||
private subscriptions: DiffSubscriptions,
|
||||
private logger: Pick<Logger, "debug" | "error">,
|
||||
) {}
|
||||
async subscribe(
|
||||
msg: DiffSubscribeRequest,
|
||||
ownership: SessionDelivery,
|
||||
emit: (message: SessionOutboundMessage) => void,
|
||||
): Promise<void> {
|
||||
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<void> {
|
||||
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 },
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -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<string, number>;
|
||||
private page = -1;
|
||||
private pending: Promise<Map<string, string | null> | 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<HighlightContext> {
|
||||
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<Map<string, string | null> | 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<string, string | null> | null,
|
||||
ref: string,
|
||||
path: string,
|
||||
): Promise<string | null> {
|
||||
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;
|
||||
}
|
||||
}
|
||||
@@ -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);
|
||||
@@ -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) => {
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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<void> | 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":
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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<CheckoutDiffSubscriptionRequest, "signal">,
|
||||
): Promise<CheckoutDiffSnapshotPayload>;
|
||||
@@ -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 },
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<string | null> {
|
||||
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<string | null> {
|
||||
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<boolean> {
|
||||
}
|
||||
}
|
||||
|
||||
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<ReadonlyMap<
|
||||
* binary-only (no textual hunks). Throws on git failure (e.g. an unknown sha),
|
||||
* which the caller maps to a typed checkout error.
|
||||
*/
|
||||
// COMPAT(diffSubscription): added in v0.15.0, remove after 2027-05-06.
|
||||
export async function getCommitFileDiff({
|
||||
cwd,
|
||||
sha,
|
||||
@@ -3588,6 +3578,21 @@ async function resolveCheckoutDiffRefs(
|
||||
if (compare.mode === "uncommitted") {
|
||||
return { baseRef: "HEAD", includeUntracked: true };
|
||||
}
|
||||
const bestBaseRef = await resolveCheckoutDiffBase(cwd, compare, context);
|
||||
if (!bestBaseRef) return null;
|
||||
return {
|
||||
baseRef: (await tryResolveMergeBase(cwd, bestBaseRef)) ?? bestBaseRef,
|
||||
targetRef: compare.mode === "base" ? "HEAD" : undefined,
|
||||
includeUntracked: compare.mode === "all",
|
||||
};
|
||||
}
|
||||
|
||||
export async function resolveCheckoutDiffBase(
|
||||
cwd: string,
|
||||
compare: CheckoutDiffCompare,
|
||||
context?: CheckoutContext,
|
||||
): Promise<string | null> {
|
||||
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 });
|
||||
|
||||
@@ -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<string | null> {
|
||||
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;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user