Mirror_Repos/anomalyco_opencode
2
0
Fork 0
Code Issues Pull Requests Actions 86 Packages Projects Releases Wiki Activity

fix(core): reload changed skill sources

Browse Source
This commit is contained in:
Kit Langton
2026-08-06 20:40:30 -04:00
parent 1eb3a43add
commit a6d120ccd1
4 changed files with 338 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/core/src/filesystem/watcher.ts
+23 -6
View File
@@ -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<Native, NativeInterface>()("@opencod
export interface Interface {
readonly subscribe: (input: WatchInput) => Effect.Effect<Stream.Stream<Update>>
readonly acquire: (input: WatchInput) => Effect.Effect<Stream.Stream<Update>, 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<Update>(), (pubsub) =>
PubSub.shutdown(pubsub),
)
const pubsub = yield* Effect.acquireRelease(PubSub.unbounded<Update>(), (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]),
})
packages/core/src/skill.ts
+109 -30
View File
@@ -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<string, { skills: Info[]; paths: readonly string[] }>()
const cacheLock = Semaphore.makeUnsafe(1)
const watched = new Set<string>()
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<Data, Draft>({
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<string, { skills: Info[]; directories: readonly string[] }>()
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<ID, Info>()
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<ID, Info>()
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],
})
packages/core/test/filesystem/watcher.test.ts
+19 -8
View File
@@ -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<void>()
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(
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
packages/core/test/skill.test.ts
+187 -2
View File
@@ -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)
}),
),
),
)
})