Merge pull request #41149 from BerriAI/litellm_strict_provider_identity

test: add strict stateless provider replay identity

.circleci/config.yml

@@ -2915,6 +2915,25 @@ jobs:
               exit 1
             fi
 
+  provider_replay_harness:
+    docker:
+      - *python312_image
+    working_directory: ~/project
+    resource_class: medium
+    steps:
+      - setup_litellm_test_deps
+      - run:
+          name: Test provider replay harness
+          command: |
+            mkdir -p test-results/provider-replay-harness
+            uv run --no-sync pytest -q --noconftest -o addopts= -o pythonpath=tests/e2e -p no:rerunfailures \
+              --junitxml=test-results/provider-replay-harness/junit.xml \
+              tests/e2e/test_provider_edge.py tests/e2e/test_fixture_bundle.py \
+              tests/e2e/test_fixture_canonical.py tests/e2e/test_fixture_mode.py \
+              tests/code_coverage_tests/test_provider_replay_harness.py
+      - store_test_results:
+          path: test-results/provider-replay-harness
+
   integration_contracts:
     parameters:
       suite:
@@ -2967,6 +2986,7 @@ workflows:
               only:
                 - main
                 - /litellm_.*/
+      - provider_replay_harness
       - base_sdk_install:
           filters: *main_branches
       - local_testing_part1:

tests/code_coverage_tests/test_provider_replay_harness.py

