Merge pull request #41520 from BerriAI/litellm_/attribution-investigation-a81211

fix(e2e): bind provider-cache recordings to the deployment's test, not the serving process

tests/code_coverage_tests/test_provider_cache.py

@@ -8,6 +8,7 @@ import shutil
 import socket
 import subprocess
 import struct
+import sys
 import threading
 import time
 import uuid
@@ -17,15 +18,17 @@ from contextlib import contextmanager
 from dataclasses import dataclass, replace
 from http.client import HTTPConnection
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
 from typing import Final
 from urllib.parse import urlsplit
 
 import pytest
-from pydantic import JsonValue
-from e2e_http import NetworkError, PreparedForward, RawResponse, StreamChunk, StreamHead, forward, prepare_forward
-from models import LiteLLMParamsBody, ModelMode
+from pydantic import JsonValue, TypeAdapter
+from e2e_http import NetworkError, PreparedForward, RawResponse, StreamChunk, StreamHead, forward, prepare_forward, without_retries
+from models import LiteLLMParamsBody, ModelMode, ModelNewBody
 from botocore.credentials import Credentials
 from botocore.eventstream import EventStreamBuffer
+from fixture_bundle import slug_for_test
 from provider_cache import (
     SIGNATURE_HEADERS,
     CacheEdge,
@@ -35,7 +38,9 @@ from provider_cache import (
     ResponseStore,
     cacheable_endpoint,
     request_identity,
+    scoped_edge_base,
     slotted_key,
+    split_test_segment,
     successful_response,
 )
 from provider_cache_redis import PUBLISH, RedisCommands, RedisResponseStore, configured_cache, redis_store
@@ -46,9 +51,16 @@ from provider_cache_routing import (
     bedrock_region,
     route_cache_model,
 )
-from fixture_mode import SESSION_TEST_KEY
-from provider_edge import EDGE_MOUNTS, configured_cache_backend, resolve_mount, start_provider_edge
+from fixture_mode import SESSION_TEST_KEY, current_test_key, registration_owner
+from provider_edge import (
+    EDGE_MOUNTS,
+    configured_cache_backend,
+    provider_edge_api_base,
+    resolve_mount,
+    start_provider_edge,
+)
 from provider_edge_bedrock import bedrock_signer
+from proxy_client import build_proxy_client
 from redis.exceptions import ConnectionError as RedisConnectionError
 
 SECRET: Final = b"synthetic-cache-hmac-key-for-tests"
@@ -57,6 +69,7 @@ SUCCESS: Final = b'{"id":"provider-fixed-id","choices":[{"message":{"content":"h
 HEADERS: Final = {"content-type": "application/json", "authorization": "Bearer synthetic-account-one"}
 TEST_KEY: Final = "tests/e2e/synthetic_suite.py::TestCase::test_case"
 OTHER_TEST_KEY: Final = "tests/e2e/synthetic_suite.py::TestCase::test_other_case"
+TEST_SLUG: Final = slug_for_test(TEST_KEY)
 
 
 def marked(marker: str) -> bytes:
@@ -173,11 +186,11 @@ def store(redis_url: str) -> RedisResponseStore:
     return redis_store(redis_url, "test-" + uuid.uuid4().hex)
 
 
-def cache_edge(store: ResponseStore, test_key: str = TEST_KEY) -> CacheEdge:
+def cache_edge(store: ResponseStore) -> CacheEdge:
     """A cache edge standing in for one pytest process. A fresh instance over the
     same store is the next build running the same test: the recordings survive,
     the per-test FIFO slot counters start over."""
-    return CacheEdge(store, SECRET, test_key=lambda: test_key)
+    return CacheEdge(store, SECRET)
 
 
 def slot_key(
@@ -186,12 +199,13 @@ def slot_key(
 ) -> str:
     prepared: Final = prepare_forward("POST", url, headers, body)
     assert isinstance(prepared, PreparedForward)
-    return slotted_key(SECRET, request_identity(SECRET, test_key, "POST", url, prepared.headers, body), slot)
+    identity: Final = request_identity(SECRET, slug_for_test(test_key), "POST", url, prepared.headers, body)
+    return slotted_key(SECRET, identity, slot)
 
 
-def bedrock_cache_edge(store: ResponseStore, test_key: str = TEST_KEY) -> CacheEdge:
+def bedrock_cache_edge(store: ResponseStore) -> CacheEdge:
     return CacheEdge(
-        store, SECRET, test_key=lambda: test_key,
+        store, SECRET,
         policies={BEDROCK_MOUNT: MountPolicy(
             sign=bedrock_signer("us-east-1", lambda: STATIC_CREDENTIALS), unkeyed_headers=SIGNATURE_HEADERS,
         )},
@@ -199,11 +213,14 @@ def bedrock_cache_edge(store: ResponseStore, test_key: str = TEST_KEY) -> CacheE
 
 
 @contextmanager
-def edge(cache: CacheEdge, provider: Provider) -> Generator[str, None, None]:
+def edge(cache: CacheEdge, provider: Provider, test_key: str | None = TEST_KEY) -> Generator[str, None, None]:
+    """The URL a deployment registered by ``test_key`` would carry, or the bare
+    mount URL for None, which is what a registration made outside any test gets."""
     upstream: Final = f"http://127.0.0.1:{provider.server_port}"
     running: Final = start_provider_edge(cache, mounts={"openai": upstream})
+    base: Final = running.edge.api_base("openai")
     try:
-        yield running.edge.api_base("openai") + "/v1/chat/completions"
+        yield f"{base if test_key is None else scoped_edge_base(base, test_key)}/v1/chat/completions"
     finally:
         running.shutdown()
 
@@ -213,7 +230,7 @@ def bedrock_edge(cache: CacheEdge, provider: Provider, action: str = "converse")
     upstream: Final = f"http://127.0.0.1:{provider.server_port}"
     running: Final = start_provider_edge(cache, mounts={BEDROCK_MOUNT: upstream})
     try:
-        yield f"{running.edge.api_base(BEDROCK_MOUNT)}/model/{BEDROCK_MODEL}/{action}"
+        yield f"{scoped_edge_base(running.edge.api_base(BEDROCK_MOUNT), TEST_KEY)}/model/{BEDROCK_MODEL}/{action}"
     finally:
         running.shutdown()
 
@@ -288,7 +305,7 @@ def test_expiry_does_not_slide(store: RedisResponseStore, provider: Provider) ->
     url: Final = f"http://127.0.0.1:{provider.server_port}/v1/chat/completions"
 
     def drain() -> None:
-        head = cache_edge(short).forward("openai", "POST", url, dict(HEADERS), BODY, 5)
+        head = cache_edge(short).forward("openai", "POST", url, dict(HEADERS), BODY, 5, test_key=TEST_SLUG)
         assert isinstance(head, StreamHead)
         assert b"".join(step.data for step in head.steps if isinstance(step, StreamChunk)) == SUCCESS
 
@@ -312,7 +329,7 @@ def test_concurrent_builds_publish_one_recording_atomically(
     edges: Final = tuple(cache_edge(store) for _ in range(5))
 
     def drain(cache: CacheEdge) -> bytes:
-        head = cache.forward("openai", "POST", url, dict(HEADERS), BODY, 5)
+        head = cache.forward("openai", "POST", url, dict(HEADERS), BODY, 5, test_key=TEST_SLUG)
         assert isinstance(head, StreamHead)
         return b"".join(step.data for step in head.steps if isinstance(step, StreamChunk))
 
@@ -401,7 +418,7 @@ def test_corrupt_entry_is_replaced_by_same_successful_request(store: RedisRespon
     assert store.publish(key, lease, payload)
     caches: Final = tuple(cache_edge(store) for _ in range(2))
     for cache in caches:
-        head = cache.forward("openai", "POST", upstream, dict(HEADERS), BODY, 5)
+        head = cache.forward("openai", "POST", upstream, dict(HEADERS), BODY, 5, test_key=TEST_SLUG)
         assert isinstance(head, StreamHead)
         assert b"".join(step.data for step in head.steps if isinstance(step, StreamChunk)) == SUCCESS
     assert len(provider.hits) == 1
@@ -471,28 +488,192 @@ def test_another_test_never_reuses_this_tests_recording(
     with edge(cache_edge(store), provider) as url:
         call(url)
     assert len(provider.hits) == 1
-    with edge(cache_edge(store, OTHER_TEST_KEY), provider) as url:
+    with edge(cache_edge(store), provider, OTHER_TEST_KEY) as url:
         call(url)
     assert len(provider.hits) == 2
-    with edge(cache_edge(store, OTHER_TEST_KEY), provider) as url:
+    with edge(cache_edge(store), provider, OTHER_TEST_KEY) as url:
         call(url)
     assert len(provider.hits) == 2
 
 
-def test_calls_outside_any_test_are_never_cached(
-    store: RedisResponseStore, provider: Provider,
+def test_a_request_without_a_test_segment_is_never_cached(
+    store: RedisResponseStore, provider: Provider, monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    url: Final = f"http://127.0.0.1:{provider.server_port}/v1/chat/completions"
-    cache: Final = CacheEdge(store, SECRET, test_key=lambda: SESSION_TEST_KEY)
-    for _ in range(2):
-        head = cache.forward("openai", "POST", url, dict(HEADERS), BODY, 5)
-        assert isinstance(head, StreamHead)
-        assert b"".join(step.data for step in head.steps if isinstance(step, StreamChunk)) == SUCCESS
+    """The bare mount URL is what a deployment registered outside any test would
+    carry. The serving process is inside a test here, and that must not count:
+    the edge never names the test from its own process state."""
+    monkeypatch.setenv("PYTEST_CURRENT_TEST", f"{TEST_KEY} (call)")
+    cache: Final = cache_edge(store)
+    with edge(cache, provider, test_key=None) as url:
+        assert call(url).body == SUCCESS
+        assert call(url).body == SUCCESS
     assert len(provider.hits) == 2
     assert dict(cache.counters.counts) == {
         "bypass": 2, "mount:openai:bypass": 2,
         "upstream_attempts": 2, "mount:openai:upstream_attempts": 2,
     }
+    with edge(cache_edge(store), provider) as url:
+        assert call(url).body == SUCCESS
+    assert len(provider.hits) == 3
+
+
+def test_attribution_comes_from_the_deployment_path_not_the_serving_process(
+    store: RedisResponseStore, provider: Provider, monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Under xdist the process serving a call is unrelated to the test that made
+    it: the proxy is a separate pod, and the compat matrix's shared aliases had
+    every worker's edge answering every other worker's cells. The recording must
+    land under the test whose deployment the request came through, whatever
+    ``PYTEST_CURRENT_TEST`` says in the edge's own process."""
+    monkeypatch.setenv("PYTEST_CURRENT_TEST", f"{OTHER_TEST_KEY} (call)")
+    monkeypatch.setenv("E2E_PROVIDER_CACHE_METRICS_DIR", "unused-but-enables-the-probe")
+    first: Final = cache_edge(store)
+    with edge(first, provider) as url:
+        assert call(url).body == SUCCESS
+    assert len(provider.hits) == 1
+    assert dict(first.probe.rows[0])["test_key"] == TEST_SLUG
+    monkeypatch.setenv("PYTEST_CURRENT_TEST", f"{TEST_KEY} (call)")
+    with edge(cache_edge(store), provider, OTHER_TEST_KEY) as url:
+        assert call(url).body == SUCCESS
+    assert len(provider.hits) == 2
+    with edge(cache_edge(store), provider) as url:
+        assert call(url).body == SUCCESS
+    assert len(provider.hits) == 2
+
+
+@pytest.mark.parametrize("upstream_path,expected", [
+    (f"t/{TEST_SLUG}/v1/chat/completions", (TEST_SLUG, "v1/chat/completions")),
+    (f"t/{TEST_SLUG}/model/{BEDROCK_MODEL}/converse-stream", (TEST_SLUG, f"model/{BEDROCK_MODEL}/converse-stream")),
+    ("v1/chat/completions", (None, "v1/chat/completions")),
+    (f"model/{BEDROCK_MODEL}/invoke", (None, f"model/{BEDROCK_MODEL}/invoke")),
+    ("t//v1/chat/completions", (None, "v1/chat/completions")),
+    ("t", (None, "")),
+])
+def test_the_test_segment_is_read_off_the_path_and_never_reaches_the_provider(
+    upstream_path: str, expected: tuple[str | None, str],
+) -> None:
+    assert split_test_segment(upstream_path) == expected
+    assert split_test_segment(scoped_edge_base("", TEST_KEY).lstrip("/") + "/v1/chat/completions") == (
+        TEST_SLUG, "v1/chat/completions",
+    )
+
+
+def test_the_cache_edge_base_is_scoped_to_the_registering_test(
+    redis_url: str, monkeypatch: pytest.MonkeyPatch, tmp_path,
+) -> 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()
+
+    def base_for(test_key: str) -> str | None:
+        return provider_edge_api_base(
+            "openai", mode_raw="live", bundle_dir=tmp_path, bind_host="127.0.0.1", advertise_host="127.0.0.1",
+            test_key=test_key,
+        )
+
+    try:
+        scoped: Final = base_for(TEST_KEY)
+        assert scoped is not None and scoped.endswith(f"/openai/t/{TEST_SLUG}")
+        assert base_for(OTHER_TEST_KEY) != scoped
+        assert base_for(SESSION_TEST_KEY) is None
+        monkeypatch.setenv("E2E_PROVIDER_CACHE", "0")
+        configured_cache.cache_clear()
+        assert base_for(TEST_KEY) is None
+    finally:
+        configured_cache.cache_clear()
+
+
+@pytest.mark.parametrize("provider_live", (False, True))
+def test_a_registration_carries_its_owners_segment_unless_it_is_provider_live(
+    provider_live: bool, provider: Provider, redis_url: str, 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", "registration-" + uuid.uuid4().hex)
+    configured_cache.cache_clear()
+    provider.status = 401
+    provider.response = b"{}"
+    url: Final = f"http://127.0.0.1:{provider.server_port}"
+    proxy: Final = build_proxy_client(base_url=url, control_plane_base_url=url, replica_urls=(url,), master_key="owner")
+    try:
+        with without_retries(), pytest.raises(AssertionError):
+            proxy.create_model("owned", LiteLLMParamsBody(model="openai/synthetic"), provider_live=provider_live)
+    finally:
+        configured_cache.cache_clear()
+    ((path, body),) = provider.hits
+    assert path == "/model/new"
+    sent: Final = ModelNewBody.model_validate_json(body)
+    if provider_live:
+        assert sent.litellm_params.api_base is None
+        return
+    assert sent.litellm_params.api_base is not None
+    assert sent.litellm_params.api_base.endswith(f"/openai/t/{slug_for_test(current_test_key())}/v1")
+
+
+OWNER_PROBE: Final = """
+import json
+import os
+
+import pytest
+from fixture_mode import registration_owner
+
+
+@pytest.fixture(scope="session")
+def session_owner() -> str:
+    return registration_owner()
+
+
+@pytest.fixture(scope="module")
+def module_owner() -> str:
+    return registration_owner()
+
+
+@pytest.fixture(scope="class")
+def class_owner() -> str:
+    return registration_owner()
+
+
+@pytest.fixture
+def function_owner() -> str:
+    return registration_owner()
+
+
+class TestOwners:
+    def test_probe(self, session_owner: str, module_owner: str, class_owner: str, function_owner: str) -> None:
+        owners = {
+            "session": session_owner,
+            "module": module_owner,
+            "class": class_owner,
+            "function": function_owner,
+            "call": registration_owner(),
+        }
+        with open(os.environ["OWNER_PROBE_OUT"], "w") as out:
+            json.dump(owners, out)
+"""
+
+
+def test_a_fixture_owns_what_it_registers_at_the_node_it_is_scoped_to(tmp_path: Path) -> None:
+    probe: Final = tmp_path / "test_owner_probe.py"
+    probe.write_text(OWNER_PROBE)
+    out: Final = tmp_path / "owners.json"
+    run: Final = subprocess.run(
+        [sys.executable, "-m", "pytest", "-q", "-p", "no:cacheprovider", "-p", "fixture_mode", "--noconftest",
+         "-o", "addopts=", probe.name],
+        cwd=tmp_path,
+        env={**os.environ, "PYTHONPATH": str(Path(__file__).resolve().parents[1] / "e2e"), "OWNER_PROBE_OUT": str(out)},
+        capture_output=True, text=True, timeout=120, check=False,
+    )
+    assert run.returncode == 0, run.stdout + run.stderr
+    assert TypeAdapter(dict[str, str]).validate_json(out.read_text()) == {
+        "session": SESSION_TEST_KEY,
+        "module": "test_owner_probe.py",
+        "class": "test_owner_probe.py::TestOwners",
+        "function": "test_owner_probe.py::TestOwners::test_probe",
+        "call": "test_owner_probe.py::TestOwners::test_probe",
+    }
 
 
 def test_counters_attribute_every_outcome_to_its_mount(
@@ -505,8 +686,8 @@ def test_counters_attribute_every_outcome_to_its_mount(
     cache: Final = cache_edge(store)
     running: Final = start_provider_edge(cache, mounts={"openai": upstream, "anthropic": upstream})
     try:
-        call(running.edge.api_base("openai") + "/v1/chat/completions")
-        call(running.edge.api_base("anthropic") + "/v1/messages")
+        call(scoped_edge_base(running.edge.api_base("openai"), TEST_KEY) + "/v1/chat/completions")
+        call(scoped_edge_base(running.edge.api_base("anthropic"), TEST_KEY) + "/v1/messages")
     finally:
         running.shutdown()
     counts: Final = dict(cache.counters.counts)
@@ -531,7 +712,7 @@ def test_a_rejection_says_whether_the_body_was_cut_short_or_simply_unfinished(
     provider.response = b'data: {"choices":[{"index":0,"delta":{"content":"hi"},"finish_reason":"stop"}]}\n\ndata: [DONE]\n\n'
     running: Final = start_provider_edge(cut_short, mounts={"openai": upstream})
     try:
-        forward("POST", running.edge.api_base("openai") + "/v1/chat/completions",
+        forward("POST", scoped_edge_base(running.edge.api_base("openai"), TEST_KEY) + "/v1/chat/completions",
                 headers=HEADERS, body=MARKED, timeout=5)
     finally:
         running.shutdown()
@@ -542,7 +723,7 @@ def test_a_rejection_says_whether_the_body_was_cut_short_or_simply_unfinished(
     provider.response = b'{"choices":[{"index":0,"message":{"content":"hi"}}]}'
     second: Final = start_provider_edge(unfinished, mounts={"openai": upstream})
     try:
-        call(second.edge.api_base("openai") + "/v1/chat/completions", MARKED)
+        call(scoped_edge_base(second.edge.api_base("openai"), TEST_KEY) + "/v1/chat/completions", MARKED)
     finally:
         second.shutdown()
 
@@ -551,7 +732,7 @@ def test_a_rejection_says_whether_the_body_was_cut_short_or_simply_unfinished(
     provider.response = b'{"message":"Too many requests"}'
     third: Final = start_provider_edge(refused, mounts={"openai": upstream})
     try:
-        call(third.edge.api_base("openai") + "/v1/chat/completions", MARKED)
+        call(scoped_edge_base(third.edge.api_base("openai"), TEST_KEY) + "/v1/chat/completions", MARKED)
     finally:
         third.shutdown()
 
@@ -586,7 +767,7 @@ def openai_edge(cache: CacheEdge, provider: Provider, path: str) -> Generator[st
     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") + path
+        yield scoped_edge_base(running.edge.api_base("openai"), TEST_KEY) + path
     finally:
         running.shutdown()
 
@@ -787,7 +968,7 @@ class TestBedrockSigning:
 
         def signing_edge() -> CacheEdge:
             return CacheEdge(
-                store, SECRET, test_key=lambda: TEST_KEY,
+                store, SECRET,
                 policies={BEDROCK_MOUNT: MountPolicy(sign=varying, unkeyed_headers=SIGNATURE_HEADERS)},
             )
 
@@ -1065,7 +1246,8 @@ def test_connection_failure_releases_capture_lease(store: RedisResponseStore) ->
         unavailable.bind(("127.0.0.1", 0))
         url: Final = f"http://127.0.0.1:{unavailable.getsockname()[1]}/v1/chat/completions"
         cache: Final = cache_edge(store)
-        assert isinstance(cache.forward("openai", "POST", url, dict(HEADERS), BODY, 0.2), NetworkError)
+        head: Final = cache.forward("openai", "POST", url, dict(HEADERS), BODY, 0.2, test_key=TEST_SLUG)
+        assert isinstance(head, NetworkError)
         key: Final = slot_key(url)
         lease: Final = store.lookup(key)
         assert isinstance(lease, CaptureLease)
@@ -1076,7 +1258,7 @@ def test_connection_failure_releases_capture_lease(store: RedisResponseStore) ->
 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 = cache_edge(store)
-    head: Final = cache.forward("openai", "POST", url, dict(HEADERS), BODY, 5)
+    head: Final = cache.forward("openai", "POST", url, dict(HEADERS), BODY, 5, test_key=TEST_SLUG)
     assert isinstance(head, StreamHead)
     head.steps.close()
     key: Final = slot_key(url)
@@ -1094,7 +1276,7 @@ def test_effective_account_change_cannot_reuse_cache(
         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("openai", "POST", url, dict(HEADERS), BODY, 5)
+        head = cache.forward("openai", "POST", url, dict(HEADERS), BODY, 5, test_key=TEST_SLUG)
         assert isinstance(head, StreamHead)
         assert b"".join(step.data for step in head.steps if isinstance(step, StreamChunk)) == SUCCESS
     assert len(provider.hits) == 2

tests/e2e/PROVIDER_CACHE.md

@@ -1,6 +1,6 @@
 # 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, as do Anthropic-on-Bedrock registrations that carry no AWS identity of their own. Existing custom API bases, named credentials, mocked models and realtime WebSocket deployments keep their existing routing. Other provider protocols remain live
+`E2E_PROVIDER_CACHE=1` enables automatic response reuse in the live E2E mode. Standard OpenAI and Anthropic model registrations made from inside a test, or while a module- or class-scoped fixture sets one up for its tests, use the provider edge, as do Anthropic-on-Bedrock registrations that carry no AWS identity of their own. A registration made from a session-scoped fixture or outside any test, or with `provider_live=True`, keeps its real provider path. 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`, `/v1/messages`, `/v1/embeddings` and `/v1/responses` on the OpenAI and Anthropic mounts, SSE streams included, and for `/model/{id}/converse` and `/model/{id}/invoke` on a Bedrock mount. Unsupported endpoints pass through. Each endpoint family has its own completeness rule, so a truncated embedding or a Responses run that never reached `response.completed` is not stored
 
@@ -10,13 +10,13 @@ Two details of that rule are worth knowing before changing it. A ConverseStream
 
 ## Request identity
 
-A recording belongs to one test. The key is a keyed digest over the test's node id, the method, the URL, the effective outbound headers (including authentication and HTTP-library defaults), body presence and the body bytes, with one normalization: a 12-hex-digit run, the shape `unique_marker()` mints, is replaced by a placeholder in both the URL and a UTF-8 body. Nothing else is normalized away. No prompts, JSON values or credentials are rewritten, and the rule is the one `fixture_canonical.py` already applies for record/replay, so there is a single definition of what a marker is
+A recording belongs to one test, and the test is named by the deployment rather than by the process. A deployment registered from inside a test gets the cache edge's mount URL with a test segment appended, `{edge}/{mount}/t/{slug}`, where the slug is `slug_for_test` of the registering test's node id, and the edge reads that segment off every request before forwarding. A deployment a module- or class-scoped fixture sets up is owned by that module or class instead: every test in it shares the deployment, `--dist loadfile` keeps those tests in one worker, and the slot index below keeps their calls apart. A session-scoped fixture runs in every worker, so its deployment has no owner and stays live; `driver_models` in `quota_management/spend_tracking/conftest.py` is the main one. The owner is read off the fixture request in `fixture_mode.registration_owner`, never off the process's `PYTEST_CURRENT_TEST`, which during a shared fixture's setup names whichever test happened to ask first. The key is a keyed digest over that slug, the method, the upstream URL, the effective outbound headers (including authentication and HTTP-library defaults), body presence and the body bytes, with one normalization: a 12-hex-digit run, the shape `unique_marker()` mints, is replaced by a placeholder in both the URL and a UTF-8 body. Nothing else is normalized away. No prompts, JSON values or credentials are rewritten, and the rule is the one `fixture_canonical.py` already applies for record/replay, so there is a single definition of what a marker is
 
 Requests that differ only by their markers therefore share a canonical identity, which is what makes the cache reusable across builds: every e2e test salts its prompt afresh, so an exact-byte key would miss on every call. Within one test, calls that share a canonical identity are still recorded and replayed separately, by a FIFO slot index appended to the key. That matters because a replayed response carries the recorded provider response id, `LiteLLM_SpendLogs.request_id` is that id, and one shared recording answering two calls would collapse two spend rows into one
 
-Two different tests never share a recording, and a provider call made outside any test (fixtures, session setup) is never cached, because the identity has no test node id to bind to
+Two different tests never share a recording. A request that reaches the edge without a test segment is forwarded live and never cached, and the edge never names the test from its own process's `PYTEST_CURRENT_TEST`. It used to, and that was wrong whenever the calling test and the serving process differed: the proxy is a separate pod, and under xdist the Claude Code compat matrix registered its shared aliases from every worker, each pointing at that worker's edge, so the router spread one worker's calls across all of them and each call was keyed on whatever test the serving worker was in. Builds 234 and 235 of the e2e pipeline, same commit, credited the same Bedrock request to unrelated tests 92% of the time, which is why that mount never converged
 
-A client that varies its own request between runs defeats that identity without breaking any rule, and the Claude Code compat cells did. The CLI sends a device id and a session id in `metadata.user_id`, and its system prompt names both its memory directory and its working directory, adding the branch and recent commits when that directory is a git repository. Driven with a fresh HOME and the checkout as its working directory, every cell sent different bytes every build. The fix belongs in the driver rather than here: `claude_code/cli_driver.py` pins the config directory, the working directory and both identifiers, which is why the cache needs no rule for any of it. Normalizing them instead would have hidden a real defect class, since a rule cannot tell a client's own churn from a value a test means to assert on
+The Claude Code compat cells are not cached. Their aliases are registered once per worker session and shared by every cell, so no call to them belongs to one test, and the matrix exists to prove the real CLI against real providers; `claude_code/conftest.py` registers them with `provider_live=True`. The driver still pins the CLI's config directory, working directory, device id and session id (`_driver_unit_tests/test_request_determinism.py` holds that), so a CLI-driven deployment registered by one test would send stable bytes. Normalizing those values in the key instead would hide a real defect class, since a rule cannot tell a client's own churn from a value a test means to assert on
 
 Provider `Set-Cookie` headers are dropped before validation and never recorded: the edge already withholds them from the proxy, and OpenAI responses always carry Cloudflare bot-management cookies
 
@@ -62,4 +62,4 @@ Provider remaining-quota headers describe the captured response. Metrics derived
 
 ## Qualification
 
-`tests/code_coverage_tests/test_provider_cache.py` exercises local HTTP providers and disposable real Redis, including the marker-canonical key, the FIFO slot index, per-test isolation, SigV4 re-signing against a local upstream, and each endpoint's completeness rule. 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/code_coverage_tests/test_provider_cache.py` exercises local HTTP providers and disposable real Redis, including the marker-canonical key, the FIFO slot index, per-test isolation, attribution from the deployment's test segment whatever the serving process is running, SigV4 re-signing against a local upstream, and each endpoint's completeness rule. 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/claude_code/conftest.py

@@ -600,10 +600,13 @@ def _build_control_plane_client(proxy_config: ProxyConfig):
 
 def _register_deployment(proxy, deployment: CompatDeployment) -> str:
     """Register one deployment and return its proxy-assigned model_id
-    once it is servable on the data plane."""
+    once it is servable on the data plane. The aliases are shared by every
+    cell and, under xdist, by every worker, so no call to them belongs to
+    one test and none is cached: the matrix exists to reach real providers."""
     return proxy.create_model(
         deployment.model_name,
         deployment.litellm_params,
+        provider_live=True,
     )
 
 

tests/e2e/conftest.py

@@ -37,6 +37,7 @@ from e2e_config import (
 from e2e_db import RESET_OPT_IN_ENV, reset_spend_logs, run_spend_log_cleanup
 from e2e_http import unwrap
 from fixture_mode import fixture_mode_collection_error, fixture_report_lines
+from fixture_mode import pytest_fixture_setup as pytest_fixture_setup
 from idp import Identity, Keycloak, keycloak_from_env
 from junit_properties import attach_result_properties
 from lifecycle import ProxyClientProvider, ResourceManager
@@ -124,8 +125,7 @@ def pytest_configure(config: pytest.Config) -> None:
     )
     config.addinivalue_line(
         "markers",
-        "cli_determinism: drives the real claude CLI for several seconds, which widens the window in which "
-        "another test's in-flight upstream call is attributed to it; deselected unless E2E_CLI_DETERMINISM is set",
+        "cli_determinism: drives the real claude CLI for several seconds; deselected unless E2E_CLI_DETERMINISM is set",
     )
     config.addinivalue_line(
         "markers",

tests/e2e/e2e_config.py

@@ -13,7 +13,7 @@ from pathlib import Path
 from typing import Final
 
 from dotenv import load_dotenv
-from fixture_mode import deterministic_marker, parse_fixture_mode
+from fixture_mode import deterministic_marker, parse_fixture_mode, registration_owner
 from provider_edge import provider_edge_api_base
 
 # Local runs keep provider / DataDog keys in tests/e2e/.env (see CONTRIBUTING.md).
@@ -200,13 +200,17 @@ def datadog_mcp_url(*, toolsets: str = "core") -> str:
 def provider_edge_base(mount: str) -> str | None:
     """The api_base an edge-wired deployment should register with, using this
     process's fixture-mode and edge-host configuration: None in live mode, the
-    shared edge server's mount URL in record and replay."""
+    shared edge server's mount URL in record and replay, and with the shared
+    cache on, the cache edge's mount URL scoped to the node that owns the
+    deployment: the running test, or the module or class whose fixture is
+    setting it up."""
     return provider_edge_api_base(
         mount,
         mode_raw=FIXTURE_MODE_RAW,
         bundle_dir=FIXTURE_DIR,
         bind_host=PROVIDER_EDGE_BIND_HOST,
         advertise_host=PROVIDER_EDGE_ADVERTISE_HOST,
+        test_key=registration_owner(),
         forward_timeout=REQUEST_TIMEOUT,
     )
 

tests/e2e/fixture_mode.py

@@ -14,11 +14,14 @@ from __future__ import annotations
 
 import hashlib
 import os
+from collections.abc import Generator
+from contextvars import ContextVar
 from dataclasses import dataclass
 from datetime import datetime
 from pathlib import Path
 from typing import Final, Literal, assert_never
 
+import pytest
 from fixture_bundle import (
     FreshBundle,
     StaleBundle,
@@ -59,6 +62,31 @@ def current_test_key() -> str:
     return raw.rsplit(" (", 1)[0]
 
 
+REGISTRATION_OWNER: Final[ContextVar[str | None]] = ContextVar("registration_owner", default=None)
+
+
+def registration_owner() -> str:
+    """The pytest node that owns a deployment registered right now. While a
+    fixture is being set up that is the node the fixture is scoped to: the module
+    or class for a fixture its tests share, and ``session`` for a session- or
+    package-scoped one, which every xdist worker sets up and no node can own.
+    Anywhere else it is the running test."""
+    owner = REGISTRATION_OWNER.get()
+    return current_test_key() if owner is None else owner
+
+
+@pytest.hookimpl(wrapper=True)
+def pytest_fixture_setup(request: pytest.FixtureRequest) -> Generator[None, object, object]:
+    node: Final = request.node  # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType]  # pytest: untyped
+    assert isinstance(node, pytest.Item | pytest.Collector)
+    owner: Final = SESSION_TEST_KEY if request.scope in ("session", "package") else node.nodeid
+    token: Final = REGISTRATION_OWNER.set(owner)
+    try:
+        return (yield)
+    finally:
+        REGISTRATION_OWNER.reset(token)
+
+
 class ReplayMiss(AssertionError):
     """Replay had no recorded interaction for a provider call the proxy made.
     The suite drifted from the bundle (or the bundle from the suite): re-record."""

tests/e2e/provider_cache.py

@@ -27,8 +27,8 @@ from e2e_http import (
     prepare_forward,
     primed_steps,
 )
+from fixture_bundle import slug_for_test
 from fixture_canonical import MARKER_PATTERN, MARKER_PLACEHOLDER
-from fixture_mode import SESSION_TEST_KEY, current_test_key
 from pydantic import BaseModel, ConfigDict, JsonValue, TypeAdapter, ValidationError
 
 LIFETIME_SECONDS: Final = 86_400
@@ -58,6 +58,19 @@ EVENT_TYPE_HEADER: Final = ":event-type"
 EVENTSTREAM_HEADERS: Final[TypeAdapter[dict[str, str]]] = TypeAdapter(dict[str, str])
 OPENAI_JSON_PATHS: Final = frozenset({"/v1/chat/completions", "/v1/messages", "/v1/embeddings", "/v1/responses"})
 JSON_VALUE: Final[TypeAdapter[JsonValue]] = TypeAdapter(JsonValue)
+TEST_SEGMENT: Final = "t"
+
+
+def scoped_edge_base(base: str, test_key: str) -> str:
+    return f"{base}/{TEST_SEGMENT}/{slug_for_test(test_key)}"
+
+
+def split_test_segment(upstream_path: str) -> tuple[str | None, str]:
+    head, _, rest = upstream_path.partition("/")
+    if head != TEST_SEGMENT:
+        return None, upstream_path
+    slug, _, remainder = rest.partition("/")
+    return slug or None, remainder
 
 
 @dataclass(frozen=True, slots=True)
@@ -517,7 +530,6 @@ class CacheEdge:
     wait_seconds: float = 2.0
     clock: Callable[[], float] = time.monotonic
     sleep: Callable[[float], None] = time.sleep
-    test_key: Callable[[], str] = current_test_key
 
     def lookup(self, key: str) -> CacheLookup:
         deadline: Final = self.clock() + self.wait_seconds
@@ -555,9 +567,9 @@ class CacheEdge:
 
     def forward(
         self, mount: str, method: str, url: str, headers: dict[str, str], body: bytes | None, timeout: float,
+        *, test_key: str | None,
     ) -> StreamHead | NetworkError:
-        test_key: Final = self.test_key()
-        if test_key == SESSION_TEST_KEY or not cacheable_endpoint(mount, method, url, body):
+        if test_key is None or not cacheable_endpoint(mount, method, url, body):
             self.count(mount, "bypass")
             self.count(mount, "upstream_attempts")
             return forward_stream(

tests/e2e/provider_edge.py

@@ -88,13 +88,21 @@ from fixture_canonical import (
 )
 from fixture_mode import (
     FIXTURE_MODES,
+    SESSION_TEST_KEY,
     InvalidFixtureMode,
     ReplayMiss,
     current_test_key,
     parse_fixture_mode,
 )
 from fixture_profile import IneligibleRequest, MatchProfile, match_profile, strict_identity
-from provider_cache import SIGNATURE_HEADERS, CacheEdge, MountPolicy, is_bedrock
+from provider_cache import (
+    SIGNATURE_HEADERS,
+    CacheEdge,
+    MountPolicy,
+    is_bedrock,
+    scoped_edge_base,
+    split_test_segment,
+)
 from provider_cache_routing import LIVE_PROVIDER_REQUIRED
 from pydantic import JsonValue, TypeAdapter
 
@@ -778,14 +786,14 @@ def _handle_record(
 
 def _handle_live(
     method: str, url: str, headers: Mapping[str, str], body: bytes | None, timeout: float,
-    cache: CacheEdge | None = None, mount: str = "",
+    cache: CacheEdge | None = None, mount: str = "", test_key: str | 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)
-        if cache is None else cache.forward(mount, method, url, forwarded, body, timeout)
+        if cache is None else cache.forward(mount, method, url, forwarded, body, timeout, test_key=test_key)
     )
     match head:
         case NetworkError(message=message):
@@ -826,7 +834,7 @@ def handle_edge_request(
         return _text_reply(404, f"unknown provider mount {unknown!r}; known mounts: {', '.join(sorted(mounts))}")
     mount: Final = resolved.mount
     upstream_base: Final = resolved.upstream_base
-    upstream_path: Final = resolved.upstream_path
+    test_key, upstream_path = split_test_segment(resolved.upstream_path)
     profile: Final = (
         backend.recorder.profile
         if isinstance(backend, RecordEdge)
@@ -858,7 +866,7 @@ def handle_edge_request(
         case CacheEdge():
             return _handle_live(
                 method, _upstream_url(upstream_base, upstream_path, split.query), headers, body, timeout,
-                backend, mount,
+                backend, mount, test_key,
             )
         case LiveEdge():
             return _handle_live(
@@ -1093,19 +1101,26 @@ def provider_edge_api_base(
     bundle_dir: Path,
     bind_host: str,
     advertise_host: str,
+    test_key: str,
     forward_timeout: float = 60.0,
 ) -> str | None:
     """The api_base a suite gives an edge-wired deployment: None in live mode
     (the deployment keeps its real provider api_base) and the process-wide edge
-    server's mount URL in record and replay, booting the server on first use."""
+    server's mount URL in record and replay, booting the server on first use.
+    With the shared cache configured, live mode answers with the cache edge's
+    mount URL scoped to ``test_key``, the node that owns the deployment, and
+    None for a deployment no node owns, since a call nobody can attribute is
+    never cached."""
     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":
-            if configured_cache_backend() is not None:
-                return _shared_cache_edge(bind_host, advertise_host, forward_timeout).api_base(mount)
-            return None
+            if configured_cache_backend() is None or test_key == SESSION_TEST_KEY:
+                return None
+            return scoped_edge_base(
+                _shared_cache_edge(bind_host, advertise_host, forward_timeout).api_base(mount), test_key
+            )
         case "record" | "replay":
             if is_bedrock(mount):
                 return None

tests/e2e/proxy_client.py

@@ -613,6 +613,8 @@ class ProxyClient:
         model_name: str,
         litellm_params: LiteLLMParamsBody,
         mode: ModelMode | None = None,
+        *,
+        provider_live: bool = False,
     ) -> str:
         """Register a deployment under `model_name` and return its proxy-assigned
         model_id, once the model is actually servable on the data plane."""
@@ -621,15 +623,20 @@ class ProxyClient:
                 model_name=model_name,
                 litellm_params=litellm_params,
                 model_info=ModelInfoBody(mode=mode),
-            )
+            ),
+            provider_live=provider_live,
         )
 
-    def register_model(self, body: ModelNewBody, listed_for: str | None = None) -> str:
+    def register_model(
+        self, body: ModelNewBody, listed_for: str | None = None, *, provider_live: bool = False
+    ) -> str:
         """`create_model` for deployments that carry more than a mode: access groups,
         team scoping, a pinned id. `listed_for` is the virtual key whose /v1/models
         view must list the deployment before it counts as servable, because a
         team-scoped deployment is listed to its own team and to nobody else, master
-        key included; leave it unset for a proxy-wide model.
+        key included; leave it unset for a proxy-wide model. `provider_live` keeps
+        the deployment on its real provider path whatever the cache setting, for a
+        deployment shared across tests or workers, which no one test could own.
 
         /model/new is a control-plane route; the data plane (which serves /chat,
         /ocr, ...) only picks the new model up on its next DB reload, so a call
@@ -650,7 +657,8 @@ class ProxyClient:
                 headers=self.management_headers(),
                 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,
+                    enabled=os.environ.get("E2E_PROVIDER_CACHE", "0") == "1" and not provider_live,
+                    mode=body.model_info.mode,
                 )}),
                 response_type=ModelNewResponse,
             )

tests/e2e/pytest.ini

@@ -10,5 +10,5 @@ markers =
     weekly: real-provider anomaly load test that spends real money; deselected unless E2E_WEEKLY_ANOMALY is set
     managed_files: needs a proxy running with require_managed_files enabled; deselected unless E2E_MANAGED_FILES_STACK is set
     prompt_caching_stack: needs a proxy running with router_settings.optional_pre_call_checks including prompt_caching; deselected unless E2E_PROMPT_CACHING_STACK is set
-    cli_determinism: drives the real claude CLI for several seconds, which widens the window in which another test's in-flight upstream call is attributed to it; deselected unless E2E_CLI_DETERMINISM is set
+    cli_determinism: drives the real claude CLI for several seconds; deselected unless E2E_CLI_DETERMINISM is set
     redis_chaos: load test that pauses the proxy's Redis outright mid-run; needs a proxy booted from gateway/redis_chaos_ci_config.yml on the same host, and is deselected unless E2E_REDIS_CHAOS is set

tests/e2e/test_provider_edge.py

@@ -1264,6 +1264,7 @@ class TestApiBaseSeam:
                     bundle_dir=tmp_path / "bundle",
                     bind_host="127.0.0.1",
                     advertise_host="127.0.0.1",
+                    test_key="tests/e2e/synthetic_suite.py::test_case",
                 )
                 is None
             )
@@ -1276,6 +1277,7 @@ class TestApiBaseSeam:
                 bundle_dir=tmp_path / "bundle",
                 bind_host="127.0.0.1",
                 advertise_host="127.0.0.1",
+                test_key="tests/e2e/synthetic_suite.py::test_case",
             )
 
     def test_unknown_mount_raises_naming_the_known_mounts(self, tmp_path: Path) -> None:
@@ -1286,6 +1288,7 @@ class TestApiBaseSeam:
                 bundle_dir=tmp_path / "bundle",
                 bind_host="127.0.0.1",
                 advertise_host="127.0.0.1",
+                test_key="tests/e2e/synthetic_suite.py::test_case",
             )
 
     @pytest.mark.parametrize("mode_raw", ["record", "replay"])
@@ -1301,15 +1304,18 @@ class TestApiBaseSeam:
             bundle_dir=tmp_path / "bundle",
             bind_host="127.0.0.1",
             advertise_host="127.0.0.1",
+            test_key="tests/e2e/synthetic_suite.py::test_case",
         ) is None
 
     def test_record_mode_boots_one_shared_edge_and_prepares_the_bundle(self, tmp_path: Path) -> None:
         root = tmp_path / "bundle"
         first = provider_edge_api_base(
-            "openai", mode_raw="record", bundle_dir=root, bind_host="127.0.0.1", advertise_host="127.0.0.1"
+            "openai", mode_raw="record", bundle_dir=root, bind_host="127.0.0.1", advertise_host="127.0.0.1",
+            test_key="tests/e2e/synthetic_suite.py::test_case",
         )
         second = provider_edge_api_base(
-            "anthropic", mode_raw="record", bundle_dir=root, bind_host="127.0.0.1", advertise_host="127.0.0.1"
+            "anthropic", mode_raw="record", bundle_dir=root, bind_host="127.0.0.1", advertise_host="127.0.0.1",
+            test_key="tests/e2e/synthetic_suite.py::test_case",
         )
         assert first is not None and second is not None
         assert first.endswith("/openai")