From bd906d468d03f0118703e53089e3d87bdb0ee888 Mon Sep 17 00:00:00 2001 From: Dax Raad Date: Wed, 29 Jul 2026 09:49:38 -0400 Subject: [PATCH] refactor(core): make watcher subscription effectful --- packages/core/src/config.ts | 3 +- .../core/src/filesystem/location-watcher.ts | 18 ++++----- packages/core/src/filesystem/watcher.ts | 22 +++++++---- packages/core/src/plugin/supervisor.ts | 3 +- packages/core/test/filesystem/watcher.test.ts | 37 +++++++++++-------- 5 files changed, 49 insertions(+), 34 deletions(-) diff --git a/packages/core/src/config.ts b/packages/core/src/config.ts index 7ded4cc66ce..05ba4b0db07 100644 --- a/packages/core/src/config.ts +++ b/packages/core/src/config.ts @@ -393,7 +393,8 @@ export const layer = (options?: Options) => Layer.effect( const key = JSON.stringify(target) if (watched.has(key)) continue watched.add(key) - yield* watcher.subscribe(target).pipe( + const stream = yield* watcher.subscribe(target) + yield* stream.pipe( Stream.runForEach((update) => PubSub.publish(updates, update)), Effect.forkScoped({ startImmediately: true }), ) diff --git a/packages/core/src/filesystem/location-watcher.ts b/packages/core/src/filesystem/location-watcher.ts index 25c98320599..934e99f3bf2 100644 --- a/packages/core/src/filesystem/location-watcher.ts +++ b/packages/core/src/filesystem/location-watcher.ts @@ -47,13 +47,12 @@ const layer = Layer.effect( const home = path.resolve(location.directory) === path.resolve(os.homedir()) if (!home && location.vcs) { - yield* watcher - .subscribe({ - path: location.directory, - type: "directory", - ignore: [...Ignore.PATTERNS, ...config, ...protecteds(location.directory)], - }) - .pipe(Stream.runForEach(publish), Effect.forkScoped) + const updates = yield* watcher.subscribe({ + path: location.directory, + type: "directory", + ignore: [...Ignore.PATTERNS, ...config, ...protecteds(location.directory)], + }) + yield* updates.pipe(Stream.runForEach(publish), Effect.forkScoped) } if (home) { yield* Effect.logInfo("location watcher skipped home directory", { directory: location.directory }) @@ -68,9 +67,8 @@ const layer = Layer.effect( const ignore = (yield* fs.readDirectoryEntries(vcs).pipe(Effect.catch(() => Effect.succeed([])))).flatMap( (entry) => (entry.name === "HEAD" ? [] : [entry.name]), ) - yield* watcher - .subscribe({ path: vcs, type: "directory", ignore }) - .pipe(Stream.runForEach(publish), Effect.forkScoped) + const updates = yield* watcher.subscribe({ path: vcs, type: "directory", ignore }) + yield* updates.pipe(Stream.runForEach(publish), Effect.forkScoped) } } }).pipe( diff --git a/packages/core/src/filesystem/watcher.ts b/packages/core/src/filesystem/watcher.ts index f7fef530297..7ff3aa5e5c5 100644 --- a/packages/core/src/filesystem/watcher.ts +++ b/packages/core/src/filesystem/watcher.ts @@ -59,7 +59,7 @@ export interface NativeInterface { export class Native extends Context.Service()("@opencode/Watcher/Native") {} export interface Interface { - readonly subscribe: (input: WatchInput) => Stream.Stream + readonly subscribe: (input: WatchInput) => Effect.Effect> } export const Options = Schema.Struct({ @@ -83,7 +83,7 @@ export const layer = (options?: Options) => Service, Effect.gen(function* () { if (options?.enabled === false) { - return Service.of({ subscribe: () => Stream.empty }) + return Service.of({ subscribe: () => Effect.succeed(Stream.empty) }) } const native = yield* Native @@ -131,11 +131,19 @@ export const layer = (options?: Options) => const subscribe = (input: WatchInput) => { const target = path.resolve(input.path) const ignore = [...new Set(input.type === "directory" ? (input.ignore ?? []) : [])].toSorted() - return Stream.unwrap( - RcMap.get(watchers, { type: input.type, target, ignore }).pipe( - Effect.map((pubsub) => Stream.fromPubSub(pubsub)), - ), - ) + return Effect.gen(function* () { + yield* Effect.logInfo("watcher subscribe", { + path: target, + type: input.type, + ignores: ignore.length, + }) + return Stream.unwrap( + Effect.gen(function* () { + const pubsub = yield* RcMap.get(watchers, { type: input.type, target, ignore }) + return Stream.fromPubSub(pubsub) + }), + ) + }) } return Service.of({ subscribe }) diff --git a/packages/core/src/plugin/supervisor.ts b/packages/core/src/plugin/supervisor.ts index 39bb868678c..bcca1f53fac 100644 --- a/packages/core/src/plugin/supervisor.ts +++ b/packages/core/src/plugin/supervisor.ts @@ -253,7 +253,8 @@ const layer = Layer.effect( // inside), so don't watch what can't trigger anything. if (yield* fs.isDir(operation.target)) continue watched.add(operation.target) - yield* watcher.subscribe({ path: operation.target, type: "file" }).pipe( + const updates = yield* watcher.subscribe({ path: operation.target, type: "file" }) + yield* updates.pipe( Stream.runForEach(() => PubSub.publish(configuredChanges, undefined)), Effect.catchCause((cause) => Effect.logError("configured plugin watch failed", { target: operation.target, cause }), diff --git a/packages/core/test/filesystem/watcher.test.ts b/packages/core/test/filesystem/watcher.test.ts index 4ff39443e1a..22dddaf0178 100644 --- a/packages/core/test/filesystem/watcher.test.ts +++ b/packages/core/test/filesystem/watcher.test.ts @@ -30,9 +30,12 @@ describe("Watcher.testLayer", () => { Effect.gen(function* () { const watcher = yield* Watcher.Service const test = yield* Watcher.Test - const received = yield* watcher - .subscribe({ path: "/root", type: "directory" }) - .pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped({ startImmediately: true })) + const updates = yield* watcher.subscribe({ path: "/root", type: "directory" }) + const received = yield* updates.pipe( + Stream.take(1), + Stream.runCollect, + Effect.forkScoped({ startImmediately: true }), + ) yield* Effect.yieldNow yield* test.emit({ type: "update", path: "/root/file.md" }) @@ -72,9 +75,10 @@ describe("Watcher lifecycle", () => { const interrupted = yield* Deferred.make() yield* Effect.gen(function* () { const watcher = yield* Watcher.Service - const consumer = yield* watcher - .subscribe({ path: "/pending", type: "directory" }) - .pipe(Stream.runDrain, Effect.forkScoped({ startImmediately: true })) + const consumer = yield* watcher.subscribe({ path: "/pending", type: "directory" }).pipe( + Effect.flatMap(Stream.runDrain), + Effect.forkScoped({ startImmediately: true }), + ) yield* Deferred.await(started) yield* Fiber.interrupt(consumer) expect(yield* Deferred.isDone(interrupted)).toBe(true) @@ -95,9 +99,10 @@ describe("Watcher lifecycle", () => { return Effect.gen(function* () { const watcher = yield* Watcher.Service const consume = () => - watcher - .subscribe({ path: "/shared", type: "directory" }) - .pipe(Stream.runDrain, Effect.forkScoped({ startImmediately: true })) + watcher.subscribe({ path: "/shared", type: "directory" }).pipe( + Effect.flatMap(Stream.runDrain), + Effect.forkScoped({ startImmediately: true }), + ) const first = yield* consume() const second = yield* consume() yield* Effect.yieldNow @@ -117,9 +122,8 @@ describe("Watcher lifecycle", () => { return Effect.gen(function* () { const consumer = yield* Effect.gen(function* () { const watcher = yield* Watcher.Service - const consumer = yield* watcher - .subscribe({ path: "/active", type: "directory" }) - .pipe(Stream.runDrain, Effect.forkScoped({ startImmediately: true })) + const updates = yield* watcher.subscribe({ path: "/active", type: "directory" }) + const consumer = yield* updates.pipe(Stream.runDrain, Effect.forkScoped({ startImmediately: true })) yield* Effect.yieldNow expect(counts.subscribes).toBe(1) expect(counts.unsubscribes).toBe(0) @@ -250,9 +254,12 @@ describeWatcher("LocationWatcher", () => { const watcher = yield* Watcher.Service const target = path.join(directory, "opencode.json") const sibling = path.join(directory, "other.json") - const update = yield* watcher - .subscribe({ path: target, type: "file" }) - .pipe(Stream.take(1), Stream.runHead, Effect.forkScoped({ startImmediately: true })) + const updates = yield* watcher.subscribe({ path: target, type: "file" }) + const update = yield* updates.pipe( + Stream.take(1), + Stream.runHead, + Effect.forkScoped({ startImmediately: true }), + ) yield* fs.writeFileString(sibling, "sibling") const writes = yield* Effect.suspend(() => fs.writeFileString(target, `target-${Math.random()}`)).pipe( Effect.repeat(Schedule.spaced("10 millis")),