test: cover persisted updates and warmed authorization policies

.circleci/scripts/run_integration.sh

@@ -7,6 +7,8 @@ mkdir -p "$results"
 integration_identity="$(.venv/bin/python -c 'import uuid; print(uuid.uuid4().hex)')"
 upstream_pid=""
 proxy_pid=""
+peer_pid=""
+launched_pid=""
 guard_created=false
 guard_installed=false
 guard6_created=false
@@ -15,9 +17,9 @@ cleanup() {
   original_status=$?
   trap - EXIT INT TERM
   sudo .venv/bin/python .circleci/scripts/stop_integration_processes.py \
-    "$integration_identity" "$(id -u)" "$proxy_pid" "$upstream_pid" \
+    "$integration_identity" "$(id -u)" "$proxy_pid" "$peer_pid" "$upstream_pid" \
     > "$results/process-cleanup.txt" 2>&1 || original_status=1
-  for owned_pid in "$proxy_pid" "$upstream_pid"; do
+  for owned_pid in "$peer_pid" "$proxy_pid" "$upstream_pid"; do
     if [ -n "$owned_pid" ]; then
       kill -- "-$owned_pid" 2>/dev/null || true
       for _ in {1..50}; do
@@ -60,8 +62,10 @@ export LITELLM_MASTER_KEY=sk-integration-master LITELLM_SALT_KEY=sk-integration-
 export LITELLM_MODE=PRODUCTION LITELLM_LOCAL_MODEL_COST_MAP=True
 export STORE_MODEL_IN_DB=True AWS_EC2_METADATA_DISABLED=true DO_NOT_TRACK=1
 export INTEGRATION_PROXY_URL=http://127.0.0.1:4000
+export INTEGRATION_PEER_URL=""
 export INTEGRATION_UPSTREAM_URL=http://127.0.0.1:8190
 export INTEGRATION_MASTER_KEY="$LITELLM_MASTER_KEY"
+export INTEGRATION_SEED="$((16#$(git rev-parse --short=8 HEAD)))"
 
 uv run --no-sync prisma generate --schema litellm/proxy/schema.prisma > "$results/prisma-generate.log" 2>&1
 
@@ -93,40 +97,29 @@ awk '$3 == "REJECT" && $1 > 0 { rejected=1 } END { exit !rejected }' "$results/e
 setsid env -i PATH="$PATH" HOME="$HOME" PYTHONPATH="$PYTHONPATH" INTEGRATION_RUN_ID="$integration_identity" \
   .venv/bin/python -m integration._support.upstream > "$results/upstream.log" 2>&1 &
 upstream_pid=$!
-setsid env -i PATH="$PATH" HOME="$HOME" PYTHONPATH="$PYTHONPATH" INTEGRATION_RUN_ID="$integration_identity" \
-  DATABASE_URL="$DATABASE_URL" REDIS_HOST="$REDIS_HOST" REDIS_PORT="$REDIS_PORT" \
-  LITELLM_MASTER_KEY="$LITELLM_MASTER_KEY" LITELLM_SALT_KEY="$LITELLM_SALT_KEY" \
-  LITELLM_MODE=PRODUCTION LITELLM_LOCAL_MODEL_COST_MAP=True STORE_MODEL_IN_DB=True \
-  AWS_EC2_METADATA_DISABLED=true DO_NOT_TRACK=1 \
-  .venv/bin/litellm --config tests/integration/proxy_config.yaml --host 127.0.0.1 --port 4000 --num_workers 1 --telemetry False \
-  --use_prisma_db_push --enforce_prisma_migration_check \
-  > "$results/proxy.log" 2>&1 &
-proxy_pid=$!
-
-.venv/bin/python - <<'PY'
-import time
-import httpx
-
-deadline = time.monotonic() + 90
-with httpx.Client(trust_env=False, timeout=2) as client:
-    while True:
-        try:
-            provider = client.get("http://127.0.0.1:8190/health")
-            proxy = client.get("http://127.0.0.1:4000/health/readiness")
-            if provider.status_code == proxy.status_code == 200:
-                cache = client.get("http://127.0.0.1:4000/cache/ping", headers={"Authorization": "Bearer sk-integration-master"})
-                cache.raise_for_status()
-                assert cache.json()["status"] == "healthy", cache.text
-                assert cache.json()["cache_type"] == "redis", cache.text
-                assert cache.json()["ping_response"] is True, cache.text
-                assert cache.json()["set_cache_response"] == "success", cache.text
-                break
-        except httpx.TransportError:
-            pass
-        if time.monotonic() >= deadline:
-            raise SystemExit("Integration services did not become ready")
-        time.sleep(0.2)
-PY
+start_proxy() {
+  local port="$1"
+  local log_name="$2"
+  setsid env -i PATH="$PATH" HOME="$HOME" PYTHONPATH="$PYTHONPATH" INTEGRATION_RUN_ID="$integration_identity" \
+    DATABASE_URL="$DATABASE_URL" REDIS_HOST="$REDIS_HOST" REDIS_PORT="$REDIS_PORT" \
+    LITELLM_MASTER_KEY="$LITELLM_MASTER_KEY" LITELLM_SALT_KEY="$LITELLM_SALT_KEY" \
+    LITELLM_MODE=PRODUCTION LITELLM_LOCAL_MODEL_COST_MAP=True STORE_MODEL_IN_DB=True \
+    AWS_EC2_METADATA_DISABLED=true DO_NOT_TRACK=1 \
+    .venv/bin/python -m integration._support.proxy --config tests/integration/proxy_config.yaml \
+    --host 127.0.0.1 --port "$port" --num_workers 1 --telemetry False \
+    --use_prisma_db_push --enforce_prisma_migration_check \
+    > "$results/$log_name" 2>&1 &
+  launched_pid=$!
+}
+start_proxy 4000 proxy.log
+proxy_pid="$launched_pid"
+.venv/bin/python .circleci/scripts/wait_integration_services.py
+if [ "$suite" = management ]; then
+  export INTEGRATION_PEER_URL=http://127.0.0.1:4001
+  start_proxy 4001 peer.log
+  peer_pid="$launched_pid"
+  .venv/bin/python .circleci/scripts/wait_integration_services.py
+fi
 
 if [ "$suite" = providers ]; then
   INTEGRATION_RUN_ID="$integration_identity" .venv/bin/python -m pytest --noconftest -o addopts= \
@@ -141,7 +134,9 @@ fi
 timeout --signal=TERM --kill-after=20s 11m env -i PATH="$PATH" HOME="$HOME" PYTHONPATH="$PYTHONPATH" \
   INTEGRATION_RUN_ID="$integration_identity" \
   DATABASE_URL="$DATABASE_URL" REDIS_HOST="$REDIS_HOST" REDIS_PORT="$REDIS_PORT" \
-  INTEGRATION_PROXY_URL="$INTEGRATION_PROXY_URL" INTEGRATION_UPSTREAM_URL="$INTEGRATION_UPSTREAM_URL" \
+  INTEGRATION_PROXY_URL="$INTEGRATION_PROXY_URL" INTEGRATION_PEER_URL="$INTEGRATION_PEER_URL" \
+  INTEGRATION_UPSTREAM_URL="$INTEGRATION_UPSTREAM_URL" \
   INTEGRATION_MASTER_KEY="$INTEGRATION_MASTER_KEY" LITELLM_MODE=PRODUCTION \
+  INTEGRATION_SEED="$INTEGRATION_SEED" \
   LITELLM_LOCAL_MODEL_COST_MAP=True AWS_EC2_METADATA_DISABLED=true DO_NOT_TRACK=1 \
   .venv/bin/python tests/integration/run.py "$suite" --results "$results"

.circleci/scripts/wait_integration_services.py

@@ -0,0 +1,43 @@
+import os
+import time
+from typing import Final
+
+import httpx
+from redis import Redis
+
+
+def main() -> None:
+    primary: Final = os.environ["INTEGRATION_PROXY_URL"]
+    peer: Final = os.environ.get("INTEGRATION_PEER_URL")
+    proxies: Final = (primary, peer) if peer else (primary,)
+    deadline: Final = time.monotonic() + 90
+    headers: Final = {"Authorization": f"Bearer {os.environ['INTEGRATION_MASTER_KEY']}"}
+    with httpx.Client(trust_env=False, timeout=2) as client, Redis(
+        host=os.environ["REDIS_HOST"], port=int(os.environ["REDIS_PORT"]), socket_timeout=2
+    ) as cache:
+        while True:
+            try:
+                ready: Final = (
+                    client.get(f"{os.environ['INTEGRATION_UPSTREAM_URL']}/health").status_code == 200
+                    and all(client.get(f"{url}/health/readiness").status_code == 200 for url in proxies)
+                )
+                if ready:
+                    for url in proxies:
+                        response: Final = client.get(f"{url}/cache/ping", headers=headers)
+                        response.raise_for_status()
+                        result: Final = response.json()
+                        assert result["status"] == "healthy", result
+                        assert result["cache_type"] == "redis", result
+                        assert result["ping_response"] is True, result
+                        assert result["set_cache_response"] == "success", result
+                    if cache.pubsub_numsub("litellm_proxy.auth_cache_invalidation")[0][1] >= len(proxies):
+                        return
+            except httpx.TransportError:
+                pass
+            if time.monotonic() >= deadline:
+                raise SystemExit("Integration services or auth-cache subscribers did not become ready")
+            time.sleep(0.2)
+
+
+if __name__ == "__main__":
+    main()

tests/e2e/coverage_registry/mgmt.yaml

@@ -91,3 +91,15 @@
 - {id: mgmt.mcp_toolset.update.clear_persists, module: mgmt, tier: P0, surface: api, assertions: [clear_persists], source: "mcp_management_endpoints.py:3098", fail_before_fix: proven, rationale: "An explicit null clears the stored description; the update used to drop null and keep the old value"}
 - {id: mgmt.mcp_toolset.delete.persists, module: mgmt, tier: P0, surface: api, assertions: [persists], source: "mcp_management_endpoints.py:3149", rationale: "A deleted toolset is gone by id and from the list on every replica"}
 - {id: mgmt.key.update.preserves_independent_fields, module: mgmt, tier: P0, surface: api, assertions: [preserves_independent_fields], source: "proxy/management_endpoints/key_management_endpoints.py", rationale: "A partial key update preserves independent policy and metadata through serving"}
+- {"id": "mgmt.key.update.generated_sequences_preserve_state", "module": "mgmt", "tier": "P0", "surface": "api", "assertions": ["persists", "serves_request"], "source": "management_endpoints/key_management_endpoints.py", "rationale": "generated sequences preserve state"}
+- {"id": "mgmt.key.update.false_zero_and_empty_values_affect_serving", "module": "mgmt", "tier": "P0", "surface": "api", "assertions": ["persists", "serves_request"], "source": "management_endpoints/key_management_endpoints.py", "rationale": "false zero and empty values affect serving"}
+- {"id": "mgmt.key.update.project_clear_preserves_scope", "module": "mgmt", "tier": "P0", "surface": "api", "assertions": ["persists", "serves_request"], "source": "management_endpoints/key_management_endpoints.py", "rationale": "project clear preserves scope"}
+- {"id": "mgmt.key.update.invalid_batch_is_atomic", "module": "mgmt", "tier": "P0", "surface": "api", "assertions": ["persists", "serves_request"], "source": "management_endpoints/key_management_endpoints.py", "rationale": "invalid batch is atomic"}
+- {"id": "mgmt.key.update.two_workers_enforce_warmed_policy", "module": "mgmt", "tier": "P0", "surface": "api", "assertions": ["persists", "serves_request"], "source": "management_endpoints", "rationale": "two workers enforce warmed policy"}
+- {"id": "mgmt.user.scim.deactivation_includes_nullable_blocked_keys", "module": "mgmt", "tier": "P0", "surface": "api", "assertions": ["persists", "serves_request"], "source": "management_endpoints", "rationale": "deactivation includes nullable blocked keys"}
+- {"id": "mgmt.team.member_update.demoted_role_cannot_write", "module": "mgmt", "tier": "P0", "surface": "api", "assertions": ["persists", "serves_request"], "source": "management_endpoints", "rationale": "demoted role cannot write"}
+- {"id": "mgmt.model.block.changes_serving_and_preserves_control", "module": "mgmt", "tier": "P0", "surface": "api", "assertions": ["persists", "serves_request"], "source": "management_endpoints", "rationale": "changes serving and preserves control"}
+- {"id": "mgmt.router_settings.update.changes_observed_attempt_count", "module": "mgmt", "tier": "P0", "surface": "api", "assertions": ["persists", "serves_request"], "source": "management_endpoints", "rationale": "changes observed attempt count"}
+- {"id": "mgmt.credential.update.saved_value_reaches_wire", "module": "mgmt", "tier": "P0", "surface": "api", "assertions": ["persists", "serves_request"], "source": "management_endpoints", "rationale": "saved value reaches wire"}
+- {"id": "mgmt.key.update.denied_request_preserves_effective_state", "module": "mgmt", "tier": "P0", "surface": "api", "assertions": ["persists", "serves_request"], "source": "management_endpoints/key_management_endpoints.py", "rationale": "Denied multi-field writes leave effective key state unchanged"}
+- {"id": "mgmt.key.update.expiry_changes_reach_warmed_workers", "module": "mgmt", "tier": "P0", "surface": "api", "assertions": ["persists", "serves_request"], "source": "management_endpoints/key_management_endpoints.py", "rationale": "Expiry update and explicit clear affect both warmed workers before cache TTL"}

tests/integration/README.md

@@ -4,6 +4,10 @@ These tests exercise a running gateway, PostgreSQL and Redis with an owned local
 
 Use `tests/integration/run.py management`, `accounting` or `providers` to run a selected group. Set `INTEGRATION_PROXY_URL`, `INTEGRATION_UPSTREAM_URL`, `INTEGRATION_MASTER_KEY` and `DATABASE_URL` to an isolated test deployment. The runner selects the new domain directories explicitly; the legacy OCI and sandbox selections remain separate
 
+Management also requires `INTEGRATION_PEER_URL`, `REDIS_HOST` and `REDIS_PORT`. CircleCI starts two directly addressed proxy processes sharing only that job's stores. The test-only CLI wrapper supplies enterprise route entitlement, following the existing behavior suite's convention. It does not qualify license validation; run it with one worker and no reload
+
+The generated lifecycle models use 20 examples, eight steps, generation and shrinking, with isolated resources per example. HTTP operation caps include generation and shrinking and exempt cleanup. Local qualification defaults to seed 4106601; CircleCI derives its exploration seed from the checked-out revision. Use `--seed` to reproduce a run. Actual installed Hypothesis version, settings and seed are written beside the execution manifest
+
 Reuse the existing canned provider handlers through `_support/upstream.py`. It rejects internal request fields and exposes actual received requests for independent assertions. Register every created resource for cleanup immediately, keep expected values independent of production calculations, and assert readback plus the runtime effect of a change
 
 The CircleCI workflow starts its own database and Redis, restricts test-phase egress to its owned services and writes JUnit plus an executed-node manifest. Missing setup, skipped tests, failed cleanup or a selected test without a passed call fail qualification. Existing GitHub Actions jobs do not own these tests

tests/integration/_support/client.py

@@ -94,6 +94,42 @@ class Scenario:
         self.cleanups.callback(self.delete_key, token)
         return token
 
+    def team(self, **fields: JsonValue) -> str:
+        created: Final = self.gateway.post("/team/new", {"team_alias": f"integration-{uuid.uuid4().hex}", **fields})
+        identity: Final = string_value(created["team_id"])
+        self.cleanups.callback(self.delete_team, identity)
+        return identity
+
+    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,)) == []
+
+    def project(self, team_id: str, **fields: JsonValue) -> str:
+        created: Final = self.gateway.post(
+            "/project/new", {"team_id": team_id, "project_alias": f"integration-{uuid.uuid4().hex}", **fields}
+        )
+        identity: Final = string_value(created["project_id"])
+        self.cleanups.callback(self.delete_project, identity)
+        return identity
+
+    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,)) == []
+
+    def user(self, **fields: JsonValue) -> str:
+        created: Final = self.gateway.post(
+            "/user/new", {"user_id": f"integration-{uuid.uuid4().hex}", "auto_create_key": False, **fields}
+        )
+        identity: Final = string_value(created["user_id"])
+        self.cleanups.callback(self.delete_user, identity)
+        return identity
+
+    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,)) == []
+
     def delete_key(self, token: str) -> None:
         self.gateway.post("/key/delete", {"keys": [token]})
         response: Final = self.gateway.request("GET", "/key/info", params={"key": sha256(token.encode()).hexdigest()})

tests/integration/_support/generation.py

@@ -0,0 +1,52 @@
+from typing import Final
+from dataclasses import dataclass
+from collections.abc import Iterator, Sequence
+from contextlib import contextmanager
+
+import httpx
+from hypothesis import Phase, settings
+
+from integration._support.client import Gateway
+
+LIFECYCLE_SETTINGS: Final = settings(
+    max_examples=20,
+    stateful_step_count=8,
+    deadline=None,
+    database=None,
+    phases=(Phase.generate, Phase.shrink),
+    print_blob=True,
+)
+
+
+@dataclass(slots=True)
+class RequestBudget:
+    limit: int
+    requests: int = 0
+    cleaning: bool = False
+
+    def observe(self, _request: httpx.Request) -> None:
+        if self.cleaning:
+            return
+        self.requests += 1
+        assert self.requests <= self.limit, f"Generated HTTP operation budget exceeded: {self.limit}"
+
+    @contextmanager
+    def cleanup(self) -> Iterator[None]:
+        self.cleaning = True
+        try:
+            yield
+        finally:
+            self.cleaning = False
+
+
+@contextmanager
+def bounded_http_requests(gateways: Sequence[Gateway], limit: int) -> Iterator[RequestBudget]:
+    budget: Final = RequestBudget(limit)
+    for gateway in gateways:
+        gateway.client.event_hooks["request"].append(budget.observe)
+    try:
+        yield budget
+    finally:
+        for gateway in gateways:
+            gateway.client.event_hooks["request"].remove(budget.observe)
+        print(f"Generated HTTP operations: {budget.requests}/{budget.limit}; cleanup excluded")

