test(e2e): verify cached answers and upstream request count

tests/e2e/provider_edge.py

@@ -45,18 +45,16 @@ import hashlib
 import re
 import threading
 from collections import deque
-from collections.abc import Mapping, Sequence
-from contextlib import closing
+from collections.abc import Generator, Mapping, Sequence
+from contextlib import closing, contextmanager
 from dataclasses import dataclass, field
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from itertools import islice
 from pathlib import Path
 from types import MappingProxyType
-from typing import Final, Generator, Literal, assert_never
+from typing import Final, Literal, assert_never
 from urllib.parse import parse_qsl, urlsplit
 
-from pydantic import JsonValue, TypeAdapter
-
 from e2e_http import (
     NetworkError,
     StreamChunk,
@@ -94,6 +92,7 @@ from fixture_mode import (
     current_test_key,
     parse_fixture_mode,
 )
+from pydantic import JsonValue, TypeAdapter
 
 EDGE_MOUNTS: Final[Mapping[str, str]] = MappingProxyType(
     {
@@ -495,7 +494,29 @@ class ReplayEdge:
     source: ReplaySource
 
 
-type EdgeBackend = RecordEdge | ReplayEdge
+@dataclass(frozen=True, slots=True)
+class LiveEdge:
+    pass
+
+
+type EdgeBackend = RecordEdge | ReplayEdge | LiveEdge
+
+
+@dataclass(slots=True)
+class ProviderRequestObservation:
+    marker: str
+    _count: int = field(default=0, init=False)
+    _lock: threading.Lock = field(default_factory=threading.Lock, init=False)
+
+    def observe(self, body: bytes | None) -> None:
+        if body is not None and self.marker.encode() in body:
+            with self._lock:
+                self._count += 1
+
+    @property
+    def count(self) -> int:
+        with self._lock:
+            return self._count
 
 
 @dataclass(frozen=True, slots=True)
@@ -721,6 +742,24 @@ def _handle_record(
             assert_never(head)
 
 
+def _handle_live(
+    method: str, url: str, headers: Mapping[str, str], body: bytes | None, timeout: float
+) -> EdgeOutcome:
+    forwarded: Final = {
+        name: value for name, value in headers.items() if name.lower() not in _REQUEST_DROPPED_HEADERS
+    }
+    head: Final = forward_stream(method, url, headers=forwarded, body=body, timeout=timeout)
+    match head:
+        case NetworkError(message=message):
+            return _recorded_outcome(_network_error_response(message))
+        case StreamHead() if _is_streamed(head.headers):
+            return EdgeStream(head.status_code, _filtered_response_headers(head.headers), head.steps)
+        case StreamHead():
+            return _recorded_outcome(_drain_to_response(head))
+        case _:
+            assert_never(head)
+
+
 def _handle_replay(source: ReplaySource, request: RecordedRequest) -> EdgeOutcome:
     try:
         interaction: Final = source.next_interaction(request)
@@ -753,6 +792,10 @@ def handle_edge_request(
         method, split.path, split.query, body, _header_value(headers, "content-type")
     )
     match backend:
+        case LiveEdge():
+            return _handle_live(
+                method, _upstream_url(upstream_base, upstream_path, split.query), headers, body, timeout
+            )
         case RecordEdge():
             return _handle_record(
                 backend,
@@ -792,6 +835,8 @@ class _EdgeHandler(BaseHTTPRequestHandler):
         assert isinstance(edge_server, _EdgeHTTPServer)
         length: Final = int(self.headers.get("content-length") or "0")
         body: Final = self.rfile.read(length) if length else None
+        if edge_server.observation is not None:
+            edge_server.observation.observe(body)
         outcome: Final = handle_edge_request(
             edge_server.backend,
             edge_server.mounts,
@@ -857,11 +902,13 @@ class _EdgeHTTPServer(ThreadingHTTPServer):
         backend: EdgeBackend,
         mounts: Mapping[str, str],
         forward_timeout: float,
+        observation: ProviderRequestObservation | None,
     ) -> None:
         super().__init__(bind, _EdgeHandler)
         self.backend: Final = backend
         self.mounts: Final = mounts
         self.forward_timeout: Final = forward_timeout
+        self.observation: Final = observation
 
 
 @dataclass(frozen=True, slots=True)
@@ -890,13 +937,14 @@ def start_provider_edge(
     bind_host: str = "127.0.0.1",
     advertise_host: str | None = None,
     forward_timeout: float = 60.0,
+    observation: ProviderRequestObservation | None = None,
 ) -> RunningEdge:
     """Boot an edge server on an OS-assigned port in a daemon thread.
     ``advertise_host`` is what api_base URLs name (it differs from the bind
     host when the proxy runs in a container and reaches the host machine via
     a gateway address like host.docker.internal)."""
     server: Final = _EdgeHTTPServer(
-        (bind_host, 0), backend=backend, mounts=mounts, forward_timeout=forward_timeout
+        (bind_host, 0), backend=backend, mounts=mounts, forward_timeout=forward_timeout, observation=observation
     )
     thread: Final = threading.Thread(target=server.serve_forever, name="e2e-provider-edge", daemon=True)
     thread.start()
@@ -979,3 +1027,40 @@ def provider_edge_api_base(
             return _shared_edge(mode, bundle_dir, bind_host, advertise_host, forward_timeout).api_base(mount)
         case _:
             assert_never(mode)
+
+
+def _observed_backend(mode_raw: str, bundle_dir: Path) -> EdgeBackend:
+    mode: Final = parse_fixture_mode(mode_raw)
+    match mode:
+        case InvalidFixtureMode(value=value):
+            raise ValueError(f"E2E_FIXTURE_MODE={value!r} is not one of {', '.join(FIXTURE_MODES)}")
+        case "live":
+            return LiveEdge()
+        case "record":
+            return RecordEdge(_shared_recorder(bundle_dir), threading.Lock())
+        case "replay":
+            return ReplayEdge(_shared_replay_source(bundle_dir))
+        case _:
+            assert_never(mode)
+
+
+@contextmanager
+def observed_provider_edge(
+    observation: ProviderRequestObservation,
+    *,
+    mode_raw: str,
+    bundle_dir: Path,
+    bind_host: str,
+    advertise_host: str,
+    forward_timeout: float = 60.0,
+    mounts: Mapping[str, str] = EDGE_MOUNTS,
+) -> Generator[ProviderEdge, None, None]:
+    running: Final = start_provider_edge(
+        _observed_backend(mode_raw, bundle_dir), mounts=mounts,
+        bind_host=bind_host, advertise_host=advertise_host,
+        forward_timeout=forward_timeout, observation=observation,
+    )
+    try:
+        yield running.edge
+    finally:
+        running.shutdown()

tests/e2e/router/test_reliability_cache_e2e.py

@@ -1,37 +1,98 @@
-"""Live e2e: the response cache returns a cached answer on an exact repeat.
+"""An exact cache hit preserves the full choices and usage without another provider call.
 
-The same unique prompt is sent twice to the real `gpt-5.5` deployment under the
-same key: the first call is a cache miss (the proxy computes and stores the entry,
-and returns no x-litellm-cache-key), the second is an exact hit (the proxy serves
-from cache and returns x-litellm-cache-key). This relies on the standard Redis
-response cache being enabled on the proxy under test.
+Response IDs, creation timestamps and proxy headers are transport metadata;
+compare every field within choices and usage, including provider extensions.
 """
 
 from __future__ import annotations
 
+from typing import Final
+
 import pytest
-
 from complexity_router_client import ComplexityRouterClient
-from e2e_config import unique_marker
-from reliability_support import chat_override
+from e2e_config import (
+    FIXTURE_DIR,
+    FIXTURE_MODE_RAW,
+    PROVIDER_EDGE_ADVERTISE_HOST,
+    PROVIDER_EDGE_BIND_HOST,
+    REQUEST_TIMEOUT,
+    unique_marker,
+)
+from lifecycle import ResourceManager
+from models import ChatBody, ChatMessage, ChatResponse, LiteLLMParamsBody
+from provider_edge import ProviderRequestObservation, observed_provider_edge
+from pydantic import BaseModel, JsonValue
 
-pytestmark = pytest.mark.e2e
+pytestmark = [pytest.mark.e2e, pytest.mark.replayable]
+
+
+class _CacheChatBody(ChatBody):
+    ttl: int = 600
+
+
+class _CachedAnswer(BaseModel):
+    model: str
+    choices: tuple[dict[str, JsonValue], ...]
+    usage: dict[str, JsonValue]
 
 
 class TestReliabilityCache:
     @pytest.mark.covers("reliability.cache.exact.returns_cached")
-    def test_exact_cache_returns_cached(self, client: ComplexityRouterClient, scoped_key: str) -> None:
-        prompt = f"cache probe {unique_marker()}"
+    def test_exact_cache_returns_cached(
+        self, client: ComplexityRouterClient, resources: ResourceManager, scoped_key: str
+    ) -> None:
+        marker: Final = unique_marker()
+        model: Final = f"e2e-cache-{marker}"
+        prompt: Final = f"Reply with a short sentence about a blue lantern. Request marker: {marker}"
+        observation: Final = ProviderRequestObservation(marker)
 
-        first = chat_override(client.proxy, scoped_key, "gpt-5.5", prompt, cache=None)
-        assert first.status_code == 200, f"first call should succeed, got {first.status_code}: {first.body[:300]}"
-        assert "x-litellm-cache-key" not in first.headers, (
-            "first (uncached) call must not report a cache-key header"
-        )
+        with observed_provider_edge(
+            observation,
+            mode_raw=FIXTURE_MODE_RAW,
+            bundle_dir=FIXTURE_DIR,
+            bind_host=PROVIDER_EDGE_BIND_HOST,
+            advertise_host=PROVIDER_EDGE_ADVERTISE_HOST,
+            forward_timeout=REQUEST_TIMEOUT,
+        ) as edge:
+            model_id: Final = client.proxy.create_model(
+                model,
+                LiteLLMParamsBody(
+                    model="openai/gpt-5.6",
+                    api_key="os.environ/OPENAI_API_KEY",
+                    api_base=f"{edge.api_base('openai')}/v1",
+                ),
+            )
+            resources.defer(lambda: client.proxy.delete_model(model_id))
+            body: Final = _CacheChatBody(
+                model=model,
+                messages=[ChatMessage(role="user", content=prompt)],
+                max_completion_tokens=512,
+                reasoning_effort="none",
+                cache=None,
+            )
+            first: Final = client.proxy.transport.send(
+                "/chat/completions", headers=client.proxy.transport.bearer(scoped_key), json=body
+            )
+            assert first.status_code == 200, f"first call should succeed, got {first.status_code}: {first.body[:300]}"
+            assert "x-litellm-cache-key" not in first.headers, "first call must be a cache miss"
+            answer: Final = ChatResponse.model_validate_json(first.body)
+            assert len(answer.choices) == 1
+            choice: Final = answer.choices[0]
+            assert choice.message is not None and choice.message.role == "assistant"
+            assert choice.message.content is not None and choice.message.content.strip(), "first answer is empty"
+            assert choice.finish_reason == "stop"
+            assert answer.usage is not None
+            assert answer.usage.prompt_tokens is not None and answer.usage.prompt_tokens > 0
+            assert answer.usage.completion_tokens is not None and answer.usage.completion_tokens > 0
+            assert answer.usage.total_tokens == answer.usage.prompt_tokens + answer.usage.completion_tokens
+            assert observation.count == 1, "first miss must invoke the provider exactly once"
 
-        second = chat_override(client.proxy, scoped_key, "gpt-5.5", prompt, cache=None)
-        assert second.status_code == 200, f"second call should succeed, got {second.status_code}: {second.body[:300]}"
-        assert "x-litellm-cache-key" in second.headers, (
-            "second identical call should hit the response cache and report a cache-key header "
-            "(requires the proxy's Redis response cache to be enabled)"
-        )
+            second: Final = client.proxy.transport.send(
+                "/chat/completions", headers=client.proxy.transport.bearer(scoped_key), json=body
+            )
+            assert second.status_code == 200, f"second call should succeed, got {second.status_code}: {second.body[:300]}"
+            assert second.headers.get("x-litellm-cache-key"), "identical request must hit the response cache"
+            assert _CachedAnswer.model_validate_json(second.body) == _CachedAnswer.model_validate_json(first.body), (
+                "cache hit changed the answer, finish reason or usage"
+            )
+            assert observation.count == 1, "two successful requests must invoke the provider exactly once"

tests/e2e/test_provider_edge.py

@@ -34,10 +34,7 @@ from pathlib import Path
 from typing import Final
 
 import pytest
-from pydantic import TypeAdapter
-
 from e2e_http import RawResponse, StreamChunk, forward
-from fixture_canonical import canonicalize
 from fixture_bundle import (
     BundleRecorder,
     Interaction,
@@ -49,6 +46,7 @@ from fixture_bundle import (
     prepare_bundle,
     slug_for_test,
 )
+from fixture_canonical import canonicalize
 from fixture_mode import current_test_key
 from provider_edge import (
     REPLAY_MISS_STATUS,
@@ -56,15 +54,18 @@ from provider_edge import (
     EdgeReply,
     EdgeStream,
     ProviderEdge,
+    ProviderRequestObservation,
     RecordEdge,
     ReplayEdge,
     ReplaySource,
     edge_request,
     handle_edge_request,
+    observed_provider_edge,
     provider_edge_api_base,
     replay_leftover_error,
     start_provider_edge,
 )
+from pydantic import TypeAdapter
 
 CHAT_PATH = "/openai/v1/chat/completions"
 UPLOAD_PATH = "/openai/v1/files"
@@ -1290,3 +1291,62 @@ class TestApiBaseSeam:
         assert second.endswith("/anthropic")
         assert first.rsplit("/", 1)[0] == second.rsplit("/", 1)[0]
         assert (root / "manifest.json").is_file()
+
+
+class TestProviderRequestObservation:
+    def test_live_counts_repeated_marker_calls_without_recording(self, tmp_path: Path) -> None:
+        observation: Final = ProviderRequestObservation("observed-lantern")
+        with fake_provider() as provider:
+            with observed_provider_edge(
+                observation, mode_raw="live", bundle_dir=tmp_path / "unused",
+                bind_host="127.0.0.1", advertise_host="127.0.0.1",
+                mounts={"openai": provider_url(provider)},
+            ) as edge:
+                assert observation.count == 0
+                unrelated: Final = call_edge(edge, "POST", CHAT_PATH, body=chat_body("other-lantern"))
+                assert unrelated.status_code == 200
+                assert observation.count == 0
+                first: Final = call_edge(edge, "POST", CHAT_PATH, body=chat_body("observed-lantern"))
+                assert first.status_code == 200
+                assert json_object(first.body)["echo"] == chat_body("observed-lantern").decode()
+                assert observation.count == 1
+                second: Final = call_edge(edge, "POST", CHAT_PATH, body=chat_body("observed-lantern"))
+                assert second.status_code == 200
+                assert observation.count == 2
+            assert len(provider.hits) == 3
+        assert not (tmp_path / "unused").exists()
+
+    def test_record_and_replay_count_each_matching_call(self, tmp_path: Path) -> None:
+        with fake_provider() as provider:
+            for mode, observation in (
+                ("record", ProviderRequestObservation("observed-lantern")),
+                ("replay", ProviderRequestObservation("observed-lantern")),
+            ):
+                with observed_provider_edge(
+                    observation, mode_raw=mode, bundle_dir=tmp_path / "bundle",
+                    bind_host="127.0.0.1", advertise_host="127.0.0.1",
+                    mounts={"openai": provider_url(provider)},
+                ) as edge:
+                    assert observation.count == 0
+                    for expected, response in (
+                        (index, call_edge(edge, "POST", CHAT_PATH, body=chat_body("observed-lantern")))
+                        for index in (1, 2)
+                    ):
+                        assert response.status_code == 200
+                        assert json_object(response.body)["hit"] == expected
+                        assert observation.count == expected
+                assert len(provider.hits) == 2
+        assert replay_leftover_error(
+            mode_raw="replay", bundle_dir=tmp_path / "bundle", test_key=current_test_key()
+        ) is None
+
+    def test_failed_provider_attempt_is_counted(self, tmp_path: Path) -> None:
+        observation: Final = ProviderRequestObservation("observed-lantern")
+        with observed_provider_edge(
+            observation, mode_raw="live", bundle_dir=tmp_path / "unused",
+            bind_host="127.0.0.1", advertise_host="127.0.0.1",
+            mounts={"openai": "http://127.0.0.1:9"},
+        ) as edge:
+            response: Final = call_edge(edge, "POST", CHAT_PATH, body=chat_body("observed-lantern"))
+            assert response.status_code == 502
+            assert observation.count == 1