Merge remote-tracking branch 'origin/main' into fix/usage-source-cli-auth-locations

packages/server/src/server/test-fixtures/reconciliation-registries.ts

@@ -0,0 +1,117 @@
+import type pino from "pino";
+import { createPersistedProjectRecord } from "../workspace-registry.js";
+import type {
+  PersistedProjectRecord,
+  PersistedWorkspaceRecord,
+  ProjectRegistry,
+  WorkspaceRegistry,
+} from "../workspace-registry.js";
+
+export function createTestRegistries() {
+  let archiveGate: {
+    started: ReturnType<typeof deferred>;
+    release: ReturnType<typeof deferred>;
+  } | null = null;
+  const holdNextArchive = () => {
+    archiveGate = { started: deferred(), release: deferred() };
+    return archiveGate;
+  };
+  const projects = new Map<string, PersistedProjectRecord>();
+  const workspaces = new Map<string, PersistedWorkspaceRecord>();
+
+  const projectRegistry: ProjectRegistry = {
+    initialize: async () => {},
+    existsOnDisk: async () => true,
+    list: async () => Array.from(projects.values()),
+    get: async (id: string) => projects.get(id) ?? null,
+    getOrCreateActiveByRoot: async (input) => {
+      const existing = Array.from(projects.values()).find(
+        (project) => !project.archivedAt && project.rootPath === input.rootPath,
+      );
+      if (existing) return existing;
+      const record = createPersistedProjectRecord({
+        projectId: `prj_${projects.size}`,
+        rootPath: input.rootPath,
+        kind: input.kind,
+        displayName: input.displayName,
+        createdAt: input.timestamp,
+        updatedAt: input.timestamp,
+      });
+      projects.set(record.projectId, record);
+      return record;
+    },
+    update: async (id, updater) => {
+      const current = projects.get(id);
+      if (!current) return null;
+      const updated = updater(current);
+      projects.set(id, updated);
+      return updated;
+    },
+    upsert: async (record: PersistedProjectRecord) => {
+      projects.set(record.projectId, record);
+    },
+    archive: async (id: string, archivedAt: string) => {
+      const existing = projects.get(id);
+      if (existing) {
+        projects.set(id, { ...existing, archivedAt, updatedAt: archivedAt });
+      }
+    },
+    remove: async (id: string) => {
+      projects.delete(id);
+    },
+  };
+
+  const workspaceRegistry: WorkspaceRegistry = {
+    initialize: async () => {},
+    existsOnDisk: async () => true,
+    list: async () => Array.from(workspaces.values()),
+    get: async (id: string) => workspaces.get(id) ?? null,
+    update: async (id, updater) => {
+      const existing = workspaces.get(id);
+      if (!existing) return null;
+      const updated = updater(existing);
+      workspaces.set(id, updated);
+      return updated;
+    },
+    upsert: async (record: PersistedWorkspaceRecord) => {
+      workspaces.set(record.workspaceId, record);
+    },
+    archive: async (id: string, archivedAt: string) => {
+      const gate = archiveGate;
+      archiveGate = null;
+      if (gate) {
+        gate.started.resolve();
+        await gate.release.promise;
+      }
+      const existing = workspaces.get(id);
+      if (!existing || existing.archivedAt) return false;
+      workspaces.set(id, { ...existing, archivedAt, updatedAt: archivedAt });
+      return true;
+    },
+    remove: async (id: string) => {
+      workspaces.delete(id);
+    },
+  };
+
+  return { projects, workspaces, projectRegistry, workspaceRegistry, holdNextArchive };
+}
+
+export function createTestLogger() {
+  const logger = {
+    child: () => logger,
+    trace: () => undefined,
+    debug: () => undefined,
+    info: () => undefined,
+    warn: () => undefined,
+    error: () => undefined,
+  };
+  return logger as unknown as pino.Logger;
+}
+
+export function deferred(): { promise: Promise<void>; resolve(): void } {
+  let resolve!: () => void;
+  const promise = new Promise<void>((accept) => {
+    resolve = accept;
+  });
+  return { promise, resolve };
+}

