fix(streaming): report zero cost for explicitly free calls (#51093)

* fix(streaming): report zero cost for explicitly free calls Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(streaming): require a stored entry for free model pricing Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * refactor(streaming): reuse declared pricing lookup Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(streaming): preserve selected free pricing entries Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(streaming): recognize explicitly free pricing tiers Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(streaming): cover invalid pricing tier fallbacks Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: kenneth <kenneth@berri.ai> Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
2026-11-12 08:42:18 +00:00 · 2026-11-11 12:00:11 -08:00 · 2026-11-11 12:00:11 -08:00 · c9d3c8b52f
commit c9d3c8b52f
parent 95ecfbbac6
11 changed files with 448 additions and 13 deletions
--- a/litellm/cost_calculator.py
+++ b/litellm/cost_calculator.py
@ -964,6 +964,19 @@ def _bedrock_registration_has_prices(entry: Mapping[str, object]) -> bool:
    )


+def _bedrock_cost_registration(model: str | None) -> tuple[str, Mapping[str, object]] | None:
+    if model is None:
+        return None
+    return next(
+        (
+            (key, entry)
+            for key, entry in runtime_model_cost_registrations(model, "bedrock")
+            if _bedrock_registration_has_prices(entry)
+        ),
+        None,
+    )
+
+
 def _bedrock_pricing_info(
    model: str | None,
    invoked_model_id: str | None,
@ -972,14 +985,7 @@ def _bedrock_pricing_info(
 ) -> ModelInfo | None:
    if explicit_pricing or custom_llm_provider != "bedrock" or not model:
        return None
-    registration: Final = next(
-        (
-            (key, entry)
-            for key, entry in runtime_model_cost_registrations(model, "bedrock")
-            if _bedrock_registration_has_prices(entry)
-        ),
-        None,
-    )
+    registration: Final = _bedrock_cost_registration(model)
    profile: Final = _cost_map_model_info(invoked_model_id, "bedrock") if invoked_model_id else None
    if profile is None or not _declares_rates(profile):
        return _cost_map_model_info(registration[0], None) if registration is not None else None
@ -2770,7 +2776,14 @@ def pricing_entry_for_cost_calc(
        custom_pricing is True or base_model is not None,
    )
    if bedrock_model_info is not None:
-        return bedrock_model_info["key"], bedrock_model_info
+        key: Final = bedrock_model_info["key"]
+        declared_pricing: Final = declared_cost_map_pricing(key, model or key, custom_llm_provider)
+        registration: Final = _bedrock_cost_registration(model)
+        declared_keys: Final = frozenset(declared_pricing) | (
+            frozenset(registration[1]) if registration is not None else frozenset[str]()
+        )
+        selected_entry: Final[Mapping[str, object]] = bedrock_model_info
+        return key, MappingProxyType({name: value for name, value in selected_entry.items() if name in declared_keys})
    candidates: Final = (selected_model, _get_response_model(completion_response), model)
    resolved: Final = next(
        (info for info in (_cost_map_model_info(name, custom_llm_provider) for name in candidates if name) if info),
@ -2778,7 +2791,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"], declared_cost_map_pricing(resolved["key"], model or resolved["key"], custom_llm_provider)


 def ocr_cost(
--- a/litellm/litellm_core_utils/litellm_logging.py
+++ b/litellm/litellm_core_utils/litellm_logging.py
@ -135,6 +135,7 @@ from litellm.litellm_core_utils.llm_cost_calc.usage_object_transformation import
 )
 from litellm.litellm_core_utils.llm_cost_calc.zero_cost_diagnostic import (
    diagnose_zero_cost,
+    is_explicitly_free_usage,
    zero_cost_warning,
 )
 from litellm.litellm_core_utils.llm_response_utils.convert_dict_to_response import LiteLLMResponseObjectHandler
@ -2417,6 +2418,33 @@ class Logging(LiteLLMLoggingBaseClass):
            usage=usage,
        )

+    def is_usage_explicitly_free(self, result: object) -> bool:
+        usage: Final = get_usage_object(completion_response=result)
+        if usage is None or self.model_call_details.get("response_cost_failure_debug_information") is not None:
+            return False
+        custom_pricing: Final = self.custom_pricing_for(result)
+        if custom_pricing and self.custom_model_info is not None:
+            return is_explicitly_free_usage(usage, self.custom_model_info)
+        if custom_pricing and self.custom_cost_per_token is not None:
+            return is_explicitly_free_usage(usage, self.custom_cost_per_token)
+        custom_llm_provider: Final = self.model_call_details.get("custom_llm_provider")
+        pricing: Final = pricing_entry_for_cost_calc(
+            model=self.model,
+            completion_response=result,
+            custom_llm_provider=self.get_router_pricing_provider() or custom_llm_provider,
+            custom_pricing=custom_pricing,
+            base_model=get_base_model_from_metadata(model_call_details=self.model_call_details),
+            router_model_id=self.get_router_model_id(),
+            region_name=_resolve_mantle_region_for_cost(
+                custom_llm_provider=custom_llm_provider,
+                litellm_params=self.model_call_details.get("litellm_params"),
+            ),
+            litellm_logging_obj=self,
+        )
+        if pricing is None:
+            return False
+        return is_explicitly_free_usage(usage, pricing[1])
+
    def spilled_over_ptu_for(self, result: object) -> bool:
        result_hidden_params: Final = getattr(result, HIDDEN_PARAMS_ATTR, None) or MappingProxyType({})
        additional_headers: Final = (
--- a/litellm/litellm_core_utils/llm_cost_calc/zero_cost_diagnostic.py
+++ b/litellm/litellm_core_utils/llm_cost_calc/zero_cost_diagnostic.py
@ -1,10 +1,12 @@
 from collections.abc import Mapping
 from functools import reduce
+from types import MappingProxyType
 from typing import Final

 from pydantic import TypeAdapter, ValidationError
 from typing_extensions import assert_never

+from litellm.litellm_core_utils.llm_cost_calc.tiered_pricing import select_tier_for_input
 from litellm.litellm_core_utils.llm_cost_calc.utils import parse_prompt_tokens_details
 from litellm.types.utils import (
    CompletionTokensDetailsWrapper,
@ -23,6 +25,7 @@ _TEXT_OUTPUT_RATE: Final = "output_cost_per_token"
 _AUDIO_OUTPUT_RATE: Final = "output_cost_per_audio_token"
 _RATE_KEY_MARKERS: Final = ("cost", "pricing")
 _NESTED_PRICING: Final = TypeAdapter(Mapping[str, object] | tuple[object, ...])
+_PRICING_TIERS: Final = TypeAdapter(tuple[Mapping[str, object], ...])
 _MAX_PRICING_DEPTH: Final = 4


@ -105,6 +108,33 @@ def _is_explicit_zero(value: object) -> bool:
    return not isinstance(value, bool) and isinstance(value, (int, float)) and value == 0


+def _pricing_for_usage(usage: Usage, pricing_entry: Mapping[str, object]) -> Mapping[str, object]:
+    tiered_pricing: Final = pricing_entry.get("tiered_pricing")
+    if not isinstance(tiered_pricing, list):
+        return pricing_entry
+    try:
+        tiers: Final = _PRICING_TIERS.validate_python(tiered_pricing)
+    except ValidationError:
+        return pricing_entry
+    tier: Final = select_tier_for_input(tiers, usage.prompt_tokens)
+    if tier is None or _TEXT_INPUT_RATE not in tier:
+        return pricing_entry
+    cache_rate: Final = tier.get(_CACHE_READ_INPUT_RATE)
+    return MappingProxyType(
+        {
+            **pricing_entry,
+            **tier,
+            _CACHE_READ_INPUT_RATE: cache_rate if cache_rate is not None else tier[_TEXT_INPUT_RATE],
+        }
+    )
+
+
+def is_explicitly_free_usage(usage: Usage, pricing_entry: Mapping[str, object]) -> bool:
+    selected_pricing: Final = _pricing_for_usage(usage, pricing_entry)
+    used_keys: Final = tuple(_effective_pricing_key(key, selected_pricing) for key in used_pricing_keys(usage))
+    return bool(used_keys) and all(_is_explicit_zero(selected_pricing.get(key)) for key in used_keys)
+
+
 def diagnose_zero_cost(
    usage: Usage,
    pricing_model: str,
--- a/litellm/litellm_core_utils/streaming_chunk_builder_utils.py
+++ b/litellm/litellm_core_utils/streaming_chunk_builder_utils.py
@ -1500,7 +1500,11 @@ def _stamp_streaming_usage_cost(usage: Usage, response: ModelResponse, logging_o
    computed_cost: Final = logging_obj.response_cost_calculator(result=response)
    if isinstance(computed_cost, bool) or not isinstance(computed_cost, (int, float)) or computed_cost < 0:
        return
-    if computed_cost > 0 or isinstance(getattr(usage, "cost", None), (int, float)):
+    if (
+        computed_cost > 0
+        or isinstance(getattr(usage, "cost", None), (int, float))
+        or logging_obj.is_usage_explicitly_free(response)
+    ):
        stamp_litellm_bill_on_usage(response, usage, float(computed_cost))


--- a/litellm/litellm_core_utils/streaming_handler.py
+++ b/litellm/litellm_core_utils/streaming_handler.py
@ -114,7 +114,12 @@ def _stamp_usage_cost(
    usage: Final[object] = getattr(usage_chunk, "usage", None)
    if usage is None or isinstance(bill, bool) or not isinstance(bill, (int, float)):
        return
-    if bill <= 0 and not isinstance(getattr(usage, "cost", None), (int, float)):
+    complete_usage: Final = getattr(complete_response, "usage", None)
+    if (
+        bill <= 0
+        and not isinstance(getattr(usage, "cost", None), (int, float))
+        and not (bill == 0 and getattr(complete_usage, "cost", None) == 0)
+    ):
        return
    chunk_hidden: Final = get_hidden_params(usage_chunk)
    if chunk_hidden is not None and UPSTREAM_USAGE_COST in complete_hidden:
--- a/litellm/responses/streaming_iterator.py
+++ b/litellm/responses/streaming_iterator.py
@ -1931,7 +1931,11 @@ def stamp_responses_usage_cost(
        return
    if isinstance(bill, bool) or not isinstance(bill, (int, float)):
        return
-    if bill > 0 or isinstance(usage_obj.cost, (int, float)):
+    if (
+        bill > 0
+        or isinstance(usage_obj.cost, (int, float))
+        or (bill == 0 and logging_obj.is_usage_explicitly_free(priced))
+    ):
        stamp_litellm_bill_on_usage(response_obj, usage_obj, float(bill))


--- a/tests/unit/litellm_core_utils/llm_cost_calc/test_zero_cost_diagnostic.py
+++ b/tests/unit/litellm_core_utils/llm_cost_calc/test_zero_cost_diagnostic.py
@ -6,6 +6,7 @@ import pytest
 from litellm.litellm_core_utils.llm_cost_calc.zero_cost_diagnostic import (
    ZERO_COST_COUNTER_NAME,
    diagnose_zero_cost,
+    is_explicitly_free_usage,
    used_pricing_keys,
    zero_cost_warning,
 )
@ -18,6 +19,85 @@ PRICED_ENTRY: Final = {"input_cost_per_token": 1e-06, "output_cost_per_token": 2
 TEXT_USAGE: Final = Usage(prompt_tokens=10, completion_tokens=20, total_tokens=30)


+@pytest.mark.parametrize(
+    ("usage", "rates", "expected"),
+    [
+        (TEXT_USAGE, FREE_ENTRY, True),
+        (TEXT_USAGE, PRICED_ENTRY, False),
+        (TEXT_USAGE, {}, False),
+        (TEXT_USAGE, {"input_cost_per_token": 0.0}, False),
+        (TEXT_USAGE, {"input_cost_per_token": False, "output_cost_per_token": 0.0}, False),
+        (Usage(prompt_tokens=10, completion_tokens=0), {"input_cost_per_token": 0.0}, True),
+        (Usage(prompt_tokens=0, completion_tokens=0), FREE_ENTRY, False),
+        (
+            Usage(prompt_tokens=10, completion_tokens=0, prompt_tokens_details={"cached_tokens": 10}),
+            FREE_ENTRY,
+            False,
+        ),
+        (
+            Usage(prompt_tokens=10, completion_tokens=0, prompt_tokens_details={"cached_tokens": 10}),
+            {"input_cost_per_token": 0.0},
+            True,
+        ),
+    ],
+)
+def test_free_usage_requires_explicit_zero_rates_for_every_used_token_kind(
+    usage: Usage, rates: Mapping[str, object], expected: bool
+) -> None:
+    assert is_explicitly_free_usage(usage, rates) is expected
+
+
+@pytest.mark.parametrize(
+    ("input_tokens", "tier_output_rate", "expected"),
+    [(10, 0.0, True), (11, 0.0, False), (10, 0.001, False), (10, None, False)],
+)
+def test_free_usage_uses_the_selected_tier_and_preserves_missing_output_prices(
+    input_tokens: int, tier_output_rate: float | None, expected: bool
+) -> None:
+    rates: Final = {
+        "input_cost_per_token": 0.01,
+        "tiered_pricing": [
+            {"range": [0, 10], "input_cost_per_token": 0.0, "output_cost_per_token": tier_output_rate},
+            {"range": [10, 20], "input_cost_per_token": 0.01, "output_cost_per_token": 0.02},
+        ],
+    }
+
+    assert is_explicitly_free_usage(Usage(prompt_tokens=input_tokens, completion_tokens=5), rates) is expected
+
+
+@pytest.mark.parametrize("cache_rate", (None, 0.01))
+def test_free_cached_usage_uses_the_selected_tier_cache_rate_or_its_input_rate(cache_rate: float | None) -> None:
+    rates: Final = {
+        "input_cost_per_token": 0.01,
+        "cache_read_input_token_cost": 0.02,
+        "tiered_pricing": [
+            {"range": [0, 20], "input_cost_per_token": 0.0, "cache_read_input_token_cost": cache_rate}
+        ],
+    }
+    usage: Final = Usage(prompt_tokens=10, completion_tokens=0, prompt_tokens_details={"cached_tokens": 10})
+
+    assert is_explicitly_free_usage(usage, rates) is (cache_rate is None)
+
+
+@pytest.mark.parametrize(
+    "tiers",
+    [
+        ["invalid tier"],
+        [],
+        [{"range": [0, 20], "output_cost_per_token": 0.0}],
+        [{"range": "invalid range", "input_cost_per_token": 0.0, "output_cost_per_token": 0.0}],
+    ],
+)
+def test_invalid_or_inactive_tiers_do_not_make_paid_usage_free(tiers: object) -> None:
+    rates: Final = {
+        "input_cost_per_token": 0.01,
+        "output_cost_per_token": 0.02,
+        "tiered_pricing": tiers,
+    }
+
+    assert is_explicitly_free_usage(Usage(prompt_tokens=10, completion_tokens=5), rates) is False
+
+
 def test_missing_pricing_key_names_every_rate_the_usage_needs() -> None:
    diagnostic = diagnose_zero_cost(
        usage=TEXT_USAGE, pricing_model="dep-1", pricing_entry=PER_SECOND_ENTRY, calculation_failed=False
--- a/tests/unit/litellm_core_utils/test_litellm_logging.py
+++ b/tests/unit/litellm_core_utils/test_litellm_logging.py
@ -79,6 +79,64 @@ from typing import List, Optional
 from litellm.types.utils import LiteLLMCommonStrings


+@pytest.mark.parametrize("metadata_key", ("metadata", "litellm_metadata"))
+def test_free_usage_uses_deployment_metadata_prices(metadata_key: str) -> None:
+    obj: Final = Logging(
+        model="metadata-free-deployment",
+        messages=[],
+        stream=True,
+        call_type="completion",
+        start_time=1.0,
+        litellm_call_id="metadata-free",
+        function_id="test",
+    )
+    obj.litellm_params = {metadata_key: {"model_info": {"input_cost_per_token": 0.0, "output_cost_per_token": 0.0}}}
+    response: Final = ModelResponse(
+        model="metadata-free-deployment", usage=Usage(prompt_tokens=10, completion_tokens=5, total_tokens=15)
+    )
+
+    assert obj.is_usage_explicitly_free(response) is True
+
+
+@pytest.mark.parametrize("pricing_source", ("model_info", "token_rates"))
+@pytest.mark.parametrize("rate", (0.0, 0.001))
+@pytest.mark.parametrize("calculation_failed", (False, True))
+def test_per_call_free_prices_override_catalog_unless_pricing_failed(
+    monkeypatch: pytest.MonkeyPatch, pricing_source: str, rate: float, calculation_failed: bool
+) -> None:
+    model: Final = "per-call-free-price-test"
+    monkeypatch.setitem(
+        litellm.model_cost,
+        model,
+        {"litellm_provider": "openai", "mode": "chat", "input_cost_per_token": 0.01, "output_cost_per_token": 0.02},
+    )
+    obj: Final = Logging(
+        model=model,
+        messages=[],
+        stream=True,
+        call_type="completion",
+        start_time=1.0,
+        litellm_call_id="per-call-free",
+        function_id="test",
+    )
+    obj.litellm_params = {"input_cost_per_token": rate, "output_cost_per_token": rate}
+    if pricing_source == "model_info":
+        obj.set_per_call_pricing(
+            {
+                **litellm.get_model_info(model=model, custom_llm_provider="openai"),
+                "input_cost_per_token": rate,
+                "output_cost_per_token": rate,
+            }
+        )
+    else:
+        obj.custom_cost_per_token = {"input_cost_per_token": rate, "output_cost_per_token": rate}
+    if calculation_failed:
+        obj.model_call_details["response_cost_failure_debug_information"] = {"error": "pricing failed"}
+    response: Final = ModelResponse(model=model, usage=Usage(prompt_tokens=10, completion_tokens=5, total_tokens=15))
+
+    assert obj.is_usage_explicitly_free(response) is (rate == 0.0 and not calculation_failed)
+
+
@pytest.fixture
 def logging_obj():
    return LitellmLogging(
--- a/tests/unit/litellm_core_utils/test_streaming_chunk_builder_utils.py
+++ b/tests/unit/litellm_core_utils/test_streaming_chunk_builder_utils.py
@ -10,6 +10,7 @@ import respx

 import litellm
 from litellm import ChatCompletionUsageBlock, stream_chunk_builder
+from litellm.litellm_core_utils.litellm_logging import Logging
 from litellm.litellm_core_utils.streaming_chunk_builder_utils import (
    ChunkProcessor,
    stream_chunk_builder_text_completion,
@ -31,6 +32,84 @@ from litellm.types.utils import (
 )


+@pytest.mark.parametrize("routed_provider", (False, True))
+@pytest.mark.parametrize(
+    ("profile_rates", "registered_rates", "expected_cost"),
+    [
+        (
+            {"input_cost_per_token": 0.01, "output_cost_per_token": 0.02},
+            {"input_cost_per_token": 0.0, "output_cost_per_token": 0.0},
+            0.0,
+        ),
+        (
+            {"input_cost_per_token": 0.01, "output_cost_per_token": 0.02},
+            {"input_cost_per_token": 0.001, "output_cost_per_token": 0.002},
+            0.02,
+        ),
+        ({"input_cost_per_token": 0.01, "output_cost_per_token": 0.02}, {"input_cost_per_token": 0.0}, 0.1),
+        ({"input_cost_per_token": 0.0}, {"input_cost_per_token": 0.0}, None),
+    ],
+)
+def test_bedrock_stream_cost_uses_registered_rates_over_profile_prices(
+    monkeypatch: pytest.MonkeyPatch,
+    profile_rates: dict[str, float],
+    registered_rates: dict[str, float],
+    expected_cost: float | None,
+    routed_provider: bool,
+) -> None:
+    model: Final = "openai.stream-pricing-test"
+    profile: Final = f"us.{model}"
+    monkeypatch.setattr(
+        litellm,
+        "model_cost",
+        {
+            **litellm.model_cost,
+            profile: {"litellm_provider": "bedrock", "mode": "chat", **profile_rates},
+        },
+    )
+    litellm.register_model({model: registered_rates}, persist_across_reloads=False)
+    logging_obj: Final = Logging(
+        model=model,
+        messages=[],
+        stream=True,
+        call_type="completion",
+        start_time=1.0,
+        litellm_call_id="bedrock-free-stream",
+        function_id="test",
+    )
+    logging_obj.optional_params = {}
+    logging_obj.litellm_params = (
+        {"metadata": {"model_info": {"pricing_provider": "bedrock"}}} if routed_provider else {}
+    )
+    logging_obj.model_call_details.update(
+        custom_llm_provider="openai" if routed_provider else "bedrock", model_id=profile
+    )
+    chunks: Final = [
+        ModelResponseStream(
+            id="bedrock-free-stream",
+            created=1,
+            model=model,
+            choices=[StreamingChoices(index=0, delta=Delta(content="hi"), finish_reason="stop")],
+        ),
+        ModelResponseStream(
+            id="bedrock-free-stream",
+            created=1,
+            model=model,
+            choices=[],
+            usage=Usage(prompt_tokens=10, completion_tokens=5, total_tokens=15),
+        ),
+    ]
+
+    response: Final = stream_chunk_builder(chunks, logging_obj=logging_obj)
+
+    assert response is not None
+    assert response.usage.prompt_tokens == 10
+    assert response.usage.completion_tokens == 5
+    assert getattr(response.usage, "cost", None) == (
+        pytest.approx(expected_cost) if expected_cost is not None else None
+    )
+
+
 def test_get_combined_tool_content():
    chunks = [
        ModelResponseStream(
--- a/tests/unit/litellm_core_utils/test_streaming_handler.py
+++ b/tests/unit/litellm_core_utils/test_streaming_handler.py
@ -7649,6 +7649,85 @@ def test_chunk_creator_preserves_hidden_provider_specific_fields_from_parsed_chu
    assert assembled._hidden_params["provider_specific_fields"] == {"traffic_type": "ON_DEMAND_FLEX"}


+@pytest.mark.parametrize("sync_mode", [True, False])
+@pytest.mark.parametrize(
+    ("rates", "expected_cost"),
+    [
+        ({"input_cost_per_token": 0.0, "output_cost_per_token": 0.0}, 0.0),
+        ({"input_cost_per_token": 1e-6, "output_cost_per_token": 2e-6}, 20e-6),
+        ({"input_cost_per_token": 0.0}, None),
+        ({}, None),
+        (
+            {
+                "input_cost_per_token": 0.01,
+                "output_cost_per_token": 0.02,
+                "tiered_pricing": [
+                    {"range": [0, 128000], "input_cost_per_token": 0.0, "output_cost_per_token": 0.0}
+                ],
+            },
+            0.0,
+        ),
+        (
+            {
+                "tiered_pricing": [
+                    {"range": [0, 128000], "input_cost_per_token": 0.0, "output_cost_per_token": 0.0}
+                ]
+            },
+            0.0,
+        ),
+    ],
+)
+async def test_stream_usage_distinguishes_free_from_unpriced_calls(
+    monkeypatch: pytest.MonkeyPatch, sync_mode: bool, rates: dict[str, object], expected_cost: float | None
+) -> None:
+    model: Final = "stream-zero-cost-regression"
+    monkeypatch.setitem(litellm.model_cost, model, {"litellm_provider": "openai", "mode": "chat", **rates})
+    logging_obj: Final = Logging(
+        model=model,
+        messages=[{"role": "user", "content": "hi"}],
+        stream=True,
+        call_type="completion" if sync_mode else "acompletion",
+        start_time=1.0,
+        litellm_call_id="free-stream",
+        function_id="test",
+    )
+    logging_obj.optional_params = {}
+    logging_obj.litellm_params = {}
+    logging_obj.stream_options = {"include_usage": True}
+    logging_obj.model_call_details["custom_llm_provider"] = "openai"
+    chunks: Final = [
+        ModelResponseStream(
+            id="free-stream",
+            created=1,
+            model=model,
+            choices=[StreamingChoices(index=0, delta=Delta(content="hi"), finish_reason="stop")],
+        ),
+        ModelResponseStream(
+            id="free-stream",
+            created=1,
+            model=model,
+            choices=[],
+            usage=Usage(prompt_tokens=10, completion_tokens=5, total_tokens=15),
+        ),
+    ]
+    stream: Final = CustomStreamWrapper(
+        completion_stream=ModelResponseListIterator(model_responses=chunks),
+        model=model,
+        custom_llm_provider="openai",
+        logging_obj=logging_obj,
+        stream_options={"include_usage": True},
+    )
+    emitted: Final = tuple(stream) if sync_mode else tuple([chunk async for chunk in stream])
+    usage_chunks: Final = tuple(chunk for chunk in emitted if getattr(chunk, "usage", None) is not None)
+
+    assert len(usage_chunks) == 1
+    assert usage_chunks[0].usage.prompt_tokens == 10
+    assert usage_chunks[0].usage.completion_tokens == 5
+    assert getattr(usage_chunks[0].usage, "cost", None) == (
+        pytest.approx(expected_cost) if expected_cost is not None else None
+    )
+
+
 def test_the_usage_chunk_carries_the_bill_the_builder_priced_for_the_spend_row() -> None:
    from litellm.litellm_core_utils.streaming_handler import (
        _stamp_usage_cost,  # pyright: ignore[reportPrivateUsage]  # the helper under test
--- a/tests/unit/responses/test_streaming_iterator.py
+++ b/tests/unit/responses/test_streaming_iterator.py
@ -861,6 +861,61 @@ def _responses_api_response_with_usage() -> ResponsesAPIResponse:
    )


+@pytest.mark.parametrize(
+    ("rates", "expected_cost"),
+    [
+        ({"input_cost_per_token": 0.0, "output_cost_per_token": 0.0}, 0.0),
+        ({"input_cost_per_token": 1e-6, "output_cost_per_token": 2e-6}, 140e-6),
+        ({"input_cost_per_token": 0.0}, None),
+        ({}, None),
+        (
+            {
+                "input_cost_per_token": 0.01,
+                "output_cost_per_token": 0.02,
+                "tiered_pricing": [
+                    {"range": [0, 128000], "input_cost_per_token": 0.0, "output_cost_per_token": 0.0}
+                ],
+            },
+            0.0,
+        ),
+        (
+            {
+                "tiered_pricing": [
+                    {"range": [0, 128000], "input_cost_per_token": 0.0, "output_cost_per_token": 0.0}
+                ]
+            },
+            0.0,
+        ),
+    ],
+)
+def test_responses_stream_usage_distinguishes_free_from_unpriced_calls(
+    monkeypatch: pytest.MonkeyPatch, rates: dict[str, object], expected_cost: float | None
+) -> None:
+    from litellm.responses.streaming_iterator import stamp_responses_usage_cost
+
+    model: Final = "responses-zero-cost-regression"
+    monkeypatch.setitem(litellm.model_cost, model, {"litellm_provider": "openai", "mode": "chat", **rates})
+    response: Final = _responses_api_response_with_usage().model_copy(update={"model": model})
+    logging_obj: Final = LiteLLMLoggingObj(
+        model=model,
+        messages=[],
+        stream=True,
+        call_type="responses",
+        start_time=1.0,
+        litellm_call_id="free-response",
+        function_id="test",
+    )
+    logging_obj.optional_params = {}
+    logging_obj.litellm_params = {}
+    logging_obj.model_call_details["custom_llm_provider"] = "openai"
+
+    stamp_responses_usage_cost(response, logging_obj)
+
+    assert getattr(response.usage, "cost", None) == (
+        pytest.approx(expected_cost) if expected_cost is not None else None
+    )
+
+
 def test_stamp_responses_usage_cost_stamps_computed_cost():
    from litellm.responses.streaming_iterator import stamp_responses_usage_cost