more robust cost calculation

routstr/payment/cost_calculation.py

@@ -316,6 +316,44 @@ def _resolve_provider_fee(model_id: str) -> float:
     return float(providers[0].provider_fee)
 
 
+def _rate_weighted_input_fraction(
+    response_data: dict,
+    input_tokens: int,
+    cache_read_tokens: int,
+    cache_creation_tokens: int,
+    output_tokens: int,
+) -> float:
+    """Fraction of a lump-sum cost attributable to the input side.
+
+    The upstream gave a single total USD figure with no input/output breakdown,
+    so the split has to be reconstructed. A raw token-count split (input + cache
+    vs output) overstates the input share whenever cheap cache-read tokens
+    dominate — a cache read costs ~10x less than a regular input token yet counts
+    the same. Weight each bucket by its per-token rate instead. Fall back to a
+    token-count split when rates are unavailable (fixed pricing / unknown model).
+    """
+    input_weight = float(input_tokens + cache_read_tokens + cache_creation_tokens)
+    output_weight = float(output_tokens)
+
+    try:
+        rates = _get_pricing_rates(response_data)
+    except ValueError:
+        rates = None
+    if rates is not None:
+        input_rate, output_rate, cache_read_rate, cache_creation_rate = rates
+        input_weight = (
+            input_tokens * input_rate
+            + cache_read_tokens * cache_read_rate
+            + cache_creation_tokens * cache_creation_rate
+        )
+        output_weight = output_tokens * output_rate
+
+    total_weight = input_weight + output_weight
+    if total_weight <= 0:
+        return 0.0
+    return input_weight / total_weight
+
+
 def _calculate_from_usd_cost(
     usd_cost: float,
     input_usd: float,
@@ -339,15 +377,16 @@ def _calculate_from_usd_cost(
         input_msats = int((input_usd * sats_per_usd) * 1000)
         output_msats = int((output_usd * sats_per_usd) * 1000)
     else:
-        effective_input_tokens = (
-            input_tokens + cache_read_tokens + cache_creation_tokens
-        )
-        total_tokens = effective_input_tokens + output_tokens
-        input_msats = (
-            int(cost_in_msats * effective_input_tokens / total_tokens)
-            if total_tokens > 0
-            else 0
+        # No explicit split: weight by per-token rates so cheap cache reads do
+        # not inflate the input share (a raw token-count split would).
+        input_fraction = _rate_weighted_input_fraction(
+            response_data,
+            input_tokens,
+            cache_read_tokens,
+            cache_creation_tokens,
+            output_tokens,
         )
+        input_msats = int(cost_in_msats * input_fraction)
         output_msats = cost_in_msats - input_msats
 
     logger.info(

routstr/payment/models.py

@@ -85,45 +85,139 @@ class Model(BaseModel):
         return hash(self.id)
 
 
+def _litellm_entry(*keys: str) -> dict | None:
+    """First litellm cost-map entry matching any of the given exact keys."""
+    import litellm
+
+    for key in keys:
+        candidate = litellm.model_cost.get(key)
+        if isinstance(candidate, dict):
+            return candidate
+    return None
+
+
+def _family_tokens(lowered_id: str) -> list[str]:
+    """Candidate family tokens derived from an id, most-specific first.
+
+    The provider prefix (``deepseek`` in ``deepseek/x``) and the leading segment
+    of the model name (``deepseek`` in ``deepseek-v4-flash``). Tokens shorter
+    than 4 chars are dropped so short noise like ``gpt`` cannot match unrelated
+    litellm keys.
+    """
+    provider, sep, rest = lowered_id.partition("/")
+    name = rest if sep else provider
+    tokens: list[str] = []
+    if sep and len(provider) >= 4:
+        tokens.append(provider)
+    leading = name.split("-", 1)[0]
+    if len(leading) >= 4 and leading not in tokens:
+        tokens.append(leading)
+    return tokens
+
+
+def _family_reference(lowered_id: str) -> dict | None:
+    """First litellm entry sharing a family token AND carrying a cache-read rate.
+
+    Generic, data-driven fallback for vanity ids that proxies invent
+    (``deepseek-v4-flash``) which match no litellm key directly. Any provider
+    whose litellm snapshots price cache reads (deepseek, anthropic, ...) is
+    matched without a hand-maintained family list. The family token must be the
+    key's root segment (``deepseek/...`` / ``deepseek-...``), not appear anywhere
+    in it — that excludes reseller-prefixed snapshots (``deepinfra/deepseek/...``,
+    ``novita/deepseek/...``) whose cache markup differs from the native provider.
+    Keys are scanned in sorted order so the chosen reference is deterministic.
+    """
+    import litellm
+
+    for token in _family_tokens(lowered_id):
+        for key in sorted(litellm.model_cost):
+            lowered_key = key.lower()
+            if not (
+                lowered_key == token
+                or lowered_key.startswith(token + "/")
+                or lowered_key.startswith(token + "-")
+            ):
+                continue
+            entry = litellm.model_cost.get(key)
+            if not isinstance(entry, dict):
+                continue
+            ref_input = entry.get("input_cost_per_token")
+            ref_read = entry.get("cache_read_input_token_cost")
+            if (
+                isinstance(ref_input, (int, float))
+                and ref_input > 0
+                and isinstance(ref_read, (int, float))
+                and ref_read > 0
+            ):
+                return entry
+    return None
+
+
 def backfill_cache_pricing(model_id: str, pricing: Pricing) -> Pricing:
     """Fill missing cache rates from litellm's bundled cost map.
 
-    The OpenRouter model feed omits ``input_cache_read``/``input_cache_write``
-    for many models (most DeepSeek entries, openai/gpt-4o, ...). Without a
-    cache rate, billing falls back to the full input rate, which overcharges
-    cache reads (DeepSeek hits are 10x cheaper) and undercharges Anthropic
-    cache writes (1.25x). litellm ships per-model USD rates keyed by the exact
-    OpenRouter id (deepseek/deepseek-chat) or by the bare model name
-    (gpt-4o, claude-sonnet-4-5), so both spellings are tried.
+    The OpenRouter model feed omits ``input_cache_read`` / ``input_cache_write``
+    for many models. Without a cache rate, billing falls back to the full input
+    rate, overcharging cache reads (DeepSeek hits are ~10x cheaper). Two
+    strategies are tried in order:
+
+    1. **Exact match** — litellm ships absolute per-token USD rates keyed by the
+       upstream id (``deepseek/deepseek-chat``) or its bare model name
+       (``gpt-4o``); both spellings are tried and copied directly.
+    2. **Family ratio** — vanity ids proxies invent (``deepseek-v4-flash``) match
+       no litellm key. A reference entry for the same provider/family is found by
+       generic scan (no hand-maintained list) and its ``cache_read/input``
+       *ratio* is scaled by THIS model's own input price — correct even when the
+       vanity model is priced differently from the reference.
 
     Rates already present (e.g. provided by OpenRouter) are authoritative and
-    never overwritten. Unknown models are returned unchanged.
+    never overwritten. Models matching no family are returned unchanged.
     """
     needs_read = (pricing.input_cache_read or 0.0) <= 0.0
     needs_write = (pricing.input_cache_write or 0.0) <= 0.0
     if not (needs_read or needs_write):
         return pricing
 
-    import litellm
-
-    info: dict | None = None
-    for key in (model_id, model_id.split("/", 1)[-1]):
-        candidate = litellm.model_cost.get(key)
-        if isinstance(candidate, dict):
-            info = candidate
-            break
-    if info is None:
-        return pricing
-
     updated = Pricing.parse_obj(pricing.dict())
+
+    # 1. Exact match: absolute per-token USD rates apply directly.
+    info = _litellm_entry(model_id, model_id.split("/", 1)[-1])
+    if info is not None:
+        if needs_read:
+            read_rate = info.get("cache_read_input_token_cost")
+            if isinstance(read_rate, (int, float)) and read_rate > 0:
+                updated.input_cache_read = float(read_rate)
+                needs_read = False
+        if needs_write:
+            write_rate = info.get("cache_creation_input_token_cost")
+            if isinstance(write_rate, (int, float)) and write_rate > 0:
+                updated.input_cache_write = float(write_rate)
+                needs_write = False
+
+    if not (needs_read or needs_write):
+        return updated
+
+    # 2. Family ratio for vanity ids (or fields the exact entry left unpriced).
+    if pricing.prompt <= 0.0:
+        return updated
+    reference = _family_reference(model_id.lower())
+    if reference is None:
+        logger.debug(
+            "No litellm cache-rate reference for model family",
+            extra={"model_id": model_id},
+        )
+        return updated
+    ref_input = reference.get("input_cost_per_token")
+    if not isinstance(ref_input, (int, float)) or ref_input <= 0:
+        return updated
     if needs_read:
-        read_rate = info.get("cache_read_input_token_cost")
-        if isinstance(read_rate, (int, float)) and read_rate > 0:
-            updated.input_cache_read = float(read_rate)
+        ref_read = reference.get("cache_read_input_token_cost")
+        if isinstance(ref_read, (int, float)) and ref_read > 0:
+            updated.input_cache_read = pricing.prompt * (ref_read / ref_input)
     if needs_write:
-        write_rate = info.get("cache_creation_input_token_cost")
-        if isinstance(write_rate, (int, float)) and write_rate > 0:
-            updated.input_cache_write = float(write_rate)
+        ref_write = reference.get("cache_creation_input_token_cost")
+        if isinstance(ref_write, (int, float)) and ref_write > 0:
+            updated.input_cache_write = pricing.prompt * (ref_write / ref_input)
     return updated
 
 

routstr/upstream/base.py

@@ -39,6 +39,7 @@ from ..payment.models import (
     list_models,
 )
 from ..payment.price import sats_usd_price
+from ..payment.usage import normalize_usage
 from ..wallet import recieve_token, send_token
 from . import messages_dispatch
 from .cache_breakpoints import (
@@ -166,6 +167,11 @@ class BaseUpstreamProvider:
         ``cache_creation_input_tokens`` fields are left in place for clients
         that want the breakdown.
 
+        Only Anthropic-shaped responses that lack ``prompt_tokens`` need their
+        additive cache fields folded into ``input_tokens``. OpenAI/litellm/
+        DeepSeek-shaped responses report ``prompt_tokens`` as an inclusive
+        total already, so it must never be inflated by top-level cache fields.
+
         For Anthropic-shaped responses (``input_tokens`` present), the cache
         fields are forced to ``0`` when the upstream omitted them, so the
         client always sees a consistent shape.
@@ -185,20 +191,13 @@ class BaseUpstreamProvider:
         except (TypeError, ValueError):
             return
         extra = cache_read + cache_creation
-        if extra <= 0:
+        if extra <= 0 or "prompt_tokens" in usage:
             return
         if "input_tokens" in usage:
             try:
                 usage["input_tokens"] = int(usage.get("input_tokens") or 0) + extra
             except (TypeError, ValueError):
                 pass
-        if "prompt_tokens" in usage:
-            try:
-                usage["prompt_tokens"] = (
-                    int(usage.get("prompt_tokens") or 0) + extra
-                )
-            except (TypeError, ValueError):
-                pass
 
     def _apply_provider_field(self, response_json: object) -> None:
         """Stamp the routstr ``provider`` field onto an upstream response payload.
@@ -1661,6 +1660,26 @@ class BaseUpstreamProvider:
                         total_cost, _coerce_usd(usage_or_root.get(field))
                     )
 
+            def _absorb_usage(usage: object) -> None:
+                nonlocal input_tokens, output_tokens
+                nonlocal cache_read_input_tokens, cache_creation_input_tokens
+                normalized = normalize_usage(usage)
+                if normalized is None:
+                    return
+                input_tokens += normalized.input_tokens
+                output_tokens += normalized.output_tokens
+                # Anthropic message streams can restate the same cumulative
+                # cache snapshot across events; keep the existing max() behavior
+                # while allowing OpenAI/litellm/DeepSeek fields to be normalized.
+                cache_read_input_tokens = max(
+                    cache_read_input_tokens,
+                    normalized.cache_read_tokens,
+                )
+                cache_creation_input_tokens = max(
+                    cache_creation_input_tokens,
+                    normalized.cache_write_tokens,
+                )
+
             async def finalize_without_usage() -> bytes | None:
                 nonlocal usage_finalized
                 if usage_finalized:
@@ -1726,62 +1745,11 @@ class BaseUpstreamProvider:
                                             changed = True
 
                                         if usage := msg.get("usage"):
-                                            input_tokens += usage.get("input_tokens", 0)
-                                            output_tokens += usage.get(
-                                                "output_tokens", 0
-                                            )
-                                            # Anthropic's `message_start.usage`
-                                            # carries the cumulative cache
-                                            # snapshot for the prompt — pick
-                                            # the max() so subsequent
-                                            # `message_delta.usage` events
-                                            # (which only restate the same
-                                            # numbers) don't double-count.
-                                            cache_read_input_tokens = max(
-                                                cache_read_input_tokens,
-                                                int(
-                                                    usage.get(
-                                                        "cache_read_input_tokens", 0
-                                                    )
-                                                    or 0
-                                                ),
-                                            )
-                                            cache_creation_input_tokens = max(
-                                                cache_creation_input_tokens,
-                                                int(
-                                                    usage.get(
-                                                        "cache_creation_input_tokens",
-                                                        0,
-                                                    )
-                                                    or 0
-                                                ),
-                                            )
+                                            _absorb_usage(usage)
                                             _absorb_usd(usage)
 
                                         if usage := data.get("usage"):
-                                            input_tokens += usage.get("input_tokens", 0)
-                                            output_tokens += usage.get(
-                                                "output_tokens", 0
-                                            )
-                                            cache_read_input_tokens = max(
-                                                cache_read_input_tokens,
-                                                int(
-                                                    usage.get(
-                                                        "cache_read_input_tokens", 0
-                                                    )
-                                                    or 0
-                                                ),
-                                            )
-                                            cache_creation_input_tokens = max(
-                                                cache_creation_input_tokens,
-                                                int(
-                                                    usage.get(
-                                                        "cache_creation_input_tokens",
-                                                        0,
-                                                    )
-                                                    or 0
-                                                ),
-                                            )
+                                            _absorb_usage(usage)
                                             _absorb_usd(usage)
                                         # Some upstreams attach cost fields at
                                         # the event root rather than nested

routstr/upstream/messages_dispatch.py

@@ -30,6 +30,7 @@ import litellm
 from ..core import get_logger
 from ..core.exceptions import UpstreamError
 from ..payment.models import Model
+from ..payment.usage import normalize_usage
 
 logger = get_logger(__name__)
 
@@ -308,10 +309,12 @@ def annotate_event(event: dict, requested_model: str | None) -> AnnotatedEvent:
     def _accumulate(usage: dict) -> None:
         nonlocal in_tokens, out_tokens, cache_read_tokens, cache_create_tokens
         nonlocal total_cost, input_cost, output_cost
-        in_tokens += int(usage.get("input_tokens") or 0)
-        out_tokens += int(usage.get("output_tokens") or 0)
-        cache_read_tokens += int(usage.get("cache_read_input_tokens") or 0)
-        cache_create_tokens += int(usage.get("cache_creation_input_tokens") or 0)
+        normalized = normalize_usage(usage)
+        if normalized is not None:
+            in_tokens += normalized.input_tokens
+            out_tokens += normalized.output_tokens
+            cache_read_tokens += normalized.cache_read_tokens
+            cache_create_tokens += normalized.cache_write_tokens
         total_cost += _coerce_float(usage.get("total_cost"))
         input_cost += _coerce_float(usage.get("input_cost"))
         output_cost += _coerce_float(usage.get("output_cost"))

tests/unit/test_cache_pricing.py

@@ -117,6 +117,46 @@ def test_backfill_unknown_model_unchanged() -> None:
     assert result.input_cache_write == 0.0
 
 
+def test_backfill_vanity_deepseek_id_uses_generic_family_ratio() -> None:
+    """The reported bug: a proxy exposes ``deepseek-v4-flash`` — a vanity id
+    litellm has never heard of. Without a family fallback its cache reads bill at
+    the full input rate (~10x overcharge on hits). The DeepSeek family's read
+    discount is found by generic litellm scan (no hard-coded list) and its
+    cache/input *ratio* applied to the model's own input price."""
+    pricing = Pricing(prompt=5e-07, completion=1.5e-06)
+
+    result = backfill_cache_pricing("deepseek-v4-flash", pricing)
+
+    ref = litellm.model_cost["deepseek/deepseek-chat"]
+    ratio = ref["cache_read_input_token_cost"] / ref["input_cost_per_token"]
+    assert result.input_cache_read == pytest.approx(pricing.prompt * ratio)
+    assert result.input_cache_read < pricing.prompt  # sanity: it's a discount
+
+
+def test_backfill_vanity_id_scales_to_own_input_price() -> None:
+    """Family fallback uses the *ratio*, not the reference's absolute rate, so a
+    vanity model priced differently from the reference is billed against its own
+    input price — the 10x input gap is preserved in the derived read rate."""
+    cheap = backfill_cache_pricing(
+        "deepseek-mini", Pricing(prompt=1e-07, completion=2e-07)
+    )
+    pricey = backfill_cache_pricing(
+        "deepseek-max", Pricing(prompt=1e-06, completion=2e-06)
+    )
+
+    assert pricey.input_cache_read == pytest.approx(cheap.input_cache_read * 10)
+
+
+def test_backfill_vanity_id_no_family_match_unchanged() -> None:
+    """A vanity id whose family token matches no litellm key stays untouched."""
+    result = backfill_cache_pricing(
+        "mystery-flash-9", Pricing(prompt=1e-06, completion=2e-06)
+    )
+
+    assert result.input_cache_read == 0.0
+    assert result.input_cache_write == 0.0
+
+
 def test_provider_fee_applies_to_backfilled_cache_rates() -> None:
     """Backfill happens before the provider fee, so cache rates carry the
     same markup as every other price component."""

tests/unit/test_cost_calculation_caching.py

@@ -431,3 +431,38 @@ async def test_null_usage_block(mock_fixed_pricing: None) -> None:
     assert isinstance(result, MaxCostData)
     assert result.input_tokens == 0
     assert result.cache_read_input_tokens == 0
+
+
+# ============================================================================
+# USD-path input/output split — weight by rate, not raw token count
+# ============================================================================
+def test_usd_split_weights_by_rate_not_token_count() -> None:
+    """A lump-sum USD cost with no explicit input/output breakdown is split by
+    per-token *rate*. Cheap cache-read tokens (~10x discount) must not inflate
+    the input share the way a raw token-count split does."""
+    from routstr.payment.cost_calculation import _rate_weighted_input_fraction
+
+    # rates: (input, output, cache_read, cache_write); relative scale only.
+    with patch(
+        "routstr.payment.cost_calculation._get_pricing_rates",
+        return_value=(1.0, 4.0, 0.1, 0.1),
+    ):
+        frac = _rate_weighted_input_fraction({"model": "x"}, 100, 900, 0, 100)
+
+    # input weight = 100*1 + 900*0.1 = 190; output = 100*4 = 400; total = 590.
+    assert frac == pytest.approx(190 / 590)
+    # The old raw token-count split gave (100+900)/1100 ≈ 0.909 — wildly skewed.
+    assert frac < 0.5
+
+
+def test_usd_split_falls_back_to_token_count_without_rates() -> None:
+    """When rates are unavailable (fixed pricing / unknown model) the split
+    degrades gracefully to the raw token-count proportion."""
+    from routstr.payment.cost_calculation import _rate_weighted_input_fraction
+
+    with patch(
+        "routstr.payment.cost_calculation._get_pricing_rates", return_value=None
+    ):
+        frac = _rate_weighted_input_fraction({"model": "x"}, 100, 900, 0, 100)
+
+    assert frac == pytest.approx(1000 / 1100)

tests/unit/test_messages_litellm_dispatch.py

@@ -649,6 +649,110 @@ async def test_streaming_handles_iterator_yielding_raw_sse_bytes() -> None:
     assert combined["model"] == "openai/gpt-4o-mini"
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("usage", "expected"),
+    [
+        (
+            {
+                "prompt_tokens": 10_000,
+                "completion_tokens": 500,
+                "prompt_cache_hit_tokens": 9_000,
+                "prompt_cache_miss_tokens": 1_000,
+            },
+            {
+                "input_tokens": 1_000,
+                "output_tokens": 500,
+                "cache_read_input_tokens": 9_000,
+                "cache_creation_input_tokens": 0,
+            },
+        ),
+        (
+            {
+                "prompt_tokens": 10_000,
+                "completion_tokens": 100,
+                "prompt_tokens_details": {
+                    "cached_tokens": 5_000,
+                    "cache_creation_tokens": 2_000,
+                },
+            },
+            {
+                "input_tokens": 3_000,
+                "output_tokens": 100,
+                "cache_read_input_tokens": 5_000,
+                "cache_creation_input_tokens": 2_000,
+            },
+        ),
+    ],
+)
+async def test_streaming_litellm_messages_normalizes_cache_usage_dialects(
+    usage: dict[str, Any], expected: dict[str, int]
+) -> None:
+    provider = _make_provider()
+    key = _make_key()
+    model = _make_model()
+    session = _make_session()
+    body = _anthropic_request_body(stream=True)
+
+    async def fake_chunks() -> AsyncIterator[dict]:
+        yield {
+            "type": "message_start",
+            "message": {
+                "id": "msg_1",
+                "type": "message",
+                "role": "assistant",
+                "model": "openai/gpt-4o-mini",
+                "content": [],
+            },
+        }
+        yield {"type": "message_delta", "delta": {}, "usage": usage}
+        yield {"type": "message_stop"}
+
+    captured: dict[str, Any] = {}
+
+    async def fake_adjust(
+        fresh_key: Any, combined_data: Any, sess: Any, max_cost: int, usage: Any = None
+    ) -> dict:
+        captured["combined_data"] = json.loads(json.dumps(combined_data))
+        return {"total_msats": 999, "total_usd": 0.0001}
+
+    fake_session = MagicMock()
+    fake_session.get = AsyncMock(return_value=key)
+
+    class FakeSessionCtx:
+        async def __aenter__(self) -> Any:
+            return fake_session
+
+        async def __aexit__(self, *args: Any) -> None:
+            return None
+
+    with (
+        patch(
+            "litellm.anthropic.messages.acreate",
+            new=AsyncMock(return_value=fake_chunks()),
+        ),
+        patch(
+            "routstr.upstream.base.adjust_payment_for_tokens",
+            new=AsyncMock(side_effect=fake_adjust),
+        ),
+        patch("routstr.upstream.base.create_session", new=lambda: FakeSessionCtx()),
+    ):
+        result = await provider._forward_messages_via_litellm(
+            request_body=body,
+            key=key,
+            session=session,
+            max_cost_for_model=10_000,
+            model_obj=model,
+        )
+        assert isinstance(result, StreamingResponse)
+        async for _ in result.body_iterator:
+            pass
+
+    combined = captured["combined_data"]
+    for field, value in expected.items():
+        assert combined["usage"][field] == value
+
+
 # ---------------------------------------------------------------------------
 # x-cashu non-streaming
 # ---------------------------------------------------------------------------

tests/unit/test_streaming_sse_providers.py

@@ -20,6 +20,7 @@ comment ever reaches the client. That invariant is exactly what the buggy
 
 import json
 from collections.abc import AsyncGenerator
+from typing import cast
 from unittest.mock import AsyncMock, MagicMock
 
 import pytest
@@ -88,6 +89,15 @@ def _data_payloads(out: list[bytes]) -> list[bytes]:
     return payloads
 
 
+def _last_adjustment_input() -> dict:
+    mock = cast(AsyncMock, base.adjust_payment_for_tokens)
+    await_args = mock.await_args
+    assert await_args is not None
+    adjustment_input = await_args.args[1]
+    assert isinstance(adjustment_input, dict)
+    return adjustment_input
+
+
 def _assert_clean(out: list[bytes]) -> list[dict]:
     """Core invariant: every data line is [DONE] or valid JSON; no comments leak."""
     blob = b"".join(out)
@@ -107,6 +117,103 @@ def _assert_clean(out: list[bytes]) -> list[dict]:
     return objs
 
 
+def test_fold_cache_does_not_inflate_inclusive_prompt_tokens() -> None:
+    """OpenAI/litellm/DeepSeek prompt_tokens already includes cache tokens."""
+    usage = {
+        "prompt_tokens": 10000,
+        "completion_tokens": 100,
+        "cache_read_input_tokens": 5000,
+        "cache_creation_input_tokens": 2000,
+    }
+
+    BaseUpstreamProvider._fold_cache_into_input_tokens(usage)
+
+    assert usage["prompt_tokens"] == 10000
+    assert usage["cache_read_input_tokens"] == 5000
+    assert usage["cache_creation_input_tokens"] == 2000
+
+
+def test_fold_cache_adds_only_anthropic_additive_input_tokens() -> None:
+    """Anthropic native input_tokens excludes cache fields and remains folded."""
+    usage = {
+        "input_tokens": 300,
+        "output_tokens": 100,
+        "cache_read_input_tokens": 500,
+        "cache_creation_input_tokens": 2000,
+    }
+
+    BaseUpstreamProvider._fold_cache_into_input_tokens(usage)
+
+    assert usage["input_tokens"] == 2800
+
+
+@pytest.mark.asyncio
+async def test_deepseek_streaming_usage_preserved_for_cost_adjustment() -> None:
+    """Streaming chat usage keeps DeepSeek cache fields for normalize_usage."""
+    usage = {
+        "prompt_tokens": 10000,
+        "completion_tokens": 500,
+        "prompt_cache_hit_tokens": 9000,
+        "prompt_cache_miss_tokens": 1000,
+    }
+    chunks = [
+        b'data: {"id":"x","model":"deepseek-chat","choices":[{"delta":{"content":"ok"}}]}\n\n',
+        b"data: "
+        + json.dumps(
+            {
+                "id": "x",
+                "model": "deepseek-chat",
+                "choices": [],
+                "usage": usage,
+            }
+        ).encode()
+        + b"\n\n",
+        b"data: [DONE]\n\n",
+    ]
+
+    out = await _drive(chunks)
+
+    _assert_clean(out)
+    adjustment_input = _last_adjustment_input()
+    for key, value in usage.items():
+        assert adjustment_input["usage"][key] == value
+
+
+@pytest.mark.asyncio
+async def test_openai_litellm_streaming_usage_preserved_for_cost_adjustment() -> None:
+    """Streaming chat usage keeps nested and top-level cache fields."""
+    usage = {
+        "prompt_tokens": 10000,
+        "completion_tokens": 100,
+        "cache_read_input_tokens": 5000,
+        "cache_creation_input_tokens": 2000,
+        "prompt_tokens_details": {
+            "cached_tokens": 5000,
+            "cache_creation_tokens": 2000,
+        },
+    }
+    chunks = [
+        b"data: "
+        + json.dumps(
+            {
+                "id": "x",
+                "model": "claude-litellm",
+                "choices": [],
+                "usage": usage,
+            }
+        ).encode()
+        + b"\n\n",
+        b"data: [DONE]\n\n",
+    ]
+
+    out = await _drive(chunks)
+
+    _assert_clean(out)
+    adjustment_input = _last_adjustment_input()
+    for key, value in usage.items():
+        assert adjustment_input["usage"][key] == value
+
+
 @pytest.mark.asyncio
 async def test_openai_style_plain_stream() -> None:
     """OpenAI / Groq / Fireworks / xAI / Perplexity: plain data + [DONE]."""