packages/server/src/server/test-fixtures/reconciliation-responsiveness.ts

@@ -0,0 +1,75 @@
+import path from "node:path";
+import { stat } from "node:fs/promises";
+import { setImmediate as nextTurn } from "node:timers/promises";
+import { measureMainThreadWork } from "../../test-utils/main-thread-work.js";
+import { createTestRegistries, createTestLogger } from "./reconciliation-registries.js";
+import {
+  createPersistedProjectRecord,
+  createPersistedWorkspaceRecord,
+} from "../workspace-registry.js";
+import { WorkspaceReconciliationService } from "../workspace-reconciliation-service.js";
+import { deriveProjectKey } from "../project-key.js";
+
+const rootPath = process.argv[2];
+if (!rootPath) throw new Error("Project root is required");
+const timestamp = "2025-01-01T00:00:00.000Z";
+const { projects, workspaces, projectRegistry, workspaceRegistry } = createTestRegistries();
+projects.set(
+  "p1",
+  createPersistedProjectRecord({
+    projectId: "p1",
+    rootPath,
+    projectKey: deriveProjectKey({
+      rootPath,
+      remoteUrl: null,
+      worktreeRoot: null,
+      mainRepoRoot: null,
+    }),
+    kind: "non_git",
+    displayName: "project",
+    createdAt: timestamp,
+    updatedAt: timestamp,
+  }),
+);
+for (let i = 0; i < 20000; i++) {
+  const workspaceId = `w${i}`;
+  workspaces.set(
+    workspaceId,
+    createPersistedWorkspaceRecord({
+      workspaceId,
+      projectId: "p1",
+      cwd: path.join(rootPath, workspaceId),
+      kind: "directory",
+      displayName: workspaceId,
+      createdAt: timestamp,
+      updatedAt: timestamp,
+    }),
+  );
+}
+const service = new WorkspaceReconciliationService({
+  projectRegistry,
+  workspaceRegistry,
+  logger: createTestLogger(),
+});
+
+// Compile before launch and measure in plain Node, outside the test runner's
+// thread and TypeScript loader. Keep the first real reconciliation cold.
+// Initialize async filesystem support using the bundle, leaving every project
+// and workspace path untouched until the first reconciliation.
+await stat(process.argv[1]);
+await nextTurn();
+try {
+  const measured = await measureMainThreadWork("workspace reconciliation main-thread work", () =>
+    service.runOnce(),
+  );
+  process.stdout.write(
+    JSON.stringify({
+      blockingMs: measured.blockingMs,
+      filesystemRequests: measured.filesystemRequests,
+      changesApplied: measured.result.changesApplied.length,
+      allArchived: [...workspaces.values()].every((workspace) => workspace.archivedAt !== null),
+    }),
+  );
+} finally {
+  service.dispose();
+}

packages/server/src/server/workspace-reconciliation-service.test.ts

@@ -1,27 +1,45 @@
-import { measureMainThreadWork } from "../test-utils/main-thread-work.js";
-import { execFileSync } from "node:child_process";
+import { execFile, execFileSync } from "node:child_process";
+import { promisify } from "node:util";
+import { fileURLToPath } from "node:url";
+import { build } from "esbuild";
+import {
+  createTestRegistries,
+  createTestLogger,
+  deferred,
+} from "./test-fixtures/reconciliation-registries.js";
 import { mkdirSync, mkdtempSync, realpathSync, rmSync, writeFileSync } from "node:fs";
 import { tmpdir } from "node:os";
 import path from "node:path";
 import type { ProjectCheckoutLitePayload } from "@getpaseo/protocol/messages";
 import type pino from "pino";
-import { afterEach, describe, expect, test } from "vitest";
+import { afterAll, beforeAll, afterEach, describe, expect, test } from "vitest";
 import {
   createPersistedProjectRecord,
   createPersistedWorkspaceRecord,
 } from "./workspace-registry.js";
