Merge pull request #41346 from BerriAI/litellm_e2e_provider_cache

feat(e2e): reuse exact provider responses for 24 hours

.circleci/config.yml

@@ -9,7 +9,7 @@ commands:
     parameters:
       category:
         type: enum
-        enum: ["backend", "client"]
+        enum: ["backend", "client", "provider-harness"]
         default: "backend"
     steps:
       - run:
@@ -2918,19 +2918,30 @@ jobs:
   provider_replay_harness:
     docker:
       - *python312_image
+      - image: redis@sha256:e2debfb7956fa12c7ddc79d7e645c8cf26b30c99a6e9161ea9bf4171e1668a5f
     working_directory: ~/project
     resource_class: medium
+    environment:
+      E2E_CACHE_TEST_REDIS_URL: redis://127.0.0.1:6379/0
+      E2E_PROVIDER_CACHE: "0"
+      E2E_FIXTURE_MODE: live
     steps:
+      - checkout
+      - skip_if_unrelated_changes:
+          category: provider-harness
       - setup_litellm_test_deps
+      - wait_for_service:
+          url: tcp://localhost:6379
       - run:
-          name: Test provider replay harness
+          name: Test provider capture and 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
+              tests/code_coverage_tests/test_provider_replay_harness.py \
+              tests/code_coverage_tests/test_provider_cache.py
       - store_test_results:
           path: test-results/provider-replay-harness
 

.circleci/scripts/classify_changes.sh

@@ -1,13 +1,19 @@
 #!/usr/bin/env bash
 set -uo pipefail
 
-category="${1:?usage: classify_changes.sh <backend|client|ui>}"
+category="${1:?usage: classify_changes.sh <backend|client|ui|provider-harness>}"
 
 has_client=false
 has_backend=false
 has_ci=false
+has_provider_harness=false
 while IFS= read -r file || [ -n "$file" ]; do
   [ -n "$file" ] || continue
