test(core): add runner scenario DSL

packages/core/test/lib/runner-scenario.ts

@@ -0,0 +1,278 @@
+import {
+  LLMClient,
+  LLMEvent,
+  type LLMClientShape,
+  type LLMClientService,
+  type LLMError,
+  type LLMRequest,
+} from "@opencode-ai/llm"
+import { Cause, Deferred, Effect, Exit, Fiber, Layer, Option, Queue, Ref, Stream } from "effect"
+
+class RunnerEndedError extends Error {
+  constructor() {
+    super("Runner completed before the next LLM request")
+  }
+}
+
+class RunnerScenarioUsedError extends Error {
+  constructor() {
+    super("RunnerScenario.run may only be called once")
+  }
+}
+
+class RunnerLLMRequestError extends Error {
+  constructor(readonly id: number) {
+    super(`Runner LLM request #${id} was not handled by the interaction`)
+  }
+}
+
+class RunnerLLMResponseClosedError extends Error {
+  constructor(readonly id: number) {
+    super(`Runner LLM request #${id} is no longer awaiting a response`)
+  }
+}
+
+export interface RunnerLLMCall {
+  readonly request: LLMRequest
+  readonly respond: {
+    readonly stop: () => Effect.Effect<void, Error>
+    readonly text: (text: string, options?: { readonly id?: string }) => Effect.Effect<void, Error>
+    readonly toolCall: (
+      name: string,
+      input: unknown,
+      options?: { readonly id?: string; readonly providerExecuted?: boolean },
+    ) => Effect.Effect<void, Error>
+    readonly events: (...events: ReadonlyArray<LLMEvent>) => Effect.Effect<void, Error>
+    readonly stream: (stream: Stream.Stream<LLMEvent, LLMError>) => Effect.Effect<void, Error>
+    readonly fail: (error: LLMError) => Effect.Effect<void, Error>
+  }
+}
+
+interface State {
+  readonly started: boolean
+  readonly accepting: boolean
+  readonly ended: boolean
+  readonly nextID: number
+  readonly calls: ReadonlyMap<number, "pending" | "responded" | "closed">
+  readonly requests: ReadonlyArray<LLMRequest>
+}
+
+type Registration =
+  | { readonly _tag: "Registered"; readonly id: number }
+  | { readonly _tag: "Unexpected"; readonly error: Error }
+
+interface PendingCall {
+  readonly id: number
+  readonly call: RunnerLLMCall
+}
+
+interface RunnerLLMInternal {
+  readonly llm: RunnerLLM
+  readonly begin: Effect.Effect<void, RunnerScenarioUsedError>
+  readonly finishInteraction: Effect.Effect<void, Error>
+  readonly end: Effect.Effect<void>
+}
+
+export interface RunnerLLM {
+  readonly layer: Layer.Layer<LLMClientService>
+  readonly next: () => Effect.Effect<RunnerLLMCall, Error>
+  readonly requests: Effect.Effect<ReadonlyArray<LLMRequest>>
+}
+
+const events = (...events: ReadonlyArray<LLMEvent>) => Stream.fromIterable(events)
+
+const stop = () =>
+  events(
+    LLMEvent.stepStart({ index: 0 }),
+    LLMEvent.stepFinish({ index: 0, reason: "stop" }),
+    LLMEvent.finish({ reason: "stop" }),
+  )
+
+const text = (value: string, options?: { readonly id?: string }) => {
+  const id = options?.id ?? "text"
+  return events(
+    LLMEvent.stepStart({ index: 0 }),
+    LLMEvent.textStart({ id }),
+    LLMEvent.textDelta({ id, text: value }),
+    LLMEvent.textEnd({ id }),
+    LLMEvent.stepFinish({ index: 0, reason: "stop" }),
+    LLMEvent.finish({ reason: "stop" }),
+  )
+}
+
+const toolCall = (
+  name: string,
+  input: unknown,
+  options?: { readonly id?: string; readonly providerExecuted?: boolean },
+) => {
+  const id = options?.id ?? `call-${name}`
+  return events(
+    LLMEvent.stepStart({ index: 0 }),
+    LLMEvent.toolCall({ id, name, input, providerExecuted: options?.providerExecuted }),
+    LLMEvent.stepFinish({ index: 0, reason: "tool-calls" }),
+    LLMEvent.finish({ reason: "tool-calls" }),
+  )
+}
+
+const makeRunnerLLM = Effect.gen(function* () {
+  const calls = yield* Queue.unbounded<PendingCall, Cause.Done>()
+  const state = yield* Ref.make<State>({
+    started: false,
+    accepting: false,
+    ended: false,
+    nextID: 1,
+    calls: new Map(),
+    requests: [],
+  })
+
+  const respond = (
+    id: number,
+    response: Deferred.Deferred<Stream.Stream<LLMEvent, LLMError>>,
+    stream: Stream.Stream<LLMEvent, LLMError>,
+  ) =>
+    Effect.uninterruptible(
+      Effect.gen(function* () {
+        const pending = yield* Ref.modify(state, (current) => {
+          if (current.calls.get(id) !== "pending") return [false, current]
+          return [true, { ...current, calls: new Map(current.calls).set(id, "responded") }]
+        })
+        if (!pending) return yield* Effect.fail(new RunnerLLMResponseClosedError(id))
+        yield* Deferred.succeed(response, stream)
+      }),
+    )
+
+  const stream = (request: LLMRequest) =>
+    Stream.unwrap(
+      Effect.uninterruptibleMask((restore) =>
+        Effect.gen(function* () {
+          const response = yield* Deferred.make<Stream.Stream<LLMEvent, LLMError>>()
+          const registered = yield* Ref.modify<State, Registration>(state, (current) => {
+            if (!current.accepting) {
+              const error = new RunnerLLMRequestError(current.nextID)
+              return [
+                { _tag: "Unexpected" as const, error },
+                { ...current, nextID: current.nextID + 1, requests: [...current.requests, request] },
+              ]
+            }
+            return [
+              { _tag: "Registered" as const, id: current.nextID },
+              {
+                ...current,
+                nextID: current.nextID + 1,
+                calls: new Map(current.calls).set(current.nextID, "pending"),
+                requests: [...current.requests, request],
+              },
+            ]
+          })
+          if (registered._tag === "Unexpected") return yield* Effect.fail(registered.error)
+          const reply = (stream: Stream.Stream<LLMEvent, LLMError>) => respond(registered.id, response, stream)
+          yield* Queue.offer(calls, {
+            id: registered.id,
+            call: {
+              request,
+              respond: {
+                stop: () => reply(stop()),
+                text: (value, options) => reply(text(value, options)),
+                toolCall: (name, input, options) => reply(toolCall(name, input, options)),
+                events: (...input) => reply(events(...input)),
+                stream: reply,
+                fail: (error) => reply(Stream.fail(error)),
+              },
+            },
+          })
+          return yield* restore(Deferred.await(response)).pipe(
+            Effect.onInterrupt(() =>
+              Ref.update(state, (current) =>
+                current.calls.get(registered.id) === "pending"
+                  ? { ...current, calls: new Map(current.calls).set(registered.id, "closed") }
+                  : current,
+              ),
+            ),
+          )
+        }),
+      ),
+    )
+
+  const next = (): Effect.Effect<RunnerLLMCall, Error> =>
+    Effect.gen(function* () {
+      if ((yield* Ref.get(state)).ended) return yield* Effect.fail(new RunnerEndedError())
+      const pending = yield* Queue.take(calls).pipe(Effect.mapError(() => new RunnerEndedError()))
+      const current = yield* Ref.get(state)
+      if (current.ended) return yield* Effect.fail(new RunnerEndedError())
+      if (current.calls.get(pending.id) === "pending") return pending.call
+      return yield* next()
+    })
+
+  return {
+    llm: {
+      layer: Layer.succeed(
+        LLMClient.Service,
+        LLMClient.Service.of({
+          prepare: () => Effect.die("RunnerLLM.prepare is not implemented"),
+          stream: stream as LLMClientShape["stream"],
+          generate: () => Effect.die("RunnerLLM.generate is not implemented"),
+        }),
+      ),
+      next,
+      requests: Ref.get(state).pipe(Effect.map((state) => state.requests)),
+    },
+    begin: Ref.modify<State, Effect.Effect<void, RunnerScenarioUsedError>>(state, (current) =>
+      current.started
+        ? [Effect.fail(new RunnerScenarioUsedError()), current]
+        : [Effect.void, { ...current, started: true, accepting: true }],
+    ).pipe(Effect.flatten),
+    finishInteraction: Effect.gen(function* () {
+      const unanswered = yield* Ref.modify(state, (current) => [
+        [...current.calls].find(([, status]) => status === "pending")?.[0],
+        { ...current, accepting: false },
+      ])
+      if (unanswered !== undefined) return yield* Effect.fail(new RunnerLLMRequestError(unanswered))
+    }),
+    end: Effect.gen(function* () {
+      yield* Ref.update(state, (current) => ({ ...current, accepting: false, ended: true }))
+      yield* Queue.end(calls)
+    }),
+  } satisfies RunnerLLMInternal
+})
+
+export interface RunnerScenario<RunError, RunRequirements> {
+  readonly llm: RunnerLLM
+  readonly run: <A, E, R>(
+    interaction: () => Effect.gen.Return<A, E, R>,
+  ) => Effect.Effect<A, RunError | E | Error, RunRequirements | R>
+}
+
+export namespace RunnerScenario {
+  export const make = <RunResult, RunError, RunRequirements>(
+    start: (llm: RunnerLLM) => Effect.Effect<RunResult, RunError, RunRequirements>,
+  ): Effect.Effect<RunnerScenario<RunError, RunRequirements>> =>
+    Effect.gen(function* () {
+      const internal = yield* makeRunnerLLM
+      const run = <A, E, R>(
+        interaction: () => Effect.gen.Return<A, E, R>,
+      ): Effect.Effect<A, RunError | E | Error, RunRequirements | R> =>
+        Effect.uninterruptibleMask((restore) =>
+          Effect.gen(function* () {
+            yield* internal.begin
+            const runner = yield* start(internal.llm).pipe(Effect.ensuring(internal.end), Effect.forkChild)
+            return yield* restore(
+              Effect.gen(function* () {
+                const interactionExit = yield* Effect.gen(interaction).pipe(Effect.exit)
+                if (Exit.isSuccess(interactionExit)) {
+                  yield* internal.finishInteraction
+                  yield* Fiber.join(runner)
+                  return interactionExit.value
+                }
+                const error = Cause.findErrorOption(interactionExit.cause)
+                if (Option.isSome(error) && error.value instanceof RunnerEndedError) yield* Fiber.join(runner)
+                return yield* Effect.failCause(interactionExit.cause)
+              }),
+            ).pipe(Effect.ensuring(Fiber.interrupt(runner).pipe(Effect.andThen(internal.end))))
+          }),
+        )
+      return {
+        llm: internal.llm,
+        run,
+      }
+    })
+}

