resolve review comments

consolidate streaming

Makefile

@@ -98,7 +98,7 @@ docker-down:
 lint:
 	@echo "🔍 Running linting checks..."
 	$(RUFF) check .
-	$(MYPY) routstr/ --ignore-missing-imports
+	$(MYPY) .
 
 format:
 	@echo "✨ Formatting code..."
@@ -107,7 +107,7 @@ format:
 
 type-check:
 	@echo "🔎 Running type checks..."
-	$(MYPY) routstr/ --ignore-missing-imports
+	$(MYPY) .
 
 # Development setup
 dev-setup:
@@ -234,7 +234,7 @@ ci-test:
 ci-lint:
 	@echo "🤖 Running CI linting..."
 	$(RUFF) check . --exit-non-zero-on-fix
-	$(MYPY) routstr/ --ignore-missing-imports --no-error-summary
+	$(MYPY) . --no-error-summary
 
 # Debug helpers
 test-debug:

routstr/auth.py

@@ -709,12 +709,18 @@ async def revert_pay_for_request(
 
 
 async def adjust_payment_for_tokens(
-    key: ApiKey, response_data: dict, session: AsyncSession, deducted_max_cost: int
+    key: ApiKey,
+    response_data: dict,
+    session: AsyncSession,
+    deducted_max_cost: int,
 ) -> dict:
     """
     Adjusts the payment based on token usage in the response.
     This is called after the initial payment and the upstream request is complete.
     Returns cost data to be included in the response.
+
+    The response's usage object is normalized with the default union parser in
+    ``calculate_cost``.
     """
     billing_key = await get_billing_key(key, session)
     model = response_data.get("model", "unknown")
@@ -797,7 +803,7 @@ async def adjust_payment_for_tokens(
                     extra={"error": str(e), "fee_msats": fee_msats},
                 )
 
-    match await calculate_cost(response_data, deducted_max_cost, session):
+    match await calculate_cost(response_data, deducted_max_cost):
         case MaxCostData() as cost:
             logger.debug(
                 "Using max cost data (no token adjustment)",
@@ -1299,6 +1305,35 @@ async def periodic_key_reset() -> None:
             logger.error(f"Error in periodic_key_reset: {e}")
 
 
+async def periodic_dead_key_prune() -> None:
+    """Periodically prune dead API keys. Interval <= 0 disables it.
+
+    See ``prune_dead_api_keys`` for eligibility.
+    """
+    from .core.db import create_session, prune_dead_api_keys
+
+    interval = settings.dead_key_prune_interval_seconds
+    if interval <= 0:
+        logger.info("Dead-key pruning disabled (interval <= 0)")
+        return
+
+    while True:
+        try:
+            await asyncio.sleep(interval)
+        except asyncio.CancelledError:
+            break
+
+        try:
+            async with create_session() as session:
+                await prune_dead_api_keys(
+                    session, settings.dead_key_min_age_seconds
+                )
+        except asyncio.CancelledError:
+            break
+        except Exception as e:
+            logger.error(f"Error in periodic_dead_key_prune: {e}")
+
+
 STALE_RESERVATION_SWEEP_INTERVAL_SECONDS: int = 60
 
 

routstr/core/db.py

@@ -9,9 +9,10 @@ from typing import AsyncGenerator
 from alembic import command
 from alembic.config import Config
 from alembic.util.exc import CommandError
-from sqlalchemy import UniqueConstraint
+from sqlalchemy import UniqueConstraint, delete
 from sqlalchemy.exc import OperationalError
 from sqlalchemy.ext.asyncio.engine import create_async_engine
+from sqlalchemy.orm import aliased
 from sqlmodel import Field, Relationship, SQLModel, col, func, select, update
 from sqlmodel.ext.asyncio.session import AsyncSession
 
@@ -125,6 +126,63 @@ async def release_stale_reservations(
     return released
 
 
+async def prune_dead_api_keys(session: AsyncSession, min_age_seconds: int) -> int:
+    """Delete dead parentless API keys; return the count removed.
+
+    Dead = 0 balance/reservation/spend/requests, older than the grace period,
+    no parent, no children, no pending invoice. Cashu rows are unlinked (not
+    deleted) first to keep the audit trail.
+    """
+    cutoff = int(time.time()) - min_age_seconds
+
+    child = aliased(ApiKey)
+    has_children = (
+        select(child.hashed_key).where(
+            col(child.parent_key_hash) == col(ApiKey.hashed_key)
+        )
+    ).exists()
+    pending_invoice = (
+        select(LightningInvoice.id)
+        .where(col(LightningInvoice.api_key_hash) == col(ApiKey.hashed_key))
+        .where(col(LightningInvoice.status) == "pending")
+    ).exists()
+
+    eligible_hashes = (
+        select(ApiKey.hashed_key)
+        .where(col(ApiKey.balance) == 0)
+        .where(col(ApiKey.reserved_balance) == 0)
+        .where(col(ApiKey.total_spent) == 0)
+        .where(col(ApiKey.total_requests) == 0)
+        .where(col(ApiKey.parent_key_hash).is_(None))
+        .where(
+            (col(ApiKey.created_at).is_(None)) | (col(ApiKey.created_at) < cutoff)
+        )
+        .where(~pending_invoice)
+        .where(~has_children)
+    )
+
+    # Unlink transactions rather than cascade-deleting them, so the financial
+    # audit trail survives. The eligibility predicate is re-evaluated inside both
+    # statements so a key that gained balance mid-run is left untouched.
+    await session.exec(  # type: ignore[call-overload]
+        update(CashuTransaction)
+        .where(col(CashuTransaction.api_key_hashed_key).in_(eligible_hashes))
+        .values(api_key_hashed_key=None)
+    )
+
+    result = await session.exec(  # type: ignore[call-overload]
+        delete(ApiKey).where(col(ApiKey.hashed_key).in_(eligible_hashes))
+    )
+    await session.commit()
+
+    pruned = int(result.rowcount or 0)
+    logger.info(
+        "Pruned dead API keys",
+        extra={"pruned_keys": pruned, "min_age_seconds": min_age_seconds},
+    )
+    return pruned
+
+
 class ModelRow(SQLModel, table=True):  # type: ignore
     __tablename__ = "models"
     id: str = Field(primary_key=True)

routstr/core/main.py

@@ -11,7 +11,11 @@ from starlette.exceptions import HTTPException
 from starlette.responses import Response as StarletteResponse
 from starlette.types import Scope
 
-from ..auth import periodic_key_reset, periodic_stale_reservation_sweep
+from ..auth import (
+    periodic_dead_key_prune,
+    periodic_key_reset,
+    periodic_stale_reservation_sweep,
+)
 from ..balance import balance_router, deprecated_wallet_router
 from ..lightning import lightning_router, periodic_invoice_watcher
 from ..nostr import (
@@ -24,6 +28,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
@@ -55,6 +60,7 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
     model_maps_refresh_task = None
     key_reset_task = None
     stale_reservation_task = None
+    dead_key_prune_task = None
     auto_topup_task = None
     refund_sweep_task = None
     routstr_fee_task = None
@@ -65,6 +71,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()
 
@@ -126,6 +137,7 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
         stale_reservation_task = asyncio.create_task(
             periodic_stale_reservation_sweep()
         )
+        dead_key_prune_task = asyncio.create_task(periodic_dead_key_prune())
         auto_topup_task = asyncio.create_task(periodic_auto_topup())
         refund_sweep_task = asyncio.create_task(periodic_refund_sweep())
         routstr_fee_task = asyncio.create_task(periodic_routstr_fee_payout())
@@ -165,6 +177,8 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
             key_reset_task.cancel()
         if stale_reservation_task is not None:
             stale_reservation_task.cancel()
+        if dead_key_prune_task is not None:
+            dead_key_prune_task.cancel()
         if auto_topup_task is not None:
             auto_topup_task.cancel()
         if refund_sweep_task is not None:
@@ -196,6 +210,8 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
                 tasks_to_wait.append(key_reset_task)
             if stale_reservation_task is not None:
                 tasks_to_wait.append(stale_reservation_task)
+            if dead_key_prune_task is not None:
+                tasks_to_wait.append(dead_key_prune_task)
             if auto_topup_task is not None:
                 tasks_to_wait.append(auto_topup_task)
             if refund_sweep_task is not None:

routstr/core/settings.py

@@ -73,6 +73,14 @@ class Settings(BaseSettings):
     stale_reservation_timeout_seconds: int = Field(
         default=300, env="STALE_RESERVATION_TIMEOUT_SECONDS"
     )
+    # Background prune of dead (zero balance, never used) API keys.
+    # Interval 0 disables it; min-age is a grace period (default 1 week).
+    dead_key_prune_interval_seconds: int = Field(
+        default=3600, env="DEAD_KEY_PRUNE_INTERVAL_SECONDS"
+    )
+    dead_key_min_age_seconds: int = Field(
+        default=604_800, env="DEAD_KEY_MIN_AGE_SECONDS"
+    )
 
     # Network
     cors_origins: list[str] = Field(default_factory=lambda: ["*"], env="CORS_ORIGINS")

routstr/payment/cost_calculation.py

@@ -3,9 +3,17 @@ import math
 from pydantic.v1 import BaseModel
 
 from ..core import get_logger
-from ..core.db import AsyncSession
 from ..core.settings import settings
 from .price import sats_usd_price
+from .usage import normalize_usage, parse_token_count
+
+__all__ = [
+    "CostData",
+    "CostDataError",
+    "MaxCostData",
+    "calculate_cost",
+    "parse_token_count",
+]
 
 logger = get_logger(__name__)
 
@@ -34,7 +42,8 @@ class CostDataError(BaseModel):
 
 
 async def calculate_cost(
-    response_data: dict, max_cost: int, session: AsyncSession
+    response_data: dict,
+    max_cost: int,
 ) -> CostData | MaxCostData | CostDataError:
     """Calculate the cost of an API request based on token usage.
 
@@ -44,6 +53,9 @@ async def calculate_cost(
 
     Returns:
         Cost data or error information
+
+    The response's usage object is normalized with the default union parser;
+    this function holds no vendor-dialect knowledge of its own.
     """
     logger.debug(
         "Starting cost calculation",
@@ -54,8 +66,9 @@ async def calculate_cost(
         },
     )
 
-    # Check for usage data
-    if "usage" not in response_data or response_data["usage"] is None:
+    usage = normalize_usage(response_data.get("usage"))
+
+    if usage is None:
         logger.warning(
             "No usage data in response — billing at MaxCostData with zero "
             "tokens. Dashboard will show this request as `(0+0)`. Most "
@@ -84,16 +97,14 @@ async def calculate_cost(
             cache_creation_msats=0,
         )
 
-    usage_data = response_data["usage"]
+    usage_data = response_data.get("usage") or {}
+    if not isinstance(usage_data, dict):
+        usage_data = {}
 
-    # Extract token counts
-    input_tokens = _extract_token_pair(usage_data, "prompt_tokens", "input_tokens")
-    output_tokens = _extract_token_pair(usage_data, "completion_tokens", "output_tokens")
-
-    # Extract cache tokens (handles OpenAI vs Anthropic formats)
-    cache_read_tokens, cache_creation_tokens, input_tokens = _extract_cache_tokens(
-        usage_data, input_tokens
-    )
+    input_tokens = usage.input_tokens
+    output_tokens = usage.output_tokens
+    cache_read_tokens = usage.cache_read_tokens
+    cache_creation_tokens = usage.cache_write_tokens
 
     # Try USD cost first
     usd_cost = _resolve_usd_cost(usage_data, response_data)
@@ -202,22 +213,6 @@ async def calculate_cost(
 # ============================================================================
 
 
-def parse_token_count(value: object) -> int:
-    """Parse a token count from various formats (int, float, str, bool)."""
-    if isinstance(value, bool):
-        return 0
-    if isinstance(value, int):
-        return max(0, value)
-    if isinstance(value, float):
-        return max(0, int(value))
-    if isinstance(value, str):
-        try:
-            return max(0, int(float(value)))
-        except ValueError:
-            return 0
-    return 0
-
-
 def _coerce_usd(value: object) -> float:
     """Coerce a value to USD float, handling various formats safely."""
     if value is None or isinstance(value, bool):
@@ -230,61 +225,6 @@ def _coerce_usd(value: object) -> float:
         return 0.0
 
 
-def _extract_token_pair(
-    usage_data: dict, standard_field: str, alt_field: str
-) -> int:
-    """Extract token count trying two field names in order."""
-    value = parse_token_count(usage_data.get(standard_field, 0))
-    if value > 0:
-        return value
-    return parse_token_count(usage_data.get(alt_field, 0))
-
-
-def _extract_cache_tokens(usage_data: dict, input_tokens: int) -> tuple[int, int, int]:
-    """Extract cache tokens, normalizing Anthropic, OpenAI, and DeepSeek shapes.
-
-    Each provider reports prompt caching differently; this normalizes them into
-    non-overlapping buckets so cached tokens are never billed at the full input
-    rate (and never double-counted):
-
-    - Anthropic: ``cache_read_input_tokens`` / ``cache_creation_input_tokens``
-      are already mutually exclusive with ``input_tokens`` (the three sum to the
-      total prompt), so ``input_tokens`` is left untouched.
-    - OpenAI: ``prompt_tokens_details.cached_tokens`` is a cache READ that is
-      INCLUDED in ``prompt_tokens``, so it is subtracted out of ``input_tokens``.
-    - DeepSeek: ``prompt_cache_hit_tokens`` / ``prompt_cache_miss_tokens`` where
-      ``prompt_tokens = hit + miss``. The hit is a cache READ included in
-      ``prompt_tokens``, so it is subtracted out; DeepSeek has no cache-write
-      charge (``cache_creation`` stays 0).
-
-    Returns: (cache_read_tokens, cache_creation_tokens, adjusted_input_tokens)
-    """
-    cache_read = parse_token_count(usage_data.get("cache_read_input_tokens", 0))
-    cache_creation = parse_token_count(
-        usage_data.get("cache_creation_input_tokens", 0)
-    )
-
-    # OpenAI: cache read is included in input_tokens, subtract it
-    prompt_details = usage_data.get("prompt_tokens_details")
-    if isinstance(prompt_details, dict) and not cache_read:
-        openai_cached = parse_token_count(prompt_details.get("cached_tokens", 0))
-        if openai_cached:
-            cache_read = openai_cached
-            input_tokens = max(0, input_tokens - cache_read)
-
-    # DeepSeek: prompt_tokens = prompt_cache_hit_tokens + prompt_cache_miss_tokens.
-    # The hit is a cache read included in input_tokens, subtract it.
-    if not cache_read:
-        deepseek_hit = parse_token_count(
-            usage_data.get("prompt_cache_hit_tokens", 0)
-        )
-        if deepseek_hit:
-            cache_read = deepseek_hit
-            input_tokens = max(0, input_tokens - cache_read)
-
-    return cache_read, cache_creation, input_tokens
-
-
 def _resolve_usd_cost(usage_data: dict, response_data: dict) -> float:
     """Resolve USD cost with clear priority order.
 
@@ -478,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/payment/models.py

@@ -85,6 +85,48 @@ class Model(BaseModel):
         return hash(self.id)
 
 
+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.
+
+    Rates already present (e.g. provided by OpenRouter) are authoritative and
+    never overwritten. Unknown models 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())
+    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)
+    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)
+    return updated
+
+
 def _has_valid_pricing(model: dict) -> bool:
     """Check if model has valid pricing (not free, no negative values)."""
     pricing = model.get("pricing", {})

routstr/payment/usage.py

@@ -0,0 +1,131 @@
+"""Vendor-agnostic normalization of upstream usage objects.
+
+Upstream providers report token usage in vendor dialects that differ in field
+names and in whether cached tokens are included in the input count:
+
+* OpenAI / Azure / xAI / Groq / Moonshot / Qwen / Gemini-compat: cache reads in
+  ``prompt_tokens_details.cached_tokens``, included in ``prompt_tokens``.
+* OpenRouter: same as OpenAI plus cache *writes* in
+  ``prompt_tokens_details.cache_write_tokens``, also included in
+  ``prompt_tokens``.
+* litellm-normalized: same nesting, but names the write field
+  ``prompt_tokens_details.cache_creation_tokens`` (and additionally mirrors the
+  Anthropic top-level fields), with ``prompt_tokens`` as the grand total.
+* Anthropic native: ``cache_read_input_tokens`` / ``cache_creation_input_tokens``
+  top-level, additive to (not included in) ``input_tokens``.
+* DeepSeek: ``prompt_cache_hit_tokens`` / ``prompt_cache_miss_tokens``, with
+  ``prompt_tokens = hit + miss``.
+
+What decides whether cached tokens must be subtracted out of the input count is
+**which prompt field the vendor uses**, not which cache field appears:
+
+* ``prompt_tokens`` present -> cached + cache-write tokens are *included* in it
+  (OpenAI family, DeepSeek, OpenRouter, litellm); subtract both so
+  ``input_tokens`` holds only the regular-rate portion.
+* only ``input_tokens`` (Anthropic native) -> cached tokens are *additive*;
+  leave ``input_tokens`` untouched.
+
+``normalize_usage`` maps all of them onto one canonical ``NormalizedUsage``
+shape so billing code needs no vendor knowledge. The known dialects' field
+names do not collide, so a single union parser is safe; a vendor whose fields
+would genuinely conflict needs a dedicated branch here.
+"""
+
+from pydantic.v1 import BaseModel
+
+
+class NormalizedUsage(BaseModel):
+    """Canonical token usage: input_tokens never includes cached tokens."""
+
+    input_tokens: int = 0
+    output_tokens: int = 0
+    cache_read_tokens: int = 0
+    cache_write_tokens: int = 0
+
+
+def parse_token_count(value: object) -> int:
+    """Parse a token count from various formats (int, float, str, bool)."""
+    if isinstance(value, bool):
+        return 0
+    if isinstance(value, int):
+        return max(0, value)
+    if isinstance(value, float):
+        return max(0, int(value))
+    if isinstance(value, str):
+        try:
+            return max(0, int(float(value)))
+        except ValueError:
+            return 0
+    return 0
+
+
+def _first_token_count(usage_data: dict, *fields: str) -> int:
+    """Return the first positive token count among the given fields."""
+    for field in fields:
+        value = parse_token_count(usage_data.get(field, 0))
+        if value > 0:
+            return value
+    return 0
+
+
+def _extract_cache_tokens(usage_data: dict) -> tuple[int, int]:
+    """Pull (cache_read, cache_write) across all known dialects.
+
+    Precedence (highest first), independent for reads and writes:
+
+    * Anthropic top-level: ``cache_read_input_tokens`` /
+      ``cache_creation_input_tokens``.
+    * Nested ``prompt_tokens_details``: ``cached_tokens`` for reads;
+      ``cache_creation_tokens`` (litellm) or ``cache_write_tokens``
+      (OpenRouter) for writes.
+    * DeepSeek: ``prompt_cache_hit_tokens`` for reads (no write concept).
+    """
+    cache_read = parse_token_count(usage_data.get("cache_read_input_tokens", 0))
+    cache_write = parse_token_count(usage_data.get("cache_creation_input_tokens", 0))
+
+    prompt_details = usage_data.get("prompt_tokens_details")
+    if isinstance(prompt_details, dict):
+        if not cache_read:
+            cache_read = parse_token_count(prompt_details.get("cached_tokens", 0))
+        if not cache_write:
+            cache_write = _first_token_count(
+                prompt_details, "cache_creation_tokens", "cache_write_tokens"
+            )
+
+    if not cache_read:
+        # DeepSeek: prompt_tokens = prompt_cache_hit_tokens + prompt_cache_miss_tokens
+        cache_read = parse_token_count(usage_data.get("prompt_cache_hit_tokens", 0))
+
+    return cache_read, cache_write
+
+
+def normalize_usage(usage_data: object) -> NormalizedUsage | None:
+    """Map a vendor usage dict onto the canonical shape, or None if absent.
+
+    Cached reads and writes are subtracted from the input count exactly once,
+    only for dialects that report a ``prompt_tokens`` grand total that already
+    includes them (OpenAI family, DeepSeek, OpenRouter, litellm). Anthropic
+    native reports them additively under ``input_tokens`` and is left untouched.
+    """
+    if not isinstance(usage_data, dict):
+        return None
+
+    output_tokens = _first_token_count(
+        usage_data, "completion_tokens", "output_tokens"
+    )
+    cache_read, cache_write = _extract_cache_tokens(usage_data)
+
+    # ``prompt_tokens`` is the inclusive grand total; ``input_tokens`` (Anthropic
+    # native) excludes cached tokens. The field chosen decides whether to subtract.
+    if "prompt_tokens" in usage_data:
+        input_tokens = parse_token_count(usage_data.get("prompt_tokens", 0))
+        input_tokens = max(0, input_tokens - cache_read - cache_write)
+    else:
+        input_tokens = parse_token_count(usage_data.get("input_tokens", 0))
+
+    return NormalizedUsage(
+        input_tokens=input_tokens,
+        output_tokens=output_tokens,
+        cache_read_tokens=cache_read,
+        cache_write_tokens=cache_write,
+    )

routstr/upstream/base.py

@@ -35,11 +35,16 @@ from ..payment.models import (
     Pricing,
     _calculate_usd_max_costs,
     _update_model_sats_pricing,
+    backfill_cache_pricing,
     list_models,
 )
 from ..payment.price import sats_usd_price
 from ..wallet import recieve_token, send_token
 from . import messages_dispatch
+from .cache_breakpoints import (
+    inject_anthropic_cache_breakpoints,
+    is_explicit_cache_model,
+)
 from .count_tokens import count_tokens_locally
 from .litellm_routing import detect_litellm_prefix
 
@@ -432,6 +437,19 @@ class BaseUpstreamProvider:
 
         return body
 
+    def _upstream_accepts_cache_control(self) -> bool:
+        """True when this upstream accepts explicit ``cache_control`` markers.
+
+        Only OpenRouter (documents Anthropic + Alibaba explicit caching) and the
+        native Anthropic API accept the markers. Stamping them toward an
+        automatic-cache or non-supporting upstream risks a 400, so injection is
+        confined to these. Base URL is also checked so an OpenRouter endpoint
+        configured through the generic provider is still recognised.
+        """
+        if self.provider_type in ("openrouter", "anthropic"):
+            return True
+        return "openrouter.ai" in (self.base_url or "")
+
     def prepare_request_body(
         self, body: bytes | None, model_obj: Model
     ) -> bytes | None:
@@ -501,6 +519,27 @@ class BaseUpstreamProvider:
                 data["stream_options"] = merged
                 changed = True
 
+        # Explicit-cache models (Anthropic Claude, Alibaba Qwen / deepseek-v3.2)
+        # cache nothing without ``cache_control`` markers in the body. Clients
+        # that don't recognise a routstr URL as one of these never send them, so
+        # caching silently never engages over routstr even though it works
+        # against OpenRouter directly. Stamp the standard breakpoints so caching
+        # works by default, deferring to any client-set markers. Gated to
+        # upstreams that accept the markers (OpenRouter / Anthropic) so they
+        # never leak to an automatic-cache provider that would reject them.
+        if (
+            "messages" in data
+            and isinstance(data.get("messages"), list)
+            and self._upstream_accepts_cache_control()
+            and is_explicit_cache_model(
+                model_obj.id,
+                model_obj.forwarded_model_id,
+                model_obj.canonical_slug,
+            )
+        ):
+            if inject_anthropic_cache_breakpoints(data):
+                changed = True
+
         if changed:
             return json.dumps(data).encode()
         return body
@@ -727,7 +766,9 @@ class BaseUpstreamProvider:
                     except Exception:
                         pass
 
-            def _process_event(raw_event: bytes) -> Iterator[bytes]:
+            def _process_event(
+                raw_event: bytes, final: bool = False
+            ) -> Iterator[bytes]:
                 """Process one complete SSE event block (lines up to a blank line).
 
                 Handles arbitrary upstream framing across every supported
@@ -833,6 +874,12 @@ 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
@@ -851,7 +898,13 @@ class BaseUpstreamProvider:
                 # boundary-independent for every provider.
                 buffer = b""
                 async for chunk in response.aiter_bytes():
-                    buffer += chunk.replace(b"\r\n", b"\n")
+                    # 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")
                     while b"\n\n" in buffer:
                         raw_event, buffer = buffer.split(b"\n\n", 1)
                         for out in _process_event(raw_event):
@@ -859,7 +912,7 @@ class BaseUpstreamProvider:
 
                 # Flush any trailing event that lacked a final blank line.
                 if buffer.strip():
-                    for out in _process_event(buffer):
+                    for out in _process_event(buffer, final=True):
                         yield out
 
                 async with create_session() as session:
@@ -1018,7 +1071,10 @@ class BaseUpstreamProvider:
                 response_json["id"] = f"chatcmpl-{uuid.uuid4()}"
 
             cost_data = await adjust_payment_for_tokens(
-                key, response_json, session, deducted_max_cost
+                key,
+                response_json,
+                session,
+                deducted_max_cost,
             )
 
             await session.refresh(key)
@@ -1162,7 +1218,9 @@ class BaseUpstreamProvider:
                     except Exception:
                         pass
 
-            def _process_event(raw_event: bytes) -> Iterator[bytes]:
+            def _process_event(
+                raw_event: bytes, final: bool = False
+            ) -> Iterator[bytes]:
                 """Process one complete SSE event block for the Responses API.
 
                 Buffers full events (delimited by a blank line) so parsing is
@@ -1232,6 +1290,11 @@ 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(
@@ -1244,14 +1307,20 @@ class BaseUpstreamProvider:
                 # delimiter so parsing is independent of byte boundaries.
                 buffer = b""
                 async for chunk in response.aiter_bytes():
-                    buffer += chunk.replace(b"\r\n", b"\n")
+                    # 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")
                     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):
+                    for out in _process_event(buffer, final=True):
                         yield out
 
                 # Always emit a cost-bearing data chunk
@@ -1436,7 +1505,10 @@ class BaseUpstreamProvider:
                 response_json["id"] = f"chatcmpl-{uuid.uuid4()}"
 
             cost_data = await adjust_payment_for_tokens(
-                key, response_json, session, deducted_max_cost
+                key,
+                response_json,
+                session,
+                deducted_max_cost,
             )
 
             await session.refresh(key)
@@ -1631,7 +1703,10 @@ class BaseUpstreamProvider:
                             "usage": None,
                         }
                         cost_data = await adjust_payment_for_tokens(
-                            fresh_key, fallback, new_session, max_cost_for_model
+                            fresh_key,
+                            fallback,
+                            new_session,
+                            max_cost_for_model,
                         )
                         usage_finalized = True
                         return f"event: cost\ndata: {json.dumps({'cost': cost_data})}\n\n".encode()
@@ -1850,7 +1925,10 @@ class BaseUpstreamProvider:
                 response_json["usage"] = {"input_tokens": input_tokens}
 
             cost_data = await adjust_payment_for_tokens(
-                key, response_json, session, deducted_max_cost
+                key,
+                response_json,
+                session,
+                deducted_max_cost,
             )
 
             self.inject_cost_metadata(response_json, cost_data, key)
@@ -1952,7 +2030,10 @@ class BaseUpstreamProvider:
             response_json["model"] = requested_model
 
         cost_data = await adjust_payment_for_tokens(
-            key, response_json, session, max_cost_for_model
+            key,
+            response_json,
+            session,
+            max_cost_for_model,
         )
         self.inject_cost_metadata(response_json, cost_data, key)
 
@@ -3025,44 +3106,46 @@ class BaseUpstreamProvider:
             extra={"model": model, "has_usage": "usage" in response_data},
         )
 
-        async with create_session() as session:
-            match await calculate_cost(response_data, max_cost_for_model, session):
-                case MaxCostData() as cost:
-                    logger.debug(
-                        "Using max cost pricing",
-                        extra={"model": model, "max_cost_msats": cost.total_msats},
-                    )
-                    return cost
-                case CostData() as cost:
-                    logger.debug(
-                        "Using token-based pricing",
-                        extra={
-                            "model": model,
-                            "total_cost_msats": cost.total_msats,
-                            "input_msats": cost.input_msats,
-                            "output_msats": cost.output_msats,
-                        },
-                    )
-                    return cost
-                case CostDataError() as error:
-                    logger.error(
-                        "Cost calculation error",
-                        extra={
-                            "model": model,
-                            "error_message": error.message,
-                            "error_code": error.code,
-                        },
-                    )
-                    raise HTTPException(
-                        status_code=400,
-                        detail={
-                            "error": {
-                                "message": error.message,
-                                "type": "invalid_request_error",
-                                "code": error.code,
-                            }
-                        },
-                    )
+        match await calculate_cost(
+            response_data,
+            max_cost_for_model,
+        ):
+            case MaxCostData() as cost:
+                logger.debug(
+                    "Using max cost pricing",
+                    extra={"model": model, "max_cost_msats": cost.total_msats},
+                )
+                return cost
+            case CostData() as cost:
+                logger.debug(
+                    "Using token-based pricing",
+                    extra={
+                        "model": model,
+                        "total_cost_msats": cost.total_msats,
+                        "input_msats": cost.input_msats,
+                        "output_msats": cost.output_msats,
+                    },
+                )
+                return cost
+            case CostDataError() as error:
+                logger.error(
+                    "Cost calculation error",
+                    extra={
+                        "model": model,
+                        "error_message": error.message,
+                        "error_code": error.code,
+                    },
+                )
+                raise HTTPException(
+                    status_code=400,
+                    detail={
+                        "error": {
+                            "message": error.message,
+                            "type": "invalid_request_error",
+                            "code": error.code,
+                        }
+                    },
+                )
         return None
 
     async def send_refund(
@@ -4562,14 +4645,19 @@ class BaseUpstreamProvider:
     def _apply_provider_fee_to_model(self, model: Model) -> Model:
         """Apply provider fee to model's USD pricing and calculate max costs.
 
+        Cache rates missing from the upstream pricing feed are backfilled from
+        litellm's cost map first, so they carry the provider fee like every
+        other price component.
+
         Args:
             model: Model object to update
 
         Returns:
             Model with provider fee applied to pricing and max costs calculated
         """
+        base_pricing = backfill_cache_pricing(model.id, model.pricing)
         adjusted_pricing = Pricing.parse_obj(
-            {k: v * self.provider_fee for k, v in model.pricing.dict().items()}
+            {k: v * self.provider_fee for k, v in base_pricing.dict().items()}
         )
 
         temp_model = Model(

routstr/upstream/cache_breakpoints.py

@@ -0,0 +1,157 @@
+"""Inject explicit prompt-cache breakpoints into OpenAI-shaped requests.
+
+Some upstreams cache *explicitly*: the request must carry
+``cache_control: {"type": "ephemeral"}`` markers on the content blocks that
+should be cached. Two model families use this identical wire format:
+
+* **Anthropic Claude** — direct or via OpenRouter's ``anthropic/*`` models.
+* **Alibaba's explicit-cache models on OpenRouter** — ``qwen/qwen3-max``,
+  ``qwen/qwen-plus``, ``qwen/qwen3.6-plus``, ``qwen/qwen3-coder-plus``,
+  ``qwen/qwen3-coder-flash`` and ``deepseek/deepseek-v3.2`` — which OpenRouter
+  documents as using "the same syntax as Anthropic explicit caching".
+
+Every other provider routstr proxies (OpenAI, Azure, xAI/Grok, Groq, Moonshot,
+default DeepSeek, Gemini implicit, Fireworks) caches *automatically* and needs
+no markers — they are left untouched.
+
+A client that doesn't know it is talking to one of these models *through*
+routstr (e.g. an OpenAI-compatible coding agent pointed at a routstr URL) never
+emits the markers — it only adds them when it recognises the provider as
+OpenRouter. So caching silently never engages over routstr even though the same
+client caches fine talking to OpenRouter directly.
+
+This module restores caching by stamping the standard breakpoints onto the
+forwarded body — the system prompt, the last tool, and the last conversation
+message (the format allows up to four; we use three, matching the common
+agent convention) — but only when the client supplied none of its own, so
+explicit client control always wins. The caller is responsible for only
+applying this toward an upstream that accepts the markers (OpenRouter /
+Anthropic), so they never leak to a provider that would reject them.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+# The single ephemeral marker stamped onto each chosen breakpoint. A 5-minute
+# TTL (the default for ``ephemeral``) — deliberately not the 1h tier, which
+# carries a higher cache-write premium and should stay opt-in.
+EPHEMERAL_CACHE_CONTROL: dict[str, str] = {"type": "ephemeral"}
+
+# Alibaba's explicit-cache models on OpenRouter. Matched as substrings of the
+# model id (any spelling routstr carries). Snapshot endpoints that OpenRouter
+# documents as *not* supporting explicit caching (e.g. ``qwen3.5-plus-02-15``)
+# are different families and deliberately absent from this list.
+_ALIBABA_EXPLICIT_CACHE_SLUGS: tuple[str, ...] = (
+    "qwen3-max",
+    "qwen-plus",
+    "qwen3.6-plus",
+    "qwen3-coder-plus",
+    "qwen3-coder-flash",
+    "deepseek-v3.2",
+)
+
+
+def is_explicit_cache_model(model_id: str | None, *fallbacks: str | None) -> bool:
+    """True when the target model uses the explicit ``cache_control`` dialect.
+
+    Covers the Claude family (broadly — every Claude model supports it) and
+    Alibaba's documented explicit-cache models, across the id spellings routstr
+    carries: the OpenRouter id (``anthropic/claude-...``, ``qwen/qwen3-max``),
+    the bare upstream id, and any forwarded/canonical alias.
+    """
+    for candidate in (model_id, *fallbacks):
+        if not candidate:
+            continue
+        lowered = candidate.lower()
+        if "claude" in lowered or "anthropic/" in lowered:
+            return True
+        if any(slug in lowered for slug in _ALIBABA_EXPLICIT_CACHE_SLUGS):
+            return True
+    return False
+
+
+def _has_cache_control(obj: Any) -> bool:
+    """Recursively detect any client-supplied ``cache_control`` marker."""
+    if isinstance(obj, dict):
+        if "cache_control" in obj:
+            return True
+        return any(_has_cache_control(v) for v in obj.values())
+    if isinstance(obj, list):
+        return any(_has_cache_control(v) for v in obj)
+    return False
+
+
+def body_has_cache_control(data: dict) -> bool:
+    """True when the request already carries cache_control on messages/tools."""
+    return _has_cache_control(data.get("messages")) or _has_cache_control(
+        data.get("tools")
+    )
+
+
+def _stamp_text_content(message: dict) -> bool:
+    """Add the ephemeral marker to a message's last text block.
+
+    A string content is promoted to the array form Anthropic requires for
+    cache markers; an existing array gets the marker on its last text part.
+    Returns True when a marker was placed.
+    """
+    content = message.get("content")
+    if isinstance(content, str):
+        if not content:
+            return False
+        message["content"] = [
+            {
+                "type": "text",
+                "text": content,
+                "cache_control": dict(EPHEMERAL_CACHE_CONTROL),
+            }
+        ]
+        return True
+    if isinstance(content, list):
+        for part in reversed(content):
+            if isinstance(part, dict) and part.get("type") == "text":
+                part["cache_control"] = dict(EPHEMERAL_CACHE_CONTROL)
+                return True
+    return False
+
+
+def _stamp_system_prompt(messages: list) -> None:
+    for message in messages:
+        if isinstance(message, dict) and message.get("role") in (
+            "system",
+            "developer",
+        ):
+            _stamp_text_content(message)
+            return
+
+
+def _stamp_last_tool(tools: Any) -> None:
+    if isinstance(tools, list) and tools and isinstance(tools[-1], dict):
+        tools[-1]["cache_control"] = dict(EPHEMERAL_CACHE_CONTROL)
+
+
+def _stamp_last_conversation_message(messages: list) -> None:
+    for message in reversed(messages):
+        if isinstance(message, dict) and message.get("role") in ("user", "assistant"):
+            if _stamp_text_content(message):
+                return
+
+
+def inject_anthropic_cache_breakpoints(data: dict) -> bool:
+    """Stamp ephemeral cache breakpoints onto an OpenAI-shaped chat body.
+
+    Mutates ``data`` in place (the established convention in
+    ``prepare_request_body``) and returns True when anything changed. No-ops
+    when the body isn't chat-shaped or the client already set cache_control.
+    """
+    messages = data.get("messages")
+    if not isinstance(messages, list) or not messages:
+        return False
+    if body_has_cache_control(data):
+        return False
+
+    _stamp_system_prompt(messages)
+    _stamp_last_tool(data.get("tools"))
+    _stamp_last_conversation_message(messages)
+    return True

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},
+        )

routstr/wallet.py

@@ -441,7 +441,11 @@ async def credit_balance(
             .where(col(db.ApiKey.hashed_key) == key.hashed_key)
             .values(balance=(db.ApiKey.balance) + amount)
         )
-        await session.exec(stmt)  # type: ignore[call-overload]
+        result = await session.exec(stmt)  # type: ignore[call-overload]
+        # If pruning removed this key after redemption, do not commit a no-op
+        # balance update and pretend the top-up succeeded.
+        if (getattr(result, "rowcount", 0) or 0) == 0:
+            raise ValueError("API key disappeared before credit could be recorded")
         await session.commit()
         await session.refresh(key)
 
@@ -580,11 +584,19 @@ 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 settings.cashu_mints
+            for mint_url in mint_urls
             for unit in units
         ]
 

tests/integration/test_prune_dead_api_keys.py

@@ -0,0 +1,283 @@
+"""
+Tests for prune_dead_api_keys — the janitor that removes provably-dead 0/0/0
+API keys (funded keys fully refunded/expired without ever being used, plus bare
+orphans), while protecting keys that are still meaningful.
+"""
+
+import time
+import uuid
+from collections.abc import Awaitable, Callable
+from typing import Any, cast
+
+import pytest
+from sqlalchemy.sql.dml import Update
+from sqlmodel import col, update
+
+from routstr.core.db import (
+    ApiKey,
+    CashuTransaction,
+    LightningInvoice,
+    create_session,
+    prune_dead_api_keys,
+)
+
+OLD = 100  # min_age_seconds used by the tests
+NOW = int(time.time())
+LONG_AGO = NOW - 10_000  # well past the grace period
+
+
+async def _exists(key_hash: str) -> bool:
+    async with create_session() as session:
+        return (await session.get(ApiKey, key_hash)) is not None
+
+
+def _dead_key(created_at: int | None) -> ApiKey:
+    return ApiKey(
+        hashed_key=f"dead_{uuid.uuid4().hex}",
+        balance=0,
+        reserved_balance=0,
+        total_spent=0,
+        total_requests=0,
+        created_at=created_at,
+    )
+
+
+@pytest.mark.asyncio
+async def test_prunes_old_refunded_zero_key(patched_db_engine: None) -> None:
+    """A funded-then-refunded key (0/0/0, NULL parent, old) is pruned."""
+    key = _dead_key(LONG_AGO)
+    async with create_session() as session:
+        session.add(key)
+        await session.commit()
+
+    async with create_session() as session:
+        pruned = await prune_dead_api_keys(session, OLD)
+
+    assert pruned == 1
+    assert not await _exists(key.hashed_key)
+
+
+@pytest.mark.asyncio
+async def test_grace_period_protects_fresh_key(patched_db_engine: None) -> None:
+    """A dead-looking but recently created key is protected by the grace period."""
+    key = _dead_key(int(time.time()))
+    async with create_session() as session:
+        session.add(key)
+        await session.commit()
+
+    async with create_session() as session:
+        pruned = await prune_dead_api_keys(session, OLD)
+
+    assert pruned == 0
+    assert await _exists(key.hashed_key)
+
+
+@pytest.mark.asyncio
+async def test_used_key_never_pruned(patched_db_engine: None) -> None:
+    """Keys with any spend/requests or live balance are never pruned."""
+    spent = _dead_key(LONG_AGO)
+    spent.total_spent = 1
+    requested = _dead_key(LONG_AGO)
+    requested.total_requests = 1
+    funded = _dead_key(LONG_AGO)
+    funded.balance = 1000
+    reserved = _dead_key(LONG_AGO)
+    reserved.reserved_balance = 500
+
+    async with create_session() as session:
+        for k in (spent, requested, funded, reserved):
+            session.add(k)
+        await session.commit()
+
+    async with create_session() as session:
+        pruned = await prune_dead_api_keys(session, OLD)
+
+    assert pruned == 0
+    for k in (spent, requested, funded, reserved):
+        assert await _exists(k.hashed_key)
+
+
+@pytest.mark.asyncio
+async def test_parent_and_child_keys_are_not_pruned(
+    patched_db_engine: None,
+) -> None:
+    """Pruning must not orphan child keys or delete valid children."""
+    parent = _dead_key(LONG_AGO)
+    child = ApiKey(
+        hashed_key=f"child_{uuid.uuid4().hex}",
+        balance=0,
+        reserved_balance=0,
+        total_spent=0,
+        total_requests=0,
+        created_at=LONG_AGO,
+        parent_key_hash=parent.hashed_key,
+    )
+    async with create_session() as session:
+        session.add(parent)
+        session.add(child)
+        await session.commit()
+
+    async with create_session() as session:
+        pruned = await prune_dead_api_keys(session, OLD)
+
+    assert pruned == 0
+    assert await _exists(parent.hashed_key)
+    assert await _exists(child.hashed_key)
+
+
+@pytest.mark.asyncio
+async def test_pending_invoice_protects_key(patched_db_engine: None) -> None:
+    """A key referenced by a pending topup invoice is never pruned mid-topup."""
+    key = _dead_key(LONG_AGO)
+    invoice = LightningInvoice(
+        id=f"inv_{uuid.uuid4().hex}",
+        bolt11=f"lnbc_{uuid.uuid4().hex}",
+        amount_sats=10,
+        description="topup",
+        payment_hash=uuid.uuid4().hex,
+        status="pending",
+        api_key_hash=key.hashed_key,
+        purpose="topup",
+        expires_at=NOW + 10_000,
+    )
+    async with create_session() as session:
+        session.add(key)
+        session.add(invoice)
+        await session.commit()
+
+    async with create_session() as session:
+        pruned = await prune_dead_api_keys(session, OLD)
+
+    assert pruned == 0
+    assert await _exists(key.hashed_key)
+
+
+@pytest.mark.asyncio
+async def test_paid_invoice_does_not_protect_key(patched_db_engine: None) -> None:
+    """A settled (non-pending) invoice does not keep a dead key alive."""
+    key = _dead_key(LONG_AGO)
+    invoice = LightningInvoice(
+        id=f"inv_{uuid.uuid4().hex}",
+        bolt11=f"lnbc_{uuid.uuid4().hex}",
+        amount_sats=10,
+        description="topup",
+        payment_hash=uuid.uuid4().hex,
+        status="paid",
+        api_key_hash=key.hashed_key,
+        purpose="topup",
+        expires_at=NOW - 1,
+    )
+    async with create_session() as session:
+        session.add(key)
+        session.add(invoice)
+        await session.commit()
+
+    async with create_session() as session:
+        pruned = await prune_dead_api_keys(session, OLD)
+
+    assert pruned == 1
+    assert not await _exists(key.hashed_key)
+
+
+@pytest.mark.asyncio
+async def test_key_that_becomes_meaningful_during_prune_survives(
+    patched_db_engine: None, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Revalidate before unlink/delete so a late top-up cannot be pruned."""
+    key = _dead_key(LONG_AGO)
+    txn = CashuTransaction(
+        id=uuid.uuid4().hex,
+        token="cashuABC",
+        amount=21,
+        unit="sat",
+        type="in",
+        source="apikey",
+        api_key_hashed_key=key.hashed_key,
+    )
+    async with create_session() as session:
+        session.add(key)
+        session.add(txn)
+        await session.commit()
+
+    async with create_session() as session:
+        original_exec = cast(Callable[..., Awaitable[Any]], session.exec)
+        topped_up = False
+
+        async def exec_with_late_topup(
+            statement: Any, *args: Any, **kwargs: Any
+        ) -> Any:
+            nonlocal topped_up
+            if (
+                not topped_up
+                and isinstance(statement, Update)
+                and getattr(statement.table, "name", None) == "cashu_transactions"
+            ):
+                topped_up = True
+                async with create_session() as topup_session:
+                    await topup_session.exec(  # type: ignore[call-overload]
+                        update(ApiKey)
+                        .where(col(ApiKey.hashed_key) == key.hashed_key)
+                        .values(balance=42)
+                    )
+                    await topup_session.commit()
+            return await original_exec(statement, *args, **kwargs)
+
+        monkeypatch.setattr(session, "exec", exec_with_late_topup)
+        pruned = await prune_dead_api_keys(session, OLD)
+
+    assert pruned == 0
+    assert await _exists(key.hashed_key)
+    async with create_session() as session:
+        surviving = await session.get(CashuTransaction, txn.id)
+        assert surviving is not None
+        assert surviving.api_key_hashed_key == key.hashed_key
+
+
+@pytest.mark.asyncio
+async def test_transaction_audit_trail_preserved(patched_db_engine: None) -> None:
+    """Pruning a refunded key keeps its cashu_transactions, unlinked from the key."""
+    key = _dead_key(LONG_AGO)
+    txn = CashuTransaction(
+        id=uuid.uuid4().hex,
+        token="cashuABC",
+        amount=21,
+        unit="sat",
+        type="in",
+        source="apikey",
+        api_key_hashed_key=key.hashed_key,
+    )
+    async with create_session() as session:
+        session.add(key)
+        session.add(txn)
+        await session.commit()
+
+    async with create_session() as session:
+        pruned = await prune_dead_api_keys(session, OLD)
+
+    assert pruned == 1
+    assert not await _exists(key.hashed_key)
+
+    async with create_session() as session:
+        surviving = await session.get(CashuTransaction, txn.id)
+        assert surviving is not None, "Financial audit row must survive key deletion"
+        assert surviving.api_key_hashed_key is None, "Link must be nulled, not dangling"
+        assert surviving.amount == 21
+
+
+@pytest.mark.asyncio
+async def test_periodic_prune_disabled_returns_immediately(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Non-positive intervals disable the janitor."""
+    from unittest.mock import AsyncMock
+
+    from routstr import auth
+    from routstr.core.settings import settings
+
+    monkeypatch.setattr(settings, "dead_key_prune_interval_seconds", 0)
+    sleep_mock = AsyncMock()
+    monkeypatch.setattr(auth.asyncio, "sleep", sleep_mock)
+
+    await auth.periodic_dead_key_prune()
+
+    sleep_mock.assert_not_called()

tests/unit/test_cache_breakpoints.py

@@ -0,0 +1,189 @@
+"""Tests for Anthropic cache-breakpoint injection on forwarded requests.
+
+Anthropic prompt caching is explicit; a client that doesn't recognise a routstr
+URL as Anthropic-backed never sends ``cache_control`` markers, so caching never
+engages over routstr. ``prepare_request_body`` must stamp the standard
+breakpoints for Anthropic-family models while always deferring to client-set
+markers and never touching automatic-cache providers.
+"""
+
+import json
+import os
+
+import pytest
+
+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.cache_breakpoints import (
+    body_has_cache_control,
+    inject_anthropic_cache_breakpoints,
+    is_explicit_cache_model,
+)
+
+
+def _chat_body() -> dict:
+    return {
+        "model": "anthropic/claude-sonnet-4.5",
+        "stream": True,
+        "messages": [
+            {"role": "system", "content": "You are concise."},
+            {"role": "user", "content": "Hello"},
+        ],
+        "tools": [
+            {"type": "function", "function": {"name": "a"}},
+            {"type": "function", "function": {"name": "b"}},
+        ],
+    }
+
+
+# ---------------------------------------------------------------------------
+# Model detection
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    "model_id,expected",
+    [
+        ("anthropic/claude-sonnet-4.5", True),
+        ("claude-haiku-4-5-20251001", True),
+        # Alibaba explicit-cache models share Anthropic's wire format
+        ("qwen/qwen3-max", True),
+        ("qwen/qwen3-coder-plus", True),
+        ("deepseek/deepseek-v3.2", True),
+        # Automatic-cache providers need no markers
+        ("openai/gpt-4o", False),
+        ("google/gemini-2.5-flash", False),
+        ("deepseek/deepseek-chat", False),
+        ("qwen/qwen3.5-plus-02-15", False),  # snapshot, no explicit caching
+        (None, False),
+    ],
+)
+def test_is_explicit_cache_model(model_id: str | None, expected: bool) -> None:
+    assert is_explicit_cache_model(model_id) is expected
+
+
+def test_is_explicit_cache_model_uses_fallbacks() -> None:
+    # routstr id is opaque but a forwarded/canonical alias reveals the family.
+    assert is_explicit_cache_model("model-xyz", None, "anthropic/claude-opus-4.1")
+
+
+# ---------------------------------------------------------------------------
+# Breakpoint placement
+# ---------------------------------------------------------------------------
+
+
+def test_injects_three_breakpoints() -> None:
+    data = _chat_body()
+    assert inject_anthropic_cache_breakpoints(data) is True
+
+    # system prompt promoted to array form with a marker
+    system = data["messages"][0]["content"]
+    assert system == [
+        {
+            "type": "text",
+            "text": "You are concise.",
+            "cache_control": {"type": "ephemeral"},
+        }
+    ]
+    # last tool marked
+    assert data["tools"][-1]["cache_control"] == {"type": "ephemeral"}
+    assert "cache_control" not in data["tools"][0]
+    # last user message marked
+    user = data["messages"][1]["content"]
+    assert user[-1]["cache_control"] == {"type": "ephemeral"}
+
+
+def test_defers_to_client_supplied_cache_control() -> None:
+    data = _chat_body()
+    data["messages"][1]["content"] = [
+        {"type": "text", "text": "Hello", "cache_control": {"type": "ephemeral"}}
+    ]
+    assert body_has_cache_control(data) is True
+    # No additional stamping when the client already controls caching.
+    assert inject_anthropic_cache_breakpoints(data) is False
+    assert "cache_control" not in data["tools"][-1]
+
+
+def test_marks_last_text_part_of_array_content() -> None:
+    data = _chat_body()
+    data["messages"][1]["content"] = [
+        {"type": "text", "text": "first"},
+        {"type": "image_url", "image_url": {"url": "x"}},
+        {"type": "text", "text": "last"},
+    ]
+    inject_anthropic_cache_breakpoints(data)
+    parts = data["messages"][1]["content"]
+    assert parts[2]["cache_control"] == {"type": "ephemeral"}
+    assert "cache_control" not in parts[0]
+
+
+def test_noop_without_messages() -> None:
+    assert inject_anthropic_cache_breakpoints({"prompt": "x"}) is False
+
+
+# ---------------------------------------------------------------------------
+# prepare_request_body integration
+# ---------------------------------------------------------------------------
+
+
+def _model(model_id: str):  # 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=200000,
+        architecture=Architecture(
+            modality="text->text",
+            input_modalities=["text"],
+            output_modalities=["text"],
+            tokenizer="Claude",
+            instruct_type=None,
+        ),
+        pricing=Pricing(prompt=0.0, completion=0.0),
+    )
+
+
+def _openrouter_provider() -> "GenericUpstreamProvider":
+    # OpenRouter endpoint via the generic provider — recognised by base URL.
+    return GenericUpstreamProvider(base_url="https://openrouter.ai/api/v1")
+
+
+@pytest.mark.parametrize(
+    "model_id", ["anthropic/claude-sonnet-4.5", "qwen/qwen3-max", "deepseek/deepseek-v3.2"]
+)
+def test_prepare_request_body_injects_for_explicit_models(model_id: str) -> None:
+    provider = _openrouter_provider()
+    body = json.dumps(_chat_body()).encode()
+    out = provider.prepare_request_body(body, _model(model_id))
+    assert out is not None
+    data = json.loads(out)
+    assert body_has_cache_control(data) is True
+    assert data["tools"][-1]["cache_control"] == {"type": "ephemeral"}
+
+
+def test_prepare_request_body_skips_for_automatic_provider_model() -> None:
+    provider = _openrouter_provider()
+    body = json.dumps(_chat_body()).encode()
+    out = provider.prepare_request_body(body, _model("openai/gpt-4o"))
+    assert out is not None
+    data = json.loads(out)
+    assert body_has_cache_control(data) is False
+
+
+def test_prepare_request_body_skips_when_upstream_rejects_markers() -> None:
+    # Claude id but a non-OpenRouter/Anthropic upstream → must NOT inject,
+    # since the markers could be rejected by an upstream that doesn't accept them.
+    from routstr.upstream import GenericUpstreamProvider
+
+    provider = GenericUpstreamProvider(base_url="https://some-gateway.example/v1")
+    body = json.dumps(_chat_body()).encode()
+    out = provider.prepare_request_body(body, _model("anthropic/claude-sonnet-4.5"))
+    assert out is not None
+    data = json.loads(out)
+    assert body_has_cache_control(data) is False

tests/unit/test_cache_pricing.py

@@ -0,0 +1,245 @@
+"""Tests for cache-aware pricing of cached input tokens.
+
+Specifies two things:
+
+1. ``backfill_cache_pricing`` — when the OpenRouter model feed omits cache
+   rates (it does for most DeepSeek models and e.g. openai/gpt-4o), they are
+   filled from litellm's bundled cost map instead of silently billing cache
+   reads at the full input rate. Existing OpenRouter values are never
+   overwritten, and provider fees apply to backfilled rates like any other.
+2. ``calculate_cost`` — cached tokens are billed at the cache rates from the
+   model's sats_pricing; the full input rate remains only as the documented
+   last resort when no cache rate could be resolved anywhere.
+"""
+
+import os
+from unittest.mock import Mock, patch
+
+import litellm
+import pytest
+
+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.core.settings import settings
+from routstr.payment.cost_calculation import CostData, calculate_cost
+from routstr.payment.models import (
+    Architecture,
+    Model,
+    Pricing,
+    backfill_cache_pricing,
+)
+from routstr.upstream import GenericUpstreamProvider
+
+
+def _make_model(model_id: str, pricing: Pricing) -> Model:
+    return Model(
+        id=model_id,
+        name=model_id,
+        created=0,
+        description="",
+        context_length=64000,
+        architecture=Architecture(
+            modality="text->text",
+            input_modalities=["text"],
+            output_modalities=["text"],
+            tokenizer="Other",
+            instruct_type=None,
+        ),
+        pricing=pricing,
+    )
+
+
+# ============================================================================
+# backfill_cache_pricing — litellm as fallback source for missing cache rates
+# ============================================================================
+
+
+def test_backfill_deepseek_cache_read_from_litellm() -> None:
+    """deepseek/deepseek-chat has no input_cache_read on OpenRouter; litellm
+    knows the real rate (10x cheaper than input)."""
+    pricing = Pricing(prompt=2.8e-07, completion=4.2e-07)
+
+    result = backfill_cache_pricing("deepseek/deepseek-chat", pricing)
+
+    expected = litellm.model_cost["deepseek/deepseek-chat"][
+        "cache_read_input_token_cost"
+    ]
+    assert result.input_cache_read == expected
+    assert result.input_cache_read < pricing.prompt  # sanity: it's a discount
+
+
+def test_backfill_strips_vendor_prefix_for_litellm_lookup() -> None:
+    """OpenRouter ids are vendor-prefixed (openai/gpt-4o); litellm keys most
+    non-DeepSeek models without the prefix (gpt-4o)."""
+    pricing = Pricing(prompt=2.5e-06, completion=1e-05)
+
+    result = backfill_cache_pricing("openai/gpt-4o", pricing)
+
+    expected = litellm.model_cost["gpt-4o"]["cache_read_input_token_cost"]
+    assert result.input_cache_read == expected
+
+
+def test_backfill_fills_cache_write_rate() -> None:
+    """Anthropic cache writes cost more than input (1.25x); billing them at
+    the input rate undercharges. litellm carries the write rate."""
+    pricing = Pricing(prompt=3e-06, completion=1.5e-05)
+
+    result = backfill_cache_pricing("anthropic/claude-sonnet-4-5", pricing)
+
+    expected = litellm.model_cost["claude-sonnet-4-5"][
+        "cache_creation_input_token_cost"
+    ]
+    assert result.input_cache_write == expected
+    assert result.input_cache_write > pricing.prompt  # sanity: write premium
+
+
+def test_backfill_never_overwrites_openrouter_rates() -> None:
+    """When OpenRouter provides a cache rate, it is authoritative."""
+    pricing = Pricing(
+        prompt=2.1e-07, completion=7.9e-07, input_cache_read=1.3e-07
+    )
+
+    result = backfill_cache_pricing("deepseek/deepseek-chat", pricing)
+
+    assert result.input_cache_read == 1.3e-07
+
+
+def test_backfill_unknown_model_unchanged() -> None:
+    """Models litellm doesn't know stay untouched (last-resort fallback to
+    the input rate happens later, at billing time)."""
+    pricing = Pricing(prompt=1e-06, completion=2e-06)
+
+    result = backfill_cache_pricing("artificial-dumbness/dumb-1", pricing)
+
+    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."""
+    provider = GenericUpstreamProvider(
+        base_url="http://upstream.example", provider_fee=2.0
+    )
+    model = _make_model(
+        "deepseek/deepseek-chat", Pricing(prompt=2.8e-07, completion=4.2e-07)
+    )
+
+    adjusted = provider._apply_provider_fee_to_model(model)
+
+    litellm_read = litellm.model_cost["deepseek/deepseek-chat"][
+        "cache_read_input_token_cost"
+    ]
+    assert adjusted.pricing.input_cache_read == pytest.approx(litellm_read * 2.0)
+    assert adjusted.pricing.prompt == pytest.approx(2.8e-07 * 2.0)
+
+
+# ============================================================================
+# calculate_cost — cached tokens billed at cache rates
+# ============================================================================
+
+
+@pytest.fixture(autouse=True)
+def patch_sats_usd_price() -> None:  # type: ignore[misc]
+    with patch("routstr.payment.cost_calculation.sats_usd_price", return_value=5.0e-5):
+        yield
+
+
+@pytest.fixture
+def model_pricing(monkeypatch: pytest.MonkeyPatch) -> Mock:
+    """Model-based pricing: 1 msat per input token, 2 per output token,
+    0.1 per cache-read token, 1.25 per cache-write token."""
+    monkeypatch.setattr(settings, "fixed_pricing", False)
+    model = Mock()
+    model.sats_pricing = Pricing(
+        prompt=0.001,
+        completion=0.002,
+        input_cache_read=0.0001,
+        input_cache_write=0.00125,
+    )
+    return model
+
+
+@pytest.mark.asyncio
+async def test_deepseek_cache_hits_billed_at_cache_rate(model_pricing: Mock) -> None:
+    """The reported overcharge scenario: a 10k-token prompt with 90% cache
+    hits costs 2900 msats at honest rates, not the 11000 msats that billing
+    every prompt token at the full input rate would charge."""
+    response = {
+        "model": "deepseek-chat",
+        "usage": {
+            "prompt_tokens": 10000,
+            "completion_tokens": 500,
+            "prompt_cache_hit_tokens": 9000,
+            "prompt_cache_miss_tokens": 1000,
+        },
+    }
+
+    with patch("routstr.proxy.get_model_instance", return_value=model_pricing):
+        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.
+    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
+
+
+@pytest.mark.asyncio
+async def test_anthropic_cache_write_billed_at_write_rate(model_pricing: Mock) -> None:
+    """Cache writes carry their premium rate (1.25x input here), instead of
+    being silently billed at the plain input rate."""
+    response = {
+        "model": "claude-sonnet-4-5",
+        "usage": {
+            "input_tokens": 300,
+            "output_tokens": 100,
+            "cache_read_input_tokens": 500,
+            "cache_creation_input_tokens": 2000,
+        },
+    }
+
+    with patch("routstr.proxy.get_model_instance", return_value=model_pricing):
+        result = await calculate_cost(response, max_cost=100000)
+
+    assert isinstance(result, CostData)
+    # 300 @ 1 + 500 @ 0.1 + 2000 @ 1.25 + 100 @ 2
+    assert result.cache_read_msats == 50
+    assert result.cache_creation_msats == 2500
+    assert result.total_msats == 3050
+
+
+@pytest.mark.asyncio
+async def test_missing_cache_rate_falls_back_to_input_rate(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Documented last resort: when no cache rate could be resolved anywhere
+    (OpenRouter and litellm both silent), cache reads bill at the input rate —
+    never cheaper, never free."""
+    monkeypatch.setattr(settings, "fixed_pricing", False)
+    model = Mock()
+    model.sats_pricing = Pricing(prompt=0.001, completion=0.002)
+
+    response = {
+        "model": "dumb-1",
+        "usage": {
+            "prompt_tokens": 10000,
+            "completion_tokens": 500,
+            "prompt_cache_hit_tokens": 9000,
+            "prompt_cache_miss_tokens": 1000,
+        },
+    }
+
+    with patch("routstr.proxy.get_model_instance", return_value=model):
+        result = await calculate_cost(response, max_cost=100000)
+
+    assert isinstance(result, CostData)
+    # 1000 @ 1 + 9000 @ 1 (fallback) + 500 @ 2
+    assert result.total_msats == 11000

tests/unit/test_cost_calculation_caching.py

@@ -1,10 +1,11 @@
 """Tests for cache token handling in cost calculation.
 
-Covers OpenAI vs Anthropic caching formats, edge cases, and billing accuracy.
+Covers OpenAI, Anthropic and DeepSeek caching formats, dialect precedence,
+edge cases, and billing accuracy.
 """
 
 import os
-from unittest.mock import AsyncMock, patch
+from unittest.mock import patch
 
 import pytest
 
@@ -16,12 +17,6 @@ from routstr.core.settings import settings
 from routstr.payment.cost_calculation import CostData, MaxCostData, calculate_cost
 
 
-@pytest.fixture
-def mock_session() -> AsyncMock:
-    """Mock AsyncSession for cost calculation tests."""
-    return AsyncMock()
-
-
 @pytest.fixture(autouse=True)
 def mock_fixed_pricing(monkeypatch: pytest.MonkeyPatch) -> None:
     """Mock settings and price to use fixed pricing."""
@@ -41,7 +36,7 @@ def patch_sats_usd_price() -> None:  # type: ignore[misc]
 # Test 1: OpenAI Cache Format
 # ============================================================================
 @pytest.mark.asyncio
-async def test_openai_cache_subtraction(mock_session: AsyncMock) -> None:
+async def test_openai_cache_subtraction() -> None:
     """OpenAI includes cached_tokens in prompt_tokens, subtract them."""
     response = {
         "model": "gpt-4",
@@ -53,7 +48,7 @@ async def test_openai_cache_subtraction(mock_session: AsyncMock) -> None:
             }
         }
     }
