test(integration): regression tests for July cost tracking, budgeting and spend bugs (#42694)

* test(integration): streamed Bedrock Messages usage cost equals the recorded spend (Pylon #6667) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): echoed cost-map model info is not persisted as deployment overrides (Pylon #6844) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): reset sweep runs on one pod per tick while replicas share the lease (Pylon #6521) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): bedrock post-call guardrail scans streamed Anthropic Messages tool use without 500 (Pylon #6503) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): realtime cached audio tokens bill at the audio cache-read rate (Pylon #6704) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): legacy GET /spend/logs returns at most the 10000 most recent rows and flags truncation (Pylon #6752) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): itemize Responses API cache write tokens as cache creation cost (Pylon #6454) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): migration entrypoint deploys pending migrations before proxy startup (Pylon #6649) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): opted-in team keys stop at the owner's personal budget (Pylon #6641) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): guardrail information stays in the spend log when the caller sends metadata on /v1/messages (Pylon #6614) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): key model allowlist is enforced on Bedrock passthrough routes (Pylon #6419) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): JWT mapped key backfills a null user email from token claims (Pylon #6266) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): scheduled budget reset recovers from a transient DB transport failure (Pylon #6582) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): team key lists models granted through a team access group (Pylon #6044) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): JWT subject without team claim lands in the configured default team (Pylon #5895) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): end-user spend lands for a key without user_id when the auth cache is Redis (Pylon #6021) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): stale-low redis counter still blocks team member over budget (Pylon #5824) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): prompt-carrying spend rows are written in byte-bounded statements (Pylon #6083) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): logs UI session_total_spend sums every round of a multi-round session (Pylon #5928) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): config.yaml guardrails are served by the guardrail usage detail and overview (Pylon #5813) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): plain chat request skips the object permission lookup (Pylon #5965) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): register july accounting regression contracts Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): isolate cost map override clear on owned proxy Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): make reset lease claim and db relay refusal deterministic Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): poll pg_stat settle, bound unbanned relay refusals, clear reset lease on teardown Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): bound relay refusals so the budget sweep can reconnect Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: kerry <kerry@berri.ai> Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
2026-09-26 01:12:21 +00:00 · 2026-09-23 09:51:56 -07:00 · 2026-09-23 09:51:56 -07:00 · 514bc181d6
commit 514bc181d6
parent a286ebf42e
22 changed files with 1830 additions and 5 deletions
--- a/tests/integration/_support/database.py
+++ b/tests/integration/_support/database.py
@ -8,7 +8,9 @@ from pydantic import JsonValue, TypeAdapter
 ROWS: Final = TypeAdapter(list[dict[str, JsonValue]])


-def read_rows(query: str, parameters: tuple[str, ...]) -> list[dict[str, JsonValue]]:
-    with psycopg.connect(os.environ["DATABASE_URL"], row_factory=dict_row) as connection:
+def read_rows(
+    query: str, parameters: tuple[str, ...], *, database_url: str | None = None
+) -> list[dict[str, JsonValue]]:
+    with psycopg.connect(database_url or os.environ["DATABASE_URL"], row_factory=dict_row) as connection:
        connection.execute("SET TRANSACTION READ ONLY")
        return ROWS.validate_python(connection.execute(query, parameters).fetchall())
--- a/tests/integration/_support/database_relay.py
+++ b/tests/integration/_support/database_relay.py
@ -0,0 +1,95 @@
+import asyncio
+import socket
+import threading
+from collections.abc import Generator
+from contextlib import contextmanager
+from typing import Final
+from urllib.parse import urlsplit, urlunsplit
+
+from pydantic import TypeAdapter
+
+PORT: Final = TypeAdapter(int)
+
+
+def _free_port() -> int:
+    with socket.socket() as reserve:
+        reserve.bind(("127.0.0.1", 0))
+        return PORT.validate_python(reserve.getsockname()[1])
+
+
+class DatabaseRelay:
+    def __init__(self, upstream_host: str, upstream_port: int, trigger: bytes) -> None:
+        self.port: Final = _free_port()
+        self._upstream_host: Final = upstream_host
+        self._upstream_port: Final = upstream_port
+        self._trigger: Final = trigger
+        self._loop: Final = asyncio.new_event_loop()
+        self._armed: Final = threading.Event()
+        self.tripped: Final = threading.Event()
+        self.refused = 0
+        self._writers: tuple[asyncio.StreamWriter, ...] = ()
+        self._ready: Final = threading.Event()
+        self._thread: Final = threading.Thread(target=self._run, daemon=True)
+
+    def arm(self) -> None:
+        self._armed.set()
+
+    def start(self) -> None:
+        self._thread.start()
+        assert self._ready.wait(10), "Database relay did not start"
+
+    def stop(self) -> None:
+        self._loop.call_soon_threadsafe(self._loop.stop)
+        self._thread.join(10)
+
+    def _run(self) -> None:
+        asyncio.set_event_loop(self._loop)
+        self._loop.run_until_complete(asyncio.start_server(self._serve, "127.0.0.1", self.port))
+        self._ready.set()
+        self._loop.run_forever()
+
+    def _drop_all(self) -> None:
+        for writer in self._writers:
+            writer.close()
+        self._writers = ()
+
+    async def _serve(self, client_reader: asyncio.StreamReader, client_writer: asyncio.StreamWriter) -> None:
+        if self.tripped.is_set() and self.refused < 5:
+            self.refused += 1
+            client_writer.close()
+            return
+        server_reader, server_writer = await asyncio.open_connection(self._upstream_host, self._upstream_port)
+        self._writers = (*self._writers, client_writer, server_writer)
+
+        async def forward(reader: asyncio.StreamReader, writer: asyncio.StreamWriter, inspect: bool) -> None:
+            try:
+                while chunk := await reader.read(65536):
+                    if inspect and self._armed.is_set() and not self.tripped.is_set() and self._trigger in chunk:
+                        self.tripped.set()
+                        self._drop_all()
+                        return
+                    writer.write(chunk)
+                    await writer.drain()
+            except (ConnectionError, asyncio.IncompleteReadError):
+                return
+            finally:
+                writer.close()
+
+        await asyncio.gather(
+            forward(client_reader, server_writer, True),
+            forward(server_reader, client_writer, False),
+        )
+
+
+@contextmanager
+def database_relay(database_url: str, trigger: bytes) -> Generator[tuple[DatabaseRelay, str]]:
+    parts: Final = urlsplit(database_url)
+    assert parts.hostname is not None and parts.port is not None, database_url
+    relay: Final = DatabaseRelay(parts.hostname, parts.port, trigger)
+    relay.start()
+    credentials: Final = f"{parts.username}:{parts.password}@" if parts.username else ""
+    relayed: Final = urlunsplit(parts._replace(netloc=f"{credentials}127.0.0.1:{relay.port}"))
+    try:
+        yield relay, relayed
+    finally:
+        relay.stop()
--- a/tests/integration/authorization/test_access_group_model_listing.py
+++ b/tests/integration/authorization/test_access_group_model_listing.py
@ -0,0 +1,51 @@
+import uuid
+from collections.abc import Iterator
+from contextlib import contextmanager
+from typing import Final
+
+import httpx
+import pytest
+
+from tests.integration._support.client import Gateway, eventually, object_value, string_value
+
+
+@contextmanager
+def _team_access_group(gateway: Gateway, team_id: str, model: str) -> Iterator[str]:
+    created: Final = gateway.request(
+        "POST",
+        "/v1/access_group",
+        {
+            "access_group_name": f"integration-{uuid.uuid4().hex}",
+            "access_model_names": [model],
+            "assigned_team_ids": [team_id],
+        },
+    )
+    assert created.status_code == 201, created.text
+    identity: Final = string_value(created.json()["access_group_id"])
+    try:
+        yield identity
+    finally:
+        deleted: Final = gateway.request("DELETE", f"/v1/access_group/{identity}")
+        assert deleted.status_code == 204, deleted.text
+
+
+def _listed_model_ids(response: httpx.Response) -> tuple[str, ...]:
+    entries: Final = response.json()["data"]
+    assert isinstance(entries, list), response.text
+    return tuple(string_value(object_value(entry)["id"]) for entry in entries)
+
+
+@pytest.mark.covers("authorization.access_groups.team_key_lists_models_granted_through_team_access_group")
+def test_team_key_lists_models_granted_through_team_access_group(gateway: Gateway) -> None:
+    with gateway.scenario() as scenario:
+        model: Final = scenario.model()
+        team_id: Final = scenario.team(models=["no-default-models"])
+        with _team_access_group(gateway, team_id, model):
+            key: Final = scenario.key(team_id=team_id)
+            response: Final = eventually(
+                lambda: gateway.request("GET", "/v1/models", key=key),
+                lambda value: value.status_code == 200 and _listed_model_ids(value) == (model,),
+                return_last_on_timeout=True,
+            )
+            assert response.status_code == 200, response.text
+            assert _listed_model_ids(response) == (model,), response.text
--- a/tests/integration/authorization/test_bedrock_passthrough_model_access.py
+++ b/tests/integration/authorization/test_bedrock_passthrough_model_access.py
@ -0,0 +1,65 @@
+import json
+from typing import Final
+
+import pytest
+from integration._support.client import Gateway
+from integration._support.upstream import _aws_event_frame
+from integration._support.wire import Reply, Request, wire_server
+
+_MODEL_ID: Final = "anthropic.claude-sonnet-5-v1:0"
+_ACTIONS: Final = ("converse", "invoke", "converse-stream", "invoke-with-response-stream")
+_REQUEST_BODY: Final = {"messages": [{"role": "user", "content": [{"text": "synthetic passthrough allowlist"}]}]}
+_CONVERSE_RESPONSE: Final = json.dumps(
+    {
+        "output": {"message": {"role": "assistant", "content": [{"text": "bedrock allowlist control"}]}},
+        "stopReason": "end_turn",
+        "usage": {"inputTokens": 11, "outputTokens": 4, "totalTokens": 15},
+        "metrics": {"latencyMs": 1},
+    }
+).encode()
+_STREAM_BYTES: Final = b"".join(
+    _aws_event_frame(kind, payload, "sc", "u")
+    for kind, payload in (
+        ("messageStart", {"role": "assistant"}),
+        ("contentBlockDelta", {"delta": {"text": "bedrock allowlist control"}, "contentBlockIndex": 0}),
+        ("messageStop", {"stopReason": "end_turn"}),
+        ("metadata", {"usage": {"inputTokens": 11, "outputTokens": 4, "totalTokens": 15}}),
+    )
+)
+
+
+def bedrock_peer(request: Request) -> Reply:
+    assert request.method == "POST", request.target
+    assert json.loads(request.body)["messages"] == _REQUEST_BODY["messages"], request.body
+    if request.target.endswith("-stream"):
+        return Reply(body=_STREAM_BYTES, content_type="application/vnd.amazon.eventstream")
+    return Reply(body=_CONVERSE_RESPONSE)
+
+
+@pytest.mark.covers("authz.key_models.bedrock_passthrough_route_model_is_enforced")
+def test_key_scoped_to_one_model_cannot_call_another_through_bedrock_passthrough_routes(gateway: Gateway) -> None:
+    with wire_server(bedrock_peer) as wire, gateway.scenario() as scenario:
+        allowed: Final = scenario.model(
+            model=f"bedrock/{_MODEL_ID}",
+            api_base=wire.url,
+            aws_access_key_id="AKIASCRIPTEDPROVIDER",
+            aws_secret_access_key="scripted-secret",
+            aws_region_name="us-east-1",
+        )
+        denied: Final = scenario.model(
+            model=f"bedrock/{_MODEL_ID}",
+            api_base=wire.url,
+            aws_access_key_id="AKIASCRIPTEDPROVIDER",
+            aws_secret_access_key="scripted-secret",
+            aws_region_name="us-east-1",
+        )
+        key: Final = scenario.key(models=[allowed])
+        for action in _ACTIONS:
+            response: Final = gateway.request("POST", f"/bedrock/model/{denied}/{action}", _REQUEST_BODY, key=key)
+            assert response.status_code == 403, f"{action}: {response.status_code} {response.text}"
+            assert response.json()["error"]["type"] == "key_model_access_denied", f"{action}: {response.text}"
+            assert wire.drain() == (), f"{action} reached the provider: {response.text}"
+        for action in _ACTIONS:
+            served: Final = gateway.request("POST", f"/bedrock/model/{allowed}/{action}", _REQUEST_BODY, key=key)
+            assert served.status_code == 200, f"{action}: {served.status_code} {served.text}"
+            assert tuple(request.target for request in wire.drain()) == (f"/model/{_MODEL_ID}/{action}",), served.text
--- a/tests/integration/authorization/test_jwt_default_team_provisioning.py
+++ b/tests/integration/authorization/test_jwt_default_team_provisioning.py
@ -0,0 +1,86 @@
+import json
+import time
+import uuid
+from pathlib import Path
+from typing import Final
+
+import jwt
+import pytest
+import yaml
+from cryptography.hazmat.primitives.asymmetric import rsa
+
+from tests.integration._support.client import Gateway, eventually
+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
+
+KEY_ID: Final = "integration-jwt-signing-key"
+TEAM_BUDGET: Final = 25.0
+
+
+def _jwks_reply(public_jwk: str) -> Reply:
+    return Reply(body=json.dumps({"keys": [{**json.loads(public_jwk), "kid": KEY_ID}]}).encode())
+
+
+@pytest.mark.covers("authorization.jwt.new_subject_without_team_claim_joins_default_team")
+def test_jwt_subject_without_team_claim_is_provisioned_into_configured_default_team(
+    gateway: Gateway, tmp_path: Path
+) -> None:
+    private_key: Final = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+    public_jwk: Final = jwt.algorithms.RSAAlgorithm.to_jwk(private_key.public_key())
+
+    def respond(request: Request) -> Reply:
+        assert request.method == "GET", request
+        return _jwks_reply(public_jwk)
+
+    with wire_server(respond) as jwks, gateway.scenario() as scenario:
+        team: Final = scenario.team()
+        config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text())
+        config["general_settings"] = {
+            **config["general_settings"],
+            "enable_jwt_auth": True,
+            "litellm_jwtauth": {"user_id_jwt_field": "sub", "user_id_upsert": True},
+        }
+        config["litellm_settings"] = {
+            **config["litellm_settings"],
+            "default_internal_user_params": {
+                "user_role": "internal_user",
+                "teams": [{"team_id": team, "user_role": "user", "max_budget_in_team": TEAM_BUDGET}],
+            },
+        }
+        path: Final = tmp_path / "jwt_default_team.yaml"
+        path.write_text(yaml.safe_dump(config))
+        subject: Final = f"integration-jwt-{uuid.uuid4().hex}"
+        token: Final = jwt.encode(
+            {"sub": subject, "iat": int(time.time()), "exp": int(time.time()) + 300},
+            private_key,
+            algorithm="RS256",
+            headers={"kid": KEY_ID},
+        )
+        with owned_proxy(gateway, tmp_path, {"JWT_PUBLIC_KEY_URL": jwks.url}, config=path) as candidate:
+            model: Final = scenario.model()
+            scenario.cleanups.callback(scenario.delete_user, subject)
+            response: Final = candidate.request(
+                "POST",
+                "/v1/chat/completions",
+                {"model": model, "messages": [{"role": "user", "content": "default team control"}]},
+                key=token,
+            )
+            assert response.status_code == 200, response.text
+            assert response.json()["choices"][0]["message"]["content"] == (
+                "Hello! This is a mock response from the fake OpenAI endpoint."
+            ), response.text
+            assert read_rows(
+                'SELECT user_id, user_role, teams FROM "LiteLLM_UserTable" WHERE user_id = %s', (subject,)
+            ) == [{"user_id": subject, "user_role": "internal_user", "teams": [team]}]
+            memberships: Final = eventually(
+                lambda: read_rows(
+                    'SELECT m.team_id, b.max_budget FROM "LiteLLM_TeamMembership" m '
+                    'JOIN "LiteLLM_BudgetTable" b ON b.budget_id = m.budget_id WHERE m.user_id = %s',
+                    (subject,),
+                ),
+                lambda rows: len(rows) == 1,
+            )
+            assert memberships == [{"team_id": team, "max_budget": TEAM_BUDGET}]
+            roster: Final = read_rows('SELECT members_with_roles FROM "LiteLLM_TeamTable" WHERE team_id = %s', (team,))
+            assert {"user_id": subject, "role": "user", "user_email": None} in roster[0]["members_with_roles"], roster
--- a/tests/integration/authorization/test_jwt_mapped_key_email_backfill.py
+++ b/tests/integration/authorization/test_jwt_mapped_key_email_backfill.py
@ -0,0 +1,112 @@
+import json
+import time
+import uuid
+from hashlib import sha256
+from pathlib import Path
+from typing import Final
+
+import jwt
+import pytest
+from cryptography.hazmat.primitives.asymmetric import rsa
+from integration._support.client import Gateway, eventually
+from integration._support.database import read_rows
+from integration._support.process import owned_proxy
+from integration._support.wire import Reply, Request, wire_server
+from jwt.algorithms import RSAAlgorithm
+
+AUDIENCE: Final = "litellm-integration"
+KEY_ID: Final = "integration-signing-key"
+CLIENT_CLAIM: Final = "client_id"
+
+
+def _proxy_config(directory: Path, model: str, upstream_url: str) -> Path:
+    config: Final = directory / "jwt_mapped_key_config.yaml"
+    config.write_text(
+        json.dumps(
+            {
+                "model_list": [
+                    {
+                        "model_name": model,
+                        "litellm_params": {
+                            "model": "openai/" + model,
+                            "api_base": upstream_url + "/v1",
+                            "api_key": "sk-upstream",
+                        },
+                    }
+                ],
+                "general_settings": {
+                    "master_key": "os.environ/LITELLM_MASTER_KEY",
+                    "database_url": "os.environ/DATABASE_URL",
+                    "store_model_in_db": True,
+                    "proxy_batch_write_at": 1,
+                    "proxy_batch_polling_interval": 1,
+                    "enable_jwt_auth": True,
+                    "litellm_jwtauth": {
+                        "user_id_jwt_field": "sub",
+                        "user_email_jwt_field": "email",
+                        "virtual_key_claim_field": CLIENT_CLAIM,
+                    },
+                },
+                "router_settings": {"disable_cooldowns": True},
+            }
+        )
+    )
+    return config
+
+
+def _signed_token(private_key: rsa.RSAPrivateKey, user_id: str, email: str, client_id: str) -> str:
+    now: Final = int(time.time())
+    return jwt.encode(
+        {"sub": user_id, "email": email, CLIENT_CLAIM: client_id, "aud": AUDIENCE, "iat": now, "exp": now + 300},
+        private_key,
+        algorithm="RS256",
+        headers={"kid": KEY_ID},
+    )
+
+
+@pytest.mark.covers("authorization.jwt.mapped_key_backfills_null_user_email_from_claims")
+def test_jwt_mapped_key_request_backfills_null_user_email_from_token_claims(gateway: Gateway, tmp_path: Path) -> None:
+    private_key: Final = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+    public_jwk: Final = json.loads(RSAAlgorithm.to_jwk(private_key.public_key()))
+    jwks: Final = json.dumps({"keys": [{**public_jwk, "kid": KEY_ID, "use": "sig", "alg": "RS256"}]}).encode()
+
+    def respond(request: Request) -> Reply:
+        assert request.target == "/jwks", request
+        return Reply(body=jwks)
+
+    model: Final = "integration-jwt-" + uuid.uuid4().hex
+    with wire_server(respond) as issuer:
+        config: Final = _proxy_config(tmp_path, model, gateway.upstream_url)
+        overrides: Final = {"JWT_PUBLIC_KEY_URL": issuer.url + "/jwks", "JWT_AUDIENCE": AUDIENCE}
+        with owned_proxy(gateway, tmp_path, overrides, config=config) as candidate, candidate.scenario() as scenario:
+            user: Final = scenario.user()
+            key: Final = scenario.key(user_id=user, models=[model])
+            client_id: Final = "integration-client-" + uuid.uuid4().hex
+            mapping: Final = candidate.post(
+                "/jwt/key/mapping/new", {"jwt_claim_name": CLIENT_CLAIM, "jwt_claim_value": client_id, "key": key}
+            )
+            scenario.cleanups.callback(candidate.post, "/jwt/key/mapping/delete", {"id": mapping["id"]})
+            assert read_rows('SELECT user_email FROM "LiteLLM_UserTable" WHERE user_id = %s', (user,)) == [
+                {"user_email": None}
+            ]
+            email: Final = f"{user}@integration.example"
+            token: Final = _signed_token(private_key, user, email, client_id)
+            response: Final = candidate.request(
+                "POST",
+                "/v1/chat/completions",
+                {"model": model, "messages": [{"role": "user", "content": "jwt email backfill control"}]},
+                key=token,
+            )
+            assert response.status_code == 200, response.text
+            assert read_rows('SELECT user_email FROM "LiteLLM_UserTable" WHERE user_id = %s', (user,)) == [
+                {"user_email": email}
+            ]
+            spend_rows: Final = eventually(
+                lambda: read_rows(
+                    'SELECT api_key, "user" FROM "LiteLLM_SpendLogs" WHERE request_id = %s',
+                    (str(response.json()["id"]),),
+                ),
+                lambda rows: len(rows) == 1,
+                seconds=70,
+            )
+            assert spend_rows == [{"api_key": sha256(key.encode()).hexdigest(), "user": user}]
--- a/tests/integration/authorization/test_object_permission_lookup.py
+++ b/tests/integration/authorization/test_object_permission_lookup.py
@ -0,0 +1,74 @@
+import time
+from typing import Final
+
+import pytest
+
+from tests.integration._support.client import Gateway, eventually
+from tests.integration._support.database import read_rows
+
+PLAIN_REQUESTS: Final = 10
+STATS_FLUSH_WINDOW_SECONDS: Final = 11.0
+
+
+def _object_permission_reads() -> int:
+    rows: Final = read_rows(
+        "SELECT seq_scan + idx_scan AS reads FROM pg_stat_user_tables WHERE relname = %s",
+        ("LiteLLM_ObjectPermissionTable",),
+    )
+    reads: Final = rows[0]["reads"]
+    assert isinstance(reads, int), rows
+    return reads
+
+
+def _settled_object_permission_reads(previous: int, unchanged_since: float) -> int:
+    changed: Final = eventually(
+        lambda: _object_permission_reads() != previous,
+        lambda drifted: drifted,
+        seconds=STATS_FLUSH_WINDOW_SECONDS - (time.monotonic() - unchanged_since),
+        return_last_on_timeout=True,
+    )
+    if not changed:
+        return previous
+    return _settled_object_permission_reads(_object_permission_reads(), time.monotonic())
+
+
+def _assert_plain_chat_served(gateway: Gateway, model: str, key: str) -> None:
+    response: Final = gateway.request(
+        "POST",
+        "/v1/chat/completions",
+        {"model": model, "messages": [{"role": "user", "content": "no vector stores"}]},
+        key=key,
+    )
+    assert response.status_code == 200, response.text
+    assert response.json()["choices"][0]["message"]["content"] == (
+        "Hello! This is a mock response from the fake OpenAI endpoint."
+    )
+
+
+def _assert_forbidden_vector_store_denied(gateway: Gateway, model: str, key: str) -> None:
+    response: Final = gateway.request(
+        "POST",
+        "/v1/chat/completions",
+        {"model": model, "messages": [{"role": "user", "content": "forbidden"}], "vector_store_ids": ["vs_forbidden"]},
+        key=key,
+    )
+    assert response.status_code == 401, response.text
+    assert response.json()["error"]["type"] == "key_vector_store_access_denied", response.text
+
+
+@pytest.mark.covers("authorization.vector_store.plain_request_skips_object_permission_lookup")
+def test_chat_request_without_vector_stores_does_not_read_object_permission_table(gateway: Gateway) -> None:
+    with gateway.scenario() as scenario:
+        model: Final = scenario.model()
+        key: Final = scenario.key(models=[model], object_permission={"vector_stores": ["vs_allowed"]})
+        _assert_plain_chat_served(gateway, model, key)
+        before_control: Final = _object_permission_reads()
+        _assert_forbidden_vector_store_denied(gateway, model, key)
+        eventually(_object_permission_reads, lambda reads: reads > before_control, seconds=15)
+        baseline: Final = _settled_object_permission_reads(_object_permission_reads(), time.monotonic())
+        for _ in range(PLAIN_REQUESTS):
+            _assert_plain_chat_served(gateway, model, key)
+        after: Final = _settled_object_permission_reads(_object_permission_reads(), time.monotonic())
+        assert after - baseline < PLAIN_REQUESTS, (
+            f"{PLAIN_REQUESTS} plain chat requests added {after - baseline} object permission reads"
+        )
--- a/tests/integration/database/test_migration_entrypoint.py
+++ b/tests/integration/database/test_migration_entrypoint.py
@ -0,0 +1,99 @@
+import os
+import shutil
+import subprocess
+import sys
+import uuid
+from collections.abc import Iterator
+from contextlib import contextmanager
+from pathlib import Path
+from typing import Final
+from urllib.parse import urlsplit, urlunsplit
+
+import psycopg
+import pytest
+from integration._support.client import Gateway
+from integration._support.process import owned_proxy
+from psycopg import sql
+from psycopg.rows import dict_row
+
+REPO_ROOT: Final = Path(__file__).resolve().parents[3]
+PRISMA_DIR: Final = REPO_ROOT / "litellm-proxy-extras" / "litellm_proxy_extras"
+MISSING_MIGRATION: Final = "20260626120000_add_mcp_tool_search_enabled"
+SHIPPED_MIGRATIONS: Final = tuple(sorted(path.name for path in (PRISMA_DIR / "migrations").iterdir() if path.is_dir()))
+
+
+@contextmanager
+def fresh_database() -> Iterator[str]:
+    name: Final = f"integration_upgrade_{uuid.uuid4().hex}"
+    admin_url: Final = os.environ["DATABASE_URL"]
+    parsed: Final = urlsplit(admin_url)
+    with psycopg.connect(admin_url, autocommit=True) as admin:
+        admin.execute(sql.SQL("CREATE DATABASE {}").format(sql.Identifier(name)))
+        try:
+            yield urlunsplit(parsed._replace(path=f"/{name}"))
+        finally:
+            admin.execute(sql.SQL("DROP DATABASE {} WITH (FORCE)").format(sql.Identifier(name)))
+
+
+def deploy_older_schema(database_url: str, directory: Path) -> None:
+    older: Final = directory / "older-release"
+    (older / "migrations").mkdir(parents=True)
+    shutil.copy(PRISMA_DIR / "schema.prisma", older / "schema.prisma")
+    shutil.copy(PRISMA_DIR / "migrations" / "migration_lock.toml", older / "migrations" / "migration_lock.toml")
+    for name in (name for name in SHIPPED_MIGRATIONS if name < MISSING_MIGRATION):
+        shutil.copytree(PRISMA_DIR / "migrations" / name, older / "migrations" / name)
+    subprocess.run(
+        [sys.executable, "-I", "-m", "prisma", "migrate", "deploy", "--schema", str(older / "schema.prisma")],
+        check=True,
+        capture_output=True,
+        text=True,
+        timeout=300,
+        env={**os.environ, "DATABASE_URL": database_url},
+    )
+
+
+def applied_migrations(database_url: str) -> tuple[str, ...]:
+    with psycopg.connect(database_url, row_factory=dict_row) as connection:
+        rows: Final = connection.execute(
+            'SELECT migration_name FROM "_prisma_migrations" '
+            "WHERE finished_at IS NOT NULL AND rolled_back_at IS NULL ORDER BY migration_name"
+        ).fetchall()
+    return tuple(str(row["migration_name"]) for row in rows)
+
+
+def object_permission_columns(database_url: str) -> tuple[str, ...]:
+    with psycopg.connect(database_url, row_factory=dict_row) as connection:
+        rows: Final = connection.execute(
+            "SELECT column_name FROM information_schema.columns "
+            "WHERE table_name = 'LiteLLM_ObjectPermissionTable' AND column_name = 'mcp_tool_search_enabled'"
+        ).fetchall()
+    return tuple(str(row["column_name"]) for row in rows)
+
+
+@pytest.mark.covers("other.database.migrations.entrypoint_deploys_pending_migrations_before_startup")
+def test_migration_entrypoint_upgrades_an_older_schema_so_the_proxy_serves_mcp_tools(
+    gateway: Gateway, tmp_path: Path
+) -> None:
+    with fresh_database() as database_url:
+        deploy_older_schema(database_url, tmp_path)
+        assert object_permission_columns(database_url) == ()
+        assert applied_migrations(database_url) == tuple(
+            name for name in SHIPPED_MIGRATIONS if name < MISSING_MIGRATION
+        )
+        entrypoint: Final = subprocess.run(
+            [sys.executable, "-I", "-m", "litellm.proxy.prisma_migration"],
+            capture_output=True,
+            text=True,
+            timeout=300,
+            cwd=REPO_ROOT,
+            env={**os.environ, "DATABASE_URL": database_url},
+        )
+        assert entrypoint.returncode == 0, entrypoint.stdout + entrypoint.stderr
+        assert object_permission_columns(database_url) == ("mcp_tool_search_enabled",), entrypoint.stdout
+        assert applied_migrations(database_url) == SHIPPED_MIGRATIONS, entrypoint.stdout
+        with owned_proxy(
+            gateway, tmp_path, {"DATABASE_URL": database_url, "DISABLE_SCHEMA_UPDATE": "true"}
+        ) as upgraded:
+            tools: Final = upgraded.request("GET", "/mcp-rest/tools/list")
+            assert tools.status_code == 200, tools.text
+            assert tools.json()["tools"] == [], tools.text
--- a/tests/integration/management/test_guardrail_usage_config_guardrail.py
+++ b/tests/integration/management/test_guardrail_usage_config_guardrail.py
@ -0,0 +1,77 @@
+import uuid
+from pathlib import Path
+from typing import Final
+
+import pytest
+import yaml
+from pydantic import JsonValue
+
+from tests.integration._support.client import Gateway, object_value, string_value
+from tests.integration._support.process import owned_proxy
+
+WINDOW: Final = {"start_date": "2026-01-01", "end_date": "2026-01-07"}
+
+
+def listed_guardrail(gateway: Gateway, guardrail_name: str) -> dict[str, JsonValue]:
+    listed: Final = gateway.get("/v2/guardrails/list")["guardrails"]
+    assert isinstance(listed, list), listed
+    matches: Final = tuple(object_value(row) for row in listed if object_value(row)["guardrail_name"] == guardrail_name)
+    assert len(matches) == 1, f"{guardrail_name} appears {len(matches)} times in {listed}"
+    return matches[0]
+
+
+@pytest.mark.covers("mgmt.guardrails.usage.config_yaml_guardrail_has_detail_and_overview_row")
+def test_config_yaml_guardrail_is_served_by_usage_detail_and_overview(gateway: Gateway, tmp_path: Path) -> None:
+    guardrail_name: Final = "tool-permission-" + uuid.uuid4().hex
+    config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text())
+    config["guardrails"] = [
+        {
+            "guardrail_name": guardrail_name,
+            "litellm_params": {
+                "guardrail": "tool_permission",
+                "mode": "post_call",
+                "default_on": False,
+                "rules": [{"id": "deny_delete", "tool_name": "(?i)^.*(delete|drop).*", "decision": "deny"}],
+                "default_action": "allow",
+                "on_disallowed_action": "block",
+            },
+            "guardrail_info": {"type": "Tool Permission", "description": "declared in config.yaml"},
+        }
+    ]
+    path: Final = tmp_path / "guardrail.yaml"
+    path.write_text(yaml.safe_dump(config))
+    with owned_proxy(gateway, tmp_path, {}, config=path) as candidate:
+        guardrail_id: Final = string_value(listed_guardrail(candidate, guardrail_name)["guardrail_id"])
+        detail: Final = candidate.request("GET", f"/guardrails/usage/detail/{guardrail_id}", params=WINDOW)
+        assert detail.status_code == 200, detail.text
+        body: Final = object_value(detail.json())
+        assert {
+            "guardrail_id": body["guardrail_id"],
+            "guardrail_name": body["guardrail_name"],
+            "provider": body["provider"],
+            "type": body["type"],
+            "description": body["description"],
+            "requestsEvaluated": body["requestsEvaluated"],
+            "failRate": body["failRate"],
+        } == {
+            "guardrail_id": guardrail_id,
+            "guardrail_name": guardrail_name,
+            "provider": "tool_permission",
+            "type": "Tool Permission",
+            "description": "declared in config.yaml",
+            "requestsEvaluated": 0,
+            "failRate": 0.0,
+        }, detail.text
+        overview: Final = candidate.request("GET", "/guardrails/usage/overview", params=WINDOW)
+        assert overview.status_code == 200, overview.text
+        rows: Final = object_value(overview.json())["rows"]
+        assert isinstance(rows, list), overview.text
+        config_rows: Final = tuple(object_value(row) for row in rows if object_value(row)["id"] == guardrail_id)
+        assert len(config_rows) == 1, overview.text
+        assert (config_rows[0]["name"], config_rows[0]["provider"], config_rows[0]["requestsEvaluated"]) == (
+            guardrail_name,
+            "tool_permission",
+            0,
+        ), overview.text
+        missing: Final = candidate.request("GET", f"/guardrails/usage/detail/{uuid.uuid4()}", params=WINDOW)
+        assert missing.status_code == 404, missing.text
--- a/tests/integration/observability/test_guardrail_effects.py
+++ b/tests/integration/observability/test_guardrail_effects.py
@ -5,7 +5,8 @@ from typing import Final

 import pytest
 import yaml
