Keep Git status and diffs current across platforms (#2775)

* fix(server): make Git observation cross-platform

Platform-specific watchers and forced full refreshes amplified Git processes and could miss nested changes. Share checkout and repository observation internally, scope final-state refreshes, and bound degraded polling so streamed summaries and diffs stay correct under bursts.

* fix(server): refresh Git worktree state on metadata changes

* fix(server): keep Git watcher runtime inputs current

* fix(server): normalize watcher runtime boundaries

* fix(server): reconcile Git observer topology

* test(server): report Git watcher event batches

* fix(server): close Git observation gaps

* chore(ci): remove temporary Git observation jobs
This commit is contained in:
Mohamed Boudra
2026-08-02 12:20:20 +02:00
committed by GitHub
parent f888bae04d
commit 0a106dd35f
16 changed files with 3558 additions and 631 deletions
+1 -1
View File
@@ -1 +1 @@
sha256-5t8wlskctuVk/xt8rAUDN9sexSimURt3it9mmMRXaUQ=
sha256-UunIQIS/Y2LhJWqhZMcU1CM9aVOwQUoKKR5z/kj+JDE=
+275 -2
View File
@@ -9509,6 +9509,280 @@
"node": "^20.19.0 || >=22.12.0"
}
},
"node_modules/@parcel/watcher": {
"version": "2.6.0",
"resolved": "https://registry.npmjs.org/@parcel/watcher/-/watcher-2.6.0.tgz",
"integrity": "sha512-7FNeNl8NCE7aINx7WXiKQrPYZWC/hvrTsmk6zmxbI7LTXE7hVek/n8AfVgpe2y82zl3w0HvCHN0bVKMBoJcC0w==",
"hasInstallScript": true,
"license": "MIT",
"dependencies": {
"detect-libc": "^2.0.3",
"is-glob": "^4.0.3",
"node-addon-api": "^7.0.0",
"picomatch": "^4.0.4"
},
"engines": {
"node": ">= 10.0.0"
},
"funding": {
"type": "opencollective",
"url": "https://opencollective.com/parcel"
},
"optionalDependencies": {
"@parcel/watcher-android-arm64": "2.6.0",
"@parcel/watcher-darwin-arm64": "2.6.0",
"@parcel/watcher-darwin-x64": "2.6.0",
"@parcel/watcher-freebsd-x64": "2.6.0",
"@parcel/watcher-linux-arm-glibc": "2.6.0",
"@parcel/watcher-linux-arm-musl": "2.6.0",
"@parcel/watcher-linux-arm64-glibc": "2.6.0",
"@parcel/watcher-linux-arm64-musl": "2.6.0",
"@parcel/watcher-linux-x64-glibc": "2.6.0",
"@parcel/watcher-linux-x64-musl": "2.6.0",
"@parcel/watcher-win32-arm64": "2.6.0",
"@parcel/watcher-win32-x64": "2.6.0"
}
},
"node_modules/@parcel/watcher-android-arm64": {
"version": "2.6.0",
"resolved": "https://registry.npmjs.org/@parcel/watcher-android-arm64/-/watcher-android-arm64-2.6.0.tgz",
"integrity": "sha512-trgpLSCKRC/huFjXX/Smh+0sWe4+YtKfktIToiMl59ghz7z+qkH6kMvNnUbLyRs9N11t8l4svSCs1+5B3rOAhA==",
"cpu": [
"arm64"
],
"license": "MIT",
"optional": true,
"os": [
"android"
],
"engines": {
"node": ">= 10.0.0"
},
"funding": {
"type": "opencollective",
"url": "https://opencollective.com/parcel"
}
},
"node_modules/@parcel/watcher-darwin-arm64": {
"version": "2.6.0",
"resolved": "https://registry.npmjs.org/@parcel/watcher-darwin-arm64/-/watcher-darwin-arm64-2.6.0.tgz",
"integrity": "sha512-Y3QV0gl7Q1zbfueunkWIERICbEojQFCgpyG7YqOGNFLsckXyI1xu9mAIUpKY9QBYzBtSkN8dBPwd3yiAO9ovMw==",
"cpu": [
"arm64"
],
"license": "MIT",
"optional": true,
"os": [
"darwin"
],
"engines": {
"node": ">= 10.0.0"
},
"funding": {
"type": "opencollective",
"url": "https://opencollective.com/parcel"
}
},
"node_modules/@parcel/watcher-darwin-x64": {
"version": "2.6.0",
"resolved": "https://registry.npmjs.org/@parcel/watcher-darwin-x64/-/watcher-darwin-x64-2.6.0.tgz",
"integrity": "sha512-Ohv6OpzhUfKYD7Beb8kDvG0jbIxORCYY1JRdZnaBtnjjkJxgD7ZVL0nw2sCYd0yTMKTvz3nnTnOF3cDifK+kvw==",
"cpu": [
"x64"
],
"license": "MIT",
"optional": true,
"os": [
"darwin"
],
"engines": {
"node": ">= 10.0.0"
},
"funding": {
"type": "opencollective",
"url": "https://opencollective.com/parcel"
}
},
"node_modules/@parcel/watcher-freebsd-x64": {
"version": "2.6.0",
"resolved": "https://registry.npmjs.org/@parcel/watcher-freebsd-x64/-/watcher-freebsd-x64-2.6.0.tgz",
"integrity": "sha512-5HmXvDgs8VK+74jF9y9/2FE3/OnlcKmc56tjmSrEuZjpSZOGL+fvAu+HKJBdPs9uwoP2hE6TlSUpXZ/C5jUFmQ==",
"cpu": [
"x64"
],
"license": "MIT",
"optional": true,
"os": [
"freebsd"
],
"engines": {
"node": ">= 10.0.0"
},
"funding": {
"type": "opencollective",
"url": "https://opencollective.com/parcel"
}
},
"node_modules/@parcel/watcher-linux-arm-glibc": {
"version": "2.6.0",
"resolved": "https://registry.npmjs.org/@parcel/watcher-linux-arm-glibc/-/watcher-linux-arm-glibc-2.6.0.tgz",
"integrity": "sha512-Ps/hui3A+vMbjdqlqAowK2ZL8+BO8dBjxeWXj6npTBs3jx4wWmbPpaLuqwrQrSqIVMCnpWo238bJ1U37GhQOYg==",
"cpu": [
"arm"
],
"license": "MIT",
"optional": true,
"os": [
"linux"
],
"engines": {
"node": ">= 10.0.0"
},
"funding": {
"type": "opencollective",
"url": "https://opencollective.com/parcel"
}
},
"node_modules/@parcel/watcher-linux-arm-musl": {
"version": "2.6.0",
"resolved": "https://registry.npmjs.org/@parcel/watcher-linux-arm-musl/-/watcher-linux-arm-musl-2.6.0.tgz",
"integrity": "sha512-9c6AUHgHoG+IY88MRIHupztQiQnrbqHYQjkM2btA+Bf/wQnQMuiD0Wfk1EVv3TlNT3x41uU71rn6E4xh/+zvkw==",
"cpu": [
"arm"
],
"license": "MIT",
"optional": true,
"os": [
"linux"
],
"engines": {
"node": ">= 10.0.0"
},
"funding": {
"type": "opencollective",
"url": "https://opencollective.com/parcel"
}
},
"node_modules/@parcel/watcher-linux-arm64-glibc": {
"version": "2.6.0",
"resolved": "https://registry.npmjs.org/@parcel/watcher-linux-arm64-glibc/-/watcher-linux-arm64-glibc-2.6.0.tgz",
"integrity": "sha512-yHRqS2owEXe6Hic9z6Mh1ECsCd+ODVOGvZDyciqRd21+v+o+DnXMOrw50DSpIG2sb8GPEaPPmfeCAWKPJdq46g==",
"cpu": [
"arm64"
],
"license": "MIT",
"optional": true,
"os": [
"linux"
],
"engines": {
"node": ">= 10.0.0"
},
"funding": {
"type": "opencollective",
"url": "https://opencollective.com/parcel"
}
},
"node_modules/@parcel/watcher-linux-arm64-musl": {
"version": "2.6.0",
"resolved": "https://registry.npmjs.org/@parcel/watcher-linux-arm64-musl/-/watcher-linux-arm64-musl-2.6.0.tgz",
"integrity": "sha512-WhB2e/V7rqdHHWZusBSPuy5Ei8S6lSz6FE5TKKQz5h3a0O+C+mhY7vxU9b/stqvMb8beLnPY82ZrFTLKs+SrKA==",
"cpu": [
"arm64"
],
"license": "MIT",
"optional": true,
"os": [
"linux"
],
"engines": {
"node": ">= 10.0.0"
},
"funding": {
"type": "opencollective",
"url": "https://opencollective.com/parcel"
}
},
"node_modules/@parcel/watcher-linux-x64-glibc": {
"version": "2.6.0",
"resolved": "https://registry.npmjs.org/@parcel/watcher-linux-x64-glibc/-/watcher-linux-x64-glibc-2.6.0.tgz",
"integrity": "sha512-ulGE6x6Oz6iAwg75T8YQSoguBWasniIbX+QWpaYPcCnDOpdWX3k+4xbEYPZVLxOuoJI+svJJPD3sEj8G7lrQ3A==",
"cpu": [
"x64"
],
"license": "MIT",
"optional": true,
"os": [
"linux"
],
"engines": {
"node": ">= 10.0.0"
},
"funding": {
"type": "opencollective",
"url": "https://opencollective.com/parcel"
}
},
"node_modules/@parcel/watcher-linux-x64-musl": {
"version": "2.6.0",
"resolved": "https://registry.npmjs.org/@parcel/watcher-linux-x64-musl/-/watcher-linux-x64-musl-2.6.0.tgz",
"integrity": "sha512-tkBYKt7YQrjIJWYDnto2YgO8MRkjlMTSNoRHzsXinBqbLdeOM3L32wPZJvIZxqaLMfSlS/4sUjH/6STVP/XDLw==",
"cpu": [
"x64"
],
"license": "MIT",
"optional": true,
"os": [
"linux"
],
"engines": {
"node": ">= 10.0.0"
},
"funding": {
"type": "opencollective",
"url": "https://opencollective.com/parcel"
}
},
"node_modules/@parcel/watcher-win32-arm64": {
"version": "2.6.0",
"resolved": "https://registry.npmjs.org/@parcel/watcher-win32-arm64/-/watcher-win32-arm64-2.6.0.tgz",
"integrity": "sha512-gIZAP23jaHjGWasY/TY6yL7NHFClf0Ga7FN+iINvk+KN94rhm94lYZhFsbYFNcA04/onvGD9kKmiJLJB2HbNwQ==",
"cpu": [
"arm64"
],
"license": "MIT",
"optional": true,
"os": [
"win32"
],
"engines": {
"node": ">= 10.0.0"
},
"funding": {
"type": "opencollective",
"url": "https://opencollective.com/parcel"
}
},
"node_modules/@parcel/watcher-win32-x64": {
"version": "2.6.0",
"resolved": "https://registry.npmjs.org/@parcel/watcher-win32-x64/-/watcher-win32-x64-2.6.0.tgz",
"integrity": "sha512-cA+/pXV2YkfxlIcXOQ5fSWqAzzPyD78/x5qbK/I0vUkrlYHA8TIz+MXjAbGouguKVSI4bOmkTSJ1/poVSsgt+A==",
"cpu": [
"x64"
],
"license": "MIT",
"optional": true,
"os": [
"win32"
],
"engines": {
"node": ">= 10.0.0"
},
"funding": {
"type": "opencollective",
"url": "https://opencollective.com/parcel"
}
},
"node_modules/@pinojs/redact": {
"version": "0.4.0",
"resolved": "https://registry.npmjs.org/@pinojs/redact/-/redact-0.4.0.tgz",
@@ -22991,7 +23265,6 @@
"version": "2.1.1",
"resolved": "https://registry.npmjs.org/is-extglob/-/is-extglob-2.1.1.tgz",
"integrity": "sha512-SbKbANkN603Vi4jEZv49LeVJMn4yGwsbzZworEoyEiutsN3nJYdbO36zfhGJ6QEDpOZIFkDtnq5JRxmvl3jsoQ==",
"dev": true,
"license": "MIT",
"engines": {
"node": ">=0.10.0"
@@ -23045,7 +23318,6 @@
"version": "4.0.3",
"resolved": "https://registry.npmjs.org/is-glob/-/is-glob-4.0.3.tgz",
"integrity": "sha512-xelSayHH36ZgE7ZWhli7pW34hNbNl8Ojv5KVmkJD4hBdD3th8Tfk9vYasLM+mXWOZhFkgZfxhLSnrwRr4elSSg==",
"dev": true,
"license": "MIT",
"dependencies": {
"is-extglob": "^2.1.1"
@@ -37924,6 +38196,7 @@
"@isaacs/ttlcache": "^2.1.4",
"@modelcontextprotocol/sdk": "^1.20.1",
"@opencode-ai/sdk": "1.14.46",
"@parcel/watcher": "^2.6.0",
"@xterm/headless": "^6.0.0",
"ai": "5.0.78",
"ajv": "^8.20.0",
+2
View File
@@ -45,6 +45,7 @@
"typecheck": "tsgo -p tsconfig.server.typecheck.json --noEmit",
"generate:config-schema": "tsx scripts/generate-config-schema.ts",
"measure:agent-tools": "tsx scripts/measure-agent-tools-context.ts",
"measure:git-observation": "tsx scripts/measure-git-observation.ts",
"speech:models": "tsx scripts/list-speech-models.ts",
"speech:download": "tsx scripts/download-speech-models.ts",
"speech:transcribe:local": "tsx scripts/transcribe-local-wav.ts",
@@ -75,6 +76,7 @@
"@isaacs/ttlcache": "^2.1.4",
"@modelcontextprotocol/sdk": "^1.20.1",
"@opencode-ai/sdk": "1.14.46",
"@parcel/watcher": "^2.6.0",
"@xterm/headless": "^6.0.0",
"ai": "5.0.78",
"ajv": "^8.20.0",
@@ -0,0 +1,414 @@
import { execFileSync } from "node:child_process";
import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";
import { writeFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import path from "node:path";
import parcelWatcher from "@parcel/watcher";
import type pino from "pino";
import {
getCheckoutDiff,
getCheckoutSnapshotFacts,
getCheckoutWorktreeState,
} from "../src/utils/checkout-git.js";
import {
startGitCommandMetrics,
stopGitCommandMetrics,
type GitCommandMetricsSnapshot,
} from "../src/utils/run-git-command.js";
import { CheckoutDiffManager } from "../src/server/checkout-diff-manager.js";
import {
WorkspaceGitServiceImpl,
type WorkspaceGitRuntimeSnapshot,
} from "../src/server/workspace-git-service.js";
interface ProducerCounts {
structural: number;
worktree: number;
detailedDiff: number;
}
interface PhaseResult {
elapsedMs: number;
gitCommands: number;
failedGitCommands: number;
maxConcurrentGitCommands: number;
byCommand: Record<string, number>;
producers: ProducerCounts;
queueAfter: {
inFlight: number;
queued: number;
};
}
function createLogger(): pino.Logger {
const logger = {
child: () => logger,
debug: () => {},
warn: () => {},
};
return logger as unknown as pino.Logger;
}
function runGit(cwd: string, args: string[]): void {
execFileSync("git", args, { cwd, stdio: "pipe" });
}
function sleep(ms: number): Promise<void> {
return new Promise((resolve) => setTimeout(resolve, ms));
}
async function waitFor(predicate: () => boolean, label: string, timeoutMs = 8_000): Promise<void> {
const deadline = Date.now() + timeoutMs;
while (!predicate()) {
if (Date.now() >= deadline) {
throw new Error(`Timed out waiting for ${label}`);
}
await sleep(25);
}
}
function summarizeCommands(metrics: GitCommandMetricsSnapshot): Record<string, number> {
const counts = new Map<string, number>();
for (const command of metrics.commands) {
const key = command.args.slice(0, 3).join(" ");
counts.set(key, (counts.get(key) ?? 0) + 1);
}
return Object.fromEntries([...counts].sort(([left], [right]) => left.localeCompare(right)));
}
function subtractCounts(after: ProducerCounts, before: ProducerCounts): ProducerCounts {
return {
structural: after.structural - before.structural,
worktree: after.worktree - before.worktree,
detailedDiff: after.detailedDiff - before.detailedDiff,
};
}
function snapshotCounts(counts: ProducerCounts): ProducerCounts {
return { ...counts };
}
function createTrackedSubscriber(failWorktreeRoot: string | null) {
const subscriptions = new Set<parcelWatcher.AsyncSubscription>();
const subscribe = async (...args: Parameters<typeof parcelWatcher.subscribe>) => {
if (failWorktreeRoot && path.resolve(args[0]) === failWorktreeRoot) {
throw new Error("measurement watcher setup failure");
}
const subscription = await parcelWatcher.subscribe(...args);
subscriptions.add(subscription);
return {
unsubscribe: async () => {
await subscription.unsubscribe();
subscriptions.delete(subscription);
},
};
};
return { subscribe, subscriptions };
}
function createMeasuredService(input: {
repoDir: string;
paseoHome: string;
failWorktreeWatch?: boolean;
}) {
const counts: ProducerCounts = { structural: 0, worktree: 0, detailedDiff: 0 };
const watcher = createTrackedSubscriber(input.failWorktreeWatch ? input.repoDir : null);
const service = new WorkspaceGitServiceImpl({
logger: createLogger(),
paseoHome: input.paseoHome,
deps: {
subscribe: watcher.subscribe,
getWorkspaceGitSelfHealPhaseMs: () => 60_000,
getCheckoutSnapshotFacts: async (...args) => {
counts.structural += 1;
return getCheckoutSnapshotFacts(...args);
},
getCheckoutWorktreeState: async (...args) => {
counts.worktree += 1;
return getCheckoutWorktreeState(...args);
},
getCheckoutDiff: async (...args) => {
counts.detailedDiff += 1;
return getCheckoutDiff(...args);
},
},
});
return { counts, service, watcher };
}
async function measurePhase(input: {
service: WorkspaceGitServiceImpl;
counts: ProducerCounts;
action: () => Promise<void>;
}): Promise<PhaseResult> {
const countsBefore = snapshotCounts(input.counts);
const startedAtMs = Date.now();
startGitCommandMetrics();
let metrics: GitCommandMetricsSnapshot;
try {
await input.action();
await waitFor(() => {
const serviceMetrics = input.service.getMetrics();
return (
serviceMetrics.workspaceRefreshInFlightCount === 0 &&
serviceMetrics.workspaceRefreshQueuedCount === 0
);
}, "observation queue to drain");
} finally {
metrics = stopGitCommandMetrics();
}
const serviceMetrics = input.service.getMetrics();
return {
elapsedMs: Date.now() - startedAtMs,
gitCommands: metrics.total,
failedGitCommands: metrics.failed,
maxConcurrentGitCommands: metrics.maxConcurrent,
byCommand: summarizeCommands(metrics),
producers: subtractCounts(input.counts, countsBefore),
queueAfter: {
inFlight: serviceMetrics.workspaceRefreshInFlightCount,
queued: serviceMetrics.workspaceRefreshQueuedCount,
},
};
}
async function closeMeasuredService(input: {
service: WorkspaceGitServiceImpl;
subscriptions: Set<parcelWatcher.AsyncSubscription>;
}): Promise<void> {
input.service.dispose();
await waitFor(() => input.subscriptions.size === 0, "watchers to close");
}
async function main(): Promise<void> {
const tempDir = mkdtempSync(path.join(tmpdir(), "paseo-git-observation-measurement-"));
const repoDir = path.join(tempDir, "repo");
const paseoHome = path.join(tempDir, "paseo-home");
const trackedPath = path.join(repoDir, "tracked.txt");
const ignoredDir = path.join(repoDir, "build");
mkdirSync(repoDir, { recursive: true });
mkdirSync(ignoredDir, { recursive: true });
writeFileSync(path.join(ignoredDir, "probe.txt"), "ignored\n");
writeFileSync(path.join(repoDir, ".gitignore"), "build/\n");
writeFileSync(trackedPath, "base\n");
runGit(repoDir, ["init", "-b", "main"]);
runGit(repoDir, ["config", "user.email", "measurement@paseo.local"]);
runGit(repoDir, ["config", "user.name", "Paseo Measurement"]);
runGit(repoDir, ["add", ".gitignore", "tracked.txt"]);
runGit(repoDir, ["commit", "-m", "fixture"]);
runGit(repoDir, ["checkout", "-b", "feature"]);
const phases: Record<string, PhaseResult> = {};
let healthy: ReturnType<typeof createMeasuredService> | null = null;
let degraded: ReturnType<typeof createMeasuredService> | null = null;
let summarySubscription: { unsubscribe: () => void } | null = null;
let diffSubscription: { unsubscribe: () => void } | null = null;
let diffManager: CheckoutDiffManager | null = null;
try {
healthy = createMeasuredService({ repoDir, paseoHome });
let latestSummary: WorkspaceGitRuntimeSnapshot | null = null;
let latestDiffAdditions = 0;
startGitCommandMetrics();
const bootstrapStartedAtMs = Date.now();
const bootstrapCounts = snapshotCounts(healthy.counts);
summarySubscription = healthy.service.registerWorkspace({ cwd: repoDir }, (snapshot) => {
latestSummary = snapshot;
});
diffManager = new CheckoutDiffManager({
logger: createLogger(),
paseoHome,
workspaceGitService: healthy.service,
});
const openedDiff = await diffManager.subscribe(
{ cwd: repoDir, compare: { mode: "uncommitted" } },
(payload) => {
latestDiffAdditions = payload.files[0]?.additions ?? 0;
},
);
diffSubscription = openedDiff;
await waitFor(
() =>
healthy?.service.peekSnapshot(repoDir) !== null &&
healthy.service.getMetrics().workspaceObservationSetupInFlightCount === 0 &&
healthy.watcher.subscriptions.size === 2,
"healthy bootstrap",
);
await sleep(250);
const bootstrapMetrics = stopGitCommandMetrics();
const bootstrapServiceMetrics = healthy.service.getMetrics();
phases.bootstrap = {
elapsedMs: Date.now() - bootstrapStartedAtMs,
gitCommands: bootstrapMetrics.total,
failedGitCommands: bootstrapMetrics.failed,
maxConcurrentGitCommands: bootstrapMetrics.maxConcurrent,
byCommand: summarizeCommands(bootstrapMetrics),
producers: subtractCounts(healthy.counts, bootstrapCounts),
queueAfter: {
inFlight: bootstrapServiceMetrics.workspaceRefreshInFlightCount,
queued: bootstrapServiceMetrics.workspaceRefreshQueuedCount,
},
};
phases.legacyFullRefreshReference = await measurePhase({
service: healthy.service,
counts: healthy.counts,
action: async () => {
await Promise.all([
healthy?.service.getSnapshot(repoDir, {
force: true,
includeForge: false,
reason: "measurement-legacy-full-refresh-reference",
}),
healthy?.service.getCheckoutDiff(
repoDir,
{ mode: "uncommitted", includeStructured: true },
{ force: true, reason: "measurement-legacy-full-refresh-reference" },
),
]);
},
});
phases.idle = await measurePhase({
service: healthy.service,
counts: healthy.counts,
action: () => sleep(750),
});
phases.activeDiff = await measurePhase({
service: healthy.service,
counts: healthy.counts,
action: async () => {
writeFileSync(trackedPath, "first\nsecond\n");
await waitFor(
() => (latestSummary?.git.diffStat?.additions ?? 0) >= 2 && latestDiffAdditions >= 2,
"active diff payload",
);
await sleep(1_200);
},
});
phases.editBurst = await measurePhase({
service: healthy.service,
counts: healthy.counts,
action: async () => {
for (let index = 0; index < 100; index += 1) {
writeFileSync(
trackedPath,
Array.from({ length: index + 2 }, (_, line) => `burst-${index}-${line}`).join("\n") +
"\n",
);
}
await waitFor(
() => (latestSummary?.git.diffStat?.additions ?? 0) >= 100 && latestDiffAdditions >= 100,
"edit burst final payload",
);
await sleep(1_200);
},
});
phases.ignoredStorm = await measurePhase({
service: healthy.service,
counts: healthy.counts,
action: async () => {
for (let index = 0; index < 100; index += 1) {
writeFileSync(path.join(ignoredDir, `artifact-${index}.txt`), `${index}\n`);
}
await sleep(1_200);
},
});
if (phases.idle.gitCommands !== 0 || phases.ignoredStorm.gitCommands !== 0) {
throw new Error("Idle and ignored-storm phases must perform zero Git commands");
}
if (phases.editBurst.producers.worktree > 2 || phases.editBurst.producers.detailedDiff > 2) {
throw new Error("One hundred edits exceeded two final-state refresh generations");
}
diffSubscription.unsubscribe();
diffSubscription = null;
summarySubscription.unsubscribe();
summarySubscription = null;
diffManager.dispose();
diffManager = null;
await closeMeasuredService({
service: healthy.service,
subscriptions: healthy.watcher.subscriptions,
});
healthy = null;
degraded = createMeasuredService({ repoDir, paseoHome, failWorktreeWatch: true });
startGitCommandMetrics();
const degradedBootstrapStartedAtMs = Date.now();
const degradedBootstrapCounts = snapshotCounts(degraded.counts);
summarySubscription = degraded.service.registerWorkspace({ cwd: repoDir }, () => {});
await waitFor(
() =>
degraded?.service.peekSnapshot(repoDir) !== null &&
degraded.service.getMetrics().workspaceObservationSetupInFlightCount === 0,
"degraded bootstrap",
);
await sleep(250);
const degradedBootstrapMetrics = stopGitCommandMetrics();
const degradedBootstrapServiceMetrics = degraded.service.getMetrics();
phases.degradedBootstrap = {
elapsedMs: Date.now() - degradedBootstrapStartedAtMs,
gitCommands: degradedBootstrapMetrics.total,
failedGitCommands: degradedBootstrapMetrics.failed,
maxConcurrentGitCommands: degradedBootstrapMetrics.maxConcurrent,
byCommand: summarizeCommands(degradedBootstrapMetrics),
producers: subtractCounts(degraded.counts, degradedBootstrapCounts),
queueAfter: {
inFlight: degradedBootstrapServiceMetrics.workspaceRefreshInFlightCount,
queued: degradedBootstrapServiceMetrics.workspaceRefreshQueuedCount,
},
};
phases.degradedWatcher = await measurePhase({
service: degraded.service,
counts: degraded.counts,
action: async () => {
const refreshCount = degraded?.counts.worktree ?? 0;
await waitFor(
() => (degraded?.counts.worktree ?? 0) > refreshCount,
"degraded watcher poll",
8_000,
);
},
});
const report = {
generatedAt: new Date().toISOString(),
phases,
};
const outputPath = process.env.PASEO_GIT_OBSERVATION_REPORT?.trim();
if (outputPath) {
await writeFile(path.resolve(outputPath), `${JSON.stringify(report, null, 2)}\n`, "utf8");
}
process.stdout.write(`GIT_OBSERVATION_MEASUREMENT ${JSON.stringify(report)}\n`);
} finally {
stopGitCommandMetrics();
diffSubscription?.unsubscribe();
summarySubscription?.unsubscribe();
diffManager?.dispose();
if (healthy) {
await closeMeasuredService({
service: healthy.service,
subscriptions: healthy.watcher.subscriptions,
}).catch(() => {});
}
if (degraded) {
await closeMeasuredService({
service: degraded.service,
subscriptions: degraded.watcher.subscriptions,
}).catch(() => {});
}
rmSync(tempDir, { recursive: true, force: true });
}
}
main().catch((error) => {
process.stderr.write(`${error instanceof Error ? error.stack : String(error)}\n`);
process.exitCode = 1;
});
@@ -12,7 +12,7 @@ vi.mock("./checkout-git-utils.js", () => ({
import type pino from "pino";
import { CheckoutDiffManager } from "./checkout-diff-manager.js";
import type { WorkspaceGitService } from "./workspace-git-service.js";
import type { WorkspaceGitRuntimeSnapshot, WorkspaceGitService } from "./workspace-git-service.js";
interface Deferred<T> {
promise: Promise<T>;
@@ -27,6 +27,36 @@ function createDeferred<T>(): Deferred<T> {
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<{
cwd: string;
@@ -36,6 +66,9 @@ function createPendingManager() {
}> = [];
const workspaceGitService = {
getCheckoutDiff: async () => ({ diff: "", structured: [] }),
getSnapshot: async () => createWorkspaceSnapshot(),
peekSnapshot: () => null,
registerWorkspace: () => ({ unsubscribe: () => {} }),
requestWorkingTreeWatch: (cwd: string, onChange: () => void) => {
const pending = createDeferred<{ repoRoot: string | null; unsubscribe: () => void }>();
const watch = {
@@ -79,7 +112,9 @@ describe("CheckoutDiffManager", () => {
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 {
@@ -91,7 +126,13 @@ describe("CheckoutDiffManager", () => {
const workspaceGitService = {
subscribe: vi.fn(),
peekSnapshot: vi.fn(),
getSnapshot: vi.fn(),
registerWorkspace: vi.fn(
(_params: { cwd: string }, listener: (snapshot: WorkspaceGitRuntimeSnapshot) => void) => {
onWorkspaceSnapshot = listener;
return { unsubscribe: workspaceUnsubscribe };
},
),
getSnapshot: vi.fn(async () => createWorkspaceSnapshot()),
getCheckoutDiff:
options?.getCheckoutDiffImplementation ?? vi.fn(async () => ({ diff: "", structured: [] })),
refresh: vi.fn(),
@@ -117,6 +158,8 @@ describe("CheckoutDiffManager", () => {
mockRequestWorkingTreeWatch,
unsubscribe,
getOnChange: () => onChange,
getOnWorkspaceSnapshot: () => onWorkspaceSnapshot,
workspaceUnsubscribe,
};
}
@@ -301,6 +344,127 @@ describe("CheckoutDiffManager", () => {
});
});
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 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,
});
});
});
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,
});
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();
});
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 });
@@ -1,6 +1,6 @@
import type pino from "pino";
import type { SubscribeCheckoutDiffRequest, SessionOutboundMessage } from "./messages.js";
import type { WorkspaceGitService } from "./workspace-git-service.js";
import type { WorkspaceGitRuntimeSnapshot, WorkspaceGitService } from "./workspace-git-service.js";
import { expandTilde } from "../utils/path.js";
import { toCheckoutError } from "./checkout-git-utils.js";
@@ -8,7 +8,11 @@ const CHECKOUT_DIFF_WATCH_DEBOUNCE_MS = 150;
type CheckoutDiffWorkspace = Pick<
WorkspaceGitService,
"getCheckoutDiff" | "requestWorkingTreeWatch"
| "getCheckoutDiff"
| "getSnapshot"
| "peekSnapshot"
| "registerWorkspace"
| "requestWorkingTreeWatch"
>;
export type CheckoutDiffCompareInput = SubscribeCheckoutDiffRequest["compare"];
@@ -32,9 +36,15 @@ interface CheckoutDiffWatchTarget {
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;
@@ -162,9 +172,51 @@ export class CheckoutDiffManager {
}
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,
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,
@@ -179,13 +231,16 @@ export class CheckoutDiffManager {
}
}
private scheduleTargetRefresh(target: CheckoutDiffWatchTarget): void {
private scheduleTargetRefresh(target: CheckoutDiffWatchTarget, force = true): void {
target.pendingDebounceForce ||= force;
if (target.debounceTimer) {
clearTimeout(target.debounceTimer);
}
target.debounceTimer = setTimeout(() => {
target.debounceTimer = null;
void this.refreshTarget(target);
const pendingForce = target.pendingDebounceForce;
target.pendingDebounceForce = false;
void this.refreshTarget(target, pendingForce);
}, CHECKOUT_DIFF_WATCH_DEBOUNCE_MS);
}
@@ -235,19 +290,22 @@ export class CheckoutDiffManager {
}
}
private async refreshTarget(target: CheckoutDiffWatchTarget): Promise<void> {
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: true,
reason: "working-tree-watch",
force: currentForce,
...(currentForce ? { reason: "working-tree-watch" } : {}),
});
target.latestPayload = snapshot;
const fingerprint = JSON.stringify(snapshot);
@@ -257,6 +315,7 @@ export class CheckoutDiffManager {
listener(snapshot);
}
}
currentForce = target.refreshQueuedForce;
} while (target.refreshQueued);
})();
@@ -281,9 +340,15 @@ export class CheckoutDiffManager {
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,
@@ -293,6 +358,27 @@ export class CheckoutDiffManager {
}
private async openTarget(target: CheckoutDiffWatchTarget): Promise<void> {
if (target.compare.mode === "base") {
const snapshot =
this.workspaceGitService.peekSnapshot(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;
return;
}
const { repoRoot, unsubscribe } = await this.workspaceGitService.requestWorkingTreeWatch(
target.cwd,
() => this.scheduleTargetRefresh(target),
@@ -0,0 +1,383 @@
import { mkdirSync, mkdtempSync, readFileSync, realpathSync, rmSync, writeFileSync } from "node:fs";
import { tmpdir } from "node:os";
import path from "node:path";
import parcelWatcher from "@parcel/watcher";
import type pino from "pino";
import { afterEach, expect, test, vi } from "vitest";
import type { CheckoutSnapshotFacts, CheckoutStatusGit } from "../utils/checkout-git.js";
import { CheckoutDiffManager } from "./checkout-diff-manager.js";
import { WorkspaceGitServiceImpl } from "./workspace-git-service.js";
function createLogger(): pino.Logger {
const logger = {
child: () => logger,
debug: vi.fn(),
warn: vi.fn(),
};
return logger as unknown as pino.Logger;
}
function createFacts(cwd: string): CheckoutSnapshotFacts {
return {
isGit: true,
worktreeRoot: cwd,
currentBranch: "main",
remoteUrl: null,
absoluteGitDir: path.join(cwd, ".git"),
gitCommonDir: path.join(cwd, ".git"),
paseoWorktree: { isPaseoOwnedWorktree: false },
storedBaseRef: null,
resolvedBaseRef: "main",
mainRepoRoot: null,
comparisonBaseRef: null,
branchRemoteName: null,
branchMergeRef: null,
pullRequestLookupTarget: { headRef: "main" },
};
}
function createStatus(cwd: string): CheckoutStatusGit {
return {
isGit: true,
repoRoot: cwd,
mainRepoRoot: null,
currentBranch: "main",
isDirty: false,
baseRef: "main",
aheadBehind: { ahead: 0, behind: 0 },
aheadOfOrigin: null,
behindOfOrigin: null,
hasRemote: false,
remoteUrl: null,
isPaseoOwnedWorktree: false,
};
}
const cleanup: Array<() => Promise<void>> = [];
afterEach(async () => {
while (cleanup.length > 0) {
await cleanup.pop()?.();
}
});
test("native recursive observation updates tracked state and prunes ignored storms", async () => {
const tempDir = realpathSync(mkdtempSync(path.join(tmpdir(), "paseo-git-observation-")));
const repoDir = path.join(tempDir, "repo");
const trackedPath = path.join(repoDir, "src", "tracked.txt");
const ignoredDir = path.join(repoDir, "build");
const remainingIgnoredDir = path.join(repoDir, "cache");
const newlyTrackedPath = path.join(ignoredDir, "tracked.txt");
mkdirSync(path.join(repoDir, ".git"), { recursive: true });
mkdirSync(path.dirname(trackedPath), { recursive: true });
mkdirSync(ignoredDir, { recursive: true });
mkdirSync(remainingIgnoredDir, { recursive: true });
writeFileSync(trackedPath, "base\n");
writeFileSync(newlyTrackedPath, "base\n");
let activeWatcherCount = 0;
let watcherStartCount = 0;
let onWorkingTreeWatcherStopped: (() => void) | null = null;
const deliveredEvents: Array<{
directory: string;
events: parcelWatcher.Event[];
}> = [];
const subscribe: typeof parcelWatcher.subscribe = async (directory, callback, options) => {
const subscription = await parcelWatcher.subscribe(
directory,
(error, events) => {
deliveredEvents.push({ directory, events });
callback(error, events);
},
options,
);
activeWatcherCount += 1;
watcherStartCount += 1;
return {
unsubscribe: async () => {
await subscription.unsubscribe();
activeWatcherCount -= 1;
if (directory === repoDir) {
const onStopped = onWorkingTreeWatcherStopped;
onWorkingTreeWatcherStopped = null;
onStopped?.();
}
},
};
};
const getCheckoutSnapshotFacts = vi.fn(async (cwd: string) => createFacts(cwd));
const getCheckoutStatus = vi.fn(async (cwd: string) => createStatus(cwd));
const getCheckoutShortstat = vi.fn(async () => ({ additions: 0, deletions: 0 }));
let observedPath = trackedPath;
let observedRelativePath = "src/tracked.txt";
const getCheckoutWorktreeState = vi.fn(async () => {
const additions = readFileSync(observedPath, "utf8").trim().split("\n").length;
return { isDirty: true, diffStat: { additions, deletions: 0 } };
});
const getCheckoutDiff = vi.fn(async () => {
const additions = readFileSync(observedPath, "utf8").trim().split("\n").length;
return {
diff: "",
structured: [
{ path: observedRelativePath, additions, deletions: 0, status: "modified" as const },
],
};
});
let buildIgnored = true;
const runGitCommand = vi.fn(async (args: string[]) => {
if (args[0] === "rev-parse") {
return {
stdout: `${repoDir}\n`,
stderr: "",
truncated: false,
exitCode: 0,
signal: null,
};
}
if (args[0] === "ls-files") {
return {
stdout: `${buildIgnored ? "build/\n" : ""}cache/\n`,
stderr: "",
truncated: false,
exitCode: 0,
signal: null,
};
}
throw new Error(`Unexpected Git command: ${args.join(" ")}`);
});
const service = new WorkspaceGitServiceImpl({
logger: createLogger(),
paseoHome: path.join(tempDir, "paseo-home"),
deps: {
subscribe,
getCheckoutSnapshotFacts,
getCheckoutStatus,
getCheckoutShortstat,
getCheckoutWorktreeState,
getCheckoutDiff,
runGitCommand,
getWorkspaceGitSelfHealPhaseMs: () => 7_000,
} as never,
});
const diffManager = new CheckoutDiffManager({
logger: createLogger(),
paseoHome: path.join(tempDir, "paseo-home"),
workspaceGitService: service,
});
const summaryListener = vi.fn();
const diffListener = vi.fn();
const summarySubscription = service.registerWorkspace({ cwd: repoDir }, summaryListener);
const diffSubscription = await diffManager.subscribe(
{ cwd: repoDir, compare: { mode: "uncommitted" } },
diffListener,
);
cleanup.push(async () => {
diffSubscription.unsubscribe();
summarySubscription.unsubscribe();
diffManager.dispose();
service.dispose();
await vi.waitFor(() => expect(activeWatcherCount).toBe(0), { timeout: 5_000 });
rmSync(tempDir, { recursive: true, force: true });
});
await vi.waitFor(
() => {
expect(activeWatcherCount).toBe(2);
expect(service.peekSnapshot(repoDir)).not.toBeNull();
expect(service.getMetrics()).toMatchObject({
workspaceObservationSetupInFlightCount: 0,
workspaceRefreshInFlightCount: 0,
workspaceRefreshQueuedCount: 0,
});
},
{ timeout: 5_000 },
);
// Parcel may deliver a startup batch before subscribe resolves. Let its
// already-scheduled consumer debounce finish inside the bootstrap phase.
await new Promise((resolve) => setTimeout(resolve, 250));
getCheckoutSnapshotFacts.mockClear();
getCheckoutStatus.mockClear();
getCheckoutShortstat.mockClear();
getCheckoutWorktreeState.mockClear();
getCheckoutDiff.mockClear();
runGitCommand.mockClear();
deliveredEvents.length = 0;
await new Promise((resolve) => setTimeout(resolve, 250));
expect(runGitCommand).not.toHaveBeenCalled();
expect(getCheckoutWorktreeState).not.toHaveBeenCalled();
expect(getCheckoutDiff, JSON.stringify(deliveredEvents)).not.toHaveBeenCalled();
expect(service.getMetrics().workspaceRefreshQueuedCount).toBe(0);
for (let index = 0; index < 100; index += 1) {
writeFileSync(path.join(ignoredDir, `artifact-${index}.txt`), `${index}\n`);
}
await new Promise((resolve) => setTimeout(resolve, 750));
expect(runGitCommand).not.toHaveBeenCalled();
expect(getCheckoutSnapshotFacts).not.toHaveBeenCalled();
expect(getCheckoutStatus).not.toHaveBeenCalled();
expect(getCheckoutShortstat).not.toHaveBeenCalled();
expect(getCheckoutWorktreeState).not.toHaveBeenCalled();
expect(getCheckoutDiff).not.toHaveBeenCalled();
expect(service.getMetrics().workspaceRefreshQueuedCount).toBe(0);
writeFileSync(trackedPath, "first\nsecond\n");
await vi.waitFor(
() => {
expect(summaryListener).toHaveBeenLastCalledWith(
expect.objectContaining({
git: expect.objectContaining({
isDirty: true,
diffStat: { additions: 2, deletions: 0 },
}),
}),
);
expect(diffListener).toHaveBeenLastCalledWith({
cwd: repoDir,
files: [
{
path: "src/tracked.txt",
additions: 2,
deletions: 0,
status: "modified",
},
],
error: null,
});
expect(service.getMetrics().workspaceRefreshQueuedCount).toBe(0);
},
{ timeout: 5_000 },
);
writeFileSync(trackedPath, "first\nsecond\nthird\n");
await vi.waitFor(
() => {
expect(summaryListener).toHaveBeenLastCalledWith(
expect.objectContaining({
git: expect.objectContaining({
diffStat: { additions: 3, deletions: 0 },
}),
}),
);
expect(diffListener).toHaveBeenLastCalledWith(
expect.objectContaining({
files: [expect.objectContaining({ additions: 3 })],
}),
);
expect(service.getMetrics()).toMatchObject({
workspaceRefreshInFlightCount: 0,
workspaceRefreshQueuedCount: 0,
});
},
{ timeout: 5_000 },
);
expect(getCheckoutSnapshotFacts).not.toHaveBeenCalled();
expect(getCheckoutStatus).not.toHaveBeenCalled();
buildIgnored = false;
observedPath = newlyTrackedPath;
observedRelativePath = "build/tracked.txt";
let editedDuringWatcherHandoff = false;
onWorkingTreeWatcherStopped = () => {
editedDuringWatcherHandoff = true;
writeFileSync(newlyTrackedPath, "first\nsecond\n");
};
writeFileSync(path.join(repoDir, ".git", "index"), "force-added\n");
await vi.waitFor(
() => {
expect(editedDuringWatcherHandoff).toBe(true);
expect(watcherStartCount).toBe(3);
expect(activeWatcherCount).toBe(2);
expect(summaryListener).toHaveBeenLastCalledWith(
expect.objectContaining({
git: expect.objectContaining({
isDirty: true,
diffStat: { additions: 2, deletions: 0 },
}),
}),
);
expect(diffListener).toHaveBeenLastCalledWith({
cwd: repoDir,
files: [
{
path: "build/tracked.txt",
additions: 2,
deletions: 0,
status: "modified",
},
],
error: null,
});
expect(service.getMetrics()).toMatchObject({
workspaceRefreshInFlightCount: 0,
workspaceRefreshQueuedCount: 0,
});
},
{ timeout: 8_000 },
);
getCheckoutSnapshotFacts.mockClear();
getCheckoutStatus.mockClear();
getCheckoutShortstat.mockClear();
getCheckoutWorktreeState.mockClear();
getCheckoutDiff.mockClear();
runGitCommand.mockClear();
summaryListener.mockClear();
diffListener.mockClear();
writeFileSync(newlyTrackedPath, "first\nsecond\nthird\nfourth\n");
await vi.waitFor(
() => {
expect(summaryListener).toHaveBeenLastCalledWith(
expect.objectContaining({
git: expect.objectContaining({
isDirty: true,
diffStat: { additions: 4, deletions: 0 },
}),
}),
);
expect(diffListener).toHaveBeenLastCalledWith({
cwd: repoDir,
files: [
{
path: "build/tracked.txt",
additions: 4,
deletions: 0,
status: "modified",
},
],
error: null,
});
expect(service.getMetrics()).toMatchObject({
workspaceRefreshInFlightCount: 0,
workspaceRefreshQueuedCount: 0,
});
},
{ timeout: 5_000 },
);
getCheckoutSnapshotFacts.mockClear();
getCheckoutStatus.mockClear();
getCheckoutShortstat.mockClear();
getCheckoutWorktreeState.mockClear();
getCheckoutDiff.mockClear();
runGitCommand.mockClear();
for (let index = 0; index < 100; index += 1) {
writeFileSync(path.join(remainingIgnoredDir, `artifact-${index}.txt`), `${index}\n`);
}
await new Promise((resolve) => setTimeout(resolve, 750));
expect(runGitCommand).not.toHaveBeenCalled();
expect(getCheckoutSnapshotFacts).not.toHaveBeenCalled();
expect(getCheckoutStatus).not.toHaveBeenCalled();
expect(getCheckoutShortstat).not.toHaveBeenCalled();
expect(getCheckoutWorktreeState).not.toHaveBeenCalled();
expect(getCheckoutDiff).not.toHaveBeenCalled();
expect(service.getMetrics().workspaceRefreshQueuedCount).toBe(0);
}, 15_000);
@@ -0,0 +1,929 @@
import path from "node:path";
import type pino from "pino";
import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
import type { CheckoutSnapshotFacts, CheckoutStatusGit } from "../utils/checkout-git.js";
import { CheckoutDiffManager } from "./checkout-diff-manager.js";
import { WorkspaceGitServiceImpl } from "./workspace-git-service.js";
const REPO_CWD = path.resolve("/tmp/paseo-observation-repo");
const GIT_DIR = path.join(REPO_CWD, ".git");
interface WatchEvent {
path: string;
type: "create" | "update" | "delete";
}
interface WatchRecord {
directory: string;
callback: (error: Error | null, events: WatchEvent[]) => void;
ignore: Array<string | RegExp>;
unsubscribe: ReturnType<typeof vi.fn>;
}
function createWatcherHarness(harnessOptions?: { failDirectories?: Set<string> }) {
const records: WatchRecord[] = [];
const subscribe = vi.fn(
async (
directory: string,
callback: WatchRecord["callback"],
options?: { ignore?: Array<string | RegExp> },
) => {
if (harnessOptions?.failDirectories?.has(directory)) {
throw new Error(`watch failed: ${directory}`);
}
const unsubscribe = vi.fn(async () => {});
records.push({
directory,
callback,
ignore: options?.ignore ?? [],
unsubscribe,
});
return { unsubscribe };
},
);
return { records, subscribe };
}
function createCheckoutFacts(cwd: string): CheckoutSnapshotFacts {
return {
isGit: true,
worktreeRoot: cwd,
currentBranch: "main",
remoteUrl: null,
absoluteGitDir: path.join(cwd, ".git"),
gitCommonDir: path.join(cwd, ".git"),
paseoWorktree: { isPaseoOwnedWorktree: false },
storedBaseRef: null,
resolvedBaseRef: "main",
mainRepoRoot: null,
comparisonBaseRef: null,
branchRemoteName: null,
branchMergeRef: null,
pullRequestLookupTarget: { headRef: "main" },
};
}
function createCheckoutStatus(
cwd: string,
overrides?: Partial<CheckoutStatusGit>,
): CheckoutStatusGit {
return {
isGit: true,
repoRoot: cwd,
mainRepoRoot: null,
currentBranch: "main",
isDirty: false,
baseRef: "main",
aheadBehind: { ahead: 0, behind: 0 },
aheadOfOrigin: null,
behindOfOrigin: null,
hasRemote: false,
remoteUrl: null,
isPaseoOwnedWorktree: false,
...overrides,
};
}
function createLogger(): pino.Logger {
const logger = {
child: () => logger,
debug: vi.fn(),
warn: vi.fn(),
};
return logger as unknown as pino.Logger;
}
function createDeferred<T>() {
let resolve!: (value: T | PromiseLike<T>) => void;
let reject!: (reason?: unknown) => void;
const promise = new Promise<T>((resolvePromise, rejectPromise) => {
resolve = resolvePromise;
reject = rejectPromise;
});
return { promise, reject, resolve };
}
function createService(
watcher: ReturnType<typeof createWatcherHarness>,
overrides?: Record<string, unknown>,
) {
const defaultGetCheckoutStatus = vi.fn(async (cwd: string) => createCheckoutStatus(cwd));
const defaultGetCheckoutShortstat = vi.fn(async () => null);
const getCheckoutStatus =
(overrides?.getCheckoutStatus as typeof defaultGetCheckoutStatus | undefined) ??
defaultGetCheckoutStatus;
const getCheckoutShortstat =
(overrides?.getCheckoutShortstat as typeof defaultGetCheckoutShortstat | undefined) ??
defaultGetCheckoutShortstat;
return new WorkspaceGitServiceImpl({
logger: createLogger(),
paseoHome: "/tmp/paseo-home",
deps: {
subscribe: watcher.subscribe,
getCheckoutSnapshotFacts: vi.fn(async (cwd: string) => createCheckoutFacts(cwd)),
getCheckoutStatus,
getCheckoutShortstat,
getCheckoutWorktreeState: vi.fn(async (cwd: string) => {
const status = await getCheckoutStatus(cwd);
if (!status.isGit) {
throw new Error("Expected a git checkout");
}
return {
isDirty: status.isDirty,
diffStat: await getCheckoutShortstat(),
};
}),
resolveAbsoluteGitDir: vi.fn(async () => GIT_DIR),
hasOriginRemote: vi.fn(async () => false),
runGitCommand: vi.fn(async () => ({
stdout: `${REPO_CWD}\n`,
stderr: "",
truncated: false,
exitCode: 0,
signal: null,
})),
...overrides,
} as never,
});
}
async function flushPromises(): Promise<void> {
for (let index = 0; index < 8; index += 1) {
await Promise.resolve();
}
}
describe("WorkspaceGitService checkout observation", () => {
beforeEach(() => {
vi.useFakeTimers();
});
afterEach(() => {
vi.useRealTimers();
});
test("shares one recursive checkout observer between cwd-equivalent consumers", async () => {
const watcher = createWatcherHarness();
const runGitCommand = vi.fn(async (args: string[]) => ({
stdout: args[0] === "rev-parse" ? `${REPO_CWD}\n` : "",
stderr: "",
truncated: false,
exitCode: 0,
signal: null,
}));
const service = createService(watcher, { runGitCommand });
const firstListener = vi.fn();
const secondListener = vi.fn();
const first = await service.requestWorkingTreeWatch(REPO_CWD, firstListener);
const second = await service.requestWorkingTreeWatch(path.join(REPO_CWD, "."), secondListener);
expect(first.repoRoot).toBe(REPO_CWD);
expect(second.repoRoot).toBe(REPO_CWD);
expect(watcher.subscribe).toHaveBeenCalledTimes(1);
expect(runGitCommand).toHaveBeenCalledTimes(2);
expect(watcher.records[0]).toMatchObject({ directory: REPO_CWD });
expect(watcher.records[0]?.ignore).toContain(GIT_DIR);
watcher.records[0]?.callback(null, [
{ path: path.join(REPO_CWD, "tracked.txt"), type: "update" },
]);
expect(firstListener).toHaveBeenCalledTimes(1);
expect(secondListener).toHaveBeenCalledTimes(1);
first.unsubscribe();
expect(watcher.records[0]?.unsubscribe).not.toHaveBeenCalled();
second.unsubscribe();
await vi.waitFor(() => {
expect(watcher.records[0]?.unsubscribe).toHaveBeenCalledTimes(1);
});
service.dispose();
});
test("a second workspace client adds no observation or Git work", async () => {
const watcher = createWatcherHarness();
const getCheckoutSnapshotFacts = vi.fn(async (cwd: string) => createCheckoutFacts(cwd));
const getCheckoutStatus = vi.fn(async (cwd: string) => createCheckoutStatus(cwd));
const runGitCommand = vi.fn(async (args: string[]) => ({
stdout: args[0] === "rev-parse" ? `${REPO_CWD}\n` : "",
stderr: "",
truncated: false,
exitCode: 0,
signal: null,
}));
const service = createService(watcher, {
getCheckoutSnapshotFacts,
getCheckoutStatus,
runGitCommand,
});
const first = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn());
const second = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn());
await vi.waitFor(() => {
expect(service.peekSnapshot(REPO_CWD)).not.toBeNull();
expect(service.getMetrics().workspaceObservationSetupInFlightCount).toBe(0);
expect(service.getMetrics().workspaceRefreshInFlightCount).toBe(0);
});
expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(1);
expect(getCheckoutStatus).toHaveBeenCalledTimes(1);
expect(runGitCommand).toHaveBeenCalledTimes(2);
expect(watcher.records.filter((record) => record.directory === REPO_CWD)).toHaveLength(1);
expect(watcher.records.filter((record) => record.directory === GIT_DIR)).toHaveLength(1);
first.unsubscribe();
second.unsubscribe();
service.dispose();
});
test("an observer abandoned during async setup is closed", async () => {
const watcher = createWatcherHarness();
const openedSubscription = createDeferred<{ unsubscribe: () => Promise<void> }>();
const unsubscribeWatcher = vi.fn(async () => {});
watcher.subscribe.mockImplementationOnce(async () => openedSubscription.promise);
const service = createService(watcher);
const subscription = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn());
await vi.waitFor(() => {
expect(watcher.subscribe).toHaveBeenCalledTimes(1);
});
subscription.unsubscribe();
openedSubscription.resolve({ unsubscribe: unsubscribeWatcher });
await vi.waitFor(() => {
expect(unsubscribeWatcher).toHaveBeenCalledTimes(1);
expect(service.getMetrics().workingTreeWatchTargetCount).toBe(0);
});
service.dispose();
});
test("a tracked edit refreshes summary and active uncommitted diff without structural or forge work", async () => {
const watcher = createWatcherHarness();
const getCheckoutSnapshotFacts = vi.fn(async (cwd: string) => createCheckoutFacts(cwd));
let diffStat = { additions: 1, deletions: 0 };
let diffFile = { path: "tracked.txt", additions: 1, deletions: 0, status: "modified" };
const getCheckoutStatus = vi.fn(async (cwd: string) =>
createCheckoutStatus(cwd, { isDirty: true }),
);
const getCheckoutShortstat = vi.fn(async () => diffStat);
const getPullRequestStatus = vi.fn();
const getCheckoutDiff = vi.fn(async () => ({ diff: "", structured: [diffFile] }));
const service = createService(watcher, {
getCheckoutSnapshotFacts,
getCheckoutStatus,
getCheckoutShortstat,
getPullRequestStatus,
getCheckoutDiff,
});
const diffManager = new CheckoutDiffManager({
logger: createLogger(),
paseoHome: "/tmp/paseo-home",
workspaceGitService: service,
});
const summaryListener = vi.fn();
const diffListener = vi.fn();
await service.getSnapshot(REPO_CWD);
const summarySubscription = service.registerWorkspace({ cwd: REPO_CWD }, summaryListener);
const diffSubscription = await diffManager.subscribe(
{ cwd: REPO_CWD, compare: { mode: "uncommitted" } },
diffListener,
);
await vi.waitFor(() => {
expect(watcher.subscribe).toHaveBeenCalledWith(
REPO_CWD,
expect.any(Function),
expect.any(Object),
);
expect(service.getMetrics().workingTreeWatchTargetCount).toBe(1);
expect(service.getMetrics().workspaceObservationSetupInFlightCount).toBe(0);
expect(service.getMetrics().workspaceRefreshInFlightCount).toBe(0);
});
diffStat = { additions: 4, deletions: 2 };
diffFile = { path: "tracked.txt", additions: 4, deletions: 2, status: "modified" };
watcher.records[0]?.callback(null, [
{ path: path.join(REPO_CWD, "tracked.txt"), type: "update" },
]);
await vi.advanceTimersByTimeAsync(1_000);
await vi.waitFor(() => {
expect(summaryListener).toHaveBeenLastCalledWith(
expect.objectContaining({
git: expect.objectContaining({
isDirty: true,
diffStat: { additions: 4, deletions: 2 },
}),
}),
);
expect(diffListener).toHaveBeenLastCalledWith({
cwd: REPO_CWD,
files: [{ path: "tracked.txt", additions: 4, deletions: 2, status: "modified" }],
error: null,
});
});
await flushPromises();
expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(1);
expect(getPullRequestStatus).not.toHaveBeenCalled();
diffSubscription.unsubscribe();
summarySubscription.unsubscribe();
diffManager.dispose();
service.dispose();
});
test("worktree and metadata events invalidate only their matching diff projections", async () => {
const watcher = createWatcherHarness();
const getCheckoutDiff = vi.fn(async () => ({ diff: "", structured: [] }));
const service = createService(watcher, { getCheckoutDiff });
const subscription = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn());
await vi.waitFor(() => {
expect(service.peekSnapshot(REPO_CWD)).not.toBeNull();
expect(service.getMetrics().workspaceObservationSetupInFlightCount).toBe(0);
});
await service.getCheckoutDiff(REPO_CWD, { mode: "uncommitted" });
await service.getCheckoutDiff(REPO_CWD, { mode: "base", baseRef: "main" });
expect(getCheckoutDiff).toHaveBeenCalledTimes(2);
watcher.records
.find((record) => record.directory === REPO_CWD)
?.callback(null, [{ path: path.join(REPO_CWD, "tracked.txt"), type: "update" }]);
await service.getCheckoutDiff(REPO_CWD, { mode: "uncommitted" });
await service.getCheckoutDiff(REPO_CWD, { mode: "base", baseRef: "main" });
expect(getCheckoutDiff).toHaveBeenCalledTimes(3);
expect(getCheckoutDiff.mock.calls[2]?.[1]).toMatchObject({ mode: "uncommitted" });
watcher.records
.find((record) => record.directory === GIT_DIR)
?.callback(null, [{ path: path.join(GIT_DIR, "HEAD"), type: "update" }]);
await service.getCheckoutDiff(REPO_CWD, { mode: "base", baseRef: "main" });
expect(getCheckoutDiff).toHaveBeenCalledTimes(4);
expect(getCheckoutDiff.mock.calls[3]?.[1]).toMatchObject({ mode: "base" });
subscription.unsubscribe();
service.dispose();
});
test("metadata refreshes notify ref-dependent consumers when the summary is unchanged", async () => {
const watcher = createWatcherHarness();
const getCheckoutSnapshotFacts = vi.fn(async (cwd: string) => createCheckoutFacts(cwd));
const service = createService(watcher, { getCheckoutSnapshotFacts });
const listener = vi.fn();
const subscription = service.registerWorkspace({ cwd: REPO_CWD }, listener);
await vi.waitFor(() => {
expect(service.getMetrics().workspaceObservationSetupInFlightCount).toBe(0);
expect(service.getMetrics().workspaceRefreshInFlightCount).toBe(0);
});
listener.mockClear();
watcher.records
.find((record) => record.directory === GIT_DIR)
?.callback(null, [{ path: path.join(GIT_DIR, "packed-refs"), type: "update" }]);
await vi.advanceTimersByTimeAsync(1_000);
await vi.waitFor(() => {
expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(2);
});
expect(listener).toHaveBeenCalledTimes(1);
subscription.unsubscribe();
service.dispose();
});
test("repository metadata observation ignores root and pruned-directory noise", async () => {
const watcher = createWatcherHarness();
const getCheckoutSnapshotFacts = vi.fn(async (cwd: string) => createCheckoutFacts(cwd));
const getCheckoutStatus = vi.fn(async (cwd: string) => createCheckoutStatus(cwd));
const service = createService(watcher, { getCheckoutSnapshotFacts, getCheckoutStatus });
const listener = vi.fn();
const subscription = service.registerWorkspace({ cwd: REPO_CWD }, listener);
await vi.waitFor(() => {
expect(service.getMetrics().workspaceObservationSetupInFlightCount).toBe(0);
expect(service.getMetrics().workspaceRefreshInFlightCount).toBe(0);
});
getCheckoutSnapshotFacts.mockClear();
getCheckoutStatus.mockClear();
listener.mockClear();
watcher.records
.find((record) => record.directory === GIT_DIR)
?.callback(null, [
{ path: GIT_DIR, type: "update" },
{ path: path.join(GIT_DIR, "objects", "pack", "temporary.pack"), type: "update" },
]);
await vi.advanceTimersByTimeAsync(1_000);
await flushPromises();
expect(getCheckoutSnapshotFacts).not.toHaveBeenCalled();
expect(getCheckoutStatus).not.toHaveBeenCalled();
expect(listener).not.toHaveBeenCalled();
subscription.unsubscribe();
service.dispose();
});
test("metadata-only changes refresh worktree summary and active uncommitted diff", async () => {
const watcher = createWatcherHarness();
let isDirty = true;
let diffStat = { additions: 3, deletions: 1 };
let diffFiles = [{ path: "tracked.txt", additions: 3, deletions: 1, status: "modified" }];
const getCheckoutStatus = vi.fn(async (cwd: string) => createCheckoutStatus(cwd, { isDirty }));
const getCheckoutShortstat = vi.fn(async () => diffStat);
const getCheckoutDiff = vi.fn(async () => ({ diff: "", structured: diffFiles }));
const service = createService(watcher, {
getCheckoutStatus,
getCheckoutShortstat,
getCheckoutDiff,
});
const diffManager = new CheckoutDiffManager({
logger: createLogger(),
paseoHome: "/tmp/paseo-home",
workspaceGitService: service,
});
const summaryListener = vi.fn();
const diffListener = vi.fn();
const summarySubscription = service.registerWorkspace({ cwd: REPO_CWD }, summaryListener);
const diffSubscription = await diffManager.subscribe(
{ cwd: REPO_CWD, compare: { mode: "uncommitted" } },
diffListener,
);
await vi.waitFor(() => {
expect(service.getMetrics().workspaceObservationSetupInFlightCount).toBe(0);
expect(service.getMetrics().workspaceRefreshInFlightCount).toBe(0);
});
isDirty = false;
diffStat = { additions: 0, deletions: 0 };
diffFiles = [];
watcher.records
.find((record) => record.directory === GIT_DIR)
?.callback(null, [{ path: path.join(GIT_DIR, "index"), type: "update" }]);
await vi.advanceTimersByTimeAsync(1_000);
await vi.waitFor(() => {
expect(summaryListener).toHaveBeenLastCalledWith(
expect.objectContaining({
git: expect.objectContaining({
isDirty: false,
diffStat: { additions: 0, deletions: 0 },
}),
}),
);
expect(diffListener).toHaveBeenLastCalledWith({
cwd: REPO_CWD,
files: [],
error: null,
});
});
diffSubscription.unsubscribe();
summarySubscription.unsubscribe();
diffManager.dispose();
service.dispose();
});
test("base diff projections follow metadata but ignore volatile worktree updates", async () => {
const watcher = createWatcherHarness();
const getCheckoutDiff = vi.fn(async () => ({ diff: "", structured: [] }));
const getCheckoutWorktreeState = vi.fn(async () => ({
isDirty: true,
diffStat: { additions: 4, deletions: 2 },
}));
const service = createService(watcher, { getCheckoutDiff, getCheckoutWorktreeState });
const diffManager = new CheckoutDiffManager({
logger: createLogger(),
paseoHome: "/tmp/paseo-home",
workspaceGitService: service,
});
const diffSubscription = await diffManager.subscribe(
{ cwd: REPO_CWD, compare: { mode: "base", baseRef: "main" } },
vi.fn(),
);
await vi.waitFor(() => {
expect(service.getMetrics().workspaceObservationSetupInFlightCount).toBe(0);
});
expect(getCheckoutDiff).toHaveBeenCalledTimes(1);
watcher.records
.find((record) => record.directory === REPO_CWD)
?.callback(null, [{ path: path.join(REPO_CWD, "tracked.txt"), type: "update" }]);
await vi.advanceTimersByTimeAsync(1_000);
await flushPromises();
expect(getCheckoutDiff).toHaveBeenCalledTimes(1);
watcher.records
.find((record) => record.directory === GIT_DIR)
?.callback(null, [{ path: path.join(GIT_DIR, "packed-refs"), type: "update" }]);
await vi.advanceTimersByTimeAsync(1_000);
await vi.advanceTimersByTimeAsync(150);
await vi.waitFor(() => {
expect(getCheckoutDiff).toHaveBeenCalledTimes(2);
});
diffSubscription.unsubscribe();
diffManager.dispose();
service.dispose();
});
test("an event burst during an in-flight refresh produces one final follow-up", async () => {
const watcher = createWatcherHarness();
const firstRefresh = createDeferred<CheckoutStatusGit>();
const getCheckoutStatus = vi
.fn<() => Promise<CheckoutStatusGit>>()
.mockResolvedValueOnce(createCheckoutStatus(REPO_CWD))
.mockImplementationOnce(() => firstRefresh.promise)
.mockResolvedValue(createCheckoutStatus(REPO_CWD, { isDirty: true }));
const getCheckoutShortstat = vi
.fn()
.mockResolvedValueOnce({ additions: 0, deletions: 0 })
.mockResolvedValueOnce({ additions: 1, deletions: 0 })
.mockResolvedValue({ additions: 100, deletions: 25 });
const service = createService(watcher, { getCheckoutStatus, getCheckoutShortstat });
const listener = vi.fn();
await service.getSnapshot(REPO_CWD);
const subscription = service.registerWorkspace({ cwd: REPO_CWD }, listener);
await vi.waitFor(() => {
expect(service.getMetrics().workspaceObservationSetupInFlightCount).toBe(0);
});
const edit = { path: path.join(REPO_CWD, "tracked.txt"), type: "update" as const };
watcher.records[0]?.callback(null, [edit]);
await vi.advanceTimersByTimeAsync(1_000);
await vi.waitFor(() => {
expect(getCheckoutStatus).toHaveBeenCalledTimes(2);
});
for (let event = 0; event < 100; event += 1) {
watcher.records[0]?.callback(null, [edit]);
}
await vi.advanceTimersByTimeAsync(1_000);
expect(service.getMetrics().workspaceRefreshQueuedCount).toBe(1);
firstRefresh.resolve(createCheckoutStatus(REPO_CWD, { isDirty: true }));
await vi.waitFor(() => {
expect(getCheckoutStatus).toHaveBeenCalledTimes(3);
expect(service.getMetrics().workspaceRefreshInFlightCount).toBe(0);
});
expect(getCheckoutStatus).toHaveBeenCalledTimes(3);
expect(listener).toHaveBeenLastCalledWith(
expect.objectContaining({
git: expect.objectContaining({
isDirty: true,
diffStat: { additions: 100, deletions: 25 },
}),
}),
);
expect(service.getMetrics().workspaceRefreshQueuedCount).toBe(0);
subscription.unsubscribe();
service.dispose();
});
test("an event queued during a failed refresh still reaches final state", async () => {
const watcher = createWatcherHarness();
const failedRefresh = createDeferred<CheckoutStatusGit>();
const getCheckoutStatus = vi
.fn<() => Promise<CheckoutStatusGit>>()
.mockResolvedValueOnce(createCheckoutStatus(REPO_CWD))
.mockImplementationOnce(() => failedRefresh.promise)
.mockResolvedValue(createCheckoutStatus(REPO_CWD, { isDirty: true }));
const getCheckoutShortstat = vi
.fn()
.mockResolvedValueOnce({ additions: 0, deletions: 0 })
.mockResolvedValue({ additions: 42, deletions: 7 });
const service = createService(watcher, { getCheckoutStatus, getCheckoutShortstat });
const listener = vi.fn();
await service.getSnapshot(REPO_CWD);
const subscription = service.registerWorkspace({ cwd: REPO_CWD }, listener);
await vi.waitFor(() => {
expect(service.getMetrics().workspaceObservationSetupInFlightCount).toBe(0);
});
const edit = { path: path.join(REPO_CWD, "tracked.txt"), type: "update" as const };
watcher.records[0]?.callback(null, [edit]);
await vi.advanceTimersByTimeAsync(1_000);
await vi.waitFor(() => {
expect(getCheckoutStatus).toHaveBeenCalledTimes(2);
});
watcher.records[0]?.callback(null, [edit]);
await vi.advanceTimersByTimeAsync(1_000);
expect(service.getMetrics().workspaceRefreshQueuedCount).toBe(1);
failedRefresh.reject(new Error("transient Git read failure"));
await vi.waitFor(() => {
expect(getCheckoutStatus).toHaveBeenCalledTimes(3);
expect(service.getMetrics().workspaceRefreshInFlightCount).toBe(0);
});
expect(listener).toHaveBeenLastCalledWith(
expect.objectContaining({
git: expect.objectContaining({
isDirty: true,
diffStat: { additions: 42, deletions: 7 },
}),
}),
);
subscription.unsubscribe();
service.dispose();
});
test("ten worktrees share one repository metadata and fetch observer while retaining checkout state", async () => {
const watcher = createWatcherHarness();
const fetch = createDeferred<void>();
const runGitFetch = vi.fn(() => fetch.promise);
const commonGitDir = path.resolve("/tmp/paseo-shared-repository.git");
const worktrees = Array.from({ length: 10 }, (_, index) =>
path.resolve(`/tmp/paseo-shared-worktree-${index}`),
);
const getCheckoutSnapshotFacts = vi.fn(
async (cwd: string): Promise<CheckoutSnapshotFacts> => ({
...createCheckoutFacts(cwd),
remoteUrl: "https://github.com/acme/shared.git",
absoluteGitDir: path.join(commonGitDir, "worktrees", path.basename(cwd)),
gitCommonDir: commonGitDir,
}),
);
const getCheckoutStatus = vi.fn(async (cwd: string) => createCheckoutStatus(cwd));
const runGitCommand = vi.fn(async (_args: string[], options: { cwd: string }) => ({
stdout: `${options.cwd}\n`,
stderr: "",
truncated: false,
exitCode: 0,
signal: null,
}));
const service = createService(watcher, {
getCheckoutSnapshotFacts,
getCheckoutStatus,
runGitCommand,
runGitFetch,
});
const subscriptions = worktrees.map((cwd) => service.registerWorkspace({ cwd }, vi.fn()));
await vi.waitFor(() => {
expect(service.getMetrics().workspaceObservationSetupInFlightCount).toBe(0);
expect(service.getMetrics().workingTreeWatchSetupInFlightCount).toBe(0);
expect(runGitFetch).toHaveBeenCalledTimes(1);
});
expect(service.getMetrics()).toMatchObject({
workspaceTargetCount: 10,
repositoryTargetCount: 1,
repositoryWorkspaceLinkCount: 10,
workingTreeWatchTargetCount: 10,
});
expect(watcher.records.filter((record) => record.directory === commonGitDir)).toHaveLength(1);
expect(watcher.records.filter((record) => worktrees.includes(record.directory))).toHaveLength(
10,
);
subscriptions[0]?.unsubscribe();
fetch.resolve();
await vi.waitFor(() => {
expect(service.getMetrics().fetchInFlightCount).toBe(0);
});
await vi.advanceTimersByTimeAsync(180_000);
await vi.waitFor(() => {
expect(runGitFetch).toHaveBeenCalledTimes(2);
});
expect(runGitFetch).toHaveBeenLastCalledWith(worktrees[1]);
for (const subscription of subscriptions.slice(1)) {
subscription.unsubscribe();
}
service.dispose();
});
test("watcher setup failure uses scoped non-overlapping polling", async () => {
const watcher = createWatcherHarness({ failDirectories: new Set([REPO_CWD]) });
const blockedPoll = createDeferred<CheckoutStatusGit>();
const getCheckoutSnapshotFacts = vi.fn(async (cwd: string) => createCheckoutFacts(cwd));
const getCheckoutStatus = vi
.fn<() => Promise<CheckoutStatusGit>>()
.mockResolvedValueOnce(createCheckoutStatus(REPO_CWD))
.mockImplementationOnce(() => blockedPoll.promise)
.mockResolvedValue(createCheckoutStatus(REPO_CWD, { isDirty: true }));
const getPullRequestStatus = vi.fn();
const service = createService(watcher, {
getCheckoutSnapshotFacts,
getCheckoutStatus,
getPullRequestStatus,
});
const subscription = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn());
await vi.waitFor(() => {
expect(watcher.subscribe).toHaveBeenCalledWith(
REPO_CWD,
expect.any(Function),
expect.any(Object),
);
expect(service.getMetrics().workingTreeWatchTargetCount).toBe(1);
expect(service.getMetrics().workspaceObservationSetupInFlightCount).toBe(0);
expect(service.getMetrics().workspaceRefreshInFlightCount).toBe(0);
expect(service.peekSnapshot(REPO_CWD)).not.toBeNull();
expect(getCheckoutStatus).toHaveBeenCalledTimes(1);
});
await vi.advanceTimersByTimeAsync(7_000);
await vi.waitFor(() => {
expect(getCheckoutStatus).toHaveBeenCalledTimes(2);
});
await vi.advanceTimersByTimeAsync(15_000);
expect(getCheckoutStatus).toHaveBeenCalledTimes(2);
expect(service.getMetrics().workspaceRefreshQueuedCount).toBe(0);
expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(1);
expect(getPullRequestStatus).not.toHaveBeenCalled();
blockedPoll.resolve(createCheckoutStatus(REPO_CWD, { isDirty: true }));
await vi.waitFor(() => {
expect(service.getMetrics().workspaceRefreshInFlightCount).toBe(0);
});
await vi.advanceTimersByTimeAsync(5_000);
await vi.waitFor(() => {
expect(getCheckoutStatus).toHaveBeenCalledTimes(3);
});
subscription.unsubscribe();
service.dispose();
});
test("non-Git fallback promotes an externally initialized checkout", async () => {
const watcher = createWatcherHarness();
let isGit = false;
const getCheckoutSnapshotFacts = vi.fn(
async (cwd: string): Promise<CheckoutSnapshotFacts> =>
isGit ? createCheckoutFacts(cwd) : { isGit: false },
);
const getCheckoutStatus = vi.fn(async (cwd: string) =>
isGit ? createCheckoutStatus(cwd) : ({ isGit: false } as const),
);
const runGitCommand = vi.fn(async (args: string[]) => {
if (args[0] === "rev-parse") {
if (!isGit) {
throw new Error("not a git repository");
}
return {
stdout: `${REPO_CWD}\n`,
stderr: "",
truncated: false,
exitCode: 0,
signal: null,
};
}
if (args[0] === "ls-files") {
return {
stdout: "",
stderr: "",
truncated: false,
exitCode: 0,
signal: null,
};
}
throw new Error(`Unexpected Git command: ${args.join(" ")}`);
});
const service = createService(watcher, {
getCheckoutSnapshotFacts,
getCheckoutStatus,
getWorkspaceGitSelfHealPhaseMs: () => 60_000,
runGitCommand,
});
const summarySubscription = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn());
await vi.waitFor(() => {
expect(service.peekSnapshot(REPO_CWD)?.git.isGit).toBe(false);
expect(service.getMetrics()).toMatchObject({
workspaceObservationSetupInFlightCount: 0,
workspaceRefreshInFlightCount: 0,
});
});
expect(watcher.records.filter((record) => record.directory === GIT_DIR)).toHaveLength(0);
isGit = true;
await vi.advanceTimersByTimeAsync(5_000);
await vi.waitFor(() => {
expect(service.peekSnapshot(REPO_CWD)?.git.isGit).toBe(true);
expect(service.getMetrics()).toMatchObject({
repositoryTargetCount: 1,
repositoryWorkspaceLinkCount: 1,
workspaceObservationSetupInFlightCount: 0,
workspaceRefreshInFlightCount: 0,
});
});
expect(watcher.records.filter((record) => record.directory === GIT_DIR)).toHaveLength(1);
expect(watcher.records.filter((record) => record.directory === REPO_CWD)).toHaveLength(1);
const factsCallCount = getCheckoutSnapshotFacts.mock.calls.length;
await vi.advanceTimersByTimeAsync(5_000);
expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(factsCallCount);
summarySubscription.unsubscribe();
service.dispose();
});
test("watcher runtime error switches to scoped polling", async () => {
const watcher = createWatcherHarness();
const getCheckoutSnapshotFacts = vi.fn(async (cwd: string) => createCheckoutFacts(cwd));
const getCheckoutStatus = vi.fn(async (cwd: string) => createCheckoutStatus(cwd));
const service = createService(watcher, {
getCheckoutSnapshotFacts,
getCheckoutStatus,
});
const subscription = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn());
await vi.waitFor(() => {
expect(service.peekSnapshot(REPO_CWD)).not.toBeNull();
expect(service.getMetrics().workspaceObservationSetupInFlightCount).toBe(0);
});
const checkoutWatcher = watcher.records.find((record) => record.directory === REPO_CWD);
expect(checkoutWatcher).toBeDefined();
checkoutWatcher?.callback(new Error("watcher stopped"), []);
await vi.waitFor(() => {
expect(checkoutWatcher?.unsubscribe).toHaveBeenCalledTimes(1);
});
await vi.advanceTimersByTimeAsync(5_000);
await vi.waitFor(() => {
expect(getCheckoutStatus).toHaveBeenCalledTimes(2);
});
expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(1);
expect(service.getMetrics().workspaceRefreshQueuedCount).toBe(0);
subscription.unsubscribe();
service.dispose();
});
test("a missed watcher event converges through the deterministic audit", async () => {
const watcher = createWatcherHarness();
let isDirty = false;
let diffStat = { additions: 0, deletions: 0 };
let diffFiles: Array<{
path: string;
additions: number;
deletions: number;
status: string;
}> = [];
const getCheckoutStatus = vi.fn(async (cwd: string) => createCheckoutStatus(cwd, { isDirty }));
const getCheckoutShortstat = vi.fn(async () => diffStat);
const getCheckoutDiff = vi.fn(async () => ({ diff: "", structured: diffFiles }));
const service = createService(watcher, {
getCheckoutStatus,
getCheckoutShortstat,
getCheckoutDiff,
getWorkspaceGitSelfHealPhaseMs: () => 10_000,
});
const diffManager = new CheckoutDiffManager({
logger: createLogger(),
paseoHome: "/tmp/paseo-home",
workspaceGitService: service,
});
const listener = vi.fn();
const diffListener = vi.fn();
const subscription = service.registerWorkspace({ cwd: REPO_CWD }, listener);
const diffSubscription = await diffManager.subscribe(
{ cwd: REPO_CWD, compare: { mode: "uncommitted" } },
diffListener,
);
await vi.waitFor(() => {
expect(service.peekSnapshot(REPO_CWD)).not.toBeNull();
expect(service.getMetrics().workspaceObservationSetupInFlightCount).toBe(0);
});
isDirty = true;
diffStat = { additions: 7, deletions: 3 };
diffFiles = [{ path: "tracked.txt", additions: 7, deletions: 3, status: "modified" }];
await vi.advanceTimersByTimeAsync(70_000);
await vi.waitFor(() => {
expect(listener).toHaveBeenLastCalledWith(
expect.objectContaining({
git: expect.objectContaining({
isDirty: true,
diffStat: { additions: 7, deletions: 3 },
}),
}),
);
expect(diffListener).toHaveBeenLastCalledWith({
cwd: REPO_CWD,
files: [{ path: "tracked.txt", additions: 7, deletions: 3, status: "modified" }],
error: null,
});
});
expect(getCheckoutStatus).toHaveBeenCalledTimes(2);
diffSubscription.unsubscribe();
subscription.unsubscribe();
diffManager.dispose();
service.dispose();
});
});
@@ -1,6 +1,5 @@
import { execFileSync } from "node:child_process";
import { mkdirSync, mkdtempSync, realpathSync, rmSync, writeFileSync } from "node:fs";
import { readdir } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join, resolve as resolvePath } from "node:path";
import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
@@ -20,10 +19,10 @@ import {
} from "../utils/checkout-git.js";
import { runGitCommand as runGitCommandReal } from "../utils/run-git-command.js";
import {
getWorkspaceGitSelfHealPhaseMs,
WorkspaceGitServiceImpl,
type WorkspaceGitRuntimeSnapshot,
} from "./workspace-git-service.js";
import { isPlatform } from "../test-utils/platform.js";
const REPO_CWD = resolvePath("/tmp/repo");
@@ -36,13 +35,6 @@ function createLogger() {
return logger;
}
function createWatcher() {
return {
close: vi.fn(),
on: vi.fn().mockReturnThis(),
};
}
function createDeferred<T>() {
let resolve!: (value: T | PromiseLike<T>) => void;
let reject!: (reason?: unknown) => void;
@@ -53,6 +45,10 @@ function createDeferred<T>() {
return { promise, resolve, reject };
}
function createAsyncSubscription() {
return { unsubscribe: vi.fn(() => Promise.resolve()) };
}
async function flushPromises(): Promise<void> {
for (let i = 0; i < 5; i += 1) {
await Promise.resolve();
@@ -313,9 +309,11 @@ function createGitHubServiceStub(): ForgeService {
}
interface CreateServiceOptions {
subscribe?: ReturnType<typeof vi.fn>;
getCheckoutSnapshotFacts?: ReturnType<typeof vi.fn>;
getCheckoutStatus?: ReturnType<typeof vi.fn>;
getCheckoutShortstat?: ReturnType<typeof vi.fn>;
getCheckoutWorktreeState?: ReturnType<typeof vi.fn>;
getPullRequestStatus?: ReturnType<typeof vi.fn>;
getCheckoutDiff?: ReturnType<typeof vi.fn>;
resolveBranchCheckout?: ReturnType<typeof vi.fn>;
@@ -327,21 +325,20 @@ interface CreateServiceOptions {
hasOriginRemote?: ReturnType<typeof vi.fn>;
runGitFetch?: ReturnType<typeof vi.fn>;
runGitCommand?: ReturnType<typeof vi.fn>;
watch?: ReturnType<typeof vi.fn>;
readdir?: ReturnType<typeof vi.fn>;
now?: () => Date;
getWorkspaceGitSelfHealPhaseMs?: (cwd: string) => number;
}
function buildDefaultServiceDeps() {
return {
watch: (() => createWatcher()) as never,
readdir: vi.fn(async () => []),
subscribe: vi.fn(async () => ({ unsubscribe: vi.fn(async () => {}) })),
getCheckoutSnapshotFacts: vi.fn(async (cwd: string) => createCheckoutFacts(cwd)),
getCheckoutStatus: vi.fn(async (cwd: string) => createCheckoutStatus(cwd)),
getCheckoutShortstat: vi.fn(async () => ({
additions: 1,
deletions: 0,
})),
getCheckoutWorktreeState: vi.fn(),
getPullRequestStatus: vi.fn(async () => createPullRequestStatusResult()),
getCheckoutDiff: vi.fn(async () => ({ diff: "", structured: [] })),
resolveBranchCheckout: vi.fn(async () => ({ kind: "not-found" })),
@@ -351,7 +348,7 @@ function buildDefaultServiceDeps() {
forgeOverrides: { github: createGitHubServiceStub() },
resolveAbsoluteGitDir: vi.fn(async () => join(REPO_CWD, ".git")),
hasOriginRemote: vi.fn(async () => false),
runGitFetch: vi.fn(async () => {}),
runGitFetch: vi.fn(() => createDeferred<void>().promise),
runGitCommand: vi.fn(async () => ({
stdout: `${REPO_CWD}\n`,
stderr: "",
@@ -359,6 +356,7 @@ function buildDefaultServiceDeps() {
exitCode: 0,
signal: null,
})),
getWorkspaceGitSelfHealPhaseMs: vi.fn(() => 30_000),
now: () => new Date("2026-04-12T00:00:00.000Z"),
};
}
@@ -366,11 +364,24 @@ function buildDefaultServiceDeps() {
function buildServiceDeps(options?: CreateServiceOptions) {
const { github, ...rest } = options ?? {};
const defaults = buildDefaultServiceDeps();
return {
const deps = {
...defaults,
...rest,
forgeOverrides: github ? { github } : defaults.forgeOverrides,
};
deps.getCheckoutWorktreeState =
options?.getCheckoutWorktreeState ??
vi.fn(async (cwd: string) => {
const status = await deps.getCheckoutStatus(cwd);
if (!status.isGit) {
throw new Error("Expected a git checkout");
}
return {
isDirty: status.isDirty,
diffStat: await deps.getCheckoutShortstat(),
};
});
return deps;
}
function createService(options?: CreateServiceOptions) {
@@ -494,8 +505,10 @@ describe("WorkspaceGitServiceImpl primitive refresh entrypoint", () => {
expect(listener).not.toHaveBeenCalled();
expect(service.peekSnapshot(REPO_CWD)).toBeNull();
await vi.waitFor(() => {
expect(getCheckoutStatus).toHaveBeenCalledTimes(1);
});
await flushPromises();
expect(getCheckoutStatus).toHaveBeenCalledTimes(1);
expect(service.peekSnapshot(REPO_CWD)).toBeNull();
@@ -572,8 +585,10 @@ describe("WorkspaceGitServiceImpl primitive refresh entrypoint", () => {
const first = service.getSnapshot(REPO_CWD);
const second = service.getSnapshot(join(REPO_CWD, "."));
await vi.waitFor(() => {
expect(getCheckoutStatus).toHaveBeenCalledTimes(1);
});
await flushPromises();
expect(getCheckoutStatus).toHaveBeenCalledTimes(1);
checkoutStatusDeferred.resolve(createCheckoutStatus(REPO_CWD));
@@ -848,11 +863,14 @@ describe("WorkspaceGitServiceImpl primitive refresh entrypoint", () => {
getPullRequestStatus,
now: () => new Date(nowMs),
});
const subscription = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn());
await flushPromises();
const initialSnapshotReady = createDeferred<void>();
const subscription = service.registerWorkspace({ cwd: REPO_CWD }, () => {
initialSnapshotReady.resolve();
});
await initialSnapshotReady.promise;
nowMs = 60_000;
await vi.advanceTimersByTimeAsync(60_000);
nowMs = 120_000;
await vi.advanceTimersByTimeAsync(120_000);
await flushPromises();
expect(getCheckoutStatus).toHaveBeenCalledTimes(2);
@@ -868,50 +886,108 @@ describe("WorkspaceGitServiceImpl primitive refresh entrypoint", () => {
service.dispose();
});
test("self-heal retries workspace observation setup while a listener remains active", async () => {
test("self-heal phase is stable across process restarts", () => {
expect(getWorkspaceGitSelfHealPhaseMs("/tmp/repo")).toBe(54_185);
expect(getWorkspaceGitSelfHealPhaseMs("/tmp/staggered-repo")).toBe(9_817);
});
test("self-heal staggers workspaces and settles into a 120 second cadence", async () => {
vi.setSystemTime(0);
const otherRepoCwd = resolvePath("/tmp/staggered-repo");
const refreshes = new Map<string, number[]>();
const getCheckoutStatus = vi.fn(async (cwd: string) => {
const timestamps = refreshes.get(cwd) ?? [];
timestamps.push(Date.now());
refreshes.set(cwd, timestamps);
return createCheckoutStatus(cwd, { hasRemote: false, remoteUrl: null });
});
const service = createService({
getCheckoutSnapshotFacts: vi.fn(async (cwd: string) =>
createCheckoutFacts(cwd, { remoteUrl: null }),
),
getCheckoutStatus,
getWorkspaceGitSelfHealPhaseMs: (cwd) => (cwd === REPO_CWD ? 10_000 : 40_000),
now: () => new Date(Date.now()),
});
const first = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn());
const second = service.registerWorkspace({ cwd: otherRepoCwd }, vi.fn());
await Promise.all([service.getSnapshot(REPO_CWD), service.getSnapshot(otherRepoCwd)]);
await vi.advanceTimersByTimeAsync(300_000);
await flushPromises();
const firstSelfHeals = refreshes.get(REPO_CWD)?.slice(1) ?? [];
const secondSelfHeals = refreshes.get(otherRepoCwd)?.slice(1) ?? [];
expect(firstSelfHeals).toHaveLength(2);
expect(secondSelfHeals).toHaveLength(2);
expect(firstSelfHeals[0]).not.toBe(secondSelfHeals[0]);
expect(firstSelfHeals[1] - firstSelfHeals[0]).toBe(120_000);
expect(secondSelfHeals[1] - secondSelfHeals[0]).toBe(120_000);
first.unsubscribe();
second.unsubscribe();
service.dispose();
});
test("self-heal retries observation setup even when the git snapshot is still recent", async () => {
let nowMs = 0;
const getCheckoutSnapshotFacts = vi
.fn<(cwd: string) => Promise<CheckoutSnapshotFacts>>()
.mockRejectedValueOnce(new Error("git facts temporarily unavailable"))
.mockImplementation(async (cwd: string) => createCheckoutFacts(cwd));
const watch = vi.fn(() => createWatcher() as never);
.mockImplementation(async (cwd: string) => createCheckoutFacts(cwd, { remoteUrl: null }));
const getCheckoutStatus = vi.fn(async (cwd: string) =>
createCheckoutStatus(cwd, { remoteUrl: null }),
);
const subscribe = vi.fn(async () => createAsyncSubscription());
const service = createService({
getCheckoutSnapshotFacts,
watch,
getCheckoutStatus,
subscribe,
now: () => new Date(nowMs),
});
const subscription = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn());
await flushPromises();
await vi.waitFor(() => {
expect(getCheckoutSnapshotFacts).toHaveBeenCalled();
expect(service.getMetrics().workspaceObservationSetupInFlightCount).toBe(0);
expect(service.getMetrics().workspaceRefreshInFlightCount).toBe(0);
});
expect(getCheckoutSnapshotFacts).toHaveBeenCalled();
expect(watch).not.toHaveBeenCalled();
const factsCallsBeforeSelfHeal = getCheckoutSnapshotFacts.mock.calls.length;
expect(subscribe).not.toHaveBeenCalled();
nowMs = 30_000;
await service.getSnapshot(REPO_CWD, { force: true, reason: "recovered-git-read" });
expect(getCheckoutStatus).toHaveBeenCalledTimes(1);
expect(subscribe).not.toHaveBeenCalled();
nowMs = 60_000;
await vi.advanceTimersByTimeAsync(60_000);
await flushPromises();
expect(getCheckoutSnapshotFacts.mock.calls.length).toBeGreaterThan(factsCallsBeforeSelfHeal);
expect(getCheckoutSnapshotFacts).toHaveBeenLastCalledWith(REPO_CWD, expect.anything());
expect(subscribe).toHaveBeenCalledTimes(2);
expect(getCheckoutStatus).toHaveBeenCalledTimes(1);
subscription.unsubscribe();
service.dispose();
});
test("stale workspace watcher callbacks do not refresh after unsubscribe", async () => {
const watchCallbacks: Array<() => void> = [];
const watch = vi.fn(
(_watchPath: string, _options: { recursive: boolean }, callback: () => void) => {
watchCallbacks.push(callback);
return createWatcher() as never;
const watchCallbacks: Array<{
path: string;
callback: (error: Error | null, events: Array<{ path: string; type: "update" }>) => void;
}> = [];
const subscribe = vi.fn(
async (watchPath: string, callback: (typeof watchCallbacks)[number]["callback"]) => {
watchCallbacks.push({ path: watchPath, callback });
return createAsyncSubscription();
},
);
const getCheckoutStatus = vi.fn(async (cwd: string) => createCheckoutStatus(cwd));
const service = createService({
getCheckoutStatus,
resolveAbsoluteGitDir: vi.fn(async () => join(REPO_CWD, ".git")),
watch,
subscribe,
});
const subscription = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn());
@@ -920,10 +996,12 @@ describe("WorkspaceGitServiceImpl primitive refresh entrypoint", () => {
await vi.waitFor(() => {
expect(watchCallbacks.length).toBeGreaterThan(0);
});
const workingTreeCallback = watchCallbacks.find((entry) => entry.path === REPO_CWD)?.callback;
expect(workingTreeCallback).toBeTypeOf("function");
const callsBeforeStaleCallback = getCheckoutStatus.mock.calls.length;
subscription.unsubscribe();
watchCallbacks[0]?.();
workingTreeCallback?.(null, [{ path: join(REPO_CWD, "file.ts"), type: "update" }]);
await vi.advanceTimersByTimeAsync(1_000);
await flushPromises();
@@ -1390,17 +1468,23 @@ describe("WorkspaceGitServiceImpl primitive refresh entrypoint", () => {
test("multiple subscribers on the same target share one self-heal timer", async () => {
let nowMs = 0;
const getCheckoutStatus = vi.fn(async (cwd: string) => createCheckoutStatus(cwd));
const getCheckoutSnapshotFacts = vi.fn(async (cwd: string) =>
createCheckoutFacts(cwd, { remoteUrl: null }),
);
const getCheckoutStatus = vi.fn(async (cwd: string) =>
createCheckoutStatus(cwd, { remoteUrl: null }),
);
const service = createService({
getCheckoutSnapshotFacts,
getCheckoutStatus,
now: () => new Date(nowMs),
});
const first = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn());
const second = service.registerWorkspace({ cwd: join(REPO_CWD, ".") }, vi.fn());
await flushPromises();
await service.getSnapshot(REPO_CWD);
nowMs = 60_000;
await vi.advanceTimersByTimeAsync(60_000);
nowMs = 120_000;
await vi.advanceTimersByTimeAsync(120_000);
await flushPromises();
expect(getCheckoutStatus).toHaveBeenCalledTimes(2);
@@ -1420,11 +1504,12 @@ describe("WorkspaceGitServiceImpl primitive refresh entrypoint", () => {
const subscription = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn());
subscription.unsubscribe();
nowMs = 60_000;
await vi.advanceTimersByTimeAsync(60_000);
nowMs = 600_000;
await vi.advanceTimersByTimeAsync(600_000);
await flushPromises();
expect(getCheckoutStatus).toHaveBeenCalledTimes(0);
expect(vi.getTimerCount()).toBe(0);
service.dispose();
});
@@ -1439,8 +1524,8 @@ describe("WorkspaceGitServiceImpl primitive refresh entrypoint", () => {
service.registerWorkspace({ cwd: REPO_CWD }, vi.fn());
service.dispose();
nowMs = 60_000;
await vi.advanceTimersByTimeAsync(60_000);
nowMs = 600_000;
await vi.advanceTimersByTimeAsync(600_000);
await flushPromises();
expect(getCheckoutStatus).toHaveBeenCalledTimes(0);
@@ -1453,15 +1538,19 @@ describe("WorkspaceGitServiceImpl primitive refresh entrypoint", () => {
.fn<() => Promise<CheckoutStatusGit>>()
.mockImplementationOnce(async () => createCheckoutStatus(REPO_CWD))
.mockImplementationOnce(async () => selfHealRefresh.promise);
const getCheckoutSnapshotFacts = vi.fn(async (cwd: string) =>
createCheckoutFacts(cwd, { remoteUrl: null }),
);
const service = createService({
getCheckoutSnapshotFacts,
getCheckoutStatus,
now: () => new Date(nowMs),
});
const subscription = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn());
await flushPromises();
await service.getSnapshot(REPO_CWD);
nowMs = 60_000;
await vi.advanceTimersByTimeAsync(60_000);
nowMs = 120_000;
await vi.advanceTimersByTimeAsync(120_000);
await flushPromises();
const directRead = service.getSnapshot(REPO_CWD);
await flushPromises();
@@ -1812,8 +1901,10 @@ describe("WorkspaceGitServiceImpl D2 read methods", () => {
const first = service.resolveRepoRoot(REPO_CWD);
const second = service.resolveRepoRoot(join(REPO_CWD, "."));
await vi.waitFor(() => {
expect(getCheckoutStatus).toHaveBeenCalledTimes(1);
});
await flushPromises();
expect(getCheckoutStatus).toHaveBeenCalledTimes(1);
checkoutDeferred.resolve(createCheckoutStatus(REPO_CWD));
await expect(Promise.all([first, second])).resolves.toEqual([REPO_CWD, REPO_CWD]);
@@ -2024,13 +2115,7 @@ describe("WorkspaceGitServiceImpl D2 read methods", () => {
service.dispose();
});
// POSIX-only: this asserts Linux working-tree walker behavior around ignored directories.
test.skipIf(isPlatform("win32"))(
"Linux working tree walker excludes gitignored directories",
async () => {
const originalPlatform = process.platform;
Object.defineProperty(process, "platform", { configurable: true, value: "linux" });
test("working tree observation prunes gitignored directories", async () => {
const tempDir = realpathSync(mkdtempSync(join(tmpdir(), "workspace-git-service-ignored-")));
const repoDir = join(tempDir, "repo");
mkdirSync(join(repoDir, "ignored", "deep"), { recursive: true });
@@ -2041,15 +2126,10 @@ describe("WorkspaceGitServiceImpl D2 read methods", () => {
writeFileSync(join(repoDir, "ignored", "deep", "log.txt"), "noise\n");
writeFileSync(join(repoDir, "kept", "file.txt"), "keep\n");
const watchedPaths: string[] = [];
const watchSpy = (watchPath: string) => {
watchedPaths.push(watchPath);
return { close: vi.fn(), on: vi.fn().mockReturnThis() };
};
const subscribe = vi.fn(async () => createAsyncSubscription());
const service = createService({
watch: watchSpy as never,
readdir: readdir as never,
subscribe,
runGitCommand: runGitCommandReal as never,
getCheckoutSnapshotFacts: getCheckoutSnapshotFactsUncached as never,
getCheckoutStatus: getCheckoutStatusUncached as never,
@@ -2060,18 +2140,17 @@ describe("WorkspaceGitServiceImpl D2 read methods", () => {
const subscription = await service.requestWorkingTreeWatch(repoDir, vi.fn());
const ignoredRoot = join(repoDir, "ignored");
expect(watchedPaths.filter((path) => path.startsWith(ignoredRoot))).toEqual([]);
expect(watchedPaths).toContain(repoDir);
expect(watchedPaths).toContain(join(repoDir, "kept"));
expect(subscribe).toHaveBeenCalledTimes(1);
expect(subscribe).toHaveBeenCalledWith(repoDir, expect.any(Function), {
ignore: [join(repoDir, ".git"), ignoredRoot],
});
subscription.unsubscribe();
} finally {
service.dispose();
rmSync(tempDir, { recursive: true, force: true });
Object.defineProperty(process, "platform", { configurable: true, value: originalPlatform });
}
},
);
});
test("onWorkspaceStateMayHaveChanged invalidates github cache and schedules a forced github-inclusive refresh", async () => {
const github = createGitHubServiceStub();
@@ -1,7 +1,6 @@
import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
import os from "node:os";
import path, { join } from "node:path";
import type { FSWatcher } from "node:fs";
import type pino from "pino";
import type { ForgeService } from "../services/forge-service.js";
import type {
@@ -13,7 +12,6 @@ import {
WorkspaceGitServiceImpl,
type WorkspaceGitRuntimeSnapshot,
} from "./workspace-git-service.js";
import { isPlatform } from "../test-utils/platform.js";
const REPO_CWD = path.resolve("/tmp/repo");
@@ -182,21 +180,6 @@ function createPullRequestStatusResult(
};
}
function createWatcher(): FSWatcher & { close: ReturnType<typeof vi.fn> } {
const watcher = {
close: vi.fn(),
on: vi.fn().mockReturnThis(),
};
return watcher as unknown as FSWatcher & { close: ReturnType<typeof vi.fn> };
}
function createDirent(name: string, isDirectory: boolean) {
return {
name,
isDirectory: () => isDirectory,
};
}
async function flushPromises(): Promise<void> {
for (let i = 0; i < 6; i++) {
await Promise.resolve();
@@ -213,6 +196,10 @@ function createDeferred<T>() {
return { promise, resolve, reject };
}
function createAsyncSubscription() {
return { unsubscribe: vi.fn(() => Promise.resolve()) };
}
function createGitHubServiceStub(): ForgeService {
return {
listPullRequests: vi.fn(async () => []),
@@ -255,30 +242,31 @@ function createGitHubServiceStub(): ForgeService {
}
interface CreateServiceTestOptions {
subscribe?: ReturnType<typeof vi.fn>;
getCheckoutStatus?: ReturnType<typeof vi.fn>;
getCheckoutSnapshotFacts?: ReturnType<typeof vi.fn>;
getCheckoutShortstat?: ReturnType<typeof vi.fn>;
getCheckoutWorktreeState?: ReturnType<typeof vi.fn>;
getPullRequestStatus?: ReturnType<typeof vi.fn>;
github?: ForgeService;
resolveAbsoluteGitDir?: ReturnType<typeof vi.fn>;
hasOriginRemote?: ReturnType<typeof vi.fn>;
runGitFetch?: ReturnType<typeof vi.fn>;
runGitCommand?: ReturnType<typeof vi.fn>;
readdir?: ReturnType<typeof vi.fn>;
watch?: ReturnType<typeof vi.fn>;
getWorkspaceGitSelfHealPhaseMs?: (cwd: string) => number;
now?: () => Date;
}
function buildDefaultTestServiceDeps() {
return {
watch: (() => createWatcher()) as unknown as typeof import("node:fs").watch,
readdir: vi.fn(async () => []),
subscribe: vi.fn(async () => ({ unsubscribe: vi.fn(async () => {}) })),
getCheckoutSnapshotFacts: vi.fn(async (cwd: string) => createCheckoutSnapshotFacts(cwd)),
getCheckoutStatus: vi.fn(async (cwd: string) => createCheckoutStatus(cwd)),
getCheckoutShortstat: vi.fn(async () => ({
additions: 1,
deletions: 0,
})),
getCheckoutWorktreeState: vi.fn(),
getPullRequestStatus: vi.fn(async () => createPullRequestStatusResult()),
forgeOverrides: { github: createGitHubServiceStub() },
resolveAbsoluteGitDir: vi.fn(async () => join(REPO_CWD, ".git")),
@@ -291,15 +279,29 @@ function buildDefaultTestServiceDeps() {
exitCode: 0,
signal: null,
})),
getWorkspaceGitSelfHealPhaseMs: vi.fn(() => 30_000),
now: () => new Date("2026-04-12T00:00:00.000Z"),
};
}
function createService(options?: CreateServiceTestOptions) {
const deps = { ...buildDefaultTestServiceDeps(), ...options };
deps.getCheckoutWorktreeState =
options?.getCheckoutWorktreeState ??
vi.fn(async (cwd: string) => {
const status = await deps.getCheckoutStatus(cwd);
if (!status.isGit) {
throw new Error("Expected a git checkout");
}
return {
isDirty: status.isDirty,
diffStat: await deps.getCheckoutShortstat(),
};
});
return new WorkspaceGitServiceImpl({
logger: createLogger() as unknown as pino.Logger,
paseoHome: "/tmp/paseo-test",
deps: { ...buildDefaultTestServiceDeps(), ...options },
deps,
});
}
@@ -480,9 +482,9 @@ describe("WorkspaceGitServiceImpl", () => {
const firstSnapshotPromise = service.getSnapshot(REPO_CWD);
const secondSnapshotPromise = service.getSnapshot(join(REPO_CWD, "."));
await flushPromises();
await vi.waitFor(() => {
expect(getCheckoutStatus).toHaveBeenCalledTimes(1);
});
expect(getPullRequestStatus).toHaveBeenCalledTimes(0);
expect(resolveAbsoluteGitDir).toHaveBeenCalledTimes(0);
@@ -563,7 +565,8 @@ describe("WorkspaceGitServiceImpl", () => {
});
test("repo-level fetch intervals are shared for workspaces in the same repo", async () => {
const runGitFetch = vi.fn(async () => {});
const firstFetch = createDeferred<void>();
const runGitFetch = vi.fn(() => firstFetch.promise);
const hasOriginRemote = vi.fn(async () => true);
const getCheckoutSnapshotFacts = vi.fn(async (cwd: string) => ({
...createCheckoutSnapshotFacts(cwd),
@@ -587,12 +590,20 @@ describe("WorkspaceGitServiceImpl", () => {
await vi.waitFor(() => {
expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(2);
expect(runGitFetch).toHaveBeenCalledTimes(1);
expect(service.getMetrics().workspaceRefreshInFlightCount).toBe(0);
});
expect(resolveAbsoluteGitDir).toHaveBeenCalledTimes(0);
expect(hasOriginRemote).toHaveBeenCalledTimes(0);
expect(runGitFetch).toHaveBeenCalledTimes(1);
firstFetch.resolve();
await flushPromises();
await vi.advanceTimersByTimeAsync(1_000);
await vi.waitFor(() => {
expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(4);
});
await vi.advanceTimersByTimeAsync(180_000);
await flushPromises();
@@ -771,69 +782,31 @@ describe("WorkspaceGitServiceImpl", () => {
service.dispose();
});
// POSIX-only: this asserts Linux recursive-watch fallback behavior.
test.skipIf(isPlatform("win32"))("watches nested repository directories on Linux", async () => {
const originalPlatform = process.platform;
Object.defineProperty(process, "platform", {
configurable: true,
value: "linux",
});
test("watches nested repository changes through one recursive checkout subscription", async () => {
const unsubscribe = vi.fn(async () => {});
const subscribe = vi.fn(async () => ({ unsubscribe }));
const service = createService({ subscribe });
const watchCalls: Array<{ path: string; close: ReturnType<typeof vi.fn> }> = [];
const watch = vi.fn((watchPath: string) => {
const watcher = createWatcher();
watchCalls.push({ path: watchPath, close: watcher.close });
return watcher;
});
const readdir = vi.fn(async (directory: string) => {
if (directory === REPO_CWD) {
return [
createDirent("packages", true),
createDirent(".git", true),
createDirent("README.md", false),
];
}
if (directory === path.join(REPO_CWD, "packages")) {
return [createDirent("server", true), createDirent("app", true)];
}
if (directory === path.join(REPO_CWD, "packages", "server")) {
return [createDirent("src", true)];
}
if (directory === path.join(REPO_CWD, "packages", "server", "src")) {
return [createDirent("server", true)];
}
return [];
});
const service = createService({ watch, readdir });
const subscription = await service.requestWorkingTreeWatch(
path.join(REPO_CWD, "packages", "server"),
vi.fn(),
);
expect(subscription.repoRoot).toBe(REPO_CWD);
expect(watchCalls.map((entry) => entry.path).sort()).toEqual([
REPO_CWD,
join(REPO_CWD, ".git"),
join(REPO_CWD, "packages"),
join(REPO_CWD, "packages", "app"),
join(REPO_CWD, "packages", "server"),
join(REPO_CWD, "packages", "server", "src"),
join(REPO_CWD, "packages", "server", "src", "server"),
]);
expect(subscribe).toHaveBeenCalledTimes(1);
expect(subscribe).toHaveBeenCalledWith(REPO_CWD, expect.any(Function), {
ignore: [join(REPO_CWD, ".git")],
});
subscription.unsubscribe();
expect(unsubscribe).toHaveBeenCalledTimes(1);
service.dispose();
Object.defineProperty(process, "platform", {
configurable: true,
value: originalPlatform,
});
});
test("requestWorkingTreeWatch reference-counts watchers by cwd", async () => {
const watchers = [createWatcher(), createWatcher()];
const watch = vi.fn().mockReturnValueOnce(watchers[0]).mockReturnValueOnce(watchers[1]);
const service = createService({ watch });
const unsubscribe = vi.fn(async () => {});
const subscribe = vi.fn(async () => ({ unsubscribe }));
const service = createService({ subscribe });
const firstListener = vi.fn();
const secondListener = vi.fn();
@@ -842,39 +815,22 @@ describe("WorkspaceGitServiceImpl", () => {
expect(first.repoRoot).toBe(REPO_CWD);
expect(second.repoRoot).toBe(REPO_CWD);
expect(watch).toHaveBeenCalledTimes(2);
expect(subscribe).toHaveBeenCalledTimes(1);
first.unsubscribe();
expect(watchers[0].close).not.toHaveBeenCalled();
expect(watchers[1].close).not.toHaveBeenCalled();
expect(unsubscribe).not.toHaveBeenCalled();
second.unsubscribe();
expect(watchers[0].close).toHaveBeenCalledTimes(1);
expect(watchers[1].close).toHaveBeenCalledTimes(1);
expect(unsubscribe).toHaveBeenCalledTimes(1);
service.dispose();
});
test("sets a 5-second fallback polling interval when recursive watch is unavailable", async () => {
if (process.platform === "linux") {
// On Linux, recursive watch is never attempted — the service uses per-directory
// watchers from the start. This scenario only applies to macOS/Windows where
// recursive watch is tried first and may fail.
return;
}
const recursiveUnsupported = new Error("recursive unsupported");
const watch = vi
.fn()
.mockImplementationOnce((_watchPath: string, options: { recursive: boolean }) => {
if (options.recursive) {
throw recursiveUnsupported;
}
return createWatcher();
})
.mockImplementationOnce(() => createWatcher());
const service = createService({ watch });
test("sets a bounded polling fallback when recursive observation is unavailable", async () => {
const subscribe = vi.fn(async () => {
throw new Error("watcher unavailable");
});
const service = createService({ subscribe });
const subscription = await service.requestWorkingTreeWatch(REPO_CWD, vi.fn());
expect(vi.getTimerCount()).toBe(1);
@@ -884,13 +840,13 @@ describe("WorkspaceGitServiceImpl", () => {
});
test("non-git directories fall back to watching cwd with polling", async () => {
const watch = vi.fn(() => createWatcher());
const subscribe = vi.fn(async () => createAsyncSubscription());
const runGitCommand = vi.fn(async () => {
throw new Error("not a git repository");
});
const resolveAbsoluteGitDir = vi.fn(async () => null);
const service = createService({
watch,
subscribe,
runGitCommand,
resolveAbsoluteGitDir,
});
@@ -899,12 +855,9 @@ describe("WorkspaceGitServiceImpl", () => {
const subscription = await service.requestWorkingTreeWatch(plainCwd, vi.fn());
expect(subscription.repoRoot).toBeNull();
const expectedRecursive = process.platform !== "linux";
expect(watch).toHaveBeenCalledWith(
plainCwd,
{ recursive: expectedRecursive },
expect.any(Function),
);
expect(subscribe).toHaveBeenCalledWith(plainCwd, expect.any(Function), {
ignore: [join(plainCwd, ".git")],
});
expect(vi.getTimerCount()).toBe(1);
subscription.unsubscribe();
@@ -912,20 +865,20 @@ describe("WorkspaceGitServiceImpl", () => {
});
test("working tree changes notify watch listeners immediately", async () => {
const watchCallbacks: Array<() => void> = [];
const watch = vi.fn(
(_watchPath: string, _options: { recursive: boolean }, callback: () => void) => {
const watchCallbacks: Array<
(error: Error | null, events: Array<{ path: string; type: "update" }>) => void
> = [];
const subscribe = vi.fn(async (_path: string, callback: (typeof watchCallbacks)[number]) => {
watchCallbacks.push(callback);
return createWatcher();
},
);
const service = createService({ watch });
return createAsyncSubscription();
});
const service = createService({ subscribe });
const listener = vi.fn();
const subscription = await service.requestWorkingTreeWatch(REPO_CWD, listener);
expect(watchCallbacks).toHaveLength(2);
expect(watchCallbacks).toHaveLength(1);
watchCallbacks[0]?.();
watchCallbacks[0]?.(null, [{ path: join(REPO_CWD, "file.ts"), type: "update" }]);
expect(listener).toHaveBeenCalledTimes(1);
@@ -934,40 +887,50 @@ describe("WorkspaceGitServiceImpl", () => {
});
test("working tree changes force a fresh diff stat for workspace subscribers", async () => {
const watchCallbacks: Array<{ path: string; callback: () => void }> = [];
const watch = vi.fn(
(watchPath: string, _options: { recursive: boolean }, callback: () => void) => {
const watchCallbacks: Array<{
path: string;
callback: (error: Error | null, events: Array<{ path: string; type: "update" }>) => void;
}> = [];
const subscribe = vi.fn(
async (watchPath: string, callback: (typeof watchCallbacks)[number]["callback"]) => {
watchCallbacks.push({ path: watchPath, callback });
return createWatcher();
return createAsyncSubscription();
},
);
const getCheckoutShortstat = vi
.fn()
.mockResolvedValueOnce({ additions: 1, deletions: 0 })
.mockResolvedValueOnce({ additions: 8, deletions: 3 });
const service = createService({ getCheckoutShortstat, watch });
const getCheckoutWorktreeState = vi.fn(async () => ({
isDirty: true,
diffStat: { additions: 8, deletions: 3 },
}));
const backgroundFetch = createDeferred<void>();
const service = createService({
getCheckoutWorktreeState,
runGitFetch: vi.fn(() => backgroundFetch.promise),
subscribe,
});
const workspaceListener = vi.fn();
const initialSnapshot = await service.getSnapshot(REPO_CWD);
const workspaceSubscription = service.registerWorkspace({ cwd: REPO_CWD }, workspaceListener);
const diffSubscription = await service.requestWorkingTreeWatch(REPO_CWD, vi.fn());
await vi.waitFor(() => {
expect(service.getMetrics().repositoryWorkspaceLinkCount).toBe(1);
});
expect(initialSnapshot.git.diffStat).toEqual({ additions: 1, deletions: 0 });
const repoRootWatch = watchCallbacks.find((entry) => entry.path === REPO_CWD);
expect(repoRootWatch).toBeDefined();
repoRootWatch?.callback();
repoRootWatch?.callback(null, [{ path: join(REPO_CWD, "file.ts"), type: "update" }]);
await vi.advanceTimersByTimeAsync(1_000);
await flushPromises();
expect(getCheckoutShortstat).toHaveBeenLastCalledWith(
expect(getCheckoutWorktreeState).toHaveBeenCalledWith(
REPO_CWD,
expect.objectContaining({ paseoHome: "/tmp/paseo-test" }),
{ force: true },
);
expect(workspaceListener).toHaveBeenCalledWith(
createSnapshot(REPO_CWD, {
git: { diffStat: { additions: 8, deletions: 3 } },
git: { isDirty: true, diffStat: { additions: 8, deletions: 3 } },
}),
);
File diff suppressed because it is too large Load Diff
+16
View File
@@ -2664,6 +2664,22 @@ export async function getCheckoutShortstat(
return getOrLoadCheckoutShortstat(cwd, context, options);
}
export interface CheckoutWorktreeState {
isDirty: boolean;
diffStat: CheckoutShortstat | null;
}
export async function getCheckoutWorktreeState(
cwd: string,
context: CheckoutContext,
): Promise<CheckoutWorktreeState> {
const [isDirty, diffStat] = await Promise.all([
isWorkingTreeDirty(cwd, context),
getCheckoutShortstat(cwd, context, { force: true }),
]);
return { isDirty, diffStat };
}
export function getCachedCheckoutShortstat(cwd: string): CheckoutShortstat | null | undefined {
return shortstatCache.get(getShortstatCacheKey(cwd));
}
@@ -0,0 +1,119 @@
import { appendFile } from "node:fs/promises";
interface GitCommandTraceContext {
id: string;
args: string[];
cwd: string;
submittedAtMs: number;
startedAtMs: number | null;
spawnedAtMs: number | null;
pid: number | null;
submittedQueue: GitCommandTraceQueueState;
startedQueue: GitCommandTraceQueueState | null;
settled: boolean;
}
interface GitCommandTraceQueueState {
active: number;
pending: number;
}
interface GitCommandTraceSettlement {
outcome: "closed" | "spawn_error" | "timed_out";
exitCode?: number | null;
signal?: NodeJS.Signals | null;
truncated?: boolean;
}
let nextTraceId = 1;
let pendingTraceWrite = Promise.resolve();
function appendTrace(event: Record<string, unknown>): void {
const tracePath = process.env.PASEO_GIT_TRACE_FILE?.trim();
if (!tracePath) return;
pendingTraceWrite = pendingTraceWrite
.then(() => appendFile(tracePath, `${JSON.stringify(event)}\n`, "utf8"))
.catch(() => {
// Diagnostic tracing must never change command execution.
});
}
export async function flushGitCommandTrace(): Promise<void> {
await pendingTraceWrite;
}
export function submitGitCommandTrace(
args: string[],
cwd: string,
queue: GitCommandTraceQueueState,
): GitCommandTraceContext | null {
if (!process.env.PASEO_GIT_TRACE_FILE?.trim()) return null;
const submittedAtMs = Date.now();
const context: GitCommandTraceContext = {
id: `${process.pid}:${nextTraceId++}`,
args: [...args],
cwd,
submittedAtMs,
startedAtMs: null,
spawnedAtMs: null,
pid: null,
submittedQueue: { ...queue },
startedQueue: null,
settled: false,
};
return context;
}
export function startGitCommandTrace(
context: GitCommandTraceContext | null,
queue: GitCommandTraceQueueState,
): void {
if (!context) return;
context.startedAtMs = Date.now();
context.startedQueue = { ...queue };
}
export function spawnGitCommandTrace(
context: GitCommandTraceContext | null,
pid: number | undefined,
): void {
if (!context) return;
context.spawnedAtMs = Date.now();
context.pid = pid ?? null;
}
export function settleGitCommandTrace(
context: GitCommandTraceContext | null,
settlement: GitCommandTraceSettlement,
): void {
if (!context || context.settled) return;
context.settled = true;
const atMs = Date.now();
appendTrace({
kind: "git_command",
event: "settled",
id: context.id,
atMs,
submittedAtMs: context.submittedAtMs,
startedAtMs: context.startedAtMs,
args: context.args,
cwd: context.cwd,
queueWaitMs: context.startedAtMs === null ? null : context.startedAtMs - context.submittedAtMs,
spawnCallMs:
context.startedAtMs === null || context.spawnedAtMs === null
? null
: context.spawnedAtMs - context.startedAtMs,
pid: context.pid,
queue: {
submitted: context.submittedQueue,
started: context.startedQueue,
},
durationMs: atMs - context.submittedAtMs,
...settlement,
});
}
@@ -0,0 +1,53 @@
import { mkdtemp, readFile, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import path from "node:path";
import { afterEach, describe, expect, it, vi } from "vitest";
import { flushGitCommandTrace } from "./git-command-trace.js";
import { runGitCommand } from "./run-git-command.js";
describe("git command trace", () => {
const tempDirectories: string[] = [];
afterEach(async () => {
vi.unstubAllEnvs();
await Promise.all(
tempDirectories.splice(0).map((directory) => rm(directory, { recursive: true, force: true })),
);
});
it("writes one asynchronous settlement record for a real git process", async () => {
const directory = await mkdtemp(path.join(tmpdir(), "paseo-git-trace-"));
tempDirectories.push(directory);
const tracePath = path.join(directory, "git.jsonl");
vi.stubEnv("PASEO_GIT_TRACE_FILE", tracePath);
await runGitCommand(["--version"], { cwd: directory });
await flushGitCommandTrace();
const events = (await readFile(tracePath, "utf8"))
.trim()
.split("\n")
.map((line) => JSON.parse(line) as Record<string, unknown>);
expect(events).toHaveLength(1);
expect(events[0]).toMatchObject({
kind: "git_command",
event: "settled",
args: ["--version"],
cwd: directory,
queue: {
submitted: { active: 0, pending: 0 },
started: expect.objectContaining({
active: expect.any(Number),
pending: expect.any(Number),
}),
},
queueWaitMs: expect.any(Number),
spawnCallMs: expect.any(Number),
pid: expect.any(Number),
outcome: "closed",
exitCode: 0,
signal: null,
durationMs: expect.any(Number),
});
});
});
@@ -6,6 +6,12 @@ import {
GitCommandRuntimeMetricsWindow,
type GitCommandRuntimeMetricsSnapshot,
} from "./git-command-runtime-metrics.js";
import {
settleGitCommandTrace,
spawnGitCommandTrace,
startGitCommandTrace,
submitGitCommandTrace,
} from "./git-command-trace.js";
import { spawnProcess } from "./spawn.js";
const DEFAULT_TIMEOUT_MS = 30_000;
@@ -135,10 +141,18 @@ export function runGitCommand(
args: string[],
options: GitCommandOptions,
): Promise<GitCommandResult> {
const commandTrace = submitGitCommandTrace(args, options.cwd, {
active: gitLimit.activeCount,
pending: gitLimit.pendingCount,
});
const runtimeMetric = gitRuntimeMetrics.submit(getGitOperation(args));
const promise = gitLimit(
() =>
new Promise<GitCommandResult>((resolve, reject) => {
startGitCommandTrace(commandTrace, {
active: gitLimit.activeCount,
pending: gitLimit.pendingCount,
});
gitRuntimeMetrics.start(runtimeMetric);
const timeout = options.timeout ?? DEFAULT_TIMEOUT_MS;
const maxOutputBytes = options.maxOutputBytes ?? DEFAULT_MAX_OUTPUT_BYTES;
@@ -173,6 +187,7 @@ export function runGitCommand(
shell: false,
stdio: ["ignore", "pipe", "pipe"],
});
spawnGitCommandTrace(commandTrace, child.pid);
let settled = false;
let metricFinished = false;
@@ -199,6 +214,11 @@ export function runGitCommand(
const timer = setTimeout(() => {
const error = new Error(`Git command timed out after ${timeout}ms: ${command}`);
child.kill("SIGKILL");
settleGitCommandTrace(commandTrace, {
outcome: "timed_out",
exitCode: null,
signal: "SIGKILL",
});
finishMetricOnce(
{
args,
@@ -255,6 +275,11 @@ export function runGitCommand(
});
child.on("error", (error) => {
settleGitCommandTrace(commandTrace, {
outcome: "spawn_error",
exitCode: null,
signal: null,
});
finishMetricOnce({
args,
cwd: options.cwd,
@@ -278,6 +303,12 @@ export function runGitCommand(
});
child.on("close", (exitCode, signal) => {
settleGitCommandTrace(commandTrace, {
outcome: "closed",
exitCode,
signal,
truncated,
});
const result: GitCommandResult = {
stdout: Buffer.concat(stdoutChunks).toString("utf8"),
stderr: Buffer.concat(stderrChunks).toString("utf8"),
+23
View File
@@ -14,11 +14,30 @@
import { nodeFileTrace } from "@vercel/nft";
import { glob } from "node:fs/promises";
import { createRequire } from "node:module";
import path from "node:path";
import { pathToFileURL } from "node:url";
const REPO_ROOT = path.resolve(import.meta.dirname, "..");
function toTracePath(absolutePath) {
return path.relative(REPO_ROOT, absolutePath).split(path.sep).join("/");
}
function parcelWatcherNativePackagePath() {
const requireFromRepo = createRequire(path.join(REPO_ROOT, "package.json"));
requireFromRepo("@parcel/watcher");
const nativeAddonPath = Object.keys(requireFromRepo.cache)
.map(toTracePath)
.find((file) => /^node_modules\/@parcel\/watcher-[^/]+\/watcher\.node$/.test(file));
if (!nativeAddonPath) {
throw new Error("@parcel/watcher loaded without a platform native addon");
}
return path.posix.dirname(nativeAddonPath);
}
const { sherpaPlatformPackageName } = await import(
pathToFileURL(
path.join(
@@ -65,6 +84,10 @@ const additionalInputs = [
// the Nix derivation builds for one platform at a time and ships only
// its own binaries.
`node_modules/node-pty/prebuilds/${process.platform}-${process.arch}/**`,
// @parcel/watcher dynamically resolves its optional platform package. npm's
// build closure can contain multiple libc variants, so copy the package the
// wrapper actually loaded rather than every installed optional package.
`${parcelWatcherNativePackagePath()}/**`,
// sherpa-onnx-node dynamically resolves a platform-specific native package.
// Copy the wrapper plus the host platform package explicitly.
"node_modules/sherpa-onnx-node/**",