diff --git a/packages/core/src/plugin/supervisor.ts b/packages/core/src/plugin/supervisor.ts index 2b14c42db1..ff8253a457 100644 --- a/packages/core/src/plugin/supervisor.ts +++ b/packages/core/src/plugin/supervisor.ts @@ -202,6 +202,16 @@ function discoverDirectory(fs: FSUtil.Interface, directory: string) { }) } +const sourceDirectories = ["plugin", "plugins"] as const + +function isPluginSource(entries: readonly Config.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 @@ -239,12 +249,25 @@ const layer = Layer.effect( }), ) }) - const updates = yield* bus + const sourceChanges = config + .changes() + .pipe( + Stream.filterEffect((update) => + Effect.map(config.entries(), (entries) => isPluginSource(entries, update.path)), + ), + // Make accepted filesystem work visible to flush before coalescing the burst. + Stream.mapEffect(() => Effect.sync(() => ++observed)), + Stream.debounce("100 millis"), + ) + const busUpdates = bus .subscribe([Event.Updated, SdkPlugins.Updated]) - .pipe(Stream.toQueue({ capacity: 1, strategy: "sliding" })) + .pipe(Stream.mapEffect(() => Effect.sync(() => ++observed))) + const updates = yield* Stream.merge(busUpdates, sourceChanges).pipe( + Stream.toQueue({ capacity: 1, strategy: "sliding" }), + ) const signals = yield* Stream.concat( Stream.succeed(0), - Stream.fromQueue(updates).pipe(Stream.mapEffect(() => Effect.sync(() => ++observed))), + Stream.fromQueue(updates), ).pipe(Stream.broadcast({ capacity: 1, strategy: "sliding", replay: 1 })) const attempt = (target: number) => activate(target).pipe( diff --git a/packages/core/test/config/plugin.test.ts b/packages/core/test/config/plugin.test.ts index 64fb43336c..151a715972 100644 --- a/packages/core/test/config/plugin.test.ts +++ b/packages/core/test/config/plugin.test.ts @@ -3,7 +3,6 @@ import path from "path" import { pathToFileURL } from "url" import { describe, expect } from "bun:test" import { Plugin as EffectPlugin } from "@opencode-ai/plugin/effect" -import { Config as ConfigSchema } from "@opencode-ai/schema/config" import { Agent } from "@opencode-ai/core/agent" import { Catalog } from "@opencode-ai/core/catalog" import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" @@ -17,7 +16,7 @@ import { PluginSupervisor } from "@opencode-ai/core/plugin/supervisor" import { Model } from "@opencode-ai/core/model" import { Provider } from "@opencode-ai/core/provider" import { AbsolutePath } from "@opencode-ai/core/schema" -import { Effect, Logger } from "effect" +import { Effect, Fiber, Logger, Stream } from "effect" import { Database } from "../../src/database/database" import { tmpdir } from "../fixture/tmpdir" import { testEffect } from "../lib/effect" @@ -173,18 +172,19 @@ describe("PluginSupervisor config", () => { expect(first).toBeDefined() 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* bus.publish(ConfigSchema.Event.Updated, {}) - yield* waitUntil( - Effect.gen(function* () { - const current = (yield* plugins.list()).find((plugin) => plugin.id === "mutable-plugin")?.id - return current === first && (yield* agents.get(Agent.ID.make("mutable")))?.description === "second" - }), - ) + yield* Fiber.join(changed).pipe(Effect.timeout("5 seconds")) + + const current = (yield* plugins.list()).find((plugin) => plugin.id === "mutable-plugin")?.id + expect(current).toBe(first) + expect((yield* agents.get(Agent.ID.make("mutable")))?.description).toBe("second") }), false, async (directory) => { @@ -320,11 +320,3 @@ export default Plugin.define({ }) ` } - -const waitUntil = Effect.fnUntraced(function* (condition: Effect.Effect) { - for (let attempt = 0; attempt < 200; attempt++) { - if (yield* condition) return - yield* Effect.sleep("10 millis") - } - return yield* Effect.die("Timed out waiting for plugin reload") -})