Merge remote-tracking branch 'origin/litellm_internal_staging' into litellm_shadcn_next_0814

ci_cd/generate_model_prices_schema.py

@@ -96,6 +96,7 @@ ARRAY_KEYS: dict[str, JsonSchema] = {
                 "output_cost_per_token": NONNEG_NUMBER,
                 "output_cost_per_reasoning_token": NONNEG_NUMBER,
                 "cache_read_input_token_cost": NONNEG_NUMBER,
+                "cache_creation_input_token_cost": NONNEG_NUMBER,
                 "input_cost_per_query": NONNEG_NUMBER,
             },
             "additionalProperties": False,

litellm/batches/batch_utils.py

@@ -5,7 +5,7 @@ from typing import Any, Final, Literal
 
 import litellm
 from litellm._logging import verbose_logger
-from litellm.litellm_core_utils.llm_cost_calc.utils import _parse_prompt_tokens_details
+from litellm.litellm_core_utils.llm_cost_calc.utils import parse_prompt_tokens_details
 from litellm.types.llms.openai import Batch
 from litellm.types.utils import CallTypes, ModelInfo, Usage
 from litellm.utils import token_counter
@@ -101,7 +101,7 @@ def _iter_successful_output_line_stats(
             continue
         response_body = _get_response_from_batch_job_output_file(entry, custom_llm_provider)
         usage = _get_batch_job_usage_from_response_body(response_body, custom_llm_provider)
-        prompt_details = _parse_prompt_tokens_details(usage)
+        prompt_details = parse_prompt_tokens_details(usage)
         raw_model = response_body.get("model")
         response_model = raw_model if isinstance(raw_model, str) and raw_model else None
         if model_info is not None or custom_llm_provider in ("anthropic", "bedrock"):

litellm/cost_calculator.py

@@ -26,11 +26,11 @@ from litellm.litellm_core_utils.llm_cost_calc.utils import (
     _generic_cost_per_character,
     _get_regional_uplift_multiplier,
     _get_service_tier_cost_key,
-    _parse_prompt_tokens_details,
     calculate_cost_component,
     generic_cost_per_token,
     get_billable_input_tokens,
     get_token_type_cost_breakdown,
+    parse_prompt_tokens_details,
     select_cost_metric_for_model,
 )
 from litellm.llms.anthropic.cost_calculation import (
@@ -645,7 +645,11 @@ def cost_per_token(
     else:
         model_info: Final = _cached_get_model_info_helper(model=model, custom_llm_provider=custom_llm_provider)
 
-        if (model_info.get("input_cost_per_token") or 0.0) > 0 or (model_info.get("output_cost_per_token") or 0.0) > 0:
+        if (
+            (model_info.get("input_cost_per_token") or 0.0) > 0
+            or (model_info.get("output_cost_per_token") or 0.0) > 0
+            or model_info.get("tiered_pricing") is not None
+        ):
             return generic_cost_per_token(
                 model=model,
                 usage=usage_block,
@@ -2159,7 +2163,7 @@ def batch_cost_calculator(
     if input_cost_per_token_batches:
         total_prompt_cost = usage.prompt_tokens * input_cost_per_token_batches
     elif input_cost_per_token:
-        details: Final = _parse_prompt_tokens_details(usage)
+        details: Final = parse_prompt_tokens_details(usage)
         cache_read_tokens: Final = details["cache_hit_tokens"]
         cache_creation_tokens: Final = details["cache_creation_tokens"]
 

litellm/litellm_core_utils/llm_cost_calc/tiered_pricing.py

@@ -1,5 +1,5 @@
 """
-Provider-neutral graduated tiered pricing calculation.
+Provider-neutral tiered pricing calculation.
 
 Shared by provider cost calculators (e.g. Dashscope) and the proxy budget
 reservation logic so neither has to depend on the other.
@@ -25,80 +25,6 @@ def _coerce_cost_per_token(value: float | str | None) -> float:
     return float(value)
 
 
-def calculate_tiered_cost(
-    tokens: int,
-    tiered_pricing: list[dict],
-    cost_key: str,
-    fallback_cost_key: str | None = None,
-) -> float:
-    """
-    Calculate cost for a given number of tokens based on a true tiered pricing structure.
-
-    This function iterates through sorted pricing tiers, calculates the cost for the
-    number of tokens that fall into each tier's range, and sums them up to get the total cost.
-
-    Args:
-        tokens (int): The total number of tokens to calculate the cost for.
-        tiered_pricing (List[dict]): A list of dictionaries, where each dictionary
-            represents a pricing tier.
-        cost_key (str): The key in the tier dictionary that holds the per-token cost
-            (e.g., 'input_cost_per_token').
-        fallback_cost_key (Optional[str], optional): A fallback key to use if the
-            primary `cost_key` is not found in a tier. Defaults to None.
-
-    Returns:
-        float: The total calculated cost for the given tokens.
-
-    Example:
-        >>> tiered_pricing = [
-        ...     {"range": [0, 100000], "input_cost_per_token": 0.0001},
-        ...     {"range": [100000, 500000], "input_cost_per_token": 0.00005},
-        ... ]
-
-        Calculating cost for 150,000 tokens:
-        (100,000 * 0.0001) + (50,000 * 0.00005) = $12.5
-    """
-    if not tiered_pricing or tokens <= 0:
-        return 0.0
-
-    total_cost = 0.0
-    tokens_processed = 0
-
-    sorted_tiers: Final = sorted(tiered_pricing, key=lambda x: x.get("range", [0, 0])[0])
-
-    for tier in sorted_tiers:
-        if tokens_processed >= tokens:
-            break
-
-        tier_range = tier.get("range", [])
-        if len(tier_range) != 2:
-            continue
-
-        range_start, range_end = tier_range
-
-        if tokens <= range_start:
-            continue
-
-        tier_start = max(range_start, tokens_processed)
-        tier_end = min(range_end, tokens)
-
-        if tier_end > tier_start:
-            tokens_in_tier = tier_end - tier_start
-            cost_per_token = tier.get(cost_key) or tier.get(fallback_cost_key, 0)
-            total_cost += tokens_in_tier * _coerce_cost_per_token(cost_per_token)
-            tokens_processed = tier_end
-
-    # After loop, check if any tokens remain (i.e., tokens > highest tier's end range)
-    # and charge them at the last tier's rate.
-    if tokens_processed < tokens and sorted_tiers:
-        last_tier: Final = sorted_tiers[-1]
-        remaining_tokens: Final = tokens - tokens_processed
-        cost_per_token = last_tier.get(cost_key) or last_tier.get(fallback_cost_key, 0)
-        total_cost += remaining_tokens * _coerce_cost_per_token(cost_per_token)
-
-    return total_cost
-
-
 def select_tier_for_input(
     tiered_pricing: list[dict],
     input_tokens: int,
@@ -134,6 +60,12 @@ def tier_rate(
     cost_key: str,
     fallback_cost_key: str | None = None,
 ) -> float:
-    """Read a per-token rate from a tier, coercing YAML string costs to float."""
-    raw: Final = tier.get(cost_key) or tier.get(fallback_cost_key, 0)
-    return _coerce_cost_per_token(raw)
+    """Read a per-token rate from a tier, coercing YAML string costs to float.
+
+    A rate that is explicitly present wins over the fallback, an explicit zero
+    included, so a tier can declare a token type free.
+    """
+    primary: Final = tier.get(cost_key)
+    if primary is not None:
+        return _coerce_cost_per_token(primary)
+    return _coerce_cost_per_token(tier.get(fallback_cost_key, 0))

litellm/litellm_core_utils/llm_cost_calc/utils.py

@@ -8,6 +8,10 @@ from typing import Any, Final, Literal, TypedDict, cast
 
 import litellm
 from litellm._logging import verbose_logger
+from litellm.litellm_core_utils.llm_cost_calc.tiered_pricing import (
+    select_tier_for_input,
+    tier_rate,
+)
 from litellm.types.utils import (
     CacheCreationTokenDetails,
     CallTypes,
@@ -95,7 +99,7 @@ def get_billable_input_tokens(usage: Usage) -> int:
     Returns the number of billable input tokens.
     Subtracts cached tokens from prompt tokens if applicable.
     """
-    details: Final = _parse_prompt_tokens_details(usage)
+    details: Final = parse_prompt_tokens_details(usage)
     return usage.prompt_tokens - details["cache_hit_tokens"]
 
 
@@ -207,6 +211,57 @@ def _parse_above_token_threshold(key: str) -> float:
     return float(threshold_str.replace("k", "")) * (1000 if "k" in threshold_str else 1)
 
 
+def _select_priced_tier(model_info: ModelInfo, usage: Usage) -> dict | None:
+    tiered_pricing: Final = model_info.get("tiered_pricing")
+    if not isinstance(tiered_pricing, list) or not tiered_pricing:
+        return None
+
+    tier: Final = select_tier_for_input(tiered_pricing=tiered_pricing, input_tokens=usage.prompt_tokens)
+    if tier is None or "input_cost_per_token" not in tier:
+        return None
+    return tier
+
+
+def _get_tiered_reasoning_rate(model_info: ModelInfo, usage: Usage) -> float | None:
+    tier: Final = _select_priced_tier(model_info=model_info, usage=usage)
+    if tier is None:
+        return None
+    if "output_cost_per_reasoning_token" not in tier and "output_cost_per_token" not in tier:
+        return None
+    return tier_rate(tier, "output_cost_per_reasoning_token", "output_cost_per_token")
+
+
+def _get_tiered_base_costs(model_info: ModelInfo, usage: Usage) -> tuple[float, float, float, float, float] | None:
+    """
+    Resolve the base rates from a model's ``tiered_pricing`` table, if it has one.
+
+    Tiered pricing is all-or-nothing: one tier is picked from the request's input tokens
+    and every token of the request is billed at that tier's rate. Rates the tier does not
+    declare fall back to the tier's input rate, so a request never mixes tiers.
+
+    An output rate is the exception: a tier table that spells out only input rates would
+    otherwise serve every completion for free, so the model's own output rate stands in.
+    """
+    tier: Final = _select_priced_tier(model_info=model_info, usage=usage)
+    if tier is None:
+        return None
+
+    cache_creation_cost: Final = tier_rate(tier, "cache_creation_input_token_cost", "input_cost_per_token")
+    completion_cost: Final = (
+        tier_rate(tier, "output_cost_per_token")
+        if "output_cost_per_token" in tier
+        else _get_cost_per_unit(model_info, "output_cost_per_token") or 0.0
+    )
+    return (
+        tier_rate(tier, "input_cost_per_token"),
+        completion_cost,
+        cache_creation_cost,
+        tier_rate(tier, "cache_creation_input_token_cost_above_1hr", "cache_creation_input_token_cost")
+        or cache_creation_cost,
+        tier_rate(tier, "cache_read_input_token_cost", "input_cost_per_token"),
+    )
+
+
 def _get_token_base_cost(
     model_info: ModelInfo,
     usage: Usage,
@@ -226,6 +281,10 @@ def _get_token_base_cost(
     Returns:
         Tuple[float, float, float, float] - (prompt_cost, completion_cost, cache_creation_cost, cache_read_cost)
     """
+    tiered_base_costs: Final = _get_tiered_base_costs(model_info=model_info, usage=usage)
+    if tiered_base_costs is not None:
+        return tiered_base_costs
+
     # Get service tier aware cost keys
     input_cost_key: Final = _get_service_tier_cost_key("input_cost_per_token", service_tier)
     output_cost_key: Final = _get_service_tier_cost_key("output_cost_per_token", service_tier)
@@ -470,7 +529,7 @@ class PromptTokensDetailsResult(TypedDict):
     audio_length_seconds: float
 
 
-def _parse_prompt_tokens_details(usage: Usage) -> PromptTokensDetailsResult:
+def parse_prompt_tokens_details(usage: Usage) -> PromptTokensDetailsResult:
     cache_hit_tokens: Final = cast(int | None, getattr(usage.prompt_tokens_details, "cached_tokens", 0)) or 0
     cache_creation_tokens: Final = (
         cast(
@@ -540,7 +599,7 @@ class CompletionTokensDetailsResult(TypedDict):
     video_tokens: int
 
 
-def _parse_completion_tokens_details(usage: Usage) -> CompletionTokensDetailsResult:
+def parse_completion_tokens_details(usage: Usage) -> CompletionTokensDetailsResult:
     audio_tokens: Final = (
         cast(
             int | None,
@@ -777,7 +836,7 @@ def generic_cost_per_token(
         audio_length_seconds=0.0,
     )
     if usage.prompt_tokens_details:
-        prompt_tokens_details = _parse_prompt_tokens_details(usage)
+        prompt_tokens_details = parse_prompt_tokens_details(usage)
 
     ## EDGE CASE - text tokens not set or includes cached tokens (double-counting)
     ## Some providers (like xAI) report text_tokens = prompt_tokens (including cached)
@@ -832,7 +891,7 @@ def generic_cost_per_token(
     video_tokens = 0
     is_text_tokens_total = False
     if usage.completion_tokens_details is not None:
-        completion_tokens_details: Final = _parse_completion_tokens_details(usage)
+        completion_tokens_details: Final = parse_completion_tokens_details(usage)
         audio_tokens = completion_tokens_details["audio_tokens"]
         text_tokens = completion_tokens_details["text_tokens"]
         reasoning_tokens = completion_tokens_details["reasoning_tokens"]
@@ -869,10 +928,15 @@ def generic_cost_per_token(
 
     ## REASONING COST
     if not is_text_tokens_total and reasoning_tokens and reasoning_tokens > 0:
-        _output_cost_per_reasoning_token = _resolve_reasoning_token_cost(
-            model_info=model_info,
-            service_tier=service_tier,
-            completion_base_cost=completion_base_cost,
+        tiered_reasoning_rate: Final = _get_tiered_reasoning_rate(model_info=model_info, usage=usage)
+        _output_cost_per_reasoning_token = (
+            tiered_reasoning_rate
+            if tiered_reasoning_rate is not None
+            else _resolve_reasoning_token_cost(
+                model_info=model_info,
+                service_tier=service_tier,
+                completion_base_cost=completion_base_cost,
+            )
         )
         completion_cost += float(reasoning_tokens) * _output_cost_per_reasoning_token
 
@@ -952,26 +1016,29 @@ def get_token_type_cost_breakdown(
     )
 
     reasoning_tokens = (
-        _parse_completion_tokens_details(usage)["reasoning_tokens"]
-        if usage.completion_tokens_details is not None
-        else 0
+        parse_completion_tokens_details(usage)["reasoning_tokens"] if usage.completion_tokens_details is not None else 0
     )
     if not reasoning_tokens:
         reasoning_tokens = _coerce_token_count(getattr(usage, "reasoning_tokens", 0))
 
-    # Reasoning is billed at the explicit per-reasoning-token rate when the model
-    # defines one, otherwise at the standard output-token rate - this mirrors how the
-    # total completion cost is computed, so the breakdown can never diverge from it.
-    reasoning_rate = _get_cost_per_unit(model_info, "output_cost_per_reasoning_token", None)
-    if reasoning_rate is None:
-        reasoning_rate = completion_base_cost
+    # Reasoning is billed at the selected tier's reasoning rate for tiered models,
+    # else at the explicit per-reasoning-token rate when the model defines one,
+    # otherwise at the standard output-token rate - this mirrors how the total
+    # completion cost is computed, so the breakdown can never diverge from it.
+    tiered_reasoning_rate: Final = _get_tiered_reasoning_rate(model_info=model_info, usage=usage)
+    flat_reasoning_rate: Final = _get_cost_per_unit(model_info, "output_cost_per_reasoning_token", None)
+    reasoning_rate: Final = (
+        tiered_reasoning_rate
+        if tiered_reasoning_rate is not None
+        else (flat_reasoning_rate if flat_reasoning_rate is not None else completion_base_cost)
+    )
     reasoning_cost = float(reasoning_tokens) * reasoning_rate
 
     cache_read_tokens = 0
     cache_creation_tokens = 0
     cache_creation_token_details: CacheCreationTokenDetails | None = None
     if usage.prompt_tokens_details is not None:
-        prompt_tokens_details: Final = _parse_prompt_tokens_details(usage)
+        prompt_tokens_details: Final = parse_prompt_tokens_details(usage)
         cache_read_tokens = prompt_tokens_details["cache_hit_tokens"]
         cache_creation_tokens = prompt_tokens_details["cache_creation_tokens"]
         cache_creation_token_details = prompt_tokens_details["cache_creation_token_details"]

litellm/llms/anthropic/cost_calculation.py

@@ -10,10 +10,10 @@ from pydantic import BaseModel, ValidationError
 from litellm.litellm_core_utils.llm_cost_calc.utils import (
     _get_token_base_cost,
     _get_web_search_requests,
-    _parse_prompt_tokens_details,
     calculate_cache_writing_cost,
     generic_cost_per_token,
     get_provider_specific_geo_multiplier,
+    parse_prompt_tokens_details,
 )
 
 if TYPE_CHECKING:
@@ -33,7 +33,7 @@ def _compute_cache_only_cost(model_info: "ModelInfo", usage: "Usage", service_ti
     if usage.prompt_tokens_details is None:
         return 0.0
 
-    prompt_tokens_details: Final = _parse_prompt_tokens_details(usage)
+    prompt_tokens_details: Final = parse_prompt_tokens_details(usage)
     (
         _,
         _,

litellm/llms/dashscope/cost_calculator.py

@@ -1,108 +1,111 @@
 """
 Cost calculator for Dashscope Chat models.
 
-Handles tiered pricing and prompt caching scenarios.
+Alibaba Model Studio tiered pricing is all-or-nothing: the tier is picked from the
+total input tokens of a single request, and every token of that request (input,
+cached, cache-creation, output, reasoning) is billed at that one tier's rate.
+See https://help.aliyun.com/zh/model-studio/billing-for-model-studio
 """
 
 from dataclasses import dataclass
 from typing import Final
 
-from litellm.litellm_core_utils.llm_cost_calc.tiered_pricing import calculate_tiered_cost
+from litellm.litellm_core_utils.llm_cost_calc.tiered_pricing import select_tier_for_input, tier_rate
+from litellm.litellm_core_utils.llm_cost_calc.utils import (
+    parse_completion_tokens_details,
+    parse_prompt_tokens_details,
+)
 from litellm.types.utils import ModelInfo, Usage
 from litellm.utils import get_model_info
 
 
-@dataclass
+@dataclass(frozen=True, slots=True)
 class TokenBreakdown:
-    """Token breakdown for cost calculation."""
-
     text_tokens: int
     cached_tokens: int
+    cache_creation_tokens: int
     completion_tokens: int
     reasoning_tokens: int
 
+    @property
+    def total_input_tokens(self) -> int:
+        return self.text_tokens + self.cached_tokens + self.cache_creation_tokens
+
 
 def _extract_token_breakdown(usage: Usage) -> TokenBreakdown:
-    """Extract token counts from usage, handling cached and reasoning tokens."""
-    cached_tokens = 0
-    if usage.prompt_tokens_details and hasattr(usage.prompt_tokens_details, "cached_tokens"):
-        cached_tokens = usage.prompt_tokens_details.cached_tokens or 0
+    prompt_details: Final = parse_prompt_tokens_details(usage)
+    cached_tokens: Final = prompt_details["cache_hit_tokens"]
+    cache_creation_tokens: Final = prompt_details["cache_creation_tokens"]
+    text_tokens: Final = max(usage.prompt_tokens - cached_tokens - cache_creation_tokens, 0)
 
-    text_tokens: Final = usage.prompt_tokens - cached_tokens
+    reasoning_tokens: Final = parse_completion_tokens_details(usage)["reasoning_tokens"]
+    completion_tokens: Final = max((usage.completion_tokens or 0) - reasoning_tokens, 0)
 
-    reasoning_tokens = 0
-    if (
-        hasattr(usage, "completion_tokens_details")
-        and usage.completion_tokens_details
-        and hasattr(usage.completion_tokens_details, "reasoning_tokens")
-    ):
-        reasoning_tokens = usage.completion_tokens_details.reasoning_tokens or 0
+    return TokenBreakdown(
+        text_tokens=text_tokens,
+        cached_tokens=cached_tokens,
+        cache_creation_tokens=cache_creation_tokens,
+        completion_tokens=completion_tokens,
+        reasoning_tokens=reasoning_tokens,
+    )
 
-    completion_tokens: Final = (usage.completion_tokens or 0) - reasoning_tokens
 
-    return TokenBreakdown(text_tokens, cached_tokens, completion_tokens, reasoning_tokens)
+def _flat_rate(model_info: ModelInfo, cost_key: str, fallback_cost_key: str) -> float:
+    value: Final = model_info.get(cost_key)
+    if value is None:
+        return float(model_info.get(fallback_cost_key) or 0.0)
+    return float(value)
 
 
 def _calculate_prompt_cost(
     breakdown: TokenBreakdown,
     model_info: ModelInfo,
-    tiered_pricing: list[dict] | None,
+    tier: dict | None,
 ) -> float:
-    """Calculate total prompt cost including cached tokens."""
-    if tiered_pricing:
-        text_cost: Final = calculate_tiered_cost(
-            tokens=breakdown.text_tokens,
-            tiered_pricing=tiered_pricing,
-            cost_key="input_cost_per_token",
+    if tier is not None:
+        return (
+            (breakdown.text_tokens * tier_rate(tier, "input_cost_per_token"))
+            + (breakdown.cached_tokens * tier_rate(tier, "cache_read_input_token_cost", "input_cost_per_token"))
+            + (
+                breakdown.cache_creation_tokens
+                * tier_rate(tier, "cache_creation_input_token_cost", "input_cost_per_token")
+            )
         )
-        cache_cost = calculate_tiered_cost(
-            tokens=breakdown.cached_tokens,
-            tiered_pricing=tiered_pricing,
-            cost_key="cache_read_input_token_cost",
-            fallback_cost_key="input_cost_per_token",
-        )
-        return text_cost + cache_cost
 
     input_cost: Final = float(model_info.get("input_cost_per_token") or 0.0)
+    cache_read_cost: Final = _flat_rate(model_info, "cache_read_input_token_cost", "input_cost_per_token")
+    cache_creation_cost: Final = _flat_rate(model_info, "cache_creation_input_token_cost", "input_cost_per_token")
 
-    # For cache_cost, first try the specific key, then fall back to input_cost.
-    cache_cost_val: Final = model_info.get("cache_read_input_token_cost")
-    if cache_cost_val is None:
-        cache_cost = input_cost
-    else:
-        cache_cost = float(cache_cost_val)
-
-    return (breakdown.text_tokens * input_cost) + (breakdown.cached_tokens * cache_cost)
+    return (
+        (breakdown.text_tokens * input_cost)
+        + (breakdown.cached_tokens * cache_read_cost)
+        + (breakdown.cache_creation_tokens * cache_creation_cost)
+    )
 
 
 def _calculate_completion_cost(
     breakdown: TokenBreakdown,
     model_info: ModelInfo,
-    tiered_pricing: list[dict] | None,
+    tier: dict | None,
 ) -> float:
-    """Calculate total completion cost including reasoning tokens."""
-    if tiered_pricing:
-        completion_cost: Final = calculate_tiered_cost(
-            tokens=breakdown.completion_tokens,
-            tiered_pricing=tiered_pricing,
-            cost_key="output_cost_per_token",
-        )
-        reasoning_cost = calculate_tiered_cost(
-            tokens=breakdown.reasoning_tokens,
-            tiered_pricing=tiered_pricing,
-            cost_key="output_cost_per_reasoning_token",
-            fallback_cost_key="output_cost_per_token",
-        )
-        return completion_cost + reasoning_cost
-
-    output_cost: Final = float(model_info.get("output_cost_per_token") or 0.0)
-
-    # For reasoning_cost, first try the specific key, then fall back to output_cost.
-    reasoning_cost_val: Final = model_info.get("output_cost_per_reasoning_token")
-    if reasoning_cost_val is None:
-        reasoning_cost = output_cost
-    else:
-        reasoning_cost = float(reasoning_cost_val)
+    # A tier that declares output rates keeps the request on them, all-or-nothing. A tier table
+    # spelling out only input rates would serve every completion for free, so there the model's
+    # own output rates stand in
+    tier_declares_output: Final = tier is not None and "output_cost_per_token" in tier
+    output_cost: Final = (
+        tier_rate(tier, "output_cost_per_token")
+        if tier_declares_output
+        else float(model_info.get("output_cost_per_token") or 0.0)
+    )
+    tier_declares_reasoning: Final = tier is not None and "output_cost_per_reasoning_token" in tier
+    model_reasoning_rate: Final = None if tier_declares_output else model_info.get("output_cost_per_reasoning_token")
+    reasoning_cost: Final = (
+        tier_rate(tier, "output_cost_per_reasoning_token", "output_cost_per_token")
+        if tier_declares_reasoning
+        else float(model_reasoning_rate)
+        if model_reasoning_rate is not None
+        else output_cost
+    )
 
     return (breakdown.completion_tokens * output_cost) + (breakdown.reasoning_tokens * reasoning_cost)
 
@@ -122,11 +125,15 @@ def cost_per_token(model: str, usage: Usage) -> tuple[float, float]:
     """
     model_info: Final = get_model_info(model=model, custom_llm_provider="dashscope")
     breakdown: Final = _extract_token_breakdown(usage)
-    tiered_pricing = model_info.get("tiered_pricing") if isinstance(model_info.get("tiered_pricing"), list) else None
-
-    prompt_cost = _calculate_prompt_cost(breakdown=breakdown, model_info=model_info, tiered_pricing=tiered_pricing)
-    completion_cost: Final = _calculate_completion_cost(
-        breakdown=breakdown, model_info=model_info, tiered_pricing=tiered_pricing
+    raw_tiers: Final = model_info.get("tiered_pricing")
+    tiered_pricing: Final = raw_tiers if isinstance(raw_tiers, list) else None
+    tier: Final = (
+        select_tier_for_input(tiered_pricing=tiered_pricing, input_tokens=breakdown.total_input_tokens)
+        if tiered_pricing
+        else None
     )
 
+    prompt_cost: Final = _calculate_prompt_cost(breakdown=breakdown, model_info=model_info, tier=tier)
+    completion_cost: Final = _calculate_completion_cost(breakdown=breakdown, model_info=model_info, tier=tier)
+
     return prompt_cost, completion_cost

litellm/llms/vertex_ai/files/transformation.py

@@ -7,6 +7,7 @@ import re
 import time
 from collections.abc import Callable, Iterable, Iterator, Mapping
 from typing import Any, Final, TypedDict
+from urllib.parse import quote, unquote
 
 import httpx
 from httpx import Headers, Response
@@ -43,6 +44,9 @@ from litellm.llms.vertex_ai.gemini.transformation import _transform_request_body
 from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import (
     VertexGeminiConfig,
 )
+from litellm.llms.vertex_ai.gemini_embeddings.batch_embed_content_transformation import (
+    transform_openai_input_gemini_embed_content,
+)
 from litellm.types.files import StreamingMediaUploadConfig
 from litellm.types.llms.openai import (
     AllMessageValues,
@@ -54,14 +58,28 @@ from litellm.types.llms.openai import (
     OpenAIFilesPurpose,
     PathLike,
 )
-from litellm.types.llms.vertex_ai import GcsBucketResponse
-from litellm.types.utils import LlmProviders, ModelResponse
+from litellm.types.llms.vertex_ai import GcsBucketResponse, GeminiEmbeddingInput
+from litellm.types.utils import (
+    Embedding,
+    EmbeddingResponse,
+    LlmProviders,
+    ModelResponse,
+    Usage,
+)
 
 from ..common_utils import VertexAIError
 from ..vertex_llm_base import VertexBase
 
 _GCP_LABEL_VALUE_MAX_LEN: Final = 63
 _CUSTOM_ID_RAW_LABEL_PREFIX: Final = "b32_"
+_VERTEX_BATCH_KEY_FIELD: Final = "key"
+_MANAGED_GCS_MODEL_PATH_PATTERN: Final = re.compile(r"publishers/[^/]+/models/([^/?]+)")
+_EMBED_REQUEST_FIELD_BY_GEMINI_PARAM: Final = (
+    ("outputDimensionality", "output_dimensionality"),
+    ("taskType", "task_type"),
+    ("title", "title"),
+)
+_VERTEX_BATCH_FANNED_OUT_KEY_PATTERN: Final = re.compile(r"(?P<custom_id>[^#]*)#(?P<index>\d+)/(?P<total>\d+)")
 
 
 class _GcsObjectMetadataJson(TypedDict, total=False):
@@ -164,8 +182,26 @@ def _set_litellm_batch_custom_id_labels(labels: dict[str, str], custom_id: objec
         labels[f"litellm_custom_id_raw_{index}"] = raw_label_chunk
 
 
-def _get_litellm_batch_custom_id_from_labels(labels: Mapping[str, object]) -> str:
+def _get_litellm_batch_custom_id(vertex_output_row: Mapping[str, object]) -> str:
+    """
+    Resolve the OpenAI `custom_id` for a Vertex batch output row.
+
+    Embedding rows carry it in the top-level `key` field that Vertex echoes back;
+    `generateContent` rows have no such field, so it is smuggled through request
+    labels instead (see `_set_litellm_batch_custom_id_labels`).
+    """
+    key = vertex_output_row.get(_VERTEX_BATCH_KEY_FIELD)
+    if key is not None:
+        return unquote(str(key))
+    request_data = vertex_output_row.get("request")
+    labels = request_data.get("labels") if isinstance(request_data, Mapping) else None
+    return _get_litellm_batch_custom_id_from_labels(labels)
+
+
+def _get_litellm_batch_custom_id_from_labels(labels: Mapping[str, object] | None) -> str:
     """Prefer encoded custom_id when present (see _set_litellm_batch_custom_id_labels)."""
+    if not labels:
+        return "unknown"
     raw: Final = labels.get("litellm_custom_id_raw")
     if raw:
         raw_chunks: Final = [str(raw)]
@@ -182,17 +218,311 @@ def _get_litellm_batch_custom_id_from_labels(labels: Mapping[str, object]) -> st
     return str(labels.get("litellm_custom_id", "unknown"))
 
 
-def _openai_batch_jsonl_entry_to_vertex_wrapped_request(
+def _is_vertex_embeddings_batch_output_row(vertex_output_row: Mapping[str, Any]) -> bool:
+    """
+    Whether a Vertex batch output row came from an `EmbedContentRequest`.
+
+    Successful rows hold the vector under `response.embedding.values`; failed rows only
+    carry `status`, so they are recognized from the singular `content` that the
+    embeddings request shape echoes back.
+    """
+    if "request" not in vertex_output_row:
+        return False
+    response = vertex_output_row.get("response")
+    if isinstance(response, dict) and isinstance(response.get("embedding"), dict):
+        return True
+    request_data = vertex_output_row.get("request")
+    return bool(vertex_output_row.get("status")) and isinstance(request_data, dict) and "content" in request_data
+
+
+def _openai_batch_output_row(
+    custom_id: str,
+    body: Mapping[str, Any] | None = None,
+    error_code: str | None = None,
+    error_message: str = "",
+) -> _OpenAIBatchOutputRow:
+    """
+    One row of an OpenAI batch output file. Per the OpenAI Batch spec, failed rows set
+    `response` to null and populate `error` instead.
+    """
+    return {
+        "id": f"batch_req_{uuid.uuid4()}",
+        "custom_id": custom_id,
+        "response": None
+        if body is None
+        else {
+            "status_code": 200,
+            "request_id": body.get("id", ""),
+            "body": body,
+        },
+        "error": None if error_code is None else {"code": error_code, "message": error_message},
+    }
+
+
+def _split_vertex_batch_key(vertex_output_row: Mapping[str, Any]) -> tuple[str, int, int]:
+    """
+    Resolve `(custom_id, index within that custom_id, group size)` for a Vertex batch
+    output row.
+
+    A `/v1/embeddings` entry whose `input` is an array fans out into one Vertex row per
+    element, tagged `<percent-encoded custom_id>#<index>/<total>` (see
+    `_vertex_batch_embeddings_key`), so the rows can be reassembled into a single OpenAI
+    response.
+    """
+    key = vertex_output_row.get(_VERTEX_BATCH_KEY_FIELD)
+    if key is None:
+        return _get_litellm_batch_custom_id(vertex_output_row), 0, 1
+    match = _VERTEX_BATCH_FANNED_OUT_KEY_PATTERN.fullmatch(str(key))
+    if match is None:
+        return unquote(str(key)), 0, 1
+    return unquote(match["custom_id"]), int(match["index"]), int(match["total"])
+
+
+def _embedding_prompt_token_count(vertex_response: Mapping[str, Any]) -> int:
+    """
+    Prompt tokens billed for one Vertex Gemini Embedding batch row.
+
+    Live rows report usage under `usageMetadata`; the documented `tokenCount` is kept as
+    a fallback.
+    """
+    usage_metadata = vertex_response.get("usageMetadata")
+    if isinstance(usage_metadata, Mapping):
+        return int(usage_metadata.get("promptTokenCount") or 0)
+    return int(vertex_response.get("tokenCount") or 0)
+
+
+def _vertex_embeddings_rows_to_openai_batch_output_row(
+    custom_id: str,
+    vertex_output_rows: tuple[Mapping[str, Any], ...],
+    element_indices: tuple[int, ...],
+    element_count: int,
+    model: str | None,
+) -> _OpenAIBatchOutputRow:
+    """
+    Transforms the Vertex Gemini Embedding batch output rows belonging to one OpenAI
+    batch entry into an OpenAI batch output row holding an `/v1/embeddings` response.
+
+    Example Vertex jsonl
+    {"key": "id_1", "request": {...}, "response": {"embedding": {"values": [-0.015, 0.024]}, "usageMetadata": {"promptTokenCount": 2}}}
+
+    An entry that asked for several embeddings at once maps to several rows here, which
+    become the indexed elements of a single `data` array. One failed or missing element
+    fails the whole entry, since an OpenAI batch row is either a response or an error and
+    a partial `data` array would silently shift the remaining embeddings onto the wrong
+    input positions. Rows carry no `modelVersion`, so the model comes from the batch they
+    belong to.
+    """
+    status = next((row["status"] for row in vertex_output_rows if row.get("status")), "")
+    if status:
+        return _openai_batch_output_row(
+            custom_id=custom_id,
+            error_code="vertex_ai_error",
+            error_message=status,
+        )
+
+    if element_indices != tuple(range(element_count)):
+        return _openai_batch_output_row(
+            custom_id=custom_id,
+            error_code="vertex_ai_error",
+            error_message=(
+                f"Vertex returned embeddings for input positions {list(element_indices)} "
+                f"of the {element_count} requested"
+            ),
+        )
+
+    responses = tuple(row["response"] for row in vertex_output_rows)
+    token_count = sum(_embedding_prompt_token_count(response) for response in responses)
+    body = EmbeddingResponse(
+        model=model or "",
+        data=[
+            Embedding(
+                embedding=response["embedding"]["values"],
+                index=index,
+                object="embedding",
+            )
+            for index, response in enumerate(responses)
+        ],
+        usage=Usage(prompt_tokens=token_count, total_tokens=token_count),
+    ).model_dump()
+    return _openai_batch_output_row(custom_id=custom_id, body=body)
+
+
+def _transform_vertex_embeddings_batch_output_to_openai(
+    vertex_output_rows: Iterable[Mapping[str, Any]],
+    model: str | None,
+) -> tuple[_OpenAIBatchOutputRow, ...]:
+    """
+    Transforms a whole Vertex Gemini Embedding batch output into OpenAI batch output
+    rows, one per OpenAI batch entry, in the order the entries first appear.
+
+    Rows are grouped rather than mapped one to one because a single entry can fan out
+    into several Vertex rows, and Vertex returns them in arbitrary order.
+    """
+    keyed_rows = tuple((_split_vertex_batch_key(row), row) for row in vertex_output_rows)
+    grouped_rows = {
+        custom_id: tuple(group)
+        for custom_id, group in itertools.groupby(sorted(keyed_rows, key=lambda kr: kr[0]), key=lambda kr: kr[0][0])
+    }
+    return tuple(
+        _vertex_embeddings_rows_to_openai_batch_output_row(
+            custom_id=custom_id,
+            vertex_output_rows=tuple(row for _, row in grouped_rows[custom_id]),
+            element_indices=tuple(index for (_, index, _), _ in grouped_rows[custom_id]),
+            element_count=max(total for (_, _, total), _ in grouped_rows[custom_id]),
+            model=model,
+        )
+        for custom_id in dict.fromkeys(custom_id for (custom_id, _, _), _ in keyed_rows)
+    )
+
+
+def _model_from_managed_gcs_url(url: str) -> str | None:
+    """
+    Extracts the model from a LiteLLM-managed Vertex batch GCS url.
+
+    Batch inputs and their sibling outputs are stored under
+    `.../publishers/google/models/<model>/...`, which is the only place the model of an
+    embeddings batch output row can be recovered from; unlike `generateContent`
+    responses, embedding rows carry no `modelVersion`.
+    """
+    match = _MANAGED_GCS_MODEL_PATH_PATTERN.search(unquote(url))
+    return match.group(1) if match else None
+
+
+def _is_embeddings_batch_entry(openai_entry: Mapping[str, Any]) -> bool:
+    """
+    Whether an OpenAI batch JSONL line targets the embeddings endpoint.
+
+    OpenAI puts the target route on each line's `url` (e.g. `/v1/embeddings`); Vertex
+    has no equivalent per-line field, so the route decides which Vertex request shape
+    the line has to be translated into.
+    """
+    url = openai_entry.get("url")
+    if not isinstance(url, str):
+        return False
+    path = url.split("?")[0].rstrip("/")
+    return path == "embeddings" or path.endswith("/embeddings")
+
+
+def _openai_embedding_input_elements(
+    embedding_input: GeminiEmbeddingInput,
+) -> tuple[str | list[str], ...]:
+    """
+    Split an OpenAI `input` into the elements that each get their own embedding.
+
+    A string is one embedding, a flat array is one embedding per element, and a nested
+    array is one combined embedding per inner array, matching the online
+    `batchEmbedContents` path.
+    """
+    if isinstance(embedding_input, list):
+        return tuple(embedding_input)
+    return (embedding_input,)
+
+
+def _vertex_batch_embeddings_key(custom_id: str, index: int, total: int) -> str:
+    """
+    The top-level `key` Vertex echoes back on an embeddings row.
+
+    An entry asking for several embeddings needs several Vertex rows, so its key also
+    carries the element index and the group size; `_split_vertex_batch_key` reads them
+    back out. The `custom_id` is percent-encoded so that a customer one ending in
+    `#<index>/<total>` cannot be mistaken for that tag, which would merge two entries.
+    """
+    encoded_custom_id = quote(custom_id, safe="")
+    return encoded_custom_id if total < 2 else f"{encoded_custom_id}#{index}/{total}"
+
+
+def _vertex_embeddings_row(key: str | None, embed_content_request: Mapping[str, Any]) -> Mapping[str, Any]:
+    """
+    One Vertex Gemini Embedding batch input row.
+
+    The config fields live inside the `EmbedContentRequest` under their snake_case batch
+    names, and the OpenAI `custom_id` rides along in the top-level `key` that Vertex
+    echoes back.
+    """
+    request = {
+        "content": embed_content_request["content"],
+        **{
+            request_field: embed_content_request[gemini_param]
+            for gemini_param, request_field in _EMBED_REQUEST_FIELD_BY_GEMINI_PARAM
+            if gemini_param in embed_content_request
+        },
+    }
+    if key is None:
+        return {"request": request}
+    return {_VERTEX_BATCH_KEY_FIELD: key, "request": request}
+
+
+def _openai_batch_jsonl_entry_to_vertex_embeddings_rows(
+    openai_entry: Mapping[str, Any],
+) -> tuple[Mapping[str, Any], ...]:
+    """
+    Transforms a single OpenAI `/v1/embeddings` batch entry into Vertex Gemini Embedding
+    batch rows, one per requested embedding.
+
+    Example Vertex jsonl
+    {"key": "id_1", "request": {"content": {"parts": [{"text": "Hello World"}]}, "output_dimensionality": 768, "task_type": "RETRIEVAL_DOCUMENT"}}
+
+    Note that `content` is singular (an `EmbedContentRequest`, not a
+    `GenerateContentRequest`) and that the `custom_id` round-trips through the top-level
+    `key`. An `EmbedContentRequest` returns exactly one vector, so an entry whose `input`
+    is an array fans out into one row per element and is reassembled on the way back.
+    The docs put the per-row config in an `embed_content_config` sibling of `request`,
+    but the API rejects that key outright and fails the whole batch job, so the config
+    fields go inside the `EmbedContentRequest` itself.
+
+    API Ref: https://cloud.google.com/vertex-ai/generative-ai/docs/embeddings/batch-prediction-genai-embeddings
+    """
+    openai_request_body = openai_entry.get("body")
+    if not isinstance(openai_request_body, dict):
+        raise TypeError(
+            "`body` on /v1/embeddings batch requests must be a JSON object, but was missing or not an object"
+        )
+    embedding_input = openai_request_body.get("input")
+    if embedding_input is None:
+        raise ValueError("`input` is required on /v1/embeddings batch requests, but was not provided")
+
+    elements = _openai_embedding_input_elements(embedding_input)
+    if not elements:
+        raise ValueError("`input` on /v1/embeddings batch requests must not be empty")
+
+    embed_content_requests = tuple(
+        transform_openai_input_gemini_embed_content(
+            input=element,
+            model=openai_request_body.get("model", ""),
+            optional_params=openai_request_body,
+        )
+        for element in elements
+    )
+    custom_id = openai_entry.get("custom_id")
+    return tuple(
+        _vertex_embeddings_row(
+            key=None
+            if custom_id is None
+            else _vertex_batch_embeddings_key(
+                custom_id=str(custom_id),
+                index=index,
+                total=len(embed_content_requests),
+            ),
+            embed_content_request=embed_content_request,
+        )
+        for index, embed_content_request in enumerate(embed_content_requests)
+    )
+
+
+def _openai_batch_jsonl_entry_to_vertex_rows(
     openai_entry: dict[str, Any],
     map_openai_to_vertex_params: Callable[[dict[str, Any]], dict[str, Any]],
-) -> dict[str, Any]:
+) -> tuple[Mapping[str, Any], ...]:
     """
-    Transforms a single OpenAI JSONL batch entry into its Vertex wrapped request.
+    Transforms a single OpenAI JSONL batch entry into the Vertex rows it maps to.
 
     jsonl body for vertex is {"request": <request_body>}
     Example Vertex jsonl
     {"request":{"contents": [{"role": "user", "parts": [{"text": "What is the relation between the following video and image samples?"}, {"fileData": {"fileUri": "gs://cloud-samples-data/generative-ai/video/animals.mp4", "mimeType": "video/mp4"}}, {"fileData": {"fileUri": "gs://cloud-samples-data/generative-ai/image/cricket.jpeg", "mimeType": "image/jpeg"}}]}]}}
     """
+    if _is_embeddings_batch_entry(openai_entry):
+        return _openai_batch_jsonl_entry_to_vertex_embeddings_rows(openai_entry)
+
     openai_request_body: Final = openai_entry.get("body") or {}
     vertex_request_body: Final = _transform_request_body(
         messages=openai_request_body.get("messages", []),
@@ -209,7 +539,7 @@ def _openai_batch_jsonl_entry_to_vertex_wrapped_request(
             vertex_request_body["labels"] = {}
         _set_litellm_batch_custom_id_labels(vertex_request_body["labels"], custom_id)
 
-    return {"request": vertex_request_body}
+    return ({"request": vertex_request_body},)
 
 
 def _iter_stripped_lines(raw_lines: Iterable[str | bytes]) -> Iterator[str]:
@@ -312,10 +642,10 @@ class _OpenAIToVertexBatchUploadStream(BaseFileUploadStream):
     def _iter_vertex_jsonl_chunks(self) -> Iterator[bytes]:
         first = True
         for entry in _iter_openai_jsonl_entries(self._openai_file_content):
-            wrapped = _openai_batch_jsonl_entry_to_vertex_wrapped_request(entry, self._map_openai_to_vertex_params)
-            prefix = b"" if first else b"\n"
-            first = False
-            yield prefix + json.dumps(wrapped).encode("utf-8")
+            for wrapped in _openai_batch_jsonl_entry_to_vertex_rows(entry, self._map_openai_to_vertex_params):
+                prefix = b"" if first else b"\n"
+                first = False
+                yield prefix + json.dumps(wrapped).encode("utf-8")
 
     def iter_bytes(self) -> Iterator[bytes]:
         return self._iter_vertex_jsonl_chunks()
@@ -667,6 +997,7 @@ class VertexAIFilesConfig(VertexBase, BaseFilesConfig):
                 transformed_content: Final = self._try_transform_vertex_batch_output_to_openai(
                     content=content,
                     logging_obj=logging_obj,
+                    model=_model_from_managed_gcs_url(str(raw_response.request.url)),
                 )
                 if transformed_content != content:
                     # Create a new response with transformed content and updated Content-Length
@@ -688,7 +1019,10 @@ class VertexAIFilesConfig(VertexBase, BaseFilesConfig):
         return HttpxBinaryResponseContent(response=raw_response)
 
     def _try_transform_vertex_batch_output_to_openai(
-        self, content: bytes, logging_obj: LiteLLMLoggingObj | None = None
+        self,
+        content: bytes,
+        logging_obj: LiteLLMLoggingObj | None = None,
+        model: str | None = None,
     ) -> bytes:
         """
         Try to transform Vertex AI batch output to OpenAI format.
@@ -730,7 +1064,7 @@ class VertexAIFilesConfig(VertexBase, BaseFilesConfig):
             # first line is not valid UTF-8/JSON) raises and falls through to the
             # passthrough below, leaving the content untouched.
             first_row: Final = _parse_vertex_batch_output_row(first_line)
-            is_vertex_batch_output: Final = (
+            is_vertex_batch_output: Final = _is_vertex_embeddings_batch_output_row(first_row) or (
                 "request" in first_row
                 and "response" in first_row
                 and "processed_time" in first_row
@@ -763,11 +1097,23 @@ class VertexAIFilesConfig(VertexBase, BaseFilesConfig):
                 request=httpx.Request(method="POST", url="https://example.com"),
             )
 
+            all_lines = itertools.chain((first_line,), lines)
+
+            # Embedding rows are grouped by `custom_id` rather than transformed one at a
+            # time, since an entry that asked for several embeddings comes back as
+            # several rows, in arbitrary order.
+            if _is_vertex_embeddings_batch_output_row(first_row):
+                openai_outputs = _transform_vertex_embeddings_batch_output_to_openai(
+                    vertex_output_rows=(json.loads(line) for line in all_lines),
+                    model=model,
+                )
+                return b"\n".join(json.dumps(openai_output).encode("utf-8") for openai_output in openai_outputs)
+
             # Transform each row straight into the output buffer, so peak memory
             # stays at ~one row plus the output. If any row fails, return the
             # original content unchanged.
             output = bytearray()
-            for line in itertools.chain([first_line], lines):
+            for line in all_lines:
                 try:
                     openai_output = self._transform_single_vertex_batch_output_to_openai(
                         vertex_output=_parse_vertex_batch_output_row(line),
@@ -798,25 +1144,18 @@ class VertexAIFilesConfig(VertexBase, BaseFilesConfig):
         Transform a single Vertex AI batch output line to OpenAI format.
         Uses the existing VertexGeminiConfig transformation for the response.
         """
-        # Extract custom_id from request labels (prefer raw for OpenAI round-trip)
-        request_data: Final = vertex_output.get("request", {})
-        labels: Final[Mapping[str, object]] = request_data.get("labels", {}) or {}
-        custom_id: Final = _get_litellm_batch_custom_id_from_labels(labels)
+        custom_id: Final = _get_litellm_batch_custom_id(vertex_output)
 
         # Check if there's an error
         status: Final = vertex_output.get("status", "")
         has_error: Final = bool(status)
 
         if has_error:
-            return {
-                "id": f"batch_req_{uuid.uuid4()}",
-                "custom_id": custom_id,
-                "response": None,
-                "error": {
-                    "code": "vertex_ai_error",
-                    "message": status,
-                },
-            }
+            return _openai_batch_output_row(
+                custom_id=custom_id,
+                error_code="vertex_ai_error",
+                error_message=status,
+            )
 
         # Transform successful response using existing transformation
         vertex_response: Final = vertex_output.get("response", {})
@@ -842,24 +1181,11 @@ class VertexAIFilesConfig(VertexBase, BaseFilesConfig):
             response_dict: Final = transformed_response.model_dump()
 
             # Return in OpenAI batch format
-            return {
-                "id": f"batch_req_{uuid.uuid4()}",
-                "custom_id": custom_id,
-                "response": {
-                    "status_code": 200,
-                    "request_id": response_dict.get("id", ""),
-                    "body": response_dict,
-                },
-                "error": None,
-            }
+            return _openai_batch_output_row(custom_id=custom_id, body=response_dict)
 
         except Exception as e:
-            return {
-                "id": f"batch_req_{uuid.uuid4()}",
-                "custom_id": custom_id,
-                "response": None,
-                "error": {
-                    "code": "transformation_error",
-                    "message": f"Failed to transform response: {e}",
-                },
-            }
+            return _openai_batch_output_row(
+                custom_id=custom_id,
+                error_code="transformation_error",
+                error_message=f"Failed to transform response: {e}",
+            )

litellm/proxy/management_endpoints/model_management_endpoints.py

@@ -342,16 +342,25 @@ def _validate_ptu_model_info(model_info: Mapping[str, object]) -> None:
         )
 
 
-# The six mirrored pricing fields plus the three remaining fields
+# The mirrored per-token pricing fields plus the three remaining fields
 # Router._inherit_builtin_cache_pricing back-fills from the public cost map. An unset field is
 # what that back-fill targets, so a field left out here is one a PTU deployment still bills.
-_PTU_ZEROED_PRICING_FIELDS: Final = SPECIAL_MODEL_INFO_PARAMS + (
+# tiered_pricing is the one mirrored field that is a table of ranges, not a rate, so it is stored
+# empty (see _PTU_EMPTIED_PRICING_FIELDS): its tiers outrank the zeros written beside them, so
+# dropping it would leave the cost map's tiers billing the traffic the reserved capacity covers.
+_PTU_ZEROED_PRICING_FIELDS: Final = tuple(f for f in SPECIAL_MODEL_INFO_PARAMS if f != "tiered_pricing") + (
     "cache_creation_input_token_cost_above_1hr",
     "cache_creation_input_token_cost_above_200k_tokens",
     "cache_read_input_token_cost_above_200k_tokens",
 )
-_PTU_ZEROED_PRICING: Final[Mapping[str, float]] = MappingProxyType(dict.fromkeys(_PTU_ZEROED_PRICING_FIELDS, 0.0))
-_NO_PRICING_OVERRIDE: Final[Mapping[str, float]] = MappingProxyType({})
+_PTU_EMPTIED_PRICING_FIELDS: Final = frozenset({"tiered_pricing"})
+_PTU_ZEROED_PRICING: Final[Mapping[str, float | tuple[()]]] = MappingProxyType(
+    {
+        **dict.fromkeys(_PTU_ZEROED_PRICING_FIELDS, 0.0),
+        **dict.fromkeys(_PTU_EMPTIED_PRICING_FIELDS, ()),
+    }
+)
+_NO_PRICING_OVERRIDE: Final[Mapping[str, float | tuple[()]]] = MappingProxyType({})
 _EMPTY_MODEL_INFO: Final[Mapping[str, object]] = _NO_PRICING_OVERRIDE
 # Rate fields only. CustomPricingLiteLLMParams also carries settings that are not charges
 # (an embedding's output_vector_size, the regional uplift multipliers), and zeroing one of
@@ -364,6 +373,8 @@ def _is_nonzero_price(value: object) -> bool:
 
 
 def _is_zero_price(value: object) -> bool:
+    if isinstance(value, (list, tuple)):
+        return not value
     return isinstance(value, (int, float)) and not isinstance(value, bool) and value == 0
 
 
@@ -378,7 +389,12 @@ def _raise_if_ptu_deployment_is_priced(*, model_info: Mapping[str, object], supp
         return
     if model_info.get("ptu_count") is None or model_info.get("cost_per_ptu_per_hour") is None:
         return
-    priced: Final = tuple(sorted(field for field in _CUSTOM_PRICING_FIELDS if _is_nonzero_price(supplied.get(field))))
+    priced: Final = tuple(
+        sorted(
+            tuple(field for field in _CUSTOM_PRICING_FIELDS if _is_nonzero_price(supplied.get(field)))
+            + tuple(field for field in _PTU_EMPTIED_PRICING_FIELDS if supplied.get(field))
+        )
+    )
     if not priced:
         return
     raise HTTPException(
@@ -395,7 +411,7 @@ def _ptu_zeroed_pricing(
     model_info: Mapping[str, object],
     litellm_params: Mapping[str, object],
     supplied: Mapping[str, object],
-) -> Mapping[str, float]:
+) -> Mapping[str, float | tuple[()]]:
     """The pricing a PTU deployment must carry, empty unless one is being stored.
 
     Reserved capacity is already billed by the flat cost the rollup writes, so charging the
@@ -432,7 +448,7 @@ def _ptu_pricing_delta(
     model_info: Mapping[str, object],
     litellm_params: Mapping[str, object],
     patch: updateDeployment,
-) -> tuple[Mapping[str, float], frozenset[str]]:
+) -> tuple[Mapping[str, float | tuple[()]], frozenset[str]]:
     """The pricing a patch must write into both blobs, and the pricing it must drop from them.
 
     A patch that takes the deployment off PTU takes the zeroed pricing with it, since the zeros
@@ -454,7 +470,7 @@ def _ptu_pricing_delta(
         return _NO_PRICING_OVERRIDE, frozenset()
     return _NO_PRICING_OVERRIDE, frozenset(
         field
-        for field in _CUSTOM_PRICING_FIELDS.union(_PTU_ZEROED_PRICING_FIELDS)
+        for field in _CUSTOM_PRICING_FIELDS.union(_PTU_ZEROED_PRICING_FIELDS, _PTU_EMPTIED_PRICING_FIELDS)
         if _is_zero_price(model_info.get(field)) or _is_zero_price(litellm_params.get(field))
     )
 
@@ -466,11 +482,16 @@ def _ptu_priced_deployment(model_params: Deployment) -> Deployment:
     override: Final = _ptu_zeroed_pricing(model_info=model_info, litellm_params=litellm_params, supplied=litellm_params)
     if not override:
         return model_params
+    # model_copy validates nothing, so the emptied tier table has to arrive as the list the field
+    # declares or Pydantic warns on every later dump of it
+    stored: Final = MappingProxyType(
+        {key: [] if isinstance(value, tuple) else value for key, value in override.items()}
+    )
     return model_params.model_copy(
         update=MappingProxyType(
             {
-                "litellm_params": model_params.litellm_params.model_copy(update=override),
-                "model_info": model_params.model_info.model_copy(update=override),
+                "litellm_params": model_params.litellm_params.model_copy(update=stored),
+                "model_info": model_params.model_info.model_copy(update=stored),
             }
         )
     )

litellm/router.py

@@ -7635,6 +7635,39 @@ class Router:
                 if backend_value is not None:
                     model_info[field] = backend_value
 
+    @staticmethod
+    def _inherit_builtin_tiered_output_rate(
+        model_info: dict, backend_model: str, custom_llm_provider: str | None
+    ) -> None:
+        """Fill a missing entry-level output rate on a deployment entry whose tier
+        table omits one, from the backend model's built-in cost map entry.
+
+        A deployment's custom pricing is registered as its own standalone
+        ``litellm.model_cost`` entry holding only the supplied fields, and the
+        tiered-cost output fallback reads that same entry, so a tier table that
+        spells out only input-side rates would bill every completion at 0.
+
+        A user-specified ``output_cost_per_token`` always wins. No-op without a
+        tier table, when every tier declares its own output rate, or when the
+        backend model has no canonical entry or no flat output rate:
+        ``get_model_info`` synthesizes a zero for tiered-only backends, and
+        storing that zero would mark the deployment as explicitly priced free.
+        """
+        tiers: Final = model_info.get("tiered_pricing")
+        if not isinstance(tiers, list) or not tiers:
+            return
+        if model_info.get("output_cost_per_token") is not None:
+            return
+        if all(isinstance(tier, dict) and "output_cost_per_token" in tier for tier in tiers):
+            return
+        try:
+            backend_info: Final = litellm.get_model_info(model=backend_model, custom_llm_provider=custom_llm_provider)
+        except Exception:  # noqa: BLE001  # get_model_info raises plain Exception for an unmapped backend model
+            return
+        backend_rate: Final = backend_info.get("output_cost_per_token")
+        if backend_rate:
+            model_info["output_cost_per_token"] = backend_rate
+
     def _create_deployment(
         self,
         deployment_info: dict,
@@ -7670,6 +7703,11 @@ class Router:
                     backend_model=deployment.litellm_params.model,
                     custom_llm_provider=deployment.litellm_params.custom_llm_provider,
                 )
+            Router._inherit_builtin_tiered_output_rate(
+                model_info=_model_info,
+                backend_model=deployment.litellm_params.model,
+                custom_llm_provider=deployment.litellm_params.custom_llm_provider,
+            )
 
             ## REGISTER MODEL INFO IN LITELLM MODEL COST MAP
             Router._register_deployment_in_model_cost(
@@ -8368,6 +8406,11 @@ class Router:
                 backend_model=deployment.litellm_params.model,
                 custom_llm_provider=deployment.litellm_params.custom_llm_provider,
             )
+        Router._inherit_builtin_tiered_output_rate(
+            model_info=_model_info_dict,
+            backend_model=deployment.litellm_params.model,
+            custom_llm_provider=deployment.litellm_params.custom_llm_provider,
+        )
 
         # Register custom pricing in litellm.model_cost.
         # Mirrors _create_deployment() logic to ensure dynamically-added deployments
@@ -8598,6 +8641,11 @@ class Router:
                 backend_model=deployment.litellm_params.model,
                 custom_llm_provider=deployment.litellm_params.custom_llm_provider,
             )
+        Router._inherit_builtin_tiered_output_rate(
+            model_info=model_info,
+            backend_model=deployment.litellm_params.model,
+            custom_llm_provider=deployment.litellm_params.custom_llm_provider,
+        )
         return model_info
 
     @staticmethod

litellm/types/utils.py

@@ -3262,6 +3262,7 @@ class MirroredPricingParams(BaseModel):
     output_cost_per_character: float | None = None
     cache_read_input_token_cost: float | None = None
     cache_creation_input_token_cost: float | None = None
+    tiered_pricing: list[dict[str, Any]] | None = None
 
 
 class CustomPricingLiteLLMParams(MirroredPricingParams):
@@ -3329,7 +3330,6 @@ class CustomPricingLiteLLMParams(MirroredPricingParams):
     output_cost_per_audio_per_second: float | None = None
     search_context_cost_per_query: dict[str, Any] | None = None
     citation_cost_per_token: float | None = None
-    tiered_pricing: list[dict[str, Any]] | None = None
     cache_read_input_token_cost_above_272k_tokens: float | None = None
     cache_read_input_token_cost_above_512k_tokens: float | None = None
     input_cost_per_image_token: float | None = None

model_prices_and_context_window.schema.json

@@ -731,6 +731,10 @@
                 "type": "number",
                 "minimum": 0
               },
+              "cache_creation_input_token_cost": {
+                "type": "number",
+                "minimum": 0
+              },
               "input_cost_per_query": {
                 "type": "number",
                 "minimum": 0

tests/test_litellm/litellm_core_utils/llm_cost_calc/test_llm_cost_calc_utils.py

@@ -587,6 +587,246 @@ def test_generic_cost_per_token_honors_non_standard_above_threshold():
         litellm.model_cost.pop(model, None)
 
 
+def test_generic_cost_per_token_tiered_pricing_charges_cache_creation_at_tier_rate():
+    """Regression for LIT-4375: a tier's cache_creation_input_token_cost must be billed
+    on the generic (provider-agnostic) path, not silently dropped."""
+    model = "litellm-test-tiered-cache-creation"
+    custom_llm_provider = "openrouter"
+    litellm.register_model(
+        {
+            model: {
+                "litellm_provider": custom_llm_provider,
+                "mode": "chat",
+                "tiered_pricing": [
+                    {
+                        "range": [0, 256000],
+                        "input_cost_per_token": 3.25e-07,
+                        "output_cost_per_token": 1.95e-06,
+                        "cache_creation_input_token_cost": 4.063e-07,
+                        "cache_read_input_token_cost": 3.25e-08,
+                    },
+                    {
+                        "range": [256000, 1000000],
+                        "input_cost_per_token": 6.5e-07,
+                        "output_cost_per_token": 3.9e-06,
+                        "cache_creation_input_token_cost": 8.125e-07,
+                        "cache_read_input_token_cost": 6.5e-08,
+                    },
+                ],
+            }
+        }
+    )
+
+    try:
+        usage = Usage(
+            prompt_tokens=300000,  # 200k new + 60k cache creation + 40k cache read
+            completion_tokens=1000,
+            total_tokens=301000,
+            prompt_tokens_details=PromptTokensDetailsWrapper(
+                cached_tokens=40000, cache_creation_tokens=60000
+            ),
+        )
+        prompt_cost, completion_cost = generic_cost_per_token(
+            model=model,
+            usage=usage,
+            custom_llm_provider=custom_llm_provider,
+        )
+
+        expected_prompt = (
+            (200000 * 6.5e-07) + (60000 * 8.125e-07) + (40000 * 6.5e-08)
+        )
+        assert round(prompt_cost, 10) == round(expected_prompt, 10)
+        assert round(completion_cost, 10) == round(1000 * 3.9e-06, 10)
+    finally:
+        litellm.model_cost.pop(model, None)
+
+
+def test_generic_cost_per_token_tiered_pricing_is_all_or_nothing():
+    """Tiered pricing bills the whole request at the tier picked from its input tokens,
+    for any provider, and falls back to flat pricing when no tier matches."""
+    model = "litellm-test-tiered-all-or-nothing"
+    custom_llm_provider = "openrouter"
+    litellm.register_model(
+        {
+            model: {
+                "litellm_provider": custom_llm_provider,
+                "mode": "chat",
+                "input_cost_per_token": 1e-06,
+                "output_cost_per_token": 2e-06,
+                "tiered_pricing": [
+                    {
+                        "range": [0, 32000],
+                        "input_cost_per_token": 4.6e-07,
+                        "output_cost_per_token": 2.3e-06,
+                    },
+                    {
+                        "range": [32000, 128000],
+                        "input_cost_per_token": 7e-07,
+                        "output_cost_per_token": 3.5e-06,
+                    },
+                ],
+            }
+        }
+    )
+
+    try:
+        usage = Usage(prompt_tokens=40000, completion_tokens=1000, total_tokens=41000)
+        prompt_cost, completion_cost = generic_cost_per_token(
+            model=model,
+            usage=usage,
+            custom_llm_provider=custom_llm_provider,
+        )
+        assert round(prompt_cost, 10) == round(40000 * 7e-07, 10)
+        assert round(completion_cost, 10) == round(1000 * 3.5e-06, 10)
+
+        boundary_usage = Usage(prompt_tokens=32000, completion_tokens=10, total_tokens=32010)
+        boundary_prompt_cost, _ = generic_cost_per_token(
+            model=model,
+            usage=boundary_usage,
+            custom_llm_provider=custom_llm_provider,
+        )
+        assert round(boundary_prompt_cost, 10) == round(32000 * 4.6e-07, 10)
+
+        empty_prompt_usage = Usage(prompt_tokens=0, completion_tokens=100, total_tokens=100)
+        empty_prompt_cost, empty_completion_cost = generic_cost_per_token(
+            model=model,
+            usage=empty_prompt_usage,
+            custom_llm_provider=custom_llm_provider,
+        )
+        assert empty_prompt_cost == 0.0
+        assert round(empty_completion_cost, 10) == round(100 * 2e-06, 10)
+    finally:
+        litellm.model_cost.pop(model, None)
+
+
+def test_generic_cost_per_token_tier_without_an_output_rate_bills_the_model_rate():
+    """Regression: a tier table that spells out only input rates served every completion for
+    free, since a tier's missing output rate has no tier-level fallback to stand in for it."""
+    model = "litellm-test-tiered-input-only"
+    custom_llm_provider = "openrouter"
+    litellm.register_model(
+        {
+            model: {
+                "litellm_provider": custom_llm_provider,
+                "mode": "chat",
+                "output_cost_per_token": 2e-06,
+                "output_cost_per_reasoning_token": 5e-06,
+                "tiered_pricing": [{"range": [0, 128000], "input_cost_per_token": 1e-03}],
+            }
+        }
+    )
+
+    try:
+        usage = Usage(
+            prompt_tokens=13,
+            completion_tokens=182,
+            total_tokens=195,
+            completion_tokens_details=CompletionTokensDetailsWrapper(reasoning_tokens=100),
+        )
+        prompt_cost, completion_cost = generic_cost_per_token(
+            model=model,
+            usage=usage,
+            custom_llm_provider=custom_llm_provider,
+        )
+        assert round(prompt_cost, 12) == round(13 * 1e-03, 12)
+        assert round(completion_cost, 12) == round((82 * 2e-06) + (100 * 5e-06), 12)
+    finally:
+        litellm.model_cost.pop(model, None)
+
+
+def test_router_deployment_with_input_only_tiers_bills_completions_at_the_backend_rate():
+    """Regression: the router registers a deployment's custom pricing as a standalone
+    model_cost entry holding only the supplied fields, so an input-only tier table left
+    the output-rate fallback nothing to read and billed every completion at 0."""
+    from litellm import Router
+
+    model_id = "litellm-test-router-tiered-input-only"
+    backend_model = "anthropic/claude-haiku-4-5"
+    backend_output_rate = litellm.get_model_info(backend_model)["output_cost_per_token"]
+    Router(
+        model_list=[
+            {
+                "model_name": "tiered-input-only",
+                "litellm_params": {
+                    "model": backend_model,
+                    "api_key": "sk-test",
+                    "tiered_pricing": [
+                        {"range": [0, 3000], "input_cost_per_token": 3.25e-07},
+                        {"range": [3000, 128000], "input_cost_per_token": 8.125e-07},
+                    ],
+                },
+                "model_info": {"id": model_id},
+            }
+        ]
+    )
+
+    try:
+        usage = Usage(prompt_tokens=21, completion_tokens=4, total_tokens=25)
+        prompt_cost, completion_cost = generic_cost_per_token(
+            model=model_id,
+            usage=usage,
+            custom_llm_provider="anthropic",
+        )
+        assert round(prompt_cost, 12) == round(21 * 3.25e-07, 12)
+        assert round(completion_cost, 12) == round(4 * backend_output_rate, 12)
+        assert backend_output_rate > 0
+    finally:
+        litellm.model_cost.pop(model_id, None)
+
+
+def test_generic_cost_per_token_tiered_pricing_bills_reasoning_at_tier_rate():
+    """Regression: a tier's output_cost_per_reasoning_token must price reasoning tokens
+    on the generic path and in the logged breakdown, not the tier's plain output rate."""
+    model = "litellm-test-tiered-reasoning"
+    custom_llm_provider = "openrouter"
+    litellm.register_model(
+        {
+            model: {
+                "litellm_provider": custom_llm_provider,
+                "mode": "chat",
+                "tiered_pricing": [
+                    {
+                        "range": [0, 256000],
+                        "input_cost_per_token": 4e-07,
+                        "output_cost_per_token": 1.2e-06,
+                        "output_cost_per_reasoning_token": 4e-06,
+                    },
+                    {
+                        "range": [256000, 1000000],
+                        "input_cost_per_token": 1.2e-06,
+                        "output_cost_per_token": 3.6e-06,
+                        "output_cost_per_reasoning_token": 1.2e-05,
+                    },
+                ],
+            }
+        }
+    )
+
+    try:
+        usage = Usage(
+            prompt_tokens=1000,
+            completion_tokens=500,
+            total_tokens=1500,
+            completion_tokens_details=CompletionTokensDetailsWrapper(reasoning_tokens=400),
+        )
+        prompt_cost, completion_cost = generic_cost_per_token(
+            model=model,
+            usage=usage,
+            custom_llm_provider=custom_llm_provider,
+        )
+        assert round(prompt_cost, 12) == round(1000 * 4e-07, 12)
+        assert round(completion_cost, 12) == round((100 * 1.2e-06) + (400 * 4e-06), 12)
+
+        breakdown = get_token_type_cost_breakdown(
+            model=model,
+            custom_llm_provider=custom_llm_provider,
+            usage=usage,
+        )
+        assert round(breakdown.reasoning_cost, 12) == round(400 * 4e-06, 12)
+    finally:
+        litellm.model_cost.pop(model, None)
+
+
 def test_generic_cost_per_token_gpt55():
     """gpt-5.5: base pricing — $5/1M input, $30/1M output, $0.50/1M cached input."""
     model = "gpt-5.5"

tests/test_litellm/llms/dashscope/test_dashscope_cost_calculator.py

@@ -2,13 +2,12 @@
 Test suite for Dashscope cost calculation functionality.
 
 Tests the cost calculation for Dashscope models including:
-- Correctly calculates graduated tiered pricing.
+- All-or-nothing tiered pricing, selected by the request's total input tokens.
 - Falls back to flat-rate pricing for non-tiered models.
-- Handles interactions with cached tokens.
-- Correctly calculates costs for token counts exceeding the highest defined tier.
+- Handles cache read and cache creation tokens.
+- Correctly prices requests exceeding the highest defined tier.
 """
 
-import json
 import math
 import os
 import sys
@@ -22,7 +21,11 @@ import litellm
 from litellm.llms.dashscope.cost_calculator import (
     cost_per_token as dashscope_cost_per_token,
 )
-from litellm.types.utils import Usage, PromptTokensDetailsWrapper
+from litellm.types.utils import (
+    CompletionTokensDetailsWrapper,
+    PromptTokensDetailsWrapper,
+    Usage,
+)
 
 
 class TestDashscopeCostCalculator:
@@ -41,7 +44,6 @@ class TestDashscopeCostCalculator:
         """
         usage = Usage(prompt_tokens=1000, completion_tokens=500)
 
-        # We call the specific calculator for dashscope
         prompt_cost, completion_cost = dashscope_cost_per_token(
             model="qwen-max", usage=usage
         )
@@ -55,7 +57,7 @@ class TestDashscopeCostCalculator:
 
     def test_dashscope_tiered_pricing_within_first_tier(self):
         """
-        Tests the dashscope tiered pricing when token count is entirely within the first tier.
+        Tests the dashscope tiered pricing when the request's input falls in the first tier.
         Uses 'dashscope/qwen-flash' as a real-world example.
         """
         # Tier 1 for qwen-flash is [0, 256,000] tokens
@@ -73,10 +75,10 @@ class TestDashscopeCostCalculator:
         assert math.isclose(prompt_cost, expected_prompt_cost, rel_tol=1e-10)
         assert math.isclose(completion_cost, expected_completion_cost, rel_tol=1e-10)
 
-    def test_dashscope_tiered_pricing_spanning_multiple_tiers(self):
+    def test_dashscope_tiered_pricing_bills_whole_request_at_selected_tier(self):
         """
-        Tests the dashscope tiered pricing with the corrected graduated calculation logic.
-        This is the most important test for validating the fix.
+        Regression: Model Studio tiered pricing is all-or-nothing, not graduated. An input
+        above the first tier's range must bill every token at the higher tier's rate.
         """
         # Tiering for qwen-flash: Tier 1: [0, 256k], Tier 2: [256k, 1M]
         usage = Usage(prompt_tokens=300000, completion_tokens=300000)
@@ -88,23 +90,54 @@ class TestDashscopeCostCalculator:
         tier_1 = model_info["tiered_pricing"][0]
         tier_2 = model_info["tiered_pricing"][1]
 
-        # Expected prompt cost: (256,000 tokens * tier_1_price) + (44,000 tokens * tier_2_price)
-        expected_prompt_cost = (256000 * tier_1["input_cost_per_token"]) + (
-            44000 * tier_2["input_cost_per_token"]
-        )
-
-        # Expected completion cost: (256,000 tokens * tier_1_price) + (44,000 tokens * tier_2_price)
-        expected_completion_cost = (256000 * tier_1["output_cost_per_token"]) + (
-            44000 * tier_2["output_cost_per_token"]
-        )
+        expected_prompt_cost = 300000 * tier_2["input_cost_per_token"]
+        expected_completion_cost = 300000 * tier_2["output_cost_per_token"]
 
         assert math.isclose(prompt_cost, expected_prompt_cost, rel_tol=1e-10)
         assert math.isclose(completion_cost, expected_completion_cost, rel_tol=1e-10)
 
+        graduated_prompt_cost = (256000 * tier_1["input_cost_per_token"]) + (
+            44000 * tier_2["input_cost_per_token"]
+        )
+        assert prompt_cost > graduated_prompt_cost
+
+    def test_dashscope_tiered_pricing_boundary_stays_in_lower_tier(self):
+        """
+        A request of exactly range_end tokens stays in the lower tier, matching the
+        official `0 < Token <= 256K` phrasing.
+        """
+        usage = Usage(prompt_tokens=256000, completion_tokens=1000)
+        prompt_cost, completion_cost = dashscope_cost_per_token(
+            model="qwen-flash", usage=usage
+        )
+
+        tier_1 = litellm.get_model_info("dashscope/qwen-flash")["tiered_pricing"][0]
+
+        assert math.isclose(
+            prompt_cost, 256000 * tier_1["input_cost_per_token"], rel_tol=1e-10
+        )
+        assert math.isclose(
+            completion_cost, 1000 * tier_1["output_cost_per_token"], rel_tol=1e-10
+        )
+
+    def test_dashscope_tiered_pricing_output_uses_input_selected_tier(self):
+        """
+        The tier is chosen by input volume only: a small input with a huge output stays
+        on the first tier's output rate.
+        """
+        usage = Usage(prompt_tokens=1000, completion_tokens=400000)
+        _, completion_cost = dashscope_cost_per_token(model="qwen-flash", usage=usage)
+
+        tier_1 = litellm.get_model_info("dashscope/qwen-flash")["tiered_pricing"][0]
+
+        assert math.isclose(
+            completion_cost, 400000 * tier_1["output_cost_per_token"], rel_tol=1e-10
+        )
+
     def test_dashscope_tiered_pricing_with_caching(self):
         """
-        Tests tiered pricing with cached tokens. This replaces the old, incorrect test.
-        Uses qwen3-coder-plus, which has cache-specific pricing defined.
+        Tests tiered pricing with cached tokens: the tier is selected from the total
+        input (text + cached), and cache reads bill at that tier's cache rate.
         """
         usage = Usage(
             prompt_tokens=50000,  # 10k cached + 40k new
@@ -115,28 +148,43 @@ class TestDashscopeCostCalculator:
 
         prompt_cost, _ = dashscope_cost_per_token(model="qwen3-coder-plus", usage=usage)
 
-        model_info = litellm.get_model_info("dashscope/qwen3-coder-plus")
-        tier_1 = model_info["tiered_pricing"][0]
-        tier_2 = model_info["tiered_pricing"][1]
+        # 50k total input falls in qwen3-coder-plus tier 2 ([32k, 128k])
+        tier_2 = litellm.get_model_info("dashscope/qwen3-coder-plus")["tiered_pricing"][1]
 
-        # 10k cached tokens are all in the first tier
-        expected_cache_cost = 10000 * tier_1["cache_read_input_token_cost"]
-
-        # 40k new tokens: 32k in tier 1, and the remaining 8k in tier 2
-        expected_text_cost = (32000 * tier_1["input_cost_per_token"]) + (
-            8000 * tier_2["input_cost_per_token"]
+        expected_prompt_cost = (40000 * tier_2["input_cost_per_token"]) + (
+            10000 * tier_2["cache_read_input_token_cost"]
         )
 
-        expected_total_prompt_cost = expected_cache_cost + expected_text_cost
+        assert math.isclose(prompt_cost, expected_prompt_cost, rel_tol=1e-10)
 
-        assert math.isclose(prompt_cost, expected_total_prompt_cost, rel_tol=1e-10)
+    def test_dashscope_tiered_pricing_exceeding_highest_tier(self):
+        """
+        Requests above the highest declared range bill entirely at the last tier's rate.
+        """
+        usage = Usage(
+            prompt_tokens=1200000, completion_tokens=1000
+        )  # Max defined range for qwen-flash is 1M
 
-    def _register_string_valued_tiered_model(self, model_key: str) -> None:
-        """Register a model whose tier costs are strings, mimicking YAML config parsing."""
+        prompt_cost, _ = dashscope_cost_per_token(model="qwen-flash", usage=usage)
+
+        tier_2 = litellm.get_model_info("dashscope/qwen-flash")["tiered_pricing"][1]
+
+        assert math.isclose(
+            prompt_cost, 1200000 * tier_2["input_cost_per_token"], rel_tol=1e-10
+        )
+
+    def _register_tiered_model(self, model_key: str, tiered_pricing: list[dict]) -> None:
         litellm.model_cost[model_key] = {
             "litellm_provider": "dashscope",
             "mode": "chat",
-            "tiered_pricing": [
+            "tiered_pricing": tiered_pricing,
+        }
+
+    def _register_string_valued_tiered_model(self, model_key: str) -> None:
+        """Register a model whose tier costs are strings, mimicking YAML config parsing."""
+        self._register_tiered_model(
+            model_key,
+            [
                 {
                     "range": [0, 1000],
                     "input_cost_per_token": "4e-07",
@@ -148,12 +196,12 @@ class TestDashscopeCostCalculator:
                     "output_cost_per_token": "3.2e-06",
                 },
             ],
-        }
+        )
 
     def test_dashscope_tiered_pricing_string_costs_within_tier(self):
         """
-        Regression: YAML-parsed tier costs can be strings (e.g. "4e-07"). Costs that
-        fall entirely within a single tier must still be computed as floats.
+        Regression: YAML-parsed tier costs can be strings (e.g. "4e-07") and must still
+        be computed as floats.
         """
         self._register_string_valued_tiered_model("dashscope/qwen-str-tier-test")
 
@@ -162,18 +210,13 @@ class TestDashscopeCostCalculator:
             model="qwen-str-tier-test", usage=usage
         )
 
-        expected_prompt_cost = 500 * float("4e-07")
-        expected_completion_cost = 200 * float("1.6e-06")
-
-        assert prompt_cost > 0
-        assert completion_cost > 0
-        assert math.isclose(prompt_cost, expected_prompt_cost, rel_tol=1e-10)
-        assert math.isclose(completion_cost, expected_completion_cost, rel_tol=1e-10)
+        assert math.isclose(prompt_cost, 500 * float("4e-07"), rel_tol=1e-10)
+        assert math.isclose(completion_cost, 200 * float("1.6e-06"), rel_tol=1e-10)
 
     def test_dashscope_tiered_pricing_string_costs_exceeding_highest_tier(self):
         """
-        Regression: string-valued tier costs must also be coerced in the
-        remaining-tokens path that charges tokens above the highest tier.
+        Regression: string-valued tier costs must also be coerced on the last-tier
+        fallback path used by requests above the highest range.
         """
         self._register_string_valued_tiered_model("dashscope/qwen-str-tier-test")
 
@@ -182,43 +225,265 @@ class TestDashscopeCostCalculator:
             model="qwen-str-tier-test", usage=usage
         )
 
-        # prompt: 1000 @ tier1 + 1000 @ tier2 + 500 remaining @ tier2 rate
-        expected_prompt_cost = (
-            (1000 * float("4e-07")) + (1000 * float("8e-07")) + (500 * float("8e-07"))
-        )
-        # completion: 1000 @ tier1 + 1000 @ tier2 + 1000 remaining @ tier2 rate
-        expected_completion_cost = (
-            (1000 * float("1.6e-06")) + (1000 * float("3.2e-06")) + (1000 * float("3.2e-06"))
+        assert math.isclose(prompt_cost, 2500 * float("8e-07"), rel_tol=1e-10)
+        assert math.isclose(completion_cost, 3000 * float("3.2e-06"), rel_tol=1e-10)
+
+    def test_dashscope_tiered_cache_creation_tokens_use_tier_rate(self):
+        """
+        Regression (tiered cache creation): cache-creation tokens must bill at the
+        selected tier's cache_creation_input_token_cost, not the input rate.
+        """
+        self._register_tiered_model(
+            "dashscope/qwen-cache-write-test",
+            [
+                {
+                    "range": [0, 256000],
+                    "input_cost_per_token": 3.25e-07,
+                    "output_cost_per_token": 1.95e-06,
+                    "cache_creation_input_token_cost": 4.063e-07,
+                    "cache_read_input_token_cost": 3.25e-08,
+                },
+                {
+                    "range": [256000, 1000000],
+                    "input_cost_per_token": 6.5e-07,
+                    "output_cost_per_token": 3.9e-06,
+                    "cache_creation_input_token_cost": 8.125e-07,
+                    "cache_read_input_token_cost": 6.5e-08,
+                },
+            ],
         )
 
-        assert prompt_cost > 0
-        assert completion_cost > 0
-        assert math.isclose(prompt_cost, expected_prompt_cost, rel_tol=1e-10)
-        assert math.isclose(completion_cost, expected_completion_cost, rel_tol=1e-10)
-
-    def test_dashscope_tiered_pricing_exceeding_highest_tier(self):
-        """
-        Tests tiered pricing when token count exceeds the highest defined tier range.
-        This replaces the old, incorrect test and validates the new fallback logic.
-        """
         usage = Usage(
-            prompt_tokens=1200000, completion_tokens=1000
-        )  # Max defined range for qwen-flash is 1M
+            prompt_tokens=300000,  # 200k new + 60k cache creation + 40k cache read
+            completion_tokens=1000,
+            prompt_tokens_details=PromptTokensDetailsWrapper(
+                cached_tokens=40000, cache_creation_tokens=60000
+            ),
+        )
 
-        prompt_cost, _ = dashscope_cost_per_token(model="qwen-flash", usage=usage)
-
-        model_info = litellm.get_model_info("dashscope/qwen-flash")
-        tier_1 = model_info["tiered_pricing"][0]
-        tier_2 = model_info["tiered_pricing"][1]
-
-        # Expected cost: (tier_1_tokens * tier_1_price) + (tokens_up_to_max_range_in_tier_2 * tier_2_price) + (remaining_tokens * tier_2_price)
-        tokens_in_tier_2_range = 1000000 - 256000
-        remaining_tokens_over_max = 1200000 - 1000000
+        prompt_cost, _ = dashscope_cost_per_token(
+            model="qwen-cache-write-test", usage=usage
+        )
 
         expected_prompt_cost = (
-            (256000 * tier_1["input_cost_per_token"])
-            + (tokens_in_tier_2_range * tier_2["input_cost_per_token"])
-            + (remaining_tokens_over_max * tier_2["input_cost_per_token"])
+            (200000 * 6.5e-07) + (60000 * 8.125e-07) + (40000 * 6.5e-08)
         )
 
         assert math.isclose(prompt_cost, expected_prompt_cost, rel_tol=1e-10)
+
+    def test_dashscope_tiered_cache_creation_falls_back_to_tier_input_rate(self):
+        """
+        Tiers without a cache_creation_input_token_cost bill cache-creation tokens at
+        that tier's input rate.
+        """
+        self._register_tiered_model(
+            "dashscope/qwen-no-cache-write-test",
+            [
+                {
+                    "range": [0, 256000],
+                    "input_cost_per_token": 3.25e-07,
+                    "output_cost_per_token": 1.95e-06,
+                }
+            ],
+        )
+
+        usage = Usage(
+            prompt_tokens=10000,
+            completion_tokens=100,
+            prompt_tokens_details=PromptTokensDetailsWrapper(cache_creation_tokens=4000),
+        )
+
+        prompt_cost, _ = dashscope_cost_per_token(
+            model="qwen-no-cache-write-test", usage=usage
+        )
+
+        assert math.isclose(prompt_cost, 10000 * 3.25e-07, rel_tol=1e-10)
+
+    def test_dashscope_flat_cache_creation_tokens_use_flat_rate(self):
+        """Flat-priced models bill cache-creation tokens at their cache-creation rate."""
+        litellm.model_cost["dashscope/qwen-flat-cache-write-test"] = {
+            "litellm_provider": "dashscope",
+            "mode": "chat",
+            "input_cost_per_token": 3.25e-07,
+            "output_cost_per_token": 1.95e-06,
+            "cache_creation_input_token_cost": 4.063e-07,
+            "cache_read_input_token_cost": 3.25e-08,
+        }
+
+        usage = Usage(
+            prompt_tokens=10000,
+            completion_tokens=100,
+            prompt_tokens_details=PromptTokensDetailsWrapper(
+                cached_tokens=2000, cache_creation_tokens=3000
+            ),
+        )
+
+        prompt_cost, _ = dashscope_cost_per_token(
+            model="qwen-flat-cache-write-test", usage=usage
+        )
+
+        expected_prompt_cost = (
+            (5000 * 3.25e-07) + (3000 * 4.063e-07) + (2000 * 3.25e-08)
+        )
+
+        assert math.isclose(prompt_cost, expected_prompt_cost, rel_tol=1e-10)
+
+    def test_dashscope_tier_without_an_output_rate_bills_the_model_rate(self):
+        """
+        Regression: a tier declaring only an input rate served every completion for free,
+        since a missing tier output rate had no tier-level fallback to stand in for it.
+        """
+        litellm.model_cost["dashscope/qwen-input-only-tier-test"] = {
+            "litellm_provider": "dashscope",
+            "mode": "chat",
+            "output_cost_per_token": 1.6e-06,
+            "tiered_pricing": [{"range": [0, 1000], "input_cost_per_token": 4e-07}],
+        }
+
+        usage = Usage(prompt_tokens=500, completion_tokens=200)
+        prompt_cost, completion_cost = dashscope_cost_per_token(
+            model="qwen-input-only-tier-test", usage=usage
+        )
+
+        assert math.isclose(prompt_cost, 500 * 4e-07, rel_tol=1e-10)
+        assert math.isclose(completion_cost, 200 * 1.6e-06, rel_tol=1e-10)
+
+    def test_dashscope_tier_without_an_output_rate_bills_the_model_reasoning_rate(self):
+        """
+        Regression: a tier declaring only an input rate billed reasoning tokens at the model's
+        plain output rate, ignoring the model's dedicated reasoning rate.
+        """
+        litellm.model_cost["dashscope/qwen-input-only-reasoning-test"] = {
+            "litellm_provider": "dashscope",
+            "mode": "chat",
+            "output_cost_per_token": 1.6e-06,
+            "output_cost_per_reasoning_token": 4e-06,
+            "tiered_pricing": [{"range": [0, 1000], "input_cost_per_token": 4e-07}],
+        }
+
+        usage = Usage(
+            prompt_tokens=500,
+            completion_tokens=200,
+            completion_tokens_details=CompletionTokensDetailsWrapper(reasoning_tokens=150),
+        )
+        _, completion_cost = dashscope_cost_per_token(
+            model="qwen-input-only-reasoning-test", usage=usage
+        )
+
+        assert math.isclose(
+            completion_cost, (50 * 1.6e-06) + (150 * 4e-06), rel_tol=1e-10
+        )
+
+    def test_dashscope_tier_output_rate_wins_over_the_model_reasoning_rate(self):
+        """
+        A tier declaring its own output rate keeps reasoning tokens on that tier rather than
+        mixing in a model-level reasoning rate.
+        """
+        litellm.model_cost["dashscope/qwen-tier-output-reasoning-test"] = {
+            "litellm_provider": "dashscope",
+            "mode": "chat",
+            "output_cost_per_reasoning_token": 4e-06,
+            "tiered_pricing": [
+                {
+                    "range": [0, 1000],
+                    "input_cost_per_token": 4e-07,
+                    "output_cost_per_token": 1.6e-06,
+                }
+            ],
+        }
+
+        usage = Usage(
+            prompt_tokens=500,
+            completion_tokens=200,
+            completion_tokens_details=CompletionTokensDetailsWrapper(reasoning_tokens=150),
+        )
+        _, completion_cost = dashscope_cost_per_token(
+            model="qwen-tier-output-reasoning-test", usage=usage
+        )
+
+        assert math.isclose(completion_cost, 200 * 1.6e-06, rel_tol=1e-10)
+
+    def test_dashscope_model_zero_reasoning_rate_bills_reasoning_free(self):
+        """
+        Regression: a model declaring an explicit zero reasoning rate had it treated as
+        missing, billing reasoning tokens at the plain output rate instead of free.
+        """
+        litellm.model_cost["dashscope/qwen-zero-reasoning-test"] = {
+            "litellm_provider": "dashscope",
+            "mode": "chat",
+            "input_cost_per_token": 4e-07,
+            "output_cost_per_token": 1.6e-06,
+            "output_cost_per_reasoning_token": 0,
+        }
+
+        usage = Usage(
+            prompt_tokens=500,
+            completion_tokens=200,
+            completion_tokens_details=CompletionTokensDetailsWrapper(reasoning_tokens=150),
+        )
+        _, completion_cost = dashscope_cost_per_token(
+            model="qwen-zero-reasoning-test", usage=usage
+        )
+
+        assert math.isclose(completion_cost, 50 * 1.6e-06, rel_tol=1e-10)
+
+    def test_dashscope_tier_zero_reasoning_rate_bills_reasoning_free(self):
+        """
+        Regression: a tier declaring an explicit zero reasoning rate had it treated as
+        missing, billing reasoning tokens at the tier's output rate instead of free.
+        """
+        litellm.model_cost["dashscope/qwen-tier-zero-reasoning-test"] = {
+            "litellm_provider": "dashscope",
+            "mode": "chat",
+            "tiered_pricing": [
+                {
+                    "range": [0, 1000],
+                    "input_cost_per_token": 4e-07,
+                    "output_cost_per_token": 1.6e-06,
+                    "output_cost_per_reasoning_token": 0,
+                }
+            ],
+        }
+
+        usage = Usage(
+            prompt_tokens=500,
+            completion_tokens=200,
+            completion_tokens_details=CompletionTokensDetailsWrapper(reasoning_tokens=150),
+        )
+        _, completion_cost = dashscope_cost_per_token(
+            model="qwen-tier-zero-reasoning-test", usage=usage
+        )
+
+        assert math.isclose(completion_cost, 50 * 1.6e-06, rel_tol=1e-10)
+
+    def test_dashscope_tiered_pricing_zero_input_falls_back_to_flat_rates(self):
+        """
+        No tier can be selected without input tokens, so an empty-prompt request must
+        not be charged at the most expensive tier.
+        """
+        litellm.model_cost["dashscope/qwen-zero-input-test"] = {
+            "litellm_provider": "dashscope",
+            "mode": "chat",
+            "input_cost_per_token": 4e-07,
+            "output_cost_per_token": 1.6e-06,
+            "tiered_pricing": [
+                {
+                    "range": [0, 1000],
+                    "input_cost_per_token": 4e-07,
+                    "output_cost_per_token": 1.6e-06,
+                },
+                {
+                    "range": [1000, 2000],
+                    "input_cost_per_token": 8e-07,
+                    "output_cost_per_token": 3.2e-06,
+                },
+            ],
+        }
+
+        usage = Usage(prompt_tokens=0, completion_tokens=500)
+        prompt_cost, completion_cost = dashscope_cost_per_token(
+            model="qwen-zero-input-test", usage=usage
+        )
+
+        assert prompt_cost == 0.0
+        assert math.isclose(completion_cost, 500 * 1.6e-06, rel_tol=1e-10)

tests/test_litellm/llms/vertex_ai/files/test_vertex_ai_files_streaming.py

@@ -37,7 +37,7 @@ from litellm.llms.vertex_ai.files.transformation import (
     _get_litellm_batch_custom_id_from_labels,
     _iter_openai_jsonl_entries,
     _iter_openai_jsonl_lines,
-    _openai_batch_jsonl_entry_to_vertex_wrapped_request,
+    _openai_batch_jsonl_entry_to_vertex_rows,
 )
 from litellm.types.llms.openai import CreateFileRequest
 
@@ -84,8 +84,9 @@ def _reference_vertex_jsonl_string(cfg: VertexAIFilesConfig, content: str) -> st
     transform, so the streaming path can be checked against it for parity."""
     entries = [json.loads(line) for line in content.splitlines() if line.strip()]
     return "\n".join(
-        json.dumps(_openai_batch_jsonl_entry_to_vertex_wrapped_request(entry, cfg._map_openai_to_vertex_params))
+        json.dumps(row)
         for entry in entries
+        for row in _openai_batch_jsonl_entry_to_vertex_rows(entry, cfg._map_openai_to_vertex_params)
     )
 
 

tests/test_litellm/proxy/management_endpoints/test_ptu_model_settings.py

@@ -15,6 +15,7 @@ from litellm.proxy._types import (
     ReconcileOutcome,
     UserAPIKeyAuth,
 )
+from litellm.litellm_core_utils.llm_cost_calc.utils import generic_cost_per_token
 from litellm.proxy.auth.auth_checks import _is_model_cost_zero
 from litellm.proxy.management_endpoints.model_management_endpoints import (
     _PTU_ZEROED_PRICING_FIELDS,
@@ -37,6 +38,7 @@ from litellm.types.router import (
     updateDeployment,
     updateLiteLLMParams,
 )
+from litellm.types.utils import Usage
 
 
 def test_model_info_accepts_valid_ptu_fields():
@@ -762,7 +764,10 @@ class TestPtuDeploymentsAreNotBilledPerToken:
         assert self._zeroed(model_info={"ptu_count": 15}) == {}
 
     def test_every_field_the_cost_map_could_fill_is_zeroed(self):
-        assert self._zeroed(model_info=self.PTU) == dict.fromkeys(_PTU_ZEROED_PRICING_FIELDS, 0.0)
+        assert self._zeroed(model_info=self.PTU) == {
+            **dict.fromkeys(_PTU_ZEROED_PRICING_FIELDS, 0.0),
+            "tiered_pricing": (),
+        }
 
     def test_nothing_is_zeroed_while_the_feature_is_disabled(self, monkeypatch):
         monkeypatch.delenv(PTU_COST_ATTRIBUTION_ENV_VAR, raising=False)
@@ -777,6 +782,54 @@ class TestPtuDeploymentsAreNotBilledPerToken:
         assert exc.value.status_code == 400
         assert field in str(exc.value.detail)
 
+    def test_a_tiered_price_the_caller_supplies_is_refused(self):
+        """Tier rates bill the traffic per token just as surely as a flat rate does."""
+        with pytest.raises(HTTPException) as exc:
+            self._zeroed(model_info=self.PTU, supplied={"tiered_pricing": [{"range": [0, 100], "input_cost_per_token": 1e-06}]})
+        assert exc.value.status_code == 400
+        assert "tiered_pricing" in str(exc.value.detail)
+
+    def test_tiered_pricing_already_on_the_row_is_emptied_not_zeroed(self):
+        """tiered_pricing is a table of ranges, so the zero the other fields store would not even
+        validate. Dropping it instead would fall back to the cost map's tiers, whose rates outrank
+        the zeros written beside them, so it is stored empty."""
+        tiers = [{"range": [0, 128000], "input_cost_per_token": 3e-06}]
+        priced = _ptu_priced_deployment(
+            Deployment(
+                model_name="tiered",
+                litellm_params=LiteLLM_Params(model="openai/gpt-4o"),
+                model_info=ModelInfo(
+                    id="dep-tiered",
+                    team_id="t",
+                    tiered_pricing=tiers,
+                    ptu_effective_from=datetime.datetime(2020, 1, 1, tzinfo=datetime.timezone.utc),
+                    **self.PTU,
+                ),
+            )
+        )
+        assert priced.litellm_params.tiered_pricing == []
+        assert priced.model_info.tiered_pricing == []
+
+        written = update_db_model(
+            db_model=Deployment(
+                model_name="tiered",
+                litellm_params=LiteLLM_Params(model="openai/gpt-4o", tiered_pricing=tiers),
+                model_info=ModelInfo(id="dep-tiered", team_id="t"),
+            ),
+            updated_patch=updateDeployment(
+                model_info=ModelInfo(
+                    id="dep-tiered",
+                    team_id="t",
+                    ptu_effective_from=datetime.datetime(2020, 1, 1, tzinfo=datetime.timezone.utc),
+                    **self.PTU,
+                )
+            ),
+        )
+        for blob in ("model_info", "litellm_params"):
+            stored = json.loads(written[blob])
+            assert stored["tiered_pricing"] == [], blob
+            assert stored["input_cost_per_token"] == 0, blob
+
     def test_a_price_the_caller_supplies_as_zero_is_accepted(self):
         assert self._zeroed(model_info={**self.PTU, "input_cost_per_token": 0}, supplied={"input_cost_per_token": 0})[
             "input_cost_per_token"
@@ -910,6 +963,32 @@ class TestPtuDeploymentsAreNotBilledPerToken:
         charged = {k: v for k, v in registered.items() if "cost" in k and k != "cost_per_ptu_per_hour" and v}
         assert charged == {}
 
+    def test_the_cost_map_tiers_contribute_no_price_to_a_priced_ptu_deployment(self):
+        """A tier table outranks the zeroed flat rates wherever cost is read, so leaving the
+        deployment's own table unset bills the reserved capacity's traffic at the map's tiers."""
+        priced = _ptu_priced_deployment(
+            Deployment(
+                model_name="ptu-deployment",
+                litellm_params=LiteLLM_Params(model="dashscope/qwen-flash", api_key="fake-key"),
+                model_info=ModelInfo(
+                    id="dep-ptu",
+                    team_id="team-1",
+                    ptu_effective_from=datetime.datetime(2020, 1, 1, tzinfo=datetime.timezone.utc),
+                    **self.PTU,
+                ),
+            )
+        )
+        router = Router(model_list=[priced.to_json(exclude_none=True)])
+        registered = router.get_deployment_model_info(model_id="dep-ptu", model_name="dashscope/qwen-flash")
+        assert registered is not None
+        assert registered["tiered_pricing"] == []
+        assert generic_cost_per_token(
+            model="dashscope/qwen-flash",
+            usage=Usage(prompt_tokens=1000, completion_tokens=100, total_tokens=1100),
+            custom_llm_provider="dashscope",
+            model_info=registered,
+        ) == (0.0, 0.0)
+
     def test_the_zeroed_pricing_does_not_waive_budget_enforcement(self):
         """A zero price otherwise tells auth the model is free and skips every budget check."""
         priced = _ptu_priced_deployment(
@@ -1098,8 +1177,10 @@ class TestPtuDeploymentsAreNotBilledPerToken:
             )
 
         written = add_team_model_to_db.call_args.kwargs["model_params"]
-        assert all(getattr(written.model_info, field, None) == 0 for field in SPECIAL_MODEL_INFO_PARAMS)
+        assert all(getattr(written.model_info, field, None) == 0 for field in SPECIAL_MODEL_INFO_PARAMS if field != "tiered_pricing")
+        assert written.model_info.tiered_pricing == []
         assert all(written.litellm_params.get(field) == 0 for field in _PTU_ZEROED_PRICING_FIELDS)
+        assert written.litellm_params.tiered_pricing == []
 
     @pytest.mark.asyncio
     async def test_model_new_refuses_a_priced_ptu_deployment(self):

tests/test_litellm/test_cost_calculator.py

@@ -42,6 +42,26 @@ def test_cost_per_token_duplicate_openai_prefix_matches_model_cost(monkeypatch):
     assert prompt_usd + completion_usd > 0
 
 
+def test_cost_per_token_tiered_only_model_bills_at_tier_rate(monkeypatch):
+    """
+    Regression: models that publish only tiered_pricing (no top-level per-token rates),
+    e.g. volcengine doubao-seed-2.0, must reach the generic tiered path instead of
+    recording zero spend.
+    """
+    monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
+    monkeypatch.setattr(litellm, "model_cost", litellm.get_model_cost_map(url=""))
+
+    prompt_usd, completion_usd = cost_per_token(
+        model="volcengine/doubao-seed-2-0-pro-260215",
+        prompt_tokens=40000,
+        completion_tokens=500,
+        custom_llm_provider="volcengine",
+    )
+
+    assert prompt_usd == pytest.approx(40000 * 7e-07)
+    assert completion_usd == pytest.approx(500 * 3.5e-06)
+
+
 def test_cost_per_token_non_string_model_does_not_hang():
     """
     The provider-prefix dedup loop must not spin forever when `model` is a

tests/test_litellm/test_model_prices_schema.py

@@ -100,6 +100,24 @@ def test_schema_accepts_minimal_and_unknown_optional_fields(committed_schema: di
     assert validator.is_valid({"some-model": {"litellm_provider": "openai", "brand_new_field": {"nested": True}}})
 
 
+def test_schema_accepts_cache_creation_cost_inside_a_pricing_tier(committed_schema: dict):
+    validator = build_validator(committed_schema)
+    entry = {
+        "litellm_provider": "dashscope",
+        "mode": "chat",
+        "tiered_pricing": [
+            {
+                "range": [0, 256000],
+                "input_cost_per_token": 3.25e-07,
+                "output_cost_per_token": 1.95e-06,
+                "cache_creation_input_token_cost": 4.063e-07,
+                "cache_read_input_token_cost": 3.25e-08,
+            }
+        ],
+    }
+    assert validator.is_valid({"some-model": entry})
+
+
 DATED_VARIANT = re.compile(r"^(.*?)-(\d{4}-\d{2}-\d{2})$")
 SERVICE_TIER_SUFFIXES = ("_flex", "_priority")
 

tests/test_litellm/test_router_model_cost_isolation.py

@@ -1529,3 +1529,58 @@ def test_strategy_router_alias_pricing_never_enters_model_cost(monkeypatch):
     finally:
         litellm.model_cost = saved_model_cost
         _invalidate_model_cost_lowercase_map()
+
+
+def test_inherit_builtin_tiered_output_rate_fills_the_backend_flat_rate():
+    """
+    A deployment entry whose custom tiers publish only input rates would bill
+    completions at 0, so the backend model's flat output rate is copied in at
+    registration.
+    """
+    model_info = {"tiered_pricing": [{"range": [0, 3000], "input_cost_per_token": 3.25e-07}]}
+
+    Router._inherit_builtin_tiered_output_rate(
+        model_info=model_info,
+        backend_model="claude-haiku-4-5",
+        custom_llm_provider="anthropic",
+    )
+
+    backend_rate = litellm.get_model_info(model="claude-haiku-4-5", custom_llm_provider="anthropic")[
+        "output_cost_per_token"
+    ]
+    assert backend_rate > 0
+    assert model_info["output_cost_per_token"] == backend_rate
+
+
+def test_inherit_builtin_tiered_output_rate_never_stores_a_synthesized_zero():
+    """
+    Regression: get_model_info reports output_cost_per_token 0 for a backend that
+    only publishes tiered rates (e.g. dashscope/qwen-flash), and storing that zero
+    would mark the deployment as explicitly priced free.
+    """
+    backend_info = litellm.get_model_info(model="qwen-flash", custom_llm_provider="dashscope")
+    assert backend_info["output_cost_per_token"] == 0
+
+    model_info = {"tiered_pricing": [{"range": [0, 3000], "input_cost_per_token": 3.25e-07}]}
+    Router._inherit_builtin_tiered_output_rate(
+        model_info=model_info,
+        backend_model="qwen-flash",
+        custom_llm_provider="dashscope",
+    )
+
+    assert "output_cost_per_token" not in model_info
+
+
+def test_inherit_builtin_tiered_output_rate_leaves_a_user_rate_alone():
+    model_info = {
+        "tiered_pricing": [{"range": [0, 3000], "input_cost_per_token": 3.25e-07}],
+        "output_cost_per_token": 9e-07,
+    }
+
+    Router._inherit_builtin_tiered_output_rate(
+        model_info=model_info,
+        backend_model="claude-haiku-4-5",
+        custom_llm_provider="anthropic",
+    )
+
+    assert model_info["output_cost_per_token"] == 9e-07

tests/test_litellm/test_utils.py

@@ -1021,6 +1021,7 @@ def test_aaamodel_prices_and_context_window_json_is_valid():
                             "input_cost_per_token": {"type": "number"},
                             "output_cost_per_token": {"type": "number"},
                             "cache_read_input_token_cost": {"type": "number"},
+                            "cache_creation_input_token_cost": {"type": "number"},
                             "output_cost_per_reasoning_token": {"type": "number"},
                             "max_results_range": {
                                 "type": "array",

tests/test_litellm/types/test_router.py

@@ -46,12 +46,30 @@ def test_custom_pricing_params_keeps_every_field_it_had():
 
 @pytest.mark.parametrize("field", SPECIAL_MODEL_INFO_PARAMS)
 def test_deployment_mirrors_pricing_from_litellm_params_onto_model_info(field):
+    value = [{"range": [0, 128000], "input_cost_per_token": 3e-06}] if field == "tiered_pricing" else 3e-06
     deployment = Deployment(
         model_name="my-model",
-        litellm_params=LiteLLM_Params(model="gpt-4o", **{field: 3e-06}),
+        litellm_params=LiteLLM_Params(model="gpt-4o", **{field: value}),
     )
-    assert getattr(deployment.model_info, field) == 3e-06
-    assert deployment.model_info.model_dump(exclude_none=True)[field] == 3e-06
+    assert getattr(deployment.model_info, field) == value
+    assert deployment.model_info.model_dump(exclude_none=True)[field] == value
+
+
+def test_deployment_mirrors_tiered_pricing_onto_model_info():
+    """
+    Regression: tiered_pricing set under a deployment's litellm_params was silently
+    ignored at cost time because the Deployment mirror excluded it, so the logging
+    path never flagged the deployment as custom-priced.
+    """
+    tiers = [
+        {"range": [0, 3000], "input_cost_per_token": 3.25e-07, "output_cost_per_token": 1.95e-06},
+        {"range": [3000, 128000], "input_cost_per_token": 6.5e-07, "output_cost_per_token": 3.9e-06},
+    ]
+    deployment = Deployment(
+        model_name="my-model",
+        litellm_params=LiteLLM_Params(model="anthropic/claude-haiku-4-5", tiered_pricing=tiers),
+    )
+    assert deployment.model_info.tiered_pricing == tiers
 
 
 def test_unset_pricing_is_still_absent_from_dumps():

ui/litellm-dashboard/src/lib/http/schema.d.ts

@@ -35830,6 +35830,10 @@ export interface components {
             team_public_model_name?: string | null;
             /** Tier */
             tier?: ("free" | "paid") | null;
+            /** Tiered Pricing */
+            tiered_pricing?: {
+                [key: string]: unknown;
+            }[] | null;
             /** Updated At */
             updated_at?: string | null;
             /** Updated By */