feat(ai): add first-class TestLLM controls (#45828)

packages/ai/README.md

@@ -214,22 +214,40 @@ the requests sent by code under test:
 import { Effect } from "effect"
 import { TestLLM } from "@opencode-ai/ai/testing"
 
-const testLLM = TestLLM.layer({
-  fallback: TestLLM.text("Hello from the test model", "text-1"),
-})
-
-// TestLLM.clientLayer provides LLMClient.Service and consumes TestLLM.Service.
 const programWithTestClient = Effect.gen(function* () {
+  const test = yield* TestLLM.Test
+  yield* test.push(TestLLM.text("Hello from the test model", "text-1"))
   const result = yield* program
-  const test = yield* TestLLM.Service
-  console.log(test.requests)
+  console.log(yield* test.requests())
   return result
-}).pipe(Effect.provide(TestLLM.clientLayer), Effect.provide(testLLM))
+}).pipe(Effect.provide(TestLLM.testLayer()))
 ```
 
-`TestLLM.push(...)` scripts one-shot responses, `TestLLM.always(...)` changes the fallback, and
-`TestLLM.wait(...)` lets concurrent tests wait until a request has arrived. Every received canonical request is
-available on the yielded `TestLLM.Service`.
+`testLayer()` provides the same object under `LLMClient.Service` and `TestLLM.Test`. Production consumes the
+normal client; tests use the additional controls. Each layer build has fresh state.
+
+- `test.push(...)` queues one-shot responses in execution order. Each argument is one response.
+- `test.always(response)` installs a repeatable fallback. The layer's `fallback` option sets its initial value.
+- `test.serve(request => response)` installs a request-dependent fallback. `always` and `serve` replace each
+  other without changing queued replies; queued replies take precedence.
+- `test.requests()` returns an array snapshot. `transformRequest` changes only the recorded observation;
+  `serve` receives the original canonical request.
+- `test.wait(count)` waits for request arrivals, not output or completion, and supports concurrent waiters.
+- `test.gate()` returns a scoped gate with countable `started` notifications and a `release` Effect. Release
+  unblocks all requests captured by that gate; closing its scope also releases it. Effect-aware test runners
+  already provide Scope.
+
+Constructing `stream()` or `generate()` does not record a request, invoke a responder, or consume a script.
+Each execution does. An exhausted queue without a fallback defects immediately rather than waiting for a
+future reply.
+
+Responses remain canonical event arrays or arbitrary `Stream<LLMEvent, AIError>` values. The client consumes
+supplied streams directly, preserving failure identity, finalizers, incomplete output, and post-finish tails;
+it does not repair or truncate them.
+
+The published legacy `Service`, `layer`, `clientLayer`, and module-level controls remain available as adapters
+over the same implementation, including the legacy live `requests` array. New tests should use `Test` and
+`testLayer`.
 
 ## Caching
 

packages/ai/src/testing.ts

@@ -1,6 +1,6 @@
 export * as TestLLM from "./testing.js"
 
-import { LLMClient, type Interface as LLMClientShape } from "./route/client.js"
+import { LLMClient } from "./route/client.js"
 import {
   LLMEvent,
   LLMResponse,
@@ -16,13 +16,33 @@ export type Response = readonly LLMEvent[] | Stream.Stream<LLMEvent, AIError>
 
 export type Gate = Readonly<{ started: Effect.Effect<void>; release: Effect.Effect<void> }>
 
+type ClientInterface = Context.Service.Shape<typeof LLMClient.Service>
+
+export type Responder = (request: LLMRequest) => Response
+
+export interface TestInterface extends ClientInterface {
+  /** Returns a snapshot of requests observed at execution time. */
+  readonly requests: () => Effect.Effect<readonly LLMRequest[]>
+  readonly push: (...responses: readonly Response[]) => Effect.Effect<void>
+  /** Replaces the fallback without changing queued responses. */
+  readonly always: (response: Response) => Effect.Effect<void>
+  /** Answers requests after the one-shot queue is exhausted; receives the original request. */
+  readonly serve: (responder: Responder) => Effect.Effect<void>
+  /** Waits for request arrivals, not output or completion. */
+  readonly wait: (count: number) => Effect.Effect<void>
+  readonly gate: () => Effect.Effect<Gate, never, Scope.Scope>
+}
+
+export class Test extends Context.Service<Test, TestInterface>()("@opencode/ai/TestLLM/Test") {}
+
+/** @deprecated Use TestInterface through Test and testLayer. */
 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
+  readonly client: ClientInterface
 }
 
 export interface LayerOptions {
@@ -31,6 +51,7 @@ export interface LayerOptions {
   readonly fallback?: Response
 }
 
+/** @deprecated Use Test and testLayer for normal client methods and test controls. */
 export class Service extends Context.Service<Service, Interface>()("@opencode/ai/TestLLM") {}
 
 export const complete = (
@@ -80,59 +101,64 @@ export const hangAfter = (...events: readonly LLMEvent[]) => Stream.concat(Strea
 
 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 make = (options: LayerOptions) =>
+  Effect.sync(() => {
+    const requests: LLMRequest[] = []
+    const responses: Response[] = []
+    let started = Deferred.makeUnsafe<void>()
+    let fallback: Response | Responder | undefined = 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 stream: ClientInterface["stream"] = (request) =>
+      Stream.suspend(() => {
+        const count = 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({
-        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")
-            }),
-          ),
+        try {
+          const response = responses.shift() ?? (typeof fallback === "function" ? fallback(request) : fallback)
+          if (!response) return Stream.die(new Error(`TestLLM has no response for request ${count}`))
+          const streamed = toStream(response)
+          if (!gate) return streamed
+          return Stream.unwrap(
+            Queue.offer(gate.started, undefined).pipe(Effect.andThen(gate.release.await), Effect.as(streamed)),
+          )
+        } finally {
+          // Waiters can resume synchronously; assign the reply and gate before notifying them.
+          Deferred.doneUnsafe(waiting, Effect.void)
+        }
       })
-
-      return Service.of({
-        requests,
-        push: (...input) =>
-          Effect.sync(() => {
-            responses.push(...input)
+    const test = Test.of({
+      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")
           }),
-        always: (response) =>
-          Effect.sync(() => {
-            fallback = response
-          }),
-        wait,
-        gate: Effect.gen(function* () {
+        ),
+      requests: () => Effect.sync(() => [...requests]),
+      push: (...input) =>
+        Effect.sync(() => {
+          responses.push(...input)
+        }),
+      always: (response) =>
+        Effect.sync(() => {
+          fallback = response
+        }),
+      serve: (responder) =>
+        Effect.sync(() => {
+          fallback = responder
+        }),
+      wait,
+      gate: () =>
+        Effect.gen(function* () {
           const gate = {
             started: yield* Effect.acquireRelease(Queue.unbounded<void>(), Queue.shutdown),
             release: yield* Latch.make(),
@@ -147,11 +173,36 @@ export const layer = (options: LayerOptions = {}) =>
             release,
           }
         }),
-        client,
-      })
-    }),
+    })
+
+    return { test, requests }
+  })
+
+/** Provides one shared implementation under the normal client and test-control tags. */
+export const testLayer = (options: LayerOptions = {}) =>
+  Layer.effectContext(
+    Effect.map(make(options), (implementation) =>
+      Context.make(LLMClient.Service, implementation.test).pipe(Context.add(Test, implementation.test)),
+    ),
   )
 
+/** @deprecated Use testLayer; retained for published callers of the legacy control interface. */
+export const layer = (options: LayerOptions = {}) =>
+  Layer.effect(
+    Service,
+    Effect.map(make(options), (implementation) =>
+      Service.of({
+        requests: implementation.requests,
+        push: implementation.test.push,
+        always: implementation.test.always,
+        wait: implementation.test.wait,
+        gate: implementation.test.gate(),
+        client: implementation.test,
+      }),
+    ),
+  )
+
+/** @deprecated testLayer provides LLMClient.Service directly. */
 export const clientLayer = Layer.effect(
   LLMClient.Service,
   Effect.map(Service, (service) => service.client),

packages/ai/test/exports.test.ts

@@ -32,6 +32,8 @@ describe("public exports", () => {
     expect(Provider.make).toBeFunction()
     expect(ProviderSubpath.make).toBe(Provider.make)
     expect(TestLLM.layer).toBeFunction()
+    expect(TestLLM.testLayer).toBeFunction()
+    expect(TestLLM.Test.of).toBeFunction()
   })
 
   test("route barrel exposes route-authoring APIs", () => {

packages/ai/test/testing.test.ts

@@ -0,0 +1,331 @@
+import { describe, expect } from "bun:test"
+import { AIError, LanguageModel, LLM, LLMClient, LLMEvent, LLMRequest, RateLimitError } from "../src/index.js"
+import { OpenAIChat } from "../src/protocols/openai-chat.js"
+import { TestLLM } from "../src/testing.js"
+import { Effect, Fiber, Latch, Stream } from "effect"
+import { testEffect } from "./lib/effect.js"
+
+const request = LLM.request({
+  model: LanguageModel.make({ id: "fictional-model", provider: "fixture", route: OpenAIChat.route }),
+  prompt: "Say hello",
+})
+const legacy = testEffect(TestLLM.layer())
+const it = testEffect(TestLLM.testLayer())
+
+describe("TestLLM legacy client", () => {
+  legacy.effect("does not observe requests or consume responses until execution", () =>
+    Effect.gen(function* () {
+      const llm = yield* TestLLM.Service
+      yield* llm.push(TestLLM.text("first", "first"), TestLLM.text("second", "second"))
+
+      llm.client.stream(request)
+      llm.client.generate(request)
+      expect(llm.requests).toEqual([])
+
+      expect((yield* llm.client.generate(request)).text).toBe("first")
+      expect((yield* llm.client.generate(request)).text).toBe("second")
+      expect(llm.requests).toEqual([request, request])
+    }),
+  )
+
+  legacy.effect("assigns and records a fresh response for each execution", () =>
+    Effect.gen(function* () {
+      const llm = yield* TestLLM.Service
+      yield* llm.push(
+        TestLLM.text("first", "first"),
+        TestLLM.text("second", "second"),
+        TestLLM.text("third", "third"),
+        TestLLM.text("fourth", "fourth"),
+      )
+      const stream = llm.client.stream(request)
+      const generate = llm.client.generate(request)
+
+      expect(yield* Stream.runCollect(stream)).toEqual(TestLLM.text("first", "first"))
+      expect(yield* Stream.runCollect(stream)).toEqual(TestLLM.text("second", "second"))
+      expect((yield* generate).text).toBe("third")
+      expect((yield* generate).text).toBe("fourth")
+      expect(llm.requests).toEqual([request, request, request, request])
+    }),
+  )
+
+  legacy.effect("keeps module-level controls and clientLayer on the same backing state", () =>
+    Effect.gen(function* () {
+      const llm = yield* TestLLM.Service
+      const requests = llm.requests
+      yield* TestLLM.push(TestLLM.text("queued", "queued"))
+      yield* TestLLM.always(TestLLM.text("fallback", "fallback"))
+      expect((yield* LLMClient.generate(request).pipe(Effect.provide(TestLLM.clientLayer))).text).toBe("queued")
+      yield* TestLLM.wait(1)
+      expect(requests).toEqual([request])
+      requests.length = 0
+      expect((yield* llm.client.generate(request)).text).toBe("fallback")
+      expect(llm.requests).toBe(requests)
+      expect(requests).toEqual([request])
+    }),
+  )
+})
+
+describe("TestLLM first-class client", () => {
+  it.effect("provides the same object under normal and test tags with snapshot observations", () =>
+    Effect.gen(function* () {
+      const llm = yield* TestLLM.Test
+      const client = yield* LLMClient.Service
+      expect(client).toBe(llm)
+      const before = yield* llm.requests()
+      yield* llm.push(TestLLM.text("hello", "answer"))
+      const generate = client.generate(request)
+      client.stream(request)
+      expect(yield* llm.requests()).toEqual([])
+
+      expect((yield* generate).text).toBe("hello")
+      expect(before).toEqual([])
+      expect(yield* llm.requests()).toEqual([request])
+      expect(yield* llm.requests()).not.toBe(yield* llm.requests())
+    }),
+  )
+
+  it.effect("prioritizes queued replies over request-dependent and constant fallbacks", () =>
+    Effect.gen(function* () {
+      const llm = yield* TestLLM.Test
+      const served: LLMRequest[] = []
+      yield* llm.always(TestLLM.text("old fallback", "old"))
+      yield* llm.push(TestLLM.text("first", "first"), TestLLM.text("second", "second"))
+      yield* llm.serve((request) => {
+        served.push(request)
+        return TestLLM.text(request.promptCacheKey ?? "default", "served")
+      })
+
+      expect((yield* LLMClient.generate(request)).text).toBe("first")
+      expect((yield* LLMClient.generate(request)).text).toBe("second")
+      expect(served).toEqual([])
+      const selected = LLMRequest.update(request, { promptCacheKey: "selected" })
+      expect((yield* LLMClient.generate(selected)).text).toBe("selected")
+      expect((yield* LLMClient.generate(request)).text).toBe("default")
+      expect(served).toEqual([selected, request])
+
+      yield* llm.push(TestLLM.text("queued again", "queued"))
+      yield* llm.always(TestLLM.text("constant", "constant"))
+      expect((yield* LLMClient.generate(request)).text).toBe("queued again")
+      expect((yield* LLMClient.generate(request)).text).toBe("constant")
+      expect((yield* LLMClient.generate(request)).text).toBe("constant")
+      expect(served).toEqual([selected, request])
+    }),
+  )
+
+  testEffect(
+    TestLLM.testLayer({
+      transformRequest: (request) => LLMRequest.update(request, { promptCacheKey: "observation" }),
+    }),
+  ).effect("transforms observations without changing the request passed to the responder", () =>
+    Effect.gen(function* () {
+      const llm = yield* TestLLM.Test
+      yield* llm.serve((input) => {
+        expect(input).toBe(request)
+        return TestLLM.text("original", "answer")
+      })
+      const generate = llm.generate(request)
+      expect(yield* llm.requests()).toEqual([])
+      expect((yield* generate).text).toBe("original")
+      expect(yield* llm.requests()).toEqual([LLMRequest.update(request, { promptCacheKey: "observation" })])
+    }),
+  )
+
+  it.effect("broadcasts request-arrival waits and satisfies waits registered afterward", () =>
+    Effect.gen(function* () {
+      const llm = yield* TestLLM.Test
+      yield* llm.always(TestLLM.stop())
+      const first = yield* llm.wait(2).pipe(Effect.forkChild({ startImmediately: true }))
+      const second = yield* llm.wait(2).pipe(Effect.forkChild({ startImmediately: true }))
+      yield* llm.generate(request)
+      expect(first.pollUnsafe()).toBeUndefined()
+      expect(second.pollUnsafe()).toBeUndefined()
+      yield* llm.generate(request)
+      yield* Fiber.join(first)
+      yield* Fiber.join(second)
+      yield* llm.wait(2)
+      expect(yield* llm.requests()).toHaveLength(2)
+    }),
+  )
+  ;(["queued", "served"] as const).forEach((mode) => {
+    it.effect(`assigns ${mode} replies before resuming request-arrival continuations`, () =>
+      Effect.gen(function* () {
+        const llm = yield* TestLLM.Test
+        const responses = [TestLLM.text("first", "first"), TestLLM.text("second", "second")]
+        yield* mode === "queued" ? llm.push(...responses) : llm.serve(() => responses.shift() ?? [])
+        const later = yield* llm
+          .wait(1)
+          .pipe(Effect.andThen(llm.generate(request)), Effect.forkChild({ startImmediately: true }))
+
+        expect((yield* llm.generate(request)).text).toBe("first")
+        expect((yield* Fiber.join(later)).text).toBe("second")
+        expect(yield* llm.requests()).toEqual([request, request])
+      }),
+    )
+  })
+
+  it.effect("notifies arrival waiters even when the responder defects", () =>
+    Effect.gen(function* () {
+      const llm = yield* TestLLM.Test
+      const defect = new Error("Broken fixture responder")
+      yield* llm.serve(() => {
+        throw defect
+      })
+      const waiter = yield* llm.wait(1).pipe(Effect.forkChild({ startImmediately: true }))
+      expect(yield* llm.generate(request).pipe(Effect.catchDefect(Effect.succeed))).toBe(defect)
+      yield* Fiber.join(waiter)
+    }),
+  )
+
+  it.effect("builds independent state even when the same layer is provided concurrently", () => {
+    const layer = TestLLM.testLayer()
+    const run = Effect.gen(function* () {
+      const llm = yield* TestLLM.Test
+      expect(yield* llm.requests()).toEqual([])
+      yield* llm.push(TestLLM.text("one", "answer"))
+      expect((yield* LLMClient.generate(request)).text).toBe("one")
+      return yield* llm.requests()
+    }).pipe(Effect.provide(layer))
+    return Effect.gen(function* () {
+      expect(yield* Effect.all([run, run], { concurrency: "unbounded" })).toEqual([[request], [request]])
+    })
+  })
+
+  it.effect("counts concurrent starts on one gate without serializing their response assignment", () =>
+    Effect.gen(function* () {
+      const llm = yield* TestLLM.Test
+      yield* llm.push(TestLLM.text("first", "first"), TestLLM.text("second", "second"))
+      const generate = llm.generate(request)
+      const gate = yield* llm.gate()
+      const first = yield* generate.pipe(Effect.forkChild({ startImmediately: true }))
+      yield* gate.started
+      const second = yield* generate.pipe(Effect.forkChild({ startImmediately: true }))
+      yield* gate.started
+      yield* llm.wait(2)
+      expect(first.pollUnsafe()).toBeUndefined()
+      expect(second.pollUnsafe()).toBeUndefined()
+      yield* gate.release
+      expect((yield* Fiber.join(first)).text).toBe("first")
+      expect((yield* Fiber.join(second)).text).toBe("second")
+    }),
+  )
+
+  it.effect("does not clear a replacement gate when the previous gate is released", () =>
+    Effect.gen(function* () {
+      const llm = yield* TestLLM.Test
+      yield* llm.always(TestLLM.stop())
+      const previous = yield* llm.gate()
+      const first = yield* llm.generate(request).pipe(Effect.forkChild({ startImmediately: true }))
+      yield* previous.started
+      const next = yield* llm.gate()
+      yield* previous.release
+      yield* Fiber.join(first)
+
+      const second = yield* llm.generate(request).pipe(Effect.forkChild({ startImmediately: true }))
+      yield* next.started
+      expect(second.pollUnsafe()).toBeUndefined()
+      yield* next.release
+      yield* Fiber.join(second)
+    }),
+  )
+
+  it.effect("releases a gate when its deliberately narrower scope closes", () =>
+    Effect.gen(function* () {
+      const llm = yield* TestLLM.Test
+      yield* llm.always(TestLLM.stop())
+      // Only the gate is scoped here; its release must happen before the test ends.
+      const run = yield* Effect.scoped(
+        Effect.gen(function* () {
+          const gate = yield* llm.gate()
+          const run = yield* llm.generate(request).pipe(Effect.forkChild({ startImmediately: true }))
+          yield* gate.started
+          return run
+        }),
+      )
+      yield* Fiber.join(run)
+      yield* llm.generate(request)
+      expect(yield* llm.requests()).toHaveLength(2)
+    }),
+  )
+
+  it.effect("keeps an executed response consumed after interruption and permits later requests", () =>
+    Effect.gen(function* () {
+      const llm = yield* TestLLM.Test
+      yield* llm.push(TestLLM.text("interrupted", "first"), TestLLM.text("next", "second"))
+      const gate = yield* llm.gate()
+      const run = yield* llm.generate(request).pipe(Effect.forkChild({ startImmediately: true }))
+      yield* gate.started
+      yield* Fiber.interrupt(run)
+      yield* gate.release
+      expect((yield* llm.generate(request)).text).toBe("next")
+      expect(yield* llm.requests()).toHaveLength(2)
+    }),
+  )
+
+  it.effect("consumes a supplied stream's post-finish tail and runs its finalizer", () =>
+    Effect.gen(function* () {
+      const llm = yield* TestLLM.Test
+      const tail = yield* Latch.make()
+      const release = yield* Latch.make()
+      const finalized = yield* Latch.make()
+      yield* llm.push(
+        Stream.unwrap(
+          Effect.gen(function* () {
+            yield* Effect.addFinalizer(() => finalized.open)
+            return Stream.fromIterable(TestLLM.text("complete", "answer")).pipe(
+              Stream.concat(Stream.fromEffect(tail.open.pipe(Effect.andThen(release.await))).pipe(Stream.drain)),
+            )
+          }),
+        ),
+      )
+      const run = yield* llm.generate(request).pipe(Effect.forkChild({ startImmediately: true }))
+      yield* tail.await
+      expect(run.pollUnsafe()).toBeUndefined()
+      yield* release.open
+      expect((yield* Fiber.join(run)).text).toBe("complete")
+      yield* finalized.await
+    }),
+  )
+
+  it.effect("preserves irregular events, ordinary EOF, typed failures, and responder defects", () =>
+    Effect.gen(function* () {
+      const llm = yield* TestLLM.Test
+      const events = [LLMEvent.textDelta({ id: "without-start", text: "partial" })]
+      yield* llm.push(events, [])
+      expect(yield* Stream.runCollect(llm.stream(request))).toEqual(events)
+      expect(yield* Stream.runCollect(llm.stream(request))).toEqual([])
+
+      const failure = new AIError({ reason: new RateLimitError({ message: "Try later" }) })
+      const observed: LLMEvent[] = []
+      yield* llm.serve(() => TestLLM.failAfter(failure, ...events))
+      expect(
+        yield* llm.stream(request).pipe(
+          Stream.runForEach((event) => Effect.sync(() => observed.push(event))),
+          Effect.flip,
+        ),
+      ).toBe(failure)
+      expect(observed).toEqual(events)
+      expect(yield* llm.generate(request).pipe(Effect.flip)).toBe(failure)
+
+      const defect = new Error("Broken fixture responder")
+      yield* llm.serve(() => {
+        throw defect
+      })
+      expect(yield* llm.generate(request).pipe(Effect.catchDefect(Effect.succeed))).toBe(defect)
+      yield* llm.push(TestLLM.text("recovered", "answer"))
+      expect((yield* llm.generate(request)).text).toBe("recovered")
+    }),
+  )
+
+  it.effect("defects on unexpected requests instead of waiting for a late script", () =>
+    Effect.gen(function* () {
+      const llm = yield* TestLLM.Test
+      const defect = yield* llm.generate(request).pipe(Effect.catchDefect(Effect.succeed))
+      expect(defect).toBeInstanceOf(Error)
+      if (!(defect instanceof Error)) return
+      expect(defect.message).toBe("TestLLM has no response for request 1")
+      expect(yield* llm.requests()).toEqual([request])
+      yield* llm.push(TestLLM.stop())
+      yield* llm.generate(request)
+    }),
+  )
+})

packages/ai/tsconfig.types.json

@@ -5,5 +5,5 @@
     "noEmit": true,
     "rootDir": "."
   },
-  "include": ["test/**/*.types.ts"]
+  "include": ["test/**/*.types.ts", "test/testing.test.ts"]
 }

packages/core/test/generate.test.ts

@@ -65,7 +65,7 @@ const aisdk = Layer.mock(AISDK.Service, {
   },
   model: () => Effect.succeed(runtime),
 })
-const client = TestLLM.clientLayer.pipe(Layer.provide(TestLLM.layer({ fallback: TestLLM.text("OK", "generate") })))
+const client = TestLLM.testLayer({ 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))))

packages/core/test/session-step.test.ts

@@ -1,5 +1,5 @@
 import { expect } from "bun:test"
-import { LanguageModel, LLM, LLMClient, LLMEvent } from "@opencode-ai/ai"
+import { LanguageModel, LLM, LLMEvent } from "@opencode-ai/ai"
 import { OpenAIChat } from "@opencode-ai/ai/protocols/openai-chat"
 import { TestLLM } from "@opencode-ai/ai/testing"
 import { Agent } from "@opencode-ai/core/agent"
@@ -29,7 +29,7 @@ const it = testEffect(
     AppNodeBuilder.build(LayerNode.group([Database.node, Bus.node, SessionProjector.node, ToolOutput.node]), [
       [Bus.node, Bus.configured({ persist: true })],
     ]),
-    TestLLM.layer(),
+    TestLLM.testLayer(),
   ),
 )
 
@@ -37,7 +37,7 @@ for (const finish of ["stop", "content-filter"] as const) {
   it.effect(`settles ${finish} with snapshot files and nonzero usage after its tool`, () =>
     Effect.gen(function* () {
       const db = (yield* Database.Service).db
-      const llm = yield* TestLLM.Service
+      const llm = yield* TestLLM.Test
       const sessionID = Session.ID.create()
       const assistantMessageID = SessionMessage.ID.create()
       const start = Snapshot.ID.make("before")
@@ -45,7 +45,6 @@ for (const finish of ["stop", "content-filter"] as const) {
       const files = [RelativePath.make("changed.ts")]
       let captures = 0
       const steps = yield* SessionStep.make.pipe(
-        Effect.provideService(LLMClient.Service, llm.client),
         Effect.provide(
           Layer.mock(Snapshot.Service)({
             capture: () => Effect.sync(() => (captures++ === 0 ? start : end)),
@@ -111,7 +110,7 @@ for (const finish of ["stop", "content-filter"] as const) {
         })
         .pipe(Effect.exit)
       expect(Exit.isSuccess(result)).toBe(finish === "stop")
-      expect(llm.requests).toHaveLength(1)
+      expect(yield* llm.requests()).toHaveLength(1)
       expect(captures).toBe(2)
       const message = yield* db
         .select()

packages/sdk/test/embedded.test.ts

@@ -1,7 +1,7 @@
 import fs from "fs/promises"
 import path from "path"
 import { expect } from "bun:test"
-import { LanguageModel, LLMClient, LLMResponse, type LLMRequest } from "@opencode-ai/ai"
+import { LanguageModel, LLMClient } from "@opencode-ai/ai"
 import { OpenAIChat } from "@opencode-ai/ai/protocols"
 import { TestLLM } from "@opencode-ai/ai/testing"
 import { llmClient } from "@opencode-ai/core/effect/app-node-platform"
@@ -40,8 +40,8 @@ for (const selection of ["explicit", "default"] as const) {
     withEmbedded("opencode-embedded-generate-", (fixture) =>
       Effect.gen(function* () {
         const release = yield* Latch.make()
-        const llm = yield* TestLLM.Service.pipe(
-          Effect.provide(TestLLM.layer({ fallback: TestLLM.text("ready", "answer") })),
+        const llm = yield* TestLLM.Test.pipe(
+          Effect.provide(TestLLM.testLayer({ fallback: TestLLM.text("ready", "answer") })),
         )
         const supervisor = Layer.effect(
           PluginSupervisor.Service,
@@ -71,7 +71,7 @@ for (const selection of ["explicit", "default"] as const) {
           },
           {
             overrides: [
-              [llmClient, Layer.succeed(LLMClient.Service, llm.client)],
+              [llmClient, Layer.succeed(LLMClient.Service, llm)],
               [PluginSupervisor.node, { ...PluginSupervisor.node, implementation: supervisor }],
             ],
           },
@@ -92,8 +92,9 @@ for (const selection of ["explicit", "default"] as const) {
         })
 
         expect(result.text).toBe("ready")
-        expect(llm.requests).toHaveLength(1)
-        expect(llm.requests[0]?.model).toMatchObject({ provider: "custom", id: "fictional-chat" })
+        const requests = yield* llm.requests()
+        expect(requests).toHaveLength(1)
+        expect(requests[0]?.model).toMatchObject({ provider: "custom", id: "fictional-chat" })
       }),
     ),
   )
@@ -644,17 +645,13 @@ const workspaceModelScenario = (fixture: Fixture, policy: "eager" | "lazy") =>
     const modelStarted = yield* Deferred.make<void>()
     yield* Effect.addFinalizer(() => Deferred.succeed(createRelease, undefined).pipe(Effect.asVoid))
     const model = LanguageModel.make({ id: "workspace-test", provider: "test", route: OpenAIChat.route })
-    const client = TestLLM.clientLayer.pipe(
-      Layer.provide(
-        TestLLM.layer({
-          fallback: TestLLM.text("ready", "answer"),
-          transformRequest: (request) => {
-            Deferred.doneUnsafe(modelStarted, Effect.void)
-            return request
-          },
-        }),
-      ),
-    )
+    const client = TestLLM.testLayer({
+      fallback: TestLLM.text("ready", "answer"),
+      transformRequest: (request) => {
+        Deferred.doneUnsafe(modelStarted, Effect.void)
+        return request
+      },
+    })
     const models = Layer.mock(SessionRunnerModel.Service, {
       resolve: () =>
         Effect.succeed(
@@ -752,27 +749,13 @@ it.live(
         // The first tool-advertising request selects the shell tool; everything else
         // (including title generation, which carries no tools) answers with text.
         let toolIssued = false
-        const respond = (request: LLMRequest) => {
+        const llm = yield* TestLLM.Test.pipe(Effect.provide(TestLLM.testLayer()))
+        yield* llm.serve((request) => {
           const wantsTool = !toolIssued && request.tools.some((tool) => tool.name === "shell")
           if (!wantsTool) return TestLLM.text("done", "answer")
           toolIssued = true
           return TestLLM.tool("call-shell", "shell", { command: "echo hi" })
-        }
-        const client = Layer.succeed(
-          LLMClient.Service,
-          LLMClient.Service.of({
-            stream: (request) => Stream.fromIterable(respond(request)),
-            generate: (request) =>
-              Stream.fromIterable(respond(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("test response ended without a terminal finish event")
-                }),
-              ),
-          }),
-        )
+        })
         const models = Layer.mock(SessionRunnerModel.Service, {
           resolve: () =>
             Effect.succeed(
@@ -807,7 +790,7 @@ it.live(
           },
           {
             overrides: [
-              [llmClient, client],
+              [llmClient, Layer.succeed(LLMClient.Service, llm)],
               [SessionRunnerModel.node, models],
             ],
           },