From 1100bf74a6d43ac587b7f8a7d80f607c767ffc43 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Wed, 2 Sep 2026 21:24:51 -0400 Subject: [PATCH] refactor(core): reconcile current watcher policy --- .../core/src/filesystem/location-watcher.ts | 14 ++-- packages/core/test/filesystem/watcher.test.ts | 73 ++++++++++++++++++- 2 files changed, 77 insertions(+), 10 deletions(-) diff --git a/packages/core/src/filesystem/location-watcher.ts b/packages/core/src/filesystem/location-watcher.ts index 57edda0bc1e0..100ee9dbfcd3 100644 --- a/packages/core/src/filesystem/location-watcher.ts +++ b/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), diff --git a/packages/core/test/filesystem/watcher.test.ts b/packages/core/test/filesystem/watcher.test.ts index 7b73020eddaa..be96f666cabd 100644 --- a/packages/core/test/filesystem/watcher.test.ts +++ b/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, config: Layer.Layer = 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( watcher?: Layer.Layer config?: Layer.Layer plugins?: LayerNode.Replacement + replacements?: LayerNode.Replacements }, ) { return Effect.acquireRelease( @@ -172,7 +176,16 @@ function withTmp( ({ 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() + const release = yield* Deferred.make() + const subscribed = yield* Deferred.make() + 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(