diff --git a/packages/core/src/plugin.ts b/packages/core/src/plugin.ts index e22453306f4..a6ead780fd6 100644 --- a/packages/core/src/plugin.ts +++ b/packages/core/src/plugin.ts @@ -161,7 +161,7 @@ const layer = Layer.effect( yield* Effect.addFinalizer((exit) => Effect.gen(function* () { active.clear() - yield* State.batch(Scope.close(scope, exit)) + yield* State.batch(Scope.close(scope, exit), { flush: false }) }), ) diff --git a/packages/core/src/state.ts b/packages/core/src/state.ts index daa9bec86e0..273e128eeb6 100644 --- a/packages/core/src/state.ts +++ b/packages/core/src/state.ts @@ -32,6 +32,7 @@ export interface Transformable { type Batch = { active: boolean + readonly flush: boolean readonly reloads: Set } @@ -40,14 +41,15 @@ const CurrentBatch = Context.Reference("@opencode/State/Curre }) const reloadDebounce = 500 -export function batch(effect: Effect.Effect) { +/** flush: false is terminal teardown: states whose transforms are removed stop rebuilding, including pending reloads. */ +export function batch(effect: Effect.Effect, options: { readonly flush?: boolean } = {}) { return Effect.gen(function* () { const current = yield* CurrentBatch - if (current?.active) return yield* effect - const batch: Batch = { active: true, reloads: new Set() } + if (current?.active && options.flush !== false) return yield* effect + const batch: Batch = { active: true, flush: options.flush !== false, reloads: new Set() } const exit = yield* effect.pipe(Effect.provideService(CurrentBatch, batch), Effect.exit) batch.active = false - yield* Effect.forEach(batch.reloads, (reload) => reload(), { discard: true }) + if (batch.flush) yield* Effect.forEach(batch.reloads, (reload) => reload(), { discard: true }) return yield* exit }) } @@ -81,6 +83,7 @@ export function create(options: Options): Inte let generation = 0 let requestedAt = 0 let running = false + let closed = false let waiters: { generation: number; done: Deferred.Deferred }[] = [] const semaphore = Semaphore.makeUnsafe(1) @@ -90,6 +93,7 @@ export function create(options: Options): Inte }) const materialize = Effect.fnUntraced(function* () { + if (closed) return const next = options.initial() const api = options.draft(next) for (const transform of transforms) { @@ -122,6 +126,7 @@ export function create(options: Options): Inte }) const reload = Effect.fnUntraced(function* () { + if (closed) return const done = Deferred.makeUnsafe() const clock = yield* Clock.Clock generation++ @@ -131,7 +136,7 @@ export function create(options: Options): Inte running = true yield* rebuild().pipe(Effect.forkDetach) } - return yield* Deferred.await(done) + yield* Deferred.await(done) }) return { @@ -152,6 +157,11 @@ export function create(options: Options): Inte return Effect.gen(function* () { const batch = yield* CurrentBatch if (batch?.active) { + // Detached debounced reloads must also stay quiet after teardown. + if (!batch.flush) { + closed = true + return + } batch.reloads.add(materializeReload) return } diff --git a/packages/core/test/plugin.test.ts b/packages/core/test/plugin.test.ts index aba6f008dd4..53c2323215b 100644 --- a/packages/core/test/plugin.test.ts +++ b/packages/core/test/plugin.test.ts @@ -184,6 +184,39 @@ describe("Plugin", () => { }), ) + it.effect("emits rebuilt state when disabling one plugin while another remains enabled", () => + Effect.gen(function* () { + const plugins = yield* Plugin.Service + const agents = yield* Agent.Service + const bus = yield* Bus.Service + const definitions = ["first", "second"].map((id) => + versioned( + EffectPlugin.define({ + id, + effect: (ctx) => ctx.agent.transform((draft) => draft.update(id, () => {})), + }), + ), + ) + yield* plugins.activate(definitions) + + const observed: string[][] = [] + const unsubscribe = yield* bus.listen((event) => + event.type === Agent.Event.Updated.type + ? agents.list().pipe( + Effect.flatMap((items) => Effect.sync(() => observed.push(items.map((item) => item.id)))), + Effect.asVoid, + ) + : Effect.void, + ) + yield* Effect.addFinalizer(() => unsubscribe) + + yield* plugins.activate(definitions.slice(1)) + expect(yield* agents.get(Agent.ID.make("first"))).toBeUndefined() + expect(yield* agents.get(Agent.ID.make("second"))).toBeDefined() + expect(observed).toEqual([["second"]]) + }), + ) + it.effect("rejects duplicate IDs before replacing active plugins", () => Effect.gen(function* () { const plugins = yield* Plugin.Service diff --git a/packages/core/test/state.test.ts b/packages/core/test/state.test.ts index 39163a18d65..b76a9b58a62 100644 --- a/packages/core/test/state.test.ts +++ b/packages/core/test/state.test.ts @@ -133,6 +133,64 @@ describe("State", () => { }), ) + it.effect("discards teardown rebuilds and pending reloads while still running cleanup", () => + Effect.gen(function* () { + let finalized = 0 + let disposed = 0 + const state = State.create({ + initial: () => ({ values: [] as string[] }), + draft: (draft) => ({ add: (item: string) => draft.values.push(item) }), + finalize: () => Effect.sync(() => finalized++), + }) + const scope = yield* Scope.make() + yield* Scope.addFinalizer( + scope, + Effect.sync(() => disposed++), + ) + const registration = yield* state.transform((draft) => draft.add("value")).pipe(Scope.provide(scope)) + expect(finalized).toBe(1) + + const pending = yield* state.reload().pipe(Effect.forkChild({ startImmediately: true })) + yield* TestClock.adjust("250 millis") + yield* State.batch(Scope.close(scope, Exit.void), { flush: false }) + expect(disposed).toBe(1) + expect(finalized).toBe(1) + + yield* TestClock.adjust("500 millis") + yield* Fiber.join(pending) + yield* registration.dispose + yield* state.reload() + expect(finalized).toBe(1) + }), + ) + + it.effect("keeps teardown suppression separate from an enclosing live batch", () => + Effect.gen(function* () { + const finalized: string[] = [] + const closing = State.create({ + initial: () => ({}), + draft: (draft) => draft, + finalize: () => Effect.sync(() => finalized.push("closing")), + }) + const live = State.create({ + initial: () => ({}), + draft: (draft) => draft, + finalize: () => Effect.sync(() => finalized.push("live")), + }) + const scope = yield* Scope.make() + yield* closing.transform(() => {}).pipe(Scope.provide(scope)) + finalized.length = 0 + + yield* State.batch( + Effect.gen(function* () { + yield* live.transform(() => {}) + yield* State.batch(Scope.close(scope, Exit.void), { flush: false }) + }), + ) + expect(finalized).toEqual(["live"]) + }), + ) + it.effect("debounces reload bursts", () => Effect.gen(function* () { let finalized = 0 diff --git a/packages/sdk/test/embedded.test.ts b/packages/sdk/test/embedded.test.ts index 49b70ef3eb8..81389661c0d 100644 --- a/packages/sdk/test/embedded.test.ts +++ b/packages/sdk/test/embedded.test.ts @@ -172,6 +172,87 @@ it.live( 15_000, ) +it.live( + "evicts a Location without triggering connected client refetches", + () => + withEmbedded("opencode-embedded-quiet-eviction-", (fixture) => + Effect.gen(function* () { + const opencode = yield* fixture.sdk.OpenCode.create({ + config: { directory: fixture.directory, project: false, content: "{}" }, + }) + const ref = location(fixture) + const connected = yield* Latch.make(false) + const booted = yield* Deferred.make() + const boots = yield* Ref.make(0) + const updates = yield* Ref.make([]) + + yield* opencode.plugin({ + id: `quiet-eviction-${crypto.randomUUID()}`, + effect: (ctx) => + Effect.gen(function* () { + yield* Ref.update(boots, (count) => count + 1) + yield* ctx.catalog.transform((catalog) => catalog.provider.update("eviction-test", () => {})) + yield* ctx.agent.transform((agents) => agents.update("eviction-test", () => {})) + yield* ctx.command.transform((commands) => + commands.add({ name: "eviction-test", execute: () => Effect.void }), + ) + }), + }) + const subscriber = yield* opencode.events.subscribe().pipe( + Stream.runForEach((event) => + Effect.gen(function* () { + if (event.type === "server.connected") { + yield* connected.open + return + } + if (event.location?.directory !== fixture.directory) return + if (event.type === "plugin.updated") { + yield* Deferred.succeed(booted, undefined) + return + } + if ( + event.type !== "catalog.updated" && + event.type !== "agent.updated" && + event.type !== "command.updated" + ) + return + yield* Ref.update(updates, (types) => [...types, event.type]) + // A connected consumer re-reads invalidated resources through the real router. + if (event.type === "catalog.updated") { + yield* opencode.model.list({ location: ref }) + yield* opencode.provider.list({ location: ref }) + return + } + if (event.type === "agent.updated") { + yield* opencode.agent.list({ location: ref }) + return + } + yield* opencode.command.list({ location: ref }) + }), + ), + Effect.forkScoped, + ) + yield* connected.await + yield* opencode.plugin.list({ location: ref }) + yield* Deferred.await(booted).pipe(Effect.timeout("5 seconds")) + expect(yield* Ref.get(updates)).toEqual( + expect.arrayContaining(["catalog.updated", "agent.updated", "command.updated"]), + ) + yield* Ref.set(updates, []) + + yield* opencode.debug.location.evict({ location: ref }) + // Allow the live event stream to deliver teardown notifications and any refetches. + yield* Effect.sleep("200 millis") + + expect(yield* Ref.get(updates)).toEqual([]) + expect(yield* Ref.get(boots)).toBe(1) + expect(yield* opencode.debug.location.list()).toEqual([]) + expect(subscriber.pollUnsafe()).toBeUndefined() + }), + ), + 15_000, +) + it.live( "keeps SDK plugin registration isolated between embedded hosts", () =>