-import type {
-  PersistedProjectRecord,
-  PersistedWorkspaceRecord,
-  ProjectRegistry,
-  WorkspaceRegistry,
-} from "./workspace-registry.js";
 import {
   type ReconciliationChange,
   WorkspaceReconciliationService,
 } from "./workspace-reconciliation-service.js";
 import { deriveProjectKey } from "./project-key.js";
 
+let probeDirectory: string;
+let probePath: string;
+beforeAll(async () => {
+  probeDirectory = mkdtempSync(path.join(tmpdir(), "reconciliation-probe-"));
+  probePath = path.join(probeDirectory, "reconciliation.mjs");
+  await build({
+    entryPoints: [
+      fileURLToPath(new URL("./test-fixtures/reconciliation-responsiveness.ts", import.meta.url)),
+    ],
+    outfile: probePath,
+    bundle: true,
+    platform: "node",
+    format: "esm",
+  });
+});
+afterAll(() => rmSync(probeDirectory, { recursive: true, force: true }));
+
 function canonicalLocalProjectKey(rootPath: string): string {
   return deriveProjectKey({
     rootPath,
@@ -31,107 +49,6 @@ function canonicalLocalProjectKey(rootPath: string): string {
   });
 }
 
-function createTestRegistries() {
-  let archiveGate: {
-    started: ReturnType<typeof deferred>;
-    release: ReturnType<typeof deferred>;
-  } | null = null;
-  const holdNextArchive = () => {
-    archiveGate = { started: deferred(), release: deferred() };
-    return archiveGate;
-  };
-  const projects = new Map<string, PersistedProjectRecord>();
-  const workspaces = new Map<string, PersistedWorkspaceRecord>();
-
-  const projectRegistry: ProjectRegistry = {
-    initialize: async () => {},
-    existsOnDisk: async () => true,
-    list: async () => Array.from(projects.values()),
-    get: async (id: string) => projects.get(id) ?? null,
-    getOrCreateActiveByRoot: async (input) => {
-      const existing = Array.from(projects.values()).find(
-        (project) => !project.archivedAt && project.rootPath === input.rootPath,
-      );
-      if (existing) return existing;
-      const record = createPersistedProjectRecord({
-        projectId: `prj_${projects.size}`,
-        rootPath: input.rootPath,
-        kind: input.kind,
-        displayName: input.displayName,
-        createdAt: input.timestamp,
-        updatedAt: input.timestamp,
-      });
-      projects.set(record.projectId, record);
-      return record;
-    },
-    update: async (id, updater) => {
-      const current = projects.get(id);
-      if (!current) return null;
-      const updated = updater(current);
-      projects.set(id, updated);
-      return updated;
-    },
-    upsert: async (record: PersistedProjectRecord) => {
-      projects.set(record.projectId, record);
-    },
-    archive: async (id: string, archivedAt: string) => {
-      const existing = projects.get(id);
-      if (existing) {
-        projects.set(id, { ...existing, archivedAt, updatedAt: archivedAt });
-      }
-    },
-    remove: async (id: string) => {
-      projects.delete(id);
-    },
-  };
-
-  const workspaceRegistry: WorkspaceRegistry = {
-    initialize: async () => {},
-    existsOnDisk: async () => true,
-    list: async () => Array.from(workspaces.values()),
-    get: async (id: string) => workspaces.get(id) ?? null,
-    update: async (id, updater) => {
-      const existing = workspaces.get(id);
-      if (!existing) return null;
-      const updated = updater(existing);
-      workspaces.set(id, updated);
-      return updated;
-    },
-    upsert: async (record: PersistedWorkspaceRecord) => {
-      workspaces.set(record.workspaceId, record);
-    },
-    archive: async (id: string, archivedAt: string) => {
-      const gate = archiveGate;
-      archiveGate = null;
-      if (gate) {
-        gate.started.resolve();
-        await gate.release.promise;
-      }
-      const existing = workspaces.get(id);
-      if (!existing || existing.archivedAt) return false;
-      workspaces.set(id, { ...existing, archivedAt, updatedAt: archivedAt });
-      return true;
-    },
-    remove: async (id: string) => {
-      workspaces.delete(id);
-    },
-  };
-
-  return { projects, workspaces, projectRegistry, workspaceRegistry, holdNextArchive };
-}
-
-function createTestLogger() {
-  const logger = {
-    child: () => logger,
-    trace: () => undefined,
-    debug: () => undefined,
-    info: () => undefined,
-    warn: () => undefined,
-    error: () => undefined,
-  };
-  return logger as unknown as pino.Logger;
-}
-
 interface CapturedLogRecord {
   message: string;
   payload: unknown;
@@ -210,14 +127,6 @@ function createCheckout(
   };
 }
 
-function deferred(): { promise: Promise<void>; resolve(): void } {
-  let resolve!: () => void;
-  const promise = new Promise<void>((accept) => {
-    resolve = accept;
-  });
-  return { promise, resolve };
-}
-
 class TestCheckouts {
   readonly reads: string[] = [];
   private readonly checkouts = new Map<string, ProjectCheckoutLitePayload>();
@@ -626,53 +535,30 @@ describe("WorkspaceReconciliationService", () => {
   test("keeps the event loop responsive while inspecting 20000 real workspace paths", async () => {
     const rootPath = realpathSync(mkdtempSync(path.join(tmpdir(), "reconcile-many-")));
     tempDirs.push(rootPath);
-    const { projects, workspaces, projectRegistry, workspaceRegistry } = createTestRegistries();
-    projects.set(
-      "p1",
-      createPersistedProjectRecord({
-        projectId: "p1",
-        rootPath,
-        projectKey: canonicalLocalProjectKey(rootPath),
-        kind: "non_git",
-        displayName: "project",
-        createdAt: timestamp,
-        updatedAt: timestamp,
-      }),
-    );
-    for (let i = 0; i < 20000; i++) {
-      const workspaceId = `w${i}`;
-      workspaces.set(
-        workspaceId,
-        createPersistedWorkspaceRecord({
-          workspaceId,
-          projectId: "p1",
-          cwd: path.join(rootPath, workspaceId),
-          kind: "directory",
-          displayName: workspaceId,
-          createdAt: timestamp,
-          updatedAt: timestamp,
-        }),
-      );
-    }
-    const service = new WorkspaceReconciliationService({
-      projectRegistry,
-      workspaceRegistry,
-      logger: createTestLogger(),
+    // Unrelated runner work must not consume the reconciliation budget.
+    const unrelatedWork = new Promise<void>((resolve) => {
+      setImmediate(() => {
+        const cpu = Reflect.get(process, "threadCpuUsage") as typeof process.cpuUsage;
+        const start = cpu();
+        while (cpu(start).user + cpu(start).system < 80_000) {}
+        resolve();
+      });
     });
-    try {
-      const { result, blockingMs, filesystemRequests } = await measureMainThreadWork(
-        "workspace reconciliation main-thread work",
-        () => service.runOnce(),
-      );
-      expect(blockingMs).toBeLessThan(20);
-      expect(filesystemRequests).toBe(20001);
-      expect(result.changesApplied).toHaveLength(20000);
-      expect([...workspaces.values()].every((workspace) => workspace.archivedAt !== null)).toBe(
-        true,
-      );
-    } finally {
-      service.dispose();
-    }
+    const { stdout, stderr } = await promisify(execFile)(process.execPath, [probePath, rootPath], {
+      timeout: 25_000,
+    });
+    process.stderr.write(stderr);
+    const { changesApplied, blockingMs, filesystemRequests, allArchived } = JSON.parse(stdout) as {
+      changesApplied: number;
+      blockingMs: number;
+      filesystemRequests: number;
+      allArchived: boolean;
+    };
+    await unrelatedWork;
+    expect(blockingMs).toBeLessThan(20);
+    expect(filesystemRequests).toBe(20001);
+    expect(changesApplied).toBe(20000);
+    expect(allArchived).toBe(true);
   }, 30000);
 
   test("preserves workspace archival that lands during boot reconciliation", async () => {