refactor(core): reconcile current watcher policy

packages/core/src/filesystem/location-watcher.ts

@@ -50,16 +50,14 @@ const layer = Layer.effect(
       ),
     )
     const lock = Semaphore.makeUnsafe(1)
-    let requested = 0
     let stopped = false
     let active: { path: string; scope: Scope.Closeable } | undefined
-    const reconcile = (ignore: readonly string[]) => {
-      const request = ++requested
-      return lock.withPermit(
+    const reconcile = () =>
+      lock.withPermit(
         Effect.gen(function* () {
-          if (stopped || request !== requested) return
+          if (stopped) return
           const resolved = yield* target
-          if (stopped || request !== requested) return
+          const ignore = policy.current()
           const next = resolved && !resolved.aliases.some((alias) => ignore.includes(alias)) ? resolved.path : undefined
           if (active?.path === next) return
           if (active) yield* Scope.close(active.scope, Exit.void)
@@ -79,12 +77,10 @@ const layer = Layer.effect(
           )
         }).pipe(Effect.withSpan("LocationWatcher.reconcile", { attributes: { directory: location.directory } })),
       )
-    }
     yield* Effect.addFinalizer(() =>
       lock.withPermit(
         Effect.gen(function* () {
           stopped = true
-          requested++
           if (active) yield* Scope.close(active.scope, Exit.void)
           active = undefined
         }),
@@ -93,7 +89,7 @@ const layer = Layer.effect(
     yield* policy.observe(reconcile)
     yield* Effect.gen(function* () {
       yield* Plugin.awaitActivation
-      yield* reconcile(policy.current())
+      yield* reconcile()
     }).pipe(
       Effect.catchCauseIf(
         (cause) => !Cause.hasInterrupts(cause),

packages/core/test/filesystem/watcher.test.ts

@@ -16,6 +16,7 @@ import { Watcher } from "@opencode-ai/core/filesystem/watcher"
 import { FileSystem } from "@opencode-ai/schema/filesystem"
 import { Document, Event, Info, type Entry } from "@opencode-ai/schema/config"
 import { Location } from "@opencode-ai/core/location"
+import { Plugin } from "@opencode-ai/core/plugin"
 import { PluginSupervisor } from "@opencode-ai/core/plugin/supervisor"
 import { AbsolutePath } from "@opencode-ai/core/schema"
 import { location } from "../fixture/location"
@@ -125,6 +126,7 @@ function provide(
   watcher?: Layer.Layer<Watcher.Service>,
   config: Layer.Layer<Config.Service> = configLayer,
   plugins?: LayerNode.Replacement,
+  replacements: LayerNode.Replacements = [],
 ) {
   const locationLayer = Layer.succeed(
     Location.Service,
@@ -137,6 +139,7 @@ function provide(
       Location.node.replace(locationLayer),
       plugins ?? PluginSupervisor.node.replace(Layer.empty),
       ...(watcher ? ([Watcher.node.replace(watcher)] as const) : []),
+      ...replacements,
     ],
   )
   return Effect.provide(built)
@@ -150,6 +153,7 @@ function withTmp<A, E, R>(
     watcher?: Layer.Layer<Watcher.Service>
     config?: Layer.Layer<Config.Service>
     plugins?: LayerNode.Replacement
+    replacements?: LayerNode.Replacements
   },
 ) {
   return Effect.acquireRelease(
@@ -172,7 +176,16 @@ function withTmp<A, E, R>(
     ({ tmp }) => Effect.promise(() => tmp[Symbol.asyncDispose]()),
   ).pipe(
     Effect.flatMap(({ tmp, vcs }) =>
-      f(tmp.path, vcs).pipe(provide(tmp.path, vcs, options?.watcher, options?.config ?? configLayer, options?.plugins)),
+      f(tmp.path, vcs).pipe(
+        provide(
+          tmp.path,
+          vcs,
+          options?.watcher,
+          options?.config ?? configLayer,
+          options?.plugins,
+          options?.replacements,
+        ),
+      ),
     ),
   )
 }
@@ -293,6 +306,64 @@ describe("LocationWatcher subscriptions", () => {
     })
   })
 
+  it.live("uses the policy changed while target discovery was suspended", () =>
+    Effect.gen(function* () {
+      const fs = yield* FSUtil.Service
+      const discovering = yield* Deferred.make<void>()
+      const release = yield* Deferred.make<void>()
+      const subscribed = yield* Deferred.make<void>()
+      const subscriptions: Watcher.WatchInput[] = []
+      let released = 0
+      yield* withTmp(
+        (directory) =>
+          Effect.gen(function* () {
+            const policy = yield* LocationWatcherPolicy.Service
+            yield* Deferred.await(discovering)
+            const update = yield* policy
+              .transform((editor) => editor.add([".hg"]))
+              .pipe(Effect.forkScoped({ startImmediately: true }))
+            expect(policy.current()).toEqual([".hg"])
+            yield* Deferred.succeed(release, undefined)
+            const registration = yield* Fiber.join(update)
+            expect(subscriptions).toEqual([])
+
+            yield* registration.dispose
+            yield* Deferred.await(subscribed)
+            yield* policy.reload()
+            expect(subscriptions).toEqual([{ path: path.join(directory, ".hg", "branch"), type: "file" }])
+            expect(released).toBe(0)
+          }),
+        {
+          vcs: "hg",
+          replacements: [
+            Plugin.node.replace(Layer.mock(Plugin.Service, { awaitActivation: Effect.void })),
+            FSUtil.node.replace(
+              Layer.succeed(FSUtil.Service, {
+                ...fs,
+                realPath: (target) =>
+                  Deferred.succeed(discovering, undefined).pipe(
+                    Effect.andThen(Deferred.await(release)),
+                    Effect.andThen(fs.realPath(target)),
+                  ),
+              }),
+            ),
+          ],
+          watcher: Layer.succeed(
+            Watcher.Service,
+            Watcher.Service.of({
+              subscribe: (input) =>
+                Effect.sync(() => subscriptions.push(input)).pipe(
+                  Effect.andThen(Deferred.succeed(subscribed, undefined)),
+                  Effect.as(Stream.never.pipe(Stream.ensuring(Effect.sync(() => released++)))),
+                ),
+            }),
+          ),
+        },
+      )
+      expect(released).toBe(1)
+    }),
+  )
+
   it.live("does not start before configured policy is ready", () => {
     const subscriptions: Watcher.WatchInput[] = []
     const watcher = Layer.succeed(