@@ -0,0 +1,329 @@
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+import threading
+from pathlib import Path
+from typing import Final
+
+import pytest
+from fixture_bundle import BundleRecorder, LoadedBundle, load_bundle, prepare_bundle
+from fixture_mode import current_test_key
+from fixture_profile import MatchProfile
+from provider_edge import REPLAY_MISS_STATUS, RecordEdge, ReplayEdge, ReplaySource
+from test_provider_edge import (
+    CHAT_PATH,
+    SSE_CHUNKS,
+    STREAM_BODY,
+    UPLOAD_PATH,
+    call_edge,
+    chunked_provider,
+    fake_provider,
+    json_object,
+    provider_url,
+    raw_stream_post,
+    running_edge,
+    this_tests_files,
+)
+
+
+class TestStrictIdentity:
+    @pytest.mark.parametrize("path", [CHAT_PATH, "/anthropic/v1/messages"])
+    def test_roundtrip_rejects_semantic_changes(self, tmp_path: Path, path: str) -> None:
+        recorder: Final = prepare_bundle(tmp_path / "strict", profile="stateless_v1")
+        assert isinstance(recorder, BundleRecorder)
+        original: Final = (
+            b'{"model":"synthetic","messages":[{"role":"user",'
+            b'"content":"2031-04-05 00000000-0000-0000-0000-000000000001"}],"options":[1,2]}'
+        )
+        headers: Final = {
+            "content-type": "application/json",
+            "accept": "application/json",
+            "anthropic-version": "2023-06-01",
+            "anthropic-beta": "feature-a",
+            "openai-beta": "feature-b",
+            "authorization": "Bearer synthetic-secret-one",
+        }
+        query: Final = "?part=one&part=two&blank="
+        with fake_provider() as provider:
+            mounts: Final = {"openai": provider_url(provider), "anthropic": provider_url(provider)}
+            with running_edge(RecordEdge(recorder, threading.Lock()), mounts) as edge:
+                captured: Final = call_edge(edge, "POST", path + query, body=original, headers=headers)
+                assert captured.status_code == 200
+                assert json_object(captured.body)["echo"] == original.decode()
+            loaded: Final = load_bundle(recorder.root, profile="stateless_v1")
+            assert isinstance(loaded, LoadedBundle)
+            assert loaded.manifest.match_profile == "stateless_v1"
+            source: Final = ReplaySource(loaded)
+            with running_edge(ReplayEdge(source), mounts) as edge:
+                cases: Final = (
+                    (original.replace(b"2031-04-05", b"2032-06-07"), headers, query, "body"),
+                    (original.replace(b"000000000001", b"000000000002"), headers, query, "body"),
+                    (original.replace(b"[1,2]", b"[2,1]"), headers, query, "body"),
+                    (original.replace(b"synthetic", b"other"), headers, query, "body"),
+                    (original, headers, "?part=three&part=two&blank=", "query"),
+                    (original, headers, "?part=two&part=one&blank=", "query"),
+                    *(
+                        (original, {k: v for k, v in headers.items() if k != name}, query, "headers")
+                        for name in ("accept", "anthropic-version", "anthropic-beta", "openai-beta")
+                    ),
+                    *(
+                        (original, {**headers, name: value}, query, "headers")
+                        for name in ("accept", "anthropic-version", "anthropic-beta", "openai-beta")
+                        for value in ("different", "")
+                    ),
+                    (original, {**headers, "authorization": "Basic synthetic-secret-two"}, query, "auth"),
+                    (original, {k: v for k, v in headers.items() if k != "authorization"}, query, "auth"),
+                )
+                for rejected, reason in (
+                    (call_edge(edge, "POST", path + changed_query, body=body, headers=changed_headers), reason)
+                    for body, changed_headers, changed_query, reason in cases
+                ):
+                    assert rejected.status_code == REPLAY_MISS_STATUS
+                    assert reason in rejected.body.decode()
+                    assert b"synthetic-secret" not in rejected.body
+                reordered: Final = json.dumps(dict(reversed(list(json_object(original).items())))).encode()
+                accepted: Final = call_edge(
+                    edge, "POST", path + query, body=reordered, headers={k.upper(): v for k, v in headers.items()}
+                )
+                assert accepted.status_code == 200
+                assert accepted.body == captured.body
+            assert source.leftover_error(current_test_key()) is None
+            assert len(provider.hits) == 1
+        assert "synthetic-secret" not in "".join(file.read_text() for file in recorder.root.rglob("*.json"))
+
+    @pytest.mark.parametrize(
+        "body",
+        [
+            b'{"value":null}',
+            b'{"value":""}',
+            b'{"value":false}',
+            b'{"value":0}',
+            b'{"value":[]}',
+            b'{"value":{}}',
+            b'{"value":0.123456789012345678901}',
+            b'{"value":0.123456789012345678902}',
+            b'{"value":1e400}',
+            b'{"value":1}',
+            b'{"value":1e0}',
+            b'{"value":-0}',
+            b'{"value":1e9999999999999999999}',
+        ],
+    )
+    def test_json_values_remain_distinct(self, tmp_path: Path, body: bytes) -> None:
+        recorder: Final = prepare_bundle(tmp_path / "strict", profile="stateless_v1")
+        assert isinstance(recorder, BundleRecorder)
+        with fake_provider() as provider:
+            mounts: Final = {"openai": provider_url(provider)}
+            with running_edge(RecordEdge(recorder, threading.Lock()), mounts) as edge:
+                assert (
+                    call_edge(
+                        edge, "POST", CHAT_PATH, body=body, headers={"content-type": "application/json"}
+                    ).status_code
+                    == 200
+                )
+            loaded: Final = load_bundle(recorder.root, profile="stateless_v1")
+            assert isinstance(loaded, LoadedBundle)
+            with running_edge(ReplayEdge(ReplaySource(loaded)), mounts) as edge:
+                values: Final = (
+                    b"{}",
+                    b'{"value":null}',
+                    b'{"value":""}',
+                    b'{"value":false}',
+                    b'{"value":0}',
+                    b'{"value":[]}',
+                    b'{"value":{}}',
+                    b'{"value":0.123456789012345678901}',
+                    b'{"value":0.123456789012345678902}',
+                    b'{"value":1e400}',
+                    b'{"value":1}',
+                    b'{"value":1e0}',
+                    b'{"value":-0}',
+                    b'{"value":1e9999999999999999999}',
+                )
+                for rejected in (
+                    call_edge(edge, "POST", CHAT_PATH, body=value, headers={"content-type": "application/json"})
+                    for value in values
+                    if value != body
+                ):
+                    assert rejected.status_code == REPLAY_MISS_STATUS
+                    assert b"body" in rejected.body
+                assert (
+                    call_edge(
+                        edge, "POST", CHAT_PATH, body=body, headers={"content-type": "application/json"}
+                    ).status_code
+                    == 200
+                )
+            assert len(provider.hits) == 1
+
+    @pytest.mark.parametrize(
+        "path,body,headers",
+        [
+            (UPLOAD_PATH, b"{}", {"content-type": "application/json"}),
+            (CHAT_PATH + "?part=%FF", b"{}", {"content-type": "application/json"}),
+            (CHAT_PATH + "?part=%FE", b"{}", {"content-type": "application/json"}),
+            (CHAT_PATH, b"opaque", {"content-type": "application/octet-stream"}),
+            (CHAT_PATH, b"--boundary", {"content-type": "multipart/form-data; boundary=boundary"}),
+            (CHAT_PATH, b'{"x":1,"x":2}', {"content-type": "application/json"}),
+            (CHAT_PATH, b"{}", {"content-type": "application/json", "x-custom-behavior": "synthetic-private-value"}),
+        ],
+    )
+    def test_ineligible_capture_never_calls_provider(
+        self, tmp_path: Path, path: str, body: bytes, headers: dict[str, str]
+    ) -> None:
+        recorder: Final = prepare_bundle(tmp_path / "strict", profile="stateless_v1")
+        assert isinstance(recorder, BundleRecorder)
+        with fake_provider() as provider:
+            with running_edge(RecordEdge(recorder, threading.Lock()), {"openai": provider_url(provider)}) as edge:
+                result: Final = call_edge(edge, "POST", path, body=body, headers=headers)
+                assert result.status_code == REPLAY_MISS_STATUS
+                assert b"eligibility error" in result.body
+                assert b"synthetic-private-value" not in result.body
+            assert provider.hits == []
+        assert this_tests_files(recorder.root) == []
+
+    def test_destination_is_part_of_actual_http_identity(self, tmp_path: Path) -> None:
+        recorder: Final = prepare_bundle(tmp_path / "strict", profile="stateless_v1")
+        assert isinstance(recorder, BundleRecorder)
+        with fake_provider() as provider:
+            with running_edge(RecordEdge(recorder, threading.Lock()), {"openai": provider_url(provider)}) as edge:
+                assert (
+                    call_edge(
+                        edge, "POST", CHAT_PATH, body=b"{}", headers={"content-type": "application/json"}
+                    ).status_code
+                    == 200
+                )
+            loaded: Final = load_bundle(recorder.root, profile="stateless_v1")
+            assert isinstance(loaded, LoadedBundle)
+            with running_edge(ReplayEdge(ReplaySource(loaded)), {"openai": provider_url(provider) + "/other"}) as edge:
+                result: Final = call_edge(
+                    edge, "POST", CHAT_PATH, body=b"{}", headers={"content-type": "application/json"}
+                )
+                assert result.status_code == REPLAY_MISS_STATUS
+                assert b"upstream" in result.body
+            assert len(provider.hits) == 1
+
+    def test_credentials_are_not_identity_and_fresh_process_replays(self, tmp_path: Path) -> None:
+        recorder: Final = prepare_bundle(tmp_path / "strict", profile="stateless_v1")
+        assert isinstance(recorder, BundleRecorder)
+        headers: Final = {
+            "content-type": "application/json",
+            "authorization": "bEaReR synthetic-token",
+            "x-api-key": "synthetic-api-key",
+            "cookie": "synthetic-cookie",
+        }
+        path: Final = CHAT_PATH + "?api_key=synthetic-query-secret&part=one&part=two"
+        body: Final = b'{"model":"synthetic","messages":[]}'
+        with fake_provider(echo_request=False) as provider:
+            mounts: Final = {"openai": provider_url(provider)}
+            with running_edge(RecordEdge(recorder, threading.Lock()), mounts) as edge:
+                captured: Final = call_edge(edge, "POST", path, body=body, headers=headers)
+                assert captured.status_code == 200
+            seen_headers, seen_body = provider.requests[0]
+            assert {k.lower(): v for k, v in seen_headers.items()}.items() >= headers.items()
+            assert seen_body == body
+            assert provider.hits == ["POST " + path.removeprefix("/openai")]
+            artifacts: Final = "".join(file.read_text() for file in recorder.root.rglob("*.json"))
+            for secret in ("synthetic-token", "synthetic-api-key", "synthetic-cookie", "synthetic-query-secret"):
+                assert secret not in artifacts
+            child: Final = subprocess.run(
+                [
+                    sys.executable,
+                    "-c",
+                    """
+import json, sys
+from pathlib import Path
+from fixture_bundle import LoadedBundle, load_bundle
+from provider_edge import ProviderRequestObservation, observed_provider_edge, replay_leftover_error
+from test_provider_edge import call_edge
+from fixture_profile import MatchProfile
+from fixture_mode import current_test_key
+loaded = load_bundle(Path(sys.argv[1]), profile="stateless_v1")
+assert isinstance(loaded, LoadedBundle)
+with observed_provider_edge(ProviderRequestObservation("synthetic"), mode_raw="replay", bundle_dir=Path(sys.argv[1]), bind_host="127.0.0.1", advertise_host="127.0.0.1", mounts={"openai": sys.argv[2]}) as edge:
+    response = call_edge(edge, "POST", sys.argv[3], body=sys.argv[4].encode(), headers=json.loads(sys.argv[5]))
+    assert response.status_code == 200
+    print(response.body.decode())
+assert replay_leftover_error(mode_raw="replay", bundle_dir=Path(sys.argv[1]), test_key=current_test_key()) is None
+""",
+                    str(recorder.root),
+                    provider_url(provider),
+                    path.replace("synthetic-query-secret", "new-query-credential"),
+                    body.decode(),
+                    json.dumps({**headers, "authorization": "Bearer another-credential", "x-api-key": "another-key"}),
+                ],
+                env={
+                    **os.environ,
+                    "PYTHONPATH": str(Path(__file__).resolve().parents[1] / "e2e"),
+                    "E2E_REPLAY_MATCH_PROFILE": "stateless_v1",
+                },
+                capture_output=True,
+                text=True,
+                timeout=30,
+            )
+            assert child.returncode == 0, child.stderr
+            assert child.stdout.strip().encode() == captured.body
+            assert len(provider.hits) == 1
+
+    @pytest.mark.parametrize("profile,other", [("legacy", "stateless_v1"), ("stateless_v1", "legacy")])
+    def test_profiles_cannot_load_each_others_bundles(
+        self, tmp_path: Path, profile: MatchProfile, other: MatchProfile
+    ) -> None:
+        from fixture_bundle import UnreadableBundle
+
+        recorder: Final = prepare_bundle(tmp_path / profile, profile=profile)
+        assert isinstance(recorder, BundleRecorder)
+        mismatch: Final = load_bundle(recorder.root, profile=other)
+        assert isinstance(mismatch, UnreadableBundle)
+        assert "profile mismatch" in mismatch.reason
+        assert "re-record" in mismatch.reason
+
+    @pytest.mark.parametrize("abort_after", [None, 2])
+    def test_strict_stream_preserves_chunks_and_truncation(self, tmp_path: Path, abort_after: int | None) -> None:
+        recorder: Final = prepare_bundle(tmp_path / "strict", profile="stateless_v1")
+        assert isinstance(recorder, BundleRecorder)
+        with chunked_provider(abort_after=abort_after) as provider:
+            mounts: Final = {"anthropic": provider_url(provider)}
+            with running_edge(RecordEdge(recorder, threading.Lock()), mounts) as edge:
+                _, captured, captured_ending = raw_stream_post(edge.port, "/anthropic/v1/messages", STREAM_BODY)
+            loaded: Final = load_bundle(recorder.root, profile="stateless_v1")
+            assert isinstance(loaded, LoadedBundle)
+            source: Final = ReplaySource(loaded)
+            with running_edge(ReplayEdge(source), mounts) as edge:
+                _, replayed, ending = raw_stream_post(edge.port, "/anthropic/v1/messages", STREAM_BODY)
+            assert captured == replayed == list(SSE_CHUNKS[:abort_after])
+            assert ending == captured_ending
+            assert (ending == "terminated") == (abort_after is None)
+            assert source.leftover_error(current_test_key()) is None
+            assert len(provider.hits) == 1
+
+    def test_auth_scheme_survives_missing_credentials(self, tmp_path: Path) -> None:
+        recorder: Final = prepare_bundle(tmp_path / "strict", profile="stateless_v1")
+        assert isinstance(recorder, BundleRecorder)
+        headers: Final = {"content-type": "application/json", "authorization": "Bearer"}
+        with fake_provider() as provider:
+            mounts: Final = {"openai": provider_url(provider)}
+            with running_edge(RecordEdge(recorder, threading.Lock()), mounts) as edge:
+                assert call_edge(edge, "POST", CHAT_PATH, body=b"{}", headers=headers).status_code == 200
+            loaded: Final = load_bundle(recorder.root, profile="stateless_v1")
+            assert isinstance(loaded, LoadedBundle)
+            with running_edge(ReplayEdge(ReplaySource(loaded)), mounts) as edge:
+                for result in (
+                    call_edge(edge, "POST", CHAT_PATH, body=b"{}", headers={**headers, "authorization": scheme})
+                    for scheme in ("Basic", "Digest")
+                ):
+                    assert result.status_code == REPLAY_MISS_STATUS
+                    assert b"auth" in result.body
+                assert (
+                    call_edge(
+                        edge,
+                        "POST",
+                        CHAT_PATH,
+                        body=b"{}",
+                        headers={**headers, "authorization": "bEaReR synthetic-token"},
+                    ).status_code
+                    == 200
+                )
+            assert len(provider.hits) == 1

