refactor(payment): extract shared _empty_cost helper; drop .gitignore change

Address PR #561 review:
- Extract the all-zero refund cost object into _empty_cost(), reused by
  both the no-usage-data path and the truly-empty USD-cost path.
- Revert the .gitignore additions; those belong in a separate PR.

routstr/core/main.py

@@ -24,7 +24,6 @@ from ..payment.models import models_router, update_sats_pricing
 from ..payment.price import update_prices_periodically
 from ..proxy import initialize_upstreams, proxy_router, refresh_model_maps_periodically
 from ..upstream.auto_topup import periodic_auto_topup
-from ..upstream.deepseek_v4_pricing_shim import register_deepseek_v4_pricing
 from ..upstream.litellm_routing import configure_litellm
 from ..wallet import periodic_payout, periodic_refund_sweep, periodic_routstr_fee_payout
 from .admin import admin_router
@@ -66,11 +65,6 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
         # debug logging) before any upstream provider dispatches a request.
         configure_litellm()
 
-        # TEMPORARY: backfill DeepSeek V4 pricing missing from litellm's cost
-        # map (BerriAI/litellm#30430). Remove this call and
-        # deepseek_v4_pricing_shim.py once litellm ships these models.
-        register_deepseek_v4_pricing()
-
         # Run database migrations on startup
         run_migrations()
 

routstr/payment/cost_calculation.py

@@ -41,6 +41,28 @@ class CostDataError(BaseModel):
     code: str
 
 
+def _empty_cost(cls: type[CostData] = CostData) -> CostData:
+    """Build an all-zero cost object — a full refund for an empty response.
+
+    Shared by the two paths that must not bill: an upstream response with no
+    usage data at all, and one that reports a USD cost but carries zero tokens
+    in every bucket.
+    """
+    return cls(
+        base_msats=0,
+        input_msats=0,
+        output_msats=0,
+        total_msats=0,
+        total_usd=0.0,
+        input_tokens=0,
+        output_tokens=0,
+        cache_read_input_tokens=0,
+        cache_creation_input_tokens=0,
+        cache_read_msats=0,
+        cache_creation_msats=0,
+    )
+
+
 async def calculate_cost(
     response_data: dict,
     max_cost: int,
@@ -83,19 +105,7 @@ async def calculate_cost(
                 else None,
             },
         )
-        return MaxCostData(
-            base_msats=0,
-            input_msats=0,
-            output_msats=0,
-            total_msats=0,
-            total_usd=0.0,
-            input_tokens=0,
-            output_tokens=0,
-            cache_read_input_tokens=0,
-            cache_creation_input_tokens=0,
-            cache_read_msats=0,
-            cache_creation_msats=0,
-        )
+        return _empty_cost(MaxCostData)
 
     usage_data = response_data.get("usage") or {}
     if not isinstance(usage_data, dict):
