From 856c5694586bc98dcf8fd6ad35e90ec3a4f102c9 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Mon, 27 Jul 2026 15:14:08 -0400 Subject: [PATCH] feat(core): reload agents from config change feed (#39167) --- packages/core/src/config/plugin/agent.ts | 43 +++-- packages/core/test/config/agent.test.ts | 215 ++++++++++++++++++++++- packages/core/test/lib/clock.ts | 38 ++++ 3 files changed, 282 insertions(+), 14 deletions(-) create mode 100644 packages/core/test/lib/clock.ts diff --git a/packages/core/src/config/plugin/agent.ts b/packages/core/src/config/plugin/agent.ts index 0cb4c43f1d1..c6d8a50d5ee 100644 --- a/packages/core/src/config/plugin/agent.ts +++ b/packages/core/src/config/plugin/agent.ts @@ -20,6 +20,8 @@ const legacySources = [ { pattern: "{agent,agents}/**/*.md", primary: false }, { pattern: "{mode,modes}/*.md", primary: true }, ] as const +// Keep in sync with the legacySources patterns and the name-strip regex in decode. +const sourceDirectories = ["agent", "agents", "mode", "modes"] as const const decodeAgent = Schema.decodeUnknownOption(ConfigAgent.Info) const decodeLegacyAgent = Schema.decodeUnknownOption(ConfigAgentV1.Info) const decodeConfig = Schema.decodeUnknownOption(Config.Info) @@ -67,7 +69,26 @@ export const Plugin = define({ }) }).pipe(Effect.map((documents) => documents.flat())) }) - const loaded = { documents: yield* load() } + const loaded = { documents: [] as Config.Document[] } + const reload = load().pipe( + Effect.tap((documents) => Effect.sync(() => (loaded.documents = documents))), + Effect.andThen(ctx.agent.reload()), + ) + // One merged trigger stream serializes reloads and shares one debounce + // window; subscribing before the initial scan means updates racing the + // scan still trigger a rebuild. + const sourceChanges = config + .changes() + .pipe( + Stream.filterEffect((update) => Effect.map(config.entries(), (entries) => isAgentSource(entries, update.path))), + ) + const configUpdates = ctx.event.subscribe().pipe(Stream.filter((event) => event.type === "config.updated")) + yield* Stream.merge(sourceChanges, configUpdates).pipe( + Stream.debounce("100 millis"), + Stream.runForEach(() => reload), + Effect.forkScoped({ startImmediately: true }), + ) + loaded.documents = yield* load() yield* ctx.agent.transform((draft) => { const permissions = expandPermissions( loaded.documents.flatMap((document) => document.info.permissions ?? []), @@ -113,19 +134,19 @@ export const Plugin = define({ } } }) - yield* ctx.event.subscribe().pipe( - Stream.filter((event) => event.type === "config.updated"), - Stream.runForEach(() => - load().pipe( - Effect.tap((documents) => Effect.sync(() => (loaded.documents = documents))), - Effect.andThen(ctx.agent.reload()), - ), - ), - Effect.forkScoped({ startImmediately: true }), - ) }), }) +// Matches anything at or under /{agent,agents,mode,modes}. No file-suffix +// check: directory-level events such as renames carry no per-file paths. +function isAgentSource(entries: Config.Entry[], file: string) { + return entries.some( + (entry) => + entry.type === "directory" && + sourceDirectories.some((name) => FSUtil.contains(path.join(entry.path, name), file)), + ) +} + function expandPermissions(rules: Permission.Ruleset, home: string): Permission.Ruleset { // Expand only resources tools resolve as filesystem paths. Bash resources are raw shell text: // rewriting `$HOME/private/**` would miss `$HOME/private/key`, and safe expansion needs shell-aware parsing. diff --git a/packages/core/test/config/agent.test.ts b/packages/core/test/config/agent.test.ts index 161748e2306..3a35ec03a97 100644 --- a/packages/core/test/config/agent.test.ts +++ b/packages/core/test/config/agent.test.ts @@ -1,8 +1,9 @@ import { describe, expect, test } from "bun:test" import fs from "fs/promises" import path from "path" -import { Effect, Schema, Stream } from "effect" +import { Effect, Fiber, Schema, Stream } from "effect" import { Agent } from "@opencode-ai/core/agent" +import { Bus } from "@opencode-ai/core/bus" import { Config } from "@opencode-ai/core/config" import { ConfigAgentPlugin } from "@opencode-ai/core/config/plugin/agent" import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" @@ -12,11 +13,12 @@ import { Global } from "@opencode-ai/util/global" import { Permission } from "@opencode-ai/core/permission" import { AbsolutePath } from "@opencode-ai/core/schema" import { ConfigMigrateV1 } from "@opencode-ai/core/v1/config/migrate" +import { advance, drain } from "../lib/clock" import { tmpdir } from "../fixture/tmpdir" import { testEffect } from "../lib/effect" import { agentHost, host } from "../plugin/host" -const it = testEffect(AppNodeBuilder.build(LayerNode.group([Agent.node, FSUtil.node, Global.node]))) +const it = testEffect(AppNodeBuilder.build(LayerNode.group([Agent.node, Bus.node, FSUtil.node, Global.node]))) const decode = Schema.decodeUnknownSync(Config.Info) const defaultPermissions = [ { action: "*", resource: "*", effect: "allow" }, @@ -272,7 +274,7 @@ Use native v2 fields.`, type: "document", info: decode({ agents: { reviewer: { description: "JSON description" } } }), }), - new Config.Directory({ type: "directory", path: AbsolutePath.make(tmp.path) }), + directoryEntry(tmp.path), ] yield* ConfigAgentPlugin.Plugin.effect(host({ agent: agentHost(agents) })).pipe( @@ -298,8 +300,215 @@ Use native v2 fields.`, ), ), ) + + for (const testCase of sourceCases()) { + it.effect(`rebuilds agents when a source file is ${testCase.name}`, () => + Effect.acquireRelease( + Effect.promise(() => tmpdir()), + (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()), + ).pipe( + Effect.flatMap((tmp) => + Effect.gen(function* () { + const directory = path.join(tmp.path, testCase.source) + yield* Effect.promise(() => fs.mkdir(directory, { recursive: true })) + yield* testCase.prepare(directory) + + const agents = yield* Agent.Service + const bus = yield* Bus.Service + const configTest = yield* Config.Test + yield* ConfigAgentPlugin.Plugin.effect(host({ agent: agentHost(agents) })) + + // Verify inside the subscription so the update event is a read barrier: + // committed state must be visible at event delivery time. + let received = 0 + const changed = yield* bus.subscribe(Agent.Event.Updated).pipe( + Stream.take(1), + Stream.tap(() => Effect.sync(() => received++)), + Stream.mapEffect(() => testCase.verify(agents)), + Stream.runDrain, + Effect.forkScoped({ startImmediately: true }), + ) + yield* Effect.yieldNow + + const updates = yield* testCase.mutate(directory) + yield* Effect.forEach(updates, (update) => configTest.emitChange(update), { discard: true }) + yield* advance(() => received === 1) + yield* Fiber.join(changed) + }).pipe(Effect.provide(Config.testLayer([directoryEntry(tmp.path)]))), + ), + ), + ) + } + + it.effect("coalesces updates inside the debounce window into one rebuild", () => + Effect.acquireRelease( + Effect.promise(() => tmpdir()), + (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()), + ).pipe( + Effect.flatMap((tmp) => + Effect.gen(function* () { + const directory = path.join(tmp.path, "agents") + yield* Effect.promise(() => fs.mkdir(directory, { recursive: true })) + + const agents = yield* Agent.Service + const configTest = yield* Config.Test + let reloads = 0 + yield* ConfigAgentPlugin.Plugin.effect( + host({ + agent: { + ...agentHost(agents), + reload: () => agents.reload().pipe(Effect.tap(() => Effect.sync(() => reloads++))), + }, + }), + ) + + yield* Effect.promise(() => fs.writeFile(path.join(directory, "reviewer.md"), "Review once")) + yield* configTest.emitChange({ type: "create", path: path.join(directory, "reviewer.md") }) + yield* configTest.emitChange({ type: "update", path: path.join(directory, "reviewer.md") }) + yield* configTest.emitChange({ type: "update", path: path.join(directory, "reviewer.md") }) + yield* advance(() => reloads >= 1) + expect(reloads).toBe(1) + + yield* Effect.promise(() => fs.writeFile(path.join(directory, "reviewer.md"), "Review twice")) + yield* configTest.emitChange({ type: "update", path: path.join(directory, "reviewer.md") }) + yield* advance(() => reloads >= 2) + expect(reloads).toBe(2) + expect(yield* agents.get(Agent.ID.make("reviewer"))).toMatchObject({ system: "Review twice" }) + }).pipe(Effect.provide(Config.testLayer([directoryEntry(tmp.path)]))), + ), + ), + ) + + it.effect("ignores updates outside agent source directories", () => + Effect.acquireRelease( + Effect.promise(() => tmpdir()), + (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()), + ).pipe( + Effect.flatMap((tmp) => + Effect.gen(function* () { + const directory = path.join(tmp.path, "agents") + yield* Effect.promise(() => fs.mkdir(directory, { recursive: true })) + + const agents = yield* Agent.Service + const configTest = yield* Config.Test + let reloads = 0 + yield* ConfigAgentPlugin.Plugin.effect( + host({ + agent: { + ...agentHost(agents), + reload: () => agents.reload().pipe(Effect.tap(() => Effect.sync(() => reloads++))), + }, + }), + ) + + yield* configTest.emitChange({ type: "create", path: path.join(tmp.path, "commands", "review.md") }) + yield* configTest.emitChange({ type: "update", path: path.join(tmp.path, "opencode.json") }) + yield* drain + expect(reloads).toBe(0) + + // The feed stays live after unrelated updates. + yield* Effect.promise(() => fs.writeFile(path.join(directory, "reviewer.md"), "Review related")) + yield* configTest.emitChange({ type: "create", path: path.join(directory, "reviewer.md") }) + yield* advance(() => reloads >= 1) + expect(yield* agents.get(Agent.ID.make("reviewer"))).toMatchObject({ system: "Review related" }) + }).pipe(Effect.provide(Config.testLayer([directoryEntry(tmp.path)]))), + ), + ), + ) }) +function directoryEntry(directory: string) { + return new Config.Directory({ type: "directory", path: AbsolutePath.make(directory) }) +} + +function sourceCases() { + return [ + { + name: "created", + source: "agents", + prepare: () => Effect.void, + mutate: (directory: string) => + Effect.promise(async () => { + const file = path.join(directory, "reviewer.md") + await fs.writeFile(file, "Review changes") + return [{ type: "create" as const, path: file }] + }), + verify: (agents: Agent.Interface) => + Effect.gen(function* () { + expect(yield* agents.get(Agent.ID.make("reviewer"))).toMatchObject({ system: "Review changes" }) + }), + }, + { + name: "created in a legacy modes directory", + source: "modes", + prepare: () => Effect.void, + mutate: (directory: string) => + Effect.promise(async () => { + const file = path.join(directory, "plan.md") + await fs.writeFile(file, "Make a plan") + return [{ type: "create" as const, path: file }] + }), + verify: (agents: Agent.Interface) => + Effect.gen(function* () { + expect(yield* agents.get(Agent.ID.make("plan"))).toMatchObject({ system: "Make a plan", mode: "primary" }) + }), + }, + { + name: "updated", + source: "agents", + prepare: (directory: string) => + Effect.promise(() => fs.writeFile(path.join(directory, "reviewer.md"), "Review first")), + mutate: (directory: string) => + Effect.promise(async () => { + const file = path.join(directory, "reviewer.md") + await fs.writeFile(file, "Review updated") + return [{ type: "update" as const, path: file }] + }), + verify: (agents: Agent.Interface) => + Effect.gen(function* () { + expect(yield* agents.get(Agent.ID.make("reviewer"))).toMatchObject({ system: "Review updated" }) + }), + }, + { + name: "renamed", + source: "agents", + prepare: (directory: string) => + Effect.promise(() => fs.writeFile(path.join(directory, "reviewer.md"), "Review renamed")), + mutate: (directory: string) => + Effect.promise(async () => { + const previous = path.join(directory, "reviewer.md") + const next = path.join(directory, "release.md") + await fs.rename(previous, next) + return [ + { type: "delete" as const, path: previous }, + { type: "create" as const, path: next }, + ] + }), + verify: (agents: Agent.Interface) => + Effect.gen(function* () { + expect(yield* agents.get(Agent.ID.make("reviewer"))).toBeUndefined() + expect(yield* agents.get(Agent.ID.make("release"))).toMatchObject({ system: "Review renamed" }) + }), + }, + { + name: "deleted", + source: "agents", + prepare: (directory: string) => + Effect.promise(() => fs.writeFile(path.join(directory, "reviewer.md"), "Review deleted")), + mutate: (directory: string) => + Effect.promise(async () => { + const file = path.join(directory, "reviewer.md") + await fs.unlink(file) + return [{ type: "delete" as const, path: file }] + }), + verify: (agents: Agent.Interface) => + Effect.gen(function* () { + expect(yield* agents.get(Agent.ID.make("reviewer"))).toBeUndefined() + }), + }, + ] as const +} + function loadHomePermissions(home: string) { return Effect.gen(function* () { const agents = yield* Agent.Service diff --git a/packages/core/test/lib/clock.ts b/packages/core/test/lib/clock.ts new file mode 100644 index 00000000000..8a46eee2057 --- /dev/null +++ b/packages/core/test/lib/clock.ts @@ -0,0 +1,38 @@ +import { Effect } from "effect" +import { TestClock } from "effect/testing" + +// Defers on a real macrotask so pubsub delivery, fiber hops, and filesystem +// work can complete between TestClock advances. +const settle = Effect.promise(() => new Promise((resolve) => setTimeout(resolve, 1))) + +// One advance step: let pending real work finish, then fire due TestClock timers. +const tick = Effect.gen(function* () { + yield* settle + yield* TestClock.adjust("500 millis") +}) + +/** + * Drives every pending TestClock timer to completion: stream debounces and + * State's 500ms reload debounce register their sleeps from separate fiber hops + * that a single adjust can miss, so the loop alternates real-macrotask settles + * with adjusts until the condition holds. Extra adjusts are harmless when + * nothing is pending. + */ +export const advance = Effect.fnUntraced(function* (condition: () => boolean) { + for (let attempt = 0; attempt < 100; attempt++) { + if (condition()) return + yield* tick + } + return yield* Effect.die(new Error("condition never became true after 100 advances")) +}) + +/** + * Advances far enough that any pending debounced work would have completed, + * so a no-op assertion afterwards is meaningful. + */ +export const drain = Effect.gen(function* () { + for (let round = 0; round < 4; round++) { + yield* tick + } + yield* settle +})