refactor(core): separate skill source observation

packages/core/src/config/plugin/skill.ts

@@ -2,17 +2,14 @@ export * as ConfigSkillPlugin from "./skill.js"
 
 import { define } from "@opencode-ai/plugin/effect/plugin"
 import type { Entry } from "@opencode-ai/schema/config"
-import { FSUtil } from "@opencode-ai/util/fs-util"
 import { Global } from "@opencode-ai/util/global"
 import path from "path"
-import { Effect, FiberMap, PubSub, Semaphore, Stream } from "effect"
+import { Effect, Stream } from "effect"
 import { Config } from "../../config.js"
-import { Watcher } from "../../filesystem/watcher.js"
 import { Location } from "../../location.js"
 import { AbsolutePath } from "../../schema.js"
 import { Skill } from "../../skill.js"
-import { SkillDiscovery } from "../../skill/discovery.js"
-import { SkillFile } from "./skill-file.js"
+import { SkillSourceObserver } from "../../skill/source-observer.js"
 
 type Source = Skill.DirectorySource | Skill.UrlSource
 
@@ -20,59 +17,11 @@ export const Plugin = define({
   id: "opencode.config.skill",
   effect: Effect.fn(function* (ctx) {
     const config = yield* Config.Service
-    const discovery = yield* SkillDiscovery.Service
-    const fs = yield* FSUtil.Service
     const global = yield* Global.Service
     const location = yield* Location.Service
-    const watcher = yield* Watcher.Service
-    const loaded: { entries: Entry[]; skills: Skill.Info[] } = {
+    const loaded: { entries: Entry[] } = {
       entries: yield* config.entries(),
-      skills: [],
     }
-    const watches = yield* FiberMap.make<string>()
-    const changes = yield* PubSub.sliding<string>(1)
-    const lock = Semaphore.makeUnsafe(1)
-
-    const watch = Effect.fn("ConfigSkillPlugin.watch")(function* (directory: string, type: Watcher.WatchInput["type"]) {
-      const target = path.resolve(directory)
-      const updates = yield* watcher.subscribe({ path: target, type })
-      yield* FiberMap.run(
-        watches,
-        `${type}:${target}`,
-        updates.pipe(Stream.runForEach((update) => PubSub.publish(changes, update.path).pipe(Effect.asVoid))),
-        { onlyIfMissing: true, startImmediately: true },
-      )
-    })
-
-    function firstMissing(target: string): Effect.Effect<string | undefined> {
-      const parent = path.dirname(target)
-      if (parent === target) return Effect.undefined
-      return fs.isDir(parent).pipe(Effect.flatMap((exists) => (exists ? Effect.succeed(target) : firstMissing(parent))))
-    }
-
-    const watchDirectory: (directory: string) => Effect.Effect<string[]> = Effect.fn(
-      "ConfigSkillPlugin.watchDirectory",
-    )(function* (directory: string) {
-      const target = path.resolve(directory)
-      const resolved = yield* fs.realPath(directory).pipe(Effect.orElseSucceed(() => undefined))
-      if (resolved) {
-        yield* watch(resolved, "directory")
-        if (resolved !== target) yield* watch(target, "file")
-        return resolved === target ? [target] : [target, resolved]
-      }
-      const missing = yield* firstMissing(target)
-      if (missing) yield* watch(missing, "file")
-      if (
-        yield* fs.realPath(directory).pipe(
-          Effect.as(true),
-          Effect.orElseSucceed(() => false),
-        )
-      ) {
-        if (missing) yield* FiberMap.remove(watches, `file:${path.resolve(missing)}`)
-        return yield* watchDirectory(directory)
-      }
-      return [target]
-    })
 
     const sources = () => {
       const result: Source[] = []
@@ -107,85 +56,16 @@ export const Plugin = define({
       return result
     }
 
-    const load = Effect.fn("ConfigSkillPlugin.load")(function* (source: Source) {
-      const directories =
-        source.type === "directory"
-          ? [source.path]
-          : yield* discovery.pull(source.url).pipe(
-              Effect.catchCause((cause) =>
-                Effect.logWarning("failed to load skill source", {
-                  source: Skill.Source.key(source),
-                  cause,
-                }).pipe(Effect.as([] as AbsolutePath[])),
-              ),
-            )
-      const roots = (yield* Effect.forEach(directories, watchDirectory)).flat()
-      const skills: Skill.Info[] = []
-      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.orElseSucceed(() => [] as string[]))
-        for (const filepath of files.toSorted()) {
-          const resolved = yield* fs.realPath(filepath).pipe(Effect.orElseSucceed(() => filepath))
-          if (!roots.some((root) => FSUtil.contains(root, resolved))) yield* watch(path.dirname(resolved), "directory")
-          const content = yield* fs.readFileStringSafe(filepath).pipe(Effect.orElseSucceed(() => undefined))
-          if (!content) continue
-          const parsed = SkillFile.parse(directory, filepath, content)
-          if (parsed._tag === "Skipped") {
-            yield* Effect.logDebug("skill file skipped", {
-              filepath,
-              reason: parsed.reason,
-              ...(parsed.reason === "frontmatter" ? { issue: parsed.issue } : {}),
-            })
-            continue
-          }
-          skills.push(parsed.skill)
-        }
-      }
-      yield* Effect.logDebug("skill source loaded", {
-        source: Skill.Source.key(source),
-        type: source.type,
-        directories,
-        skills: skills.map((skill) => skill.id),
-      })
-      return skills
-    })
-
-    const refresh = Effect.fn("ConfigSkillPlugin.refresh")(function* (file?: string) {
-      yield* lock.withPermit(
-        Effect.gen(function* () {
-          yield* FiberMap.clear(watches)
-          const skills = new Map<Skill.ID, Skill.Info>()
-          const current = sources()
-          for (const source of current) {
-            for (const skill of yield* load(source)) skills.set(skill.id, skill)
-          }
-          loaded.skills = Array.from(skills.values())
-          if (file) {
-            yield* Effect.logInfo("skills rescanned", {
-              file,
-              sources: current.map(Skill.Source.key),
-              skills: loaded.skills.map((skill) => skill.id),
-            })
-          }
-        }),
-      )
-    })
-
-    yield* Stream.fromPubSub(changes).pipe(
-      Stream.runForEach((file) => refresh(file).pipe(Effect.andThen(ctx.skill.reload()))),
-      Effect.forkScoped({ startImmediately: true }),
-    )
-    yield* refresh()
+    const observer = yield* SkillSourceObserver.make({ sources, onChange: ctx.skill.reload })
     yield* ctx.skill.transform((draft) => {
-      for (const skill of loaded.skills) draft.add(skill)
+      for (const skill of observer.list()) draft.add(skill)
     })
     yield* ctx.event.subscribe().pipe(
       Stream.filter((event) => event.type === "config.updated"),
       Stream.runForEach(() =>
         config.entries().pipe(
           Effect.tap((entries) => Effect.sync(() => (loaded.entries = entries))),
-          Effect.andThen(refresh()),
+          Effect.andThen(observer.refresh()),
           Effect.andThen(ctx.skill.reload()),
         ),
       ),

packages/core/src/skill/source-observer.ts

@@ -0,0 +1,143 @@
+export * as SkillSourceObserver from "./source-observer.js"
+
+import { FSUtil } from "@opencode-ai/util/fs-util"
+import path from "path"
+import { Effect, FiberMap, PubSub, Semaphore, Stream } from "effect"
+import { Watcher } from "../filesystem/watcher.js"
+import { AbsolutePath } from "../schema.js"
+import { Skill } from "../skill.js"
+import { SkillDiscovery } from "./discovery.js"
+import { SkillFile } from "../config/plugin/skill-file.js"
+
+type Source = Skill.DirectorySource | Skill.UrlSource
+
+// Sources are read inside each rescan; the caller owns their interpretation and
+// publishes domain updates after filesystem-triggered snapshots are committed.
+export const make = Effect.fn("SkillSourceObserver.make")(function* (input: {
+  readonly sources: () => readonly Source[]
+  readonly onChange: () => Effect.Effect<void>
+}) {
+  const discovery = yield* SkillDiscovery.Service
+  const fs = yield* FSUtil.Service
+  const watcher = yield* Watcher.Service
+  const loaded: { skills: Skill.Info[] } = { skills: [] }
+  const watches = yield* FiberMap.make<string>()
+  const changes = yield* PubSub.sliding<string>(1)
+  const lock = Semaphore.makeUnsafe(1)
+
+  const watch = Effect.fn("SkillSourceObserver.watch")(function* (directory: string, type: Watcher.WatchInput["type"]) {
+    const target = path.resolve(directory)
+    const updates = yield* watcher.subscribe({ path: target, type })
+    yield* FiberMap.run(
+      watches,
+      `${type}:${target}`,
+      updates.pipe(Stream.runForEach((update) => PubSub.publish(changes, update.path).pipe(Effect.asVoid))),
+      { onlyIfMissing: true, startImmediately: true },
+    )
+  })
+
+  function firstMissing(target: string): Effect.Effect<string | undefined> {
+    const parent = path.dirname(target)
+    if (parent === target) return Effect.undefined
+    return fs.isDir(parent).pipe(Effect.flatMap((exists) => (exists ? Effect.succeed(target) : firstMissing(parent))))
+  }
+
+  const watchDirectory: (directory: string) => Effect.Effect<string[]> = Effect.fn(
+    "SkillSourceObserver.watchDirectory",
+  )(function* (directory: string) {
+    const target = path.resolve(directory)
+    const resolved = yield* fs.realPath(directory).pipe(Effect.orElseSucceed(() => undefined))
+    if (resolved) {
+      yield* watch(resolved, "directory")
+      if (resolved !== target) yield* watch(target, "file")
+      return resolved === target ? [target] : [target, resolved]
+    }
+    const missing = yield* firstMissing(target)
+    if (missing) yield* watch(missing, "file")
+    if (
+      yield* fs.realPath(directory).pipe(
+        Effect.as(true),
+        Effect.orElseSucceed(() => false),
+      )
+    ) {
+      if (missing) yield* FiberMap.remove(watches, `file:${path.resolve(missing)}`)
+      return yield* watchDirectory(directory)
+    }
+    return [target]
+  })
+
+  const load = Effect.fn("SkillSourceObserver.load")(function* (source: Source) {
+    const directories =
+      source.type === "directory"
+        ? [source.path]
+        : yield* discovery.pull(source.url).pipe(
+            Effect.catchCause((cause) =>
+              Effect.logWarning("failed to load skill source", {
+                source: Skill.Source.key(source),
+                cause,
+              }).pipe(Effect.as([] as AbsolutePath[])),
+            ),
+          )
+    const roots = (yield* Effect.forEach(directories, watchDirectory)).flat()
+    const skills: Skill.Info[] = []
+    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.orElseSucceed(() => [] as string[]))
+      for (const filepath of files.toSorted()) {
+        const resolved = yield* fs.realPath(filepath).pipe(Effect.orElseSucceed(() => filepath))
+        if (!roots.some((root) => FSUtil.contains(root, resolved))) yield* watch(path.dirname(resolved), "directory")
+        const content = yield* fs.readFileStringSafe(filepath).pipe(Effect.orElseSucceed(() => undefined))
+        if (!content) continue
+        const parsed = SkillFile.parse(directory, filepath, content)
+        if (parsed._tag === "Skipped") {
+          yield* Effect.logDebug("skill file skipped", {
+            filepath,
+            reason: parsed.reason,
+            ...(parsed.reason === "frontmatter" ? { issue: parsed.issue } : {}),
+          })
+          continue
+        }
+        skills.push(parsed.skill)
+      }
+    }
+    yield* Effect.logDebug("skill source loaded", {
+      source: Skill.Source.key(source),
+      type: source.type,
+      directories,
+      skills: skills.map((skill) => skill.id),
+    })
+    return skills
+  })
+
+  const refresh = Effect.fn("SkillSourceObserver.refresh")(function* (file?: string) {
+    yield* lock.withPermit(
+      Effect.gen(function* () {
+        yield* FiberMap.clear(watches)
+        const skills = new Map<Skill.ID, Skill.Info>()
+        const current = input.sources()
+        for (const source of current) {
+          for (const skill of yield* load(source)) skills.set(skill.id, skill)
+        }
+        loaded.skills = Array.from(skills.values())
+        if (file) {
+          yield* Effect.logInfo("skills rescanned", {
+            file,
+            sources: current.map(Skill.Source.key),
+            skills: loaded.skills.map((skill) => skill.id),
+          })
+        }
+      }),
+    )
+  })
+
+  yield* Stream.fromPubSub(changes).pipe(
+    Stream.runForEach((file) => refresh(file).pipe(Effect.andThen(() => input.onChange()))),
+    Effect.forkScoped({ startImmediately: true }),
+  )
+  yield* refresh()
+  return {
+    list: (): readonly Skill.Info[] => loaded.skills,
+    refresh: () => refresh(),
+  }
+})

packages/core/test/config/skill.test.ts

@@ -9,6 +9,7 @@ import {
   Directory as ConfigDirectory,
   Document,
   type Entry,
+  Event,
   Info,
 } from "@opencode-ai/schema/config"
 import { ConfigSkillPlugin } from "@opencode-ai/core/config/plugin/skill"
@@ -179,6 +180,58 @@ metadata:
 })
 
 describe("ConfigSkillPlugin.Plugin", () => {
+  it.live("reinterprets config entries before refreshing and publishing skills", () =>
+    Effect.gen(function* () {
+      const tmp = yield* Effect.acquireDisposable(Effect.promise(() => tmpdir()))
+      const first = path.join(tmp.path, "first")
+      const second = path.join(tmp.path, "second")
+      yield* Effect.promise(async () => {
+        await fs.mkdir(path.join(first, "review"), { recursive: true })
+        await fs.mkdir(path.join(second, "deploy"), { recursive: true })
+        await write(first, "review", "First")
+        await write(second, "deploy", "Second")
+      })
+      yield* Effect.gen(function* () {
+        const config = yield* Config.Test
+        const skill = yield* Skill.Service
+        const bus = yield* Bus.Service
+        const watcher = yield* Watcher.Test
+        yield* ConfigSkillPlugin.Plugin.effect(
+          host({
+            skill: {
+              list: () => Effect.die("unused skill.list"),
+              transform: skill.transform,
+              reload: skill.reload,
+            },
+            event: { subscribe: () => bus.subscribe(Event.Updated) },
+          }),
+        ).pipe(
+          Effect.provideService(Global.Service, Global.Service.of({ ...Global.make(), home: tmp.path })),
+          Effect.provideService(
+            Location.Service,
+            Location.Service.of(location({ directory: AbsolutePath.make(tmp.path) })),
+          ),
+        )
+        expect((yield* skill.list()).map((item) => item.id)).toEqual([Skill.ID.make("review")])
+
+        const updated = yield* Deferred.make<Skill.Info[]>()
+        yield* bus.subscribe(Skill.Event.Updated).pipe(
+          Stream.runForEach(() => skill.list().pipe(Effect.flatMap((skills) => Deferred.succeed(updated, skills)))),
+          Effect.forkScoped({ startImmediately: true }),
+        )
+        yield* config.setEntries([new Document({ type: "document", info: decode({ skills: [second] }) })])
+        yield* bus.publish(Event.Updated, {})
+        expect(yield* Deferred.await(updated).pipe(Effect.timeout("2 seconds"))).toMatchObject([
+          { id: "deploy", description: "Second" },
+        ])
+        expect(yield* watcher.subscriptions()).toEqual([
+          { path: first, type: "directory" },
+          { path: second, type: "directory" },
+        ])
+      }).pipe(Effect.provide(Config.testLayer([new Document({ type: "document", info: decode({ skills: [first] }) })])))
+    }),
+  )
+
   it.live("maps config entry types to skill directories", () =>
     Effect.acquireRelease(
       Effect.promise(() => tmpdir()),

packages/core/test/skill-source-observer.test.ts

@@ -0,0 +1,167 @@
+import fs from "fs/promises"
+import path from "path"
+import { createServer } from "node:http"
+import { describe, expect } from "bun:test"
+import { NodeHttpServer } from "@effect/platform-node"
+import { Deferred, Effect, Layer } from "effect"
+import { HttpServer, HttpServerRequest, HttpServerResponse } from "effect/unstable/http"
+import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
+import { Watcher } from "@opencode-ai/core/filesystem/watcher"
+import { AbsolutePath } from "@opencode-ai/core/schema"
+import { Skill } from "@opencode-ai/core/skill"
+import { SkillDiscovery } from "@opencode-ai/core/skill/discovery"
+import { SkillSourceObserver } from "@opencode-ai/core/skill/source-observer"
+import { FSUtil } from "@opencode-ai/util/fs-util"
+import { Global } from "@opencode-ai/util/global"
+import { LayerNode } from "@opencode-ai/util/effect/layer-node"
+import { tmpdir } from "./fixture/tmpdir"
+import { it } from "./lib/effect"
+
+describe("SkillSourceObserver", () => {
+  it.live("rebuilds watches on every refresh and releases them when the observer scope closes", () =>
+    Effect.gen(function* () {
+      const tmp = yield* Effect.acquireDisposable(Effect.promise(() => tmpdir()))
+      const first = path.join(tmp.path, "first")
+      const second = path.join(tmp.path, "second")
+      yield* Effect.promise(async () => {
+        await fs.mkdir(first)
+        await fs.mkdir(second)
+        await fs.writeFile(path.join(first, "review.md"), "# First")
+        await fs.writeFile(path.join(second, "deploy.md"), "# Second")
+      })
+      const started: string[] = []
+      const stopped: string[] = []
+      const active = new Set<(update: Watcher.Update) => void>()
+      const native = Watcher.Native.of({
+        subscribe: (input) =>
+          Effect.sync(() => {
+            started.push(input.target)
+            active.add(input.publish)
+            return {
+              unsubscribe: () => {
+                stopped.push(input.target)
+                active.delete(input.publish)
+                return Promise.resolve()
+              },
+            }
+          }),
+      })
+      const current = {
+        sources: [
+          Skill.DirectorySource.make({ type: "directory", path: AbsolutePath.make(first) }),
+          Skill.DirectorySource.make({ type: "directory", path: AbsolutePath.make(first) }),
+        ],
+      }
+      yield* Effect.gen(function* () {
+        yield* Effect.gen(function* () {
+          const changed = yield* Deferred.make<readonly Skill.Info[]>()
+          const observer = yield* SkillSourceObserver.make({
+            sources: () => {
+              // Source interpretation still happens after the old watches are released.
+              expect(active.size).toBe(0)
+              return current.sources
+            },
+            onChange: (): Effect.Effect<void> => Deferred.succeed(changed, observer.list()).pipe(Effect.asVoid),
+          })
+          expect(observer.list().map((skill) => skill.id)).toEqual([Skill.ID.make("review")])
+          expect(started).toEqual([first])
+          expect(stopped).toEqual([])
+          expect(active.size).toBe(1)
+
+          yield* observer.refresh()
+          expect(started).toEqual([first, first])
+          expect(stopped).toEqual([first])
+          expect(active.size).toBe(1)
+
+          current.sources = [Skill.DirectorySource.make({ type: "directory", path: AbsolutePath.make(second) })]
+          yield* observer.refresh()
+          expect(observer.list().map((skill) => skill.id)).toEqual([Skill.ID.make("deploy")])
+          expect(started).toEqual([first, first, second])
+          expect(stopped).toEqual([first, first])
+          expect(yield* Deferred.isDone(changed)).toBe(false)
+
+          const snapshot = observer.list()
+          const file = path.join(second, "deploy.md")
+          yield* Effect.promise(() => fs.writeFile(file, "# Updated"))
+          yield* Effect.sync(() => active.forEach((publish) => publish({ path: file, type: "update" })))
+          expect(yield* Deferred.await(changed).pipe(Effect.timeout("2 seconds"))).toMatchObject([
+            { id: "deploy", content: "# Updated" },
+          ])
+          expect(observer.list()[0]?.content).toBe("# Updated")
+          expect(snapshot[0]?.content).toBe("# Second")
+          expect(started).toEqual([first, first, second, second])
+          expect(stopped).toEqual([first, first, second])
+          expect(active.size).toBe(1)
+        }).pipe(Effect.scoped)
+
+        // The Watcher layer remains alive; only the observer's consumers were disposed.
+        expect(active.size).toBe(0)
+        expect(stopped).toEqual(started)
+      }).pipe(
+        Effect.provide(Watcher.layer().pipe(Layer.provide(Layer.succeed(Watcher.Native, native)))),
+        Effect.provide(AppNodeBuilder.build(LayerNode.group([FSUtil.node, SkillDiscovery.node]))),
+      )
+      expect(stopped).toEqual(started)
+    }),
+  )
+
+  it.live("pulls URL sources through SkillDiscovery on manual and filesystem refreshes", () =>
+    Effect.gen(function* () {
+      const tmp = yield* Effect.acquireDisposable(Effect.promise(() => tmpdir()))
+      const catalog = { version: "1", content: "# First", requests: [] as string[] }
+      const server = yield* NodeHttpServer.make(createServer, { host: "127.0.0.1", port: 0 })
+      const base = new URL("/catalog/", HttpServer.formatAddress(server.address)).href
+      yield* server.serve(
+        Effect.gen(function* () {
+          const request = yield* HttpServerRequest.HttpServerRequest
+          catalog.requests.push(request.url)
+          if (request.url === "/catalog/index.json") {
+            return HttpServerResponse.text(
+              JSON.stringify({ skills: [{ name: "review", version: catalog.version, files: ["SKILL.md"] }] }),
+            )
+          }
+          return HttpServerResponse.text(catalog.content)
+        }),
+      )
+      yield* Effect.gen(function* () {
+        const changed = yield* Deferred.make<void>()
+        const observer = yield* SkillSourceObserver.make({
+          sources: () => [Skill.UrlSource.make({ type: "url", url: base })],
+          onChange: () => Deferred.succeed(changed, undefined).pipe(Effect.asVoid),
+        })
+        expect(observer.list()).toMatchObject([{ id: "review", content: "# First" }])
+        expect(FSUtil.contains(tmp.path, observer.list()[0].location)).toBe(true)
+
+        catalog.version = "2"
+        catalog.content = "# Second"
+        yield* observer.refresh()
+        expect(observer.list()).toMatchObject([{ id: "review", content: "# Second" }])
+        expect(yield* Deferred.isDone(changed)).toBe(false)
+
+        catalog.version = "3"
+        catalog.content = "# Third"
+        const watcher = yield* Watcher.Test
+        yield* watcher.emit({ path: observer.list()[0].location, type: "update" })
+        yield* Deferred.await(changed).pipe(Effect.timeout("2 seconds"))
+        expect(observer.list()).toMatchObject([{ id: "review", content: "# Third" }])
+        expect(catalog.requests).toEqual([
+          "/catalog/index.json",
+          "/catalog/review/SKILL.md",
+          "/catalog/index.json",
+          "/catalog/review/SKILL.md",
+          "/catalog/index.json",
+          "/catalog/review/SKILL.md",
+        ])
+      }).pipe(
+        Effect.provide(
+          Layer.mergeAll(
+            AppNodeBuilder.build(LayerNode.group([FSUtil.node, SkillDiscovery.node]), [
+              [Global.node, Global.layerWith({ cache: tmp.path })],
+            ]),
+            Watcher.testLayer,
+          ),
+        ),
+      )
+    }),
+  )
+})