feat(e2e): canonical content-based match keys for record-and-replay

Replay previously matched interactions by transport verb and path in
recorded order, so a request whose body drifted from the recording
silently replayed the stale response, and reordering two independent
calls broke replay even though both were recorded. Match keys are now
canonical: fixture_canonical.py strips volatile headers and credential
fields, replaces unique markers, generated ids, uuids, and timestamps
with fixed placeholders, sorts object keys, and hashes what remains, so
a key is stable across runs and machines while any real content drift
is a hard ReplayMiss naming the computed key, the closest recorded key
with its file, and a content diff, with no fallthrough to a live call.
Matching is order-independent across distinct keys and FIFO within one
key. Recording now also redacts credential body and form fields (not
just auth headers) so provider keys never land in bundles.

Resolves LIT-5741

tests/e2e/CLAUDE.md

@@ -75,11 +75,11 @@ Mark live tests with `@pytest.mark.e2e` (on the class or the module). Pure cover
 
 `E2E_FIXTURE_MODE` selects the transport every client is built on: `live` (the default, and what an unset variable means: nothing changes), `record` (run against the live proxy and write every interaction to a fixture bundle), or `replay` (serve every interaction back from the bundle with no HTTP at all, so a replay run needs no proxy and cannot bill a provider). The seam is `select_transport` in `fixture_transport.py`, applied inside `build_proxy_client`; both transports fulfil the same `Transport` protocol, so no test or client changes shape in any mode
 
