diff --git a/packages/client/src/effect/api/api.ts b/packages/client/src/effect/api/api.ts index ebf112861692..583d78f9427b 100644 --- a/packages/client/src/effect/api/api.ts +++ b/packages/client/src/effect/api/api.ts @@ -1840,9 +1840,25 @@ export type FileFindInput = { export type FileFindOutput = { readonly location: Location.PublicRef; readonly data: ReadonlyArray } export type FileFindOperation = (input: FileFindInput) => Effect.Effect +export type FileWatchInput = { + readonly location?: { readonly directory?: string | undefined } | undefined + readonly path: string +} +export type FileWatchOutput = void +export type FileWatchOperation = (input: FileWatchInput) => Effect.Effect + +export type FileUnwatchInput = { + readonly location?: { readonly directory?: string | undefined } | undefined + readonly path: string +} +export type FileUnwatchOutput = void +export type FileUnwatchOperation = (input: FileUnwatchInput) => Effect.Effect + export interface FileApi { readonly list: FileListOperation readonly find: FileFindOperation + readonly watch: FileWatchOperation + readonly unwatch: FileUnwatchOperation } export type CommandListInput = { readonly location?: { readonly directory?: string | undefined } | undefined } diff --git a/packages/client/src/effect/generated/client.ts b/packages/client/src/effect/generated/client.ts index 69c1136c0dac..7612a10ba8a7 100644 --- a/packages/client/src/effect/generated/client.ts +++ b/packages/client/src/effect/generated/client.ts @@ -195,6 +195,10 @@ import type { FileListOutput, FileFindInput, FileFindOutput, + FileWatchInput, + FileWatchOutput, + FileUnwatchInput, + FileUnwatchOutput, CommandListInput, CommandListOutput, SkillListInput, @@ -1276,7 +1280,26 @@ const EndpointFileFind = (raw: RawClient["server.fs"]) => (input: FileFindInput) }).pipe(Effect.mapError(mapClientError)), ) -const adaptGroupFile = (raw: RawClient["server.fs"]) => ({ list: EndpointFileList(raw), find: EndpointFileFind(raw) }) +const EndpointFileWatch = (raw: RawClient["server.fs"]) => (input: FileWatchInput) => + preserveEffect()( + raw["fs.watch"]({ query: { location: input["location"] }, payload: { path: input["path"] } }).pipe( + Effect.mapError(mapClientError), + ), + ) + +const EndpointFileUnwatch = (raw: RawClient["server.fs"]) => (input: FileUnwatchInput) => + preserveEffect()( + raw["fs.unwatch"]({ query: { location: input["location"] }, payload: { path: input["path"] } }).pipe( + Effect.mapError(mapClientError), + ), + ) + +const adaptGroupFile = (raw: RawClient["server.fs"]) => ({ + list: EndpointFileList(raw), + find: EndpointFileFind(raw), + watch: EndpointFileWatch(raw), + unwatch: EndpointFileUnwatch(raw), +}) const EndpointCommandList = (raw: RawClient["server.command"]) => (input?: CommandListInput) => preserveEffect()( diff --git a/packages/client/src/promise/generated/client.ts b/packages/client/src/promise/generated/client.ts index 8c19478c024b..d29a52685795 100644 --- a/packages/client/src/promise/generated/client.ts +++ b/packages/client/src/promise/generated/client.ts @@ -193,6 +193,10 @@ import type { FileFindOutput, FileWriteInput, FileWriteOutput, + FileWatchInput, + FileWatchOutput, + FileUnwatchInput, + FileUnwatchOutput, CommandListInput, CommandListOutput, SkillListInput, @@ -1734,6 +1738,32 @@ export function make(options: ClientOptions) { }, requestOptions, ), + watch: (input: FileWatchInput, requestOptions?: RequestOptions) => + request( + { + method: "POST", + path: `/api/experimental/fs/watch`, + query: { location: input["location"] }, + body: { path: input["path"] }, + successStatus: 204, + declaredStatuses: [400, 401, 404, 503], + empty: true, + }, + requestOptions, + ), + unwatch: (input: FileUnwatchInput, requestOptions?: RequestOptions) => + request( + { + method: "DELETE", + path: `/api/experimental/fs/watch`, + query: { location: input["location"] }, + body: { path: input["path"] }, + successStatus: 204, + declaredStatuses: [400, 401, 404], + empty: true, + }, + requestOptions, + ), }, command: { list: (input?: CommandListInput, requestOptions?: RequestOptions) => diff --git a/packages/client/src/promise/generated/types.ts b/packages/client/src/promise/generated/types.ts index 8aba08b66962..1b322792149f 100644 --- a/packages/client/src/promise/generated/types.ts +++ b/packages/client/src/promise/generated/types.ts @@ -5611,6 +5611,20 @@ export type FileWriteInput = { export type FileWriteOutput = { location: LocationPublicRef; data: FileSystemWrite } +export type FileWatchInput = { + readonly location?: { readonly location?: { readonly directory?: string | undefined } | undefined }["location"] + readonly path: { readonly path: string }["path"] +} + +export type FileWatchOutput = void + +export type FileUnwatchInput = { + readonly location?: { readonly location?: { readonly directory?: string | undefined } | undefined }["location"] + readonly path: { readonly path: string }["path"] +} + +export type FileUnwatchOutput = void + export type CommandListInput = { readonly location?: { readonly location?: { readonly directory?: string | undefined } | undefined }["location"] } diff --git a/packages/client/test/promise.test.ts b/packages/client/test/promise.test.ts index 3283ddaaaa13..2a975d57ffc5 100644 --- a/packages/client/test/promise.test.ts +++ b/packages/client/test/promise.test.ts @@ -49,7 +49,7 @@ test("exposes every standard HTTP API group", () => { expect(Object.keys(client.websearch)).toEqual(["providers", "query"]) expect(Object.keys(client.voice)).toEqual(["models", "session"]) expect(Object.keys(client.usage)).toEqual(["limits"]) - expect(Object.keys(client.file)).toEqual(["read", "list", "find", "write"]) + expect(Object.keys(client.file)).toEqual(["read", "list", "find", "write", "watch", "unwatch"]) expect(Object.keys(client.vcs)).toEqual(["init", "get", "base", "status", "branch", "diff"]) expect(Object.keys(client.pty)).toEqual(["list", "create", "get", "update", "remove", "connect"]) expect(Object.keys(client.pty.connect)).toEqual(["token"]) diff --git a/packages/core/src/filesystem/file-watch.ts b/packages/core/src/filesystem/file-watch.ts new file mode 100644 index 000000000000..011e2319bed3 --- /dev/null +++ b/packages/core/src/filesystem/file-watch.ts @@ -0,0 +1,164 @@ +export * as FileWatch from "./file-watch.js" + +import { makeGlobalNode } from "@opencode/util/effect/app-node" +import { FSUtil } from "@opencode/util/fs-util" +import { FileSystem } from "@opencode/schema/filesystem" +import type { Location } from "@opencode/schema/location" +import { + Cause, + Clock, + Context, + Duration, + Effect, + Exit, + Layer, + Schedule, + Schema, + Scope, + Semaphore, + Stream, +} from "effect" +import { Bus } from "../bus.js" +import { Watcher } from "./watcher.js" + +export const TTL = Duration.hours(24) +export const SWEEP_INTERVAL = Duration.minutes(5) +export const MAX_PER_LOCATION = 256 + +export class DisabledError extends Schema.TaggedError()("FileWatch.DisabledError", {}) { + override get message() { + return "File watching is disabled" + } +} + +export type Input = { + readonly location: Location.Ref + /** Absolute path of the file to watch. */ + readonly file: string +} + +export interface Interface { + /** Starts watching a file, or resets its expiry when it is already watched. */ + readonly watch: (input: Input) => Effect.Effect + readonly unwatch: (input: Input) => Effect.Effect +} + +export class Service extends Context.Service()("@opencode/FileWatch") {} + +export const Options = Schema.Struct({ + enabled: Schema.optional(Schema.Boolean), +}) +export type Options = typeof Options.Type + +type Entry = { expires: number; scope: Scope.Closeable } + +// Global rather than location-scoped so watches outlive idle location eviction. +export const layer = (options?: Options) => + Layer.effect( + Service, + Effect.gen(function* () { + if (options?.enabled === false) { + return Service.of({ watch: () => Effect.fail(new DisabledError()), unwatch: () => Effect.void }) + } + const watcher = yield* Watcher.Service + const bus = yield* Bus.Service + const fs = yield* FSUtil.Service + const scope = yield* Effect.scope + const lock = yield* Semaphore.make(1) + const locations = new Map>() + const locationKey = (location: Location.Ref) => JSON.stringify([location.directory, location.workspaceID]) + + const remove = (key: string, file: string) => { + const files = locations.get(key) + const entry = files?.get(file) + if (!files || !entry) return + files.delete(file) + if (files.size === 0) locations.delete(key) + return entry + } + + const stop = (key: string, file: string) => + Effect.suspend(() => { + const entry = remove(key, file) + if (!entry) return Effect.void + return Scope.close(entry.scope, Exit.void) + }) + + const start = Effect.fnUntraced(function* (key: string, input: Input) { + const child = yield* Scope.fork(scope) + yield* Effect.gen(function* () { + const updates = yield* watcher.subscribe({ path: input.file, type: "file" }) + yield* Stream.runForEach(updates, () => + Effect.gen(function* () { + const exists = yield* fs.existsSafe(input.file) + yield* bus.publish( + FileSystem.Event.Changed, + { file: input.file, event: exists ? "change" : "unlink" }, + { location: input.location }, + ) + }), + ) + }).pipe( + Effect.catchCauseIf( + (cause) => !Cause.hasInterrupts(cause), + (cause) => Effect.logError("file watch failed", { file: input.file, cause }), + ), + // The watch ended or failed (e.g. its directory vanished): forget it so the next watch starts a new one. + // Interrupts skip this. The scope closes from another fiber because closing it interrupts this one. + Effect.andThen( + lock.withPermit( + Effect.suspend(() => { + if (locations.get(key)?.get(input.file)?.scope !== child) return Effect.void + remove(key, input.file) + return Scope.close(child, Exit.void).pipe(Effect.forkIn(scope), Effect.asVoid) + }), + ), + ), + Effect.forkIn(child, { startImmediately: true }), + ) + return child + }) + + const sweep = lock.withPermit( + Effect.gen(function* () { + const now = yield* Clock.currentTimeMillis + const expired = [...locations].flatMap(([key, files]) => + [...files].filter(([, entry]) => entry.expires <= now).map(([file]) => [key, file] as const), + ) + yield* Effect.forEach(expired, ([key, file]) => stop(key, file), { discard: true }) + }), + ) + yield* sweep.pipe(Effect.repeat(Schedule.spaced(SWEEP_INTERVAL)), Effect.forkScoped) + + return Service.of({ + watch: (input) => + lock.withPermit( + Effect.gen(function* () { + const key = locationKey(input.location) + const expires = (yield* Clock.currentTimeMillis) + Duration.toMillis(TTL) + const existing = locations.get(key)?.get(input.file) + if (existing) { + existing.expires = expires + return + } + const full = locations.get(key) + if (full && full.size >= MAX_PER_LOCATION) { + const soonest = [...full].reduce((a, b) => (b[1].expires < a[1].expires ? b : a)) + yield* stop(key, soonest[0]) + } + const entry = { expires, scope: yield* start(key, input) } + const files = locations.get(key) ?? new Map() + files.set(input.file, entry) + locations.set(key, files) + }), + ), + unwatch: (input) => lock.withPermit(stop(locationKey(input.location), input.file)), + }) + }), + ) + +export function configured(options?: Options) { + return makeGlobalNode({ service: Service, layer: layer(options), deps: [Watcher.node, Bus.node, FSUtil.node] }) +} + +export const node = configured() diff --git a/packages/core/src/filesystem/watcher.ts b/packages/core/src/filesystem/watcher.ts index f33413bfb023..ae359e36658e 100644 --- a/packages/core/src/filesystem/watcher.ts +++ b/packages/core/src/filesystem/watcher.ts @@ -203,20 +203,27 @@ export const nativeLayer = Layer.succeed( Native.of({ subscribe: (input) => { if (input.type === "file" || input.type === "entries") { - return Effect.sync(() => { - const directory = input.type === "file" ? path.dirname(input.target) : input.target - const names = new Set(input.type === "file" ? [path.basename(input.target)] : input.names) - const subscription = watch(directory, { recursive: false }, (_event, file) => { + const directory = input.type === "file" ? path.dirname(input.target) : input.target + const names = new Set(input.type === "file" ? [path.basename(input.target)] : input.names) + return Effect.try(() => + watch(directory, { recursive: false }, (_event, file) => { if (file && !names.has(file)) return for (const name of file ? [file] : names) { input.publish({ path: path.join(directory, name), type: "update" }) } - }) - subscription.on("error", (error: unknown) => - Effect.runFork(Effect.logError("watcher callback failed", { path: directory, error })), - ) - return { unsubscribe: () => Promise.resolve(subscription.close()), backend: "node" } - }) + }), + ).pipe( + Effect.map((subscription): Subscription => { + subscription.on("error", (error: unknown) => + Effect.runFork(Effect.logError("watcher callback failed", { path: directory, error })), + ) + return { unsubscribe: () => Promise.resolve(subscription.close()), backend: "node" } + }), + // A missing directory (ENOENT) ends subscriber streams instead of crashing the watch. + Effect.catch((error) => + Effect.logError("failed to watch directory", { path: directory, error }).pipe(Effect.as(undefined)), + ), + ) } return subscribeDirectory(watcher(), getBackend(), input.target, input.ignore, input.publish) }, diff --git a/packages/core/test/filesystem/file-watch.test.ts b/packages/core/test/filesystem/file-watch.test.ts new file mode 100644 index 000000000000..52efcb852826 --- /dev/null +++ b/packages/core/test/filesystem/file-watch.test.ts @@ -0,0 +1,213 @@ +import { describe, expect } from "bun:test" +import fs from "fs/promises" +import path from "path" +import { Deferred, Duration, Effect, Fiber, Layer, Schedule, Stream } from "effect" +import { TestClock } from "effect/testing" +import { AppNodeBuilder } from "@opencode/core/effect/app-node-builder" +import { Bus } from "@opencode/core/bus" +import { FileWatch } from "@opencode/core/filesystem/file-watch" +import { Watcher } from "@opencode/core/filesystem/watcher" +import { FileSystem } from "@opencode/schema/filesystem" +import { AbsolutePath } from "@opencode/core/schema" +import { FSUtil } from "@opencode/util/fs-util" +import { LayerNode } from "@opencode/util/effect/layer-node" +import { tmpdirScoped } from "../fixture/tmpdir" +import { testEffect } from "../lib/effect" + +const services = AppNodeBuilder.build(LayerNode.group([FSUtil.node, Bus.node])) +const it = testEffect(Layer.provideMerge(FileWatch.layer(), Watcher.testLayer).pipe(Layer.provideMerge(services))) +const native = testEffect( + FileWatch.layer().pipe( + Layer.provideMerge(Watcher.layer().pipe(Layer.provide(Watcher.nativeLayer))), + Layer.provideMerge(services), + ), +) +const describeNative = process.env.CI ? describe.skip : describe + +const location = { directory: AbsolutePath.make("/project") } +const file = (name: string) => ({ location, file: path.join(location.directory, name) }) + +const subscriptions = Effect.gen(function* () { + const test = yield* Watcher.Test + return (yield* test.subscriptions()).length +}) + +describe("FileWatch", () => { + it.effect("shares one watch per file and resubscribes after unwatch", () => + Effect.gen(function* () { + const fileWatch = yield* FileWatch.Service + yield* fileWatch.watch(file("a.ts")) + yield* fileWatch.watch(file("a.ts")) + expect(yield* subscriptions).toBe(1) + yield* fileWatch.unwatch(file("a.ts")) + yield* fileWatch.watch(file("a.ts")) + expect(yield* subscriptions).toBe(2) + }), + ) + + it.effect("expires a watch after 24 hours unless it is watched again", () => + Effect.gen(function* () { + const fileWatch = yield* FileWatch.Service + yield* fileWatch.watch(file("a.ts")) + yield* fileWatch.watch(file("b.ts")) + yield* TestClock.adjust(Duration.hours(23)) + yield* fileWatch.watch(file("a.ts")) + yield* TestClock.adjust(Duration.hours(2)) + expect(yield* subscriptions).toBe(2) + yield* fileWatch.watch(file("a.ts")) + expect(yield* subscriptions).toBe(2) + yield* fileWatch.watch(file("b.ts")) + expect(yield* subscriptions).toBe(3) + }), + ) + + it.effect("evicts the watch closest to expiring when a location is full", () => + Effect.gen(function* () { + const fileWatch = yield* FileWatch.Service + yield* fileWatch.watch(file("oldest.ts")) + yield* TestClock.adjust(Duration.minutes(1)) + yield* Effect.forEach( + Array.from({ length: FileWatch.MAX_PER_LOCATION - 1 }, (_, index) => file(`${index}.ts`)), + (input) => fileWatch.watch(input), + { discard: true }, + ) + yield* fileWatch.watch({ location: { directory: AbsolutePath.make("/other") }, file: "/other/a.ts" }) + expect(yield* subscriptions).toBe(FileWatch.MAX_PER_LOCATION + 1) + yield* fileWatch.watch(file("newest.ts")) + yield* fileWatch.watch(file("0.ts")) + expect(yield* subscriptions).toBe(FileWatch.MAX_PER_LOCATION + 2) + yield* fileWatch.watch(file("oldest.ts")) + expect(yield* subscriptions).toBe(FileWatch.MAX_PER_LOCATION + 3) + }), + ) + + it.effect("publishes filesystem.changed on the watch's location", () => + Effect.gen(function* () { + const fileWatch = yield* FileWatch.Service + const test = yield* Watcher.Test + const bus = yield* Bus.Service + const tmp = yield* tmpdirScoped() + const target = path.join(tmp.path, "a.ts") + yield* Effect.promise(() => fs.writeFile(target, "a")) + const input = { location: { directory: AbsolutePath.make(tmp.path) }, file: target } + const events = yield* bus + .subscribe(FileSystem.Event.Changed) + .pipe(Stream.take(2), Stream.runCollect, Effect.forkScoped({ startImmediately: true })) + yield* fileWatch.watch(input) + yield* test.emit({ path: target, type: "update" }) + yield* Effect.promise(() => fs.rm(target)) + yield* test.emit({ path: target, type: "update" }) + const received = yield* Fiber.join(events) + expect(received.map((event) => ({ directory: event.location?.directory, ...event.data }))).toEqual([ + { directory: AbsolutePath.make(tmp.path), file: target, event: "change" }, + { directory: AbsolutePath.make(tmp.path), file: target, event: "unlink" }, + ]) + }), + ) +}) + +describe("FileWatch dead watches", () => { + const counts = { subscribes: 0 } + // A native watch that cannot start, like one on a directory that no longer exists. + const failing = testEffect( + FileWatch.layer().pipe( + Layer.provide( + Watcher.layer().pipe( + Layer.provide( + Layer.succeed( + Watcher.Native, + Watcher.Native.of({ subscribe: () => Effect.sync(() => void counts.subscribes++) }), + ), + ), + ), + ), + Layer.provide(services), + ), + ) + failing.effect("starts a new watch when the previous one ended", () => + Effect.gen(function* () { + const fileWatch = yield* FileWatch.Service + yield* fileWatch.watch(file("a.ts")) + yield* Effect.repeat(Effect.yieldNow, { times: 10 }) + yield* fileWatch.watch(file("a.ts")) + expect(counts.subscribes).toBe(2) + }), + ) + + const failures = { subscribes: 0 } + const broken = testEffect( + FileWatch.layer().pipe( + Layer.provide( + Layer.succeed( + Watcher.Service, + Watcher.Service.of({ + subscribe: () => Effect.sync(() => (failures.subscribes++, Stream.die(new Error("watch failed")))), + }), + ), + ), + Layer.provide(services), + ), + ) + broken.effect("starts a new watch when the previous one failed", () => + Effect.gen(function* () { + const fileWatch = yield* FileWatch.Service + yield* fileWatch.watch(file("a.ts")) + yield* Effect.repeat(Effect.yieldNow, { times: 10 }) + yield* fileWatch.watch(file("a.ts")) + expect(failures.subscribes).toBe(2) + }), + ) +}) + +describe("FileWatch disabled", () => { + const disabled = testEffect( + FileWatch.layer({ enabled: false }).pipe(Layer.provide(Watcher.testLayer), Layer.provide(services)), + ) + disabled.effect("fails watch requests", () => + Effect.gen(function* () { + const fileWatch = yield* FileWatch.Service + const error = yield* fileWatch.watch(file("a.ts")).pipe(Effect.flip) + expect(error).toBeInstanceOf(FileWatch.DisabledError) + }), + ) +}) + +describeNative("FileWatch native", () => { + native.live("reports in-place edits and atomic temp + rename saves", () => + Effect.gen(function* () { + const fileWatch = yield* FileWatch.Service + const bus = yield* Bus.Service + const tmp = yield* tmpdirScoped() + const target = path.join(tmp.path, "a.ts") + yield* Effect.promise(() => fs.writeFile(target, "start")) + yield* fileWatch.watch({ location: { directory: AbsolutePath.make(tmp.path) }, file: target }) + + const changed = (write: () => Promise) => + Effect.gen(function* () { + const seen = yield* Deferred.make() + const listener = yield* bus.subscribe(FileSystem.Event.Changed).pipe( + Stream.filter((event) => event.data.file === target && event.data.event === "change"), + Stream.runForEach(() => Deferred.succeed(seen, undefined)), + Effect.forkScoped({ startImmediately: true }), + ) + // Native watches arm asynchronously, so repeat the write until one is seen. + const writes = yield* Effect.promise(write).pipe( + Effect.repeat(Schedule.spaced("20 millis")), + Effect.forkScoped({ startImmediately: true }), + ) + yield* Deferred.await(seen).pipe( + Effect.timeoutOrElse({ duration: "5 seconds", orElse: () => Effect.fail(new Error("no change event")) }), + Effect.ensuring(Fiber.interrupt(writes)), + Effect.ensuring(Fiber.interrupt(listener)), + ) + }) + + yield* changed(() => fs.writeFile(target, `inplace-${Math.random()}`)) + yield* changed(async () => { + const temp = path.join(tmp.path, `.a.ts.${Math.random().toString(36).slice(2)}.tmp`) + await fs.writeFile(temp, `atomic-${Math.random()}`) + await fs.rename(temp, target) + }) + }), + ) +}) diff --git a/packages/core/test/filesystem/watcher.test.ts b/packages/core/test/filesystem/watcher.test.ts index c880c31ec72d..b50ac8b9978a 100644 --- a/packages/core/test/filesystem/watcher.test.ts +++ b/packages/core/test/filesystem/watcher.test.ts @@ -54,6 +54,16 @@ function countingNative() { } describe("Watcher lifecycle", () => { + it.live("ends the stream when the watched directory is missing", () => + withTmp((directory) => + Effect.gen(function* () { + const watcher = yield* Watcher.Service + const updates = yield* watcher.subscribe({ path: path.join(directory, "missing", "a.ts"), type: "file" }) + expect(Option.isNone(yield* Stream.runHead(updates))).toBe(true) + }).pipe(Effect.provide(AppNodeBuilder.build(Watcher.node))), + ), + ) + it.effect("signals readiness after acquisition and buffers updates published by the ready callback", () => Effect.gen(function* () { const publish = yield* Deferred.make<(update: Watcher.Update) => void>() diff --git a/packages/protocol/openapi.json b/packages/protocol/openapi.json index 9e362f333b29..bc4fbe407cf3 100644 --- a/packages/protocol/openapi.json +++ b/packages/protocol/openapi.json @@ -9564,6 +9564,209 @@ } } }, + "/api/experimental/fs/watch": { + "post": { + "tags": ["filesystem"], + "operationId": "experimental.fs.watch", + "parameters": [ + { + "name": "location", + "in": "query", + "schema": { + "anyOf": [ + { + "type": "object", + "properties": { + "directory": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ] + } + }, + "additionalProperties": false + }, + { + "type": "null" + } + ] + }, + "required": false, + "style": "deepObject", + "explode": true + } + ], + "security": [], + "responses": { + "204": { + "description": "" + }, + "400": { + "description": "InvalidRequestError", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/InvalidRequestErrorEncoded" + } + } + } + }, + "401": { + "description": "UnauthorizedError", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/UnauthorizedErrorEncoded" + } + } + } + }, + "404": { + "description": "FileNotFoundError | LocationNotFoundError", + "content": { + "application/json": { + "schema": { + "anyOf": [ + { + "$ref": "#/components/schemas/FileNotFoundErrorEncoded" + }, + { + "$ref": "#/components/schemas/LocationNotFoundErrorEncoded" + } + ] + } + } + } + }, + "503": { + "description": "ServiceUnavailableError", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ServiceUnavailableErrorEncoded" + } + } + } + } + }, + "description": "Watch one file relative to the requested location for 24 hours, publishing filesystem.changed on that location when it changes. Watching an already watched file resets its expiry. Paths follow read confinement. Experimental: may change without compatibility guarantees.", + "summary": "Watch file", + "requestBody": { + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "path": { + "type": "string", + "description": "A file path relative to the requested location." + } + }, + "required": ["path"], + "additionalProperties": false + } + } + }, + "required": true + } + }, + "delete": { + "tags": ["filesystem"], + "operationId": "experimental.fs.unwatch", + "parameters": [ + { + "name": "location", + "in": "query", + "schema": { + "anyOf": [ + { + "type": "object", + "properties": { + "directory": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ] + } + }, + "additionalProperties": false + }, + { + "type": "null" + } + ] + }, + "required": false, + "style": "deepObject", + "explode": true + } + ], + "security": [], + "responses": { + "204": { + "description": "" + }, + "400": { + "description": "InvalidRequestError", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/InvalidRequestErrorEncoded" + } + } + } + }, + "401": { + "description": "UnauthorizedError", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/UnauthorizedErrorEncoded" + } + } + } + }, + "404": { + "description": "LocationNotFoundError", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/LocationNotFoundErrorEncoded" + } + } + } + } + }, + "description": "Stop watching one file relative to the requested location. Experimental: may change without compatibility guarantees.", + "summary": "Unwatch file", + "requestBody": { + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "path": { + "type": "string", + "description": "A file path relative to the requested location." + } + }, + "required": ["path"], + "additionalProperties": false + } + } + }, + "required": true + } + } + }, "/api/command": { "get": { "tags": ["command"], diff --git a/packages/protocol/src/groups/fs.ts b/packages/protocol/src/groups/fs.ts index 612e8d392115..f41fa74bc83c 100644 --- a/packages/protocol/src/groups/fs.ts +++ b/packages/protocol/src/groups/fs.ts @@ -3,7 +3,7 @@ import { Location } from "@opencode/schema/location" import { PositiveInt } from "@opencode/schema/schema" import { Schema } from "effect" import { HttpApiEndpoint, HttpApiGroup, HttpApiSchema, OpenApi } from "effect/http-api" -import { FileNotFoundError } from "../errors.js" +import { FileNotFoundError, ServiceUnavailableError } from "../errors.js" import { LocationQuery, locationQueryOpenApi } from "./location.js" const ListQuery = Schema.Struct({ @@ -21,6 +21,10 @@ const WriteQuery = Schema.Struct({ }), }) +const WatchPayload = Schema.Struct({ + path: Schema.String.annotate({ description: "A file path relative to the requested location." }), +}) + const FindQuery = Schema.Struct({ ...LocationQuery.fields, query: FileSystem.FindInput.fields.query, @@ -91,6 +95,39 @@ export const FileSystemGroup = HttpApiGroup.make("server.fs") }), ), ) + .add( + HttpApiEndpoint.post("fs.watch", "/api/experimental/fs/watch", { + query: LocationQuery, + payload: WatchPayload, + success: HttpApiSchema.NoContent, + error: [FileNotFoundError, ServiceUnavailableError], + }) + .annotateMerge(locationQueryOpenApi) + .annotateMerge( + OpenApi.annotations({ + identifier: "experimental.fs.watch", + summary: "Watch file", + description: + "Watch one file relative to the requested location for 24 hours, publishing filesystem.changed on that location when it changes. Watching an already watched file resets its expiry. Paths follow read confinement. Experimental: may change without compatibility guarantees.", + }), + ), + ) + .add( + HttpApiEndpoint.delete("fs.unwatch", "/api/experimental/fs/watch", { + query: LocationQuery, + payload: WatchPayload, + success: HttpApiSchema.NoContent, + }) + .annotateMerge(locationQueryOpenApi) + .annotateMerge( + OpenApi.annotations({ + identifier: "experimental.fs.unwatch", + summary: "Unwatch file", + description: + "Stop watching one file relative to the requested location. Experimental: may change without compatibility guarantees.", + }), + ), + ) .annotateMerge( OpenApi.annotations({ title: "filesystem", diff --git a/packages/server/src/handlers/fs.ts b/packages/server/src/handlers/fs.ts index 11225d42b716..41d009a18ba8 100644 --- a/packages/server/src/handlers/fs.ts +++ b/packages/server/src/handlers/fs.ts @@ -1,14 +1,18 @@ import { FileSystem } from "@opencode/core/filesystem" +import { FileWatch } from "@opencode/core/filesystem/file-watch" +import { Location } from "@opencode/core/location" import { RelativePath } from "@opencode/core/schema" -import { FileNotFoundError } from "@opencode/protocol/errors" +import { FileNotFoundError, ServiceUnavailableError } from "@opencode/protocol/errors" import { Effect, Option, Stream } from "effect" import { type HttpServerRequest, HttpServerResponse } from "effect/http" import { HttpApiBuilder } from "effect/http-api" +import path from "path" import { Api } from "../api" import { response } from "../location" export const FileSystemHandler = HttpApiBuilder.group(Api, "server.fs", (handlers) => Effect.gen(function* () { + const fileWatch = yield* FileWatch.Service return handlers .handleRaw("fs.read", (ctx) => Effect.gen(function* () { @@ -54,6 +58,39 @@ export const FileSystemHandler = HttpApiBuilder.group(Api, "server.fs", (handler }), ), ) + .handle("fs.watch", (ctx) => + Effect.gen(function* () { + const fs = yield* FileSystem.Service + const location = yield* Location.Service + // Same confinement as read: the file must exist inside the location. + yield* fs + .read({ path: RelativePath.make(ctx.payload.path) }) + .pipe( + Effect.mapError( + (error) => new FileNotFoundError({ path: error.path, message: `File not found: ${error.path}` }), + ), + ) + yield* fileWatch + .watch({ + location: { directory: location.directory, workspaceID: location.workspaceID }, + file: path.resolve(location.directory, ctx.payload.path), + }) + .pipe( + Effect.mapError( + (error) => new ServiceUnavailableError({ message: error.message, service: "filewatcher" }), + ), + ) + }), + ) + .handle("fs.unwatch", (ctx) => + Effect.gen(function* () { + const location = yield* Location.Service + yield* fileWatch.unwatch({ + location: { directory: location.directory, workspaceID: location.workspaceID }, + file: path.resolve(location.directory, ctx.payload.path), + }) + }), + ) }), ) diff --git a/packages/server/src/routes.ts b/packages/server/src/routes.ts index c0d08f90ff0f..19122cfe789f 100644 --- a/packages/server/src/routes.ts +++ b/packages/server/src/routes.ts @@ -34,6 +34,7 @@ import { SdkPlugins } from "@opencode/core/plugin/sdk" import { WellKnown } from "@opencode/core/wellknown" import { Workspace } from "@opencode/core/workspace" import { Watcher } from "@opencode/core/filesystem/watcher" +import { FileWatch } from "@opencode/core/filesystem/file-watch" import { HttpRouter } from "effect/http" import { HttpApiBuilder } from "effect/http-api" import { Context, Effect, Layer, Option } from "effect" @@ -77,6 +78,7 @@ const applicationServiceNodes = [ LocationActivity.node, SessionRestart.node, Workspace.node, + FileWatch.node, ] as const const applicationServices = LayerNode.group(applicationServiceNodes) @@ -122,6 +124,7 @@ function makeRoutes( App.node.replace(App.configured(options.app)), ModelsDev.node.replace(ModelsDev.configured(options.models)), Watcher.node.replace(Watcher.configured({ enabled: options.fs?.filewatcher })), + FileWatch.node.replace(FileWatch.configured({ enabled: options.fs?.filewatcher })), FileSystemSearch.node.replace(FileSystemSearch.configured({ fff: options.fs?.fff })), Global.node.replace(Global.layerWith(options.config?.directory ? { config: options.config.directory } : {})), Config.node.replace( diff --git a/packages/server/test/fs.test.ts b/packages/server/test/fs.test.ts index 75bb33051415..432ee5a2713e 100644 --- a/packages/server/test/fs.test.ts +++ b/packages/server/test/fs.test.ts @@ -335,3 +335,32 @@ it.live("opens no file for HEAD and closes it when a download is aborted", () => ).toBe(0) }), ) + +it.live("confines watch requests like read and reports a disabled watcher", () => + Effect.gen(function* () { + const tmp = yield* tmpdirScoped() + const project = path.join(tmp.path, "project") + yield* Effect.promise(async () => { + await fs.mkdir(project) + await fs.writeFile(path.join(project, "a.ts"), "a") + await fs.writeFile(path.join(tmp.path, "outside.ts"), "outside") + }) + const server = yield* startServer(path.join(tmp.path, "config")) + const watch = (method: string, file: string) => + Effect.promise(async () => { + const url = new URL("/api/experimental/fs/watch", server.base) + url.searchParams.set("location[directory]", project) + const response = await fetch(url, { + method, + headers: { ...server.headers, "content-type": "application/json" }, + body: JSON.stringify({ path: file }), + }) + return response.status + }) + + expect(yield* watch("POST", "../outside.ts")).toBe(404) + expect(yield* watch("POST", "missing.ts")).toBe(404) + expect(yield* watch("POST", "a.ts")).toBe(503) + expect(yield* watch("DELETE", "a.ts")).toBe(204) + }), +)