feat(core): reload configured plugins from source edits (#39224)
This commit is contained in:
parent
775f24f049
commit
debdea40ea
2 changed files with 90 additions and 22 deletions
|
|
@ -1,7 +1,8 @@
|
||||||
export * as PluginSupervisor from "./supervisor"
|
export * as PluginSupervisor from "./supervisor"
|
||||||
|
|
||||||
|
import type { Plugin as PluginDefinition } from "@opencode-ai/plugin/effect/plugin"
|
||||||
import { Event } from "@opencode-ai/schema/config"
|
import { Event } from "@opencode-ai/schema/config"
|
||||||
import { Context, Deferred, Effect, Layer, Option, Schema, Semaphore, Stream } from "effect"
|
import { Context, Deferred, Effect, Layer, Option, PubSub, Schema, Semaphore, Stream } from "effect"
|
||||||
import path from "path"
|
import path from "path"
|
||||||
import { fileURLToPath, pathToFileURL } from "url"
|
import { fileURLToPath, pathToFileURL } from "url"
|
||||||
import { Agent } from "../agent"
|
import { Agent } from "../agent"
|
||||||
|
|
@ -14,6 +15,7 @@ import { httpClient } from "@opencode-ai/util/effect/app-node-platform"
|
||||||
import { Bus } from "../bus"
|
import { Bus } from "../bus"
|
||||||
import { FileMutation } from "../file-mutation"
|
import { FileMutation } from "../file-mutation"
|
||||||
import { FileSystem } from "../filesystem"
|
import { FileSystem } from "../filesystem"
|
||||||
|
import { Watcher } from "../filesystem/watcher"
|
||||||
import { Form } from "../form"
|
import { Form } from "../form"
|
||||||
import { FSUtil } from "@opencode-ai/util/fs-util"
|
import { FSUtil } from "@opencode-ai/util/fs-util"
|
||||||
import { Global } from "@opencode-ai/util/global"
|
import { Global } from "@opencode-ai/util/global"
|
||||||
|
|
@ -45,9 +47,9 @@ const PluginModule = Schema.Struct({
|
||||||
default: Schema.Union([
|
default: Schema.Union([
|
||||||
Schema.Struct({
|
Schema.Struct({
|
||||||
id: Schema.String,
|
id: Schema.String,
|
||||||
effect: Schema.declare<import("@opencode-ai/plugin/effect/plugin").Plugin["effect"]>(
|
effect: Schema.declare<PluginDefinition["effect"]>(
|
||||||
(input): input is import("@opencode-ai/plugin/effect/plugin").Plugin["effect"] => typeof input === "function",
|
(input): input is PluginDefinition["effect"] => typeof input === "function",
|
||||||
),
|
),
|
||||||
}),
|
}),
|
||||||
Schema.Struct({
|
Schema.Struct({
|
||||||
id: Schema.String,
|
id: Schema.String,
|
||||||
|
|
@ -226,11 +228,43 @@ const layer = Layer.effect(
|
||||||
const sdk = yield* SdkPlugins.Service
|
const sdk = yield* SdkPlugins.Service
|
||||||
const config = yield* Config.Service
|
const config = yield* Config.Service
|
||||||
const bus = yield* Bus.Service
|
const bus = yield* Bus.Service
|
||||||
|
const watcher = yield* Watcher.Service
|
||||||
|
const fs = yield* FSUtil.Service
|
||||||
const lock = Semaphore.makeUnsafe(1)
|
const lock = Semaphore.makeUnsafe(1)
|
||||||
const ready = yield* Deferred.make<void>()
|
const ready = yield* Deferred.make<void>()
|
||||||
let observed = 0
|
let observed = 0
|
||||||
let applied = -1
|
let applied = -1
|
||||||
|
|
||||||
|
// Configured local plugin files can live outside config roots, where the
|
||||||
|
// config change feed cannot see them; watch those entrypoints directly.
|
||||||
|
// Watches start on first sighting and are never torn down individually:
|
||||||
|
// a stale watch after a config edit costs one deduped fs handle and a
|
||||||
|
// no-op activation, and every watch dies with this layer's scope.
|
||||||
|
const configuredChanges = yield* PubSub.unbounded<void>()
|
||||||
|
const watched = new Set<string>()
|
||||||
|
const watchConfiguredSources = Effect.fn("PluginSupervisor.watchConfiguredSources")(function* (
|
||||||
|
entries: readonly Config.Entry[],
|
||||||
|
operations: readonly Operation[],
|
||||||
|
) {
|
||||||
|
for (const operation of operations) {
|
||||||
|
if (operation.type !== "add" || !path.isAbsolute(operation.target)) continue
|
||||||
|
if (watched.has(operation.target)) continue
|
||||||
|
// The config change feed already covers {plugin,plugins} directories.
|
||||||
|
if (isPluginSource(entries, operation.target)) continue
|
||||||
|
// Directory targets can't hot-reload (their stat mtime ignores edits
|
||||||
|
// inside), so don't watch what can't trigger anything.
|
||||||
|
if (yield* fs.isDir(operation.target)) continue
|
||||||
|
watched.add(operation.target)
|
||||||
|
yield* watcher.subscribe({ path: operation.target, type: "file" }).pipe(
|
||||||
|
Stream.runForEach(() => PubSub.publish(configuredChanges, undefined)),
|
||||||
|
Effect.catchCause((cause) =>
|
||||||
|
Effect.logError("configured plugin watch failed", { target: operation.target, cause }),
|
||||||
|
),
|
||||||
|
Effect.forkScoped({ startImmediately: true }),
|
||||||
|
)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
const activate = Effect.fn("PluginSupervisor.activate")(function* (target: number) {
|
const activate = Effect.fn("PluginSupervisor.activate")(function* (target: number) {
|
||||||
yield* lock.withPermit(
|
yield* lock.withPermit(
|
||||||
Effect.gen(function* () {
|
Effect.gen(function* () {
|
||||||
|
|
@ -240,7 +274,9 @@ const layer = Layer.effect(
|
||||||
// Combine internal plugins with host-contributed SDK plugins in boot order.
|
// Combine internal plugins with host-contributed SDK plugins in boot order.
|
||||||
const pre = [...internal.pre.map((plugin) => ({ ...plugin, version: "internal" })), ...sdk.all()]
|
const pre = [...internal.pre.map((plugin) => ({ ...plugin, version: "internal" })), ...sdk.all()]
|
||||||
const post = internal.post.map((plugin) => ({ ...plugin, version: "internal" }))
|
const post = internal.post.map((plugin) => ({ ...plugin, version: "internal" }))
|
||||||
const operations = yield* scan(yield* config.entries())
|
const entries = yield* config.entries()
|
||||||
|
const operations = yield* scan(entries)
|
||||||
|
yield* watchConfiguredSources(entries, operations)
|
||||||
// Apply config operations and load enabled package plugins into one ordered generation.
|
// Apply config operations and load enabled package plugins into one ordered generation.
|
||||||
const plugins = yield* resolve(pre, post, operations)
|
const plugins = yield* resolve(pre, post, operations)
|
||||||
// Replace the active generation in one scoped, batched activation.
|
// Replace the active generation in one scoped, batched activation.
|
||||||
|
|
@ -249,26 +285,22 @@ const layer = Layer.effect(
|
||||||
}),
|
}),
|
||||||
)
|
)
|
||||||
})
|
})
|
||||||
const sourceChanges = config
|
const sourceChanges = config.changes().pipe(
|
||||||
.changes()
|
Stream.filterEffect((update) => Effect.map(config.entries(), (entries) => isPluginSource(entries, update.path))),
|
||||||
.pipe(
|
Stream.merge(Stream.fromPubSub(configuredChanges)),
|
||||||
Stream.filterEffect((update) =>
|
// Make accepted filesystem work visible to flush before coalescing the burst.
|
||||||
Effect.map(config.entries(), (entries) => isPluginSource(entries, update.path)),
|
Stream.mapEffect(() => Effect.sync(() => ++observed)),
|
||||||
),
|
Stream.debounce("100 millis"),
|
||||||
// Make accepted filesystem work visible to flush before coalescing the burst.
|
)
|
||||||
Stream.mapEffect(() => Effect.sync(() => ++observed)),
|
|
||||||
Stream.debounce("100 millis"),
|
|
||||||
)
|
|
||||||
const busUpdates = bus
|
const busUpdates = bus
|
||||||
.subscribe([Event.Updated, SdkPlugins.Updated])
|
.subscribe([Event.Updated, SdkPlugins.Updated])
|
||||||
.pipe(Stream.mapEffect(() => Effect.sync(() => ++observed)))
|
.pipe(Stream.mapEffect(() => Effect.sync(() => ++observed)))
|
||||||
const updates = yield* Stream.merge(busUpdates, sourceChanges).pipe(
|
const updates = yield* Stream.merge(busUpdates, sourceChanges).pipe(
|
||||||
Stream.toQueue({ capacity: 1, strategy: "sliding" }),
|
Stream.toQueue({ capacity: 1, strategy: "sliding" }),
|
||||||
)
|
)
|
||||||
const signals = yield* Stream.concat(
|
const signals = yield* Stream.concat(Stream.succeed(0), Stream.fromQueue(updates)).pipe(
|
||||||
Stream.succeed(0),
|
Stream.broadcast({ capacity: 1, strategy: "sliding", replay: 1 }),
|
||||||
Stream.fromQueue(updates),
|
)
|
||||||
).pipe(Stream.broadcast({ capacity: 1, strategy: "sliding", replay: 1 }))
|
|
||||||
const attempt = (target: number) =>
|
const attempt = (target: number) =>
|
||||||
activate(target).pipe(
|
activate(target).pipe(
|
||||||
Effect.map(() => observed === target),
|
Effect.map(() => observed === target),
|
||||||
|
|
@ -329,6 +361,7 @@ export const node = makeLocationNode({
|
||||||
Shell.node,
|
Shell.node,
|
||||||
Skill.node,
|
Skill.node,
|
||||||
Tool.node,
|
Tool.node,
|
||||||
|
Watcher.node,
|
||||||
WebSearch.node,
|
WebSearch.node,
|
||||||
WellKnown.node,
|
WellKnown.node,
|
||||||
],
|
],
|
||||||
|
|
|
||||||
|
|
@ -195,6 +195,41 @@ describe("PluginSupervisor config", () => {
|
||||||
),
|
),
|
||||||
)
|
)
|
||||||
|
|
||||||
|
it.live("reloads a configured plugin when its source file changes", () =>
|
||||||
|
withLocation(
|
||||||
|
{ plugins: ["-*", "./external/mutable.ts"] },
|
||||||
|
Effect.gen(function* () {
|
||||||
|
yield* ready()
|
||||||
|
const agents = yield* Agent.Service
|
||||||
|
const bus = yield* Bus.Service
|
||||||
|
const location = yield* Location.Service
|
||||||
|
const file = path.join(location.directory, "external", "mutable.ts")
|
||||||
|
|
||||||
|
expect((yield* agents.get(Agent.ID.make("mutable")))?.description).toBe("first")
|
||||||
|
|
||||||
|
const changed = yield* bus
|
||||||
|
.subscribe(Plugin.Event.Updated)
|
||||||
|
.pipe(Stream.take(1), Stream.runDrain, Effect.forkScoped({ startImmediately: true }))
|
||||||
|
yield* Effect.promise(async () => {
|
||||||
|
await fs.writeFile(file, mutablePlugin("second"))
|
||||||
|
const modified = new Date(Date.now() + 5_000)
|
||||||
|
await fs.utimes(file, modified, modified)
|
||||||
|
})
|
||||||
|
yield* Fiber.join(changed).pipe(Effect.timeout("5 seconds"))
|
||||||
|
|
||||||
|
expect((yield* agents.get(Agent.ID.make("mutable")))?.description).toBe("second")
|
||||||
|
}),
|
||||||
|
false,
|
||||||
|
async (directory) => {
|
||||||
|
// Outside any {plugin,plugins} config-source directory, so only the
|
||||||
|
// configured-entrypoint watch can observe the edit.
|
||||||
|
const external = path.join(directory, "external")
|
||||||
|
await fs.mkdir(external, { recursive: true })
|
||||||
|
await fs.writeFile(path.join(external, "mutable.ts"), mutablePlugin("first"))
|
||||||
|
},
|
||||||
|
),
|
||||||
|
)
|
||||||
|
|
||||||
it.live("applies explicit removals after auto-discovery", () =>
|
it.live("applies explicit removals after auto-discovery", () =>
|
||||||
withLocation(
|
withLocation(
|
||||||
{ plugins: ["-*"] },
|
{ plugins: ["-*"] },
|
||||||
|
|
@ -251,9 +286,9 @@ describe("PluginSupervisor config", () => {
|
||||||
expect((yield* registry.list()).map((plugin) => String(plugin.id))).not.toContain("opencode.variant")
|
expect((yield* registry.list()).map((plugin) => String(plugin.id))).not.toContain("opencode.variant")
|
||||||
|
|
||||||
const catalog = yield* Catalog.Service
|
const catalog = yield* Catalog.Service
|
||||||
expect(
|
expect((yield* catalog.model.get(Provider.ID.make("configured"), Model.ID.make("glm-5.2")))?.variants).toEqual([
|
||||||
(yield* catalog.model.get(Provider.ID.make("configured"), Model.ID.make("glm-5.2")))?.variants,
|
expect.objectContaining({ id: "high", headers: { custom: "true" } }),
|
||||||
).toEqual([expect.objectContaining({ id: "high", headers: { custom: "true" } })])
|
])
|
||||||
}),
|
}),
|
||||||
),
|
),
|
||||||
)
|
)
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue