diff --git a/tests/integration/management/test_auto_router_databricks_classifier_writes.py b/tests/integration/management/test_auto_router_databricks_classifier_writes.py
index af486d90a7e..d17ce947671 100644
--- a/tests/integration/management/test_auto_router_databricks_classifier_writes.py
+++ b/tests/integration/management/test_auto_router_databricks_classifier_writes.py
@@ -3,6 +3,7 @@ import json
 import threading
 import uuid
 from collections.abc import Callable, Mapping, Sequence
+from pathlib import Path
 from queue import SimpleQueue
 from typing import Final
 
@@ -64,6 +65,21 @@ _ROWS_QUERY: Final = (
 )
 
 
+def _number(value: JsonValue) -> float:
+    assert isinstance(value, (int, float)) and not isinstance(value, bool), value
+    return float(value)
+
+
+def _endpoint_cost() -> float:
+    prices: Final = object_value(
+        json.loads(Path("model_prices_and_context_window.json").read_text())[f"databricks/{_ENDPOINT}"]
+    )
+    usage: Final = object_value(_ENDPOINT_ANSWER["usage"])
+    return _number(usage["input_tokens"]) * _number(prices["input_cost_per_token"]) + _number(
+        usage["output_tokens"]
+    ) * _number(prices["output_cost_per_token"])
+
+
 def _classifier(request: Request) -> Reply:
     answer: Final = _ENDPOINT_ANSWER if request.target.endswith("/invocations") else _AI_DECIDE_ANSWER
     return Reply(body=json.dumps(answer).encode())
@@ -221,11 +237,14 @@ def test_validate_names_each_databricks_classifier_misconfiguration(gateway: Gat
 
 @pytest.mark.parametrize(
     ("model", "target", "cost"),
-    [(_MODEL, _ROUTE, {}), (_ENDPOINT, f"/{_ENDPOINT}/invocations", {"classifier_cost": 0.0})],
+    [
+        (_MODEL, _ROUTE, {}),
+        (_ENDPOINT, f"/{_ENDPOINT}/invocations", {"classifier_cost": pytest.approx(_endpoint_cost())}),
+    ],
     ids=["ai_decide", "serving_endpoint"],
 )
 def test_test_routing_classifies_through_databricks_without_calling_the_tier_it_picked(
-    gateway: Gateway, model: str, target: str, cost: dict[str, JsonValue]
+    gateway: Gateway, model: str, target: str, cost: dict[str, object]
 ) -> None:
     prompt: Final = f"design a distributed cache {uuid.uuid4().hex}"
     with (
diff --git a/tests/integration/mcp/test_mcp_access_matrix.py b/tests/integration/mcp/test_mcp_access_matrix.py
index e59f3e52a11..762fbee384d 100644
--- a/tests/integration/mcp/test_mcp_access_matrix.py
+++ b/tests/integration/mcp/test_mcp_access_matrix.py
@@ -181,7 +181,6 @@ def test_same_tool_name_on_two_servers_routes_by_prefix(gateway: Gateway) -> Non
 
 _PROBE: Final = "catalog-probe"
 _ECHO: Final = "catalog-echo"
-_UNLISTED: Final = ""
 _GUARDRAIL_CODE: Final = (
     "def apply_guardrail(inputs, request_data, input_type):\n"
     f'    if "{_PROBE}" not in list(inputs.get("texts") or []):\n'
@@ -217,13 +216,26 @@ def _echoed_description(outcome: Outcome) -> str:
     return found.group(1)
 
 
+def _add_described_anew_per_fetch(described: str) -> ScriptedTool:
+    return ScriptedTool("add", lambda _: text_result("3"), description=lambda _: f"{described} {uuid.uuid4().hex}")
+
+
+def _fetched(described: str) -> re.Pattern[str]:
+    return re.compile(rf"{re.escape(described)} [0-9a-f]{{32}}")
+
+
+def _listed_description(listing: Outcome, described: str) -> str:
+    found: Final = _fetched(described).search(listing.raw)
+    assert found is not None, listing.raw
+    return found.group(0)
+
+
 @pytest.mark.parametrize("subject", SUBJECTS)
 def test_each_subjects_call_is_evaluated_only_against_the_catalog_its_own_listing_served(
     echo_rig: Gateway, subject: Subject
 ) -> None:
     described: Final = "Adds under grant " + uuid.uuid4().hex[:8]
-    tool: Final = ScriptedTool("add", lambda _: text_result("3"), description=described)
-    with scripted_peer(tool) as peer, echo_rig.scenario() as scenario:
+    with scripted_peer(_add_described_anew_per_fetch(described)) as peer, echo_rig.scenario() as scenario:
         group: Final = "grp" + uuid.uuid4().hex[:8]
         alias: Final = "cat" + uuid.uuid4().hex[:8]
         identity: Final = register_mcp(scenario, peer, alias, mcp_access_groups=[group])
@@ -235,8 +247,10 @@ def test_each_subjects_call_is_evaluated_only_against_the_catalog_its_own_listin
         cold: Final = _echoed_description(reach.call(f"{alias}-add", {"probe": _PROBE}))
         listed: Final = reach.list_tools()
         assert listed.ok and f"{alias}-add" in listed.tools, listed.raw
+        served: Final = _listed_description(listed, described)
         warm: Final = _echoed_description(reach.call(f"{alias}-add", {"probe": _PROBE}))
-        assert (cold, warm) == (_UNLISTED, described), (cold, warm)
+        assert _fetched(described).fullmatch(cold) and cold != served, (cold, served)
+        assert warm == served, (cold, served, warm)
         assert tool_calls(peer.drain()) == (), "a blocked probe reached the peer"
 
 
@@ -244,8 +258,7 @@ def test_end_users_of_one_key_share_its_catalog_slot_because_the_identity_exclud
     echo_rig: Gateway,
 ) -> None:
     described: Final = "Adds for end users " + uuid.uuid4().hex[:8]
-    tool: Final = ScriptedTool("add", lambda _: text_result("3"), description=described)
-    with scripted_peer(tool) as peer, echo_rig.scenario() as scenario:
+    with scripted_peer(_add_described_anew_per_fetch(described)) as peer, echo_rig.scenario() as scenario:
         alias: Final = "eu" + uuid.uuid4().hex[:8]
         identity: Final = register_mcp(scenario, peer, alias)
         granted: Final = grant(scenario, "end_user", (identity,), (identity,))
@@ -254,10 +267,12 @@ def test_end_users_of_one_key_share_its_catalog_slot_because_the_identity_exclud
             echo_rig, granted.key, "mcp", alias, {"x-litellm-end-user-id": "integration-" + uuid.uuid4().hex[:10]}
         )
         assert second.initialize().ok
-        assert _echoed_description(second.call(f"{alias}-add", {"probe": _PROBE})) == _UNLISTED
+        cold: Final = _echoed_description(second.call(f"{alias}-add", {"probe": _PROBE}))
         listed: Final = first.list_tools()
         assert listed.ok and f"{alias}-add" in listed.tools, listed.raw
-        assert _echoed_description(second.call(f"{alias}-add", {"probe": _PROBE})) == described, (
+        served: Final = _listed_description(listed, described)
+        assert _fetched(described).fullmatch(cold) and cold != served, (cold, served)
+        assert _echoed_description(second.call(f"{alias}-add", {"probe": _PROBE})) == served, (
             "the end-user header is intentionally not part of the catalog identity: one key, one slot"
         )
         assert tool_calls(peer.drain()) == (), "a blocked probe reached the peer"
diff --git a/tests/integration/mcp/test_mcp_credentials.py b/tests/integration/mcp/test_mcp_credentials.py
index 63f1b0024c0..08dd5234b68 100644
--- a/tests/integration/mcp/test_mcp_credentials.py
+++ b/tests/integration/mcp/test_mcp_credentials.py
@@ -448,7 +448,6 @@ def test_deprecated_string_x_mcp_auth_lists_a_byok_server_for_a_key_without_a_us
 
 _PROBE: Final = "catalog-probe"
 _ECHO: Final = "catalog-echo"
-_UNLISTED: Final = ""
 _GUARDRAIL_CODE: Final = (
     "def apply_guardrail(inputs, request_data, input_type):\n"
     f'    if "{_PROBE}" not in list(inputs.get("texts") or []):\n'
@@ -488,6 +487,28 @@ _PROBE_ARGUMENTS: Final = {"probe": _PROBE}
 _HEADERS: Final = TypeAdapter(dict[str, str])
 
 
+def _add_described_anew_per_fetch_for(header: str) -> ScriptedTool:
+    return ScriptedTool(
+        "add",
+        lambda _: text_result("3"),
+        description=lambda headers: f"Adds for {headers.get(header, 'nobody')} at {uuid.uuid4().hex}",
+    )
+
+
+def _fetched_for(credential: str) -> re.Pattern[str]:
+    return re.compile(rf"Adds for {re.escape(credential)} at [0-9a-f]{{32}}")
+
+
+def _listed_description(listing: Outcome, credential: str) -> str:
+    found: Final = _fetched_for(credential).search(listing.raw)
+    assert found is not None, listing.raw
+    return found.group(0)
+
+
+def _sent_header(request: dict[str, object], name: str) -> str | None:
+    return _HEADERS.validate_python(request["headers"]).get(name)
+
+
 def _store_byok_credential(scenario: Scenario, identity: str, key: str, secret: str) -> None:
     stored: Final = scenario.gateway.client.post(
         f"/v1/mcp/server/{identity}/user-credential", json={"credential": secret}, headers={"x-litellm-api-key": key}
@@ -499,7 +520,7 @@ def _store_byok_credential(scenario: Scenario, identity: str, key: str, secret:
 
 
 def test_rotating_the_credential_drops_the_callers_listing_until_it_lists_again(echo_rig: Gateway) -> None:
-    with mcp_peer() as peer, echo_rig.scenario() as scenario:
+    with scripted_peer(_add_described_anew_per_fetch_for("authorization")) as peer, echo_rig.scenario() as scenario:
         first: Final = "cred-" + uuid.uuid4().hex
         second: Final = "cred-" + uuid.uuid4().hex
         alias: Final = "rot" + uuid.uuid4().hex[:8]
@@ -508,53 +529,62 @@ def test_rotating_the_credential_drops_the_callers_listing_until_it_lists_again(
         )
         key: Final = scenario.key(object_permission={"mcp_servers": [identity]})
         caller: Final = McpCaller(echo_rig, key, "mcp", alias)
-        assert caller.list_tools().ok
-        assert _echoed_description(caller.call(f"{alias}-add", _PROBE_ARGUMENTS)) == "Add two integers"
+        listed: Final = caller.list_tools()
+        assert listed.ok, listed.raw
+        served: Final = _listed_description(listed, f"Bearer {first}")
+        assert _echoed_description(caller.call(f"{alias}-add", _PROBE_ARGUMENTS)) == served
         rotated: Final = echo_rig.request(
             "PUT", "/v1/mcp/server", {"server_id": identity, "credentials": {"auth_value": second}}
         )
         assert rotated.status_code == 202, rotated.text
         eventually(
-            lambda: _echoed_description(caller.call(f"{alias}-add", _PROBE_ARGUMENTS)), lambda seen: seen == _UNLISTED
+            lambda: _echoed_description(caller.call(f"{alias}-add", _PROBE_ARGUMENTS)),
+            lambda seen: _fetched_for(f"Bearer {second}").fullmatch(seen) is not None,
         )
         peer.drain()
-        assert caller.list_tools().ok
-        assert _echoed_description(caller.call(f"{alias}-add", _PROBE_ARGUMENTS)) == "Add two integers"
+        relisting: Final = caller.list_tools()
+        assert relisting.ok, relisting.raw
+        served_again: Final = _listed_description(relisting, f"Bearer {second}")
+        assert _echoed_description(caller.call(f"{alias}-add", _PROBE_ARGUMENTS)) == served_again
         relisted: Final = _listings(peer)
         assert len(relisted) == 1, relisted
-        assert _header(relisted[0], b"authorization") == f"Bearer {second}".encode(), relisted[0]["headers"]
+        assert _sent_header(relisted[0], "authorization") == f"Bearer {second}", relisted[0]["headers"]
 
 
 def test_byok_callers_are_evaluated_against_their_own_listing_and_the_stored_secret_never_keys_the_slot(
     echo_rig: Gateway,
 ) -> None:
-    with mcp_peer() as peer, echo_rig.scenario() as scenario:
+    with scripted_peer(_add_described_anew_per_fetch_for("x-api-key")) as peer, echo_rig.scenario() as scenario:
         alias: Final = "byok" + uuid.uuid4().hex[:8]
         identity: Final = register_mcp(scenario, peer, alias, auth_type="api_key", is_byok=True)
         owner_key: Final = scenario.key(user_id=scenario.user(), object_permission={"mcp_servers": [identity]})
         stranger_key: Final = scenario.key(user_id=scenario.user(), object_permission={"mcp_servers": [identity]})
         owner_secret: Final = "byok-" + uuid.uuid4().hex
+        stranger_secret: Final = "byok-" + uuid.uuid4().hex
         replacement: Final = "byok-" + uuid.uuid4().hex
         _store_byok_credential(scenario, identity, owner_key, owner_secret)
-        _store_byok_credential(scenario, identity, stranger_key, "byok-" + uuid.uuid4().hex)
+        _store_byok_credential(scenario, identity, stranger_key, stranger_secret)
         owner: Final = McpCaller(echo_rig, owner_key, "mcp", alias)
         stranger: Final = McpCaller(echo_rig, stranger_key, "mcp", alias)
         peer.drain()
-        assert owner.list_tools().ok
+        listed: Final = owner.list_tools()
+        assert listed.ok, listed.raw
         listings: Final = _listings(peer)
-        assert [_header(item, b"x-api-key") for item in listings] == [owner_secret.encode()], listings
+        assert [_sent_header(item, "x-api-key") for item in listings] == [owner_secret], listings
+        served: Final = _listed_description(listed, owner_secret)
         own: Final = _echoed_description(owner.call(f"{alias}-add", _PROBE_ARGUMENTS))
         other: Final = _echoed_description(stranger.call(f"{alias}-add", _PROBE_ARGUMENTS))
-        assert (own, other) == ("Add two integers", _UNLISTED), (own, other)
+        assert own == served, (own, served)
+        assert _fetched_for(stranger_secret).fullmatch(other), (other, served)
         _store_byok_credential(scenario, identity, owner_key, replacement)
-        assert _echoed_description(owner.call(f"{alias}-add", _PROBE_ARGUMENTS)) == "Add two integers", (
-            "the slot is keyed by the client-supplied header, never by the stored credential"
-        )
         sent: Final = eventually(
             lambda: (owner.call(f"{alias}-add", ADD).ok, tool_calls(peer.drain())),
-            lambda value: any(_header(call, b"x-api-key") == replacement.encode() for call in value[1]),
+            lambda value: any(_sent_header(call, "x-api-key") == replacement for call in value[1]),
         )
         assert sent[0], sent
+        assert _echoed_description(owner.call(f"{alias}-add", _PROBE_ARGUMENTS)) == served, (
+            "the slot is keyed by the client-supplied header, never by the stored credential"
+        )
 
 
 def test_callers_with_different_server_scoped_auth_headers_are_evaluated_against_their_own_listings(
@@ -585,12 +615,7 @@ def test_callers_with_different_server_scoped_auth_headers_are_evaluated_against
 
 
 def test_deprecated_string_x_mcp_auth_callers_on_a_user_less_key_own_separate_listings(echo_rig: Gateway) -> None:
-    tool: Final = ScriptedTool(
-        "add",
-        lambda _: text_result("3"),
-        description=lambda headers: "Adds for " + headers.get("authorization", "nobody"),
-    )
-    with scripted_peer(tool) as peer, echo_rig.scenario() as scenario:
+    with scripted_peer(_add_described_anew_per_fetch_for("authorization")) as peer, echo_rig.scenario() as scenario:
         alias: Final = "legacy" + uuid.uuid4().hex[:8]
         identity: Final = register_mcp(scenario, peer, alias, auth_type="bearer_token")
         key: Final = scenario.key(object_permission={"mcp_servers": [identity]})
@@ -598,28 +623,33 @@ def test_deprecated_string_x_mcp_auth_callers_on_a_user_less_key_own_separate_li
         second_token: Final = "second-" + uuid.uuid4().hex
         first: Final = McpCaller(echo_rig, key, "mcp", alias, {"x-mcp-auth": f"Bearer {first_token}"})
         second: Final = McpCaller(echo_rig, key, "mcp", alias, {"x-mcp-auth": f"Bearer {second_token}"})
-        assert _echoed_description(first.call(f"{alias}-add", _PROBE_ARGUMENTS)) == _UNLISTED
+        first_cold: Final = _echoed_description(first.call(f"{alias}-add", _PROBE_ARGUMENTS))
+        assert _fetched_for(f"Bearer {first_token}").fullmatch(first_cold), first_cold
         peer.drain()
-        assert first.list_tools().ok
+        first_listing: Final = first.list_tools()
+        assert first_listing.ok, first_listing.raw
         listings: Final = _listings(peer)
         assert len(listings) == 1, listings
-        listed_with: Final = _HEADERS.validate_python(listings[0]["headers"])
-        assert listed_with.get("authorization") == f"Bearer {first_token}", listed_with
+        assert _sent_header(listings[0], "authorization") == f"Bearer {first_token}", listings[0]["headers"]
+        first_served: Final = _listed_description(first_listing, f"Bearer {first_token}")
         warm: Final = eventually(
             lambda: _echoed_description(first.call(f"{alias}-add", _PROBE_ARGUMENTS)),
-            lambda seen: seen != _UNLISTED,
+            lambda seen: seen == first_served,
         )
-        assert warm == f"Adds for Bearer {first_token}", warm
-        assert _echoed_description(second.call(f"{alias}-add", _PROBE_ARGUMENTS)) == _UNLISTED
-        assert second.list_tools().ok
+        assert warm == first_served != first_cold, (first_cold, first_served, warm)
+        second_cold: Final = _echoed_description(second.call(f"{alias}-add", _PROBE_ARGUMENTS))
+        assert _fetched_for(f"Bearer {second_token}").fullmatch(second_cold), (second_cold, first_served)
+        second_listing: Final = second.list_tools()
+        assert second_listing.ok, second_listing.raw
+        second_served: Final = _listed_description(second_listing, f"Bearer {second_token}")
         seen: Final = eventually(
             lambda: (
                 _echoed_description(first.call(f"{alias}-add", _PROBE_ARGUMENTS)),
                 _echoed_description(second.call(f"{alias}-add", _PROBE_ARGUMENTS)),
             ),
-            lambda pair: _UNLISTED not in pair,
+            lambda pair: pair == (first_served, second_served),
         )
-        assert seen == (f"Adds for Bearer {first_token}", f"Adds for Bearer {second_token}"), seen
+        assert seen == (first_served, second_served), seen
         assert tool_calls(peer.drain()) == (), "a blocked probe reached the peer"
 
 
diff --git a/tests/integration/mcp/test_mcp_llm_endpoints.py b/tests/integration/mcp/test_mcp_llm_endpoints.py
index f751d1f8f4c..4b967802893 100644
--- a/tests/integration/mcp/test_mcp_llm_endpoints.py
+++ b/tests/integration/mcp/test_mcp_llm_endpoints.py
@@ -1,5 +1,6 @@
 import asyncio
 import json
+import re
 import uuid
 from collections.abc import Callable, Generator, Iterator, Mapping, Sequence
 from concurrent.futures import ThreadPoolExecutor
@@ -1408,10 +1409,6 @@ REPORT: Final[tuple[str, dict[str, JsonValue]]] = (
     "Write one report",
     {"type": "object", "properties": {"query": {"type": "string"}, "format": {}}},
 )
-COLD: Final[tuple[str, dict[str, JsonValue]]] = (
-    "",
-    {"type": "object", "properties": {}, "additionalProperties": False},
-)
 RELOAD_FAST: Final = {"PROXY_CONFIG_RELOAD_INTERVAL_SECONDS": "5"}
 CALL_ID: Final = "x-litellm-call-id"
 T = TypeVar("T")
@@ -1577,12 +1574,14 @@ def _echoed(value: JsonValue) -> Echo:
 
 
 @contextmanager
-def _bridge_rig(hooked: Hooked, bridge: Bridge) -> Generator[BridgeRig, None, None]:
+def _bridge_rig(
+    hooked: Hooked, bridge: Bridge, lookup_description: str | Callable[[Mapping[str, str]], str] = LOOKUP[0]
+) -> Generator[BridgeRig, None, None]:
     alias: Final = "brg" + uuid.uuid4().hex[:8]
     lookup: Final = ScriptedTool(
         "lookup",
         lambda params: text_result("found:" + json.dumps(JSON_OBJECT.validate_python(params)["arguments"])),
-        description=LOOKUP[0],
+        description=lookup_description,
         input_schema=LOOKUP[1],
     )
     report: Final = ScriptedTool(
@@ -1753,23 +1752,35 @@ def test_bridge_hook_sees_the_definition_of_the_tool_filtered_for_that_request(
         assert _spend_row(key, found.call_id)["status"] == "success"
 
 
+def _fetched_lookup(description: JsonValue) -> bool:
+    return isinstance(description, str) and re.fullmatch(rf"{LOOKUP[0]} [0-9a-f]{{32}}", description) is not None
+
+
 @pytest.mark.parametrize("bridge", BRIDGES)
 def test_direct_call_after_bridge_only_discovery_stays_cold(hooked: Hooked, bridge: Bridge) -> None:
-    with _bridge_rig(hooked, bridge) as rig:
+    with _bridge_rig(hooked, bridge, lambda _: f"{LOOKUP[0]} {uuid.uuid4().hex}") as rig:
         key: Final = _bridge_key(rig)
         workers: Final = _both_workers(rig.hooked.proxy)
         bridged: Final = _ask(rig, key, HOOK_PROBE, [rig.mcp("lookup")], False, "sync")
-        assert _echoed(bridged.text) == _served(LOOKUP), bridged
+        discovered: Final = _echoed(bridged.text)
+        offered: Final = rig.upstream_by_prompt()
+        assert _fetched_lookup(discovered[0]) and discovered[1] == _served(LOOKUP)[1], bridged
+        assert set(offered) == {HOOK_PROBE} and all(
+            definitions == ((rig.tool("lookup"), *discovered),) for definitions in offered[HOOK_PROBE]
+        ), (discovered, offered)
         direct: Final = eventually(
             lambda: tuple(
                 _on_worker(rig.hooked.proxy, lambda client: _direct_probe(rig, key, "lookup", client)) for _ in range(6)
             ),
             lambda seen: frozenset(pid for pid, _ in seen) == workers,
         )
-        assert all(echo == COLD for _, echo in direct) and frozenset(pid for pid, _ in direct) == workers, (
-            direct,
-            workers,
-        )
+        assert (
+            all(
+                _fetched_lookup(description) and description != discovered[0] and parameters == discovered[1]
+                for _, (description, parameters) in direct
+            )
+            and frozenset(pid for pid, _ in direct) == workers
+        ), (direct, discovered, workers)
         assert rig.peer_calls() == ()
 
 
diff --git a/tests/integration/mcp/test_mcp_oauth_flows.py b/tests/integration/mcp/test_mcp_oauth_flows.py
index 90a462472b2..0e174ac37aa 100644
--- a/tests/integration/mcp/test_mcp_oauth_flows.py
+++ b/tests/integration/mcp/test_mcp_oauth_flows.py
@@ -19,7 +19,7 @@ import jwt
 import pytest
 import yaml
 from cryptography.hazmat.primitives.asymmetric.rsa import RSAPrivateKey, generate_private_key
-from integration._support.client import Gateway, eventually, gateway_from_environment
+from integration._support.client import Gateway, Scenario, eventually, gateway_from_environment
 from integration._support.database import read_rows
 from integration._support.mcp import (
     ENTRY_POINTS,
@@ -786,13 +786,20 @@ def test_dcr_bridge_relays_client_registration_and_advertises_gateway_endpoints(
         assert "S256" in issuer.json()["code_challenge_methods_supported"]
 
 
-def _ui_session_cookie(gateway: Gateway, user_id: str) -> dict[str, str]:
-    claims: Final = {"user_id": user_id, "login_method": "username_password", "exp": int(time.time()) + 600}
-    return {"token": jwt.encode(claims, gateway.key, algorithm="HS256")}
+def _ui_session_cookie(scenario: Scenario, user_id: str) -> dict[str, str]:
+    session_key: Final = scenario.key(user_id=user_id, duration="10m")
+    claims: Final = {
+        "user_id": user_id,
+        "key": session_key,
+        "login_method": "username_password",
+        "exp": int(time.time()) + 600,
+    }
+    return {"token": jwt.encode(claims, scenario.gateway.key, algorithm="HS256")}
 
 
 def _gateway_session_sign_in(
     gateway: Gateway,
+    scenario: Scenario,
     user_id: str,
     resource: str | None = None,
     *,
@@ -806,7 +813,7 @@ def _gateway_session_sign_in(
     assert registered.status_code in (200, 201), registered.text
     client_id: Final = str(_response_object(registered)["client_id"])
     pkce: Final = _Pkce(secrets.token_urlsafe(48))
-    cookies: Final = _ui_session_cookie(gateway, user_id)
+    cookies: Final = _ui_session_cookie(scenario, user_id)
     started: Final = gateway.client.get(
         authorization_endpoint,
         params={
@@ -843,8 +850,8 @@ def _gateway_session_sign_in(
     return _response_object(issued), client_id
 
 
-def _gateway_session_bearer(gateway: Gateway, user_id: str, resource: str | None = None) -> str:
-    issued, _ = _gateway_session_sign_in(gateway, user_id, resource)
+def _gateway_session_bearer(gateway: Gateway, scenario: Scenario, user_id: str, resource: str | None = None) -> str:
+    issued, _ = _gateway_session_sign_in(gateway, scenario, user_id, resource)
     return _issued_token(issued)
 
 
@@ -887,7 +894,7 @@ def test_spec_client_discovers_the_aggregate_authorization_server_and_signs_in_t
         assert registered.status_code in (200, 201), registered.text
         client_id: Final = str(_response_object(registered)["client_id"])
         pkce: Final = _Pkce(secrets.token_urlsafe(48))
-        cookies: Final = _ui_session_cookie(gateway, member)
+        cookies: Final = _ui_session_cookie(scenario, member)
         started: Final = gateway.client.get(
             authorization_endpoint,
             params={
@@ -997,7 +1004,7 @@ def test_per_server_resource_scopes_the_session_bearer_to_the_named_server(gatew
         assert registered.status_code in (200, 201), registered.text
         client_id: Final = str(_response_object(registered)["client_id"])
         pkce: Final = _Pkce(secrets.token_urlsafe(48))
-        cookies: Final = _ui_session_cookie(gateway, member)
+        cookies: Final = _ui_session_cookie(scenario, member)
         started: Final = gateway.client.get(
             "/authorize/mcp-session",
             params={
@@ -1110,7 +1117,7 @@ def test_rotated_session_refresh_token_cannot_be_replayed_after_revocation(gatew
         alias: Final = "rt6" + uuid.uuid4().hex[:10]
         identity: Final = register_mcp(scenario, peer, alias)
         member: Final = scenario.member(scenario.team(object_permission={"mcp_servers": [identity]}))
-        issued, client_id = _gateway_session_sign_in(gateway, member)
+        issued, client_id = _gateway_session_sign_in(gateway, scenario, member)
         r1: Final = str(issued["refresh_token"])
         a1: Final = str(issued["access_token"])
         assert r1.startswith("llm_srefresh_"), r1
@@ -1207,7 +1214,7 @@ def test_lite_logout_revokes_the_cli_refresh_token_sent_the_way_the_cli_sends_it
         assert registered.status_code in (200, 201), registered.text
         client_id: Final = str(_response_object(registered)["client_id"])
         pkce: Final = _Pkce(secrets.token_urlsafe(64))
-        cookies: Final = _ui_session_cookie(gateway, member)
+        cookies: Final = _ui_session_cookie(scenario, member)
         consent: Final = gateway.client.get(
             str(contract["authorization_endpoint"]),
             params={
@@ -1362,6 +1369,7 @@ def test_gateway_session_bearer_of_a_team_member_is_served_the_team_toolset_on_i
         token_endpoint: Final = str(authorization_body["token_endpoint"])
         issued, client_id = _gateway_session_sign_in(
             gateway,
+            scenario,
             member,
             resource,
             registration_endpoint=str(authorization_body["registration_endpoint"]),
@@ -1413,7 +1421,7 @@ def test_resource_scoped_session_bearer_opens_a_team_toolset_inside_its_server_a
         inside_id: Final = create_toolset(scenario, ((inside_server, "add"),), toolset_name=inside_name)
         outside_id: Final = create_toolset(scenario, ((outside_server, "add"),), toolset_name=outside_name)
         member: Final = scenario.member(scenario.team(object_permission={"mcp_toolsets": [inside_id, outside_id]}))
-        bearer: Final = _gateway_session_bearer(gateway, member, resource=f"{_base(gateway)}/{inside}/mcp")
+        bearer: Final = _gateway_session_bearer(gateway, scenario, member, resource=f"{_base(gateway)}/{inside}/mcp")
         assert bearer.startswith("llm_session_"), bearer[:16]
         listed: Final = _toolset_rpc(gateway, bearer, inside_name, "tools/list", {})
         assert listed.tools == (f"{inside}-add",), listed.raw
@@ -1430,7 +1438,6 @@ def test_resource_scoped_session_bearer_opens_a_team_toolset_inside_its_server_a
 
 _PROBE: Final = "catalog-probe"
 _ECHO: Final = "catalog-echo"
-_UNLISTED: Final = ""
 _GUARDRAIL_CODE: Final = (
     "def apply_guardrail(inputs, request_data, input_type):\n"
     f'    if "{_PROBE}" not in list(inputs.get("texts") or []):\n'
@@ -1466,8 +1473,22 @@ def _echoed_description(outcome: Outcome) -> str:
     return found.group(1)
 
 
+def _add_described_for_the_exchanged_subject(auth: AuthorizationServer) -> ScriptedTool:
+    def described(headers: Mapping[str, str]) -> str:
+        token: Final = headers.get("authorization", "").removeprefix("Bearer ")
+        with auth.lock:
+            grant: Final = auth.access_tokens.get(token, {})
+        return "Add for " + grant.get("subject", "nobody")
+
+    return ScriptedTool("add", lambda _: text_result("must not run"), description=described)
+
+
 def test_token_exchange_callers_with_different_subject_tokens_own_separate_listings(echo_rig: Gateway) -> None:
-    with mcp_peer() as peer, oauth_server() as auth, echo_rig.scenario() as scenario:
+    with (
+        oauth_server() as auth,
+        scripted_peer(_add_described_for_the_exchanged_subject(auth)) as peer,
+        echo_rig.scenario() as scenario,
+    ):
         alias: Final = "te" + uuid.uuid4().hex[:8]
         identity: Final = register_mcp(
             scenario,
@@ -1479,19 +1500,20 @@ def test_token_exchange_callers_with_different_subject_tokens_own_separate_listi
         )
         key: Final = scenario.key(object_permission={"mcp_servers": [identity]})
         first_subject: Final = "subject-" + uuid.uuid4().hex
+        second_subject: Final = "subject-" + uuid.uuid4().hex
         first: Final = McpCaller(echo_rig, key, "mcp", alias, {"Authorization": f"Bearer {first_subject}"})
-        second: Final = McpCaller(echo_rig, key, "mcp", alias, {"Authorization": "Bearer subject-" + uuid.uuid4().hex})
+        second: Final = McpCaller(echo_rig, key, "mcp", alias, {"Authorization": f"Bearer {second_subject}"})
         auth.drain()
         assert first.list_tools().ok
         assert [request["subject_token"] for request in auth.token_requests()] == [first_subject]
         probe: Final = {"probe": _PROBE}
         own: Final = _echoed_description(first.call(f"{alias}-add", probe))
         other: Final = _echoed_description(second.call(f"{alias}-add", probe))
-        assert (own, other) == ("Add two integers", _UNLISTED), (
+        assert (own, other) == (f"Add for exchanged:{first_subject}", f"Add for exchanged:{second_subject}"), (
             "the caller bearer is part of the identity on a token-exchange server: one subject, one slot"
         )
         assert second.list_tools().ok
-        assert _echoed_description(second.call(f"{alias}-add", probe)) == "Add two integers"
+        assert _echoed_description(second.call(f"{alias}-add", probe)) == f"Add for exchanged:{second_subject}"
         assert tool_calls(peer.drain()) == (), "a blocked probe reached the peer"
 
 
diff --git a/tests/integration/mcp/test_mcp_resilience.py b/tests/integration/mcp/test_mcp_resilience.py
index c335012d787..145994f1b29 100644
--- a/tests/integration/mcp/test_mcp_resilience.py
+++ b/tests/integration/mcp/test_mcp_resilience.py
@@ -1,4 +1,5 @@
 import itertools
+import re
 import uuid
 from collections.abc import Mapping
 from concurrent.futures import ThreadPoolExecutor
@@ -37,6 +38,7 @@ pytestmark: Final = pytest.mark.shares_worker_state
 _ECHO: Final = "catalog-echo:"
 _PROBE: Final = "catalog-probe"
 _DESCRIPTION: Final = "Look up one record"
+_FETCHED: Final = re.compile(rf"{_DESCRIPTION} [0-9a-f]{{32}}")
 _GUARDRAIL_CODE: Final = (
     "def apply_guardrail(inputs, request_data, input_type):\n"
     '    texts = list(inputs.get("texts") or [])\n'
@@ -69,6 +71,20 @@ class _RpcReply(BaseModel):
     result: _Result
 
 
+class _ListedTool(BaseModel):
+    name: str
+    description: str | None = None
+
+
+class _ListedTools(BaseModel):
+    tools: tuple[_ListedTool, ...]
+
+
+class _ListReply(BaseModel):
+    id: int
+    result: _ListedTools
+
+
 class _SessionsReport(BaseModel):
     worker_pid: int
 
@@ -207,9 +223,13 @@ def test_peer_restart_on_the_same_url_is_picked_up_without_gateway_restart(gatew
             assert len(tool_calls(replacement.drain())) >= 1
 
 
-def _rpc_reply(raw: str) -> _RpcReply:
+def _rpc_payload(raw: str) -> str:
     data: Final = tuple(line[5:].strip() for line in raw.splitlines() if line.startswith("data:"))
-    return _RpcReply.model_validate_json(data[-1] if data else raw)
+    return data[-1] if data else raw
+
+
+def _rpc_reply(raw: str) -> _RpcReply:
+    return _RpcReply.model_validate_json(_rpc_payload(raw))
 
 
 def _call_params(call: Mapping[str, object]) -> Mapping[str, object]:
@@ -220,9 +240,18 @@ def _call_nonce(call: Mapping[str, object]) -> str:
     return _STRINGS.validate_python(_call_params(call)["arguments"])["nonce"]
 
 
-def _listed(caller: McpCaller, name: str) -> None:
+def _listed(caller: McpCaller, name: str) -> Outcome:
     listing: Final = eventually(caller.list_tools, lambda outcome: name in outcome.tools, seconds=45)
     assert listing.error is None, (caller.gateway.client.base_url, listing.raw)
+    return listing
+
+
+def _listed_description(caller: McpCaller, name: str) -> str:
+    listing: Final = _listed(caller, name)
+    tools: Final = _ListReply.model_validate_json(_rpc_payload(listing.raw)).result.tools
+    described: Final = next(tool.description for tool in tools if tool.name == name)
+    assert described is not None and _FETCHED.fullmatch(described), listing.raw
+    return described
 
 
 @dataclass(frozen=True, slots=True)
@@ -251,15 +280,18 @@ def _probed_description(worker: _Worker, name: str) -> str | None:
     return _ECHOED.validate_json(carrier.split(_ECHO, 1)[1])["description"]
 
 
-def _catalog_is_cold(worker: _Worker, name: str) -> bool:
-    return not _probed_description(worker, name)
+def _catalog_is_cold(worker: _Worker, name: str, listings: tuple[str, ...]) -> bool:
+    probed: Final = _probed_description(worker, name)
+    return probed is not None and _FETCHED.fullmatch(probed) is not None and probed not in listings
 
 
 @pytest.mark.timeout(600)
 def test_worker_restart_cools_its_listed_catalog_while_the_sibling_worker_keeps_serving(
     gateway: Gateway, echo_config: Path, tmp_path: Path
 ) -> None:
-    tool: Final = ScriptedTool("lookup", lambda _: text_result("found"), description=_DESCRIPTION)
+    tool: Final = ScriptedTool(
+        "lookup", lambda _: text_result("found"), description=lambda _: f"{_DESCRIPTION} {uuid.uuid4().hex}"
+    )
     with scripted_peer(tool) as peer, gateway.scenario() as scenario:
         alias: Final = "cold" + uuid.uuid4().hex[:8]
         identity: Final = register_mcp(scenario, peer, alias)
@@ -272,21 +304,23 @@ def test_worker_restart_cools_its_listed_catalog_while_the_sibling_worker_keeps_
                 assert first.pid != sibling.pid
                 _served(first, name, "first-unlisted")
                 _served(sibling, name, "sibling-unlisted")
-                assert _catalog_is_cold(first, name) and _catalog_is_cold(sibling, name)
-                _listed(first.caller, name)
-                assert _probed_description(first, name) == _DESCRIPTION
-                assert _catalog_is_cold(sibling, name), "a listing on one worker warmed its sibling"
-                _listed(sibling.caller, name)
-                assert _probed_description(sibling, name) == _DESCRIPTION
+                assert _catalog_is_cold(first, name, ()) and _catalog_is_cold(sibling, name, ())
+                first_listing: Final = _listed_description(first.caller, name)
+                assert _probed_description(first, name) == first_listing
+                assert _catalog_is_cold(sibling, name, (first_listing,)), "a listing on one worker warmed its sibling"
+                sibling_listing: Final = _listed_description(sibling.caller, name)
+                assert _probed_description(sibling, name) == sibling_listing
             _served(sibling, name, "sibling-alone")
             with owned_proxy_process(gateway, tmp_path / "restarted", {}, config=echo_config) as restarted_proxy:
                 restarted: Final = _worker(restarted_proxy.gateway, key, alias)
                 assert restarted.pid not in (first.pid, sibling.pid)
-                assert _catalog_is_cold(restarted, name), "a restarted worker kept the old process's catalog"
-                assert _probed_description(sibling, name) == _DESCRIPTION
+                assert _catalog_is_cold(restarted, name, (first_listing, sibling_listing)), (
+                    "a restarted worker kept the old process's catalog"
+                )
+                assert _probed_description(sibling, name) == sibling_listing
                 _served(restarted, name, "restarted-unlisted")
-                _listed(restarted.caller, name)
-                assert _probed_description(restarted, name) == _DESCRIPTION
+                restarted_listing: Final = _listed_description(restarted.caller, name)
+                assert _probed_description(restarted, name) == restarted_listing
         calls: Final = tool_calls(peer.drain())
         assert [_call_nonce(call) for call in calls] == [
             "first-unlisted",
diff --git a/tests/integration/mcp/test_mcp_toolsets.py b/tests/integration/mcp/test_mcp_toolsets.py
index f275bd0e427..fc040810970 100644
--- a/tests/integration/mcp/test_mcp_toolsets.py
+++ b/tests/integration/mcp/test_mcp_toolsets.py
@@ -538,7 +538,6 @@ def test_owned_proxy_loss_retains_exit_status_and_first_http_failure(gateway: Ga
 
 _PROBE: Final = "catalog-probe"
 _ECHO: Final = "catalog-echo"
-_UNLISTED: Final = ""
 _GUARDRAIL_CODE: Final = (
     "def apply_guardrail(inputs, request_data, input_type):\n"
     f'    if "{_PROBE}" not in list(inputs.get("texts") or []):\n'
@@ -574,9 +573,21 @@ def _echoed_description(outcome: Outcome) -> str:
     return found.group(1)
 
 
+def _fetched(described: str) -> re.Pattern[str]:
+    return re.compile(rf"{re.escape(described)} [0-9a-f]{{32}}")
+
+
+def _listed_description(listing: Outcome, described: str) -> str:
+    found: Final = _fetched(described).search(listing.raw)
+    assert found is not None, listing.raw
+    return found.group(0)
+
+
 def test_a_team_keys_toolset_route_listing_feeds_its_own_calls_but_not_a_team_mates(echo_rig: Gateway) -> None:
     described: Final = "Adds for the team " + uuid.uuid4().hex[:8]
-    tool: Final = ScriptedTool("add", lambda _: text_result("9"), description=described)
+    tool: Final = ScriptedTool(
+        "add", lambda _: text_result("9"), description=lambda _: f"{described} {uuid.uuid4().hex}"
+    )
     with scripted_peer(tool) as peer, echo_rig.scenario() as scenario:
         alias: Final = "lit6029echo" + uuid.uuid4().hex[:6]
         server_id: Final = register_mcp(scenario, peer, alias)
@@ -587,10 +598,14 @@ def test_a_team_keys_toolset_route_listing_feeds_its_own_calls_but_not_a_team_ma
         _assert_team_grants_only(echo_rig, team_id, key, granted_id)
         listed: Final = _toolset_rpc(echo_rig, _bearer(key), granted_name, "tools/list", {})
         assert listed.ok and listed.tools == (f"{alias}-add",), listed.raw
+        served: Final = _listed_description(listed, described)
         probe: Final[dict[str, object]] = {"name": f"{alias}-add", "arguments": {"probe": _PROBE}}
         own: Final = _echoed_description(_toolset_rpc(echo_rig, _bearer(key), granted_name, "tools/call", probe))
         mate: Final = _echoed_description(_toolset_rpc(echo_rig, _bearer(team_mate), granted_name, "tools/call", probe))
-        assert (own, mate) == (described, _UNLISTED), (
-            "the slot is keyed by the hashed key, so a team-mate that never listed is handed nothing"
+        assert own == served, (own, served)
+        assert _fetched(described).fullmatch(mate) and mate != served, (
+            "the slot is keyed by the hashed key, so a team-mate that never listed is not handed the key's listing",
+            mate,
+            served,
         )
         assert tool_calls(peer.drain()) == (), "a blocked probe reached the peer"
diff --git a/tests/integration/observability/test_guardrail_effects.py b/tests/integration/observability/test_guardrail_effects.py
index 537eb2bbf69..4f5c04bddcc 100644
--- a/tests/integration/observability/test_guardrail_effects.py
+++ b/tests/integration/observability/test_guardrail_effects.py
@@ -1085,7 +1085,8 @@ def test_request_selected_mcp_guardrail_blocks_direct_and_virtual_calls(gateway:
         mcp_peer() as peer,
         candidate.scenario() as scenario,
     ):
-        identity = register_mcp(scenario, peer, "guardrail" + uuid.uuid4().hex)
+        server = "guardrail" + uuid.uuid4().hex
+        identity = register_mcp(scenario, peer, server)
         permission = {"mcp_servers": [identity], "mcp_tool_search_enabled": True}
         key = scenario.key(object_permission=permission)
         key_selected = scenario.key(object_permission=permission, guardrails=[guardrail])
@@ -1116,8 +1117,10 @@ def test_request_selected_mcp_guardrail_blocks_direct_and_virtual_calls(gateway:
                 )
                 calls = tuple(item for item in peer.drain() if item["body"].get("method") == "tools/call")
                 if expected is None:
-                    assert response.status_code == 400, response.text
-                    assert "integration resolved add denied" in response.text, response.text
+                    assert response.status_code == 403, response.text
+                    assert response.json() == {
+                        "detail": {"error": f"Tool add is unavailable after MCP guardrail checks for server {server}."}
+                    }, response.text
                     assert calls == (), "pre-call denial must prevent upstream execution"
                 else:
                     assert response.status_code == 200, response.text
diff --git a/tests/integration/providers/test_databricks_decisions_wire.py b/tests/integration/providers/test_databricks_decisions_wire.py
index f105092e9a3..7b54895adaf 100644
--- a/tests/integration/providers/test_databricks_decisions_wire.py
+++ b/tests/integration/providers/test_databricks_decisions_wire.py
@@ -1,3 +1,4 @@
+import json
 import threading
 import time
 import uuid
@@ -67,7 +68,12 @@ _NON_BARE_NAMES: Final[tuple[tuple[str, str], ...]] = (
 )
 _DATABRICKS_ERRORS: Final[tuple[tuple[str, str, int, dict[str, JsonValue]], ...]] = (
     (_MODEL, _ROUTE, 400, {"error_code": "BAD_REQUEST", "message": "Invalid input: questions must not be empty"}),
-    (_MODEL, _ROUTE, 403, {"error_code": "PERMISSION_DENIED", "message": "ai_decide is not enabled for this workspace"}),
+    (
+        _MODEL,
+        _ROUTE,
+        403,
+        {"error_code": "PERMISSION_DENIED", "message": "ai_decide is not enabled for this workspace"},
+    ),
     (
         _ENDPOINT,
         f"/{_ENDPOINT}/invocations",
@@ -89,6 +95,15 @@ def _number(value: JsonValue) -> float:
     return float(value)
 
 
+def _endpoint_spend() -> float:
+    prices: Final = object_value(
+        json.loads(Path("model_prices_and_context_window.json").read_text())[f"databricks/{_ENDPOINT}"]
+    )
+    return _number(_USAGE["input_tokens"]) * _number(prices["input_cost_per_token"]) + _number(
+        _USAGE["output_tokens"]
+    ) * _number(prices["output_cost_per_token"])
+
+
 def _register(scenario: Scenario, body: dict[str, JsonValue], *, status: int = 200) -> ScenarioHandle:
     handle: Final = register_scenario(
         f"databricks-{uuid.uuid4().hex[:12]}", JsonResponse(content_type="application/json", body=body, status=status)
@@ -300,9 +315,14 @@ def test_an_openai_format_request_at_v1_decisions_reaches_the_serving_endpoint_a
             row["model_group"],
             row["api_base"],
             _number(row["spend"]),
-        ) == ("success", "adecisions", "databricks", deployment, f"{handle.api_base()}/{_ENDPOINT}/invocations", 0.0), (
-            row
-        )
+        ) == (
+            "success",
+            "adecisions",
+            "databricks",
+            deployment,
+            f"{handle.api_base()}/{_ENDPOINT}/invocations",
+            pytest.approx(_endpoint_spend(), rel=1e-9),
+        ), row
 
 
 def test_an_openai_format_request_at_v1_decisions_reaches_the_ai_decide_route_without_a_model_field(
@@ -381,6 +401,7 @@ def test_a_noul_probability_reaches_the_caller_as_the_system_one_noul_answer(gat
 
 
 def test_a_wildcard_deployment_sends_each_request_to_the_named_endpoint_and_bills_each_once(gateway: Gateway) -> None:
+    expected_spend: Final = _endpoint_spend()
     with gateway.scenario() as scenario:
         handle: Final = _register(scenario, _ANSWER_BODY)
         prefix: Final = _wildcard_deployment(scenario, api_base=handle.api_base(), api_key=_API_KEY)
@@ -408,9 +429,14 @@ def test_a_wildcard_deployment_sends_each_request_to_the_named_endpoint_and_bill
                 row["model_group"],
                 row["api_base"],
                 _number(row["spend"]),
-            ) == ("success", "asystemone", "databricks", model, f"{handle.api_base()}/{_ENDPOINT}/invocations", 0.0), (
-                row
-            )
+            ) == (
+                "success",
+                "asystemone",
+                "databricks",
+                model,
+                f"{handle.api_base()}/{_ENDPOINT}/invocations",
+                pytest.approx(expected_spend, rel=1e-9),
+            ), row
 
 
 def test_a_wildcard_deployment_sends_each_ai_decide_request_to_the_route_and_logs_each_once(gateway: Gateway) -> None:
diff --git a/tests/integration/providers/test_decisions_openai_format_wire.py b/tests/integration/providers/test_decisions_openai_format_wire.py
index 2f82561d1b1..0c40fe06199 100644
--- a/tests/integration/providers/test_decisions_openai_format_wire.py
+++ b/tests/integration/providers/test_decisions_openai_format_wire.py
@@ -187,7 +187,16 @@ _PROVIDERS: Final = (
         "cloudflare/@cf/cloudflare/clef",
         takes_images=True,
     ),
-    _Provider("hosted_vllm", "hosted_vllm/Qwen/Qwen3-0.6B", "/v1/systemone", "Qwen/Qwen3-0.6B", None, False, None),
+    _Provider(
+        "hosted_vllm",
+        "hosted_vllm/Qwen/Qwen3-0.6B",
+        "/v1/systemone",
+        "Qwen/Qwen3-0.6B",
+        None,
+        False,
+        None,
+        takes_images=True,
+    ),
     _Provider(
         "databricks",
         "databricks/ai_decide",
@@ -205,7 +214,7 @@ _PROVIDERS: Final = (
         "databricks-openjev-qwen35-4b",
         _API_KEY,
         False,
-        None,
+        "databricks/databricks-openjev-qwen35-4b",
     ),
     _Provider(
         "azure_ai", "azure_ai/decision-1", "/providers/microsoft/v1/systemone", "decision-1", _API_KEY, False, None
diff --git a/tests/integration/providers/test_idle_upstream_keepalive_edges_wire.py b/tests/integration/providers/test_idle_upstream_keepalive_edges_wire.py
index fc27f0f6e1e..a5628d7764c 100644
--- a/tests/integration/providers/test_idle_upstream_keepalive_edges_wire.py
+++ b/tests/integration/providers/test_idle_upstream_keepalive_edges_wire.py
@@ -33,15 +33,14 @@ pytestmark: Final = pytest.mark.timeout(int(2 * graceful_stop_seconds() + 120 +
 
 REUSE_ROWS: Final = (
     Row("reuse-chat-openai-sdk-sync", openai_deployment, chat_sdk),
-    Row("reuse-messages-rust-anthropic-sdk-sync", anthropic_deployment, messages_sdk, rust=True),
+    Row("reuse-messages-anthropic-sdk-sync", anthropic_deployment, messages_sdk),
 )
 QUIET_ROWS: Final = (
     Row("quiet-chat-openai-sdk-async-stream", openai_deployment, chat_sdk_stream, respond=quiet_stream_answer),
     Row(
-        "quiet-messages-rust-anthropic-sdk-async-stream",
+        "quiet-messages-anthropic-sdk-async-stream",
         anthropic_deployment,
         messages_sdk_stream,
-        rust=True,
         respond=quiet_stream_answer,
     ),
 )
diff --git a/tests/integration/providers/test_idle_upstream_keepalive_override_wire.py b/tests/integration/providers/test_idle_upstream_keepalive_override_wire.py
index 8b5a69d729e..ddcf1ed7741 100644
--- a/tests/integration/providers/test_idle_upstream_keepalive_override_wire.py
+++ b/tests/integration/providers/test_idle_upstream_keepalive_override_wire.py
@@ -27,7 +27,7 @@ pytestmark: Final = pytest.mark.timeout(int(2 * graceful_stop_seconds() + 120 +
 
 ROWS: Final = (
     Row("override-chat-openai-sdk-sync", openai_deployment, chat_sdk),
-    Row("override-messages-rust-anthropic-sdk-sync", anthropic_deployment, messages_sdk, rust=True),
+    Row("override-messages-anthropic-sdk-sync", anthropic_deployment, messages_sdk),
     Row("override-ocr-mistral-rust-bridge", ocr_deployment, ocr, rust=True),
 )
 
diff --git a/tests/integration/providers/test_idle_upstream_reconnect_rust_messages_wire.py b/tests/integration/providers/test_idle_upstream_reconnect_rust_messages_wire.py
deleted file mode 100644
index 8d92fb609e9..00000000000
--- a/tests/integration/providers/test_idle_upstream_reconnect_rust_messages_wire.py
+++ /dev/null
@@ -1,56 +0,0 @@
-from collections.abc import Iterator
-from typing import Final
-
-import pytest
-from integration._support.client import Gateway
-from integration._support.idle_gateway import Drop
-from integration._support.process import graceful_stop_seconds
-from integration.providers._idle_upstream_support import (
-    IDLE_SECONDS,
-    Row,
-    StalePools,
-    answered,
-    anthropic_deployment,
-    expected_answers,
-    landed,
-    landed_once_undropped,
-    messages_sdk,
-    messages_sdk_stream,
-    served_by_rust,
-    spend_statuses,
-    stale_pools,
-)
-
-pytestmark: Final = pytest.mark.timeout(int(2 * graceful_stop_seconds() + 120 + IDLE_SECONDS))
-
-ROWS: Final = (
-    Row("messages-rust-anthropic-sdk-sync", anthropic_deployment, messages_sdk, rust=True),
-    Row("messages-rust-anthropic-sdk-async-stream", anthropic_deployment, messages_sdk_stream, rust=True),
-    Row(
-        "messages-rust-upstream-resets-stale-connection",
-        anthropic_deployment,
-        messages_sdk,
-        drop=Drop.RESET,
-        rust=True,
-    ),
-)
-
-
-@pytest.fixture(scope="module")
-def pools(session_gateway: Gateway, tmp_path_factory: pytest.TempPathFactory) -> Iterator[StalePools]:
-    with stale_pools(
-        session_gateway, tmp_path_factory.mktemp("idle-upstream-rust"), ROWS, {"LITELLM_RUST": "1"}
-    ) as value:
-        yield value
-
-
-@pytest.mark.parametrize("name", [row.name for row in ROWS])
-def test_rust_messages_after_upstream_idle_past_keepalive_open_a_fresh_connection_on_each_worker(
-    pools: StalePools, name: str
-) -> None:
-    row: Final = pools.rows[name]
-    sents, arrivals = pools.probe(name)
-    assert answered(sents) == expected_answers(row, sents), sents
-    assert landed(arrivals) == landed_once_undropped(sents), arrivals
-    assert served_by_rust(sents) == (row.rust,) * len(sents), sents
-    assert spend_statuses(sents) == ("success",) * len(sents), sents
diff --git a/tests/integration/providers/test_inferenceapis_wire.py b/tests/integration/providers/test_inferenceapis_wire.py
index 8288c59ad5d..5f400d05957 100644
--- a/tests/integration/providers/test_inferenceapis_wire.py
+++ b/tests/integration/providers/test_inferenceapis_wire.py
@@ -30,6 +30,7 @@ _WEATHER_TOOL: Final[JsonValue] = {
     },
 }
 _COST_MAP_PATH: Final = Path(__file__).resolve().parents[3] / "model_prices_and_context_window.json"
+_ENDPOINT_SUPPORT_PATH: Final = Path(__file__).resolve().parents[3] / "litellm/provider_endpoints_support_backup.json"
 _COST_MAP: Final = TypeAdapter(dict[str, dict[str, JsonValue]])
 _JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue])
 _JSON_LIST: Final = TypeAdapter(list[JsonValue])
@@ -512,6 +513,12 @@ def _lists_inferenceapis(endpoint: Mapping[str, JsonValue]) -> bool:
     return any(object_value(provider)["slug"] == "inferenceapis" for provider in providers)
 
 
+def _bundled_inferenceapis_endpoints() -> frozenset[str]:
+    providers: Final = object_value(_JSON_OBJECT.validate_json(_ENDPOINT_SUPPORT_PATH.read_bytes())["providers"])
+    flags: Final = object_value(object_value(providers["inferenceapis"])["endpoints"])
+    return frozenset(key for key, supported in flags.items() if supported)
+
+
 def test_public_routes_list_the_provider_with_its_endpoints_and_fields(gateway: Gateway) -> None:
     providers: Final = gateway.request("GET", "/public/providers")
     assert providers.status_code == 200, providers.text
@@ -538,7 +545,8 @@ def test_public_routes_list_the_provider_with_its_endpoints_and_fields(gateway:
         for endpoint in _JSON_LIST.validate_python(_JSON_OBJECT.validate_json(endpoints_response.content)["endpoints"])
     )
     listed: Final = frozenset(str(endpoint["key"]) for endpoint in endpoints if _lists_inferenceapis(endpoint))
-    assert listed == {"chat_completions", "messages", "responses"}
+    assert {"chat_completions", "messages", "responses"} <= listed, listed
+    assert listed == _bundled_inferenceapis_endpoints()
 
 
 def test_openai_and_anthropic_sdk_clients_reach_the_vendor(gateway: Gateway) -> None:
diff --git a/tests/integration/providers/test_responses_bridge_tool_search_undeferred_wire.py b/tests/integration/providers/test_responses_bridge_tool_search_undeferred_wire.py
index a7fc2b379ea..f6037dd4835 100644
--- a/tests/integration/providers/test_responses_bridge_tool_search_undeferred_wire.py
+++ b/tests/integration/providers/test_responses_bridge_tool_search_undeferred_wire.py
@@ -113,10 +113,8 @@ def _namespace_tool() -> dict[str, JsonValue]:
     }
 
 
-def _search_tools(route: Route, search: Search) -> tuple[dict[str, JsonValue], ...]:
-    if search == "absent":
-        return ()
-    return (_tool_search(execution="client"),) if route == "anthropic" else (_tool_search(),)
+def _search_tools(search: Search) -> tuple[dict[str, JsonValue], ...]:
+    return () if search == "absent" else (_tool_search(),)
 
 
 def _gemini_frame(marker: str) -> dict[str, JsonValue]:
@@ -406,7 +404,6 @@ _UNDEFERRING: Final[tuple[tuple[Route, Search], ...]] = (
 _KEEPING: Final[tuple[tuple[Route, Search], ...]] = (
     ("unlisted_claude", "absent"),
     ("anthropic", "absent"),
-    ("anthropic", "dropped"),
     ("azure_ai", "absent"),
 )
 
@@ -428,7 +425,7 @@ def test_without_a_search_tool_on_the_wire_every_tool_arrives_without_defer_load
     call: Final = _Call(route, stream, uuid.uuid4().hex)
     with wire_server(_vendor(route)) as wire, gateway.scenario() as scenario:
         model: Final = _deployment(scenario, route, wire.url)
-        tools: Final = (*_search_tools(route, search), _function_tool(), _namespace_tool())
+        tools: Final = (*_search_tools(search), _function_tool(), _namespace_tool())
         response: Final = _raw(gateway, _responses_body(model, call.marker, tools, stream))
         _assert_served(response, call.marker, stream)
         request, body = _only_request(wire)
@@ -442,7 +439,7 @@ def test_a_claude_route_sent_no_server_search_tool_keeps_each_tools_own_defer_lo
     call: Final = _Call(route, stream, uuid.uuid4().hex)
     with wire_server(_vendor(route)) as wire, gateway.scenario() as scenario:
         model: Final = _deployment(scenario, route, wire.url)
-        tools: Final = (*_search_tools(route, search), _function_tool(), _namespace_tool(), _loaded_tool())
+        tools: Final = (*_search_tools(search), _function_tool(), _namespace_tool(), _loaded_tool())
         response: Final = _raw(gateway, _responses_body(model, call.marker, tools, stream))
         _assert_served(response, call.marker, stream)
         request, body = _only_request(wire)
@@ -456,6 +453,26 @@ def test_a_claude_route_sent_no_server_search_tool_keeps_each_tools_own_defer_lo
         ], sent
 
 
+@pytest.mark.parametrize("stream", [False, True], ids=["sync", "stream"])
+def test_a_client_search_tool_on_anthropic_goes_out_as_a_function_and_holds_back_the_deferred_tools(
+    gateway: Gateway, stream: bool
+) -> None:
+    marker: Final = uuid.uuid4().hex
+    with wire_server(_claude) as wire, gateway.scenario() as scenario:
+        model: Final = _deployment(scenario, "anthropic", wire.url)
+        tools: Final = (_tool_search(execution="client"), _function_tool(), _namespace_tool(), _loaded_tool())
+        response: Final = _raw(gateway, _responses_body(model, marker, tools, stream))
+        _assert_served(response, marker, stream)
+        request, body = _only_request(wire)
+        _assert_target("anthropic", stream, request, body)
+        sent: Final = _ITEMS.validate_python(body["tools"])
+        assert [(tool["name"], tool.get("defer_loading")) for tool in sent] == [
+            ("tool_search", None),
+            ("send_email", None),
+        ], sent
+        assert "defer_loading" not in request.body.decode(), body
+
+
 def test_a_false_defer_loading_never_reaches_an_openai_compatible_wire_as_a_key(gateway: Gateway) -> None:
     call: Final = _Call("mistral", False, uuid.uuid4().hex)
     with wire_server(_mistral) as wire, gateway.scenario() as scenario:
@@ -682,7 +699,7 @@ def test_a_streamed_answer_closes_its_message_item_with_every_streamed_delta(gat
 
 
 async def _send(client: httpx.AsyncClient, key: str, model: str, call: _Call) -> _Served:
-    tools: Final = (*_search_tools(call.route, "dropped"), _function_tool(), _namespace_tool())
+    tools: Final = (*_search_tools("dropped"), _function_tool(), _namespace_tool())
     body: Final = _responses_body(model, call.marker, tools, call.stream)
     async with client.stream(
         "POST", "/v1/responses", json=body, headers={"Authorization": f"Bearer {key}"}
diff --git a/tests/integration/providers/test_responses_bridge_tool_search_wire.py b/tests/integration/providers/test_responses_bridge_tool_search_wire.py
index 03cd4ce9dc4..57d9655b7d5 100644
--- a/tests/integration/providers/test_responses_bridge_tool_search_wire.py
+++ b/tests/integration/providers/test_responses_bridge_tool_search_wire.py
@@ -835,15 +835,17 @@ def test_a_claude_the_cost_map_does_not_list_gets_the_tool_dropped_on_the_anthro
         _assert_dropped_on_wire(body, _DEFERRED_NAMES)
 
 
-def test_client_execution_drops_the_tool_search_tool(gateway: Gateway) -> None:
+def test_client_execution_offers_tool_search_as_a_function_and_holds_back_the_deferred_tools(gateway: Gateway) -> None:
     marker: Final = uuid.uuid4().hex
     with wire_server(_Vendor().respond) as wire, gateway.scenario() as scenario:
         model: Final = _deployment(scenario, "anthropic", wire.url, gateway.upstream_url)
         tools: Final = (_tool_search(execution="client"), _function_tool(defer_loading=False), _namespace_tool())
         answer: Final = _raw_answer(gateway, _responses_body(model, marker, tools, False))
         assert _output_types(answer.payload) == ["message", "function_call"], answer
-        _, body = _only_request(wire)
-        _assert_dropped_on_wire(body, _DEFERRED_NAMES)
+        request, body = _only_request(wire)
+        assert request.target == "/v1/messages", request.target
+        _assert_dropped_on_wire(body, ("tool_search", "get_weather"))
+        assert "defer_loading" not in request.body.decode(), body
 
 
 @pytest.mark.parametrize("drop_params", [False, True], ids=["refused", "dropped"])
diff --git a/tests/integration/providers/test_rust_anthropic_safeguards_beta_chaos.py b/tests/integration/providers/test_rust_anthropic_safeguards_beta_chaos.py
index f90d68c1a63..4ddfd16f931 100644
--- a/tests/integration/providers/test_rust_anthropic_safeguards_beta_chaos.py
+++ b/tests/integration/providers/test_rust_anthropic_safeguards_beta_chaos.py
@@ -45,7 +45,7 @@ def _clients(stack: ExitStack, base_url: str, count: int) -> tuple[httpx.Client,
 def _post(client: httpx.Client, key: str, marker: str) -> int:
     body: Final = safeguarded(messages_body(ANTHROPIC_DEPLOYMENT, marker), list(SAFEGUARDS))
     response: Final = client.post("/v1/messages", json=body, headers={"Authorization": f"Bearer {key}"})
-    assert response.status_code != 200 or rust_routed(response.headers), dict(response.headers)
+    assert response.status_code != 200 or not rust_routed(response.headers), dict(response.headers)
     return response.status_code
 
 
diff --git a/tests/integration/providers/test_rust_anthropic_safeguards_beta_wire.py b/tests/integration/providers/test_rust_anthropic_safeguards_beta_wire.py
index c13d4da7dbb..2941f8003ee 100644
--- a/tests/integration/providers/test_rust_anthropic_safeguards_beta_wire.py
+++ b/tests/integration/providers/test_rust_anthropic_safeguards_beta_wire.py
@@ -155,7 +155,7 @@ def test_sdk_messages_with_safeguards_go_upstream_with_the_dangerous_tool_use_be
             extra_body={"safeguards": list(SAFEGUARDS), "tools": [dict(BASH_TOOL)]},
         )
     message: Final = raw.parse()
-    assert rust_routed(raw.headers), dict(raw.headers)
+    assert not rust_routed(raw.headers), dict(raw.headers)
     received: Final = json_object(message.model_dump_json())
     assert message.id == f"msg_{marker}", received
     assert received.get("safeguard_results") == list(SAFEGUARD_RESULTS), received
@@ -192,7 +192,7 @@ async def test_sdk_streamed_messages_with_safeguards_go_upstream_with_the_beta(
         ) as stream:
             text: Final = "".join([piece async for piece in stream.text_stream])
             final: Final = await stream.get_final_message()
-            assert rust_routed(stream.response.headers), dict(stream.response.headers)
+            assert not rust_routed(stream.response.headers), dict(stream.response.headers)
     assert (final.id, text) == (f"msg_{marker}", REPLY_TEXT), (final, text)
     sent: Final = one_delivery(rig.wire, marker)
     assert (sent.target, sent.beta, sent.safeguards) == (
@@ -209,7 +209,7 @@ def test_raw_messages_merge_the_safeguards_beta_into_the_client_betas(rig: _Rig,
     body: Final = safeguarded(messages_body(ANTHROPIC_DEPLOYMENT, marker, stream=stream), list(SAFEGUARDS))
     response: Final = _post(rig.gateway, body, beta=CLIENT_BETAS)
     assert response.status_code == 200, response.text
-    assert rust_routed(response.headers), dict(response.headers)
+    assert not rust_routed(response.headers), dict(response.headers)
     assert f"msg_{marker}" in response.text and "safeguard_results" in response.text, response.text
     sent: Final = one_delivery(rig.wire, marker)
     assert (sent.target, sent.beta) == (ANTHROPIC_TARGET, MERGED_BETAS), sent
@@ -221,7 +221,7 @@ def test_null_or_absent_safeguards_add_no_beta(rig: _Rig, extra: Mapping[str, Js
     marker: Final = new_marker()
     response: Final = _post(rig.gateway, {**messages_body(ANTHROPIC_DEPLOYMENT, marker), **extra})
     assert response.status_code == 200, response.text
-    assert rust_routed(response.headers), dict(response.headers)
+    assert not rust_routed(response.headers), dict(response.headers)
     assert f"msg_{marker}" in response.text and "safeguard_results" not in response.text, response.text
     sent: Final = one_delivery(rig.wire, marker)
     assert (sent.beta, sent.safeguards) == (None, None), sent
@@ -253,7 +253,7 @@ def test_malformed_safeguards_still_add_the_beta_and_relay_the_vendor_verdict(
     control: Final = new_marker()
     after: Final = _post(rig.gateway, messages_body(ANTHROPIC_DEPLOYMENT, control))
     assert after.status_code == 200, after.text
-    assert rust_routed(after.headers), dict(after.headers)
+    assert not rust_routed(after.headers), dict(after.headers)
     assert one_delivery(rig.wire, control).beta is None
     assert rig.gateway.request("GET", "/health/liveliness").status_code == 200
 
@@ -276,7 +276,7 @@ def test_duplicated_safeguards_keys_resolve_to_the_last_value(rig: _Rig, last: s
         },
     )
     assert response.status_code == 200, response.text
-    assert rust_routed(response.headers), dict(response.headers)
+    assert not rust_routed(response.headers), dict(response.headers)
     assert ("safeguard_results" in response.text) == (last == "list"), response.text
     sent: Final = one_delivery(rig.wire, marker)
     assert sent.beta == (DANGEROUS_TOOL_USE if last == "list" else None), sent
@@ -304,7 +304,7 @@ def test_vertex_messages_with_safeguards_keep_the_beta_from_the_shared_helper(
     body: Final = safeguarded(messages_body(VERTEX_DEPLOYMENT, marker, stream=stream), list(SAFEGUARDS))
     response: Final = _post(rig.gateway, {**body, "tools": tools})
     assert response.status_code == 200, response.text
-    assert rust_routed(response.headers), dict(response.headers)
+    assert not rust_routed(response.headers), dict(response.headers)
     assert f"msg_{marker}" in response.text and "safeguard_results" in response.text, response.text
     sent: Final = one_delivery(rig.wire, marker)
     assert sent.target == (VERTEX_STREAM_TARGET if stream else VERTEX_TARGET), sent
@@ -326,7 +326,7 @@ def test_deepseek_messages_forward_safeguards_without_adding_a_beta(rig: _Rig, b
         beta=beta,
     )
     assert response.status_code == 200, response.text
-    assert rust_routed(response.headers), dict(response.headers)
+    assert not rust_routed(response.headers), dict(response.headers)
     assert f"msg_{marker}" in response.text, response.text
     sent: Final = one_delivery(rig.wire, marker)
     assert (sent.target, sent.api_key, sent.safeguards) == (DEEPSEEK_TARGET, DEEPSEEK_KEY, list(SAFEGUARDS)), sent
@@ -370,7 +370,7 @@ def _wave(pool: ThreadPoolExecutor, gateway: Gateway, markers: tuple[str, ...],
 def _assert_served(answer: _Answer) -> None:
     assert answer.status == 200, answer
     assert REPLY_TEXT in answer.text, answer
-    assert answer.rust == (answer.path == "/v1/messages"), answer
+    assert not answer.rust, answer
     if answer.path == "/v1/messages":
         assert f"msg_{answer.marker}" in answer.text and "safeguard_results" in answer.text, answer
 
diff --git a/tests/integration/providers/test_vertex_credentials_source_wire.py b/tests/integration/providers/test_vertex_credentials_source_wire.py
index c66bb10f9a9..f3bc86eca19 100644
--- a/tests/integration/providers/test_vertex_credentials_source_wire.py
+++ b/tests/integration/providers/test_vertex_credentials_source_wire.py
@@ -416,7 +416,7 @@ def vertex_claude_peer(request: Request) -> Reply:
 
 
 @pytest.mark.timeout(2 * graceful_stop_seconds() + 120)
-def test_rust_messages_route_loads_a_claude_key_file_and_names_unloadable_credentials(
+def test_rust_enabled_messages_route_loads_a_claude_key_file_and_names_unloadable_credentials(
     gateway: Gateway, tmp_path: Path
 ) -> None:
     token_url: Final = gateway.upstream_url.rstrip("/")
@@ -452,7 +452,7 @@ def test_rust_messages_route_loads_a_claude_key_file_and_names_unloadable_creden
             candidate,
             wire,
             ask(str(tmp_path / "missing.json"), "rust refuses a missing file"),
-            "failed to open application credentials file",
+            FILE_MESSAGE,
             str(tmp_path),
         )
         refused(
diff --git a/tests/integration/spend/golden/daily_activity_team_paginated.json b/tests/integration/spend/golden/daily_activity_team_paginated.json
index 4d91b0da209..18ee30a74e6 100644
--- a/tests/integration/spend/golden/daily_activity_team_paginated.json
+++ b/tests/integration/spend/golden/daily_activity_team_paginated.json
@@ -177,35 +177,6 @@
         "endpoints": {
           "/v1/chat/completions": {
             "api_key_breakdown": {
-              "__ptu_flat_cost__": {
-                "metadata": {
-                  "key_alias": null,
-                  "key_exists": false,
-                  "team_id": null,
-                  "user_email": null,
-                  "user_id": null
-                },
-                "metrics": {
-                  "api_requests": 0,
-                  "autorouter_savings_spend": 0.0,
-                  "cache_creation_input_tokens": 0,
-                  "cache_read_input_tokens": 0,
-                  "completion_tokens": 0,
-                  "compression_saved_tokens": 0,
-                  "compression_savings_spend": 0.0,
-                  "failed_requests": 0,
-                  "flat_cost": 0.0,
-                  "ptu_hours": 0.0,
-                  "gateway_injected_caching_savings_spend": 0.0,
-                  "prompt_caching_savings_spend": 0.0,
-                  "prompt_tokens": 0,
-                  "spend": 1000.0,
-                  "successful_requests": 0,
-                  "timed_requests": 0,
-                  "total_response_time_ms": 0,
-                  "total_tokens": 0
-                }
-              },
               "key-a": {
                 "metadata": {
                   "key_alias": "alias-a",
diff --git a/tests/integration/spend/golden/daily_activity_user_paginated.json b/tests/integration/spend/golden/daily_activity_user_paginated.json
index 00389289ebe..4f7b62b3147 100644
--- a/tests/integration/spend/golden/daily_activity_user_paginated.json
+++ b/tests/integration/spend/golden/daily_activity_user_paginated.json
@@ -177,35 +177,6 @@
         "endpoints": {
           "/v1/chat/completions": {
             "api_key_breakdown": {
-              "__ptu_flat_cost__": {
-                "metadata": {
-                  "key_alias": null,
-                  "key_exists": false,
-                  "team_id": null,
-                  "user_email": null,
-                  "user_id": null
-                },
-                "metrics": {
-                  "api_requests": 0,
-                  "autorouter_savings_spend": 0.0,
-                  "cache_creation_input_tokens": 0,
-                  "cache_read_input_tokens": 0,
-                  "completion_tokens": 0,
-                  "compression_saved_tokens": 0,
-                  "compression_savings_spend": 0.0,
-                  "failed_requests": 0,
-                  "flat_cost": 0.0,
-                  "ptu_hours": 0.0,
-                  "gateway_injected_caching_savings_spend": 0.0,
-                  "prompt_caching_savings_spend": 0.0,
-                  "prompt_tokens": 0,
-                  "spend": 1000.0,
-                  "successful_requests": 0,
-                  "timed_requests": 0,
-                  "total_response_time_ms": 0,
-                  "total_tokens": 0
-                }
-              },
               "key-a": {
                 "metadata": {
                   "key_alias": "alias-a",
diff --git a/tests/integration/spend/test_shutdown_flush.py b/tests/integration/spend/test_shutdown_flush.py
index 5c2e9cf98da..174403d1873 100644
--- a/tests/integration/spend/test_shutdown_flush.py
+++ b/tests/integration/spend/test_shutdown_flush.py
@@ -16,7 +16,7 @@ import httpx
 import psycopg
 import pytest
 import yaml
-from integration._support.client import Gateway, delete_key_if_present, eventually, string_value
+from integration._support.client import Gateway, delete_key_if_present, eventually, object_value, string_value
 from integration._support.database import read_rows
 from integration._support.process import OwnedProxy, owned_proxy_process
 from integration._support.wire import Reply, Request, wire_server
@@ -25,7 +25,6 @@ from psycopg import sql
 REQUESTS_WHILE_BLOCKED: Final = 6
 CANCEL_LOG_LINE: Final = "in-flight scheduled job(s) for shutdown"
 BATCH_DRAINED_LOG_LINE: Final = f"flushed {REQUESTS_WHILE_BLOCKED} daily spend update items from in-memory queue"
-MODEL_INSERT_ARRIVED_LOG_LINE: Final = "path=/model/new"
 COMMIT_DELAY_SECONDS: Final = 15
 BURST_REQUESTS: Final = 30
 
@@ -356,8 +355,15 @@ class _Shutdown:
             for outcome in pool.map(lambda _: self.chat(), range(requests)):
                 assert outcome is None
 
-    def logged(self, line: str, times: int = 1) -> bool:
-        return self.owned.log.read_text(errors="replace").count(line) >= times
+    def logged(self, line: str) -> bool:
+        return line in self.owned.log.read_text(errors="replace")
+
+    def other_requests_in_flight(self) -> int:
+        response: Final = self.owned.gateway.request("GET", "/health/backlog")
+        assert response.status_code == 200, response.text
+        in_flight: Final = object_value(response.json())["in_flight_requests"]
+        assert isinstance(in_flight, int)
+        return in_flight - 1
 
     def chat_while_spend_update_is_blocked(self, blocker: psycopg.Connection, table: str) -> None:
         blocker.execute(f'LOCK TABLE "{table}" IN EXCLUSIVE MODE')
@@ -463,7 +469,7 @@ def test_daily_spend_batch_cancelled_while_waiting_for_a_pool_connection_is_writ
         eventually(lambda: _waiting_on("LiteLLM_ProxyModelTable"), lambda waiting: waiting == 1, seconds=30)
         shutdown.chat_while_spend_update_is_blocked(memberships, "LiteLLM_TeamMembership")
         second: Final = shutdown.start_blocked_model_insert()
-        eventually(lambda: shutdown.logged(MODEL_INSERT_ARRIVED_LOG_LINE, times=2), lambda seen: seen, seconds=30)
+        eventually(shutdown.other_requests_in_flight, lambda in_flight: in_flight == 2, seconds=30)
         memberships.rollback()
         eventually(lambda: _waiting_on("LiteLLM_ProxyModelTable"), lambda waiting: waiting == 2, seconds=30)
         shutdown.terminate_once(lambda: shutdown.logged(BATCH_DRAINED_LOG_LINE), models.rollback)
diff --git a/tests/integration/translation/chat_completions/bases/openai.py b/tests/integration/translation/chat_completions/bases/openai.py
index 21886e4c774..25d8121cf31 100644
--- a/tests/integration/translation/chat_completions/bases/openai.py
+++ b/tests/integration/translation/chat_completions/bases/openai.py
@@ -1,4 +1,5 @@
 from typing import Final
+from unittest.mock import ANY
 
 from integration._support.client import object_value
 from integration.translation.case import TranslationTestCase
@@ -420,6 +421,7 @@ GPT_6_LUNA_TEST_CASE: Final = TranslationTestCase(
                             "summary": [],
                         }
                     ],
+                    "provider_specific_fields": {"phase": "commentary", "openai_responses_replay": ANY},
                 },
             }
         ],
@@ -630,6 +632,7 @@ GPT_5_4_MINI_TEST_CASE: Final = TranslationTestCase(
                             "type": "function",
                         },
                     ],
+                    "provider_specific_fields": {"phase": "commentary", "openai_responses_replay": ANY},
                 },
             }
         ],
@@ -729,6 +732,7 @@ GPT_6_SOL_TEST_CASE: Final = TranslationTestCase(
                         },
                     ],
                     "content": None,
+                    "provider_specific_fields": {"openai_responses_replay": ANY},
                 },
             }
         ],
@@ -886,6 +890,7 @@ GPT_6_1_SOL_TEST_CASE: Final = TranslationTestCase(
                             "summary": [],
                         }
                     ],
+                    "provider_specific_fields": {"phase": "commentary", "openai_responses_replay": ANY},
                 },
             }
         ],
@@ -1042,6 +1047,7 @@ GPT_6_ASTRA_TEST_CASE: Final = TranslationTestCase(
                             "summary": [],
                         }
                     ],
+                    "provider_specific_fields": {"phase": "commentary", "openai_responses_replay": ANY},
                 },
             }
         ],
@@ -1201,6 +1207,7 @@ GPT_5_6_TERRA_TEST_CASE: Final = TranslationTestCase(
                             "summary": [],
                         }
                     ],
+                    "provider_specific_fields": {"phase": "commentary", "openai_responses_replay": ANY},
                 },
             }
         ],
diff --git a/tests/integration/translation/chat_completions/cross_model/openai.py b/tests/integration/translation/chat_completions/cross_model/openai.py
index 0793daa96b1..46e9277b7cb 100644
--- a/tests/integration/translation/chat_completions/cross_model/openai.py
+++ b/tests/integration/translation/chat_completions/cross_model/openai.py
@@ -1,5 +1,7 @@
+from collections.abc import Mapping
 from dataclasses import replace
 from typing import Final
+from unittest.mock import ANY
 
 from integration._support.client import list_value, object_value
 from integration.translation.case import (
@@ -29,6 +31,30 @@ from integration.translation.templates.openai import (
     OPENAI_RESPONSES_AGENT_REQUEST,
     OPENAI_RESPONSES_AGENT_RESPONSE,
 )
+from pydantic import JsonValue
+
+
+def _without_replay_record(response: Mapping[str, JsonValue]) -> dict[str, JsonValue]:
+    choices: Final = response["choices"]
+    assert isinstance(choices, list) and isinstance(choices[0], dict), choices
+    message: Final = choices[0]["message"]
+    assert isinstance(message, dict), message
+    fields: Final = message["provider_specific_fields"]
+    assert isinstance(fields, dict), fields
+    kept: Final = {key: value for key, value in fields.items() if key != "openai_responses_replay"}
+    return {
+        **response,
+        "choices": [
+            {
+                **choices[0],
+                "message": {
+                    **{key: value for key, value in message.items() if key != "provider_specific_fields"},
+                    **({"provider_specific_fields": kept} if kept else {}),
+                },
+            }
+        ],
+    }
+
 
 GPT_5_MINI_AGENT_FOLLOWUP_TEST_CASE: Final = replace(
     GPT_5_MINI_TEST_CASE,
@@ -559,7 +585,7 @@ GPT_5_4_MINI_AGENT_FOLLOWUP_TEST_CASE: Final = replace(
     GPT_5_4_MINI_TEST_CASE,
     scenario="agent_followup",
     litellm_request=followup_request(
-        GPT_5_4_MINI_TEST_CASE.litellm_request, GPT_5_4_MINI_TEST_CASE.expected_litellm_response
+        GPT_5_4_MINI_TEST_CASE.litellm_request, _without_replay_record(GPT_5_4_MINI_TEST_CASE.expected_litellm_response)
     ),
     expected_provider_endpoint="/responses",
     expected_provider_request={
@@ -570,6 +596,7 @@ GPT_5_4_MINI_AGENT_FOLLOWUP_TEST_CASE: Final = replace(
             {
                 "type": "message",
                 "role": "assistant",
+                "phase": "commentary",
                 "content": [
                     {
                         "type": "output_text",
@@ -679,6 +706,7 @@ GPT_5_4_MINI_AGENT_FOLLOWUP_TEST_CASE: Final = replace(
                     "caller in `src/worker.py` (`remaining = retries_remaining(3, attempts)`), and "
                     "`tests/test_retry.py` failed with `assert 4 == 2` for `retries_remaining(3, 1)`.",
                     "role": "assistant",
+                    "provider_specific_fields": {"phase": "final_answer", "openai_responses_replay": ANY},
                 },
             }
         ],
@@ -1310,6 +1338,7 @@ GPT_5_6_LUNA_AGENT_TEST_CASE: Final = replace(
                         }
                     ],
                     "content": None,
+                    "provider_specific_fields": {"openai_responses_replay": ANY},
                 },
             }
         ],
@@ -1343,7 +1372,8 @@ GPT_5_6_LUNA_AGENT_FOLLOWUP_TEST_CASE: Final = replace(
     GPT_5_6_LUNA_AGENT_TEST_CASE,
     scenario="agent_followup",
     litellm_request=followup_request(
-        GPT_5_6_LUNA_AGENT_TEST_CASE.litellm_request, GPT_5_6_LUNA_AGENT_TEST_CASE.expected_litellm_response
+        GPT_5_6_LUNA_AGENT_TEST_CASE.litellm_request,
+        _without_replay_record(GPT_5_6_LUNA_AGENT_TEST_CASE.expected_litellm_response),
     ),
     expected_provider_endpoint="/responses",
     expected_provider_request={
@@ -1456,6 +1486,7 @@ GPT_5_6_LUNA_AGENT_FOLLOWUP_TEST_CASE: Final = replace(
                     "of 2 and potentially negative values are not clamped. It is called by "
                     "`src/worker.py:12`, and `tests/test_retry.py::test_retries_remaining` failed.",
                     "role": "assistant",
+                    "provider_specific_fields": {"phase": "final_answer", "openai_responses_replay": ANY},
                 },
             }
         ],
@@ -1819,6 +1850,7 @@ GPT_5_6_SOL_AGENT_TEST_CASE: Final = replace(
                         },
                     ],
                     "content": None,
+                    "provider_specific_fields": {"openai_responses_replay": ANY},
                 },
             }
         ],
@@ -1852,7 +1884,8 @@ GPT_5_6_SOL_AGENT_FOLLOWUP_TEST_CASE: Final = replace(
     GPT_5_6_SOL_AGENT_TEST_CASE,
     scenario="agent_followup",
     litellm_request=followup_request(
-        GPT_5_6_SOL_AGENT_TEST_CASE.litellm_request, GPT_5_6_SOL_AGENT_TEST_CASE.expected_litellm_response
+        GPT_5_6_SOL_AGENT_TEST_CASE.litellm_request,
+        _without_replay_record(GPT_5_6_SOL_AGENT_TEST_CASE.expected_litellm_response),
     ),
     expected_provider_endpoint="/responses",
     expected_provider_request={
@@ -1961,6 +1994,7 @@ GPT_5_6_SOL_AGENT_FOLLOWUP_TEST_CASE: Final = replace(
                     "`tests/test_retry.py` failed because `retries_remaining(3, 1)` returned `4` "
                     "instead of `2`.",
                     "role": "assistant",
+                    "provider_specific_fields": {"phase": "final_answer", "openai_responses_replay": ANY},
                 },
             }
         ],
@@ -2278,6 +2312,7 @@ GPT_5_6_TERRA_AGENT_TEST_CASE: Final = replace(
                         },
                     ],
                     "content": None,
+                    "provider_specific_fields": {"openai_responses_replay": ANY},
                 },
             }
         ],
@@ -2311,7 +2346,8 @@ GPT_5_6_TERRA_AGENT_FOLLOWUP_TEST_CASE: Final = replace(
     GPT_5_6_TERRA_AGENT_TEST_CASE,
     scenario="agent_followup",
     litellm_request=followup_request(
-        GPT_5_6_TERRA_AGENT_TEST_CASE.litellm_request, GPT_5_6_TERRA_AGENT_TEST_CASE.expected_litellm_response
+        GPT_5_6_TERRA_AGENT_TEST_CASE.litellm_request,
+        _without_replay_record(GPT_5_6_TERRA_AGENT_TEST_CASE.expected_litellm_response),
     ),
     expected_provider_endpoint="/responses",
     expected_provider_request={
@@ -2420,6 +2456,7 @@ GPT_5_6_TERRA_AGENT_FOLLOWUP_TEST_CASE: Final = replace(
                     "`tests/test_retry.py` fails because `retries_remaining(3, 1)` returns `4` instead "
                     "of `2`.",
                     "role": "assistant",
+                    "provider_specific_fields": {"phase": "final_answer", "openai_responses_replay": ANY},
                 },
             }
         ],
@@ -2737,6 +2774,7 @@ GPT_6_ASTRA_AGENT_TEST_CASE: Final = replace(
                         },
                     ],
                     "content": None,
+                    "provider_specific_fields": {"openai_responses_replay": ANY},
                 },
             }
         ],
@@ -2769,7 +2807,8 @@ GPT_6_ASTRA_AGENT_FOLLOWUP_TEST_CASE: Final = replace(
     GPT_6_ASTRA_AGENT_TEST_CASE,
     scenario="agent_followup",
     litellm_request=followup_request(
-        GPT_6_ASTRA_AGENT_TEST_CASE.litellm_request, GPT_6_ASTRA_AGENT_TEST_CASE.expected_litellm_response
+        GPT_6_ASTRA_AGENT_TEST_CASE.litellm_request,
+        _without_replay_record(GPT_6_ASTRA_AGENT_TEST_CASE.expected_litellm_response),
     ),
     expected_provider_endpoint="/responses",
     expected_provider_request={
@@ -2878,6 +2917,7 @@ GPT_6_ASTRA_AGENT_FOLLOWUP_TEST_CASE: Final = replace(
                     "and `tests/test_retry.py::test_retries_remaining` failed because it returned 4 "
                     "instead of 2.",
                     "role": "assistant",
+                    "provider_specific_fields": {"phase": "final_answer", "openai_responses_replay": ANY},
                 },
             }
         ],
@@ -3194,6 +3234,7 @@ GPT_6_LUNA_AGENT_TEST_CASE: Final = replace(
                         },
                     ],
                     "content": None,
+                    "provider_specific_fields": {"openai_responses_replay": ANY},
                 },
             }
         ],
@@ -3227,7 +3268,8 @@ GPT_6_LUNA_AGENT_FOLLOWUP_TEST_CASE: Final = replace(
     GPT_6_LUNA_AGENT_TEST_CASE,
     scenario="agent_followup",
     litellm_request=followup_request(
-        GPT_6_LUNA_AGENT_TEST_CASE.litellm_request, GPT_6_LUNA_AGENT_TEST_CASE.expected_litellm_response
+        GPT_6_LUNA_AGENT_TEST_CASE.litellm_request,
+        _without_replay_record(GPT_6_LUNA_AGENT_TEST_CASE.expected_litellm_response),
     ),
     expected_provider_endpoint="/responses",
     expected_provider_request={
@@ -3336,6 +3378,7 @@ GPT_6_LUNA_AGENT_FOLLOWUP_TEST_CASE: Final = replace(
                     "`tests/test_retry.py` failed because `retries_remaining(3, 1)` returned 4 instead "
                     "of 2.",
                     "role": "assistant",
+                    "provider_specific_fields": {"phase": "final_answer", "openai_responses_replay": ANY},
                 },
             }
         ],
@@ -3587,7 +3630,7 @@ GPT_6_SOL_AGENT_FOLLOWUP_TEST_CASE: Final = replace(
     GPT_6_SOL_TEST_CASE,
     scenario="agent_followup",
     litellm_request=followup_request(
-        GPT_6_SOL_TEST_CASE.litellm_request, GPT_6_SOL_TEST_CASE.expected_litellm_response
+        GPT_6_SOL_TEST_CASE.litellm_request, _without_replay_record(GPT_6_SOL_TEST_CASE.expected_litellm_response)
     ),
     expected_provider_endpoint="/responses",
     expected_provider_request={
@@ -3694,6 +3737,7 @@ GPT_6_SOL_AGENT_FOLLOWUP_TEST_CASE: Final = replace(
                     "clamping the result at zero; `src/worker.py` calls it. `tests/test_retry.py` "
                     "failed because `retries_remaining(3, 1)` returned 4 instead of 2.",
                     "role": "assistant",
+                    "provider_specific_fields": {"phase": "final_answer", "openai_responses_replay": ANY},
                 },
             }
         ],
@@ -4007,6 +4051,7 @@ GPT_6_1_SOL_AGENT_TEST_CASE: Final = replace(
                         },
                     ],
                     "content": None,
+                    "provider_specific_fields": {"openai_responses_replay": ANY},
                 },
             }
         ],
@@ -4039,7 +4084,8 @@ GPT_6_1_SOL_AGENT_FOLLOWUP_TEST_CASE: Final = replace(
     GPT_6_1_SOL_AGENT_TEST_CASE,
     scenario="agent_followup",
     litellm_request=followup_request(
-        GPT_6_1_SOL_AGENT_TEST_CASE.litellm_request, GPT_6_1_SOL_AGENT_TEST_CASE.expected_litellm_response
+        GPT_6_1_SOL_AGENT_TEST_CASE.litellm_request,
+        _without_replay_record(GPT_6_1_SOL_AGENT_TEST_CASE.expected_litellm_response),
     ),
     expected_provider_endpoint="/responses",
     expected_provider_request={
@@ -4148,6 +4194,7 @@ GPT_6_1_SOL_AGENT_FOLLOWUP_TEST_CASE: Final = replace(
                     "`src/worker.py:12`. `tests/test_retry.py::test_retries_remaining` failed because "
                     "it returned 4 instead of 2; no files were changed.",
                     "role": "assistant",
+                    "provider_specific_fields": {"phase": "final_answer", "openai_responses_replay": ANY},
                 },
             }
         ],