tests/integration/_support/proxy.py

@@ -0,0 +1,16 @@
+"""Run the normal single-process CLI with the existing behavior-suite test entitlement."""
+
+from unittest.mock import patch
+
+from litellm import run_server
+
+
+def main() -> None:
+    with patch(  # test-quality-ok: route entitlement only; license validation is outside these HTTP/DB contracts
+        "litellm.proxy.auth.litellm_license.LicenseCheck.is_premium", return_value=True
+    ):
+        run_server()
+
+
+if __name__ == "__main__":
+    main()

tests/integration/_support/upstream.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import argparse
 from dataclasses import dataclass, field
+from collections import deque
 from queue import SimpleQueue
 from typing import Final
 
@@ -40,6 +41,7 @@ class Observation:
 @dataclass(frozen=True, slots=True)
 class Provider:
     observations: SimpleQueue[Observation] = field(default_factory=SimpleQueue)
+    scripts: dict[str, deque[int]] = field(default_factory=dict)
 
     async def chat(self, request: Request) -> Response:
         body: Final = JSON_OBJECT.validate_json(await request.body())
@@ -57,8 +59,34 @@ class Provider:
             for message in messages
         ):
             return JSONResponse({"error": {"message": "Invalid selected message contract"}}, status_code=400)
+        script: Final = self.scripts.get(str(body["model"]))
+        if script is not None:
+            if not script:
+                return JSONResponse({"error": {"message": "Script exhausted", "type": "api_error"}}, status_code=500)
+            status: Final = script.popleft()
+            if status != 200:
+                return JSONResponse(
+                    {"error": {"message": "Controlled provider failure", "type": "api_error", "code": str(status)}},
+                    status_code=status,
+                )
         return await chat_completions(request)
 
