Recover daemon workers stalled by workspace Git pressure (#3033)

* fix(server): bound workspace git pipelines

* fix(server): bound workspace git watcher setup

* fix(server): recover unresponsive daemon workers

* fix(server): tolerate transient daemon heartbeat delays

A five-second watchdog threshold can replace a healthy worker during a short event-loop delay. Keep recovery bounded while allowing a 15-second heartbeat gap, and preserve process-level coverage for workers that ignore graceful termination.

* test(server): normalize workspace fixture paths

Workspace Git entry points resolve paths before invoking dependencies. Keep admission-control fixtures keyed by the same canonical paths so their assertions and deferred gates work on Windows.

---------

Co-authored-by: Timi Ajiboye <trypio00@gmail.com>
Co-authored-by: Mohamed Boudra <boudra.moha@gmail.com>

packages/server/scripts/supervisor.logging.test.ts

@@ -13,9 +13,11 @@ const supervisorPath = fileURLToPath(new URL("./supervisor.ts", import.meta.url)
 async function runSupervisorFixture(options: {
   workerSource: string;
   restartOnCrash?: boolean;
+  timeoutMs?: number;
 }): Promise<{
   code: number | null;
   signal: NodeJS.Signals | null;
+  elapsedMs: number;
   log: string;
   stdout: string;
   stderr: string;
@@ -47,6 +49,7 @@ async function runSupervisorFixture(options: {
     `,
   );
 
+  const startedAt = Date.now();
   const child = spawn(process.execPath, ["--import", "tsx", runnerPath], {
     cwd: repoRoot,
     env: { ...process.env },
@@ -71,7 +74,7 @@ async function runSupervisorFixture(options: {
     const timeout = setTimeout(() => {
       child.kill("SIGKILL");
       reject(new Error("supervisor fixture timed out"));
-    }, 10000);
+    }, options.timeoutMs ?? 10_000);
 
     child.on("error", (error) => {
       clearTimeout(timeout);
@@ -84,7 +87,7 @@ async function runSupervisorFixture(options: {
   });
 
   const log = await readFile(logPath, "utf8");
-  return { code, signal, log, stdout, stderr };
+  return { code, signal, elapsedMs: Date.now() - startedAt, log, stdout, stderr };
 }
 
 describe("supervisor durable logging", () => {
@@ -186,6 +189,152 @@ describe("supervisor durable logging", () => {
     expect(result.log).toContain('"workerPid":');
   });
 
+  test("keeps a worker alive while heartbeats continue", async () => {
+    const result = await runSupervisorFixture({
+      timeoutMs: 10_000,
+      workerSource: `
+        setInterval(() => {
+          process.send?.({ type: "paseo:worker-heartbeat" });
+        }, 250);
+        setTimeout(() => {
+          process.send?.({ type: "paseo:shutdown", reason: "healthy_heartbeat_test_complete" });
+        }, 6000);
+      `,
+    });
+
+    expect(result.code).toBe(0);
+    expect(result.signal).toBeNull();
+    expect(result.log).toContain('"reason":"healthy_heartbeat_test_complete"');
+    expect(result.log).not.toContain('"msg":"Worker heartbeat timed out; restarting worker"');
+  }, 10_000);
+
+  test("tolerates a transient seven-second heartbeat pause", async () => {
+    const result = await runSupervisorFixture({
+      timeoutMs: 10_000,
+      workerSource: `
+        import { existsSync, writeFileSync } from "node:fs";
+
+        const marker = process.argv[1] + ".started";
+        if (!existsSync(marker)) {
+          writeFileSync(marker, "started");
+          let heartbeatCount = 0;
+          const heartbeat = setInterval(() => {
+            process.send?.({ type: "paseo:worker-heartbeat" });
+            heartbeatCount += 1;
+            if (heartbeatCount === 3) clearInterval(heartbeat);
+          }, 100);
+          setTimeout(() => {
+            process.send?.({ type: "paseo:shutdown", reason: "heartbeat_pause_tolerated" });
+          }, 7_000);
+          setInterval(() => {}, 1_000);
+        } else {
+          process.send?.({ type: "paseo:shutdown", reason: "unexpected_heartbeat_restart" });
+          setInterval(() => {}, 1_000);
+        }
+      `,
+    });
+
+    expect(result.code).toBe(0);
+    expect(result.signal).toBeNull();
+    expect(result.log).toContain('"reason":"heartbeat_pause_tolerated"');
+    expect(result.log).not.toContain('"reason":"unexpected_heartbeat_restart"');
+    expect(result.log).not.toContain('"msg":"Worker heartbeat timed out; restarting worker"');
+  }, 10_000);
+
+  test.skipIf(isPlatform("win32"))(
+    "forces shutdown when a worker ignores SIGTERM",
+    async () => {
+      const result = await runSupervisorFixture({
+        timeoutMs: 15_000,
+        workerSource: `
+          process.on("SIGTERM", () => {});
+          process.send?.({ type: "paseo:shutdown", reason: "stalled_worker_shutdown" });
+          setInterval(() => {}, 1_000);
+        `,
+      });
+
+      expect(result.code).toBe(0);
+      expect(result.signal).toBeNull();
+      expect(result.log).toContain('"reason":"stalled_worker_shutdown"');
+      expect(result.log).toContain('"msg":"Worker did not exit after SIGTERM; forcing SIGKILL"');
+      expect(result.log).toContain('"signal":"SIGKILL"');
+    },
+    20_000,
+  );
+
+  // POSIX-only: the watchdog uses SIGKILL after its graceful shutdown window.
+  test.skipIf(isPlatform("win32"))(
+    "restarts a worker that stops heartbeating",
+    async () => {
+      const result = await runSupervisorFixture({
+        timeoutMs: 35_000,
+        workerSource: `
+          import { existsSync, writeFileSync } from "node:fs";
+
+          const marker = process.argv[1] + ".started";
+          if (!existsSync(marker)) {
+            writeFileSync(marker, "started");
+            let heartbeatCount = 0;
+            const heartbeat = setInterval(() => {
+              process.send?.({ type: "paseo:worker-heartbeat" });
+              heartbeatCount += 1;
+              if (heartbeatCount === 3) clearInterval(heartbeat);
+            }, 100);
+            process.on("SIGTERM", () => {});
+            setInterval(() => {}, 1000);
+          } else {
+            process.send?.({ type: "paseo:shutdown", reason: "watchdog_test_complete" });
+            setInterval(() => {}, 1000);
+          }
+        `,
+      });
+
+      expect(result.code).toBe(0);
+      expect(result.signal).toBeNull();
+      expect(result.log).toContain('"msg":"Worker heartbeat timed out; restarting worker"');
+      expect(result.log).toContain('"msg":"Worker did not exit after SIGTERM; forcing SIGKILL"');
+      expect(result.log).toContain('"signal":"SIGKILL"');
+    },
+    40_000,
+  );
+
+  test.skipIf(isPlatform("win32"))(
+    "restarts after worker exit while a descendant retains the worker stdio",
+    async () => {
+      const result = await runSupervisorFixture({
+        timeoutMs: 7_000,
+        workerSource: `
+          import { spawn } from "node:child_process";
+          import { existsSync, writeFileSync } from "node:fs";
+
+          const marker = process.argv[1] + ".started";
+          if (!existsSync(marker)) {
+            writeFileSync(marker, "started");
+            const descendant = spawn(
+              process.execPath,
+              ["-e", "process.on('SIGTERM', () => {}); setTimeout(() => process.exit(0), 4000)"],
+              { detached: true, stdio: ["ignore", "inherit", "inherit"] },
+            );
+            descendant.unref();
+            process.on("SIGTERM", () => process.exit(0));
+            process.send?.({ type: "paseo:restart", reason: "stdio_descendant" });
+            setInterval(() => {}, 1000);
+          } else {
+            process.send?.({ type: "paseo:shutdown", reason: "stdio_restart_complete" });
+            setInterval(() => {}, 1000);
+          }
+        `,
+      });
+
+      expect(result.code).toBe(0);
+      expect(result.signal).toBeNull();
+      expect(result.elapsedMs).toBeLessThan(2_500);
+      expect(result.log).toContain('"reason":"stdio_descendant"');
+      expect(result.log).toContain("Restarting worker");
+    },
+    7_000,
+  );
+
   // POSIX-only: Windows reports the worker self-kill as an exit code, not SIGKILL.
   test.skipIf(isPlatform("win32"))(
     "logs worker signal exits even when the worker cannot log",

packages/server/scripts/supervisor.ts

@@ -2,6 +2,11 @@ import { fork, spawn, type ChildProcess } from "child_process";
 import { mkdirSync } from "node:fs";
 import path from "node:path";
 import { createStream as createRotatingFileStream } from "rotating-file-stream";
+import { signalProcessTree } from "../src/utils/tree-kill.js";
+
+const WORKER_HEARTBEAT_INTERVAL_MS = 1_000;
+const WORKER_HEARTBEAT_TIMEOUT_MS = 15_000;
+const WORKER_TERMINATION_GRACE_MS = 10_000;
 
 interface SupervisorLogFileOptions {
   path: string;
@@ -29,6 +34,10 @@ interface SupervisorHeartbeatMessage {
   type: "paseo:supervisor-heartbeat";
 }
 
+interface WorkerHeartbeatMessage {
+  type: "paseo:worker-heartbeat";
+}
+
 interface SupervisorOptions {
   name: string;
   startupMessage: string;
@@ -84,6 +93,15 @@ function parseLifecycleMessage(msg: unknown): WorkerLifecycleMessage | null {
   return null;
 }
 
+function isWorkerHeartbeatMessage(msg: unknown): msg is WorkerHeartbeatMessage {
+  return (
+    typeof msg === "object" &&
+    msg !== null &&
+    "type" in msg &&
+    (msg as { type?: unknown }).type === "paseo:worker-heartbeat"
+  );
+}
+
 function toRotatingFileStreamSize(size: string): string {
   const trimmed = size.trim();
   const match = trimmed.match(/^(\d+)\s*([bBkKmMgG])?$/);
@@ -120,6 +138,7 @@ export function runSupervisor(options: SupervisorOptions): SupervisorController
   let restarting = false;
   let shuttingDown = false;
   let exiting = false;
+  let forceKillTimer: NodeJS.Timeout | null = null;
   const logStream = createSupervisorLogStream(options.logFile);
 
   const writeDurableChunk = (chunk: string | Buffer): void => {
@@ -169,6 +188,40 @@ export function runSupervisor(options: SupervisorOptions): SupervisorController
       });
   };
 
+  const clearForceKillTimer = (): void => {
+    if (forceKillTimer) {
+      clearTimeout(forceKillTimer);
+      forceKillTimer = null;
+    }
+  };
+
+  const scheduleForceKill = (reason: string): void => {
+    if (!child) {
+      return;
+    }
+    const currentChild = child;
+    clearForceKillTimer();
+    forceKillTimer = setTimeout(() => {
+      forceKillTimer = null;
+      if (child !== currentChild) {
+        return;
+      }
+      writeLifecycleLog("Worker did not exit after SIGTERM; forcing SIGKILL", {
+        reason,
+        supervisorPid: process.pid,
+        workerPid: currentChild.pid ?? null,
+      });
+      void signalProcessTree(currentChild, "SIGKILL").catch((error) => {
+        writeLifecycleLog("Failed to force-kill worker process tree", {
+          error: error instanceof Error ? error.message : String(error),
+          supervisorPid: process.pid,
+          workerPid: currentChild.pid ?? null,
+        });
+      });
+    }, WORKER_TERMINATION_GRACE_MS);
+    forceKillTimer.unref();
+  };
+
   const spawnWorker = () => {
     let workerEntry: string;
     try {
@@ -197,6 +250,7 @@ export function runSupervisor(options: SupervisorOptions): SupervisorController
     }
 
     const currentChild = child;
+    let lastWorkerHeartbeatAt = Date.now();
     const heartbeat = setInterval(() => {
       const message: SupervisorHeartbeatMessage = { type: "paseo:supervisor-heartbeat" };
       if (currentChild.connected) {
@@ -210,9 +264,26 @@ export function runSupervisor(options: SupervisorOptions): SupervisorController
       } else {
         writeLifecycleLog("Worker heartbeat skipped because IPC channel is disconnected");
       }
-    }, 1000);
+    }, WORKER_HEARTBEAT_INTERVAL_MS);
     heartbeat.unref();
 
+    const workerWatchdog = setInterval(() => {
+      if (child !== currentChild || restarting || shuttingDown) {
+        return;
+      }
+      const heartbeatAgeMs = Date.now() - lastWorkerHeartbeatAt;
+      if (heartbeatAgeMs < WORKER_HEARTBEAT_TIMEOUT_MS) {
+        return;
+      }
+      writeLifecycleLog("Worker heartbeat timed out; restarting worker", {
+        heartbeatAgeMs,
+        supervisorPid: process.pid,
+        workerPid: currentChild.pid ?? null,
+      });
+      requestRestart("worker_heartbeat_timeout");
+    }, WORKER_HEARTBEAT_INTERVAL_MS);
+    workerWatchdog.unref();
+
     child.on("disconnect", () => {
       writeLifecycleLog("Worker IPC channel disconnected");
     });
@@ -228,6 +299,10 @@ export function runSupervisor(options: SupervisorOptions): SupervisorController
     });
 
     child.on("message", (msg: unknown) => {
+      if (isWorkerHeartbeatMessage(msg)) {
+        lastWorkerHeartbeatAt = Date.now();
+        return;
+      }
       const lifecycleMessage = parseLifecycleMessage(msg);
       if (!lifecycleMessage) {
         return;
@@ -256,8 +331,10 @@ export function runSupervisor(options: SupervisorOptions): SupervisorController
       requestRestart(reason);
     });
 
-    child.on("close", (code, signal) => {
+    child.on("exit", (code, signal) => {
       clearInterval(heartbeat);
+      clearInterval(workerWatchdog);
+      clearForceKillTimer();
       const exitDescriptor = describeExit(code, signal);
       writeLifecycleLog("Worker exited", { code, signal, exit: exitDescriptor });
 
@@ -308,6 +385,7 @@ export function runSupervisor(options: SupervisorOptions): SupervisorController
     writeLifecycleLog("Restart requested", { reason });
     log(`${reason}. Stopping worker for restart...`);
     signalWorker("SIGTERM", reason);
+    scheduleForceKill(reason);
   };
 
   const requestShutdown = (reason: string) => {
@@ -323,6 +401,7 @@ export function runSupervisor(options: SupervisorOptions): SupervisorController
       return;
     }
     signalWorker("SIGTERM", reason);
+    scheduleForceKill(reason);
   };
 
   const forwardSignal = (signal: NodeJS.Signals) => {

packages/server/src/server/daemon-worker.ts

@@ -27,6 +27,10 @@ interface SupervisorHeartbeatMessage {
   type: "paseo:supervisor-heartbeat";
 }
 
+interface WorkerHeartbeatMessage {
+  type: "paseo:worker-heartbeat";
+}
+
 interface BootstrapResult {
   paseoHome: string;
   logger: ReturnType<typeof createRootLogger>;
@@ -228,6 +232,13 @@ async function main() {
     const supervisorPid = process.ppid;
     let lastSupervisorHeartbeatAt = Date.now();
     let supervisorExitRequested = false;
+    const sendWorkerHeartbeat = () => {
+      try {
+        process.send?.({ type: "paseo:worker-heartbeat" } satisfies WorkerHeartbeatMessage);
+      } catch {
+        // The disconnect handler below owns supervisor-loss shutdown.
+      }
+    };
     const exitAfterSupervisorLoss = (reason: string) => {
       if (supervisorExitRequested) {
         return;
@@ -261,6 +272,10 @@ async function main() {
     });
     process.on("disconnect", () => exitAfterSupervisorLoss("ipc_disconnect_event"));
 
+    sendWorkerHeartbeat();
+    const workerHeartbeat = setInterval(sendWorkerHeartbeat, 1_000);
+    workerHeartbeat.unref();
+
     const timer = setInterval(() => {
       const ipcConnected = typeof process.connected === "boolean" ? process.connected : true;
       const heartbeatExpired = Date.now() - lastSupervisorHeartbeatAt > 3500;

packages/server/src/server/test-utils/workspace-git-service-stub.ts

@@ -82,6 +82,10 @@ export function createNoopWorkspaceGitService(
       workingTreeWatchSetupInFlightCount: 0,
       workspaceRefreshInFlightCount: 0,
       workspaceRefreshQueuedCount: 0,
+      workspaceRefreshAdmissionActiveCount: 0,
+      workspaceRefreshAdmissionPendingCount: 0,
+      workspaceObservationSetupAdmissionActiveCount: 0,
+      workspaceObservationSetupAdmissionPendingCount: 0,
       fetchInFlightCount: 0,
       snapshotUpdatedListenerCount: 0,
       watcherErrorCallbackCount: 0,

packages/server/src/server/websocket-server.ts

@@ -246,6 +246,10 @@ function createFallbackWorkspaceGitService(): WorkspaceGitService {
       workingTreeWatchSetupInFlightCount: 0,
       workspaceRefreshInFlightCount: 0,
       workspaceRefreshQueuedCount: 0,
+      workspaceRefreshAdmissionActiveCount: 0,
+      workspaceRefreshAdmissionPendingCount: 0,
+      workspaceObservationSetupAdmissionActiveCount: 0,
+      workspaceObservationSetupAdmissionPendingCount: 0,
       fetchInFlightCount: 0,
       snapshotUpdatedListenerCount: 0,
       watcherErrorCallbackCount: 0,

packages/server/src/server/workspace-git-service.observation.test.ts

@@ -1093,7 +1093,8 @@ describe("WorkspaceGitService checkout observation", () => {
       expect(watcher.subscribe).toHaveBeenCalledTimes(1);
     });
     await vi.advanceTimersByTimeAsync(35_000);
-    expect(service.getMetrics().workingTreeWatchSetupInFlightCount).toBe(1);
+    expect(service.getMetrics().workingTreeWatchSetupInFlightCount).toBe(0);
+    expect(service.getMetrics().workspaceObservationSetupAdmissionActiveCount).toBe(0);
     expect(getWatcherSubscribeCallCount(watcher, REPO_CWD)).toBe(1);
 
     openedSubscription.resolve({ unsubscribe: erroredUnsubscribe });

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

@@ -9,6 +9,9 @@ import type {
   PullRequestStatusResult,
 } from "../utils/checkout-git.js";
 import {
+  WORKSPACE_GIT_OBSERVATION_SETUP_CONCURRENCY,
+  WORKSPACE_GIT_REFRESH_CONCURRENCY,
+  WORKSPACE_GIT_WATCHER_SUBSCRIBE_TIMEOUT_MS,
   WorkspaceGitServiceImpl,
   type WorkspaceGitRuntimeSnapshot,
 } from "./workspace-git-service.js";
@@ -507,6 +510,279 @@ describe("WorkspaceGitServiceImpl", () => {
     service.dispose();
   });
 
+  test("bounds refresh generations and re-enqueues hot workspace successors fairly", async () => {
+    const hotCwds = Array.from({ length: WORKSPACE_GIT_REFRESH_CONCURRENCY }, (_, index) =>
+      path.resolve(`/tmp/hot-repo-${index}`),
+    );
+    const fifthCwd = path.resolve("/tmp/fifth-repo");
+    const firstReadGates = new Map(
+      [...hotCwds, fifthCwd].map((cwd) => [cwd, createDeferred<CheckoutStatusGit>()]),
+    );
+    const pendingFirstReads = new Set(firstReadGates.keys());
+    const startedReads: string[] = [];
+    const getCheckoutStatus = vi.fn((cwd: string) => {
+      startedReads.push(cwd);
+      const firstRead = firstReadGates.get(cwd);
+      if (firstRead && pendingFirstReads.delete(cwd)) {
+        return firstRead.promise;
+      }
+      return Promise.resolve(createCheckoutStatus(cwd, { currentBranch: "successor" }));
+    });
+    const service = createService({ getCheckoutStatus });
+
+    const hotInitialReads = hotCwds.map((cwd) => service.getSnapshot(cwd, { includeForge: false }));
+    await vi.waitFor(() => {
+      expect(startedReads).toEqual(hotCwds);
+    });
+
+    const fifthRead = service.getSnapshot(fifthCwd, { includeForge: false });
+    const hotSuccessorReads = hotCwds.map((cwd) =>
+      service.getSnapshot(cwd, {
+        force: true,
+        includeForge: false,
+        reason: "hot-successor",
+      }),
+    );
+    expect(service.getMetrics()).toMatchObject({
+      workspaceRefreshAdmissionActiveCount: WORKSPACE_GIT_REFRESH_CONCURRENCY,
+      workspaceRefreshAdmissionPendingCount: 1,
+      workspaceRefreshQueuedCount: WORKSPACE_GIT_REFRESH_CONCURRENCY,
+    });
+
+    firstReadGates.get(hotCwds[0])?.resolve(createCheckoutStatus(hotCwds[0]));
+    await vi.waitFor(() => {
+      expect(startedReads).toContain(fifthCwd);
+    });
+    expect(startedReads.filter((cwd) => cwd === hotCwds[0])).toHaveLength(1);
+
+    for (const [cwd, gate] of firstReadGates) {
+      gate.resolve(createCheckoutStatus(cwd));
+    }
+    await Promise.all([...hotInitialReads, fifthRead, ...hotSuccessorReads]);
+
+    for (const cwd of hotCwds) {
+      expect(startedReads.filter((startedCwd) => startedCwd === cwd)).toHaveLength(2);
+    }
+    expect(startedReads.filter((cwd) => cwd === fifthCwd)).toHaveLength(1);
+
+    service.dispose();
+  });
+
+  test("bounds workspace observation setup pipelines", async () => {
+    const cwds = Array.from(
+      { length: WORKSPACE_GIT_OBSERVATION_SETUP_CONCURRENCY + 2 },
+      (_, index) => `/tmp/observation-repo-${index}`,
+    );
+    const setupGate = createDeferred<void>();
+    let activeSetups = 0;
+    let maxActiveSetups = 0;
+    const subscribe = vi.fn(async () => {
+      activeSetups += 1;
+      maxActiveSetups = Math.max(maxActiveSetups, activeSetups);
+      await setupGate.promise;
+      activeSetups -= 1;
+      return createAsyncSubscription();
+    });
+    const runGitCommand = vi.fn(async (_args: string[], options: { cwd: string }) => ({
+      stdout: `${options.cwd}\n`,
+      stderr: "",
+      truncated: false,
+      exitCode: 0,
+      signal: null,
+    }));
+    const service = createService({ subscribe, runGitCommand });
+
+    await Promise.all(cwds.map((cwd) => service.getSnapshot(cwd, { includeForge: false })));
+    const subscriptions = cwds.map((cwd) => service.registerWorkspace({ cwd }, vi.fn()));
+
+    await vi.waitFor(() => {
+      expect(subscribe).toHaveBeenCalledTimes(WORKSPACE_GIT_OBSERVATION_SETUP_CONCURRENCY);
+    });
+    expect(maxActiveSetups).toBe(WORKSPACE_GIT_OBSERVATION_SETUP_CONCURRENCY);
+    expect(service.getMetrics()).toMatchObject({
+      workspaceObservationSetupAdmissionActiveCount: WORKSPACE_GIT_OBSERVATION_SETUP_CONCURRENCY,
+      workspaceObservationSetupAdmissionPendingCount: 2,
+    });
+
+    setupGate.resolve();
+    await vi.waitFor(() => {
+      expect(service.getMetrics().workspaceObservationSetupInFlightCount).toBe(0);
+    });
+    expect(maxActiveSetups).toBe(WORKSPACE_GIT_OBSERVATION_SETUP_CONCURRENCY);
+
+    for (const subscription of subscriptions) {
+      subscription.unsubscribe();
+    }
+    service.dispose();
+  });
+
+  test("watcher subscription deadlines release observation admission slots", async () => {
+    const cwds = Array.from(
+      { length: WORKSPACE_GIT_OBSERVATION_SETUP_CONCURRENCY + 2 },
+      (_, index) => path.resolve(`/tmp/deadline-repo-${index}`),
+    );
+    const stalledSubscriptions = Array.from(
+      { length: WORKSPACE_GIT_OBSERVATION_SETUP_CONCURRENCY },
+      () => createDeferred<ReturnType<typeof createAsyncSubscription>>(),
+    );
+    const subscribedPaths: string[] = [];
+    const subscribe = vi.fn((watchPath: string) => {
+      subscribedPaths.push(watchPath);
+      const stalled = stalledSubscriptions.shift();
+      return stalled?.promise ?? Promise.resolve(createAsyncSubscription());
+    });
+    const runGitCommand = vi.fn(async (_args: string[], options: { cwd: string }) => ({
+      stdout: `${options.cwd}\n`,
+      stderr: "",
+      truncated: false,
+      exitCode: 0,
+      signal: null,
+    }));
+    const service = createService({ subscribe, runGitCommand });
+
+    await Promise.all(cwds.map((cwd) => service.getSnapshot(cwd, { includeForge: false })));
+    const subscriptions = cwds.map((cwd) => service.registerWorkspace({ cwd }, vi.fn()));
+
+    await vi.waitFor(() => {
+      expect(subscribe).toHaveBeenCalledTimes(WORKSPACE_GIT_OBSERVATION_SETUP_CONCURRENCY);
+    });
+
+    await vi.advanceTimersByTimeAsync(WORKSPACE_GIT_WATCHER_SUBSCRIBE_TIMEOUT_MS);
+    await vi.waitFor(() => {
+      expect(subscribedPaths).toEqual(expect.arrayContaining(cwds));
+      expect(service.getMetrics().workspaceObservationSetupInFlightCount).toBe(0);
+    });
+
+    for (const subscription of subscriptions) {
+      subscription.unsubscribe();
+    }
+    service.dispose();
+  });
+
+  test("dispose settles queued refresh and observation work and cleans late watchers", async () => {
+    const cwds = Array.from({ length: WORKSPACE_GIT_REFRESH_CONCURRENCY + 1 }, (_, index) =>
+      path.resolve(`/tmp/dispose-repo-${index}`),
+    );
+    const refreshGates = new Map(cwds.map((cwd) => [cwd, createDeferred<CheckoutStatusGit>()]));
+    const getCheckoutStatus = vi.fn((cwd: string) => {
+      const gate = refreshGates.get(cwd);
+      if (!gate) {
+        throw new Error(`Unexpected cwd: ${cwd}`);
+      }
+      return gate.promise;
+    });
+    const watcherGates = Array.from({ length: WORKSPACE_GIT_OBSERVATION_SETUP_CONCURRENCY }, () =>
+      createDeferred<ReturnType<typeof createAsyncSubscription>>(),
+    );
+    const lateSubscriptions = watcherGates.map(() => createAsyncSubscription());
+    const lateUnsubscribes = lateSubscriptions.map(({ unsubscribe }) => unsubscribe);
+    const subscribe = vi.fn(
+      () => watcherGates[subscribe.mock.calls.length - 1]?.promise ?? Promise.reject(),
+    );
+    const runGitCommand = vi.fn(async (_args: string[], options: { cwd: string }) => ({
+      stdout: `${options.cwd}\n`,
+      stderr: "",
+      truncated: false,
+      exitCode: 0,
+      signal: null,
+    }));
+    const service = createService({ getCheckoutStatus, subscribe, runGitCommand });
+
+    cwds.forEach((cwd) => service.registerWorkspace({ cwd }, vi.fn()));
+    await vi.waitFor(() => {
+      expect(service.getMetrics()).toMatchObject({
+        workspaceRefreshAdmissionActiveCount: WORKSPACE_GIT_REFRESH_CONCURRENCY,
+        workspaceRefreshAdmissionPendingCount: 1,
+        workspaceObservationSetupAdmissionActiveCount: WORKSPACE_GIT_OBSERVATION_SETUP_CONCURRENCY,
+        workspaceObservationSetupAdmissionPendingCount:
+          cwds.length - WORKSPACE_GIT_OBSERVATION_SETUP_CONCURRENCY,
+      });
+    });
+    const queuedRefreshSettlement = service.getSnapshot(cwds.at(-1)!).then(
+      () => ({ status: "fulfilled" as const, error: null }),
+      (error: unknown) => ({ status: "rejected" as const, error }),
+    );
+
+    service.dispose();
+
+    await expect(queuedRefreshSettlement).resolves.toMatchObject({
+      status: "rejected",
+      error: { name: "AbortError" },
+    });
+    await vi.waitFor(() => {
+      expect(service.getMetrics()).toMatchObject({
+        workspaceRefreshAdmissionPendingCount: 0,
+        workspaceObservationSetupAdmissionActiveCount: 0,
+        workspaceObservationSetupAdmissionPendingCount: 0,
+      });
+    });
+    expect(subscribe).toHaveBeenCalledTimes(WORKSPACE_GIT_OBSERVATION_SETUP_CONCURRENCY);
+
+    watcherGates.forEach((gate, index) => gate.resolve(lateSubscriptions[index]!));
+    refreshGates.forEach((gate, cwd) => gate.resolve(createCheckoutStatus(cwd)));
+    await vi.waitFor(() => {
+      for (const unsubscribe of lateUnsubscribes) {
+        expect(unsubscribe).toHaveBeenCalledTimes(1);
+      }
+      expect(service.getMetrics().workspaceRefreshAdmissionActiveCount).toBe(0);
+    });
+  });
+
+  test("cleans a watcher subscription that resolves after its deadline", async () => {
+    const lateWatcher = createDeferred<ReturnType<typeof createAsyncSubscription>>();
+    const lateSubscription = createAsyncSubscription();
+    const subscribe = vi
+      .fn()
+      .mockImplementationOnce(() => lateWatcher.promise)
+      .mockResolvedValue(createAsyncSubscription());
+    const service = createService({ subscribe });
+
+    await service.getSnapshot(REPO_CWD, { includeForge: false });
+    const workspaceSubscription = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn());
+    await vi.waitFor(() => {
+      expect(subscribe).toHaveBeenCalledTimes(1);
+    });
+
+    await vi.advanceTimersByTimeAsync(WORKSPACE_GIT_WATCHER_SUBSCRIBE_TIMEOUT_MS);
+    await vi.waitFor(() => {
+      expect(service.getMetrics().workspaceObservationSetupInFlightCount).toBe(0);
+    });
+    lateWatcher.resolve(lateSubscription);
+
+    await vi.waitFor(() => {
+      expect(lateSubscription.unsubscribe).toHaveBeenCalledTimes(1);
+    });
+    workspaceSubscription.unsubscribe();
+    service.dispose();
+    expect(lateSubscription.unsubscribe).toHaveBeenCalledTimes(1);
+  });
+
+  test("rejects new work after disposal", async () => {
+    const getCheckoutSnapshotFacts = vi.fn(async (cwd: string) => createCheckoutSnapshotFacts(cwd));
+    const runGitCommand = vi.fn();
+    const service = createService({ getCheckoutSnapshotFacts, runGitCommand });
+
+    service.dispose();
+
+    expect(() => service.registerWorkspace({ cwd: REPO_CWD }, vi.fn())).toThrow(
+      "WorkspaceGitService is disposed",
+    );
+    await expect(service.getSnapshot(REPO_CWD)).rejects.toThrow("WorkspaceGitService is disposed");
+    await expect(service.refresh(REPO_CWD)).rejects.toThrow("WorkspaceGitService is disposed");
+    await expect(service.requestWorkingTreeWatch(REPO_CWD, vi.fn())).rejects.toThrow(
+      "WorkspaceGitService is disposed",
+    );
+    expect(() => service.scheduleRefreshForCwd(REPO_CWD)).toThrow(
+      "WorkspaceGitService is disposed",
+    );
+    expect(() => service.onWorkspaceStateMayHaveChanged(REPO_CWD)).toThrow(
+      "WorkspaceGitService is disposed",
+    );
+    expect(() => service.invalidateForge(REPO_CWD)).toThrow("WorkspaceGitService is disposed");
+    expect(getCheckoutSnapshotFacts).not.toHaveBeenCalled();
+    expect(runGitCommand).not.toHaveBeenCalled();
+  });
+
   test("multiple listeners on the same workspace share one observation setup", async () => {
     const getPullRequestStatus = vi.fn(async () => createPullRequestStatusResult());
     const getCheckoutSnapshotFacts = vi.fn(async (cwd: string) => createCheckoutSnapshotFacts(cwd));

packages/server/src/server/workspace-git-service.ts

@@ -3,6 +3,7 @@ import { readFile } from "node:fs/promises";
 import { join, resolve } from "node:path";
 import parcelWatcher from "@parcel/watcher";
 import { LRUCache } from "lru-cache";
+import pLimit from "p-limit";
 import type pino from "pino";
 import type { ProjectCheckoutLitePayload } from "@getpaseo/protocol/messages";
 import { parseGitRemoteLocation } from "@getpaseo/protocol/git-remote";
@@ -58,6 +59,11 @@ const FORGE_PR_STATUS_POLL_FAST_INTERVAL_MS = 20_000;
 const FORGE_PR_STATUS_POLL_SLOW_INTERVAL_MS = 120_000;
 const FORGE_PR_STATUS_POLL_ERROR_BACKOFF_CAP_MS = 300_000;
 const DEGRADED_GIT_POLL_INTERVAL_MS = 5_000;
+// Keep whole workspace pipelines below the lower-level Git process pool so daemon control work
+// retains subprocess and event-loop headroom during large workspace reconciliation bursts.
+export const WORKSPACE_GIT_REFRESH_CONCURRENCY = 4;
+export const WORKSPACE_GIT_OBSERVATION_SETUP_CONCURRENCY = 2;
+export const WORKSPACE_GIT_WATCHER_SUBSCRIBE_TIMEOUT_MS = 10_000;
 const WATCH_RECOVERY_BASE_DELAY_MS = 30_000;
 const WATCH_RECOVERY_MAX_ATTEMPTS = 3;
 // Auxiliary reads may reuse cached values within this window; snapshots do not expire on read.
@@ -196,6 +202,10 @@ export interface WorkspaceGitServiceMetrics {
   workingTreeWatchSetupInFlightCount: number;
   workspaceRefreshInFlightCount: number;
   workspaceRefreshQueuedCount: number;
+  workspaceRefreshAdmissionActiveCount: number;
+  workspaceRefreshAdmissionPendingCount: number;
+  workspaceObservationSetupAdmissionActiveCount: number;
+  workspaceObservationSetupAdmissionPendingCount: number;
   fetchInFlightCount: number;
   snapshotUpdatedListenerCount: number;
   watcherErrorCallbackCount: number;
@@ -340,6 +350,22 @@ export function subscribeToWorkspaceFileChanges(
   });
 }
 
+class WorkspaceGitServiceDisposedError extends Error {
+  constructor() {
+    super("WorkspaceGitService is disposed");
+    this.name = "WorkspaceGitServiceDisposedError";
+  }
+}
+
+class WorkspaceGitWatcherSubscriptionTimeoutError extends Error {
+  constructor(watchPath: string) {
+    super(
+      `Watcher subscription for ${watchPath} timed out after ${WORKSPACE_GIT_WATCHER_SUBSCRIBE_TIMEOUT_MS}ms`,
+    );
+    this.name = "WorkspaceGitWatcherSubscriptionTimeoutError";
+  }
+}
+
 interface WorkspaceGitTarget {
   cwd: string;
   listeners: Set<WorkspaceGitListener>;
@@ -460,6 +486,16 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
   private readonly worktreesRoot: string | undefined;
   private readonly deps: WorkspaceGitServiceDependencies;
   private readonly forgeResolver: ForgeResolver;
+  private readonly workspaceRefreshLimit = pLimit({
+    concurrency: WORKSPACE_GIT_REFRESH_CONCURRENCY,
+    rejectOnClear: true,
+  });
+  private readonly workspaceObservationSetupLimit = pLimit({
+    concurrency: WORKSPACE_GIT_OBSERVATION_SETUP_CONCURRENCY,
+    rejectOnClear: true,
+  });
+  private readonly disposeController = new AbortController();
+  private disposed = false;
   private readonly snapshotUpdatedListeners = new Set<WorkspaceGitSnapshotUpdatedListener>();
   private readonly workspaceTargets = new Map<string, WorkspaceGitTarget>();
   private readonly repoTargets = new Map<string, RepoGitTarget>();
@@ -507,6 +543,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
   }
 
   resolveForge(cwd: string): Promise<ForgeResolution | null> {
+    this.assertNotDisposed();
     return this.forgeResolver.resolve(resolve(cwd));
   }
 
@@ -514,6 +551,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
     params: { cwd: string },
     listener: WorkspaceGitListener,
   ): WorkspaceGitSubscription {
+    this.assertNotDisposed();
     const cwd = resolve(params.cwd);
     const target = this.ensureWorkspaceTarget(cwd);
     target.listeners.add(listener);
@@ -533,6 +571,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
   }
 
   onSnapshotUpdated(listener: WorkspaceGitSnapshotUpdatedListener): WorkspaceGitSubscription {
+    this.assertNotDisposed();
     this.snapshotUpdatedListeners.add(listener);
     return {
       unsubscribe: () => {
@@ -583,6 +622,12 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
       workingTreeWatchSetupInFlightCount: this.workingTreeWatchSetups.size,
       workspaceRefreshInFlightCount,
       workspaceRefreshQueuedCount,
+      workspaceRefreshAdmissionActiveCount: this.workspaceRefreshLimit.activeCount,
+      workspaceRefreshAdmissionPendingCount: this.workspaceRefreshLimit.pendingCount,
+      workspaceObservationSetupAdmissionActiveCount:
+        this.workspaceObservationSetupLimit.activeCount,
+      workspaceObservationSetupAdmissionPendingCount:
+        this.workspaceObservationSetupLimit.pendingCount,
       fetchInFlightCount,
       snapshotUpdatedListenerCount: this.snapshotUpdatedListeners.size,
       watcherErrorCallbackCount: this.watcherErrorCallbackCount,
@@ -593,6 +638,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
     cwd: string,
     options?: WorkspaceGitSnapshotOptions,
   ): Promise<WorkspaceGitRuntimeSnapshot> {
+    this.assertNotDisposed();
     cwd = resolve(cwd);
     const request = this.normalizeRefreshRequest(options, "getSnapshot", true);
     const target = this.ensureWorkspaceTarget(cwd);
@@ -604,6 +650,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
   }
 
   async getCheckout(cwd: string): Promise<ProjectCheckoutLitePayload> {
+    this.assertNotDisposed();
     const normalizedCwd = resolve(cwd);
     const status = await this.deps.getCheckoutStatus(normalizedCwd, {
       paseoHome: this.paseoHome,
@@ -640,6 +687,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
     options: CheckoutDiffCompare,
     readOptions?: WorkspaceGitReadOptions,
   ): Promise<CheckoutDiffResult> {
+    this.assertNotDisposed();
     const normalizedCwd = resolve(cwd);
     const normalizedOptions = this.normalizeCheckoutDiffOptions(options);
     const key = this.buildCheckoutDiffCacheKey(normalizedCwd, normalizedOptions);
@@ -689,6 +737,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
     ref: string,
     options?: WorkspaceGitReadOptions,
   ): Promise<WorkspaceGitBranchValidationResult> {
+    this.assertNotDisposed();
     const normalizedCwd = resolve(cwd);
     const normalizedRef = ref.trim();
     const key = JSON.stringify(["branch-validation", normalizedCwd, normalizedRef]);
@@ -698,6 +747,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
   }
 
   hasLocalBranch(cwd: string, branch: string, options?: WorkspaceGitReadOptions): Promise<boolean> {
+    this.assertNotDisposed();
     const normalizedCwd = resolve(cwd);
     const normalizedBranch = branch.trim();
     const ref = `refs/heads/${normalizedBranch}`;
@@ -717,6 +767,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
     options?: WorkspaceGitBranchSuggestionsOptions,
     readOptions?: WorkspaceGitReadOptions,
   ): Promise<WorkspaceGitBranchSuggestion[]> {
+    this.assertNotDisposed();
     const normalizedCwd = resolve(cwd);
     const query = options?.query ?? "";
     const limit = options?.limit;
@@ -731,6 +782,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
     options?: WorkspaceGitStashListOptions,
     readOptions?: WorkspaceGitReadOptions,
   ): Promise<WorkspaceGitStashEntry[]> {
+    this.assertNotDisposed();
     const normalizedCwd = resolve(cwd);
     const paseoOnly = options?.paseoOnly !== false;
     const key = JSON.stringify(["stashes", normalizedCwd, paseoOnly]);
@@ -747,6 +799,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
     cwdOrRepoRoot: string,
     options?: WorkspaceGitReadOptions,
   ): Promise<WorkspaceGitWorktreeInfo[]> {
+    this.assertNotDisposed();
     const repoRoot = await this.resolveRepoRoot(cwdOrRepoRoot, options);
     const key = JSON.stringify(["worktrees", repoRoot]);
     return this.readAuxiliaryCache(this.worktreeListCache, key, options, () =>
@@ -773,6 +826,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
     cwdOrRepoRoot: string,
     options?: WorkspaceGitReadOptions,
   ): Promise<string> {
+    this.assertNotDisposed();
     const cwd = resolve(cwdOrRepoRoot);
     const key = JSON.stringify(["default-branch", cwd]);
     return this.readAuxiliaryCache(this.defaultBranchCache, key, options, async () => {
@@ -798,6 +852,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
   }
 
   async refresh(cwd: string, _options?: { priority?: "normal" | "high" }): Promise<void> {
+    this.assertNotDisposed();
     cwd = resolve(cwd);
     const target = this.ensureWorkspaceTarget(cwd);
     await this.refreshWorkspaceTarget(target, {
@@ -816,6 +871,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
     cwd: string,
     onChange: () => void,
   ): Promise<{ repoRoot: string | null; unsubscribe: () => void }> {
+    this.assertNotDisposed();
     cwd = resolve(cwd);
     const target = await this.ensureWorkingTreeWatchTarget(cwd);
     target.listeners.add(onChange);
@@ -829,6 +885,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
   }
 
   scheduleRefreshForCwd(cwd: string): void {
+    this.assertNotDisposed();
     cwd = resolve(cwd);
     const target = this.workspaceTargets.get(cwd);
     if (target) {
@@ -837,6 +894,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
   }
 
   onWorkspaceStateMayHaveChanged(cwd: string): void {
+    this.assertNotDisposed();
     const normalizedCwd = resolve(cwd);
     const target = this.workspaceTargets.get(normalizedCwd);
     if (!target || target.closed) {
@@ -856,10 +914,19 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
    * git mutations to force a fresh forge status on the next refresh.
    */
   invalidateForge(cwd: string): void {
+    this.assertNotDisposed();
     this.forgeResolver.invalidate(resolve(cwd));
   }
 
   dispose(): void {
+    if (this.disposed) {
+      return;
+    }
+    this.disposed = true;
+    this.disposeController.abort(new WorkspaceGitServiceDisposedError());
+    this.workspaceRefreshLimit.clearQueue();
+    this.workspaceObservationSetupLimit.clearQueue();
+
     for (const target of this.workspaceTargets.values()) {
       this.closeWorkspaceTarget(target);
     }
@@ -880,7 +947,14 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
     this.snapshotUpdatedListeners.clear();
   }
 
+  private assertNotDisposed(): void {
+    if (this.disposed) {
+      throw new WorkspaceGitServiceDisposedError();
+    }
+  }
+
   private ensureWorkspaceTarget(cwd: string): WorkspaceGitTarget {
+    this.assertNotDisposed();
     const existingTarget = this.workspaceTargets.get(cwd);
     if (existingTarget) {
       return existingTarget;
@@ -895,6 +969,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
     options: WorkspaceGitReadOptions | undefined,
     load: () => Promise<T>,
   ): Promise<T> {
+    this.assertNotDisposed();
     if (options?.force && !options.reason) {
       throw new Error("WorkspaceGitService forced read requires a reason");
     }
@@ -951,6 +1026,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
   }
 
   private ensureWorkingTreeWatchTarget(cwd: string): Promise<WorkingTreeWatchTarget> {
+    this.assertNotDisposed();
     const targetCwd = this.workingTreeWatchAliases.get(cwd);
     if (targetCwd) {
       const existingTarget = this.workingTreeWatchTargets.get(targetCwd);
@@ -1067,9 +1143,16 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
       return;
     }
 
-    target.observationSetupPromise = Promise.resolve()
-      .then(() => this.setupWorkspaceObservation(target))
+    target.observationSetupPromise = this.workspaceObservationSetupLimit(async () => {
+      if (!this.isActiveObservedWorkspaceTarget(target)) {
+        return;
+      }
+      await this.setupWorkspaceObservation(target);
+    })
       .catch((error) => {
+        if (this.disposed || !this.isActiveObservedWorkspaceTarget(target)) {
+          return;
+        }
         this.logger.warn(
           { err: error, cwd: target.cwd },
           "Failed to set up workspace git observation",
@@ -1183,6 +1266,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
     this.workingTreeWatchTargets.set(cwd, target);
     this.workingTreeWatchAliases.set(cwd, cwd);
     await this.startWorkingTreeSubscription(target);
+    this.assertNotDisposed();
 
     if (repoRoot === null) {
       this.startWorkingTreeWatchFallback(target, "not_a_git_checkout");
@@ -1191,6 +1275,95 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
     return target;
   }
 
+  private async subscribeWithDeadline(
+    watchPath: string,
+    callback: parcelWatcher.SubscribeCallback,
+    options: parcelWatcher.Options,
+    onSubscribeSettled: () => void,
+    shouldAbandonSubscription: () => boolean,
+  ): Promise<parcelWatcher.AsyncSubscription> {
+    this.assertNotDisposed();
+    const signal = this.disposeController.signal;
+    let outcome: "pending" | "accepted" | "expired" = "pending";
+    let timeout: NodeJS.Timeout | null = null;
+    let removeAbortListener = () => {};
+    let unsubscribePromise: Promise<void> | null = null;
+    let subscriptionPromise: Promise<parcelWatcher.AsyncSubscription>;
+    try {
+      subscriptionPromise = this.deps
+        .subscribe(watchPath, callback, options)
+        .finally(onSubscribeSettled);
+    } catch (error) {
+      onSubscribeSettled();
+      throw error;
+    }
+    void subscriptionPromise.then(
+      (subscription) => {
+        if ((outcome === "expired" || signal.aborted) && !shouldAbandonSubscription()) {
+          unsubscribePromise ??= this.unsubscribeWatcherSubscription(subscription, watchPath);
+          return unsubscribePromise;
+        }
+        return undefined;
+      },
+      () => undefined,
+    );
+    const timeoutPromise = new Promise<never>((_resolve, reject) => {
+      timeout = setTimeout(() => {
+        outcome = "expired";
+        reject(new WorkspaceGitWatcherSubscriptionTimeoutError(watchPath));
+      }, WORKSPACE_GIT_WATCHER_SUBSCRIBE_TIMEOUT_MS);
+    });
+    const disposalPromise = new Promise<never>((_resolve, reject) => {
+      const rejectForDisposal = () => {
+        outcome = "expired";
+        reject(signal.reason);
+      };
+      if (signal.aborted) {
+        rejectForDisposal();
+        return;
+      }
+      signal.addEventListener("abort", rejectForDisposal, { once: true });
+      removeAbortListener = () => signal.removeEventListener("abort", rejectForDisposal);
+    });
+
+    try {
+      const subscription = await Promise.race([
+        subscriptionPromise,
+        timeoutPromise,
+        disposalPromise,
+      ]);
+      if (signal.aborted) {
+        outcome = "expired";
+        if (!shouldAbandonSubscription()) {
+          unsubscribePromise ??= this.unsubscribeWatcherSubscription(subscription, watchPath);
+          await unsubscribePromise;
+        }
+        throw signal.reason;
+      }
+      outcome = "accepted";
+      return subscription;
+    } finally {
+      if (outcome === "pending") {
+        outcome = "expired";
+      }
+      if (timeout) {
+        clearTimeout(timeout);
+      }
+      removeAbortListener();
+    }
+  }
+
+  private async unsubscribeWatcherSubscription(
+    subscription: parcelWatcher.AsyncSubscription,
+    watchPath: string,
+  ): Promise<void> {
+    try {
+      await subscription.unsubscribe();
+    } catch (error) {
+      this.logger.warn({ err: error, watchPath }, "Failed to stop watcher subscription");
+    }
+  }
+
   private async startWorkingTreeSubscription(
     target: WorkingTreeWatchTarget,
     options?: { replaceFallback?: boolean },
@@ -1198,8 +1371,14 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
     const ignore = [join(target.watchPath, ".git"), ...target.ignoredDirectories];
     let watcherErrored = false;
     let subscribeSettled = false;
+    const markSubscribeSettled = () => {
+      subscribeSettled = true;
+      if (watcherErrored) {
+        this.scheduleWorkingTreeWatchRecovery(target);
+      }
+    };
     try {
-      const subscription = await this.deps.subscribe(
+      const subscription = await this.subscribeWithDeadline(
         target.watchPath,
         (error, events) => {
           if (error) {
@@ -1227,10 +1406,10 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
           this.notifyWorkingTreeChanged(target, "working-tree-watch");
         },
         { ignore },
+        markSubscribeSettled,
+        () => watcherErrored,
       );
-      subscribeSettled = true;
       if (watcherErrored) {
-        this.scheduleWorkingTreeWatchRecovery(target);
         return false;
       }
       if (
@@ -1238,7 +1417,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
         (target.fallbackPolling && !options?.replaceFallback) ||
         target.subscription
       ) {
-        await subscription.unsubscribe();
+        await this.unsubscribeWatcherSubscription(subscription, target.watchPath);
         return false;
       }
       target.subscription = subscription;
@@ -1251,11 +1430,12 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
       }
       return true;
     } catch (error) {
-      subscribeSettled = true;
       if (watcherErrored) {
-        this.scheduleWorkingTreeWatchRecovery(target);
         return false;
       }
+      if (this.disposed || target.closed) {
+        throw error;
+      }
       this.logger.warn(
         { err: error, cwd: target.cwd },
         "Failed to start working tree watcher; using degraded polling",
@@ -1271,7 +1451,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
     target: WorkingTreeWatchTarget,
     reason: WorkingTreeWatchFallbackReason,
   ): void {
-    if (target.fallbackPolling) {
+    if (this.disposed || target.closed || target.fallbackPolling) {
       return;
     }
     target.fallbackPolling = true;
@@ -1606,8 +1786,14 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
     const matchesRepoGitRoot = createRealpathAwarePathMatcher(target.repoGitRoot);
     let watcherErrored = false;
     let subscribeSettled = false;
+    const markSubscribeSettled = () => {
+      subscribeSettled = true;
+      if (watcherErrored) {
+        this.scheduleRepoMetadataWatchRecovery(target);
+      }
+    };
     try {
-      const subscription = await this.deps.subscribe(
+      const subscription = await this.subscribeWithDeadline(
         target.repoGitRoot,
         (error, events) => {
           if (error) {
@@ -1644,10 +1830,10 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
           }
         },
         { ignore },
+        markSubscribeSettled,
+        () => watcherErrored,
       );
-      subscribeSettled = true;
       if (watcherErrored) {
-        this.scheduleRepoMetadataWatchRecovery(target);
         return false;
       }
       if (
@@ -1655,7 +1841,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
         (target.fallbackPolling && !options?.replaceFallback) ||
         this.repoTargets.get(target.repoGitRoot) !== target
       ) {
-        await subscription.unsubscribe();
+        await this.unsubscribeWatcherSubscription(subscription, target.repoGitRoot);
         return false;
       }
       target.subscription = subscription;
@@ -1668,11 +1854,12 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
       }
       return true;
     } catch (error) {
-      subscribeSettled = true;
       if (watcherErrored) {
-        this.scheduleRepoMetadataWatchRecovery(target);
         return false;
       }
+      if (this.disposed || target.closed) {
+        throw error;
+      }
       this.logger.warn(
         { err: error, repoGitRoot: target.repoGitRoot },
         "Failed to start repository metadata watcher; using degraded polling",
@@ -2280,6 +2467,9 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
     try {
       await this.requestWorkspaceSnapshot(target, request);
     } catch (error) {
+      if (this.disposed || target.closed) {
+        return;
+      }
       this.logger.warn(
         { err: error, cwd: target.cwd, reason: request.reason },
         "Failed to refresh workspace git snapshot",
@@ -2419,7 +2609,16 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
         });
       }
       try {
-        snapshot = await this.refreshSnapshot(target, request);
+        const admittedSnapshot = await this.workspaceRefreshLimit(() => {
+          if (target.closed || this.workspaceTargets.get(target.cwd) !== target) {
+            return null;
+          }
+          return this.refreshSnapshot(target, request);
+        });
+        if (!admittedSnapshot) {
+          break;
+        }
+        snapshot = admittedSnapshot;
         this.rememberSnapshot(target, snapshot, {
           notify: request.notify,
           forceEmit: request.force,
@@ -2429,6 +2628,10 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
         failure = { error };
       }
 
+      if (this.disposed || target.closed || this.workspaceTargets.get(target.cwd) !== target) {
+        break;
+      }
+
       const state = target.refreshState;
       if (state.status !== "in-flight" || !state.queued) {
         break;

packages/server/src/utils/tree-kill.ts

@@ -38,13 +38,13 @@ export async function terminateWithTreeKill(
   }
 
   const exitPromise = waitForProcessExit(child);
-  await signalTreeOrChild(child, options.gracefulSignal ?? "SIGTERM");
+  await signalProcessTree(child, options.gracefulSignal ?? "SIGTERM");
   if (await waitForExitOrTimeout(exitPromise, options.gracefulTimeoutMs)) {
     return "terminated";
   }
 
   options.onForceSignal?.();
-  await signalTreeOrChild(child, options.forceSignal ?? "SIGKILL");
+  await signalProcessTree(child, options.forceSignal ?? "SIGKILL");
   if (options.forceTimeoutMs === undefined) {
     return "killed";
   }
@@ -53,7 +53,7 @@ export async function terminateWithTreeKill(
     : "kill-timeout";
 }
 
-function signalTreeOrChild(child: TreeKillTarget, signal: NodeJS.Signals): Promise<void> {
+export function signalProcessTree(child: TreeKillTarget, signal: NodeJS.Signals): Promise<void> {
   if (isProcessExited(child)) {
     return Promise.resolve();
   }