-from integration._support.client import Gateway
+from integration._support.client import Gateway, eventually, object_value
+from integration._support.database import read_rows
 from integration._support.mcp import mcp_peer, register_mcp, tool_names
 from integration._support.process import owned_proxy
 from integration._support.wire import Reply, Request, wire_server
@ -88,6 +89,91 @@ def test_guardrail_rewrites_system_and_user_in_actual_anthropic_request(gateway:
            assert len(policy.drain()) == len(upstream.drain()) == 1


+@pytest.mark.covers("other.observability.guardrails.anthropic_messages_caller_metadata_keeps_guardrail_spend_log")
+def test_anthropic_messages_with_caller_metadata_keeps_guardrail_information_in_spend_log(
+    gateway: Gateway, tmp_path: Path
+) -> None:
+    identity: Final = "guardrail" + uuid.uuid4().hex
+    prompt: Final = "synthetic allowed prompt " + identity
+    caller_metadata: Final = {"user_id": "device-account-session"}
+
+    def guardrail(request: Request) -> Reply:
+        assert request.target == "/beta/litellm_basic_guardrail_api"
+        assert json.loads(request.body)["texts"] == [prompt]
+        return Reply(body=json.dumps({"action": "NONE"}).encode())
+
+    def provider(request: Request) -> Reply:
+        assert request.target == "/v1/messages"
+        body: Final = json.loads(request.body)
+        assert body["messages"] == [{"role": "user", "content": prompt}]
+        assert body["metadata"] == caller_metadata
+        return Reply(
+            body=json.dumps(
+                {
+                    "id": identity,
+                    "type": "message",
+                    "role": "assistant",
+                    "model": "claude-sonnet-4-5-20250929",
+                    "content": [{"type": "text", "text": "permitted response"}],
+                    "stop_reason": "end_turn",
+                    "stop_sequence": None,
+                    "usage": {"input_tokens": 11, "output_tokens": 4},
+                }
+            ).encode()
+        )
+
+    with wire_server(guardrail) as policy, wire_server(provider) as upstream:
+        config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text())
+        config["guardrails"] = [
+            {
+                "guardrail_name": identity,
+                "litellm_params": {
+                    "guardrail": "generic_guardrail_api",
+                    "mode": "pre_call",
+                    "default_on": True,
+                    "api_base": policy.url,
+                    "api_key": "synthetic-guardrail-key",
+                },
+            }
+        ]
+        path: Final = tmp_path / "caller-metadata.yaml"
+        path.write_text(yaml.safe_dump(config))
+        with owned_proxy(gateway, tmp_path, {}, config=path) as candidate, candidate.scenario() as scenario:
+            model: Final = scenario.model(
+                model="anthropic/claude-sonnet-4-5-20250929", api_base=upstream.url, api_key="synthetic-anthropic-key"
+            )
+            response: Final = candidate.request(
+                "POST",
+                "/v1/messages",
+                {
+                    "model": model,
+                    "max_tokens": 16,
+                    "messages": [{"role": "user", "content": prompt}],
+                    "metadata": caller_metadata,
+                },
+            )
+            assert response.status_code == 200, response.text
+            assert response.json()["content"] == [{"type": "text", "text": "permitted response"}], response.text
+            assert response.headers["x-litellm-applied-guardrails"] == identity, dict(response.headers)
+            assert len(policy.drain()) == len(upstream.drain()) == 1
+            rows: Final = eventually(
+                lambda: read_rows(
+                    'SELECT call_type, metadata FROM "LiteLLM_SpendLogs" WHERE model_group=%s',
+                    (model,),
+                ),
+                lambda values: len(values) == 1,
+                seconds=70,
+            )
+            assert rows[0]["call_type"] == "anthropic_messages", rows[0]
+            saved: Final = object_value(rows[0]["metadata"])
+            entries: Final = saved["guardrail_information"]
+            assert isinstance(entries, list) and len(entries) == 1, saved
+            entry: Final = object_value(entries[0])
+            assert entry["guardrail_name"] == identity, saved
+            assert entry["guardrail_mode"] == "pre_call", saved
+            assert entry["guardrail_status"] == "success", saved
+
+
@pytest.mark.covers("other.observability.guardrails.denial_prevents_provider_with_allowed_control")
 def test_guardrail_denial_prevents_provider_and_preserves_allowed_control(gateway: Gateway, tmp_path: Path) -> None:
    identity: Final = "guardrail" + uuid.uuid4().hex
