fix(server): wait for native watcher delivery before declaring it silent (#8107)

* test(server): expose macOS observation startup failures

* test(server): repeat macOS startup to capture the intermittent failure

* fix(server): probe watcher liveness until native event delivery starts

* test(server): use the production deadline for watcher startup regressions

---------

Co-authored-by: paseo-bot[bot] <266920839+paseo-bot[bot]@users.noreply.github.com>

packages/server/src/server/watcher-liveness-canary.test.ts

@@ -1,4 +1,5 @@
-import { mkdtemp, rm } from "node:fs/promises";
+import { watchFile, unwatchFile } from "node:fs";
+import { mkdtemp, readFile, rm, stat } from "node:fs/promises";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { afterEach, expect, test } from "vitest";
@@ -31,3 +32,39 @@ test("rejects when the watcher never reports the canary", async () => {
 
   await expect(canary.verify()).rejects.toThrow("did not report its liveness canary");
 });
+
+test("verifies a watcher that starts delivering after the initial canary write", async () => {
+  const watchRoot = await mkdtemp(join(tmpdir(), "paseo-watcher-canary-"));
+  cleanupPaths.push(watchRoot);
+  const canary = createWatcherLivenessCanary(watchRoot);
+  const verification = canary.verify();
+  const verified = expect(verification).resolves.toBeUndefined();
+
+  // The initial write happens before the observer starts delivering changes.
+  await expect.poll(() => readFile(canary.path, "utf8")).toBe("paseo watcher liveness canary\n");
+  watchFile(canary.path, { interval: 10 }, () => {
+    canary.filterEvents([{ path: canary.path, type: "update" }]);
+  });
+  try {
+    await verified;
+    await expect(readFile(canary.path)).rejects.toMatchObject({ code: "ENOENT" });
+  } finally {
+    unwatchFile(canary.path);
+  }
+});
+
+test("aborting verification stops pulses before removing the canary", async () => {
+  const watchRoot = await mkdtemp(join(tmpdir(), "paseo-watcher-canary-"));
+  cleanupPaths.push(watchRoot);
+  const canary = createWatcherLivenessCanary(watchRoot);
+  const controller = new AbortController();
+  const verification = canary.verify(controller.signal);
+  const cancelled = expect(verification).rejects.toThrow("stop watching");
+  await expect.poll(() => readFile(canary.path, "utf8")).toBe("paseo watcher liveness canary\n");
+  const initialWrite = (await stat(canary.path)).mtimeMs;
+  await expect.poll(async () => (await stat(canary.path)).mtimeMs).toBeGreaterThan(initialWrite);
+
+  controller.abort(new Error("stop watching"));
+  await cancelled;
+  await expect(readFile(canary.path)).rejects.toMatchObject({ code: "ENOENT" });
+});

packages/server/src/server/watcher-liveness-canary.ts

@@ -1,9 +1,11 @@
 import { randomUUID } from "node:crypto";
 import { rm, writeFile } from "node:fs/promises";
 import { join } from "node:path";
+import { setInterval } from "node:timers/promises";
 import type { FileChange } from "./file-observer/index.js";
 
 export const WATCHER_LIVENESS_CANARY_TIMEOUT_MS = 10_000;
+const CANARY_CONTENTS = "paseo watcher liveness canary\n";
 
 export interface WatcherLivenessCanary {
   readonly path: string;
@@ -32,7 +34,11 @@ export function createWatcherLivenessCanary(
       return filtered;
     },
     async verify(signal) {
-      await writeFile(canaryPath, "paseo watcher liveness canary\n", { flag: "wx" });
+      await writeFile(canaryPath, CANARY_CONTENTS, { flag: "wx" });
+      // Native registration can return before macOS starts its FSEvents stream.
+      // Keep probing until an event round-trips instead of relying on the first write.
+      const pulses = new AbortController();
+      const pulsing = pulseCanary(canaryPath, pulses.signal);
       let timeout: NodeJS.Timeout | null = null;
       let removeAbortListener = () => {};
       try {
@@ -55,12 +61,24 @@ export function createWatcherLivenessCanary(
           signal.addEventListener("abort", rejectForAbort, { once: true });
           removeAbortListener = () => signal.removeEventListener("abort", rejectForAbort);
         });
-        await Promise.race([reported, timeoutPromise, abortPromise]);
+        await Promise.race([reported, timeoutPromise, abortPromise, pulsing]);
       } finally {
         if (timeout) clearTimeout(timeout);
         removeAbortListener();
+        pulses.abort();
+        await Promise.allSettled([pulsing]);
         await rm(canaryPath, { force: true });
       }
     },
   };
 }
+
+async function pulseCanary(path: string, signal: AbortSignal): Promise<void> {
+  try {
+    for await (const _ of setInterval(100, undefined, { signal })) {
+      await writeFile(path, CANARY_CONTENTS);
+    }
+  } catch (error) {
+    if (!signal.aborted) throw error;
+  }
+}

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

@@ -12,12 +12,7 @@ import { WorkspaceGitServiceImpl } from "./workspace-git-service.js";
 import type { FileChange, FileObserver, SubscribeToFileChanges } from "./file-observer/index.js";
 
 function createLogger(): pino.Logger {
-  const logger = {
-    child: () => logger,
-    debug: vi.fn(),
-    warn: vi.fn(),
-  };
-  return logger as unknown as pino.Logger;
+  return pino({ level: "warn" });
 }
 
 function createFacts(cwd: string): CheckoutSnapshotFacts {
@@ -232,7 +227,14 @@ test("recursive observation updates tracked state and prunes ignored storms", as
 
   await vi.waitFor(
     () => {
-      expect(activeWatcherCount).toBe(2);
+      expect(
+        activeWatcherCount,
+        JSON.stringify({
+          observer: observer.getDiagnostics(),
+          service: service.getMetrics(),
+          deliveredEvents,
+        }),
+      ).toBe(2);
       expect(workspaceRuntimeStore.get(repoDir)).not.toBeNull();
       expect(service.getMetrics()).toMatchObject({
         workspaceObservationSetupInFlightCount: 0,