test: add strict stateless provider replay identity

.github/workflows/test-code-quality.yml

@@ -83,6 +83,14 @@ jobs:
       - name: test_e2e_changed_gate
         run: uv run --no-sync pytest -q --noconftest -p no:cacheprovider -c /dev/null tests/code_coverage_tests/test_e2e_changed_gate.py tests/code_coverage_tests/test_e2e_idp_stack.py
 
+      - name: test_provider_replay_harness
+        run: |
+          pwd
+          uv run --no-sync pytest -q --noconftest -o addopts= -o pythonpath=tests/e2e -p no:rerunfailures \
+            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
+
       - name: router_code_coverage
         run: uv run --no-sync python ./tests/code_coverage_tests/router_code_coverage.py
 

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 `test_provider_replay_harness` code-quality step 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 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 @@ def record(self, *, test_key: str, request: RecordedRequest, response: RecordedR
         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 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 @@ def pretty_content(self) -> str:
 
 
 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 @@
     check_freshness,
     format_age,
 )
+from fixture_profile import match_profile
 
 type FixtureMode = Literal["live", "record", "replay"]
 
@@ -82,14 +83,15 @@ 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):
             return f"E2E_FIXTURE_MODE={value!r} is not one of {', '.join(FIXTURE_MODES)}"
         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,14 +112,15 @@ 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":
             return []
         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 [