diff --git a/tests/integration/_support/upstream.py b/tests/integration/_support/upstream.py
index 89a28cffb1d..f7af39bcdfb 100644
--- a/tests/integration/_support/upstream.py
+++ b/tests/integration/_support/upstream.py
@@ -24,7 +24,7 @@ from _fake_openai_endpoint_server import (
     moderations,
     triton_embeddings,
 )
-from integration.cost_calculation.cost_map import COST_MAP_ENTRIES
+from integration.cost_calculation.cost_map import COST_MAP_ENTRIES, FALLBACK_GENERALIZATIONS
 from integration.cost_calculation.cost_tracking_case import (
     BinaryResponse,
     EventStreamEvent,
@@ -275,7 +275,7 @@ class Provider:
         return JSONResponse({"deleted": deleted}, status_code=200 if deleted else 404)
 
     async def cost_map(self, _request: Request) -> Response:
-        return JSONResponse(COST_MAP_ENTRIES)
+        return JSONResponse({**COST_MAP_ENTRIES, "fallback_generalizations": FALLBACK_GENERALIZATIONS})
 
     async def oauth_token(self, _request: Request) -> Response:
         return JSONResponse(
diff --git a/tests/integration/cost_calculation/bridge_cost_parity/case.py b/tests/integration/cost_calculation/bridge_cost_parity/case.py
index a3970a156a7..f2f03a69378 100644
--- a/tests/integration/cost_calculation/bridge_cost_parity/case.py
+++ b/tests/integration/cost_calculation/bridge_cost_parity/case.py
@@ -1,7 +1,11 @@
 from dataclasses import dataclass
-from typing import Final
+from collections.abc import Sequence
+from collections.abc import Mapping
+from types import MappingProxyType
+from typing import Final, Literal
 
-from integration.cost_calculation.cost_tracking_case import CostTrackingTestCase, ExactExpected
+from integration.cost_calculation.cost_tracking_case import CostTrackingTestCase, Deployment, ExactExpected
+from pydantic import JsonValue
 
 ROW_FIELDS: Final = frozenset({"spend", "prompt_tokens", "completion_tokens"})
 _EXPECTED_FIELDS: Final = frozenset(ExactExpected.model_fields)
@@ -32,6 +36,7 @@ class BridgeCostParityTestCase:
     native: CostTrackingTestCase
     bridged: CostTrackingTestCase
     quirk: ProviderCostQuirk | None = None
+    known_bug: str | None = None
 
     def __post_init__(self) -> None:
         if self.native.endpoint == self.bridged.endpoint:
@@ -61,3 +66,75 @@ class BridgeCostParityTestCase:
     @property
     def id(self) -> str:
         return self.bridged.name
+
+
+BridgeEndpoint = Literal["/v1/chat/completions", "/v1/responses", "/v1/messages"]
+
+_ENDPOINT_SUFFIX: Final[Mapping[str, str]] = MappingProxyType(
+    {"/v1/chat/completions": "chat_bridged", "/v1/responses": "responses_bridged", "/v1/messages": "messages_bridged"}
+)
+
+
+def _prompt_text(request: Mapping[str, JsonValue]) -> str:
+    """The user prompt of a plain base request, whichever client shape it was written in."""
+    prompt: Final = request.get("input")
+    if isinstance(prompt, str):
+        return prompt
+    messages: Final = request.get("messages")
+    if not isinstance(messages, list):
+        raise ValueError("a bridge base request needs a string input or a messages list")
+    for message in reversed(messages):
+        if not isinstance(message, dict) or message.get("role") != "user":
+            continue
+        content: Final = message.get("content")
+        if isinstance(content, str):
+            return content
+        if isinstance(content, list):
+            return "".join(str(part["text"]) for part in content if isinstance(part, dict) and "text" in part)
+    raise ValueError("a bridge base request needs a user message")
+
+
+def _bridged_request(base: CostTrackingTestCase, endpoint: BridgeEndpoint) -> dict[str, JsonValue]:
+    streamed: Final = base.request.get("stream") is True
+    text: Final = _prompt_text(base.request)
+    if endpoint == "/v1/responses":
+        return {"model": "$MODEL", "input": text, **({"stream": True} if streamed else {})}
+    messages: Final[list[JsonValue]] = [{"role": "user", "content": text}]
+    if endpoint == "/v1/messages":
+        return {"model": "$MODEL", "max_tokens": 412, "messages": messages, **({"stream": True} if streamed else {})}
+    return {
+        "model": "$MODEL",
+        "messages": messages,
+        **({"stream": True, "stream_options": {"include_usage": True}} if streamed else {}),
+    }
+
+
+def bridge_pairs(
+    bases: Sequence[CostTrackingTestCase],
+    endpoints: Sequence[BridgeEndpoint],
+    *,
+    deployment: Deployment | None = None,
+    known_bug: str | None = None,
+) -> tuple[BridgeCostParityTestCase, ...]:
+    """One parity pair per base and bridged endpoint: the same stored provider response and deployment, billed through
+    the base's own endpoint and through each client endpoint LiteLLM bridges into it. `deployment` pins both legs to a
+    deployment the base does not carry (an `openai/responses/` deployment for the chat -> Responses bridge)."""
+    pinned: Final = tuple(
+        base if deployment is None else base.model_copy(update={"deployment": deployment}) for base in bases
+    )
+    return tuple(
+        BridgeCostParityTestCase(
+            native=base,
+            bridged=base.model_copy(
+                update={
+                    "name": f"{base.name}-{_ENDPOINT_SUFFIX[endpoint]}",
+                    "endpoint": endpoint,
+                    "request": _bridged_request(base, endpoint),
+                }
+            ),
+            known_bug=known_bug,
+        )
+        for base in pinned
+        for endpoint in endpoints
+        if endpoint != base.endpoint
+    )
diff --git a/tests/integration/cost_calculation/bridge_cost_parity/test_basic.py b/tests/integration/cost_calculation/bridge_cost_parity/test_basic.py
index 9f0ca44636d..b9b735cc119 100644
--- a/tests/integration/cost_calculation/bridge_cost_parity/test_basic.py
+++ b/tests/integration/cost_calculation/bridge_cost_parity/test_basic.py
@@ -6,6 +6,8 @@ provider's own API and once through a client API LiteLLM bridges into it. Both L
 case's row, and they must agree with each other except where a documented provider quirk says otherwise.
 """
 
+from typing import Final
+
 import pytest
 from integration._support.client import Gateway
 from integration.cost_calculation.bridge_cost_parity.case import BridgeCostParityTestCase
@@ -13,7 +15,15 @@ from integration.cost_calculation.bridge_cost_parity.runner import assert_bridge
 from integration.cost_calculation.catalog import BRIDGE_COST_PARITY_CASES
 
 
-@pytest.mark.parametrize("case", BRIDGE_COST_PARITY_CASES, ids=[case.id for case in BRIDGE_COST_PARITY_CASES])
+def _marks(case: BridgeCostParityTestCase) -> tuple[pytest.MarkDecorator, ...]:
+    """A known bridge bug is intermittent (it billed the native row in 1 of 4 local runs), so the xfail is not strict."""
+    return (pytest.mark.xfail(strict=False, reason=f"BUG: {case.known_bug}"),) if case.known_bug is not None else ()
+
+
+_CASES: Final = tuple(pytest.param(case, marks=_marks(case), id=case.id) for case in BRIDGE_COST_PARITY_CASES)
+
+
+@pytest.mark.parametrize("case", _CASES)
 @pytest.mark.timeout(180)
 def test_bridged_request_bills_the_same_row_as_the_native_request(
     case: BridgeCostParityTestCase, gateway: Gateway
diff --git a/tests/integration/cost_calculation/cases/anthropic.py b/tests/integration/cost_calculation/cases/anthropic.py
index 702420e36e8..3d74ac4e2f3 100644
--- a/tests/integration/cost_calculation/cases/anthropic.py
+++ b/tests/integration/cost_calculation/cases/anthropic.py
@@ -15,7 +15,7 @@ from integration.cost_calculation.cost_tracking_case import (
     RecountRates,
     SseResponse,
 )
-from integration.cost_calculation.bridge_cost_parity.case import BridgeCostParityTestCase
+from integration.cost_calculation.bridge_cost_parity.case import BridgeCostParityTestCase, bridge_pairs
 from integration.cost_calculation.stream_parity.case import StreamParityTestCase, sse_frames
 
 
@@ -6954,7 +6954,22 @@ CLAUDE_HAIKU_5_5_TIERED_CACHE_WRITE_1H_ABOVE_100K: Final = with_usage(
     ),
 )
 
+CLAUDE_SONNET_5_GENERALIZED_ALIAS: Final = CLAUDE_SONNET_5_INPUT_TEXT.model_copy(
+    update={"name": "claude-sonnet-5-generalized_alias", "alias": "claude-sonnet-5-$MARKER"}
+)
+
+CLAUDE_SONNET_5_GENERALIZED_ALIAS_STREAM: Final = CLAUDE_SONNET_5_STREAM.model_copy(
+    update={"name": "claude-sonnet-5-generalized_alias_stream", "alias": "claude-sonnet-5-$MARKER"}
+)
+
+CLAUDE_SONNET_5_SLASH_ALIAS_STREAM: Final = CLAUDE_SONNET_5_STREAM.model_copy(
+    update={"name": "claude-sonnet-5-slash_alias_stream", "alias": "anthropic/claude-sonnet-5-$MARKER"}
+)
+
 CASES: Final[tuple[CostTrackingTestCase, ...]] = (
+    CLAUDE_SONNET_5_GENERALIZED_ALIAS,
+    CLAUDE_SONNET_5_GENERALIZED_ALIAS_STREAM,
+    CLAUDE_SONNET_5_SLASH_ALIAS_STREAM,
     CLAUDE_HAIKU_5_5_TIERED_INPUT_ABOVE_100K,
     CLAUDE_HAIKU_5_5_TIERED_AT_100K_BOUNDARY,
     CLAUDE_HAIKU_5_5_TIERED_CACHE_READ_ABOVE_100K,
@@ -7118,6 +7133,15 @@ PARITY: Final[tuple[StreamParityTestCase, ...]] = (
 )
 
 BRIDGE_COST_PARITY: Final[tuple[BridgeCostParityTestCase, ...]] = (
+    *bridge_pairs(
+        (
+            CLAUDE_SONNET_5_MESSAGES_INPUT_TEXT,
+            CLAUDE_SONNET_5_MESSAGES_CACHE_READ,
+            CLAUDE_SONNET_5_MESSAGES_STREAM,
+            CLAUDE_SONNET_5_MESSAGES_STREAM_CACHE_READ,
+        ),
+        ("/v1/responses", "/v1/chat/completions"),
+    ),
     BridgeCostParityTestCase(
         native=CLAUDE_SONNET_5_MESSAGES_INPUT_TEXT, bridged=CLAUDE_SONNET_5_RESPONSES_BRIDGED_JSON
     ),
diff --git a/tests/integration/cost_calculation/cases/azure.py b/tests/integration/cost_calculation/cases/azure.py
index 96185896b21..6cf7cb23d2e 100644
--- a/tests/integration/cost_calculation/cases/azure.py
+++ b/tests/integration/cost_calculation/cases/azure.py
@@ -19,7 +19,7 @@ from integration.cost_calculation.cost_tracking_case import (
     SseResponse,
     WavUpload,
 )
-from integration.cost_calculation.bridge_cost_parity.case import BridgeCostParityTestCase
+from integration.cost_calculation.bridge_cost_parity.case import BridgeCostParityTestCase, bridge_pairs
 from integration.cost_calculation.stream_parity.case import StreamParityTestCase, sse_frames
 
 
@@ -6170,7 +6170,9 @@ AZURE_GPT_6_LUNA_CACHE_WRITE_5M: Final = with_response_usage(
         "total_tokens": 13308,
     },
     ExactExpected(spend=0.002327, input_cost=0.00208, output_cost=0.000247, prompt_tokens=12928, completion_tokens=380),
-    choice_updates={"content_filter_result": {"error": {"code": "content_filter_error", "message": "The contents are not filtered"}}},
+    choice_updates={
+        "content_filter_result": {"error": {"code": "content_filter_error", "message": "The contents are not filtered"}}
+    },
 )
 AZURE_GPT_6_ASTRA_CACHE_WRITE_5M: Final = with_response_usage(
     AZURE_GPT_6_ASTRA_CACHE_READ,
@@ -6317,6 +6319,16 @@ PARITY: Final[tuple[StreamParityTestCase, ...]] = (
 )
 
 BRIDGE_COST_PARITY: Final[tuple[BridgeCostParityTestCase, ...]] = (
+    *bridge_pairs((AZURE_GPT_5_6_INPUT_TEXT, AZURE_GPT_5_6_CACHE_READ, AZURE_GPT_5_6_STREAM), ("/v1/messages",)),
+    *bridge_pairs(
+        (AZURE_GPT_5_6_REASONING,),
+        ("/v1/messages",),
+        known_bug=(
+            "the /v1/messages -> chat bridge usually drops completion_tokens_details.reasoning_tokens on azure, so"
+            " reasoning tokens bill at output_cost_per_token instead of output_cost_per_reasoning_token"
+            " (output_cost 0.058176 instead of 0.065484)"
+        ),
+    ),
     BridgeCostParityTestCase(native=AZURE_GPT_6_SOL_INPUT_TEXT, bridged=AZURE_GPT_6_SOL_MESSAGES_BRIDGED_JSON),
     BridgeCostParityTestCase(native=AZURE_GPT_6_LUNA_INPUT_TEXT, bridged=AZURE_GPT_6_LUNA_MESSAGES_BRIDGED_JSON),
     BridgeCostParityTestCase(native=AZURE_GPT_6_ASTRA_INPUT_TEXT, bridged=AZURE_GPT_6_ASTRA_MESSAGES_BRIDGED_JSON),
diff --git a/tests/integration/cost_calculation/cases/bedrock_converse.py b/tests/integration/cost_calculation/cases/bedrock_converse.py
index 3c353a107f4..f78bdfecfa3 100644
--- a/tests/integration/cost_calculation/cases/bedrock_converse.py
+++ b/tests/integration/cost_calculation/cases/bedrock_converse.py
@@ -17,7 +17,7 @@ from integration.cost_calculation.cost_tracking_case import (
     RecountExpected,
     RecountRates,
 )
-from integration.cost_calculation.bridge_cost_parity.case import BridgeCostParityTestCase
+from integration.cost_calculation.bridge_cost_parity.case import BridgeCostParityTestCase, bridge_pairs
 from integration.cost_calculation.stream_parity.case import StreamParityTestCase
 
 # anthropic.claude-sonnet-5-v1:0
@@ -6238,6 +6238,14 @@ PARITY: Final[tuple[StreamParityTestCase, ...]] = (
 )
 
 BRIDGE_COST_PARITY: Final[tuple[BridgeCostParityTestCase, ...]] = (
+    *bridge_pairs(
+        (
+            ANTHROPIC_CLAUDE_SONNET_5_V1_0_INPUT_TEXT,
+            ANTHROPIC_CLAUDE_SONNET_5_V1_0_CACHE_READ,
+            ANTHROPIC_CLAUDE_SONNET_5_V1_0_STREAM,
+        ),
+        ("/v1/responses", "/v1/messages"),
+    ),
     BridgeCostParityTestCase(
         native=US_ANTHROPIC_CLAUDE_SONNET_5_INPUT_TEXT, bridged=US_ANTHROPIC_CLAUDE_SONNET_5_RESPONSES_BRIDGED_JSON
     ),
diff --git a/tests/integration/cost_calculation/cases/fireworks_ai.py b/tests/integration/cost_calculation/cases/fireworks_ai.py
index 6e6c2c9abbf..98178f89b7c 100644
--- a/tests/integration/cost_calculation/cases/fireworks_ai.py
+++ b/tests/integration/cost_calculation/cases/fireworks_ai.py
@@ -5,6 +5,9 @@ PARITY pairs the streamed case with its plain twin where both bill the same row.
 from typing import Final
 
 from integration.cost_calculation.cost_tracking_case import (
+    ALWAYS_OPEN_OFF_PEAK_WINDOW,
+    Deployment,
+    OffPeakPricing,
     CostTrackingTestCase,
     ExactExpected,
     JsonResponse,
@@ -3089,7 +3092,39 @@ FIREWORKS_EMBEDDINGS_V1: Final = CostTrackingTestCase(
 )
 
 
+_DEEPSEEK_V4P1_FLASH_OFF_PEAK: Final = Deployment(
+    off_peak_pricing=OffPeakPricing(
+        hours_utc=ALWAYS_OPEN_OFF_PEAK_WINDOW, input_cost_per_token=3e-07, output_cost_per_token=1.2e-06
+    )
+)
+
+FIREWORKS_AI_ACCOUNTS_FIREWORKS_MODELS_DEEPSEEK_V4P1_FLASH_OFF_PEAK: Final = (
+    FIREWORKS_AI_ACCOUNTS_FIREWORKS_MODELS_DEEPSEEK_V4P1_FLASH_INPUT_TEXT.model_copy(
+        update={
+            "name": "fireworks_ai-accounts-fireworks-models-deepseek-v4p1-flash-off_peak",
+            "deployment": _DEEPSEEK_V4P1_FLASH_OFF_PEAK,
+            "expected": ExactExpected(
+                spend=0.0010464, input_cost=0.000552, output_cost=0.0004944, prompt_tokens=1840, completion_tokens=412
+            ),
+        }
+    )
+)
+
+FIREWORKS_AI_ACCOUNTS_FIREWORKS_MODELS_DEEPSEEK_V4P1_FLASH_OFF_PEAK_STREAM: Final = (
+    FIREWORKS_AI_ACCOUNTS_FIREWORKS_MODELS_DEEPSEEK_V4P1_FLASH_STREAM.model_copy(
+        update={
+            "name": "fireworks_ai-accounts-fireworks-models-deepseek-v4p1-flash-off_peak_stream",
+            "deployment": _DEEPSEEK_V4P1_FLASH_OFF_PEAK,
+            "expected": ExactExpected(
+                spend=0.0010464, input_cost=0.000552, output_cost=0.0004944, prompt_tokens=1840, completion_tokens=412
+            ),
+        }
+    )
+)
+
 CASES: Final[tuple[CostTrackingTestCase, ...]] = (
+    FIREWORKS_AI_ACCOUNTS_FIREWORKS_MODELS_DEEPSEEK_V4P1_FLASH_OFF_PEAK,
+    FIREWORKS_AI_ACCOUNTS_FIREWORKS_MODELS_DEEPSEEK_V4P1_FLASH_OFF_PEAK_STREAM,
     FIREWORKS_AI_ACCOUNTS_FIREWORKS_MODELS_DEEPSEEK_V4P1_FLASH_INPUT_TEXT,
     FIREWORKS_AI_ACCOUNTS_FIREWORKS_MODELS_DEEPSEEK_V4P1_FLASH_FALLBACK_CACHE_READ_AT_HALF_INPUT_RATE,
     FIREWORKS_AI_ACCOUNTS_FIREWORKS_MODELS_DEEPSEEK_V4P1_FLASH_STREAM,
diff --git a/tests/integration/cost_calculation/cases/gemini.py b/tests/integration/cost_calculation/cases/gemini.py
index 2f9ffcb4daa..a4538b87db3 100644
--- a/tests/integration/cost_calculation/cases/gemini.py
+++ b/tests/integration/cost_calculation/cases/gemini.py
@@ -4,6 +4,7 @@ PARITY pairs the streamed case with its plain twin where both bill the same row.
 
 from typing import Final
 
+from integration.cost_calculation.bridge_cost_parity.case import BridgeCostParityTestCase, bridge_pairs
 from integration.cost_calculation.cost_tracking_case import (
     CostTrackingTestCase,
     ExactExpected,
@@ -3626,3 +3627,15 @@ PARITY: Final[tuple[StreamParityTestCase, ...]] = (
     ),
     StreamParityTestCase(plain=GEMINI_GEMINI_3_8_FLASH_TOOL_CALL, streamed=GEMINI_GEMINI_3_8_FLASH_STREAM_TOOL_CALL),
 )
+
+BRIDGE_COST_PARITY: Final[tuple[BridgeCostParityTestCase, ...]] = (
+    *bridge_pairs(
+        (
+            GEMINI_GEMINI_3_8_FLASH_INPUT_TEXT,
+            GEMINI_GEMINI_3_8_FLASH_CACHE_READ,
+            GEMINI_GEMINI_3_8_FLASH_REASONING,
+            GEMINI_GEMINI_3_8_FLASH_STREAM,
+        ),
+        ("/v1/responses", "/v1/messages"),
+    ),
+)
diff --git a/tests/integration/cost_calculation/cases/misc.py b/tests/integration/cost_calculation/cases/misc.py
index 76b06f5441b..c3dc0225d0d 100644
--- a/tests/integration/cost_calculation/cases/misc.py
+++ b/tests/integration/cost_calculation/cases/misc.py
@@ -5,6 +5,8 @@ PARITY pairs the streamed case with its plain twin where both bill the same row.
 from typing import Final
 
 from integration.cost_calculation.cost_tracking_case import (
+    ALWAYS_OPEN_OFF_PEAK_WINDOW,
+    OffPeakPricing,
     CostTrackingTestCase,
     Deployment,
     EventStreamEvent,
@@ -1697,7 +1699,22 @@ AZURE_AI_CLAUDE_HAIKU_5_5_STREAM: Final = CLAUDE_HAIKU_5_5_STREAM.model_copy(
 )
 
 
+PERPLEXITY_SONAR_NEXT_OFF_PEAK: Final = PERPLEXITY_SONAR_NEXT_NO_SEARCH.model_copy(
+    update={
+        "name": "perplexity-sonar-next-off_peak",
+        "deployment": Deployment(
+            off_peak_pricing=OffPeakPricing(
+                hours_utc=ALWAYS_OPEN_OFF_PEAK_WINDOW, input_cost_per_token=2.26e-06, output_cost_per_token=2.1e-06
+            )
+        ),
+        "expected": ExactExpected(
+            spend=0.0050236, input_cost=0.0041584, output_cost=0.0008652, prompt_tokens=1840, completion_tokens=412
+        ),
+    }
+)
+
 CASES: Final[tuple[CostTrackingTestCase, ...]] = (
+    PERPLEXITY_SONAR_NEXT_OFF_PEAK,
     NOVA_NEXT_TRANSCRIPTIONS_PER_SECOND,
     AMAZON_NOVA_CANVAS_NEXT_IMAGES_ONE,
     COHERE_EMBEDDINGS_V5,
diff --git a/tests/integration/cost_calculation/cases/openai.py b/tests/integration/cost_calculation/cases/openai.py
index fe4b736959c..9a3885af259 100644
--- a/tests/integration/cost_calculation/cases/openai.py
+++ b/tests/integration/cost_calculation/cases/openai.py
@@ -23,7 +23,7 @@ from integration.cost_calculation.cost_tracking_case import (
 )
 from pydantic import JsonValue
 from integration.cost_calculation.stream_parity.case import StreamParityTestCase, sse_frames
-from integration.cost_calculation.bridge_cost_parity.case import BridgeCostParityTestCase
+from integration.cost_calculation.bridge_cost_parity.case import BridgeCostParityTestCase, bridge_pairs
 
 
 # gpt-5.3-codex
@@ -11184,7 +11184,17 @@ GPT_6_ASTRA_CACHE_WRITE_5M: Final = with_response_usage(
 )
 
 
+GPT_5_6_SLASH_ALIAS: Final = GPT_5_6_INPUT_TEXT.model_copy(
+    update={"name": "gpt-5.6-slash_alias", "alias": "openai/custom/gpt-5.6-$MARKER"}
+)
+
+GPT_5_6_SLASH_ALIAS_STREAM: Final = GPT_5_6_STREAM.model_copy(
+    update={"name": "gpt-5.6-slash_alias_stream", "alias": "openai/custom/gpt-5.6-$MARKER"}
+)
+
 CASES: Final[tuple[CostTrackingTestCase, ...]] = (
+    GPT_5_6_SLASH_ALIAS,
+    GPT_5_6_SLASH_ALIAS_STREAM,
     GPT_6_SOL_RESPONSES_DEPLOYMENT_JSON,
     GPT_6_SOL_CHAT_BRIDGED_JSON,
     GPT_6_LUNA_RESPONSES_DEPLOYMENT_JSON,
@@ -11424,6 +11434,27 @@ PARITY: Final[tuple[StreamParityTestCase, ...]] = (
 )
 
 BRIDGE_COST_PARITY: Final[tuple[BridgeCostParityTestCase, ...]] = (
+    *bridge_pairs(
+        (
+            GPT_5_6_RESPONSES_NATIVE_JSON,
+            GPT_5_6_RESPONSES_CACHE_READ,
+            GPT_5_6_RESPONSES_REASONING,
+            GPT_5_6_RESPONSES_STREAM,
+            GPT_5_6_RESPONSES_STREAM_CACHE_READ,
+        ),
+        ("/v1/messages",),
+    ),
+    *bridge_pairs(
+        (
+            GPT_5_6_RESPONSES_NATIVE_JSON,
+            GPT_5_6_RESPONSES_CACHE_READ,
+            GPT_5_6_RESPONSES_REASONING,
+            GPT_5_6_RESPONSES_STREAM,
+            GPT_5_6_RESPONSES_STREAM_CACHE_READ,
+        ),
+        ("/v1/chat/completions",),
+        deployment=Deployment(model="openai/responses/gpt-5.6"),
+    ),
     BridgeCostParityTestCase(native=GPT_6_SOL_RESPONSES_NATIVE_JSON, bridged=GPT_6_SOL_MESSAGES_BRIDGED_JSON),
     BridgeCostParityTestCase(native=GPT_6_LUNA_RESPONSES_NATIVE_JSON, bridged=GPT_6_LUNA_MESSAGES_BRIDGED_JSON),
     BridgeCostParityTestCase(native=GPT_6_ASTRA_RESPONSES_NATIVE_JSON, bridged=GPT_6_ASTRA_MESSAGES_BRIDGED_JSON),
diff --git a/tests/integration/cost_calculation/cases/vertex_ai.py b/tests/integration/cost_calculation/cases/vertex_ai.py
index 526703bdb24..966a24bcba2 100644
--- a/tests/integration/cost_calculation/cases/vertex_ai.py
+++ b/tests/integration/cost_calculation/cases/vertex_ai.py
@@ -12,6 +12,7 @@ from integration.cost_calculation.cases.anthropic import (
     CLAUDE_SONNET_5_5_INPUT_TEXT,
     CLAUDE_SONNET_5_5_STREAM,
 )
+from integration.cost_calculation.bridge_cost_parity.case import BridgeCostParityTestCase, bridge_pairs
 from integration.cost_calculation.cost_tracking_case import (
     CostTrackingTestCase,
     ExactExpected,
@@ -3157,3 +3158,8 @@ PARITY: Final[tuple[StreamParityTestCase, ...]] = (
     StreamParityTestCase(plain=VERTEX_AI_CLAUDE_OPUS_5_5_INPUT_TEXT, streamed=VERTEX_AI_CLAUDE_OPUS_5_5_STREAM),
     StreamParityTestCase(plain=VERTEX_AI_CLAUDE_FABLE_5_1_INPUT_TEXT, streamed=VERTEX_AI_CLAUDE_FABLE_5_1_STREAM),
 )
+
+BRIDGE_COST_PARITY: Final[tuple[BridgeCostParityTestCase, ...]] = bridge_pairs(
+    (GEMINI_3_8_FLASH_INPUT_TEXT, GEMINI_3_8_FLASH_CACHE_READ, GEMINI_3_8_FLASH_STREAM),
+    ("/v1/responses", "/v1/messages"),
+)
diff --git a/tests/integration/cost_calculation/catalog.py b/tests/integration/cost_calculation/catalog.py
index 15dec65c260..66eb7497bd7 100644
--- a/tests/integration/cost_calculation/catalog.py
+++ b/tests/integration/cost_calculation/catalog.py
@@ -62,6 +62,8 @@ BRIDGE_COST_PARITY_CASES: Final[tuple[BridgeCostParityTestCase, ...]] = (
     *bedrock_converse.BRIDGE_COST_PARITY,
     *azure.BRIDGE_COST_PARITY,
     *anthropic.BRIDGE_COST_PARITY,
+    *gemini.BRIDGE_COST_PARITY,
+    *vertex_ai.BRIDGE_COST_PARITY,
 )
 _PARITY_LEGS: Final = frozenset(leg.name for pair in PARITY_CASES for leg in (pair.plain, pair.streamed)) | frozenset(
     leg.name for pair in BRIDGE_COST_PARITY_CASES for leg in (pair.native, pair.bridged)
diff --git a/tests/integration/cost_calculation/conftest.py b/tests/integration/cost_calculation/conftest.py
index 574d8aaf9af..730f231e098 100644
--- a/tests/integration/cost_calculation/conftest.py
+++ b/tests/integration/cost_calculation/conftest.py
@@ -13,7 +13,7 @@ from cryptography.hazmat.primitives.asymmetric import rsa
 from integration._support.client import JSON_OBJECT, Scenario, eventually, object_value, string_value
 from integration._support.database import read_rows
 from integration._support.upstream import ScenarioHandle, delete_scenario, register_scenario
-from integration.cost_calculation.cost_tracking_case import CostTrackingTestCase, StoredResponse
+from integration.cost_calculation.cost_tracking_case import ALIAS_MARKER, CostTrackingTestCase, StoredResponse
 from pydantic import BaseModel, ConfigDict
 
 
@@ -289,7 +289,10 @@ def register_scenario_deployment(
         case.response if response is None else response,
     )
     scenario.cleanups.callback(delete_scenario, handle)
-    registered_model_name: Final = f"cost-{marker}{marker_suffix}-{run_marker}"
+    unique_marker: Final = f"{marker}{marker_suffix}-{run_marker}"
+    registered_model_name: Final = (
+        case.alias.replace(ALIAS_MARKER, unique_marker) if case.alias is not None else f"cost-{unique_marker}"
+    )
     parameters: Final = {
         "model": case.litellm_model,
         "api_key": case.api_key,
@@ -330,7 +333,14 @@ def register_scenario_deployment(
             {
                 "model_name": registered_model_name,
                 "litellm_params": parameters,
-                "model_info": ({"base_model": case.base_model} if case.base_model is not None else {}),
+                "model_info": {
+                    **({"base_model": case.base_model} if case.base_model is not None else {}),
+                    **(
+                        {"off_peak_pricing": case.deployment.off_peak_pricing.model_dump()}
+                        if case.deployment is not None and case.deployment.off_peak_pricing is not None
+                        else {}
+                    ),
+                },
             }
         ),
     )
diff --git a/tests/integration/cost_calculation/cost_map.py b/tests/integration/cost_calculation/cost_map.py
index ff877ee2299..78d025855b5 100644
--- a/tests/integration/cost_calculation/cost_map.py
+++ b/tests/integration/cost_calculation/cost_map.py
@@ -1335,3 +1335,13 @@ COST_MAP_ENTRIES: Final[dict[str, dict[str, JsonValue]]] = {
         "cache_read_input_token_cost_above_272k_tokens": 4.8e-07,
     },
 }
+
+FALLBACK_GENERALIZATIONS: Final[dict[str, JsonValue]] = {
+    "rules": [
+        {
+            "name": "claude-family-baseline",
+            "pattern": "claude-[a-z]+-\\d+(?:[-.]\\d+)?(?:-\\d{8})?",
+            "model_info": {"mode": "chat", "max_input_tokens": 200000, "max_output_tokens": 64000},
+        }
+    ]
+}
diff --git a/tests/integration/cost_calculation/cost_tracking_case.py b/tests/integration/cost_calculation/cost_tracking_case.py
index fc730524f8e..038c8d6b9e3 100644
--- a/tests/integration/cost_calculation/cost_tracking_case.py
+++ b/tests/integration/cost_calculation/cost_tracking_case.py
@@ -9,6 +9,7 @@ from integration.cost_calculation.cost_map import COST_MAP_ENTRIES
 from pydantic import BaseModel, ConfigDict, Field, JsonValue, field_validator, model_validator
 
 PRIOR_RESPONSE_ID_MARKER: Final = "$PRIOR_RESPONSE_ID"
+ALIAS_MARKER: Final = "$MARKER"
 
 
 class SearchContextCostPerQuery(BaseModel):
@@ -132,6 +133,17 @@ class CostMapEntry(BaseModel):
     provider_specific_entry: ProviderSpecificEntry | None = None
 
 
+class OffPeakPricing(BaseModel):
+    model_config = ConfigDict(frozen=True, extra="forbid")
+
+    hours_utc: str
+    input_cost_per_token: float
+    output_cost_per_token: float
+
+
+ALWAYS_OPEN_OFF_PEAK_WINDOW: Final = "00:00-00:00"
+
+
 class Deployment(BaseModel):
     model_config = ConfigDict(frozen=True, extra="forbid")
 
@@ -140,6 +152,7 @@ class Deployment(BaseModel):
     input_cost_per_token: float | None = None
     output_cost_per_token: float | None = None
     search_cost_per_query: float | None = None
+    off_peak_pricing: OffPeakPricing | None = None
 
 
 class WavUpload(BaseModel):
@@ -286,6 +299,7 @@ class CostTrackingTestCase(BaseModel):
     covers: str
     known_bug: str | None = None
     model: str
+    alias: str | None = None
     endpoint: (
         Literal[
             "/v1/chat/completions",
@@ -313,6 +327,13 @@ class CostTrackingTestCase(BaseModel):
     disconnect_after_frames: int | None = Field(default=None, ge=1)
     upstream_body: dict[str, JsonValue] | None = None
 
+    @field_validator("alias")
+    @classmethod
+    def validate_alias(cls, value: str | None) -> str | None:
+        if value is not None and ALIAS_MARKER not in value:
+            raise ValueError(f"alias must contain {ALIAS_MARKER} so every run registers a unique model_name")
+        return value
+
     @property
     def rates(self) -> CostMapEntry:
         return COST_MAP[self.model]
@@ -664,9 +685,7 @@ def with_response_usage(
     if not isinstance(base.response, JsonResponse):
         raise TypeError(f"{base.name}: with_response_usage needs a JsonResponse base")
     body: Final = {**base.response.body, "usage": dict(usage)}
-    response_body: Final = (
-        _with_choice_updates(body, choice_updates) if choice_updates is not None else body
-    )
+    response_body: Final = _with_choice_updates(body, choice_updates) if choice_updates is not None else body
     return base.model_copy(
         update={
             "name": name,
diff --git a/tests/integration/cost_calculation/test_cost_tracking.py b/tests/integration/cost_calculation/test_cost_tracking.py
index ce30cc7f643..463d9840127 100644
--- a/tests/integration/cost_calculation/test_cost_tracking.py
+++ b/tests/integration/cost_calculation/test_cost_tracking.py
@@ -5,7 +5,6 @@ from __future__ import annotations
 import io
 import json
 import struct
-import time
 import uuid
 import wave
 import zlib
@@ -16,10 +15,11 @@ from typing import Final, cast
 
 import httpx
 import pytest
-from integration._support.client import JSON_OBJECT, Gateway, string_value
+from integration._support.client import JSON_OBJECT, Gateway, Scenario, string_value
 from integration._support.upstream import delete_scenario, register_scenario
 from integration.cost_calculation.assertions import assert_exact, assert_recount
 from integration.cost_calculation.conftest import (
+    CostRow,
     approx_equal,
     poll_cost_row,
     poll_failure_row,
@@ -29,6 +29,7 @@ from integration.cost_calculation.conftest import (
     register_scenario_deployment,
 )
 from integration.cost_calculation.catalog import STANDALONE_CASES, data_errors
+from integration.cost_calculation.stream_parity.runner import flush_spend_logs
 from integration.cost_calculation.cost_tracking_case import (
     BinaryResponse,
     CostTrackingTestCase,
@@ -42,6 +43,14 @@ if _data_errors := data_errors():
     raise ValueError("\n".join(_data_errors))
 
 
+def _settled_rows(scenario: Scenario, gateway: Gateway, key: str, count: int) -> tuple[CostRow, ...]:
+    """Every row the key will ever get, read after a flush request so a late duplicate cannot hide behind `>= count`."""
+    flush_identity: Final = flush_spend_logs(scenario, gateway, key)
+    settled: Final = tuple(row for row in read_rows_now(key) if row.model_id != flush_identity)
+    assert len(settled) == count, f"expected exactly {count} SpendLogs rows for the key, got {len(settled)}: {settled}"
+    return settled
+
+
 def _marks(case: CostTrackingTestCase) -> tuple[pytest.MarkDecorator, ...]:
     bug: Final = (
         (pytest.mark.xfail(strict=True, reason=f"BUG: {case.known_bug}"),) if case.known_bug is not None else ()
@@ -234,6 +243,7 @@ def test_case_bills_expected_cost(gateway: Gateway, case: CostTrackingTestCase)
             assert isinstance(expected, RecountExpected)
             assert row.status == "success", f"{case.name}: disconnect row status was {row.status}"
             assert_recount(case.name, expected, row)
+            _settled_rows(scenario, gateway, key, 1)
             return
         responses: Final = tuple(
             (
@@ -255,6 +265,7 @@ def test_case_bills_expected_cost(gateway: Gateway, case: CostTrackingTestCase)
             )
             row: Final = poll_failure_row(key)
             assert row.spend == 0, f"{case.name}: failure spend was {row.spend}"
+            _settled_rows(scenario, gateway, key, 1)
             return
         assert response.is_success, f"{case.name}: proxy returned {response.status_code}: {response.text[:400]}"
         if case.upstream_body is not None:
@@ -273,19 +284,14 @@ def test_case_bills_expected_cost(gateway: Gateway, case: CostTrackingTestCase)
             )
         if case.response.content_type == "text/event-stream":
             _assert_stream_has_no_error(response.text)
-        rows: Final = poll_rows(key, len(responses) + (prior_response_id is not None))
+        row_count: Final = len(responses) + (prior_response_id is not None)
+        rows: Final = poll_rows(key, row_count)
         if isinstance(expected, RecountExpected):
             row: Final = rows[0]
             assert_recount(case.name, expected, row)
+            _settled_rows(scenario, gateway, key, row_count)
             return
         assert isinstance(expected, ExactExpected)
-        if fallback_deployment is not None:
-            assert deployment is not None
-            time.sleep(3)
-            settled_rows: Final = read_rows_now(key)
-            assert len(settled_rows) == 1
-            assert settled_rows[0].status == "success"
-            assert settled_rows[0].model_id == deployment.identity
         if isinstance(case.response, BinaryResponse):
             header: Final = response.headers.get("x-litellm-response-cost")
             if header is not None:
@@ -329,3 +335,8 @@ def test_case_bills_expected_cost(gateway: Gateway, case: CostTrackingTestCase)
             assert rollups.daily_team.prompt_tokens == expected.prompt_tokens * 3
             assert rollups.daily_team.completion_tokens == expected.completion_tokens * 3
             assert rollups.daily_team.api_requests == 3
+        settled: Final = _settled_rows(scenario, gateway, key, row_count)
+        if fallback_deployment is not None:
+            assert deployment is not None
+            assert settled[0].status == "success"
+            assert settled[0].model_id == deployment.identity