fix(core): subscribe before debouncing config plugin updates (#46874)

packages/core/src/config/plugin/agent.ts

@@ -4,7 +4,7 @@ import { define } from "@opencode-ai/plugin/effect/plugin"
 import { Document, Info, type Entry } from "@opencode-ai/schema/config"
 import { ConfigAgent } from "@opencode-ai/schema/config/agent"
 import path from "path"
-import { Effect, Option, Schema, Stream } from "effect"
+import { Effect, Option, PubSub, Schema, Stream } from "effect"
 import { Agent } from "../../agent.js"
 import { Config } from "../../config.js"
 import { ConfigMarkdown } from "../markdown.js"
@@ -59,16 +59,25 @@ export const Plugin = define({
       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(
+    // One trigger feed serializes reloads and shares one debounce window;
+    // subscribing before the initial scan means updates racing the scan still
+    // trigger a rebuild. Each source is subscribed eagerly on its own fiber
+    // (Stream.merge and Stream.debounce both open upstream a fiber hop later)
+    // so no update slips through while the debounce starts its pull.
+    const changes = yield* PubSub.sliding<void>(1)
+    const notify = () => PubSub.publish(changes, undefined)
+    yield* config.changes().pipe(
+      Stream.filterEffect((update) => Effect.map(config.entries(), (entries) => isAgentSource(entries, update.path))),
+      Stream.runForEach(notify),
+      Effect.forkScoped({ startImmediately: true }),
+    )
+    yield* ctx.event.subscribe().pipe(
+      Stream.filter((event) => event.type === "config.updated"),
+      Stream.runForEach(notify),
+      Effect.forkScoped({ startImmediately: true }),
+    )
+    const updates = yield* PubSub.subscribe(changes)
+    yield* Stream.fromSubscription(updates).pipe(
       Stream.debounce("100 millis"),
       Stream.runForEach(() => reload),
       Effect.forkScoped({ startImmediately: true }),

packages/core/src/config/plugin/command.ts

@@ -8,7 +8,7 @@ import { Model } from "@opencode-ai/schema/model"
 import { Provider } from "@opencode-ai/schema/provider"
 import { AppProcess } from "@opencode-ai/util/process"
 import path from "path"
-import { Effect, Option, Schema, Stream } from "effect"
+import { Effect, Option, PubSub, Schema, Stream } from "effect"
 import { ChildProcess } from "effect/unstable/process"
 import { Config } from "../../config.js"
 import { Location } from "../../location.js"
@@ -40,18 +40,25 @@ export const Plugin = define({
       Effect.tap((documents) => Effect.sync(() => (loaded.documents = documents))),
       Effect.andThen(ctx.command.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) => isCommandSource(entries, update.path)),
-        ),
-      )
-    const configUpdates = ctx.event.subscribe().pipe(Stream.filter((event) => event.type === "config.updated"))
-    yield* Stream.merge(sourceChanges, configUpdates).pipe(
+    // One trigger feed serializes reloads and shares one debounce window;
+    // subscribing before the initial scan means updates racing the scan still
+    // trigger a rebuild. Each source is subscribed eagerly on its own fiber
+    // (Stream.merge and Stream.debounce both open upstream a fiber hop later)
+    // so no update slips through while the debounce starts its pull.
+    const changes = yield* PubSub.sliding<void>(1)
+    const notify = () => PubSub.publish(changes, undefined)
+    yield* config.changes().pipe(
+      Stream.filterEffect((update) => Effect.map(config.entries(), (entries) => isCommandSource(entries, update.path))),
+      Stream.runForEach(notify),
+      Effect.forkScoped({ startImmediately: true }),
+    )
+    yield* ctx.event.subscribe().pipe(
+      Stream.filter((event) => event.type === "config.updated"),
+      Stream.runForEach(notify),
+      Effect.forkScoped({ startImmediately: true }),
+    )
+    const updates = yield* PubSub.subscribe(changes)
+    yield* Stream.fromSubscription(updates).pipe(
       Stream.debounce("100 millis"),
       Stream.runForEach(() => reload),
       Effect.forkScoped({ startImmediately: true }),

packages/core/test/config/agent.test.ts

@@ -5,7 +5,7 @@ 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 { Directory, Document, Info } from "@opencode-ai/schema/config"
+import { Directory, Document, Event, Info } from "@opencode-ai/schema/config"
 import { ConfigAgentPlugin } from "@opencode-ai/core/config/plugin/agent"
 import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
 import { LayerNode } from "@opencode-ai/util/effect/layer-node"
@@ -501,6 +501,30 @@ Use native v2 fields.`,
     ),
   )
 
+  it.effect("rebuilds on a config update published immediately after startup", () =>
+    Effect.gen(function* () {
+      const agents = yield* Agent.Service
+      const bus = yield* Bus.Service
+      let reloads = 0
+      // No directory entries, so startup has no filesystem hop that could hide a late subscription.
+      yield* ConfigAgentPlugin.Plugin.effect(
+        host({
+          agent: {
+            ...agentHost(agents),
+            reload: () => agents.reload().pipe(Effect.tap(() => Effect.sync(() => reloads++))),
+          },
+          event: { subscribe: () => bus.subscribe(Event.Updated) },
+        }),
+      )
+
+      // Published in the same fiber step as startup: the subscription must
+      // already be open when the plugin effect returns.
+      yield* bus.publish(Event.Updated, {})
+      yield* advance(() => reloads >= 1)
+      expect(reloads).toBe(1)
+    }).pipe(Effect.provide(Config.testLayer([]))),
+  )
+
   it.effect("ignores updates outside agent source directories", () =>
     Effect.acquireDisposable(Effect.promise(() => tmpdir())).pipe(
       Effect.flatMap((tmp) =>

packages/core/test/config/command.test.ts

@@ -244,6 +244,31 @@ Review files`,
     ),
   )
 
+  it.effect("rebuilds on a config update published immediately after startup", () =>
+    Effect.gen(function* () {
+      const command = yield* Command.Service
+      const bus = yield* Bus.Service
+      let reloads = 0
+      // No directory entries, so startup has no filesystem hop that could hide a late subscription.
+      yield* ConfigCommandPlugin.Plugin.effect(
+        host({
+          command: {
+            list: () => Effect.die("unused command.list"),
+            transform: command.transform,
+            reload: () => command.reload().pipe(Effect.tap(() => Effect.sync(() => reloads++))),
+          },
+          event: { subscribe: () => bus.subscribe(Event.Updated) },
+        }),
+      )
+
+      // Published in the same fiber step as startup: the subscription must
+      // already be open when the plugin effect returns.
+      yield* bus.publish(Event.Updated, {})
+      yield* advance(() => reloads >= 1)
+      expect(reloads).toBe(1)
+    }).pipe(Effect.provide(Config.testLayer([]))),
+  )
+
   it.effect("ignores updates outside command source directories", () =>
     Effect.acquireDisposable(Effect.promise(() => tmpdir())).pipe(
       Effect.flatMap((tmp) =>