+  case "$file" in
+    tests/e2e/*/*.py) : ;;
+    tests/e2e/*.py | tests/code_coverage_tests/test_provider_cache.py | tests/code_coverage_tests/test_provider_replay_harness.py | tests/test_litellm/test_circleci_path_filter.py | .circleci/* | pyproject.toml | uv.lock)
+      has_provider_harness=true ;;
+  esac
   case "$file" in
     ui/* | tests/e2e/ui/*) has_client=true ;;
     docs/* | *.md | *.mdx) : ;;
@@ -17,6 +23,9 @@ while IFS= read -r file || [ -n "$file" ]; do
 done
 
 case "$category" in
+  provider-harness)
+    [ "$has_provider_harness" = true ] && echo run || echo skip
+    ;;
   backend)
     [ "$has_backend" = true ] && echo run || echo skip
     ;;

.circleci/scripts/path_filter.sh

@@ -1,7 +1,7 @@
 #!/usr/bin/env bash
 set -uo pipefail
 
-category="${1:?usage: path_filter.sh <backend|client>}"
+category="${1:?usage: path_filter.sh <backend|client|provider-harness>}"
 here="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
 
 run_full() {
@@ -36,5 +36,5 @@ if [ "$decision" = run ]; then
   run_full "$category-relevant changes detected"
 fi
 
-echo "path-filter[$category]: only unrelated (docs/client) changes detected; halting job as successful"
+echo "path-filter[$category]: only unrelated changes detected; halting job as successful"
 circleci-agent step halt

tests/code_coverage_tests/test_provider_cache.py

@@ -0,0 +1,461 @@
+from __future__ import annotations
+
+import os
+import shutil
+import socket
+import subprocess
+import threading
+import time
+import uuid
+from collections.abc import Generator
+from concurrent.futures import ThreadPoolExecutor
+from contextlib import contextmanager
+from dataclasses import dataclass, replace
+from http.client import HTTPConnection
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from typing import Final
+from urllib.parse import urlsplit
+
+import pytest
+from e2e_http import NetworkError, PreparedForward, RawResponse, StreamChunk, StreamHead, forward, prepare_forward
+from models import LiteLLMParamsBody
+from provider_cache import CacheEdge, CacheHit, CaptureLease, exact_key, successful_response
+from provider_cache_redis import PUBLISH, RedisCommands, RedisResponseStore, configured_cache, redis_store
+from provider_cache_routing import LIVE_PROVIDER_REQUIRED, route_cache_model
+from provider_edge import configured_cache_backend, start_provider_edge
+from redis.exceptions import ConnectionError as RedisConnectionError
+
+SECRET: Final = b"synthetic-cache-hmac-key-for-tests"
+BODY: Final = b'{"model":"test","messages":[{"role":"user","content":"hello"}]}'
+SUCCESS: Final = b'{"id":"provider-fixed-id","choices":[{"message":{"content":"hello"},"finish_reason":"stop"}],"usage":{"prompt_tokens":1,"completion_tokens":1,"total_tokens":2}}'
+HEADERS: Final = {"content-type": "application/json", "authorization": "Bearer synthetic-account-one"}
+
+
+class Provider(ThreadingHTTPServer):
+    hits: tuple[tuple[str, bytes], ...] = ()
+    response: bytes = SUCCESS
+    status: int = 200
+    delay: float = 0
+    stream: bool = False
+    truncated: bool = False
+
+
+class Handler(BaseHTTPRequestHandler):
+    protocol_version = "HTTP/1.1"
+
+    def do_POST(self) -> None:
+        server: Final = self.server
+        assert isinstance(server, Provider)
+        body: Final = self.rfile.read(int(self.headers.get("content-length", "0")))
+        server.hits += ((self.path, body),)
+        time.sleep(server.delay)
+        self.send_response(server.status)
+        if server.stream:
+            self.send_header("content-type", "text/event-stream")
+            self.send_header("transfer-encoding", "chunked")
+            self.end_headers()
+            self.wfile.write(b"%x\r\n%s\r\n" % (len(server.response), server.response))
+            if server.truncated:
+                self.close_connection = True
+                return
+            self.wfile.write(b"0\r\n\r\n")
+            return
+        self.send_header("content-type", "application/json")
+        self.send_header("content-length", str(len(server.response)))
+        self.end_headers()
+        self.wfile.write(server.response)
+
+    def log_message(self, format: str, *args: object) -> None:
+        pass
+
+
+@pytest.fixture
+def provider() -> Generator[Provider, None, None]:
+    server: Final = Provider(("127.0.0.1", 0), Handler)
+    thread: Final = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    try:
+        yield server
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join(timeout=5)
+
+
+@pytest.fixture(scope="module")
+def redis_url(tmp_path_factory: pytest.TempPathFactory) -> Generator[str, None, None]:
+    configured: Final = os.environ.get("E2E_CACHE_TEST_REDIS_URL")
+    if configured:
+        yield configured
+        return
+    binary: Final = shutil.which("redis-server")
+    assert binary is not None, "Set E2E_CACHE_TEST_REDIS_URL or install Redis for cache integration checks"
+    root: Final = tmp_path_factory.mktemp("provider-cache-redis")
+    with socket.socket() as probe:
+        probe.bind(("127.0.0.1", 0))
+        port: Final = probe.getsockname()[1]
+    with (root / "redis.log").open("wb") as log:
+        process: Final = subprocess.Popen(
+            [binary, "--bind", "127.0.0.1", "--port", str(port), "--save", "", "--appendonly", "no", "--dir", str(root)],
+            stdout=log, stderr=subprocess.STDOUT,
+        )
+        try:
+            deadline: Final = time.monotonic() + 5
+            while True:
+                try:
+                    with socket.create_connection(("127.0.0.1", port), timeout=0.1):
+                        break
+                except OSError:
+                    assert process.poll() is None and time.monotonic() < deadline
+                    time.sleep(0.02)
+            yield f"redis://127.0.0.1:{port}/0"
+        finally:
+            process.terminate()
+            process.wait(timeout=5)
+
+
+@pytest.fixture
+def store(redis_url: str) -> RedisResponseStore:
+    return redis_store(redis_url, "test-" + uuid.uuid4().hex)
+
+
+@contextmanager
+def edge(cache: CacheEdge, provider: Provider) -> Generator[str, None, None]:
+    upstream: Final = f"http://127.0.0.1:{provider.server_port}"
+    running: Final = start_provider_edge(cache, mounts={"openai": upstream})
+    try:
+        yield running.edge.api_base("openai") + "/v1/chat/completions"
+    finally:
+        running.shutdown()
+
+
+def call(url: str, body: bytes = BODY, headers: dict[str, str] = HEADERS) -> RawResponse:
+    result: Final = forward("POST", url, headers=headers, body=body, timeout=5)
+    assert isinstance(result, RawResponse), result
+    return result
+
+
+def test_success_is_reusable_across_fresh_edges(store: RedisResponseStore, provider: Provider) -> None:
+    with edge(CacheEdge(store, SECRET), provider) as url:
+        assert call(url).body == SUCCESS
+        assert call(url).body == SUCCESS
+    with edge(CacheEdge(store, SECRET), provider) as other:
+        assert call(other).body == SUCCESS
+    assert len(provider.hits) == 1
+
+
+@pytest.mark.parametrize("body", [BODY + b" ", BODY.replace(b"hello", b"Hello"), BODY.replace(b"test", b"test2")])
+def test_any_body_change_calls_live(store: RedisResponseStore, provider: Provider, body: bytes) -> None:
+    with edge(CacheEdge(store, SECRET), provider) as url:
+        call(url)
+        call(url, body)
+        call(url, body)
+    assert len(provider.hits) == 2
+
+
+@pytest.mark.parametrize("name,value", [("authorization", "Bearer another-account"), ("x-request-id", "one"), ("anthropic-version", "new")])
+def test_changed_header_cannot_reuse(store: RedisResponseStore, provider: Provider, name: str, value: str) -> None:
+    with edge(CacheEdge(store, SECRET), provider) as url:
+        call(url)
+        call(url, headers=HEADERS | {name: value})
+        call(url + "?x=1")
+    assert len(provider.hits) == 3
+
+
+@pytest.mark.parametrize("status,response", [(429, b'{"error":"rate limited"}'), (500, b'failed'), (200, b'{"error":"bad"}'), (200, b'not json')])
+def test_failed_provider_responses_never_enter_cache(store: RedisResponseStore, provider: Provider, status: int, response: bytes) -> None:
+    provider.status = status
+    provider.response = response
+    with edge(CacheEdge(store, SECRET), provider) as url:
+        assert call(url).status_code == status
+        assert call(url).body == response
+    assert len(provider.hits) == 2
+
+
+def test_expiry_does_not_slide(store: RedisResponseStore, provider: Provider) -> None:
+    short: Final = replace(store, lifetime_ms=250)
+    with edge(CacheEdge(short, SECRET), provider) as url:
+        call(url)
+        call(url)
+        time.sleep(0.3)
+        call(url)
+        call(url)
+    assert len(provider.hits) == 2
+
+
+def test_concurrent_requests_publish_atomically(store: RedisResponseStore, provider: Provider) -> None:
+    provider.delay = 0.15
+    with edge(CacheEdge(store, SECRET), provider) as url:
+        with ThreadPoolExecutor(max_workers=5) as executor:
+            replies: Final = tuple(executor.map(lambda _: call(url).body, range(5)))
+    assert replies == (SUCCESS,) * 5
+    assert len(provider.hits) == 1
+
+
+@pytest.mark.parametrize("age_past_expiry_ms", [0, 1])
+def test_expired_response_is_rejected_without_physical_eviction(
+    store: RedisResponseStore, age_past_expiry_ms: int,
+) -> None:
+    response_key: Final = store.keys("expired")[0]
+    retained: Final = store.client.eval(
+        """
+local clock = redis.call('TIME')
+local expires = clock[1] * 1000 + math.floor(clock[2] / 1000) - tonumber(ARGV[1])
+redis.call('HSET', KEYS[1], 'captured', expires - 86400000, 'expires', expires, 'payload', 'old-response')
+return redis.call('PTTL', KEYS[1])
+""",
+        1, response_key, age_past_expiry_ms,
+    )
+    assert retained == -1
+    replacement: Final = store.lookup("expired")
+    assert isinstance(replacement, CaptureLease)
+    assert replacement.expires_at_ms - replacement.captured_at_ms == 86_400_000
+    assert store.publish("expired", replacement, b"fresh-response")
+    hit: Final = store.lookup("expired")
+    assert isinstance(hit, CacheHit) and hit.payload == b"fresh-response"
+
+
+@pytest.mark.parametrize("truncated", [False, True])
+def test_stream_completion_controls_publication(store: RedisResponseStore, provider: Provider, truncated: bool) -> None:
+    provider.stream = True
+    provider.truncated = truncated
+    provider.response = b'data: {"choices":[{"index":0,"delta":{"content":"hello"},"finish_reason":"stop"}]}\n\ndata: [DONE]\n\n'
+    with edge(CacheEdge(store, SECRET), provider) as url:
+        for _ in range(2):
+            result: Final = forward("POST", url, headers=HEADERS, body=BODY, timeout=5)
+            if truncated:
+                assert isinstance(result, NetworkError)
+            else:
+                assert isinstance(result, RawResponse) and result.body == provider.response
+    assert len(provider.hits) == (2 if truncated else 1)
+
+
+def test_store_outage_preserves_provider_success(provider: Provider) -> None:
+    with socket.socket() as probe:
+        probe.bind(("127.0.0.1", 0))
+        port: Final = probe.getsockname()[1]
+        unavailable: Final = redis_store(f"redis://127.0.0.1:{port}/0", "unavailable")
+        with edge(CacheEdge(unavailable, SECRET), provider) as url:
+            assert call(url).body == SUCCESS
+            assert call(url).body == SUCCESS
+    assert len(provider.hits) == 2
+
+
+def test_old_lease_cannot_overwrite_new_owner(store: RedisResponseStore) -> None:
+    short: Final = replace(store, lease_ms=50)
+    old: Final = short.lookup("key")
+    assert isinstance(old, CaptureLease)
+    time.sleep(0.08)
+    current: Final = short.lookup("key")
+    assert isinstance(current, CaptureLease)
+    assert not short.publish("key", old, b"old")
+    assert short.publish("key", current, b"new")
+    hit: Final = short.lookup("key")
+    assert isinstance(hit, CacheHit) and hit.payload == b"new"
+
+
+def test_identity_preserves_values_and_never_contains_credentials() -> None:
+    variants: Final = (b'{}', b'{"a":null}', b'{"a":false}', b'{"a":0}', b'{"a":0.0}', b'{"a":"0"}', b' { }', None, b'')
+    keys: Final = tuple(exact_key(SECRET, "POST", "https://example.invalid/v1/chat/completions", HEADERS, body) for body in variants)
+    assert len(set(keys)) == len(variants)
+    assert all(len(key) == 64 and "synthetic-account" not in key for key in keys)
+
+
+@pytest.mark.parametrize("payload", [b"corrupt response", '{"response":"{}","signature":"é"}'.encode()])
+def test_corrupt_entry_is_replaced_by_same_successful_request(store: RedisResponseStore, provider: Provider, payload: bytes) -> None:
+    upstream: Final = f"http://127.0.0.1:{provider.server_port}/v1/chat/completions"
+    prepared: Final = prepare_forward("POST", upstream, HEADERS, BODY)
+    assert isinstance(prepared, PreparedForward)
+    key: Final = exact_key(SECRET, "POST", upstream, prepared.headers, BODY)
+    lease: Final = store.lookup(key)
+    assert isinstance(lease, CaptureLease)
+    assert store.publish(key, lease, payload)
+    cache: Final = CacheEdge(store, SECRET)
+    for _ in range(2):
+        head = cache.forward("POST", upstream, HEADERS, BODY, 5)
+        assert isinstance(head, StreamHead)
+        assert b"".join(step.data for step in head.steps if isinstance(step, StreamChunk)) == SUCCESS
+    assert len(provider.hits) == 1
+    assert dict(cache.counters.counts) == {
+        "corrupt": 1, "misses": 1, "upstream_attempts": 1, "writes": 1, "hits": 1,
+    }
+
+
+@pytest.mark.parametrize("payload", [
+    b'data: {}\n\ndata: [DONE]\n\n',
+    b'data: {"choices":[{"index":0,"delta":{}}]}\n\ndata: [DONE]\n\n',
+    b'data: {"choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}\n\ndata: [DONE]',
+    b'data: {"error":{"message":"failed"}}\n\ndata: [DONE]\n\n',
+])
+def test_malformed_success_stream_is_never_cached(store: RedisResponseStore, provider: Provider, payload: bytes) -> None:
+    provider.stream = True
+    provider.response = payload
+    with edge(CacheEdge(store, SECRET), provider) as url:
+        assert call(url).body == payload
+        assert call(url).body == payload
+    assert len(provider.hits) == 2
+
+
+def test_anthropic_stream_requires_start_finish_and_stop() -> None:
+    start: Final = b'data: {"type":"message_start","message":{}}\n\n'
+    finish: Final = b'data: {"type":"message_delta","delta":{"stop_reason":"end_turn"}}\n\n'
+    stop: Final = b'data: {"type":"message_stop"}\n\n'
+    url: Final = "https://example.invalid/v1/messages"
+    headers: Final = {"content-type": "text/event-stream"}
+    assert successful_response(url, 200, headers, start + finish + stop)
+    assert not successful_response(url, 200, headers, start + stop)
+    assert not successful_response(url, 200, headers, finish + stop)
+    assert not successful_response(url, 200, headers, start + finish)
+
+
+@pytest.mark.parametrize("provider,suffix", [("openai", "/v1"), ("anthropic", "")])
+def test_normal_registration_routes_supported_providers(provider: str, suffix: str) -> None:
+    params: Final = LiteLLMParamsBody(model=f"{provider}/test", api_key="os.environ/SYNTHETIC_KEY", timeout=12)
+    routed: Final = route_cache_model(params, lambda mount: f"http://edge.invalid/{mount}", enabled=True)
+    assert routed.api_base == f"http://edge.invalid/{provider}{suffix}"
+    assert routed.model_dump(exclude={"api_base"}) == params.model_dump(exclude={"api_base"})
+    assert params.api_base is None
+
+
+@pytest.mark.parametrize("params", [
+    LiteLLMParamsBody(model="bedrock/test"),
+    LiteLLMParamsBody(model="azure/test"),
+    LiteLLMParamsBody(model="openai/test", api_base="https://custom.invalid/v1"),
+    LiteLLMParamsBody(model="openai/test", api_base=""),
+    LiteLLMParamsBody(model="openai/test", litellm_credential_name="named-credential"),
+    LiteLLMParamsBody(model="openai/test", mock_response="synthetic"),
+])
+def test_registration_preserves_unsupported_or_explicit_routes(params: LiteLLMParamsBody) -> None:
+    def unexpected_edge(mount: str) -> str:
+        pytest.fail(f"should not start edge for {mount}")
+    assert route_cache_model(params, unexpected_edge, enabled=True) is params
+
+
+def test_rollback_and_live_only_policy_keep_direct_provider_route() -> None:
+    params: Final = LiteLLMParamsBody(model="openai/test")
+    assert route_cache_model(params, lambda _: "http://edge.invalid", enabled=False) is params
+    assert route_cache_model(params, lambda _: "http://edge.invalid", enabled=True, mode="realtime") is params
+    token: Final = LIVE_PROVIDER_REQUIRED.set(True)
+    try:
+        assert route_cache_model(params, lambda _: "http://edge.invalid", enabled=True) is params
+    finally:
+        LIVE_PROVIDER_REQUIRED.reset(token)
+    assert route_cache_model(params, lambda _: "http://edge.invalid", enabled=True).api_base == "http://edge.invalid/v1"
+
+
+@dataclass(frozen=True)
+class PublishOutage:
+    client: RedisCommands
+
+    def eval(self, script: str, numkeys: int, *args: str | bytes | int) -> object:
+        if script == PUBLISH:
+            raise RedisConnectionError("synthetic publication outage")
+        return self.client.eval(script, numkeys, *args)
+
+
+def test_write_outage_preserves_success_without_hidden_retry(store: RedisResponseStore, provider: Provider) -> None:
+    unavailable: Final = replace(store, client=PublishOutage(store.client))
+    cache: Final = CacheEdge(unavailable, SECRET)
+    with edge(cache, provider) as url:
+        assert call(url).body == SUCCESS
+        assert call(url).body == SUCCESS
+    assert len(provider.hits) == 2
+    assert dict(cache.counters.counts)["write_failures"] == 2
+    with edge(CacheEdge(store, SECRET), provider) as url:
+        assert call(url).body == SUCCESS
+        assert call(url).body == SUCCESS
+    assert len(provider.hits) == 3
+
+
+def test_connection_failure_releases_capture_lease(store: RedisResponseStore) -> None:
+    with socket.socket() as unavailable:
+        unavailable.bind(("127.0.0.1", 0))
+        url: Final = f"http://127.0.0.1:{unavailable.getsockname()[1]}/v1/chat/completions"
+        cache: Final = CacheEdge(store, SECRET)
+        assert isinstance(cache.forward("POST", url, HEADERS, BODY, 0.2), NetworkError)
+        prepared: Final = prepare_forward("POST", url, HEADERS, BODY)
+        assert isinstance(prepared, PreparedForward)
+        key: Final = exact_key(SECRET, "POST", url, prepared.headers, BODY)
+        slot: Final = store.lookup(key)
+        assert isinstance(slot, CaptureLease)
+        assert store.release(key, slot)
+        assert dict(cache.counters.counts)["rejected"] == 1
+
+
+def test_close_before_first_chunk_releases_lease(store: RedisResponseStore, provider: Provider) -> None:
+    url: Final = f"http://127.0.0.1:{provider.server_port}/v1/chat/completions"
+    cache: Final = CacheEdge(store, SECRET)
+    head: Final = cache.forward("POST", url, HEADERS, BODY, 5)
+    assert isinstance(head, StreamHead)
+    head.steps.close()
+    prepared: Final = prepare_forward("POST", url, HEADERS, BODY)
+    assert isinstance(prepared, PreparedForward)
+    key: Final = exact_key(SECRET, "POST", url, prepared.headers, BODY)
+    slot: Final = store.lookup(key)
+    assert isinstance(slot, CaptureLease)
+    assert store.release(key, slot)
+
+
+def test_effective_account_change_cannot_reuse_cache(
+    store: RedisResponseStore, provider: Provider, monkeypatch: pytest.MonkeyPatch, tmp_path,
+) -> None:
+    url: Final = f"http://127.0.0.1:{provider.server_port}/v1/chat/completions"
+    cache: Final = CacheEdge(store, SECRET)
+    for account in ("account-a", "account-b", "account-b"):
+        netrc = tmp_path / account
+        netrc.write_text(f"machine 127.0.0.1 login {account} password synthetic\n")
+        monkeypatch.setenv("NETRC", str(netrc))
+        head = cache.forward("POST", url, HEADERS, BODY, 5)
+        assert isinstance(head, StreamHead)
+        assert b"".join(step.data for step in head.steps if isinstance(step, StreamChunk)) == SUCCESS
+    assert len(provider.hits) == 2
+    assert dict(cache.counters.counts)["hits"] == 1
+
+
+def test_enabled_environment_reuses_store_across_fresh_backends(
+    redis_url: str, provider: Provider, monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("E2E_PROVIDER_CACHE", "1")
+    monkeypatch.setenv("E2E_PROVIDER_CACHE_REDIS_URL", redis_url)
+    monkeypatch.setenv("E2E_PROVIDER_CACHE_HMAC_KEY", SECRET.decode())
+    monkeypatch.setenv("E2E_PROVIDER_CACHE_NAMESPACE", "environment-" + uuid.uuid4().hex)
+    configured_cache.cache_clear()
+    try:
+        for _ in range(2):
+            backend = configured_cache_backend()
+            assert isinstance(backend, CacheEdge)
+            with edge(backend, provider) as url:
+                assert call(url).body == SUCCESS
+            configured_cache.cache_clear()
+        assert len(provider.hits) == 1
+        monkeypatch.setenv("E2E_PROVIDER_CACHE", "0")
+        assert configured_cache_backend() is None
+    finally:
+        configured_cache.cache_clear()
+
+
+@pytest.mark.parametrize("known_mount", (True, False))
+def test_duplicate_headers_bypass_cache_and_count_live_calls(
+    store: RedisResponseStore, provider: Provider, known_mount: bool,
+) -> None:
+    cache: Final = CacheEdge(store, SECRET)
+    with edge(cache, provider) as url:
+        parsed: Final = urlsplit(url)
+        for _ in range(2):
+            connection = HTTPConnection(str(parsed.hostname), parsed.port, timeout=5)
+            try:
+                connection.putrequest("POST", parsed.path if known_mount else "/unknown/v1/chat/completions")
+                connection.putheader("content-length", str(len(BODY)))
+                connection.putheader("content-type", "application/json")
+                connection.putheader("x-duplicate", "first")
+                connection.putheader("x-duplicate", "second")
+                connection.endheaders(BODY)
+                response = connection.getresponse()
+                assert response.status == (200 if known_mount else 404)
+                payload = response.read()
+                assert payload == SUCCESS if known_mount else b"unknown provider mount" in payload
+            finally:
+                connection.close()
+    assert len(provider.hits) == (2 if known_mount else 0)
+    assert dict(cache.counters.counts)["duplicate_header_bypass"] == 2
+    assert dict(cache.counters.counts).get("upstream_attempts", 0) == (2 if known_mount else 0)

tests/e2e/PROVIDER_CACHE.md

@@ -0,0 +1,33 @@
+# Shared provider-response cache
+
+`E2E_PROVIDER_CACHE=1` enables automatic response reuse in the live E2E mode. Standard OpenAI and Anthropic model registrations use the provider edge. Existing custom API bases, named credentials, mocked models and realtime WebSocket deployments keep their existing routing. Other provider protocols remain live
+
+The edge caches complete successful POST responses for `/v1/chat/completions` and `/v1/messages`, including streams. Unsupported endpoints pass through. It matches the method, original URL, effective outbound headers (including authentication and HTTP-library defaults), body presence and exact body bytes using a full keyed digest. It sends the same prepared request used for matching. No prompts, random markers, JSON values or credentials are normalized away
+
+An eligible miss calls the provider. A complete successful response is stored immediately even if a later test assertion fails. Provider errors, malformed responses, truncated streams and cancelled captures are not stored. Cache reads, writes and lease failures fall through to normal provider behavior; they introduce no provider retry. An already-started response cannot be restarted after a delivery failure
+
+Recordings are shared across workers and builds through dedicated Redis, separate from the candidate's own cache. They expire 86,400 seconds after capture starts, based on Redis time. Reads never extend expiry. There is no scheduled recapture: the next miss calls the provider again. Bounded coordination reduces duplicate concurrent calls, but slow or failed captures may lead to extra live calls after the wait expires
+
+## Configuration
+
+The trusted runner receives:
+
+- `E2E_PROVIDER_CACHE`: `1` to enable, `0` to use the normal live path
+- `E2E_PROVIDER_CACHE_REDIS_URL`: authenticated dedicated Redis URL
+- `E2E_PROVIDER_CACHE_HMAC_KEY`: dedicated secret containing at least 32 bytes
+- `E2E_PROVIDER_CACHE_NAMESPACE`: shared environment namespace, independent of build and candidate revision
+- `E2E_PROVIDER_CACHE_METRICS_DIR`: optional per-process counter artifact directory
+
+Do not give cache credentials to candidate deployments. Counter artifacts contain no recorded payloads or credentials. Hits count shared-cache responses; upstream attempts count actual forwards from the edge. Existing application-cache observations still count requests arriving at the edge, including shared-cache hits
+
+Tests that require real provider timing, limits or state use `@pytest.mark.provider_live`. The marker keeps newly registered models on live routes without weakening their assertions. Ordinary assertion failures still fail E2E. The shared cache does not modify provider response IDs or make the proxy aware of replay
+
+## Recorded response semantics
+
+Replay preserves the original response ID, usage and end-to-end headers. The proxy can therefore deduplicate repeated provider IDs when storing spend-log rows, just as it does when a live upstream returns the same ID twice. One spend-log row per invocation is not guaranteed for identical recorded responses. Existing spend reconciliation requests use distinct prompt markers and retain their distinct-ID and row-count assertions; accounting tests are not automatically excluded from caching
+
+Provider remaining-quota headers describe the captured response. Metrics derived from them are historical on a cache hit, not a measurement of current provider capacity. Gateway-generated API-key quota headers are a separate contract. A test of fresh provider quota or timing must use the live-provider policy; replay can still exercise how the proxy processes the recorded headers
+
+## Qualification
+
+`tests/code_coverage_tests/test_provider_cache.py` exercises local HTTP providers and disposable real Redis. CI runs these checks with the existing provider-edge and replay harness tests. These component checks do not establish Buildkite deployment, full-suite cross-build reuse or a genuine 24-hour expiry observation; those require separate runtime evidence

tests/e2e/conftest.py

@@ -22,7 +22,6 @@ from typing import Final
 
 import pytest
 import requests
-
 from e2e_config import (
     CONTROL_PLANE_BASE_URL,
     FIXTURE_DIR,
@@ -41,6 +40,7 @@ from idp import Identity, Keycloak, keycloak_from_env
 from junit_properties import attach_result_properties
 from lifecycle import ProxyClientProvider, ResourceManager
 from models import TeamNewBody, UserNewBody, UserNewResponse
+from provider_cache_routing import LIVE_PROVIDER_REQUIRED
 from provider_edge import replay_leftover_error
 from proxy_client import ProxyClient, build_proxy_client
 
@@ -85,6 +85,7 @@ def jwt_identity(idp: Keycloak, resources: ResourceManager, proxy: ProxyClient)
 
 
 def pytest_configure(config: pytest.Config) -> None:
+    config.addinivalue_line("markers", "provider_live: requires actual provider timing, limits or state; bypass shared cache")
     config.addinivalue_line(
         "markers",
         "e2e: live test that requires a running proxy and real provider keys",
@@ -192,11 +193,13 @@ def _proxy_fail_reason() -> str | None:
     return None
 
 
+@pytest.hookimpl(tryfirst=True)
 def pytest_runtest_setup(item: pytest.Item) -> None:
     """Hard-fail `e2e`-marked tests unless a proxy answers its liveness probe.
     Unmarked tests (unit coverage of the harness) don't touch the proxy, so they
     run even when none is up. Never skip for a missing proxy. Replay mode needs
     the proxy too: only provider-bound traffic replays from the bundle."""
+    LIVE_PROVIDER_REQUIRED.set(item.get_closest_marker("provider_live") is not None)
     if item.get_closest_marker("e2e") is None:
         return
     reason = _proxy_fail_reason()
@@ -235,6 +238,7 @@ def pytest_runtest_teardown(item: pytest.Item) -> Generator[None, None, None]:
     yield so fixture finalizers replay their recorded calls first. Failed tests
     are left alone - their own failure already explains any unconsumed tail."""
     result = yield
+    LIVE_PROVIDER_REQUIRED.set(False)
     if not item.stash.get(_CALL_PASSED, False):
         return result
     reason = replay_leftover_error(

tests/e2e/e2e_http.py

@@ -853,6 +853,7 @@ def _stream_steps(resp: requests.Response) -> Generator[StreamStep, None, None]:
     the chunks already delivered are exactly what makes a mid-stream failure
     different from a request that never streamed at all."""
     try:
+        yield StreamChunk(b"")
         for piece in cast("Iterator[bytes]", resp.iter_content(chunk_size=None)):
             if piece:
                 yield StreamChunk(data=piece)
@@ -862,6 +863,44 @@ def _stream_steps(resp: requests.Response) -> Generator[StreamStep, None, None]:
         resp.close()
 
 
+def primed_steps(steps: Generator[StreamStep, None, None]) -> Generator[StreamStep, None, None]:
+    first: Final = next(steps)
+    assert isinstance(first, StreamChunk) and first.data == b""
+    return steps
+
+
+@dataclass(frozen=True, slots=True, repr=False)
+class PreparedForward:
+    request: requests.PreparedRequest
+    url: str
+    headers: dict[str, str]
+
+
+def prepare_forward(
+    method: str, url: str, headers: dict[str, str], body: bytes | None,
+) -> PreparedForward | NetworkError:
+    try:
+        with requests.Session() as session:
+            request: Final = session.prepare_request(requests.Request(method, url, headers=headers, data=body))
+    except requests.RequestException as exc:
+        return NetworkError(message=str(exc))
+    assert request.url is not None
+    return PreparedForward(request, request.url, dict(request.headers))
+
+
+def forward_prepared_stream(prepared: PreparedForward, timeout: float) -> StreamHead | NetworkError:
+    try:
+        with requests.Session() as session:
+            settings: Final = session.merge_environment_settings(prepared.url, {}, True, None, None)
+            resp: Final = session.send(prepared.request, timeout=timeout, allow_redirects=False, **settings)
+    except requests.RequestException as exc:
+        return NetworkError(message=str(exc))
+    return StreamHead(
+        resp.status_code, {name.lower(): value for name, value in resp.headers.items()},
+        primed_steps(_stream_steps(resp)),
+    )
+
+
 def open_stream(url: URL, *, headers: BaseModel, json: BaseModel, timeout: float = 60.0) -> StreamHead | NetworkError:
     """POST a streaming request and return the moment its response head arrives,
     leaving the body unread behind ``StreamHead.steps``. For a test that must keep
@@ -907,5 +946,5 @@ def forward_stream(
     return StreamHead(
         status_code=resp.status_code,
         headers={name.lower(): value for name, value in resp.headers.items()},
-        steps=_stream_steps(resp),
+        steps=primed_steps(_stream_steps(resp)),
     )

tests/e2e/llm_translation/test_messages_e2e.py

@@ -24,10 +24,10 @@ from models import (
     AnthropicAssistantTurn,
     AnthropicContentBlock,
     AnthropicCustomTool,
+    AnthropicMessagesBody,
     AnthropicToolChoice,
     AnthropicToolResultBlock,
     AnthropicToolResultTurn,
-    AnthropicMessagesBody,
     ChatMessage,
     JsonSchemaProperty,
     LiteLLMParamsBody,
@@ -165,6 +165,7 @@ class TestAnthropicMessages:
         )
 
     @pytest.mark.covers("llm.messages.anthropic.basic.stream.works")
+    @pytest.mark.provider_live
     def test_messages_streams_completion(
         self, endpoints_client: EndpointsClient, resources: ResourceManager
     ) -> None:

tests/e2e/llm_translation/test_together_ai_e2e.py

@@ -744,6 +744,7 @@ class TestTogetherMessages:
         assert "22" in text, f"the model never saw the tool result: {response.content}"
 
     @pytest.mark.covers("llm.messages.together_ai.basic.stream.works")
+    @pytest.mark.provider_live
     def test_streams_text_deltas(
         self, client: PassthroughClient, resources: ResourceManager, reasoning_tool_backend: str
     ) -> None:

tests/e2e/provider_cache.py

@@ -0,0 +1,297 @@
+from __future__ import annotations
+
+import base64
+import hashlib
+import hmac
+import io
+import threading
+import time
+from collections.abc import Callable, Generator, Mapping
+from contextlib import closing
+from dataclasses import dataclass, field
+from typing import Final, Literal, Protocol
+from urllib.parse import urlsplit
+
+from e2e_http import (
+    NetworkError,
+    StreamChunk,
+    StreamHead,
+    StreamStep,
+    StreamTruncation,
+    forward_prepared_stream,
+    forward_stream,
+    prepare_forward,
+    primed_steps,
+)
+from pydantic import BaseModel, ConfigDict, JsonValue, TypeAdapter, ValidationError
+
+LIFETIME_SECONDS: Final = 86_400
+MAX_REQUEST_BYTES: Final = 256 * 1024
+MAX_RESPONSE_BYTES: Final = 8 * 1024 * 1024
+JSON_VALUE: Final[TypeAdapter[JsonValue]] = TypeAdapter(JsonValue)
+
+
+@dataclass(frozen=True, slots=True)
+class CacheHit:
+    payload: bytes
+    valid_until: float
+
+
+@dataclass(frozen=True, slots=True)
+class CaptureLease:
+    token: str
+    captured_at_ms: int
+    expires_at_ms: int
+
+
+@dataclass(frozen=True, slots=True)
+class CacheBusy:
+    pass
+
+
+@dataclass(frozen=True, slots=True)
+class CacheUnavailable:
+    pass
+
+
+type CacheLookup = CacheHit | CaptureLease | CacheBusy | CacheUnavailable
+
+
+class ResponseStore(Protocol):
+    def lookup(self, key: str) -> CacheLookup: ...
+
+    def publish(self, key: str, lease: CaptureLease, payload: bytes) -> bool: ...
+
+    def release(self, key: str, lease: CaptureLease) -> bool: ...
+
+    def discard(self, key: str, payload: bytes) -> bool: ...
+
+
+class CachedResponse(BaseModel):
+    model_config = ConfigDict(frozen=True, extra="forbid", strict=True)
+    format_version: Literal[1] = 1
+    request_key: str
+    status_code: int
+    headers: dict[str, str]
+    chunks: tuple[str, ...]
+
+
+class SignedResponse(BaseModel):
+    model_config = ConfigDict(frozen=True, extra="forbid", strict=True)
+    response: str
+    signature: str
+
+
+def exact_key(secret: bytes, method: str, url: str, headers: Mapping[str, str], body: bytes | None) -> str:
+    fields: Final = (
+        b"provider-cache-exact-v1", method.encode(), url.encode(),
+        *(part.encode() for pair in sorted(headers.items()) for part in pair),
+        b"no-body" if body is None else b"body", b"" if body is None else body,
+    )
+    encoded: Final = b"".join(len(part).to_bytes(8, "big") + part for part in fields)
+    return hmac.new(secret, encoded, hashlib.sha256).hexdigest()
+
+
+def cacheable_endpoint(method: str, url: str, body: bytes | None) -> bool:
+    return (
+        method == "POST"
+        and urlsplit(url).path in {"/v1/chat/completions", "/v1/messages"}
+        and body is not None
+        and len(body) <= MAX_REQUEST_BYTES
+    )
+
+
+def successful_response(url: str, status: int, headers: Mapping[str, str], body: bytes) -> bool:
+    if not 200 <= status < 300 or len(body) > MAX_RESPONSE_BYTES:
+        return False
+    if any(name.lower() == "set-cookie" for name in headers):
+        return False
+    streaming: Final = "text/event-stream" in headers.get("content-type", "").lower()
+    if streaming:
+        try:
+            text: Final = body.decode("utf-8").replace("\r\n", "\n")
+            if not text.endswith("\n\n"):
+                return False
+            events: Final = tuple(
+                "\n".join(line[5:].removeprefix(" ") for line in event.split("\n") if line.startswith("data:"))
+                for event in text.split("\n\n") if any(line.startswith("data:") for line in event.split("\n"))
+            )
+            values: Final = tuple(JSON_VALUE.validate_json(event) for event in events if event != "[DONE]")
+        except (UnicodeDecodeError, ValidationError):
+            return False
+        if not values or any(not isinstance(value, dict) or "error" in value or value.get("type") == "error" for value in values):
+            return False
+        if urlsplit(url).path == "/v1/chat/completions":
+            return events[-1] == "[DONE]" and "[DONE]" not in events[:-1] and complete_chat_stream(values)
+        return (
+            "[DONE]" not in events
+            and isinstance(values[0], dict) and values[0].get("type") == "message_start"
+            and isinstance(values[-1], dict) and values[-1].get("type") == "message_stop"
+            and any(
+                isinstance(value, dict) and value.get("type") == "message_delta"
+                and isinstance(delta := value.get("delta"), dict) and isinstance(delta.get("stop_reason"), str)
+                for value in values
+            )
+        )
+    try:
+        value: Final = JSON_VALUE.validate_json(body)
+    except ValidationError:
+        return False
+    if not isinstance(value, dict) or "error" in value:
+        return False
+    if urlsplit(url).path == "/v1/messages":
+        return value.get("type") == "message" and isinstance(value.get("content"), list) and isinstance(value.get("stop_reason"), str)
+    choices: Final = value.get("choices")
+    return isinstance(choices, list) and bool(choices) and all(
+        isinstance(choice, dict) and isinstance(choice.get("message"), dict) and isinstance(choice.get("finish_reason"), str)
+        for choice in choices
+    )
+
+
+def complete_chat_stream(values: tuple[JsonValue, ...]) -> bool:
+    if any(not isinstance(value, dict) or not isinstance(value.get("choices"), list) for value in values):
+        return False
+    choices: Final = tuple(
+        choice for value in values if isinstance(value, dict)
+        if isinstance(items := value.get("choices"), list) for choice in items
+    )
+    if not choices or any(
+        not isinstance(choice, dict) or type(choice.get("index")) is not int
+        or not isinstance(choice.get("delta"), dict)
+        for choice in choices
+    ):
+        return False
+    indices: Final = frozenset(choice["index"] for choice in choices if isinstance(choice, dict))
+    return all(
+        isinstance(tuple(choice for choice in choices if isinstance(choice, dict) and choice["index"] == index)[-1].get("finish_reason"), str)
+        for index in indices
+    )
+
+
+def encode_response(secret: bytes, response: CachedResponse) -> bytes:
+    raw: Final = response.model_dump_json()
+    return SignedResponse(response=raw, signature=hmac.new(secret, raw.encode(), hashlib.sha256).hexdigest()).model_dump_json().encode()
+
+
+def decode_response(secret: bytes, key: str, payload: bytes, url: str) -> CachedResponse | None:
+    if len(payload) > 2 * MAX_RESPONSE_BYTES:
+        return None
+    try:
+        signed: Final = SignedResponse.model_validate_json(payload)
+        if not hmac.compare_digest(signed.signature.encode(), hmac.new(secret, signed.response.encode(), hashlib.sha256).hexdigest().encode()):
+            return None
+        response: Final = CachedResponse.model_validate_json(signed.response)
+        chunks: Final = tuple(base64.b64decode(chunk, validate=True) for chunk in response.chunks)
+    except (ValidationError, ValueError):
+        return None
+    if response.request_key != key or not successful_response(url, response.status_code, response.headers, b"".join(chunks)):
+        return None
+    return response
+
+
+@dataclass(slots=True)
+class CacheCounters:
+    counts: tuple[tuple[str, int], ...] = ()
+    lock: threading.Lock = field(default_factory=threading.Lock)
+
+    def increment(self, name: str) -> None:
+        with self.lock:
+            current: Final = dict(self.counts)
+            self.counts = tuple((current | {name: current.get(name, 0) + 1}).items())
+
+
+@dataclass(slots=True)
+class ResponseCapture:
+    buffer: io.BytesIO = field(default_factory=io.BytesIO)
+    size: int = 0
+    eligible: bool = True
+
+    def observe(self, step: StreamStep) -> None:
+        if not self.eligible:
+            return
+        if isinstance(step, StreamTruncation) or self.size + len(step.data) + 8 > MAX_RESPONSE_BYTES:
+            self.eligible = False
+            self.buffer.close()
+            return
+        self.buffer.write(len(step.data).to_bytes(8, "big"))
+        self.buffer.write(step.data)
+        self.size += len(step.data) + 8
+
+    def chunks(self) -> tuple[bytes, ...]:
+        self.buffer.seek(0)
+        return tuple(self.buffer.read(int.from_bytes(size, "big")) for size in iter(lambda: self.buffer.read(8), b""))
+
+
+def response_steps(response: CachedResponse) -> Generator[StreamStep, None, None]:
+    for chunk in response.chunks:
+        yield StreamChunk(base64.b64decode(chunk, validate=True))
+
+
+@dataclass(frozen=True, slots=True)
+class CacheEdge:
+    store: ResponseStore
+    secret: bytes = field(repr=False)
+    counters: CacheCounters = field(default_factory=CacheCounters)
+    wait_seconds: float = 2.0
+    clock: Callable[[], float] = time.monotonic
+    sleep: Callable[[float], None] = time.sleep
+
+    def lookup(self, key: str) -> CacheLookup:
+        deadline: Final = self.clock() + self.wait_seconds
+        while isinstance(result := self.store.lookup(key), CacheBusy) and self.clock() < deadline:
+            self.sleep(min(0.05, max(0, deadline - self.clock())))
+        return result
+
+    def forward(self, method: str, url: str, headers: dict[str, str], body: bytes | None, timeout: float) -> StreamHead | NetworkError:
+        if not cacheable_endpoint(method, url, body):
+            self.counters.increment("bypass")
+            self.counters.increment("upstream_attempts")
+            return forward_stream(method, url, headers=headers, body=body, timeout=timeout)
+        prepared: Final = prepare_forward(method, url, headers, body)
+        if isinstance(prepared, NetworkError):
+            self.counters.increment("rejected")
+            return prepared
+        key: Final = exact_key(self.secret, method, url, prepared.headers, body)
+        found: Final = self.lookup(key)
+        if isinstance(found, CacheHit):
+            response: Final = decode_response(self.secret, key, found.payload, url)
+            if response is not None and self.clock() < found.valid_until:
+                self.counters.increment("hits")
+                return StreamHead(response.status_code, response.headers, response_steps(response))
+            self.counters.increment("corrupt" if response is None else "expired")
+            self.store.discard(key, found.payload)
+        capture_slot: Final = self.lookup(key) if isinstance(found, CacheHit) else found
+        self.counters.increment("misses")
+        if isinstance(capture_slot, CacheUnavailable):
+            self.counters.increment("cache_errors")
+        self.counters.increment("upstream_attempts")
+        head: Final = forward_prepared_stream(prepared, timeout)
+        if not isinstance(capture_slot, CaptureLease):
+            return head
+        if isinstance(head, NetworkError):
+            self.store.release(key, capture_slot)
+            self.counters.increment("rejected")
+            return head
+        return StreamHead(head.status_code, head.headers, primed_steps(self.capture(key, capture_slot, url, head)))
+
+    def capture(self, key: str, lease: CaptureLease, url: str, head: StreamHead) -> Generator[StreamStep, None, None]:
+        capture: Final = ResponseCapture()
+        try:
+            with closing(head.steps):
+                yield StreamChunk(b"")
+                for step in head.steps:
+                    yield step
+                    capture.observe(step)
+            chunks: Final = capture.chunks() if capture.eligible else ()
+            if not capture.eligible or not successful_response(url, head.status_code, head.headers, b"".join(chunks)):
+                self.counters.increment("rejected")
+                return
+            response: Final = CachedResponse(
+                request_key=key, status_code=head.status_code, headers=head.headers,
+                chunks=tuple(base64.b64encode(chunk).decode("ascii") for chunk in chunks),
+            )
+            published: Final = self.store.publish(key, lease, encode_response(self.secret, response))
+            self.counters.increment("writes" if published else "write_failures")
+        finally:
+            self.store.release(key, lease)
+            capture.buffer.close()

tests/e2e/provider_cache_redis.py

@@ -0,0 +1,154 @@
+from __future__ import annotations
+
+import atexit
+import functools
+import json
+import logging
+import os
+import re
+import time
+import uuid
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Final, Protocol, cast
+
+from provider_cache import LIFETIME_SECONDS, CacheBusy, CacheEdge, CacheHit, CacheLookup, CacheUnavailable, CaptureLease
+from pydantic import TypeAdapter, ValidationError
+from redis import Redis
+from redis.exceptions import RedisError
+
+REDIS_ARRAY: Final[TypeAdapter[list[bytes]]] = TypeAdapter(list[bytes])
+
+LOOKUP: Final = """
+local clock = redis.call('TIME')
+local now = clock[1] * 1000 + math.floor(clock[2] / 1000)
+local row = redis.call('HMGET', KEYS[1], 'captured', 'expires', 'payload')
+if row[3] then
+  local captured = tonumber(row[1])
+  local expires = tonumber(row[2])
+  if captured and expires and captured <= now and expires > now
+     and expires - captured == tonumber(ARGV[2]) then
+    return {'hit', row[3], tostring(expires - now)}
+  end
+  redis.call('DEL', KEYS[1])
+end
+if redis.call('SET', KEYS[2], ARGV[1], 'NX', 'PX', ARGV[3]) then
+  return {'lease', tostring(now), tostring(now + tonumber(ARGV[2]))}
+end
+return {'busy'}
+"""
+
+PUBLISH: Final = """
+if redis.call('GET', KEYS[2]) ~= ARGV[1] then return 0 end
+local clock = redis.call('TIME')
+local now = clock[1] * 1000 + math.floor(clock[2] / 1000)
+local captured = tonumber(ARGV[2])
+local expires = tonumber(ARGV[3])
+if captured > now or expires <= now or expires - captured ~= tonumber(ARGV[5]) then return 0 end
+if redis.call('EXISTS', KEYS[1]) == 1 then return 0 end
+redis.call('HSET', KEYS[1], 'captured', ARGV[2], 'expires', ARGV[3], 'payload', ARGV[4])
+redis.call('PEXPIREAT', KEYS[1], expires)
+redis.call('DEL', KEYS[2])
+return 1
+"""
+
+RELEASE: Final = """
+if redis.call('GET', KEYS[1]) ~= ARGV[1] then return 0 end
+return redis.call('DEL', KEYS[1])
+"""
+
+DISCARD: Final = """
+if redis.call('HGET', KEYS[1], 'payload') ~= ARGV[1] then return 0 end
+return redis.call('DEL', KEYS[1])
+"""
+
+
+class RedisCommands(Protocol):
+    def eval(self, script: str, numkeys: int, *args: str | bytes | int) -> object: ...
+
+
+@dataclass(frozen=True, slots=True)
+class RedisResponseStore:
+    client: RedisCommands
+    namespace: str
+    lifetime_ms: int = LIFETIME_SECONDS * 1000
+    lease_ms: int = 120_000
+
+    def keys(self, key: str) -> tuple[str, str]:
+        prefix: Final = f"e2e-provider-cache:v1:{self.namespace}:{{{key}}}"
+        return prefix + ":response", prefix + ":lease"
+
+    def lookup(self, key: str) -> CacheLookup:
+        token: Final = uuid.uuid4().hex
+        started: Final = time.monotonic()
+        try:
+            result: Final = self.client.eval(LOOKUP, 2, *self.keys(key), token, self.lifetime_ms, self.lease_ms)
+        except (RedisError, OSError):
+            return CacheUnavailable()
+        try:
+            parts: Final = tuple(REDIS_ARRAY.validate_python(result, strict=True))
+        except ValidationError:
+            return CacheUnavailable()
+        if len(parts) == 3 and parts[0] == b"hit" and parts[2].isdigit():
+            return CacheHit(parts[1], started + int(parts[2]) / 1000)
+        if len(parts) == 3 and parts[0] == b"lease" and parts[1].isdigit() and parts[2].isdigit():
+            return CaptureLease(token, int(parts[1]), int(parts[2]))
+        if parts == (b"busy",):
+            return CacheBusy()
+        return CacheUnavailable()
+
+    def publish(self, key: str, lease: CaptureLease, payload: bytes) -> bool:
+        try:
+            result: Final = self.client.eval(
+                PUBLISH, 2, *self.keys(key), lease.token, lease.captured_at_ms, lease.expires_at_ms, payload, self.lifetime_ms,
+            )
+        except (RedisError, OSError):
+            return False
+        return result == 1
+
+    def release(self, key: str, lease: CaptureLease) -> bool:
+        try:
+            result: Final = self.client.eval(RELEASE, 1, self.keys(key)[1], lease.token)
+        except (RedisError, OSError):
+            return False
+        return result == 1
+
+    def discard(self, key: str, payload: bytes) -> bool:
+        try:
+            result: Final = self.client.eval(DISCARD, 1, self.keys(key)[0], payload)
+        except (RedisError, OSError):
+            return False
+        return result == 1
+
+
+def redis_store(url: str, namespace: str) -> RedisResponseStore:
+    client: Final = Redis.from_url(url, socket_timeout=0.25, socket_connect_timeout=0.25, decode_responses=False)
+    return RedisResponseStore(cast(RedisCommands, client), namespace)
+
+
+def write_metrics(cache: CacheEdge) -> None:
+    report: Final = json.dumps({"provider_cache": dict(cache.counters.counts)})
+    directory: Final = os.environ.get("E2E_PROVIDER_CACHE_METRICS_DIR")
+    if directory:
+        try:
+            root: Final = Path(directory)
+            root.mkdir(parents=True, exist_ok=True)
+            (root / f"{os.getpid()}.json").write_text(report + "\n")
+        except OSError:
+            logging.getLogger(__name__).warning("provider cache metrics artifact unavailable")
+    logging.getLogger(__name__).info("%s", report)
+
+
+@functools.lru_cache(maxsize=1)
+def configured_cache() -> CacheEdge | None:
+    if os.environ.get("E2E_PROVIDER_CACHE", "0") == "0":
+        return None
+    if os.environ.get("E2E_PROVIDER_CACHE") != "1":
+        raise ValueError("E2E_PROVIDER_CACHE must be 0 or 1")
+    secret: Final = os.environ.get("E2E_PROVIDER_CACHE_HMAC_KEY", "").encode()
+    namespace: Final = os.environ.get("E2E_PROVIDER_CACHE_NAMESPACE", "")
+    if len(secret) < 32 or re.fullmatch(r"[a-zA-Z0-9_-]{1,64}", namespace) is None:
+        raise ValueError("provider cache requires a dedicated key and namespace")
+    cache: Final = CacheEdge(redis_store(os.environ["E2E_PROVIDER_CACHE_REDIS_URL"], namespace), secret)
+    atexit.register(write_metrics, cache)
+    return cache

tests/e2e/provider_cache_routing.py

@@ -0,0 +1,23 @@
+from __future__ import annotations
+
+from collections.abc import Callable
+from contextvars import ContextVar
+from typing import Final
+
+from models import LiteLLMParamsBody, ModelMode
+
+LIVE_PROVIDER_REQUIRED: Final[ContextVar[bool]] = ContextVar("live_provider_required", default=False)
+
+
+def route_cache_model(
+    params: LiteLLMParamsBody, base_for: Callable[[str], str | None], *, enabled: bool, mode: ModelMode | None = None,
+) -> LiteLLMParamsBody:
+    if not enabled or mode == "realtime" or LIVE_PROVIDER_REQUIRED.get() or params.api_base is not None or params.mock_response is not None:
+        return params
+    provider: Final = params.model.partition("/")[0]
+    if provider not in {"openai", "anthropic"} or params.litellm_credential_name is not None:
+        return params
+    base: Final = base_for(provider)
+    if base is None:
+        return params
+    return params.model_copy(update={"api_base": f"{base}/v1" if provider == "openai" else base})

tests/e2e/provider_edge.py

@@ -42,6 +42,7 @@ import base64
 import difflib
 import functools
 import hashlib
+import os
 import re
 import threading
 from collections import deque
@@ -93,6 +94,8 @@ from fixture_mode import (
     parse_fixture_mode,
 )
 from fixture_profile import IneligibleRequest, MatchProfile, match_profile, strict_identity
+from provider_cache import CacheEdge
+from provider_cache_routing import LIVE_PROVIDER_REQUIRED
 from pydantic import JsonValue, TypeAdapter
 
 EDGE_MOUNTS: Final[Mapping[str, str]] = MappingProxyType(
@@ -506,7 +509,7 @@ class LiveEdge:
     pass
 
 
-type EdgeBackend = RecordEdge | ReplayEdge | LiveEdge
+type EdgeBackend = RecordEdge | ReplayEdge | LiveEdge | CacheEdge
 
 
 @dataclass(slots=True)
@@ -750,12 +753,16 @@ def _handle_record(
 
 
 def _handle_live(
-    method: str, url: str, headers: Mapping[str, str], body: bytes | None, timeout: float
+    method: str, url: str, headers: Mapping[str, str], body: bytes | None, timeout: float,
+    cache: CacheEdge | None = None,
 ) -> 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)
+    head: Final = (
+        forward_stream(method, url, headers=forwarded, body=body, timeout=timeout)
+        if cache is None else cache.forward(method, url, forwarded, body, timeout)
+    )
     match head:
         case NetworkError(message=message):
             return _recorded_outcome(_network_error_response(message))
@@ -821,6 +828,10 @@ def handle_edge_request(
         else edge_request(method, split.path, split.query, body, _header_value(headers, "content-type"))
     )
     match backend:
+        case CacheEdge():
+            return _handle_live(
+                method, _upstream_url(upstream_base, upstream_path, split.query), headers, body, timeout, backend,
+            )
         case LiveEdge():
             return _handle_live(
                 method, _upstream_url(upstream_base, upstream_path, split.query), headers, body, timeout
@@ -871,11 +882,19 @@ class _EdgeHandler(BaseHTTPRequestHandler):
             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):
+        if strict and len({name.lower() for name in self.headers}) != len(self.headers):
             self._write_reply(_text_reply(REPLAY_MISS_STATUS, "stateless_v1 eligibility error: duplicate headers"))
             return
+        duplicate_headers: Final = len({name.lower() for name in self.headers}) != len(self.headers)
+        selected_backend: Final = (
+            LiveEdge() if isinstance(edge_server.backend, CacheEdge) and duplicate_headers else edge_server.backend
+        )
+        if isinstance(edge_server.backend, CacheEdge) and duplicate_headers:
+            edge_server.backend.counters.increment("duplicate_header_bypass")
+            if urlsplit(self.path).path.lstrip("/").partition("/")[0] in edge_server.mounts:
+                edge_server.backend.counters.increment("upstream_attempts")
         outcome: Final = handle_edge_request(
-            edge_server.backend,
+            selected_backend,
             edge_server.mounts,
             self.command,
             self.path,
@@ -908,12 +927,12 @@ class _EdgeHandler(BaseHTTPRequestHandler):
         shuts down write-side first: the proxy sees a graceful close mid-message,
         which is the incomplete chunked read a provider hanging up produces, and not
         the reset that could discard the chunks already in flight."""
-        self.send_response(stream.status_code)
-        for name, value in stream.headers.items():
-            self.send_header(name, value)
-        self.send_header("transfer-encoding", "chunked")
-        self.end_headers()
         with closing(stream.steps) as steps:
+            self.send_response(stream.status_code)
+            for name, value in stream.headers.items():
+                self.send_header(name, value)
+            self.send_header("transfer-encoding", "chunked")
+            self.end_headers()
             for step in steps:
                 match step:
                     case StreamChunk(data=data):
@@ -923,7 +942,7 @@ class _EdgeHandler(BaseHTTPRequestHandler):
                         return
                     case _:
                         assert_never(step)
-        self.wfile.write(b"0\r\n\r\n")
+            self.wfile.write(b"0\r\n\r\n")
 
     def log_message(self, format: str, *args: object) -> None:
         """Silence the per-request stderr line BaseHTTPRequestHandler emits."""
@@ -1056,6 +1075,8 @@ def provider_edge_api_base(
         case InvalidFixtureMode(value=value):
             raise ValueError(f"E2E_FIXTURE_MODE={value!r} is not one of {', '.join(FIXTURE_MODES)}")
         case "live":
+            if configured_cache_backend() is not None:
+                return _shared_cache_edge(bind_host, advertise_host, forward_timeout).api_base(mount)
             return None
         case "record" | "replay":
             if mount not in EDGE_MOUNTS:
@@ -1073,7 +1094,7 @@ def _observed_backend(mode_raw: str, bundle_dir: Path) -> EdgeBackend:
         case InvalidFixtureMode(value=value):
             raise ValueError(f"E2E_FIXTURE_MODE={value!r} is not one of {', '.join(FIXTURE_MODES)}")
         case "live":
-            return LiveEdge()
+            return configured_cache_backend() or LiveEdge()
         case "record":
             return RecordEdge(_shared_recorder(bundle_dir, match_profile()), threading.Lock())
         case "replay":
@@ -1082,6 +1103,24 @@ def _observed_backend(mode_raw: str, bundle_dir: Path) -> EdgeBackend:
             assert_never(mode)
 
 
+def configured_cache_backend() -> CacheEdge | None:
+    if LIVE_PROVIDER_REQUIRED.get() or os.environ.get("E2E_PROVIDER_CACHE", "0") == "0":
+        return None
+    from provider_cache_redis import configured_cache
+
+    return configured_cache()
+
+
+@functools.lru_cache(maxsize=8)
+def _shared_cache_edge(bind_host: str, advertise_host: str, forward_timeout: float) -> ProviderEdge:
+    backend: Final = configured_cache_backend()
+    assert backend is not None
+    return start_provider_edge(
+        backend, mounts=EDGE_MOUNTS, bind_host=bind_host,
+        advertise_host=advertise_host, forward_timeout=forward_timeout,
+    ).edge
+
+
 @contextmanager
 def observed_provider_edge(
     observation: ProviderRequestObservation,

tests/e2e/proxy_client.py

@@ -8,6 +8,7 @@ ProxyClient's key/customer methods for cleanup. Read-backs are eventually consis
 
 from __future__ import annotations
 
+import os
 import time
 import warnings
 from collections.abc import Callable, Mapping
@@ -26,6 +27,7 @@ from e2e_config import (
     PROXY_REPLICA_URLS,
     REQUEST_TIMEOUT,
     SLOW_PROVIDER_TIMEOUT_SECONDS,
+    provider_edge_base,
     settle_propagation,
 )
 from e2e_http import (
@@ -93,6 +95,7 @@ from models import (
     UserDeleteBody,
     UserDeleteResponse,
 )
+from provider_cache_routing import route_cache_model
 from pydantic import BaseModel
 from transport import HttpTransport, SplitTransport, Transport, is_control_plane_path
 
@@ -645,7 +648,10 @@ class ProxyClient:
             self.transport.post(
                 "/model/new",
                 headers=self.management_headers(),
-                json=body,
+                json=body.model_copy(update={"litellm_params": route_cache_model(
+                    body.litellm_params, provider_edge_base,
+                    enabled=os.environ.get("E2E_PROVIDER_CACHE", "0") == "1", mode=body.model_info.mode,
+                )}),
                 response_type=ModelNewResponse,
             )
         ).model_id

tests/test_litellm/test_circleci_path_filter.py

@@ -49,6 +49,20 @@ CI = [".github/workflows/test-litellm-ui-unit.yml"]
 @pytest.mark.parametrize(
     "category,changed,expected",
     [
+        ("provider-harness", ["tests/e2e/provider_cache.py"], "run"),
+        ("provider-harness", ["tests/e2e/conftest.py"], "run"),
+        ("provider-harness", ["tests/e2e/e2e_http.py"], "run"),
+        ("provider-harness", ["tests/code_coverage_tests/test_provider_cache.py"], "run"),
+        ("provider-harness", ["tests/code_coverage_tests/test_provider_replay_harness.py"], "run"),
+        ("provider-harness", [".circleci/config.yml"], "run"),
+        ("provider-harness", [".circleci/scripts/classify_changes.sh"], "run"),
+        ("provider-harness", ["pyproject.toml"], "run"),
+        ("provider-harness", ["uv.lock"], "run"),
+        ("provider-harness", ["tests/e2e/PROVIDER_CACHE.md"], "skip"),
+        ("provider-harness", ["tests/e2e/ui/test_example.py"], "skip"),
+        ("provider-harness", ["tests/e2e/quota_management/test_quota.py"], "skip"),
+        ("provider-harness", ["litellm/main.py"], "skip"),
+        ("provider-harness", ["ui/litellm-dashboard/src/App.tsx"], "skip"),
         # docs-only: skip everything
         ("backend", DOCS, "skip"),
         ("client", DOCS, "skip"),