From 9de4a3516e65d38aafabfee4bfc7ce4087cde0ba Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Mon, 10 Aug 2026 18:43:07 -0400 Subject: [PATCH 1/2] refactor(core): plugin supervisor consumes sources, config side owns discovery and watching --- packages/core/src/config/plugin/source.ts | 175 ++++++++++++ packages/core/src/plugin/supervisor.ts | 330 +++++++--------------- packages/core/test/config/plugin.test.ts | 28 ++ 3 files changed, 310 insertions(+), 223 deletions(-) create mode 100644 packages/core/src/config/plugin/source.ts diff --git a/packages/core/src/config/plugin/source.ts b/packages/core/src/config/plugin/source.ts new file mode 100644 index 000000000000..9b689b9c9a0e --- /dev/null +++ b/packages/core/src/config/plugin/source.ts @@ -0,0 +1,175 @@ +export * as ConfigPluginSource from "./source" + +import { Directory, Document, type Entry } from "@opencode-ai/schema/config" +import { ConfigPlugin } from "@opencode-ai/schema/config/plugin" +import { FSUtil } from "@opencode-ai/util/fs-util" +import { LayerNode } from "@opencode-ai/util/effect/layer-node" +import { Context, Effect, Layer, Option, PubSub, Scope, Stream } from "effect" +import path from "path" +import { fileURLToPath } from "url" +import { Config } from "../../config" +import { Watcher } from "../../filesystem/watcher" +import { Location } from "../../location" + +export type Operation = + | { + readonly type: "add" + readonly target: string + readonly options: Record + readonly mtime?: number + } + | { + readonly type: "remove" + readonly target: string + } + +export interface Interface { + readonly operations: () => Effect.Effect + readonly changes: () => Stream.Stream +} + +export class Service extends Context.Service()("@opencode/ConfigPluginSource") {} + +export type Options = { + readonly dynamic?: boolean +} + +export const layer = (options?: Options) => + Layer.effect( + Service, + Effect.gen(function* () { + if (options?.dynamic === false) { + return Service.of({ + operations: () => Effect.succeed([]), + changes: () => Stream.empty, + }) + } + + const config = yield* Config.Service + const watcher = yield* Watcher.Service + const fs = yield* FSUtil.Service + const location = yield* Location.Service + const configuredChanges = yield* PubSub.unbounded() + const watched = new Set() + + // 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 watchConfiguredSources = Effect.fn("ConfigPluginSource.watchConfiguredSources")(function* ( + entries: readonly 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) + const updates = yield* watcher.subscribe({ path: operation.target, type: "file" }) + yield* updates.pipe( + Stream.runForEach(() => PubSub.publish(configuredChanges, undefined)), + Effect.catchCause((cause) => + Effect.logError("configured plugin watch failed", { target: operation.target, cause }), + ), + Effect.forkScoped({ startImmediately: true }), + ) + } + }) + + return Service.of({ + operations: Effect.fn("ConfigPluginSource.operations")(function* () { + const entries = yield* config.entries() + const operations = yield* scan(fs, location, entries) + yield* watchConfiguredSources(entries, operations) + return operations + }), + changes: () => + Stream.merge( + config.changes().pipe( + Stream.filterEffect((update) => + Effect.map(config.entries(), (entries) => isPluginSource(entries, update.path)), + ), + Stream.map(() => undefined), + ), + Stream.fromPubSub(configuredChanges), + ), + }) + }), + ) + +export const requirements = LayerNode.group([Config.node, FSUtil.node, Location.node, Watcher.node]) + +function parse(input: ConfigPlugin.Plugin): Operation { + if (typeof input !== "string") { + return { type: "add", target: input.package, options: input.options ?? {} } + } + if (!input.startsWith("-")) return { type: "add", target: input, options: {} } + if (input.length === 1) throw new Error("Plugin remove operation requires a target") + return { type: "remove", target: input.slice(1) } +} + +const scan = Effect.fn("ConfigPluginSource.scan")(function* ( + fs: FSUtil.Interface, + location: Location.Interface, + entries: readonly Entry[], +) { + const discovered = yield* Effect.forEach( + entries.filter((entry): entry is Directory => entry.type === "directory"), + (entry) => discoverDirectory(fs, entry.path), + ).pipe(Effect.map((items) => items.flat())) + const configured = entries + .filter((entry): entry is Document => entry.type === "document") + .flatMap((entry) => + (entry.info.plugins ?? []).map(parse).map((operation) => { + if (operation.type === "remove") return operation + const directory = entry.path ? path.dirname(entry.path) : location.directory + const target = operation.target.startsWith("file://") + ? fileURLToPath(operation.target) + : operation.target.startsWith("./") || operation.target.startsWith("../") + ? path.resolve(directory, operation.target) + : operation.target + return { ...operation, target } + }), + ) + // Explicit config is applied last so it can remove auto-discovered packages. + return yield* Effect.forEach([...discovered, ...configured], (operation) => { + if (operation.type === "remove" || !path.isAbsolute(operation.target)) return Effect.succeed(operation) + return fs.stat(operation.target).pipe( + Effect.map((info) => ({ + ...operation, + mtime: Option.getOrElse(info.mtime, () => new Date(0)).getTime(), + })), + Effect.catch(() => Effect.succeed(operation)), + ) + }) +}) + +function discoverDirectory(fs: FSUtil.Interface, directory: string) { + return Effect.gen(function* () { + const files = yield* fs + .scan("{plugin,plugins}/*.{ts,js}", { + cwd: directory, + absolute: true, + include: "file", + dot: true, + symlink: true, + }) + .pipe(Effect.orElseSucceed(() => [])) + return files.sort().map((target): Operation => ({ type: "add", target, options: {} })) + }) +} + +const sourceDirectories = ["plugin", "plugins"] as const + +function isPluginSource(entries: readonly Entry[], file: string) { + return entries.some( + (entry) => + entry.type === "directory" && + sourceDirectories.some((directory) => FSUtil.contains(path.join(entry.path, directory), file)), + ) +} diff --git a/packages/core/src/plugin/supervisor.ts b/packages/core/src/plugin/supervisor.ts index 9343424ac25b..ddb7b1ca3d89 100644 --- a/packages/core/src/plugin/supervisor.ts +++ b/packages/core/src/plugin/supervisor.ts @@ -1,15 +1,14 @@ export * as PluginSupervisor from "./supervisor" import type { Plugin as PluginDefinition } from "@opencode-ai/plugin/effect/plugin" -import { Directory, Document, Event, type Entry } from "@opencode-ai/schema/config" -import { ConfigPlugin } from "@opencode-ai/schema/config/plugin" -import { Context, Deferred, Effect, Layer, Option, PubSub, Schema, Stream } from "effect" +import { Event } from "@opencode-ai/schema/config" +import { Context, Deferred, Effect, Layer, Schema, Stream } from "effect" import path from "path" -import { fileURLToPath, pathToFileURL } from "url" +import { pathToFileURL } from "url" import { Agent } from "../agent" import { Catalog } from "../catalog" import { Command } from "../command" -import { Config } from "../config" +import { ConfigPluginSource } from "../config/plugin/source" import { Credential } from "../credential" import { makeLocationNode } from "@opencode-ai/util/effect/app-node" import { httpClient } from "@opencode-ai/util/effect/app-node-platform" @@ -18,14 +17,11 @@ import { Environment } from "../environment" import { FileMutation } from "../file-mutation" import { Formatter } from "../formatter" import { FileSystem } from "../filesystem" -import { Watcher } from "../filesystem/watcher" import { Form } from "../form" -import { FSUtil } from "@opencode-ai/util/fs-util" import { Global } from "@opencode-ai/util/global" import { Image } from "../image" import { Integration } from "../integration" import { KV } from "../kv" -import { Location } from "../location" import { LocationMutation } from "../location-mutation" import { ModelsDev } from "../models-dev" import { Npm } from "@opencode-ai/util/npm" @@ -63,65 +59,10 @@ const PluginModule = Schema.Struct({ ]), }) -type Operation = - | { - readonly type: "add" - readonly target: string - readonly options: Record - readonly mtime?: number - } - | { - readonly type: "remove" - readonly target: string - } - -function parse(input: ConfigPlugin.Plugin): Operation { - if (typeof input !== "string") { - return { type: "add", target: input.package, options: input.options ?? {} } - } - if (!input.startsWith("-")) return { type: "add", target: input, options: {} } - if (input.length === 1) throw new Error("Plugin remove operation requires a target") - return { type: "remove", target: input.slice(1) } -} - -const scan = Effect.fn("PluginSupervisor.scan")(function* (entries: readonly Entry[]) { - const fs = yield* FSUtil.Service - const location = yield* Location.Service - const discovered = yield* Effect.forEach( - entries.filter((entry): entry is Directory => entry.type === "directory"), - (entry) => discoverDirectory(fs, entry.path), - ).pipe(Effect.map((items) => items.flat())) - const configured = entries - .filter((entry): entry is Document => entry.type === "document") - .flatMap((entry) => - (entry.info.plugins ?? []).map(parse).map((operation) => { - if (operation.type === "remove") return operation - const directory = entry.path ? path.dirname(entry.path) : location.directory - const target = operation.target.startsWith("file://") - ? fileURLToPath(operation.target) - : operation.target.startsWith("./") || operation.target.startsWith("../") - ? path.resolve(directory, operation.target) - : operation.target - return { ...operation, target } - }), - ) - // Explicit config is applied last so it can remove auto-discovered packages. - return yield* Effect.forEach([...discovered, ...configured], (operation) => { - if (operation.type === "remove" || !path.isAbsolute(operation.target)) return Effect.succeed(operation) - return fs.stat(operation.target).pipe( - Effect.map((info) => ({ - ...operation, - mtime: Option.getOrElse(info.mtime, () => new Date(0)).getTime(), - })), - Effect.catch(() => Effect.succeed(operation)), - ) - }) -}) - const resolve = Effect.fn("PluginSupervisor.resolve")(function* ( pre: readonly Plugin.Versioned[], post: readonly Plugin.Versioned[], - operations: readonly Operation[], + operations: readonly ConfigPluginSource.Operation[], ) { const matches = (selector: string, target: string) => selector === "*" || (selector.endsWith(".*") ? target.startsWith(selector.slice(0, -1)) : selector === target) @@ -168,7 +109,9 @@ const resolve = Effect.fn("PluginSupervisor.resolve")(function* ( ] }) -const load = Effect.fn("PluginSupervisor.load")(function* (operation: Extract) { +const load = Effect.fn("PluginSupervisor.load")(function* ( + operation: Extract, +) { const npm = yield* Npm.Service const entrypoint = path.isAbsolute(operation.target) ? pathToFileURL(operation.target).href @@ -192,171 +135,112 @@ const load = Effect.fn("PluginSupervisor.load")(function* (operation: Extract [])) - return files.sort().map((target): Operation => ({ type: "add", target, options: {} })) - }) -} - -const sourceDirectories = ["plugin", "plugins"] as const - -function isPluginSource(entries: readonly Entry[], file: string) { - return entries.some( - (entry) => - entry.type === "directory" && - sourceDirectories.some((directory) => FSUtil.contains(path.join(entry.path, directory), file)), - ) -} - export interface Interface { /** Wait for the initial plugin generation and startup updates to settle. */ readonly flush: Effect.Effect } -export class Service extends Context.Service()("@opencode/PluginSupervisor") {} - -const layer = Layer.effect( - Service, - Effect.gen(function* () { - const registry = yield* Plugin.Service - const sdk = yield* SdkPlugins.Service - const config = yield* Config.Service - const bus = yield* Bus.Service - const watcher = yield* Watcher.Service - const fs = yield* FSUtil.Service - const ready = { current: yield* Deferred.make() } - let observed = 0 +export const Options = Schema.Struct({ + /** Set false to activate only precompiled (internal and SDK) plugins, skipping config-declared + * packages and plugin directories entirely: no filesystem scan, npm install, or disk import. */ + dynamic: Schema.optional(Schema.Boolean), +}) +export type Options = typeof Options.Type - // 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() - const watched = new Set() - const watchConfiguredSources = Effect.fn("PluginSupervisor.watchConfiguredSources")(function* ( - entries: readonly 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) - const updates = yield* watcher.subscribe({ path: operation.target, type: "file" }) - yield* updates.pipe( - Stream.runForEach(() => PubSub.publish(configuredChanges, undefined)), - Effect.catchCause((cause) => - Effect.logError("configured plugin watch failed", { target: operation.target, cause }), - ), - Effect.forkScoped({ startImmediately: true }), - ) - } - }) +export class Service extends Context.Service()("@opencode/PluginSupervisor") {} - const activate = Effect.fn("PluginSupervisor.activate")(function* () { - // Resolve OpenCode's internal plugins with their privileged Location services. - const internal = yield* PluginInternal.list() - // Combine internal plugins with host-contributed SDK plugins in boot order. - const pre = [...internal.pre.map((plugin) => ({ ...plugin, version: "internal" })), ...sdk.all()] - const post = internal.post.map((plugin) => ({ ...plugin, version: "internal" })) - 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. - const plugins = yield* resolve(pre, post, operations) - // Replace the active generation in one scoped, batched activation. - yield* registry.activate(plugins) - }) - const updates = Stream.merge( - config.changes().pipe( - Stream.filterEffect((update) => - Effect.map(config.entries(), (entries) => isPluginSource(entries, update.path)), +const makeLayer = (options?: Options) => + Layer.effect( + Service, + Effect.gen(function* () { + const registry = yield* Plugin.Service + const sdk = yield* SdkPlugins.Service + const sources = yield* ConfigPluginSource.Service + const bus = yield* Bus.Service + const ready = { current: yield* Deferred.make() } + let observed = 0 + + const activate = Effect.fn("PluginSupervisor.activate")(function* () { + // Resolve OpenCode's internal plugins with their privileged Location services. + const internal = yield* PluginInternal.list() + // Combine internal plugins with host-contributed SDK plugins in boot order. + const pre = [...internal.pre.map((plugin) => ({ ...plugin, version: "internal" })), ...sdk.all()] + const post = internal.post.map((plugin) => ({ ...plugin, version: "internal" })) + const operations = yield* sources.operations() + // Apply config operations and load enabled package plugins into one ordered generation. + const plugins = yield* resolve(pre, post, operations) + // Replace the active generation in one scoped, batched activation. + yield* registry.activate(plugins) + }) + const updates = Stream.merge(sources.changes(), bus.subscribe([Event.Updated, SdkPlugins.Updated])).pipe( + // Make accepted work visible to flush before coalescing the burst. + Stream.mapEffect(() => + Effect.gen(function* () { + observed++ + if (yield* Deferred.isDone(ready.current)) ready.current = yield* Deferred.make() + return observed + }), ), - Stream.merge(Stream.fromPubSub(configuredChanges)), - ), - bus.subscribe([Event.Updated, SdkPlugins.Updated]), - ).pipe( - // Make accepted work visible to flush before coalescing the burst. - Stream.mapEffect(() => - Effect.gen(function* () { - observed++ - if (yield* Deferred.isDone(ready.current)) ready.current = yield* Deferred.make() - return observed - }), - ), - ) - yield* Stream.concat(Stream.succeed(0), updates).pipe( - // Keep observing updates while activation runs, retaining only the latest generation request. - Stream.buffer({ capacity: 1, strategy: "sliding" }), - Stream.debounce("100 millis"), - Stream.runForEach((target) => - Effect.gen(function* () { - yield* activate() - if (observed === target) yield* Deferred.succeed(ready.current, undefined) - }).pipe(Effect.catchCause((cause) => Effect.logError("failed to reload plugins", { cause }))), - ), - Effect.forkScoped({ startImmediately: true }), - ) - return Service.of({ flush: Effect.suspend(() => Deferred.await(ready.current)) }) - }), -) - -const nodeLayer = layer as Layer.Layer + ) + yield* Stream.concat(Stream.succeed(0), updates).pipe( + // Keep observing updates while activation runs, retaining only the latest generation request. + Stream.buffer({ capacity: 1, strategy: "sliding" }), + Stream.debounce("100 millis"), + Stream.runForEach((target) => + Effect.gen(function* () { + yield* activate() + if (observed === target) yield* Deferred.succeed(ready.current, undefined) + }).pipe(Effect.catchCause((cause) => Effect.logError("failed to reload plugins", { cause }))), + ), + Effect.forkScoped({ startImmediately: true }), + ) + return Service.of({ flush: Effect.suspend(() => Deferred.await(ready.current)) }) + }), + ).pipe(Layer.provide(ConfigPluginSource.layer(options))) -export const node = makeLocationNode({ - service: Service, - layer: nodeLayer, - deps: [ - Plugin.node, - SdkPlugins.node, - Agent.node, - Catalog.node, - Command.node, - Config.node, - Credential.node, - Bus.node, - Environment.node, - FileMutation.node, - Formatter.node, - FileSystem.node, - FSUtil.node, - Global.node, - httpClient, - Image.node, - Integration.node, - KV.node, - Location.node, - LocationMutation.node, - ModelsDev.node, - Npm.node, - Permission.node, - PluginRuntime.node, - Form.node, - ReadToolFileSystem.node, - Reference.node, - Ripgrep.node, - SessionInstructions.node, - Shell.node, - Skill.node, - Tool.node, - Watcher.node, - WebSearch.node, - WellKnown.node, - ], -}) +export function configured(options?: Options) { + return makeLocationNode({ + service: Service, + layer: makeLayer(options), + deps: nodeDeps, + }) +} -export { layer } +const nodeDeps = [ + Plugin.node, + SdkPlugins.node, + Agent.node, + Catalog.node, + Command.node, + ConfigPluginSource.requirements, + Credential.node, + Bus.node, + Environment.node, + FileMutation.node, + Formatter.node, + FileSystem.node, + Global.node, + httpClient, + Image.node, + Integration.node, + KV.node, + LocationMutation.node, + ModelsDev.node, + Npm.node, + Permission.node, + PluginRuntime.node, + Form.node, + ReadToolFileSystem.node, + Reference.node, + Ripgrep.node, + SessionInstructions.node, + Shell.node, + Skill.node, + Tool.node, + WebSearch.node, + WellKnown.node, +] as const + +export const node = configured() + +export const layer = makeLayer() diff --git a/packages/core/test/config/plugin.test.ts b/packages/core/test/config/plugin.test.ts index 65a91c6f5053..a51ce830d5f3 100644 --- a/packages/core/test/config/plugin.test.ts +++ b/packages/core/test/config/plugin.test.ts @@ -24,6 +24,11 @@ import { testEffect } from "../lib/effect" const it = testEffect( AppNodeBuilder.build(LayerNode.group([Database.node, Bus.node, SdkPlugins.node, LocationServiceMap.node])), ) +const staticIt = testEffect( + AppNodeBuilder.build(LayerNode.group([Database.node, Bus.node, SdkPlugins.node, LocationServiceMap.node]), [ + [PluginSupervisor.node, PluginSupervisor.configured({ dynamic: false })], + ]), +) describe("PluginSupervisor config", () => { it.live("applies selectors in order", () => @@ -157,6 +162,29 @@ describe("PluginSupervisor config", () => { ), ) + staticIt.live("uses only internal and SDK plugins when dynamic imports are disabled", () => + Effect.gen(function* () { + const sdk = yield* SdkPlugins.Service + yield* sdk.register(EffectPlugin.define({ id: "static-sdk", effect: () => Effect.void })) + yield* withLocation( + { plugins: ["-*", path.join(import.meta.dir, "../plugin/fixtures/config-promise-plugin.ts")] }, + Effect.gen(function* () { + yield* ready() + const plugins = yield* Plugin.Service + const ids = (yield* plugins.list()).map((plugin) => String(plugin.id)) + expect(ids).toContain("opencode.agent") + expect(ids).toContain("static-sdk") + expect(ids).not.toContain("config-promise-plugin") + + const agents = yield* Agent.Service + expect(yield* agents.get(Agent.ID.make("directory"))).toBeUndefined() + expect(yield* agents.get(Agent.ID.make("configured"))).toBeUndefined() + }), + true, + ) + }), + ) + it.live("reloads an auto-discovered plugin when its file changes", () => withLocation( undefined, From d2b479bcb75762e04a44e48b67df8936391656d5 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Mon, 10 Aug 2026 18:50:03 -0400 Subject: [PATCH 2/2] refactor(core): hoist plugin source wiring --- packages/core/src/config/plugin/source.ts | 140 +++++++++--------- packages/core/src/plugin/internal.ts | 37 +++++ packages/core/src/plugin/supervisor.ts | 171 +++++++--------------- packages/core/test/config/plugin.test.ts | 5 +- 4 files changed, 162 insertions(+), 191 deletions(-) diff --git a/packages/core/src/config/plugin/source.ts b/packages/core/src/config/plugin/source.ts index 9b689b9c9a0e..8123a0949a45 100644 --- a/packages/core/src/config/plugin/source.ts +++ b/packages/core/src/config/plugin/source.ts @@ -3,7 +3,7 @@ export * as ConfigPluginSource from "./source" import { Directory, Document, type Entry } from "@opencode-ai/schema/config" import { ConfigPlugin } from "@opencode-ai/schema/config/plugin" import { FSUtil } from "@opencode-ai/util/fs-util" -import { LayerNode } from "@opencode-ai/util/effect/layer-node" +import { makeLocationNode } from "@opencode-ai/util/effect/app-node" import { Context, Effect, Layer, Option, PubSub, Scope, Stream } from "effect" import path from "path" import { fileURLToPath } from "url" @@ -30,79 +30,83 @@ export interface Interface { export class Service extends Context.Service()("@opencode/ConfigPluginSource") {} -export type Options = { - readonly dynamic?: boolean -} +export const layer = Layer.effect( + Service, + Effect.gen(function* () { + const config = yield* Config.Service + const watcher = yield* Watcher.Service + const fs = yield* FSUtil.Service + const location = yield* Location.Service + const configuredChanges = yield* PubSub.unbounded() + const watched = new Set() -export const layer = (options?: Options) => - Layer.effect( - Service, - Effect.gen(function* () { - if (options?.dynamic === false) { - return Service.of({ - operations: () => Effect.succeed([]), - changes: () => Stream.empty, - }) + // 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 watchConfiguredSources = Effect.fn("ConfigPluginSource.watchConfiguredSources")(function* ( + entries: readonly 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) + const updates = yield* watcher.subscribe({ path: operation.target, type: "file" }) + yield* updates.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 config = yield* Config.Service - const watcher = yield* Watcher.Service - const fs = yield* FSUtil.Service - const location = yield* Location.Service - const configuredChanges = yield* PubSub.unbounded() - const watched = new Set() - - // 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 watchConfiguredSources = Effect.fn("ConfigPluginSource.watchConfiguredSources")(function* ( - entries: readonly 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) - const updates = yield* watcher.subscribe({ path: operation.target, type: "file" }) - yield* updates.pipe( - Stream.runForEach(() => PubSub.publish(configuredChanges, undefined)), - Effect.catchCause((cause) => - Effect.logError("configured plugin watch failed", { target: operation.target, cause }), - ), - Effect.forkScoped({ startImmediately: true }), - ) - } - }) - - return Service.of({ - operations: Effect.fn("ConfigPluginSource.operations")(function* () { - const entries = yield* config.entries() - const operations = yield* scan(fs, location, entries) - yield* watchConfiguredSources(entries, operations) - return operations - }), - changes: () => - Stream.merge( - config.changes().pipe( - Stream.filterEffect((update) => - Effect.map(config.entries(), (entries) => isPluginSource(entries, update.path)), - ), - Stream.map(() => undefined), + return Service.of({ + operations: Effect.fn("ConfigPluginSource.operations")(function* () { + const entries = yield* config.entries() + const operations = yield* scan(fs, location, entries) + yield* watchConfiguredSources(entries, operations) + return operations + }), + changes: () => + Stream.merge( + config.changes().pipe( + Stream.filterEffect((update) => + Effect.map(config.entries(), (entries) => isPluginSource(entries, update.path)), ), - Stream.fromPubSub(configuredChanges), + Stream.map(() => undefined), ), - }) - }), - ) + Stream.fromPubSub(configuredChanges), + ), + }) + }), +) + +export const node = makeLocationNode({ + service: Service, + layer, + deps: [Config.node, FSUtil.node, Watcher.node, Location.node], +}) -export const requirements = LayerNode.group([Config.node, FSUtil.node, Location.node, Watcher.node]) +export const empty = makeLocationNode({ + service: Service, + layer: Layer.succeed( + Service, + Service.of({ + operations: () => Effect.succeed([]), + changes: () => Stream.never, + }), + ), + deps: [], +}) function parse(input: ConfigPlugin.Plugin): Operation { if (typeof input !== "string") { diff --git a/packages/core/src/plugin/internal.ts b/packages/core/src/plugin/internal.ts index a76d6994abef..2d2c4b3f27b4 100644 --- a/packages/core/src/plugin/internal.ts +++ b/packages/core/src/plugin/internal.ts @@ -1,6 +1,8 @@ export * as PluginInternal from "./internal" import type { Plugin } from "@opencode-ai/plugin/effect/plugin" +import { LayerNode } from "@opencode-ai/util/effect/layer-node" +import { httpClient } from "@opencode-ai/util/effect/app-node-platform" import { Context, Effect, Scope } from "effect" import { HttpClient } from "effect/unstable/http" import { Agent } from "../agent" @@ -137,6 +139,41 @@ type ContextServices = A extends Context.Context ? R : never export type Requirements = ContextServices>> +export const requirements = LayerNode.group([ + Agent.node, + Catalog.node, + Command.node, + Config.node, + Credential.node, + Bus.node, + Environment.node, + FileMutation.node, + Formatter.node, + FileSystem.node, + FSUtil.node, + Global.node, + httpClient, + Image.node, + Integration.node, + KV.node, + Location.node, + LocationMutation.node, + ModelsDev.node, + Npm.node, + Permission.node, + PluginRuntime.node, + Form.node, + ReadToolFileSystem.node, + Reference.node, + WebSearch.node, + Ripgrep.node, + SessionInstructions.node, + Shell.node, + Skill.node, + Tool.node, + WellKnown.node, +]) + export type InternalPlugin = Plugin const pre = [ diff --git a/packages/core/src/plugin/supervisor.ts b/packages/core/src/plugin/supervisor.ts index ddb7b1ca3d89..4e60e4a434b8 100644 --- a/packages/core/src/plugin/supervisor.ts +++ b/packages/core/src/plugin/supervisor.ts @@ -5,40 +5,13 @@ import { Event } from "@opencode-ai/schema/config" import { Context, Deferred, Effect, Layer, Schema, Stream } from "effect" import path from "path" import { pathToFileURL } from "url" -import { Agent } from "../agent" -import { Catalog } from "../catalog" -import { Command } from "../command" import { ConfigPluginSource } from "../config/plugin/source" -import { Credential } from "../credential" import { makeLocationNode } from "@opencode-ai/util/effect/app-node" -import { httpClient } from "@opencode-ai/util/effect/app-node-platform" import { Bus } from "../bus" -import { Environment } from "../environment" -import { FileMutation } from "../file-mutation" -import { Formatter } from "../formatter" -import { FileSystem } from "../filesystem" -import { Form } from "../form" -import { Global } from "@opencode-ai/util/global" -import { Image } from "../image" -import { Integration } from "../integration" -import { KV } from "../kv" -import { LocationMutation } from "../location-mutation" -import { ModelsDev } from "../models-dev" import { Npm } from "@opencode-ai/util/npm" -import { Permission } from "../permission" import { Plugin } from "../plugin" import { PluginPromise } from "../plugin/promise" -import { Reference } from "../reference" -import { Ripgrep } from "../ripgrep" -import { SessionInstructions } from "../session/instructions" -import { Shell } from "../shell" -import { Skill } from "../skill" -import { ReadToolFileSystem } from "../tool/read-filesystem" -import { Tool } from "../tool" -import { WebSearch } from "../websearch" -import { WellKnown } from "../wellknown" import { PluginInternal } from "./internal" -import { PluginRuntime } from "./runtime" import { SdkPlugins } from "./sdk" import { importModule } from "@opencode-ai/util/runtime-import" @@ -140,107 +113,63 @@ export interface Interface { readonly flush: Effect.Effect } -export const Options = Schema.Struct({ - /** Set false to activate only precompiled (internal and SDK) plugins, skipping config-declared - * packages and plugin directories entirely: no filesystem scan, npm install, or disk import. */ - dynamic: Schema.optional(Schema.Boolean), -}) -export type Options = typeof Options.Type - export class Service extends Context.Service()("@opencode/PluginSupervisor") {} -const makeLayer = (options?: Options) => - Layer.effect( - Service, - Effect.gen(function* () { - const registry = yield* Plugin.Service - const sdk = yield* SdkPlugins.Service - const sources = yield* ConfigPluginSource.Service - const bus = yield* Bus.Service - const ready = { current: yield* Deferred.make() } - let observed = 0 - - const activate = Effect.fn("PluginSupervisor.activate")(function* () { - // Resolve OpenCode's internal plugins with their privileged Location services. - const internal = yield* PluginInternal.list() - // Combine internal plugins with host-contributed SDK plugins in boot order. - const pre = [...internal.pre.map((plugin) => ({ ...plugin, version: "internal" })), ...sdk.all()] - const post = internal.post.map((plugin) => ({ ...plugin, version: "internal" })) - const operations = yield* sources.operations() - // Apply config operations and load enabled package plugins into one ordered generation. - const plugins = yield* resolve(pre, post, operations) - // Replace the active generation in one scoped, batched activation. - yield* registry.activate(plugins) - }) - const updates = Stream.merge(sources.changes(), bus.subscribe([Event.Updated, SdkPlugins.Updated])).pipe( - // Make accepted work visible to flush before coalescing the burst. - Stream.mapEffect(() => - Effect.gen(function* () { - observed++ - if (yield* Deferred.isDone(ready.current)) ready.current = yield* Deferred.make() - return observed - }), - ), - ) - yield* Stream.concat(Stream.succeed(0), updates).pipe( - // Keep observing updates while activation runs, retaining only the latest generation request. - Stream.buffer({ capacity: 1, strategy: "sliding" }), - Stream.debounce("100 millis"), - Stream.runForEach((target) => - Effect.gen(function* () { - yield* activate() - if (observed === target) yield* Deferred.succeed(ready.current, undefined) - }).pipe(Effect.catchCause((cause) => Effect.logError("failed to reload plugins", { cause }))), - ), - Effect.forkScoped({ startImmediately: true }), - ) - return Service.of({ flush: Effect.suspend(() => Deferred.await(ready.current)) }) - }), - ).pipe(Layer.provide(ConfigPluginSource.layer(options))) - -export function configured(options?: Options) { - return makeLocationNode({ - service: Service, - layer: makeLayer(options), - deps: nodeDeps, - }) -} +export const layer = Layer.effect( + Service, + Effect.gen(function* () { + const registry = yield* Plugin.Service + const sdk = yield* SdkPlugins.Service + const sources = yield* ConfigPluginSource.Service + const bus = yield* Bus.Service + const ready = { current: yield* Deferred.make() } + let observed = 0 + + const activate = Effect.fn("PluginSupervisor.activate")(function* () { + // Resolve OpenCode's internal plugins with their privileged Location services. + const internal = yield* PluginInternal.list() + // Combine internal plugins with host-contributed SDK plugins in boot order. + const pre = [...internal.pre.map((plugin) => ({ ...plugin, version: "internal" })), ...sdk.all()] + const post = internal.post.map((plugin) => ({ ...plugin, version: "internal" })) + const operations = yield* sources.operations() + // Apply config operations and load enabled package plugins into one ordered generation. + const plugins = yield* resolve(pre, post, operations) + // Replace the active generation in one scoped, batched activation. + yield* registry.activate(plugins) + }) + const updates = Stream.merge(sources.changes(), bus.subscribe([Event.Updated, SdkPlugins.Updated])).pipe( + // Make accepted work visible to flush before coalescing the burst. + Stream.mapEffect(() => + Effect.gen(function* () { + observed++ + if (yield* Deferred.isDone(ready.current)) ready.current = yield* Deferred.make() + return observed + }), + ), + ) + yield* Stream.concat(Stream.succeed(0), updates).pipe( + // Keep observing updates while activation runs, retaining only the latest generation request. + Stream.buffer({ capacity: 1, strategy: "sliding" }), + Stream.debounce("100 millis"), + Stream.runForEach((target) => + Effect.gen(function* () { + yield* activate() + if (observed === target) yield* Deferred.succeed(ready.current, undefined) + }).pipe(Effect.catchCause((cause) => Effect.logError("failed to reload plugins", { cause }))), + ), + Effect.forkScoped({ startImmediately: true }), + ) + return Service.of({ flush: Effect.suspend(() => Deferred.await(ready.current)) }) + }), +) const nodeDeps = [ Plugin.node, SdkPlugins.node, - Agent.node, - Catalog.node, - Command.node, - ConfigPluginSource.requirements, - Credential.node, + ConfigPluginSource.node, Bus.node, - Environment.node, - FileMutation.node, - Formatter.node, - FileSystem.node, - Global.node, - httpClient, - Image.node, - Integration.node, - KV.node, - LocationMutation.node, - ModelsDev.node, Npm.node, - Permission.node, - PluginRuntime.node, - Form.node, - ReadToolFileSystem.node, - Reference.node, - Ripgrep.node, - SessionInstructions.node, - Shell.node, - Skill.node, - Tool.node, - WebSearch.node, - WellKnown.node, + PluginInternal.requirements, ] as const -export const node = configured() - -export const layer = makeLayer() +export const node = makeLocationNode({ service: Service, layer, deps: nodeDeps }) diff --git a/packages/core/test/config/plugin.test.ts b/packages/core/test/config/plugin.test.ts index a51ce830d5f3..bf6b0cf52f6d 100644 --- a/packages/core/test/config/plugin.test.ts +++ b/packages/core/test/config/plugin.test.ts @@ -5,6 +5,7 @@ import { describe, expect } from "bun:test" import { Plugin as EffectPlugin } from "@opencode-ai/plugin/effect" import { Agent } from "@opencode-ai/core/agent" import { Catalog } from "@opencode-ai/core/catalog" +import { ConfigPluginSource } from "@opencode-ai/core/config/plugin/source" 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" @@ -26,7 +27,7 @@ const it = testEffect( ) const staticIt = testEffect( AppNodeBuilder.build(LayerNode.group([Database.node, Bus.node, SdkPlugins.node, LocationServiceMap.node]), [ - [PluginSupervisor.node, PluginSupervisor.configured({ dynamic: false })], + [ConfigPluginSource.node, ConfigPluginSource.empty], ]), ) @@ -162,7 +163,7 @@ describe("PluginSupervisor config", () => { ), ) - staticIt.live("uses only internal and SDK plugins when dynamic imports are disabled", () => + staticIt.live("uses only internal and SDK plugins when the static source is wired", () => Effect.gen(function* () { const sdk = yield* SdkPlugins.Service yield* sdk.register(EffectPlugin.define({ id: "static-sdk", effect: () => Effect.void }))