packages/core/test/runner-scenario.test.ts

@@ -0,0 +1,266 @@
+import { describe, expect } from "bun:test"
+import { LLM, LLMError, LLMEvent, Model, TransportReason } from "@opencode-ai/llm"
+import * as OpenAIChat from "@opencode-ai/llm/protocols/openai-chat"
+import { Deferred, Effect, Fiber, Stream } from "effect"
+import { it } from "./lib/effect"
+import { RunnerScenario } from "./lib/runner-scenario"
+
+const model = Model.make({ id: "fake-model", provider: "fake", route: OpenAIChat.route })
+const request = (text: string) => LLM.request({ model, prompt: text })
+const textOf = (events: ReadonlyArray<LLMEvent>) => events.filter(LLMEvent.is.textDelta).map((event) => event.text)
+
+describe("RunnerScenario", () => {
+  it.effect("drives provider requests in execution order", () =>
+    Effect.gen(function* () {
+      const output: LLMEvent[][] = []
+      const scenario = yield* RunnerScenario.make((llm) =>
+        Effect.gen(function* () {
+          output.push(Array.from(yield* LLM.stream(request("Echo this")).pipe(Stream.runCollect)))
+          output.push(Array.from(yield* LLM.stream(request("Continue")).pipe(Stream.runCollect)))
+        }).pipe(Effect.provide(llm.layer)),
+      )
+
+      yield* scenario.run(function* () {
+        const first = yield* scenario.llm.next()
+        expect(first.request.messages[0]?.content).toEqual([{ type: "text", text: "Echo this" }])
+        yield* first.respond.toolCall("echo", { text: "hello" })
+
+        const second = yield* scenario.llm.next()
+        expect(second.request.messages[0]?.content).toEqual([{ type: "text", text: "Continue" }])
+        yield* second.respond.text("Done")
+      })
+
+      expect(output[0]?.find(LLMEvent.is.toolCall)).toMatchObject({ name: "echo", input: { text: "hello" } })
+      expect(textOf(output[1] ?? [])).toEqual(["Done"])
+    }),
+  )
+
+  it.effect("pauses the runner until the pending request receives a response", () =>
+    Effect.gen(function* () {
+      const completed = yield* Deferred.make<void>()
+      const scenario = yield* RunnerScenario.make((llm) =>
+        LLM.stream(request("Wait")).pipe(
+          Stream.runDrain,
+          Effect.andThen(Deferred.succeed(completed, undefined)),
+          Effect.provide(llm.layer),
+        ),
+      )
+
+      yield* scenario.run(function* () {
+        const call = yield* scenario.llm.next()
+        expect(Deferred.isDoneUnsafe(completed)).toBe(false)
+        yield* call.respond.stop()
+      })
+
+      expect(Deferred.isDoneUnsafe(completed)).toBe(true)
+    }),
+  )
+
+  it.effect("accepts raw failing streams for lifecycle edge cases", () =>
+    Effect.gen(function* () {
+      const failure = new LLMError({
+        module: "test",
+        method: "stream",
+        reason: new TransportReason({ message: "unavailable" }),
+      })
+      const scenario = yield* RunnerScenario.make((llm) =>
+        LLM.stream(request("Fail")).pipe(Stream.runDrain, Effect.provide(llm.layer)),
+      )
+
+      expect(
+        yield* scenario
+          .run(function* () {
+            const call = yield* scenario.llm.next()
+            yield* call.respond.stream(
+              Stream.concat(Stream.make(LLMEvent.stepStart({ index: 0 })), Stream.fail(failure)),
+            )
+          })
+          .pipe(Effect.flip),
+      ).toBe(failure)
+    }),
+  )
+
+  it.effect("fails when an interaction leaves a request unanswered", () =>
+    Effect.gen(function* () {
+      const scenario = yield* RunnerScenario.make((llm) =>
+        LLM.stream(request("Wait")).pipe(Stream.runDrain, Effect.provide(llm.layer)),
+      )
+
+      const failure = yield* scenario
+        .run(function* () {
+          yield* scenario.llm.next()
+        })
+        .pipe(Effect.flip)
+
+      expect(failure).toMatchObject({ message: "Runner LLM request #1 was not handled by the interaction" })
+    }),
+  )
+
+  it.effect("fails when the runner starts another request before the interaction finishes", () =>
+    Effect.gen(function* () {
+      const scenario = yield* RunnerScenario.make((llm) =>
+        Effect.gen(function* () {
+          yield* LLM.stream(request("First")).pipe(Stream.runDrain)
+          yield* LLM.stream(request("Unexpected")).pipe(Stream.runDrain)
+        }).pipe(Effect.provide(llm.layer)),
+      )
+
+      const failure = yield* scenario
+        .run(function* () {
+          const call = yield* scenario.llm.next()
+          yield* call.respond.stop()
+        })
+        .pipe(Effect.flip)
+
+      expect(failure).toMatchObject({ message: "Runner LLM request #2 was not handled by the interaction" })
+    }),
+  )
+
+  it.effect("rejects a second response to the same request", () =>
+    Effect.gen(function* () {
+      const scenario = yield* RunnerScenario.make((llm) =>
+        LLM.stream(request("Once")).pipe(Stream.runDrain, Effect.provide(llm.layer)),
+      )
+      let duplicate: Error | undefined
+
+      yield* scenario.run(function* () {
+        const call = yield* scenario.llm.next()
+        yield* call.respond.stop()
+        duplicate = yield* call.respond.stop().pipe(Effect.flip, Effect.orDie)
+      })
+      expect(duplicate?.message).toBe("Runner LLM request #1 is no longer awaiting a response")
+    }),
+  )
+
+  it.effect("does not wait for the runner after a duplicate response", () =>
+    Effect.gen(function* () {
+      const scenario = yield* RunnerScenario.make((llm) =>
+        LLM.stream(request("Once")).pipe(Stream.runDrain, Effect.andThen(Effect.never), Effect.provide(llm.layer)),
+      )
+
+      expect(
+        yield* scenario
+          .run(function* () {
+            const call = yield* scenario.llm.next()
+            yield* call.respond.stop()
+            yield* call.respond.stop()
+          })
+          .pipe(Effect.flip),
+      ).toMatchObject({ message: "Runner LLM request #1 is no longer awaiting a response" })
+    }),
+  )
+
+  it.effect("surfaces a runner failure while waiting for its next request", () =>
+    Effect.gen(function* () {
+      const failure = new Error("runner failed")
+      const scenario = yield* RunnerScenario.make(() => Effect.fail<Error>(failure))
+
+      expect(
+        yield* scenario
+          .run(function* () {
+            yield* scenario.llm.next()
+          })
+          .pipe(Effect.flip),
+      ).toBe(failure)
+    }),
+  )
+
+  it.effect("preserves a runner failure when its pending request closes", () =>
+    Effect.gen(function* () {
+      const fail = yield* Deferred.make<void>()
+      const closed = yield* Deferred.make<void>()
+      const failure = new Error("runner failed")
+      const scenario = yield* RunnerScenario.make((llm) =>
+        Effect.raceFirst(
+          LLM.stream(request("Fail pending")).pipe(
+            Stream.runDrain,
+            Effect.ensuring(Deferred.succeed(closed, undefined)),
+          ),
+          Deferred.await(fail).pipe(Effect.andThen(Effect.fail(failure))),
+        ).pipe(Effect.provide(llm.layer)),
+      )
+
+      expect(
+        yield* scenario
+          .run(function* () {
+            const call = yield* scenario.llm.next()
+            yield* Deferred.succeed(fail, undefined)
+            yield* Deferred.await(closed)
+            expect((yield* call.respond.stop().pipe(Effect.flip)).message).toBe(
+              "Runner LLM request #1 is no longer awaiting a response",
+            )
+            yield* scenario.llm.next()
+          })
+          .pipe(Effect.flip),
+      ).toBe(failure)
+    }),
+  )
+
+  it.effect("rejects a stale response after the runner closes a request", () =>
+    Effect.gen(function* () {
+      const release = yield* Deferred.make<void>()
+      const closed = yield* Deferred.make<void>()
+      const scenario = yield* RunnerScenario.make((llm) =>
+        Effect.raceFirst(
+          LLM.stream(request("Close pending")).pipe(
+            Stream.runDrain,
+            Effect.ensuring(Deferred.succeed(closed, undefined)),
+          ),
+          Deferred.await(release),
+        ).pipe(Effect.provide(llm.layer)),
+      )
+
+      yield* scenario.run(function* () {
+        const call = yield* scenario.llm.next()
+        yield* Deferred.succeed(release, undefined)
+        yield* Deferred.await(closed)
+        expect((yield* call.respond.stop().pipe(Effect.flip)).message).toBe(
+          "Runner LLM request #1 is no longer awaiting a response",
+        )
+      })
+    }),
+  )
+
+  it.effect("interrupts a runner that hangs after its final response", () =>
+    Effect.gen(function* () {
+      const started = yield* Deferred.make<void>()
+      const scenario = yield* RunnerScenario.make((llm) =>
+        LLM.stream(request("Hang")).pipe(
+          Stream.runDrain,
+          Effect.andThen(Deferred.succeed(started, undefined)),
+          Effect.andThen(Effect.never),
+          Effect.provide(llm.layer),
+        ),
+      )
+      const run = yield* scenario
+        .run(function* () {
+          const call = yield* scenario.llm.next()
+          yield* call.respond.stop()
+        })
+        .pipe(Effect.forkChild)
+      yield* Deferred.await(started)
+
+      yield* Fiber.interrupt(run)
+
+      expect(run.pollUnsafe()).toBeDefined()
+    }),
+  )
+
+  it.effect("rejects running one scenario twice", () =>
+    Effect.gen(function* () {
+      const scenario = yield* RunnerScenario.make((llm) =>
+        LLM.stream(request("Once")).pipe(Stream.runDrain, Effect.provide(llm.layer)),
+      )
+      yield* scenario.run(function* () {
+        yield* (yield* scenario.llm.next()).respond.stop()
+      })
+      const failure = yield* scenario.run(function* (): Effect.gen.Return<void> {}).pipe(Effect.exit)
+      expect(failure._tag).toBe("Failure")
+      if (failure._tag !== "Failure") throw new Error("Expected second run to fail")
+      const error = failure.cause.reasons.find((reason) => reason._tag === "Fail")
+      expect(error?._tag).toBe("Fail")
+      if (error?._tag !== "Fail") throw new Error("Expected a typed failure")
+      expect(error.error).toMatchObject({ message: "RunnerScenario.run may only be called once" })
+    }),
+  )
+})