tests/e2e/CONTRIBUTING.md

@@ -236,3 +236,15 @@ Before you push
 4. Capture screenshots of the test run and attach them to the PR as proof
 
 5. If a test fails because it surfaced a real issue in the product, flag that explicitly in the PR rather than reworking the test until it passes
+
+### Strict stateless replay matching
+
+Set `E2E_REPLAY_MATCH_PROFILE=stateless_v1` for both recording and replay to bind OpenAI `/v1/chat/completions` and Anthropic `/v1/messages` requests to their upstream destination, ordered query pairs, semantic headers and literal JSON content. The default remains `legacy`. Strict bundles use format 5 and cannot load as legacy bundles; select the matching profile or re-record with `E2E_FIXTURE_MODE=record`. Missing profile metadata never enrolls a legacy bundle in strict matching
+
+Strict matching preserves dates, UUIDs, hashes, model names, tool arguments, array order and omitted/null/empty/false/zero values. JSON object key order and header name casing may change. The strict body uses tagged JSON values so number precision and JSON types survive persistence, including exact numeric spelling and numbers larger than a floating-point value. Invalid UTF-8 query values fail eligibility. Duplicate JSON keys, unsupported endpoints, non-JSON bodies and unknown semantic headers fail eligibility before contacting a provider
+
+The semantic header set is `content-type`, `accept`, `anthropic-version`, `anthropic-beta` and `openai-beta`, including missing versus present values. Authorization records presence and the case-insensitive scheme; `x-api-key` records presence only. Credential values and cookies are excluded. Credential query values are redacted while their position and field name remain in the identity. Never use real customer inputs in fixture qualification
+
+Excluded transport and telemetry headers are `host`, `content-length`, `connection`, `accept-encoding`, `user-agent`, `traceparent`, `tracestate`, `x-request-id`, `x-client-request-id` and `x-stainless-*`. Inbound transfer-encoding is unsupported; send JSON with content-length framing. The destination represents host identity and the relay carries original body bytes. Replay does not verify credentials, SDK timeout/retry behavior, transport performance, model availability or stateful remote IDs. Live relay uses original request bytes and header values, never the stored identity
+
+Strict replay harness regression tests live in `tests/code_coverage_tests/test_provider_replay_harness.py`. The CircleCI `provider_replay_harness` job runs them alongside the existing legacy harness files with `--noconftest -o pythonpath=tests/e2e`; they need only synthetic HTTP providers and temporary fixture storage

