fix(e2e): own a shared fixture's deployment by the fixture's node, not the first test

A deployment registered while a module- or class-scoped fixture is being set up
was bound to whichever test asked for the fixture first, so every later test in
the module shared that partition. A session-scoped fixture is set up by every
xdist worker, so its deployment could never have one owner at all.

The e2e conftest now wraps pytest_fixture_setup and records the node the fixture
is scoped to: registrations made during a module or class fixture's setup carry
that node's slug, and a session- or package-scoped one has no owner and stays
live. The registration seam test moves from tests/e2e to the cache harness tests
beside the rest of the attribution coverage.

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,13 +18,14 @@ 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
@@ -49,7 +51,7 @@ from provider_cache_routing import (
     bedrock_region,
     route_cache_model,
 )
-from fixture_mode import SESSION_TEST_KEY
+from fixture_mode import SESSION_TEST_KEY, current_test_key, registration_owner
 from provider_edge import (
     EDGE_MOUNTS,
     configured_cache_backend,
@@ -58,6 +60,7 @@ from provider_edge import (
     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"
@@ -582,6 +585,97 @@ def test_the_cache_edge_base_is_scoped_to_the_registering_test(
         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(
     store: RedisResponseStore, provider: Provider,
 ) -> None:

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 made from inside a test use the provider edge, as do Anthropic-on-Bedrock registrations that carry no AWS identity of their own. A registration made 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
+`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,7 +10,7 @@ Two details of that rule are worth knowing before changing it. A ConverseStream
 
 ## Request identity
 
-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. 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
+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
 

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

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 current_test_key, 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).
@@ -201,14 +201,16 @@ 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, and with the shared
-    cache on, the cache edge's mount URL scoped to the running test."""
+    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=current_test_key(),
+        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_edge.py

@@ -1108,8 +1108,9 @@ def provider_edge_api_base(
     (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.
     With the shared cache configured, live mode answers with the cache edge's
-    mount URL scoped to ``test_key``, the test registering the deployment, and
-    None outside any test, since a call nobody can attribute is never cached."""
+    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):

tests/e2e/test_proxy_client.py

@@ -26,8 +26,6 @@ from typing import Final, cast
 import pytest
 from e2e_config import parse_replica_urls
 from e2e_http import NoBody, Result, Success, without_retries
-from fixture_bundle import slug_for_test
-from fixture_mode import current_test_key
 from idp import Keycloak
 from lifecycle import ResourceManager
 from management.jwt_actors import ActorFactory
@@ -40,7 +38,6 @@ from models import (
     KeyInfoResponse,
     KeyUpdateBody,
     LiteLLMParamsBody,
-    ModelNewBody,
     McpServerCreateBody,
     McpServerUpdateBody,
     ModelListEntry,
@@ -584,36 +581,3 @@ def test_partial_updates_preserve_explicit_null_at_the_http_boundary(operation:
         )
         assert json.loads(bodies.get_nowait()) == expected
         assert bodies.empty()
-
-
-@pytest.mark.parametrize("provider_live", (False, True))
-def test_registration_binds_the_deployment_to_this_test_unless_it_is_provider_live(
-    provider_live: bool, monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    """With the shared cache on, a deployment registered from inside a test carries
-    this test's segment in its api_base, which is how the edge knows whose recording
-    a call belongs to. `provider_live` is the opt-out for a deployment no single test
-    owns: it goes to the proxy exactly as written."""
-    from provider_cache_redis import configured_cache
-
-    monkeypatch.setenv("E2E_PROVIDER_CACHE", "1")
-    monkeypatch.setenv("E2E_PROVIDER_CACHE_REDIS_URL", "redis://127.0.0.1:1/0")
-    monkeypatch.setenv("E2E_PROVIDER_CACHE_HMAC_KEY", "synthetic-cache-hmac-key-for-tests")
-    monkeypatch.setenv("E2E_PROVIDER_CACHE_NAMESPACE", "registration-seam")
-    configured_cache.cache_clear()
-    bodies: Final[SimpleQueue[bytes]] = SimpleQueue()
-    try:
-        with caller_boundary(status=401, bodies=bodies) as (bootstrap, _), without_retries():
-            with pytest.raises(AssertionError):
-                bootstrap.proxy.create_model(
-                    "owned", LiteLLMParamsBody(model="openai/synthetic"), provider_live=provider_live
-                )
-    finally:
-        configured_cache.cache_clear()
-    sent: Final = ModelNewBody.model_validate_json(bodies.get_nowait())
-    assert bodies.empty()
-    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")