test(integration): give each owned proxy a private database and Redis by default (#47213)

* test(integration): give each owned proxy a private database and Redis by default

owned_proxy and owned_proxy_process now boot on a scratch database and a
per-proxy redis-server unless the caller opts into the worker's state with
state=OwnedState.SHARED_WITH_WORKER, passes its own DATABASE_URL, or strips
DATABASE_URL. Under the replica profile the private database gets the
writer and reader roles so the reader path stays covered.

Gateway carries the database_url its proxy writes to, and the Scenario
cleanup checks and delete_key_if_present read from it, so they keep
checking something on a private database.

The 189 test files with owned proxy sites that depend on the worker's
database or Redis opt in with a shares_worker_state module marker, which a
module-scoped conftest fixture turns into the shared state. The 7 helper
modules with such sites opt in per call, since conftest fixtures and other
test files reach them.

* test(integration): reach session fixtures with the shared marker and keep the worker retry default

The module-scoped fixture that applied shares_worker_state ran after
session-scoped fixtures, so proxies those booted went private in marked
files. The marker is now applied around each marked item's setup, call
and teardown, which covers fixtures of every scope.

Owned proxies on the worker database inherited the worker gateway's
router_settings.num_retries of 0 from it. A private proxy now gets the same
default in its config copy unless the config sets num_retries itself.

Five providers files read the owned proxy's spend rows or response cache
through the worker's database or Redis and opt into sharing.

tests/integration/_support/actors.py

@@ -15,7 +15,7 @@ from pydantic import JsonValue
 from litellm.proxy.utils import PrismaClient, hash_token
 from tests.integration._support.client import Gateway, Scenario, string_value
 from tests.integration._support.prisma_clients import connected_proxy_prisma
-from tests.integration._support.process import owned_proxy
+from tests.integration._support.process import OwnedState, owned_proxy
 
 TEAM_ALPHA: Final = "behavior-pin-team-alpha"
 TEAM_BETA: Final = "behavior-pin-team-beta"
@@ -124,7 +124,7 @@ def owned_settings_client(
     config: Final = {**base, "litellm_settings": {**base["litellm_settings"], **litellm_settings}}
     path: Final = directory / "owned-settings.yaml"
     path.write_text(yaml.safe_dump(config))
-    with owned_proxy(client.gateway, directory, {}, config=path) as candidate:
+    with owned_proxy(client.gateway, directory, {}, config=path, state=OwnedState.SHARED_WITH_WORKER) as candidate:
         yield GatewayClient(candidate, client.replacements)
 
 

tests/integration/_support/bedrock_discovery.py

@@ -23,7 +23,7 @@ from integration._support.aws_control_plane import (
     verified_scope,
 )
 from integration._support.client import JSON_OBJECT, Gateway, Scenario, eventually, held, object_value, string_value
-from integration._support.process import OwnedProxy, owned_proxy_process
+from integration._support.process import OwnedProxy, OwnedState, owned_proxy_process
 from litellm.litellm_core_utils.aws_partition import get_aws_dns_suffix
 from pydantic import JsonValue
 
@@ -115,6 +115,7 @@ def discovery_proxy(
         config=discovery_config(parent, directory, check_provider_endpoint=check_provider_endpoint),
         remove_environment=aws_environment_names(),
         workers=workers,
+        state=OwnedState.SHARED_WITH_WORKER,
     ) as owned:
         wait_for_every_worker(owned.log, workers)
         yield owned

tests/integration/_support/client.py

@@ -35,9 +35,9 @@ def list_value(value: JsonValue) -> Sequence[JsonValue]:
 
 def delete_key_if_present(candidate: Gateway, key: str) -> None:
     digest: Final = sha256(key.encode()).hexdigest()
-    if read_rows('SELECT token FROM "LiteLLM_VerificationToken" WHERE token=%s', (digest,)):
+    if candidate.rows('SELECT token FROM "LiteLLM_VerificationToken" WHERE token=%s', (digest,)):
         candidate.post("/key/delete", {"keys": [key]})
-    assert read_rows('SELECT token FROM "LiteLLM_VerificationToken" WHERE token=%s', (digest,)) == []
+    assert candidate.rows('SELECT token FROM "LiteLLM_VerificationToken" WHERE token=%s', (digest,)) == []
 
 
 def eventually(
@@ -95,6 +95,10 @@ class Gateway:
     client: httpx.Client
     key: str
     upstream_url: str
+    database_url: str | None = None
+
+    def rows(self, query: str, parameters: tuple[str, ...]) -> list[dict[str, JsonValue]]:
+        return read_rows(query, parameters, database_url=self.database_url)
 
     def request(
         self,
@@ -175,7 +179,7 @@ class Scenario:
 
     def delete_team(self, identity: str) -> None:
         self.gateway.post("/team/delete", {"team_ids": [identity]})
-        assert read_rows('SELECT team_id FROM "LiteLLM_TeamTable" WHERE team_id = %s', (identity,)) == []
+        assert self.gateway.rows('SELECT team_id FROM "LiteLLM_TeamTable" WHERE team_id = %s', (identity,)) == []
 
     def project(self, team_id: str, **fields: JsonValue) -> str:
         created: Final = self.gateway.post(
@@ -188,7 +192,9 @@ class Scenario:
     def delete_project(self, identity: str) -> None:
         response: Final = self.gateway.request("DELETE", "/project/delete", {"project_ids": [identity]})
         assert response.status_code == 200, response.text
-        assert read_rows('SELECT project_id FROM "LiteLLM_ProjectTable" WHERE project_id = %s', (identity,)) == []
+        assert (
+            self.gateway.rows('SELECT project_id FROM "LiteLLM_ProjectTable" WHERE project_id = %s', (identity,)) == []
+        )
 
     def budget(self, **fields: JsonValue) -> str:
         created: Final = self.gateway.post("/budget/new", fields)
@@ -198,7 +204,7 @@ class Scenario:
 
     def delete_budget(self, identity: str) -> None:
         self.gateway.post("/budget/delete", {"id": identity})
-        assert read_rows('SELECT budget_id FROM "LiteLLM_BudgetTable" WHERE budget_id = %s', (identity,)) == []
+        assert self.gateway.rows('SELECT budget_id FROM "LiteLLM_BudgetTable" WHERE budget_id = %s', (identity,)) == []
 
     def organization(self, **fields: JsonValue) -> str:
         created: Final = self.gateway.post(
@@ -212,7 +218,9 @@ class Scenario:
         response: Final = self.gateway.request("DELETE", "/organization/delete", {"organization_ids": [identity]})
         assert response.status_code == 200, response.text
         assert (
-            read_rows('SELECT organization_id FROM "LiteLLM_OrganizationTable" WHERE organization_id = %s', (identity,))
+            self.gateway.rows(
+                'SELECT organization_id FROM "LiteLLM_OrganizationTable" WHERE organization_id = %s', (identity,)
+            )
             == []
         )
         self.delete_budget(budget_id)
@@ -236,7 +244,7 @@ class Scenario:
     def delete_user(self, identity: str) -> None:
         response: Final = self.gateway.request("POST", "/user/delete", {"user_ids": [identity]})
         assert response.status_code == 200 and response.json() == 1, response.text
-        assert read_rows('SELECT user_id FROM "LiteLLM_UserTable" WHERE user_id = %s', (identity,)) == []
+        assert self.gateway.rows('SELECT user_id FROM "LiteLLM_UserTable" WHERE user_id = %s', (identity,)) == []
 
     def member(self, team_id: str, role: str = "user") -> str:
         """Create an internal user and add them to ``team_id``; deleting the user later removes the membership."""
@@ -247,7 +255,7 @@ class Scenario:
     def delete_key(self, token: str) -> None:
         self.gateway.post("/key/delete", {"keys": [token]})
         hashed: Final = sha256(token.encode()).hexdigest()
-        assert read_rows('SELECT token FROM "LiteLLM_VerificationToken" WHERE token = %s', (hashed,)) == []
+        assert self.gateway.rows('SELECT token FROM "LiteLLM_VerificationToken" WHERE token = %s', (hashed,)) == []
         info: Final = object_value(self.gateway.get("/key/info", {"key": hashed})["info"])
         assert info["status"] == "deleted", f"Deleted key still served as live: {info['status']}"
 
@@ -265,7 +273,9 @@ class Scenario:
             lambda served: identity not in served,
             seconds=2 * PROXY_CONFIG_RELOAD_INTERVAL_SECONDS,
         )
-        assert read_rows('SELECT model_id FROM "LiteLLM_ProxyModelTable" WHERE model_id = %s', (identity,)) == []
+        assert (
+            self.gateway.rows('SELECT model_id FROM "LiteLLM_ProxyModelTable" WHERE model_id = %s', (identity,)) == []
+        )
 
     def model(self, *, model_info: Mapping[str, JsonValue] | None = None, **parameters: JsonValue) -> str:
         name: Final = f"integration-{uuid.uuid4().hex}"

tests/integration/_support/process.py

@@ -7,9 +7,9 @@ import subprocess
 import sys
 import time
 import uuid
-from collections.abc import Generator, Iterable, Iterator, Mapping
+from collections.abc import Callable, Generator, Iterable, Iterator, Mapping
 from concurrent.futures import ThreadPoolExecutor
-from contextlib import contextmanager, suppress
+from contextlib import AbstractContextManager, contextmanager, suppress
 from dataclasses import dataclass
 from enum import Enum
 from datetime import UTC, datetime, timedelta
@@ -17,7 +17,7 @@ from ipaddress import IPv4Address
 from itertools import accumulate
 from pathlib import Path
 from types import MappingProxyType
-from typing import TYPE_CHECKING, Final
+from typing import TYPE_CHECKING, Final, TypeAlias
 from urllib.parse import urlsplit
 
 import httpx
@@ -33,6 +33,8 @@ from tests.integration._support.budget import import_seconds, readiness_for, rea
 from tests.integration._support.client import GATEWAY_LIMITS, Gateway, gateway_from_environment
 from tests.integration._support.database import scratch_database
 from tests.integration._support.ports import allocate_port
+from tests.integration._support.redis_process import owned_redis
+from tests.integration._support.workers import gateway_profile, replica_urls
 
 if TYPE_CHECKING:
     import pytest
@@ -43,6 +45,75 @@ LEGACY_MIGRATE_DEPLOY: Final = ("--use_legacy_migration_resolver",)
 CONFIGURATION: Final = TypeAdapter(dict[str, object])
 _DEPLOYMENT_IDS: Final = TypeAdapter(dict[str, list[str]])
 MCP_COVERAGE_CONFIG: Final = Path(__file__).resolve().parents[1] / "mcp_coverage.toml"
+OWNED_STATE_VARIABLE: Final = "INTEGRATION_OWNED_STATE"
+
+
+class OwnedState(Enum):
+    PRIVATE = "private"
+    SHARED_WITH_WORKER = "shared_with_worker"
+
+
+@dataclass(frozen=True, slots=True)
+class PrivateState:
+    database_url: str
+    environment: Mapping[str, str]
+
+
+PrivateStateFactory: TypeAlias = Callable[[Path, Mapping[str, str]], AbstractContextManager[PrivateState]]
+
+
+def boots_private_state(
+    state: OwnedState | None,
+    overrides: Mapping[str, str],
+    remove_environment: tuple[str, ...],
+    inherited: Mapping[str, str],
+) -> bool:
+    chosen: Final = state or OwnedState(inherited.get(OWNED_STATE_VARIABLE, OwnedState.PRIVATE.value))
+    return chosen is OwnedState.PRIVATE and "DATABASE_URL" not in overrides and "DATABASE_URL" not in remove_environment
+
+
+@contextmanager
+def private_state(directory: Path, inherited: Mapping[str, str]) -> Iterator[PrivateState]:
+    with scratch_database("owned") as database, owned_redis(directory, inherited=inherited) as cache:
+        replica: Final = replica_urls(database, inherited) if gateway_profile(inherited).replica else None
+        yield PrivateState(
+            database,
+            MappingProxyType(
+                {
+                    "DATABASE_URL": database if replica is None else replica["INTEGRATION_PROXY_DATABASE_URL"],
+                    **(
+                        {}
+                        if replica is None
+                        else {"DATABASE_URL_READ_REPLICA": replica["INTEGRATION_PROXY_READ_REPLICA_URL"]}
+                    ),
+                    "REDIS_HOST": cache.host,
+                    "REDIS_PORT": str(cache.port),
+                }
+            ),
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class OwnedEnvironment:
+    overrides: Mapping[str, str]
+    database_url: str | None
+
+
+@contextmanager
+def owned_environment(
+    directory: Path,
+    overrides: Mapping[str, str],
+    remove_environment: tuple[str, ...],
+    state: OwnedState | None,
+    inherited: Mapping[str, str],
+    private: PrivateStateFactory = private_state,
+) -> Iterator[OwnedEnvironment]:
+    if not boots_private_state(state, overrides, remove_environment, inherited):
+        yield OwnedEnvironment(overrides, None)
+        return
+    with private(directory, inherited) as booted:
+        kept: Final = {name: value for name, value in booted.environment.items() if name not in remove_environment}
+        yield OwnedEnvironment(MappingProxyType({**kept, **overrides}), booted.database_url)
 
 
 def proxy_database_environment(inherited: Mapping[str, str] = os.environ) -> Mapping[str, str]:
@@ -368,6 +439,7 @@ def owned_proxy(
     database_setup: tuple[str, ...] = DB_PUSH,
     coverage_file: Path | None = None,
     inherited: Mapping[str, str] = os.environ,
+    state: OwnedState | None = None,
 ) -> Iterator[Gateway]:
     with owned_proxy_process(
         gateway,
@@ -379,20 +451,21 @@ def owned_proxy(
         database_setup=database_setup,
         coverage_file=coverage_file,
         inherited=inherited,
+        state=state,
     ) as owned:
         yield owned.gateway
 
 
 @contextmanager
 def isolated_gateway(directory: Path, environment: "pytest.MonkeyPatch") -> Iterator[Gateway]:
-    with gateway_from_environment() as shared, scratch_database() as database:
+    with gateway_from_environment() as shared, owned_proxy(shared, directory, {}, state=OwnedState.PRIVATE) as owned:
-        environment.setenv("DATABASE_URL", database)
+        assert owned.database_url is not None
-        environment.setenv("INTEGRATION_PROXY_DATABASE_URL", database)
+        environment.setenv("DATABASE_URL", owned.database_url)
+        environment.setenv("INTEGRATION_PROXY_DATABASE_URL", owned.database_url)
         environment.delenv("INTEGRATION_PROXY_READ_REPLICA_URL", raising=False)
         environment.delenv("DATABASE_URL_READ_REPLICA", raising=False)
-        with owned_proxy(shared, directory, {}) as owned:
+        owned.post("/config/update", {"router_settings": {"num_retries": 0}})
-            owned.post("/config/update", {"router_settings": {"num_retries": 0}})
+        yield owned
-            yield owned
 
 
 def _stop(process: subprocess.Popen[bytes]) -> None:
@@ -634,6 +707,32 @@ def _proxy_root() -> Path:
     return Path(os.environ.get("INTEGRATION_PROXY_ROOT") or Path(__file__).resolve().parents[3])
 
 
+def _configuration_copy(config: Path, directory: Path, prefix: str) -> Path:
+    checkouts: Final = (_harness_root(), _proxy_root().resolve())
+    inside_checkout: Final = any(config.resolve().is_relative_to(checkout) for checkout in checkouts)
+    target: Final = directory if inside_checkout else config.parent
+    target.mkdir(parents=True, exist_ok=True)
+    return target / f"{prefix}-{uuid.uuid4().hex}.yaml"
+
+
+def private_configuration(config: Path, directory: Path) -> Path:
+    loaded: Final[object] = yaml.safe_load(config.read_text())
+    if not isinstance(loaded, dict):
+        return config
+    settings: Final = CONFIGURATION.validate_python(loaded)
+    router: Final = settings.get("router_settings") or {}
+    if not isinstance(router, dict) or "num_retries" in router:
+        return config
+    destination: Final = _configuration_copy(config, directory, "private-config")
+    destination.write_text(
+        yaml.safe_dump(
+            {**settings, "router_settings": {**CONFIGURATION.validate_python(router), "num_retries": 0}},
+            sort_keys=False,
+        )
+    )
+    return destination
+
+
 def worker_configuration(config: Path, directory: Path, inherited: Mapping[str, str] = os.environ) -> Path:
     if inherited.get("INTEGRATION_WORKER_ISOLATION") != "1":
         return config
@@ -644,11 +743,7 @@ def worker_configuration(config: Path, directory: Path, inherited: Mapping[str,
         return f"http://127.0.0.1:{inherited['INTEGRATION_PROVIDER_PORT']}"
 
     contents: Final = re.sub(r"http://127[.]0[.]0[.]1:(8190|8191)(?![0-9])", endpoint, config.read_text())
-    checkouts: Final = (_harness_root(), _proxy_root().resolve())
+    destination: Final = _configuration_copy(config, directory, "worker-config")
-    inside_checkout: Final = any(config.resolve().is_relative_to(checkout) for checkout in checkouts)
-    target: Final = directory if inside_checkout else config.parent
-    target.mkdir(parents=True, exist_ok=True)
-    destination: Final = target / f"worker-config-{uuid.uuid4().hex}.yaml"
     loaded: Final[object] = yaml.safe_load(contents)
     if not isinstance(loaded, dict):
         destination.write_text(contents)
@@ -753,42 +848,58 @@ def owned_proxy_process(
     healthcheck: WorkerHealthcheck = WorkerHealthcheck.STRICT,
     inherited: Mapping[str, str] = os.environ,
     client_timeout: float = 15,
+    state: OwnedState | None = None,
 ) -> Iterator[OwnedProxy]:
     assert "--timeout_worker_healthcheck" not in extra_arguments, "Choose the worker healthcheck with `healthcheck`"
     root: Final = _proxy_root()
-    environment: Final = _proxy_environment(
+    with owned_environment(directory, overrides, remove_environment, state, inherited) as owned:
-        gateway,
+        configured: Final = worker_configuration(
-        {**overrides, **({"COVERAGE_FILE": str(coverage_file)} if coverage_file else {})},
+            config or root / "tests/integration/proxy_config.yaml", directory, inherited
-        remove_environment,
+        )
-        inherited,
+        owned_configuration: Final = (
-    )
+            configured if owned.database_url is None else private_configuration(configured, directory)
-    output: Final = Path(inherited.get("INTEGRATION_RESULTS_DIR", str(directory)))
+        )
-    output.mkdir(parents=True, exist_ok=True)
+        environment: Final = _proxy_environment(
-    command: Final = (
+            gateway,
-        sys.executable,
+            {**owned.overrides, **({"COVERAGE_FILE": str(coverage_file)} if coverage_file else {})},
-        "-m",
+            remove_environment,
-        *(("coverage", "run", f"--rcfile={MCP_COVERAGE_CONFIG}", "-m") if coverage_file else ()),
+            inherited,
-        "integration._support.proxy",
+        )
-        "--config",
+        output: Final = Path(inherited.get("INTEGRATION_RESULTS_DIR", str(directory)))
-        str(worker_configuration(config or root / "tests/integration/proxy_config.yaml", directory, inherited)),
+        output.mkdir(parents=True, exist_ok=True)
-        "--host",
+        command: Final = (
-        "127.0.0.1",
+            sys.executable,
-        "--num_workers",
+            "-m",
-        str(workers),
+            *(("coverage", "run", f"--rcfile={MCP_COVERAGE_CONFIG}", "-m") if coverage_file else ()),
-        "--timeout_worker_healthcheck",
+            "integration._support.proxy",
-        str(healthcheck_seconds(healthcheck)),
+            "--config",
-        *database_setup,
+            str(owned_configuration),
-        *extra_arguments,
+            "--host",
-    )
+            "127.0.0.1",
-    launch: Final = _launch_until_ready(command, root, environment, output, workers)
+            "--num_workers",
-    process: Final = launch.process
+            str(workers),
-    try:
+            "--timeout_worker_healthcheck",
-        with httpx.Client(
+            str(healthcheck_seconds(healthcheck)),
-            base_url=f"http://127.0.0.1:{launch.port}", timeout=client_timeout, trust_env=False, limits=GATEWAY_LIMITS
+            *database_setup,
-        ) as client:
+            *extra_arguments,
-            yield OwnedProxy(Gateway(client, gateway.key, gateway.upstream_url), process, launch.log, healthcheck)
+        )
-    finally:
+        launch: Final = _launch_until_ready(command, root, environment, output, workers)
-        _stop_launch(launch)
+        process: Final = launch.process
+        try:
+            with httpx.Client(
+                base_url=f"http://127.0.0.1:{launch.port}",
+                timeout=client_timeout,
+                trust_env=False,
+                limits=GATEWAY_LIMITS,
+            ) as client:
+                yield OwnedProxy(
+                    Gateway(client, gateway.key, gateway.upstream_url, owned.database_url),
+                    process,
+                    launch.log,
+                    healthcheck,
+                )
+        finally:
+            _stop_launch(launch)
 
 
 @contextmanager

tests/integration/_support/prometheus_series.py

@@ -19,7 +19,7 @@ import httpx
 import yaml
 from integration._support.client import Gateway, Scenario, eventually, gateway_from_environment, object_value
 from integration._support.database import read_rows
-from integration._support.process import OwnedProxy, owned_proxy_process
+from integration._support.process import OwnedProxy, OwnedState, owned_proxy_process
 from integration._support.responses_vendor import ResponsesVendor, same_response
 from integration._support.wire import Reply, Request, Wire, wire_server
 from prometheus_client.mmap_dict import MmapedDict
@@ -207,6 +207,7 @@ def series_cap_rig(
                 config=config,
                 remove_environment=() if shared_samples else ("PROMETHEUS_MULTIPROC_DIR",),
                 workers=workers,
+                state=OwnedState.SHARED_WITH_WORKER,
             )
         )
         scenario: Final = stack.enter_context(owned.gateway.scenario())

tests/integration/_support/safeguards_rig.py

@@ -30,7 +30,7 @@ from integration._support.client import (
     string_value,
 )
 from integration._support.database import read_rows
-from integration._support.process import OwnedProxy, owned_proxy_process
+from integration._support.process import OwnedProxy, OwnedState, owned_proxy_process
 from integration._support.wire import Reply, Request, Wire, wire_server
 from pydantic import JsonValue, TypeAdapter
 
@@ -313,6 +313,7 @@ def safeguards_rig(
             config=path,
             remove_environment=remove_environment,
             workers=workers,
+            state=OwnedState.SHARED_WITH_WORKER,
         ) as owned:
             yield Rig(wire, scripts, owned)
 

tests/integration/_support/workers.py

@@ -86,9 +86,9 @@ def worker_endpoints() -> Mapping[str, str]:
     )
 
 
-def _replica_urls(database: str) -> Mapping[str, str]:
+def replica_urls(database: str, inherited: Mapping[str, str] = os.environ) -> Mapping[str, str]:
-    writer: Final = urlsplit(os.environ["INTEGRATION_PROXY_DATABASE_URL"])
+    writer: Final = urlsplit(inherited["INTEGRATION_PROXY_DATABASE_URL"])
-    reader: Final = urlsplit(os.environ["INTEGRATION_PROXY_READ_REPLICA_URL"])
+    reader: Final = urlsplit(inherited["INTEGRATION_PROXY_READ_REPLICA_URL"])
     database_path: Final = urlsplit(database).path
     assert writer.username is not None and reader.username is not None
     with psycopg.connect(database, autocommit=True) as admin:
@@ -164,7 +164,7 @@ def worker_services(
                 "INTEGRATION_PROXY_READ_REPLICA_URL": "",
                 "REDIS_HOST": cache.host,
                 "REDIS_PORT": str(cache.port),
-                **(_replica_urls(database) if replica else {}),
+                **(replica_urls(database, inherited) if replica else {}),
             }
         )
 

tests/integration/authorization/test_access_group_budget.py

@@ -22,6 +22,8 @@ from integration._support.process import owned_proxy
 from integration._support.wire import Reply, Request, Wire, wire_server
 from pydantic import BaseModel, TypeAdapter
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 _INPUT_RATE: Final = 0.000001
 _OUTPUT_RATE: Final = 0.00001
 _MAX_BUDGET: Final = 0.0001

tests/integration/authorization/test_alias_chain_budget_waiver.py

@@ -62,7 +62,7 @@ from integration.authorization._hidden_alias_budget import (
 )
 from pydantic import JsonValue, TypeAdapter
 
-pytestmark: Final = pytest.mark.timeout(240)
+pytestmark: Final = [pytest.mark.timeout(240), pytest.mark.shares_worker_state]
 
 PAID_INPUT_COST_PER_TOKEN: Final = 0.001
 PAID_OUTPUT_COST_PER_TOKEN: Final = 0.002

tests/integration/authorization/test_claude_code_gateway_oauth.py

@@ -23,6 +23,8 @@ from integration._support.process import owned_proxy_process
 from integration._support.wire import Reply, Request, Wire, wire_server
 from pydantic import BaseModel, ConfigDict, JsonValue, TypeAdapter
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 _CLIENT_ID: Final = "scripted-cli-client"
 _CLIENT_SECRET: Final = "scripted-cli-secret"
 _AUTHORIZATION_CODE: Final = "scripted-auth-code"

tests/integration/authorization/test_cli_sso_login_ui_disabled.py

@@ -34,7 +34,7 @@ from tests.integration._support.process import OwnedProxy, group_members, owned_
 from tests.integration._support.provider import SharedProvider
 from tests.integration._support.wire import Reply, Request, Wire, wire_server
 
-pytestmark: Final = pytest.mark.timeout(300)
+pytestmark: Final = [pytest.mark.timeout(300), pytest.mark.shares_worker_state]
 
 CLI_SOURCE: Final = "litellm-cli"
 CLI_STATE_PREFIX: Final = "litellm-session-token"

tests/integration/authorization/test_end_user_block.py

@@ -12,6 +12,8 @@ from integration._support.database import read_rows
 from integration._support.process import owned_proxy
 from pydantic import BaseModel, JsonValue
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 
 class _BlockedCustomerInfo(BaseModel):
     user_id: str

tests/integration/authorization/test_hidden_alias_budget_bypass.py

@@ -62,7 +62,7 @@ from integration.authorization._hidden_alias_budget import (
 )
 from pydantic import JsonValue
 
-pytestmark: Final = pytest.mark.timeout(240)
+pytestmark: Final = [pytest.mark.timeout(240), pytest.mark.shares_worker_state]
 
 # $0 comes from litellm's own cost-map row (model_prices_and_context_window.json, added 2025-10-14 in 900a57f5b1f):
 # Lemonade is a local inference server, so there is no vendor price for the row to drift from. A row that gains a

tests/integration/authorization/test_hidden_alias_budget_bypass_chaos.py

@@ -32,7 +32,7 @@ from integration.authorization._hidden_alias_budget import (
     upstream_requests,
 )
 
-pytestmark: Final = pytest.mark.timeout(240)
+pytestmark: Final = [pytest.mark.timeout(240), pytest.mark.shares_worker_state]
 
 _STARTED_WORKER: Final = re.compile(r"Started server process \[(\d+)\]")
 _WAVE: Final = 8

tests/integration/authorization/test_jwt_auto_register_map_existing_key.py

@@ -24,7 +24,7 @@ from tests.integration._support.wire import Reply, Request, wire_server
 KEY_ID: Final = "integration-jwt-map-existing-key"
 MAPPING_INSERT: Final = b'INSERT INTO "public"."LiteLLM_JWTKeyMapping"'
 
-pytestmark = pytest.mark.timeout(240)
+pytestmark: Final = [pytest.mark.timeout(240), pytest.mark.shares_worker_state]
 
 
 def _hash(key: str) -> str:

tests/integration/authorization/test_jwt_default_team_provisioning.py

@@ -14,6 +14,8 @@ from tests.integration._support.database import read_rows
 from tests.integration._support.process import owned_proxy
 from tests.integration._support.wire import Reply, Request, wire_server
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 KEY_ID: Final = "integration-jwt-signing-key"
 TEAM_BUDGET: Final = 25.0
 

tests/integration/authorization/test_jwt_key_mapping_lifecycle.py

@@ -36,7 +36,7 @@ _KEY_ID: Final = "integration-signing-key"
 _UPSTREAM_KEY: Final = "synthetic-openai-key"
 _CHANNEL: Final = "litellm_proxy.auth_cache_invalidation"
 _CHAT_BODY: Final = TypeAdapter(dict[str, JsonValue])
-pytestmark: Final = pytest.mark.timeout(180)
+pytestmark: Final = [pytest.mark.timeout(180), pytest.mark.shares_worker_state]
 
 
 class _Usage(BaseModel):

tests/integration/authorization/test_jwt_mapped_key_email_backfill.py

@@ -14,6 +14,8 @@ from integration._support.process import owned_proxy
 from integration._support.wire import Reply, Request, wire_server
 from jwt.algorithms import RSAAlgorithm
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 AUDIENCE: Final = "litellm-integration"
 KEY_ID: Final = "integration-signing-key"
 CLIENT_CLAIM: Final = "client_id"

tests/integration/authorization/test_key_guardrail_opt_out.py

@@ -3,6 +3,7 @@ from pathlib import Path
 from typing import Final
 
 import httpx
+import pytest
 import yaml
 
 from integration._support.client import Gateway, Scenario, object_value, string_value
@@ -16,6 +17,8 @@ from integration.authorization._guardrail_opt_out import (
     stored_metadata,
 )
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 _KEY_ROUTES: Final = ["/key/generate", "/key/update", "/key/regenerate", "/v1/chat/completions"]
 
 

tests/integration/authorization/test_key_guardrail_opt_out_chaos.py

@@ -9,6 +9,7 @@ from queue import SimpleQueue
 from typing import Final
 
 import httpx
+import pytest
 
 from integration._support.client import Gateway, eventually, string_value
 from integration._support.database import read_rows
@@ -24,6 +25,8 @@ from integration.authorization._guardrail_opt_out import (
     upstream_observations,
 )
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 
 class _GuardrailSink:
     """Test-owned guardrail endpoint that can be stopped and restarted on the same port."""

tests/integration/authorization/test_key_guardrail_opt_out_runtime.py

@@ -5,6 +5,7 @@ from pathlib import Path
 from typing import Final
 
 import httpx
+import pytest
 from anthropic import Anthropic
 from openai import AsyncOpenAI, OpenAI
 
@@ -20,6 +21,8 @@ from integration.authorization._guardrail_opt_out import (
     upstream_hits,
 )
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 
 def _wire_hits(wire: Wire, marker: str) -> int:
     return sum(1 for request in wire.drain() if marker.encode() in request.body)

tests/integration/authorization/test_rag_query_vector_store_allowlist.py

@@ -25,6 +25,8 @@ from openai.types.chat import ChatCompletion
 from pydantic import JsonValue, TypeAdapter
 from redis import Redis
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 CONFIG_STORE_ID: Final = "vs_integration_config_store"
 PROXY_CONFIG: Final = Path(__file__).resolve().parents[1] / "proxy_config.yaml"
 REMOVE_OPENAI_API_BASE: Final = ("OPENAI_API_BASE",)

tests/integration/authorization/test_rag_team_provider_credentials.py

@@ -89,7 +89,7 @@ AWS_AMBIENT_CREDENTIALS: Final = (
     "AWS_WEB_IDENTITY_TOKEN_FILE",
 )
 
-pytestmark: Final = pytest.mark.timeout(OWNED_PROXY_CELL_SECONDS)
+pytestmark: Final = [pytest.mark.timeout(OWNED_PROXY_CELL_SECONDS), pytest.mark.shares_worker_state]
 
 Signature = tuple[str, str, str, str | None]
 

tests/integration/authorization/test_responses_compact_id_ownership.py

@@ -50,6 +50,8 @@ from integration.authorization._responses_compact import (
 )
 from pydantic import JsonValue
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 _CLIENTS: Final = ("sync", "async", "httpx")
 
 

tests/integration/authorization/test_responses_compact_id_ownership_chaos.py

@@ -40,6 +40,8 @@ from integration.authorization._responses_compact import (
     vendor_id,
 )
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 _Kind = Literal["owner", "other_user", "other_team", "unmanaged"]
 _KINDS: Final[tuple[_Kind, ...]] = ("owner", "other_user", "other_team", "unmanaged")
 _REFUSING: Final[tuple[_Kind, ...]] = ("other_user", "other_team", "unmanaged")

tests/integration/authorization/test_responses_id_ownership.py

@@ -19,6 +19,8 @@ from integration._support.worker_readiness import wait_for_worker_readiness
 from openai.types.responses import ResponseInputParam
 from pydantic import BaseModel, ConfigDict, JsonValue, TypeAdapter
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 _MODEL: Final = "gpt-5"
 _API_KEY: Final = "synthetic-responses-key"
 _JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue])

tests/integration/authorization/test_sso_blank_identity.py

@@ -30,7 +30,7 @@ from tests.integration._support.database_relay import dropped_connection_relay
 from tests.integration._support.process import OwnedProxy, owned_proxy_process
 from tests.integration._support.wire import Reply, Request, Wire, wire_server
 
-pytestmark: Final = pytest.mark.timeout(300)
+pytestmark: Final = [pytest.mark.timeout(300), pytest.mark.shares_worker_state]
 
 CLIENT_ID: Final = "integration-entra-client"
 CLIENT_SECRET: Final = "integration-entra-secret"

tests/integration/authorization/test_tool_search_replay_guardrails.py

@@ -18,7 +18,7 @@ from pydantic import JsonValue
 
 from tests.integration.caching.response_cache_case import await_entry, redis_store
 
-pytestmark: Final = pytest.mark.timeout(2 * graceful_stop_seconds() + 120)
+pytestmark: Final = [pytest.mark.timeout(2 * graceful_stop_seconds() + 120), pytest.mark.shares_worker_state]
 
 _POLICY: Final = "integration-tool-policy"
 _BLOCK: Final = "tool-permission-block"

tests/integration/caching/test_cache_max_messages.py

@@ -19,6 +19,8 @@ from pydantic import JsonValue, TypeAdapter
 from redis import Redis
 from redis.exceptions import ResponseError
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 _Turns = Callable[[str], tuple[list[JsonValue], list[JsonValue]]]
 _INDEXES: Final = TypeAdapter(tuple[bytes, ...])
 _COUNT: Final = TypeAdapter(tuple[int])

tests/integration/caching/test_response_cache_chaos.py

@@ -61,6 +61,8 @@ from tests.integration.caching.response_cache_case import (
     slowed,
 )
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 RecordProperty = Callable[[str, object], None]
 
 WORKERS: Final = 2

tests/integration/compatibility/_status_code_audit.py

@@ -248,7 +248,7 @@ def owned_gateway(
         gateway_from_environment() as gateway,
         owned_proxy_process(gateway, directory, environment or {}, config=path, workers=workers) as owned,
     ):
-        yield Gateway(owned.gateway.client, owned.gateway.key, gateway.upstream_url)
+        yield owned.gateway
 
 
 def stream_finished(endpoint: str, lines: Sequence[str]) -> bool:

tests/integration/compatibility/test_codex_model_catalog.py

@@ -32,6 +32,8 @@ from tests.integration._support.database import read_rows
 from tests.integration._support.process import owned_proxy
 from tests.integration._support.wire import Reply, Request, Wire, wire_server
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 T = TypeVar("T")
 
 CLIENT_VERSION: Final = "0.159.3"

tests/integration/compatibility/test_missing_body_param_status.py

@@ -43,6 +43,8 @@ from tests.integration.cost_calculation.cost_tracking_case import (
     SseResponse,
 )
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 _Route = tuple[str, str, tuple[str, ...], dict[str, JsonValue], str]
 _EVAL_DATA_SOURCE_CONFIG: Final[dict[str, JsonValue]] = {
     "type": "custom",

tests/integration/compatibility/test_openai_sdk_client_paths_chaos.py

@@ -26,6 +26,8 @@ from pydantic import JsonValue
 
 from tests.integration.cost_calculation.cost_tracking_case import JsonResponse, RoutedResponse, SseResponse
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 _Kind: TypeAlias = Literal["chat", "chat_stream", "messages", "responses", "completions", "moderations"]
 
 _ALL_KINDS: Final[tuple[_Kind, ...]] = ("chat", "chat_stream", "messages", "responses", "completions", "moderations")

tests/integration/compatibility/test_persisted_toolsets.py

@@ -12,6 +12,8 @@ from integration._support.database import read_rows
 from integration._support.mcp import call_tool, mcp_peer, register_mcp, tool_names
 from integration._support.process import owned_proxy
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 
 @pytest.mark.covers("other.compatibility.mcp.persisted_tool_names_survive_candidate_startup")
 def test_existing_toolset_format_loads_before_start_and_keeps_sibling_denied(gateway: Gateway, tmp_path: Path) -> None:

tests/integration/compatibility/test_status_code_followups.py

@@ -207,7 +207,7 @@ def router_default_proxy(tmp_path_factory: pytest.TempPathFactory) -> Iterator[G
             },
         )
         with owned_proxy_process(gateway, directory, {}, config=config) as owned:
-            yield Gateway(owned.gateway.client, owned.gateway.key, gateway.upstream_url)
+            yield owned.gateway
 
 
 def test_ocr_multipart_without_file_returns_400(gateway: Gateway) -> None:

tests/integration/compatibility/test_status_code_followups_chaos.py

@@ -41,7 +41,7 @@ from pydantic import JsonValue
 
 from tests.integration.cost_calculation.cost_tracking_case import StoredResponse
 
-pytestmark: Final = pytest.mark.timeout(480)
+pytestmark: Final = [pytest.mark.timeout(480), pytest.mark.shares_worker_state]
 
 _STARTED_WORKER: Final = re.compile(r"Started server process \[(\d+)\]")
 _MARKER_HEADER: Final = "x-litellm-spend-logs-metadata"

tests/integration/configuration/test_codex_catalog_boot.py

@@ -16,6 +16,8 @@ from pydantic import JsonValue
 from tests.integration._support.client import JSON_OBJECT, Gateway, Scenario, eventually, object_value, string_value
 from tests.integration._support.process import owned_proxy, owned_proxy_process
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 CLIENT_VERSION: Final = "0.159.3"
 FRESH_CONNECTION: Final = {"Connection": "close"}
 CONSISTENT_READS: Final = 8

tests/integration/configuration/test_effective_settings.py

@@ -23,6 +23,8 @@ from tests.integration._support.database import read_rows, scratch_database
 from tests.integration._support.process import owned_proxy
 from tests.integration._support.wire import Reply, Request, Wire, wire_server
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 _JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue])
 
 

tests/integration/conftest.py

@@ -5,7 +5,7 @@ import json
 import os
 import re
 from collections.abc import AsyncIterator, Generator, Iterator, Mapping, Sequence
-from contextlib import ExitStack
+from contextlib import ExitStack, contextmanager
 from copy import copy
 from functools import partial
 from importlib.metadata import version
@@ -23,6 +23,7 @@ from tests.integration._support.client import Gateway, Scenario, eventually, gat
 from tests.integration._support.generation import LIFECYCLE_SETTINGS
 from tests.integration._support.manifest import OWNED_DIRECTORIES
 from tests.integration._support.ordering import collected_case, fixture_groups, scheduling_groups
+from tests.integration._support.process import OWNED_STATE_VARIABLE, OwnedState
 from tests.integration._support.provider import SharedProvider, shared_provider
 from tests.integration._support.routing import RoutingPlugin
 from tests.integration._support.sdk_models import build_sdk_models
@@ -69,6 +70,7 @@ def pytest_addoption(parser: pytest.Parser) -> None:
 def pytest_configure(config: pytest.Config) -> None:
     config.addinivalue_line("markers", "integration: owned real-service integration contracts")
     config.addinivalue_line("markers", "covers(*ids): legacy contract IDs kept for existing tests, not enforced")
+    config.addinivalue_line("markers", "shares_worker_state: owned proxies here share the worker DB and Redis")
     config.stash[REPORTS] = []
     config.stash[SHARD_FILES] = _shard_files(SHARD_PATH.validate_python(config.getoption("integration_shard_files")))
     config.pluginmanager.register(IntegrationReportPlugin(config))
@@ -295,6 +297,32 @@ def pytest_sessionfinish(session: pytest.Session, exitstatus: int) -> None:
         session.exitstatus = pytest.ExitCode.OK
 
 
+@contextmanager
+def _owned_state(item: pytest.Item) -> Iterator[None]:
+    with pytest.MonkeyPatch.context() as environment:
+        if item.get_closest_marker("shares_worker_state") is not None:
+            environment.setenv(OWNED_STATE_VARIABLE, OwnedState.SHARED_WITH_WORKER.value)
+        yield
+
+
+@pytest.hookimpl(wrapper=True)
+def pytest_runtest_setup(item: pytest.Item) -> Generator[None, None, None]:
+    with _owned_state(item):
+        return (yield)
+
+
+@pytest.hookimpl(wrapper=True)
+def pytest_runtest_call(item: pytest.Item) -> Generator[None, None, None]:
+    with _owned_state(item):
+        return (yield)
+
+
+@pytest.hookimpl(wrapper=True)
+def pytest_runtest_teardown(item: pytest.Item, nextitem: pytest.Item | None) -> Generator[None, None, None]:
+    with _owned_state(item):
+        return (yield)
+
+
 @pytest.fixture
 def gateway() -> Iterator[Gateway]:
     with gateway_from_environment() as value:

tests/integration/cost_calculation/test_base_model_cross_provider_spend.py

@@ -16,7 +16,7 @@ from integration._support.process import OwnedProxy, owned_proxy_process
 from integration._support.wire import Reply, Request, Wire, wire_server
 from pydantic import JsonValue
 
-pytestmark: Final = pytest.mark.timeout(240)
+pytestmark: Final = [pytest.mark.timeout(240), pytest.mark.shares_worker_state]
 
 _MODEL: Final = "cc-subscription-sol"
 _INPUT_TOKENS: Final = 1000

tests/integration/cost_calculation/test_bridged_previous_response_id_spend.py

@@ -17,7 +17,7 @@ from integration._support.workers import cost_map_environment
 from integration.cost_calculation.cost_tracking_case import COST_MAP
 from pydantic import JsonValue, TypeAdapter
 
-pytestmark: Final = pytest.mark.timeout(240)
+pytestmark: Final = [pytest.mark.timeout(240), pytest.mark.shares_worker_state]
 
 _JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue])
 _MESSAGES: Final = TypeAdapter(list[dict[str, JsonValue]])

