fix(files): refresh the open file when it changes on disk

apps/server/src/auth/RpcAuthorization.ts

@@ -78,6 +78,8 @@ export const RPC_REQUIRED_SCOPES = {
   [WS_METHODS.sourceControlPublishRepository]: AuthOrchestrationOperateScope,
   [WS_METHODS.projectsListEntries]: AuthOrchestrationReadScope,
   [WS_METHODS.projectsReadFile]: AuthOrchestrationReadScope,
+  // Watching a file you are already allowed to read is still reading.
+  [WS_METHODS.subscribeProjectFileChanges]: AuthOrchestrationReadScope,
   [WS_METHODS.projectsSearchContents]: AuthOrchestrationReadScope,
   [WS_METHODS.projectsSearchEntries]: AuthOrchestrationReadScope,
   [WS_METHODS.projectsWriteFile]: AuthOrchestrationOperateScope,

apps/server/src/server.test.ts

@@ -4784,6 +4784,53 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
     }).pipe(Effect.provide(NodeHttpServer.layerTest), TestClock.withLive),
   );
 
+  it.effect("routes websocket rpc subscribeProjectFileChanges for edits made outside the app", () =>
+    Effect.gen(function* () {
+      const fs = yield* FileSystem.FileSystem;
+      const path = yield* Path.Path;
+      const workspaceDir = yield* fs.makeTempDirectoryScoped({ prefix: "t3-ws-project-watch-" });
+      const filePath = path.join(workspaceDir, "package.json");
+      const before = '{ "description": "An awesome horse platform" }\n';
+      const after = '{ "description": "An awesome course platform" }\n';
+      yield* fs.writeFileString(filePath, before);
+
+      yield* buildAppUnderTest();
+
+      const wsUrl = yield* getWsServerUrl("/ws");
+      const result = yield* Effect.scoped(
+        withWsRpcClient(wsUrl, (client) =>
+          Effect.gen(function* () {
+            const watcher = yield* client[WS_METHODS.subscribeProjectFileChanges]({
+              cwd: workspaceDir,
+              relativePath: "package.json",
+            }).pipe(Stream.runHead, Effect.forkChild());
+            // `fs.watch` registration is not observable, so rewrite until the
+            // subscription reports rather than racing a fixed startup delay.
+            // The pause outlasts the watcher's debounce window.
+            const writer = yield* fs
+              .writeFileString(filePath, after)
+              .pipe(
+                Effect.orDie,
+                Effect.delay(Duration.millis(300)),
+                Effect.forever,
+                Effect.forkChild(),
+              );
+            const event = yield* Fiber.join(watcher).pipe(Effect.timeout(Duration.seconds(10)));
+            yield* Fiber.interrupt(writer);
+            const file = yield* client[WS_METHODS.projectsReadFile]({
+              cwd: workspaceDir,
+              relativePath: "package.json",
+            });
+            return { event, file };
+          }),
+        ),
+      );
+
+      assert.deepEqual(result.event, Option.some({ relativePath: "package.json" }));
+      assert.equal(result.file.contents, after);
+    }).pipe(Effect.provide(NodeHttpServer.layerTest), TestClock.withLive),
+  );
+
   it.effect("routes websocket rpc projects.searchEntries excludes gitignored files", () =>
     Effect.gen(function* () {
       const fs = yield* FileSystem.FileSystem;

apps/server/src/workspace/WorkspaceFileSystem.test.ts

@@ -1,9 +1,13 @@
 import * as NodeServices from "@effect/platform-node/NodeServices";
 import { it, describe, expect } from "@effect/vitest";
+import * as Duration from "effect/Duration";
 import * as Effect from "effect/Effect";
+import * as Fiber from "effect/Fiber";
 import * as FileSystem from "effect/FileSystem";
 import * as Layer from "effect/Layer";
+import * as Option from "effect/Option";
 import * as Path from "effect/Path";
+import * as Stream from "effect/Stream";
 
 import * as ServerConfig from "../config.ts";
 import * as VcsDriverRegistry from "../vcs/VcsDriverRegistry.ts";
@@ -265,4 +269,85 @@ it.layer(TestLayer, { excludeTestServices: true })("WorkspaceFileSystemLive", (i
       }),
     );
   });