+    async def script(self, request: Request) -> Response:
+        name: Final = request.path_params["model"]
+        if request.method in {"DELETE", "GET"} and name not in self.scripts:
+            return JSONResponse({"error": "Script not found"}, status_code=404)
+        if request.method == "GET":
+            return JSONResponse({"remaining": list(self.scripts[name])})
+        if request.method == "DELETE":
+            remaining: Final = self.scripts.pop(name)
+            return JSONResponse({"remaining": list(remaining)})
+        body: Final = JSON_OBJECT.validate_json(await request.body())
+        statuses: Final = body.get("statuses")
+        if not isinstance(statuses, list) or not statuses or any(type(value) is not int for value in statuses):
+            return JSONResponse({"error": "A nonempty list of HTTP status codes is required"}, status_code=400)
+        self.scripts[name] = deque(int(str(value)) for value in statuses)
+        return JSONResponse({"configured": len(statuses)})
+
     async def observed(self, _request: Request) -> Response:
         values: Final = tuple(self.observations.get() for _ in range(self.observations.qsize()))
         return JSONResponse(
@@ -74,6 +102,7 @@ class Provider:
             routes=[
                 Route("/health", health),
                 Route("/__observations", self.observed),
+                Route("/__scripts/{model}", self.script, methods=["POST", "DELETE", "GET"]),
                 Route("/v1/chat/completions", self.chat, methods=["POST"]),
                 Route("/v1/completions", completions, methods=["POST"]),
                 Route("/v1/embeddings", embeddings, methods=["POST"]),

tests/integration/authorization/test_warmed_policy.py

@@ -0,0 +1,197 @@
+from contextlib import ExitStack
+from hashlib import sha256
+from typing import Final
+import os
+
+import psycopg
+import pytest
+from hypothesis import strategies as st
+from hypothesis.stateful import RuleBasedStateMachine, invariant, rule, run_state_machine_as_test
+
+from integration._support.client import Gateway, eventually, object_value
+from integration._support.database import read_rows
+from integration._support.generation import LIFECYCLE_SETTINGS, bounded_http_requests
+
+
+def assert_serving(gateway: Gateway, model: str, key: str, status: int, error_type: str = "auth_error") -> None:
+    response: Final = eventually(
+        lambda: gateway.request(
+            "POST", "/v1/chat/completions",
+            {"model": model, "messages": [{"role": "user", "content": "warmed policy control"}]}, key=key,
+        ),
+        lambda value: value.status_code == status,
+        seconds=3,
+    )
+    if status == 200:
+        assert response.json()["usage"]["total_tokens"] == 40
+        assert response.json()["choices"][0]["message"]["content"] == (
+            "Hello! This is a mock response from the fake OpenAI endpoint."
+        )
+    else:
+        assert response.json()["error"]["type"] == error_type
+
+
+@pytest.mark.covers("mgmt.key.update.two_workers_enforce_warmed_policy")
+def test_generated_policy_changes_reach_both_warmed_workers(gateway: Gateway, peer: Gateway) -> None:
+    class Policies(RuleBasedStateMachine):
+        def __init__(self) -> None:
+            super().__init__()
+            self.resources = ExitStack()
+            try:
+                scenario = self.resources.enter_context(gateway.scenario())
+                self.models = (scenario.model(), scenario.model())
+                self.allowed = 0
+                self.blocked = False
+                self.key = scenario.key(models=[self.models[0]], blocked=False)
+                self.control = scenario.key(models=list(self.models))
+                for worker in (gateway, peer):
+                    assert_serving(worker, self.models[0], self.key, 200)
+                    assert_serving(worker, self.models[1], self.control, 200)
+            except BaseException:
+                with budget.cleanup():
+                    self.resources.close()
+                raise
+
+        @rule(index=st.integers(min_value=0, max_value=1))
+        def model_grant(self, index: int) -> None:
+            gateway.post("/key/update", {"key": self.key, "models": [self.models[index]]})
+            self.allowed = index
+
+        @rule(blocked=st.booleans())
+        def block(self, blocked: bool) -> None:
+            gateway.post("/key/update", {"key": self.key, "blocked": blocked})
+            self.blocked = blocked
+
+        @invariant()
+        def both_workers_enforce_policy(self) -> None:
+            rows: Final = read_rows(
+                'SELECT models, blocked FROM "LiteLLM_VerificationToken" WHERE token = %s',
+                (sha256(self.key.encode()).hexdigest(),),
+            )
+            assert rows == [{"models": [self.models[self.allowed]], "blocked": self.blocked}]
+            for worker in (gateway, peer):
+                for index, model in enumerate(self.models):
+                    status: Final = 401 if self.blocked else 200 if index == self.allowed else 403
+                    kind: Final = "auth_error" if self.blocked else "key_model_access_denied"
+                    assert_serving(worker, model, self.key, status, kind)
+                assert_serving(worker, self.models[1], self.control, 200)
+
+        def teardown(self) -> None:
+            with budget.cleanup():
+                self.resources.close()
+
+    with bounded_http_requests((gateway, peer), limit=3000) as budget:
+        run_state_machine_as_test(Policies, settings=LIFECYCLE_SETTINGS)
+
+
+@pytest.mark.covers("mgmt.user.scim.deactivation_includes_nullable_blocked_keys")
+def test_scim_deactivation_blocks_null_and_false_keys_but_preserves_other_owners(gateway: Gateway) -> None:
+    with gateway.scenario() as scenario:
+        model: Final = scenario.model()
+        user: Final = scenario.user(user_role="internal_user")
+        other: Final = scenario.user(user_role="internal_user")
+        null_key: Final = scenario.key(user_id=user, models=[model])
+        false_key: Final = scenario.key(user_id=user, models=[model], blocked=False)
+        manual: Final = scenario.key(user_id=user, models=[model], blocked=True)
+        control: Final = scenario.key(user_id=other, models=[model])
+        team: Final = scenario.team(models=[model])
+        service: Final = gateway.post("/key/service-account/generate", {"team_id": team, "models": [model]})
+        service_key: Final = service["key"]
+        assert isinstance(service_key, str)
+        scenario.cleanups.callback(scenario.delete_key, service_key)
+        with psycopg.connect(os.environ["DATABASE_URL"]) as connection:
+            connection.execute(
+                'UPDATE "LiteLLM_VerificationToken" SET blocked = NULL WHERE token = %s',
+                (sha256(null_key.encode()).hexdigest(),),
+            )
+        assert read_rows(
+            'SELECT user_id, blocked FROM "LiteLLM_VerificationToken" WHERE token = %s',
+            (sha256(null_key.encode()).hexdigest(),),
+        ) == [{"user_id": user, "blocked": None}]
+        assert read_rows(
+            'SELECT user_id FROM "LiteLLM_VerificationToken" WHERE token = %s',
+            (sha256(service_key.encode()).hexdigest(),),
+        ) == [{"user_id": None}]
+        for token in (null_key, false_key, control, service_key):
+            assert_serving(gateway, model, token, 200)
+        for active in (False, True):
+            response: Final = gateway.request(
+                "PATCH", f"/scim/v2/Users/{user}",
+                {"schemas": ["urn:ietf:params:scim:api:messages:2.0:PatchOp"],
+                 "Operations": [{"op": "replace", "path": "active", "value": active}]},
+            )
+            assert response.status_code == 200, response.text
+            for token in (null_key, false_key):
+                rows: Final = read_rows(
+                    'SELECT blocked, metadata FROM "LiteLLM_VerificationToken" WHERE token = %s',
+                    (sha256(token.encode()).hexdigest(),),
+                )
+                assert rows[0]["blocked"] is not active
+                assert object_value(rows[0]["metadata"]).get("scim_blocked") is (None if active else True)
+                assert_serving(gateway, model, token, 200 if active else 401)
+            assert_serving(gateway, model, manual, 401)
+            for token in (control, service_key):
+                assert_serving(gateway, model, token, 200)
+
+
+@pytest.mark.covers("mgmt.team.member_update.demoted_role_cannot_write")
+def test_warmed_team_role_demotion_prevents_later_management_writes(gateway: Gateway) -> None:
+    with gateway.scenario() as scenario:
+        model: Final = scenario.model()
+        user: Final = scenario.user(user_role="internal_user")
+        team: Final = scenario.team(models=[model], members_with_roles=[{"user_id": user, "role": "admin"}])
+        control_team: Final = scenario.team(models=[model])
+        caller: Final = scenario.key(
+            user_id=user, team_id=team, models=[model], allowed_routes=["/team/update", "/v1/chat/completions"]
+        )
+        gateway.chat(model, key=caller)
+        changed: Final = gateway.request("POST", "/team/update", {"team_id": team, "team_alias": "permitted"}, key=caller)
+        assert changed.status_code == 200, changed.text
+        unrelated_before: Final = read_rows(
+            'SELECT team_alias FROM "LiteLLM_TeamTable" WHERE team_id = %s', (control_team,)
+        )
+        unrelated: Final = gateway.request(
+            "POST", "/team/update", {"team_id": control_team, "team_alias": "must-not-persist"}, key=caller
+        )
+        assert unrelated.status_code == 403, unrelated.text
+        assert read_rows(
+            'SELECT team_alias FROM "LiteLLM_TeamTable" WHERE team_id = %s', (control_team,)
+        ) == unrelated_before
+        gateway.post("/team/member_update", {"team_id": team, "user_id": user, "role": "user"})
+        for target in (team, control_team):
+            before: Final = read_rows('SELECT team_alias FROM "LiteLLM_TeamTable" WHERE team_id = %s', (target,))
+            denied: Final = gateway.request(
+                "POST", "/team/update", {"team_id": target, "team_alias": "must-not-persist"}, key=caller
+            )
+            assert denied.status_code == 403, denied.text
+            assert read_rows('SELECT team_alias FROM "LiteLLM_TeamTable" WHERE team_id = %s', (target,)) == before
+        roster: Final = read_rows('SELECT members_with_roles FROM "LiteLLM_TeamTable" WHERE team_id = %s', (team,))
+        members: Final = roster[0]["members_with_roles"]
+        assert isinstance(members, list)
+        assert next(object_value(member)["role"] for member in members if object_value(member)["user_id"] == user) == "user"
+        assert_serving(gateway, model, caller, 200)
+
+
+@pytest.mark.covers("mgmt.key.update.expiry_changes_reach_warmed_workers")
+def test_expiry_and_explicit_clear_reach_both_warmed_workers(gateway: Gateway, peer: Gateway) -> None:
+    with gateway.scenario() as scenario:
+        model: Final = scenario.model()
+        key: Final = scenario.key(models=[model], duration="1h")
+        control: Final = scenario.key(models=[model])
+        for worker in (gateway, peer):
+            assert_serving(worker, model, key, 200)
+        gateway.post("/key/update", {"key": key, "duration": "0s"})
+        assert read_rows(
+            "SELECT expires <= timezone('UTC', now()) AS expired FROM \"LiteLLM_VerificationToken\" WHERE token = %s",
+            (sha256(key.encode()).hexdigest(),),
+        ) == [{"expired": True}]
+        for worker in (gateway, peer):
+            assert_serving(worker, model, key, 401, "expired_key")
+            assert_serving(worker, model, control, 200)
+        gateway.post("/key/update", {"key": key, "duration": None})
+        assert read_rows(
+            'SELECT expires IS NULL AS cleared FROM "LiteLLM_VerificationToken" WHERE token = %s',
+            (sha256(key.encode()).hexdigest(),),
+        ) == [{"cleared": True}]
+        for worker in (gateway, peer):
+            assert_serving(worker, model, key, 200)

tests/integration/configuration/test_effective_settings.py

@@ -0,0 +1,123 @@
+import uuid
+from typing import Final
+
+import httpx
+import pytest
+
+from integration._support.client import Gateway, object_value, string_value
+from integration._support.database import read_rows
+
+
+def model_identity(gateway: Gateway, alias: str) -> str:
+    entries: Final = gateway.get("/model/info")["data"]
+    assert isinstance(entries, list)
+    entry: Final = next(object_value(value) for value in entries if object_value(value)["model_name"] == alias)
+    return string_value(object_value(entry["model_info"])["id"])
+
+
+@pytest.mark.covers("mgmt.model.block.changes_serving_and_preserves_control")
+def test_model_block_changes_actual_route_and_leaves_other_route_working(gateway: Gateway) -> None:
+    with gateway.scenario() as scenario:
+        model: Final = scenario.model()
+        other: Final = scenario.model()
+        identity: Final = model_identity(gateway, model)
+        gateway.chat(model)
+        gateway.chat(other)
+        gateway.post("/model/block", {"model_id": identity})
+        assert read_rows(
+            'SELECT blocked FROM "LiteLLM_ProxyModelTable" WHERE model_id = %s', (identity,)
+        ) == [{"blocked": True}]
+        response: Final = gateway.request(
+            "POST", "/v1/chat/completions",
+            {"model": model, "messages": [{"role": "user", "content": "blocked deployment"}]},
+        )
+        assert response.status_code == 403, response.text
+        assert response.json()["error"]["type"] == "permission_error"
+        assert response.json()["error"]["message"] == "litellm.PermissionDeniedError: Model is blocked"
+        assert object_value(gateway.chat(other)["usage"])["total_tokens"] == 40
+        gateway.post("/model/unblock", {"model_id": identity})
+        assert read_rows(
+            'SELECT blocked FROM "LiteLLM_ProxyModelTable" WHERE model_id = %s', (identity,)
+        ) == [{"blocked": False}]
+        assert object_value(gateway.chat(model)["usage"])["total_tokens"] == 40
+
+
+@pytest.mark.covers("mgmt.router_settings.update.changes_observed_attempt_count")
+def test_saved_retry_setting_controls_real_attempts_and_restores(gateway: Gateway) -> None:
+    with httpx.Client(base_url=gateway.upstream_url, timeout=5, trust_env=False) as upstream, gateway.scenario() as scenario:
+        original: Final = object_value(gateway.get("/router/settings")["current_values"])["num_retries"]
+        provider_model: Final = f"retry-{uuid.uuid4().hex}"
+        model: Final = scenario.model(model=f"openai/{provider_model}", input_cost_per_token=0, output_cost_per_token=0)
+
+        def remove_script() -> None:
+            response: Final = upstream.delete(f"/__scripts/{provider_model}")
+            assert response.status_code in (200, 404), response.text
+            assert upstream.get(f"/__scripts/{provider_model}").status_code == 404
+
+        scenario.cleanups.callback(remove_script)
+        try:
+            for generation, retries in enumerate((0, 1, original)):
+                gateway.post("/config/update", {"router_settings": {"num_retries": retries}})
+                assert object_value(gateway.get("/router/settings")["current_values"])["num_retries"] == retries
+                configured: Final = upstream.post(f"/__scripts/{provider_model}", json={"statuses": [500, 200]})
+                assert configured.status_code == 200, configured.text
+                upstream.get("/__observations").raise_for_status()
+                response: Final = gateway.request(
+                    "POST", "/v1/chat/completions",
+                    {"model": model, "messages": [{"role": "user", "content": f"{provider_model} attempt {generation}"}]},
+                )
+                observed: Final = upstream.get("/__observations")
+                observed.raise_for_status()
+                requests: Final = observed.json()["requests"]
+                assert len(requests) == (1 if retries == 0 else 2), (response.status_code, response.text, requests)
+                assert all(value["body"]["model"] == provider_model for value in requests)
+                assert response.status_code == (500 if retries == 0 else 200), response.text
+                if retries != 0:
+                    assert response.json()["usage"]["total_tokens"] == 40
+                remaining: Final = upstream.delete(f"/__scripts/{provider_model}")
+                assert remaining.status_code == 200, remaining.text
+                assert remaining.json()["remaining"] == ([200] if retries == 0 else [])
+        finally:
+            gateway.post("/config/update", {"router_settings": {"num_retries": original}})
+            assert object_value(gateway.get("/router/settings")["current_values"])["num_retries"] == original
+
+
+@pytest.mark.covers("mgmt.credential.update.saved_value_reaches_wire")
+def test_credential_value_update_and_model_reload_reach_provider(gateway: Gateway) -> None:
+    with gateway.scenario() as scenario, httpx.Client(base_url=gateway.upstream_url, timeout=5, trust_env=False) as upstream:
+        name: Final = f"credential-{uuid.uuid4().hex}"
+        gateway.post("/credentials", {
+            "credential_name": name, "credential_values": {"api_key": "synthetic-credential-first"}, "credential_info": {}
+        })
+
+        def remove_credential() -> None:
+            response: Final = gateway.request("DELETE", f"/credentials/{name}")
+            assert response.status_code == 200, response.text
+            assert read_rows(
+                'SELECT credential_name FROM "LiteLLM_CredentialsTable" WHERE credential_name = %s', (name,)
+            ) == []
+
+        scenario.cleanups.callback(remove_credential)
+        model: Final = scenario.model(api_key=None, litellm_credential_name=name)
+        identity: Final = model_identity(gateway, model)
+        for value in ("synthetic-credential-first", "synthetic-credential-second"):
+            patched: Final = gateway.request("PATCH", f"/credentials/{name}", {
+                "credential_name": name, "credential_values": {"api_key": value}, "credential_info": {}
+            })
+            assert patched.status_code == 200, patched.text
+            rows: Final = read_rows(
+                'SELECT credential_values FROM "LiteLLM_CredentialsTable" WHERE credential_name = %s', (name,)
+            )
+            assert len(rows) == 1
+            stored: Final = object_value(rows[0]["credential_values"])
+            assert isinstance(stored["api_key"], str) and stored["api_key"] != value
+            for reload in (False, True):
+                if reload:
+                    response: Final = gateway.request("PATCH", f"/model/{identity}/update", {"model_info": {"description": value}})
+                    assert response.status_code == 200, response.text
+                upstream.get("/__observations").raise_for_status()
+                assert object_value(gateway.chat(model, text=f"{name} {value} reload={reload}")["usage"])["total_tokens"] == 40
+                observed: Final = upstream.get("/__observations")
+                observed.raise_for_status()
+                assert len(observed.json()["requests"]) == 1, (value, reload, observed.text)
+                assert observed.json()["requests"][0]["authorization"] == f"Bearer {value}"

tests/integration/conftest.py

@@ -2,14 +2,18 @@ from __future__ import annotations
 
 import json
 import os
+from importlib.metadata import version
 from collections.abc import Generator, Iterator
 from pathlib import Path
 from typing import Final
 
 import pytest
+import httpx
+from redis import Redis
 
-from integration._support.client import Gateway, gateway_from_environment
+from integration._support.client import Gateway, eventually, gateway_from_environment
 from integration._support.manifest import OWNED_DIRECTORIES, contracts
+from integration._support.generation import LIFECYCLE_SETTINGS
 
 COLLECTED: Final = pytest.StashKey[tuple[str, ...]]()
 REPORTS: Final = pytest.StashKey[list[pytest.TestReport]]()
@@ -66,7 +70,17 @@ def pytest_sessionfinish(session: pytest.Session, exitstatus: int) -> None:
     output: Final = Path(destination)
     output.mkdir(parents=True, exist_ok=True)
     (output / "execution.json").write_text(
-        json.dumps({"collected": collected, "passed": passed, "complete": complete, "exitstatus": exitstatus}, indent=2)
+        json.dumps({
+            "collected": collected, "passed": passed, "complete": complete, "exitstatus": exitstatus,
+            "hypothesis_version": version("hypothesis"),
+            "hypothesis_seed": session.config.getoption("hypothesis_seed"),
+            "generation": {
+                "max_examples": LIFECYCLE_SETTINGS.max_examples,
+                "stateful_step_count": LIFECYCLE_SETTINGS.stateful_step_count,
+                "database": str(LIFECYCLE_SETTINGS.database),
+                "phases": [phase.name for phase in LIFECYCLE_SETTINGS.phases],
+            },
+        }, indent=2)
         + "\n"
     )
     if not complete and exitstatus == 0:
@@ -77,3 +91,13 @@ def pytest_sessionfinish(session: pytest.Session, exitstatus: int) -> None:
 def gateway() -> Iterator[Gateway]:
     with gateway_from_environment() as value:
         yield value
+
+
+@pytest.fixture
+def peer(gateway: Gateway) -> Iterator[Gateway]:
+    url: Final = os.environ["INTEGRATION_PEER_URL"]
+    assert url.rstrip("/") != str(gateway.client.base_url).rstrip("/")
+    with Redis(host=os.environ["REDIS_HOST"], port=int(os.environ["REDIS_PORT"])) as cache:
+        eventually(lambda: cache.pubsub_numsub("litellm_proxy.auth_cache_invalidation")[0][1], lambda count: count >= 2)
+    with httpx.Client(base_url=url, timeout=15, trust_env=False) as client:
+        yield Gateway(client, gateway.key, gateway.upstream_url)

tests/integration/contracts.json

@@ -41,6 +41,40 @@
     ],
     "tests/integration/pricing/test_configured_prices.py::test_loaded_router_preserves_cached_defaults_during_real_requests": [
       "quota_management.spend_tracking.default_prices.loaded_router_preserves_cached_defaults"
+    ],
+    "tests/integration/management/test_partial_update_sequences.py::test_generated_partial_updates_preserve_persisted_and_effective_state": [
+      "mgmt.key.update.generated_sequences_preserve_state"
+    ],
+    "tests/integration/management/test_partial_update_sequences.py::test_zero_false_and_empty_values_are_not_treated_as_omission": [
+      "mgmt.key.update.false_zero_and_empty_values_affect_serving"
+    ],
+    "tests/integration/management/test_partial_update_sequences.py::test_project_omission_clear_and_invalid_update_have_distinct_effects": [
+      "mgmt.key.update.project_clear_preserves_scope",
+      "mgmt.key.update.invalid_batch_is_atomic"
+    ],
+    "tests/integration/authorization/test_warmed_policy.py::test_generated_policy_changes_reach_both_warmed_workers": [
+      "mgmt.key.update.two_workers_enforce_warmed_policy"
+    ],
+    "tests/integration/authorization/test_warmed_policy.py::test_scim_deactivation_blocks_null_and_false_keys_but_preserves_other_owners": [
+      "mgmt.user.scim.deactivation_includes_nullable_blocked_keys"
+    ],
+    "tests/integration/authorization/test_warmed_policy.py::test_warmed_team_role_demotion_prevents_later_management_writes": [
+      "mgmt.team.member_update.demoted_role_cannot_write"
+    ],
+    "tests/integration/configuration/test_effective_settings.py::test_model_block_changes_actual_route_and_leaves_other_route_working": [
+      "mgmt.model.block.changes_serving_and_preserves_control"
+    ],
+    "tests/integration/configuration/test_effective_settings.py::test_saved_retry_setting_controls_real_attempts_and_restores": [
+      "mgmt.router_settings.update.changes_observed_attempt_count"
+    ],
+    "tests/integration/configuration/test_effective_settings.py::test_credential_value_update_and_model_reload_reach_provider": [
+      "mgmt.credential.update.saved_value_reaches_wire"
+    ],
+    "tests/integration/management/test_partial_update_sequences.py::test_denied_key_update_preserves_saved_grants_and_serving": [
+      "mgmt.key.update.denied_request_preserves_effective_state"
+    ],
+    "tests/integration/authorization/test_warmed_policy.py::test_expiry_and_explicit_clear_reach_both_warmed_workers": [
+      "mgmt.key.update.expiry_changes_reach_warmed_workers"
     ]
   }
 }