@@ -109,6 +119,27 @@ async def calculate_cost(
     # Try USD cost first
     usd_cost = _resolve_usd_cost(usage_data, response_data)
     if usd_cost > 0:
+        truly_empty = (
+            input_tokens == 0
+            and output_tokens == 0
+            and cache_read_tokens == 0
+            and cache_creation_tokens == 0
+        )
+        if truly_empty:
+            logger.warning(
+                "Upstream reported a USD cost but the response carries no "
+                "tokens at all (input, output, cache-read and cache-creation "
+                "are all zero) — refunding in full rather than billing the "
+                "USD-derived cost for an empty response.",
+                extra={
+                    "model": response_data.get("model", "unknown"),
+                    "usd_cost": usd_cost,
+                    "usage_keys": sorted(usage_data.keys())
+                    if isinstance(usage_data, dict)
+                    else None,
+                },
+            )
+            return _empty_cost()
         if input_tokens == 0 and output_tokens == 0:
             logger.warning(
                 "Upstream reported a USD cost but no token counts — "
@@ -418,21 +449,10 @@ def _calculate_from_tokens(
         },
     )
 
-    # Fold the cache-read/write cost into the visible ``input_msats`` so a
-    # dashboard that renders I / O / T sees ``input + output == total``
-    # exactly. This mirrors ``_fold_cache_into_input_tokens`` (which rolls the
-    # cache token counts into the visible prompt total). The standalone
-    # ``cache_read_msats`` / ``cache_creation_msats`` fields stay populated for
-    # clients that want the breakdown; nothing sums the components to derive
-    # ``total_msats`` (it is computed independently above), so this is
-    # display-only and does not change what is billed.
-    visible_output_msats = int(calc_output_msats)
-    visible_input_msats = token_based_cost - visible_output_msats
-
     return CostData(
         base_msats=0,
-        input_msats=visible_input_msats,
-        output_msats=visible_output_msats,
+        input_msats=int(calc_input_msats),
+        output_msats=int(calc_output_msats),
         total_msats=token_based_cost,
         total_usd=total_usd,
         input_tokens=input_tokens,

routstr/upstream/base.py

@@ -766,9 +766,7 @@ class BaseUpstreamProvider:
                     except Exception:
                         pass
 
-            def _process_event(
-                raw_event: bytes, final: bool = False
-            ) -> Iterator[bytes]:
+            def _process_event(raw_event: bytes) -> Iterator[bytes]:
                 """Process one complete SSE event block (lines up to a blank line).
 
                 Handles arbitrary upstream framing across every supported
@@ -874,12 +872,6 @@ class BaseUpstreamProvider:
                         return
                     yield prefix + b"data: " + json.dumps(obj).encode() + b"\n\n"
                 else:
-                    if final:
-                        # Final flush of a truncated tail: the upstream closed
-                        # mid-event, so ``data`` is incomplete JSON. Emitting it
-                        # as a ``data:`` frame would hand the client invalid
-                        # JSON (the "unexpected token" parse error). Drop it.
-                        return
                     # Non-JSON data payload (partial fragment already reassembled
                     # by buffering, or a provider control string). Re-prefix each
                     # line so multi-line ``data`` stays valid SSE framing - a bare
@@ -898,13 +890,7 @@ class BaseUpstreamProvider:
                 # boundary-independent for every provider.
                 buffer = b""
                 async for chunk in response.aiter_bytes():
-                    # Normalize the *joined* buffer, not each chunk in
-                    # isolation: a CRLF event delimiter can straddle two
-                    # ``aiter_bytes`` chunks (``...\r`` then ``\n...``). A
-                    # per-chunk replace would leave a stray ``\r`` and the
-                    # ``\n\n`` split would miss the delimiter, merging two
-                    # events into one frame and breaking SSE clients.
-                    buffer = (buffer + chunk).replace(b"\r\n", b"\n")
+                    buffer += chunk.replace(b"\r\n", b"\n")
                     while b"\n\n" in buffer:
                         raw_event, buffer = buffer.split(b"\n\n", 1)
                         for out in _process_event(raw_event):
@@ -912,7 +898,7 @@ class BaseUpstreamProvider:
 
                 # Flush any trailing event that lacked a final blank line.
                 if buffer.strip():
-                    for out in _process_event(buffer, final=True):
+                    for out in _process_event(buffer):
                         yield out
 
                 async with create_session() as session:
@@ -1218,9 +1204,7 @@ class BaseUpstreamProvider:
                     except Exception:
                         pass
 
-            def _process_event(
-                raw_event: bytes, final: bool = False
-            ) -> Iterator[bytes]:
+            def _process_event(raw_event: bytes) -> Iterator[bytes]:
                 """Process one complete SSE event block for the Responses API.
 
                 Buffers full events (delimited by a blank line) so parsing is
@@ -1290,11 +1274,6 @@ class BaseUpstreamProvider:
 
                     yield prefix + b"data: " + json.dumps(obj).encode() + b"\n\n"
                 else:
-                    if final:
-                        # Final flush of a truncated tail: upstream closed
-                        # mid-event, so ``data`` is incomplete JSON. Dropping it
-                        # avoids handing the client an invalid ``data:`` frame.
-                        return
                     # Re-prefix each line so multi-line ``data`` stays valid SSE
                     # framing for the client.
                     body = b"".join(
@@ -1307,20 +1286,14 @@ class BaseUpstreamProvider:
                 # delimiter so parsing is independent of byte boundaries.
                 buffer = b""
                 async for chunk in response.aiter_bytes():
-                    # Normalize the *joined* buffer, not each chunk in
-                    # isolation: a CRLF event delimiter can straddle two
-                    # ``aiter_bytes`` chunks (``...\r`` then ``\n...``). A
-                    # per-chunk replace would leave a stray ``\r`` and the
-                    # ``\n\n`` split would miss the delimiter, merging two
-                    # events into one frame and breaking SSE clients.
-                    buffer = (buffer + chunk).replace(b"\r\n", b"\n")
+                    buffer += chunk.replace(b"\r\n", b"\n")
                     while b"\n\n" in buffer:
                         raw_event, buffer = buffer.split(b"\n\n", 1)
                         for out in _process_event(raw_event):
                             yield out
 
                 if buffer.strip():
-                    for out in _process_event(buffer, final=True):
+                    for out in _process_event(buffer):
                         yield out
 
                 # Always emit a cost-bearing data chunk

routstr/upstream/deepseek_v4_pricing_shim.py

@@ -1,66 +0,0 @@
-"""TEMPORARY: local DeepSeek V4 pricing shim.
-
-litellm's bundled cost map does not yet ship ``deepseek-v4-flash`` /
-``deepseek-v4-pro``. Without an entry, ``backfill_cache_pricing`` cannot find a
-``cache_read_input_token_cost`` and cache reads fall back to the full input
-rate — a ~60% overcharge on cache hits (DeepSeek hits are ~0.2x input).
-
-This module injects the missing entries into ``litellm.model_cost`` at startup
-so the existing backfill path resolves them. Rates mirror the open upstream PR
-https://github.com/BerriAI/litellm/pull/26380 (issue
-https://github.com/BerriAI/litellm/issues/30430).
-
-=== REMOVAL (once litellm ships these models) ===
-Delete this file and the single ``register_deepseek_v4_pricing()`` call in
-``routstr/core/main.py``. Nothing else depends on it. Entries are only added
-when absent, so a stale shim is harmless after upstream lands — but remove it.
-"""
-
-import litellm
-
-from ..core import get_logger
-
-logger = get_logger(__name__)
-
-# USD per token. Source: BerriAI/litellm PR #26380.
-_DEEPSEEK_V4_RATES: dict[str, dict[str, float]] = {
-    "deepseek-v4-flash": {
-        "input_cost_per_token": 1.4e-07,
-        "output_cost_per_token": 2.8e-07,
-        "cache_read_input_token_cost": 2.8e-08,
-        "cache_creation_input_token_cost": 0.0,
-        "input_cost_per_token_cache_hit": 2.8e-08,
-    },
-    "deepseek-v4-pro": {
-        "input_cost_per_token": 1.74e-06,
-        "output_cost_per_token": 3.48e-06,
-        "cache_read_input_token_cost": 1.4e-07,
-        "cache_creation_input_token_cost": 0.0,
-        "input_cost_per_token_cache_hit": 1.4e-07,
-    },
-}
-
-
-def register_deepseek_v4_pricing() -> None:
-    """Inject DeepSeek V4 pricing into ``litellm.model_cost`` if absent.
-
-    Idempotent and non-destructive: a key already present in the cost map
-    (e.g. once litellm ships it) is left untouched. Registers both the bare
-    (``deepseek-v4-flash``) and prefixed (``deepseek/deepseek-v4-flash``)
-    spellings since ``backfill_cache_pricing`` tries both.
-    """
-    added = []
-    for bare, rates in _DEEPSEEK_V4_RATES.items():
-        for key in (bare, f"deepseek/{bare}"):
-            if key in litellm.model_cost:
-                continue
-            entry: dict[str, object] = dict(rates)
-            entry["litellm_provider"] = "deepseek"
-            entry["mode"] = "chat"
-            litellm.model_cost[key] = entry
-            added.append(key)
-    if added:
-        logger.info(
-            "Registered temporary DeepSeek V4 pricing shim",
-            extra={"models": added},
-        )

routstr/upstream/openrouter.py

@@ -1,4 +1,3 @@
-import json
 from typing import TYPE_CHECKING
 
 import httpx
@@ -19,38 +18,6 @@ class OpenRouterUpstreamProvider(BaseUpstreamProvider):
     supports_anthropic_messages = True
     litellm_provider_prefix = "openrouter/"
 
-    def prepare_request_body(
-        self, body: bytes | None, model_obj: Model
-    ) -> bytes | None:
-        """Set provider.require_parameters on tool-use requests.
-
-        Without it OpenRouter can route a tool call to an endpoint that doesn't
-        support function calling and 404 with "No endpoints found that support
-        tool use". We leave a client-supplied value untouched.
-        """
-        body = super().prepare_request_body(body, model_obj)
-        if not body:
-            return body
-
-        try:
-            data = json.loads(body)
-        except json.JSONDecodeError:
-            return body
-
-        if not isinstance(data, dict) or not data.get("tools"):
-            return body
-
-        provider = data.get("provider")
-        if not isinstance(provider, dict):
-            provider = {}
-
-        if "require_parameters" in provider:
-            return body
-
-        provider["require_parameters"] = True
-        data["provider"] = provider
-        return json.dumps(data).encode()
-
     def _apply_provider_field(self, response_json: object) -> None:
         """Stamp the ``provider`` field for OpenRouter responses.
 

tests/unit/test_cache_pricing.py

@@ -181,14 +181,10 @@ async def test_deepseek_cache_hits_billed_at_cache_rate(model_pricing: Mock) ->
         result = await calculate_cost(response, max_cost=100000)
 
     assert isinstance(result, CostData)
-    # 1000 input @ 1 msat + 9000 cache reads @ 0.1 msat + 500 output @ 2 msat.
-    # input_msats folds the cache-read cost in (1000 + 900) so a dashboard
-    # rendering I/O/T sees input + output == total; the cache portion stays
-    # visible in cache_read_msats.
+    # 1000 input @ 1 msat + 9000 cache reads @ 0.1 msat + 500 output @ 2 msat
+    assert result.input_msats == 1000
     assert result.cache_read_msats == 900
     assert result.output_msats == 1000
-    assert result.input_msats == 1900
-    assert result.input_msats + result.output_msats == result.total_msats
     assert result.total_msats == 2900
 
 

tests/unit/test_cost_calculation_caching.py

@@ -404,6 +404,67 @@ async def test_deepseek_malformed_hit_tokens_coerce_to_zero() -> None:
     assert result.cache_read_input_tokens == 0
 
 
+# ============================================================================
+# Truly-empty response with a non-zero USD cost → full refund
+#
+# When an upstream reports a USD cost but the response carries NO tokens at all
+# (input, output, cache-read and cache-creation all zero), billing the
+# USD-derived cost charges the user for nothing. Refund in full. The gate is
+# tightened relative to PR #489: a cache-read/-creation-only turn legitimately
+# reports zero prompt/completion tokens with a real cost and must still bill.
+# ============================================================================
+@pytest.mark.asyncio
+async def test_truly_empty_usd_cost_response_is_refunded(
+    mock_fixed_pricing: None,
+) -> None:
+    """0 input + 0 output + 0 cache tokens with a non-zero USD cost → refund."""
+    response = {
+        "model": "gpt-4",
+        "usage": {
+            "prompt_tokens": 0,
+            "completion_tokens": 0,
+            "total_cost": 0.01,  # non-zero USD cost despite no tokens
+        },
+    }
+    result = await calculate_cost(response, max_cost=100000)
+
+    assert isinstance(result, CostData)
+    assert result.total_msats == 0  # full refund
+    assert result.input_msats == 0
+    assert result.output_msats == 0
+    assert result.total_usd == 0.0
+    assert result.input_tokens == 0
+    assert result.output_tokens == 0
+    assert result.cache_read_input_tokens == 0
+    assert result.cache_creation_input_tokens == 0
+
+
+@pytest.mark.asyncio
+async def test_cache_read_only_usd_cost_response_is_billed(
+    mock_fixed_pricing: None,
+) -> None:
+    """Cache-read-only turn (0 prompt/completion, non-zero cost) still bills."""
+    response = {
+        "model": "claude-3-5-sonnet",
+        "usage": {
+            "input_tokens": 0,
+            "output_tokens": 0,
+            "cache_read_input_tokens": 1000,  # real cached usage
+            "cache_creation_input_tokens": 0,
+            "total_cost": 0.01,  # non-zero USD cost
+        },
+    }
+    result = await calculate_cost(response, max_cost=100000)
+
+    assert isinstance(result, CostData)
+    # NOT refunded — the USD cost is billed in full. Pinning the exact value
+    # guards against any future regression that would over-refund a cache-only
+    # turn (the bug in PR #489, which refunded whenever prompt+completion == 0).
+    assert result.total_msats == 200000
+    assert result.total_usd == 0.01
+    assert result.cache_read_input_tokens == 1000
+
+
 # ============================================================================
 # Test 13: Missing Usage Block
 # ============================================================================

tests/unit/test_require_parameters.py

@@ -1,95 +0,0 @@
-import json
-import os
-
-os.environ.setdefault("UPSTREAM_BASE_URL", "http://test")
-os.environ.setdefault("UPSTREAM_API_KEY", "test")
-os.environ.setdefault("LIGHTNING_ADDRESS", "test@stm.to")
-
-from routstr.upstream import GenericUpstreamProvider
-from routstr.upstream.openrouter import OpenRouterUpstreamProvider
-
-
-def _model(model_id: str = "openai/gpt-4o"):  # type: ignore[no-untyped-def]
-    from routstr.payment.models import Architecture, Model, Pricing
-
-    return Model(
-        id=model_id,
-        name=model_id,
-        created=0,
-        description="",
-        context_length=128000,
-        architecture=Architecture(
-            modality="text->text",
-            input_modalities=["text"],
-            output_modalities=["text"],
-            tokenizer="GPT",
-            instruct_type=None,
-        ),
-        pricing=Pricing(prompt=0.0, completion=0.0),
-    )
-
-
-def _tool_body() -> dict:
-    return {
-        "model": "openai/gpt-4o",
-        "messages": [{"role": "user", "content": "What's the weather?"}],
-        "tools": [
-            {
-                "type": "function",
-                "function": {"name": "get_weather", "parameters": {}},
-            }
-        ],
-    }
-
-
-def _prepare(provider, body: dict) -> dict:  # type: ignore[no-untyped-def]
-    out = provider.prepare_request_body(json.dumps(body).encode(), _model())
-    assert out is not None
-    return json.loads(out)
-
-
-def test_injects_require_parameters_for_tool_request() -> None:
-    data = _prepare(OpenRouterUpstreamProvider(api_key="test"), _tool_body())
-    assert data["provider"]["require_parameters"] is True
-
-
-def test_generic_provider_on_openrouter_url_is_left_alone() -> None:
-    # Only OpenRouterUpstreamProvider injects; a generic provider pointed at the
-    # same base URL doesn't.
-    provider = GenericUpstreamProvider(base_url="https://openrouter.ai/api/v1")
-    data = _prepare(provider, _tool_body())
-    assert "provider" not in data
-
-
-def test_no_injection_without_tools() -> None:
-    body = {"model": "openai/gpt-4o", "messages": [{"role": "user", "content": "hi"}]}
-    data = _prepare(OpenRouterUpstreamProvider(api_key="test"), body)
-    assert "provider" not in data
-
-
-def test_empty_tools_list_does_not_inject() -> None:
-    body = _tool_body()
-    body["tools"] = []
-    data = _prepare(OpenRouterUpstreamProvider(api_key="test"), body)
-    assert "provider" not in data
-
-
-def test_direct_provider_does_not_inject() -> None:
-    provider = GenericUpstreamProvider(base_url="https://api.openai.com/v1")
-    data = _prepare(provider, _tool_body())
-    assert "provider" not in data
-
-
-def test_keeps_client_set_require_parameters() -> None:
-    body = _tool_body()
-    body["provider"] = {"require_parameters": False}
-    data = _prepare(OpenRouterUpstreamProvider(api_key="test"), body)
-    assert data["provider"]["require_parameters"] is False
-
-
-def test_preserves_other_provider_fields() -> None:
-    body = _tool_body()
-    body["provider"] = {"order": ["openai", "azure"]}
-    data = _prepare(OpenRouterUpstreamProvider(api_key="test"), body)
-    assert data["provider"]["order"] == ["openai", "azure"]
-    assert data["provider"]["require_parameters"] is True

tests/unit/test_streaming_sse_providers.py

@@ -337,75 +337,3 @@ async def test_multiline_non_json_data_each_line_prefixed() -> None:
             continue
         assert line.startswith(b"data: "), f"bare line leaked to client: {line!r}"
     assert b"data: line one" in blob and b"data: line two" in blob
-
-
-@pytest.mark.asyncio
-async def test_crlf_delimiter_split_across_chunk_boundary() -> None:
-    """CRLF event delimiter straddling two TCP reads must not merge events.
-
-    Regression: a per-chunk ``replace(b"\\r\\n", b"\\n")`` left a stray ``\\r``
-    when a ``\\r\\n`` of the ``\\r\\n\\r\\n`` delimiter landed at the very end of
-    one ``aiter_bytes`` chunk and the matching ``\\n`` opened the next. The
-    ``\\n\\n`` split then missed the boundary, glued two events into one frame
-    with two ``data:`` lines, and the client's ``JSON.parse`` threw on the
-    concatenated payload (the "unexpected token"/"Extra data" crash).
-    """
-    e1 = b'data: {"id":"x","choices":[{"delta":{"content":"a"}}]}'
-    e2 = b'data: {"id":"x","choices":[{"delta":{"content":"b"}}]}'
-    chunks = [
-        e1 + b"\r\n\r",  # delimiter cut mid-CRLF
-        b"\n" + e2 + b"\r\n\r\n",
-        b"data: [DONE]\r\n\r\n",
-    ]
-    out = await _drive(chunks)
-
-    # Client-accurate check: a real SSE client concatenates all ``data:`` lines
-    # *within one event* (events are ``\n\n``-delimited) before parsing. A
-    # merged frame would surface here as two objects glued into one payload,
-    # which ``_assert_clean`` (per-line) would miss.
-    blob = b"".join(out)
-    contents: list[str] = []
-    for event in blob.split(b"\n\n"):
-        datas = [
-            ln[len(b"data: ") :]
-            for ln in event.split(b"\n")
-            if ln.startswith(b"data: ")
-        ]
-        if not datas:
-            continue
-        payload = b"".join(datas)
-        if payload.strip() == b"[DONE]":
-            continue
-        obj = json.loads(payload)  # raises if two events were merged into one
-        for c in obj.get("choices", []):
-            if "delta" in c:
-                contents.append(c["delta"]["content"])
-    assert contents == ["a", "b"]
-
-
-@pytest.mark.asyncio
-async def test_truncated_json_tail_on_connection_close() -> None:
-    """A stream that drops mid-event must not emit the partial JSON downstream.
-
-    Regression: the end-of-stream flush ran ``_process_event`` on the leftover
-    buffer unconditionally. When the upstream connection closed mid-event the
-    leftover was incomplete JSON, which fell through to the raw-forward path and
-    handed the client a ``data: {partial`` frame -> ``Unterminated string`` parse
-    error. The truncated tail must be dropped instead.
-    """
-    chunks = [
-        b'data: {"id":"x","choices":[{"delta":{"content":"ok"}}]}\n\n',
-        b'data: {"id":"x","choices":[{"delta":{"con',  # connection dies here
-    ]
-    out = await _drive(chunks)
-    objs = _assert_clean(out)  # raises if the partial tail leaked as a data frame
-    contents = [
-        c["delta"]["content"]
-        for o in objs
-        for c in o.get("choices", [])
-        if "delta" in c
-    ]
-    # The one complete chunk is delivered; the truncated fragment is dropped
-    # entirely (no second delta), and _assert_clean above guarantees nothing
-    # non-JSON ever reached the client.
-    assert contents == ["ok"]