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.
2026-04-26 09:00:27 -04:00
parent 412a1bec44
commit 18d618d051
7 changed files with 454 additions and 129 deletions
@@ -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<FakeDraft, FakeDraft, FakeChunk>({
          .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<FakeDraft, FakeDraft, FakeChunk>({
  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] })
@@ -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<Response>
+
+const handlerLayer = (handler: Handler): Layer.Layer<HttpClient.HttpClient> =>
+  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<HttpClient.HttpClient>) =>
+  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<string>) =>
+  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 })
+    }),
+  )
@@ -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<unknown>
+): 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>): string =>
+  lines.map((line) => `${line}\n\n`).join("")
@@ -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" })
    }),
  )
@@ -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"])
+    }),
+  )
 })
@@ -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<typeof RequestSnapshot>

 const ResponseSnapshot = Schema.Struct({
  status: Schema.Number,
@@ -25,6 +32,7 @@ const Interaction = Schema.Struct({
  request: RequestSnapshot,
  response: ResponseSnapshot,
 })
+type Interaction = Schema.Schema.Type<typeof Interaction>

 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<string> = [
+  "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<string> = ["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<string>
+  /**
+   * 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<string>
+  /**
+   * 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<string, unknown>)
+        .toSorted()
+        .map((key) => [key, canonicalize((value as Record<string, unknown>)[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<string, string>, allow: ReadonlyArray<string>) => {
+  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<string>,
+) => {
+  const merged = lowerHeaders(response.headers as Record<string, string>, 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<HttpClient.HttpClient> =>
+export const layer = (
+  name: string,
+  options: RecordReplayOptions = {},
+): Layer.Layer<HttpClient.HttpClient> =>
  Layer.effect(
    HttpClient.HttpClient,
    Effect.gen(function* () {
@@ -101,22 +177,50 @@ export const layer = (name: string): Layer.Layer<HttpClient.HttpClient> =>
      const fileSystem = yield* FileSystem.FileSystem
      const file = fixturePath(name)
      const dir = path.dirname(file)
-      const recorded: Array<typeof Interaction.Type> = []
+      const requestHeadersAllow = options.requestHeaders ?? DEFAULT_REQUEST_HEADERS
+      const responseHeadersAllow = options.responseHeaders ?? DEFAULT_RESPONSE_HEADERS
+      const match = options.match ?? defaultMatcher
+      const recorded = yield* Ref.make<ReadonlyArray<Interaction>>([])
+
+      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<HttpClient.HttpClient> =>
          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))
        })
@@ -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<A, E, R> = Effect.Effect<A, E, R> | (() => Effect.Effect<A, E, R>)

 type RecordedTestsOptions = {
  readonly prefix: string
  readonly requires?: ReadonlyArray<string>
+  readonly options?: RecordReplayOptions
 }

 type RecordedCaseOptions = {
  readonly cassette?: string
  readonly requires?: ReadonlyArray<string>
+  readonly options?: RecordReplayOptions
 }

-const cassettes = new Set<string>()
-
 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<string>()
+
  const run = <A, E>(
    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 = <A, E>(