fix(server): keep workspace reconciliation responsive on slow paths

Bound asynchronous directory inspection without releasing timed-out filesystem slots. Preserve unavailable records and make workspace archival atomic and idempotent so concurrent user archival cannot be repeated by a scan.
2026-10-27 03:32:55 +00:00 · 2026-10-26 15:23:07 +01:00
parent e3bbed3c73
commit 776248b2c4
6 changed files with 597 additions and 114 deletions
@@ -358,7 +358,7 @@ function createNoopWorkspaceRegistry(): WorkspaceRegistry {
    get: async () => null,
    update: async () => null,
    upsert: async () => {},
-    archive: async () => {},
+    archive: async () => false,
    remove: async () => {},
  };
 }
@@ -495,6 +495,10 @@ describe("observed workspace placement", () => {

    expect((await observed.placement("workspace-one"))?.archivedAt).toEqual(expect.any(String));
    expect(observed.workspaceBatches).toEqual([["workspace-one"]]);
+    const archived = await observed.placement("workspace-one");
+    await observed.advanceBy(RESCAN_INTERVAL_MS);
+    expect(await observed.placement("workspace-one")).toEqual(archived);
+    expect(observed.workspaceBatches).toEqual([["workspace-one"]]);
    observed.dispose();
  });

@@ -31,6 +31,14 @@ 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>();

@@ -55,6 +63,13 @@ function createTestRegistries() {
      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);
    },
