diff --git a/packages/core/src/config/plugin/agent.ts b/packages/core/src/config/plugin/agent.ts index facfff811b9..263777876ca 100644 --- a/packages/core/src/config/plugin/agent.ts +++ b/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(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 }), diff --git a/packages/core/src/config/plugin/command.ts b/packages/core/src/config/plugin/command.ts index 1aea02cca2e..fb82ab55703 100644 --- a/packages/core/src/config/plugin/command.ts +++ b/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(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 }), diff --git a/packages/core/test/config/agent.test.ts b/packages/core/test/config/agent.test.ts index 2df2cd93d35..7dcd90773e6 100644 --- a/packages/core/test/config/agent.test.ts +++ b/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) => diff --git a/packages/core/test/config/command.test.ts b/packages/core/test/config/command.test.ts index 8bc526b45b8..c8764ead26a 100644 --- a/packages/core/test/config/command.test.ts +++ b/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) =>