-    result = await calculate_cost(response, max_cost=100000, session=mock_session)
+    result = await calculate_cost(response, max_cost=100000)
 
     assert isinstance(result, CostData)
     assert result.input_tokens == 1000  # 2000 - 1000
@@ -65,7 +60,7 @@ async def test_openai_cache_subtraction(mock_session: AsyncMock) -> None:
 # Test 2: Anthropic Cache Format
 # ============================================================================
 @pytest.mark.asyncio
-async def test_anthropic_cache_additive(mock_session: AsyncMock, mock_fixed_pricing: None) -> None:
+async def test_anthropic_cache_additive(mock_fixed_pricing: None) -> None:
     """Anthropic cache tokens are separate (additive) from input_tokens."""
     response = {
         "model": "claude-3-5-sonnet",
@@ -76,7 +71,7 @@ async def test_anthropic_cache_additive(mock_session: AsyncMock, mock_fixed_pric
             "cache_read_input_tokens": 0,
         }
     }
-    result = await calculate_cost(response, max_cost=100000, session=mock_session)
+    result = await calculate_cost(response, max_cost=100000)
 
     assert isinstance(result, CostData)
     assert result.input_tokens == 500
@@ -89,7 +84,7 @@ async def test_anthropic_cache_additive(mock_session: AsyncMock, mock_fixed_pric
 # Test 3: Invalid Cache (Edge Case)
 # ============================================================================
 @pytest.mark.asyncio
-async def test_cache_read_exceeds_prompt_tokens(mock_session: AsyncMock, mock_fixed_pricing: None) -> None:
+async def test_cache_read_exceeds_prompt_tokens(mock_fixed_pricing: None) -> None:
     """Handle buggy upstream reporting cached > prompt_tokens."""
     response = {
         "model": "gpt-4",
@@ -101,7 +96,7 @@ async def test_cache_read_exceeds_prompt_tokens(mock_session: AsyncMock, mock_fi
             }
         }
     }