@@ -85,17 +100,23 @@ function createTestRegistries() {
      workspaces.set(record.workspaceId, record);
    },
    archive: async (id: string, archivedAt: string) => {
-      const existing = workspaces.get(id);
-      if (existing) {
-        workspaces.set(id, { ...existing, archivedAt, updatedAt: archivedAt });
+      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 };
+  return { projects, workspaces, projectRegistry, workspaceRegistry, holdNextArchive };
 }

 function createTestLogger() {
@@ -117,6 +138,7 @@ interface CapturedLogRecord {

 function createCapturingLogger() {
  const infoRecords: CapturedLogRecord[] = [];
+  const warnRecords: CapturedLogRecord[] = [];
  const logger = {
    child: () => logger,
    trace: () => undefined,
@@ -124,10 +146,12 @@ function createCapturingLogger() {
    info: (payload: unknown, message?: string) => {
      infoRecords.push({ payload, message: message ?? "" });
    },
-    warn: () => undefined,
+    warn: (payload: unknown, message?: string) => {
+      warnRecords.push({ payload, message: message ?? "" });
+    },
    error: () => undefined,
  };
-  return { logger: logger as unknown as pino.Logger, infoRecords };
+  return { logger: logger as unknown as pino.Logger, infoRecords, warnRecords };
 }

 function createWorkspaceGitServiceStub(
@@ -241,6 +265,354 @@ describe("WorkspaceReconciliationService", () => {
    tempDirs.length = 0;
  });

+  test("does not emit a duplicate archive when user archival commits before reconciliation's archive", async () => {
+    const rootPath = realpathSync(mkdtempSync(path.join(tmpdir(), "reconcile-archive-commit-")));
+    tempDirs.push(rootPath);
+    const { projects, workspaces, projectRegistry, workspaceRegistry, holdNextArchive } =
+      createTestRegistries();
+    projects.set(
+      "p1",
+      createPersistedProjectRecord({
+        projectId: "p1",
+        rootPath,
+        projectKey: canonicalLocalProjectKey(rootPath),
+        kind: "non_git",
+        displayName: "project",
+        createdAt: timestamp,
+        updatedAt: timestamp,
+      }),
+    );
+    workspaces.set(
+      "w1",
+      createPersistedWorkspaceRecord({
+        workspaceId: "w1",
+        projectId: "p1",
+        cwd: path.join(rootPath, "missing"),
+        kind: "directory",
+        displayName: "workspace",
+        createdAt: timestamp,
+        updatedAt: timestamp,
+      }),
+    );
+    const callbacks: string[] = [];
+    const service = new WorkspaceReconciliationService({
+      projectRegistry,
+      workspaceRegistry,
+      logger: createTestLogger(),
+      onWorkspaceArchived: (id) => {
+        callbacks.push(id);
+      },
+    });
+    const gate = holdNextArchive();
+    try {
+      const scan = service.runOnce();
+      await gate.started.promise;
+      expect(await workspaceRegistry.archive("w1", timestamp)).toBe(true);
+      gate.release.resolve();
+      expect((await scan).changesApplied).toEqual([]);
+      expect(callbacks).toEqual([]);
+      expect(workspaces.get("w1")?.archivedAt).toBe(timestamp);
+    } finally {
+      gate.release.resolve();
+      service.dispose();
+    }
+  });
+
+  test("preserves paths while two inspections saturate capacity, logs once, and recovers after they drain", async () => {
+    const rootPath = realpathSync(mkdtempSync(path.join(tmpdir(), "reconcile-saturated-")));
+    tempDirs.push(rootPath);
+    const { projects, workspaces, projectRegistry, workspaceRegistry } = createTestRegistries();
+    const { logger, warnRecords, infoRecords } = createCapturingLogger();
+    projects.set(
+      "p1",
+      createPersistedProjectRecord({
+        projectId: "p1",
+        rootPath,
+        projectKey: canonicalLocalProjectKey(rootPath),
+        kind: "non_git",
+        displayName: "project",
+        createdAt: timestamp,
+        updatedAt: timestamp,
+      }),
+    );
+    const ids = ["hung1", "hung2", "healthy"];
+    for (const workspaceId of ids) {
+      workspaces.set(
+        workspaceId,
+        createPersistedWorkspaceRecord({
+          workspaceId,
+          projectId: "p1",
+          cwd: path.join(rootPath, workspaceId),
+          kind: "directory",
+          displayName: workspaceId,
+          createdAt: timestamp,
+          updatedAt: timestamp,
+        }),
+      );
+    }
+    const release = deferred();
+    const calls: string[] = [];
+    let recovered = false;
+    let ticks = 0;
+    const probe = setInterval(() => ticks++, 1);
+    const service = new WorkspaceReconciliationService({
+      projectRegistry,
+      workspaceRegistry,
+      logger,
+      inspectionTimeoutMs: 30,
+      inspectDirectory: async (targetPath) => {
+        calls.push(targetPath);
+        if (!recovered && path.basename(targetPath).startsWith("hung")) await release.promise;
+        return "directory";
+      },
+    });
+    try {
+      const result = await service.runOnce();
+      expect(ticks).toBeGreaterThan(0);
+      expect(result.durationMs).toBeLessThan(1000);
+      expect(result.changesApplied).toEqual([]);
+      expect(calls).toEqual(ids.slice(0, 2).map((id) => path.join(rootPath, id)));
+      expect([...workspaces.values()].map((workspace) => workspace.archivedAt)).toEqual([
+        null,
+        null,
+        null,
+      ]);
+      expect(warnRecords.filter((record) => record.message.includes("saturated"))).toEqual([
+        {
+          message: "Workspace directory inspection saturated; preserving uninspected workspaces",
+          payload: { paths: ids.slice(0, 2).map((id) => path.join(rootPath, id)) },
+        },
+      ]);
+      recovered = true;
+      release.resolve();
+      await new Promise<void>((resolve) => setImmediate(resolve));
+      expect(infoRecords).toEqual([
+        {
+          message:
+            "Timed-out workspace directory inspections drained; reconciliation capacity restored",
+          payload: {},
+        },
+      ]);
+      calls.length = 0;
+      await service.runOnce();
+      expect(calls).toEqual([...ids.map((id) => path.join(rootPath, id)), rootPath]);
+      expect(warnRecords.filter((record) => record.message.includes("saturated"))).toHaveLength(1);
+    } finally {
+      release.resolve();
+      clearInterval(probe);
+      service.dispose();
+    }
+  });
+
+  test("times out a hung inspection without archiving it or blocking healthy paths, and retains its IO slot", async () => {
+    const rootPath = realpathSync(mkdtempSync(path.join(tmpdir(), "reconcile-hung-")));
+    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 (const workspaceId of ["hung", "healthy", "healthy2", "healthy3"]) {
+      workspaces.set(
+        workspaceId,
+        createPersistedWorkspaceRecord({
+          workspaceId,
+          projectId: "p1",
+          cwd: path.join(rootPath, workspaceId),
+          kind: "directory",
+          displayName: workspaceId,
+          createdAt: timestamp,
+          updatedAt: timestamp,
+        }),
+      );
+    }
+    const calls: string[] = [];
+    let outstanding = 0;
+    let maximumOutstanding = 0;
+    const healthyInspected = deferred();
+    const service = new WorkspaceReconciliationService({
+      projectRegistry,
+      workspaceRegistry,
+      logger: createTestLogger(),
+      inspectionTimeoutMs: 30,
+      inspectDirectory: async (targetPath) => {
+        calls.push(targetPath);
+        maximumOutstanding = Math.max(maximumOutstanding, ++outstanding);
+        if (path.basename(targetPath) === "hung") return new Promise(() => {});
+        healthyInspected.resolve();
+        await new Promise<void>((resolve) => setImmediate(resolve));
+        outstanding--;
+        return "directory";
+      },
+    });
+    try {
+      const scan = service.runOnce();
+      await Promise.race([
+        healthyInspected.promise,
+        scan.then(() => {
+          throw new Error("Reconciliation did not use its directory inspection capability");
+        }),
+      ]);
+      expect(outstanding).toBeLessThanOrEqual(2);
+      expect(workspaces.get("hung")?.archivedAt).toBeNull();
+      const result = await scan;
+      expect(result.changesApplied).toEqual([]);
+      await service.runOnce();
+      expect(calls.filter((cwd) => path.basename(cwd) === "hung")).toHaveLength(1);
+      expect(calls.filter((cwd) => path.basename(cwd) === "healthy3")).toHaveLength(2);
+      expect(maximumOutstanding).toBe(2);
+      expect([...workspaces.values()].map((workspace) => workspace.archivedAt)).toEqual([
+        null,
+        null,
+        null,
+        null,
+      ]);
+    } finally {
+      service.dispose();
+    }
+  });
+
+  test.each(["missing", "directory"] as const)(
+    "does not process or archive a workspace archived during %s inspection",
+    async (state) => {
+      const rootPath = realpathSync(
+        mkdtempSync(path.join(tmpdir(), "reconcile-inspection-archive-")),
+      );
+      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,
+        }),
+      );
+      const cwd = path.join(rootPath, "missing");
+      workspaces.set(
+        "w1",
+        createPersistedWorkspaceRecord({
+          workspaceId: "w1",
+          projectId: "p1",
+          cwd,
+          kind: "directory",
+          displayName: "workspace",
+          createdAt: timestamp,
+          updatedAt: timestamp,
+        }),
+      );
+      const started = deferred();
+      const release = deferred();
+      const archivedCallbacks: string[] = [];
+      const checkouts = new TestCheckouts();
+      const service = new WorkspaceReconciliationService({
+        projectRegistry,
+        workspaceRegistry,
+        logger: createTestLogger(),
+        workspaceGitService: checkouts,
+        onWorkspaceArchived: (id) => {
+          archivedCallbacks.push(id);
+        },
+        inspectDirectory: async (targetPath) => {
+          if (targetPath === cwd) {
+            started.resolve();
+            await release.promise;
+            return state;
+          }
+          return "directory";
+        },
+      });
+      try {
+        const scan = service.runOnce();
+        await Promise.race([
+          started.promise,
+          scan.then(() => {
+            throw new Error("Reconciliation did not use its directory inspection capability");
+          }),
+        ]);
+        await workspaceRegistry.archive("w1", "2026-01-01T00:00:00.000Z");
+        const archived = workspaces.get("w1");
+        release.resolve();
+        const result = await scan;
+        expect(workspaces.get("w1")).toEqual(archived);
+        expect(archivedCallbacks).toEqual([]);
+        expect(checkouts.reads).toEqual([rootPath]);
+        expect(result.changesApplied).toEqual([]);
+      } finally {
+        release.resolve();
+        service.dispose();
+      }
+    },
+  );
+
+  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(),
+    });
+    let last = performance.now();
+    let longestGap = 0;
+    const probe = setInterval(() => {
+      const now = performance.now();
+      longestGap = Math.max(longestGap, now - last);
+      last = now;
+    }, 1);
+    let result;
+    try {
+      result = await service.runOnce();
+      await new Promise<void>((resolve) => setTimeout(resolve, 1));
+    } finally {
+      clearInterval(probe);
+      service.dispose();
+    }
+    expect(longestGap).toBeLessThan(20);
+    expect(result.changesApplied).toHaveLength(20000);
+    expect([...workspaces.values()].every((workspace) => workspace.archivedAt !== null)).toBe(true);
+  }, 30000);
+
  test("preserves workspace archival that lands during boot reconciliation", async () => {
    const workspaceRoot = realpathSync(mkdtempSync(path.join(tmpdir(), "reconcile-archive-race-")));
    tempDirs.push(workspaceRoot);
@@ -1,4 +1,6 @@
-import { statSync, watch as watchPath } from "node:fs";
+import { watch as watchPath } from "node:fs";
+import { stat } from "node:fs/promises";
+import { setImmediate as yieldToLoop } from "node:timers/promises";
 import type { ProjectCheckoutLitePayload } from "@getpaseo/protocol/messages";
 import type pino from "pino";
 import type {
@@ -19,6 +21,8 @@ import { deriveProjectKey } from "./project-key.js";

 const DEFAULT_RESCAN_INTERVAL_MS = 5 * 60_000;
 const DEFAULT_DEBOUNCE_MS = 100;
+const INSPECTION_CONCURRENCY = 2;
+const DEFAULT_INSPECTION_TIMEOUT_MS = 5000;

 export type ProjectUpdate =
  | { kind: "upsert"; project: PersistedProjectRecord }
@@ -95,6 +99,8 @@ export interface WorkspaceReconciliationServiceOptions {
  clock?: ReconciliationClock;
  rescanIntervalMs?: number;
  debounceMs?: number;
+  inspectDirectory?: (targetPath: string) => Promise<DirectoryState>;
+  inspectionTimeoutMs?: number;
 }

 interface ProjectReconciliationInput {
@@ -110,9 +116,16 @@ interface CachedCheckoutRead {
  checkout: Promise<ProjectCheckoutLitePayload>;
 }

-type DirectoryState = "directory" | "missing" | "unreadable";
+export type DirectoryState = "directory" | "missing" | "unreadable";

 export class WorkspaceReconciliationService {
+  // Timed-out fs operations retain their slot until the OS returns. Never use
+  // the deadline to admit more work into libuv's shared threadpool.
+  private readonly timedOutPaths = new Set<string>();
+  private inspectionSaturated = false;
+  private readonly pendingInspections = new Map<string, Promise<DirectoryState>>();
+  private readonly directoryInspection: (targetPath: string) => Promise<DirectoryState>;
+  private readonly inspectionTimeoutMs: number;
  private readonly serverId: string | undefined;
  private readonly projectRegistry: ProjectRegistry;
  private readonly workspaceRegistry: WorkspaceRegistry;
@@ -136,6 +149,10 @@ export class WorkspaceReconciliationService {
  private reconcileQueuedMode: "metadata" | "full" | null = null;

  constructor(options: WorkspaceReconciliationServiceOptions) {
+    this.directoryInspection =
+      options.inspectDirectory ??
+      (async (targetPath) => ((await stat(targetPath)).isDirectory() ? "directory" : "missing"));
+    this.inspectionTimeoutMs = options.inspectionTimeoutMs ?? DEFAULT_INSPECTION_TIMEOUT_MS;
    this.serverId = options.serverId;
    this.projectRegistry = options.projectRegistry;
    this.workspaceRegistry = options.workspaceRegistry;
@@ -197,16 +214,23 @@ export class WorkspaceReconciliationService {
      this.workspaceRegistry.list(),
    ]);
    const workspacesByProject = new Map<string, PersistedWorkspaceRecord[]>();
-    for (const workspace of workspaces) {
-      if (workspace.archivedAt || this.inspectDirectory(workspace.cwd) !== "directory") continue;
+    const workspaceStates = await mapBounded(
+      workspaces.filter((w) => !w.archivedAt),
+      async (workspace) => ({ workspace, state: await this.inspectDirectory(workspace.cwd) }),
+    );
+    for (const { workspace, state } of workspaceStates) {
+      if (state !== "directory") continue;
+      if ((await this.workspaceRegistry.get(workspace.workspaceId))?.archivedAt !== null) continue;
      const siblings = workspacesByProject.get(workspace.projectId) ?? [];
      siblings.push(workspace);
      workspacesByProject.set(workspace.projectId, siblings);
    }
+    const projectStates = await mapBounded(
+      projects.filter((p) => !p.archivedAt),
+      async (project) => ({ project, state: await this.inspectDirectory(project.rootPath) }),
+    );
    await this.reconcileGitMetadataForProjects(
-      projects.filter(
-        (project) => !project.archivedAt && this.inspectDirectory(project.rootPath) === "directory",
-      ),
+      projectStates.filter(({ state }) => state === "directory").map(({ project }) => project),
      workspacesByProject,
      changes,
    );
@@ -223,22 +247,26 @@ export class WorkspaceReconciliationService {

    const activeProjects = allProjects.filter((p) => !p.archivedAt);
    const activeWorkspaces = allWorkspaces.filter((w) => !w.archivedAt);
-    const workspaceDirectoryStates = activeWorkspaces.map((workspace) => ({
+    const workspaceDirectoryStates = await mapBounded(activeWorkspaces, async (workspace) => ({
      workspace,
-      state: this.inspectDirectory(workspace.cwd),
+      state: await this.inspectDirectory(workspace.cwd),
+    }));
+    // Read project roots after workspaces: losing a volume can withhold an
+    // archive, never turn a workspace's absence into proof of deletion.
+    const projectStates = await mapBounded(activeProjects, async (project) => ({
+      project,
+      state: await this.inspectDirectory(project.rootPath),
    }));
-    // Project roots are read after the workspace directories, so a volume that
-    // goes away mid-pass leaves its project unreachable rather than its workspaces
-    // alone. The skew can only withhold an archive, never produce one.
    const reachableProjectIds = new Set(
-      activeProjects
-        .filter((project) => this.inspectDirectory(project.rootPath) === "directory")
-        .map((project) => project.projectId),
+      projectStates
+        .filter(({ state }) => state === "directory")
+        .map(({ project }) => project.projectId),
    );

    const workspacesByProject = new Map<string, PersistedWorkspaceRecord[]>();
    for (const { workspace, state } of workspaceDirectoryStates) {
      if (state !== "directory") continue;
+      if ((await this.workspaceRegistry.get(workspace.workspaceId))?.archivedAt !== null) continue;
      const list = workspacesByProject.get(workspace.projectId) ?? [];
      list.push(workspace);
      workspacesByProject.set(workspace.projectId, list);
@@ -255,26 +283,29 @@ export class WorkspaceReconciliationService {
          state === "missing" && reachableProjectIds.has(workspace.projectId),
      )
      .map(({ workspace }) => workspace);
-    await Promise.all(
-      missingWorkspaces.map(async (workspace) => {
-        const timestamp = new Date().toISOString();
-        await this.workspaceRegistry.archive(workspace.workspaceId, timestamp);
-        await this.onWorkspaceArchived?.(workspace.workspaceId);
-        changes.push({
-          kind: "workspace_archived",
-          workspaceId: workspace.workspaceId,
-          directory: workspace.cwd,
-          reason: "directory_missing",
-        });
-
-        // Update the in-memory list for the project orphan check below
-        const siblings = workspacesByProject.get(workspace.projectId);
-        if (siblings) {
-          const updated = siblings.filter((w) => w.workspaceId !== workspace.workspaceId);
-          workspacesByProject.set(workspace.projectId, updated);
-        }
-      }),
-    );
+    await mapBounded(missingWorkspaces, async (workspace) => {
+      const current = await this.workspaceRegistry.get(workspace.workspaceId);
+      const project = await this.projectRegistry.get(workspace.projectId);
+      if (
+        !current ||
+        current.archivedAt ||
+        current.cwd !== workspace.cwd ||
+        current.projectId !== workspace.projectId ||
+        !project ||
+        project.archivedAt ||
+        project.rootPath !== activeProjects.find((p) => p.projectId === project.projectId)?.rootPath
+      )
+        return;
+      const timestamp = new Date().toISOString();
+      if (!(await this.workspaceRegistry.archive(workspace.workspaceId, timestamp))) return;
+      await this.onWorkspaceArchived?.(workspace.workspaceId);
+      changes.push({
+        kind: "workspace_archived",
+        workspaceId: workspace.workspaceId,
+        directory: workspace.cwd,
+        reason: "directory_missing",
+      });
+    });

    // 2. Reconcile mutable git metadata without changing identity or membership.
    //    Projects persist until explicitly removed, even when they currently have
@@ -351,13 +382,17 @@ export class WorkspaceReconciliationService {
  }

  private async reconcileProject(input: ProjectReconciliationInput): Promise<void> {
-    const { project, siblings, currentGit, readCheckout, changes } = input;
+    const { siblings, currentGit, readCheckout, changes } = input;
    const workspaceCheckouts = await Promise.all(
-      siblings.map(async (workspace) => ({
-        workspace,
-        checkout: await readCheckout(workspace.cwd),
-      })),
+      siblings.map(async (workspace) => {
+        const current = await this.workspaceRegistry.get(workspace.workspaceId);
+        if (!current || current.cwd !== workspace.cwd || current.projectId !== workspace.projectId)
+          return null;
+        return { workspace: current, checkout: await readCheckout(current.cwd) };
+      }),
    );
+    const project = await this.projectRegistry.get(input.project.projectId);
+    if (!project || project.archivedAt || project.rootPath !== input.project.rootPath) return;
    const projectUpdates: Partial<Pick<PersistedProjectRecord, "kind" | "projectKey">> = {};
    const mappedKind = deriveProjectKind(currentGit);
    const projectKey = deriveProjectKey({
@@ -377,40 +412,47 @@ export class WorkspaceReconciliationService {

    if (Object.keys(projectUpdates).length > 0) {
      const timestamp = new Date().toISOString();
-      await this.projectRegistry.upsert({
-        ...project,
-        ...projectUpdates,
-        updatedAt: timestamp,
-      });
-      changes.push({
-        kind: "project_updated",
-        projectId: project.projectId,
-        directory: project.rootPath,
-        fields: projectUpdates,
+      let applied = false;
+      await this.projectRegistry.update(project.projectId, (current) => {
+        if (current.archivedAt || current.rootPath !== project.rootPath) return current;
+        applied = true;
+        return { ...current, ...projectUpdates, updatedAt: timestamp };
      });
+      if (applied)
+        changes.push({
+          kind: "project_updated",
+          projectId: project.projectId,
+          directory: project.rootPath,
+          fields: projectUpdates,
+        });
    }

    await Promise.all(
-      workspaceCheckouts.map(async ({ workspace, checkout: wsGit }) => {
+      workspaceCheckouts.map(async (entry) => {
+        if (!entry) return;
+        const { workspace, checkout: wsGit } = entry;
        const timestamp = new Date().toISOString();
-        const update = reconcileWorkspacePlacement({
-          workspace,
-          checkout: wsGit,
-          updatedAt: timestamp,
+        if (!reconcileWorkspacePlacement({ workspace, checkout: wsGit, updatedAt: timestamp }))
+          return;
+        let fields: Partial<MutableWorkspacePlacement> | null = null;
+        await this.workspaceRegistry.update(workspace.workspaceId, (current) => {
+          if (current.cwd !== workspace.cwd || current.projectId !== workspace.projectId)
+            return current;
+          const update = reconcileWorkspacePlacement({
+            workspace: current,
+            checkout: wsGit,
+            updatedAt: timestamp,
+          });
+          if (!update) return current;
+          fields = update.fields;
+          return { ...current, ...fields, updatedAt: timestamp };
        });
-        if (!update) return;
-
-        const updated = await this.workspaceRegistry.update(workspace.workspaceId, (current) => ({
-          ...current,
-          ...update.fields,
-          updatedAt: timestamp,
-        }));
-        if (!updated) return;
+        if (!fields) return;
        changes.push({
          kind: "workspace_updated",
          workspaceId: workspace.workspaceId,
          directory: workspace.cwd,
-          fields: update.fields,
+          fields,
        });
      }),
    );
@@ -537,16 +579,58 @@ export class WorkspaceReconciliationService {
    return this.workspaceGitService.getCheckout(cwd);
  }

-  private inspectDirectory(targetPath: string): DirectoryState {
+  private async inspectDirectory(targetPath: string): Promise<DirectoryState> {
+    let pending = this.pendingInspections.get(targetPath);
+    if (!pending) {
+      if (this.pendingInspections.size >= INSPECTION_CONCURRENCY || this.disposed)
+        return "unreadable";
+      pending = Promise.resolve()
+        .then(() => this.directoryInspection(targetPath))
+        .catch((error) => {
+          if (isMissingPathError(error)) return "missing" as const;
+          this.logger.warn(
+            { err: error, targetPath },
+            "Skipped workspace reconciliation after directory inspection failed",
+          );
+          return "unreadable" as const;
+        })
+        .finally(() => {
+          this.pendingInspections.delete(targetPath);
+          this.timedOutPaths.delete(targetPath);
+          if (this.inspectionSaturated && this.timedOutPaths.size === 0) {
+            this.inspectionSaturated = false;
+            this.logger.info(
+              {},
+              "Timed-out workspace directory inspections drained; reconciliation capacity restored",
+            );
+          }
+        });
+      this.pendingInspections.set(targetPath, pending);
+    }
+    let timer: ReturnType<typeof setTimeout> | undefined;
    try {
-      return statSync(targetPath).isDirectory() ? "directory" : "missing";
-    } catch (error) {
-      if (isMissingPathError(error)) return "missing";
-      this.logger.warn(
-        { err: error, targetPath },
-        "Skipped workspace reconciliation after directory inspection failed",
-      );
-      return "unreadable";
+      return await Promise.race([
+        pending,
+        new Promise<DirectoryState>((resolve) => {
+          timer = setTimeout(() => {
+            this.timedOutPaths.add(targetPath);
+            if (!this.inspectionSaturated && this.timedOutPaths.size === INSPECTION_CONCURRENCY) {
+              this.inspectionSaturated = true;
+              this.logger.warn(
+                { paths: [...this.timedOutPaths] },
+                "Workspace directory inspection saturated; preserving uninspected workspaces",
+              );
+            }
+            this.logger.warn(
+              { targetPath },
+              "Skipped workspace reconciliation after directory inspection timed out",
+            );
+            resolve("unreadable");
+          }, this.inspectionTimeoutMs);
+        }),
+      ]);
+    } finally {
+      clearTimeout(timer);
    }
  }
 }
@@ -555,3 +639,21 @@ function isMissingPathError(error: unknown): boolean {
  if (typeof error !== "object" || error === null || !("code" in error)) return false;
  return error.code === "ENOENT" || error.code === "ENOTDIR";
 }
+
+// Two workers bound admission; yielding also bounds passes whose adapters settle
+// immediately (including unavailable paths after both IO slots time out).
+async function mapBounded<T, R>(items: T[], visit: (item: T) => Promise<R>): Promise<R[]> {
+  const results: R[] = [];
+  let next = 0;
+  await Promise.all(
+    Array.from({ length: Math.min(INSPECTION_CONCURRENCY, items.length) }, async () => {
+      let count = 0;
+      while (next < items.length) {
+        const index = next++;
+        results[index] = await visit(items[index]!);
+        if (++count % 32 === 0) await yieldToLoop();
+      }
+    }),
+  );
+  return results;
+}
@@ -474,7 +474,7 @@ describe("workspace registries", () => {
    expect(await workspaceRegistry.list()).toEqual([]);
  });

-  test("refreshes workspace archive timestamps when an archive is repeated", async () => {
+  test("preserves the first workspace archive and reports whether a queued archive applied", async () => {
    await workspaceRegistry.initialize();
    await workspaceRegistry.upsert(
      createPersistedWorkspaceRecord({
@@ -488,12 +488,24 @@ describe("workspace registries", () => {
      }),
    );

-    await workspaceRegistry.archive("workspace-one", "2026-03-02T00:00:00.000Z");
-    await workspaceRegistry.archive("workspace-one", "2026-03-03T00:00:00.000Z");
-
-    expect(await workspaceRegistry.get("workspace-one")).toMatchObject({
-      archivedAt: "2026-03-03T00:00:00.000Z",
-      updatedAt: "2026-03-03T00:00:00.000Z",
+    const events: WorkspaceMutation[] = [];
+    workspaceRegistry.subscribeToMutations((event) => {
+      events.push(event);
+    });
+    expect(
+      await Promise.all([
+        workspaceRegistry.archive("workspace-one", "2026-03-02T00:00:00.000Z"),
+        workspaceRegistry.archive("workspace-one", "2026-03-03T00:00:00.000Z"),
+      ]),
+    ).toEqual([true, false]);
+    expect(events.map((event) => event.kind)).toEqual(["archive"]);
+    const reloaded = new FileBackedWorkspaceRegistry(
+      path.join(tmpDir, "projects", "workspaces.json"),
+      logger,
+    );
+    expect(await reloaded.get("workspace-one")).toMatchObject({
+      archivedAt: "2026-03-02T00:00:00.000Z",
+      updatedAt: "2026-03-02T00:00:00.000Z",
    });
  });

@@ -167,7 +167,7 @@ export interface WorkspaceRegistry {
    workspaceId: string,
    archivedAt: string,
    context?: WorkspaceArchiveContext,
-  ): Promise<void>;
+  ): Promise<boolean>;
  remove(workspaceId: string): Promise<void>;
  /** Central lifecycle seam for daemon-global workspace observers. */
  subscribeToMutations?(
@@ -247,20 +247,6 @@ class FileBackedRegistry<TRecord extends RegistryRecord> {
    });
  }

-  async archive(id: string, archivedAt: string): Promise<void> {
-    await this.archiveIfPresent(id, archivedAt);
-  }
-
-  protected async archiveIfPresent(id: string, archivedAt: string): Promise<TRecord | null> {
-    return this.mutateCache((records) => {
-      const existing = records.get(id);
-      if (!existing) return null;
-      const next = this.schema.parse({ ...existing, updatedAt: archivedAt, archivedAt });
-      records.set(id, next);
-      return next;
-    });
-  }
-
  protected async archiveIfActive(id: string, archivedAt: string): Promise<TRecord | null> {
    return this.mutateCache((records) => {
      const existing = records.get(id);
@@ -485,7 +471,7 @@ export class FileBackedProjectRegistry
    return project;
  }

-  override async archive(projectId: string, archivedAt: string): Promise<void> {
+  async archive(projectId: string, archivedAt: string): Promise<void> {
    const project = await this.archiveIfActive(projectId, archivedAt);
    if (!project) return;
    await this.notifyMutation({ kind: "archive", projectId, project });
@@ -565,21 +551,28 @@ export class FileBackedWorkspaceRegistry
    });
  }

-  override async archive(
+  async archive(
    workspaceId: string,
    archivedAt: string,
    context?: WorkspaceArchiveContext,
-  ): Promise<void> {
-    const workspace = await super.update(workspaceId, (existing) => ({
-      ...existing,
-      updatedAt: archivedAt,
-      archivedAt,
-      ...(context?.autoArchivedChangeRequestUrl
-        ? { autoArchivedChangeRequestUrl: context.autoArchivedChangeRequestUrl }
-        : {}),
-    }));
-    if (!workspace) return;
+  ): Promise<boolean> {
+    const [workspace] = await this.mutateMany((records) => {
+      const existing = records.get(workspaceId);
+      if (!existing || existing.archivedAt) return [];
+      return [
+        {
+          ...existing,
+          updatedAt: archivedAt,
+          archivedAt,
+          ...(context?.autoArchivedChangeRequestUrl
+            ? { autoArchivedChangeRequestUrl: context.autoArchivedChangeRequestUrl }
+            : {}),
+        },
+      ];
+    });
+    if (!workspace) return false;
    await this.notifyMutation({ kind: "archive", workspaceId, workspace });
+    return true;
  }

  override async remove(workspaceId: string): Promise<void> {