@ -242,6 +328,110 @@ def test_bedrock_passthrough_converse_guardrail_ignores_denied_term_in_tool_defi
            assert [json.loads(request.body)["texts"] for request in policy.drain()] == [[allowed], [denied]]


+@pytest.mark.covers("other.observability.guardrails.bedrock_post_call_scans_streamed_anthropic_messages_tool_use")
+def test_bedrock_guardrail_streams_anthropic_messages_tool_use_instead_of_chunk_builder_500(
+    gateway: Gateway, tmp_path: Path
+) -> None:
+    identity: Final = "guardrail" + uuid.uuid4().hex
+    guardrail_id: Final = "synthetic" + uuid.uuid4().hex[:8]
+    spoken: Final = "Checking the forecast"
+    frames: Final = (
+        'event: message_start\ndata: {"type": "message_start", "message": {"id": "msg_synthetic", "type": "message", '
+        '"role": "assistant", "model": "claude-sonnet-4-5-20250929", "content": [], "stop_reason": null, '
+        '"stop_sequence": null, "usage": {"input_tokens": 11, "output_tokens": 1}}}\n\n',
+        'event: content_block_start\ndata: {"type": "content_block_start", "index": 0, '
+        '"content_block": {"type": "text", "text": ""}}\n\n',
+        'event: content_block_delta\ndata: {"type": "content_block_delta", "index": 0, '
+        f'"delta": {{"type": "text_delta", "text": "{spoken}"}}}}\n\n',
+        'event: content_block_stop\ndata: {"type": "content_block_stop", "index": 0}\n\n',
+        'event: content_block_start\ndata: {"type": "content_block_start", "index": 1, '
+        '"content_block": {"type": "tool_use", "id": "toolu_synthetic", "name": "lookup_weather", "input": {}}}\n\n',
+        'event: content_block_delta\ndata: {"type": "content_block_delta", "index": 1, '
+        '"delta": {"type": "input_json_delta", "partial_json": "{\\"city\\": \\"Paris\\"}"}}\n\n',
+        'event: content_block_stop\ndata: {"type": "content_block_stop", "index": 1}\n\n',
+        'event: message_delta\ndata: {"type": "message_delta", "delta": {"stop_reason": "tool_use", '
+        '"stop_sequence": null}, "usage": {"output_tokens": 9}}\n\n',
+        'event: message_stop\ndata: {"type": "message_stop"}\n\n',
+    )
+    tools: Final = [
+        {
+            "name": "lookup_weather",
+            "description": "Look up the forecast for a city",
+            "input_schema": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]},
+        }
+    ]
+
+    def guardrail(request: Request) -> Reply:
+        assert request.target == f"/guardrail/{guardrail_id}/version/DRAFT/apply", request.target
+        body: Final = json.loads(request.body)
+        assert body["source"] == "OUTPUT", body
+        assert body["content"] == [{"text": {"text": spoken}}], body
+        return Reply(body=json.dumps({"action": "NONE", "outputs": [], "assessments": []}).encode())
+
+    def provider(request: Request) -> Reply:
+        assert request.target == "/v1/messages"
+        body: Final = json.loads(request.body)
+        assert body["stream"] is True, body
+        assert body["tools"] == tools, body
+        return Reply(content_type="text/event-stream", chunks=tuple(frame.encode() for frame in frames))
+
+    with wire_server(guardrail) as policy, wire_server(provider) as upstream:
+        config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text())
+        config["guardrails"] = [
+            {
+                "guardrail_name": identity,
+                "litellm_params": {
+                    "guardrail": "bedrock",
+                    "mode": "post_call",
+                    "default_on": True,
+                    "mask_response_content": True,
+                    "guardrailIdentifier": guardrail_id,
+                    "guardrailVersion": "DRAFT",
+                    "aws_region_name": "us-east-1",
+                    "aws_access_key_id": "AKIASYNTHETICGUARDRAIL",
+                    "aws_secret_access_key": "synthetic-secret",
+                    "aws_bedrock_runtime_endpoint": policy.url,
+                },
+            }
+        ]
+        path: Final = tmp_path / "bedrock-stream.yaml"
+        path.write_text(yaml.safe_dump(config))
+        with owned_proxy(gateway, tmp_path, {}, config=path) as candidate, candidate.scenario() as scenario:
+            model: Final = scenario.model(
+                model="anthropic/claude-sonnet-4-5-20250929", api_base=upstream.url, api_key="synthetic-anthropic-key"
+            )
+            response: Final = candidate.request(
+                "POST",
+                "/v1/messages",
+                {
+                    "model": model,
+                    "max_tokens": 64,
+                    "stream": True,
+                    "tools": tools,
+                    "messages": [{"role": "user", "content": f"What is the weather in Paris? {identity}"}],
+                },
+            )
+            assert response.status_code == 200, response.text
+            head, separator, tail = response.text.partition("\n\n")
+            assert separator == "\n\n", response.text
+            assert head.startswith("event: message_start\ndata: "), response.text
+            assert json.loads(head.removeprefix("event: message_start\ndata: ")) == {
+                "type": "message_start",
+                "message": {
+                    "id": "msg_synthetic",
+                    "type": "message",
+                    "role": "assistant",
+                    "model": model,
+                    "content": [],
+                    "stop_reason": None,
+                    "stop_sequence": None,
+                    "usage": {"input_tokens": 11, "output_tokens": 1},
+                },
+            }, response.text
+            assert tail == "".join(frames[1:]), response.text
+            assert len(policy.drain()) == len(upstream.drain()) == 1
+
+
@pytest.mark.covers("other.mcp.guardrails.request_selection_blocks_resolved_tool_without_execution")
 def test_request_selected_mcp_guardrail_blocks_direct_and_virtual_calls(gateway: Gateway, tmp_path: Path) -> None:
    guardrail = "mcp-policy-" + uuid.uuid4().hex
--- a/tests/integration/pricing/test_configured_prices.py
+++ b/tests/integration/pricing/test_configured_prices.py
@ -8,8 +8,10 @@ import pytest
 import yaml
 from pydantic import JsonValue

+from litellm import get_model_info
 from tests.integration._support.client import Gateway, eventually, object_value, string_value
 from tests.integration._support.database import read_rows
+from tests.integration._support.process import owned_proxy


@pytest.mark.covers("quota_management.spend_tracking.custom_price.matches_input_rates")
@ -169,6 +171,49 @@ def test_saving_echoed_model_info_does_not_freeze_cost_map_price_into_deployment
        assert {key: value for key, value in stored.items() if key in COST_MAP_DISPLAY_PRICING_KEYS} == {}, stored


+def displayed_model_info(gateway: Gateway, model: str) -> dict[str, JsonValue]:
+    entries: Final = gateway.get("/model/info")["data"]
+    assert isinstance(entries, list)
+    target: Final = next(object_value(entry) for entry in entries if object_value(entry)["model_name"] == model)
+    return object_value(target["model_info"])
+
+
+@pytest.mark.covers("pricing.model_update.echoed_cost_map_metadata_is_not_persisted_as_override")
+def test_saving_echoed_model_info_does_not_persist_cost_map_metadata_as_overrides(gateway: Gateway) -> None:
+    catalog_entry: Final = get_model_info("openai/gpt-4o-mini")
+    with gateway.scenario() as scenario:
+        model: Final = scenario.model()
+        displayed: Final = displayed_model_info(gateway, model)
+        identity: Final = string_value(displayed["id"])
+        assert displayed["key"] == catalog_entry["key"], displayed
+        assert displayed["max_input_tokens"] == catalog_entry["max_input_tokens"], displayed
+        saved: Final = gateway.request(
+            "PATCH", f"/model/{identity}/update", {"model_info": {**displayed, "description": "echoed ui save"}}
+        )
+        assert saved.status_code == 200, saved.text
+        stored: Final = persisted_model_info(identity)
+        assert stored["description"] == "echoed ui save", stored
+        assert {key: value for key, value in stored.items() if key in catalog_entry} == {}, stored
+
+
+@pytest.mark.covers("pricing.model_update.echoing_cost_map_value_back_clears_stored_override")
+def test_saving_the_cost_map_value_back_over_a_stored_override_clears_it(gateway: Gateway, tmp_path: Path) -> None:
+    catalog_limit: Final = get_model_info("openai/gpt-4o-mini")["max_input_tokens"]
+    assert isinstance(catalog_limit, int) and catalog_limit != 4321, catalog_limit
+    with owned_proxy(gateway, tmp_path, {}) as candidate, candidate.scenario() as scenario:
+        overridden: Final = scenario.model(model_info={"max_input_tokens": 4321})
+        displayed: Final = displayed_model_info(candidate, overridden)
+        identity: Final = string_value(displayed["id"])
+        assert displayed["max_input_tokens"] == 4321, displayed
+        assert persisted_model_info(identity)["max_input_tokens"] == 4321
+        saved: Final = candidate.request(
+            "PATCH", f"/model/{identity}/update", {"model_info": {**displayed, "max_input_tokens": catalog_limit}}
+        )
+        assert saved.status_code == 200, saved.text
+        stored: Final = persisted_model_info(identity)
+        assert "max_input_tokens" not in stored, stored
+
+
@pytest.mark.covers("quota_management.spend_tracking.default_prices.loaded_router_preserves_cached_defaults")
 def test_loaded_router_preserves_cached_defaults_during_real_requests(gateway: Gateway, tmp_path: Path) -> None:
    from litellm import Router
--- a/tests/integration/pricing/test_realtime_cached_audio_pricing.py
+++ b/tests/integration/pricing/test_realtime_cached_audio_pricing.py
@ -0,0 +1,131 @@
+import asyncio
+import json
+import os
+import uuid
+from hashlib import sha256
+from typing import Final
+
+import pytest
+import websockets
+from pydantic import BaseModel, ConfigDict, JsonValue
+
+import litellm
+from tests.integration._support.client import JSON_OBJECT, Gateway, eventually
+from tests.integration._support.database import read_rows
+from tests.integration._support.upstream import delete_scenario, register_scenario
+from tests.integration.cost_calculation.cost_tracking_case import RealtimeResponse
+
+
+class RealtimeRates(BaseModel):
+    model_config = ConfigDict(frozen=True)
+
+    input_cost_per_token: float
+    input_cost_per_audio_token: float
+    cache_read_input_token_cost: float
+    cache_read_input_audio_token_cost: float | None = None
+    output_cost_per_token: float
+    output_cost_per_audio_token: float
+
+
+MODEL: Final = "gpt-realtime-2"
+RATES: Final = RealtimeRates.model_validate(litellm.get_model_info(MODEL, custom_llm_provider="openai"))
+TEXT_RATE: Final = RATES.input_cost_per_token
+AUDIO_RATE: Final = RATES.input_cost_per_audio_token
+CACHED_TEXT_RATE: Final = RATES.cache_read_input_token_cost
+CACHED_AUDIO_RATE: Final = (
+    RATES.cache_read_input_token_cost
+    if RATES.cache_read_input_audio_token_cost is None
+    else RATES.cache_read_input_audio_token_cost
+)
+OUTPUT_TEXT_RATE: Final = RATES.output_cost_per_token
+OUTPUT_AUDIO_RATE: Final = RATES.output_cost_per_audio_token
+INPUT_TEXT_TOKENS: Final = 116
+INPUT_AUDIO_TOKENS: Final = 167
+CACHED_TEXT_TOKENS: Final = 64
+CACHED_AUDIO_TOKENS: Final = 128
+INPUT_TOKENS: Final = INPUT_TEXT_TOKENS + INPUT_AUDIO_TOKENS
+OUTPUT_TEXT_TOKENS: Final = 8
+OUTPUT_AUDIO_TOKENS: Final = 12
+OUTPUT_TOKENS: Final = OUTPUT_TEXT_TOKENS + OUTPUT_AUDIO_TOKENS
+EXPECTED_INPUT_COST: Final = (
+    (INPUT_TEXT_TOKENS - CACHED_TEXT_TOKENS) * TEXT_RATE
+    + CACHED_TEXT_TOKENS * CACHED_TEXT_RATE
+    + (INPUT_AUDIO_TOKENS - CACHED_AUDIO_TOKENS) * AUDIO_RATE
+    + CACHED_AUDIO_TOKENS * CACHED_AUDIO_RATE
+)
+EXPECTED_OUTPUT_COST: Final = OUTPUT_TEXT_TOKENS * OUTPUT_TEXT_RATE + OUTPUT_AUDIO_TOKENS * OUTPUT_AUDIO_RATE
+
+
+def cached_audio_response_done() -> RealtimeResponse:
+    return RealtimeResponse(
+        content_type="application/x-realtime",
+        events=(
+            {
+                "type": "response.done",
+                "event_id": "evt_$REQUEST_ID",
+                "response": {
+                    "id": "resp_$REQUEST_ID",
+                    "object": "realtime.response",
+                    "status": "completed",
+                    "output": [],
+                    "usage": {
+                        "total_tokens": INPUT_TOKENS + OUTPUT_TOKENS,
+                        "input_tokens": INPUT_TOKENS,
+                        "output_tokens": OUTPUT_TOKENS,
+                        "input_token_details": {
+                            "text_tokens": INPUT_TEXT_TOKENS,
+                            "audio_tokens": INPUT_AUDIO_TOKENS,
+                            "cached_tokens": CACHED_TEXT_TOKENS + CACHED_AUDIO_TOKENS,
+                            "cached_tokens_details": {
+                                "text_tokens": CACHED_TEXT_TOKENS,
+                                "audio_tokens": CACHED_AUDIO_TOKENS,
+                            },
+                        },
+                        "output_token_details": {
+                            "text_tokens": OUTPUT_TEXT_TOKENS,
+                            "audio_tokens": OUTPUT_AUDIO_TOKENS,
+                        },
+                    },
+                },
+            },
+        ),
+    )
+
+
+async def _one_realtime_turn(proxy_url: str, key: str, model: str) -> dict[str, JsonValue]:
+    async with websockets.connect(
+        f"{proxy_url.replace('http://', 'ws://').replace('https://', 'wss://')}/v1/realtime?model={model}",
+        additional_headers={"Authorization": f"Bearer {key}"},
+    ) as websocket:
+        session: Final = JSON_OBJECT.validate_json(await websocket.recv())
+        await websocket.send(json.dumps({"type": "response.create"}))
+        async for message in websocket:
+            if JSON_OBJECT.validate_json(message).get("type") == "response.done":
+                return session
+        raise AssertionError(f"websocket closed before response.done for {model}")
+
+
+@pytest.mark.covers("pricing.realtime.cached_audio_tokens_bill_at_audio_cache_read_rate")
+def test_realtime_cached_audio_tokens_bill_at_audio_cache_read_rate_not_full_audio_rate(gateway: Gateway) -> None:
+    with gateway.scenario() as scenario:
+        scenario_id: Final = f"realtime-cached-audio-{uuid.uuid4().hex[:12]}"
+        handle: Final = register_scenario(scenario_id, cached_audio_response_done())
+        scenario.cleanups.callback(delete_scenario, handle)
+        key: Final = scenario.key()
+        model: Final = scenario.model(
+            model=f"openai/{MODEL}", api_key=scenario_id, api_base=gateway.upstream_url.rstrip("/")
+        )
+        session: Final = asyncio.run(_one_realtime_turn(os.environ["INTEGRATION_PROXY_URL"].rstrip("/"), key, model))
+        assert session.get("type") == "session.created", session
+        rows: Final = eventually(
+            lambda: read_rows(
+                'SELECT spend, prompt_tokens, completion_tokens, call_type FROM "LiteLLM_SpendLogs" WHERE api_key = %s',
+                (sha256(key.encode()).hexdigest(),),
+            ),
+            lambda values: len(values) == 1,
+            seconds=70,
+        )
+        assert rows[0]["call_type"] == "_arealtime", rows
+        assert rows[0]["prompt_tokens"] == INPUT_TOKENS, rows
+        assert rows[0]["completion_tokens"] == OUTPUT_TOKENS, rows
+        assert float(str(rows[0]["spend"])) == pytest.approx(EXPECTED_INPUT_COST + EXPECTED_OUTPUT_COST, rel=1e-6), rows
--- a/tests/integration/spend/test_cache_and_quota.py
+++ b/tests/integration/spend/test_cache_and_quota.py
@ -1,19 +1,27 @@
 import json
+import os
 import threading
 import uuid
+from collections.abc import Generator
 from concurrent.futures import ThreadPoolExecutor
-from contextlib import ExitStack
+from contextlib import ExitStack, contextmanager
 from hashlib import sha256
+from pathlib import Path
 from typing import Final
+from urllib.parse import urlsplit, urlunsplit

 import httpx
+import psycopg
 import pytest
 from hypothesis import strategies as st
 from hypothesis.stateful import RuleBasedStateMachine, rule, run_state_machine_as_test
-from integration._support.client import Gateway, eventually
+from integration._support.client import Gateway, eventually, string_value
 from integration._support.database import read_rows
+from integration._support.database_relay import database_relay
 from integration._support.generation import LIFECYCLE_SETTINGS, bounded_http_requests
+from integration._support.process import owned_proxy
 from integration._support.wire import Reply, Request, wire_server
+from psycopg import sql


@pytest.mark.covers("quota_management.response_cache.generated_sequences_preserve_content_and_accounting")
@ -214,6 +222,99 @@ def test_key_budget_at_boundary_blocks_provider_then_explicit_reset_restores(gat
        assert upstream.get("/__observations").json()["requests"] == []


+RESET_SWEEP_QUERY: Final = b'"LiteLLM_VerificationToken"."budget_reset_at" < $'
+
+
+@contextmanager
+def scratch_database() -> Generator[str]:
+    name: Final = f"integration_{uuid.uuid4().hex}"
+    with psycopg.connect(os.environ["DATABASE_URL"], autocommit=True) as admin:
+        admin.execute(sql.SQL("CREATE DATABASE {}").format(sql.Identifier(name)))
+        try:
+            yield urlunsplit(urlsplit(os.environ["DATABASE_URL"])._replace(path=f"/{name}"))
+        finally:
+            admin.execute(sql.SQL("DROP DATABASE {} WITH (FORCE)").format(sql.Identifier(name)))
+
+
+@pytest.mark.covers("quota_management.budget.key.scheduled_reset_survives_transient_db_outage")
+@pytest.mark.timeout(300)
+def test_scheduled_budget_reset_reconnects_after_db_transport_failure_and_unblocks_key(
+    gateway: Gateway, tmp_path: Path
+) -> None:
+    with (
+        scratch_database() as scratch_url,
+        database_relay(scratch_url, RESET_SWEEP_QUERY) as (relay, relayed_url),
+        httpx.Client(base_url=gateway.upstream_url, timeout=5, trust_env=False) as upstream,
+        owned_proxy(
+            gateway,
+            tmp_path,
+            {
+                "DATABASE_URL": relayed_url,
+                "PROXY_BUDGET_RESCHEDULER_MIN_TIME": "30",
+                "PROXY_BUDGET_RESCHEDULER_MAX_TIME": "30",
+                "PRISMA_HEALTH_WATCHDOG_ENABLED": "false",
+            },
+        ) as candidate,
+    ):
+        model: Final = f"integration-{uuid.uuid4().hex}"
+        candidate.post(
+            "/model/new",
+            {
+                "model_name": model,
+                "litellm_params": {
+                    "model": "openai/gpt-4o-mini",
+                    "api_key": "integration-provider-key",
+                    "api_base": f"{gateway.upstream_url}/v1",
+                    "input_cost_per_token": 0.001,
+                    "output_cost_per_token": 0.002,
+                },
+                "model_info": {},
+            },
+        )
+        key: Final = string_value(
+            candidate.post("/key/generate", {"models": [model], "max_budget": 0.06, "budget_duration": "5s"})["key"]
+        )
+        digest: Final = sha256(key.encode()).hexdigest()
+        row_query: Final = (
+            'SELECT spend, budget_reset_at::text AS budget_reset_at FROM "LiteLLM_VerificationToken" WHERE token=%s'
+        )
+        assert candidate.chat(model, key=key, text=f"spend it {uuid.uuid4().hex}")["usage"]["total_tokens"] == 40
+        exhausted: Final = eventually(
+            lambda: read_rows(row_query, (digest,), database_url=scratch_url),
+            lambda rows: len(rows) == 1 and float(rows[0]["spend"]) >= 0.06,
+            seconds=70,
+        )
+        assert float(exhausted[0]["spend"]) == pytest.approx(0.06)
+        upstream.get("/__observations").raise_for_status()
+        denied: Final = candidate.request(
+            "POST",
+            "/v1/chat/completions",
+            {"model": model, "messages": [{"role": "user", "content": f"over budget {uuid.uuid4().hex}"}]},
+            key=key,
+        )
+        assert denied.status_code == 422 and denied.json()["error"]["type"] == "budget_exceeded", denied.text
+        assert upstream.get("/__observations").json()["requests"] == []
+        relay.arm()
+        assert relay.tripped.wait(90), "Scheduled reset sweep never reached the database"
+        eventually(lambda: relay.refused, lambda count: count >= 1, seconds=30)
+        reset: Final = eventually(
+            lambda: read_rows(row_query, (digest,), database_url=scratch_url),
+            lambda rows: len(rows) == 1 and float(rows[0]["spend"]) == 0,
+            seconds=80,
+            return_last_on_timeout=True,
+        )
+        assert len(reset) == 1 and reset[0]["spend"] == 0.0, (exhausted, reset)
+        assert str(reset[0]["budget_reset_at"]) > str(exhausted[0]["budget_reset_at"]), (exhausted, reset)
+        prompt: Final = f"after reset {uuid.uuid4().hex}"
+        recovered: Final = candidate.request(
+            "POST", "/v1/chat/completions", {"model": model, "messages": [{"role": "user", "content": prompt}]}, key=key
+        )
+        assert recovered.status_code == 200, recovered.text
+        assert recovered.json()["usage"]["total_tokens"] == 40, recovered.text
+        reached: Final = upstream.get("/__observations").json()["requests"]
+        assert len(reached) == 1 and reached[0]["body"]["messages"] == [{"role": "user", "content": prompt}], reached
+
+
@pytest.mark.covers("quota_management.budget.key.count_tokens_reserves_nothing_so_completion_within_budget_succeeds")
 def test_repeated_count_tokens_on_budgeted_key_does_not_reserve_budget_or_block_later_completion(
    gateway: Gateway,
@ -333,6 +434,7 @@ def test_different_system_messages_do_not_share_a_cached_response(gateway: Gatew
    ):
        model: Final = scenario.model()
        prompt: Final = uuid.uuid4().hex
+
        def completion_id(system: str, expected_calls: int) -> str:
            upstream.get("/__observations").raise_for_status()
            response: Final = gateway.request(
--- a/tests/integration/spend/test_end_user_spend_without_proxy_user.py
+++ b/tests/integration/spend/test_end_user_spend_without_proxy_user.py
@ -0,0 +1,29 @@
+import uuid
+from typing import Final
+
+import pytest
+from integration._support.client import Gateway, eventually
+from integration._support.database import read_rows
+
+
+@pytest.mark.covers("spend.end_user.charged_when_key_has_no_user_id_and_auth_cache_is_redis")
+def test_end_user_spend_lands_for_key_without_user_id_when_auth_cache_is_redis(gateway: Gateway) -> None:
+    with gateway.scenario() as scenario:
+        model: Final = scenario.model(input_cost_per_token=0.001, output_cost_per_token=0.002)
+        key: Final = scenario.key(models=[model])
+        end_user: Final = f"integration-end-user-{uuid.uuid4().hex}"
+        scenario.cleanups.callback(gateway.request, "POST", "/customer/delete", {"user_ids": [end_user]})
+        response: Final = gateway.request(
+            "POST",
+            "/v1/chat/completions",
+            {"model": model, "messages": [{"role": "user", "content": f"end user spend {end_user}"}], "user": end_user},
+            key=key,
+        )
+        assert response.status_code == 200, response.text
+        assert response.json()["usage"]["total_tokens"] == 40, response.text
+        charged: Final = eventually(
+            lambda: read_rows('SELECT spend FROM "LiteLLM_EndUserTable" WHERE user_id=%s', (end_user,)),
+            lambda values: len(values) == 1 and float(values[0]["spend"]) >= 0.06,
+            seconds=70,
+        )
+        assert float(charged[0]["spend"]) == pytest.approx(20 * 0.001 + 20 * 0.002)
--- a/tests/integration/spend/test_legacy_spend_logs_row_cap.py
+++ b/tests/integration/spend/test_legacy_spend_logs_row_cap.py
@ -0,0 +1,46 @@
+import os
+import uuid
+from datetime import datetime, timedelta, timezone
+from typing import Final
+
+import psycopg
+import pytest
+from integration._support.client import Gateway
+from integration._support.database import read_rows
+
+LEGACY_SPEND_LOGS_ROW_CAP: Final = 10000
+
+
+def _seed_spend_rows(user_id: str, count: int) -> tuple[str, ...]:
+    started: Final = datetime.now(timezone.utc) - timedelta(days=1)
+    with psycopg.connect(os.environ["DATABASE_URL"], autocommit=True) as connection:
+        connection.execute(
+            'INSERT INTO "LiteLLM_SpendLogs" '
+            '(request_id, call_type, "startTime", "endTime", "user", status) '
+            "SELECT %s || '-' || n, 'acompletion', %s + n * interval '1 second', %s + n * interval '1 second', %s, "
+            "'success' FROM generate_series(1, %s) AS n",
+            (user_id, started, started, user_id, count),
+        )
+    return tuple(f"{user_id}-{n}" for n in range(1, count + 1))
+
+
+def _delete_spend_rows(user_id: str) -> None:
+    with psycopg.connect(os.environ["DATABASE_URL"], autocommit=True) as connection:
+        connection.execute('DELETE FROM "LiteLLM_SpendLogs" WHERE "user" = %s', (user_id,))
+
+
+@pytest.mark.covers("spend.legacy_spend_logs.row_count_is_capped_at_the_most_recent_rows_and_flagged_truncated")
+def test_legacy_spend_logs_returns_only_the_cap_of_most_recent_rows_and_flags_truncation(gateway: Gateway) -> None:
+    user_id: Final = f"integration-cap-{uuid.uuid4().hex}"
+    with gateway.scenario() as scenario:
+        scenario.cleanups.callback(_delete_spend_rows, user_id)
+        seeded: Final = _seed_spend_rows(user_id, LEGACY_SPEND_LOGS_ROW_CAP + 1)
+        assert read_rows('SELECT count(*)::text AS total FROM "LiteLLM_SpendLogs" WHERE "user" = %s', (user_id,)) == [
+            {"total": str(LEGACY_SPEND_LOGS_ROW_CAP + 1)}
+        ]
+        response: Final = gateway.request("GET", "/spend/logs", params={"user_id": user_id})
+        assert response.status_code == 200, response.text
+        returned: Final = tuple(row["request_id"] for row in response.json())
+        assert len(returned) == LEGACY_SPEND_LOGS_ROW_CAP, f"{len(returned)} rows: {response.text[:300]}"
+        assert returned == tuple(reversed(seeded[1:])), response.text[:300]
+        assert response.headers.get("x-litellm-spend-logs-truncated") == "true", dict(response.headers)
--- a/tests/integration/spend/test_messages_stream_usage_cost.py
+++ b/tests/integration/spend/test_messages_stream_usage_cost.py
@ -0,0 +1,176 @@
+import base64
+import json
+import uuid
+from pathlib import Path
+from typing import Final
+
+import pytest
+from integration._support.client import Gateway, eventually
+from integration._support.database import read_rows
+from integration._support.process import owned_proxy
+from integration._support.upstream import _aws_event_frame
+from integration._support.wire import Reply, Request, wire_server
+from pydantic import JsonValue, TypeAdapter
+
+BEDROCK_MODEL: Final = "anthropic.claude-haiku-4-5-20251001-v1:0"
+INPUT_TOKENS: Final = 30
+CACHE_READ_TOKENS: Final = 900
+CACHE_CREATION_TOKENS: Final = 400
+OUTPUT_TOKENS: Final = 57
+INPUT_RATE: Final = 0.001
+OUTPUT_RATE: Final = 0.002
+CACHE_READ_RATE: Final = 0.0001
+CACHE_CREATION_RATE: Final = 0.00125
+STREAMED_USAGE: Final = TypeAdapter(dict[str, float])
+EXPECTED_SPEND: Final = (
+    INPUT_TOKENS * INPUT_RATE
+    + CACHE_READ_TOKENS * CACHE_READ_RATE
+    + CACHE_CREATION_TOKENS * CACHE_CREATION_RATE
+    + OUTPUT_TOKENS * OUTPUT_RATE
+)
+
+
+def _invoke_chunk(payload: dict[str, JsonValue]) -> bytes:
+    encoded: Final = base64.b64encode(json.dumps(payload, separators=(",", ":")).encode()).decode()
+    return _aws_event_frame("chunk", {"bytes": encoded}, "", "")
+
+
+def _stream(message_id: str) -> bytes:
+    return (
+        _invoke_chunk(
+            {
+                "type": "message_start",
+                "message": {
+                    "id": message_id,
+                    "type": "message",
+                    "role": "assistant",
+                    "model": BEDROCK_MODEL,
+                    "content": [],
+                    "stop_reason": None,
+                    "stop_sequence": None,
+                    "usage": {
+                        "input_tokens": INPUT_TOKENS,
+                        "cache_read_input_tokens": CACHE_READ_TOKENS,
+                        "cache_creation_input_tokens": CACHE_CREATION_TOKENS,
+                        "output_tokens": 0,
+                    },
+                },
+            }
+        )
+        + _invoke_chunk({"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}})
+        + _invoke_chunk(
+            {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "cached answer"}}
+        )
+        + _invoke_chunk({"type": "content_block_stop", "index": 0})
+        + _invoke_chunk(
+            {
+                "type": "message_delta",
+                "delta": {"stop_reason": "end_turn"},
+                "usage": {
+                    "input_tokens": INPUT_TOKENS,
+                    "cache_read_input_tokens": CACHE_READ_TOKENS,
+                    "cache_creation_input_tokens": CACHE_CREATION_TOKENS,
+                    "output_tokens": OUTPUT_TOKENS,
+                },
+            }
+        )
+        + _invoke_chunk({"type": "message_stop"})
+    )
+
+
+def _proxy_config(directory: Path, model: str, upstream_url: str) -> Path:
+    config: Final = directory / "streamed_usage_cost_config.yaml"
+    config.write_text(
+        json.dumps(
+            {
+                "model_list": [
+                    {
+                        "model_name": model,
+                        "litellm_params": {
+                            "model": f"bedrock/invoke/{BEDROCK_MODEL}",
+                            "api_base": upstream_url,
+                            "aws_access_key_id": "AKIASCRIPTEDPROVIDER",
+                            "aws_secret_access_key": "scripted-secret",
+                            "aws_region_name": "us-east-1",
+                            "input_cost_per_token": INPUT_RATE,
+                            "output_cost_per_token": OUTPUT_RATE,
+                            "cache_read_input_token_cost": CACHE_READ_RATE,
+                            "cache_creation_input_token_cost": CACHE_CREATION_RATE,
+                        },
+                    }
+                ],
+                "general_settings": {
+                    "master_key": "os.environ/LITELLM_MASTER_KEY",
+                    "database_url": "os.environ/DATABASE_URL",
+                    "store_model_in_db": True,
+                    "disable_spend_logs": False,
+                    "proxy_batch_write_at": 1,
+                    "proxy_batch_polling_interval": 1,
+                },
+                "litellm_settings": {"include_cost_in_streaming_usage": True},
+                "router_settings": {"disable_cooldowns": True},
+            }
+        )
+    )
+    return config
+
+
+def _data_events(body: str) -> tuple[dict[str, JsonValue], ...]:
+    return tuple(json.loads(line.removeprefix("data:")) for line in body.splitlines() if line.startswith("data:"))
+
+
+@pytest.mark.covers("spend.anthropic_messages_stream.streamed_usage_cost_equals_recorded_spend")
+@pytest.mark.timeout(180)
+def test_bedrock_messages_stream_usage_cost_matches_recorded_spend_with_custom_cache_rates(
+    gateway: Gateway, tmp_path: Path
+) -> None:
+    message_id: Final = f"msg_{uuid.uuid4().hex}"
+    model: Final = f"{BEDROCK_MODEL}-{uuid.uuid4().hex}"
+
+    def respond(request: Request) -> Reply:
+        assert request.target == f"/model/{BEDROCK_MODEL}/invoke-with-response-stream", request.target
+        assert json.loads(request.body)["messages"] == [{"role": "user", "content": "cached cost control"}], (
+            request.body
+        )
+        return Reply(content_type="application/vnd.amazon.eventstream", chunks=(_stream(message_id),))
+
+    with wire_server(respond) as wire:
+        config: Final = _proxy_config(tmp_path, model, wire.url)
+        with owned_proxy(gateway, tmp_path, {}, config=config) as candidate:
+            response: Final = candidate.request(
+                "POST",
+                "/v1/messages",
+                {
+                    "model": model,
+                    "messages": [{"role": "user", "content": "cached cost control"}],
+                    "max_tokens": OUTPUT_TOKENS,
+                    "stream": True,
+                },
+            )
+            assert response.status_code == 200, response.text
+            message_delta: Final = next(
+                event for event in _data_events(response.text) if event["type"] == "message_delta"
+            )
+            streamed_usage: Final = STREAMED_USAGE.validate_python(message_delta["usage"])
+            rows: Final = eventually(
+                lambda: read_rows(
+                    'SELECT status, prompt_tokens, completion_tokens, spend FROM "LiteLLM_SpendLogs" '
+                    "WHERE request_id=%s",
+                    (message_id,),
+                ),
+                lambda values: len(values) == 1,
+                seconds=70,
+            )
+        assert rows[0]["status"] == "success", rows
+        assert rows[0]["prompt_tokens"] == INPUT_TOKENS + CACHE_READ_TOKENS + CACHE_CREATION_TOKENS, rows
+        assert rows[0]["completion_tokens"] == OUTPUT_TOKENS, rows
+        recorded_spend: Final = float(str(rows[0]["spend"]))
+        assert recorded_spend == pytest.approx(EXPECTED_SPEND), rows
+        assert streamed_usage == {
+            "input_tokens": INPUT_TOKENS,
+            "cache_read_input_tokens": CACHE_READ_TOKENS,
+            "cache_creation_input_tokens": CACHE_CREATION_TOKENS,
+            "output_tokens": OUTPUT_TOKENS,
+            "cost": pytest.approx(recorded_spend),
+        }, (streamed_usage, rows, response.text)
+        assert len(wire.drain()) == 1
--- a/tests/integration/spend/test_reset_budget_leader_election.py
+++ b/tests/integration/spend/test_reset_budget_leader_election.py
@ -0,0 +1,75 @@
+import json
+import os
+from pathlib import Path
+from types import MappingProxyType
+from typing import Final
+
+import psycopg
+import pytest
+from integration._support.client import Gateway, eventually
+from integration._support.database import read_rows
+from integration._support.process import owned_proxy
+from redis import Redis
+
+RESET_LEASE_KEY: Final = "cronjob_lock:reset_budget_job"
+PEER_POD_LEASE: Final = json.dumps("integration-peer-pod-holding-the-reset-lease")
+FAST_RESET_TICK: Final = MappingProxyType(
+    {"PROXY_BUDGET_RESCHEDULER_MIN_TIME": "2", "PROXY_BUDGET_RESCHEDULER_MAX_TIME": "2"}
+)
+
+
+def _team_spend(team: str) -> float:
+    rows: Final = read_rows('SELECT spend FROM "LiteLLM_TeamTable" WHERE team_id = %s', (team,))
+    assert len(rows) == 1, rows
+    spend: Final = rows[0]["spend"]
+    assert isinstance(spend, (int, float)), rows
+    return float(spend)
+
+
+def _make_team_budget_due(team: str, spend: float) -> None:
+    with psycopg.connect(os.environ["DATABASE_URL"]) as connection:
+        connection.execute(
+            "UPDATE \"LiteLLM_TeamTable\" SET spend = %s, budget_reset_at = now() - interval '1 day' "
+            "WHERE team_id = %s",
+            (spend, team),
+        )
+
+
+@pytest.mark.covers("spend.budget_reset.one_pod_sweeps_per_tick")
+def test_reset_sweep_skips_ticks_while_another_pod_holds_the_lease_and_resumes_after_release(
+    gateway: Gateway, tmp_path: Path
+) -> None:
+    with (
+        gateway.scenario() as scenario,
+        Redis(host=os.environ["REDIS_HOST"], port=int(os.environ["REDIS_PORT"])) as cache,
+    ):
+        model: Final = scenario.model(input_cost_per_token=0.0, output_cost_per_token=0.0)
+        team: Final = scenario.team(max_budget=1.0, budget_duration="30d")
+        key: Final = scenario.key(team_id=team)
+        _make_team_budget_due(team, spend=0.5)
+        assert _team_spend(team) == 0.5
+        eventually(
+            lambda: cache.set(RESET_LEASE_KEY, PEER_POD_LEASE, ex=120, nx=True),
+            lambda claimed: claimed is True,
+            seconds=60,
+        )
+        try:
+            with owned_proxy(gateway, tmp_path, FAST_RESET_TICK) as replica:
+                response: Final = replica.request(
+                    "POST",
+                    "/v1/chat/completions",
+                    {"model": model, "messages": [{"role": "user", "content": "classification request"}]},
+                    key=key,
+                )
+                assert response.status_code == 200, response.text
+                held: Final = eventually(
+                    lambda: _team_spend(team), lambda spend: spend != 0.5, seconds=8, return_last_on_timeout=True
+                )
+                assert held == 0.5, f"team {team} was swept while another pod held the reset lease: spend={held}"
+                assert cache.get(RESET_LEASE_KEY) == PEER_POD_LEASE.encode()
+                cache.delete(RESET_LEASE_KEY)
+                swept: Final = eventually(lambda: _team_spend(team), lambda spend: spend == 0.0, seconds=15)
+                assert swept == 0.0
+                eventually(lambda: cache.get(RESET_LEASE_KEY), lambda value: value is None, seconds=15)
+        finally:
+            cache.delete(RESET_LEASE_KEY)
--- a/tests/integration/spend/test_responses_cache_write_itemization.py
+++ b/tests/integration/spend/test_responses_cache_write_itemization.py
@ -0,0 +1,98 @@
+import json
+import uuid
+from typing import Final
+
+import pytest
+
+from tests.integration._support.client import Gateway, eventually, object_value, string_value
+from tests.integration._support.database import read_rows
+from tests.integration._support.upstream import delete_scenario, register_scenario
+from tests.integration.cost_calculation.cost_tracking_case import JsonResponse
+
+INPUT_RATE: Final = 0.001
+OUTPUT_RATE: Final = 0.002
+CACHE_CREATION_RATE: Final = 0.004
+CACHE_READ_RATE: Final = 0.0001
+UNCACHED_INPUT_TOKENS: Final = 1000
+CACHE_WRITE_TOKENS: Final = 2000
+CACHED_TOKENS: Final = 8000
+INPUT_TOKENS: Final = UNCACHED_INPUT_TOKENS + CACHE_WRITE_TOKENS + CACHED_TOKENS
+OUTPUT_TOKENS: Final = 500
+
+
+def responses_cache_write_response() -> JsonResponse:
+    return JsonResponse(
+        content_type="application/json",
+        body={
+            "id": "resp_$REQUEST_ID",
+            "object": "response",
+            "created_at": 1700000000,
+            "status": "completed",
+            "model": "gpt-5.6",
+            "output": [
+                {
+                    "type": "message",
+                    "id": "msg_$REQUEST_ID",
+                    "status": "completed",
+                    "role": "assistant",
+                    "content": [{"type": "output_text", "text": "scripted response", "annotations": []}],
+                }
+            ],
+            "usage": {
+                "input_tokens": INPUT_TOKENS,
+                "output_tokens": OUTPUT_TOKENS,
+                "total_tokens": INPUT_TOKENS + OUTPUT_TOKENS,
+                "input_tokens_details": {
+                    "cached_tokens": CACHED_TOKENS,
+                    "cache_write_tokens": CACHE_WRITE_TOKENS,
+                },
+                "output_tokens_details": {"reasoning_tokens": 0},
+            },
+        },
+    )
+
+
+@pytest.mark.covers("spend.responses_api.cache_write_tokens_itemized_as_cache_creation_cost")
+def test_responses_cache_write_tokens_are_itemized_as_cache_creation_cost(gateway: Gateway) -> None:
+    with gateway.scenario() as scenario:
+        scenario_id: Final = f"responses-cache-write-{uuid.uuid4().hex[:12]}"
+        handle: Final = register_scenario(scenario_id, responses_cache_write_response())
+        scenario.cleanups.callback(delete_scenario, handle)
+        model: Final = scenario.model(
+            model="openai/gpt-5.6",
+            api_base=handle.api_base(),
+            input_cost_per_token=INPUT_RATE,
+            output_cost_per_token=OUTPUT_RATE,
+            cache_creation_input_token_cost=CACHE_CREATION_RATE,
+            cache_read_input_token_cost=CACHE_READ_RATE,
+        )
+        response: Final = gateway.request("POST", "/v1/responses", {"model": model, "input": "cache write control"})
+        assert response.status_code == 200, response.text
+        expected_cache_creation_cost: Final = CACHE_WRITE_TOKENS * CACHE_CREATION_RATE
+        expected_cache_read_cost: Final = CACHED_TOKENS * CACHE_READ_RATE
+        expected_input_cost: Final = (
+            UNCACHED_INPUT_TOKENS * INPUT_RATE + expected_cache_creation_cost + expected_cache_read_cost
+        )
+        expected_output_cost: Final = OUTPUT_TOKENS * OUTPUT_RATE
+        request_id: Final = string_value(object_value(response.json())["id"])
+        rows: Final = eventually(
+            lambda: read_rows(
+                'SELECT spend, metadata, prompt_tokens, completion_tokens FROM "LiteLLM_SpendLogs" '
+                "WHERE request_id = %s",
+                (request_id,),
+            ),
+            lambda values: len(values) == 1,
+            seconds=70,
+        )
+        assert rows[0]["prompt_tokens"] == INPUT_TOKENS, response.text
+        assert rows[0]["completion_tokens"] == OUTPUT_TOKENS, response.text
+        assert float(rows[0]["spend"]) == pytest.approx(expected_input_cost + expected_output_cost, rel=1e-6), (
+            response.text
+        )
+        metadata: Final = rows[0]["metadata"]
+        parsed: Final = json.loads(metadata) if isinstance(metadata, str) else object_value(metadata)
+        breakdown: Final = object_value(parsed["cost_breakdown"])
+        assert breakdown.get("cache_creation_cost") == pytest.approx(expected_cache_creation_cost, rel=1e-6), breakdown
+        assert breakdown.get("cache_read_cost") == pytest.approx(expected_cache_read_cost, rel=1e-6), breakdown
+        assert float(breakdown["input_cost"]) == pytest.approx(expected_input_cost, rel=1e-6), breakdown
+        assert float(breakdown["output_cost"]) == pytest.approx(expected_output_cost, rel=1e-6), breakdown
--- a/tests/integration/spend/test_session_total_spend.py
+++ b/tests/integration/spend/test_session_total_spend.py
@ -0,0 +1,98 @@
+import json
+import uuid
+from datetime import datetime, timedelta, timezone
+from typing import Final
+
+import pytest
+from integration._support.client import Gateway, eventually
+from integration._support.database import read_rows
+from integration._support.wire import Reply, Request, wire_server
+
+INPUT_COST_PER_TOKEN: Final = 0.001
+OUTPUT_COST_PER_TOKEN: Final = 0.002
+ROUND_USAGE: Final = ((10, 5), (20, 10), (30, 15))
+ROUND_SPEND: Final = tuple(
+    prompt * INPUT_COST_PER_TOKEN + completion * OUTPUT_COST_PER_TOKEN for prompt, completion in ROUND_USAGE
+)
+SESSION_SPEND: Final = sum(ROUND_SPEND)
+
+
+def _round_reply(request: Request, prompt: str, usage: tuple[int, int]) -> Reply:
+    assert request.method == "POST" and request.target == "/chat/completions", request.target
+    assert json.loads(request.body) == {
+        "model": "gpt-4o-mini",
+        "messages": [{"role": "user", "content": prompt}],
+    }, request.body
+    return Reply(
+        body=json.dumps(
+            {
+                "id": "chatcmpl-" + uuid.uuid4().hex,
+                "object": "chat.completion",
+                "created": 1,
+                "model": "gpt-4o-mini",
+                "choices": [
+                    {"index": 0, "message": {"role": "assistant", "content": "round answer"}, "finish_reason": "stop"}
+                ],
+                "usage": {"prompt_tokens": usage[0], "completion_tokens": usage[1], "total_tokens": sum(usage)},
+            }
+        ).encode()
+    )
+
+
+@pytest.mark.covers("spend.logs_ui.multi_round_session_total_spend_sums_every_round")
+def test_logs_ui_session_total_spend_sums_every_round_of_a_multi_round_session(gateway: Gateway) -> None:
+    session_id: Final = f"session-{uuid.uuid4().hex}"
+    prompts: Final = tuple(f"round-{index}-{uuid.uuid4().hex}" for index in range(len(ROUND_USAGE)))
+    usage_by_prompt: Final = dict(zip(prompts, ROUND_USAGE, strict=True))
+
+    def provider(request: Request) -> Reply:
+        prompt: Final = str(json.loads(request.body)["messages"][0]["content"])
+        return _round_reply(request, prompt, usage_by_prompt[prompt])
+
+    with wire_server(provider) as wire, gateway.scenario() as scenario:
+        alias: Final = scenario.model(
+            api_base=wire.url,
+            input_cost_per_token=INPUT_COST_PER_TOKEN,
+            output_cost_per_token=OUTPUT_COST_PER_TOKEN,
+            num_retries=0,
+        )
+        request_ids: Final = tuple(_completed_round(gateway, alias, session_id, prompt) for prompt in prompts)
+        assert len(wire.drain()) == len(ROUND_USAGE)
+        rows: Final = eventually(
+            lambda: read_rows(
+                'SELECT request_id, spend FROM "LiteLLM_SpendLogs" WHERE session_id=%s ORDER BY "startTime"',
+                (session_id,),
+            ),
+            lambda values: len(values) == len(ROUND_USAGE),
+            seconds=70,
+        )
+        assert [row["request_id"] for row in rows] == list(request_ids), rows
+        assert [float(row["spend"]) for row in rows] == pytest.approx(list(ROUND_SPEND)), rows
+        now: Final = datetime.now(timezone.utc)
+        logs: Final = gateway.request(
+            "GET",
+            "/spend/logs/ui",
+            params={
+                "session_id": session_id,
+                "start_date": (now - timedelta(days=1)).strftime("%Y-%m-%d %H:%M:%S"),
+                "end_date": (now + timedelta(days=1)).strftime("%Y-%m-%d %H:%M:%S"),
+            },
+        )
+        assert logs.status_code == 200, logs.text
+        page: Final = logs.json()
+        assert page["total"] == len(ROUND_USAGE), logs.text
+        assert sorted(row["request_id"] for row in page["data"]) == sorted(request_ids), logs.text
+        assert [row["session_total_count"] for row in page["data"]] == [len(ROUND_USAGE)] * len(ROUND_USAGE), logs.text
+        assert [row["session_total_spend"] for row in page["data"]] == pytest.approx(
+            [SESSION_SPEND] * len(ROUND_USAGE)
+        ), logs.text
+
+
+def _completed_round(gateway: Gateway, alias: str, session_id: str, prompt: str) -> str:
+    response: Final = gateway.request(
+        "POST",
+        "/v1/chat/completions",
+        {"model": alias, "messages": [{"role": "user", "content": prompt}], "litellm_trace_id": session_id},
+    )
+    assert response.status_code == 200, response.text
+    return str(response.json()["id"])
--- a/tests/integration/spend/test_spend_log_write_batching.py
+++ b/tests/integration/spend/test_spend_log_write_batching.py
@ -0,0 +1,75 @@
+import uuid
+from pathlib import Path
+from typing import Final
+
+import pytest
+import yaml
+from integration._support.client import Gateway, eventually, string_value
+from integration._support.database import read_rows
+from integration._support.process import owned_proxy
+from pydantic import JsonValue
+
+WRITE_STATEMENT_MAX_BYTES: Final = 200_000
+MESSAGES_PER_REQUEST: Final = 60
+MESSAGE_CHARACTERS: Final = 2_000
+REQUESTS: Final = 4
+
+
+def _config_storing_prompts(tmp_path: Path) -> Path:
+    config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text())
+    config["general_settings"]["store_prompts_in_spend_logs"] = True
+    path: Final = tmp_path / "store-prompts.yaml"
+    path.write_text(yaml.safe_dump(config))
+    return path
+
+
+def _prompt_messages(marker: str) -> list[JsonValue]:
+    return [
+        {"role": "user", "content": f"{marker}-{index}-".ljust(MESSAGE_CHARACTERS, "x")}
+        for index in range(MESSAGES_PER_REQUEST)
+    ]
+
+
+def _persisted(request_ids: tuple[str, ...]) -> list[dict[str, JsonValue]]:
+    placeholders: Final = ", ".join("%s" for _ in request_ids)
+    return read_rows(
+        "SELECT request_id, xmin::text AS statement, octet_length(proxy_server_request::text) AS stored_bytes "
+        f'FROM "LiteLLM_SpendLogs" WHERE request_id IN ({placeholders}) ORDER BY request_id',
+        request_ids,
+    )
+
+
+@pytest.mark.covers("quota_management.spend_tracking.prompt_rows_are_written_in_byte_bounded_statements")
+def test_prompt_carrying_spend_rows_flushed_together_are_written_in_byte_bounded_statements(
+    gateway: Gateway, tmp_path: Path
+) -> None:
+    with (
+        gateway.scenario() as scenario,
+        owned_proxy(
+            gateway,
+            tmp_path,
+            {
+                "SPEND_LOG_WRITE_BATCH_MAX_BYTES": str(WRITE_STATEMENT_MAX_BYTES),
+                "SPEND_LOG_QUEUE_POLL_INTERVAL": "15",
+            },
+            config=_config_storing_prompts(tmp_path),
+        ) as owned,
+    ):
+        model: Final = scenario.model()
+        request_ids: Final = tuple(
+            string_value(
+                owned.post(
+                    "/v1/chat/completions",
+                    {"model": model, "messages": _prompt_messages(f"integration-prompt-{uuid.uuid4().hex}")},
+                )["id"]
+            )
+            for _ in range(REQUESTS)
+        )
+        assert len(set(request_ids)) == REQUESTS, request_ids
+        rows: Final = eventually(lambda: _persisted(request_ids), lambda values: len(values) == REQUESTS, seconds=70)
+    stored_bytes: Final = tuple(row["stored_bytes"] for row in rows)
+    assert all(
+        isinstance(size, int) and WRITE_STATEMENT_MAX_BYTES // 2 < size < WRITE_STATEMENT_MAX_BYTES
+        for size in stored_bytes
+    ), rows
+    assert len({row["statement"] for row in rows}) == REQUESTS, rows
--- a/tests/integration/spend/test_team_member_spend.py
+++ b/tests/integration/spend/test_team_member_spend.py
@ -1,9 +1,12 @@
+import os
 import uuid
 from typing import Final

+import httpx
 import pytest
 from integration._support.client import Gateway, eventually, object_value
 from integration._support.database import read_rows
+from redis import Redis


@pytest.mark.covers("spend.team_member.member_without_budget_gets_membership_row_and_spend")
@ -52,3 +55,45 @@ def test_member_added_without_any_budget_is_charged_on_its_membership_row(gatewa
            if object_value(row)["user_id"] == user
        ]
        assert len(exposed) == 1 and exposed[0][1] == pytest.approx(0.06), info
+
+
+@pytest.mark.covers("spend.team_member.stale_low_redis_counter_still_blocks_member_over_budget")
+def test_member_over_budget_is_blocked_when_redis_counter_reads_stale_low(gateway: Gateway) -> None:
+    with (
+        gateway.scenario() as scenario,
+        httpx.Client(base_url=gateway.upstream_url, timeout=5, trust_env=False) as upstream,
+        Redis(host=os.environ["REDIS_HOST"], port=int(os.environ["REDIS_PORT"])) as cache,
+    ):
+        model: Final = scenario.model(input_cost_per_token=0.001, output_cost_per_token=0.002)
+        team: Final = scenario.team(models=[model])
+        user: Final = scenario.user()
+        added: Final = gateway.request(
+            "POST",
+            "/team/member_add",
+            {"team_id": team, "member": {"user_id": user, "role": "user"}, "max_budget_in_team": 0.05},
+        )
+        assert added.status_code == 200, added.text
+        key: Final = scenario.key(team_id=team, user_id=user, models=[model])
+        assert gateway.chat(model, key=key, text=f"member budget {uuid.uuid4().hex}")["usage"]["total_tokens"] == 40
+        charged: Final = eventually(
+            lambda: read_rows(
+                'SELECT spend FROM "LiteLLM_TeamMembership" WHERE team_id=%s AND user_id=%s', (team, user)
+            ),
+            lambda values: len(values) == 1 and float(values[0]["spend"]) >= 0.06,
+            seconds=70,
+        )
+        assert float(charged[0]["spend"]) == pytest.approx(0.06)
+        counter_key: Final = f"spend:team_member:{user}:{team}"
+        counted: Final = eventually(lambda: cache.get(counter_key), lambda value: value is not None, seconds=10)
+        assert float(counted) == pytest.approx(0.06), counted
+        cache.set(counter_key, "0.01")
+        upstream.get("/__observations").raise_for_status()
+        denied: Final = gateway.request(
+            "POST",
+            "/v1/chat/completions",
+            {"model": model, "messages": [{"role": "user", "content": f"stale counter {uuid.uuid4().hex}"}]},
+            key=key,
+        )
+        assert denied.status_code == 422 and denied.json()["error"]["type"] == "budget_exceeded", denied.text
+        assert upstream.get("/__observations").json()["requests"] == []
+        assert float(cache.get(counter_key)) == pytest.approx(0.06), denied.text
--- a/tests/integration/spend/test_user_budget_on_team_keys.py
+++ b/tests/integration/spend/test_user_budget_on_team_keys.py
@ -0,0 +1,54 @@
+import uuid
+from hashlib import sha256
+from pathlib import Path
+from typing import Final
+
+import httpx
+import pytest
+import yaml
+from integration._support.client import Gateway, eventually
+from integration._support.database import read_rows
+from integration._support.process import owned_proxy
+
+
+@pytest.mark.covers("spend.user_budget.opted_in_team_key_is_denied_once_owner_budget_is_exhausted")
+def test_team_key_is_denied_before_provider_once_owner_personal_budget_is_exhausted_when_opted_in(
+    gateway: Gateway, tmp_path: Path
+) -> None:
+    configuration: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text())
+    configuration["general_settings"]["apply_user_budget_to_team_keys"] = True
+    path: Final = tmp_path / "apply-user-budget-to-team-keys.yaml"
+    path.write_text(yaml.safe_dump(configuration))
+    with (
+        owned_proxy(gateway, tmp_path, {}, config=path) as candidate,
+        candidate.scenario() as scenario,
+        httpx.Client(base_url=candidate.upstream_url, timeout=5, trust_env=False) as upstream,
+    ):
+        model: Final = scenario.model(input_cost_per_token=0.001, output_cost_per_token=0.002)
+        user: Final = scenario.user(max_budget=0.06)
+        team: Final = scenario.team(models=[model])
+        added: Final = candidate.request(
+            "POST", "/team/member_add", {"team_id": team, "member": {"user_id": user, "role": "user"}}
+        )
+        assert added.status_code == 200, added.text
+        key: Final = scenario.key(team_id=team, user_id=user, models=[model])
+        first: Final = candidate.chat(model, key=key, text=f"owner budget {uuid.uuid4().hex}")
+        assert first["usage"]["total_tokens"] == 40
+        spent: Final = eventually(
+            lambda: read_rows('SELECT spend FROM "LiteLLM_UserTable" WHERE user_id=%s', (user,)),
+            lambda values: len(values) == 1 and float(values[0]["spend"]) >= 0.06,
+            seconds=70,
+        )
+        assert float(spent[0]["spend"]) == pytest.approx(0.06)
+        assert read_rows(
+            'SELECT max_budget FROM "LiteLLM_VerificationToken" WHERE token=%s', (sha256(key.encode()).hexdigest(),)
+        ) == [{"max_budget": None}]
+        upstream.get("/__observations").raise_for_status()
+        denied: Final = candidate.request(
+            "POST",
+            "/v1/chat/completions",
+            {"model": model, "messages": [{"role": "user", "content": f"over owner budget {uuid.uuid4().hex}"}]},
+            key=key,
+        )
+        assert denied.status_code == 422 and denied.json()["error"]["type"] == "budget_exceeded", denied.text
+        assert upstream.get("/__observations").json()["requests"] == []