-    result = await calculate_cost(response, max_cost=100000, session=mock_session)
+    result = await calculate_cost(response, max_cost=100000)
 
     # Should not go negative
     assert isinstance(result, CostData)
@@ -114,7 +109,7 @@ async def test_cache_read_exceeds_prompt_tokens(mock_session: AsyncMock, mock_fi
 # Test 4: Malformed Token Values
 # ============================================================================
 @pytest.mark.asyncio
-async def test_malformed_cache_tokens_coerce_to_zero(mock_session: AsyncMock, mock_fixed_pricing: None) -> None:
+async def test_malformed_cache_tokens_coerce_to_zero(mock_fixed_pricing: None) -> None:
     """Handle non-numeric cache token values."""
     response = {
         "model": "gpt-4",
@@ -127,7 +122,7 @@ async def test_malformed_cache_tokens_coerce_to_zero(mock_session: AsyncMock, mo
             }
         }
     }
-    result = await calculate_cost(response, max_cost=100000, session=mock_session)
+    result = await calculate_cost(response, max_cost=100000)
 
     # Both should coerce to 0
     assert isinstance(result, CostData)
@@ -139,7 +134,7 @@ async def test_malformed_cache_tokens_coerce_to_zero(mock_session: AsyncMock, mo
 # Test 5: Anthropic Cache Not Subtracted
 # ============================================================================
 @pytest.mark.asyncio
-async def test_anthropic_cache_not_subtracted(mock_session: AsyncMock, mock_fixed_pricing: None) -> None:
+async def test_anthropic_cache_not_subtracted(mock_fixed_pricing: None) -> None:
     """Anthropic cache fields should NOT be subtracted from input_tokens."""
     response = {
         "model": "claude-3-5-sonnet",
@@ -149,7 +144,7 @@ async def test_anthropic_cache_not_subtracted(mock_session: AsyncMock, mock_fixe
             "cache_read_input_tokens": 200,  # ← Additive, don't subtract
         }
     }
-    result = await calculate_cost(response, max_cost=100000, session=mock_session)
+    result = await calculate_cost(response, max_cost=100000)
 
     # Anthropic: input_tokens stays as-is
     assert isinstance(result, CostData)
@@ -161,7 +156,7 @@ async def test_anthropic_cache_not_subtracted(mock_session: AsyncMock, mock_fixe
 # Test 6: Only Cache Read, No Regular Input
 # ============================================================================
 @pytest.mark.asyncio
-async def test_only_cache_read_tokens(mock_session: AsyncMock, mock_fixed_pricing: None) -> None:
+async def test_only_cache_read_tokens(mock_fixed_pricing: None) -> None:
     """Handle response with only cache read tokens."""
     response = {
         "model": "gpt-4",
@@ -173,7 +168,7 @@ async def test_only_cache_read_tokens(mock_session: AsyncMock, mock_fixed_pricin
             }
         }
     }
-    result = await calculate_cost(response, max_cost=100000, session=mock_session)
+    result = await calculate_cost(response, max_cost=100000)
 
     assert isinstance(result, CostData)
     assert result.input_tokens == 0  # max(0, 0 - 1000)
@@ -185,7 +180,7 @@ async def test_only_cache_read_tokens(mock_session: AsyncMock, mock_fixed_pricin
 # Test 7: Only Cache Creation
 # ============================================================================
 @pytest.mark.asyncio
-async def test_only_cache_creation_tokens(mock_session: AsyncMock, mock_fixed_pricing: None) -> None:
+async def test_only_cache_creation_tokens(mock_fixed_pricing: None) -> None:
     """Handle response with only cache creation tokens (Anthropic)."""
     response = {
         "model": "claude-3-5-sonnet",
@@ -196,7 +191,7 @@ async def test_only_cache_creation_tokens(mock_session: AsyncMock, mock_fixed_pr
             "cache_read_input_tokens": 0,
         }
     }
-    result = await calculate_cost(response, max_cost=100000, session=mock_session)
+    result = await calculate_cost(response, max_cost=100000)
 
     assert isinstance(result, CostData)
     assert result.input_tokens == 500
@@ -209,7 +204,7 @@ async def test_only_cache_creation_tokens(mock_session: AsyncMock, mock_fixed_pr
 # Test 8: Both Cache Read and Creation
 # ============================================================================
 @pytest.mark.asyncio
-async def test_both_cache_read_and_creation(mock_session: AsyncMock, mock_fixed_pricing: None) -> None:
+async def test_both_cache_read_and_creation(mock_fixed_pricing: None) -> None:
     """Handle response with both cache read and creation."""
     response = {
         "model": "claude-3-5-sonnet",
@@ -220,7 +215,7 @@ async def test_both_cache_read_and_creation(mock_session: AsyncMock, mock_fixed_
             "cache_read_input_tokens": 500,
         }
     }
-    result = await calculate_cost(response, max_cost=100000, session=mock_session)
+    result = await calculate_cost(response, max_cost=100000)
 
     assert isinstance(result, CostData)
     assert result.input_tokens == 300
@@ -233,7 +228,7 @@ async def test_both_cache_read_and_creation(mock_session: AsyncMock, mock_fixed_
 # Test 9: Token Field Fallback
 # ============================================================================
 @pytest.mark.asyncio
-async def test_token_field_fallback_order(mock_session: AsyncMock, mock_fixed_pricing: None) -> None:
+async def test_token_field_fallback_order(mock_fixed_pricing: None) -> None:
     """Verify fallback order for token extraction."""
     # When prompt_tokens is not present, fall back to input_tokens
     response = {
@@ -243,7 +238,7 @@ async def test_token_field_fallback_order(mock_session: AsyncMock, mock_fixed_pr
             "completion_tokens": 50,
         }
     }
-    result = await calculate_cost(response, max_cost=100000, session=mock_session)
+    result = await calculate_cost(response, max_cost=100000)
 
     assert isinstance(result, CostData)
     assert result.input_tokens == 250
@@ -254,20 +249,21 @@ async def test_token_field_fallback_order(mock_session: AsyncMock, mock_fixed_pr
 # Test 10: Float Token Values
 # ============================================================================
 @pytest.mark.asyncio
-async def test_float_token_values_coerced_to_int(mock_session: AsyncMock, mock_fixed_pricing: None) -> None:
+async def test_float_token_values_coerced_to_int(mock_fixed_pricing: None) -> None:
     """Handle float token values by converting to int."""
     response = {
         "model": "gpt-4",
         "usage": {
             "prompt_tokens": 100.7,  # Float
             "completion_tokens": 50.3,  # Float
-            "cache_read_input_tokens": 25.9,  # Float
+            "prompt_tokens_details": {"cached_tokens": 25.9},  # Float
         }
     }
-    result = await calculate_cost(response, max_cost=100000, session=mock_session)
+    result = await calculate_cost(response, max_cost=100000)
 
     assert isinstance(result, CostData)
-    assert result.input_tokens == 100  # Floored
+    # cached_tokens are part of prompt_tokens (OpenAI dialect) → subtracted: 100 - 25
+    assert result.input_tokens == 75  # Floored
     assert result.output_tokens == 50  # Floored
     assert result.cache_read_input_tokens == 25  # Floored
 
@@ -276,7 +272,7 @@ async def test_float_token_values_coerced_to_int(mock_session: AsyncMock, mock_f
 # Test 11: Boolean Cache Tokens
 # ============================================================================
 @pytest.mark.asyncio
-async def test_boolean_cache_tokens_coerced_to_zero(mock_session: AsyncMock, mock_fixed_pricing: None) -> None:
+async def test_boolean_cache_tokens_coerced_to_zero(mock_fixed_pricing: None) -> None:
     """Handle boolean cache token values by coercing to zero."""
     response = {
         "model": "gpt-4",
@@ -286,7 +282,7 @@ async def test_boolean_cache_tokens_coerced_to_zero(mock_session: AsyncMock, moc
             "cache_read_input_tokens": True,  # Boolean
         }
     }
-    result = await calculate_cost(response, max_cost=100000, session=mock_session)
+    result = await calculate_cost(response, max_cost=100000)
 
     assert isinstance(result, CostData)
     assert result.cache_read_input_tokens == 0  # Boolean coerced to 0
@@ -297,7 +293,7 @@ async def test_boolean_cache_tokens_coerced_to_zero(mock_session: AsyncMock, moc
 # Test 12: Zero Cache Tokens
 # ============================================================================
 @pytest.mark.asyncio
-async def test_zero_cache_tokens(mock_session: AsyncMock, mock_fixed_pricing: None) -> None:
+async def test_zero_cache_tokens(mock_fixed_pricing: None) -> None:
     """Handle explicit zero cache tokens."""
     response = {
         "model": "gpt-4",
@@ -309,7 +305,7 @@ async def test_zero_cache_tokens(mock_session: AsyncMock, mock_fixed_pricing: No
             }
         }
     }
-    result = await calculate_cost(response, max_cost=100000, session=mock_session)
+    result = await calculate_cost(response, max_cost=100000)
 
     assert isinstance(result, CostData)
     assert result.cache_read_input_tokens == 0
@@ -317,138 +313,105 @@ async def test_zero_cache_tokens(mock_session: AsyncMock, mock_fixed_pricing: No
 
 
 # ============================================================================
-# Test 12b: DeepSeek Cache Hit/Miss Format
+# DeepSeek Cache Format
+# DeepSeek emits neither OpenAI's prompt_tokens_details nor Anthropic's
+# cache_read_input_tokens — only prompt_cache_hit_tokens and
+# prompt_cache_miss_tokens, with the documented guarantee
+# prompt_tokens = hit + miss. Hits are ~10x cheaper upstream, so billing
+# them as regular input is a large overcharge.
 # ============================================================================
 @pytest.mark.asyncio
-async def test_deepseek_cache_hit_subtraction(
-    mock_session: AsyncMock, mock_fixed_pricing: None
-) -> None:
-    """DeepSeek prompt_tokens = hit + miss; the hit is a cache read, subtract it."""
+async def test_deepseek_cache_hit_tokens_extracted() -> None:
+    """DeepSeek cache hits are extracted and removed from regular input.
+
+    Payload shape verbatim from the DeepSeek API reference (usage object).
+    """
     response = {
         "model": "deepseek-chat",
         "usage": {
-            "prompt_tokens": 10000,  # ← hit + miss
-            "completion_tokens": 200,
-            "prompt_cache_hit_tokens": 9000,  # ← cache read (0.1x upstream)
-            "prompt_cache_miss_tokens": 1000,  # ← uncached input
+            "prompt_tokens": 10000,  # = hit + miss
+            "completion_tokens": 500,
+            "total_tokens": 10500,
+            "prompt_cache_hit_tokens": 9000,
+            "prompt_cache_miss_tokens": 1000,
         },
     }
-    result = await calculate_cost(response, max_cost=100000, session=mock_session)
+    result = await calculate_cost(response, max_cost=100000)
 
     assert isinstance(result, CostData)
-    assert result.input_tokens == 1000  # 10000 - 9000
+    assert result.input_tokens == 1000  # only the cache misses
     assert result.cache_read_input_tokens == 9000
-    assert result.cache_creation_input_tokens == 0  # DeepSeek has no cache write
-    assert result.output_tokens == 200
+    assert result.output_tokens == 500
 
 
 @pytest.mark.asyncio
-async def test_deepseek_no_cache_hit(
-    mock_session: AsyncMock, mock_fixed_pricing: None
-) -> None:
-    """All-miss DeepSeek response leaves input_tokens untouched."""
+async def test_deepseek_all_tokens_cached() -> None:
+    """A fully cached DeepSeek prompt bills zero regular input tokens."""
+    response = {
+        "model": "deepseek-chat",
+        "usage": {
+            "prompt_tokens": 5000,
+            "completion_tokens": 100,
+            "prompt_cache_hit_tokens": 5000,
+            "prompt_cache_miss_tokens": 0,
+        },
+    }
+    result = await calculate_cost(response, max_cost=100000)
+
+    assert isinstance(result, CostData)
+    assert result.input_tokens == 0
+    assert result.cache_read_input_tokens == 5000
+
+
+@pytest.mark.asyncio
+async def test_dialect_precedence_never_double_subtracts() -> None:
+    """If a vendor emits both OpenAI-style and DeepSeek-style cache fields for
+    the same cached tokens, they are counted once, not subtracted twice."""
+    response = {
+        "model": "deepseek-chat",
+        "usage": {
+            "prompt_tokens": 10000,
+            "completion_tokens": 500,
+            "prompt_tokens_details": {"cached_tokens": 9000},
+            "prompt_cache_hit_tokens": 9000,
+            "prompt_cache_miss_tokens": 1000,
+        },
+    }
+    result = await calculate_cost(response, max_cost=100000)
+
+    assert isinstance(result, CostData)
+    assert result.input_tokens == 1000  # 10000 - 9000, applied exactly once
+    assert result.cache_read_input_tokens == 9000
+
+
+@pytest.mark.asyncio
+async def test_deepseek_malformed_hit_tokens_coerce_to_zero() -> None:
+    """Malformed DeepSeek cache fields degrade to billing all input at full
+    rate instead of crashing or going negative."""
     response = {
         "model": "deepseek-chat",
         "usage": {
             "prompt_tokens": 1000,
             "completion_tokens": 50,
-            "prompt_cache_hit_tokens": 0,
-            "prompt_cache_miss_tokens": 1000,
+            "prompt_cache_hit_tokens": "garbage",
+            "prompt_cache_miss_tokens": -5,
         },
     }
-    result = await calculate_cost(response, max_cost=100000, session=mock_session)
+    result = await calculate_cost(response, max_cost=100000)
 
     assert isinstance(result, CostData)
     assert result.input_tokens == 1000
     assert result.cache_read_input_tokens == 0
 
 
-@pytest.mark.asyncio
-async def test_anthropic_fields_take_priority_over_deepseek(
-    mock_session: AsyncMock, mock_fixed_pricing: None
-) -> None:
-    """Explicit cache_read_input_tokens wins; DeepSeek hit must not double-subtract."""
-    response = {
-        "model": "claude-3-5-sonnet",
-        "usage": {
-            "input_tokens": 500,
-            "output_tokens": 100,
-            "cache_read_input_tokens": 200,
-            "prompt_cache_hit_tokens": 9999,  # ← ignored, Anthropic field present
-        },
-    }
-    result = await calculate_cost(response, max_cost=100000, session=mock_session)
-
-    assert isinstance(result, CostData)
-    assert result.input_tokens == 500  # not subtracted
-    assert result.cache_read_input_tokens == 200
-
-
-# ============================================================================
-# Test 12c: DeepSeek billed at cache rate, not full input (the actual bug)
-# ============================================================================
-@pytest.mark.asyncio
-async def test_deepseek_cache_hits_billed_at_cache_rate(
-    mock_session: AsyncMock, monkeypatch: pytest.MonkeyPatch
-) -> None:
-    """10k prompt at 90% cache hit must bill hits at the cache-read rate.
-
-    Regression for the overcharge bug: cached prompt tokens were billed at the
-    full input rate. With per-token prompt rate P and cache-read rate 0.1*P:
-      honest  = 1000*P  (uncached) + 9000*0.1*P (cache read) = 1900*P
-      buggy   = 10000*P (all at full input rate)
-    """
-    from types import SimpleNamespace
-
-    monkeypatch.setattr(settings, "fixed_pricing", False)
-
-    # per-token sats rates; cache read is 0.1x the prompt rate (DeepSeek upstream)
-    prompt_rate = 1.0e-6
-    sats_pricing = SimpleNamespace(
-        prompt=prompt_rate,
-        completion=2.0e-6,
-        input_cache_read=prompt_rate * 0.1,
-        input_cache_write=0.0,
-    )
-    model_obj = SimpleNamespace(sats_pricing=sats_pricing)
-    monkeypatch.setattr(
-        "routstr.proxy.get_model_instance", lambda _id: model_obj
-    )
-
-    response = {
-        "model": "deepseek-chat",
-        "usage": {
-            "prompt_tokens": 10000,
-            "completion_tokens": 0,
-            "prompt_cache_hit_tokens": 9000,
-            "prompt_cache_miss_tokens": 1000,
-        },
-    }
-    result = await calculate_cost(response, max_cost=100000, session=mock_session)
-
-    assert isinstance(result, CostData)
-    assert result.input_tokens == 1000
-    assert result.cache_read_input_tokens == 9000
-
-    # input_rate (msats/1k) = prompt_rate * 1e6 ; per-token msats = prompt_rate*1000
-    per_tok_input_msats = prompt_rate * 1000
-    expected_input = round(1000 / 1000 * (prompt_rate * 1e6), 3)
-    expected_cache = round(9000 / 1000 * (prompt_rate * 0.1 * 1e6), 3)
-    assert result.input_msats == int(expected_input)
-    assert result.cache_read_msats == int(expected_cache)
-    # honest total (1900*P) must be far below the buggy all-input charge (10000*P)
-    buggy_total = 10000 * per_tok_input_msats
-    assert result.total_msats < buggy_total * 0.25
-
-
 # ============================================================================
 # Test 13: Missing Usage Block
 # ============================================================================
 @pytest.mark.asyncio
-async def test_missing_usage_block(mock_session: AsyncMock, mock_fixed_pricing: None) -> None:
+async def test_missing_usage_block(mock_fixed_pricing: None) -> None:
     """When usage is missing, return MaxCostData with zero tokens."""
     response = {"model": "gpt-4", "choices": [{"message": {"content": "test"}}]}
-    result = await calculate_cost(response, max_cost=100000, session=mock_session)
+    result = await calculate_cost(response, max_cost=100000)
 
     assert isinstance(result, MaxCostData)
     assert result.input_tokens == 0
@@ -460,10 +423,10 @@ async def test_missing_usage_block(mock_session: AsyncMock, mock_fixed_pricing:
 # Test 14: Null Usage Block
 # ============================================================================
 @pytest.mark.asyncio
-async def test_null_usage_block(mock_session: AsyncMock, mock_fixed_pricing: None) -> None:
+async def test_null_usage_block(mock_fixed_pricing: None) -> None:
     """When usage is null, return MaxCostData with zero tokens."""
     response = {"model": "gpt-4", "usage": None}
-    result = await calculate_cost(response, max_cost=100000, session=mock_session)
+    result = await calculate_cost(response, max_cost=100000)
 
     assert isinstance(result, MaxCostData)
     assert result.input_tokens == 0

tests/unit/test_fetch_all_balances.py

@@ -0,0 +1,73 @@
+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_messages_litellm_dispatch.py

@@ -500,7 +500,7 @@ async def test_streaming_emits_sse_and_reconciles_cost_at_end() -> None:
     captured_cost_call: dict[str, Any] = {}
 
     async def fake_adjust(
-        fresh_key: Any, combined_data: Any, sess: Any, max_cost: int
+        fresh_key: Any, combined_data: Any, sess: Any, max_cost: int, usage: Any = None
     ) -> dict:
         captured_cost_call["combined_data"] = combined_data
         captured_cost_call["max_cost"] = max_cost
@@ -589,7 +589,7 @@ async def test_streaming_handles_iterator_yielding_raw_sse_bytes() -> None:
     captured: dict[str, Any] = {}
 
     async def fake_adjust(
-        fresh_key: Any, combined_data: Any, sess: Any, max_cost: int
+        fresh_key: Any, combined_data: Any, sess: Any, max_cost: int, usage: Any = None
     ) -> dict:
         captured["combined_data"] = combined_data
         return fake_cost
@@ -841,109 +841,6 @@ async def test_x_cashu_streaming_replays_events_and_sets_refund_header() -> None
     assert "event: message_stop" in joined
 
 
-# ---------------------------------------------------------------------------
-# Prompt-caching marker pass-through (Anthropic / OpenRouter)
-# ---------------------------------------------------------------------------
-#
-# Anthropic prompt caching is activated by `cache_control: {"type": "ephemeral"}`
-# markers embedded INSIDE the request JSON (messages[].content[], system[],
-# tools[]). Anthropic and OpenRouter providers set
-# ``supports_anthropic_messages = True`` and therefore forward the body through
-# ``prepare_request_body`` (raw pass-through) — NOT the litellm translation path
-# that strips ``ANTHROPIC_ONLY_FIELDS``. These tests lock in that the caching
-# markers survive that forwarding for both /v1/messages and /chat/completions.
-
-
-def _body_with_cache_control(*, stream: bool = True) -> bytes:
-    return json.dumps(
-        {
-            "model": "claude-3-5-sonnet",
-            "system": [
-                {
-                    "type": "text",
-                    "text": "big reusable context",
-                    "cache_control": {"type": "ephemeral"},
-                }
-            ],
-            "messages": [
-                {
-                    "role": "user",
-                    "content": [
-                        {
-                            "type": "text",
-                            "text": "hello",
-                            "cache_control": {"type": "ephemeral", "ttl": "1h"},
-                        }
-                    ],
-                }
-            ],
-            "tools": [
-                {"name": "t", "cache_control": {"type": "ephemeral"}}
-            ],
-            # top-level marker some clients send; must also survive
-            "cache_control": {"type": "ephemeral"},
-            "stream": stream,
-            "max_tokens": 64,
-        }
-    ).encode()
-
-
-def test_prepare_request_body_preserves_nested_cache_control() -> None:
-    """Raw-forward path (Anthropic/OpenRouter) must keep cache_control markers."""
-    provider = _make_provider()
-    model = _make_model(model_id="claude-3-5-sonnet")
-
-    out = provider.prepare_request_body(_body_with_cache_control(stream=True), model)
-    assert out is not None
-    data = json.loads(out)
-
-    # nested markers intact
-    assert data["system"][0]["cache_control"] == {"type": "ephemeral"}
-    assert data["messages"][0]["content"][0]["cache_control"] == {
-        "type": "ephemeral",
-        "ttl": "1h",
-    }
-    assert data["tools"][0]["cache_control"] == {"type": "ephemeral"}
-    # top-level marker intact
-    assert data["cache_control"] == {"type": "ephemeral"}
-    # stream_options.include_usage injected without disturbing cache markers
-    assert data["stream_options"]["include_usage"] is True
-
-
-def test_prepare_request_body_preserves_cache_control_when_no_other_changes() -> None:
-    """Non-streaming request with no model rewrite still preserves markers.
-
-    Even when ``prepare_request_body`` makes no changes it must return a body
-    that still carries the cache_control markers (here: returns original bytes).
-    """
-    provider = _make_provider()
-    model = _make_model(model_id="claude-3-5-sonnet")
-
-    out = provider.prepare_request_body(_body_with_cache_control(stream=False), model)
-    assert out is not None
-    data = json.loads(out)
-    assert data["messages"][0]["content"][0]["cache_control"] == {
-        "type": "ephemeral",
-        "ttl": "1h",
-    }
-    assert data["cache_control"] == {"type": "ephemeral"}
-
-
-def test_native_messages_providers_skip_anthropic_field_stripping() -> None:
-    """Anthropic & OpenRouter forward natively, so they never reach the
-    litellm ANTHROPIC_ONLY_FIELDS strip that would drop cache_control."""
-    from routstr.upstream.anthropic import AnthropicUpstreamProvider
-    from routstr.upstream.openrouter import OpenRouterUpstreamProvider
-
-    assert AnthropicUpstreamProvider.supports_anthropic_messages is True
-    assert OpenRouterUpstreamProvider.supports_anthropic_messages is True
-    # `cache_control` IS in the strip list — proving it only matters on the
-    # non-native litellm path, which these providers bypass.
-    from routstr.upstream.messages_dispatch import ANTHROPIC_ONLY_FIELDS
-
-    assert "cache_control" in ANTHROPIC_ONLY_FIELDS
-
-
 # ---------------------------------------------------------------------------
 # forward_request gating
 # ---------------------------------------------------------------------------

tests/unit/test_streaming_sse_providers.py

@@ -337,3 +337,75 @@ 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_usage_normalization.py

@@ -0,0 +1,140 @@
+"""Tests for vendor-agnostic usage normalization.
+
+Specifies the seam that keeps vendor usage dialects out of generic billing
+code: a canonical ``NormalizedUsage`` shape produced by
+``routstr.payment.usage.normalize_usage`` (union parser for the known,
+non-colliding dialects). ``calculate_cost`` normalizes the response's usage
+object with this parser and needs no vendor knowledge of its own.
+"""
+
+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")
+
+import pytest
+
+from routstr.payment.usage import NormalizedUsage, normalize_usage
+
+# ============================================================================
+# The union parser: one canonical shape for all known dialects
+# ============================================================================
+
+
+@pytest.mark.parametrize(
+    "usage,expected",
+    [
+        # OpenAI: cached_tokens included in prompt_tokens → subtracted
+        (
+            {
+                "prompt_tokens": 2000,
+                "completion_tokens": 100,
+                "prompt_tokens_details": {"cached_tokens": 800},
+            },
+            NormalizedUsage(
+                input_tokens=1200,
+                output_tokens=100,
+                cache_read_tokens=800,
+                cache_write_tokens=0,
+            ),
+        ),
+        # DeepSeek: hit/miss fields, prompt_tokens = hit + miss → hit subtracted
+        (
+            {
+                "prompt_tokens": 10000,
+                "completion_tokens": 500,
+                "prompt_cache_hit_tokens": 9000,
+                "prompt_cache_miss_tokens": 1000,
+            },
+            NormalizedUsage(
+                input_tokens=1000,
+                output_tokens=500,
+                cache_read_tokens=9000,
+                cache_write_tokens=0,
+            ),
+        ),
+        # Anthropic: cache fields additive, input_tokens NOT reduced
+        (
+            {
+                "input_tokens": 300,
+                "output_tokens": 100,
+                "cache_read_input_tokens": 500,
+                "cache_creation_input_tokens": 2000,
+            },
+            NormalizedUsage(
+                input_tokens=300,
+                output_tokens=100,
+                cache_read_tokens=500,
+                cache_write_tokens=2000,
+            ),
+        ),
+        # Plain OpenAI without caching
+        (
+            {"prompt_tokens": 100, "completion_tokens": 50},
+            NormalizedUsage(input_tokens=100, output_tokens=50),
+        ),
+        # OpenRouter: cache writes nested as prompt_tokens_details.cache_write_tokens,
+        # both reads and writes included in prompt_tokens → both subtracted
+        (
+            {
+                "prompt_tokens": 10000,
+                "completion_tokens": 60,
+                "prompt_tokens_details": {
+                    "cached_tokens": 5000,
+                    "cache_write_tokens": 2000,
+                },
+            },
+            NormalizedUsage(
+                input_tokens=3000,
+                output_tokens=60,
+                cache_read_tokens=5000,
+                cache_write_tokens=2000,
+            ),
+        ),
+        # litellm-normalized Anthropic: prompt_tokens is the grand total and the
+        # write field is named cache_creation_tokens; top-level fields mirror it.
+        # prompt_tokens present → both subtracted (NOT additive like native).
+        (
+            {
+                "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,
+                },
+            },
+            NormalizedUsage(
+                input_tokens=3000,
+                output_tokens=100,
+                cache_read_tokens=5000,
+                cache_write_tokens=2000,
+            ),
+        ),
+    ],
+)
+def test_normalize_usage_dialects(usage: dict, expected: NormalizedUsage) -> None:
+    """Each known vendor dialect maps onto the same canonical shape."""
+    assert normalize_usage(usage) == expected
+
+
+def test_normalize_usage_absent_usage() -> None:
+    """Missing/invalid usage yields None so callers can bill at max cost."""
+    assert normalize_usage(None) is None
+    assert normalize_usage("not a dict") is None  # type: ignore[arg-type]
+
+
+def test_normalize_usage_never_negative() -> None:
+    """Buggy upstreams reporting more cached than prompt tokens clamp to 0."""
+    result = normalize_usage(
+        {
+            "prompt_tokens": 100,
+            "completion_tokens": 50,
+            "prompt_cache_hit_tokens": 150,
+        }
+    )
+    assert result is not None
+    assert result.input_tokens == 0
+    assert result.cache_read_tokens == 150

tests/unit/test_wallet.py

@@ -150,6 +150,7 @@ async def test_credit_balance() -> None:
     mock_key.balance = 5000000
     mock_key.hashed_key = "test_hash"
     mock_session = AsyncMock()
+    mock_session.exec.return_value.rowcount = 1
 
     # Mock session.refresh to update the balance (simulates DB reload)
     async def mock_refresh(key: ApiKey) -> None:
@@ -206,6 +207,38 @@ async def test_credit_balance_rejects_zero_amount() -> None:
     assert not mock_session.commit.called
 
 
+@pytest.mark.asyncio
+async def test_credit_balance_rejects_missing_key() -> None:
+    """A top-up must fail if the key was pruned after redemption."""
+    token_data = {
+        "token": [{"mint": "http://mint:3338", "proofs": [{"amount": 1000}]}],
+        "unit": "sat",
+    }
+    token_json = json.dumps(token_data)
+    token_b64 = base64.urlsafe_b64encode(token_json.encode()).decode()
+    token_str = f"cashuA{token_b64}"
+
+    mock_key = Mock()
+    mock_key.balance = 0
+    mock_key.hashed_key = "test_hash"
+    mock_session = AsyncMock()
+    mock_session.exec.return_value.rowcount = 0
+
+    from routstr.core.settings import settings
+
+    with patch.object(settings, "cashu_mints", ["http://mint:3338"]):
+        with patch(
+            "routstr.wallet.recieve_token",
+            return_value=(1000, "sat", "http://mint:3338"),
+        ):
+            with pytest.raises(ValueError, match="disappeared"):
+                await credit_balance(token_str, mock_key, mock_session)
+
+    # UPDATE matched nothing; committing would hide the failed credit.
+    assert mock_session.exec.called
+    assert not mock_session.commit.called
+
+
 @pytest.mark.asyncio
 async def test_swap_to_primary_mint_insufficient_for_fees() -> None:
     """Token amount is less than melt_quote.amount + melt_quote.fee_reserve."""

ui/app/providers/page.tsx

@@ -297,16 +297,13 @@ export default function ProvidersPage() {
   };
 
   const toggleProviderExpansion = (providerId: number) => {
-    const newExpanded = new Set(expandedProviders);
-    if (newExpanded.has(providerId)) {
-      newExpanded.delete(providerId);
-    } else {
-      newExpanded.add(providerId);
-    }
-    setExpandedProviders(newExpanded);
-    if (!newExpanded.has(providerId)) {
+    if (expandedProviders.has(providerId)) {
+      setExpandedProviders(new Set());
       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);
     }
   };