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 #801 from Routstr/fix/pinned-path-pricing

Browse Source 
fix: reserve pinned endpoints at the endpoint's own price and limits
This commit is contained in:
9qeklajc
2026-10-03 19:04:08 +02:00
committed by GitHub
parent 95c4fbb9e0 799d1ba411
commit df89c0a5b6
8 changed files with 244 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files
routstr/core/admin.py
+3 -6
View File
@@ -1724,14 +1724,12 @@ async def certify_upstream_provider(
status_code=503,
detail="sats/USD price is not initialized yet; retry shortly",
)
# The proxy reserves and token-bills a pinned request with the model's
# own pricing, so the cost rows use it too; the path's advertised
# endpoint rates are only compared against it in the margin row.
advertised_model = None
# The proxy reserves and token-bills a pinned endpoint at that
# endpoint's own rates, so the cost rows price it the same way.
if selected_path is not None:
from ..upstream.model_paths import apply_model_path_pricing
advertised_model = apply_model_path_pricing(
model_obj = apply_model_path_pricing(
model_obj,
selected_path,
provider.provider_fee,
@@ -1750,7 +1748,6 @@ async def certify_upstream_provider(
check_cache=payload.check_cache,
endpoint_tag=endpoint_tag,
upstream=upstream_obj,
advertised_model=advertised_model,
)
rows = pricing_rows + live_rows
routstr/proxy.py
+46
View File
@@ -19,6 +19,7 @@ from .core import get_logger
from .core.db import (
ApiKey,
AsyncSession,
ModelPathRow,
ModelRow,
UpstreamProviderRow,
create_session,
@@ -52,6 +53,7 @@ from .upstream.ehbp import forward_ehbp_request, forward_ehbp_x_cashu_request
from .upstream.helpers import init_upstreams
from .upstream.model_paths import (
ModelPathSelector,
apply_model_path_pricing,
decode_model_path,
is_openrouter_base_url,
public_model_id,
@@ -157,6 +159,44 @@ def _candidate_for_selector(
return None
async def _price_pinned_endpoint(
session: AsyncSession,
selector: ModelPathSelector,
model_obj: Model,
upstream: BaseUpstreamProvider,
) -> Model:
"""Reprice ``model_obj`` with the pinned endpoint's own rates.
An OpenRouter endpoint can cost more than the model's default listing, and
``/v1/models/paths`` quotes that endpoint's price, so the reservation and
token billing must use it too. Without a stored path row or a sats price
the request keeps the model's default pricing.
"""
from .payment import price as price_module
sats_to_usd = price_module.SATS_USD_PRICE
if upstream.db_id is None or not sats_to_usd:
return model_obj
rows = (
await session.exec(
select(ModelPathRow).where(
ModelPathRow.upstream_provider_id == upstream.db_id,
ModelPathRow.endpoint_tag == selector.endpoint_tag,
)
)
).all()
row = next(
(r for r in rows if _model_ids_match(r.model_id, selector.model_id)), None
)
if row is None:
logger.warning(
"No stored path for pinned endpoint; billing the model's default pricing",
extra={"model": selector.model_id, "endpoint": selector.endpoint_tag},
)
return model_obj
return apply_model_path_pricing(model_obj, row, upstream.provider_fee, sats_to_usd)
def get_model_instance(model_id: str) -> Model | None:
"""Get the best-ranked Model instance for a model ID."""
candidates = get_candidates(model_id)
@@ -702,6 +742,12 @@ async def _proxy(
},
}
request_body = json.dumps(request_body_dict).encode()
candidates = [
(
await _price_pinned_endpoint(session, selector, *pinned),
pinned[1],
)
]
if is_ehbp:
candidates = [
routstr/upstream/certification.py
+2 -4
View File
@@ -861,15 +861,14 @@ async def run_live_checks(
check_cache: bool = True,
endpoint_tag: str | None = None,
upstream: "BaseUpstreamProvider | None" = None,
advertised_model: "Model | None" = None,
) -> list[dict[str, Any]]:
"""Probe one upstream and build the live/derived rows.
``check_cache`` adds the prompt-cache and margin rows, which cost two or
three more completions against a long prompt. ``upstream`` shapes the
probes like the proxy's own requests; without it they assume a plain
OpenAI-compatible base URL. ``advertised_model`` carries a pinned path's
own endpoint rates for the margin row to compare against ``model``'s.
OpenAI-compatible base URL. On a pinned endpoint ``model`` carries that
endpoint's own rates, as the proxy bills it.
"""
probe = await probe_upstream(
base_url,
@@ -973,7 +972,6 @@ async def run_live_checks(
pricing_known=pricing_known,
endpoint_tag=endpoint_tag,
upstream=upstream,
advertised_model=advertised_model,
token_limit_field=probe.token_limit_field,
)
)
routstr/upstream/certification_cache.py
+2 -32
View File
@@ -446,7 +446,6 @@ def cost_margin_row(
provider_fee: float,
sats_to_usd: float,
pricing_known: bool = True,
advertised_model: Model | None = None,
) -> dict[str, Any]:
"""Configured token pricing must cover what the upstream reports charging.
@@ -457,14 +456,9 @@ def cost_margin_row(
falls below the fee-adjusted upstream cost means those paths underprice.
Upstreams that report no cost give no sample and the row stays a warn.
``model`` carries the pricing the proxy reserves and token-bills with. On
a pinned path, ``advertised_model`` carries the endpoint's own rates; a
covered margin whose advertised rates differ from the billed ones is a
warn, since ``/v1/models/paths`` then shows a price the node does not bill.
``model`` carries the pricing the proxy reserves and token-bills with,
which on a pinned endpoint is that endpoint's own rates.
"""
advertised_pricing = (
advertised_model.sats_pricing if advertised_model is not None else None
)
evidence: dict[str, Any] = {
"model_id": model.id,
"provider_fee": provider_fee,
@@ -483,7 +477,6 @@ def cost_margin_row(
samples: list[dict[str, Any]] = []
short: list[str] = []
mismatched: list[str] = []
for payload in payloads:
if not isinstance(payload, dict):
continue
@@ -496,11 +489,6 @@ def cost_margin_row(
upstream_total = _expected_usd_msats(
reported_usd, provider_fee, sats_to_usd
)
advertised_total = (
_expected_token_msats(advertised_pricing, usage)[0]
if advertised_pricing is not None
else None
)
except (ValueError, OverflowError) as exc:
evidence["error"] = f"{type(exc).__name__}: {exc}"
return certification_row(
@@ -516,10 +504,6 @@ def cost_margin_row(
"upstream_msats_with_fee": upstream_total,
"configured_msats": configured_total,
}
if advertised_total is not None:
sample["advertised_msats"] = advertised_total
if abs(advertised_total - configured_total) > COST_TOLERANCE_MSATS:
mismatched.append(f"{advertised_total} vs {configured_total}")
samples.append(sample)
if configured_total + COST_TOLERANCE_MSATS < upstream_total:
short.append(f"{configured_total} < {upstream_total}")
@@ -545,18 +529,6 @@ def cost_margin_row(
"requests lose money.",
evidence,
)
if mismatched:
return certification_row(
ROW_MARGIN,
STATUS_WARN,
TITLE_MARGIN,
f"Configured pricing covers the upstream's reported cost on "
f"{len(samples)} sampled completion(s), but this path advertises "
f"different endpoint rates (advertised vs billed msats: "
f"{'; '.join(mismatched)}); the proxy reserves and token-bills "
"pinned requests with the model's own pricing.",
evidence,
)
return certification_row(
ROW_MARGIN,
STATUS_OK,
@@ -597,7 +569,6 @@ async def run_cache_checks(
pricing_known: bool = True,
endpoint_tag: str | None = None,
upstream: "BaseUpstreamProvider | None" = None,
advertised_model: Model | None = None,
token_limit_field: str = "max_tokens",
) -> list[dict[str, Any]]:
"""Run the cache probe and build the three cache/margin rows."""
@@ -634,7 +605,6 @@ async def run_cache_checks(
provider_fee=provider_fee,
sats_to_usd=sats_to_usd,
pricing_known=pricing_known,
advertised_model=advertised_model,
),
),
]
routstr/upstream/model_paths.py
+16 -4
View File
@@ -899,14 +899,17 @@ def apply_model_path_pricing(
Direct paths already use the provider model cache and therefore carry the
same pricing as ``model``. OpenRouter endpoint rows instead contain raw,
endpoint-specific USD rates; certification compares them against the
model's own pricing, which the proxy reserves and token-bills with.
endpoint-specific USD rates and limits, which can differ from the model's
default listing; OpenRouter charges the endpoint that serves the request,
so the proxy reserves and token-bills a pinned endpoint with them, using
the same limits ``/v1/models/paths`` quotes its max cost from.
"""
if row.endpoint_tag is None:
return model
from ..payment.models import (
Pricing,
TopProvider,
_calculate_usd_max_costs,
_update_model_sats_pricing,
backfill_cache_pricing,
@@ -925,7 +928,16 @@ def apply_model_path_pricing(
pricing = Pricing.parse_obj(
{key: float(value) * provider_fee for key, value in pricing.dict().items()}
)
priced = model.copy(update={"pricing": pricing, "sats_pricing": None})
update: dict[str, Any] = {"pricing": pricing, "sats_pricing": None}
context_length = metadata.get("context_length")
max_completion_tokens = metadata.get("max_completion_tokens")
if context_length or max_completion_tokens:
update["context_length"] = context_length or model.context_length
update["top_provider"] = TopProvider(
context_length=context_length,
max_completion_tokens=max_completion_tokens,
)
priced = model.copy(update=update)
(
pricing.max_prompt_cost,
pricing.max_completion_cost,
@@ -934,7 +946,7 @@ def apply_model_path_pricing(
return _update_model_sats_pricing(priced, sats_to_usd)
except Exception as exc:
logger.warning(
"Could not apply model-path pricing for certification",
"Could not apply model-path pricing",
extra={"model_id": model.id, "path": row.path, "error": str(exc)},
)
return model
tests/integration/test_certify_endpoint.py
+6 -13
View File
@@ -328,7 +328,7 @@ async def test_certify_model_path_pins_every_completion(
@pytest.mark.integration
@pytest.mark.asyncio
@respx.mock
async def test_certify_margin_bills_model_pricing_and_reports_path_pricing(
async def test_certify_margin_bills_pinned_path_pricing(
integration_client: AsyncClient, integration_session: AsyncSession
) -> None:
base_url = "https://openrouter.ai/api/v1"
@@ -409,20 +409,13 @@ async def test_certify_margin_bills_model_pricing_and_reports_path_pricing(
assert resp.status_code == 200, resp.text
margin = _find_row(resp.json()["rows"], "cost.margin")
# The proxy reserves and token-bills a pinned request with the model's own
# pricing (``configured_msats``); the path's endpoint rates are reported
# alongside (``advertised_msats``) and differ, so the covered margin warns.
# The proxy token-bills a pinned endpoint at its own rates, not the
# model's (3, 356, 43). Those rates miss the cached call's reported cost.
assert [
(
sample["upstream_msats_with_fee"],
sample["configured_msats"],
sample["advertised_msats"],
)
(sample["upstream_msats_with_fee"], sample["configured_msats"])
for sample in margin["evidence"]["samples"]
] == [(3, 3, 3), (269, 356, 289), (26, 43, 15)]
assert margin["status"] == "warn"
assert "advertises different endpoint rates" in margin["detail"]
assert "289 vs 356" in margin["detail"]
] == [(3, 3), (269, 289), (26, 15)]
assert margin["status"] == "fail"
@pytest.mark.integration
tests/unit/test_certification_cache.py
-41
View File
@@ -333,47 +333,6 @@ class TestCostMarginRow:
assert row["status"] == STATUS_OK
def test_pinned_path_fails_when_billed_pricing_misses_cost(self) -> None:
"""The path's endpoint rates cover the cost but the model pricing the
proxy actually bills with does not: the margin must fail."""
payload = _payload({"prompt_tokens": 5, "completion_tokens": 1, "cost": 4e-6})
row = cost_margin_row(
model=_model(),
payloads=[payload],
provider_fee=1.0,
sats_to_usd=SATS_USD,
advertised_model=_model(prompt=1e-6, completion=2e-6),
)
assert row["status"] == STATUS_FAIL
sample = row["evidence"]["samples"][0]
assert sample["advertised_msats"] >= sample["upstream_msats_with_fee"]
assert sample["configured_msats"] < sample["upstream_msats_with_fee"]
def test_pinned_path_warns_when_advertised_rates_differ(self) -> None:
payload = _payload({"prompt_tokens": 5, "completion_tokens": 1, "cost": 9e-7})
row = cost_margin_row(
model=_model(),
payloads=[payload],
provider_fee=1.0,
sats_to_usd=SATS_USD,
advertised_model=_model(prompt=1e-6, completion=2e-6),
)
assert row["status"] == STATUS_WARN
assert "advertises different endpoint rates" in row["detail"]
def test_pinned_path_ok_when_advertised_rates_match(self) -> None:
payload = _payload({"prompt_tokens": 5, "completion_tokens": 1, "cost": 9e-7})
row = cost_margin_row(
model=_model(),
payloads=[payload],
provider_fee=1.0,
sats_to_usd=SATS_USD,
advertised_model=_model(),
)
assert row["status"] == STATUS_OK
sample = row["evidence"]["samples"][0]
assert sample["advertised_msats"] == sample["configured_msats"]
def test_warn_when_pricing_unknown(self) -> None:
payload = _payload({"prompt_tokens": 5, "completion_tokens": 1, "cost": 9e-7})
row = cost_margin_row(
tests/unit/test_model_path_routing.py
+169 -1
View File
@@ -50,6 +50,7 @@ async def _run_proxy(
request: MagicMock,
candidates: list[tuple[Any, Any]],
path: str = "v1/chat/completions",
session: Any = None,
) -> Any:
key = ApiKey(hashed_key="mpkey", balance=10_000)
reservation = ReservationSnapshot(
@@ -74,7 +75,7 @@ async def _run_proxy(
proxy_module, "pay_for_request", AsyncMock(return_value=reservation)
),
patch.object(proxy_module, "revert_pay_for_request", AsyncMock()),
patch_proxy_session(MagicMock()),
patch_proxy_session(session if session is not None else MagicMock()),
):
return await proxy_module.proxy(request, path)
@@ -845,3 +846,170 @@ async def test_node_fault_stays_500_without_scope_header() -> None:
assert ERROR_SCOPE_HEADER not in response.headers
body = json.loads(bytes(response.body))
assert body["error"]["code"] != UPSTREAM_UNAVAILABLE
_OPENROUTER = "https://openrouter.ai/api/v1"
_SATS_USD = 0.001
_ENDPOINT_PRICING = {"prompt": 2e-6, "completion": 4e-6}
def _priced_model(
prompt: float = 1e-6, completion: float = 2e-6, cache_read: float = 0.0
) -> Any:
from routstr.payment.models import (
Architecture,
Model,
Pricing,
_calculate_usd_max_costs,
_update_model_sats_pricing,
)
model = Model(
id=MODEL_ID,
name=MODEL_ID,
created=0,
description="",
context_length=8192,
architecture=Architecture(
modality="text",
input_modalities=["text"],
output_modalities=["text"],
tokenizer="unknown",
instruct_type=None,
),
pricing=Pricing(
prompt=prompt, completion=completion, input_cache_read=cache_read
),
)
(
model.pricing.max_prompt_cost,
model.pricing.max_completion_cost,
model.pricing.max_cost,
) = _calculate_usd_max_costs(model)
return _update_model_sats_pricing(model, _SATS_USD)
def _endpoint_row(
model_id: str = MODEL_ID,
endpoint_tag: str = "deepinfra/fp8",
pricing: dict[str, float] | None = None,
**limits: int,
) -> Any:
from routstr.core.db import ModelPathRow
return ModelPathRow(
model_id=model_id,
path=encode_model_path(_OPENROUTER, model_id, endpoint_tag),
provider_slug="openrouter",
provider_type="openrouter",
endpoint_tag=endpoint_tag,
model_metadata=json.dumps(
{"id": model_id, "pricing": pricing or _ENDPOINT_PRICING, **limits}
),
upstream_provider_id=1,
)
def _session_with_rows(rows: list[Any]) -> MagicMock:
session = MagicMock()
session.exec = AsyncMock(return_value=MagicMock(all=MagicMock(return_value=rows)))
return session
def _endpoint_selector() -> Any:
selector = decode_model_path(
encode_model_path(_OPENROUTER, MODEL_ID, "deepinfra/fp8")
)
assert selector is not None
return selector
def _openrouter_upstream() -> MagicMock:
upstream = _make_upstream(1)
upstream.base_url = _OPENROUTER
upstream.provider_fee = 1.0
return upstream
@pytest.mark.asyncio
async def test_endpoint_pin_bills_the_endpoint_pricing() -> None:
"""A pinned endpoint is reserved and billed at the rates
``/v1/models/paths`` quotes for it, not the model's default listing."""
model = _priced_model()
upstream = _openrouter_upstream()
request = _make_request(
{
"authorization": "Bearer sk-mpkey",
"x-routstr-model-path": encode_model_path(
_OPENROUTER, MODEL_ID, "deepinfra/fp8"
),
},
json.dumps({"model": MODEL_ID}).encode(),
)
with patch("routstr.payment.price.SATS_USD_PRICE", _SATS_USD):
await _run_proxy(
request, [(model, upstream)], session=_session_with_rows([_endpoint_row()])
)
billed = upstream.forward_request.await_args.args[7]
assert billed.sats_pricing.prompt == pytest.approx(2e-6 / _SATS_USD)
assert billed.sats_pricing.completion == pytest.approx(4e-6 / _SATS_USD)
assert billed.sats_pricing.max_cost > model.sats_pricing.max_cost
assert model.sats_pricing.prompt == pytest.approx(1e-6 / _SATS_USD)
@pytest.mark.asyncio
@pytest.mark.parametrize(
"rows",
[[], [_endpoint_row(model_id="other-model")]],
ids=["no-stored-path", "other-model"],
)
async def test_endpoint_pin_without_a_stored_path_keeps_model_pricing(
rows: list[Any],
) -> None:
model = _priced_model()
with patch("routstr.payment.price.SATS_USD_PRICE", _SATS_USD):
priced = await proxy_module._price_pinned_endpoint(
_session_with_rows(rows),
_endpoint_selector(),
model,
_openrouter_upstream(),
)
assert priced is model
@pytest.mark.asyncio
async def test_endpoint_pin_without_sats_price_keeps_model_pricing() -> None:
model = _priced_model()
session = _session_with_rows([_endpoint_row()])
with patch("routstr.payment.price.SATS_USD_PRICE", None):
priced = await proxy_module._price_pinned_endpoint(
session,
_endpoint_selector(),
model,
_openrouter_upstream(),
)
assert priced is model
session.exec.assert_not_awaited()
@pytest.mark.asyncio
async def test_endpoint_pin_reserves_the_max_cost_paths_quotes() -> None:
"""The reservation uses the endpoint's own context and completion limits,
the same ones ``/v1/models/paths`` quotes its max cost from."""
from routstr.upstream.model_paths import _serialize_path
row = _endpoint_row(context_length=32768, max_completion_tokens=8192)
with patch("routstr.payment.price.SATS_USD_PRICE", _SATS_USD):
priced = await proxy_module._price_pinned_endpoint(
_session_with_rows([row]),
_endpoint_selector(),
_priced_model(),
_openrouter_upstream(),
)
quoted = _serialize_path(row, 1.0)["model"]["sats_pricing"]["max_cost"]
assert priced.sats_pricing is not None and priced.top_provider is not None
assert priced.sats_pricing.max_cost == pytest.approx(quoted)
assert priced.top_provider.max_completion_tokens == 8192