deepseek cache calculation fix

routstr/auth.py

@@ -364,7 +364,6 @@ async def validate_bearer_key(
         except HTTPException:
             raise
         except Exception as e:
-            await session.rollback()
             logger.error(
                 "Cashu token redemption failed",
                 extra={

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

@@ -418,21 +418,12 @@ 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
+    visible_input_msats = int(calc_input_msats + calc_cache_read_msats)
 
     return CostData(
         base_msats=0,
         input_msats=visible_input_msats,
-        output_msats=visible_output_msats,
+        output_msats=int(calc_output_msats),
         total_msats=token_based_cost,
         total_usd=total_usd,
         input_tokens=input_tokens,

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 (
@@ -157,48 +158,56 @@ class BaseUpstreamProvider:
 
     @staticmethod
     def _fold_cache_into_input_tokens(usage: object) -> None:
-        """Fold cache token counts into ``input_tokens`` / ``prompt_tokens``.
+        """Fold additive cache token counts into Anthropic ``input_tokens``.
 
         Cost calculation has already used the per-bucket counts to bill the
-        request correctly; what the client sees in the visible token total
-        should be a single rolled-up prompt count *including* the cache
-        portion. The standalone ``cache_read_input_tokens`` /
+        request correctly; what the client sees in Anthropic-shaped visible
+        token totals should be a single rolled-up input count *including* the
+        cache portion. OpenAI-compatible ``prompt_tokens`` is already inclusive
+        (DeepSeek, OpenAI, OpenRouter, litellm), so adding cache fields there
+        would double-count.
+
+        The standalone ``cache_read_input_tokens`` /
         ``cache_creation_input_tokens`` fields are left in place for clients
         that want the breakdown.
-
-        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.
         """
         if not isinstance(usage, dict):
             return
 
-        # Normalise missing cache fields to 0 on Anthropic-shaped usage so
-        # downstream consumers can rely on them being present.
-        if "input_tokens" in usage:
-            usage.setdefault("cache_read_input_tokens", 0)
-            usage.setdefault("cache_creation_input_tokens", 0)
+        # ``prompt_tokens`` is an inclusive OpenAI-compatible grand total.
+        # Fold only native Anthropic-style usage where ``input_tokens`` excludes
+        # cache reads/writes.
+        if "input_tokens" not in usage or "prompt_tokens" in usage:
+            return
+
+        usage.setdefault("cache_read_input_tokens", 0)
+        usage.setdefault("cache_creation_input_tokens", 0)
 
         try:
             cache_read = int(usage.get("cache_read_input_tokens") or 0)
             cache_creation = int(usage.get("cache_creation_input_tokens") or 0)
+            input_tokens = int(usage.get("input_tokens") or 0)
         except (TypeError, ValueError):
             return
         extra = cache_read + cache_creation
-        if extra <= 0:
+        if extra > 0:
+            usage["input_tokens"] = input_tokens + extra
+
+    @staticmethod
+    def _add_normalized_usage_fields(response_json: object) -> None:
+        """Preserve raw usage while adding canonical fields for billing/display."""
+        if not isinstance(response_json, dict):
             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
+        usage = response_json.get("usage")
+        if not isinstance(usage, dict):
+            return
+        normalized = normalize_usage(usage)
+        if normalized is None:
+            return
+        usage.setdefault("input_tokens", normalized.input_tokens)
+        usage.setdefault("output_tokens", normalized.output_tokens)
+        usage.setdefault("cache_read_input_tokens", normalized.cache_read_tokens)
+        usage.setdefault("cache_creation_input_tokens", normalized.cache_write_tokens)
 
     def _apply_provider_field(self, response_json: object) -> None:
         """Stamp the routstr ``provider`` field onto an upstream response payload.
@@ -766,9 +775,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
@@ -860,6 +867,7 @@ class BaseUpstreamProvider:
                                 k: v for k, v in obj.items() if k != "choices"
                             }
                             usage_chunk_data["choices"] = []
+                            self._add_normalized_usage_fields(usage_chunk_data)
                             # Forward the content now, without usage, so token
                             # usage is reported exactly once (in the trailer).
                             forward = {k: v for k, v in obj.items() if k != "usage"}
@@ -871,15 +879,10 @@ class BaseUpstreamProvider:
                             )
                             return
                         usage_chunk_data = obj
+                        self._add_normalized_usage_fields(usage_chunk_data)
                         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 +901,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 +909,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:
@@ -920,6 +917,8 @@ class BaseUpstreamProvider:
                     if fresh_key:
                         cost_data: dict
                         try:
+                            if usage_chunk_data is not None:
+                                self._add_normalized_usage_fields(usage_chunk_data)
                             adjustment_input = (
                                 usage_chunk_data
                                 if usage_chunk_data is not None
@@ -1218,9 +1217,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 +1287,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 +1299,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.
 

routstr/wallet.py

@@ -35,24 +35,6 @@ async def get_balance(unit: str) -> int:
     return wallet.available_balance.amount
 
 
-async def _redeem_same_mint(
-    wallet: Wallet, token_obj: Token
-) -> tuple[int, str, str]:  # amount, unit, mint_url
-    """Redeem proofs at their own issuing mint (no cross-mint swap).
-
-    split() re-mints the incoming proofs into fresh ones we own so the sender
-    can't double-spend them. With include_fees=True the mint deducts its NUT-02
-    per-proof input fee, so we end up holding only `amount - input_fees`. Credit
-    that, not the face value, or routstr over-credits the user and its wallet
-    drifts insolvent.
-    """
-    await wallet.load_mint(keyset_id=token_obj.keysets[0])
-    wallet.verify_proofs_dleq(token_obj.proofs)
-    input_fees = wallet.get_fees_for_proofs(token_obj.proofs)
-    await wallet.split(proofs=token_obj.proofs, amount=0, include_fees=True)
-    return int(token_obj.amount) - input_fees, token_obj.unit, token_obj.mint
-
-
 async def recieve_token(
     token: str,
 ) -> tuple[int, str, str]:  # amount, unit, mint_url
@@ -66,7 +48,12 @@ async def recieve_token(
     if token_obj.mint not in settings.cashu_mints:
         return await swap_to_primary_mint(token_obj, wallet)
 
-    return await _redeem_same_mint(wallet, token_obj)
+    await wallet.load_mint(keyset_id=token_obj.keysets[0])
+
+    wallet.verify_proofs_dleq(token_obj.proofs)
+    await wallet.split(proofs=token_obj.proofs, amount=0, include_fees=True)
+
+    return token_obj.amount, token_obj.unit, token_obj.mint
 
 
 async def send(amount: int, unit: str, mint_url: str | None = None) -> tuple[int, str]:
@@ -226,9 +213,10 @@ async def swap_to_primary_mint(
         amount_msat = token_amount
     else:
         raise ValueError("Invalid unit")
-    # If the token is already from the primary mint, we don't need a cross-mint
-    # swap — redeem it same-mint. There's no melt/Lightning fee, but the mint's
-    # NUT-02 input fee still applies; _redeem_same_mint accounts for it.
+    primary_wallet = await get_wallet(settings.primary_mint, settings.primary_mint_unit)
+
+    # If the token is already from the primary mint, we don't need to swap
+    # and we definitely don't want to calculate or pay fees.
     if token_obj.mint == settings.primary_mint:
         logger.info(
             "swap_to_primary_mint: token already on primary mint, skipping swap",
@@ -238,9 +226,8 @@ async def swap_to_primary_mint(
                 "unit": token_obj.unit,
             },
         )
-        return await _redeem_same_mint(token_wallet, token_obj)
-
-    primary_wallet = await get_wallet(settings.primary_mint, settings.primary_mint_unit)
+        await token_wallet.split(proofs=token_obj.proofs, amount=0, include_fees=True)
+        return token_amount, token_obj.unit, token_obj.mint
 
     minted_amount = await _calculate_swap_amount(
         amount_msat,
@@ -580,19 +567,11 @@ async def fetch_all_balances(
             }
             return error_result
 
-    # Build the set of mints to inspect. Received tokens are stored against
-    # ``primary_mint`` (which defaults to a real mint even when ``cashu_mints``
-    # is empty), so include it as a fallback — otherwise a node that accepts
-    # payments would still report empty balances when ``cashu_mints`` is unset.
-    mint_urls: list[str] = list(settings.cashu_mints)
-    if settings.primary_mint and settings.primary_mint not in mint_urls:
-        mint_urls.append(settings.primary_mint)
-
     # Create tasks for all mint/unit combinations
     async with db.create_session() as session:
         tasks = [
             fetch_balance(session, mint_url, unit)
-            for mint_url in mint_urls
+            for mint_url in settings.cashu_mints
             for unit in units
         ]
 

tests/unit/test_auth_cashu.py

@@ -1,59 +0,0 @@
-import hashlib
-from types import SimpleNamespace
-from typing import AsyncGenerator
-from unittest.mock import AsyncMock, patch
-
-import pytest
-from fastapi import HTTPException
-from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
-from sqlalchemy.pool import StaticPool
-from sqlmodel import SQLModel
-from sqlmodel.ext.asyncio.session import AsyncSession
-
-from routstr.auth import validate_bearer_key
-from routstr.core.db import ApiKey
-
-
-def _make_engine() -> AsyncEngine:
-    return create_async_engine(
-        "sqlite+aiosqlite://",
-        poolclass=StaticPool,
-        connect_args={"check_same_thread": False},
-    )
-
-
-@pytest.fixture
-async def session() -> AsyncGenerator[AsyncSession, None]:
-    engine = _make_engine()
-    async with engine.begin() as conn:
-        await conn.run_sync(SQLModel.metadata.create_all)
-    db_session = AsyncSession(engine, expire_on_commit=False)
-    try:
-        yield db_session
-    finally:
-        await db_session.close()
-        await engine.dispose()
-
-
-@pytest.mark.asyncio
-async def test_failed_first_cashu_redemption_rolls_back_empty_api_key(
-    session: AsyncSession,
-) -> None:
-    token = "cashuAfirst_seen_but_redemption_fails"
-    hashed_key = hashlib.sha256(token.encode()).hexdigest()
-    token_obj = SimpleNamespace(mint="http://mint:3338", unit="sat")
-
-    from routstr.core.settings import settings
-
-    with (
-        patch.object(settings, "cashu_mints", ["http://mint:3338"]),
-        patch("routstr.auth.deserialize_token_from_string", return_value=token_obj),
-        patch(
-            "routstr.auth.credit_balance",
-            new=AsyncMock(side_effect=ValueError("token already spent")),
-        ),
-    ):
-        with pytest.raises(HTTPException):
-            await validate_bearer_key(token, session)
-
-    assert await session.get(ApiKey, hashed_key) is None

tests/unit/test_cache_pricing.py

@@ -182,13 +182,11 @@ async def test_deepseek_cache_hits_billed_at_cache_rate(model_pricing: Mock) ->
 
     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.
+    # Client-visible input_msats includes cache-read input cost for display,
+    # while cache_read_msats keeps the detailed breakdown.
+    assert result.input_msats == 1900
     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_fetch_all_balances.py

@@ -1,73 +0,0 @@
-from contextlib import asynccontextmanager
-from unittest.mock import AsyncMock, MagicMock, patch
-
-import pytest
-
-from routstr.wallet import fetch_all_balances
-
-
-@asynccontextmanager
-async def _fake_session():  # type: ignore[no-untyped-def]
-    yield MagicMock()
-
-
-def _patches(proof_amount: int = 1000):  # type: ignore[no-untyped-def]
-    proof = MagicMock(amount=proof_amount)
-    return [
-        patch("routstr.wallet.get_wallet", AsyncMock(return_value=MagicMock())),
-        patch(
-            "routstr.wallet.get_proofs_per_mint_and_unit",
-            MagicMock(return_value=[proof]),
-        ),
-        patch(
-            "routstr.wallet.slow_filter_spend_proofs",
-            AsyncMock(side_effect=lambda proofs, wallet: proofs),
-        ),
-        patch(
-            "routstr.wallet.db.balances_for_mint_and_unit",
-            AsyncMock(return_value=0),
-        ),
-        patch("routstr.wallet.db.create_session", _fake_session),
-    ]
-
-
-@pytest.mark.asyncio
-async def test_fetch_all_balances_falls_back_to_primary_mint() -> None:
-    """With empty cashu_mints, balances are still fetched for primary_mint."""
-    from routstr.core.settings import settings
-
-    with patch.object(settings, "cashu_mints", []), patch.object(
-        settings, "primary_mint", "http://primary:3338"
-    ):
-        for p in _patches(proof_amount=1000):
-            p.start()
-        try:
-            details, total_wallet, total_user, owner = await fetch_all_balances(
-                units=["sat"]
-            )
-        finally:
-            patch.stopall()
-
-    assert [d["mint_url"] for d in details] == ["http://primary:3338"]
-    assert total_wallet == 1000
-
-
-@pytest.mark.asyncio
-async def test_fetch_all_balances_no_duplicate_primary_mint() -> None:
-    """primary_mint already in cashu_mints is not inspected twice."""
-    from routstr.core.settings import settings
-
-    with patch.object(
-        settings, "cashu_mints", ["http://primary:3338"]
-    ), patch.object(settings, "primary_mint", "http://primary:3338"):
-        for p in _patches(proof_amount=1000):
-            p.start()
-        try:
-            details, total_wallet, _total_user, _owner = await fetch_all_balances(
-                units=["sat"]
-            )
-        finally:
-            patch.stopall()
-
-    assert [d["mint_url"] for d in details] == ["http://primary:3338"]
-    assert total_wallet == 1000

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

@@ -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
@@ -107,6 +108,76 @@ def _assert_clean(out: list[bytes]) -> list[dict]:
     return objs
 
 
+@pytest.mark.asyncio
+async def test_deepseek_usage_chunk_is_normalized_before_billing() -> None:
+    """DeepSeek stream trailers keep raw fields and add canonical billing fields."""
+    usage = {
+        "prompt_tokens": 10000,
+        "completion_tokens": 500,
+        "total_tokens": 10500,
+        "prompt_cache_hit_tokens": 9000,
+        "prompt_cache_miss_tokens": 1000,
+    }
+    chunks = [
+        b'data: {"id":"ds","model":"deepseek-chat","choices":[{"delta":{"content":"ok"}}]}\n\n',
+        b"data: "
+        + json.dumps(
+            {
+                "id": "ds",
+                "model": "deepseek-chat",
+                "choices": [],
+                "usage": usage,
+            }
+        ).encode()
+        + b"\n\n",
+        b"data: [DONE]\n\n",
+    ]
+
+    await _drive(chunks)
+
+    adjust_mock = cast(AsyncMock, base.adjust_payment_for_tokens)
+    adjustment_input = adjust_mock.call_args.args[1]
+    billed_usage = adjustment_input["usage"]
+    assert billed_usage["prompt_tokens"] == 10000
+    assert billed_usage["prompt_cache_hit_tokens"] == 9000
+    assert billed_usage["prompt_cache_miss_tokens"] == 1000
+    assert billed_usage["input_tokens"] == 1000
+    assert billed_usage["output_tokens"] == 500
+    assert billed_usage["cache_read_input_tokens"] == 9000
+    assert billed_usage["cache_creation_input_tokens"] == 0
+
+
+@pytest.mark.asyncio
+async def test_fold_cache_tokens_does_not_double_count_inclusive_prompt_tokens() -> None:
+    """Visible usage mutation must not inflate OpenAI-compatible prompt totals."""
+    usage = {
+        "prompt_tokens": 10000,
+        "completion_tokens": 500,
+        "cache_read_input_tokens": 9000,
+        "prompt_cache_hit_tokens": 9000,
+    }
+
+    BaseUpstreamProvider._fold_cache_into_input_tokens(usage)
+
+    assert usage["prompt_tokens"] == 10000
+    assert usage["cache_read_input_tokens"] == 9000
+
+
+@pytest.mark.asyncio
+async def test_fold_cache_tokens_still_rolls_up_anthropic_input_tokens() -> None:
+    """Anthropic native input_tokens excludes cache and still needs rollup."""
+    usage = {
+        "input_tokens": 1000,
+        "output_tokens": 500,
+        "cache_read_input_tokens": 9000,
+        "cache_creation_input_tokens": 200,
+    }
+
+    BaseUpstreamProvider._fold_cache_into_input_tokens(usage)
+
+    assert usage["input_tokens"] == 10200
+
+
 @pytest.mark.asyncio
 async def test_openai_style_plain_stream() -> None:
     """OpenAI / Groq / Fireworks / xAI / Perplexity: plain data + [DONE]."""
@@ -337,75 +408,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"]

tests/unit/test_wallet.py

@@ -39,8 +39,6 @@ async def test_recieve_token_valid() -> None:
 
     mock_wallet = Mock()
     mock_wallet.split = AsyncMock()
-    # Fee-free trusted mint (e.g. Minibits): nothing deducted.
-    mock_wallet.get_fees_for_proofs = Mock(return_value=0)
 
     from routstr.core.settings import settings
 
@@ -63,69 +61,6 @@ async def test_recieve_token_valid() -> None:
                 assert mint == "http://mint:3338"
 
 
-@pytest.mark.asyncio
-async def test_recieve_token_trusted_mint_deducts_input_fee() -> None:
-    """A trusted mint that charges NUT-02 input fees.
-
-    The same-mint receive (`wallet.split(..., include_fees=True)`, a NUT-03 swap
-    at the same mint — not swap_to_primary_mint) pays the mint's per-proof fee,
-    so routstr only ends up with `face - input_fee` in fresh proofs. The credited
-    amount must reflect that, otherwise routstr over-credits the user and its own
-    wallet drifts toward insolvency.
-    """
-    token_data = {
-        "token": [
-            {
-                "mint": "http://mint:3338",
-                "proofs": [
-                    {"amount": 1000, "id": "test", "secret": "secret", "C": "curve"}
-                ],
-            }
-        ],
-        "unit": "sat",
-    }
-    token_json = json.dumps(token_data)
-    token_b64 = base64.urlsafe_b64encode(token_json.encode()).decode()
-    token_str = f"cashuA{token_b64}"
-
-    mock_wallet = Mock()
-    mock_wallet.split = AsyncMock()
-    # Mock a 3-sat input fee from the Cashu wallet API.
-    mock_wallet.get_fees_for_proofs = Mock(return_value=3)
-
-    from routstr.core.settings import settings
-
-    with patch.object(settings, "cashu_mints", ["http://mint:3338"]):
-        with patch("routstr.wallet.deserialize_token_from_string") as mock_deserialize:
-            mock_token = Mock()
-            mock_token.keysets = ["keyset1"]
-            mock_token.mint = "http://mint:3338"
-            mock_token.unit = "sat"
-            mock_token.amount = 1000
-            mock_token.proofs = [{"amount": 1000}]
-            mock_deserialize.return_value = mock_token
-
-            mock_wallet.load_mint = AsyncMock()
-            mock_wallet.load_proofs = AsyncMock()
-            # Patch get_wallet directly so the module-level `_wallets` cache
-            # (keyed by mint URL) can't hand back a wallet from another test.
-            with patch(
-                "routstr.wallet.get_wallet",
-                AsyncMock(return_value=mock_wallet),
-            ):
-                amount, unit, mint = await recieve_token(token_str)
-                assert amount == 997  # 1000 face - 3 sat input fee paid on swap
-                assert unit == "sat"
-                assert mint == "http://mint:3338"
-                mock_wallet.get_fees_for_proofs.assert_called_once_with(
-                    mock_token.proofs
-                )
-                # DLEQ is verified before re-minting the incoming proofs.
-                mock_wallet.verify_proofs_dleq.assert_called_once_with(
-                    mock_token.proofs
-                )
-
-
 @pytest.mark.asyncio
 async def test_send_token() -> None:
     mock_wallet = Mock()
@@ -319,31 +254,19 @@ async def test_recieve_token_untrusted_mint() -> None:
                 assert mint == "http://mint:3338"
 
 
+@pytest.mark.asyncio
 @pytest.mark.asyncio
 async def test_swap_to_primary_mint_already_on_primary() -> None:
-    """Same-mint shortcut: the token is already on the primary mint.
-
-    No cross-mint swap (no melt/mint), but the same-mint split(include_fees=True)
-    still burns the mint's NUT-02 input fee, so the credited amount must be face
-    minus the input fee — not full face value (the over-credit bug). DLEQ is
-    verified too, matching the trusted same-mint receive path.
-    """
     from routstr.core.settings import settings
     from routstr.wallet import swap_to_primary_mint
 
     mock_token = Mock()
     mock_token.mint = settings.primary_mint
-    mock_token.keysets = ["keyset1"]
     mock_token.amount = 1000
     mock_token.unit = "sat"
-    mock_token.proofs = [{"amount": 1000}]
+    mock_token.proofs = []
 
     mock_token_wallet = Mock()
-    mock_token_wallet.load_mint = AsyncMock()
-    mock_token_wallet.load_proofs = AsyncMock()
-    mock_token_wallet.verify_proofs_dleq = Mock()
-    # Mock a 3-sat input fee from the Cashu wallet API.
-    mock_token_wallet.get_fees_for_proofs = Mock(return_value=3)
     mock_token_wallet.split = AsyncMock(return_value=None)
     mock_token_wallet.request_mint = AsyncMock()
     mock_token_wallet.melt_quote = AsyncMock()
@@ -351,11 +274,9 @@ async def test_swap_to_primary_mint_already_on_primary() -> None:
     with patch("routstr.wallet.get_wallet", AsyncMock(return_value=mock_token_wallet)):
         amount, unit, mint = await swap_to_primary_mint(mock_token, mock_token_wallet)
 
-    assert amount == 997  # 1000 face - 3 sat input fee
+    assert amount == 1000
     assert unit == "sat"
     assert mint == settings.primary_mint
-    mock_token_wallet.verify_proofs_dleq.assert_called_once_with(mock_token.proofs)
-    mock_token_wallet.get_fees_for_proofs.assert_called_once_with(mock_token.proofs)
     mock_token_wallet.split.assert_called_once()
     mock_token_wallet.request_mint.assert_not_called()
     mock_token_wallet.melt_quote.assert_not_called()

ui/app/providers/page.tsx

@@ -297,13 +297,16 @@ export default function ProvidersPage() {
   };
 
   const toggleProviderExpansion = (providerId: number) => {
-    if (expandedProviders.has(providerId)) {
-      setExpandedProviders(new Set());
+    const newExpanded = new Set(expandedProviders);
+    if (newExpanded.has(providerId)) {
+      newExpanded.delete(providerId);
+    } else {
+      newExpanded.add(providerId);
+    }
+    setExpandedProviders(newExpanded);
+    if (!newExpanded.has(providerId)) {
       setViewingModels(null);
     } else {
-      // Accordion: only one provider open at a time so switching to another
-      // provider's models auto-collapses the previously expanded one.
-      setExpandedProviders(new Set([providerId]));
       setViewingModels(providerId);
     }
   };