From a7c7473bf67f81537656afa7fc53a14cd05c690f Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Thu, 6 Aug 2026 20:40:30 -0400 Subject: [PATCH 1/3] fix(core): reload changed skill sources --- packages/core/src/filesystem/watcher.ts | 29 ++- packages/core/src/skill.ts | 139 ++++++++++--- packages/core/test/filesystem/watcher.test.ts | 27 ++- packages/core/test/skill.test.ts | 189 +++++++++++++++++- 4 files changed, 338 insertions(+), 46 deletions(-) diff --git a/packages/core/src/filesystem/watcher.ts b/packages/core/src/filesystem/watcher.ts index 7ff3aa5e5c5a..d0e065af7d5e 100644 --- a/packages/core/src/filesystem/watcher.ts +++ b/packages/core/src/filesystem/watcher.ts @@ -5,7 +5,7 @@ import { createWrapper } from "@parcel/watcher/wrapper" import type ParcelWatcher from "@parcel/watcher" import { FileSystem } from "@opencode-ai/schema/filesystem" import { makeGlobalNode } from "@opencode-ai/util/effect/app-node" -import { Cause, Context, Effect, Layer, PubSub, RcMap, Schema, Stream } from "effect" +import { Cause, Context, Effect, Layer, PubSub, RcMap, Schema, Scope, Stream } from "effect" import { lazy } from "../util/lazy" import { watch as watchFileSystem } from "node:fs" import path from "path" @@ -60,6 +60,7 @@ export class Native extends Context.Service()("@opencod export interface Interface { readonly subscribe: (input: WatchInput) => Effect.Effect> + readonly acquire: (input: WatchInput) => Effect.Effect, never, Scope.Scope> } export const Options = Schema.Struct({ @@ -83,7 +84,10 @@ export const layer = (options?: Options) => Service, Effect.gen(function* () { if (options?.enabled === false) { - return Service.of({ subscribe: () => Effect.succeed(Stream.empty) }) + return Service.of({ + subscribe: () => Effect.succeed(Stream.empty), + acquire: () => Effect.succeed(Stream.empty), + }) } const native = yield* Native @@ -92,9 +96,7 @@ export const layer = (options?: Options) => const watchers = yield* RcMap.make({ lookup: (key: Key) => Effect.gen(function* () { - const pubsub = yield* Effect.acquireRelease(PubSub.unbounded(), (pubsub) => - PubSub.shutdown(pubsub), - ) + const pubsub = yield* Effect.acquireRelease(PubSub.unbounded(), (pubsub) => PubSub.shutdown(pubsub)) const subscription = yield* Effect.acquireRelease( native.subscribe({ type: key.type, @@ -146,7 +148,21 @@ export const layer = (options?: Options) => }) } - return Service.of({ subscribe }) + const acquire = (input: WatchInput) => { + const target = path.resolve(input.path) + const ignore = [...new Set(input.type === "directory" ? (input.ignore ?? []) : [])].toSorted() + return Effect.gen(function* () { + yield* Effect.logInfo("watcher acquire", { + path: target, + type: input.type, + ignores: ignore.length, + }) + const pubsub = yield* RcMap.get(watchers, { type: input.type, target, ignore }) + return Stream.fromPubSub(pubsub) + }) + } + + return Service.of({ subscribe, acquire }) }), ) @@ -180,6 +196,7 @@ export const testLayer = Layer.effectContext( const context = yield* Layer.build(layer().pipe(Layer.provide(Layer.succeed(Native, native)))) const test = Test.of({ subscribe: Context.get(context, Service).subscribe, + acquire: Context.get(context, Service).acquire, emit: (update) => Effect.sync(() => active.forEach((publish) => publish(update))), subscriptions: () => Effect.sync(() => [...subscriptions]), }) diff --git a/packages/core/src/skill.ts b/packages/core/src/skill.ts index 521d3d6eb553..c11855ba1739 100644 --- a/packages/core/src/skill.ts +++ b/packages/core/src/skill.ts @@ -2,7 +2,7 @@ export * as Skill from "./skill" import { makeLocationNode } from "@opencode-ai/util/effect/app-node" import path from "path" -import { Context, Effect, Layer, Schema, Stream, Types } from "effect" +import { Context, Effect, Layer, Schema, Scope, Semaphore, Stream, Types } from "effect" import { FileSystem } from "@opencode-ai/schema/filesystem" import { Skill } from "@opencode-ai/schema/skill" import { Agent } from "./agent" @@ -13,6 +13,7 @@ import { Permission } from "./permission" import { AbsolutePath } from "./schema" import { SkillDiscovery } from "./skill/discovery" import { State } from "./state" +import { Watcher } from "./filesystem/watcher" export const DirectorySource = Skill.DirectorySource export type DirectorySource = Skill.DirectorySource @@ -81,6 +82,87 @@ const layer = Layer.effect( const discovery = yield* SkillDiscovery.Service const fs = yield* FSUtil.Service const bus = yield* Bus.Service + const watcher = yield* Watcher.Service + const scope = yield* Scope.Scope + const cache = new Map() + const cacheLock = Semaphore.makeUnsafe(1) + const watched = new Set() + + const invalidate = Effect.fn("Skill.invalidateFromWatcher")(function* (file: string) { + const invalidated = yield* cacheLock.withPermit( + Effect.sync(() => { + return Array.from(cache.entries()).flatMap(([key, loaded]) => { + if (!loaded.paths.some((item) => FSUtil.overlaps(item, file))) return [] + cache.delete(key) + return [[key, loaded] as const] + }) + }), + ) + if (invalidated.length === 0) return + yield* Effect.logInfo("skill cache invalidated", { + file, + sources: invalidated.map(([key]) => key), + skills: invalidated.flatMap(([, loaded]) => loaded.skills.map((skill) => skill.id)), + }) + yield* bus.publish(Skill.Event.Updated, {}).pipe(Effect.asVoid) + }) + + const watch = Effect.fn("Skill.watch")(function* ( + input: Watcher.WatchInput, + accepts: (update: Watcher.Update) => boolean = () => true, + suffix = "", + ) { + yield* Effect.uninterruptible( + Effect.gen(function* () { + const target = path.resolve(input.path) + const key = `${input.type}:${target}:${suffix}` + if (watched.has(key)) return + watched.add(key) + const updates = yield* watcher + .acquire({ ...input, path: target }) + .pipe(Effect.provideService(Scope.Scope, scope)) + yield* updates.pipe( + Stream.filter(accepts), + Stream.runForEach((update) => invalidate(update.path)), + Effect.forkIn(scope, { startImmediately: true }), + ) + }), + ) + }) + + const watchDirectory = Effect.fn("Skill.watchDirectory")(function* (directory: string) { + const target = path.resolve(directory) + const resolved = yield* fs.realPath(directory).pipe(Effect.catch(() => Effect.succeed(undefined))) + if (resolved) { + yield* watch({ path: resolved, type: "directory" }) + if (resolved !== target) { + yield* watch( + { path: path.dirname(target), type: "directory" }, + (update) => FSUtil.overlaps(target, update.path), + target, + ) + } + return resolved === target ? [target] : [target, resolved] + } + if (yield* fs.isDir(path.dirname(target))) { + yield* watch( + { path: path.dirname(target), type: "directory" }, + (update) => FSUtil.overlaps(target, update.path), + target, + ) + } + return [target] + }) + + const refresh = Effect.fn("Skill.refresh")(function* (sources: readonly Source[]) { + yield* Effect.forEach( + sources, + (source) => (source.type === "directory" ? watchDirectory(source.path) : Effect.void), + { discard: true }, + ) + yield* cacheLock.withPermit(Effect.sync(() => cache.clear())) + yield* bus.publish(Skill.Event.Updated, {}).pipe(Effect.asVoid) + }) const state = State.create({ name: "skill", @@ -92,7 +174,7 @@ const layer = Layer.effect( }, list: () => draft.sources as Source[], }), - finalize: () => bus.publish(Skill.Event.Updated, {}).pipe(Effect.asVoid), + finalize: (draft) => refresh(draft.list()), }) const load = Effect.fn("Skill.load")(function* (source: Source) { @@ -104,14 +186,22 @@ const layer = Layer.effect( directories: [], skills: [source.skill.id], }) - return { skills: [source.skill], directories: [] } + return { skills: [source.skill], paths: [] } } const directories = source.type === "directory" ? [source.path] : yield* discovery.pull(source.url) + const roots = (yield* Effect.forEach(directories, watchDirectory)).flat() + const paths = [...roots] for (const directory of directories) { const files = yield* fs .scan("{*.md,**/SKILL.md}", { cwd: directory, absolute: true, include: "file", symlink: true, dot: true }) .pipe(Effect.catch(() => Effect.succeed([] as string[]))) for (const filepath of files.toSorted()) { + const resolved = yield* fs.realPath(filepath).pipe(Effect.catch(() => Effect.succeed(filepath))) + if (!roots.some((root) => FSUtil.contains(root, resolved))) { + const external = path.dirname(resolved) + paths.push(external) + yield* watch({ path: external, type: "directory" }) + } const content = yield* fs.readFileStringSafe(filepath).pipe(Effect.catch(() => Effect.succeed(undefined))) if (!content) continue const markdown = ConfigMarkdown.parseOption(content) @@ -139,22 +229,7 @@ const layer = Layer.effect( directories, skills: skills.map((skill) => skill.id), }) - return { skills, directories } - }) - - const cache = new Map() - const invalidate = Effect.fn("Skill.invalidateFromWatcher")(function* (file: string) { - const invalidated = Array.from(cache.entries()).filter(([, loaded]) => - loaded.directories.some((directory) => FSUtil.contains(directory, file)), - ) - if (invalidated.length === 0) return - for (const [key] of invalidated) cache.delete(key) - yield* Effect.logInfo("skill cache invalidated", { - file, - sources: invalidated.map(([key]) => key), - skills: invalidated.flatMap(([, loaded]) => loaded.skills.map((skill) => skill.id)), - }) - yield* bus.publish(Skill.Event.Updated, {}).pipe(Effect.asVoid) + return { skills, paths } }) yield* bus.subscribe(FileSystem.Event.Changed).pipe( @@ -162,16 +237,20 @@ const layer = Layer.effect( Effect.forkScoped({ startImmediately: true }), ) - const list = Effect.fn("Skill.list")(function* () { - const skills = new Map() - for (const source of state.get().sources) { - const key = Source.key(source) - const loaded = cache.get(key) ?? (yield* load(source)) - cache.set(key, loaded) - for (const skill of loaded.skills) skills.set(skill.id, skill) - } - return Array.from(skills.values()) - }) + const list = Effect.fn("Skill.list")(() => + cacheLock.withPermit( + Effect.gen(function* () { + const skills = new Map() + for (const source of state.get().sources) { + const key = Source.key(source) + const loaded = cache.get(key) ?? (yield* load(source)) + cache.set(key, loaded) + for (const skill of loaded.skills) skills.set(skill.id, skill) + } + return Array.from(skills.values()) + }), + ), + ) return Service.of({ transform: state.transform, @@ -187,5 +266,5 @@ const layer = Layer.effect( export const node = makeLocationNode({ service: Service, layer, - deps: [SkillDiscovery.node, FSUtil.node, Bus.node], + deps: [SkillDiscovery.node, FSUtil.node, Bus.node, Watcher.node], }) diff --git a/packages/core/test/filesystem/watcher.test.ts b/packages/core/test/filesystem/watcher.test.ts index 002b918813e3..888dc5e28e9a 100644 --- a/packages/core/test/filesystem/watcher.test.ts +++ b/packages/core/test/filesystem/watcher.test.ts @@ -69,16 +69,28 @@ function countingNative() { } describe("Watcher lifecycle", () => { + it.effect("acquires a ready subscription before returning", () => { + const { native, counts } = countingNative() + return Effect.gen(function* () { + yield* Effect.gen(function* () { + const watcher = yield* Watcher.Service + yield* watcher.acquire({ path: "/ready", type: "directory" }) + expect(counts.subscribes).toBe(1) + expect(counts.unsubscribes).toBe(0) + }).pipe(withNative(native)) + expect(counts.unsubscribes).toBe(1) + }) + }) + it.effect("interrupting a consumer interrupts a pending acquisition", () => Effect.gen(function* () { const started = yield* Deferred.make() const interrupted = yield* Deferred.make() yield* Effect.gen(function* () { const watcher = yield* Watcher.Service - const consumer = yield* watcher.subscribe({ path: "/pending", type: "directory" }).pipe( - Effect.flatMap(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) @@ -99,10 +111,9 @@ describe("Watcher lifecycle", () => { return Effect.gen(function* () { const watcher = yield* Watcher.Service const consume = () => - watcher.subscribe({ path: "/shared", type: "directory" }).pipe( - Effect.flatMap(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 diff --git a/packages/core/test/skill.test.ts b/packages/core/test/skill.test.ts index 07e29ff37804..1209c8074991 100644 --- a/packages/core/test/skill.test.ts +++ b/packages/core/test/skill.test.ts @@ -6,11 +6,11 @@ import { Agent } from "@opencode-ai/core/agent" import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" import { LayerNode } from "@opencode-ai/util/effect/layer-node" import { Bus } from "@opencode-ai/core/bus" -import { FSUtil } from "@opencode-ai/util/fs-util" import { AbsolutePath } from "@opencode-ai/core/schema" import { Skill } from "@opencode-ai/core/skill" import { SkillDiscovery } from "@opencode-ai/core/skill/discovery" import { FileSystem } from "@opencode-ai/schema/filesystem" +import { Watcher } from "@opencode-ai/core/filesystem/watcher" import { tmpdir } from "./fixture/tmpdir" import { testEffect } from "./lib/effect" @@ -25,8 +25,15 @@ const discovery = Layer.succeed( }, }), ) +const watcherLayer = Watcher.testLayer const it = testEffect( - AppNodeBuilder.build(LayerNode.group([Skill.node, Agent.node, Bus.node]), [[SkillDiscovery.node, discovery]]), + Layer.mergeAll( + AppNodeBuilder.build(LayerNode.group([Skill.node, Agent.node, Bus.node]), [ + [SkillDiscovery.node, discovery], + [Watcher.node, watcherLayer], + ]), + watcherLayer, + ), ) function write(directory: string, name: string, description: string) { @@ -53,6 +60,13 @@ function waitForSkillUpdate() { }) } +function waitForSubscription(check: (input: Watcher.WatchInput) => boolean) { + return Effect.gen(function* () { + const watcher = yield* Watcher.Test + while (!(yield* watcher.subscriptions()).some(check)) yield* Effect.yieldNow + }).pipe(Effect.timeout("1 second")) +} + describe("Skill", () => { it.live("publishes updates when skill sources change", () => Effect.gen(function* () { @@ -234,4 +248,175 @@ metadata: ), ), ) + + it.live("clears cached skills when sources reload", () => + Effect.acquireRelease( + Effect.promise(() => tmpdir()), + (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()), + ).pipe( + Effect.flatMap((tmp) => + Effect.gen(function* () { + yield* Effect.promise(async () => { + await fs.mkdir(path.join(tmp.path, "deploy"), { recursive: true }) + await write(tmp.path, "deploy", "Initial deploy") + }) + + const skill = yield* Skill.Service + yield* skill.transform((editor) => editor.source({ type: "directory", path: AbsolutePath.make(tmp.path) })) + expect((yield* skill.list()).find((item) => item.id === "deploy")?.description).toBe("Initial deploy") + + yield* Effect.promise(() => write(tmp.path, "deploy", "Updated deploy")) + yield* skill.reload() + + expect((yield* skill.list()).find((item) => item.id === "deploy")?.description).toBe("Updated deploy") + }), + ), + ), + ) + + it.live("watches directory sources for added and changed skills", () => + Effect.acquireRelease( + Effect.promise(() => tmpdir()), + (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()), + ).pipe( + Effect.flatMap((tmp) => + Effect.gen(function* () { + yield* Effect.promise(async () => { + await fs.mkdir(path.join(tmp.path, "deploy"), { recursive: true }) + await write(tmp.path, "deploy", "Initial deploy") + }) + + const skill = yield* Skill.Service + const watcher = yield* Watcher.Test + yield* skill.transform((editor) => editor.source({ type: "directory", path: AbsolutePath.make(tmp.path) })) + expect((yield* skill.list()).map((item) => item.id)).toEqual([Skill.ID.make("deploy")]) + yield* waitForSubscription((input) => input.type === "directory" && input.path === tmp.path) + + const deploy = path.join(tmp.path, "deploy", "SKILL.md") + yield* Effect.promise(() => write(tmp.path, "deploy", "Updated deploy")) + yield* Effect.acquireUseRelease( + waitForSkillUpdate(), + ({ deferred }) => + watcher + .emit({ type: "update", path: deploy }) + .pipe(Effect.andThen(Deferred.await(deferred)), Effect.timeout("1 second")), + ({ fiber }) => Fiber.interrupt(fiber), + ) + expect((yield* skill.list()).find((item) => item.id === "deploy")?.description).toBe("Updated deploy") + + yield* Effect.promise(async () => { + await fs.mkdir(path.join(tmp.path, "review"), { recursive: true }) + await write(tmp.path, "review", "Review changes") + }) + const review = path.join(tmp.path, "review", "SKILL.md") + yield* Effect.acquireUseRelease( + waitForSkillUpdate(), + ({ deferred }) => + watcher + .emit({ type: "create", path: review }) + .pipe(Effect.andThen(Deferred.await(deferred)), Effect.timeout("1 second")), + ({ fiber }) => Fiber.interrupt(fiber), + ) + expect((yield* skill.list()).map((item) => item.id)).toEqual([ + Skill.ID.make("deploy"), + Skill.ID.make("review"), + ]) + + yield* Effect.promise(() => fs.rm(path.join(tmp.path, "review"), { recursive: true })) + yield* Effect.acquireUseRelease( + waitForSkillUpdate(), + ({ deferred }) => + watcher + .emit({ type: "delete", path: review }) + .pipe(Effect.andThen(Deferred.await(deferred)), Effect.timeout("1 second")), + ({ fiber }) => Fiber.interrupt(fiber), + ) + expect((yield* skill.list()).map((item) => item.id)).toEqual([Skill.ID.make("deploy")]) + }), + ), + ), + ) + + it.live("watches canonical directories behind symlinked skills", () => + Effect.acquireRelease( + Effect.promise(() => tmpdir()), + (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()), + ).pipe( + Effect.flatMap((tmp) => + Effect.gen(function* () { + const source = path.join(tmp.path, "source") + const target = path.join(tmp.path, "target", "bro") + const file = path.join(target, "SKILL.md") + yield* Effect.promise(async () => { + await fs.mkdir(source, { recursive: true }) + await fs.mkdir(target, { recursive: true }) + await fs.writeFile(file, "---\nname: bro\ndescription: Initial\n---\n# bro") + await fs.symlink(target, path.join(source, "bro")) + }) + + const skill = yield* Skill.Service + const watcher = yield* Watcher.Test + yield* skill.transform((editor) => editor.source({ type: "directory", path: AbsolutePath.make(source) })) + expect((yield* skill.list()).find((item) => item.id === "bro")?.description).toBe("Initial") + yield* waitForSubscription((input) => input.type === "directory" && input.path === target) + + yield* Effect.promise(() => fs.writeFile(file, "---\nname: bro\ndescription: Updated\n---\n# bro")) + yield* Effect.acquireUseRelease( + waitForSkillUpdate(), + ({ deferred }) => + watcher + .emit({ type: "update", path: file }) + .pipe(Effect.andThen(Deferred.await(deferred)), Effect.timeout("1 second")), + ({ fiber }) => Fiber.interrupt(fiber), + ) + expect((yield* skill.list()).find((item) => item.id === "bro")?.description).toBe("Updated") + }), + ), + ), + ) + + it.live("invalidates symlinked sources when their target changes", () => + Effect.acquireRelease( + Effect.promise(() => tmpdir()), + (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()), + ).pipe( + Effect.flatMap((tmp) => + Effect.gen(function* () { + const source = path.join(tmp.path, "source") + const first = path.join(tmp.path, "first") + const second = path.join(tmp.path, "second") + yield* Effect.promise(async () => { + await fs.mkdir(path.join(first, "bro"), { recursive: true }) + await fs.mkdir(path.join(second, "bro"), { recursive: true }) + await write(first, "bro", "First") + await write(second, "bro", "Second") + await fs.symlink(first, source) + }) + + const skill = yield* Skill.Service + const watcher = yield* Watcher.Test + yield* skill.transform((editor) => editor.source({ type: "directory", path: AbsolutePath.make(source) })) + expect((yield* skill.list()).find((item) => item.id === "bro")?.description).toBe("First") + yield* waitForSubscription((input) => input.type === "directory" && input.path === first) + yield* waitForSubscription((input) => input.type === "directory" && input.path === tmp.path) + + yield* Effect.promise(async () => { + await fs.unlink(source) + await fs.symlink(second, source) + }) + yield* Effect.acquireUseRelease( + waitForSkillUpdate(), + ({ deferred }) => + watcher + .emit({ type: "update", path: source }) + .pipe(Effect.andThen(Deferred.await(deferred)), Effect.timeout("1 second")), + ({ fiber }) => Fiber.interrupt(fiber), + ) + + expect((yield* skill.list()).find((item) => item.id === "bro")?.description).toBe("Second") + yield* waitForSubscription((input) => input.type === "directory" && input.path === second) + }), + ), + ), + ) }) From 1cbe9626344cf0e2a40182e875452a7e65087c5b Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Thu, 6 Aug 2026 20:53:23 -0400 Subject: [PATCH 2/3] refactor(core): simplify skill source watches --- packages/core/src/filesystem/watcher.ts | 37 ++----- packages/core/src/skill.ts | 30 ++---- packages/core/test/skill.test.ts | 125 +++++++++--------------- 3 files changed, 62 insertions(+), 130 deletions(-) diff --git a/packages/core/src/filesystem/watcher.ts b/packages/core/src/filesystem/watcher.ts index d0e065af7d5e..bc2c65ee144d 100644 --- a/packages/core/src/filesystem/watcher.ts +++ b/packages/core/src/filesystem/watcher.ts @@ -130,37 +130,18 @@ export const layer = (options?: Options) => }), }) - const subscribe = (input: WatchInput) => { + const acquire = Effect.fn("Watcher.acquire")(function* (input: WatchInput) { const target = path.resolve(input.path) const ignore = [...new Set(input.type === "directory" ? (input.ignore ?? []) : [])].toSorted() - 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) - }), - ) - }) - } - - const acquire = (input: WatchInput) => { - const target = path.resolve(input.path) - const ignore = [...new Set(input.type === "directory" ? (input.ignore ?? []) : [])].toSorted() - return Effect.gen(function* () { - yield* Effect.logInfo("watcher acquire", { - path: target, - type: input.type, - ignores: ignore.length, - }) - const pubsub = yield* RcMap.get(watchers, { type: input.type, target, ignore }) - return Stream.fromPubSub(pubsub) + yield* Effect.logInfo("watcher subscribe", { + path: target, + type: input.type, + ignores: ignore.length, }) - } + const pubsub = yield* RcMap.get(watchers, { type: input.type, target, ignore }) + return Stream.fromPubSub(pubsub) + }) + const subscribe = (input: WatchInput) => Effect.succeed(Stream.unwrap(acquire(input))) return Service.of({ subscribe, acquire }) }), diff --git a/packages/core/src/skill.ts b/packages/core/src/skill.ts index c11855ba1739..256af424bb00 100644 --- a/packages/core/src/skill.ts +++ b/packages/core/src/skill.ts @@ -107,22 +107,17 @@ const layer = Layer.effect( yield* bus.publish(Skill.Event.Updated, {}).pipe(Effect.asVoid) }) - const watch = Effect.fn("Skill.watch")(function* ( - input: Watcher.WatchInput, - accepts: (update: Watcher.Update) => boolean = () => true, - suffix = "", - ) { + const watch = Effect.fn("Skill.watch")(function* (input: Watcher.WatchInput) { yield* Effect.uninterruptible( Effect.gen(function* () { const target = path.resolve(input.path) - const key = `${input.type}:${target}:${suffix}` + const key = `${input.type}:${target}` if (watched.has(key)) return watched.add(key) const updates = yield* watcher .acquire({ ...input, path: target }) .pipe(Effect.provideService(Scope.Scope, scope)) yield* updates.pipe( - Stream.filter(accepts), Stream.runForEach((update) => invalidate(update.path)), Effect.forkIn(scope, { startImmediately: true }), ) @@ -136,30 +131,17 @@ const layer = Layer.effect( if (resolved) { yield* watch({ path: resolved, type: "directory" }) if (resolved !== target) { - yield* watch( - { path: path.dirname(target), type: "directory" }, - (update) => FSUtil.overlaps(target, update.path), - target, - ) + yield* watch({ path: path.dirname(target), type: "directory" }) } return resolved === target ? [target] : [target, resolved] } if (yield* fs.isDir(path.dirname(target))) { - yield* watch( - { path: path.dirname(target), type: "directory" }, - (update) => FSUtil.overlaps(target, update.path), - target, - ) + yield* watch({ path: path.dirname(target), type: "directory" }) } return [target] }) - const refresh = Effect.fn("Skill.refresh")(function* (sources: readonly Source[]) { - yield* Effect.forEach( - sources, - (source) => (source.type === "directory" ? watchDirectory(source.path) : Effect.void), - { discard: true }, - ) + const refresh = Effect.fn("Skill.refresh")(function* () { yield* cacheLock.withPermit(Effect.sync(() => cache.clear())) yield* bus.publish(Skill.Event.Updated, {}).pipe(Effect.asVoid) }) @@ -174,7 +156,7 @@ const layer = Layer.effect( }, list: () => draft.sources as Source[], }), - finalize: (draft) => refresh(draft.list()), + finalize: refresh, }) const load = Effect.fn("Skill.load")(function* (source: Source) { diff --git a/packages/core/test/skill.test.ts b/packages/core/test/skill.test.ts index 1209c8074991..507da25b63e6 100644 --- a/packages/core/test/skill.test.ts +++ b/packages/core/test/skill.test.ts @@ -60,11 +60,22 @@ function waitForSkillUpdate() { }) } -function waitForSubscription(check: (input: Watcher.WatchInput) => boolean) { +function expectSubscription(check: (input: Watcher.WatchInput) => boolean) { return Effect.gen(function* () { const watcher = yield* Watcher.Test - while (!(yield* watcher.subscriptions()).some(check)) yield* Effect.yieldNow - }).pipe(Effect.timeout("1 second")) + expect((yield* watcher.subscriptions()).some(check)).toBe(true) + }) +} + +function emitAndWait(update: Watcher.Update) { + return Effect.gen(function* () { + const watcher = yield* Watcher.Test + yield* Effect.acquireUseRelease( + waitForSkillUpdate(), + ({ deferred }) => watcher.emit(update).pipe(Effect.andThen(Deferred.await(deferred)), Effect.timeout("1 second")), + ({ fiber }) => Fiber.interrupt(fiber), + ) + }) } describe("Skill", () => { @@ -212,7 +223,7 @@ metadata: ), ) - it.live("invalidates cached skills and publishes updates for watcher changes", () => + it.live("clears cached skills when sources reload", () => Effect.acquireRelease( Effect.promise(() => tmpdir()), (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()), @@ -224,51 +235,47 @@ metadata: await write(tmp.path, "deploy", "Initial deploy") }) - const bus = yield* Bus.Service const skill = yield* Skill.Service yield* skill.transform((editor) => editor.source({ type: "directory", path: AbsolutePath.make(tmp.path) })) + expect((yield* skill.list()).find((item) => item.id === "deploy")?.description).toBe("Initial deploy") - expect((yield* skill.list()).find((item) => item.name === "deploy")?.description).toBe("Initial deploy") - - const file = path.join(tmp.path, "deploy", "SKILL.md") yield* Effect.promise(() => write(tmp.path, "deploy", "Updated deploy")) - expect((yield* skill.list()).find((item) => item.name === "deploy")?.description).toBe("Initial deploy") - - yield* Effect.acquireUseRelease( - waitForSkillUpdate(), - ({ deferred }) => - bus - .publish(FileSystem.Event.Changed, { file, event: "change" }) - .pipe(Effect.andThen(Deferred.await(deferred)), Effect.timeout("1 second")), - ({ fiber }) => Fiber.interrupt(fiber), - ) + yield* skill.reload() - expect((yield* skill.list()).find((item) => item.name === "deploy")?.description).toBe("Updated deploy") + expect((yield* skill.list()).find((item) => item.id === "deploy")?.description).toBe("Updated deploy") }), ), ), ) - it.live("clears cached skills when sources reload", () => + it.live("reloads project sources created after their missing parent", () => Effect.acquireRelease( Effect.promise(() => tmpdir()), (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()), ).pipe( Effect.flatMap((tmp) => Effect.gen(function* () { - yield* Effect.promise(async () => { - await fs.mkdir(path.join(tmp.path, "deploy"), { recursive: true }) - await write(tmp.path, "deploy", "Initial deploy") - }) - + const source = path.join(tmp.path, "generated", "skills") + const file = path.join(source, "deploy", "SKILL.md") const skill = yield* Skill.Service - yield* skill.transform((editor) => editor.source({ type: "directory", path: AbsolutePath.make(tmp.path) })) - expect((yield* skill.list()).find((item) => item.id === "deploy")?.description).toBe("Initial deploy") + const bus = yield* Bus.Service + yield* skill.transform((editor) => editor.source({ type: "directory", path: AbsolutePath.make(source) })) + expect(yield* skill.list()).toEqual([]) - yield* Effect.promise(() => write(tmp.path, "deploy", "Updated deploy")) - yield* skill.reload() + yield* Effect.promise(async () => { + await fs.mkdir(path.dirname(file), { recursive: true }) + await write(source, "deploy", "Deploy production") + }) + yield* Effect.acquireUseRelease( + waitForSkillUpdate(), + ({ deferred }) => + bus + .publish(FileSystem.Event.Changed, { file, event: "add" }) + .pipe(Effect.andThen(Deferred.await(deferred)), Effect.timeout("1 second")), + ({ fiber }) => Fiber.interrupt(fiber), + ) - expect((yield* skill.list()).find((item) => item.id === "deploy")?.description).toBe("Updated deploy") + expect((yield* skill.list()).map((item) => item.id)).toEqual([Skill.ID.make("deploy")]) }), ), ), @@ -287,21 +294,13 @@ metadata: }) const skill = yield* Skill.Service - const watcher = yield* Watcher.Test yield* skill.transform((editor) => editor.source({ type: "directory", path: AbsolutePath.make(tmp.path) })) expect((yield* skill.list()).map((item) => item.id)).toEqual([Skill.ID.make("deploy")]) - yield* waitForSubscription((input) => input.type === "directory" && input.path === tmp.path) + yield* expectSubscription((input) => input.type === "directory" && input.path === tmp.path) const deploy = path.join(tmp.path, "deploy", "SKILL.md") yield* Effect.promise(() => write(tmp.path, "deploy", "Updated deploy")) - yield* Effect.acquireUseRelease( - waitForSkillUpdate(), - ({ deferred }) => - watcher - .emit({ type: "update", path: deploy }) - .pipe(Effect.andThen(Deferred.await(deferred)), Effect.timeout("1 second")), - ({ fiber }) => Fiber.interrupt(fiber), - ) + yield* emitAndWait({ type: "update", path: deploy }) expect((yield* skill.list()).find((item) => item.id === "deploy")?.description).toBe("Updated deploy") yield* Effect.promise(async () => { @@ -309,28 +308,14 @@ metadata: await write(tmp.path, "review", "Review changes") }) const review = path.join(tmp.path, "review", "SKILL.md") - yield* Effect.acquireUseRelease( - waitForSkillUpdate(), - ({ deferred }) => - watcher - .emit({ type: "create", path: review }) - .pipe(Effect.andThen(Deferred.await(deferred)), Effect.timeout("1 second")), - ({ fiber }) => Fiber.interrupt(fiber), - ) + yield* emitAndWait({ type: "create", path: review }) expect((yield* skill.list()).map((item) => item.id)).toEqual([ Skill.ID.make("deploy"), Skill.ID.make("review"), ]) yield* Effect.promise(() => fs.rm(path.join(tmp.path, "review"), { recursive: true })) - yield* Effect.acquireUseRelease( - waitForSkillUpdate(), - ({ deferred }) => - watcher - .emit({ type: "delete", path: review }) - .pipe(Effect.andThen(Deferred.await(deferred)), Effect.timeout("1 second")), - ({ fiber }) => Fiber.interrupt(fiber), - ) + yield* emitAndWait({ type: "delete", path: review }) expect((yield* skill.list()).map((item) => item.id)).toEqual([Skill.ID.make("deploy")]) }), ), @@ -355,20 +340,12 @@ metadata: }) const skill = yield* Skill.Service - const watcher = yield* Watcher.Test yield* skill.transform((editor) => editor.source({ type: "directory", path: AbsolutePath.make(source) })) expect((yield* skill.list()).find((item) => item.id === "bro")?.description).toBe("Initial") - yield* waitForSubscription((input) => input.type === "directory" && input.path === target) + yield* expectSubscription((input) => input.type === "directory" && input.path === target) yield* Effect.promise(() => fs.writeFile(file, "---\nname: bro\ndescription: Updated\n---\n# bro")) - yield* Effect.acquireUseRelease( - waitForSkillUpdate(), - ({ deferred }) => - watcher - .emit({ type: "update", path: file }) - .pipe(Effect.andThen(Deferred.await(deferred)), Effect.timeout("1 second")), - ({ fiber }) => Fiber.interrupt(fiber), - ) + yield* emitAndWait({ type: "update", path: file }) expect((yield* skill.list()).find((item) => item.id === "bro")?.description).toBe("Updated") }), ), @@ -394,27 +371,19 @@ metadata: }) const skill = yield* Skill.Service - const watcher = yield* Watcher.Test yield* skill.transform((editor) => editor.source({ type: "directory", path: AbsolutePath.make(source) })) expect((yield* skill.list()).find((item) => item.id === "bro")?.description).toBe("First") - yield* waitForSubscription((input) => input.type === "directory" && input.path === first) - yield* waitForSubscription((input) => input.type === "directory" && input.path === tmp.path) + yield* expectSubscription((input) => input.type === "directory" && input.path === first) + yield* expectSubscription((input) => input.type === "directory" && input.path === tmp.path) yield* Effect.promise(async () => { await fs.unlink(source) await fs.symlink(second, source) }) - yield* Effect.acquireUseRelease( - waitForSkillUpdate(), - ({ deferred }) => - watcher - .emit({ type: "update", path: source }) - .pipe(Effect.andThen(Deferred.await(deferred)), Effect.timeout("1 second")), - ({ fiber }) => Fiber.interrupt(fiber), - ) + yield* emitAndWait({ type: "update", path: source }) expect((yield* skill.list()).find((item) => item.id === "bro")?.description).toBe("Second") - yield* waitForSubscription((input) => input.type === "directory" && input.path === second) + yield* expectSubscription((input) => input.type === "directory" && input.path === second) }), ), ), From 2e57411ffe9e7bee3624a1e4b34503b005fad4a2 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Thu, 6 Aug 2026 21:05:31 -0400 Subject: [PATCH 3/3] refactor(core): reuse watcher subscription pattern --- packages/core/src/filesystem/watcher.ts | 38 ++++----- packages/core/src/skill.ts | 77 +++++++------------ packages/core/test/filesystem/watcher.test.ts | 27 ++----- 3 files changed, 56 insertions(+), 86 deletions(-) diff --git a/packages/core/src/filesystem/watcher.ts b/packages/core/src/filesystem/watcher.ts index bc2c65ee144d..7ff3aa5e5c5a 100644 --- a/packages/core/src/filesystem/watcher.ts +++ b/packages/core/src/filesystem/watcher.ts @@ -5,7 +5,7 @@ import { createWrapper } from "@parcel/watcher/wrapper" import type ParcelWatcher from "@parcel/watcher" import { FileSystem } from "@opencode-ai/schema/filesystem" import { makeGlobalNode } from "@opencode-ai/util/effect/app-node" -import { Cause, Context, Effect, Layer, PubSub, RcMap, Schema, Scope, Stream } from "effect" +import { Cause, Context, Effect, Layer, PubSub, RcMap, Schema, Stream } from "effect" import { lazy } from "../util/lazy" import { watch as watchFileSystem } from "node:fs" import path from "path" @@ -60,7 +60,6 @@ export class Native extends Context.Service()("@opencod export interface Interface { readonly subscribe: (input: WatchInput) => Effect.Effect> - readonly acquire: (input: WatchInput) => Effect.Effect, never, Scope.Scope> } export const Options = Schema.Struct({ @@ -84,10 +83,7 @@ export const layer = (options?: Options) => Service, Effect.gen(function* () { if (options?.enabled === false) { - return Service.of({ - subscribe: () => Effect.succeed(Stream.empty), - acquire: () => Effect.succeed(Stream.empty), - }) + return Service.of({ subscribe: () => Effect.succeed(Stream.empty) }) } const native = yield* Native @@ -96,7 +92,9 @@ export const layer = (options?: Options) => const watchers = yield* RcMap.make({ lookup: (key: Key) => Effect.gen(function* () { - const pubsub = yield* Effect.acquireRelease(PubSub.unbounded(), (pubsub) => PubSub.shutdown(pubsub)) + const pubsub = yield* Effect.acquireRelease(PubSub.unbounded(), (pubsub) => + PubSub.shutdown(pubsub), + ) const subscription = yield* Effect.acquireRelease( native.subscribe({ type: key.type, @@ -130,20 +128,25 @@ export const layer = (options?: Options) => }), }) - const acquire = Effect.fn("Watcher.acquire")(function* (input: WatchInput) { + const subscribe = (input: WatchInput) => { const target = path.resolve(input.path) const ignore = [...new Set(input.type === "directory" ? (input.ignore ?? []) : [])].toSorted() - yield* Effect.logInfo("watcher subscribe", { - path: target, - type: input.type, - ignores: ignore.length, + 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) + }), + ) }) - const pubsub = yield* RcMap.get(watchers, { type: input.type, target, ignore }) - return Stream.fromPubSub(pubsub) - }) - const subscribe = (input: WatchInput) => Effect.succeed(Stream.unwrap(acquire(input))) + } - return Service.of({ subscribe, acquire }) + return Service.of({ subscribe }) }), ) @@ -177,7 +180,6 @@ export const testLayer = Layer.effectContext( const context = yield* Layer.build(layer().pipe(Layer.provide(Layer.succeed(Native, native)))) const test = Test.of({ subscribe: Context.get(context, Service).subscribe, - acquire: Context.get(context, Service).acquire, emit: (update) => Effect.sync(() => active.forEach((publish) => publish(update))), subscriptions: () => Effect.sync(() => [...subscriptions]), }) diff --git a/packages/core/src/skill.ts b/packages/core/src/skill.ts index 256af424bb00..dd0cbc29f8ba 100644 --- a/packages/core/src/skill.ts +++ b/packages/core/src/skill.ts @@ -2,7 +2,7 @@ export * as Skill from "./skill" import { makeLocationNode } from "@opencode-ai/util/effect/app-node" import path from "path" -import { Context, Effect, Layer, Schema, Scope, Semaphore, Stream, Types } from "effect" +import { Context, Effect, Layer, Schema, Scope, Stream, Types } from "effect" import { FileSystem } from "@opencode-ai/schema/filesystem" import { Skill } from "@opencode-ai/schema/skill" import { Agent } from "./agent" @@ -85,20 +85,14 @@ const layer = Layer.effect( const watcher = yield* Watcher.Service const scope = yield* Scope.Scope const cache = new Map() - const cacheLock = Semaphore.makeUnsafe(1) const watched = new Set() const invalidate = Effect.fn("Skill.invalidateFromWatcher")(function* (file: string) { - const invalidated = yield* cacheLock.withPermit( - Effect.sync(() => { - return Array.from(cache.entries()).flatMap(([key, loaded]) => { - if (!loaded.paths.some((item) => FSUtil.overlaps(item, file))) return [] - cache.delete(key) - return [[key, loaded] as const] - }) - }), + const invalidated = Array.from(cache.entries()).filter(([, loaded]) => + loaded.paths.some((item) => FSUtil.overlaps(item, file)), ) if (invalidated.length === 0) return + for (const [key] of invalidated) cache.delete(key) yield* Effect.logInfo("skill cache invalidated", { file, sources: invalidated.map(([key]) => key), @@ -107,21 +101,14 @@ const layer = Layer.effect( yield* bus.publish(Skill.Event.Updated, {}).pipe(Effect.asVoid) }) - const watch = Effect.fn("Skill.watch")(function* (input: Watcher.WatchInput) { - yield* Effect.uninterruptible( - Effect.gen(function* () { - const target = path.resolve(input.path) - const key = `${input.type}:${target}` - if (watched.has(key)) return - watched.add(key) - const updates = yield* watcher - .acquire({ ...input, path: target }) - .pipe(Effect.provideService(Scope.Scope, scope)) - yield* updates.pipe( - Stream.runForEach((update) => invalidate(update.path)), - Effect.forkIn(scope, { startImmediately: true }), - ) - }), + const watch = Effect.fn("Skill.watch")(function* (directory: string) { + const target = path.resolve(directory) + if (watched.has(target)) return + watched.add(target) + const updates = yield* watcher.subscribe({ path: target, type: "directory" }) + yield* updates.pipe( + Stream.runForEach((update) => invalidate(update.path)), + Effect.forkIn(scope, { startImmediately: true }), ) }) @@ -129,23 +116,18 @@ const layer = Layer.effect( const target = path.resolve(directory) const resolved = yield* fs.realPath(directory).pipe(Effect.catch(() => Effect.succeed(undefined))) if (resolved) { - yield* watch({ path: resolved, type: "directory" }) + yield* watch(resolved) if (resolved !== target) { - yield* watch({ path: path.dirname(target), type: "directory" }) + yield* watch(path.dirname(target)) } return resolved === target ? [target] : [target, resolved] } if (yield* fs.isDir(path.dirname(target))) { - yield* watch({ path: path.dirname(target), type: "directory" }) + yield* watch(path.dirname(target)) } return [target] }) - const refresh = Effect.fn("Skill.refresh")(function* () { - yield* cacheLock.withPermit(Effect.sync(() => cache.clear())) - yield* bus.publish(Skill.Event.Updated, {}).pipe(Effect.asVoid) - }) - const state = State.create({ name: "skill", initial: () => ({ sources: [] }), @@ -156,7 +138,8 @@ const layer = Layer.effect( }, list: () => draft.sources as Source[], }), - finalize: refresh, + finalize: () => + Effect.sync(() => cache.clear()).pipe(Effect.andThen(bus.publish(Skill.Event.Updated, {})), Effect.asVoid), }) const load = Effect.fn("Skill.load")(function* (source: Source) { @@ -182,7 +165,7 @@ const layer = Layer.effect( if (!roots.some((root) => FSUtil.contains(root, resolved))) { const external = path.dirname(resolved) paths.push(external) - yield* watch({ path: external, type: "directory" }) + yield* watch(external) } const content = yield* fs.readFileStringSafe(filepath).pipe(Effect.catch(() => Effect.succeed(undefined))) if (!content) continue @@ -219,20 +202,16 @@ const layer = Layer.effect( Effect.forkScoped({ startImmediately: true }), ) - const list = Effect.fn("Skill.list")(() => - cacheLock.withPermit( - Effect.gen(function* () { - const skills = new Map() - for (const source of state.get().sources) { - const key = Source.key(source) - const loaded = cache.get(key) ?? (yield* load(source)) - cache.set(key, loaded) - for (const skill of loaded.skills) skills.set(skill.id, skill) - } - return Array.from(skills.values()) - }), - ), - ) + const list = Effect.fn("Skill.list")(function* () { + const skills = new Map() + for (const source of state.get().sources) { + const key = Source.key(source) + const loaded = cache.get(key) ?? (yield* load(source)) + cache.set(key, loaded) + for (const skill of loaded.skills) skills.set(skill.id, skill) + } + return Array.from(skills.values()) + }) return Service.of({ transform: state.transform, diff --git a/packages/core/test/filesystem/watcher.test.ts b/packages/core/test/filesystem/watcher.test.ts index 888dc5e28e9a..002b918813e3 100644 --- a/packages/core/test/filesystem/watcher.test.ts +++ b/packages/core/test/filesystem/watcher.test.ts @@ -69,28 +69,16 @@ function countingNative() { } describe("Watcher lifecycle", () => { - it.effect("acquires a ready subscription before returning", () => { - const { native, counts } = countingNative() - return Effect.gen(function* () { - yield* Effect.gen(function* () { - const watcher = yield* Watcher.Service - yield* watcher.acquire({ path: "/ready", type: "directory" }) - expect(counts.subscribes).toBe(1) - expect(counts.unsubscribes).toBe(0) - }).pipe(withNative(native)) - expect(counts.unsubscribes).toBe(1) - }) - }) - it.effect("interrupting a consumer interrupts a pending acquisition", () => Effect.gen(function* () { const started = yield* Deferred.make() const interrupted = yield* Deferred.make() yield* Effect.gen(function* () { const watcher = yield* Watcher.Service - const consumer = yield* watcher - .subscribe({ path: "/pending", type: "directory" }) - .pipe(Effect.flatMap(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) @@ -111,9 +99,10 @@ describe("Watcher lifecycle", () => { return Effect.gen(function* () { const watcher = yield* Watcher.Service const consume = () => - watcher - .subscribe({ path: "/shared", type: "directory" }) - .pipe(Effect.flatMap(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