From 4e30cf4dbf8b145d5fc02e80e3851e2df5cde49f Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Tue, 7 Jul 2026 11:08:18 -0400 Subject: [PATCH] test(core): add runner scenario DSL --- packages/core/test/lib/runner-scenario.ts | 278 +++++++++++++++++++++ packages/core/test/runner-scenario.test.ts | 266 ++++++++++++++++++++ 2 files changed, 544 insertions(+) create mode 100644 packages/core/test/lib/runner-scenario.ts create mode 100644 packages/core/test/runner-scenario.test.ts diff --git a/packages/core/test/lib/runner-scenario.ts b/packages/core/test/lib/runner-scenario.ts new file mode 100644 index 0000000000..d8f21fdd0d --- /dev/null +++ b/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 + readonly text: (text: string, options?: { readonly id?: string }) => Effect.Effect + readonly toolCall: ( + name: string, + input: unknown, + options?: { readonly id?: string; readonly providerExecuted?: boolean }, + ) => Effect.Effect + readonly events: (...events: ReadonlyArray) => Effect.Effect + readonly stream: (stream: Stream.Stream) => Effect.Effect + readonly fail: (error: LLMError) => Effect.Effect + } +} + +interface State { + readonly started: boolean + readonly accepting: boolean + readonly ended: boolean + readonly nextID: number + readonly calls: ReadonlyMap + readonly requests: ReadonlyArray +} + +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 + readonly finishInteraction: Effect.Effect + readonly end: Effect.Effect +} + +export interface RunnerLLM { + readonly layer: Layer.Layer + readonly next: () => Effect.Effect + readonly requests: Effect.Effect> +} + +const events = (...events: ReadonlyArray) => 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() + const state = yield* Ref.make({ + started: false, + accepting: false, + ended: false, + nextID: 1, + calls: new Map(), + requests: [], + }) + + const respond = ( + id: number, + response: Deferred.Deferred>, + stream: Stream.Stream, + ) => + 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>() + const registered = yield* Ref.modify(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) => 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 => + 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, (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 { + readonly llm: RunnerLLM + readonly run: ( + interaction: () => Effect.gen.Return, + ) => Effect.Effect +} + +export namespace RunnerScenario { + export const make = ( + start: (llm: RunnerLLM) => Effect.Effect, + ): Effect.Effect> => + Effect.gen(function* () { + const internal = yield* makeRunnerLLM + const run = ( + interaction: () => Effect.gen.Return, + ): Effect.Effect => + 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, + } + }) +} diff --git a/packages/core/test/runner-scenario.test.ts b/packages/core/test/runner-scenario.test.ts new file mode 100644 index 0000000000..6be8ea6e68 --- /dev/null +++ b/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) => 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() + 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(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() + const closed = yield* Deferred.make() + 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() + const closed = yield* Deferred.make() + 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() + 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 {}).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" }) + }), + ) +})