refactor(proxy): share the cost map pricing lookup for the response cost header

The header check now reuses the same raw entry, fill-missing and rule lookup that
cost model selection uses, so the two can't drift apart. It also counts every declared
rate field, so a free model priced only per request keeps its 0.0 header

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

litellm/cost_calculator.py

@@ -979,7 +979,7 @@ def _bedrock_pricing_info(
 
 
 def cost_map_key_prices_anything(key: str) -> bool:
-    entry: Final = raw_cost_map_entry(key)
+    entry: Final = _raw_cost_map_entry(key)
     return entry is not None and _cost_map_entry_prices_anything(entry)
 
 
@@ -1102,20 +1102,19 @@ def _cost_map_model_name(model: str, custom_llm_provider: str | None, region_nam
     return _strip_unregistered_leading_segments(f"{provider_prefix}/{model}", region_name)
 
 
-def _rule_declares_rates(model: str, custom_llm_provider: str | None) -> bool:
-    rule_info: Final = unmapped_model_generalization(model, custom_llm_provider)
-    return rule_info is not None and _cost_map_entry_prices_anything(rule_info)
+def declared_cost_map_pricing(key: str, model: str, custom_llm_provider: str | None) -> Mapping[str, object]:
+    entry: Final = _raw_cost_map_entry(key)
+    if entry is None:
+        return unmapped_model_generalization(model, custom_llm_provider) or MappingProxyType({})
+    fill_missing: Final = mapped_model_fill_missing(key, entry)
+    return entry if fill_missing is None else MappingProxyType({**fill_missing, **entry})
 
 
 def _resolves_to_rates(model: str, custom_llm_provider: str | None) -> bool:
     key: Final = cached_cost_map_key(model, custom_llm_provider)
-    if key is None:
-        return False
-    entry: Final = raw_cost_map_entry(key)
-    if entry is None:
-        return _rule_declares_rates(model, custom_llm_provider)
-    fill_missing: Final = mapped_model_fill_missing(key, entry)
-    return _cost_map_entry_prices_anything(entry if fill_missing is None else {**fill_missing, **entry})
+    return key is not None and _cost_map_entry_prices_anything(
+        declared_cost_map_pricing(key, model, custom_llm_provider)
+    )
 
 
 def _served_or_priced_requested_model(
@@ -2420,6 +2419,9 @@ def completion_cost(
         raise e
 
 
+_ADDITIONAL_HEADERS: Final = TypeAdapter(Mapping[str, object])
+
+
 def get_response_cost_from_hidden_params(
     hidden_params: Mapping[str, object] | BaseModel,
 ) -> float | None:
@@ -2429,7 +2431,7 @@ def get_response_cost_from_hidden_params(
     additional_headers: Final = params.get("additional_headers")
     if not isinstance(additional_headers, Mapping):
         return None
-    headers: Final = TypeAdapter(Mapping[str, object]).validate_python(additional_headers)
+    headers: Final = _ADDITIONAL_HEADERS.validate_python(additional_headers)
     return _billable_header_cost(headers.get("llm_provider-x-litellm-response-cost"))
 
 
@@ -2623,7 +2625,7 @@ def _deployment_model_info(
 ) -> ModelInfo | None:
     if not custom_pricing:
         return None
-    registered_deployment_info: Final = raw_cost_map_entry(router_model_id) if router_model_id is not None else None
+    registered_deployment_info: Final = _raw_cost_map_entry(router_model_id) if router_model_id is not None else None
     if registered_deployment_info is not None:
         return cast(ModelInfo, registered_deployment_info)  # cast-ok: router registers deployment prices under its id
     if litellm_logging_obj is None:
@@ -2696,7 +2698,7 @@ def _priced_candidates_before_unpriced_rules(candidates: tuple[_CostCandidate, .
     )
 
 
-def raw_cost_map_entry(key: str) -> Mapping[str, object] | None:
+def _raw_cost_map_entry(key: str) -> Mapping[str, object] | None:
     raw_entry: Final = litellm.model_cost.get(key)
     return raw_entry if isinstance(raw_entry, Mapping) else None
 
@@ -2740,7 +2742,7 @@ def pricing_entry_for_cost_calc(
     )
     if resolved is None:
         return None
-    return resolved["key"], raw_cost_map_entry(resolved["key"]) or resolved
+    return resolved["key"], _raw_cost_map_entry(resolved["key"]) or resolved
 
 
 def ocr_cost(

litellm/litellm_core_utils/litellm_logging.py

@@ -56,10 +56,10 @@ from litellm.constants import (
 from litellm.cost_calculator import (
     RealtimeAPITokenUsageProcessor,
     ResponsesWebSocketTokenUsageProcessor,
+    declared_cost_map_pricing,
     get_response_cost_from_hidden_params,
     get_usage_object,
     pricing_entry_for_cost_calc,
-    raw_cost_map_entry,
     select_model_name_for_cost_calc,
 )
 from litellm.exceptions import (
@@ -216,14 +216,11 @@ from litellm.types.utils import (
     TranscriptionResponse,
     Usage,
     batch_chat_usage,
-    is_server_derived_pricing_key,
 )
 from litellm.types.videos.main import VideoObject
 from litellm.utils import (
     get_base_model_from_metadata,
-    mapped_model_fill_missing,
     print_verbose,
-    unmapped_model_generalization,
 )
 
 from ..integrations.argilla import ArgillaLogger
@@ -2615,19 +2612,8 @@ class Logging(LiteLLMLoggingBaseClass):
         model_info: Final = payload["model_map_information"]["model_map_value"]
         if model_info is None:
             return False
-        raw_entry: Final = raw_cost_map_entry(model_info["key"])
-        rule_entry: Final = (
-            unmapped_model_generalization(model_info["key"], payload["custom_llm_provider"])
-            if raw_entry is None
-            else mapped_model_fill_missing(model_info["key"], raw_entry)
-        )
-        prices: Final = MappingProxyType({**(rule_entry or {}), **(raw_entry or {})})
-        return any(
-            value is not None
-            and ("cost" in key or "pricing" in key)
-            and (is_server_derived_pricing_key(key) or key in DEPLOYMENT_SCOPED_PRICING_FIELDS)
-            for key, value in prices.items()
-        )
+        prices: Final = declared_cost_map_pricing(model_info["key"], model_info["key"], payload["custom_llm_provider"])
+        return any(value is not None and ("cost" in key or "pricing" in key) for key, value in prices.items())
 
     def prepare_response_cost(self, response: object, now: datetime.datetime) -> PreparedCost | None:
         if self.prepared_cost is not None:

tests/unit/proxy/spend_tracking/test_cost_finalization.py

@@ -680,11 +680,14 @@ async def test_finalized_header_distinguishes_missing_prices_from_free_models(
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
-    "source", ["unpriced", "cache", "non-inference", "provider", "provider-header", "per-token", "per-second"]
+    "source",
+    ["unpriced", "cache", "non-inference", "provider", "provider-header", "per-token", "per-second", "passthrough"],
 )
 @pytest.mark.parametrize("guardrail_cost", [0.0, 0.1])
 async def test_zero_cost_headers_preserve_known_free_calls_and_guardrail_charges(
-    source: Literal["unpriced", "cache", "non-inference", "provider", "provider-header", "per-token", "per-second"],
+    source: Literal[
+        "unpriced", "cache", "non-inference", "provider", "provider-header", "per-token", "per-second", "passthrough"
+    ],
     guardrail_cost: float,
 ) -> None:
     model: Final = "unmapped-header-cost"
@@ -711,6 +714,8 @@ async def test_zero_cost_headers_preserve_known_free_calls_and_guardrail_charges
             logging.custom_cost_per_token = {"input_cost_per_token": 0.0, "output_cost_per_token": 0.0}
         case "per-second":
             logging.custom_cost_per_second = 0.0
+        case "passthrough":
+            logging.model_call_details["passthrough_logging_payload"] = {"cost_per_request": 0.0}
         case _:
             pass
     finalized: Final = await finalize_response_cost(logging, result, None, NOW, guardrail_cost=guardrail_cost)
@@ -726,6 +731,35 @@ async def test_zero_cost_headers_preserve_known_free_calls_and_guardrail_charges
     )
 
 
+@pytest.mark.asyncio
+async def test_free_model_priced_only_per_request_keeps_the_zero_cost_header(monkeypatch: pytest.MonkeyPatch) -> None:
+    model: Final = "gpt-free-per-request-header"
+    monkeypatch.setitem(
+        litellm.model_cost, model, {"litellm_provider": "openai", "mode": "chat", "input_cost_per_request": 0.0}
+    )
+    logging: Final = Logging(
+        model=model,
+        messages=[],
+        stream=False,
+        call_type="acompletion",
+        start_time=NOW,
+        litellm_call_id="per-request-header",
+        function_id="per-request-header",
+    )
+    logging.update_environment_variables(
+        model=model, custom_llm_provider="openai", litellm_params={}, optional_params={}
+    )
+    result: Final = ModelResponse(model=model, usage=Usage(prompt_tokens=10, completion_tokens=5, total_tokens=15))
+    finalized: Final = await finalize_response_cost(logging, result, None, NOW)
+    assert finalized is not None and finalized.amount == 0.0
+    headers: Final = ProxyBaseLLMRequestProcessing.get_custom_headers(
+        user_api_key_dict=UserAPIKeyAuth(),
+        response_cost=finalized.amount,
+        litellm_logging_obj=logging,
+    )
+    assert headers["x-litellm-response-cost"] == "0.0"
+
+
 @pytest.mark.asyncio
 async def test_free_off_peak_pricing_keeps_the_finalized_zero_cost_header(monkeypatch: pytest.MonkeyPatch) -> None:
     model: Final = "gpt-free-off-peak-header"