feat(e2e): fail passed replays that leave recorded interactions unconsumed

tests/e2e/CLAUDE.md

@@ -77,7 +77,7 @@ Mark live tests with `@pytest.mark.e2e` (on the class or the module). Pure cover
 
 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
 
-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; 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 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
 
 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)
 

tests/e2e/conftest.py

@@ -15,7 +15,7 @@ shared fixtures build on it.
 
 import functools
 import os
-from collections.abc import Iterator
+from collections.abc import Generator, Iterator
 from datetime import datetime, timezone
 
 import pytest
@@ -27,6 +27,7 @@ from fixture_transport import (
     fixture_mode_collection_error,
     fixture_report_lines,
     parse_fixture_mode,
+    replay_leftover_error,
 )
 from junit_properties import attach_result_properties
 from lifecycle import ProxyClientProvider, ResourceManager
@@ -34,6 +35,7 @@ from proxy_client import ProxyClient, build_proxy_client
 
 
 _E2E_TEST_RAN = pytest.StashKey[bool]()
+_CALL_PASSED = pytest.StashKey[bool]()
 
 
 def pytest_configure(config: pytest.Config) -> None:
@@ -134,6 +136,36 @@ def pytest_runtest_call(item: pytest.Item) -> None:
     item.session.stash[_E2E_TEST_RAN] = True
 
 
+@pytest.hookimpl(wrapper=True)
+def pytest_runtest_makereport(
+    item: pytest.Item, call: pytest.CallInfo[None]
+) -> Generator[None, pytest.TestReport, pytest.TestReport]:
+    """Stash the call-phase outcome so teardown can tell a passed test from a
+    failed one without re-deriving it."""
+    report = yield
+    if report.when == "call":
+        item.stash[_CALL_PASSED] = report.passed
+    return report
+
+
+@pytest.hookimpl(wrapper=True)
+def pytest_runtest_teardown(item: pytest.Item) -> Generator[None, None, None]:
+    """In replay mode a passing test must consume its whole recording: leftover
+    interactions mean the test now makes fewer calls than it did at record time,
+    so the replay proved less than the bundle claims. The check runs after the
+    yield so fixture finalizers replay their recorded calls first. Failed tests
+    are left alone - their own failure already explains any unconsumed tail."""
+    result = yield
+    if not item.stash.get(_CALL_PASSED, False):
+        return result
+    reason = replay_leftover_error(
+        mode_raw=FIXTURE_MODE_RAW, bundle_dir=FIXTURE_DIR, test_key=item.nodeid
+    )
+    if reason is not None:
+        pytest.fail(reason)
+    return result
+
+
 def pytest_sessionfinish(session: pytest.Session, exitstatus: int) -> None:
     """Once the whole e2e session is done (all suites), optionally truncate the
     spend logs so the DB doesn't accumulate test rows. The truncate is destructive

tests/e2e/fixture_transport.py

@@ -332,6 +332,21 @@ class ReplaySource:
         self._cursors[slug] = index + 1
         return interaction
 
+    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):
+            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}; "
+            "re-record with E2E_FIXTURE_MODE=record"
+        )
+
 
 def _expect_result(interaction: Interaction) -> RecordedResult:
     match interaction.response:
@@ -474,6 +489,15 @@ def _shared_replay_source(root: Path) -> ReplaySource:
     return ReplaySource(bundle=loaded)
 
 
+def replay_leftover_error(*, mode_raw: str, bundle_dir: Path, test_key: str) -> str | None:
+    """Teardown-time completeness check: in replay mode a passed test with
+    unconsumed recorded interactions must fail instead of passing against a
+    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)
+
+
 def select_transport(
     live: Transport, *, mode_raw: str, bundle_dir: Path, master_key: str
 ) -> Transport:

tests/e2e/test_fixture_transport.py

@@ -50,6 +50,7 @@ from fixture_transport import (
     fixture_mode_collection_error,
     fixture_report_lines,
     parse_fixture_mode,
+    replay_leftover_error,
     select_transport,
 )
 from transport import Transport
@@ -354,6 +355,52 @@ class TestReplayTransport:
             replay.post("/model/new", headers=replay.master, json=Body(prompt="x"), response_type=Payload)
 
 
+class TestReplayLeftover:
+    def test_fully_consumed_recording_leaves_nothing(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)
+        source = replay_source(root)
+        replay: Transport = ReplayTransport(source=source, master_key="sk-1234")
+        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_unconsumed_trailing_interactions_name_the_next_call(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)
+        recording.probe("/health/liveliness", params=Query(q="1"))
+        source = replay_source(root)
+        replay: Transport = ReplayTransport(source=source, master_key="sk-1234")
+        replay.post("/model/new", headers=replay.master, json=Body(prompt="x"), response_type=Payload)
+        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 "re-record with E2E_FIXTURE_MODE=record" in error
+
+    def test_test_without_recordings_has_no_leftover(self, tmp_path: Path) -> None:
+        root = tmp_path / "bundle"
+        make_recorder(root)
+        assert replay_source(root).leftover_error("suite.py::test_never_recorded") is None
+
+    def test_inert_outside_replay_mode(self, tmp_path: Path) -> None:
+        missing = tmp_path / "missing"
+        assert replay_leftover_error(mode_raw="", bundle_dir=missing, test_key="k") is None
+        assert replay_leftover_error(mode_raw="record", bundle_dir=missing, test_key="k") is None
+
+    def test_replay_mode_reads_the_shared_bundle(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)
+        error = replay_leftover_error(mode_raw="replay", bundle_dir=root, test_key=current_test_key())
+        assert error is not None
+        assert "1 of 1 recorded interactions never consumed" in error
+
+
 class TestSelectTransport:
     def test_live_returns_the_live_transport_untouched(self, tmp_path: Path) -> None:
         fake = FakeTransport()