tests/integration/cost_calculation/test_responses_websocket_guardrail_spend.py

@@ -21,7 +21,7 @@ from integration._support.responses_ws_peer import PROVIDER_MODEL, USAGE, create
 from integration._support.tls import write_self_signed_cert
 from pydantic import JsonValue
 
-pytestmark: Final = pytest.mark.timeout(2 * graceful_stop_seconds() + 120)
+pytestmark: Final = [pytest.mark.timeout(2 * graceful_stop_seconds() + 120), pytest.mark.shares_worker_state]
 
 _INPUT_COST_PER_TOKEN: Final = 0.002
 _OUTPUT_COST_PER_TOKEN: Final = 0.004

tests/integration/database/test_reader_writer_regeneration.py

@@ -17,6 +17,8 @@ from integration._support.wire import Reply, Request, wire_server
 from psycopg import sql
 from pydantic import BaseModel, JsonValue, TypeAdapter
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 
 @pytest.mark.covers("other.database.regeneration.writer_updates_dependent_grants")
 def test_key_regeneration_uses_writer_with_a_real_readonly_reader(gateway: Gateway, tmp_path: Path) -> None:

tests/integration/management/test_batch_file_usage_caps.py

@@ -59,7 +59,7 @@ from integration.management._batch_file_caps import (
 from pydantic import JsonValue
 from redis import Redis
 
-pytestmark = pytest.mark.timeout(600)
+pytestmark: Final = [pytest.mark.timeout(600), pytest.mark.shares_worker_state]
 
 YAML_RECORDS: Final = 3
 YAML_UPLOADS: Final = 4

tests/integration/management/test_batch_file_usage_caps_chaos.py

@@ -46,7 +46,7 @@ from integration.management._batch_file_caps import (
 )
 from pydantic import JsonValue
 
-pytestmark = pytest.mark.timeout(600)
+pytestmark: Final = [pytest.mark.timeout(600), pytest.mark.shares_worker_state]
 
 WORKERS: Final = 2
 UPLOAD_CAP: Final = 5

tests/integration/management/test_budget_reset_windows.py

@@ -49,7 +49,7 @@ from integration.cost_calculation.cost_tracking_case import JsonResponse, SseRes
 from pydantic import JsonValue, TypeAdapter
 from redis import Redis
 
-pytestmark: Final = pytest.mark.timeout(240)
+pytestmark: Final = [pytest.mark.timeout(240), pytest.mark.shares_worker_state]
 
 DAY: Final = 86400
 HOUR: Final = 3600

tests/integration/management/test_budget_updates.py

@@ -9,6 +9,8 @@ from tests.integration._support.client import Gateway, string_value
 from tests.integration._support.database import read_rows
 from tests.integration._support.process import owned_proxy
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 
 def _persisted_reset_at(budget_id: str) -> datetime:
     rows: Final = read_rows(

tests/integration/management/test_claude_code_plugin_installation_preference.py

@@ -14,6 +14,8 @@ from tests.integration._support.client import JSON_OBJECT, Gateway, Scenario, ev
 from tests.integration._support.database import read_rows, write_rows
 from tests.integration._support.process import OwnedProxy, owned_proxy_process
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 PREFERENCES: Final = ("available", "auto_install", "required")
 INVALID_PREFERENCES: Final[tuple[JsonValue, ...]] = (
     "",

tests/integration/management/test_credential_federation_values.py

@@ -48,6 +48,8 @@ from tests.integration._support.database import read_rows
 from tests.integration._support.process import OwnedProxy, graceful_stop_seconds, owned_proxy_process
 from tests.integration._support.wire import Reply, Request, Wire, wire_server
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 T = TypeVar("T")
 
 

tests/integration/management/test_fallback_management_public_names.py

@@ -29,7 +29,7 @@ from tests.integration._support.process import (
 )
 from tests.integration._support.redis_process import owned_redis
 
-pytestmark: Final = pytest.mark.timeout(300)
+pytestmark: Final = [pytest.mark.timeout(300), pytest.mark.shares_worker_state]
 
 PROVIDER_KEY: Final = "integration-provider-key"
 EVICTION_WARNING: Final = "config cache eviction of router_settings failed"

tests/integration/management/test_project_lifecycle.py

@@ -31,6 +31,8 @@ from pydantic import JsonValue
 from litellm.models.user import LiteLLM_UserTable
 from litellm.proxy.auth.auth_checks import ExperimentalUIJWTToken
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 _OWNED_PROXY_SALT_KEY: Final = "sk-integration-salt"
 _OWNERSHIP_POOL_LIMIT: Final = 4
 _CONNECTION_BUDGET: Final = re.compile(

tests/integration/management/test_ptu_share_guards.py

@@ -17,7 +17,7 @@ from integration._support.process import OwnedProxy, graceful_stop_seconds, owne
 from integration._support.wire import Reply, Request, wire_server
 from pydantic import JsonValue, TypeAdapter
 
-pytestmark: Final = pytest.mark.timeout(4 * graceful_stop_seconds() + 240)
+pytestmark: Final = [pytest.mark.timeout(4 * graceful_stop_seconds() + 240), pytest.mark.shares_worker_state]
 
 _RELOAD_SECONDS: Final = 5
 _PROPAGATION_SECONDS: Final = 6 * _RELOAD_SECONDS

tests/integration/management/test_scim_admin_group.py

@@ -13,6 +13,8 @@ from integration._support.database import read_rows
 from integration._support.process import owned_proxy_process
 from pydantic import BaseModel, ConfigDict, JsonValue, TypeAdapter
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 SCIM_HEADERS: Final = {"Content-Type": "application/scim+json"}
 SCIM_CORE_USER_SCHEMA: Final = "urn:ietf:params:scim:schemas:core:2.0:User"
 SCIM_CORE_GROUP_SCHEMA: Final = "urn:ietf:params:scim:schemas:core:2.0:Group"

tests/integration/management/test_scim_group_bulk_roster.py

@@ -36,6 +36,8 @@ from integration._support.process import OwnedProxy, graceful_stop_seconds, owne
 from psycopg import sql
 from pydantic import JsonValue
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 GROUP_SCHEMA: Final = "urn:ietf:params:scim:schemas:core:2.0:Group"
 PATCH_OP_SCHEMA: Final = "urn:ietf:params:scim:api:messages:2.0:PatchOp"
 SMALL_PUSH: Final = 5

tests/integration/management/test_scim_group_pathless_patch.py

@@ -15,6 +15,8 @@ from integration._support.process import owned_proxy_process
 from integration.authorization._guardrail_opt_out import upstream_hits
 from pydantic import JsonValue
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 PATCH_OP_SCHEMA: Final = "urn:ietf:params:scim:api:messages:2.0:PatchOp"
 GROUP_SCHEMA: Final = "urn:ietf:params:scim:schemas:core:2.0:Group"
 BURST_TEAMS: Final = 10

tests/integration/management/test_search_tool_crud.py

@@ -5,12 +5,15 @@ from types import MappingProxyType
 from typing import Final
 
 import yaml
+import pytest
 from integration._support.client import Gateway, string_value
 from integration._support.database import read_rows
 from integration._support.process import owned_proxy
 from integration._support.wire import Reply, Request, wire_server
 from pydantic import JsonValue
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 _PROXY_CONFIG: Final = Path(__file__).resolve().parents[1] / "proxy_config.yaml"
 _ORIGINAL_KEY: Final = "exa-original-provider-key"
 _ROTATED_KEY: Final = "exa-rotated-provider-key"

tests/integration/management/test_sso_sign_in.py

@@ -18,7 +18,7 @@ from tests.integration._support.database import read_rows
 from tests.integration._support.process import owned_proxy
 from tests.integration._support.wire import Reply, Request, wire_server
 
-pytestmark: Final = pytest.mark.timeout(300)
+pytestmark: Final = [pytest.mark.timeout(300), pytest.mark.shares_worker_state]
 
 CLIENT_ID: Final = "integration-sso-client"
 CLIENT_SECRET: Final = "integration-sso-secret"

tests/integration/management/test_team_budget_duration_defaults.py

@@ -10,6 +10,8 @@ from tests.integration._support.client import Gateway, string_value
 from tests.integration._support.database import read_rows
 from tests.integration._support.process import owned_proxy
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 
 def _budget_row(team_id: str) -> dict[str, JsonValue]:
     rows: Final = read_rows(

tests/integration/management/test_team_delete_prometheus.py

@@ -24,6 +24,8 @@ from tests.integration._support.client import Gateway, Scenario, eventually, obj
 from tests.integration._support.database import read_rows
 from tests.integration._support.process import graceful_stop_seconds, owned_proxy_process
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 METRIC: Final = "litellm_team_members_metric"
 METRICS_ROUTE: Final = "/metrics/"
 MEMBERS: Final = 3

tests/integration/management/test_tool_policy_user.py

@@ -22,6 +22,8 @@ from pydantic import BaseModel, JsonValue
 
 from litellm.repositories.chunked_in import IN_LIST_CHUNK_SIZE
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 AUDIENCE: Final = "litellm-integration"
 KEY_ID: Final = "integration-signing-key"
 CLIENT_CLAIM: Final = "client_id"

tests/integration/management/test_vector_store_file_list_managed_ids.py

@@ -26,6 +26,8 @@ from integration._support.ports import allocate_port
 from openai import APIStatusError, AsyncOpenAI, OpenAI
 from pydantic import JsonValue, TypeAdapter
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 MANAGED_PREFIX: Final = "litellm_proxy:"
 CARRIED_PROVIDER_FILE_ID: Final = re.compile(r"(?:^|;)llm_output_file_id,([^;]+)")
 UPLOAD_FILENAME: Final = re.compile(rb'filename="([^"]+)"')

tests/integration/mcp/test_mcp_accounting_guardrails.py

@@ -36,6 +36,8 @@ from integration._support.process import owned_proxy
 from integration._support.wire import Reply, Request, Wire, wire_server
 from pydantic import JsonValue
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 DEFAULT_COST: Final = 0.25
 ADD_COST: Final = 0.5
 FORBIDDEN: Final = "forbidden-integration-word"

tests/integration/mcp/test_mcp_agent_365_guardrail.py

@@ -27,6 +27,8 @@ from integration._support.mcp import (
 from integration._support.process import owned_proxy_process
 from integration._support.wire import Reply, Request, wire_server
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 TENANT: Final = "00000000-0000-4000-8000-0000000a3650"
 REJECTED: Final = "Agent 365 guardrail rejected the tool call"
 GUARDRAIL_ROWS: Final = (

tests/integration/mcp/test_mcp_akto_logging_only.py

@@ -20,6 +20,8 @@ from integration._support.mcp import EntryPoint, McpCaller, McpPeer, echo_tool,
 from integration._support.process import owned_proxy_process
 from integration._support.wire import Reply, Request, Wire, wire_server
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 AKTO_KEY: Final = "synthetic-akto-mcp-log-key"
 BLOCK_MARK: Final = "SYNTHETIC-AKTO-BLOCK"
 RESPONSE_CHECK: Final = {"akto_connector": "litellm", "response_guardrails": "true", "ingest_data": "true"}

tests/integration/mcp/test_mcp_llm_endpoints.py

@@ -41,7 +41,7 @@ from openai.types.responses.tool_param import Mcp
 from pydantic import JsonValue, TypeAdapter
 
 
-pytestmark: Final = pytest.mark.xdist_group("mcp_llm_private_gateway")
+pytestmark: Final = [pytest.mark.xdist_group("mcp_llm_private_gateway"), pytest.mark.shares_worker_state]
 
 
 @pytest.fixture(scope="session")

tests/integration/mcp/test_mcp_management.py

@@ -27,6 +27,8 @@ from pydantic import BaseModel, JsonValue
 from litellm.models.user import LiteLLM_UserTable
 from litellm.proxy.auth.auth_checks import ExperimentalUIJWTToken
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 ADD: Final = {"a": 4, "b": 5}
 
 

tests/integration/mcp/test_mcp_oauth_flows.py

@@ -43,6 +43,8 @@ from integration._support.process import owned_proxy
 from integration._support.wire import Reply, Request, Wire, wire_server
 from pydantic import TypeAdapter
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 ADD: Final = {"a": 2, "b": 3}
 CLIENT_REDIRECT: Final = "http://127.0.0.1:9/cb"
 ACCEPT: Final = {"Accept": "application/json, text/event-stream"}

tests/integration/mcp/test_mcp_protocol_errors.py

@@ -9,6 +9,8 @@ from integration._support.client import Gateway
 from integration._support.mcp import McpPeer, call_tool, register_mcp, tool_names
 from integration._support.wire import Reply, Request, wire_server
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 
 @pytest.mark.covers("other.mcp.errors.protocol_and_malformed_results_cannot_be_empty_success")
 def test_jsonrpc_error_and_malformed_tool_result_remain_errors(gateway: Gateway) -> None:

tests/integration/mcp/test_mcp_resilience.py

@@ -32,6 +32,8 @@ from integration._support.process import owned_proxy_process
 from integration._support.wire import Reply
 from pydantic import BaseModel, TypeAdapter
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 _ECHO: Final = "catalog-echo:"
 _PROBE: Final = "catalog-probe"
 _DESCRIPTION: Final = "Look up one record"

tests/integration/mcp/test_mcp_user_env_vars.py

@@ -14,6 +14,8 @@ from integration._support.mcp import McpPeer, call_tool, mcp_peer, register_mcp,
 from integration._support.process import owned_proxy_process
 from pydantic import JsonValue, TypeAdapter
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 TOKEN: Final = "USER_TOKEN"
 WORKSPACE: Final = "WORKSPACE"
 METHODS: Final = ("GET", "POST", "DELETE")

tests/integration/mcp/test_oauth_configuration.py

@@ -19,6 +19,8 @@ from integration._support.process import owned_proxy
 from integration._support.wire import Reply, Request, Wire, wire_server
 from pydantic import TypeAdapter
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 _Upstream = Callable[[Request], Reply]
 CLIENT_REDIRECT: Final = "http://127.0.0.1:9/cb"
 

tests/integration/messages_endpoint/chat_bridge/test_chat_bridge_multi_tool_call_blocks_wire.py

@@ -26,6 +26,8 @@ from openai.types.chat.chat_completion_chunk import ChoiceDeltaToolCall
 from openai.types.responses import FunctionToolParam, ResponseFunctionToolCall, ResponseOutputItemDoneEvent
 from pydantic import JsonValue, TypeAdapter
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 _BACKEND: Final = "mistral-multi-tool"
 _PROVIDER_MODEL: Final = f"mistral/{_BACKEND}"
 _PROVIDER_KEY: Final = "integration-provider-key"

tests/integration/messages_endpoint/providers/anthropic/test_anthropic_gateway_wire.py

@@ -14,6 +14,8 @@ from integration._support.process import owned_proxy_process
 from integration._support.wire import Reply, Request, wire_server
 from pydantic import JsonValue, TypeAdapter
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 _BACKEND: Final = "claude-sonnet-4-6"
 _API_KEY: Final = "synthetic-anthropic-key"
 _JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue])

tests/integration/messages_endpoint/providers/anthropic/test_anthropic_messages_pre_content_retry_chaos.py

@@ -45,7 +45,7 @@ _STARTED_WORKER: Final = re.compile(r"Started server process \[(\d+)\]")
 _OUTAGE_STATUSES: Final = (529, 429, 503)
 _ROUTING_ENCODED_ID: Final = re.compile(r"resp_([A-Za-z0-9+/]+=*)")
 
-pytestmark = pytest.mark.timeout(240)
+pytestmark: Final = [pytest.mark.timeout(240), pytest.mark.shares_worker_state]
 
 Endpoint = Literal["messages", "chat", "responses"]
 

tests/integration/messages_endpoint/providers/anthropic/test_anthropic_safeguards_beta_wire.py

@@ -20,6 +20,8 @@ from integration._support.vertex import service_account_json
 from integration._support.wire import Reply, Request, Wire, wire_server
 from pydantic import JsonValue, TypeAdapter
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 BETA: Final = "dangerous-tool-use-2026-09-03"
 SAFEGUARDS: Final[list[JsonValue]] = [
     {"type": "dangerous_tool_use", "classifier_context": {"v": 1, "permission_mode": "auto"}}

tests/integration/messages_endpoint/providers/anthropic/test_passthrough_migration_wire.py

@@ -6,11 +6,15 @@ from hashlib import sha256
 from pathlib import Path
 from typing import Final
 
+import pytest
+
 from integration._support.client import Gateway, JsonValue, eventually
 from integration._support.database import read_rows
 from integration._support.process import owned_proxy
 from integration._support.wire import Reply, Request, Wire, wire_server
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 _MODEL: Final = "claude-sonnet-4-5-20250929"
 _KEY: Final = "synthetic-anthropic-key"
 

tests/integration/messages_endpoint/providers/anthropic/test_pdf_document_count_tokens_chaos.py

@@ -25,6 +25,8 @@ from integration._support.wire import Reply, Request, wire_server
 from integration.providers._cache_control_marks_support import anthropic_peer, owned_config
 from pydantic import JsonValue, TypeAdapter
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 _STARTED_WORKER: Final = re.compile(r"Started server process \[(\d+)\]")
 _MODEL: Final = "burst-claude"
 _PROVIDER_KEY: Final = "synthetic-provider-key"

tests/integration/messages_endpoint/providers/anthropic/test_websearch_interception_wire.py

@@ -180,6 +180,8 @@ import httpx
 import pytest
 from integration._support.client import Gateway, eventually
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 _QUERY: Final = "integration capped search"
 _TEXT_BLOCK: Final = {"type": "text", "text": "searching once more"}
 _NOT_INTERCEPTED: Final = "native tool reached the provider"

tests/integration/messages_endpoint/test_anthropic_messages_caller_tools_wire.py

@@ -15,6 +15,8 @@ from integration._support.process import owned_proxy
 from integration._support.wire import Reply, Request, Wire, wire_server
 from pydantic import JsonValue, TypeAdapter
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 CLAUDE_MODEL: Final = "caller-tools-claude"
 OPENAI_MODEL: Final = "caller-tools-openai"
 GEMINI_MODEL: Final = "caller-tools-gemini"

tests/integration/observability/_openinference_support.py

@@ -5,7 +5,7 @@ import json
 import uuid
 from collections.abc import Callable, Iterator, Mapping, Sequence
 from contextlib import AbstractContextManager, ExitStack, contextmanager, nullcontext
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from itertools import chain
 from pathlib import Path
 from typing import Final
@@ -854,11 +854,7 @@ def _rig(
                 if fresh_client_connections
                 else owned.gateway.client
             )
-            proxy: Final = Gateway(
+            proxy: Final = replace(owned.gateway, client=proxy_client)
-                client=proxy_client,
-                key=owned.gateway.key,
-                upstream_url=owned.gateway.upstream_url,
-            )
             scenario: Final = resources.enter_context(proxy.scenario())
             model: Final = scenario.model(
                 model=model_name,

tests/integration/observability/azure_dispatch_support.py

@@ -10,7 +10,7 @@ from typing import Final
 
 import yaml
 from integration._support.client import Gateway, object_value
-from integration._support.process import OwnedProxy, owned_proxy_process
+from integration._support.process import OwnedProxy, OwnedState, owned_proxy_process
 from integration._support.redis_process import OwnedRedis, owned_redis
 from integration._support.wire import Reply, Request, Wire, wire_server
 from pydantic import JsonValue
@@ -528,6 +528,7 @@ def dispatch_proxy(
         },
         config=config,
         workers=workers,
+        state=OwnedState.SHARED_WITH_WORKER,
     ) as owned:
         yield owned
 

tests/integration/observability/test_akto_logging_only.py

@@ -26,6 +26,8 @@ from integration._support.database import read_rows
 from integration._support.process import owned_proxy_process
 from integration._support.wire import Reply, Request, Wire, wire_server
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 LOG_KEY: Final = "synthetic-akto-log-key"
 INPUT_KEY: Final = "synthetic-akto-input-key"
 OUTPUT_KEY: Final = "synthetic-akto-output-key"

tests/integration/observability/test_azure_content_safety_audit.py

@@ -15,6 +15,8 @@ from integration._support.process import OwnedProxy, owned_proxy_process
 from integration._support.wire import Reply, Request, Wire, wire_server
 from pydantic import JsonValue
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 _ATTACK_MARKER: Final = "synthetic-attack-marker"
 _MODERATION_MARKER: Final = "synthetic-moderation-marker"
 

tests/integration/observability/test_azure_content_safety_endpoints.py

@@ -13,6 +13,8 @@ from integration._support.process import owned_proxy
 from integration._support.wire import Reply, Request, Wire, wire_server
 from pydantic import JsonValue
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 _ATTACK_MARKER: Final = "synthetic-attack-marker"
 
 _AZURE_TARGET_PREFIX: Final = "/contentsafety/text:shieldPrompt?api-version="

tests/integration/observability/test_azure_storage_client_ttl.py

@@ -4,6 +4,8 @@ from collections.abc import Callable
 from pathlib import Path
 from typing import Final
 
+import pytest
+
 from _azure_storage_support import (
     SINK_HOSTS,
     RecordingDataLakeSink,
@@ -18,6 +20,8 @@ from integration._support.process import owned_proxy
 from integration._support.tls import server_context, write_self_signed_cert
 from integration._support.wire import Reply, Request, wire_server
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 WORKERS: Final = 2
 FLUSH_SECONDS: Final = "1"
 

tests/integration/observability/test_bedrock_converse_stream_post_call_guardrail.py

@@ -36,6 +36,8 @@ from integration._support.process import owned_proxy_process
 from integration._support.upstream import aws_event_stream_frame
 from integration._support.wire import Reply, Request, Wire, wire_server
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 MODEL_ID: Final = "anthropic.claude-sonnet-5-v1:0"
 ROUTER_MODEL: Final = "converse-relay"
 EVENT_STREAM: Final = "application/vnd.amazon.eventstream"

tests/integration/observability/test_bedrock_invoke_stream_post_call_guardrail.py

@@ -38,6 +38,8 @@ from integration._support.process import owned_proxy_process
 from integration._support.upstream import aws_event_stream_frame
 from integration._support.wire import Reply, Request, Wire, wire_server
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 CLAUDE_ID: Final = "anthropic.claude-sonnet-5-v1:0"
 LLAMA_ID: Final = "meta.llama4-maverick-17b-instruct-v1:0"
 PROFILE_ID: Final = "arn:aws:bedrock:us-east-1:000000000000:application-inference-profile/relayprofile"

tests/integration/observability/test_cache_hit_guardrail_metrics.py

@@ -18,6 +18,8 @@ from integration._support.process import OwnedProxy, owned_proxy_process
 from integration._support.wire import Reply, Request, Wire, wire_server
 from prometheus_client.parser import text_string_to_metric_families
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 GUARDRAIL_PATH: Final = "/beta/litellm_basic_guardrail_api"
 DEPLOYMENT_FAILURE: Final = "litellm_deployment_failure_responses_total"
 DEPLOYMENT_REQUESTS: Final = "litellm_deployment_total_requests_total"

tests/integration/observability/test_cache_hit_guardrail_metrics_chaos.py

@@ -7,6 +7,7 @@ from pathlib import Path
 from typing import Final
 
 import httpx
+import pytest
 from integration._support.client import Gateway, eventually, object_value, string_value
 from integration._support.process import owned_proxy_process
 from integration._support.redis_process import owned_redis
@@ -24,6 +25,8 @@ from test_cache_hit_guardrail_metrics import (
     _rig,
 )
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 BURST: Final = 10
 
 

tests/integration/observability/test_callback_delivery.py

@@ -14,6 +14,8 @@ from integration._support.database import read_rows
 from integration._support.process import graceful_stop_seconds, owned_proxy
 from integration._support.wire import Reply, Request, wire_server
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 
 @pytest.mark.covers(
     "other.observability.callbacks.credentials_stay_out_of_event_bodies",

tests/integration/observability/test_content_filter_logging_only.py

@@ -27,6 +27,8 @@ from integration._support.otlp_sink import AuditConfigWriter, SpanSinks, recorde
 from integration._support.process import owned_proxy_process
 from integration._support.wire import Reply, Request, Wire, wire_server
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 AWS_KEY: Final = "AKIA" + "SYNTHETIC0000000"
 GITHUB_TOKEN: Final = "ghp_" + "a1" * 18
 FORBIDDEN_WORD: Final = "synthetic-forbidden-word"

tests/integration/observability/test_custom_logger_redaction.py

@@ -28,7 +28,7 @@ INSTRUCTIONS: Final = "Reply in French. redaction-instructions-canary"
 BRIDGED: Final = "bridged-redaction-canary"
 USAGE_DEPLOYMENT: Final = f"redaction-usage-{uuid.uuid4().hex}"
 REDACT: Final = {"x-litellm-enable-message-redaction": "true"}
-pytestmark = pytest.mark.timeout(240)
+pytestmark: Final = [pytest.mark.timeout(240), pytest.mark.shares_worker_state]
 
 
 def _provider(request: Request) -> Reply:

tests/integration/observability/test_guardrail_effects.py

@@ -27,6 +27,8 @@ from openai import APIError, AsyncOpenAI, OpenAI
 from openai.types.chat import ChatCompletionChunk
 from pydantic import BaseModel, JsonValue
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 
 @pytest.mark.covers("other.observability.guardrails.rewrite_reaches_correct_anthropic_positions")
 def test_guardrail_rewrites_system_and_user_in_actual_anthropic_request(gateway: Gateway, tmp_path: Path) -> None:

tests/integration/observability/test_guardrail_stream_scope.py

@@ -25,6 +25,8 @@ from pydantic import TypeAdapter
 
 import litellm
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue])
 EndpointScope: TypeAlias = Literal["streaming", "non_streaming"]
 ModelKind: TypeAlias = Literal["router", "direct"]

tests/integration/observability/test_guardrail_stream_scope_chaos.py

@@ -25,6 +25,8 @@ from integration._support.wire import Reply, Request, Wire, wire_server
 from integration._support.ports import allocate_port
 from pydantic import TypeAdapter
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 ChaosEndpoint: TypeAlias = Literal["chat", "messages", "responses"]
 CHAOS_ENDPOINTS: Final[tuple[ChaosEndpoint, ...]] = ("chat", "messages", "responses")
 CHAOS_MODELS: Final = MappingProxyType(

tests/integration/observability/test_guardrail_stream_scope_matrix.py

@@ -38,6 +38,8 @@ from integration.cost_calculation.cost_tracking_case import RealtimeResponse
 from pydantic import TypeAdapter
 from websockets.asyncio.server import ServerConnection, serve
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 Endpoint: TypeAlias = Literal["chat", "messages", "responses"]
 Mode: TypeAlias = Literal["pre_call", "during_call", "post_call", "logging_only"]
 Scope: TypeAlias = Literal["streaming", "non_streaming"]

tests/integration/observability/test_langfuse_delivery.py

@@ -33,6 +33,8 @@ from opentelemetry.proto.common.v1.common_pb2 import KeyValue
 from opentelemetry.proto.trace.v1.trace_pb2 import Span
 from pydantic import BaseModel, JsonValue, TypeAdapter
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 PUBLIC_KEY: Final = "pk-lf-integration"
 SECRET_KEY: Final = "sk-lf-integration"
 PROJECTS_PATH: Final = "/api/public/projects"

tests/integration/observability/test_langtrace_delivery.py

@@ -7,7 +7,7 @@ import uuid
 from collections.abc import Callable, Iterator, Sequence
 from concurrent.futures import ThreadPoolExecutor
 from contextlib import contextmanager
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field, replace
 from itertools import repeat
 from pathlib import Path
 from queue import SimpleQueue
@@ -659,7 +659,7 @@ def test_langtrace_survives_a_killed_worker(gateway: Gateway, tmp_path: Path) ->
                 limits=httpx.Limits(max_keepalive_connections=0),
             ) as fresh_connections,
         ):
-            proxy: Final = Gateway(fresh_connections, owned.gateway.key, owned.gateway.upstream_url)
+            proxy: Final = replace(owned.gateway, client=fresh_connections)
             with proxy.scenario() as scenario:
                 rig: Final = _Rig(proxy, scenario.model(api_base=provider.url + "/v1"), provider, _Sink(sink, key))
                 _assert_kill_and_recovery(owned, rig)

tests/integration/observability/test_logging_only_scope_chaos.py

@@ -49,6 +49,8 @@ from integration._support.ports import allocate_port
 from openai import APIConnectionError as OpenAIAPIConnectionError
 from pydantic import JsonValue
 
+pytestmark: Final = pytest.mark.shares_worker_state
+
 _EDGE_DELAY_SECONDS: Final = 5