tests/integration/management/test_partial_update_sequences.py

@@ -0,0 +1,200 @@
+from contextlib import ExitStack
+from hashlib import sha256
+from typing import Final
+
+import pytest
+from hypothesis import strategies as st
+from hypothesis.stateful import RuleBasedStateMachine, invariant, rule, run_state_machine_as_test
+from pydantic import JsonValue
+
+from integration._support.client import Gateway, object_value
+from integration._support.database import read_rows
+from integration._support.generation import LIFECYCLE_SETTINGS, bounded_http_requests
+
+
+@pytest.mark.covers("mgmt.key.update.generated_sequences_preserve_state")
+def test_generated_partial_updates_preserve_persisted_and_effective_state(gateway: Gateway) -> None:
+    class KeyUpdates(RuleBasedStateMachine):
+        def __init__(self) -> None:
+            super().__init__()
+            self.resources = ExitStack()
+            try:
+                scenario = self.resources.enter_context(gateway.scenario())
+                self.models = (scenario.model(), scenario.model())
+                self.key = scenario.key(models=[self.models[0]], key_alias="initial", metadata={"revision": "initial"})
+                self.expected: dict[str, JsonValue] = {
+                    "models": [self.models[0]], "key_alias": "initial", "metadata": {"revision": "initial"}
+                }
+                gateway.chat(self.models[0], key=self.key)
+            except BaseException:
+                with budget.cleanup():
+                    self.resources.close()
+                raise
+
+        @rule(alias=st.sampled_from(("first", "second", "", "unicode-λ")))
+        def alias(self, alias: str) -> None:
+            gateway.post("/key/update", {"key": self.key, "key_alias": alias})
+            self.expected["key_alias"] = alias
+
+        @rule(index=st.integers(min_value=0, max_value=1), both=st.booleans())
+        def grant(self, index: int, both: bool) -> None:
+            models: Final = list(self.models) if both else [self.models[index]]
+            gateway.post("/key/update", {"key": self.key, "models": models})
+            self.expected["models"] = models
+
+        @rule(value=st.sampled_from(("", "a", "different", "λ")))
+        def metadata(self, value: str) -> None:
+            gateway.post("/key/update", {"key": self.key, "metadata": {"revision": value}})
+            self.expected["metadata"] = {"revision": value}
+
+        @invariant()
+        def persisted_state_and_serving_match(self) -> None:
+            rows: Final = read_rows(
+                'SELECT models, key_alias, metadata FROM "LiteLLM_VerificationToken" WHERE token = %s',
+                (sha256(self.key.encode()).hexdigest(),),
+            )
+            assert rows == [self.expected]
+            info: Final = object_value(gateway.get("/key/info", {"key": self.key})["info"])
+            assert {field: info[field] for field in self.expected} == self.expected
+            for model in self.models:
+                response: Final = gateway.request(
+                    "POST", "/v1/chat/completions",
+                    {"model": model, "messages": [{"role": "user", "content": "generated update control"}]},
+                    key=self.key,
+                )
+                if model in self.expected["models"]:
+                    assert response.status_code == 200, response.text
+                    assert response.json()["usage"]["total_tokens"] == 40
+                else:
+                    assert response.status_code == 403, response.text
+                    assert response.json()["error"]["type"] == "key_model_access_denied"
+
+        def teardown(self) -> None:
+            with budget.cleanup():
+                self.resources.close()
+
+    with bounded_http_requests((gateway,), limit=2000) as budget:
+        run_state_machine_as_test(KeyUpdates, settings=LIFECYCLE_SETTINGS)
+
+
+@pytest.mark.covers("mgmt.key.update.false_zero_and_empty_values_affect_serving")
+def test_zero_false_and_empty_values_are_not_treated_as_omission(gateway: Gateway) -> None:
+    with gateway.scenario() as scenario:
+        models: Final = (scenario.model(), scenario.model())
+        key: Final = scenario.key(models=[models[0]], max_budget=0, metadata={"ordinary": "value"})
+        denied: Final = gateway.request(
+            "POST", "/v1/chat/completions",
+            {"model": models[0], "messages": [{"role": "user", "content": "zero budget"}]}, key=key,
+        )
+        assert denied.status_code == 429, denied.text
+        assert denied.json()["error"]["type"] == "budget_exceeded"
+        gateway.post("/key/update", {"key": key, "max_budget": 1, "models": [], "metadata": {}})
+        info: Final = object_value(gateway.get("/key/info", {"key": key})["info"])
+        assert (info["models"], info["metadata"], info["max_budget"]) == ([], {}, 1)
+        for model in models:
+            assert object_value(gateway.chat(model, key=key)["usage"])["total_tokens"] == 40
+        gateway.post("/key/update", {"key": key, "blocked": True})
+        blocked: Final = gateway.request(
+            "POST", "/v1/chat/completions",
+            {"model": models[0], "messages": [{"role": "user", "content": "blocked control"}]}, key=key,
+        )
+        assert blocked.status_code == 401, blocked.text
+        assert blocked.json()["error"]["type"] == "auth_error"
+        gateway.post("/key/update", {"key": key, "blocked": False})
+        assert object_value(gateway.chat(models[0], key=key)["usage"])["total_tokens"] == 40
+        rows: Final = read_rows(
+            'SELECT blocked, models, metadata, max_budget FROM "LiteLLM_VerificationToken" WHERE token = %s',
+            (sha256(key.encode()).hexdigest(),),
+        )
+        assert rows == [{"blocked": False, "models": [], "metadata": {}, "max_budget": 1.0}]
+        gateway.post("/key/update", {"key": key, "max_budget": 0})
+        assert read_rows(
+            'SELECT max_budget FROM "LiteLLM_VerificationToken" WHERE token = %s',
+            (sha256(key.encode()).hexdigest(),),
+        ) == [{"max_budget": 0.0}]
+        zero_after_update: Final = gateway.request(
+            "POST", "/v1/chat/completions",
+            {"model": models[0], "messages": [{"role": "user", "content": "updated zero budget"}]}, key=key,
+        )
+        assert zero_after_update.status_code == 429, zero_after_update.text
+        assert zero_after_update.json()["error"]["type"] == "budget_exceeded"
+        gateway.post("/key/update", {"key": key, "max_budget": None})
+        assert read_rows(
+            'SELECT max_budget FROM "LiteLLM_VerificationToken" WHERE token = %s',
+            (sha256(key.encode()).hexdigest(),),
+        ) == [{"max_budget": None}]
+        assert object_value(gateway.chat(models[0], key=key)["usage"])["total_tokens"] == 40
+
+
+@pytest.mark.covers("mgmt.key.update.project_clear_preserves_scope", "mgmt.key.update.invalid_batch_is_atomic")
+def test_project_omission_clear_and_invalid_update_have_distinct_effects(gateway: Gateway) -> None:
+    with gateway.scenario() as scenario:
+        model: Final = scenario.model()
+        outside: Final = scenario.model()
+        team: Final = scenario.team(models=[model])
+        project: Final = scenario.project(team, models=[model])
+        other: Final = scenario.project(team, models=[model])
+        key: Final = scenario.key(team_id=team, project_id=project, models=[model], key_alias="before", max_budget=5)
+        gateway.chat(model, key=key)
+        gateway.post("/key/update", {"key": key, "key_alias": "after"})
+        digest: Final = sha256(key.encode()).hexdigest()
+
+        def saved() -> list[dict[str, object]]:
+            return read_rows(
+                'SELECT key_alias, project_id, team_id, models, max_budget FROM "LiteLLM_VerificationToken" '
+                'WHERE token = %s', (digest,),
+            )
+
+        before: Final = saved()
+        assert before == [{"key_alias": "after", "project_id": project, "team_id": team, "models": [model], "max_budget": 5}]
+        for invalid in (other, ""):
+            rejected: Final = gateway.request(
+                "POST", "/key/update", {"key": key, "project_id": invalid, "key_alias": "must-not-persist"}
+            )
+            assert rejected.status_code == 400, rejected.text
+            assert saved() == before
+            gateway.chat(model, key=key)
+        gateway.post("/project/update", {"project_id": project, "blocked": True})
+        denied: Final = gateway.request(
+            "POST", "/v1/chat/completions", {"model": model, "messages": [{"role": "user", "content": "blocked project"}]},
+            key=key,
+        )
+        assert denied.status_code == 401, denied.text
+        assert denied.json()["error"]["type"] == "auth_error"
+        for _ in range(2):
+            gateway.post("/key/update", {"key": key, "project_id": None})
+            assert saved() == [{**before[0], "project_id": None}]
+            assert object_value(gateway.chat(model, key=key)["usage"])["total_tokens"] == 40
+            outside_request: Final = gateway.request(
+                "POST", "/v1/chat/completions",
+                {"model": outside, "messages": [{"role": "user", "content": "detached scope control"}]}, key=key,
+            )
+            assert outside_request.status_code == 403, outside_request.text
+            assert outside_request.json()["error"]["type"] == "key_model_access_denied"
+
+
+@pytest.mark.covers("mgmt.key.update.denied_request_preserves_effective_state")
+def test_denied_key_update_preserves_saved_grants_and_serving(gateway: Gateway) -> None:
+    with gateway.scenario() as scenario:
+        model: Final = scenario.model()
+        outside: Final = scenario.model()
+        owner: Final = scenario.user(user_role="internal_user")
+        other: Final = scenario.user(user_role="internal_user")
+        key: Final = scenario.key(user_id=owner, models=[model], key_alias="unchanged", max_budget=2)
+        caller: Final = scenario.key(user_id=other, models=[model], allowed_routes=["/key/update", "/v1/chat/completions"])
+        gateway.chat(model, key=key)
+        denied: Final = gateway.request(
+            "POST", "/key/update", {"key": key, "key_alias": "wrong", "models": [outside], "max_budget": 0}, key=caller
+        )
+        assert denied.status_code == 403, denied.text
+        assert read_rows(
+            'SELECT user_id, models, key_alias, max_budget FROM "LiteLLM_VerificationToken" WHERE token = %s',
+            (sha256(key.encode()).hexdigest(),),
+        ) == [{"user_id": owner, "models": [model], "key_alias": "unchanged", "max_budget": 2.0}]
+        assert object_value(gateway.chat(model, key=key)["usage"])["total_tokens"] == 40
+        rejected: Final = gateway.request(
+            "POST", "/v1/chat/completions",
+            {"model": outside, "messages": [{"role": "user", "content": "unchanged scope"}]}, key=key,
+        )
+        assert rejected.status_code == 403, rejected.text
+        assert rejected.json()["error"]["type"] == "key_model_access_denied"

tests/integration/run.py

@@ -16,6 +16,7 @@ def main() -> int:
     parser: Final = argparse.ArgumentParser()
     parser.add_argument("group", choices=tuple(GROUPS))
     parser.add_argument("--results", type=Path, default=Path("test-results/integration"))
+    parser.add_argument("--seed", type=int, default=int(os.environ.get("INTEGRATION_SEED", "4106601")))
     options: Final = parser.parse_args()
     root: Final = Path(__file__).resolve().parents[2]
     selected: Final = tuple(
@@ -51,6 +52,7 @@ def main() -> int:
             "no:rerunfailures",
             "--timeout=90",
             "--durations=15",
+            f"--hypothesis-seed={options.seed}",
             f"--junitxml={output / 'junit.xml'}",
         ],
         cwd=root,