feat(files): refresh the files tree when the app changes the workspace

apps/server/src/auth/RpcAuthorization.ts

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

apps/server/src/server.test.ts

@@ -5150,6 +5150,49 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
     }).pipe(Effect.provide(NodeHttpServer.layerTest), TestClock.withLive),
   );
 
+  it.effect(
+    "routes websocket rpc subscribeProjectEntryChanges for writes made through the app",
+    () =>
+      Effect.gen(function* () {
+        const fs = yield* FileSystem.FileSystem;
+        const path = yield* Path.Path;
+        const workspaceDir = yield* fs.makeTempDirectoryScoped({
+          prefix: "t3-ws-project-entries-watch-",
+        });
+        yield* fs.writeFileString(path.join(workspaceDir, "README.md"), "# workspace\n");
+
+        yield* buildAppUnderTest();
+
+        const wsUrl = yield* getWsServerUrl("/ws");
+        const result = yield* Effect.scoped(
+          withWsRpcClient(wsUrl, (client) =>
+            Effect.gen(function* () {
+              const watcher = yield* client[WS_METHODS.subscribeProjectEntryChanges]({
+                cwd: workspaceDir,
+              }).pipe(Stream.runHead, Effect.forkChild());
+              // A completed round trip on the same socket proves the server has
+              // read the subscription frame ahead of the write, so the signal
+              // that write raises cannot arrive before the watcher is
+              // listening. It also primes the index, so the write refreshes a
+              // live listing rather than building one from cold.
+              yield* client[WS_METHODS.projectsListEntries]({ cwd: workspaceDir });
+              yield* client[WS_METHODS.projectsWriteFile]({
+                cwd: workspaceDir,
+                relativePath: "src/created.ts",
+                contents: "export const created = true;\n",
+              });
+              const event = yield* Fiber.join(watcher).pipe(Effect.timeout(Duration.seconds(10)));
+              const listing = yield* client[WS_METHODS.projectsListEntries]({ cwd: workspaceDir });
+              return { event, listing };
+            }),
+          ),
+        );
+
+        assert.deepEqual(result.event, Option.some({}));
+        assert.isTrue(result.listing.entries.some((entry) => entry.path === "src/created.ts"));
+      }).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;

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

@@ -7,7 +7,9 @@ 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 Option from "effect/Option";
 import * as PlatformError from "effect/PlatformError";
+import * as Stream from "effect/Stream";
 import { vi } from "vite-plus/test";
 
 import * as ServerConfig from "../config.ts";
@@ -123,6 +125,34 @@ it.layer(TestLayer, { excludeTestServices: true })("WorkspaceEntries", (it) => {
     );
   });
 
