feat(core): reload agents from config change feed (#39167)

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 <root>/{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.

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

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
+})