tests/e2e/fixture_bundle.py

@@ -30,9 +30,11 @@ from datetime import datetime, timedelta, timezone
 from pathlib import Path
 from typing import Annotated, Final, Literal
 
+from fixture_profile import MatchProfile, StrictIdentity
 from pydantic import BaseModel, Field, JsonValue
 
 BUNDLE_FORMAT_VERSION: Final = 4
+STRICT_BUNDLE_FORMAT_VERSION: Final = 5
 MAX_BUNDLE_AGE: Final = timedelta(days=7)
 MANIFEST_FILENAME: Final = "manifest.json"
 
@@ -41,6 +43,7 @@ class Manifest(BaseModel):
     format_version: int
     recorded_at: datetime
     harness_version: str
+    match_profile: MatchProfile = "legacy"
 
 
 class RecordedRequest(BaseModel):
@@ -69,6 +72,7 @@ class RecordedRequest(BaseModel):
     file_name: str | None = None
     file_sha256: str | None = None
     file_bytes: int | None = None
+    strict_identity: StrictIdentity | None = None
 
 
 class RecordedHttpResponse(BaseModel):
@@ -100,9 +104,7 @@ class RecordedStreamedResponse(BaseModel):
     truncated: str | None = None
 
 
-type RecordedResponse = Annotated[
-    RecordedHttpResponse | RecordedStreamedResponse, Field(discriminator="kind")
-]
+type RecordedResponse = Annotated[RecordedHttpResponse | RecordedStreamedResponse, Field(discriminator="kind")]
 
 
 class Interaction(BaseModel):
@@ -152,6 +154,7 @@ class BundleRecorder:
     manifest, so record mode never reads (or merges into) an existing bundle."""
 
     root: Path
+    profile: MatchProfile = "legacy"
     _ordinals: dict[str, int] = field(default_factory=dict)
 
     def record(self, *, test_key: str, request: RecordedRequest, response: RecordedResponse) -> None:
@@ -162,7 +165,12 @@ class BundleRecorder:
         directory.mkdir(parents=True, exist_ok=True)
         interaction = Interaction(request=request, response=response)
         target = directory / interaction_filename(ordinal, request)
-        target.write_text(interaction.model_dump_json(indent=2), encoding="utf-8")
+        target.write_text(
+            interaction.model_dump_json(
+                indent=2, exclude={"request": {"strict_identity"}} if self.profile == "legacy" else None
+            ),
+            encoding="utf-8",
+        )
 
 
 @dataclass(frozen=True, slots=True)
@@ -171,7 +179,7 @@ class UnsafeBundleDir:
     reason: str
 
 
-def prepare_bundle(root: Path) -> BundleRecorder | UnsafeBundleDir:
+def prepare_bundle(root: Path, *, profile: MatchProfile = "legacy") -> BundleRecorder | UnsafeBundleDir:
     """Start a fresh bundle at ``root`` for record mode: wipe whatever bundle is
     there and write a new manifest. Refuses to wipe a directory that is neither
     empty nor a bundle (no manifest.json), so a mistyped E2E_FIXTURE_DIR can
