diff --git a/packages/server/src/server/workspace-git-service.runtime.integration.test.ts b/packages/server/src/server/workspace-git-service.runtime.integration.test.ts index 795e0258a5..7b304e4936 100644 --- a/packages/server/src/server/workspace-git-service.runtime.integration.test.ts +++ b/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 { diff --git a/packages/workspace-helper/src/client.ts b/packages/workspace-helper/src/client.ts index 3ca354e72a..9a701ef1f1 100644 --- a/packages/workspace-helper/src/client.ts +++ b/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 { 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 { return new Promise((resolve, reject) => { stream.once("drain", resolve); diff --git a/packages/workspace-helper/test/workspace-helper.posix.test.ts b/packages/workspace-helper/test/workspace-helper.posix.test.ts index 3513b412eb..c6809cd9b7 100644 --- a/packages/workspace-helper/test/workspace-helper.posix.test.ts +++ b/packages/workspace-helper/test/workspace-helper.posix.test.ts @@ -40,6 +40,70 @@ async function collect(chunks: AsyncIterable): Promise { 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((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): WorkspaceHelperProcess { kill: (signal) => child.kill(signal), }; } + +function eventWithin(event: Promise, label: string, timeoutMs = 5_000): Promise { + let timer: ReturnType | undefined; + return Promise.race([ + event, + new Promise((_resolve, reject) => { + timer = setTimeout(() => reject(new Error(`Timed out waiting for ${label}`)), timeoutMs); + }), + ]).finally(() => { + if (timer) clearTimeout(timer); + }); +}