display cr as i

routstr/core/main.py

@@ -24,6 +24,7 @@ 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
@@ -65,6 +66,11 @@ 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

@@ -418,10 +418,21 @@ 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=int(calc_input_msats),
-        output_msats=int(calc_output_msats),
+        input_msats=visible_input_msats,
+        output_msats=visible_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

@@ -0,0 +1,66 @@
+"""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},
+        )

tests/unit/test_cache_pricing.py

@@ -181,10 +181,14 @@ 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
-    assert result.input_msats == 1000
+    # 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.
     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_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"]