From eaf74edbba2f6cec418e6f0b4519e6818dc2923a Mon Sep 17 00:00:00 2001 From: Jeroen Ubbink Date: Thu, 11 Jun 2026 21:17:11 +0200 Subject: [PATCH 01/12] fix: bill cached input tokens at their real rates across vendor dialects MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cached prompt tokens were billed at the full input rate whenever a vendor's usage dialect or cache pricing was unknown, overcharging DeepSeek topups ~5-10x on agentic workloads (hits are 10x cheaper upstream) and silently mispricing OpenAI cached reads and Anthropic cache writes the same way. Two root causes, two fixes: - Usage dialects: DeepSeek reports prompt_cache_hit_tokens / prompt_cache_miss_tokens, which billing never parsed. Usage normalization now lives in payment/usage.py as a union parser over the known, non-colliding dialects (OpenAI prompt_tokens_details, Anthropic additive cache fields, DeepSeek hit/miss), producing one canonical NormalizedUsage. Providers expose it as an overridable BaseUpstreamProvider.normalize_usage hook — the escape hatch for future vendors whose fields genuinely conflict — and every settlement call site passes the provider's result through, so calculate_cost holds no vendor knowledge of its own. - Cache rates: the OpenRouter model feed omits input_cache_read/-write for most DeepSeek models (and e.g. openai/gpt-4o), so billing fell back to the full input rate. Missing rates are now backfilled from litellm's bundled cost map before the provider fee is applied; the input-rate fallback remains only as the documented last resort. Co-Authored-By: Claude Fable 5 --- routstr/auth.py | 13 +- routstr/payment/cost_calculation.py | 87 +++---- routstr/payment/models.py | 42 ++++ routstr/payment/usage.py | 92 +++++++ routstr/upstream/base.py | 70 +++++- tests/unit/test_cache_pricing.py | 241 +++++++++++++++++++ tests/unit/test_cost_calculation_caching.py | 95 +++++++- tests/unit/test_messages_litellm_dispatch.py | 4 +- tests/unit/test_usage_normalization.py | 206 ++++++++++++++++ 9 files changed, 779 insertions(+), 71 deletions(-) create mode 100644 routstr/payment/usage.py create mode 100644 tests/unit/test_cache_pricing.py create mode 100644 tests/unit/test_usage_normalization.py diff --git a/routstr/auth.py b/routstr/auth.py index b3f353d2..a5e1d396 100644 --- a/routstr/auth.py +++ b/routstr/auth.py @@ -20,6 +20,7 @@ from .payment.cost_calculation import ( MaxCostData, calculate_cost, ) +from .payment.usage import NormalizedUsage from .wallet import credit_balance, deserialize_token_from_string logger = get_logger(__name__) @@ -675,12 +676,20 @@ 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, + usage: NormalizedUsage | None = None, ) -> 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. + + ``usage`` carries the upstream provider's normalized token usage (its + ``normalize_usage`` hook); when omitted, the response's usage object is + normalized with the default union parser. """ billing_key = await get_billing_key(key, session) model = response_data.get("model", "unknown") @@ -763,7 +772,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, session, usage=usage): case MaxCostData() as cost: logger.debug( "Using max cost data (no token adjustment)", diff --git a/routstr/payment/cost_calculation.py b/routstr/payment/cost_calculation.py index b86cff61..3dd52854 100644 --- a/routstr/payment/cost_calculation.py +++ b/routstr/payment/cost_calculation.py @@ -6,6 +6,15 @@ from ..core import get_logger from ..core.db import AsyncSession from ..core.settings import settings from .price import sats_usd_price +from .usage import NormalizedUsage, normalize_usage, parse_token_count + +__all__ = [ + "CostData", + "CostDataError", + "MaxCostData", + "calculate_cost", + "parse_token_count", +] logger = get_logger(__name__) @@ -34,13 +43,20 @@ class CostDataError(BaseModel): async def calculate_cost( - response_data: dict, max_cost: int, session: AsyncSession + response_data: dict, + max_cost: int, + session: AsyncSession, + usage: NormalizedUsage | None = None, ) -> CostData | MaxCostData | CostDataError: """Calculate the cost of an API request based on token usage. Args: response_data: Response data containing usage information max_cost: Maximum cost in millisats + usage: Pre-normalized usage from the upstream provider's + ``normalize_usage`` hook. When omitted, the response's usage + object is normalized with the default union parser. This function + holds no vendor-dialect knowledge of its own. Returns: Cost data or error information @@ -54,8 +70,10 @@ async def calculate_cost( }, ) - # Check for usage data - if "usage" not in response_data or response_data["usage"] is None: + if 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 +102,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 +218,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,37 +230,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, handling OpenAI vs Anthropic formats. - - 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 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) - - 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. diff --git a/routstr/payment/models.py b/routstr/payment/models.py index 1c1ee52a..e62f7875 100644 --- a/routstr/payment/models.py +++ b/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", {}) diff --git a/routstr/payment/usage.py b/routstr/payment/usage.py new file mode 100644 index 00000000..bf5c2aa2 --- /dev/null +++ b/routstr/payment/usage.py @@ -0,0 +1,92 @@ +"""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: ``prompt_tokens_details.cached_tokens``, included in ``prompt_tokens`` +* Anthropic: ``cache_read_input_tokens`` / ``cache_creation_input_tokens``, + additive to (not included in) ``input_tokens`` +* DeepSeek: ``prompt_cache_hit_tokens`` / ``prompt_cache_miss_tokens``, with + ``prompt_tokens = hit + miss`` + +``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; vendors whose fields +would genuinely conflict override ``BaseUpstreamProvider.normalize_usage``. +""" + +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 normalize_usage(usage_data: object) -> NormalizedUsage | None: + """Map a vendor usage dict onto the canonical shape, or None if absent. + + Cached tokens are subtracted from the input count exactly once, only for + dialects that include them in it (OpenAI, DeepSeek). Precedence between + cache fields: Anthropic explicit > OpenAI details > DeepSeek hit/miss. + """ + if not isinstance(usage_data, dict): + return None + + input_tokens = _first_token_count(usage_data, "prompt_tokens", "input_tokens") + output_tokens = _first_token_count( + usage_data, "completion_tokens", "output_tokens" + ) + cache_write = parse_token_count(usage_data.get("cache_creation_input_tokens", 0)) + + # Anthropic: cache reads are additive, input_tokens stays untouched + cache_read = parse_token_count(usage_data.get("cache_read_input_tokens", 0)) + + if not cache_read: + # OpenAI: cached tokens are included in prompt_tokens + prompt_details = usage_data.get("prompt_tokens_details") + if isinstance(prompt_details, dict): + cache_read = parse_token_count(prompt_details.get("cached_tokens", 0)) + # DeepSeek: prompt_tokens = prompt_cache_hit_tokens + prompt_cache_miss_tokens + if not cache_read: + cache_read = parse_token_count( + usage_data.get("prompt_cache_hit_tokens", 0) + ) + if cache_read: + input_tokens = max(0, input_tokens - cache_read) + + return NormalizedUsage( + input_tokens=input_tokens, + output_tokens=output_tokens, + cache_read_tokens=cache_read, + cache_write_tokens=cache_write, + ) diff --git a/routstr/upstream/base.py b/routstr/upstream/base.py index d440509c..56cd80b8 100644 --- a/routstr/upstream/base.py +++ b/routstr/upstream/base.py @@ -35,9 +35,11 @@ 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 ..payment.usage import NormalizedUsage, normalize_usage from ..wallet import recieve_token, send_token from . import messages_dispatch from .count_tokens import count_tokens_locally @@ -101,6 +103,17 @@ class BaseUpstreamProvider: self._models_cache = [] self._models_by_id = {} + def normalize_usage(self, usage_data: object) -> NormalizedUsage | None: + """Map this provider's usage dialect onto the canonical shape. + + The default union parser covers the known dialects (OpenAI, Anthropic, + DeepSeek), whose field names do not collide. Override in a subclass + when a vendor's usage fields would conflict with another dialect or + need bespoke interpretation; billing code consumes only the canonical + result and holds no vendor knowledge. + """ + return normalize_usage(usage_data) + def get_litellm_provider_prefix(self) -> str: """Resolve the litellm provider prefix for this provider instance. @@ -880,6 +893,9 @@ class BaseUpstreamProvider: adjustment_input, session, max_cost_for_model, + usage=self.normalize_usage( + adjustment_input.get("usage") + ), ) usage_finalized = True except Exception as e: @@ -1018,7 +1034,11 @@ 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, + usage=self.normalize_usage(response_json.get("usage")), ) await session.refresh(key) @@ -1273,6 +1293,9 @@ class BaseUpstreamProvider: adjustment_input, session, max_cost_for_model, + usage=self.normalize_usage( + adjustment_input.get("usage") + ), ) usage_finalized = True except Exception as e: @@ -1436,7 +1459,11 @@ 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, + usage=self.normalize_usage(response_json.get("usage")), ) await session.refresh(key) @@ -1631,7 +1658,11 @@ 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=self.normalize_usage(fallback.get("usage")), ) usage_finalized = True return f"event: cost\ndata: {json.dumps({'cost': cost_data})}\n\n".encode() @@ -1783,6 +1814,9 @@ class BaseUpstreamProvider: combined_data, new_session, max_cost_for_model, + usage=self.normalize_usage( + combined_data.get("usage") + ), ) self.inject_cost_metadata( @@ -1850,7 +1884,11 @@ 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, + usage=self.normalize_usage(response_json.get("usage")), ) self.inject_cost_metadata(response_json, cost_data, key) @@ -1952,7 +1990,11 @@ 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, + usage=self.normalize_usage(response_json.get("usage")), ) self.inject_cost_metadata(response_json, cost_data, key) @@ -2094,6 +2136,7 @@ class BaseUpstreamProvider: fallback, new_session, max_cost_for_model, + usage=self.normalize_usage(fallback.get("usage")), ) usage_finalized = True return ( @@ -2167,6 +2210,9 @@ class BaseUpstreamProvider: combined_data, new_session, max_cost_for_model, + usage=self.normalize_usage( + combined_data.get("usage") + ), ) self.inject_cost_metadata( combined_data, cost_data, fresh_key @@ -3026,7 +3072,12 @@ class BaseUpstreamProvider: ) async with create_session() as session: - match await calculate_cost(response_data, max_cost_for_model, session): + match await calculate_cost( + response_data, + max_cost_for_model, + session, + usage=self.normalize_usage(response_data.get("usage")), + ): case MaxCostData() as cost: logger.debug( "Using max cost pricing", @@ -4562,14 +4613,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( diff --git a/tests/unit/test_cache_pricing.py b/tests/unit/test_cache_pricing.py new file mode 100644 index 00000000..9d73d231 --- /dev/null +++ b/tests/unit/test_cache_pricing.py @@ -0,0 +1,241 @@ +"""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 AsyncMock, 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, session=AsyncMock()) + + assert isinstance(result, CostData) + # 1000 input @ 1 msat + 9000 cache reads @ 0.1 msat + 500 output @ 2 msat + assert result.input_msats == 1000 + assert result.cache_read_msats == 900 + assert result.output_msats == 1000 + assert result.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, session=AsyncMock()) + + 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, session=AsyncMock()) + + assert isinstance(result, CostData) + # 1000 @ 1 + 9000 @ 1 (fallback) + 500 @ 2 + assert result.total_msats == 11000 diff --git a/tests/unit/test_cost_calculation_caching.py b/tests/unit/test_cost_calculation_caching.py index 643a8788..8b8e09ea 100644 --- a/tests/unit/test_cost_calculation_caching.py +++ b/tests/unit/test_cost_calculation_caching.py @@ -1,6 +1,7 @@ """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 @@ -316,6 +317,98 @@ async def test_zero_cache_tokens(mock_session: AsyncMock, mock_fixed_pricing: No assert result.input_tokens == 100 +# ============================================================================ +# 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_tokens_extracted(mock_session: AsyncMock) -> 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": 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) + + assert isinstance(result, CostData) + assert result.input_tokens == 1000 # only the cache misses + assert result.cache_read_input_tokens == 9000 + assert result.output_tokens == 500 + + +@pytest.mark.asyncio +async def test_deepseek_all_tokens_cached(mock_session: AsyncMock) -> 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, session=mock_session) + + 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(mock_session: AsyncMock) -> 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, session=mock_session) + + 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(mock_session: AsyncMock) -> 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": "garbage", + "prompt_cache_miss_tokens": -5, + }, + } + 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 == 0 + + # ============================================================================ # Test 13: Missing Usage Block # ============================================================================ diff --git a/tests/unit/test_messages_litellm_dispatch.py b/tests/unit/test_messages_litellm_dispatch.py index 1e770d1b..5603137e 100644 --- a/tests/unit/test_messages_litellm_dispatch.py +++ b/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 diff --git a/tests/unit/test_usage_normalization.py b/tests/unit/test_usage_normalization.py new file mode 100644 index 00000000..8401c355 --- /dev/null +++ b/tests/unit/test_usage_normalization.py @@ -0,0 +1,206 @@ +"""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) and exposed as an overridable +``BaseUpstreamProvider.normalize_usage`` hook for vendors whose fields would +genuinely conflict. ``calculate_cost`` accepts a pre-normalized usage and then +needs no vendor knowledge of its own. +""" + +import os +from unittest.mock import AsyncMock, patch + +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.usage import NormalizedUsage, normalize_usage +from routstr.upstream import BaseUpstreamProvider, GenericUpstreamProvider + + +@pytest.fixture(autouse=True) +def mock_fixed_pricing(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(settings, "fixed_pricing", True) + monkeypatch.setattr(settings, "fixed_per_1k_input_tokens", 0.001) + monkeypatch.setattr(settings, "fixed_per_1k_output_tokens", 0.001) + + +@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 + + +# ============================================================================ +# 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), + ), + ], +) +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 + + +# ============================================================================ +# The provider hook: default delegates to the union parser, subclasses +# override for vendors whose fields genuinely conflict +# ============================================================================ + + +class ArtificialDumbnessProvider(BaseUpstreamProvider): + """Fictional vendor whose usage dialect collides with nothing we know.""" + + provider_type = "artificial-dumbness" + + def normalize_usage(self, usage_data: object) -> NormalizedUsage | None: + if not isinstance(usage_data, dict): + return None + return NormalizedUsage( + input_tokens=usage_data.get("dumb_tokens_in", 0), + output_tokens=usage_data.get("dumb_tokens_out", 0), + cache_read_tokens=usage_data.get("dumb_tokens_reused", 0), + cache_write_tokens=0, + ) + + +def test_base_provider_hook_delegates_to_union_parser() -> None: + """Without an override, the provider hook equals the module parser, so + DeepSeek-dialect upstreams work through GenericUpstreamProvider with no + subclass.""" + provider = GenericUpstreamProvider(base_url="http://upstream.example") + usage = { + "prompt_tokens": 10000, + "completion_tokens": 500, + "prompt_cache_hit_tokens": 9000, + "prompt_cache_miss_tokens": 1000, + } + assert provider.normalize_usage(usage) == normalize_usage(usage) + + +@pytest.mark.asyncio +async def test_provider_override_is_honored_by_calculate_cost() -> None: + """The escape hatch: a vendor-specific override feeds calculate_cost + through the `usage` parameter, and generic billing needs no knowledge of + the vendor's field names.""" + provider = ArtificialDumbnessProvider(base_url="http://ad.example", api_key="k") + response = { + "model": "dumb-1", + "usage": { + "dumb_tokens_in": 700, + "dumb_tokens_out": 60, + "dumb_tokens_reused": 4300, + }, + } + + result = await calculate_cost( + response, + max_cost=100000, + session=AsyncMock(), + usage=provider.normalize_usage(response["usage"]), + ) + + assert isinstance(result, CostData) + assert result.input_tokens == 700 + assert result.output_tokens == 60 + assert result.cache_read_input_tokens == 4300 + + +@pytest.mark.asyncio +async def test_explicit_usage_param_wins_over_response_extraction() -> None: + """When a normalized usage is passed, calculate_cost must not re-derive + token counts from the raw response.""" + response = { + "model": "dumb-1", + "usage": {"prompt_tokens": 999999, "completion_tokens": 999999}, + } + + result = await calculate_cost( + response, + max_cost=100000, + session=AsyncMock(), + usage=NormalizedUsage(input_tokens=10, output_tokens=5), + ) + + assert isinstance(result, CostData) + assert result.input_tokens == 10 + assert result.output_tokens == 5 From 068fb3572ffa7c857aa629e22b3ca2552ff14d0c Mon Sep 17 00:00:00 2001 From: Jeroen Ubbink Date: Thu, 11 Jun 2026 21:17:11 +0200 Subject: [PATCH 02/12] refactor: drop unused session parameter from calculate_cost The session was needed when model pricing lived in the DB (73d3613) and has been dead since pricing moved to the in-memory model map (0da08fb), yet every caller was still obliged to supply one. get_x_cashu_cost even opened a DB session per x-cashu request solely to feed it. Co-Authored-By: Claude Fable 5 --- routstr/auth.py | 2 +- routstr/payment/cost_calculation.py | 2 - routstr/upstream/base.py | 84 ++++++++++----------- tests/unit/test_cache_pricing.py | 8 +- tests/unit/test_cost_calculation_caching.py | 80 +++++++++----------- tests/unit/test_usage_normalization.py | 4 +- 6 files changed, 84 insertions(+), 96 deletions(-) diff --git a/routstr/auth.py b/routstr/auth.py index a5e1d396..2c1f0207 100644 --- a/routstr/auth.py +++ b/routstr/auth.py @@ -772,7 +772,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, usage=usage): + match await calculate_cost(response_data, deducted_max_cost, usage=usage): case MaxCostData() as cost: logger.debug( "Using max cost data (no token adjustment)", diff --git a/routstr/payment/cost_calculation.py b/routstr/payment/cost_calculation.py index 3dd52854..774afd02 100644 --- a/routstr/payment/cost_calculation.py +++ b/routstr/payment/cost_calculation.py @@ -3,7 +3,6 @@ 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 NormalizedUsage, normalize_usage, parse_token_count @@ -45,7 +44,6 @@ class CostDataError(BaseModel): async def calculate_cost( response_data: dict, max_cost: int, - session: AsyncSession, usage: NormalizedUsage | None = None, ) -> CostData | MaxCostData | CostDataError: """Calculate the cost of an API request based on token usage. diff --git a/routstr/upstream/base.py b/routstr/upstream/base.py index 56cd80b8..6c2d11db 100644 --- a/routstr/upstream/base.py +++ b/routstr/upstream/base.py @@ -3071,49 +3071,47 @@ 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, - usage=self.normalize_usage(response_data.get("usage")), - ): - 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, + usage=self.normalize_usage(response_data.get("usage")), + ): + 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( diff --git a/tests/unit/test_cache_pricing.py b/tests/unit/test_cache_pricing.py index 9d73d231..14eb7965 100644 --- a/tests/unit/test_cache_pricing.py +++ b/tests/unit/test_cache_pricing.py @@ -13,7 +13,7 @@ Specifies two things: """ import os -from unittest.mock import AsyncMock, Mock, patch +from unittest.mock import Mock, patch import litellm import pytest @@ -178,7 +178,7 @@ async def test_deepseek_cache_hits_billed_at_cache_rate(model_pricing: Mock) -> } with patch("routstr.proxy.get_model_instance", return_value=model_pricing): - result = await calculate_cost(response, max_cost=100000, session=AsyncMock()) + 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 @@ -203,7 +203,7 @@ async def test_anthropic_cache_write_billed_at_write_rate(model_pricing: Mock) - } with patch("routstr.proxy.get_model_instance", return_value=model_pricing): - result = await calculate_cost(response, max_cost=100000, session=AsyncMock()) + result = await calculate_cost(response, max_cost=100000) assert isinstance(result, CostData) # 300 @ 1 + 500 @ 0.1 + 2000 @ 1.25 + 100 @ 2 @@ -234,7 +234,7 @@ async def test_missing_cache_rate_falls_back_to_input_rate( } with patch("routstr.proxy.get_model_instance", return_value=model): - result = await calculate_cost(response, max_cost=100000, session=AsyncMock()) + result = await calculate_cost(response, max_cost=100000) assert isinstance(result, CostData) # 1000 @ 1 + 9000 @ 1 (fallback) + 500 @ 2 diff --git a/tests/unit/test_cost_calculation_caching.py b/tests/unit/test_cost_calculation_caching.py index 8b8e09ea..2eb7a7c0 100644 --- a/tests/unit/test_cost_calculation_caching.py +++ b/tests/unit/test_cost_calculation_caching.py @@ -5,7 +5,7 @@ edge cases, and billing accuracy. """ import os -from unittest.mock import AsyncMock, patch +from unittest.mock import patch import pytest @@ -17,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.""" @@ -42,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", @@ -54,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 @@ -66,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", @@ -77,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 @@ -90,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", @@ -102,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) @@ -115,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", @@ -128,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) @@ -140,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", @@ -150,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) @@ -162,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", @@ -174,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) @@ -186,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", @@ -197,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 @@ -210,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", @@ -221,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 @@ -234,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 = { @@ -244,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 @@ -255,7 +249,7 @@ 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", @@ -265,7 +259,7 @@ async def test_float_token_values_coerced_to_int(mock_session: AsyncMock, mock_f "cache_read_input_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 @@ -277,7 +271,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", @@ -287,7 +281,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 @@ -298,7 +292,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", @@ -310,7 +304,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 @@ -326,7 +320,7 @@ async def test_zero_cache_tokens(mock_session: AsyncMock, mock_fixed_pricing: No # them as regular input is a large overcharge. # ============================================================================ @pytest.mark.asyncio -async def test_deepseek_cache_hit_tokens_extracted(mock_session: AsyncMock) -> None: +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). @@ -341,7 +335,7 @@ async def test_deepseek_cache_hit_tokens_extracted(mock_session: AsyncMock) -> N "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 # only the cache misses @@ -350,7 +344,7 @@ async def test_deepseek_cache_hit_tokens_extracted(mock_session: AsyncMock) -> N @pytest.mark.asyncio -async def test_deepseek_all_tokens_cached(mock_session: AsyncMock) -> None: +async def test_deepseek_all_tokens_cached() -> None: """A fully cached DeepSeek prompt bills zero regular input tokens.""" response = { "model": "deepseek-chat", @@ -361,7 +355,7 @@ async def test_deepseek_all_tokens_cached(mock_session: AsyncMock) -> None: "prompt_cache_miss_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 == 0 @@ -369,7 +363,7 @@ async def test_deepseek_all_tokens_cached(mock_session: AsyncMock) -> None: @pytest.mark.asyncio -async def test_dialect_precedence_never_double_subtracts(mock_session: AsyncMock) -> None: +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 = { @@ -382,7 +376,7 @@ async def test_dialect_precedence_never_double_subtracts(mock_session: AsyncMock "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, applied exactly once @@ -390,7 +384,7 @@ async def test_dialect_precedence_never_double_subtracts(mock_session: AsyncMock @pytest.mark.asyncio -async def test_deepseek_malformed_hit_tokens_coerce_to_zero(mock_session: AsyncMock) -> None: +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 = { @@ -402,7 +396,7 @@ async def test_deepseek_malformed_hit_tokens_coerce_to_zero(mock_session: AsyncM "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 @@ -413,10 +407,10 @@ async def test_deepseek_malformed_hit_tokens_coerce_to_zero(mock_session: AsyncM # 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 @@ -428,10 +422,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 diff --git a/tests/unit/test_usage_normalization.py b/tests/unit/test_usage_normalization.py index 8401c355..0412d55e 100644 --- a/tests/unit/test_usage_normalization.py +++ b/tests/unit/test_usage_normalization.py @@ -10,7 +10,7 @@ needs no vendor knowledge of its own. """ import os -from unittest.mock import AsyncMock, patch +from unittest.mock import patch import pytest @@ -175,7 +175,6 @@ async def test_provider_override_is_honored_by_calculate_cost() -> None: result = await calculate_cost( response, max_cost=100000, - session=AsyncMock(), usage=provider.normalize_usage(response["usage"]), ) @@ -197,7 +196,6 @@ async def test_explicit_usage_param_wins_over_response_extraction() -> None: result = await calculate_cost( response, max_cost=100000, - session=AsyncMock(), usage=NormalizedUsage(input_tokens=10, output_tokens=5), ) From cbc424e8e740a3c9c3dc8a07a6a1a0150a961711 Mon Sep 17 00:00:00 2001 From: Jeroen Ubbink Date: Thu, 11 Jun 2026 21:22:28 +0200 Subject: [PATCH 03/12] build: type-check the entire repo in make targets, matching CI CI runs 'uv run mypy .' while the Makefile only checked routstr/, so test files could pass locally and fail the pipeline. lint, type-check and ci-lint now check everything; --ignore-missing-imports is dropped since the CI invocation passes without it. Co-Authored-By: Claude Fable 5 --- Makefile | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/Makefile b/Makefile index 3a2f605c..52c96da7 100644 --- a/Makefile +++ b/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: From 439ac48216136b274f222eb51e31de3ff3fb993f Mon Sep 17 00:00:00 2001 From: 9qeklajc Date: Sat, 13 Jun 2026 21:06:56 +0200 Subject: [PATCH 04/12] update to all providers --- routstr/payment/usage.py | 87 +++++++++++++++------ tests/unit/test_cost_calculation_caching.py | 5 +- tests/unit/test_usage_normalization.py | 39 +++++++++ 3 files changed, 105 insertions(+), 26 deletions(-) diff --git a/routstr/payment/usage.py b/routstr/payment/usage.py index bf5c2aa2..647fe6b5 100644 --- a/routstr/payment/usage.py +++ b/routstr/payment/usage.py @@ -3,11 +3,27 @@ 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: ``prompt_tokens_details.cached_tokens``, included in ``prompt_tokens`` -* Anthropic: ``cache_read_input_tokens`` / ``cache_creation_input_tokens``, - additive to (not included in) ``input_tokens`` +* 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`` + ``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 @@ -52,37 +68,60 @@ def _first_token_count(usage_data: dict, *fields: str) -> int: 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 tokens are subtracted from the input count exactly once, only for - dialects that include them in it (OpenAI, DeepSeek). Precedence between - cache fields: Anthropic explicit > OpenAI details > DeepSeek hit/miss. + 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 - input_tokens = _first_token_count(usage_data, "prompt_tokens", "input_tokens") output_tokens = _first_token_count( usage_data, "completion_tokens", "output_tokens" ) - cache_write = parse_token_count(usage_data.get("cache_creation_input_tokens", 0)) + cache_read, cache_write = _extract_cache_tokens(usage_data) - # Anthropic: cache reads are additive, input_tokens stays untouched - cache_read = parse_token_count(usage_data.get("cache_read_input_tokens", 0)) - - if not cache_read: - # OpenAI: cached tokens are included in prompt_tokens - prompt_details = usage_data.get("prompt_tokens_details") - if isinstance(prompt_details, dict): - cache_read = parse_token_count(prompt_details.get("cached_tokens", 0)) - # DeepSeek: prompt_tokens = prompt_cache_hit_tokens + prompt_cache_miss_tokens - if not cache_read: - cache_read = parse_token_count( - usage_data.get("prompt_cache_hit_tokens", 0) - ) - if cache_read: - input_tokens = max(0, input_tokens - cache_read) + # ``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, diff --git a/tests/unit/test_cost_calculation_caching.py b/tests/unit/test_cost_calculation_caching.py index 2eb7a7c0..41d85385 100644 --- a/tests/unit/test_cost_calculation_caching.py +++ b/tests/unit/test_cost_calculation_caching.py @@ -256,13 +256,14 @@ async def test_float_token_values_coerced_to_int(mock_fixed_pricing: None) -> No "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) 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 diff --git a/tests/unit/test_usage_normalization.py b/tests/unit/test_usage_normalization.py index 0412d55e..f819c2cb 100644 --- a/tests/unit/test_usage_normalization.py +++ b/tests/unit/test_usage_normalization.py @@ -94,6 +94,45 @@ def patch_sats_usd_price() -> None: # type: ignore[misc] {"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: From 355e3f19ef513e9f056a7261c30cc1c20b0095ee Mon Sep 17 00:00:00 2001 From: 9qeklajc Date: Sat, 13 Jun 2026 22:35:55 +0200 Subject: [PATCH 05/12] explicit cache --- routstr/upstream/base.py | 38 ++++++ routstr/upstream/cache_breakpoints.py | 157 +++++++++++++++++++++ tests/unit/test_cache_breakpoints.py | 189 ++++++++++++++++++++++++++ 3 files changed, 384 insertions(+) create mode 100644 routstr/upstream/cache_breakpoints.py create mode 100644 tests/unit/test_cache_breakpoints.py diff --git a/routstr/upstream/base.py b/routstr/upstream/base.py index 6c2d11db..10bcbd7f 100644 --- a/routstr/upstream/base.py +++ b/routstr/upstream/base.py @@ -42,6 +42,10 @@ from ..payment.price import sats_usd_price from ..payment.usage import NormalizedUsage, normalize_usage 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 @@ -445,6 +449,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: @@ -514,6 +531,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 diff --git a/routstr/upstream/cache_breakpoints.py b/routstr/upstream/cache_breakpoints.py new file mode 100644 index 00000000..6979819b --- /dev/null +++ b/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 diff --git a/tests/unit/test_cache_breakpoints.py b/tests/unit/test_cache_breakpoints.py new file mode 100644 index 00000000..641623b1 --- /dev/null +++ b/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 From 8f5f3d9738cfb1d43e37b62da47f302b9cb49bd3 Mon Sep 17 00:00:00 2001 From: 9qeklajc Date: Sat, 13 Jun 2026 23:26:23 +0200 Subject: [PATCH 06/12] resolve review comments --- routstr/auth.py | 9 +- routstr/payment/cost_calculation.py | 13 ++- routstr/payment/usage.py | 4 +- routstr/upstream/base.py | 31 ------- tests/unit/test_usage_normalization.py | 111 +------------------------ 5 files changed, 14 insertions(+), 154 deletions(-) diff --git a/routstr/auth.py b/routstr/auth.py index 2c1f0207..2385f09d 100644 --- a/routstr/auth.py +++ b/routstr/auth.py @@ -20,7 +20,6 @@ from .payment.cost_calculation import ( MaxCostData, calculate_cost, ) -from .payment.usage import NormalizedUsage from .wallet import credit_balance, deserialize_token_from_string logger = get_logger(__name__) @@ -680,16 +679,14 @@ async def adjust_payment_for_tokens( response_data: dict, session: AsyncSession, deducted_max_cost: int, - usage: NormalizedUsage | None = None, ) -> 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. - ``usage`` carries the upstream provider's normalized token usage (its - ``normalize_usage`` hook); when omitted, the response's usage object is - normalized with the default union parser. + 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") @@ -772,7 +769,7 @@ async def adjust_payment_for_tokens( extra={"error": str(e), "fee_msats": fee_msats}, ) - match await calculate_cost(response_data, deducted_max_cost, usage=usage): + match await calculate_cost(response_data, deducted_max_cost): case MaxCostData() as cost: logger.debug( "Using max cost data (no token adjustment)", diff --git a/routstr/payment/cost_calculation.py b/routstr/payment/cost_calculation.py index 774afd02..27b04235 100644 --- a/routstr/payment/cost_calculation.py +++ b/routstr/payment/cost_calculation.py @@ -5,7 +5,7 @@ from pydantic.v1 import BaseModel from ..core import get_logger from ..core.settings import settings from .price import sats_usd_price -from .usage import NormalizedUsage, normalize_usage, parse_token_count +from .usage import normalize_usage, parse_token_count __all__ = [ "CostData", @@ -44,20 +44,18 @@ class CostDataError(BaseModel): async def calculate_cost( response_data: dict, max_cost: int, - usage: NormalizedUsage | None = None, ) -> CostData | MaxCostData | CostDataError: """Calculate the cost of an API request based on token usage. Args: response_data: Response data containing usage information max_cost: Maximum cost in millisats - usage: Pre-normalized usage from the upstream provider's - ``normalize_usage`` hook. When omitted, the response's usage - object is normalized with the default union parser. This function - holds no vendor-dialect knowledge of its own. 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", @@ -68,8 +66,7 @@ async def calculate_cost( }, ) - if usage is None: - usage = normalize_usage(response_data.get("usage")) + usage = normalize_usage(response_data.get("usage")) if usage is None: logger.warning( diff --git a/routstr/payment/usage.py b/routstr/payment/usage.py index 647fe6b5..02c90055 100644 --- a/routstr/payment/usage.py +++ b/routstr/payment/usage.py @@ -27,8 +27,8 @@ What decides whether cached tokens must be subtracted out of the input count is ``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; vendors whose fields -would genuinely conflict override ``BaseUpstreamProvider.normalize_usage``. +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 diff --git a/routstr/upstream/base.py b/routstr/upstream/base.py index 10bcbd7f..1503cc67 100644 --- a/routstr/upstream/base.py +++ b/routstr/upstream/base.py @@ -39,7 +39,6 @@ from ..payment.models import ( list_models, ) from ..payment.price import sats_usd_price -from ..payment.usage import NormalizedUsage, normalize_usage from ..wallet import recieve_token, send_token from . import messages_dispatch from .cache_breakpoints import ( @@ -107,17 +106,6 @@ class BaseUpstreamProvider: self._models_cache = [] self._models_by_id = {} - def normalize_usage(self, usage_data: object) -> NormalizedUsage | None: - """Map this provider's usage dialect onto the canonical shape. - - The default union parser covers the known dialects (OpenAI, Anthropic, - DeepSeek), whose field names do not collide. Override in a subclass - when a vendor's usage fields would conflict with another dialect or - need bespoke interpretation; billing code consumes only the canonical - result and holds no vendor knowledge. - """ - return normalize_usage(usage_data) - def get_litellm_provider_prefix(self) -> str: """Resolve the litellm provider prefix for this provider instance. @@ -931,9 +919,6 @@ class BaseUpstreamProvider: adjustment_input, session, max_cost_for_model, - usage=self.normalize_usage( - adjustment_input.get("usage") - ), ) usage_finalized = True except Exception as e: @@ -1076,7 +1061,6 @@ class BaseUpstreamProvider: response_json, session, deducted_max_cost, - usage=self.normalize_usage(response_json.get("usage")), ) await session.refresh(key) @@ -1331,9 +1315,6 @@ class BaseUpstreamProvider: adjustment_input, session, max_cost_for_model, - usage=self.normalize_usage( - adjustment_input.get("usage") - ), ) usage_finalized = True except Exception as e: @@ -1501,7 +1482,6 @@ class BaseUpstreamProvider: response_json, session, deducted_max_cost, - usage=self.normalize_usage(response_json.get("usage")), ) await session.refresh(key) @@ -1700,7 +1680,6 @@ class BaseUpstreamProvider: fallback, new_session, max_cost_for_model, - usage=self.normalize_usage(fallback.get("usage")), ) usage_finalized = True return f"event: cost\ndata: {json.dumps({'cost': cost_data})}\n\n".encode() @@ -1852,9 +1831,6 @@ class BaseUpstreamProvider: combined_data, new_session, max_cost_for_model, - usage=self.normalize_usage( - combined_data.get("usage") - ), ) self.inject_cost_metadata( @@ -1926,7 +1902,6 @@ class BaseUpstreamProvider: response_json, session, deducted_max_cost, - usage=self.normalize_usage(response_json.get("usage")), ) self.inject_cost_metadata(response_json, cost_data, key) @@ -2032,7 +2007,6 @@ class BaseUpstreamProvider: response_json, session, max_cost_for_model, - usage=self.normalize_usage(response_json.get("usage")), ) self.inject_cost_metadata(response_json, cost_data, key) @@ -2174,7 +2148,6 @@ class BaseUpstreamProvider: fallback, new_session, max_cost_for_model, - usage=self.normalize_usage(fallback.get("usage")), ) usage_finalized = True return ( @@ -2248,9 +2221,6 @@ class BaseUpstreamProvider: combined_data, new_session, max_cost_for_model, - usage=self.normalize_usage( - combined_data.get("usage") - ), ) self.inject_cost_metadata( combined_data, cost_data, fresh_key @@ -3112,7 +3082,6 @@ class BaseUpstreamProvider: match await calculate_cost( response_data, max_cost_for_model, - usage=self.normalize_usage(response_data.get("usage")), ): case MaxCostData() as cost: logger.debug( diff --git a/tests/unit/test_usage_normalization.py b/tests/unit/test_usage_normalization.py index f819c2cb..50c6c645 100644 --- a/tests/unit/test_usage_normalization.py +++ b/tests/unit/test_usage_normalization.py @@ -3,39 +3,19 @@ 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) and exposed as an overridable -``BaseUpstreamProvider.normalize_usage`` hook for vendors whose fields would -genuinely conflict. ``calculate_cost`` accepts a pre-normalized usage and then -needs no vendor knowledge of its own. +non-colliding dialects). ``calculate_cost`` normalizes the response's usage +object with this parser and needs no vendor knowledge of its own. """ import os -from unittest.mock import patch - -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 +import pytest + from routstr.payment.usage import NormalizedUsage, normalize_usage -from routstr.upstream import BaseUpstreamProvider, GenericUpstreamProvider - - -@pytest.fixture(autouse=True) -def mock_fixed_pricing(monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.setattr(settings, "fixed_pricing", True) - monkeypatch.setattr(settings, "fixed_per_1k_input_tokens", 0.001) - monkeypatch.setattr(settings, "fixed_per_1k_output_tokens", 0.001) - - -@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 - # ============================================================================ # The union parser: one canonical shape for all known dialects @@ -158,86 +138,3 @@ def test_normalize_usage_never_negative() -> None: assert result is not None assert result.input_tokens == 0 assert result.cache_read_tokens == 150 - - -# ============================================================================ -# The provider hook: default delegates to the union parser, subclasses -# override for vendors whose fields genuinely conflict -# ============================================================================ - - -class ArtificialDumbnessProvider(BaseUpstreamProvider): - """Fictional vendor whose usage dialect collides with nothing we know.""" - - provider_type = "artificial-dumbness" - - def normalize_usage(self, usage_data: object) -> NormalizedUsage | None: - if not isinstance(usage_data, dict): - return None - return NormalizedUsage( - input_tokens=usage_data.get("dumb_tokens_in", 0), - output_tokens=usage_data.get("dumb_tokens_out", 0), - cache_read_tokens=usage_data.get("dumb_tokens_reused", 0), - cache_write_tokens=0, - ) - - -def test_base_provider_hook_delegates_to_union_parser() -> None: - """Without an override, the provider hook equals the module parser, so - DeepSeek-dialect upstreams work through GenericUpstreamProvider with no - subclass.""" - provider = GenericUpstreamProvider(base_url="http://upstream.example") - usage = { - "prompt_tokens": 10000, - "completion_tokens": 500, - "prompt_cache_hit_tokens": 9000, - "prompt_cache_miss_tokens": 1000, - } - assert provider.normalize_usage(usage) == normalize_usage(usage) - - -@pytest.mark.asyncio -async def test_provider_override_is_honored_by_calculate_cost() -> None: - """The escape hatch: a vendor-specific override feeds calculate_cost - through the `usage` parameter, and generic billing needs no knowledge of - the vendor's field names.""" - provider = ArtificialDumbnessProvider(base_url="http://ad.example", api_key="k") - response = { - "model": "dumb-1", - "usage": { - "dumb_tokens_in": 700, - "dumb_tokens_out": 60, - "dumb_tokens_reused": 4300, - }, - } - - result = await calculate_cost( - response, - max_cost=100000, - usage=provider.normalize_usage(response["usage"]), - ) - - assert isinstance(result, CostData) - assert result.input_tokens == 700 - assert result.output_tokens == 60 - assert result.cache_read_input_tokens == 4300 - - -@pytest.mark.asyncio -async def test_explicit_usage_param_wins_over_response_extraction() -> None: - """When a normalized usage is passed, calculate_cost must not re-derive - token counts from the raw response.""" - response = { - "model": "dumb-1", - "usage": {"prompt_tokens": 999999, "completion_tokens": 999999}, - } - - result = await calculate_cost( - response, - max_cost=100000, - usage=NormalizedUsage(input_tokens=10, output_tokens=5), - ) - - assert isinstance(result, CostData) - assert result.input_tokens == 10 - assert result.output_tokens == 5 From ecd46975b47b5c7ca635200d16f1bf5659e173f6 Mon Sep 17 00:00:00 2001 From: Jeroen Ubbink Date: Wed, 17 Jun 2026 17:10:37 +0200 Subject: [PATCH 07/12] fix: deduct mint NUT-02 input fee when crediting trusted-mint topups recieve_token swaps the incoming proofs at the same mint with include_fees=True (paying the mint's NUT-02 per-proof input fee) but credited the full face value. On every topup from a fee-charging trusted mint, routstr over-credited the user by the fee and its own wallet drifted toward insolvency. Subtract get_fees_for_proofs(proofs) from the credited amount, mirroring the foreign-mint swap path which already accounts for it. Adds a fee-charging trusted-mint unit test (credited == face - input_fee). Co-Authored-By: Claude Opus 4.8 --- routstr/wallet.py | 8 ++++- tests/unit/test_wallet.py | 61 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 68 insertions(+), 1 deletion(-) diff --git a/routstr/wallet.py b/routstr/wallet.py index 69ff6f82..0fe984b8 100644 --- a/routstr/wallet.py +++ b/routstr/wallet.py @@ -51,9 +51,15 @@ async def recieve_token( await wallet.load_mint(keyset_id=token_obj.keysets[0]) wallet.verify_proofs_dleq(token_obj.proofs) + # Same-mint receive (not swap_to_primary_mint): split() re-mints the incoming + # proofs into fresh ones we own so the sender can't double-spend them. With + # include_fees=True the mint deducts its NUT-02 per-proof input fee, so we end + # up holding only `amount - input_fees`. Credit that, not the face value, or + # routstr over-credits the user and its wallet drifts insolvent. + input_fees = wallet.get_fees_for_proofs(token_obj.proofs) await wallet.split(proofs=token_obj.proofs, amount=0, include_fees=True) - return token_obj.amount, token_obj.unit, token_obj.mint + return token_obj.amount - input_fees, token_obj.unit, token_obj.mint async def send(amount: int, unit: str, mint_url: str | None = None) -> tuple[int, str]: diff --git a/tests/unit/test_wallet.py b/tests/unit/test_wallet.py index 245a624a..e536baf9 100644 --- a/tests/unit/test_wallet.py +++ b/tests/unit/test_wallet.py @@ -39,6 +39,8 @@ async def test_recieve_token_valid() -> None: mock_wallet = Mock() mock_wallet.split = AsyncMock() + # Fee-free trusted mint (e.g. Minibits): nothing deducted. + mock_wallet.get_fees_for_proofs = Mock(return_value=0) from routstr.core.settings import settings @@ -61,6 +63,65 @@ async def test_recieve_token_valid() -> None: assert mint == "http://mint:3338" +@pytest.mark.asyncio +async def test_recieve_token_trusted_mint_deducts_input_fee() -> None: + """A trusted mint that charges NUT-02 input fees. + + The same-mint receive (`wallet.split(..., include_fees=True)`, a NUT-03 swap + at the same mint — not swap_to_primary_mint) pays the mint's per-proof fee, + so routstr only ends up with `face - input_fee` in fresh proofs. The credited + amount must reflect that, otherwise routstr over-credits the user and its own + wallet drifts toward insolvency. + """ + token_data = { + "token": [ + { + "mint": "http://mint:3338", + "proofs": [ + {"amount": 1000, "id": "test", "secret": "secret", "C": "curve"} + ], + } + ], + "unit": "sat", + } + token_json = json.dumps(token_data) + token_b64 = base64.urlsafe_b64encode(token_json.encode()).decode() + token_str = f"cashuA{token_b64}" + + mock_wallet = Mock() + mock_wallet.split = AsyncMock() + # 21 proofs @ 100 ppk -> (21*100 + 999) // 1000 = 3 sat input fee + mock_wallet.get_fees_for_proofs = Mock(return_value=3) + + from routstr.core.settings import settings + + with patch.object(settings, "cashu_mints", ["http://mint:3338"]): + with patch("routstr.wallet.deserialize_token_from_string") as mock_deserialize: + mock_token = Mock() + mock_token.keysets = ["keyset1"] + mock_token.mint = "http://mint:3338" + mock_token.unit = "sat" + mock_token.amount = 1000 + mock_token.proofs = [{"amount": 1000}] + mock_deserialize.return_value = mock_token + + mock_wallet.load_mint = AsyncMock() + mock_wallet.load_proofs = AsyncMock() + # Patch get_wallet directly so the module-level `_wallets` cache + # (keyed by mint URL) can't hand back a wallet from another test. + with patch( + "routstr.wallet.get_wallet", + AsyncMock(return_value=mock_wallet), + ): + amount, unit, mint = await recieve_token(token_str) + assert amount == 997 # 1000 face - 3 sat input fee paid on swap + assert unit == "sat" + assert mint == "http://mint:3338" + mock_wallet.get_fees_for_proofs.assert_called_once_with( + mock_token.proofs + ) + + @pytest.mark.asyncio async def test_send_token() -> None: mock_wallet = Mock() From e9dced814859efe6f6ee2b0b8e77a0e2fa274631 Mon Sep 17 00:00:00 2001 From: 9qeklajc Date: Wed, 17 Jun 2026 17:27:08 +0200 Subject: [PATCH 08/12] rollback when token generation failed --- routstr/auth.py | 1 + tests/unit/test_auth_cashu.py | 59 +++++++++++++++++++++++++++++++++++ 2 files changed, 60 insertions(+) create mode 100644 tests/unit/test_auth_cashu.py diff --git a/routstr/auth.py b/routstr/auth.py index 79e05807..1f910fed 100644 --- a/routstr/auth.py +++ b/routstr/auth.py @@ -364,6 +364,7 @@ async def validate_bearer_key( except HTTPException: raise except Exception as e: + await session.rollback() logger.error( "Cashu token redemption failed", extra={ diff --git a/tests/unit/test_auth_cashu.py b/tests/unit/test_auth_cashu.py new file mode 100644 index 00000000..cf1c98de --- /dev/null +++ b/tests/unit/test_auth_cashu.py @@ -0,0 +1,59 @@ +import hashlib +from types import SimpleNamespace +from typing import AsyncGenerator +from unittest.mock import AsyncMock, patch + +import pytest +from fastapi import HTTPException +from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine +from sqlalchemy.pool import StaticPool +from sqlmodel import SQLModel +from sqlmodel.ext.asyncio.session import AsyncSession + +from routstr.auth import validate_bearer_key +from routstr.core.db import ApiKey + + +def _make_engine() -> AsyncEngine: + return create_async_engine( + "sqlite+aiosqlite://", + poolclass=StaticPool, + connect_args={"check_same_thread": False}, + ) + + +@pytest.fixture +async def session() -> AsyncGenerator[AsyncSession, None]: + engine = _make_engine() + async with engine.begin() as conn: + await conn.run_sync(SQLModel.metadata.create_all) + db_session = AsyncSession(engine, expire_on_commit=False) + try: + yield db_session + finally: + await db_session.close() + await engine.dispose() + + +@pytest.mark.asyncio +async def test_failed_first_cashu_redemption_rolls_back_empty_api_key( + session: AsyncSession, +) -> None: + token = "cashuAfirst_seen_but_redemption_fails" + hashed_key = hashlib.sha256(token.encode()).hexdigest() + token_obj = SimpleNamespace(mint="http://mint:3338", unit="sat") + + from routstr.core.settings import settings + + with ( + patch.object(settings, "cashu_mints", ["http://mint:3338"]), + patch("routstr.auth.deserialize_token_from_string", return_value=token_obj), + patch( + "routstr.auth.credit_balance", + new=AsyncMock(side_effect=ValueError("token already spent")), + ), + ): + with pytest.raises(HTTPException): + await validate_bearer_key(token, session) + + assert await session.get(ApiKey, hashed_key) is None From 7969a8da55da01a6868ee98181640b2d2f5fad44 Mon Sep 17 00:00:00 2001 From: Jeroen Ubbink Date: Thu, 18 Jun 2026 12:26:44 +0200 Subject: [PATCH 09/12] test: clarify mocked input-fee comment in trusted-mint test The comment described "21 proofs @ 100 ppk" arithmetic, but the mocked token has one proof and get_fees_for_proofs is hard-mocked to 3, so the math wasn't exercised. Describe what the mock actually does. Co-Authored-By: Claude Opus 4.8 --- tests/unit/test_wallet.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/unit/test_wallet.py b/tests/unit/test_wallet.py index e536baf9..e972232a 100644 --- a/tests/unit/test_wallet.py +++ b/tests/unit/test_wallet.py @@ -90,7 +90,7 @@ async def test_recieve_token_trusted_mint_deducts_input_fee() -> None: mock_wallet = Mock() mock_wallet.split = AsyncMock() - # 21 proofs @ 100 ppk -> (21*100 + 999) // 1000 = 3 sat input fee + # Mock a 3-sat input fee from the Cashu wallet API. mock_wallet.get_fees_for_proofs = Mock(return_value=3) from routstr.core.settings import settings From 75ed865a5f0cd202d97820b2d2c9ec4558b696a0 Mon Sep 17 00:00:00 2001 From: Jeroen Ubbink Date: Thu, 18 Jun 2026 15:48:49 +0200 Subject: [PATCH 10/12] fix: deduct input fee in swap_to_primary_mint same-mint shortcut MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The same-mint shortcut in swap_to_primary_mint did a same-mint split(include_fees=True) — which burns the mint's NUT-02 per-proof input fee — but returned the full token amount, over-crediting the user (the same bug already fixed for the trusted-mint receive path). It also skipped DLEQ verification that the trusted path performs. Extract the shared same-mint redeem into _redeem_same_mint (load mint, verify DLEQ, split, credit amount - input_fees) and delegate from both recieve_token and the shortcut, so the two paths can't drift again. This shortcut is reachable when PRIMARY_MINT_URL is set outside CASHU_MINTS. Co-Authored-By: Claude Opus 4.8 --- routstr/wallet.py | 43 +++++++++++++++++++++++---------------- tests/unit/test_wallet.py | 24 +++++++++++++++++++--- 2 files changed, 46 insertions(+), 21 deletions(-) diff --git a/routstr/wallet.py b/routstr/wallet.py index 0fe984b8..815da134 100644 --- a/routstr/wallet.py +++ b/routstr/wallet.py @@ -35,6 +35,24 @@ async def get_balance(unit: str) -> int: return wallet.available_balance.amount +async def _redeem_same_mint( + wallet: Wallet, token_obj: Token +) -> tuple[int, str, str]: # amount, unit, mint_url + """Redeem proofs at their own issuing mint (no cross-mint swap). + + split() re-mints the incoming proofs into fresh ones we own so the sender + can't double-spend them. With include_fees=True the mint deducts its NUT-02 + per-proof input fee, so we end up holding only `amount - input_fees`. Credit + that, not the face value, or routstr over-credits the user and its wallet + drifts insolvent. + """ + await wallet.load_mint(keyset_id=token_obj.keysets[0]) + wallet.verify_proofs_dleq(token_obj.proofs) + input_fees = wallet.get_fees_for_proofs(token_obj.proofs) + await wallet.split(proofs=token_obj.proofs, amount=0, include_fees=True) + return int(token_obj.amount) - input_fees, token_obj.unit, token_obj.mint + + async def recieve_token( token: str, ) -> tuple[int, str, str]: # amount, unit, mint_url @@ -48,18 +66,7 @@ async def recieve_token( if token_obj.mint not in settings.cashu_mints: return await swap_to_primary_mint(token_obj, wallet) - await wallet.load_mint(keyset_id=token_obj.keysets[0]) - - wallet.verify_proofs_dleq(token_obj.proofs) - # Same-mint receive (not swap_to_primary_mint): split() re-mints the incoming - # proofs into fresh ones we own so the sender can't double-spend them. With - # include_fees=True the mint deducts its NUT-02 per-proof input fee, so we end - # up holding only `amount - input_fees`. Credit that, not the face value, or - # routstr over-credits the user and its wallet drifts insolvent. - input_fees = wallet.get_fees_for_proofs(token_obj.proofs) - await wallet.split(proofs=token_obj.proofs, amount=0, include_fees=True) - - return token_obj.amount - input_fees, token_obj.unit, token_obj.mint + return await _redeem_same_mint(wallet, token_obj) async def send(amount: int, unit: str, mint_url: str | None = None) -> tuple[int, str]: @@ -219,10 +226,9 @@ async def swap_to_primary_mint( amount_msat = token_amount else: raise ValueError("Invalid unit") - primary_wallet = await get_wallet(settings.primary_mint, settings.primary_mint_unit) - - # If the token is already from the primary mint, we don't need to swap - # and we definitely don't want to calculate or pay fees. + # If the token is already from the primary mint, we don't need a cross-mint + # swap — redeem it same-mint. There's no melt/Lightning fee, but the mint's + # NUT-02 input fee still applies; _redeem_same_mint accounts for it. if token_obj.mint == settings.primary_mint: logger.info( "swap_to_primary_mint: token already on primary mint, skipping swap", @@ -232,8 +238,9 @@ async def swap_to_primary_mint( "unit": token_obj.unit, }, ) - await token_wallet.split(proofs=token_obj.proofs, amount=0, include_fees=True) - return token_amount, token_obj.unit, token_obj.mint + return await _redeem_same_mint(token_wallet, token_obj) + + primary_wallet = await get_wallet(settings.primary_mint, settings.primary_mint_unit) minted_amount = await _calculate_swap_amount( amount_msat, diff --git a/tests/unit/test_wallet.py b/tests/unit/test_wallet.py index e972232a..40a4df6c 100644 --- a/tests/unit/test_wallet.py +++ b/tests/unit/test_wallet.py @@ -120,6 +120,10 @@ async def test_recieve_token_trusted_mint_deducts_input_fee() -> None: mock_wallet.get_fees_for_proofs.assert_called_once_with( mock_token.proofs ) + # DLEQ is verified before re-minting the incoming proofs. + mock_wallet.verify_proofs_dleq.assert_called_once_with( + mock_token.proofs + ) @pytest.mark.asyncio @@ -315,19 +319,31 @@ async def test_recieve_token_untrusted_mint() -> None: assert mint == "http://mint:3338" -@pytest.mark.asyncio @pytest.mark.asyncio async def test_swap_to_primary_mint_already_on_primary() -> None: + """Same-mint shortcut: the token is already on the primary mint. + + No cross-mint swap (no melt/mint), but the same-mint split(include_fees=True) + still burns the mint's NUT-02 input fee, so the credited amount must be face + minus the input fee — not full face value (the over-credit bug). DLEQ is + verified too, matching the trusted same-mint receive path. + """ from routstr.core.settings import settings from routstr.wallet import swap_to_primary_mint mock_token = Mock() mock_token.mint = settings.primary_mint + mock_token.keysets = ["keyset1"] mock_token.amount = 1000 mock_token.unit = "sat" - mock_token.proofs = [] + mock_token.proofs = [{"amount": 1000}] mock_token_wallet = Mock() + mock_token_wallet.load_mint = AsyncMock() + mock_token_wallet.load_proofs = AsyncMock() + mock_token_wallet.verify_proofs_dleq = Mock() + # Mock a 3-sat input fee from the Cashu wallet API. + mock_token_wallet.get_fees_for_proofs = Mock(return_value=3) mock_token_wallet.split = AsyncMock(return_value=None) mock_token_wallet.request_mint = AsyncMock() mock_token_wallet.melt_quote = AsyncMock() @@ -335,9 +351,11 @@ async def test_swap_to_primary_mint_already_on_primary() -> None: with patch("routstr.wallet.get_wallet", AsyncMock(return_value=mock_token_wallet)): amount, unit, mint = await swap_to_primary_mint(mock_token, mock_token_wallet) - assert amount == 1000 + assert amount == 997 # 1000 face - 3 sat input fee assert unit == "sat" assert mint == settings.primary_mint + mock_token_wallet.verify_proofs_dleq.assert_called_once_with(mock_token.proofs) + mock_token_wallet.get_fees_for_proofs.assert_called_once_with(mock_token.proofs) mock_token_wallet.split.assert_called_once() mock_token_wallet.request_mint.assert_not_called() mock_token_wallet.melt_quote.assert_not_called() From ff9c645bb1f83eac8f4a8ecb14770d10c6e95414 Mon Sep 17 00:00:00 2001 From: 9qeklajc Date: Fri, 19 Jun 2026 11:20:56 +0200 Subject: [PATCH 11/12] make sure balance is set correctly --- routstr/wallet.py | 10 +++- tests/unit/test_fetch_all_balances.py | 73 +++++++++++++++++++++++++++ 2 files changed, 82 insertions(+), 1 deletion(-) create mode 100644 tests/unit/test_fetch_all_balances.py diff --git a/routstr/wallet.py b/routstr/wallet.py index 815da134..30f36ded 100644 --- a/routstr/wallet.py +++ b/routstr/wallet.py @@ -580,11 +580,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 ] diff --git a/tests/unit/test_fetch_all_balances.py b/tests/unit/test_fetch_all_balances.py new file mode 100644 index 00000000..dcd99107 --- /dev/null +++ b/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 From ccab5e42165d5ad6aeb4abbf519f49b0befd22e4 Mon Sep 17 00:00:00 2001 From: 9qeklajc Date: Fri, 19 Jun 2026 11:53:30 +0200 Subject: [PATCH 12/12] collapse when focus change --- ui/app/providers/page.tsx | 13 +++++-------- 1 file changed, 5 insertions(+), 8 deletions(-) diff --git a/ui/app/providers/page.tsx b/ui/app/providers/page.tsx index 4a7046a8..0a40150e 100644 --- a/ui/app/providers/page.tsx +++ b/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); } };