@@ -188,12 +196,15 @@ def prepare_bundle(root: Path) -> BundleRecorder | UnsafeBundleDir:
         shutil.rmtree(root)
     root.mkdir(parents=True)
     manifest = Manifest(
-        format_version=BUNDLE_FORMAT_VERSION,
+        format_version=BUNDLE_FORMAT_VERSION if profile == "legacy" else STRICT_BUNDLE_FORMAT_VERSION,
+        match_profile=profile,
         recorded_at=datetime.now(timezone.utc),
         harness_version=harness_version(),
     )
-    (root / MANIFEST_FILENAME).write_text(manifest.model_dump_json(indent=2), encoding="utf-8")
-    return BundleRecorder(root=root)
+    (root / MANIFEST_FILENAME).write_text(
+        manifest.model_dump_json(indent=2, exclude={"match_profile"} if profile == "legacy" else None), encoding="utf-8"
+    )
+    return BundleRecorder(root=root, profile=profile)
 
 
 @dataclass(frozen=True, slots=True)
@@ -226,25 +237,30 @@ def _read_manifest(root: Path) -> Manifest | UnreadableBundle:
         return UnreadableBundle(reason=f"{MANIFEST_FILENAME} is invalid: {exc}")
 
 
-def _supported_manifest(root: Path) -> Manifest | UnreadableBundle:
+def _supported_manifest(root: Path, profile: MatchProfile = "legacy") -> Manifest | UnreadableBundle:
     """The manifest, refused when it was written under a different format version.
     A bundle is atomic (record wipes and rewrites the whole directory and never
     merges), so a foreign version is a hard reject rather than a partial read."""
     manifest = _read_manifest(root)
     if isinstance(manifest, UnreadableBundle):
         return manifest
-    if manifest.format_version != BUNDLE_FORMAT_VERSION:
+    expected_version: Final = BUNDLE_FORMAT_VERSION if profile == "legacy" else STRICT_BUNDLE_FORMAT_VERSION
+    if manifest.match_profile != profile:
+        return UnreadableBundle(
+            reason="match profile mismatch; select the recorded E2E_REPLAY_MATCH_PROFILE or re-record"
+        )
+    if manifest.format_version != expected_version:
         return UnreadableBundle(
             reason=(
-                f"format_version {manifest.format_version} != supported {BUNDLE_FORMAT_VERSION}; "
+                f"format_version {manifest.format_version} != supported {expected_version}; "
                 "re-record with E2E_FIXTURE_MODE=record"
             )
         )
     return manifest
 
 
