fix(core): preserve plugin update order (#39372)

packages/ai/package.json

@@ -15,6 +15,7 @@
   ],
   "exports": {
     ".": "./src/index.ts",
+    "./testing": "./src/testing.ts",
     "./*": "./src/*.ts"
   },
   "devDependencies": {

packages/ai/src/testing.ts

@@ -0,0 +1,157 @@
+export * as TestLLM from "./testing"
+
+import { LLMClient, type Interface as LLMClientShape } from "./route/client"
+import {
+  LLMEvent,
+  LLMResponse,
+  type FinishReasonDetails,
+  type LLMError,
+  type LLMRequest,
+  type UsageInput,
+} from "./schema"
+import { Context, Deferred, Effect, Latch, Layer, Queue, Scope, Stream } from "effect"
+
+export type Response = readonly LLMEvent[] | Stream.Stream<LLMEvent, LLMError>
+
+export type Gate = Readonly<{ started: Effect.Effect<void>; release: Effect.Effect<void> }>
+
+export interface Interface {
+  readonly requests: LLMRequest[]
+  readonly push: (...responses: readonly Response[]) => Effect.Effect<void>
+  readonly always: (response: Response) => Effect.Effect<void>
+  readonly wait: (count: number) => Effect.Effect<void>
+  readonly gate: Effect.Effect<Gate, never, Scope.Scope>
+  readonly client: LLMClientShape
+}
+
+export interface LayerOptions {
+  readonly transformRequest?: (request: LLMRequest) => LLMRequest
+  /** Used after the one-shot response queue is exhausted. Omit to defect on unexpected requests. */
+  readonly fallback?: Response
+}
+
+export class Service extends Context.Service<Service, Interface>()("@opencode/ai/TestLLM") {}
+
+export const complete = (
+  options: { readonly reason: FinishReasonDetails; readonly usage?: UsageInput },
+  ...events: readonly LLMEvent[]
+) => [
+  LLMEvent.stepStart({ index: 0 }),
+  ...events,
+  LLMEvent.stepFinish({ index: 0, reason: options.reason, usage: options.usage }),
+  LLMEvent.finish({ reason: options.reason }),
+]
+
+export const stop = (...events: readonly LLMEvent[]) => complete({ reason: { normalized: "stop" } }, ...events)
+
+export const toolCalls = (...events: readonly LLMEvent[]) =>
+  complete({ reason: { normalized: "tool-calls" } }, ...events)
+
+const textEvents = (value: string, id: string) => [
+  LLMEvent.textStart({ id }),
+  LLMEvent.textDelta({ id, text: value }),
+  LLMEvent.textEnd({ id }),
+]
+
+export const text = (value: string, id: string) => stop(...textEvents(value, id))
+
+export const textWithUsage = (value: string, id: string, inputTokens: number) =>
+  complete(
+    { reason: { normalized: "stop" }, usage: { inputTokens, nonCachedInputTokens: inputTokens } },
+    ...textEvents(value, id),
+  )
+
+export const tool = (id: string, name: string, input: unknown) => toolCalls(LLMEvent.toolCall({ id, name, input }))
+
+export const failAfter = (error: LLMError, ...events: readonly LLMEvent[]) =>
+  Stream.fromIterable(events).pipe(Stream.concat(Stream.fail(error)))
+
+export const hangAfter = (...events: readonly LLMEvent[]) => Stream.concat(Stream.fromIterable(events), Stream.never)
+
+const toStream = (response: Response) => (Stream.isStream(response) ? response : Stream.fromIterable(response))
+
+export const layer = (options: LayerOptions = {}) =>
+  Layer.effect(
+    Service,
+    Effect.gen(function* () {
+      const requests: LLMRequest[] = []
+      const responses: Response[] = []
+      let started = Deferred.makeUnsafe<void>()
+      let fallback = options.fallback
+      let activeGate: { readonly started: Queue.Queue<void>; readonly release: Latch.Latch } | undefined
+      const wait = (count: number): Effect.Effect<void> =>
+        Effect.suspend(() =>
+          requests.length >= count ? Effect.void : Deferred.await(started).pipe(Effect.andThen(wait(count))),
+        )
+
+      const stream = ((request: LLMRequest) => {
+        requests.push(options.transformRequest?.(request) ?? request)
+        const waiting = started
+        started = Deferred.makeUnsafe()
+        Deferred.doneUnsafe(waiting, Effect.void)
+        const response = responses.shift() ?? fallback
+        if (!response) return Stream.die(new Error(`TestLLM has no response for request ${requests.length}`))
+        const streamed = toStream(response)
+        const gate = activeGate
+        if (!gate) return streamed
+        return Stream.unwrap(
+          Queue.offer(gate.started, undefined).pipe(Effect.andThen(gate.release.await), Effect.as(streamed)),
+        )
+      }) as LLMClientShape["stream"]
+      const client = LLMClient.Service.of({
+        prepare: () => Effect.die("TestLLM does not prepare provider-native requests"),
+        stream,
+        generate: (request) =>
+          stream(request).pipe(
+            Stream.runFold(LLMResponse.empty, LLMResponse.reduce),
+            Effect.flatMap((state) => {
+              const response = LLMResponse.complete(state)
+              if (response) return Effect.succeed(response)
+              return Effect.die("TestLLM response ended without a terminal finish event")
+            }),
+          ),
+      })
+
+      return Service.of({
+        requests,
+        push: (...input) =>
+          Effect.sync(() => {
+            responses.push(...input)
+          }),
+        always: (response) =>
+          Effect.sync(() => {
+            fallback = response
+          }),
+        wait,
+        gate: Effect.gen(function* () {
+          const gate = {
+            started: yield* Effect.acquireRelease(Queue.unbounded<void>(), Queue.shutdown),
+            release: yield* Latch.make(),
+          }
+          activeGate = gate
+          const release = Effect.sync(() => {
+            if (activeGate === gate) activeGate = undefined
+          }).pipe(Effect.andThen(gate.release.open), Effect.asVoid)
+          yield* Effect.addFinalizer(() => release)
+          return {
+            started: Queue.take(gate.started),
+            release,
+          }
+        }),
+        client,
+      })
+    }),
+  )
+
+export const clientLayer = Layer.effect(
+  LLMClient.Service,
+  Effect.map(Service, (service) => service.client),
+)
+
+export const push = (...responses: readonly Response[]) => Service.use((service) => service.push(...responses))
+
+export const always = (response: Response) => Service.use((service) => service.always(response))
+
+export const wait = (count: number) => Service.use((service) => service.wait(count))
+
+export const gate = Service.use((service) => service.gate)

packages/ai/test/exports.test.ts

@@ -19,6 +19,7 @@ import {
   OpenResponses,
 } from "@opencode-ai/ai/protocols"
 import * as AnthropicMessages from "@opencode-ai/ai/protocols/anthropic-messages"
+import { TestLLM } from "@opencode-ai/ai/testing"
 
 describe("public exports", () => {
   test("root exposes app-facing runtime APIs", () => {
@@ -28,6 +29,7 @@ describe("public exports", () => {
     expect(ImageInput.bytes).toBeFunction()
     expect(Provider.make).toBeFunction()
     expect(ProviderSubpath.make).toBe(Provider.make)
+    expect(TestLLM.layer).toBeFunction()
   })
 
   test("route barrel exposes route-authoring APIs", () => {

packages/core/src/plugin/supervisor.ts

@@ -277,17 +277,17 @@ const layer = Layer.effect(
       // Replace the active generation in one scoped, batched activation.
       yield* registry.activate(plugins)
     })
-    const sourceChanges = config.changes().pipe(
-      Stream.filterEffect((update) => Effect.map(config.entries(), (entries) => isPluginSource(entries, update.path))),
-      Stream.merge(Stream.fromPubSub(configuredChanges)),
-      // Make accepted filesystem work visible to flush before coalescing the burst.
+    const updates = Stream.merge(
+      config.changes().pipe(
+        Stream.filterEffect((update) => Effect.map(config.entries(), (entries) => isPluginSource(entries, update.path))),
+        Stream.merge(Stream.fromPubSub(configuredChanges)),
+      ),
+      bus.subscribe([Event.Updated, SdkPlugins.Updated]),
+    ).pipe(
+      // Make accepted work visible to flush before coalescing the burst.
       Stream.mapEffect(() => Effect.sync(() => ++observed)),
-      Stream.debounce("100 millis"),
     )
-    const busUpdates = bus
-      .subscribe([Event.Updated, SdkPlugins.Updated])
-      .pipe(Stream.mapEffect(() => Effect.sync(() => ++observed)))
-    yield* Stream.concat(Stream.succeed(0), Stream.merge(busUpdates, sourceChanges)).pipe(
+    yield* Stream.concat(Stream.succeed(0), updates).pipe(
       // Keep observing updates while activation runs, retaining only the latest generation request.
       Stream.buffer({ capacity: 1, strategy: "sliding" }),
       Stream.debounce("100 millis"),

packages/core/test/generate.test.ts

@@ -1,6 +1,7 @@
 import { expect } from "bun:test"
-import { LLMClient, LLMEvent, LLMResponse, Model } from "@opencode-ai/ai"
+import { Model } from "@opencode-ai/ai"
 import { OpenAIChat } from "@opencode-ai/ai/protocols"
+import { TestLLM } from "@opencode-ai/ai/testing"
 import { AISDK } from "@opencode-ai/core/aisdk"
 import { Catalog } from "@opencode-ai/core/catalog"
 import { Generate } from "@opencode-ai/core/generate"
@@ -9,7 +10,7 @@ import { ModelResolver } from "@opencode-ai/core/model-resolver"
 import { ID, Info, Ref } from "@opencode-ai/core/model"
 import { Provider } from "@opencode-ai/core/provider"
 import { Npm } from "@opencode-ai/util/npm"
-import { Effect, Layer, Stream } from "effect"
+import { Effect, Layer } from "effect"
 import { testEffect } from "./lib/effect"
 
 const selected = Info.make({
@@ -64,21 +65,7 @@ const aisdk = Layer.mock(AISDK.Service, {
   },
   model: () => Effect.succeed(runtime),
 })
-const client = Layer.mock(LLMClient.Service)({
-  prepare: () => Effect.die("unused"),
-  stream: () => Stream.die("unused"),
-  generate: () =>
-    Effect.sync(() => {
-      const response = LLMResponse.fromEvents([
-        LLMEvent.textStart({ id: "generate" }),
-        LLMEvent.textDelta({ id: "generate", text: "OK" }),
-        LLMEvent.textEnd({ id: "generate" }),
-        LLMEvent.finish({ reason: { normalized: "stop" } }),
-      ])
-      if (!response) throw new Error("Incomplete generate response")
-      return response
-    }),
-})
+const client = TestLLM.clientLayer.pipe(Layer.provide(TestLLM.layer({ fallback: TestLLM.text("OK", "generate") })))
 
 const resolver = ModelResolver.layer.pipe(Layer.provide(Layer.mergeAll(catalog, integrations, npm, aisdk)))
 const it = testEffect(Generate.layer.pipe(Layer.provide(Layer.merge(resolver, client))))