laantungir/routstr-core
1
0
Fork 0
mirror of https://github.com/Routstr/routstr-core.git synced 2026-10-05 12:28:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #581 from jeroenubbink/fix/generic-provider-pricing

Browse Source 
Resolve generic provider pricing instead of fabricating it
This commit is contained in:
9qeklajc
2026-07-07 14:38:32 +02:00
committed by GitHub
parent f55c3e9c8c 2baea4149e
commit 97172641f8
4 changed files with 845 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files
routstr/payment/models.py
+27 -28
View File
@@ -85,6 +85,30 @@ class Model(BaseModel):
return hash(self.id)
def litellm_cost_entry(model_id: str) -> dict | None:
"""Look up ``model_id`` in litellm's bundled cost map.
litellm ships per-model USD rates keyed by the exact OpenRouter id
(``deepseek/deepseek-chat``) or the bare model name (``gpt-4o``,
``claude-sonnet-4-5``), so both spellings are tried. Keys are lowercase, so
a mixed-case upstream id (``deepseek-ai/DeepSeek-V4-Flash``) is retried via
a case-insensitive scan. Returns the matched cost dict, or ``None``.
"""
import litellm
candidates = (model_id, model_id.split("/", 1)[-1])
for key in candidates:
info = litellm.model_cost.get(key)
if isinstance(info, dict):
return info
lowered = {c.lower() for c in candidates}
for key, info in litellm.model_cost.items():
if isinstance(key, str) and key.lower() in lowered and isinstance(info, dict):
return info
return None
def backfill_cache_pricing(model_id: str, pricing: Pricing) -> Pricing:
"""Fill missing cache rates from litellm's bundled cost map.
@@ -92,12 +116,8 @@ def backfill_cache_pricing(model_id: str, pricing: Pricing) -> Pricing:
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. litellm keys are
lowercase, but a generic upstream may report a mixed-case id
(``deepseek-ai/DeepSeek-V4-Flash``); an exact match is attempted first, then
a case-insensitive fallback so such ids still resolve.
cache writes (1.25x). The lookup (see ``litellm_cost_entry``) tries both
id spellings and a case-insensitive fallback.
Rates already present (e.g. provided by OpenRouter) are authoritative and
never overwritten. Unknown models are returned unchanged.
@@ -107,28 +127,7 @@ def backfill_cache_pricing(model_id: str, pricing: Pricing) -> Pricing:
if not (needs_read or needs_write):
return pricing
import litellm
candidates = (model_id, model_id.split("/", 1)[-1])
info: dict | None = None
for key in candidates:
candidate = litellm.model_cost.get(key)
if isinstance(candidate, dict):
info = candidate
break
if info is None:
# Case-insensitive fallback: a mixed-case upstream id (e.g.
# ``deepseek-ai/DeepSeek-V4-Flash``) won't match litellm's lowercase
# keys exactly. Build a lowercased index once and retry.
lowered = {c.lower() for c in candidates}
for key, candidate in litellm.model_cost.items():
if (
isinstance(key, str)
and key.lower() in lowered
and isinstance(candidate, dict)
):
info = candidate
break
info = litellm_cost_entry(model_id)
if info is None:
return pricing
routstr/upstream/generic.py
+91 -44
View File
@@ -5,6 +5,12 @@ from typing import TYPE_CHECKING
import httpx
from .base import BaseUpstreamProvider
from .pricing_resolver import (
FallbackPricingResolver,
ResolvedPricing,
_as_float,
estimate_context_length,
)
if TYPE_CHECKING:
from ..core.db import UpstreamProviderRow
@@ -64,6 +70,40 @@ class GenericUpstreamProvider(BaseUpstreamProvider):
"platform_url": cls.platform_url,
}
def _native_pricing(
self, model_id: str, model_spec: dict
) -> ResolvedPricing | None:
"""Read pricing/metadata from Venice's bespoke ``model_spec`` schema.
Returns ``None`` when the upstream reported no *usable* native price —
absent, non-numeric, negative, or both-zero — so the caller falls
through to the shared resolution chain instead of fabricating a number
or trusting a bogus one. This mirrors the money-safety guards the
litellm and OpenRouter rungs already apply: a both-zero price would
serve the model free, a negative one would credit the caller, and a
non-numeric string would otherwise throw and drop the whole catalog.
"""
pricing_info = model_spec.get("pricing", {})
input_usd = _as_float(pricing_info.get("input", {}).get("usd"))
output_usd = _as_float(pricing_info.get("output", {}).get("usd"))
if input_usd is None or output_usd is None:
return None
if input_usd < 0 or output_usd < 0 or (input_usd == 0 and output_usd == 0):
return None
capabilities = model_spec.get("capabilities", {})
input_modalities = ["text"]
if capabilities.get("supportsVision", False):
input_modalities.append("image")
return ResolvedPricing(
prompt=input_usd / 1_000_000,
completion=output_usd / 1_000_000,
context_length=model_spec.get("availableContextTokens"),
source="native",
input_modalities=input_modalities,
)
async def fetch_models(self) -> list[Model]:
"""Fetch models from upstream API using /models endpoint."""
from ..payment.models import Architecture, Model, Pricing, TopProvider
@@ -78,6 +118,7 @@ class GenericUpstreamProvider(BaseUpstreamProvider):
response.raise_for_status()
data = response.json()
resolver = FallbackPricingResolver()
models_list = []
for model_data in data.get("data", []):
model_id = model_data.get("id", "")
@@ -89,41 +130,44 @@ class GenericUpstreamProvider(BaseUpstreamProvider):
owned_by = model_data.get("owned_by", "unknown")
model_spec = model_data.get("model_spec", {})
context_length = 4096
if model_spec.get("availableContextTokens"):
context_length = model_spec["availableContextTokens"]
elif any(
pattern in model_id.lower() for pattern in ["32k", "32000"]
):
context_length = 32768
elif any(
pattern in model_id.lower() for pattern in ["16k", "16000"]
):
context_length = 16384
elif any(pattern in model_id.lower() for pattern in ["8k", "8000"]):
context_length = 8192
elif "gpt-4" in model_id.lower():
context_length = 8192
elif "claude" in model_id.lower():
context_length = 200000
resolved = self._native_pricing(model_id, model_spec)
if resolved is None:
resolved = await resolver.resolve(model_id)
pricing_info = model_spec.get("pricing", {})
input_pricing = pricing_info.get("input", {})
output_pricing = pricing_info.get("output", {})
if resolved is None:
# Fail closed: never invent a price. Import the model
# disabled with a warning so the operator can price it
# (the admin UI surfaces disabled remote models).
logger.warning(
f"No pricing source resolved for '{model_id}' from "
f"{self.upstream_name}; importing it disabled",
extra={"model_id": model_id, "base_url": self.base_url},
)
resolved = ResolvedPricing(
prompt=0.0,
completion=0.0,
context_length=None,
source="unresolved",
)
enabled = False
else:
enabled = True
prompt_price = input_pricing.get("usd", 0.001) / 1000000
completion_price = output_pricing.get("usd", 0.001) / 1000000
# Prefer the source's own modality string (OpenRouter ships
# one, e.g. "text+image->text"); otherwise derive it from the
# captured input/output modalities in the same "in->out" shape
# rather than flattening vision models to "text->text".
modality = resolved.modality or (
f"{'+'.join(resolved.input_modalities)}"
f"->{'+'.join(resolved.output_modalities)}"
)
capabilities = model_spec.get("capabilities", {})
input_modalities = ["text"]
output_modalities = ["text"]
if capabilities.get("supportsVision", False):
input_modalities.append("image")
modality = "text"
if capabilities.get("supportsVision", False):
modality = "text->text"
# A source can carry a price but no context (e.g. a litellm
# entry missing max_input_tokens); fall back to an id-based
# estimate so we never persist a zero-length window.
context_length = resolved.context_length or estimate_context_length(
model_id
)
spec_name = model_spec.get("name", model_name)
description = f"{spec_name}"
@@ -139,30 +183,33 @@ class GenericUpstreamProvider(BaseUpstreamProvider):
context_length=context_length,
architecture=Architecture(
modality=modality,
input_modalities=input_modalities,
output_modalities=output_modalities,
tokenizer="unknown",
instruct_type=None,
input_modalities=resolved.input_modalities,
output_modalities=resolved.output_modalities,
tokenizer=resolved.tokenizer,
instruct_type=resolved.instruct_type,
),
pricing=Pricing(
prompt=prompt_price,
completion=completion_price,
prompt=resolved.prompt,
completion=resolved.completion,
request=0.0,
image=0.0,
web_search=0.0,
internal_reasoning=0.0,
max_prompt_cost=0.001,
max_completion_cost=0.001,
max_cost=0.001,
input_cache_read=resolved.input_cache_read,
input_cache_write=resolved.input_cache_write,
),
sats_pricing=None,
per_request_limits=None,
top_provider=TopProvider(
context_length=context_length,
max_completion_tokens=context_length // 2,
is_moderated=False,
max_completion_tokens=(
resolved.max_completion_tokens
if resolved.max_completion_tokens is not None
else context_length // 2
),
is_moderated=bool(resolved.is_moderated),
),
enabled=True,
enabled=enabled,
upstream_provider_id=None,
canonical_slug=None,
)
routstr/upstream/pricing_resolver.py
+203
View File
@@ -0,0 +1,203 @@
"""Shared price/metadata resolution chain for upstream model discovery.
Most OpenAI-compatible ``/models`` responses carry no pricing. Rather than let
a provider fabricate one, this module resolves a model through decreasingly
trustworthy sources — litellm's bundled cost map (curated list prices, mirrors
provider docs), then the OpenRouter feed (resale prices, broader coverage) —
and returns ``None`` when none of them know the model, so the caller can fail
closed instead of inventing a number.
Provider-native pricing (a gateway's own ``/models`` schema, e.g. Venice's
``model_spec``) is authoritative and handled by the provider before this chain
is consulted; only the shared fallback lives here so a later refactor can hoist
it into the base provider unchanged.
"""
from __future__ import annotations
from dataclasses import dataclass, field
@dataclass
class ResolvedPricing:
"""Per-token pricing plus whatever metadata the answering source carried.
Prices are USD per token. ``source`` records provenance
(``native``/``litellm``/``openrouter``/``unresolved``) so later work can
surface where each price came from.
"""
prompt: float
completion: float
context_length: int | None
source: str
modality: str | None = None
max_completion_tokens: int | None = None
input_cache_read: float = 0.0
input_cache_write: float = 0.0
input_modalities: list[str] = field(default_factory=lambda: ["text"])
output_modalities: list[str] = field(default_factory=lambda: ["text"])
tokenizer: str = "unknown"
instruct_type: str | None = None
is_moderated: bool | None = None
def estimate_context_length(model_id: str) -> int:
"""Best-effort context window from a model id when no source reports one.
The last rung of the fallback chain, reached only for a model whose price
resolved but whose context did not (or that imported disabled). Context is
not a billing input, so a rough id-based guess is acceptable here where a
guessed *price* never would be.
"""
lowered = model_id.lower()
if any(pattern in lowered for pattern in ["32k", "32000"]):
return 32768
if any(pattern in lowered for pattern in ["16k", "16000"]):
return 16384
if any(pattern in lowered for pattern in ["8k", "8000"]):
return 8192
if "gpt-4" in lowered:
return 8192
if "claude" in lowered:
return 200000
return 4096
def _as_float(value: object) -> float | None:
"""OpenRouter reports prices as strings; coerce, ``None`` if unparseable."""
try:
return float(value) # type: ignore[arg-type]
except (TypeError, ValueError):
return None
def _as_int(value: object) -> int | None:
"""Coerce an already-numeric token count to ``int``, else ``None``."""
return int(value) if isinstance(value, (int, float)) else None
def _from_litellm(model_id: str) -> ResolvedPricing | None:
# Lazy import so the resolver stays import-light and shares the exact
# lookup semantics used by cache-rate backfill.
from ..payment.models import litellm_cost_entry
info = litellm_cost_entry(model_id)
if info is None:
return None
prompt = info.get("input_cost_per_token")
completion = info.get("output_cost_per_token")
if not isinstance(prompt, (int, float)) or not isinstance(completion, (int, float)):
return None
# A both-zero entry is litellm listing a model without a real price (free
# moderation/rerank tiers do this) — treating 0/0 as resolved would serve
# the model for free. Reject it (and any negative) so the caller falls
# through, mirroring async_fetch_openrouter_models' _has_valid_pricing.
if prompt < 0 or completion < 0 or (prompt == 0 and completion == 0):
return None
input_modalities = ["text"]
if info.get("supports_vision"):
input_modalities.append("image")
return ResolvedPricing(
prompt=float(prompt),
completion=float(completion),
# max_input_tokens is the context window; max_tokens is litellm's
# completion cap (it tracks max_output_tokens for ~94% of models), so
# it is never a context source. A missing window falls to the id-based
# estimate downstream rather than borrowing the output cap.
context_length=_as_int(info.get("max_input_tokens")),
source="litellm",
max_completion_tokens=_as_int(info.get("max_output_tokens")),
input_cache_read=float(info.get("cache_read_input_token_cost") or 0.0),
input_cache_write=float(info.get("cache_creation_input_token_cost") or 0.0),
input_modalities=input_modalities,
)
def _match_openrouter(model_id: str, feed: list[dict]) -> dict | None:
"""Find ``model_id`` in the OpenRouter feed, exact id before bare tail.
Bare-tail matching (``deepseek-chat`` ↔ ``deepseek/deepseek-chat``) is a
looser, lower-trust match — OpenRouter fans a model out across resellers —
so an exact id match always wins first. When several entries share the bare
tail, the one with the highest *combined* (prompt + completion) per-token
cost wins: the choice must be deterministic (not feed-order-dependent) and
money-safe whichever way traffic leans, since undercharging is the hazard.
Ranking on prompt alone could pick an entry that is cheap on input but dear
on output. The live feed has no such collisions today; this only governs
the latent case.
"""
bare = model_id.split("/", 1)[-1]
exact = next((m for m in feed if m.get("id") == model_id), None)
if exact is not None:
return exact
matches = [m for m in feed if m.get("id", "").split("/", 1)[-1] == bare]
if not matches:
return None
def _combined_cost(m: dict) -> float:
pricing = m.get("pricing", {})
return (_as_float(pricing.get("prompt")) or 0.0) + (
_as_float(pricing.get("completion")) or 0.0
)
return max(matches, key=_combined_cost)
def _from_openrouter(model_id: str, feed: list[dict]) -> ResolvedPricing | None:
entry = _match_openrouter(model_id, feed)
if entry is None:
return None
pricing = entry.get("pricing", {})
prompt = _as_float(pricing.get("prompt"))
completion = _as_float(pricing.get("completion"))
if prompt is None or completion is None:
return None
architecture = entry.get("architecture", {})
top_provider = entry.get("top_provider", {})
return ResolvedPricing(
prompt=prompt,
completion=completion,
context_length=_as_int(entry.get("context_length")),
source="openrouter",
modality=architecture.get("modality"),
max_completion_tokens=_as_int(top_provider.get("max_completion_tokens")),
input_cache_read=_as_float(pricing.get("input_cache_read")) or 0.0,
input_cache_write=_as_float(pricing.get("input_cache_write")) or 0.0,
input_modalities=architecture.get("input_modalities") or ["text"],
output_modalities=architecture.get("output_modalities") or ["text"],
tokenizer=architecture.get("tokenizer") or "unknown",
instruct_type=architecture.get("instruct_type"),
is_moderated=top_provider.get("is_moderated"),
)
class FallbackPricingResolver:
"""Resolves models via litellm → OpenRouter for one discovery pass.
The OpenRouter catalog is fetched at most once and only when a model
actually misses litellm, so a provider full of litellm-known models never
touches the network. Instantiate one per ``fetch_models`` call.
"""
def __init__(self) -> None:
self._openrouter_feed: list[dict] | None = None
async def resolve(self, model_id: str) -> ResolvedPricing | None:
"""Resolve ``model_id``; ``None`` if no source knows it."""
resolved = _from_litellm(model_id)
if resolved is not None:
return resolved
if self._openrouter_feed is None:
# Lazy import so tests can patch the feed at its source.
from ..payment.models import async_fetch_openrouter_models
self._openrouter_feed = await async_fetch_openrouter_models()
return _from_openrouter(model_id, self._openrouter_feed)
tests/unit/test_upstream_generic.py
+524
View File
@@ -0,0 +1,524 @@
"""Unit tests for ``GenericUpstreamProvider.fetch_models`` price/metadata resolution.
A generic upstream is any OpenAI-compatible API. Most (DeepSeek, OpenAI,
Groq, ...) answer ``/models`` with bare ``{id, object, owned_by}`` entries that
carry *no* pricing. The provider must not fabricate a price for those: it
resolves through native ``model_spec`` (Venice's bespoke schema) → litellm's
bundled cost map → the OpenRouter feed, and only when every source misses does
it import the model **disabled** with a warning rather than invent a number.
These tests drive that behaviour through the public ``fetch_models`` API. The
``/models`` HTTP call is faked at ``httpx.AsyncClient``; the OpenRouter feed is
patched at its source (``routstr.payment.models.async_fetch_openrouter_models``)
so the resolver's lazy import picks up the stub. litellm's real bundled cost map
is used unmocked — the DeepSeek rates it ships are the assertion's ground truth.
"""
from __future__ import annotations
import logging
from typing import Any
from unittest.mock import AsyncMock, patch
import pytest
from routstr.upstream.generic import GenericUpstreamProvider
class _FakeResponse:
def __init__(self, payload: dict[str, Any]) -> None:
self._payload = payload
def raise_for_status(self) -> None:
return None
def json(self) -> dict[str, Any]:
return self._payload
class _FakeAsyncClient:
"""Stand-in for ``httpx.AsyncClient`` returning a canned ``/models`` body."""
def __init__(self, payload: dict[str, Any]) -> None:
self._payload = payload
async def __aenter__(self) -> "_FakeAsyncClient":
return self
async def __aexit__(self, *exc: object) -> bool:
return False
async def get(self, url: str, headers: dict[str, str] | None = None) -> _FakeResponse:
return _FakeResponse(self._payload)
def _patch_models_endpoint(payload: dict[str, Any]) -> Any:
"""Patch the provider's ``httpx.AsyncClient`` to serve ``payload``."""
return patch(
"routstr.upstream.generic.httpx.AsyncClient",
lambda *args, **kwargs: _FakeAsyncClient(payload),
)
def _model_by_id(models: list[Any], model_id: str) -> Any:
return next(m for m in models if m.id == model_id)
# ---------------------------------------------------------------------------
# native model_spec (Venice) — must keep resolving, and capture its metadata
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_native_model_spec_resolves_and_captures_metadata() -> None:
"""A Venice-shaped ``model_spec`` is authoritative: prices/context come
straight from it and vision capability becomes an image input modality."""
payload = {
"data": [
{
"id": "venice-llama",
"owned_by": "venice",
"model_spec": {
"name": "Venice Llama",
"availableContextTokens": 65536,
"pricing": {
"input": {"usd": 0.5},
"output": {"usd": 1.5},
},
"capabilities": {"supportsVision": True},
},
}
]
}
with _patch_models_endpoint(payload):
or_feed = AsyncMock(return_value=[])
with patch(
"routstr.payment.models.async_fetch_openrouter_models", or_feed
):
models = await GenericUpstreamProvider(base_url="http://x").fetch_models()
model = _model_by_id(models, "venice-llama")
assert model.enabled is True
assert model.pricing.prompt == pytest.approx(0.5 / 1_000_000)
assert model.pricing.completion == pytest.approx(1.5 / 1_000_000)
assert model.context_length == 65536
assert "image" in model.architecture.input_modalities
# Vision capability must be reflected in the combined modality string, not
# flattened to "text->text".
assert model.architecture.modality == "text+image->text"
# A native price never needs the OpenRouter feed.
or_feed.assert_not_awaited()
# ---------------------------------------------------------------------------
# native model_spec validation — a bogus native price is not authoritative
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_native_both_zero_price_falls_through_to_litellm() -> None:
"""A native ``model_spec`` that prices both tokens at 0 is not a real price
(the same free-tier trap the litellm/OpenRouter rungs already reject). It
must not be treated as authoritative and served free; the resolver falls
through, so a litellm-known model lands on litellm's real rate instead."""
payload = {
"data": [
{
"id": "deepseek-chat",
"owned_by": "deepseek",
"model_spec": {
"pricing": {"input": {"usd": 0}, "output": {"usd": 0}},
},
}
]
}
with _patch_models_endpoint(payload):
or_feed = AsyncMock(return_value=[])
with patch("routstr.payment.models.async_fetch_openrouter_models", or_feed):
models = await GenericUpstreamProvider(base_url="http://x").fetch_models()
model = _model_by_id(models, "deepseek-chat")
assert model.enabled is True
assert model.pricing.prompt == pytest.approx(2.8e-07)
assert model.pricing.completion == pytest.approx(4.2e-07)
@pytest.mark.asyncio
async def test_native_negative_price_falls_through_to_litellm() -> None:
"""A negative native price would credit the caller's balance on every
request (a fund drain, not a discount). Reject it like any other unusable
price and fall through to the chain."""
payload = {
"data": [
{
"id": "deepseek-chat",
"owned_by": "deepseek",
"model_spec": {
"pricing": {"input": {"usd": -0.5}, "output": {"usd": -1.5}},
},
}
]
}
with _patch_models_endpoint(payload):
or_feed = AsyncMock(return_value=[])
with patch("routstr.payment.models.async_fetch_openrouter_models", or_feed):
models = await GenericUpstreamProvider(base_url="http://x").fetch_models()
model = _model_by_id(models, "deepseek-chat")
assert model.enabled is True
assert model.pricing.prompt == pytest.approx(2.8e-07)
assert model.pricing.completion == pytest.approx(4.2e-07)
@pytest.mark.asyncio
async def test_native_non_numeric_price_does_not_break_catalog() -> None:
"""A non-numeric native price (``"free"``) must not raise while parsing —
an unguarded ``"free" / 1_000_000`` throws and the outer catch drops the
*entire* provider catalog. It has to fail closed for that one model while
every other model in the same response still resolves."""
payload = {
"data": [
{
"id": "broken-price",
"owned_by": "mystery",
"model_spec": {
"pricing": {"input": {"usd": "free"}, "output": {"usd": "free"}},
},
},
{"id": "deepseek-chat", "object": "model", "owned_by": "deepseek"},
]
}
with _patch_models_endpoint(payload):
or_feed = AsyncMock(return_value=[])
with patch("routstr.payment.models.async_fetch_openrouter_models", or_feed):
models = await GenericUpstreamProvider(base_url="http://x").fetch_models()
# One malformed entry must not empty the catalog.
assert {m.id for m in models} == {"broken-price", "deepseek-chat"}
broken = _model_by_id(models, "broken-price")
assert broken.enabled is False
healthy = _model_by_id(models, "deepseek-chat")
assert healthy.enabled is True
assert healthy.pricing.prompt == pytest.approx(2.8e-07)
# ---------------------------------------------------------------------------
# litellm rescue — the money-critical case (DeepSeek bare /models)
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_bare_deepseek_resolves_via_litellm() -> None:
"""DeepSeek's ``/models`` carries no pricing. The old code fabricated
``$0.001`` + ctx 4096; the resolver must instead pull DeepSeek's real
rates from litellm's bundled cost map (``$0.28``/``$0.42`` per 1M, ctx
131072) and keep the model enabled."""
payload = {
"data": [
{"id": "deepseek-chat", "object": "model", "owned_by": "deepseek"},
]
}
with _patch_models_endpoint(payload):
or_feed = AsyncMock(return_value=[])
with patch(
"routstr.payment.models.async_fetch_openrouter_models", or_feed
):
models = await GenericUpstreamProvider(base_url="http://x").fetch_models()
model = _model_by_id(models, "deepseek-chat")
assert model.enabled is True
assert model.pricing.prompt == pytest.approx(2.8e-07)
assert model.pricing.completion == pytest.approx(4.2e-07)
assert model.context_length == 131072
# Richer metadata than the two base prices is captured too.
assert model.pricing.input_cache_read == pytest.approx(2.8e-08)
assert model.top_provider is not None
assert model.top_provider.max_completion_tokens == 8192
# litellm answered, so the OpenRouter feed is never consulted.
or_feed.assert_not_awaited()
@pytest.mark.asyncio
async def test_litellm_zero_price_entry_fails_closed(
caplog: pytest.LogCaptureFixture,
) -> None:
"""A litellm entry that lists a model but prices it at 0/0 (free-tier
moderation/rerank models do this) is not a real price — treating it as one
would silently serve the model for free. The resolver must reject a both-zero
litellm hit and fall through, so the model imports disabled, not at $0."""
payload = {
"data": [
{"id": "omni-moderation-latest", "object": "model", "owned_by": "openai"},
]
}
gen_logger = logging.getLogger("routstr.upstream.generic")
gen_logger.addHandler(caplog.handler)
try:
with _patch_models_endpoint(payload):
or_feed = AsyncMock(return_value=[])
with patch(
"routstr.payment.models.async_fetch_openrouter_models", or_feed
):
models = await GenericUpstreamProvider(
base_url="http://x"
).fetch_models()
finally:
gen_logger.removeHandler(caplog.handler)
model = _model_by_id(models, "omni-moderation-latest")
assert model.enabled is False
assert model.pricing.prompt == 0.0
assert model.pricing.completion == 0.0
assert any(
"omni-moderation-latest" in rec.getMessage()
for rec in caplog.records
if rec.levelno >= logging.WARNING
)
@pytest.mark.asyncio
async def test_litellm_output_cap_not_used_as_context() -> None:
"""litellm's ``max_tokens`` is the completion cap, not the context window
(it tracks ``max_output_tokens`` for ~94% of models). When a model reports
no ``max_input_tokens``, the resolver must not smuggle the output cap in as
the context window; it falls back to the id-based estimate instead, while
``max_tokens`` still feeds the completion limit."""
payload = {
"data": [
{
"id": "gemini/gemini-gemma-2-9b-it",
"object": "model",
"owned_by": "google",
},
]
}
with _patch_models_endpoint(payload):
or_feed = AsyncMock(return_value=[])
with patch(
"routstr.payment.models.async_fetch_openrouter_models", or_feed
):
models = await GenericUpstreamProvider(base_url="http://x").fetch_models()
model = _model_by_id(models, "gemini/gemini-gemma-2-9b-it")
assert model.enabled is True
# litellm gives this model max_input_tokens=None, max_tokens=8192 (an output
# cap). Context must come from the estimate (4096), never the 8192 cap.
assert model.context_length == 4096
# The 8192 output cap still lands where it belongs: the completion limit.
assert model.top_provider is not None
assert model.top_provider.max_completion_tokens == 8192
# ---------------------------------------------------------------------------
# OpenRouter fallback — litellm misses, OR carries a full payload
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_unknown_to_litellm_resolves_via_openrouter() -> None:
"""A model litellm has never heard of still resolves if the OpenRouter
feed lists it, pulling price + context from that entry."""
payload = {
"data": [
{"id": "exotic/model-9000", "object": "model", "owned_by": "exotic"},
]
}
or_entry = {
"id": "exotic/model-9000",
"name": "Exotic 9000",
"context_length": 65536,
"architecture": {
"modality": "text+image->text",
"input_modalities": ["text", "image"],
"output_modalities": ["text"],
"tokenizer": "Other",
"instruct_type": None,
},
"pricing": {"prompt": "0.000005", "completion": "0.000010"},
"top_provider": {
"context_length": 65536,
"max_completion_tokens": 4096,
"is_moderated": False,
},
}
with _patch_models_endpoint(payload):
or_feed = AsyncMock(return_value=[or_entry])
with patch(
"routstr.payment.models.async_fetch_openrouter_models", or_feed
):
models = await GenericUpstreamProvider(base_url="http://x").fetch_models()
model = _model_by_id(models, "exotic/model-9000")
assert model.enabled is True
assert model.pricing.prompt == pytest.approx(5e-06)
assert model.pricing.completion == pytest.approx(1e-05)
assert model.context_length == 65536
# The feed's own modality string is carried through verbatim, not recomputed.
assert model.architecture.modality == "text+image->text"
or_feed.assert_awaited()
@pytest.mark.asyncio
async def test_openrouter_bare_tail_collision_picks_highest_price() -> None:
"""When a bare model id matches several OpenRouter entries by tail
(``model`` ↔ ``a/model``, ``b/model``), the match must be deterministic and
money-safe: pick the highest-priced candidate regardless of feed order, so
ordering can never leave the node charging below true cost. (The live feed
has zero such collisions today; this guards the latent case.)"""
payload = {
"data": [
{"id": "zzz-phantom-model", "object": "model", "owned_by": "mystery"},
]
}
# Same bare tail, different resellers; the pricier one is listed *second*
# so a first-wins match would pick the cheaper (undercharging) entry.
or_feed = AsyncMock(
return_value=[
{
"id": "cheapco/zzz-phantom-model",
"context_length": 8192,
"pricing": {"prompt": "0.000001", "completion": "0.000002"},
},
{
"id": "premiumco/zzz-phantom-model",
"context_length": 8192,
"pricing": {"prompt": "0.000009", "completion": "0.000010"},
},
]
)
with _patch_models_endpoint(payload):
with patch("routstr.payment.models.async_fetch_openrouter_models", or_feed):
models = await GenericUpstreamProvider(base_url="http://x").fetch_models()
model = _model_by_id(models, "zzz-phantom-model")
assert model.pricing.prompt == pytest.approx(9e-06)
assert model.pricing.completion == pytest.approx(1e-05)
@pytest.mark.asyncio
async def test_openrouter_bare_tail_tie_breaks_on_combined_cost() -> None:
"""The bare-tail tie-break must weigh *both* rates, not prompt alone.
Given two colliding entries where one is cheaper on prompt but far dearer
on completion, ranking by prompt would pick the entry that undercharges
output-heavy traffic. Pick the highest *combined* per-token cost so the
money-safe choice holds whichever way the traffic leans."""
payload = {
"data": [
{"id": "yyy-phantom-model", "object": "model", "owned_by": "mystery"},
]
}
# dear-overall is listed first with the *lower* prompt, so a prompt-only max
# would wrongly pick the second (cheaper-overall) entry.
or_feed = AsyncMock(
return_value=[
{
"id": "dearco/yyy-phantom-model",
"context_length": 8192,
"pricing": {"prompt": "0.000001", "completion": "0.000100"},
},
{
"id": "cheapco/yyy-phantom-model",
"context_length": 8192,
"pricing": {"prompt": "0.000009", "completion": "0.000002"},
},
]
)
with _patch_models_endpoint(payload):
with patch("routstr.payment.models.async_fetch_openrouter_models", or_feed):
models = await GenericUpstreamProvider(base_url="http://x").fetch_models()
model = _model_by_id(models, "yyy-phantom-model")
assert model.pricing.prompt == pytest.approx(1e-06)
assert model.pricing.completion == pytest.approx(1e-04)
@pytest.mark.asyncio
async def test_openrouter_feed_fetched_once_per_discovery() -> None:
"""Two models both missing litellm must share a single OpenRouter fetch —
the feed is not re-downloaded per model."""
payload = {
"data": [
{"id": "exotic/model-a", "object": "model", "owned_by": "exotic"},
{"id": "exotic/model-b", "object": "model", "owned_by": "exotic"},
]
}
or_feed = AsyncMock(
return_value=[
{
"id": "exotic/model-a",
"context_length": 8192,
"pricing": {"prompt": "0.000001", "completion": "0.000002"},
},
{
"id": "exotic/model-b",
"context_length": 8192,
"pricing": {"prompt": "0.000003", "completion": "0.000004"},
},
]
)
with _patch_models_endpoint(payload):
with patch("routstr.payment.models.async_fetch_openrouter_models", or_feed):
models = await GenericUpstreamProvider(base_url="http://x").fetch_models()
assert {m.id for m in models} == {"exotic/model-a", "exotic/model-b"}
assert or_feed.await_count == 1
# ---------------------------------------------------------------------------
# fail closed — no source resolves → disabled + warned, never fabricated
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_unresolvable_model_fails_closed(
caplog: pytest.LogCaptureFixture,
) -> None:
"""When native, litellm and OpenRouter all miss, the model is imported
disabled with a warning naming it — and no price is invented (the old
``$0.001`` placeholder is gone)."""
payload = {
"data": [
{
"id": "nobody-has-priced-this-xyz",
"object": "model",
"owned_by": "mystery",
},
]
}
# routstr loggers set propagate=False, so caplog's root handler misses
# them; attach its handler to the provider logger directly.
gen_logger = logging.getLogger("routstr.upstream.generic")
gen_logger.addHandler(caplog.handler)
try:
with _patch_models_endpoint(payload):
or_feed = AsyncMock(return_value=[])
with patch(
"routstr.payment.models.async_fetch_openrouter_models", or_feed
):
models = await GenericUpstreamProvider(
base_url="http://x"
).fetch_models()
finally:
gen_logger.removeHandler(caplog.handler)
model = _model_by_id(models, "nobody-has-priced-this-xyz")
assert model.enabled is False
assert model.pricing.prompt == 0.0
assert model.pricing.completion == 0.0
assert any(
"nobody-has-priced-this-xyz" in rec.getMessage()
for rec in caplog.records
if rec.levelno >= logging.WARNING
)