diff --git a/apps/server/src/auth/RpcAuthorization.ts b/apps/server/src/auth/RpcAuthorization.ts index 6b35f0d54e18..ade44837e327 100644 --- a/apps/server/src/auth/RpcAuthorization.ts +++ b/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, diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index de3f5101f53e..f77bb6fd8485 100644 --- a/apps/server/src/server.test.ts +++ b/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; diff --git a/apps/server/src/workspace/WorkspaceFileSystem.test.ts b/apps/server/src/workspace/WorkspaceFileSystem.test.ts index cecffbc1993d..ac2d77dbe29a 100644 --- a/apps/server/src/workspace/WorkspaceFileSystem.test.ts +++ b/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,177 @@ 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, + ) { + 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("reports a file that does not exist yet when it appears", () => + Effect.gen(function* () { + const cwd = yield* makeTempDir; + // No file at this path: the watcher must still attach, so a tab whose + // file is momentarily absent keeps its live refresh. + const event = yield* awaitFirstChange( + cwd, + "later.txt", + writeTextFile(cwd, "later.txt", "created\n"), + ); + + expect(event).toEqual(Option.some({ relativePath: "later.txt" })); + }), + ); + + it.effect("follows a symlinked file to edits made through its target", () => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const cwd = yield* makeTempDir; + yield* writeTextFile(cwd, "real/config.json", '{ "value": "before" }\n'); + yield* fileSystem + .symlink(path.join(cwd, "real", "config.json"), path.join(cwd, "alias.json")) + .pipe(Effect.orDie); + + // Watching through the alias must see a write to the canonical path, + // which lives in a different directory under a different name. + // + // This discriminates on inotify platforms, where a watch on the alias's + // own directory would never hear about the target. It does not on + // macOS: FSEvents reports the alias entry as touched when its target is + // written, so this passes there even with the parent-only resolve. + const event = yield* awaitFirstChange( + cwd, + "alias.json", + writeTextFile(cwd, "real/config.json", '{ "value": "after" }\n'), + ); + + expect(event).toEqual(Option.some({ relativePath: "alias.json" })); + }), + ); + + it.effect("reports an atomic replace of the symlink's own path", () => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const cwd = yield* makeTempDir; + yield* writeTextFile(cwd, "real/config.json", '{ "value": "before" }\n'); + const alias = path.join(cwd, "alias.json"); + yield* fileSystem.symlink(path.join(cwd, "real", "config.json"), alias).pipe(Effect.orDie); + + // The editor/git pattern: write a temp file beside the alias and rename + // over it. That fires in the alias's directory, not the target's, so a + // watch that only followed the canonical path would miss it. + // + // Like the sibling symlink test, this discriminates on inotify and not + // on macOS, where FSEvents coalesces to directory granularity and + // delivers a matching event either way. + const temp = path.join(cwd, "alias.json.tmp"); + const replaceAlias = Effect.gen(function* () { + yield* fileSystem.writeFileString(temp, '{ "value": "after" }\n').pipe(Effect.orDie); + yield* fileSystem.rename(temp, alias).pipe(Effect.orDie); + }); + + const event = yield* awaitFirstChange(cwd, "alias.json", replaceAlias); + + expect(event).toEqual(Option.some({ relativePath: "alias.json" })); + }), + ); + + it.effect("rejects a watch that escapes the workspace through a symlink", () => + Effect.gen(function* () { + const workspaceFileSystem = yield* WorkspaceFileSystem.WorkspaceFileSystem; + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const cwd = yield* makeTempDir; + const outside = yield* makeTempDir; + yield* fileSystem + .writeFileString(path.join(outside, "secret.txt"), "secret\n") + .pipe(Effect.orDie); + // Lexically `link/secret.txt` sits inside the workspace; physically it + // does not, so no watcher may be pointed at it. + yield* fileSystem.symlink(outside, path.join(cwd, "link")).pipe(Effect.orDie); + + const error = yield* workspaceFileSystem + .watchFile({ cwd, relativePath: "link/secret.txt" }) + .pipe(Stream.runHead, Effect.flip); + + expect(error._tag).toBe("WorkspaceFilePathEscapeError"); + }), + ); + + 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", + ); + }), + ); + }); }); diff --git a/apps/server/src/workspace/WorkspaceFileSystem.ts b/apps/server/src/workspace/WorkspaceFileSystem.ts index e2dc9cbbb390..3612757495d5 100644 --- a/apps/server/src/workspace/WorkspaceFileSystem.ts +++ b/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; + /** + * 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") {} @@ -132,39 +149,42 @@ export const make = Effect.gen(function* () { const workspacePaths = yield* WorkspacePaths.WorkspacePaths; const workspaceEntries = yield* WorkspaceEntries.WorkspaceEntries; - const readFile: WorkspaceFileSystem["Service"]["readFile"] = Effect.fn( - "WorkspaceFileSystem.readFile", - )(function* (input) { - const target = yield* workspacePaths.resolveRelativePathWithinRoot({ - workspaceRoot: input.cwd, - relativePath: input.relativePath, - }); - + /** + * Resolve one absolute path with `realpath` and prove it sits inside the + * workspace root. Lexical containment is not enough: a symlink inside the + * workspace can point anywhere, so both ends are resolved physically and + * compared. Callers pick what they need contained — `readFile` contains the + * file it opens, `watchFile` the directory it watches. + */ + const realPathWithinRoot = Effect.fn("WorkspaceFileSystem.realPathWithinRoot")(function* ( + input: ProjectReadFileInput, + absolutePath: string, + ) { const realWorkspaceRoot = yield* Effect.tryPromise({ try: () => NodeFSP.realpath(input.cwd), catch: (cause) => new WorkspaceFileSystemOperationError({ workspaceRoot: input.cwd, relativePath: input.relativePath, - resolvedPath: target.absolutePath, + resolvedPath: absolutePath, operationPath: input.cwd, operation: "realpath-workspace-root", cause, }), }); - const realTargetPath = yield* Effect.tryPromise({ - try: () => NodeFSP.realpath(target.absolutePath), + const realPath = yield* Effect.tryPromise({ + try: () => NodeFSP.realpath(absolutePath), catch: (cause) => new WorkspaceFileSystemOperationError({ workspaceRoot: input.cwd, relativePath: input.relativePath, - resolvedPath: target.absolutePath, - operationPath: target.absolutePath, + resolvedPath: absolutePath, + operationPath: absolutePath, operation: "realpath-target", cause, }), }); - const relativeRealPath = path.relative(realWorkspaceRoot, realTargetPath); + const relativeRealPath = path.relative(realWorkspaceRoot, realPath); if ( relativeRealPath.startsWith(`..${path.sep}`) || relativeRealPath === ".." || @@ -174,10 +194,22 @@ export const make = Effect.gen(function* () { workspaceRoot: input.cwd, relativePath: input.relativePath, resolvedWorkspaceRoot: realWorkspaceRoot, - resolvedPath: realTargetPath, + resolvedPath: realPath, }); } + return realPath; + }); + + const readFile: WorkspaceFileSystem["Service"]["readFile"] = Effect.fn( + "WorkspaceFileSystem.readFile", + )(function* (input) { + const target = yield* workspacePaths.resolveRelativePathWithinRoot({ + workspaceRoot: input.cwd, + relativePath: input.relativePath, + }); + const realTargetPath = yield* realPathWithinRoot(input, target.absolutePath); + return yield* Effect.acquireUseRelease( Effect.tryPromise({ try: () => NodeFSP.open(realTargetPath, "r"), @@ -297,7 +329,80 @@ 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, + }); + + // A symlinked file can be changed from either end, and the two land in + // different directories: an edit through the target's own path fires + // beside the target, while an atomic rename-over-temp save on the alias + // fires beside the alias. Watch both, and the ordinary case collapses + // back to one watcher because the two coincide. + const aliasDirectory = yield* realPathWithinRoot(input, path.dirname(target.absolutePath)); + const aliasWatch = { + directory: aliasDirectory, + fileName: path.basename(target.absolutePath), + }; + const watched: Array = [aliasWatch]; + + // Missing leaf is not an error: a watch must still attach to a path + // that is momentarily absent, mid atomic-replace or not yet created. + const canonical = yield* Effect.tryPromise({ + try: () => NodeFSP.realpath(target.absolutePath), + catch: () => null, + }).pipe(Effect.orElseSucceed(() => null)); + if (canonical !== null) { + // Contained, not merely resolved: an escaping symlink fails here + // rather than quietly falling back to the alias-only watch. + const canonicalPath = yield* realPathWithinRoot(input, canonical); + const directory = path.dirname(canonicalPath); + const fileName = path.basename(canonicalPath); + if (directory !== aliasWatch.directory || fileName !== aliasWatch.fileName) { + watched.push({ directory, fileName }); + } + } + + const changes = watched.map(({ directory, fileName }) => + fileSystem.watch(directory).pipe( + Stream.filter( + (event) => + event.path === fileName || + path.resolve(directory, event.path) === path.join(directory, fileName), + ), + Stream.mapError( + (cause) => + new WorkspaceFileSystemOperationError({ + workspaceRoot: input.cwd, + relativePath: input.relativePath, + resolvedPath: path.join(directory, fileName), + operationPath: directory, + operation: "watch", + cause, + }), + ), + ), + ); + + const [first, second] = changes; + return (second === undefined ? first! : Stream.merge(first!, second)).pipe( + // Debounce so the file is fully written before subscribers re-read it, + // and so both watchers reporting one save stay a single event. + Stream.debounce(Duration.millis(100)), + Stream.map(() => ({ relativePath: target.relativePath })), + ); + }), + ); + + return WorkspaceFileSystem.of({ readFile, writeFile, watchFile }); }); export const layer = Layer.effect(WorkspaceFileSystem, make); diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index c3caea225704..809809bd15a4 100644 --- a/apps/server/src/ws.ts +++ b/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, diff --git a/packages/client-runtime/src/rpc/client.ts b/packages/client-runtime/src/rpc/client.ts index bfe57a6c0dd5..849273a0d0be 100644 --- a/packages/client-runtime/src/rpc/client.ts +++ b/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 = diff --git a/packages/client-runtime/src/state/projectCommands.ts b/packages/client-runtime/src/state/projectCommands.ts index 3defcc321547..b02a372d54f6 100644 --- a/packages/client-runtime/src/state/projectCommands.ts +++ b/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( 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)), diff --git a/packages/client-runtime/src/state/runtime.test.ts b/packages/client-runtime/src/state/runtime.test.ts index f36087ebf66a..c7c8f772c438 100644 --- a/packages/client-runtime/src/state/runtime.test.ts +++ b/packages/client-runtime/src/state/runtime.test.ts @@ -12,7 +12,9 @@ import { AsyncResult, Atom, AtomRegistry } from "effect/unstable/reactivity"; import { environmentRpcKey, createAtomCommandScheduler, + createInvalidatableQueryAtom, createRuntimeCommand, + createStreamRevisionAtom, scheduleAtomCommandEffect, executeAtomCommand, executeAtomQuery, @@ -498,3 +500,93 @@ describe("runtime command runner", () => { registry.dispose(); }); }); + +describe("createInvalidatableQueryAtom", () => { + it("refetches when the revision changes, and only then, inside the stale window", async () => { + const runtime = Atom.runtime(Layer.empty); + const revisions = Atom.make(0); + let reads = 0; + const query = createInvalidatableQueryAtom(runtime, { + label: "test.invalidatable", + staleTimeMs: 30_000, + idleTtlMs: 60_000, + revisions, + execute: () => Effect.sync(() => (reads += 1)), + }); + const registry = AtomRegistry.make(); + + expect(await executeAtomQuery(registry, query)).toMatchObject({ + _tag: "Success", + value: 1, + }); + + // Fresh within `staleTime`: reading again must not hit the source. + expect(await executeAtomQuery(registry, query)).toMatchObject({ + _tag: "Success", + value: 1, + }); + expect(reads).toBe(1); + + // A revision bump means the underlying data moved, so the stale window + // stops applying — this is the stale-file-viewer fix. + registry.set(revisions, 1); + + expect(await executeAtomQuery(registry, query)).toMatchObject({ + _tag: "Success", + value: 2, + }); + expect(reads).toBe(2); + registry.dispose(); + }); + + it("keeps serving the stale window when no revisions are supplied", async () => { + const runtime = Atom.runtime(Layer.empty); + let reads = 0; + const query = createInvalidatableQueryAtom(runtime, { + label: "test.plain", + staleTimeMs: 30_000, + idleTtlMs: 60_000, + execute: () => Effect.sync(() => (reads += 1)), + }); + const registry = AtomRegistry.make(); + + await executeAtomQuery(registry, query); + await executeAtomQuery(registry, query); + + expect(reads).toBe(1); + registry.dispose(); + }); +}); + +describe("createStreamRevisionAtom", () => { + it("counts emissions and ignores everything else", async () => { + const runtime = Atom.runtime(Layer.empty); + const latch = Latch.makeUnsafe(); + const revisions = createStreamRevisionAtom( + runtime, + Stream.fromArray(["changed", "changed"]).pipe(Stream.concat(Stream.fromEffect(latch.await))), + { label: "test.revisions", idleTtlMs: 60_000 }, + ); + const registry = AtomRegistry.make(); + const observed: Array = []; + let unsubscribe = () => {}; + const secondRevision = new Promise((resolve) => { + unsubscribe = registry.subscribe( + revisions, + (revision) => { + observed.push(revision); + if (revision === 2) resolve(revision); + }, + { immediate: true }, + ); + }); + + // Two emissions, two revisions — identical payloads still count as two, + // because the emission is the signal, not its contents. + expect(await secondRevision).toBe(2); + // Waiting states never rewind the revision. + expect(observed).toEqual([...observed].toSorted((a, b) => a - b)); + unsubscribe(); + registry.dispose(); + }); +}); diff --git a/packages/client-runtime/src/state/runtime.ts b/packages/client-runtime/src/state/runtime.ts index 2404feb82b22..3db1c5fc3327 100644 --- a/packages/client-runtime/src/state/runtime.ts +++ b/packages/client-runtime/src/state/runtime.ts @@ -52,6 +52,14 @@ interface EnvironmentQueryAtomOptions extends EnvironmentAtomOpt readonly staleTimeMs?: number; readonly idleTtlMs?: number; readonly refreshIntervalMs?: number; + /** + * Server-pushed staleness signal. Every emission recomputes the query, which + * re-runs `execute` outright — `staleTimeMs` only gates SWR's own background + * revalidation, so a change that lands inside the stale window still lands in + * the UI. Use this instead of polling when the server can say when the + * underlying data moved. + */ + readonly invalidate?: (input: Input) => Stream.Stream; } interface EnvironmentSubscriptionAtomOptions { @@ -505,11 +513,27 @@ export function createEnvironmentQueryAtomFamily( { initialValue: null }, ), ); + const invalidate = options.invalidate; const family = Atom.family((key: string) => { const target = parseEnvironmentRpcKey(key); const idleTtlMs = options.idleTtlMs ?? 5 * 60_000; - const queryAtom = runtime - .atom((get) => { + const revisions = + invalidate === undefined + ? undefined + : createStreamRevisionAtom( + runtime, + followStreamInEnvironment(target.environmentId, invalidate(target.input)), + { label: `${options.label}:invalidate:${key}`, idleTtlMs }, + ); + return createInvalidatableQueryAtom(runtime, { + label: `${options.label}:${key}`, + idleTtlMs, + staleTimeMs: options.staleTimeMs ?? 30_000, + ...(options.refreshIntervalMs === undefined + ? {} + : { refreshIntervalMs: options.refreshIntervalMs }), + ...(revisions === undefined ? {} : { revisions }), + execute: (get) => { const generation = Option.getOrNull( AsyncResult.value(get(rpcGenerationAtom(target.environmentId))), ); @@ -517,23 +541,82 @@ export function createEnvironmentQueryAtomFamily( return Effect.never; } return runInEnvironment(target.environmentId, options.execute(target.input)); - }) - .pipe( - Atom.swr({ - staleTime: options.staleTimeMs ?? 30_000, - revalidateOnMount: true, - }), - Atom.setIdleTTL(idleTtlMs), - ); - return ( - options.refreshIntervalMs === undefined - ? queryAtom - : queryAtom.pipe(Atom.withRefresh(options.refreshIntervalMs)) - ).pipe(Atom.setIdleTTL(idleTtlMs), Atom.withLabel(`${options.label}:${key}`)); + }, + }); }); return (target) => family(environmentRpcKey(target)); } +/** + * Counts emissions of `stream` as a plain revision number. + * + * Counting on the client keeps the revision insensitive to everything a query + * should not refetch for: waiting states, reconnects and repeated payloads all + * map back to the same number, and the atom's `Object.is` equality drops them. + */ +export function createStreamRevisionAtom( + runtime: Atom.AtomRuntime, + stream: Stream.Stream, + options: { readonly label: string; readonly idleTtlMs: number }, +): Atom.Atom { + return runtime + .atom( + stream.pipe( + Stream.mapAccum( + () => 0, + (revision: number) => [revision + 1, [revision + 1]] as const, + ), + ), + { initialValue: 0 }, + ) + .pipe( + Atom.map((result: AsyncResult.AsyncResult) => { + const revision = AsyncResult.value(result); + return Option.isSome(revision) && typeof revision.value === "number" ? revision.value : 0; + }), + Atom.setIdleTTL(options.idleTtlMs), + Atom.withLabel(options.label), + ); +} + +/** + * Stale-while-revalidate query atom that also re-executes whenever `revisions` + * changes. + * + * `staleTimeMs` only gates SWR's own background revalidation, so a revision + * bump refetches even inside the stale window. That is the point: a push that + * says the data moved must not be answered with a cached value. + */ +export function createInvalidatableQueryAtom( + runtime: Atom.AtomRuntime, + options: { + readonly label: string; + readonly execute: (get: Atom.AtomContext) => Effect.Effect; + readonly revisions?: Atom.Atom; + readonly staleTimeMs: number; + readonly idleTtlMs: number; + readonly refreshIntervalMs?: number; + }, +): Atom.Atom> { + const revisions = options.revisions; + const queryAtom = runtime + .atom((get: Atom.AtomContext) => { + if (revisions !== undefined) { + get(revisions); + } + return options.execute(get); + }) + .pipe( + Atom.swr({ staleTime: options.staleTimeMs, revalidateOnMount: true }), + Atom.setIdleTTL(options.idleTtlMs), + ); + return ( + options.refreshIntervalMs === undefined + ? queryAtom + : queryAtom.pipe(Atom.withRefresh(options.refreshIntervalMs)) + ).pipe(Atom.setIdleTTL(options.idleTtlMs), Atom.withLabel(options.label)); +} + export function createEnvironmentSubscriptionAtomFamily( runtime: Atom.AtomRuntime, options: EnvironmentSubscriptionAtomOptions, @@ -598,8 +681,17 @@ export function createEnvironmentRpcQueryAtomFamily, + ) => Stream.Stream; }, ) { + const invalidate = options.invalidate; return createEnvironmentQueryAtomFamily(runtime, { label: options.label, ...(options.staleTimeMs === undefined ? {} : { staleTimeMs: options.staleTimeMs }), @@ -607,6 +699,14 @@ export function createEnvironmentRpcQueryAtomFamily) => + invalidate(input).pipe(Stream.catchCause(() => Stream.empty)), + }), execute: (input: EnvironmentRpcInput) => request(options.tag, input), }); } diff --git a/packages/contracts/src/project.ts b/packages/contracts/src/project.ts index 757c000a065a..1a3245215391 100644 --- a/packages/contracts/src/project.ts +++ b/packages/contracts/src/project.ts @@ -205,6 +205,17 @@ export const ProjectReadFileResult = Schema.Struct({ }); export type ProjectReadFileResult = typeof ProjectReadFileResult.Type; +/** + * Emitted by `subscribeProjectFileChanges` whenever a watched workspace file + * changes on disk. It is a signal, not a payload: clients re-read the file + * through `projects.readFile` so size limits, binary detection and error + * mapping keep living in one place. + */ +export const ProjectFileChangedEvent = Schema.Struct({ + relativePath: TrimmedNonEmptyString, +}); +export type ProjectFileChangedEvent = typeof ProjectFileChangedEvent.Type; + export const ProjectFileFailure = Schema.Literals([ "workspace_path_outside_root", "resolved_path_outside_root", @@ -223,6 +234,7 @@ export const ProjectFileOperation = Schema.Literals([ "close", "make-directory", "write-file", + "watch", ]); export type ProjectFileOperation = typeof ProjectFileOperation.Type; diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index 51c65f50e1a2..c0c46385684e 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -99,6 +99,7 @@ import { RelayClientStatusSchema, } from "./relayClient.ts"; import { + ProjectFileChangedEvent, ProjectListEntriesError, ProjectListEntriesInput, ProjectListEntriesResult, @@ -304,6 +305,7 @@ export const WS_METHODS = { // Streaming subscriptions subscribeVcsStatus: "subscribeVcsStatus", + subscribeProjectFileChanges: "subscribeProjectFileChanges", subscribeTerminalEvents: "subscribeTerminalEvents", subscribeTerminalMetadata: "subscribeTerminalMetadata", subscribePreviewEvents: "subscribePreviewEvents", @@ -642,6 +644,17 @@ export const WsProjectsReadFileRpc = Rpc.make(WS_METHODS.projectsReadFile, { error: Schema.Union([ProjectReadFileError, EnvironmentAuthorizationError]), }); +/** + * Watches one workspace file and emits whenever it changes on disk, so an open + * viewer re-reads instead of pinning whatever it read when it was opened. + */ +export const WsSubscribeProjectFileChangesRpc = Rpc.make(WS_METHODS.subscribeProjectFileChanges, { + payload: ProjectReadFileInput, + success: ProjectFileChangedEvent, + error: Schema.Union([ProjectReadFileError, EnvironmentAuthorizationError]), + stream: true, +}); + export const WsProjectsWriteFileRpc = Rpc.make(WS_METHODS.projectsWriteFile, { payload: ProjectWriteFileInput, success: ProjectWriteFileResult, @@ -1028,6 +1041,7 @@ export const WsRpcGroup = RpcGroup.make( WsSourceControlPublishRepositoryRpc, WsProjectsListEntriesRpc, WsProjectsReadFileRpc, + WsSubscribeProjectFileChangesRpc, WsProjectsSearchContentsRpc, WsProjectsSearchEntriesRpc, WsProjectsWriteFileRpc,