fix(workspaces): invalidate watches after helper writes

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

@@ -732,7 +732,7 @@ test("selected workspaces with the same public cwd keep snapshots and diffs isol
 
 test("selected workspaces with the same public cwd keep observations isolated", async () => {
   const { workspaceA, workspaceB, workspaceRuntime } = await createSamePublicCwdFixture();
-  const commandRuntimeEventTimeoutMs = 30_000;
+  const commandRuntimeEventTimeoutMs = 15_000;
   await Promise.all([
     workspaceA.getSnapshot({ force: true, includeForge: false, reason: "observation-warm-a" }),
     workspaceB.getSnapshot({ force: true, includeForge: false, reason: "observation-warm-b" }),
@@ -773,7 +773,7 @@ test("selected workspaces with the same public cwd keep observations isolated",
   expect(workspaceAChanges).toBeGreaterThan(0);
   expect(workspaceBStates).toEqual([]);
   await Promise.all([observationA.unsubscribe(), observationB.unsubscribe()]);
-}, 45_000);
+}, 30_000);
 
 test("selected workspaces with the same public cwd keep mutations and caches isolated", async () => {
   const {

packages/workspace-helper/src/client.ts

@@ -100,7 +100,8 @@ export function createClient(options: {
         return { ...metadata, chunks } satisfies WorkspaceFileRead;
       },
       async write(input) {
-        const args = ["--path", requireRelativePath(input.path)];
+        const writtenPath = requireRelativePath(input.path);
+        const args = ["--path", writtenPath];
         if (input.expectedModifiedAt) args.push("--expected-modified-at", input.expectedModifiedAt);
         if (input.expectedRevision) args.push("--expected-revision", input.expectedRevision);
         const child = await command("fs-write", args);
@@ -121,7 +122,9 @@ export function createClient(options: {
         }
         const [body, diagnostics, exit] = await Promise.all([stdout, stderr, child.exited]);
         if (exit.code !== 0 || exit.signal !== null) throw helperError(diagnostics, exit);
-        return writeResultSchema.parse(JSON.parse(body));
+        const result = writeResultSchema.parse(JSON.parse(body));
+        if (result.status === "written") notifyWrite(writtenPath);
+        return result;
       },
       async subscribe(input, listener) {
         if (closing) throw new Error("Workspace files client is closed");
@@ -210,6 +213,13 @@ export function createClient(options: {
     };
   }
 
+  function notifyWrite(writtenPath: string): void {
+    for (const subscription of watchSubscriptions.values()) {
+      if (!subscriptionIncludesPath(subscription.input, writtenPath)) continue;
+      subscription.listener({ type: "changed", paths: [writtenPath] });
+    }
+  }
+
   async function requireWatcher(): Promise<WatchState> {
     await watcherRecovery;
     if (watcherStart) return watcherStart;
@@ -531,6 +541,24 @@ function requireRelativePath(value: string): string {
   return value;
 }
 
+function subscriptionIncludesPath(
+  input: LogicalWatchSubscription["input"],
+  changedPath: string,
+): boolean {
+  if (input.ignoredPaths?.some((ignoredPath) => isPathInside(changedPath, ignoredPath))) {
+    return false;
+  }
+  return input.paths.some(
+    (watchedPath) =>
+      changedPath === watchedPath ||
+      (input.recursive === true && isPathInside(changedPath, watchedPath)),
+  );
+}
+
+function isPathInside(candidate: string, directory: string): boolean {
+  return directory === "." || candidate === directory || candidate.startsWith(`${directory}/`);
+}
+
 function onceDrain(stream: NodeJS.WritableStream): Promise<void> {
   return new Promise((resolve, reject) => {
     stream.once("drain", resolve);

packages/workspace-helper/test/workspace-helper.posix.test.ts

@@ -40,6 +40,70 @@ async function collect(chunks: AsyncIterable<Uint8Array>): Promise<Buffer> {
   return Buffer.concat(collected);
 }
 
+describe("workspace-helper cross-platform observation", () => {
+  test("successful client writes notify matching live subscriptions", async () => {
+    const { client, root } = await fixture();
+    await mkdir(path.join(root, "ignored"));
+    await writeFile(path.join(root, "notes.txt"), "before\n");
+    await writeFile(path.join(root, "other.txt"), "other\n");
+    await writeFile(path.join(root, "ignored", "secret.txt"), "secret\n");
+    const exactEvents: WorkspaceWatchEvent[] = [];
+    const recursiveEvents: WorkspaceWatchEvent[] = [];
+    const unrelatedEvents: WorkspaceWatchEvent[] = [];
+    const subscriptions = await Promise.all([
+      client.files.subscribe({ paths: ["notes.txt"] }, (event) => exactEvents.push(event)),
+      client.files.subscribe(
+        { paths: ["."], recursive: true, ignoredPaths: ["ignored"] },
+        (event) => recursiveEvents.push(event),
+      ),
+      client.files.subscribe({ paths: ["other.txt"] }, (event) => unrelatedEvents.push(event)),
+    ]);
+
+    try {
+      await expect(
+        client.files.write({ path: "notes.txt", contents: Buffer.from("after\n") }),
+      ).resolves.toMatchObject({ status: "written" });
+      expect(exactEvents).toContainEqual({ type: "changed", paths: ["notes.txt"] });
+      expect(recursiveEvents).toContainEqual({ type: "changed", paths: ["notes.txt"] });
+      expect(unrelatedEvents).not.toContainEqual({ type: "changed", paths: ["notes.txt"] });
+
+      await expect(
+        client.files.write({
+          path: "ignored/secret.txt",
+          contents: Buffer.from("hidden\n"),
+        }),
+      ).resolves.toMatchObject({ status: "written" });
+      expect(recursiveEvents).not.toContainEqual({
+        type: "changed",
+        paths: ["ignored/secret.txt"],
+      });
+    } finally {
+      await Promise.all(subscriptions.map((subscription) => subscription.unsubscribe()));
+      await client.close();
+    }
+  });
+
+  test("direct filesystem writes still reach the native watcher", async () => {
+    const { client, root } = await fixture();
+    await writeFile(path.join(root, "watched.txt"), "before\n");
+    let resolveChanged!: (paths: string[]) => void;
+    const changed = new Promise<string[]>((resolve) => {
+      resolveChanged = resolve;
+    });
+    const subscription = await client.files.subscribe({ paths: ["watched.txt"] }, (event) => {
+      if (event.type === "changed") resolveChanged(event.paths);
+    });
+
+    try {
+      await writeFile(path.join(root, "watched.txt"), "after\n");
+      await expect(eventWithin(changed, "native watch change")).resolves.toEqual(["watched.txt"]);
+    } finally {
+      await subscription.unsubscribe();
+      await client.close();
+    }
+  });
+});
+
 posixDescribe("workspace-helper public capability", () => {
   test("executable describe reports the exact versioned capability contract", async () => {
     const child = spawn(process.execPath, [workspaceHelper, "describe"], {
@@ -456,3 +520,15 @@ function childProcess(child: ReturnType<typeof spawn>): WorkspaceHelperProcess {
     kill: (signal) => child.kill(signal),
   };
 }
+
+function eventWithin<T>(event: Promise<T>, label: string, timeoutMs = 5_000): Promise<T> {
+  let timer: ReturnType<typeof setTimeout> | undefined;
+  return Promise.race([
+    event,
+    new Promise<never>((_resolve, reject) => {
+      timer = setTimeout(() => reject(new Error(`Timed out waiting for ${label}`)), timeoutMs);
+    }),
+  ]).finally(() => {
+    if (timer) clearTimeout(timer);
+  });
+}