refactor(core): make watcher subscription effectful

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 }),
         )

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(

packages/core/src/filesystem/watcher.ts

@@ -59,7 +59,7 @@ export interface NativeInterface {
 export class Native extends Context.Service<Native, NativeInterface>()("@opencode/Watcher/Native") {}
 
 export interface Interface {
-  readonly subscribe: (input: WatchInput) => Stream.Stream<Update>
+  readonly subscribe: (input: WatchInput) => Effect.Effect<Stream.Stream<Update>>
 }
 
 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 })

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 }),

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<void>()
       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")),