+
+  describe("watchFile", () => {
+    /**
+     * `fs.watch` registration is not observable, so the writer keeps rewriting
+     * until the watcher reports instead of racing a fixed startup delay. It
+     * pauses longer than the debounce window between rewrites so the stream
+     * gets the quiet period it waits for.
+     */
+    const awaitFirstChange = Effect.fn("awaitFirstChange")(function* (
+      cwd: string,
+      relativePath: string,
+      rewrite: Effect.Effect<void, never, FileSystem.FileSystem | Path.Path>,
+    ) {
+      const workspaceFileSystem = yield* WorkspaceFileSystem.WorkspaceFileSystem;
+      const watcher = yield* workspaceFileSystem
+        .watchFile({ cwd, relativePath })
+        .pipe(Stream.runHead, Effect.forkChild());
+      const writer = yield* rewrite.pipe(
+        Effect.delay(Duration.millis(300)),
+        Effect.forever,
+        Effect.forkChild(),
+      );
+      const event = yield* Fiber.join(watcher).pipe(Effect.timeout(Duration.seconds(10)));
+      yield* Fiber.interrupt(writer);
+      return event;
+    });
+
+    it.effect("reports a plain on-disk write", () =>
+      Effect.gen(function* () {
+        const cwd = yield* makeTempDir;
+        yield* writeTextFile(
+          cwd,
+          "package.json",
+          '{ "description": "An awesome horse platform" }\n',
+        );
+
+        const event = yield* awaitFirstChange(
+          cwd,
+          "package.json",
+          writeTextFile(cwd, "package.json", '{ "description": "An awesome course platform" }\n'),
+        );
+
+        expect(event).toEqual(Option.some({ relativePath: "package.json" }));
+      }),
+    );
+
+    it.effect("reports an atomic replace that swaps the file's inode", () =>
+      Effect.gen(function* () {
+        const fileSystem = yield* FileSystem.FileSystem;
+        const path = yield* Path.Path;
+        const cwd = yield* makeTempDir;
+        yield* writeTextFile(cwd, "src/index.ts", "export const answer = 42;\n");
+
+        const target = path.join(cwd, "src/index.ts");
+        const temp = path.join(cwd, "src/index.ts.tmp");
+        const replace = Effect.gen(function* () {
+          yield* fileSystem.writeFileString(temp, "export const answer = 43;\n").pipe(Effect.orDie);
+          yield* fileSystem.rename(temp, target).pipe(Effect.orDie);
+        });
+
+        const event = yield* awaitFirstChange(cwd, "src/index.ts", replace);
+
+        expect(event).toEqual(Option.some({ relativePath: "src/index.ts" }));
+      }),
+    );
+
+    it.effect("rejects watches outside the workspace root", () =>
+      Effect.gen(function* () {
+        const workspaceFileSystem = yield* WorkspaceFileSystem.WorkspaceFileSystem;
+        const cwd = yield* makeTempDir;
+
+        const error = yield* workspaceFileSystem
+          .watchFile({ cwd, relativePath: "../escape.md" })
+          .pipe(Stream.runHead, Effect.flip);
+
+        expect(error.message).toContain(
+          "Workspace file path must be relative to the project root: ../escape.md",
+        );
+      }),
+    );
+  });
 });

apps/server/src/workspace/WorkspaceFileSystem.ts

@@ -10,17 +10,20 @@
 import * as NodeFSP from "node:fs/promises";
 
 import type {
+  ProjectFileChangedEvent,
   ProjectReadFileInput,
   ProjectReadFileResult,
   ProjectWriteFileInput,
   ProjectWriteFileResult,
 } from "@t3tools/contracts";
 import * as Context from "effect/Context";
+import * as Duration from "effect/Duration";
 import * as Effect from "effect/Effect";
 import * as FileSystem from "effect/FileSystem";
 import * as Layer from "effect/Layer";
 import * as Path from "effect/Path";
 import * as Schema from "effect/Schema";
+import * as Stream from "effect/Stream";
 
 import * as WorkspaceEntries from "./WorkspaceEntries.ts";
 import * as WorkspacePaths from "./WorkspacePaths.ts";
@@ -43,6 +46,7 @@ export class WorkspaceFileSystemOperationError extends Schema.TaggedErrorClass<W
       "close",
       "make-directory",
       "write-file",
+      "watch",
     ]),
     cause: Schema.Defect(),
   },
@@ -123,6 +127,19 @@ export class WorkspaceFileSystem extends Context.Service<
       ProjectWriteFileResult,
       WorkspaceFileSystemError | WorkspacePaths.WorkspacePathOutsideRootError
     >;
+    /**
+     * Emit a change event every time the file changes on disk.
+     *
+     * The stream is a signal only: subscribers re-read through `readFile`, so
+     * every size, binary and error rule stays in one place. Events are
+     * debounced because a single save is several `fs.watch` events.
+     */
+    readonly watchFile: (
+      input: ProjectReadFileInput,
+    ) => Stream.Stream<
+      ProjectFileChangedEvent,
+      WorkspaceFileSystemError | WorkspacePaths.WorkspacePathOutsideRootError
+    >;
   }
 >()("t3/workspace/WorkspaceFileSystem") {}
 