+  describe("watchEntries", () => {
+    it.effect("signals only the workspace root whose index refreshed", () =>
+      Effect.gen(function* () {
+        const workspaceEntries = yield* WorkspaceEntries.WorkspaceEntries;
+        const cwd = yield* makeTempDir({ prefix: "t3code-workspace-watch-entries-" });
+        const otherCwd = yield* makeTempDir({ prefix: "t3code-workspace-watch-entries-other-" });
+
+        // Awaiting the acquisitions attaches both subscriptions, so neither
+        // refresh below can land before its watcher is listening.
+        const watched = yield* workspaceEntries.watchEntries({ cwd });
+        const other = yield* workspaceEntries.watchEntries({ cwd: otherCwd });
+
+        yield* workspaceEntries.refresh(otherCwd);
+        expect(
+          yield* Effect.raceFirst(
+            // `watched` is raced first deliberately: were it to ignore the key
+            // it would have a signal waiting and would win outright.
+            Stream.runHead(watched).pipe(Effect.as("watched")),
+            Stream.runHead(other).pipe(Effect.as("other")),
+          ),
+        ).toBe("other");
+
+        yield* workspaceEntries.refresh(cwd);
+        expect(yield* Stream.runHead(watched)).toEqual(Option.some({}));
+      }).pipe(Effect.scoped),
+    );
+  });
+
   describe("search", () => {
     it.effect("returns files and directories relative to cwd", () =>
       Effect.gen(function* () {

apps/server/src/workspace/WorkspaceEntries.ts

@@ -6,12 +6,16 @@ import * as Context from "effect/Context";
 import * as Effect from "effect/Effect";
 import * as Layer from "effect/Layer";
 import * as Path from "effect/Path";
+import * as PubSub from "effect/PubSub";
 import * as RcMap from "effect/RcMap";
 import * as Schema from "effect/Schema";
+import type * as Scope from "effect/Scope";
+import * as Stream from "effect/Stream";
 
 import type {
   FilesystemBrowseInput,
   FilesystemBrowseResult,
+  ProjectEntriesChangedEvent,
   ProjectListEntriesInput,
   ProjectListEntriesResult,
   ProjectSearchContentsInput,
@@ -100,6 +104,17 @@ export class WorkspaceEntries extends Context.Service<
       input: ProjectSearchContentsInput,
     ) => Effect.Effect<ProjectSearchContentsResult, WorkspaceEntriesError>;
     readonly refresh: (cwd: string) => Effect.Effect<void>;
+    /**
+     * Emits once per `refresh` of the same workspace root. It is a signal, not
+     * a payload: subscribers re-read through `list`.
+     *
+     * Acquiring the subscription is the effect and consuming it is the stream,
+     * so a caller that has awaited this call is already listening and cannot
+     * miss a refresh it goes on to trigger itself.
+     */
+    readonly watchEntries: (
+      input: ProjectListEntriesInput,
+    ) => Effect.Effect<Stream.Stream<ProjectEntriesChangedEvent>, never, Scope.Scope>;
   }
 >()("t3/workspace/WorkspaceEntries") {}
 
@@ -149,11 +164,25 @@ export const make = Effect.gen(function* () {
     return yield* workspacePaths.normalizeWorkspaceRoot(cwd);
   });
 
+  /**
+   * The key both sides of `entryChanges` agree on. It falls back to the raw
+   * `cwd` so a root that cannot be normalised still pairs its publishes with
+   * its subscribers instead of silently going quiet.
+   */
+  const workspaceRootKey = (cwd: string) =>
+    normalizeWorkspaceRoot(cwd).pipe(Effect.orElseSucceed(() => cwd));
+
+  /**
+   * Unbounded rather than sliding: several workspace roots share this channel,
+   * and a sliding buffer would let a busy root evict a quiet one's signal from
+   * a subscriber that only cares about the quiet one. Every subscriber is a
+   * live stream that drains as it goes, and its scope closes with its socket.
+   */
+  const entryChanges = yield* PubSub.unbounded<string>();
+
   const refresh: WorkspaceEntries["Service"]["refresh"] = Effect.fn("WorkspaceEntries.refresh")(
     function* (cwd) {
-      const normalizedCwd = yield* normalizeWorkspaceRoot(cwd).pipe(
-        Effect.orElseSucceed(() => cwd),
-      );
+      const normalizedCwd = yield* workspaceRootKey(cwd);
       for (const variant of WorkspaceSearchIndex.WORKSPACE_SEARCH_INDEX_VARIANTS) {
         const indexKey = WorkspaceSearchIndex.workspaceSearchIndexKey(normalizedCwd, variant);
         if (!(yield* RcMap.has(workspaceSearchIndexes.rcMap, indexKey))) {
@@ -185,9 +214,25 @@ export const make = Effect.gen(function* () {
           }),
         );
       }
+      // After the indexes, never before: a subscriber that re-lists on this
+      // signal must see the refreshed listing, not the one it replaced.
+      yield* PubSub.publish(entryChanges, normalizedCwd);
     },
   );
 
+  const watchEntries: WorkspaceEntries["Service"]["watchEntries"] = Effect.fn(
+    "WorkspaceEntries.watchEntries",
+  )(function* (input) {
+    // Subscribe before resolving the key, so a refresh landing while the root
+    // is being normalised is buffered rather than dropped.
+    const subscription = yield* PubSub.subscribe(entryChanges);
+    const normalizedCwd = yield* workspaceRootKey(input.cwd);
+    return Stream.fromSubscription(subscription).pipe(
+      Stream.filter((changedCwd) => changedCwd === normalizedCwd),
+      Stream.map(() => ({})),
+    );
+  });
+
   const browse: WorkspaceEntries["Service"]["browse"] = Effect.fn("WorkspaceEntries.browse")(
     function* (input) {
       const resolvedInputPath = yield* resolveBrowseTarget(input, path);
@@ -288,7 +333,7 @@ export const make = Effect.gen(function* () {
     },
   );
 
-  return WorkspaceEntries.of({ browse, list, refresh, search, searchContents });
+  return WorkspaceEntries.of({ browse, list, refresh, search, searchContents, watchEntries });
 });
 
 export const layer = Layer.effect(WorkspaceEntries, make).pipe(

apps/server/src/ws.ts

@@ -1996,6 +1996,12 @@ const makeWsRpcLayer = (
             ),
             { "rpc.aggregate": "workspace" },
           ),
+        [WS_METHODS.subscribeProjectEntryChanges]: (input) =>
+          observeRpcStreamEffect(
+            WS_METHODS.subscribeProjectEntryChanges,
+            workspaceEntries.watchEntries(input),
+            { "rpc.aggregate": "workspace" },
+          ),
         [WS_METHODS.projectsReadFile]: (input) =>
           observeRpcEffect(
             WS_METHODS.projectsReadFile,

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

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

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

@@ -61,11 +61,15 @@ export function createProjectEnvironmentAtoms<R, E>(
       tag: WS_METHODS.projectsSearchEntries,
       staleTimeMs: 15_000,
     }),
+    // The server says when its own entry index moved, so files an agent creates
+    // or deletes during a turn land in the tree without waiting out the stale
+    // window. Changes made outside the app still need a remount.
     listEntries: createEnvironmentRpcQueryAtomFamily(runtime, {
       label: "environment-data:projects:list-entries",
       tag: WS_METHODS.projectsListEntries,
       staleTimeMs: 30_000,
       idleTtlMs: 5 * 60_000,
+      invalidate: (input) => subscribe(WS_METHODS.subscribeProjectEntryChanges, input),
     }),
     // 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

packages/contracts/src/project.ts

@@ -81,6 +81,15 @@ export const ProjectListEntriesResult = Schema.Struct({
 });
 export type ProjectListEntriesResult = typeof ProjectListEntriesResult.Type;
 
+/**
+ * Emitted by `subscribeProjectEntryChanges` whenever the server's entry index
+ * for a workspace is refreshed. Like `ProjectFileChangedEvent` it carries no
+ * payload: clients re-read the listing through `projects.listEntries` so
+ * limits, ordering and error mapping keep living in one place.
+ */
+export const ProjectEntriesChangedEvent = Schema.Struct({});
+export type ProjectEntriesChangedEvent = typeof ProjectEntriesChangedEvent.Type;
+
 export const ProjectEntriesFailure = Schema.Literals([
   "workspace_root_not_found",
   "workspace_root_create_failed",

packages/contracts/src/rpc.ts

@@ -112,6 +112,7 @@ import {
   RelayClientStatusSchema,
 } from "./relayClient.ts";
 import {
+  ProjectEntriesChangedEvent,
   ProjectFileChangedEvent,
   ProjectListEntriesError,
   ProjectListEntriesInput,
@@ -324,6 +325,7 @@ export const WS_METHODS = {
   // Streaming subscriptions
   subscribeVcsStatus: "subscribeVcsStatus",
   subscribeProjectFileChanges: "subscribeProjectFileChanges",
+  subscribeProjectEntryChanges: "subscribeProjectEntryChanges",
   subscribeTerminalEvents: "subscribeTerminalEvents",
   subscribeTerminalMetadata: "subscribeTerminalMetadata",
   subscribePreviewEvents: "subscribePreviewEvents",
@@ -656,6 +658,19 @@ export const WsProjectsListEntriesRpc = Rpc.make(WS_METHODS.projectsListEntries,
   error: Schema.Union([ProjectListEntriesError, EnvironmentAuthorizationError]),
 });
 
+/**
+ * Says when one workspace's entry list moved, so an open file tree re-lists
+ * instead of waiting out its staleness window. The signal originates from the
+ * server's own index refresh, so it covers changes made through the app -
+ * including an agent's - and nothing else.
+ */
+export const WsSubscribeProjectEntryChangesRpc = Rpc.make(WS_METHODS.subscribeProjectEntryChanges, {
+  payload: ProjectListEntriesInput,
+  success: ProjectEntriesChangedEvent,
+  error: EnvironmentAuthorizationError,
+  stream: true,
+});
+
 export const WsProjectsReadFileRpc = Rpc.make(WS_METHODS.projectsReadFile, {
   payload: ProjectReadFileInput,
   success: ProjectReadFileResult,
@@ -1075,6 +1090,7 @@ export const WsRpcGroup = RpcGroup.make(
   WsSourceControlCloneRepositoryRpc,
   WsSourceControlPublishRepositoryRpc,
   WsProjectsListEntriesRpc,
+  WsSubscribeProjectEntryChangesRpc,
   WsProjectsReadFileRpc,
   WsSubscribeProjectFileChangesRpc,
   WsProjectsSearchContentsRpc,