-def check_freshness(root: Path, *, now: datetime) -> BundleFreshness:
-    manifest = _supported_manifest(root)
+def check_freshness(root: Path, *, now: datetime, profile: MatchProfile = "legacy") -> BundleFreshness:
+    manifest = _supported_manifest(root, profile)
     if isinstance(manifest, UnreadableBundle):
         return manifest
     recorded_at = (
@@ -269,16 +285,27 @@ class LoadedBundle:
     interactions: dict[str, tuple[Interaction, ...]]
 
 
-def load_bundle(root: Path) -> LoadedBundle | UnreadableBundle:
-    manifest = _supported_manifest(root)
+def load_bundle(root: Path, *, profile: MatchProfile = "legacy") -> LoadedBundle | UnreadableBundle:
+    manifest = _supported_manifest(root, profile)
     if isinstance(manifest, UnreadableBundle):
         return manifest
-    interactions = {
-        directory.name: tuple(
-            Interaction.model_validate_json(file.read_text(encoding="utf-8"))
-            for file in sorted(directory.glob("*.json"))
-        )
-        for directory in sorted(root.iterdir())
-        if directory.is_dir()
-    }
+    try:
+        interactions = {
+            directory.name: tuple(
+                Interaction.model_validate_json(file.read_text(encoding="utf-8"))
+                for file in sorted(directory.glob("*.json"))
+            )
+            for directory in sorted(root.iterdir())
+            if directory.is_dir()
+        }
+    except (ValueError, OSError):
+        if profile == "legacy":
+            raise
+        return UnreadableBundle(reason="invalid stateless_v1 interaction; re-record with the selected profile")
+    if any(
+        (item.request.strict_identity is not None) != (profile == "stateless_v1")
+        for items in interactions.values()
+        for item in items
+    ):
+        return UnreadableBundle(reason="request identity/profile mismatch; re-record with the selected profile")
     return LoadedBundle(manifest=manifest, interactions=interactions)

tests/e2e/fixture_canonical.py

@@ -23,9 +23,8 @@ from dataclasses import dataclass
 from functools import reduce
 from typing import Final
 
-from pydantic import JsonValue
-
 from fixture_bundle import RecordedRequest
+from pydantic import JsonValue
 
 VOLATILE_HEADER_NAMES: Final[frozenset[str]] = frozenset(
     {
@@ -123,6 +122,12 @@ class CanonicalRequest:
 
 
 def canonicalize(request: RecordedRequest) -> CanonicalRequest:
+    if request.strict_identity is not None:
+        return CanonicalRequest(
+            method=request.method,
+            path=request.path,
+            content=json.dumps(request.strict_identity.model_dump(mode="json"), sort_keys=True, separators=(",", ":")),
+        )
     file_identity: Final[JsonValue | None] = (
         None
         if request.file_name is None and request.file_sha256 is None

tests/e2e/fixture_mode.py

@@ -26,6 +26,7 @@ from fixture_bundle import (
     check_freshness,
     format_age,
 )
+from fixture_profile import match_profile
 
 type FixtureMode = Literal["live", "record", "replay"]
 
@@ -82,6 +83,7 @@ def fixture_mode_collection_error(mode_raw: str, bundle_dir: Path, *, now: datet
     Called at collection time (conftest pytest_sessionstart) so a stale or missing
     bundle fails the whole run up front, naming the bundle age, instead of failing
     every test individually."""
+    match_profile()
     mode = parse_fixture_mode(mode_raw)
     match mode:
         case InvalidFixtureMode(value=value):
@@ -89,7 +91,7 @@ def fixture_mode_collection_error(mode_raw: str, bundle_dir: Path, *, now: datet
         case "live" | "record":
             return None
         case "replay":
-            freshness = check_freshness(bundle_dir, now=now)
+            freshness = check_freshness(bundle_dir, now=now, profile=match_profile())
             match freshness:
                 case FreshBundle():
                     return None
@@ -110,6 +112,7 @@ def fixture_mode_collection_error(mode_raw: str, bundle_dir: Path, *, now: datet
 def fixture_report_lines(mode_raw: str, bundle_dir: Path, *, now: datetime) -> list[str]:
     """pytest report-header lines; empty in live mode so an unset
     E2E_FIXTURE_MODE keeps today's output byte-identical."""
+    match_profile()
     mode = parse_fixture_mode(mode_raw)
     match mode:
         case InvalidFixtureMode() | "live":
@@ -117,7 +120,7 @@ def fixture_report_lines(mode_raw: str, bundle_dir: Path, *, now: datetime) -> l
         case "record":
             return [f"e2e fixture mode: record -> {bundle_dir}"]
         case "replay":
-            freshness = check_freshness(bundle_dir, now=now)
+            freshness = check_freshness(bundle_dir, now=now, profile=match_profile())
             match freshness:
                 case FreshBundle(manifest=manifest):
                     return [

tests/e2e/fixture_profile.py

@@ -0,0 +1,179 @@
+from __future__ import annotations
+
+import json
+import os
+from collections.abc import Mapping
+from dataclasses import dataclass
+from typing import Final, Literal
+from urllib.parse import parse_qsl, urlsplit
+
+from pydantic import BaseModel, JsonValue, TypeAdapter
+
+type MatchProfile = Literal["legacy", "stateless_v1"]
+
+
+@dataclass(frozen=True, slots=True)
+class NumberToken:
+    literal: str
+
+
+type ExactJson = dict[str, ExactJson] | list[ExactJson] | str | bool | NumberToken | None
+
+SEMANTIC_HEADERS: Final = frozenset({"content-type", "accept", "anthropic-version", "anthropic-beta", "openai-beta"})
+AUTH_HEADERS: Final = frozenset({"authorization", "x-api-key"})
+EXCLUDED_HEADERS: Final = frozenset(
+    {
+        "host",
+        "content-length",
+        "transfer-encoding",
+        "connection",
+        "accept-encoding",
+        "user-agent",
+        "traceparent",
+        "tracestate",
+        "x-request-id",
+        "x-client-request-id",
+        "cookie",
+    }
+)
+CREDENTIAL_QUERY: Final = frozenset(
+    {
+        "api_key",
+        "api-key",
+        "apikey",
+        "key",
+        "token",
+        "access_token",
+        "signature",
+        "password",
+        "secret",
+        "credentials",
+        "authorization",
+        "sig",
+        "client_secret",
+        "aws_access_key_id",
+        "aws_secret_access_key",
+        "aws_session_token",
+    }
+)
+JSON_VALUE: Final[TypeAdapter[ExactJson]] = TypeAdapter(ExactJson)
+
+
+def match_profile() -> MatchProfile:
+    raw: Final = os.environ.get("E2E_REPLAY_MATCH_PROFILE", "legacy")
+    if raw in ("legacy", "stateless_v1"):
+        return raw
+    raise ValueError("E2E_REPLAY_MATCH_PROFILE must be legacy or stateless_v1")
+
+
+class StrictIdentity(BaseModel):
+    upstream: str
+    mount: str
+    query: tuple[tuple[str, str], ...]
+    headers: dict[str, str]
+    auth: dict[str, str]
+    body_present: bool
+    body: JsonValue
+
+
+@dataclass(frozen=True, slots=True)
+class IneligibleRequest:
+    reason: str
+
+
+def _unique_object(pairs: list[tuple[str, ExactJson]]) -> dict[str, ExactJson]:
+    if len({key for key, _ in pairs}) != len(pairs):
+        raise ValueError("duplicate JSON object keys")
+    return dict(pairs)
+
+
+def _invalid_constant(value: str) -> ExactJson:
+    raise ValueError("nonfinite JSON number")
+
+
+def _exact_value(value: ExactJson) -> JsonValue:
+    match value:
+        case dict():
+            return {"object": {key: _exact_value(item) for key, item in value.items()}}
+        case list():
+            return {"array": [_exact_value(item) for item in value]}
+        case bool():
+            return {"boolean": value}
+        case NumberToken(literal=literal):
+            return {"number": literal}
+        case str():
+            return {"string": value}
+        case None:
+            return None
+
+
+def strict_identity(
+    *,
+    method: str,
+    path: str,
+    query: str,
+    headers: Mapping[str, str],
+    body: bytes | None,
+    mount: str,
+    upstream_base: str,
+) -> StrictIdentity | IneligibleRequest:
+    if (mount, path, method.upper()) not in {
+        ("openai", "/openai/v1/chat/completions", "POST"),
+        ("anthropic", "/anthropic/v1/messages", "POST"),
+    }:
+        return IneligibleRequest("unsupported endpoint or method")
+    lowered: Final = {key.lower(): value for key, value in headers.items()}
+    if len(lowered) != len(headers):
+        return IneligibleRequest("duplicate header names")
+    if any(
+        key not in SEMANTIC_HEADERS | AUTH_HEADERS | EXCLUDED_HEADERS and not key.startswith("x-stainless-")
+        for key in lowered
+    ):
+        return IneligibleRequest("unsupported semantic header")
+    if "transfer-encoding" in lowered:
+        return IneligibleRequest("unsupported request transfer-encoding; send a content-length framed JSON body")
+    authorization: Final = lowered.get("authorization")
+    if authorization is not None and authorization.partition(" ")[0].lower() not in {"bearer", "basic", "digest"}:
+        return IneligibleRequest("unsupported authorization scheme")
+    destination: Final = urlsplit(upstream_base)
+    if destination.username or destination.password or destination.query or destination.fragment:
+        return IneligibleRequest("upstream destination contains credentials, query or fragment")
+    if destination.scheme not in ("http", "https") or not destination.netloc:
+        return IneligibleRequest("unsupported upstream destination")
+    if body and lowered.get("content-type", "").split(";", 1)[0].strip().lower() != "application/json":
+        return IneligibleRequest("unsupported body content-type; stateless_v1 requires JSON")
+    try:
+        parsed: Final = (
+            JSON_VALUE.validate_python(
+                json.loads(
+                    body,
+                    object_pairs_hook=_unique_object,
+                    parse_constant=_invalid_constant,
+                    parse_float=NumberToken,
+                    parse_int=NumberToken,
+                )
+            )
+            if body
+            else None
+        )
+    except (ValueError, UnicodeError):
+        return IneligibleRequest("invalid JSON or duplicate JSON object keys")
+    if body and not isinstance(parsed, dict):
+        return IneligibleRequest("stateless inference requires a JSON object")
+    try:
+        query_pairs: Final = tuple(parse_qsl(query, keep_blank_values=True, errors="strict"))
+    except UnicodeError:
+        return IneligibleRequest("invalid UTF-8 query encoding")
+    return StrictIdentity(
+        upstream=upstream_base,
+        mount=mount,
+        query=tuple((key, "<credential>" if key.lower() in CREDENTIAL_QUERY else value) for key, value in query_pairs),
+        headers={key: value for key, value in lowered.items() if key in SEMANTIC_HEADERS},
+        auth={
+            key: (value.partition(" ")[0].lower() if key == "authorization" else "present")
+            for key, value in lowered.items()
+            if key in AUTH_HEADERS
+        },
+        body_present=bool(body),
+        body=_exact_value(parsed),
+    )

tests/e2e/provider_edge.py

@@ -92,6 +92,7 @@ from fixture_mode import (
     current_test_key,
     parse_fixture_mode,
 )
+from fixture_profile import IneligibleRequest, MatchProfile, match_profile, strict_identity
 from pydantic import JsonValue, TypeAdapter
 
 EDGE_MOUNTS: Final[Mapping[str, str]] = MappingProxyType(
@@ -404,6 +405,12 @@ def _miss_message(test_key: str, slug: str, canonical: CanonicalRequest, bundle:
             f"under {slug}; re-record with E2E_FIXTURE_MODE=record"
         )
     closest, closest_file = _closest_recorded(canonical, recorded)
+    if bundle.manifest.match_profile == "stateless_v1":
+        expected: Final = _JSON.validate_json(closest.content)
+        actual: Final = _JSON.validate_json(canonical.content)
+        assert isinstance(expected, dict) and isinstance(actual, dict)
+        changed: Final = ", ".join(key for key in expected if expected[key] != actual.get(key))
+        return f"stateless_v1 replay mismatch: {changed or 'method/path'}; re-record with E2E_FIXTURE_MODE=record"
     diff: Final = "\n".join(
         islice(
             difflib.unified_diff(
@@ -785,11 +792,33 @@ def handle_edge_request(
     mount, _, upstream_path = split.path.lstrip("/").partition("/")
     upstream_base: Final = mounts.get(mount)
     if upstream_base is None:
-        return _text_reply(
-            404, f"unknown provider mount {mount!r}; known mounts: {', '.join(sorted(mounts))}"
+        return _text_reply(404, f"unknown provider mount {mount!r}; known mounts: {', '.join(sorted(mounts))}")
+    profile: Final = (
+        backend.recorder.profile
+        if isinstance(backend, RecordEdge)
+        else backend.source.bundle.manifest.match_profile
+        if isinstance(backend, ReplayEdge)
+        else "legacy"
+    )
+    identity: Final = (
+        strict_identity(
+            method=method,
+            path=split.path,
+            query=split.query,
+            headers=headers,
+            body=body,
+            mount=mount,
+            upstream_base=upstream_base,
         )
-    request: Final = edge_request(
-        method, split.path, split.query, body, _header_value(headers, "content-type")
+        if profile == "stateless_v1"
+        else None
+    )
+    if isinstance(identity, IneligibleRequest):
+        return _text_reply(REPLAY_MISS_STATUS, f"stateless_v1 eligibility error: {identity.reason}")
+    request: Final = (
+        RecordedRequest(method=method.lower(), path=split.path, headers={}, strict_identity=identity)
+        if identity is not None
+        else edge_request(method, split.path, split.query, body, _header_value(headers, "content-type"))
     )
     match backend:
         case LiveEdge():
@@ -837,6 +866,14 @@ class _EdgeHandler(BaseHTTPRequestHandler):
         body: Final = self.rfile.read(length) if length else None
         if edge_server.observation is not None:
             edge_server.observation.observe(body)
+        strict: Final = (
+            isinstance(edge_server.backend, RecordEdge) and edge_server.backend.recorder.profile == "stateless_v1"
+            or isinstance(edge_server.backend, ReplayEdge)
+            and edge_server.backend.source.bundle.manifest.match_profile == "stateless_v1"
+        )
+        if strict and len({name.lower() for name in self.headers.keys()}) != len(self.headers):
+            self._write_reply(_text_reply(REPLAY_MISS_STATUS, "stateless_v1 eligibility error: duplicate headers"))
+            return
         outcome: Final = handle_edge_request(
             edge_server.backend,
             edge_server.mounts,
@@ -955,16 +992,16 @@ def start_provider_edge(
 
 
 @functools.lru_cache(maxsize=8)
-def _shared_recorder(root: Path) -> BundleRecorder:
-    prepared = prepare_bundle(root)
+def _shared_recorder(root: Path, profile: MatchProfile = "legacy") -> BundleRecorder:
+    prepared = prepare_bundle(root, profile=profile)
     if isinstance(prepared, UnsafeBundleDir):
         raise ValueError(f"E2E_FIXTURE_DIR {prepared.path} {prepared.reason}")
     return prepared
 
 
 @functools.lru_cache(maxsize=8)
-def _shared_replay_source(root: Path) -> ReplaySource:
-    loaded = load_bundle(root)
+def _shared_replay_source(root: Path, profile: MatchProfile = "legacy") -> ReplaySource:
+    loaded = load_bundle(root, profile=profile)
     if isinstance(loaded, UnreadableBundle):
         raise ValueError(f"cannot replay from {root}: {loaded.reason}")
     return ReplaySource(bundle=loaded)
@@ -977,11 +1014,12 @@ def _shared_edge(
     bind_host: str,
     advertise_host: str,
     forward_timeout: float,
+    profile: MatchProfile,
 ) -> ProviderEdge:
     backend: Final[EdgeBackend] = (
-        RecordEdge(recorder=_shared_recorder(bundle_dir), lock=threading.Lock())
+        RecordEdge(recorder=_shared_recorder(bundle_dir, profile), lock=threading.Lock())
         if mode == "record"
-        else ReplayEdge(source=_shared_replay_source(bundle_dir))
+        else ReplayEdge(source=_shared_replay_source(bundle_dir, profile))
     )
     return start_provider_edge(
         backend,
@@ -998,7 +1036,7 @@ def replay_leftover_error(*, mode_raw: str, bundle_dir: Path, test_key: str) ->
     recording it no longer matches. Inert in every other mode."""
     if parse_fixture_mode(mode_raw) != "replay":
         return None
-    return _shared_replay_source(bundle_dir).leftover_error(test_key)
+    return _shared_replay_source(bundle_dir, match_profile()).leftover_error(test_key)
 
 
 def provider_edge_api_base(
@@ -1021,10 +1059,10 @@ def provider_edge_api_base(
             return None
         case "record" | "replay":
             if mount not in EDGE_MOUNTS:
-                raise ValueError(
-                    f"unknown provider mount {mount!r}; known mounts: {', '.join(sorted(EDGE_MOUNTS))}"
-                )
-            return _shared_edge(mode, bundle_dir, bind_host, advertise_host, forward_timeout).api_base(mount)
+                raise ValueError(f"unknown provider mount {mount!r}; known mounts: {', '.join(sorted(EDGE_MOUNTS))}")
+            return _shared_edge(mode, bundle_dir, bind_host, advertise_host, forward_timeout, match_profile()).api_base(
+                mount
+            )
         case _:
             assert_never(mode)
 
@@ -1037,9 +1075,9 @@ def _observed_backend(mode_raw: str, bundle_dir: Path) -> EdgeBackend:
         case "live":
             return LiveEdge()
         case "record":
-            return RecordEdge(_shared_recorder(bundle_dir), threading.Lock())
+            return RecordEdge(_shared_recorder(bundle_dir, match_profile()), threading.Lock())
         case "replay":
-            return ReplayEdge(_shared_replay_source(bundle_dir))
+            return ReplayEdge(_shared_replay_source(bundle_dir, match_profile()))
         case _:
             assert_never(mode)
 

tests/e2e/test_provider_edge.py

@@ -31,6 +31,7 @@ from concurrent.futures import ThreadPoolExecutor
 from contextlib import contextmanager
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
+from types import MappingProxyType
 from typing import Final
 
 import pytest
@@ -81,9 +82,14 @@ def json_object(body: bytes) -> dict[str, object]:
 class _FakeProvider(ThreadingHTTPServer):
     daemon_threads = True
 
-    def __init__(self, bind: tuple[str, int]) -> None:
+    def __init__(self, bind: tuple[str, int], *, echo_request: bool = True) -> None:
         super().__init__(bind, _FakeProviderHandler)
         self.hits: list[str] = []
+        self.echo_request = echo_request
+        self.requests: tuple[tuple[Mapping[str, str], bytes], ...] = ()
+
+    def capture_request(self, headers: Mapping[str, str], body: bytes) -> None:
+        self.requests = (*self.requests, (MappingProxyType(dict(headers)), body))
 
 
 class _FakeProviderHandler(BaseHTTPRequestHandler):
@@ -101,8 +107,11 @@ class _FakeProviderHandler(BaseHTTPRequestHandler):
         length = int(self.headers.get("content-length") or "0")
         body = self.rfile.read(length) if length else b""
         provider.hits.append(f"{self.command} {self.path}")
-        payload = json.dumps(
+        provider.capture_request(dict(self.headers.items()), body)
+        payload: Final = json.dumps(
             {"echo": body.decode("utf-8"), "path": self.path, "hit": len(provider.hits)}
+            if provider.echo_request
+            else {"ok": True}
         ).encode()
         self.send_response(200)
         self.send_header("content-type", "application/json")
@@ -117,8 +126,8 @@ class _FakeProviderHandler(BaseHTTPRequestHandler):
 
 
 @contextmanager
-def fake_provider() -> Generator[_FakeProvider]:
-    server = _FakeProvider(("127.0.0.1", 0))
+def fake_provider(*, echo_request: bool = True) -> Generator[_FakeProvider]:
+    server = _FakeProvider(("127.0.0.1", 0), echo_request=echo_request)
     thread = threading.Thread(target=server.serve_forever, daemon=True)
     thread.start()
     try: