diff --git a/.circleci/config.yml b/.circleci/config.yml
index 84d8f48b4be..7c241af5853 100644
--- a/.circleci/config.yml
+++ b/.circleci/config.yml
@@ -147,6 +147,9 @@ commands:
       db_name:
         type: string
         default: circle_test
+      image:
+        type: string
+        default: postgres:14@sha256:6a70deda415ec296f977890e11aba04a0db9f632a362e3fce45e845e3db74f26
     steps:
       - run:
           name: Start PostgreSQL
@@ -157,7 +160,7 @@ commands:
               -e POSTGRES_PASSWORD=postgres \
               -e POSTGRES_DB=<< parameters.db_name >> \
               -p 5432:5432 \
-              postgres:14@sha256:6a70deda415ec296f977890e11aba04a0db9f632a362e3fce45e845e3db74f26
+              << parameters.image >>
       - wait_for_service:
           url: tcp://localhost:5432
           timeout: "60"
@@ -2912,7 +2915,50 @@ jobs:
               exit 1
             fi
 
+  integration_contracts:
+    parameters:
+      suite:
+        type: string
+    machine:
+      image: ubuntu-2204:2024.04.1
+    resource_class: large
+    working_directory: ~/project
+    steps:
+      - setup_litellm_test_deps
+      - start_postgres:
+          image: postgres:16@sha256:e17e86066e5ef83e0952a9347f5c792b7ece00972e2aa787a6986f471b3dd3d5
+      - start_redis
+      - run:
+          name: Run owned integration contracts
+          command: bash .circleci/scripts/run_integration.sh << parameters.suite >>
+          no_output_timeout: 15m
+      - run:
+          name: Stop owned database and Redis
+          when: always
+          command: |
+            mkdir -p test-results/integration-<< parameters.suite >>
+            docker logs postgres-db > test-results/integration-<< parameters.suite >>/postgres.log 2>&1 || true
+            docker logs redis-cache > test-results/integration-<< parameters.suite >>/redis.log 2>&1 || true
+            docker rm -f postgres-db redis-cache
+            test -z "$(docker ps -aq --filter name=postgres-db --filter name=redis-cache)"
+      - store_test_results:
+          path: test-results
+      - store_artifacts:
+          path: test-results
+
 workflows:
+  integration:
+    jobs:
+      - integration_contracts:
+          name: integration-<< matrix.suite >>
+          matrix:
+            parameters:
+              suite: [management, accounting, providers]
+          filters:
+            branches:
+              only:
+                - main
+                - /litellm_.*/
   build_and_test:
     jobs:
       - using_litellm_on_windows:
diff --git a/.circleci/scripts/run_integration.sh b/.circleci/scripts/run_integration.sh
new file mode 100644
index 00000000000..57863c62a52
--- /dev/null
+++ b/.circleci/scripts/run_integration.sh
@@ -0,0 +1,145 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+suite="${1:?integration suite required}"
+results="test-results/integration-${suite}"
+mkdir -p "$results"
+integration_identity="$(.venv/bin/python -c 'import uuid; print(uuid.uuid4().hex)')"
+upstream_pid=""
+proxy_pid=""
+guard_created=false
+guard_installed=false
+guard6_created=false
+guard6_installed=false
+cleanup() {
+  original_status=$?
+  trap - EXIT INT TERM
+  .venv/bin/python .circleci/scripts/stop_integration_processes.py "$integration_identity" \
+    > "$results/process-cleanup.txt" 2>&1 || original_status=1
+  for owned_pid in "$proxy_pid" "$upstream_pid"; do
+    if [ -n "$owned_pid" ]; then
+      kill -- "-$owned_pid" 2>/dev/null || true
+      for _ in {1..50}; do
+        kill -0 -- "-$owned_pid" 2>/dev/null || break
+        sleep 0.1
+      done
+      if kill -0 -- "-$owned_pid" 2>/dev/null; then
+        kill -KILL -- "-$owned_pid" 2>/dev/null || true
+        original_status=1
+      fi
+      wait "$owned_pid" 2>/dev/null || true
+    fi
+  done
+  if [ "$guard_installed" = true ]; then
+    sudo iptables -D OUTPUT -m owner --uid-owner "$(id -u)" -j integration_only || original_status=1
+  fi
+  if [ "$guard_created" = true ]; then
+    sudo iptables -F integration_only || original_status=1
+    sudo iptables -X integration_only || original_status=1
+  fi
+  if [ "$guard6_installed" = true ]; then
+    sudo ip6tables -D OUTPUT -m owner --uid-owner "$(id -u)" -j integration_only || original_status=1
+  fi
+  if [ "$guard6_created" = true ]; then
+    sudo ip6tables -F integration_only || original_status=1
+    sudo ip6tables -X integration_only || original_status=1
+  fi
+  printf '%s\n' "$original_status" > "$results/exit-status.txt"
+  exit "$original_status"
+}
+trap cleanup EXIT
+trap 'exit 130' INT
+trap 'exit 143' TERM
+
+export PATH="$PWD/.venv/bin:$PATH"
+export PYTHONPATH="$PWD:$PWD/tests:$PWD/tests/e2e"
+export DATABASE_URL="postgresql://postgres:postgres@127.0.0.1:5432/circle_test"
+export REDIS_HOST=127.0.0.1 REDIS_PORT=6379
+export LITELLM_MASTER_KEY=sk-integration-master LITELLM_SALT_KEY=sk-integration-salt
+export LITELLM_MODE=PRODUCTION LITELLM_LOCAL_MODEL_COST_MAP=True
+export STORE_MODEL_IN_DB=True AWS_EC2_METADATA_DISABLED=true DO_NOT_TRACK=1
+export INTEGRATION_PROXY_URL=http://127.0.0.1:4000
+export INTEGRATION_UPSTREAM_URL=http://127.0.0.1:8190
+export INTEGRATION_MASTER_KEY="$LITELLM_MASTER_KEY"
+
+uv run --no-sync prisma generate --schema litellm/proxy/schema.prisma > "$results/prisma-generate.log" 2>&1
+
+sudo iptables -N integration_only
+guard_created=true
+sudo iptables -A integration_only -o lo -j ACCEPT
+sudo iptables -A integration_only -m conntrack --ctstate ESTABLISHED,RELATED -j ACCEPT
+for service in postgres-db redis-cache; do
+  address="$(docker inspect --format '{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}' "$service")"
+  sudo iptables -A integration_only -d "$address" -j ACCEPT
+done
+sudo iptables -A integration_only -j REJECT
+sudo iptables -I OUTPUT 1 -m owner --uid-owner "$(id -u)" -j integration_only
+guard_installed=true
+sudo ip6tables -N integration_only
+guard6_created=true
+sudo ip6tables -A integration_only -o lo -j ACCEPT
+sudo ip6tables -A integration_only -j REJECT
+sudo ip6tables -I OUTPUT 1 -m owner --uid-owner "$(id -u)" -j integration_only
+guard6_installed=true
+
+if curl --noproxy '*' --connect-timeout 2 -s http://198.51.100.1 >/dev/null 2>&1; then
+  echo "Unexpected outbound network access" >&2
+  exit 1
+fi
+sudo iptables -L integration_only -n -v -x > "$results/egress-guard.txt"
+awk '$3 == "REJECT" && $1 > 0 { rejected=1 } END { exit !rejected }' "$results/egress-guard.txt"
+
+setsid env -i PATH="$PATH" HOME="$HOME" PYTHONPATH="$PYTHONPATH" INTEGRATION_RUN_ID="$integration_identity" \
+  .venv/bin/python -m integration._support.upstream > "$results/upstream.log" 2>&1 &
+upstream_pid=$!
+setsid env -i PATH="$PATH" HOME="$HOME" PYTHONPATH="$PYTHONPATH" INTEGRATION_RUN_ID="$integration_identity" \
+  DATABASE_URL="$DATABASE_URL" REDIS_HOST="$REDIS_HOST" REDIS_PORT="$REDIS_PORT" \
+  LITELLM_MASTER_KEY="$LITELLM_MASTER_KEY" LITELLM_SALT_KEY="$LITELLM_SALT_KEY" \
+  LITELLM_MODE=PRODUCTION LITELLM_LOCAL_MODEL_COST_MAP=True STORE_MODEL_IN_DB=True \
+  AWS_EC2_METADATA_DISABLED=true DO_NOT_TRACK=1 \
+  .venv/bin/litellm --config tests/integration/proxy_config.yaml --host 127.0.0.1 --port 4000 --num_workers 1 --telemetry False \
+  --use_prisma_db_push --enforce_prisma_migration_check \
+  > "$results/proxy.log" 2>&1 &
+proxy_pid=$!
+
+.venv/bin/python - <<'PY'
+import time
+import httpx
+
+deadline = time.monotonic() + 90
+with httpx.Client(trust_env=False, timeout=2) as client:
+    while True:
+        try:
+            provider = client.get("http://127.0.0.1:8190/health")
+            proxy = client.get("http://127.0.0.1:4000/health/readiness")
+            if provider.status_code == proxy.status_code == 200:
+                cache = client.get("http://127.0.0.1:4000/cache/ping", headers={"Authorization": "Bearer sk-integration-master"})
+                cache.raise_for_status()
+                assert cache.json()["status"] == "healthy", cache.text
+                assert cache.json()["cache_type"] == "redis", cache.text
+                assert cache.json()["ping_response"] is True, cache.text
+                assert cache.json()["set_cache_response"] == "success", cache.text
+                break
+        except httpx.TransportError:
+            pass
+        if time.monotonic() >= deadline:
+            raise SystemExit("Integration services did not become ready")
+        time.sleep(0.2)
+PY
+
+if [ "$suite" = providers ]; then
+  INTEGRATION_RUN_ID="$integration_identity" .venv/bin/python -m pytest --noconftest -p no:pytest-retry -p no:rerunfailures --timeout=30 \
+    tests/e2e/test_provider_edge.py::TestReplayMode::test_content_drift_returns_the_miss_status_naming_both_keys \
+    tests/e2e/test_provider_edge.py::TestReplayMode::test_exhausted_key_returns_the_miss_status \
+    tests/e2e/test_provider_edge.py::TestReplayLeftover::test_partially_consumed_recording_names_the_leftover \
+    tests/e2e/test_provider_edge.py::TestStreamingFidelity::test_replay_of_a_stream_makes_no_provider_connection \
+    --junitxml="$results/replay-controls.xml"
+fi
+
+timeout --signal=TERM --kill-after=20s 11m env -i PATH="$PATH" HOME="$HOME" PYTHONPATH="$PYTHONPATH" \
+  INTEGRATION_RUN_ID="$integration_identity" \
+  DATABASE_URL="$DATABASE_URL" REDIS_HOST="$REDIS_HOST" REDIS_PORT="$REDIS_PORT" \
+  INTEGRATION_PROXY_URL="$INTEGRATION_PROXY_URL" INTEGRATION_UPSTREAM_URL="$INTEGRATION_UPSTREAM_URL" \
+  INTEGRATION_MASTER_KEY="$INTEGRATION_MASTER_KEY" LITELLM_MODE=PRODUCTION \
+  LITELLM_LOCAL_MODEL_COST_MAP=True AWS_EC2_METADATA_DISABLED=true DO_NOT_TRACK=1 \
+  .venv/bin/python tests/integration/run.py "$suite" --results "$results"
diff --git a/.circleci/scripts/stop_integration_processes.py b/.circleci/scripts/stop_integration_processes.py
new file mode 100644
index 00000000000..ff4d53b9aaa
--- /dev/null
+++ b/.circleci/scripts/stop_integration_processes.py
@@ -0,0 +1,42 @@
+import os
+import sys
+from typing import Final
+
+import psutil
+
+
+def owned_processes(identity: str) -> tuple[psutil.Process, ...]:
+    owned: Final[list[psutil.Process]] = []
+    for process in psutil.process_iter(["uids"]):
+        if process.info["uids"].real != os.getuid():
+            continue
+        try:
+            if process.environ().get("INTEGRATION_RUN_ID") == identity:
+                owned.append(process)
+        except psutil.NoSuchProcess:
+            continue
+    return tuple(owned)
+
+
+def main(identity: str) -> int:
+    owned: Final = owned_processes(identity)
+    for process in owned:
+        try:
+            process.terminate()
+        except psutil.NoSuchProcess:
+            continue
+    psutil.wait_procs(owned, timeout=8)
+    remaining: Final = owned_processes(identity)
+    for process in remaining:
+        try:
+            process.kill()
+        except psutil.NoSuchProcess:
+            continue
+    psutil.wait_procs(remaining, timeout=2)
+    survivors: Final = owned_processes(identity)
+    print(f"Owned integration processes: {len(owned)}, forced: {len(remaining)}, remaining: {len(survivors)}")
+    return 1 if remaining or survivors else 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main(sys.argv[1]))
diff --git a/.github/scripts/assert_ci_coverage.py b/.github/scripts/assert_ci_coverage.py
index 411852acb98..4c66ab251de 100644
--- a/.github/scripts/assert_ci_coverage.py
+++ b/.github/scripts/assert_ci_coverage.py
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import ast
+import json
 import operator
 import pathlib
 import re
@@ -498,6 +499,69 @@ def _check_shards() -> int:
     return 0
 
 
+def _integration_ownership(repo_root: pathlib.Path = REPO_ROOT) -> tuple[frozenset[str], tuple[Finding, ...]]:
+    manifest: Final = repo_root / "tests/integration/contracts.json"
+    if not manifest.exists():
+        return frozenset(), ()
+    entries: Final = json.loads(manifest.read_text())
+    paths: Final = frozenset(node.split("::", 1)[0] for node in entries["tests"])
+    circle_path: Final = repo_root / ".circleci/config.yml"
+    circle: Final = yaml.safe_load(circle_path.read_text()) if circle_path.exists() else {}
+    steps: Final = circle.get("jobs", {}).get("integration_contracts", {}).get("steps", ())
+    invoked: Final = any(
+        ".circleci/scripts/run_integration.sh" in scalar.value
+        for scalar in _scalars(steps, "integration_contracts")
+        if scalar.key == "command"
+    )
+    scheduled: Final = frozenset(
+        suite
+        for job in circle.get("workflows", {}).get("integration", {}).get("jobs", ())
+        if isinstance(job, dict) and "integration_contracts" in job
+        for suite in job["integration_contracts"]
+        .get("matrix", {})
+        .get("parameters", {})
+        .get("suite", (job["integration_contracts"].get("suite"),))
+        if isinstance(suite, str)
+    )
+    required: Final = frozenset(
+        group
+        for group, folders in entries["groups"].items()
+        if any(any(path.startswith(f"tests/integration/{folder}/") for folder in folders) for path in paths)
+    )
+    ungrouped: Final = frozenset(
+        path
+        for path in paths
+        if sum(
+            any(path.startswith(f"tests/integration/{folder}/") for folder in folders)
+            for folders in entries["groups"].values()
+        )
+        != 1
+    )
+    gha_tokens: Final = _invoked_test_tokens(
+        scalar
+        for path in (repo_root / ".github/workflows").glob("*.y*ml")
+        for scalar in _scalars(yaml.safe_load(path.read_text()), path.name)
+    )
+    findings: Final = tuple(
+        Finding(path, "integration contract is also selected by GitHub Actions")
+        for path in paths
+        if any(_token_covers(token, path) for token in gha_tokens)
+    ) + tuple(
+        Finding(path, "canonical integration test file is missing")
+        for path in paths
+        if not (repo_root / path).is_file()
+    )
+    group_findings: Final = tuple(
+        Finding(group, "canonical integration group is not scheduled by CircleCI")
+        for group in sorted(required - scheduled)
+    ) + tuple(Finding(path, "canonical node must have exactly one integration group") for path in sorted(ungrouped))
+    if not paths or not invoked or not scheduled:
+        return frozenset(), findings + (
+            Finding(str(manifest.relative_to(repo_root)), "dedicated CircleCI runner is missing"),
+        )
+    return paths, findings + group_findings
+
+
 def main() -> int:
     if "--shards" in sys.argv[1:]:
         return _check_shards()
@@ -507,7 +571,8 @@ def main() -> int:
     allowlist = _load_allowlist()
     scalars = _all_scalars()
 
-    test_findings = _uncovered_tests(allowlist, _invoked_test_tokens(scalars))
+    integration_paths, ownership_findings = _integration_ownership()
+    test_findings = _uncovered_tests(allowlist, _invoked_test_tokens(scalars) | integration_paths) + ownership_findings
     dockerfile_findings = _uncovered_dockerfiles(allowlist, _built_dockerfile_tokens(scalars))
     stale_findings = _stale_allowlist_paths(allowlist, test_files=_test_files(), dockerfiles=_dockerfiles())
 
diff --git a/tests/e2e/coverage_registry/mgmt.yaml b/tests/e2e/coverage_registry/mgmt.yaml
index d1227fe7c0c..18fc2d18bb9 100644
--- a/tests/e2e/coverage_registry/mgmt.yaml
+++ b/tests/e2e/coverage_registry/mgmt.yaml
@@ -90,3 +90,4 @@
 - {id: mgmt.mcp_toolset.update.persists, module: mgmt, tier: P0, surface: api, assertions: [persists], source: "mcp_management_endpoints.py:3098", rationale: "Narrowing the tools to one entry reads back exactly that entry"}
 - {id: mgmt.mcp_toolset.update.clear_persists, module: mgmt, tier: P0, surface: api, assertions: [clear_persists], source: "mcp_management_endpoints.py:3098", fail_before_fix: proven, rationale: "An explicit null clears the stored description; the update used to drop null and keep the old value"}
 - {id: mgmt.mcp_toolset.delete.persists, module: mgmt, tier: P0, surface: api, assertions: [persists], source: "mcp_management_endpoints.py:3149", rationale: "A deleted toolset is gone by id and from the list on every replica"}
+- {id: mgmt.key.update.preserves_independent_fields, module: mgmt, tier: P0, surface: api, assertions: [preserves_independent_fields], source: "proxy/management_endpoints/key_management_endpoints.py", rationale: "A partial key update preserves independent policy and metadata through serving"}
diff --git a/tests/e2e/coverage_registry/other.yaml b/tests/e2e/coverage_registry/other.yaml
index 9292e5f07db..b3bddfd898c 100644
--- a/tests/e2e/coverage_registry/other.yaml
+++ b/tests/e2e/coverage_registry/other.yaml
@@ -54,3 +54,5 @@
 
 - {id: other.auth.jwt.wrong_issuer_denied, module: other, tier: P0, area: auth, assertions: [wrong_issuer_denied], source: "auth/handle_jwt.py", rationale: "A signed token with the correct audience and an unexpected issuer is rejected"}
 - {id: other.auth.jwt.wrong_audience_denied, module: other, tier: P0, area: auth, assertions: [wrong_audience_denied], source: "auth/handle_jwt.py", rationale: "A signed token from the trusted issuer intended for another app is rejected"}
+- {id: other.provider_wire.internal_parameters_filtered, module: other, tier: P0, area: provider_wire, assertions: [internal_parameters_filtered], source: "main.py", rationale: "A real provider request preserves content and public parameters without internal limiter fields"}
+- {id: other.provider_wire.validator_rejects_corruption, module: other, tier: P0, area: provider_wire, assertions: [validator_rejects_corruption], source: "tests/integration/_support/upstream.py", rationale: "The controlled transport rejects missing messages and internal fields while accepting supported metadata"}
diff --git a/tests/e2e/coverage_registry/quota_management.yaml b/tests/e2e/coverage_registry/quota_management.yaml
index ad0914d455b..66198e6649c 100644
--- a/tests/e2e/coverage_registry/quota_management.yaml
+++ b/tests/e2e/coverage_registry/quota_management.yaml
@@ -63,3 +63,6 @@
 - {id: quota_management.spend_tracking.key_attribution.health_rows_keep_service_account, module: quota_management, tier: P1, behavior: spend_tracking, variant: key_attribution, assertions: [health_rows_keep_service_account], exercised_on: [chat_completions], source: "proxy/health_check.py", rationale: "A /health probe's spend row stays keyed by the literal litellm-internal-health-check service account rather than a hash of it, so health spend never appears as an unattributed key"}
 - {id: quota_management.spend_tracking.key_attribution.retrieve_batch_cost_joins_retrieving_key, module: quota_management, tier: P1, behavior: spend_tracking, variant: key_attribution, assertions: [retrieve_batch_cost_joins_retrieving_key], exercised_on: [batches], source: "proxy/batches_endpoints/endpoints.py", rationale: "The retrieve that first sees a batch in a terminal state prices it inline and writes its {provider_batch_id}_batch_cost row against the retrieving key, so the batch each run creates is one OpenAI fails at validation within seconds and the test retrieves it by its raw provider id with the same key until it is failed; a raw id is never owned by the CheckBatchCost poller, and the row must carry that key's token hash and alias"}
 - {id: quota_management.spend_tracking.key_attribution.poller_batch_cost_joins_creating_key, module: quota_management, tier: P1, behavior: spend_tracking, variant: key_attribution, assertions: [poller_batch_cost_joins_creating_key], exercised_on: [batches], source: "enterprise/litellm_enterprise/proxy/common_utils/check_batch_cost.py", rationale: "The CheckBatchCost poller bills a completed, positive-cost batch created through a unified id against the key that created it, a different writer from the inline retrieve. No test claims this cell yet: OpenAI's completion window is 24h and both e2e stacks boot a fresh Postgres per build, so a completed batch is out of one run's reach and the managed list never shows an earlier run's batch; the cell stays visible as a gap until a run can hand a completed batch to the poller"}
+- {id: quota_management.spend_tracking.custom_price.matches_input_rates, module: quota_management, tier: P0, behavior: spend_tracking, variant: custom_price, assertions: [matches_input_rates], exercised_on: [chat_completions], source: "router.py", rationale: "Configured deployment prices reach the response cost"}
+- {id: quota_management.spend_tracking.default_prices.survive_nullable_sibling_reload, module: quota_management, tier: P0, behavior: spend_tracking, variant: default_prices, assertions: [survive_nullable_sibling_reload], exercised_on: [chat_completions], source: "router.py", rationale: "Omitted and null prices retain defaults across sibling loading and reload, with persisted request charges"}
+- {id: quota_management.spend_tracking.default_prices.loaded_router_preserves_cached_defaults, module: quota_management, tier: P0, behavior: spend_tracking, variant: default_prices, assertions: [loaded_router_preserves_cached_defaults], exercised_on: [chat_completions], source: "router.py", rationale: "YAML-loaded omitted and null model-info prices preserve cached defaults across real SDK requests and reload order"}
diff --git a/tests/integration/README.md b/tests/integration/README.md
new file mode 100644
index 00000000000..47ec35ae6c7
--- /dev/null
+++ b/tests/integration/README.md
@@ -0,0 +1,15 @@
+# Integration contracts
+
+These tests exercise a running gateway, PostgreSQL and Redis with an owned local upstream. CircleCI owns this suite. Tests are grouped by behavior, with no automatic test retries or fallback to paid provider calls
+
+Use `tests/integration/run.py management`, `accounting` or `providers` to run a selected group. Set `INTEGRATION_PROXY_URL`, `INTEGRATION_UPSTREAM_URL`, `INTEGRATION_MASTER_KEY` and `DATABASE_URL` to an isolated test deployment. The runner selects the new domain directories explicitly; the legacy OCI and sandbox selections remain separate
+
+Reuse the existing canned provider handlers through `_support/upstream.py`. It rejects internal request fields and exposes actual received requests for independent assertions. Register every created resource for cleanup immediately, keep expected values independent of production calculations, and assert readback plus the runtime effect of a change
+
+The CircleCI workflow starts its own database and Redis, restricts test-phase egress to its owned services and writes JUnit plus an executed-node manifest. Missing setup, skipped tests, failed cleanup or a selected test without a passed call fail qualification. Existing GitHub Actions jobs do not own these tests
+
+Add contract definitions to the existing `tests/e2e/coverage_registry` and map canonical node IDs to those definitions in `contracts.json`. Every node must declare the same IDs with `covers`. The runner checks exact collected and passed selections against that mapping. Registry declarations alone do not mean a test passed
+
+Provider sentinels currently use the controlled server, not live recordings. The provider shard also runs the existing strict replay controls for changed requests, exhausted interactions, leftover interactions and no provider connection. Future recorded scenarios must use that replay-only implementation; missing recordings cannot fall back to a real provider. The observation endpoint is destructive and the current selection runs serially against one owned upstream
+
+Fixtures must contain synthetic data only. Keep private incident records and source documents out of code, fixtures, logs and PR descriptions
diff --git a/tests/integration/__init__.py b/tests/integration/__init__.py
new file mode 100644
index 00000000000..e69de29bb2d
diff --git a/tests/integration/_support/__init__.py b/tests/integration/_support/__init__.py
new file mode 100644
index 00000000000..e69de29bb2d
diff --git a/tests/integration/_support/client.py b/tests/integration/_support/client.py
new file mode 100644
index 00000000000..160859709b1
--- /dev/null
+++ b/tests/integration/_support/client.py
@@ -0,0 +1,134 @@
+from __future__ import annotations
+
+import os
+import time
+import uuid
+from hashlib import sha256
+from collections.abc import Callable, Iterator, Mapping
+from contextlib import ExitStack, contextmanager
+from dataclasses import dataclass
+from typing import Final, TypeVar
+
+import httpx
+from pydantic import JsonValue, TypeAdapter
+
+from integration._support.database import read_rows
+
+JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue])
+T = TypeVar("T")
+
+
+def object_value(value: JsonValue) -> dict[str, JsonValue]:
+    return JSON_OBJECT.validate_python(value)
+
+
+def string_value(value: JsonValue) -> str:
+    assert isinstance(value, str), f"Expected a string, received {type(value).__name__}"
+    return value
+
+
+def eventually(read: Callable[[], T], satisfied: Callable[[T], bool], seconds: float = 10) -> T:
+    deadline: Final = time.monotonic() + seconds
+    while True:
+        observed: Final = read()
+        if satisfied(observed):
+            return observed
+        assert time.monotonic() < deadline, f"State did not converge: {observed!r}"
+        time.sleep(0.1)
+
+
+@dataclass(frozen=True, slots=True)
+class Gateway:
+    client: httpx.Client
+    key: str
+    upstream_url: str
+
+    def request(
+        self,
+        method: str,
+        path: str,
+        body: Mapping[str, JsonValue] | None = None,
+        *,
+        key: str | None = None,
+        params: Mapping[str, str] | None = None,
+    ) -> httpx.Response:
+        return self.client.request(
+            method,
+            path,
+            json=body,
+            params=params,
+            headers={"Authorization": f"Bearer {self.key if key is None else key}"},
+        )
+
+    def post(self, path: str, body: Mapping[str, JsonValue], *, key: str | None = None) -> dict[str, JsonValue]:
+        response: Final = self.request("POST", path, body, key=key)
+        assert response.status_code == 200, f"POST {path}: {response.status_code} {response.text}"
+        return JSON_OBJECT.validate_json(response.content)
+
+    def get(self, path: str, params: Mapping[str, str] | None = None) -> dict[str, JsonValue]:
+        response: Final = self.request("GET", path, params=params)
+        assert response.status_code == 200, f"GET {path}: {response.status_code} {response.text}"
+        return JSON_OBJECT.validate_json(response.content)
+
+    def chat(self, model: str, *, key: str | None = None, text: str = "integration control") -> dict[str, JsonValue]:
+        return self.post(
+            "/v1/chat/completions",
+            {"model": model, "messages": [{"role": "user", "content": text}]},
+            key=key,
+        )
+
+    @contextmanager
+    def scenario(self) -> Iterator[Scenario]:
+        with ExitStack() as cleanups:
+            yield Scenario(self, cleanups)
+
+
+@dataclass(frozen=True, slots=True)
+class Scenario:
+    gateway: Gateway
+    cleanups: ExitStack
+
+    def key(self, **fields: JsonValue) -> str:
+        created: Final = self.gateway.post("/key/generate", fields)
+        token: Final = string_value(created["key"])
+        self.cleanups.callback(self.delete_key, token)
+        return token
+
+    def delete_key(self, token: str) -> None:
+        self.gateway.post("/key/delete", {"keys": [token]})
+        response: Final = self.gateway.request("GET", "/key/info", params={"key": sha256(token.encode()).hexdigest()})
+        assert response.status_code == 404, f"Deleted key remains readable: {response.status_code}"
+
+    def delete_model(self, identity: str) -> None:
+        self.gateway.post("/model/delete", {"id": identity})
+        entries: Final = self.gateway.get("/model/info")["data"]
+        assert isinstance(entries, list)
+        assert all(object_value(object_value(entry)["model_info"])["id"] != identity for entry in entries)
+        assert read_rows('SELECT model_id FROM "LiteLLM_ProxyModelTable" WHERE model_id = %s', (identity,)) == []
+
+    def model(self, **parameters: JsonValue) -> str:
+        name: Final = f"integration-{uuid.uuid4().hex}"
+        created: Final = self.gateway.post(
+            "/model/new",
+            {
+                "model_name": name,
+                "litellm_params": {
+                    "model": "openai/gpt-4o-mini",
+                    "api_key": "integration-provider-key",
+                    "api_base": f"{self.gateway.upstream_url}/v1",
+                    **parameters,
+                },
+                "model_info": {},
+            },
+        )
+        identity: Final = string_value(object_value(created["model_info"])["id"])
+        self.cleanups.callback(self.delete_model, identity)
+        return name
+
+
+@contextmanager
+def gateway_from_environment() -> Iterator[Gateway]:
+    url: Final = os.environ["INTEGRATION_PROXY_URL"]
+    upstream: Final = os.environ["INTEGRATION_UPSTREAM_URL"]
+    with httpx.Client(base_url=url, timeout=15, trust_env=False) as client:
+        yield Gateway(client, os.environ["INTEGRATION_MASTER_KEY"], upstream)
diff --git a/tests/integration/_support/database.py b/tests/integration/_support/database.py
new file mode 100644
index 00000000000..283d26a632e
--- /dev/null
+++ b/tests/integration/_support/database.py
@@ -0,0 +1,14 @@
+import os
+from typing import Final
+
+import psycopg
+from psycopg.rows import dict_row
+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:
+        connection.execute("SET TRANSACTION READ ONLY")
+        return ROWS.validate_python(connection.execute(query, parameters).fetchall())
diff --git a/tests/integration/_support/manifest.py b/tests/integration/_support/manifest.py
new file mode 100644
index 00000000000..a1e7f11b49d
--- /dev/null
+++ b/tests/integration/_support/manifest.py
@@ -0,0 +1,37 @@
+import json
+from pathlib import Path
+from typing import Final
+
+from pydantic import TypeAdapter
+
+from e2e.coverage_registry.registry import load_registry
+
+MAPPING: Final = TypeAdapter(dict[str, tuple[str, ...]])
+OWNED_DIRECTORIES: Final = frozenset(
+    {
+        "management",
+        "authorization",
+        "database",
+        "pricing",
+        "spend",
+        "routing",
+        "providers",
+        "streaming",
+        "configuration",
+        "mcp",
+        "observability",
+        "compatibility",
+    }
+)
+
+
+def contracts() -> dict[str, tuple[str, ...]]:
+    document: Final = json.loads((Path(__file__).resolve().parents[1] / "contracts.json").read_bytes())
+    result: Final = MAPPING.validate_python(document["tests"])
+    if not result or any(not values for values in result.values()):
+        raise ValueError("Integration manifest must contain nodes with contract IDs")
+    registered: Final = {cell.id for cell in load_registry()}
+    unknown: Final = {identity for identities in result.values() for identity in identities} - registered
+    if unknown:
+        raise ValueError(f"Unknown coverage-registry contracts: {sorted(unknown)}")
+    return result
diff --git a/tests/integration/_support/upstream.py b/tests/integration/_support/upstream.py
new file mode 100644
index 00000000000..6f3b4784385
--- /dev/null
+++ b/tests/integration/_support/upstream.py
@@ -0,0 +1,93 @@
+from __future__ import annotations
+
+import argparse
+from dataclasses import dataclass, field
+from queue import SimpleQueue
+from typing import Final
+
+import uvicorn
+from pydantic import JsonValue, TypeAdapter
+from starlette.applications import Starlette
+from starlette.requests import Request
+from starlette.responses import JSONResponse, Response
+from starlette.routing import Route
+
+from _fake_openai_endpoint_server import chat_completions, completions, embeddings, health, moderations
+
+JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue])
+INTERNAL_FIELDS: Final = frozenset(
+    {
+        "litellm_params",
+        "litellm_logging_obj",
+        "litellm_call_id",
+        "litellm_metadata",
+        "proxy_server_request",
+        "rpm",
+        "tpm",
+        "timeout",
+        "stream_chunk_size",
+    }
+)
+
+
+@dataclass(frozen=True, slots=True)
+class Observation:
+    path: str
+    authorization: str
+    body: dict[str, JsonValue]
+
+
+@dataclass(frozen=True, slots=True)
+class Provider:
+    observations: SimpleQueue[Observation] = field(default_factory=SimpleQueue)
+
+    async def chat(self, request: Request) -> Response:
+        body: Final = JSON_OBJECT.validate_json(await request.body())
+        self.observations.put(Observation(request.url.path, request.headers.get("authorization", ""), body))
+        leaked: Final = tuple(sorted(INTERNAL_FIELDS.intersection(body)))
+        if leaked:
+            return JSONResponse({"error": {"message": f"Unexpected provider fields: {leaked}"}}, status_code=400)
+        messages: Final = body.get("messages")
+        if not isinstance(body.get("model"), str) or not isinstance(messages, list) or not messages:
+            return JSONResponse({"error": {"message": "model and nonempty messages are required"}}, status_code=400)
+        if any(
+            not isinstance(message, dict)
+            or message.get("role") not in {"system", "developer", "user", "assistant", "tool"}
+            or "content" not in message
+            for message in messages
+        ):
+            return JSONResponse({"error": {"message": "Invalid selected message contract"}}, status_code=400)
+        return await chat_completions(request)
+
+    async def observed(self, _request: Request) -> Response:
+        values: Final = tuple(self.observations.get() for _ in range(self.observations.qsize()))
+        return JSONResponse(
+            {
+                "requests": [
+                    {"path": value.path, "authorization": value.authorization, "body": value.body} for value in values
+                ]
+            }
+        )
+
+    def app(self) -> Starlette:
+        return Starlette(
+            routes=[
+                Route("/health", health),
+                Route("/__observations", self.observed),
+                Route("/v1/chat/completions", self.chat, methods=["POST"]),
+                Route("/v1/completions", completions, methods=["POST"]),
+                Route("/v1/embeddings", embeddings, methods=["POST"]),
+                Route("/v1/moderations", moderations, methods=["POST"]),
+            ]
+        )
+
+
+def main() -> None:
+    parser: Final = argparse.ArgumentParser()
+    parser.add_argument("--port", type=int, default=8190)
+    arguments: Final = parser.parse_args()
+    uvicorn.run(Provider().app(), host="127.0.0.1", port=arguments.port, access_log=False)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py
new file mode 100644
index 00000000000..908a25bf4cb
--- /dev/null
+++ b/tests/integration/conftest.py
@@ -0,0 +1,79 @@
+from __future__ import annotations
+
+import json
+import os
+from collections.abc import Generator, Iterator
+from pathlib import Path
+from typing import Final
+
+import pytest
+
+from integration._support.client import Gateway, gateway_from_environment
+from integration._support.manifest import OWNED_DIRECTORIES, contracts
+
+COLLECTED: Final = pytest.StashKey[tuple[str, ...]]()
+REPORTS: Final = pytest.StashKey[list[pytest.TestReport]]()
+
+
+def pytest_configure(config: pytest.Config) -> None:
+    config.addinivalue_line("markers", "integration: owned real-service integration contracts")
+    config.addinivalue_line("markers", "covers(*ids): independently asserted behavior contracts")
+    config.stash[REPORTS] = []
+
+
+def pytest_collection_modifyitems(config: pytest.Config, items: list[pytest.Item]) -> None:
+    manifest: Final = contracts()
+    root: Final = Path(__file__).parent
+    owned: Final = tuple(
+        item
+        for item in items
+        if item.path.is_relative_to(root) and item.path.relative_to(root).parts[0] in OWNED_DIRECTORIES
+    )
+    if owned and os.environ.get("GITHUB_ACTIONS") == "true":
+        raise pytest.UsageError("Integration contracts are owned by CircleCI")
+    for item in owned:
+        if item.nodeid not in manifest:
+            raise pytest.UsageError(f"Integration node missing from manifest: {item.nodeid}")
+        item.add_marker(pytest.mark.integration)
+        declared: Final = tuple(value for mark in item.iter_markers("covers") for value in mark.args)
+        if set(declared) != set(manifest[item.nodeid]):
+            raise pytest.UsageError(f"Contract mapping differs for {item.nodeid}")
+    config.stash[COLLECTED] = tuple(item.nodeid for item in owned)
+
+
+@pytest.hookimpl(wrapper=True)
+def pytest_runtest_makereport(
+    item: pytest.Item, call: pytest.CallInfo[None]
+) -> Generator[None, pytest.TestReport, pytest.TestReport]:
+    report: Final = yield
+    item.config.stash[REPORTS].append(report)
+    return report
+
+
+def pytest_sessionfinish(session: pytest.Session, exitstatus: int) -> None:
+    destination: Final = os.environ.get("INTEGRATION_RESULTS_DIR")
+    if destination is None:
+        return
+    collected: Final = session.config.stash.get(COLLECTED, ())
+    reports: Final = tuple(report for report in session.config.stash[REPORTS] if report.nodeid in collected)
+    passed: Final = tuple(report.nodeid for report in reports if report.when == "call" and report.passed)
+    complete: Final = (
+        exitstatus == 0
+        and bool(collected)
+        and sorted(collected) == sorted(passed)
+        and all(report.passed for report in reports)
+    )
+    output: Final = Path(destination)
+    output.mkdir(parents=True, exist_ok=True)
+    (output / "execution.json").write_text(
+        json.dumps({"collected": collected, "passed": passed, "complete": complete, "exitstatus": exitstatus}, indent=2)
+        + "\n"
+    )
+    if not complete and exitstatus == 0:
+        session.exitstatus = pytest.ExitCode.TESTS_FAILED
+
+
+@pytest.fixture
+def gateway() -> Iterator[Gateway]:
+    with gateway_from_environment() as value:
+        yield value
diff --git a/tests/integration/contracts.json b/tests/integration/contracts.json
new file mode 100644
index 00000000000..d3139ff8e06
--- /dev/null
+++ b/tests/integration/contracts.json
@@ -0,0 +1,46 @@
+{
+  "groups": {
+    "management": [
+      "management",
+      "authorization",
+      "configuration"
+    ],
+    "accounting": [
+      "pricing",
+      "spend"
+    ],
+    "database": [
+      "database"
+    ],
+    "providers": [
+      "providers",
+      "routing",
+      "streaming"
+    ],
+    "extensions": [
+      "mcp",
+      "observability",
+      "compatibility"
+    ]
+  },
+  "tests": {
+    "tests/integration/management/test_key_updates.py::test_update_preserves_independent_fields_and_serving": [
+      "mgmt.key.update.preserves_independent_fields"
+    ],
+    "tests/integration/pricing/test_configured_prices.py::test_custom_price_is_reported_and_charged": [
+      "quota_management.spend_tracking.custom_price.matches_input_rates"
+    ],
+    "tests/integration/providers/test_request_boundary.py::test_internal_request_state_does_not_reach_provider": [
+      "other.provider_wire.internal_parameters_filtered"
+    ],
+    "tests/integration/pricing/test_configured_prices.py::test_default_prices_survive_nullable_sibling_and_reload": [
+      "quota_management.spend_tracking.default_prices.survive_nullable_sibling_reload"
+    ],
+    "tests/integration/providers/test_request_boundary.py::test_upstream_rejects_corruption_and_accepts_supported_metadata": [
+      "other.provider_wire.validator_rejects_corruption"
+    ],
+    "tests/integration/pricing/test_configured_prices.py::test_loaded_router_preserves_cached_defaults_during_real_requests": [
+      "quota_management.spend_tracking.default_prices.loaded_router_preserves_cached_defaults"
+    ]
+  }
+}
diff --git a/tests/integration/management/test_key_updates.py b/tests/integration/management/test_key_updates.py
new file mode 100644
index 00000000000..6f2e850b17a
--- /dev/null
+++ b/tests/integration/management/test_key_updates.py
@@ -0,0 +1,40 @@
+from typing import Final
+from hashlib import sha256
+
+import pytest
+
+from integration._support.client import Gateway, object_value
+from integration._support.database import read_rows
+
+
+@pytest.mark.covers("mgmt.key.update.preserves_independent_fields")
+def test_update_preserves_independent_fields_and_serving(gateway: Gateway) -> None:
+    with gateway.scenario() as scenario:
+        model: Final = scenario.model()
+        key: Final = scenario.key(models=[model], key_alias="before", metadata={"retained": "value"})
+        gateway.chat(model, key=key)
+        gateway.post("/key/update", {"key": key, "key_alias": "after"})
+        info: Final = object_value(gateway.get("/key/info", {"key": key})["info"])
+        assert info["key_alias"] == "after"
+        assert info["models"] == [model]
+        assert object_value(info["metadata"])["retained"] == "value"
+        response: Final = gateway.chat(model, key=key)
+        assert object_value(response["usage"])["total_tokens"] == 40
+        replacement: Final = scenario.model()
+        gateway.post("/key/update", {"key": key, "models": [replacement]})
+        saved: Final = read_rows(
+            'SELECT key_alias, models, metadata FROM "LiteLLM_VerificationToken" WHERE token = %s',
+            (sha256(key.encode()).hexdigest(),),
+        )
+        assert len(saved) == 1
+        assert saved[0]["models"] == [replacement]
+        assert saved[0]["key_alias"] == "after"
+        denied: Final = gateway.request(
+            "POST",
+            "/v1/chat/completions",
+            {"model": model, "messages": [{"role": "user", "content": "old grant"}]},
+            key=key,
+        )
+        assert denied.status_code == 403, denied.text
+        assert object_value(object_value(denied.json())["error"])["type"] == "key_model_access_denied"
+        assert object_value(gateway.chat(replacement, key=key)["usage"])["total_tokens"] == 40
diff --git a/tests/integration/pricing/test_configured_prices.py b/tests/integration/pricing/test_configured_prices.py
new file mode 100644
index 00000000000..584669711ff
--- /dev/null
+++ b/tests/integration/pricing/test_configured_prices.py
@@ -0,0 +1,132 @@
+from typing import Final
+from pathlib import Path
+import uuid
+
+import pytest
+import yaml
+
+from integration._support.client import Gateway, eventually, object_value, string_value
+from integration._support.database import read_rows
+
+
+@pytest.mark.covers("quota_management.spend_tracking.custom_price.matches_input_rates")
+def test_custom_price_is_reported_and_charged(gateway: Gateway) -> None:
+    with gateway.scenario() as scenario:
+        model: Final = scenario.model(input_cost_per_token=0.001, output_cost_per_token=0.002)
+        response: Final = gateway.request(
+            "POST", "/v1/chat/completions", {"model": model, "messages": [{"role": "user", "content": "price control"}]}
+        )
+        assert response.status_code == 200, response.text
+        assert float(response.headers["x-litellm-response-cost"]) == pytest.approx(20 * 0.001 + 20 * 0.002)
+        entries: Final = gateway.get("/model/info")["data"]
+        assert isinstance(entries, list)
+        matching: Final = tuple(object_value(entry) for entry in entries if object_value(entry)["model_name"] == model)
+        assert len(matching) == 1
+        params: Final = object_value(matching[0]["litellm_params"])
+        assert params["input_cost_per_token"] == 0.001
+        assert params["output_cost_per_token"] == 0.002
+
+
+@pytest.mark.covers("quota_management.spend_tracking.default_prices.survive_nullable_sibling_reload")
+def test_default_prices_survive_nullable_sibling_and_reload(gateway: Gateway) -> None:
+    for registration_order in (("custom", "omitted", "nullable"), ("nullable", "omitted", "custom")):
+        with gateway.scenario() as scenario:
+            configured: Final = {
+                "custom": {"input_cost_per_token": 0.001, "output_cost_per_token": 0.002},
+                "omitted": {},
+                "nullable": {"input_cost_per_token": None, "output_cost_per_token": None},
+            }
+            rates: Final = {
+                "custom": (0.001, 0.002),
+                "omitted": (0.00000015, 0.0000006),
+                "nullable": (0.00000015, 0.0000006),
+            }
+            models: Final = {kind: scenario.model(**configured[kind]) for kind in registration_order}
+            observations: Final[list[tuple[str, float]]] = []
+            for generation in range(2):
+                entries: Final = gateway.get("/model/info")["data"]
+                assert isinstance(entries, list)
+                for kind in reversed(registration_order) if generation else registration_order:
+                    model: Final = models[kind]
+                    target: Final = next(
+                        object_value(entry) for entry in entries if object_value(entry)["model_name"] == model
+                    )
+                    info: Final = object_value(target["model_info"])
+                    assert info["input_cost_per_token"] == rates[kind][0]
+                    assert info["output_cost_per_token"] == rates[kind][1]
+                    response: Final = gateway.request(
+                        "POST",
+                        "/v1/chat/completions",
+                        {"model": model, "messages": [{"role": "user", "content": f"price {len(observations)}"}]},
+                    )
+                    assert response.status_code == 200, response.text
+                    expected: Final = 20 * rates[kind][0] + 20 * rates[kind][1]
+                    assert float(response.headers["x-litellm-response-cost"]) == pytest.approx(expected, rel=1e-6)
+                    request_id: Final = string_value(object_value(response.json())["id"])
+                    observations.append((request_id, expected))
+                target: Final = next(
+                    object_value(entry) for entry in entries if object_value(entry)["model_name"] == models["nullable"]
+                )
+                identity: Final = string_value(object_value(target["model_info"])["id"])
+                updated: Final = gateway.request(
+                    "PATCH", f"/model/{identity}/update", {"model_info": {"description": "reload price contract"}}
+                )
+                assert updated.status_code == 200, updated.text
+            for request_id, expected in observations:
+                rows: Final = eventually(
+                    lambda request_id=request_id: read_rows(
+                        'SELECT request_id, spend, 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"] == 20
+                assert rows[0]["completion_tokens"] == 20
+                assert float(rows[0]["spend"]) == pytest.approx(expected, rel=1e-6)
+
+
+@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
+
+    aliases: Final = (f"pricing-{uuid.uuid4().hex}", f"pricing-{uuid.uuid4().hex}")
+    path: Final = tmp_path / "models.yaml"
+    path.write_text(
+        yaml.safe_dump(
+            {
+                "model_list": [
+                    {
+                        "model_name": alias,
+                        "litellm_params": {
+                            "model": "openai/gpt-4o-mini",
+                            "api_key": "integration-provider-key",
+                            "api_base": f"{gateway.upstream_url}/v1",
+                        },
+                        "model_info": {"id": alias, **pricing},
+                    }
+                    for alias, pricing in zip(
+                        aliases, ({}, {"input_cost_per_token": None, "output_cost_per_token": None}), strict=True
+                    )
+                ]
+            }
+        )
+    )
+    for reverse in (False, True):
+        configured: Final = yaml.safe_load(path.read_text())["model_list"]
+        router: Final = Router(model_list=list(reversed(configured)) if reverse else configured, num_retries=0)
+        try:
+            for alias in (*aliases, *reversed(aliases)):
+                result: Final = router.completion(
+                    model=alias, messages=[{"role": "user", "content": "router price control"}]
+                )
+                assert result.usage.prompt_tokens == 20
+                assert result.usage.completion_tokens == 20
+                deployment: Final = router.get_deployment(model_id=alias)
+                assert deployment is not None
+                info: Final = router.get_router_model_info(deployment=deployment, received_model_name=alias)
+                assert info["input_cost_per_token"] == 0.00000015
+                assert info["output_cost_per_token"] == 0.0000006
+        finally:
+            router.reset()
diff --git a/tests/integration/providers/test_request_boundary.py b/tests/integration/providers/test_request_boundary.py
new file mode 100644
index 00000000000..aad10843642
--- /dev/null
+++ b/tests/integration/providers/test_request_boundary.py
@@ -0,0 +1,57 @@
+from typing import Final
+
+import httpx
+import pytest
+
+from integration._support.client import Gateway, JSON_OBJECT, object_value
+
+
+@pytest.mark.covers("other.provider_wire.internal_parameters_filtered")
+def test_internal_request_state_does_not_reach_provider(gateway: Gateway) -> None:
+    with gateway.scenario() as scenario, httpx.Client(base_url=gateway.upstream_url, trust_env=False) as upstream:
+        upstream.get("/__observations").raise_for_status()
+        model: Final = scenario.model()
+        key: Final = scenario.key(models=[model], tpm_limit=10000, rpm_limit=100)
+        result: Final = gateway.post(
+            "/v1/chat/completions",
+            {
+                "model": model,
+                "messages": [{"role": "user", "content": "wire contract"}],
+                "temperature": 0.4,
+                "max_tokens": 20,
+                "timeout": 12,
+            },
+            key=key,
+        )
+        assert object_value(result["usage"])["total_tokens"] == 40
+        observations: Final = JSON_OBJECT.validate_json(upstream.get("/__observations").content)["requests"]
+        assert isinstance(observations, list)
+        assert len(observations) == 1
+        observed: Final = object_value(observations[0])
+        body: Final = object_value(observed["body"])
+        assert body["model"] == "gpt-4o-mini"
+        assert body["messages"] == [{"role": "user", "content": "wire contract"}]
+        assert body["temperature"] == 0.4
+        assert body["max_tokens"] == 20
+        assert observed["authorization"] == "Bearer integration-provider-key"
+        assert "litellm_metadata" not in body
+        assert "litellm_params" not in body
+        assert "timeout" not in body
+        assert "tpm" not in body
+
+
+@pytest.mark.covers("other.provider_wire.validator_rejects_corruption")
+def test_upstream_rejects_corruption_and_accepts_supported_metadata(gateway: Gateway) -> None:
+    with httpx.Client(base_url=gateway.upstream_url, trust_env=False) as upstream:
+        missing: Final = upstream.post("/v1/chat/completions", json={"model": "gpt-4o-mini"})
+        assert missing.status_code == 400
+        assert (
+            object_value(object_value(missing.json())["error"])["message"] == "model and nonempty messages are required"
+        )
+        body: Final = {"model": "gpt-4o-mini", "messages": [{"role": "user", "content": "strict control"}]}
+        leaked: Final = upstream.post("/v1/chat/completions", json={**body, "litellm_metadata": {"hidden": "value"}})
+        assert leaked.status_code == 400
+        assert "litellm_metadata" in str(object_value(object_value(leaked.json())["error"])["message"])
+        valid: Final = upstream.post("/v1/chat/completions", json={**body, "metadata": {"purpose": "synthetic"}})
+        assert valid.status_code == 200, valid.text
+        assert object_value(object_value(valid.json())["usage"])["total_tokens"] == 40
diff --git a/tests/integration/proxy_config.yaml b/tests/integration/proxy_config.yaml
new file mode 100644
index 00000000000..b6f9767c210
--- /dev/null
+++ b/tests/integration/proxy_config.yaml
@@ -0,0 +1,17 @@
+model_list: []
+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
+litellm_settings:
+  enable_redis_auth_cache: true
+  cache: true
+  cache_params:
+    type: redis
+    host: os.environ/REDIS_HOST
+    port: os.environ/REDIS_PORT
+router_settings:
+  num_retries: 0
+  disable_cooldowns: true
diff --git a/tests/integration/run.py b/tests/integration/run.py
new file mode 100644
index 00000000000..6e343173a82
--- /dev/null
+++ b/tests/integration/run.py
@@ -0,0 +1,69 @@
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import subprocess
+import sys
+from pathlib import Path
+from types import MappingProxyType
+from typing import Final
+
+GROUPS: Final = MappingProxyType(json.loads(Path(__file__).with_name("contracts.json").read_text())["groups"])
+
+
+def main() -> int:
+    parser: Final = argparse.ArgumentParser()
+    parser.add_argument("group", choices=tuple(GROUPS))
+    parser.add_argument("--results", type=Path, default=Path("test-results/integration"))
+    options: Final = parser.parse_args()
+    root: Final = Path(__file__).resolve().parents[2]
+    selected: Final = tuple(
+        str(path.relative_to(root))
+        for folder in GROUPS[options.group]
+        for path in sorted((root / "tests/integration" / folder).glob("test_*.py"))
+    )
+    if not selected:
+        parser.error(f"No integration contracts selected for {options.group}")
+    output: Final = options.results.resolve()
+    output.mkdir(parents=True, exist_ok=True)
+    manifest: Final = json.loads((root / "tests/integration/contracts.json").read_text())["tests"]
+    expected: Final = sorted(node for node in manifest if node.split("::", 1)[0] in selected)
+    if not expected or set(selected) != {node.split("::", 1)[0] for node in expected}:
+        parser.error("Every selected file must have canonical manifest nodes")
+    environment: Final = {
+        **os.environ,
+        "PYTHONPATH": os.pathsep.join((str(root), str(root / "tests"), str(root / "tests/e2e"))),
+        "INTEGRATION_RESULTS_DIR": str(output),
+        "LITELLM_LOCAL_MODEL_COST_MAP": "True",
+    }
+    result: Final = subprocess.call(
+        [
+            sys.executable,
+            "-m",
+            "pytest",
+            *selected,
+            "-vv",
+            "--strict-markers",
+            "-p",
+            "no:pytest-retry",
+            "-p",
+            "no:rerunfailures",
+            "--timeout=90",
+            "--durations=15",
+            f"--junitxml={output / 'junit.xml'}",
+        ],
+        cwd=root,
+        env=environment,
+    )
+    if result != 0:
+        return result
+    evidence: Final = json.loads((output / "execution.json").read_text())
+    if not evidence["complete"] or sorted(evidence["passed"]) != expected or sorted(evidence["collected"]) != expected:
+        print("Executed integration nodes differ from the canonical manifest", file=sys.stderr)
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/tests/test_litellm/test_assert_ci_coverage.py b/tests/test_litellm/test_assert_ci_coverage.py
index d948c1a4155..69db2411742 100644
--- a/tests/test_litellm/test_assert_ci_coverage.py
+++ b/tests/test_litellm/test_assert_ci_coverage.py
@@ -9,8 +9,12 @@ the question neither covers: whether the job that globs a file then deselects it
 """
 
 import importlib.util
+import json
 import sys
 from pathlib import Path
+from typing import Final
+
+import yaml
 
 _REPO_ROOT = Path(__file__).resolve().parents[2]
 _MODULE_PATH = _REPO_ROOT / ".github" / "scripts" / "assert_ci_coverage.py"
@@ -20,6 +24,56 @@ sys.modules[_spec.name] = coverage  # @dataclass(slots=True) rebuilds via sys.mo
 _spec.loader.exec_module(coverage)
 
 
+def test_integration_manifest_requires_exclusive_scheduled_circleci_owner(tmp_path: Path) -> None:
+    test_path: Final = "tests/integration/management/test_contract.py"
+    test_file: Final = tmp_path / test_path
+    test_file.parent.mkdir(parents=True)
+    test_file.write_text("def test_contract(): pass\n")
+    (tmp_path / "tests/integration/contracts.json").write_text(
+        json.dumps({"groups": {"management": ["management"]}, "tests": {f"{test_path}::test_contract": ["mgmt.test"]}})
+    )
+    paths, findings = coverage._integration_ownership(tmp_path)
+    assert not paths
+    assert [finding.detail for finding in findings] == ["dedicated CircleCI runner is missing"]
+    circle: Final = tmp_path / ".circleci/config.yml"
+    circle.parent.mkdir()
+    circle.write_text(
+        yaml.safe_dump(
+            {
+                "jobs": {
+                    "integration_contracts": {
+                        "steps": [{"run": {"command": "bash .circleci/scripts/run_integration.sh management"}}]
+                    }
+                },
+                "workflows": {"integration": {"jobs": [{"integration_contracts": {"suite": "management"}}]}},
+            }
+        )
+    )
+    paths, findings = coverage._integration_ownership(tmp_path)
+    assert paths == frozenset({test_path})
+    assert findings == ()
+    configured: Final = yaml.safe_load(circle.read_text())
+    configured["workflows"]["integration"]["jobs"] = [
+        {"integration_contracts": {"matrix": {"parameters": {"suite": ["providers"]}}}}
+    ]
+    circle.write_text(yaml.safe_dump(configured))
+    _, findings = coverage._integration_ownership(tmp_path)
+    assert [(finding.subject, finding.detail) for finding in findings] == [
+        ("management", "canonical integration group is not scheduled by CircleCI")
+    ]
+    configured["workflows"]["integration"]["jobs"][0]["integration_contracts"]["matrix"]["parameters"]["suite"] = [
+        "management"
+    ]
+    circle.write_text(yaml.safe_dump(configured))
+    workflow: Final = tmp_path / ".github/workflows/test.yml"
+    workflow.parent.mkdir(parents=True)
+    workflow.write_text(yaml.safe_dump({"jobs": {"tests": {"steps": [{"run": "pytest tests/integration"}]}}}))
+    _, findings = coverage._integration_ownership(tmp_path)
+    assert [(finding.subject, finding.detail) for finding in findings] == [
+        (test_path, "integration contract is also selected by GitHub Actions")
+    ]
+
+
 def test_an_ancestor_directory_covers_a_file_but_does_not_name_it():
     # The whole point of the split: `tests/x` answers "does it run?" but not
     # "which shard owns it?" — accepting it for the latter is how a new child