@@ -297,7 +314,47 @@ export const make = Effect.gen(function* () {
     return { relativePath: target.relativePath };
   });
 
-  return WorkspaceFileSystem.of({ readFile, writeFile });
+  /**
+   * Watches the containing directory rather than the file itself: saves that
+   * land as rename-over-temp (git, most editors, atomic writers) replace the
+   * inode, and a file-level watch would follow the discarded one.
+   */
+  const watchFile: WorkspaceFileSystem["Service"]["watchFile"] = (input) =>
+    Stream.unwrap(
+      Effect.gen(function* () {
+        const target = yield* workspacePaths.resolveRelativePathWithinRoot({
+          workspaceRoot: input.cwd,
+          relativePath: input.relativePath,
+        });
+        const directory = path.dirname(target.absolutePath);
+        const fileName = path.basename(target.absolutePath);
+
+        return fileSystem.watch(directory).pipe(
+          Stream.filter(
+            (event) =>
+              event.path === fileName ||
+              event.path === target.absolutePath ||
+              path.resolve(directory, event.path) === target.absolutePath,
+          ),
+          // Debounce so the file is fully written before subscribers re-read it.
+          Stream.debounce(Duration.millis(100)),
+          Stream.map(() => ({ relativePath: target.relativePath })),
+          Stream.mapError(
+            (cause) =>
+              new WorkspaceFileSystemOperationError({
+                workspaceRoot: input.cwd,
+                relativePath: input.relativePath,
+                resolvedPath: target.absolutePath,
+                operationPath: directory,
+                operation: "watch",
+                cause,
+              }),
+          ),
+        );
+      }),
+    );
+
+  return WorkspaceFileSystem.of({ readFile, writeFile, watchFile });
 });
 
 export const layer = Layer.effect(WorkspaceFileSystem, make);

apps/server/src/ws.ts

@@ -1899,6 +1899,21 @@ const makeWsRpcLayer = (
             ),
             { "rpc.aggregate": "workspace" },
           ),
+        [WS_METHODS.subscribeProjectFileChanges]: (input) =>
+          observeRpcStream(
+            WS_METHODS.subscribeProjectFileChanges,
+            workspaceFileSystem.watchFile(input).pipe(
+              Stream.mapError(
+                (cause) =>
+                  new ProjectReadFileError({
+                    ...input,
+                    ...projectFileFailureContext(cause),
+                    cause,
+                  }),
+              ),
+            ),
+            { "rpc.aggregate": "workspace" },
+          ),
         [WS_METHODS.projectsWriteFile]: (input) =>
           observeRpcEffect(
             WS_METHODS.projectsWriteFile,

packages/client-runtime/src/rpc/client.ts

@@ -52,6 +52,7 @@ export type EnvironmentSubscriptionRpcTag =
   | typeof WS_METHODS.subscribeResourceTelemetry
   | typeof WS_METHODS.previewAutomationConnect
   | typeof WS_METHODS.subscribeVcsStatus
+  | typeof WS_METHODS.subscribeProjectFileChanges
   | typeof WS_METHODS.terminalAttach;
 
 export type EnvironmentStreamCommandRpcTag =

packages/client-runtime/src/state/projectCommands.ts

@@ -8,6 +8,7 @@ import {
   createEnvironmentRpcCommand,
   createEnvironmentRpcQueryAtomFamily,
 } from "./runtime.ts";
+import { subscribe } from "../rpc/client.ts";
 import {
   type CreateProjectInput,
   type DeleteProjectInput,
@@ -66,11 +67,16 @@ export function createProjectEnvironmentAtoms<R, E>(
       staleTimeMs: 30_000,
       idleTtlMs: 5 * 60_000,
     }),
+    // The server watches the open file and says when it moved, so an agent (or
+    // anything else) editing on disk lands in the viewer without polling and
+    // without waiting for a focus change. Web, desktop and mobile all read this
+    // atom, so they all stop serving stale contents together.
     readFile: createEnvironmentRpcQueryAtomFamily(runtime, {
       label: "environment-data:projects:read-file",
       tag: WS_METHODS.projectsReadFile,
       staleTimeMs: 30_000,
       idleTtlMs: 5 * 60_000,
+      invalidate: (input) => subscribe(WS_METHODS.subscribeProjectFileChanges, input),
     }),
     optimisticFile: (target: OptimisticProjectFileTarget) =>
       optimisticFileFamily(optimisticProjectFileKey(target)),