Compare commits

..
Author SHA1 Message Date
Mohamed Boudra 7d093967ef fix(server): pick the git job when its start token fires
Background jobs that already reserved a start-rate token no longer start
ahead of normal work submitted after them.
2026-11-10 02:58:52 +01:00
Mohamed Boudra 1d4b436bf6 fix(server): keep cold diffs out from behind background git fan-out
Add a background lane to the git process scheduler and run reconciliation
checkout reads, observation-driven workspace refreshes and background fetches
in it. Bound reconciliation checkout reads to four at a time. Stop the Linux
file observer's startup scan from reporting existing files as creates, which
made every cold diff compute and discard its first revision.
2026-11-10 02:49:54 +01:00
f9779a14a5 feat(claude): add Fable 5.2 and Mythos 5.2 to the model catalog (#9233)
Adds Claude Fable 5.2 and Mythos 5.2, available from Claude Code 2.1.321, and defaults both to medium effort to match Claude Code.

Co-authored-by: Mohamed Boudra <boudra.moha@gmail.com>
Co-authored-by: gudcks <84506708+gudcks@users.noreply.github.com>
2026-11-10 01:09:47 +00:00
18 changed files with 653 additions and 174 deletions
+8
View File
@@ -340,6 +340,14 @@ the number of Git processes that have started but not exited. Every Git command
including initial workspace reads, filesystem-triggered refreshes, background checks, and explicit
requests.
Commands wait in one of three lanes, drained in order: `high` (worktree creation and archive),
`normal` (everything a caller waits for, including diffs and explicit refreshes), and `background`
(workspace reconciliation, refreshes triggered by watchers, polls and fetches, and the fetches
themselves). Daemon-initiated fan-out must run in the background lane: one rescan over ~90 worktrees
queues over 1,000 commands, and in the normal lane every command a cold diff runs waited behind
them. Set the lane on the runner with `createRunGitCommand(provenance, { priority })`, or around a
whole operation with `runWithGitCommandPriority`.
Environment variables override `config.json`:
| Environment variable | Setting |
+1 -1
View File
@@ -8,7 +8,7 @@ Closing an observation must remain safe after its root is renamed or removed. Wo
`unsubscribe()` is an awaited barrier. Once it resolves, no scan, reconciliation, queued event, native handle, or callback may remain. `updateIgnore()` is the equivalent barrier for excluded paths: events queued below a newly excluded root cannot be delivered after it resolves. Treat path disappearance while scanning, attaching, or receiving a watcher error as rename/archive churn. Surface permission and resource errors so WorkspaceGit can enter bounded polling.
Linux topology repair is trailing-debounced and scoped to the renamed path. A file rename needs one classification stat; a new or moved directory scans only its subtree. Full-tree scans are reserved for startup and ignore-set replacement. The initial Linux scan conservatively emits existing files discovered before their directory watcher was installed; WorkspaceGit batches these into one refresh. Benchmark this startup cost with pre-populated trees.
Linux topology repair is trailing-debounced and scoped to the renamed path. A file rename needs one classification stat; a new or moved directory scans only its subtree. Full-tree scans are reserved for startup and ignore-set replacement. A subscription reports changes after `subscribe()` resolves, never the files that already existed, on every platform. Consumers read their initial state after `subscribe()` resolves. The Linux startup scan therefore emits nothing; only a scoped reconciliation of new topology reports files found before their directory's watcher was installed. Startup creates made every cold diff compute and discard a first revision. Benchmark startup cost with pre-populated trees.
Windows and macOS use one native recursive watcher plus an ignored-pruned, directory-indexed inventory. Native events deliver immediately. Named events trigger serialized shallow scans of their parent directories; known directory events also reconcile that subtree. This recovers concrete create and delete paths omitted by coalescing without walking the whole root for each edit. Classification is admission-controlled to 32 concurrent `fs.stat` calls, since `fs.stat` runs on the libuv threadpool the whole daemon shares. Startup and exclusion changes use an awaited full inventory barrier. Activity arms a full safety audit 30 seconds after it settles, with a 30-second minimum interval and a five-minute starvation bound; pathless native events also trigger an immediate shallow root scan. Observation fails into polling rather than retaining more than 250,000 file and directory entries per root. A repository that large is also expensive to poll, so degraded polling doubles its gap from 5 seconds to a 60-second ceiling while the workspace snapshot comes back unchanged, and resets to 5 seconds on the first change. Explicit reads still force a refresh; the ceiling only bounds how long an edit made outside Paseo stays undiscovered.
+10 -6
View File
@@ -62,10 +62,15 @@ const EXPECTED_CLAUDE_MODELS = [
model: "Opus 5",
descriptionFragment: "Previous release",
},
{
id: "claude-fable-5-2",
model: "Fable 5.2",
descriptionFragment: "Most powerful",
},
{
id: "claude-fable-5-1",
model: "Fable 5.1",
descriptionFragment: "Most powerful",
descriptionFragment: "Previous release",
},
{
id: "claude-fable-5",
@@ -199,11 +204,10 @@ function assertClaudeModels(data: ProviderModel[]): void {
);
}
const fable51Index = data.findIndex((model) => model.id === "claude-fable-5-1");
const fable5Index = data.findIndex((model) => model.id === "claude-fable-5");
assert.strictEqual(
fable5Index,
fable51Index + 1,
const fable52Index = data.findIndex((model) => model.id === "claude-fable-5-2");
assert.deepStrictEqual(
data.slice(fable52Index, fable52Index + 3).map((model) => model.id),
["claude-fable-5-2", "claude-fable-5-1", "claude-fable-5"],
"Fable models should stay adjacent and newest-first",
);
assert(
@@ -1614,6 +1614,8 @@ describe("ClaudeAgentSession features", () => {
test.each([
["supported model", "claude-opus-4-8", { type: "disabled" }, undefined],
["Mythos 5.1", "claude-mythos-5-1", { type: "adaptive", display: "summarized" }, "high"],
["Fable 5.2", "claude-fable-5-2", { type: "adaptive", display: "summarized" }, "medium"],
["Mythos 5.2", "claude-mythos-5-2", { type: "adaptive", display: "summarized" }, "medium"],
["Haiku 5.5", "claude-haiku-5-5", { type: "disabled" }, undefined],
["unsupported model", "claude-fable-5", { type: "adaptive", display: "summarized" }, "high"],
["custom model", "openrouter/anthropic/claude-opus-4-8", undefined, undefined],
@@ -5,7 +5,7 @@ export const CLAUDE_FAST_MODE_CONTROL: AgentOptionToggleControl = {
type: "toggle",
key: "fast_mode",
label: "Fast",
description: "Lower latency Opus responses at higher token cost",
description: "Lower latency responses at higher token cost",
tooltip: "Toggle fast mode",
icon: "zap",
};
@@ -58,6 +58,17 @@ export const CLAUDE_MODEL_MANIFEST = [
supportsThinkingDisabled: true,
supportsFastMode: true,
},
{
id: "claude-mythos-5-2",
label: "Mythos 5.2",
requiresDiscovery: true,
description: "Mythos 5.2 · Requires verified access",
minimumClaudeCodeVersion: "2.1.321",
defaultThinkingOptionId: "medium",
contextWindowMaxTokens: 1_000_000,
effortLevels: CLAUDE_EFFORT_LEVELS.xhigh,
supportsFastMode: true,
},
{
id: "claude-mythos-5-1",
label: "Mythos 5.1",
@@ -66,10 +77,20 @@ export const CLAUDE_MODEL_MANIFEST = [
contextWindowMaxTokens: 1_000_000,
effortLevels: CLAUDE_EFFORT_LEVELS.xhigh,
},
{
id: "claude-fable-5-2",
label: "Fable 5.2",
description: "Fable 5.2 · Most powerful model",
minimumClaudeCodeVersion: "2.1.321",
defaultThinkingOptionId: "medium",
contextWindowMaxTokens: 1_000_000,
effortLevels: CLAUDE_EFFORT_LEVELS.xhigh,
supportsFastMode: true,
},
{
id: "claude-fable-5-1",
label: "Fable 5.1",
description: "Fable 5.1 · Most powerful model",
description: "Fable 5.1 · Previous release",
contextWindowMaxTokens: 1_000_000,
effortLevels: CLAUDE_EFFORT_LEVELS.xhigh,
},
@@ -61,17 +61,23 @@ async function createClaudeConfigDirWithRawSettings(settings: string): Promise<s
return configDir;
}
function createCatalogClient(claudeCodeVersion = "2.1.293"): ClaudeAgentClient {
function createCatalogClient(claudeCodeVersion = "2.1.321"): ClaudeAgentClient {
return new ClaudeAgentClient({
logger: createTestLogger(),
resolveVersion: async () => claudeCodeVersion,
discoverModels: async () => [
{
value: "mythos",
resolvedModel: "claude-mythos-5-1",
resolvedModel: "claude-mythos-5-2",
displayName: "Mythos",
description: "",
},
{
value: "claude-mythos-5-1",
resolvedModel: "claude-mythos-5-1",
displayName: "Mythos 5.1",
description: "",
},
],
});
}
@@ -82,7 +88,9 @@ describe("getClaudeModels", () => {
expect(models.map((m) => m.id)).toEqual([
"claude-opus-5-5",
"claude-opus-5",
"claude-mythos-5-2",
"claude-mythos-5-1",
"claude-fable-5-2",
"claude-fable-5-1",
"claude-fable-5",
"claude-fable-5[1m]",
@@ -118,7 +126,9 @@ describe("getClaudeModels", () => {
new Map([
["claude-opus-5-5", 1_000_000],
["claude-opus-5", 1_000_000],
["claude-mythos-5-2", 1_000_000],
["claude-mythos-5-1", 1_000_000],
["claude-fable-5-2", 1_000_000],
["claude-fable-5-1", 1_000_000],
["claude-fable-5", 1_000_000],
["claude-fable-5[1m]", 1_000_000],
@@ -163,6 +173,12 @@ describe("getClaudeModels", () => {
expect(getClaudeModels("2.1.293").find((model) => model.isDefault)?.id).toBe("claude-opus-5-5");
});
it("gates Fable 5.2 on Claude Code 2.1.321 without changing the default", () => {
expect(getClaudeModels("2.1.320").map((model) => model.id)).not.toContain("claude-fable-5-2");
expect(getClaudeModels("2.1.321").map((model) => model.id)).toContain("claude-fable-5-2");
expect(getClaudeModels("2.1.321").find((model) => model.isDefault)?.id).toBe("claude-opus-5-5");
});
it("derives thinking options from model effort capabilities", () => {
const models = new Map(getClaudeModels().map((model) => [model.id, model]));
@@ -232,6 +248,8 @@ describe("getClaudeModels", () => {
["claude-sonnet-5-20260101", true, "high"],
["claude-fable-5", false, "high"],
["claude-fable-5-1", false, "high"],
["claude-fable-5-2", false, "medium"],
["claude-mythos-5-2", false, "medium"],
["claude-haiku-5-5", true, "medium"],
["claude-haiku-5-5-20261007", true, "medium"],
["claude-haiku-4-5", false, undefined],
@@ -284,6 +302,31 @@ describe("ClaudeAgentClient.fetchCatalog", () => {
},
);
it("offers Mythos 5.2 in place of Mythos 5.1 when Claude Code resolves mythos to 5.2", async () => {
const configDir = await createClaudeConfigDir({});
const client = new ClaudeAgentClient({
logger: createTestLogger(),
resolveVersion: async () => "2.1.321",
runtimeSettings: { env: { CLAUDE_CONFIG_DIR: configDir } },
discoverModels: async () => [
{
value: "mythos",
resolvedModel: "claude-mythos-5-2",
displayName: "Mythos",
description: "",
},
],
});
const { models } = await client.fetchCatalog({
scope: "workspace",
cwd: configDir,
force: false,
});
expect(models.filter((model) => model.id.startsWith("claude-mythos-"))).toEqual([
findClaudeModel("claude-mythos-5-2"),
]);
});
it("preserves the curated catalog and explicitly configured models after discovery failure", async () => {
const configDir = await createClaudeConfigDir({ model: "claude-mythos-5-1" });
const client = new ClaudeAgentClient({
@@ -299,8 +342,8 @@ describe("ClaudeAgentClient.fetchCatalog", () => {
cwd: configDir,
force: false,
});
expect(models.filter((model) => model.id !== "claude-mythos-5-1")).toEqual(
getClaudeModels("2.1.293").filter((model) => model.id !== "claude-mythos-5-1"),
expect(models.filter((model) => !model.id.startsWith("claude-mythos-"))).toEqual(
getClaudeModels("2.1.293").filter((model) => !model.id.startsWith("claude-mythos-")),
);
expect(models.find((model) => model.id === "claude-mythos-5-1")).toEqual({
...findClaudeModel("claude-mythos-5-1"),
@@ -360,7 +403,7 @@ describe("ClaudeAgentClient.fetchCatalog", () => {
});
expect(models.map((model) => model.id)).toEqual(
getClaudeModels("2.1.293")
.filter((model) => model.id !== "claude-mythos-5-1")
.filter((model) => !model.id.startsWith("claude-mythos-"))
.map((model) => model.id),
);
});
@@ -785,6 +828,44 @@ describe("Claude Fable 5.1 catalog", () => {
});
});
describe("Claude Fable 5.2 catalog", () => {
it("offers one 1M model with medium default effort, always-on thinking, and fast mode", () => {
const models = getClaudeModels().filter((model) => model.id.startsWith("claude-fable-5-2"));
expect(models).toHaveLength(1);
expect(models[0]).toMatchObject({
id: "claude-fable-5-2",
label: "Fable 5.2",
contextWindowMaxTokens: 1_000_000,
defaultThinkingOptionId: "medium",
});
expect(models[0].thinkingOptions?.map((option) => option.id)).toEqual([
"low",
"medium",
"high",
"xhigh",
"max",
CLAUDE_ULTRACODE_THINKING_OPTION_ID,
]);
expect(claudeManifestModelSupportsFastMode("claude-fable-5-2")).toBe(true);
});
it.each([
"claude-fable-5-2[1m]",
"claude-fable-5-2-20261110",
"claude-fable-5-2-20261110[1m]",
"Fable 5.2",
])("resolves %s to the single catalog entry", (id) => {
expect(findClaudeModel(id)?.id).toBe("claude-fable-5-2");
});
it.each(["claude-fable-5-2", "anthropic/claude-fable-5-2", "us.anthropic.claude-fable-5-2-v1:0"])(
"keeps runtime model %s from collapsing onto Fable 5",
(id) => {
expect(normalizeClaudeRuntimeModelId(id)).toBe("claude-fable-5-2");
},
);
});
describe("Claude Opus 5.5 catalog", () => {
it("offers one Opus 5.5 entry with a 1M context window", () => {
const opus55Models = getClaudeModels()
@@ -920,7 +1001,7 @@ describe("claudeManifestModelSupportsFastMode", () => {
describe("Claude Mythos 5.1 catalog", () => {
it("offers one 1M model with high default effort and always-on thinking", () => {
const models = getClaudeModels().filter((model) => model.id.startsWith("claude-mythos"));
const models = getClaudeModels().filter((model) => model.id.startsWith("claude-mythos-5-1"));
expect(models).toHaveLength(1);
expect(models[0]).toMatchObject({
id: "claude-mythos-5-1",
@@ -965,6 +1046,61 @@ describe("Claude Mythos 5.1 catalog", () => {
);
});
describe("Claude Mythos 5.2 catalog", () => {
it("stays hidden on Claude Code older than 2.1.321 even when advertised", async () => {
const configDir = await createClaudeConfigDir({});
const client = new ClaudeAgentClient({
logger: createTestLogger(),
resolveVersion: async () => "2.1.320",
runtimeSettings: { env: { CLAUDE_CONFIG_DIR: configDir } },
discoverModels: async () => [
{
value: "mythos",
resolvedModel: "claude-mythos-5-2",
displayName: "Mythos",
description: "",
},
],
});
const { models } = await client.fetchCatalog({
scope: "workspace",
cwd: configDir,
force: false,
});
expect(models.some((model) => model.id === "claude-mythos-5-2")).toBe(false);
});
it("offers one 1M model with medium default effort, always-on thinking, and fast mode", () => {
const models = getClaudeModels().filter((model) => model.id.startsWith("claude-mythos-5-2"));
expect(models).toHaveLength(1);
expect(models[0]).toMatchObject({
id: "claude-mythos-5-2",
label: "Mythos 5.2",
contextWindowMaxTokens: 1_000_000,
defaultThinkingOptionId: "medium",
});
expect(models[0].thinkingOptions?.map((option) => option.id)).toEqual([
"low",
"medium",
"high",
"xhigh",
"max",
CLAUDE_ULTRACODE_THINKING_OPTION_ID,
]);
expect(claudeManifestModelSupportsFastMode("claude-mythos-5-2")).toBe(true);
});
it.each(["anthropic/claude-mythos-5-2", "us.anthropic.claude-mythos-5-2-v1:0"])(
"normalizes runtime identity without granting first-party capabilities: %s",
(id) => {
expect(normalizeClaudeRuntimeModelId(id)).toBe("claude-mythos-5-2");
expect(normalizeClaudeManifestModelId(id)).toBeNull();
},
);
});
describe("Claude model discovery process", () => {
it("does not send a prompt or persist a session and closes after discovery", async () => {
let closed = false;
@@ -6,6 +6,115 @@ import { expect, test } from "vitest";
import { DiffOpcode, type DiffFrame } from "@getpaseo/protocol/binary-frames/index";
import { DiffSubscriptions } from "./index.js";
import { createFileObserver } from "../file-observer/index.js";
import {
configureGitProcessPolicy,
runGitCommand,
runWithGitCommandPriority,
} from "../../utils/run-git-command.js";
import {
DEFAULT_GIT_PROCESS_POLICY,
resolveGitProcessPolicy,
} from "../../utils/git-process-scheduler.js";
async function createObservedWorktreeDiff(prefix: string) {
const cwd = await mkdtemp(join(tmpdir(), prefix));
const git = (...args: string[]) => execFileSync("git", args, { cwd, encoding: "utf8" }).trim();
const observer = createFileObserver();
const subscriptions = new DiffSubscriptions({
async requestWorkingTreeWatch(root, onChange) {
const watch = await observer.subscribe(root, () => onChange(), {
ignore: [join(root, ".git")],
});
return {
repoRoot: root,
unsubscribe: async () => {
await watch.unsubscribe();
},
};
},
});
git("init", "-b", "main");
git("config", "user.email", "test@example.com");
git("config", "user.name", "Test");
for (const directory of ["src", "src/nested", "docs"]) {
await mkdir(join(cwd, directory), { recursive: true });
for (let index = 0; index < 40; index++)
await writeFile(join(cwd, directory, `${index}.ts`), `export const value = ${index};\n`);
}
git("add", ".");
git("-c", "commit.gpgsign=false", "commit", "-m", "base");
for (let index = 0; index < 10; index++)
await writeFile(join(cwd, "src", `${index}.ts`), `export const value = ${index + 100};\n`);
return {
cwd,
subscriptions,
compare: {
cwd,
base: { kind: "rev" as const, rev: "HEAD" },
head: { kind: "worktree" as const },
highlight: true,
},
async close() {
await subscriptions.close();
await observer.close();
await rm(cwd, { recursive: true, force: true });
},
};
}
test("a cold worktree diff over existing files is complete at its first revision", async () => {
const fixture = await createObservedWorktreeDiff("diff-cold-revision-");
const frames: DiffFrame[] = [];
try {
const subscription = await fixture.subscriptions.subscribe(fixture.compare, async (frame) => {
frames.push(frame);
});
await subscription.start();
await writeFile(join(fixture.cwd, "docs", "0.ts"), "export const value = 999;\n");
await expect
.poll(() => frames.filter((frame) => frame.opcode === DiffOpcode.End).length, {
timeout: 10_000,
})
.toBe(2);
expect(
frames.filter((frame) => frame.opcode === DiffOpcode.Begin).map((frame) => frame.revision),
).toEqual([1, 2]);
await subscription.release();
} finally {
await fixture.close();
}
}, 30_000);
test("a cold diff finishes ahead of a queued background Git burst", async () => {
const fixture = await createObservedWorktreeDiff("diff-cold-burst-");
// The suite lifts the start-rate cap; the burst needs the daemon's real one.
configureGitProcessPolicy(DEFAULT_GIT_PROCESS_POLICY);
try {
let burstSettledAt = Number.POSITIVE_INFINITY;
// ~3 s of queued work at 64 starts/s, the shape of a reconciliation rescan.
const burst = runWithGitCommandPriority("background", () =>
Promise.all(
Array.from({ length: 200 }, () =>
runGitCommand(["rev-parse", "HEAD"], { cwd: fixture.cwd }),
),
),
).then(() => {
burstSettledAt = performance.now();
return undefined;
});
let readyAt = Number.POSITIVE_INFINITY;
const subscription = await fixture.subscriptions.subscribe(fixture.compare, async (frame) => {
if (frame.opcode === DiffOpcode.End && frame.error === null) readyAt = performance.now();
});
await subscription.start();
await burst;
expect(readyAt).toBeLessThan(burstSettledAt);
await subscription.release();
} finally {
configureGitProcessPolicy(resolveGitProcessPolicy({ env: process.env }));
await fixture.close();
}
}, 30_000);
test("shares computation, highlights only the changed file, recovers refs and releases observations", async () => {
const cwd = await mkdtemp(join(tmpdir(), "diff-subscriptions-"));
@@ -47,6 +47,29 @@ test("observes nested files while pruning excluded directories", async () => {
await subscription.unsubscribe();
});
test("subscribing to a populated tree reports only later changes", async () => {
const root = await createRoot();
for (const directory of ["src", join("src", "nested"), "docs"]) {
await mkdir(join(root, directory), { recursive: true });
await Promise.all(
Array.from({ length: 20 }, (_, index) =>
writeFile(join(root, directory, `existing-${index}.txt`), `${index}`),
),
);
}
const events: FileChange[] = [];
const subscription = await subscribeToFileChanges(root, (error, batch) => {
expect(error).toBeNull();
events.push(...batch);
});
const laterPath = join(root, "src", "nested", "later.txt");
await writeFile(laterPath, "later");
await expect.poll(() => events.map((event) => event.path)).toContain(laterPath);
expect(new Set(events.map((event) => event.path))).toEqual(new Set([laterPath]));
await subscription.unsubscribe();
});
test("covers files populated immediately inside a newly created directory", async () => {
const root = await createRoot();
const events: FileChange[] = [];
@@ -23,6 +23,11 @@ const RECONCILIATION_DELAY_MS = 50;
const MAX_CONCURRENT_CLASSIFICATIONS = 32;
const MAX_QUEUED_CLASSIFICATIONS = 16_384;
// "baseline" establishes coverage of what already exists when the subscription
// starts, so it reports nothing. "changes" scans topology that appeared while
// observed, so files found before their directory was covered are creates.
type ReconcileMode = "baseline" | "changes";
interface DirectorySnapshot {
directories: Set<string>;
filesDiscoveredBeforeCoverage: Set<string>;
@@ -62,7 +67,7 @@ class LinuxBackend implements ObservationBackend {
async start(): Promise<void> {
this.watchDirectory(this.host.root);
await this.enqueue([this.host.root]);
await this.enqueue([this.host.root], "baseline");
}
async updateIgnore(): Promise<void> {
@@ -75,7 +80,7 @@ class LinuxBackend implements ObservationBackend {
}
}
this.cancelQueued();
await this.enqueue([this.host.root]);
await this.enqueue([this.host.root], "changes");
await Promise.allSettled(inFlight);
}
@@ -159,15 +164,17 @@ class LinuxBackend implements ObservationBackend {
this.timer = null;
const scopes = [...this.pendingScopes];
this.pendingScopes.clear();
void this.enqueue(scopes).catch((error: unknown) => this.host.fail(toError(error)));
void this.enqueue(scopes, "changes").catch((error: unknown) =>
this.host.fail(toError(error)),
);
}, RECONCILIATION_DELAY_MS);
this.timer.unref();
}
private enqueue(scopes: string[]): Promise<void> {
private enqueue(scopes: string[], mode: ReconcileMode): Promise<void> {
this.queueDepth += 1;
const result = this.reconcileTail
.then(() => this.run(scopes))
.then(() => this.run(scopes, mode))
.finally(() => {
this.queueDepth -= 1;
});
@@ -175,12 +182,12 @@ class LinuxBackend implements ObservationBackend {
return result;
}
private async run(scopes: string[]): Promise<void> {
private async run(scopes: string[], mode: ReconcileMode): Promise<void> {
if (!this.host.isActive()) return;
this.inFlight = true;
const startedAt = performance.now();
try {
await this.reconcile(this.paths.collapse(scopes));
await this.reconcile(this.paths.collapse(scopes), mode);
} catch (error) {
this.host.metrics.reconciliationFailureCount += 1;
throw error;
@@ -197,15 +204,17 @@ class LinuxBackend implements ObservationBackend {
}
}
private async reconcile(scopes: string[]): Promise<void> {
private async reconcile(scopes: string[], mode: ReconcileMode): Promise<void> {
for (const scope of scopes) {
const snapshot = await this.scan(scope);
if (!this.host.isActive()) return;
for (const directory of snapshot.directories) {
if (!this.watchSnapshotDirectory(directory)) snapshot.directories.delete(directory);
}
for (const path of snapshot.filesDiscoveredBeforeCoverage) {
this.host.queueEvent("create", path);
if (mode === "changes") {
for (const path of snapshot.filesDiscoveredBeforeCoverage) {
this.host.queueEvent("create", path);
}
}
for (const [directory, watcher] of this.watchers) {
const disappeared =
@@ -1,6 +1,3 @@
import { channel } from "node:diagnostics_channel";
import type { ChildProcess } from "node:child_process";
import { startGitCommandMetrics, getGitCommandMetrics } from "../utils/run-git-command.js";
import { AgentDirectory } from "./agent/directory.js";
import { buildStoredAgentPayload } from "./agent/agent-projections.js";
import {
@@ -21,12 +18,11 @@ import {
rmSync,
statSync,
writeFileSync,
writeSync,
} from "node:fs";
import { tmpdir } from "node:os";
import path from "node:path";
import { setImmediate as waitForImmediate } from "node:timers/promises";
import { afterEach, expect, onTestFinished, test, vi } from "vitest";
import { afterEach, expect, test, vi } from "vitest";
import { z } from "zod";
import { CLIENT_CAPS } from "@getpaseo/protocol/client-capabilities";
@@ -1105,98 +1101,21 @@ test("create_agent_request keeps requested child cwd when grouped under an exist
});
test("create_agent_request launches from an exact subdirectory in a created worktree", async () => {
// Temporary diagnostic: preserve the first invocation while separating process exit
// from stream closure. Metrics alone count exit, not the Git promise settling.
const started = performance.now();
const phases: Array<{ name: string; atMs: number }> = [];
const children: Array<{
id: number;
pid?: number;
operation?: string;
events: Array<{ name: string; atMs: number }>;
}> = [];
let observing = true;
const elapsed = () => Math.round(performance.now() - started);
const mark = (name: string) => {
if (observing) phases.push({ name, atMs: elapsed() });
};
const spawned = channel("child_process");
const observe = (message: unknown) => {
const child = (message as { process: ChildProcess }).process;
const entry: (typeof children)[number] = { id: children.length + 1, events: [] };
children.push(entry);
const record = (name: string) => {
if (observing) entry.events.push({ name, atMs: elapsed() });
};
record("created");
child.once("spawn", () => {
entry.pid = child.pid;
entry.operation =
child.spawnargs.find((arg) =>
["rev-parse", "check-ref-format", "show-ref", "for-each-ref", "worktree"].includes(arg),
) ?? "other";
record("spawn");
child.stdout?.once("end", () => record("stdout end"));
child.stdout?.once("close", () => record("stdout close"));
child.stderr?.once("end", () => record("stderr end"));
child.stderr?.once("close", () => record("stderr close"));
});
child.once("exit", () => record("exit"));
child.once("close", () => record("close"));
child.once("error", () => record("error"));
};
const report = (reason: string) => {
const metrics = getGitCommandMetrics();
writeSync(
2,
JSON.stringify({
test: "worktree subdirectory",
reason,
elapsedMs: elapsed(),
phases,
children,
submitted: metrics.submitted,
completed: metrics.completed,
active: metrics.active,
pending: metrics.pending,
operations: metrics.submissions.map(({ args }) => args.slice(0, 2).join(" ")),
}) + "\n",
);
};
spawned.subscribe(observe);
startGitCommandMetrics();
const watchdog = setTimeout(() => report("watchdog"), 25000);
watchdog.unref();
onTestFinished(() => {
clearTimeout(watchdog);
spawned.unsubscribe(observe);
mark("observer detached");
observing = false;
report("test finished");
});
const fixtureGit = (args: string[], options: { cwd: string; stdio: "pipe" }) => {
mark(`fixture git ${args[0]} start`);
const output = execFileSync("git", args, options);
mark(`fixture git ${args[0]} end`);
return output;
};
mark("repository setup");
const workdir = mkdtempSync(path.join(tmpdir(), "paseo-create-agent-worktree-cwd-"));
try {
const parent = path.join(workdir, "parent");
const child = path.join(parent, "packages", "app");
mkdirSync(child, { recursive: true });
fixtureGit(["init", "-b", "main"], { cwd: parent, stdio: "pipe" });
fixtureGit(["config", "user.email", "test@getpaseo.local"], {
execFileSync("git", ["init", "-b", "main"], { cwd: parent, stdio: "pipe" });
execFileSync("git", ["config", "user.email", "test@getpaseo.local"], {
cwd: parent,
stdio: "pipe",
});
fixtureGit(["config", "user.name", "Paseo Test"], { cwd: parent, stdio: "pipe" });
execFileSync("git", ["config", "user.name", "Paseo Test"], { cwd: parent, stdio: "pipe" });
writeFileSync(path.join(child, "README.md"), "app\n");
fixtureGit(["add", "."], { cwd: parent, stdio: "pipe" });
fixtureGit(["commit", "-m", "initial"], { cwd: parent, stdio: "pipe" });
execFileSync("git", ["add", "."], { cwd: parent, stdio: "pipe" });
execFileSync("git", ["commit", "-m", "initial"], { cwd: parent, stdio: "pipe" });
mark("session setup");
const logger = {
child: () => logger,
trace: vi.fn(),
@@ -1233,7 +1152,6 @@ test("create_agent_request launches from an exact subdirectory in a created work
resolveRepoRoot: async () => parent,
resolveDefaultBranch: async () => "main",
});
mark("project registry write");
await projectRegistry.upsert(
createPersistedProjectRecord({
projectId: "proj-parent",
@@ -1244,7 +1162,6 @@ test("create_agent_request launches from an exact subdirectory in a created work
updatedAt: "2026-05-07T00:00:00.000Z",
}),
);
mark("workspace registry write");
await workspaceRegistry.upsert(
createPersistedWorkspaceRecord({
workspaceId: "ws-parent",
@@ -1257,7 +1174,6 @@ test("create_agent_request launches from an exact subdirectory in a created work
}),
);
mark("session construction");
const emitted: SessionOutboundMessage[] = [];
const session = new Session({
workspaceRuntimeStore: createWorkspaceRuntimeStore(),
@@ -1315,7 +1231,6 @@ test("create_agent_request launches from an exact subdirectory in a created work
terminalManager: null,
});
mark("create agent request");
await session.handleMessage({
type: "create_agent_request",
requestId: "req-create-worktree-child",
@@ -1324,9 +1239,8 @@ test("create_agent_request launches from an exact subdirectory in a created work
worktree: { mode: "branch-off", newBranch: "feature/created-worktree" },
});
mark("create agent returned");
const [createdAgent] = agentManager.listAgents();
const createdWorktreeRoot = fixtureGit(["rev-parse", "--show-toplevel"], {
const createdWorktreeRoot = execFileSync("git", ["rev-parse", "--show-toplevel"], {
cwd: createdAgent!.cwd,
stdio: "pipe",
})
@@ -1342,9 +1256,7 @@ test("create_agent_request launches from an exact subdirectory in a created work
agent: { cwd: createdAgent?.cwd },
});
} finally {
mark("cleanup start");
rmSync(workdir, { recursive: true, force: true });
mark("cleanup end");
}
});
@@ -1,4 +1,9 @@
import { flushGitCommandTrace } from "../utils/git-command-trace.js";
import {
startGitCommandMetrics,
stopGitCommandMetrics,
waitForGitCommandMetricsIdle,
} from "../utils/run-git-command.js";
import { createWorkspaceRuntimeStore } from "./workspace-runtime-store/index.js";
import { mkdirSync, mkdtempSync, readFileSync, realpathSync, rmSync, writeFileSync } from "node:fs";
import { execFileSync } from "node:child_process";
@@ -856,3 +861,59 @@ test.skipIf(process.platform !== "linux")(
},
60_000,
);
test("observation-driven refreshes run their Git work in the background lane", async () => {
const root = realpathSync(mkdtempSync(path.join(tmpdir(), "paseo-git-priority-")));
const repo = path.join(root, "repo");
mkdirSync(repo);
const git = (...args: string[]) =>
execFileSync("git", ["-c", "commit.gpgsign=false", ...args], { cwd: repo, stdio: "pipe" });
git("init", "-b", "main");
git("config", "user.email", "test@example.com");
git("config", "user.name", "Test");
writeFileSync(path.join(repo, "README.md"), "base\n");
git("add", ".");
git("commit", "-m", "base");
const service = new WorkspaceGitServiceImpl({
logger: pino({ enabled: false }),
paseoHome: path.join(root, "home"),
});
const heads: Array<string | null> = [];
const subscription = service.registerWorkspace({ cwd: repo }, (snapshot) => {
heads.push(snapshot.git.currentBranch);
});
cleanup.push(async () => {
subscription.unsubscribe();
await service.dispose();
rmSync(root, { recursive: true, force: true });
});
// Registration, observation setup and the first refresh settle before the
// measured change.
startGitCommandMetrics();
await service.getSnapshot(repo, { includeForge: false });
await waitForGitCommandMetricsIdle({ quietMs: 500, timeoutMs: 10_000 });
stopGitCommandMetrics();
startGitCommandMetrics();
const deliveredBefore = heads.length;
git("switch", "-c", "feature");
await expect.poll(() => heads.slice(deliveredBefore), { timeout: 10_000 }).toContain("feature");
await waitForGitCommandMetricsIdle({ quietMs: 300, timeoutMs: 10_000 });
const observed = stopGitCommandMetrics().submissions;
expect(observed.length).toBeGreaterThan(0);
expect(new Set(observed.map((command) => command.priority))).toEqual(new Set(["background"]));
startGitCommandMetrics();
await service.getSnapshot(repo, { force: true, reason: "test-demand", includeForge: false });
await service.getCheckout(repo, { priority: "background" });
await waitForGitCommandMetricsIdle({ quietMs: 0, timeoutMs: 10_000 });
const demand = stopGitCommandMetrics().submissions;
const checkoutStart = demand.findIndex((command) => command.priority === "background");
expect(checkoutStart).toBeGreaterThan(0);
expect(new Set(demand.slice(0, checkoutStart).map((command) => command.priority))).toEqual(
new Set(["normal"]),
);
expect(new Set(demand.slice(checkoutStart).map((command) => command.priority))).toEqual(
new Set(["background"]),
);
}, 30_000);
@@ -48,6 +48,7 @@ import {
runGitCommand,
type RunGitCommand,
} from "../utils/run-git-command.js";
import type { GitProcessPriority } from "../utils/git-process-scheduler.js";
import { GitObservationPaths } from "./git-observation-paths.js";
import { branchNameFromRef } from "../utils/worktree-metadata.js";
import { listPaseoWorktrees, type PaseoWorktreeInfo } from "../utils/worktree.js";
@@ -209,7 +210,10 @@ export interface WorkspaceGitService {
/** Publishes computed snapshot changes independently of workspace UI subscriptions. */
onSnapshotUpdated(listener: WorkspaceGitSnapshotUpdatedListener): WorkspaceGitSubscription;
getCheckout(cwd: string): Promise<ProjectCheckoutLitePayload>;
getCheckout(
cwd: string,
options?: WorkspaceGitCheckoutReadOptions,
): Promise<ProjectCheckoutLitePayload>;
getSnapshot(
cwd: string,
options?: WorkspaceGitSnapshotOptions,
@@ -329,7 +333,28 @@ export type WorkspaceGitSnapshotOptions =
reason: string;
};
/**
* Who is waiting for a refresh. "demand" refreshes answer a caller; "observation"
* refreshes react to watchers, polls and fetches and run in the background Git
* lane so they never queue in front of interactive Git work.
*/
type WorkspaceGitRefreshOrigin = "demand" | "observation";
export interface WorkspaceGitCheckoutReadOptions {
/** Git lane for the fresh checkout read; defaults to the caller's ambient lane. */
priority?: GitProcessPriority;
}
// A caller waiting on a merged refresh keeps it out of the background lane.
function mergeRefreshOrigins(
pending: WorkspaceGitRefreshOrigin,
request: WorkspaceGitRefreshOrigin,
): WorkspaceGitRefreshOrigin {
return pending === "demand" ? pending : request;
}
interface WorkspaceGitRefreshRequest {
origin: WorkspaceGitRefreshOrigin;
force: boolean;
refreshStructure: boolean;
refreshWorktree: boolean;
@@ -341,6 +366,7 @@ interface WorkspaceGitRefreshRequest {
}
interface ScheduledWorkspaceGitRefreshOptions {
origin: WorkspaceGitRefreshOrigin;
force?: boolean;
scope?: "refs" | "structure" | "worktree";
includeForge?: boolean;
@@ -782,13 +808,19 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
return this.requestWorkspaceSnapshot(target, request);
}
async getCheckout(cwd: string): Promise<ProjectCheckoutLitePayload> {
async getCheckout(
cwd: string,
options?: WorkspaceGitCheckoutReadOptions,
): Promise<ProjectCheckoutLitePayload> {
this.assertNotDisposed();
const normalizedCwd = resolve(cwd);
const status = await this.deps.getCheckoutStatus(normalizedCwd, {
paseoHome: this.paseoHome,
worktreesRoot: this.worktreesRoot,
logger: this.logger,
...(options?.priority
? { runGitCommand: createRunGitCommand("checkout-read", { priority: options.priority }) }
: {}),
});
const cached = this.getCachedSnapshot(normalizedCwd);
if (cached && cached.git.isGit !== status.isGit) {
@@ -1009,6 +1041,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
await this.resumeWorkspaceIfFolderRestored(target);
}
await this.refreshWorkspaceTarget(target, {
origin: "demand",
force: false,
refreshStructure: true,
refreshWorktree: true,
@@ -1048,7 +1081,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
void this.resumeWorkspaceIfFolderRestored(target);
return;
}
this.scheduleWorkspaceRefresh(target, { queueIfBusy: false });
this.scheduleWorkspaceRefresh(target, { origin: "demand", queueIfBusy: false });
}
onWorkspaceStateMayHaveChanged(cwd: string): void {
@@ -1064,6 +1097,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
return;
}
this.scheduleWorkspaceRefresh(target, {
origin: "demand",
force: true,
includeForge: true,
reason: "external-state-change",
@@ -1292,6 +1326,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
return;
}
void this.refreshWorkspaceTarget(target, {
origin: "demand",
force: false,
refreshStructure: true,
refreshWorktree: true,
@@ -1762,6 +1797,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
target.latestFacts = null;
target.folderMissing = false;
await this.refreshWorkspaceTarget(target, {
origin: "observation",
force: false,
refreshStructure: true,
refreshWorktree: true,
@@ -1805,6 +1841,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
return false;
}
const changed = await this.refreshDegradedPollTarget(workspaceTarget, {
origin: "observation",
force: false,
refreshStructure: target.repoRoot === null,
refreshWorktree: true,
@@ -2286,7 +2323,11 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
for (const workspaceKey of target.workspaceKeys) {
const workspaceTarget = this.workspaceTargets.get(workspaceKey);
if (workspaceTarget) {
this.scheduleWorkspaceRefresh(workspaceTarget, { scope: "worktree", reason });
this.scheduleWorkspaceRefresh(workspaceTarget, {
origin: "observation",
scope: "worktree",
reason,
});
}
}
}
@@ -2869,6 +2910,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
}
}
this.scheduleWorkspaceRefresh(workspaceTarget, {
origin: "observation",
scope,
emitUnchanged: refresh.refreshBase,
reason,
@@ -2909,6 +2951,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
workingTreeTargets.add(workingTreeTarget);
}
return await this.refreshDegradedPollTarget(workspaceTarget, {
origin: "observation",
force: false,
refreshStructure: true,
refreshWorktree: true,
@@ -2940,7 +2983,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
private scheduleWorkspaceRefresh(
targetOrCwd: WorkspaceGitTarget | string,
options?: ScheduledWorkspaceGitRefreshOptions,
options: ScheduledWorkspaceGitRefreshOptions,
): void {
const target =
typeof targetOrCwd === "string"
@@ -3309,6 +3352,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
const force = options?.force === true;
return {
origin: "demand",
force,
refreshStructure: true,
refreshWorktree: true,
@@ -3332,10 +3376,11 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
}
private buildScheduledRefreshRequest(
options: ScheduledWorkspaceGitRefreshOptions | undefined,
options: ScheduledWorkspaceGitRefreshOptions,
): WorkspaceGitRefreshRequest {
const scope = options?.scope;
return {
origin: options.origin,
force: options?.force === true,
refreshStructure: scope !== "worktree" && scope !== "refs",
refreshWorktree: scope !== "structure" && scope !== "refs",
@@ -3366,6 +3411,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
request.movedRemoteRefs,
);
return {
origin: mergeRefreshOrigins(pending.origin, request.origin),
force,
refreshStructure: pending.refreshStructure || request.refreshStructure,
refreshWorktree: pending.refreshWorktree || request.refreshWorktree,
@@ -3404,7 +3450,9 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
});
}
try {
const runRefreshGitCommand = createRunGitCommand(`workspace-refresh:${request.reason}`);
const runRefreshGitCommand = createRunGitCommand(`workspace-refresh:${request.reason}`, {
priority: request.origin === "demand" ? "normal" : "background",
});
const admittedSnapshot = await this.workspaceRefreshLimit(() => {
if (
target.closed ||
@@ -3792,7 +3840,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
}
},
},
createRunGitCommand("background-fetch"),
createRunGitCommand("background-fetch", { priority: "background" }),
);
} catch (error) {
this.logger.warn(
@@ -772,6 +772,62 @@ describe("WorkspaceReconciliationService", () => {
expect(projects.get("p1")?.kind).toBe("git");
});
test("reads sibling checkouts a few at a time in the background Git lane", async () => {
const projectRoot = "/reconcile/fan-out/root";
const { projects, workspaces, projectRegistry, workspaceRegistry } = createTestRegistries();
projects.set(
"p1",
createPersistedProjectRecord({
projectId: "p1",
rootPath: projectRoot,
kind: "git",
displayName: "fan-out",
createdAt: timestamp,
updatedAt: timestamp,
}),
);
const siblingCwds = Array.from({ length: 40 }, (_, index) => `${projectRoot}-${index}`);
for (const [index, cwd] of siblingCwds.entries()) {
workspaces.set(
`w${index}`,
createPersistedWorkspaceRecord({
workspaceId: `w${index}`,
projectId: "p1",
cwd,
kind: "worktree",
displayName: `sibling-${index}`,
createdAt: timestamp,
updatedAt: timestamp,
}),
);
}
const reads: Array<{ cwd: string; priority: string | undefined }> = [];
let inFlight = 0;
let maxInFlight = 0;
const service = new WorkspaceReconciliationService({
projectRegistry,
workspaceRegistry,
logger: createTestLogger(),
inspectDirectory: async () => "directory",
workspaceGitService: {
async getCheckout(cwd, options) {
reads.push({ cwd, priority: options?.priority });
inFlight += 1;
maxInFlight = Math.max(maxInFlight, inFlight);
await new Promise((resolve) => setTimeout(resolve, 1));
inFlight -= 1;
return createCheckout(cwd, { isGit: true, currentBranch: "main", worktreeRoot: cwd });
},
},
});
await service.runOnce();
expect(reads.map((read) => read.cwd).sort()).toEqual([projectRoot, ...siblingCwds].sort());
expect(new Set(reads.map((read) => read.priority))).toEqual(new Set(["background"]));
expect(maxInFlight).toBe(4);
});
test("deduplicates equivalent project and workspace paths across legacy duplicate projects", async () => {
const projectRoot = realpathSync(mkdtempSync(path.join(tmpdir(), "reconcile-global-root-")));
const workspaceRoot = realpathSync(
@@ -1,6 +1,7 @@
import { watch as watchPath } from "node:fs";
import { stat } from "node:fs/promises";
import { setImmediate as yieldToLoop } from "node:timers/promises";
import pLimit from "p-limit";
import type { ProjectCheckoutLitePayload } from "@getpaseo/protocol/messages";
import type pino from "pino";
import type {
@@ -22,6 +23,9 @@ import { deriveProjectKey } from "./project-key.js";
const DEFAULT_RESCAN_INTERVAL_MS = 5 * 60_000;
const DEFAULT_DEBOUNCE_MS = 100;
const INSPECTION_CONCURRENCY = 2;
// Each checkout read runs ~15 Git commands. Reading every sibling at once
// queued ~1,300 commands per rescan on a project with ~90 worktrees.
const CHECKOUT_READ_CONCURRENCY = 4;
const DEFAULT_INSPECTION_TIMEOUT_MS = 5000;
export type ProjectUpdate =
@@ -346,10 +350,11 @@ export class WorkspaceReconciliationService {
changes: ReconciliationChange[],
): Promise<void> {
const checkoutReads: CachedCheckoutRead[] = [];
const checkoutReadLimit = pLimit(CHECKOUT_READ_CONCURRENCY);
const readCheckout = (cwd: string): Promise<ProjectCheckoutLitePayload> => {
const existing = checkoutReads.find((read) => areEquivalentPaths(read.cwd, cwd));
if (existing) return existing.checkout;
const checkout = this.readCheckout(cwd);
const checkout = checkoutReadLimit(() => this.readCheckout(cwd));
checkoutReads.push({ cwd, checkout });
return checkout;
};
@@ -581,7 +586,9 @@ export class WorkspaceReconciliationService {
mainRepoRoot: null,
};
}
return this.workspaceGitService.getCheckout(cwd);
// Reconciliation converges in the background; it must not queue ahead of
// Git work a user is waiting for.
return this.workspaceGitService.getCheckout(cwd, { priority: "background" });
}
private wakeInspectionWaiters(): void {
@@ -68,6 +68,64 @@ describe("GitProcessScheduler", () => {
expect(started).toEqual([0, 3, 2]);
});
test("admits normal work before a queued background burst", async () => {
const scheduler = new GitProcessScheduler({
maxProcessesPerSecond: 100,
maxProcessConcurrency: 1,
});
const releases: Array<() => void> = [];
const started: number[] = [];
const activeBackground = scheduler.run(createConcurrencyTask(0, started, releases), {
priority: "background",
});
await vi.waitFor(() => expect(started).toEqual([0]));
const queuedBackground = [2, 3, 4].map((index) =>
scheduler.run(createConcurrencyTask(index, started, releases), { priority: "background" }),
);
const normalWork = scheduler.run(createConcurrencyTask(5, started, releases));
const highWork = scheduler.run(createConcurrencyTask(6, started, releases), {
priority: "high",
});
releases.shift()?.();
await Promise.all([activeBackground, ...queuedBackground, normalWork, highWork]);
expect(started).toEqual([0, 6, 5, 2, 3, 4]);
});
test("starts normal work at the next start slot when background work saturates the rate", async () => {
vi.useFakeTimers();
const scheduler = new GitProcessScheduler({
maxProcessesPerSecond: 2,
maxProcessConcurrency: 8,
});
const started: string[] = [];
const task = (name: string) => () => {
started.push(name);
const completed = Promise.resolve();
return { result: completed, exited: completed };
};
const background = Array.from({ length: 8 }, (_, index) =>
scheduler.run(task(`background-${index}`), { priority: "background" }),
);
// Admission yields a real event-loop turn per process; let every
// background job reach the start-rate limiter.
for (let turn = 0; turn < 10; turn += 1) {
await new Promise<void>((resolve) => scheduleImmediate(resolve));
}
expect(started).toEqual(["background-0", "background-1"]);
const normal = scheduler.run(task("normal"));
while (started.length < 3) {
await vi.advanceTimersToNextTimerAsync();
}
expect(started[2]).toBe("normal");
await vi.runAllTimersAsync();
await Promise.all([...background, normal]);
});
test("yields to daemon liveness work between process starts", async () => {
const scheduler = new GitProcessScheduler({
maxProcessesPerSecond: 100,
@@ -16,7 +16,10 @@ export interface ScheduledGitProcess<T> {
exited: Promise<void>;
}
export type GitProcessPriority = "normal" | "high";
// Lanes drain in this order. "background" is for daemon-initiated fan-out
// (reconciliation, observation-driven refreshes, fetches) so it never sits in
// front of work a user is waiting for.
export type GitProcessPriority = "high" | "normal" | "background";
function parsePositiveInteger(value: string | undefined): number | undefined {
if (value === undefined) {
@@ -44,11 +47,19 @@ export function resolveGitProcessPolicy(input: {
};
}
// Admission reserves a process slot and a start-rate token; the job runs only
// once its token fires. The job is chosen when the token fires, not when the
// slot is reserved, so work submitted while background jobs hold every reserved
// slot still takes the next start.
export class GitProcessScheduler {
private readonly startThrottled;
private readonly highPriorityQueue: Array<() => void> = [];
private readonly normalPriorityQueue: Array<() => void> = [];
private readonly queues: Record<GitProcessPriority, Array<() => Promise<void>>> = {
high: [],
normal: [],
background: [],
};
private admitted = 0;
private reservedStarts = 0;
private running = 0;
private waiting = 0;
private drainScheduled = false;
@@ -59,7 +70,7 @@ export class GitProcessScheduler {
interval: 1_000,
strict: true,
});
this.startThrottled = throttle((start: () => Promise<void>) => start());
this.startThrottled = throttle(() => this.startNext());
}
get activeCount(): number {
@@ -76,17 +87,7 @@ export class GitProcessScheduler {
): Promise<T> {
this.waiting += 1;
return new Promise<T>((resolve, reject) => {
const admit = () => {
void this.startThrottled(() => this.execute(start, resolve, reject)).catch((error) => {
this.waiting = Math.max(0, this.waiting - 1);
this.admitted = Math.max(0, this.admitted - 1);
reject(error);
this.scheduleDrain();
});
};
const queue =
options?.priority === "high" ? this.highPriorityQueue : this.normalPriorityQueue;
queue.push(admit);
this.queues[options?.priority ?? "normal"].push(() => this.execute(start, resolve, reject));
if (this.admitted === 0 && !this.drainScheduled) {
this.drainOne();
} else {
@@ -95,11 +96,12 @@ export class GitProcessScheduler {
});
}
private canReserveStart(): boolean {
return this.admitted < this.policy.maxProcessConcurrency && this.waiting > this.reservedStarts;
}
private scheduleDrain(): void {
if (this.drainScheduled || this.admitted >= this.policy.maxProcessConcurrency) {
return;
}
if (this.highPriorityQueue.length === 0 && this.normalPriorityQueue.length === 0) {
if (this.drainScheduled || !this.canReserveStart()) {
return;
}
this.drainScheduled = true;
@@ -108,20 +110,28 @@ export class GitProcessScheduler {
private drainOne(): void {
this.drainScheduled = false;
if (this.admitted >= this.policy.maxProcessConcurrency) {
return;
}
const admit = this.highPriorityQueue.shift() ?? this.normalPriorityQueue.shift();
if (!admit) {
if (!this.canReserveStart()) {
return;
}
this.admitted += 1;
admit();
this.reservedStarts += 1;
void this.startThrottled();
// Spawning can be synchronous and expensive on a throttled host. Let timers,
// sockets, and child exits run before admitting another Git process.
this.scheduleDrain();
}
private startNext(): Promise<void> {
this.reservedStarts -= 1;
// Every reservation is backed by a queued job: jobs leave the queues only here.
const next =
this.queues.high.shift() ?? this.queues.normal.shift() ?? this.queues.background.shift();
if (!next) {
throw new Error("Git process start fired without a queued process");
}
return next();
}
private async execute<T>(
start: () => ScheduledGitProcess<T>,
resolve: (value: T | PromiseLike<T>) => void,
+34 -19
View File
@@ -67,8 +67,17 @@ export type RunGitCommand = (
options: GitCommandOptions,
) => Promise<GitCommandResult>;
export function createRunGitCommand(provenance: string): RunGitCommand {
return (args, options) => runGitCommandWithProvenance(args, options, provenance);
export interface GitCommandRunnerOptions {
/** Overrides the ambient `runWithGitCommandPriority` lane for every command this runner starts. */
priority?: GitProcessPriority;
}
export function createRunGitCommand(
provenance: string,
runnerOptions: GitCommandRunnerOptions = {},
): RunGitCommand {
return (args, options) =>
executeGitCommand(args, options, decodeUtf8, { provenance, priority: runnerOptions.priority });
}
export interface GitCommandMetric {
@@ -97,6 +106,7 @@ export interface GitCommandMetricsSnapshot {
export interface GitCommandSubmissionMetric {
args: string[];
cwd: string;
priority: GitProcessPriority;
}
interface GitCommandMetricsState {
@@ -188,8 +198,8 @@ function snapshotGitCommandMetricsState(state: GitCommandMetricsState): GitComma
return {
commands: [...state.commands],
submissions: state.submissions.map((submission) => ({
...submission,
args: [...submission.args],
cwd: submission.cwd,
})),
submitted: state.submitted,
started: state.started,
@@ -209,12 +219,16 @@ export function snapshotGitCommandRuntimeMetrics(): GitCommandRuntimeMetricsSnap
});
}
function submitGitCommandMetric(args: string[], cwd: string): GitCommandMetricsState | null {
function submitGitCommandMetric(
args: string[],
cwd: string,
priority: GitProcessPriority,
): GitCommandMetricsState | null {
const state = gitCommandMetricsState;
if (!state) {
return null;
}
state.submissions.push({ args: [...args], cwd });
state.submissions.push({ args: [...args], cwd, priority });
state.submitted += 1;
state.lastSubmittedAtMs = Date.now();
return state;
@@ -258,12 +272,13 @@ function getEnvOverlayKeys(envOverlay: ProcessEnvRecord | undefined): string[] {
return Object.keys(envOverlay ?? {}).sort();
}
function runGitCommandWithProvenance(
args: string[],
options: GitCommandOptions,
provenance?: string,
): Promise<GitCommandResult> {
return executeGitCommand(args, options, (output) => output.toString("utf8"), provenance);
function decodeUtf8(output: Buffer): string {
return output.toString("utf8");
}
interface GitCommandSubmission {
provenance?: string;
priority?: GitProcessPriority;
}
/** Binary stdout preserves cat-file byte framing before text decoding. */
@@ -271,21 +286,22 @@ export function runGitCommandBytes(
args: string[],
options: GitCommandOptions,
): Promise<GitCommandResult<Buffer>> {
return executeGitCommand(args, options, (output) => output);
return executeGitCommand(args, options, (output) => output, {});
}
function executeGitCommand<Output>(
args: string[],
options: GitCommandOptions,
decode: (output: Buffer) => Output,
provenance?: string,
submission: GitCommandSubmission,
): Promise<GitCommandResult<Output>> {
const metricsState = submitGitCommandMetric(args, options.cwd);
const priority = submission.priority ?? gitCommandPriority.getStore() ?? "normal";
const metricsState = submitGitCommandMetric(args, options.cwd, priority);
const commandTrace = submitGitCommandTrace(args, options.cwd, {
active: gitProcessScheduler.activeCount,
pending: gitProcessScheduler.pendingCount,
});
const runtimeMetric = gitRuntimeMetrics.submit(getGitOperation(args), provenance);
const runtimeMetric = gitRuntimeMetrics.submit(getGitOperation(args), submission.provenance);
const startCommand = () => {
let releaseProcessSlot!: () => void;
const exited = new Promise<void>((resolve) => {
@@ -558,9 +574,7 @@ function executeGitCommand<Output>(
});
return { result: resultPromise, exited };
};
const promise = gitProcessScheduler.run(startCommand, {
priority: gitCommandPriority.getStore() ?? "normal",
});
const promise = gitProcessScheduler.run(startCommand, { priority });
gitRuntimeMetrics.observeLimiter(
gitProcessScheduler.activeCount,
gitProcessScheduler.pendingCount,
@@ -568,7 +582,8 @@ function executeGitCommand<Output>(
return promise;
}
export const runGitCommand: RunGitCommand = runGitCommandWithProvenance;
export const runGitCommand: RunGitCommand = (args, options) =>
executeGitCommand(args, options, decodeUtf8, {});
function formatGitCommand(args: string[]): string {
return ["git", ...args].join(" ");