From 18d618d051520fb0143904f2e2ffe42ecf0afc8c Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Sun, 26 Apr 2026 09:00:27 -0400 Subject: [PATCH] test(llm): harden cassette matching and add streaming edge-case coverage - Structurally match recorded requests by canonical JSON so non-deterministic field ordering doesn't break replay. - Pluggable header allow-list and body redaction hook on the record/replay layer, so adapters with non-default auth (Anthropic, Bedrock) can plug in without touching this file. - Move the cassette-name dedupe set inside recordedTests() so two describe files using different prefixes can run in parallel. - Replace inline SSE template literals and per-file HTTP layers with shared test/lib helpers (sseEvents, fixedResponse, dynamicResponse, truncatedStream). - Tighten recorded-test assertions to exact text and usage so adapter parser regressions surface immediately instead of passing fuzzy length>0 checks. - Add cancellation and mid-stream transport-error tests for the OpenAI Chat adapter. - Add cross-phase patch tests that verify each phase sees an updated PatchContext and that same-order patches sort deterministically by id. --- packages/llm/test/adapter.test.ts | 124 ++++++++--- packages/llm/test/lib/http.ts | 58 ++++++ packages/llm/test/lib/sse.ts | 20 ++ .../provider/openai-chat.recorded.test.ts | 33 ++- .../llm/test/provider/openai-chat.test.ts | 136 ++++++++----- packages/llm/test/record-replay.ts | 192 ++++++++++++++---- packages/llm/test/recorded-test.ts | 20 +- 7 files changed, 454 insertions(+), 129 deletions(-) create mode 100644 packages/llm/test/lib/http.ts create mode 100644 packages/llm/test/lib/sse.ts diff --git a/packages/llm/test/adapter.test.ts b/packages/llm/test/adapter.test.ts index e9017dc416..539086b6e8 100644 --- a/packages/llm/test/adapter.test.ts +++ b/packages/llm/test/adapter.test.ts @@ -1,11 +1,22 @@ import { describe, expect } from "bun:test" -import { Effect, Layer, Schema, Stream } from "effect" -import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/unstable/http" +import { Effect, Schema, Stream } from "effect" +import { HttpClientRequest } from "effect/unstable/http" import { LLM } from "../src" import { Adapter, client } from "../src/adapter" -import { RequestExecutor } from "../src/executor" import { Patch } from "../src/patch" +import type { LLMRequest } from "../src/schema" import { testEffect } from "./lib/effect" +import { dynamicResponse } from "./lib/http" + +const mapText = (fn: (text: string) => string) => (request: LLMRequest): LLMRequest => ({ + ...request, + messages: request.messages.map((message) => ({ + ...message, + content: message.content.map((part) => + part.type === "text" ? { ...part, text: fn(part.text) } : part, + ), + })), +}) const Json = Schema.fromJsonString(Schema.Unknown) const encodeJson = Schema.encodeSync(Json) @@ -42,8 +53,7 @@ const fake = Adapter.define({ .filter((part) => part.type === "text") .map((part) => part.text), ...request.tools.map((tool) => `tool:${tool.name}:${tool.description}`), - ] - .join("\n"), + ].join("\n"), }), toHttp: (target) => Effect.succeed( @@ -68,20 +78,18 @@ const gemini = Adapter.define({ protocol: "gemini", }) -const httpLayer = Layer.succeed( - HttpClient.HttpClient, - HttpClient.make((request) => - Effect.gen(function* () { - const web = yield* HttpClientRequest.toWeb(request).pipe(Effect.orDie) - return HttpClientResponse.fromWeb( - request, - new Response(encodeJson([{ type: "text", text: `echo:${yield* Effect.promise(() => web.text())}` }, { type: "finish", reason: "stop" }])), - ) - }), +const echoLayer = dynamicResponse(({ text }) => + Effect.succeed( + new Response( + encodeJson([ + { type: "text", text: `echo:${text}` }, + { type: "finish", reason: "stop" }, + ]), + ), ), ) -const it = testEffect(RequestExecutor.layer.pipe(Layer.provide(httpLayer))) +const it = testEffect(echoLayer) describe("llm adapter", () => { it.effect("prepare applies target patches with trace", () => @@ -137,13 +145,7 @@ describe("llm adapter", () => { }), Patch.prompt("test.message", { reason: "rewrite prompt text", - apply: (request) => ({ - ...request, - messages: request.messages.map((message) => ({ - ...message, - content: message.content.map((part) => (part.type === "text" ? { ...part, text: "patched" } : part)), - })), - }), + apply: mapText(() => "patched"), }), Patch.toolSchema("test.description", { reason: "rewrite tool description", @@ -167,6 +169,59 @@ describe("llm adapter", () => { }), ) + it.effect("request patches feed into prompt-patch predicates so phases see updated context", () => + Effect.gen(function* () { + const prepared = yield* client({ + adapters: [fake], + patches: [ + // Earlier phase rewrites the provider, later phase only fires for the + // rewritten provider. If `compile` re-uses a stale PatchContext this + // test fails because the prompt patch's `when` would not match. + Patch.request("rewrite-provider", { + reason: "swap provider before prompt phase", + apply: (request) => ({ + ...request, + model: LLM.model({ ...request.model, provider: "rewritten" }), + }), + }), + Patch.prompt("rewrite-only-when-rewritten", { + reason: "rewrite prompt text only after provider swap", + when: (ctx) => ctx.model.provider === "rewritten", + apply: mapText((text) => `rewrote-${text}`), + }), + ], + }).prepare(request) + + expect(prepared.target).toEqual({ body: "rewrote-hello" }) + expect(prepared.patchTrace.map((item) => item.id)).toEqual([ + "request.rewrite-provider", + "prompt.rewrite-only-when-rewritten", + ]) + }), + ) + + it.effect("patches with the same order sort by id for deterministic application", () => + Effect.gen(function* () { + const prepared = yield* client({ + adapters: [fake], + patches: [ + Patch.prompt("zeta", { + reason: "later id", + order: 1, + apply: mapText((text) => `${text}|zeta`), + }), + Patch.prompt("alpha", { + reason: "earlier id", + order: 1, + apply: mapText((text) => `${text}|alpha`), + }), + ], + }).prepare(request) + + expect(prepared.target).toEqual({ body: "hello|alpha|zeta" }) + }), + ) + it.effect("stream patches transform raised events", () => Effect.gen(function* () { const llm = client({ @@ -185,6 +240,29 @@ describe("llm adapter", () => { }), ) + it.effect("stream patches transform multiple events per stream", () => + Effect.gen(function* () { + // Verifies stream patches run on every event, not just the first. + const seen: string[] = [] + const llm = client({ + adapters: [fake], + patches: [ + Patch.stream("test.tap", { + reason: "record every event type", + apply: (event) => { + seen.push(event.type) + return event + }, + }), + ], + }) + + yield* llm.stream(request).pipe(Stream.runDrain) + + expect(seen).toEqual(["text-delta", "request-finish"]) + }), + ) + it.effect("rejects protocol mismatch", () => Effect.gen(function* () { const error = yield* client({ adapters: [fake] }) diff --git a/packages/llm/test/lib/http.ts b/packages/llm/test/lib/http.ts new file mode 100644 index 0000000000..6ae8bb00ad --- /dev/null +++ b/packages/llm/test/lib/http.ts @@ -0,0 +1,58 @@ +import { Effect, Layer } from "effect" +import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/unstable/http" +import { RequestExecutor } from "../../src/executor" + +export type HandlerInput = { + readonly request: HttpClientRequest.HttpClientRequest + readonly text: string +} + +export type Handler = (input: HandlerInput) => Effect.Effect + +const handlerLayer = (handler: Handler): Layer.Layer => + Layer.succeed( + HttpClient.HttpClient, + HttpClient.make((request) => + Effect.gen(function* () { + const web = yield* HttpClientRequest.toWeb(request).pipe(Effect.orDie) + const text = yield* Effect.promise(() => web.text()) + const response = yield* handler({ request, text }) + return HttpClientResponse.fromWeb(request, response) + }), + ), + ) + +const executorWith = (layer: Layer.Layer) => + RequestExecutor.layer.pipe(Layer.provide(layer)) + +const SSE_HEADERS = { "content-type": "text/event-stream" } as const + +/** + * Layer that returns a single fixed response body. Use for stream-parser + * fixture tests where the request shape is irrelevant. + */ +export const fixedResponse = (body: string, init: ResponseInit = { headers: SSE_HEADERS }) => + executorWith(handlerLayer(() => Effect.succeed(new Response(body, init)))) + +/** + * Layer that builds a response per request. Useful for echo servers. + */ +export const dynamicResponse = (handler: Handler) => executorWith(handlerLayer(handler)) + +/** + * Layer that emits the supplied SSE chunks and then aborts mid-stream. Used to + * exercise transport errors that surface during parsing. + */ +export const truncatedStream = (chunks: ReadonlyArray) => + dynamicResponse(() => + Effect.sync(() => { + const encoder = new TextEncoder() + const stream = new ReadableStream({ + start(controller) { + for (const chunk of chunks) controller.enqueue(encoder.encode(chunk)) + controller.error(new Error("connection reset")) + }, + }) + return new Response(stream, { headers: SSE_HEADERS }) + }), + ) diff --git a/packages/llm/test/lib/sse.ts b/packages/llm/test/lib/sse.ts new file mode 100644 index 0000000000..3e72df0f10 --- /dev/null +++ b/packages/llm/test/lib/sse.ts @@ -0,0 +1,20 @@ +/** + * Helpers for building deterministic SSE bodies in tests. + * + * Inline template-literal SSE strings are hard to write and review when chunks + * contain JSON; this helper accepts plain values and serializes them, so test + * authors only think about the chunk shapes, not the wire format. + */ +export const sseEvents = ( + ...chunks: ReadonlyArray +): string => `${chunks.map(formatChunk).join("")}data: [DONE]\n\n` + +const formatChunk = (chunk: unknown) => + `data: ${typeof chunk === "string" ? chunk : JSON.stringify(chunk)}\n\n` + +/** + * Build an SSE body from already-serialized strings (used when the chunk shape + * itself is part of what's being tested, e.g. malformed chunks). + */ +export const sseRaw = (...lines: ReadonlyArray): string => + lines.map((line) => `${line}\n\n`).join("") diff --git a/packages/llm/test/provider/openai-chat.recorded.test.ts b/packages/llm/test/provider/openai-chat.recorded.test.ts index 704ffb0796..cf3807778d 100644 --- a/packages/llm/test/provider/openai-chat.recorded.test.ts +++ b/packages/llm/test/provider/openai-chat.recorded.test.ts @@ -54,6 +54,9 @@ const toolResultRequest = LLM.request({ generation: { maxTokens: 40, temperature: 0 }, }) +// Cassettes are deterministic — assert exact stream contents instead of fuzzy +// `length > 0` checks so adapter parsing regressions surface immediately. +// Re-record (`RECORD=true`) only when intentionally refreshing a cassette. const recorded = recordedTests({ prefix: "openai-chat", requires: ["OPENAI_API_KEY"] }) const openai = client({ adapters: [OpenAIChat.adapter] }) const openaiWithUsage = client({ adapters: [OpenAIChat.adapter.withPatches([OpenAIChat.includeUsage])] }) @@ -62,21 +65,34 @@ describe("OpenAI Chat recorded", () => { recorded.effect("streams text", () => Effect.gen(function* () { const response = yield* openaiWithUsage.generate(request) - const text = LLM.outputText(response) - expect(text.length).toBeGreaterThan(0) - expect(response.usage?.totalTokens).toBeGreaterThan(0) - expect(response.events.at(-1)?.type).toBe("request-finish") + expect(LLM.outputText(response)).toBe("Hello!") + expect(response.usage).toMatchObject({ + inputTokens: 22, + outputTokens: 2, + totalTokens: 24, + cacheReadInputTokens: 0, + reasoningTokens: 0, + }) + expect(response.events.map((event) => event.type)).toEqual([ + "text-delta", + "text-delta", + "request-finish", + ]) + expect(response.events.at(-1)).toMatchObject({ type: "request-finish", reason: "stop" }) }), ) recorded.effect("streams tool call", () => Effect.gen(function* () { const response = yield* openai.generate(toolRequest) - const toolCall = response.events.find((event) => event.type === "tool-call") expect(response.events.some((event) => event.type === "tool-input-delta")).toBe(true) - expect(toolCall).toMatchObject({ type: "tool-call", name: "get_weather", input: { city: "Paris" } }) + expect(response.events.find((event) => event.type === "tool-call")).toMatchObject({ + type: "tool-call", + name: "get_weather", + input: { city: "Paris" }, + }) expect(response.events.at(-1)).toMatchObject({ type: "request-finish", reason: "tool-calls" }) }), ) @@ -84,10 +100,9 @@ describe("OpenAI Chat recorded", () => { recorded.effect("continues after tool result", () => Effect.gen(function* () { const response = yield* openaiWithUsage.generate(toolResultRequest) - const text = LLM.outputText(response) - expect(text.toLowerCase()).toContain("sunny") - expect(response.usage?.totalTokens).toBeGreaterThan(0) + expect(LLM.outputText(response)).toBe("The weather in Paris is sunny with a temperature of 22°C.") + expect(response.usage).toMatchObject({ inputTokens: 59, outputTokens: 14, totalTokens: 73 }) expect(response.events.at(-1)).toMatchObject({ type: "request-finish", reason: "stop" }) }), ) diff --git a/packages/llm/test/provider/openai-chat.test.ts b/packages/llm/test/provider/openai-chat.test.ts index bfec00c217..d3a1bff344 100644 --- a/packages/llm/test/provider/openai-chat.test.ts +++ b/packages/llm/test/provider/openai-chat.test.ts @@ -1,11 +1,11 @@ import { describe, expect } from "bun:test" -import { Effect, Layer, Schema } from "effect" -import { HttpClient, HttpClientResponse } from "effect/unstable/http" +import { Effect, Layer, Schema, Stream } from "effect" import { LLM } from "../../src" import { client } from "../../src/adapter" -import { RequestExecutor } from "../../src/executor" import { OpenAIChat } from "../../src/provider/openai-chat" import { testEffect } from "../lib/effect" +import { fixedResponse, truncatedStream } from "../lib/http" +import { sseEvents } from "../lib/sse" const TargetJson = Schema.fromJsonString(Schema.Unknown) const encodeJson = Schema.encodeSync(TargetJson) @@ -26,27 +26,24 @@ const request = LLM.request({ const it = testEffect(Layer.empty) -const streamLayer = (body: string) => - RequestExecutor.layer.pipe( - Layer.provide( - Layer.succeed( - HttpClient.HttpClient, - HttpClient.make((request) => - Effect.succeed( - HttpClientResponse.fromWeb( - request, - new Response(body, { headers: { "content-type": "text/event-stream" } }), - ), - ), - ), - ), - ), - ) +const deltaChunk = (delta: object, finishReason: string | null = null) => ({ + id: "chatcmpl_fixture", + choices: [{ delta, finish_reason: finishReason }], + usage: null, +}) + +const usageChunk = (usage: object) => ({ + id: "chatcmpl_fixture", + choices: [], + usage, +}) describe("OpenAI Chat adapter", () => { it.effect("prepares OpenAI Chat target", () => Effect.gen(function* () { - const prepared = yield* client({ adapters: [OpenAIChat.adapter.withPatches([OpenAIChat.includeUsage])] }).prepare(request) + const prepared = yield* client({ + adapters: [OpenAIChat.adapter.withPatches([OpenAIChat.includeUsage])], + }).prepare(request) expect(prepared.target).toEqual({ model: "gpt-4o-mini", @@ -133,18 +130,23 @@ describe("OpenAI Chat adapter", () => { it.effect("parses text and usage stream fixtures", () => Effect.gen(function* () { - const body = `data: {"id":"chatcmpl_fixture","choices":[{"delta":{"role":"assistant","content":"Hello"},"finish_reason":null}],"usage":null} - -data: {"id":"chatcmpl_fixture","choices":[{"delta":{"content":"!"},"finish_reason":null}],"usage":null} - -data: {"id":"chatcmpl_fixture","choices":[{"delta":{},"finish_reason":"stop"}],"usage":null} - -data: {"id":"chatcmpl_fixture","choices":[],"usage":{"prompt_tokens":5,"completion_tokens":2,"total_tokens":7,"prompt_tokens_details":{"cached_tokens":1},"completion_tokens_details":{"reasoning_tokens":0}}} - -data: [DONE] -` - const response = yield* client({ adapters: [OpenAIChat.adapter] }).generate(request).pipe(Effect.provide(streamLayer(body))) + const body = sseEvents( + deltaChunk({ role: "assistant", content: "Hello" }), + deltaChunk({ content: "!" }), + deltaChunk({}, "stop"), + usageChunk({ + prompt_tokens: 5, + completion_tokens: 2, + total_tokens: 7, + prompt_tokens_details: { cached_tokens: 1 }, + completion_tokens_details: { reasoning_tokens: 0 }, + }), + ) + const response = yield* client({ adapters: [OpenAIChat.adapter] }) + .generate(request) + .pipe(Effect.provide(fixedResponse(body))) + expect(LLM.outputText(response)).toBe("Hello!") expect(response.events).toEqual([ { type: "text-delta", text: "Hello" }, { type: "text-delta", text: "!" }, @@ -167,26 +169,29 @@ data: [DONE] }, }, ]) - expect(response.usage?.totalTokens).toBe(7) }), ) it.effect("assembles streamed tool call input", () => Effect.gen(function* () { - const body = `data: {"id":"chatcmpl_fixture","choices":[{"delta":{"role":"assistant","tool_calls":[{"index":0,"id":"call_1","function":{"name":"lookup","arguments":"{\\"query\\""}}]},"finish_reason":null}],"usage":null} - -data: {"id":"chatcmpl_fixture","choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":":\\"weather\\"}"}}]},"finish_reason":null}],"usage":null} - -data: {"id":"chatcmpl_fixture","choices":[{"delta":{},"finish_reason":"tool_calls"}],"usage":null} - -data: [DONE] -` - const response = yield* client({ adapters: [OpenAIChat.adapter] }).generate( - LLM.request({ - ...request, - tools: [{ name: "lookup", description: "Lookup data", inputSchema: { type: "object" } }], + const body = sseEvents( + deltaChunk({ + role: "assistant", + tool_calls: [ + { index: 0, id: "call_1", function: { name: "lookup", arguments: '{"query"' } }, + ], }), - ).pipe(Effect.provide(streamLayer(body))) + deltaChunk({ tool_calls: [{ index: 0, function: { arguments: ':"weather"}' } }] }), + deltaChunk({}, "tool_calls"), + ) + const response = yield* client({ adapters: [OpenAIChat.adapter] }) + .generate( + LLM.request({ + ...request, + tools: [{ name: "lookup", description: "Lookup data", inputSchema: { type: "object" } }], + }), + ) + .pipe(Effect.provide(fixedResponse(body))) expect(response.events).toEqual([ { type: "tool-input-delta", id: "call_1", name: "lookup", text: '{"query"' }, @@ -199,15 +204,46 @@ data: [DONE] it.effect("fails on malformed stream chunks", () => Effect.gen(function* () { - const body = `data: {"id":"chatcmpl_fixture","choices":[{"delta":{"content":123},"finish_reason":null}],"usage":null} - -data: [DONE] -` + const body = sseEvents(deltaChunk({ content: 123 })) const error = yield* client({ adapters: [OpenAIChat.adapter] }) .generate(request) - .pipe(Effect.provide(streamLayer(body)), Effect.flip) + .pipe(Effect.provide(fixedResponse(body)), Effect.flip) expect(error.message).toContain("Invalid OpenAI Chat stream chunk") }), ) + + it.effect("surfaces transport errors that occur mid-stream", () => + Effect.gen(function* () { + const layer = truncatedStream([ + `data: ${JSON.stringify(deltaChunk({ role: "assistant", content: "Hello" }))}\n\n`, + ]) + const error = yield* client({ adapters: [OpenAIChat.adapter] }) + .generate(request) + .pipe(Effect.provide(layer), Effect.flip) + + expect(error.message).toContain("Failed to read OpenAI Chat stream") + }), + ) + + it.effect("short-circuits the upstream stream when the consumer takes a prefix", () => + Effect.gen(function* () { + const llm = client({ adapters: [OpenAIChat.adapter] }) + // The body has more chunks than we'll consume. If `Stream.take(1)` did + // not interrupt the upstream HTTP body the test would hang waiting for + // the rest of the stream to drain. + const body = sseEvents( + deltaChunk({ role: "assistant", content: "Hello" }), + deltaChunk({ content: " world" }), + deltaChunk({}, "stop"), + ) + + const events = Array.from( + yield* llm + .stream(request) + .pipe(Stream.take(1), Stream.runCollect, Effect.provide(fixedResponse(body))), + ) + expect(events.map((event) => event.type)).toEqual(["text-delta"]) + }), + ) }) diff --git a/packages/llm/test/record-replay.ts b/packages/llm/test/record-replay.ts index b6337b3b1b..2697f5b5d7 100644 --- a/packages/llm/test/record-replay.ts +++ b/packages/llm/test/record-replay.ts @@ -1,6 +1,12 @@ import { NodeFileSystem } from "@effect/platform-node" -import { Effect, FileSystem, Layer, Schema } from "effect" -import { FetchHttpClient, HttpClient, HttpClientError, HttpClientRequest, HttpClientResponse } from "effect/unstable/http" +import { Effect, FileSystem, Layer, Option, Ref, Schema } from "effect" +import { + FetchHttpClient, + HttpClient, + HttpClientError, + HttpClientRequest, + HttpClientResponse, +} from "effect/unstable/http" import * as fs from "node:fs" import * as path from "node:path" import { fileURLToPath } from "node:url" @@ -14,6 +20,7 @@ const RequestSnapshot = Schema.Struct({ headers: Schema.Record(Schema.String, Schema.String), body: Schema.String, }) +type RequestSnapshot = Schema.Schema.Type const ResponseSnapshot = Schema.Struct({ status: Schema.Number, @@ -25,6 +32,7 @@ const Interaction = Schema.Struct({ request: RequestSnapshot, response: ResponseSnapshot, }) +type Interaction = Schema.Schema.Type const Cassette = Schema.Struct({ version: Schema.Literal(1), @@ -32,31 +40,106 @@ const Cassette = Schema.Struct({ }) const CassetteJson = Schema.fromJsonString(Cassette) -const RequestJson = Schema.fromJsonString(RequestSnapshot) - -const decodeCassette = Schema.decodeUnknownSync(Cassette) const decodeCassetteJson = Schema.decodeUnknownSync(CassetteJson) const encodeCassetteJson = Schema.encodeSync(CassetteJson) -const encodeRequestJson = Schema.encodeSync(RequestJson) + +const JsonValue = Schema.fromJsonString(Schema.Unknown) +const decodeJson = Schema.decodeUnknownOption(JsonValue) const isRecordMode = process.env.RECORD === "true" const fixturePath = (name: string) => path.join(FIXTURES_DIR, `${name}.json`) -const requestHeaders = (headers: Headers) => - Object.fromEntries( - [...headers.entries()].filter(([name]) => ["content-type", "accept", "openai-beta"].includes(name.toLowerCase())), - ) +/** + * Default request header allow-list. Provider adapters with custom auth + * (Anthropic `x-api-key`, Bedrock SigV4, etc.) should extend this via the + * `requestHeaders` option so cassette matching uses the right keys. + */ +export const DEFAULT_REQUEST_HEADERS: ReadonlyArray = [ + "content-type", + "accept", + "openai-beta", +] -const requestSnapshot = Effect.fnUntraced(function* (request: HttpClientRequest.HttpClientRequest) { - const web = yield* HttpClientRequest.toWeb(request).pipe(Effect.orDie) - return { - method: web.method, - url: web.url, - headers: requestHeaders(web.headers), - body: yield* Effect.promise(() => web.text()), +const DEFAULT_RESPONSE_HEADERS: ReadonlyArray = ["content-type"] + +export interface RecordReplayOptions { + /** + * Lower-cased request header names that participate in cassette matching and + * are persisted to disk. Anything not in this list is dropped. + */ + readonly requestHeaders?: ReadonlyArray + /** + * Lower-cased response header names persisted to disk. Defaults to + * `content-type` only. Add `x-request-id`, rate-limit headers, etc. when a + * test depends on them. + */ + readonly responseHeaders?: ReadonlyArray + /** + * Hook to redact secrets from request bodies before they are written. Runs + * on the parsed JSON value when the body decodes as JSON; non-JSON bodies + * pass through untouched. + */ + readonly redactBody?: (body: unknown) => unknown + /** + * Custom request matcher. Defaults to `defaultMatcher`, which compares + * method, url, structurally-canonical JSON body, and the allow-listed + * headers. + */ + readonly match?: RequestMatcher +} + +export type RequestMatcher = (incoming: RequestSnapshot, recorded: RequestSnapshot) => boolean + +/** + * Sort object keys recursively so two semantically equal JSON values produce + * the same string. Arrays preserve order — provider request bodies care about + * `messages` ordering. + */ +const canonicalize = (value: unknown): unknown => { + if (Array.isArray(value)) return value.map(canonicalize) + if (value !== null && typeof value === "object") { + return Object.fromEntries( + Object.keys(value as Record) + .toSorted() + .map((key) => [key, canonicalize((value as Record)[key])]), + ) } -}) + return value +} + +const canonicalSnapshot = (snapshot: RequestSnapshot): string => + JSON.stringify({ + method: snapshot.method, + url: snapshot.url, + headers: canonicalize(snapshot.headers), + body: Option.match(decodeJson(snapshot.body), { + onNone: () => snapshot.body, + onSome: canonicalize, + }), + }) + +export const defaultMatcher: RequestMatcher = (incoming, recorded) => + canonicalSnapshot(incoming) === canonicalSnapshot(recorded) + +const lowerHeaders = (headers: Record, allow: ReadonlyArray) => { + const allowed = new Set(allow.map((name) => name.toLowerCase())) + return Object.fromEntries( + Object.entries(headers) + .map(([name, value]) => [name.toLowerCase(), value] as const) + .filter(([name]) => allowed.has(name)) + .toSorted(([a], [b]) => a.localeCompare(b)), + ) +} + +const responseHeaders = ( + response: HttpClientResponse.HttpClientResponse, + allow: ReadonlyArray, +) => { + const merged = lowerHeaders(response.headers as Record, allow) + if (!merged["content-type"]) merged["content-type"] = "text/event-stream" + return merged +} const fixtureMissing = (request: HttpClientRequest.HttpClientRequest, name: string) => new HttpClientError.HttpClientError({ @@ -74,16 +157,6 @@ const fixtureMismatch = (request: HttpClientRequest.HttpClientRequest, name: str }), }) -const responseSnapshot = (response: HttpClientResponse.HttpClientResponse, body: string) => ({ - status: response.status, - headers: headers(response), - body, -}) - -const headers = (response: HttpClientResponse.HttpClientResponse) => ({ - "content-type": response.headers["content-type"] ?? "text/event-stream", -}) - export const hasFixtureSync = (name: string) => { try { decodeCassetteJson(fs.readFileSync(fixturePath(name), "utf8")) @@ -93,7 +166,10 @@ export const hasFixtureSync = (name: string) => { } } -export const layer = (name: string): Layer.Layer => +export const layer = ( + name: string, + options: RecordReplayOptions = {}, +): Layer.Layer => Layer.effect( HttpClient.HttpClient, Effect.gen(function* () { @@ -101,22 +177,50 @@ export const layer = (name: string): Layer.Layer => const fileSystem = yield* FileSystem.FileSystem const file = fixturePath(name) const dir = path.dirname(file) - const recorded: Array = [] + const requestHeadersAllow = options.requestHeaders ?? DEFAULT_REQUEST_HEADERS + const responseHeadersAllow = options.responseHeaders ?? DEFAULT_RESPONSE_HEADERS + const match = options.match ?? defaultMatcher + const recorded = yield* Ref.make>([]) + + const snapshotRequest = (request: HttpClientRequest.HttpClientRequest) => + Effect.gen(function* () { + const web = yield* HttpClientRequest.toWeb(request).pipe(Effect.orDie) + const raw = yield* Effect.promise(() => web.text()) + const redact = options.redactBody + const body = redact + ? Option.match(decodeJson(raw), { + onNone: () => raw, + onSome: (parsed) => JSON.stringify(redact(parsed)), + }) + : raw + return { + method: web.method, + url: web.url, + headers: lowerHeaders(Object.fromEntries(web.headers.entries()), requestHeadersAllow), + body, + } + }) return HttpClient.make((request) => { if (isRecordMode) { return Effect.gen(function* () { - const currentRequest = yield* requestSnapshot(request) + const currentRequest = yield* snapshotRequest(request) const response = yield* upstream.execute(request) const body = yield* response.text - const interaction = decodeCassette({ - version: 1, - interactions: [...recorded, { request: currentRequest, response: responseSnapshot(response, body) }], - }) - recorded.splice(0, recorded.length, ...interaction.interactions) + const interaction: Interaction = { + request: currentRequest, + response: { + status: response.status, + headers: responseHeaders(response, responseHeadersAllow), + body, + }, + } + const interactions = yield* Ref.updateAndGet(recorded, (prev) => [...prev, interaction]) yield* fileSystem.makeDirectory(dir, { recursive: true }).pipe(Effect.orDie) - yield* fileSystem.writeFileString(file, encodeCassetteJson(interaction)).pipe(Effect.orDie) - return HttpClientResponse.fromWeb(request, new Response(body, responseSnapshot(response, body))) + yield* fileSystem + .writeFileString(file, encodeCassetteJson({ version: 1, interactions })) + .pipe(Effect.orDie) + return HttpClientResponse.fromWeb(request, new Response(body, interaction.response)) }) } @@ -124,11 +228,13 @@ export const layer = (name: string): Layer.Layer => const cassette = decodeCassetteJson( yield* fileSystem.readFileString(file).pipe(Effect.mapError(() => fixtureMissing(request, name))), ) - const currentRequest = encodeRequestJson(yield* requestSnapshot(request)) - const interaction = cassette.interactions.find((interaction) => encodeRequestJson(interaction.request) === currentRequest) - if (!interaction) { - return yield* fixtureMismatch(request, name) - } + const incoming = yield* snapshotRequest(request) + const incomingCanonical = canonicalSnapshot(incoming) + const interaction = + match === defaultMatcher + ? cassette.interactions.find((candidate) => canonicalSnapshot(candidate.request) === incomingCanonical) + : cassette.interactions.find((candidate) => match(incoming, candidate.request)) + if (!interaction) return yield* fixtureMismatch(request, name) return HttpClientResponse.fromWeb(request, new Response(interaction.response.body, interaction.response)) }) diff --git a/packages/llm/test/recorded-test.ts b/packages/llm/test/recorded-test.ts index 7e72ac5627..78921f6ebe 100644 --- a/packages/llm/test/recorded-test.ts +++ b/packages/llm/test/recorded-test.ts @@ -2,22 +2,26 @@ import { test, type TestOptions } from "bun:test" import { Effect, Layer } from "effect" import { RequestExecutor } from "../src/executor" import { testEffect } from "./lib/effect" -import { hasFixtureSync, layer as recordReplayLayer } from "./record-replay" +import { + hasFixtureSync, + layer as recordReplayLayer, + type RecordReplayOptions, +} from "./record-replay" type Body = Effect.Effect | (() => Effect.Effect) type RecordedTestsOptions = { readonly prefix: string readonly requires?: ReadonlyArray + readonly options?: RecordReplayOptions } type RecordedCaseOptions = { readonly cassette?: string readonly requires?: ReadonlyArray + readonly options?: RecordReplayOptions } -const cassettes = new Set() - const kebab = (value: string) => value .trim() @@ -32,6 +36,11 @@ const cassetteName = (prefix: string, name: string, options: RecordedCaseOptions options.cassette ?? `${prefix}/${kebab(name)}` export const recordedTests = (options: RecordedTestsOptions) => { + // Scoped to this `recordedTests` group rather than module-global so two + // describe files using different prefixes don't collide and parallelization + // at the file level stays safe. + const cassettes = new Set() + const run = ( name: string, caseOptions: RecordedCaseOptions, @@ -50,7 +59,10 @@ export const recordedTests = (options: RecordedTestsOptions) => { return test.skip(name, () => {}, testOptions) } - return testEffect(RequestExecutor.layer.pipe(Layer.provide(recordReplayLayer(cassette)))).live(name, body, testOptions) + const layerOptions = caseOptions.options ?? options.options + return testEffect( + RequestExecutor.layer.pipe(Layer.provide(recordReplayLayer(cassette, layerOptions))), + ).live(name, body, testOptions) } const effect = (