-A bundle (default `tests/e2e/.fixtures`, override with `E2E_FIXTURE_DIR`) is a directory: `manifest.json` carries the record timestamp, harness git version, and format version, and each test gets a subdirectory holding one JSON file per transport call in call order (`0000-post-chat-completions.json`). Auth header values are redacted on write, and file uploads store a sha256 digest instead of the bytes; response bodies are stored verbatim (a /key/generate response keeps the ephemeral virtual key it minted), which is part of why bundles are gitignored. `fixture_bundle.py` owns the format
+A bundle (default `tests/e2e/.fixtures`, override with `E2E_FIXTURE_DIR`) is a directory: `manifest.json` carries the record timestamp, harness git version, and format version, and each test gets a subdirectory holding one JSON file per transport call in call order (`0000-post-chat-completions.json`). Auth header values and credential request fields (`api_key`, `*_secret_key`, `static_headers`, and the like; the list is `fixture_canonical.py`'s) are redacted on write, and file uploads store a sha256 digest instead of the bytes; response bodies are stored verbatim (a /key/generate response keeps the ephemeral virtual key it minted), which is part of why bundles are gitignored. `fixture_bundle.py` owns the format
 
-Replay matches calls per test by transport verb and path in recorded order and raises `ReplayMiss` on any drift, naming the recorded and the actual call; a passed test must also consume its whole recording, or teardown fails it naming the first leftover interaction. Either way the fix is always to re-record with `E2E_FIXTURE_MODE=record`. Record starts fresh every time: it wipes the previous bundle (refusing to wipe a directory that is not a bundle) and never reads it. A replay bundle whose manifest is older than seven days hard-fails at collection time naming the bundle's age, so replay can never certify against fixtures that have drifted more than a week from the live proxy
+Replay matches calls per test by canonical key: `fixture_canonical.py` canonicalizes the recorded request (volatile headers and credential fields out, unique markers, generated ids, uuids, and timestamps replaced with fixed placeholders, object keys sorted) and the key is the method, path, and a content hash, so identity survives re-records and machine changes while any real content drift is a `ReplayMiss` that names the computed key, the closest recorded key with its file, and a content diff, and never falls through to a live call. Matching is order-independent across distinct keys (concurrent calls may interleave) and FIFO within one key (a poll loop replays its responses in recorded order); a passed test must also consume its whole recording, or teardown fails it naming a leftover key. Either way the fix is always to re-record with `E2E_FIXTURE_MODE=record`. Every rewrite rule lives in `fixture_canonical.py`, so a new volatile header, credential field name, or generated-id shape is one edit there. Record starts fresh every time: it wipes the previous bundle (refusing to wipe a directory that is not a bundle) and never reads it. A replay bundle whose manifest is older than seven days hard-fails at collection time naming the bundle's age, so replay can never certify against fixtures that have drifted more than a week from the live proxy
 
-Deliberately not here yet: canonical content-based match keys (LIT-5741), streaming chunk fidelity (LIT-5742), and scoping record/replay to provider-bound traffic (LIT-5745)
+Deliberately not here yet: streaming chunk fidelity (LIT-5742) and scoping record/replay to provider-bound traffic (LIT-5745)
 
 ## Typing
 

tests/e2e/fixture_bundle.py

@@ -8,10 +8,10 @@ green replay run can never certify against fixtures that have drifted more than
 a week from the live proxy.
 
 This module owns the format only. The transports that produce and consume it
-live in fixture_transport.py; canonical request matching, streaming chunk
-fidelity, and provider-scoping are follow-ups (LIT-5741/5742/5745) and are
-deliberately absent here, which is why every interaction file stores the full
-redacted request even though replay today matches by call order.
+live in fixture_transport.py and the canonical match keys they compute live in
+fixture_canonical.py (LIT-5741); streaming chunk fidelity and provider-scoping
+are follow-ups (LIT-5742/5745). Every interaction file stores the full redacted
+request because replay matches on its canonicalized content.
 """
 
 from __future__ import annotations
@@ -54,12 +54,13 @@ class Manifest(BaseModel):
 
 
 class RecordedRequest(BaseModel):
-    """The request as the transport saw it, auth header values redacted.
+    """The request as the transport saw it, auth header values and credential
+    body/form fields redacted.
 
-    Replay today only matches ``method`` (the transport verb, not the HTTP verb)
-    and ``path`` in call order; the rest is stored so LIT-5741 can move to
-    content-based match keys without re-recording. File uploads store a content
-    digest instead of the bytes."""
+    Replay matches on the canonical content key fixture_canonical.py computes
+    over ``method`` (the transport verb, not the HTTP verb), ``path``, and the
+    canonicalized headers, params, body, form, and file identity. File uploads
+    store a content digest instead of the bytes."""
 
     method: str
     path: str

tests/e2e/fixture_canonical.py

@@ -0,0 +1,150 @@
+"""Canonical request identity for replay matching (LIT-5741).
+
+Matching a replayed call against the raw recorded request never hits: unique
+markers salt prompts, model names, and tags; every run mints fresh virtual
+keys; request ids and timestamps differ on every call. Matching on transport
+verb + path alone collides: two different requests to the same route silently
+swap responses, which passes when it should miss. The canonicalizer strips
+exactly the volatile material (volatile headers, credential fields, markers,
+generated ids, timestamps) and hashes what remains with sorted object keys, so
+identity is content-based and stable across runs and machines.
+
+Every rewrite rule lives in this module, next to the transports that apply it:
+a new volatile header, credential field name, or generated-id shape is one
+edit here, never a per-suite change.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+from dataclasses import dataclass
+from functools import reduce
+from typing import Final
+
+from pydantic import JsonValue
+
+from fixture_bundle import RecordedRequest
+
+VOLATILE_HEADER_NAMES: Final[frozenset[str]] = frozenset(
+    {
+        "authorization",
+        "x-litellm-api-key",
+        "x-api-key",
+        "x-goog-api-key",
+        "x-request-id",
+        "traceparent",
+        "tracestate",
+    }
+)
+
+SECRET_FIELD_NAMES: Final[frozenset[str]] = frozenset(
+    {"api_key", "aws_access_key_id", "static_headers", "vertex_credentials"}
+)
+SECRET_FIELD_SUFFIXES: Final[tuple[str, ...]] = (
+    "_api_key",
+    "_secret_key",
+    "_secret_access_key",
+    "_session_token",
+    "_credentials",
+    "_password",
+)
+SECRET_PLACEHOLDER: Final = "<secret>"
+
+PLACEHOLDER_RULES: Final[tuple[tuple[re.Pattern[str], str], ...]] = (
+    (re.compile(r"(?<![0-9a-fA-F])[0-9a-f]{64}(?![0-9a-fA-F])"), "<sha256>"),
+    (
+        re.compile(r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}"),
+        "<uuid>",
+    ),
+    (re.compile(r"sk-[A-Za-z0-9_-]{16,}"), "<key>"),
+    (
+        re.compile(r"\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:Z|[+-]\d{2}:?\d{2})?"),
+        "<timestamp>",
+    ),
+    (re.compile(r"(?<!\d)\d{4}-\d{2}-\d{2}(?!\d)"), "<date>"),
+    (
+        re.compile(r"\b(?:chatcmpl|msgbatch|msg|resp|batch|call|req|ftjob|gen|file)[-_][A-Za-z0-9]{8,}\b"),
+        "<id>",
+    ),
+    (re.compile(r"(?<![0-9a-fA-F])[0-9a-f]{12}(?![0-9a-fA-F])"), "<marker>"),
+)
+
+
+def is_secret_field(name: str) -> bool:
+    lowered: Final = name.lower()
+    return lowered in SECRET_FIELD_NAMES or lowered.endswith(SECRET_FIELD_SUFFIXES)
+
+
+def canonical_string(value: str) -> str:
+    return reduce(lambda acc, rule: rule[0].sub(rule[1], acc), PLACEHOLDER_RULES, value)
+
+
+def _canonical_flat(fields: dict[str, str]) -> dict[str, JsonValue]:
+    return {
+        key: SECRET_PLACEHOLDER if is_secret_field(key) else canonical_string(value)
+        for key, value in fields.items()
+    }
+
+
+def _canonical_value(value: JsonValue) -> JsonValue:
+    match value:
+        case str():
+            return canonical_string(value)
+        case dict():
+            return {
+                key: SECRET_PLACEHOLDER
+                if is_secret_field(key) and item is not None
+                else _canonical_value(item)
+                for key, item in value.items()
+            }
+        case list():
+            return [_canonical_value(item) for item in value]
+        case _:
+            return value
+
+
+@dataclass(frozen=True, slots=True)
+class CanonicalRequest:
+    method: str
+    path: str
+    content: str
+
+    @property
+    def key(self) -> str:
+        digest: Final = hashlib.sha256(
+            f"{self.method} {self.path}\n{self.content}".encode()
+        ).hexdigest()[:16]
+        return f"{self.method} {self.path} #{digest}"
+
+    def pretty_content(self) -> str:
+        return json.dumps(json.loads(self.content), indent=2, sort_keys=True)
+
+
+def canonicalize(request: RecordedRequest) -> CanonicalRequest:
+    file_identity: Final[JsonValue | None] = (
+        None
+        if request.file_name is None and request.file_sha256 is None
+        else {
+            "name": None if request.file_name is None else canonical_string(request.file_name),
+            "sha256": request.file_sha256,
+            "bytes": request.file_bytes,
+        }
+    )
+    content: Final[dict[str, JsonValue]] = {
+        "headers": {
+            name.lower(): canonical_string(value)
+            for name, value in request.headers.items()
+            if name.lower() not in VOLATILE_HEADER_NAMES
+        },
+        "params": _canonical_flat(request.params),
+        "body": _canonical_value(request.body),
+        "form": None if request.form is None else _canonical_flat(request.form),
+        "file": file_identity,
+    }
+    return CanonicalRequest(
+        method=request.method,
+        path=canonical_string(request.path),
+        content=json.dumps(content, sort_keys=True, separators=(",", ":")),
+    )

tests/e2e/fixture_transport.py

@@ -7,23 +7,30 @@ HTTP, no proxy, no provider spend. Because both fulfil ``Transport``, no test
 or client changes shape; ``build_proxy_client`` picks the transport from
 ``E2E_FIXTURE_MODE`` (live | record | replay, default live).
 
-Replay matches each call by test node id and call order, verifying transport
-verb + path and failing hard on any drift (``ReplayMiss``). Canonical
-content-based match keys are LIT-5741; streaming chunk fidelity is LIT-5742;
-scoping record/replay to provider-bound traffic is LIT-5745.
+Replay matches each call by test node id and canonical content key
+(fixture_canonical.py, LIT-5741): volatile headers, credential fields, unique
+markers, generated ids, and timestamps are canonicalized out before hashing, so
+matching is order-independent across distinct keys, FIFO within a key, and a
+miss fails hard (``ReplayMiss``) printing the computed key and the closest
+recorded key without ever falling through to a live call. Streaming chunk
+fidelity is LIT-5742; scoping record/replay to provider-bound traffic is
+LIT-5745.
 """
 
 from __future__ import annotations
 
+import difflib
 import functools
 import hashlib
 import os
+from collections import deque
 from dataclasses import dataclass, field
 from datetime import datetime
+from itertools import islice
 from pathlib import Path
 from typing import Final, Literal, assert_never
 
-from pydantic import BaseModel
+from pydantic import BaseModel, JsonValue
 
 from e2e_http import AuthHeaders, BinaryStream, ProbeResult, Result, StreamingResponse
 from fixture_bundle import (
@@ -43,12 +50,14 @@ from fixture_bundle import (
     check_freshness,
     format_age,
     from_result,
+    interaction_filename,
     load_bundle,
     prepare_bundle,
     slug_for_test,
     to_json_value,
     to_result,
 )
+from fixture_canonical import CanonicalRequest, canonicalize, is_secret_field
 from transport import Transport
 
 type FixtureMode = Literal["live", "record", "replay"]
@@ -118,6 +127,27 @@ def _redact(headers: dict[str, str]) -> dict[str, str]:
     }
 
 
+def _redact_secret_fields(value: JsonValue) -> JsonValue:
+    match value:
+        case dict():
+            return {
+                key: REDACTED_VALUE
+                if is_secret_field(key) and item is not None
+                else _redact_secret_fields(item)
+                for key, item in value.items()
+            }
+        case list():
+            return [_redact_secret_fields(item) for item in value]
+        case _:
+            return value
+
+
+def _redact_flat(fields: dict[str, str]) -> dict[str, str]:
+    return {
+        key: REDACTED_VALUE if is_secret_field(key) else value for key, value in fields.items()
+    }
+
+
 def recorded_request(
     method: str,
     path: str,
@@ -133,9 +163,9 @@ def recorded_request(
         method=method,
         path=path,
         headers=_redact(_dump_flat(headers)),
-        params=_dump_flat(params),
-        body=None if body is None else to_json_value(body),
-        form=None if form is None else _dump_flat(form),
+        params=_redact_flat(_dump_flat(params)),
+        body=None if body is None else _redact_secret_fields(to_json_value(body)),
+        form=None if form is None else _redact_flat(_dump_flat(form)),
         file_name=file_name,
         file_sha256=None if file_content is None else hashlib.sha256(file_content).hexdigest(),
         file_bytes=None if file_content is None else len(file_content),
@@ -304,46 +334,106 @@ class RecordingTransport:
         return response
 
 
+def _build_pool(recorded: tuple[Interaction, ...]) -> dict[str, deque[Interaction]]:
+    keys: Final = tuple(canonicalize(interaction.request).key for interaction in recorded)
+    return {
+        key: deque(
+            interaction
+            for candidate_key, interaction in zip(keys, recorded, strict=True)
+            if candidate_key == key
+        )
+        for key in dict.fromkeys(keys)
+    }
+
+
+def _closest_recorded(
+    canonical: CanonicalRequest, recorded: tuple[Interaction, ...]
+) -> tuple[CanonicalRequest, str]:
+    candidates: Final = tuple(canonicalize(interaction.request) for interaction in recorded)
+    ratios: Final = tuple(
+        difflib.SequenceMatcher(
+            None, f"{canonical.method} {canonical.path}\n{canonical.content}",
+            f"{candidate.method} {candidate.path}\n{candidate.content}",
+        ).ratio()
+        for candidate in candidates
+    )
+    best: Final = max(range(len(candidates)), key=lambda index: ratios[index])
+    return candidates[best], interaction_filename(best, recorded[best].request)
+
+
+def _miss_message(test_key: str, slug: str, canonical: CanonicalRequest, bundle: LoadedBundle) -> str:
+    recorded: Final = bundle.interactions.get(slug, ())
+    if not recorded:
+        return (
+            f"replay miss for {test_key}: computed key {canonical.key} but nothing is recorded "
+            f"under {slug}; re-record with E2E_FIXTURE_MODE=record"
+        )
+    closest, closest_file = _closest_recorded(canonical, recorded)
+    diff: Final = "\n".join(
+        islice(
+            difflib.unified_diff(
+                closest.pretty_content().splitlines(),
+                canonical.pretty_content().splitlines(),
+                fromfile=f"closest recorded ({closest_file})",
+                tofile="test made",
+                lineterm="",
+            ),
+            60,
+        )
+    )
+    return (
+        f"replay miss for {test_key}: no recorded interaction matches key {canonical.key}; "
+        f"closest recorded key is {closest.key} ({closest_file})\n{diff}\n"
+        "re-record with E2E_FIXTURE_MODE=record"
+    )
+
+
 @dataclass(slots=True)
 class ReplaySource:
-    """One shared cursor set over a loaded bundle, so every client built in the
-    session consumes the same recorded sequence per test."""
+    """One shared pool per test over a loaded bundle, so every client built in
+    the session consumes the same recorded interactions. Calls match by
+    canonical content key: order-independent across distinct keys (concurrent
+    tests interleave calls nondeterministically), FIFO within one key (a poll
+    loop replays its recorded responses in recorded order)."""
 
     bundle: LoadedBundle
-    _cursors: dict[str, int] = field(default_factory=dict)
+    _pools: dict[str, dict[str, deque[Interaction]]] = field(default_factory=dict)
 
-    def next_interaction(self, method: str, path: str) -> Interaction:
-        test_key = current_test_key()
-        slug = slug_for_test(test_key)
-        recorded = self.bundle.interactions.get(slug, ())
-        index = self._cursors.get(slug, 0)
-        if index >= len(recorded):
+    def _pool(self, slug: str) -> dict[str, deque[Interaction]]:
+        if slug not in self._pools:
+            self._pools[slug] = _build_pool(self.bundle.interactions.get(slug, ()))
+        return self._pools[slug]
+
+    def next_interaction(self, request: RecordedRequest) -> Interaction:
+        test_key: Final = current_test_key()
+        slug: Final = slug_for_test(test_key)
+        pool: Final = self._pool(slug)
+        canonical: Final = canonicalize(request)
+        queue: Final = pool.get(canonical.key)
+        if queue is None:
+            raise ReplayMiss(_miss_message(test_key, slug, canonical, self.bundle))
+        if not queue:
             raise ReplayMiss(
-                f"replay exhausted for {test_key}: call #{index + 1} ({method} {path}) has no recorded "
-                f"interaction ({len(recorded)} recorded under {slug}); re-record with E2E_FIXTURE_MODE=record"
+                f"replay exhausted for {test_key}: every recorded interaction for key "
+                f"{canonical.key} is already consumed; re-record with E2E_FIXTURE_MODE=record"
             )
-        interaction = recorded[index]
-        if interaction.request.method != method or interaction.request.path != path:
-            raise ReplayMiss(
-                f"replay mismatch for {test_key} at call #{index + 1}: recorded "
-                f"{interaction.request.method} {interaction.request.path}, test made {method} {path}; "
-                "re-record with E2E_FIXTURE_MODE=record"
-            )
-        self._cursors[slug] = index + 1
-        return interaction
+        return queue.popleft()
 
     def leftover_error(self, test_key: str) -> str | None:
         """Non-None when the test consumed fewer interactions than were recorded,
         meaning a passing replay proved less than the bundle claims."""
-        slug = slug_for_test(test_key)
-        recorded = self.bundle.interactions.get(slug, ())
-        consumed = self._cursors.get(slug, 0)
-        if consumed >= len(recorded):
+        slug: Final = slug_for_test(test_key)
+        recorded: Final = self.bundle.interactions.get(slug, ())
+        if not recorded:
+            return None
+        leftover: Final = tuple(
+            interaction for queue in self._pool(slug).values() for interaction in queue
+        )
+        if not leftover:
             return None
-        pending = recorded[consumed]
         return (
-            f"replay incomplete for {test_key}: {len(recorded) - consumed} of {len(recorded)} recorded "
-            f"interactions never consumed, next is {pending.request.method} {pending.request.path}; "
+            f"replay incomplete for {test_key}: {len(leftover)} of {len(recorded)} recorded "
+            f"interactions never consumed, e.g. {canonicalize(leftover[0].request).key}; "
             "re-record with E2E_FIXTURE_MODE=record"
         )
 
@@ -386,7 +476,12 @@ class ReplayTransport:
     def post[R: BaseModel](
         self, path: str, *, headers: BaseModel, json: BaseModel, response_type: type[R]
     ) -> Result[R]:
-        return to_result(_expect_result(self.source.next_interaction("post", path)), response_type)
+        return to_result(
+            _expect_result(
+                self.source.next_interaction(recorded_request("post", path, headers=headers, body=json))
+            ),
+            response_type,
+        )
 
     def get[R: BaseModel](
         self,
@@ -397,7 +492,12 @@ class ReplayTransport:
         response_type: type[R],
         timeout: float | None = None,
     ) -> Result[R]:
-        return to_result(_expect_result(self.source.next_interaction("get", path)), response_type)
+        return to_result(
+            _expect_result(
+                self.source.next_interaction(recorded_request("get", path, headers=headers, params=params))
+            ),
+            response_type,
+        )
 
     def delete[R: BaseModel](
         self,
@@ -408,25 +508,46 @@ class ReplayTransport:
         response_type: type[R],
         params: BaseModel | None = None,
     ) -> Result[R]:
-        return to_result(_expect_result(self.source.next_interaction("delete", path)), response_type)
+        return to_result(
+            _expect_result(
+                self.source.next_interaction(
+                    recorded_request("delete", path, headers=headers, body=json, params=params)
+                )
+            ),
+            response_type,
+        )
 
     def patch[R: BaseModel](
         self, path: str, *, headers: BaseModel, json: BaseModel, response_type: type[R]
     ) -> Result[R]:
-        return to_result(_expect_result(self.source.next_interaction("patch", path)), response_type)
+        return to_result(
+            _expect_result(
+                self.source.next_interaction(recorded_request("patch", path, headers=headers, body=json))
+            ),
+            response_type,
+        )
 
     def put[R: BaseModel](
         self, path: str, *, headers: BaseModel, json: BaseModel, response_type: type[R]
     ) -> Result[R]:
-        return to_result(_expect_result(self.source.next_interaction("put", path)), response_type)
+        return to_result(
+            _expect_result(
+                self.source.next_interaction(recorded_request("put", path, headers=headers, body=json))
+            ),
+            response_type,
+        )
 
     def stream(self, path: str, *, headers: BaseModel, json: BaseModel) -> StreamingResponse:
-        return _expect_streaming(self.source.next_interaction("stream", path))
+        return _expect_streaming(
+            self.source.next_interaction(recorded_request("stream", path, headers=headers, body=json))
+        )
 
     def stream_binary(
         self, path: str, *, headers: BaseModel, json: BaseModel, chunk_size: int = 8192
     ) -> BinaryStream:
-        interaction = self.source.next_interaction("stream_binary", path)
+        interaction = self.source.next_interaction(
+            recorded_request("stream_binary", path, headers=headers, body=json)
+        )
         match interaction.response:
             case RecordedBinary(payload=payload):
                 return payload
@@ -444,10 +565,16 @@ class ReplayTransport:
         params: BaseModel | None = None,
         stream: bool = False,
     ) -> StreamingResponse:
-        return _expect_streaming(self.source.next_interaction("send", path))
+        return _expect_streaming(
+            self.source.next_interaction(
+                recorded_request("send", path, headers=headers, body=json, params=params)
+            )
+        )
 
     def probe(self, path: str, *, params: BaseModel) -> ProbeResult:
-        interaction = self.source.next_interaction("probe", path)
+        interaction = self.source.next_interaction(
+            recorded_request("probe", path, headers=self.master, params=params)
+        )
         match interaction.response:
             case RecordedProbe(payload=payload):
                 return payload
@@ -467,10 +594,27 @@ class ReplayTransport:
         params: BaseModel | None = None,
         response_type: type[R],
     ) -> Result[R]:
-        return to_result(_expect_result(self.source.next_interaction("upload", path)), response_type)
+        return to_result(
+            _expect_result(
+                self.source.next_interaction(
+                    recorded_request(
+                        "upload",
+                        path,
+                        headers=headers,
+                        params=params,
+                        form=form,
+                        file_name=filename,
+                        file_content=content,
+                    )
+                )
+            ),
+            response_type,
+        )
 
     def download(self, path: str, *, headers: BaseModel) -> StreamingResponse:
-        return _expect_streaming(self.source.next_interaction("download", path))
+        return _expect_streaming(
+            self.source.next_interaction(recorded_request("download", path, headers=headers))
+        )
 
 
 @functools.lru_cache(maxsize=8)

tests/e2e/test_fixture_canonical.py

@@ -0,0 +1,163 @@
+"""Harness coverage for canonical request identity (LIT-5741).
+
+No proxy and no ``e2e`` marker: pure functions over ``RecordedRequest``. Pins
+the two failure modes match keys must avoid: keying on volatile material so
+nothing ever matches (markers, virtual keys, ids, timestamps, volatile
+headers), and keying on too little so different requests collide and a test
+silently asserts against another request's response.
+"""
+
+from __future__ import annotations
+
+import pytest
+from pydantic import JsonValue
+
+from fixture_bundle import RecordedRequest
+from fixture_canonical import CanonicalRequest, canonical_string, canonicalize, is_secret_field
+
+
+def request(
+    method: str = "post",
+    path: str = "/chat/completions",
+    *,
+    headers: dict[str, str] | None = None,
+    params: dict[str, str] | None = None,
+    body: JsonValue | None = None,
+    form: dict[str, str] | None = None,
+    file_name: str | None = None,
+    file_sha256: str | None = None,
+    file_bytes: int | None = None,
+) -> RecordedRequest:
+    return RecordedRequest(
+        method=method,
+        path=path,
+        headers=headers or {},
+        params=params or {},
+        body=body,
+        form=form,
+        file_name=file_name,
+        file_sha256=file_sha256,
+        file_bytes=file_bytes,
+    )
+
+
+class TestPlaceholders:
+    @pytest.mark.parametrize(
+        ("raw", "expected"),
+        [
+            ("Reply ok. 4d5152a995b7", "Reply ok. <marker>"),
+            ("e2e-chat-stream-4d5152a995b7", "e2e-chat-stream-<marker>"),
+            ("sk-3mCXCTGmYuEEIU2i2qmVE3Xq6tSK1O0X6ZIRP1Lpw8ZlbNjt", "<key>"),
+            ("9f1c8a2e-4b3d-4f6a-8f2f-0a1b2c3d4e5f", "<uuid>"),
+            ("z" * 64, "z" * 64),
+            ("0123456789abcdef" * 4, "<sha256>"),
+            ("2026-08-19T20:57:13.363499+00:00", "<timestamp>"),
+            ("2026-08-19", "<date>"),
+            ("chatcmpl-C0LO6rRkfJlpJ2mqW9BHYo4Sm8FWl", "<id>"),
+            ("batch_688a8b7f9a08819096e0f7c88fcd07c5", "<id>"),
+            ("file-XyZ12345abc", "<id>"),
+            ("gpt-4o-mini", "gpt-4o-mini"),
+            ("max_tokens", "max_tokens"),
+            ("sk-1234", "sk-1234"),
+        ],
+    )
+    def test_rewrites_exactly_the_volatile_shapes(self, raw: str, expected: str) -> None:
+        assert canonical_string(raw) == expected
+
+
+class TestSecretFields:
+    @pytest.mark.parametrize(
+        ("name", "secret"),
+        [
+            ("api_key", True),
+            ("openai_api_key", True),
+            ("aws_secret_access_key", True),
+            ("aws_session_token", True),
+            ("vertex_credentials", True),
+            ("static_headers", True),
+            ("langfuse_secret_key", True),
+            ("model", False),
+            ("max_completion_tokens", False),
+            ("api_base", False),
+        ],
+    )
+    def test_names_that_carry_credentials(self, name: str, secret: bool) -> None:
+        assert is_secret_field(name) is secret
+
+
+class TestKeyStability:
+    def test_volatile_material_does_not_change_the_key(self) -> None:
+        """Acceptance: a suite recorded on one machine (fresh keys, that day's
+        dates, that run's markers) replays on another with no misses."""
+        first = request(
+            headers={"authorization": "Bearer sk-run-one-aaaaaaaaaaaaaaaa", "x-request-id": "req-1"},
+            params={"start_date": "2026-08-18"},
+            body={
+                "model": "e2e-chat-4d5152a995b7",
+                "messages": [{"role": "user", "content": "Reply ok. 4d5152a995b7"}],
+                "api_key": "sk-live-one-aaaaaaaaaaaaaaaa",
+            },
+        )
+        second = request(
+            headers={"authorization": "Bearer sk-run-two-bbbbbbbbbbbbbbbb", "x-request-id": "req-2"},
+            params={"start_date": "2026-08-19"},
+            body={
+                "model": "e2e-chat-1a2b3c4d5e6f",
+                "messages": [{"role": "user", "content": "Reply ok. 1a2b3c4d5e6f"}],
+                "api_key": "os.environ/OPENAI_API_KEY",
+            },
+        )
+        assert canonicalize(first).key == canonicalize(second).key
+
+    def test_serialization_order_is_not_identity(self) -> None:
+        ordered = request(body={"model": "m", "stream": True})
+        reversed_order = request(body={"stream": True, "model": "m"})
+        assert canonicalize(ordered).key == canonicalize(reversed_order).key
+
+    def test_generated_ids_in_the_path_do_not_change_the_key(self) -> None:
+        first = request("get", "/v1/batches/batch_688a8b7f9a08819096e0f7c88fcd07c5")
+        second = request("get", "/v1/batches/batch_770b9c8f0b19920107f1f8d99fde18d6")
+        assert canonicalize(first).key == canonicalize(second).key
+
+
+class TestKeyDistinctness:
+    def test_requests_differing_only_inside_canonicalized_fields_stay_distinct(self) -> None:
+        """Acceptance: a naive verb+path hash collides these; the content key
+        must not, or one test silently asserts against the other's response."""
+        first = request(body={"messages": [{"content": "Reply ok. 4d5152a995b7"}]})
+        second = request(body={"messages": [{"content": "Count to three. 4d5152a995b7"}]})
+        naive = (first.method, first.path)
+        assert naive == (second.method, second.path)
+        assert canonicalize(first).key != canonicalize(second).key
+
+    def test_a_kept_header_is_identity(self) -> None:
+        first = request(headers={"x-litellm-tags": "prod"})
+        second = request(headers={"x-litellm-tags": "shadow"})
+        assert canonicalize(first).key != canonicalize(second).key
+
+    def test_a_volatile_header_is_not_identity(self) -> None:
+        first = request(headers={"traceparent": "00-aa-bb-01", "x-api-key": "one"})
+        second = request(headers={"traceparent": "00-cc-dd-01", "x-api-key": "two"})
+        assert canonicalize(first).key == canonicalize(second).key
+
+    def test_secret_set_versus_unset_stays_distinct(self) -> None:
+        with_key = request(body={"api_key": "sk-live-aaaaaaaaaaaaaaaa"})
+        without_key = request(body={"api_key": None})
+        assert canonicalize(with_key).key != canonicalize(without_key).key
+
+    def test_file_content_is_identity(self) -> None:
+        first = request(
+            "upload", "/v1/files", file_name="batch.jsonl", file_sha256="a" * 64, file_bytes=10
+        )
+        second = request(
+            "upload", "/v1/files", file_name="batch.jsonl", file_sha256="b" * 64, file_bytes=10
+        )
+        assert canonicalize(first).key != canonicalize(second).key
+
+
+class TestKeyShape:
+    def test_key_names_method_path_and_digest(self) -> None:
+        canonical = canonicalize(request("post", "/model/new", body={"model_name": "m"}))
+        assert isinstance(canonical, CanonicalRequest)
+        assert canonical.key.startswith("post /model/new #")
+        assert len(canonical.key.rsplit("#", 1)[1]) == 16

tests/e2e/test_fixture_transport.py

@@ -5,9 +5,10 @@ the live one (dependency injection, no monkeypatching): recording must pass
 every value through unchanged while writing one redacted interaction file per
 call, and replay must serve identical values from the bundle alone - the
 fake's call log proves nothing reaches the inner transport - failing hard
-(``ReplayMiss``) on any drift in order, verb, or path. The collection-time
-gate and report header are pinned here too, including the stale message that
-names the bundle's age.
+(``ReplayMiss``) on any content drift, printing the computed canonical key and
+the closest recorded key (LIT-5741; the pure canonicalizer is pinned in
+test_fixture_canonical.py). The collection-time gate and report header are
+pinned here too, including the stale message that names the bundle's age.
 """
 
 from __future__ import annotations
@@ -16,6 +17,7 @@ import hashlib
 from dataclasses import dataclass, field
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
+from uuid import uuid4
 
 import pytest
 from pydantic import BaseModel
@@ -35,10 +37,12 @@ from fixture_bundle import (
     Interaction,
     LoadedBundle,
     Manifest,
+    RecordedResult,
     load_bundle,
     prepare_bundle,
     slug_for_test,
 )
+from fixture_canonical import canonicalize
 from fixture_transport import (
     InvalidFixtureMode,
     RecordingTransport,
@@ -50,6 +54,7 @@ from fixture_transport import (
     fixture_mode_collection_error,
     fixture_report_lines,
     parse_fixture_mode,
+    recorded_request,
     replay_leftover_error,
     select_transport,
 )
@@ -70,6 +75,17 @@ class Query(BaseModel):
     q: str
 
 
+class DeployParams(BaseModel):
+    model: str
+    api_key: str | None = None
+    aws_secret_access_key: str | None = None
+
+
+class DeployBody(BaseModel):
+    model_name: str
+    litellm_params: DeployParams
+
+
 STREAMING = StreamingResponse(
     status_code=200,
     body="",
@@ -272,6 +288,28 @@ class TestRecordingTransport:
         }
         assert "sk-secret" not in this_tests_files(root)[0].read_text(encoding="utf-8")
 
+    def test_redacts_credential_body_fields_in_the_recorded_request(self, tmp_path: Path) -> None:
+        fake = FakeTransport()
+        root = tmp_path / "bundle"
+        recording: Transport = RecordingTransport(inner=fake, recorder=make_recorder(root))
+        recording.post(
+            "/model/new",
+            headers=fake.master,
+            json=DeployBody(
+                model_name="m",
+                litellm_params=DeployParams(model="openai/gpt", api_key="sk-live-provider-secret-123456"),
+            ),
+            response_type=Payload,
+        )
+        raw = this_tests_files(root)[0].read_text(encoding="utf-8")
+        interaction = Interaction.model_validate_json(raw)
+        assert "sk-live-provider-secret-123456" not in raw
+        assert isinstance(interaction.request.body, dict)
+        params = interaction.request.body["litellm_params"]
+        assert isinstance(params, dict)
+        assert params["api_key"] == "<redacted>"
+        assert params["aws_secret_access_key"] is None
+
     def test_upload_records_a_content_digest_not_the_bytes(self, tmp_path: Path) -> None:
         fake = FakeTransport()
         root = tmp_path / "bundle"
@@ -335,25 +373,135 @@ class TestReplayTransport:
         )
         assert fake.calls == calls_after_record
 
-    def test_mismatched_call_names_recorded_and_actual(self, tmp_path: Path) -> None:
+    def test_miss_names_the_computed_key_and_the_closest_recorded_key(self, tmp_path: Path) -> None:
         fake = FakeTransport()
         root = tmp_path / "bundle"
         recording: Transport = RecordingTransport(inner=fake, recorder=make_recorder(root))
         recording.post("/model/new", headers=fake.master, json=Body(prompt="x"), response_type=Payload)
         replay: Transport = ReplayTransport(source=replay_source(root), master_key="sk-1234")
-        with pytest.raises(ReplayMiss, match=r"recorded post /model/new, test made get /v1/models"):
+        with pytest.raises(ReplayMiss) as excinfo:
             replay.get("/v1/models", headers=replay.master, params=Query(q="all"), response_type=Payload)
+        message = str(excinfo.value)
+        assert "no recorded interaction matches key get /v1/models #" in message
+        assert "closest recorded key is post /model/new #" in message
+        assert "0000-post-model-new.json" in message
+        assert "re-record with E2E_FIXTURE_MODE=record" in message
 
-    def test_exhausted_recording_names_the_call_count(self, tmp_path: Path) -> None:
+    def test_content_drift_on_the_same_route_misses_with_no_live_call(self, tmp_path: Path) -> None:
+        """The naive verb+path match replayed a stale response for a request
+        whose content had changed, silently passing; a content key must miss,
+        print both canonical forms' diff, and never reach the inner transport."""
+        fake = FakeTransport()
+        root = tmp_path / "bundle"
+        recording: Transport = RecordingTransport(inner=fake, recorder=make_recorder(root))
+        recording.post("/model/new", headers=fake.master, json=Body(prompt="x"), response_type=Payload)
+        calls_after_record = list(fake.calls)
+        replay: Transport = ReplayTransport(source=replay_source(root), master_key="sk-1234")
+        with pytest.raises(ReplayMiss) as excinfo:
+            replay.post("/model/new", headers=replay.master, json=Body(prompt="y"), response_type=Payload)
+        message = str(excinfo.value)
+        assert "no recorded interaction matches key post /model/new #" in message
+        assert "closest recorded key is post /model/new #" in message
+        assert '-    "prompt": "x"' in message
+        assert '+    "prompt": "y"' in message
+        assert fake.calls == calls_after_record
+
+    def test_exhausted_key_names_the_key(self, tmp_path: Path) -> None:
         fake = FakeTransport()
         root = tmp_path / "bundle"
         recording: Transport = RecordingTransport(inner=fake, recorder=make_recorder(root))
         recording.post("/model/new", headers=fake.master, json=Body(prompt="x"), response_type=Payload)
         replay: Transport = ReplayTransport(source=replay_source(root), master_key="sk-1234")
         replay.post("/model/new", headers=replay.master, json=Body(prompt="x"), response_type=Payload)
-        with pytest.raises(ReplayMiss, match=r"call #2 \(post /model/new\) has no recorded interaction \(1 recorded"):
+        with pytest.raises(
+            ReplayMiss, match=r"every recorded interaction for key post /model/new #\w{16} is already consumed"
+        ):
             replay.post("/model/new", headers=replay.master, json=Body(prompt="x"), response_type=Payload)
 
+    def test_replays_out_of_recorded_order_across_distinct_keys(self, tmp_path: Path) -> None:
+        """Concurrent tests interleave independent calls nondeterministically
+        (e.g. a burst of parallel chat calls), so replay matches by content,
+        never by recorded position."""
+        fake = FakeTransport()
+        root = tmp_path / "bundle"
+        recording: Transport = RecordingTransport(inner=fake, recorder=make_recorder(root))
+        recording.post("/model/new", headers=fake.master, json=Body(prompt="x"), response_type=Payload)
+        recording.post("/key/generate", headers=fake.master, json=Body(prompt="k"), response_type=Payload)
+        source = replay_source(root)
+        replay: Transport = ReplayTransport(source=source, master_key="sk-1234")
+        replay.post("/key/generate", headers=replay.master, json=Body(prompt="k"), response_type=Payload)
+        replay.post("/model/new", headers=replay.master, json=Body(prompt="x"), response_type=Payload)
+        assert source.leftover_error(current_test_key()) is None
+
+    def test_identical_requests_replay_their_responses_in_recorded_order(self, tmp_path: Path) -> None:
+        """A poll loop makes the same request repeatedly and asserts on the
+        progression, so duplicates under one key stay FIFO."""
+        root = tmp_path / "bundle"
+        recorder = make_recorder(root)
+        recorder.record(
+            test_key=current_test_key(),
+            request=recorded_request(
+                "get", "/v1/models", headers=AuthHeaders(authorization="Bearer sk-x"), params=Query(q="all")
+            ),
+            response=RecordedResult(kind="success", status_code=200, data={"value": "first"}),
+        )
+        recorder.record(
+            test_key=current_test_key(),
+            request=recorded_request(
+                "get", "/v1/models", headers=AuthHeaders(authorization="Bearer sk-x"), params=Query(q="all")
+            ),
+            response=RecordedResult(kind="success", status_code=200, data={"value": "second"}),
+        )
+        replay: Transport = ReplayTransport(source=replay_source(root), master_key="sk-1234")
+        first = replay.get("/v1/models", headers=replay.master, params=Query(q="all"), response_type=Payload)
+        second = replay.get("/v1/models", headers=replay.master, params=Query(q="all"), response_type=Payload)
+        assert first == Success(status_code=200, data=Payload(value="first"))
+        assert second == Success(status_code=200, data=Payload(value="second"))
+
+
+class TestRecordedKeySets:
+    def test_two_separate_recordings_of_one_flow_produce_identical_key_sets(
+        self, tmp_path: Path
+    ) -> None:
+        """Everything a run randomizes (markers, virtual keys, dates) must
+        canonicalize out, so separately recorded runs of the same suite agree
+        on every match key and a bundle recorded elsewhere replays here."""
+
+        def record_flow(root: Path, run_date: str) -> list[str]:
+            fake = FakeTransport()
+            recording: Transport = RecordingTransport(inner=fake, recorder=make_recorder(root))
+            marker = deterministic_marker()
+            recording.post(
+                "/model/new",
+                headers=fake.master,
+                json=DeployBody(
+                    model_name=f"e2e-chat-{marker}",
+                    litellm_params=DeployParams(model="openai/gpt", api_key=f"sk-live-{uuid4().hex}"),
+                ),
+                response_type=Payload,
+            )
+            recording.post(
+                "/chat/completions",
+                headers=recording.bearer(f"sk-{uuid4().hex}"),
+                json=Body(prompt=f"Reply with the single word ok. {marker}"),
+                response_type=Payload,
+            )
+            recording.get(
+                "/spend/logs", headers=fake.master, params=Query(q=run_date), response_type=Payload
+            )
+            loaded = load_bundle(root)
+            assert isinstance(loaded, LoadedBundle)
+            return sorted(
+                canonicalize(interaction.request).key
+                for interactions in loaded.interactions.values()
+                for interaction in interactions
+            )
+
+        first_keys = record_flow(tmp_path / "one", "2026-08-18")
+        second_keys = record_flow(tmp_path / "two", "2026-08-19")
+        assert first_keys == second_keys
+        assert len(first_keys) == 3
+
 
 class TestReplayLeftover:
     def test_fully_consumed_recording_leaves_nothing(self, tmp_path: Path) -> None:
@@ -378,7 +526,7 @@ class TestReplayLeftover:
         error = source.leftover_error(current_test_key())
         assert error is not None
         assert "1 of 2 recorded interactions never consumed" in error
-        assert "next is probe /health/liveliness" in error
+        assert "e.g. probe /health/liveliness #" in error
         assert "re-record with E2E_FIXTURE_MODE=record" in error
 
     def test_test_without_recordings_has_no_leftover(self, tmp_path: Path) -> None: