mirror of
https://github.com/Routstr/routstr-core.git
synced 2026-07-30 23:36:15 +00:00
Compare commits
5
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
14785e4cde | ||
|
|
1dceffffa7 | ||
|
|
4b74a9cf81 | ||
|
|
3be2da6728 | ||
|
|
17d690e77e |
@@ -364,7 +364,6 @@ async def validate_bearer_key(
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
await session.rollback()
|
||||
logger.error(
|
||||
"Cashu token redemption failed",
|
||||
extra={
|
||||
@@ -1305,35 +1304,6 @@ async def periodic_key_reset() -> None:
|
||||
logger.error(f"Error in periodic_key_reset: {e}")
|
||||
|
||||
|
||||
async def periodic_dead_key_prune() -> None:
|
||||
"""Periodically prune dead API keys. Interval <= 0 disables it.
|
||||
|
||||
See ``prune_dead_api_keys`` for eligibility.
|
||||
"""
|
||||
from .core.db import create_session, prune_dead_api_keys
|
||||
|
||||
interval = settings.dead_key_prune_interval_seconds
|
||||
if interval <= 0:
|
||||
logger.info("Dead-key pruning disabled (interval <= 0)")
|
||||
return
|
||||
|
||||
while True:
|
||||
try:
|
||||
await asyncio.sleep(interval)
|
||||
except asyncio.CancelledError:
|
||||
break
|
||||
|
||||
try:
|
||||
async with create_session() as session:
|
||||
await prune_dead_api_keys(
|
||||
session, settings.dead_key_min_age_seconds
|
||||
)
|
||||
except asyncio.CancelledError:
|
||||
break
|
||||
except Exception as e:
|
||||
logger.error(f"Error in periodic_dead_key_prune: {e}")
|
||||
|
||||
|
||||
STALE_RESERVATION_SWEEP_INTERVAL_SECONDS: int = 60
|
||||
|
||||
|
||||
|
||||
+1
-59
@@ -9,10 +9,9 @@ from typing import AsyncGenerator
|
||||
from alembic import command
|
||||
from alembic.config import Config
|
||||
from alembic.util.exc import CommandError
|
||||
from sqlalchemy import UniqueConstraint, delete
|
||||
from sqlalchemy import UniqueConstraint
|
||||
from sqlalchemy.exc import OperationalError
|
||||
from sqlalchemy.ext.asyncio.engine import create_async_engine
|
||||
from sqlalchemy.orm import aliased
|
||||
from sqlmodel import Field, Relationship, SQLModel, col, func, select, update
|
||||
from sqlmodel.ext.asyncio.session import AsyncSession
|
||||
|
||||
@@ -126,63 +125,6 @@ async def release_stale_reservations(
|
||||
return released
|
||||
|
||||
|
||||
async def prune_dead_api_keys(session: AsyncSession, min_age_seconds: int) -> int:
|
||||
"""Delete dead parentless API keys; return the count removed.
|
||||
|
||||
Dead = 0 balance/reservation/spend/requests, older than the grace period,
|
||||
no parent, no children, no pending invoice. Cashu rows are unlinked (not
|
||||
deleted) first to keep the audit trail.
|
||||
"""
|
||||
cutoff = int(time.time()) - min_age_seconds
|
||||
|
||||
child = aliased(ApiKey)
|
||||
has_children = (
|
||||
select(child.hashed_key).where(
|
||||
col(child.parent_key_hash) == col(ApiKey.hashed_key)
|
||||
)
|
||||
).exists()
|
||||
pending_invoice = (
|
||||
select(LightningInvoice.id)
|
||||
.where(col(LightningInvoice.api_key_hash) == col(ApiKey.hashed_key))
|
||||
.where(col(LightningInvoice.status) == "pending")
|
||||
).exists()
|
||||
|
||||
eligible_hashes = (
|
||||
select(ApiKey.hashed_key)
|
||||
.where(col(ApiKey.balance) == 0)
|
||||
.where(col(ApiKey.reserved_balance) == 0)
|
||||
.where(col(ApiKey.total_spent) == 0)
|
||||
.where(col(ApiKey.total_requests) == 0)
|
||||
.where(col(ApiKey.parent_key_hash).is_(None))
|
||||
.where(
|
||||
(col(ApiKey.created_at).is_(None)) | (col(ApiKey.created_at) < cutoff)
|
||||
)
|
||||
.where(~pending_invoice)
|
||||
.where(~has_children)
|
||||
)
|
||||
|
||||
# Unlink transactions rather than cascade-deleting them, so the financial
|
||||
# audit trail survives. The eligibility predicate is re-evaluated inside both
|
||||
# statements so a key that gained balance mid-run is left untouched.
|
||||
await session.exec( # type: ignore[call-overload]
|
||||
update(CashuTransaction)
|
||||
.where(col(CashuTransaction.api_key_hashed_key).in_(eligible_hashes))
|
||||
.values(api_key_hashed_key=None)
|
||||
)
|
||||
|
||||
result = await session.exec( # type: ignore[call-overload]
|
||||
delete(ApiKey).where(col(ApiKey.hashed_key).in_(eligible_hashes))
|
||||
)
|
||||
await session.commit()
|
||||
|
||||
pruned = int(result.rowcount or 0)
|
||||
logger.info(
|
||||
"Pruned dead API keys",
|
||||
extra={"pruned_keys": pruned, "min_age_seconds": min_age_seconds},
|
||||
)
|
||||
return pruned
|
||||
|
||||
|
||||
class ModelRow(SQLModel, table=True): # type: ignore
|
||||
__tablename__ = "models"
|
||||
id: str = Field(primary_key=True)
|
||||
|
||||
+1
-17
@@ -11,11 +11,7 @@ from starlette.exceptions import HTTPException
|
||||
from starlette.responses import Response as StarletteResponse
|
||||
from starlette.types import Scope
|
||||
|
||||
from ..auth import (
|
||||
periodic_dead_key_prune,
|
||||
periodic_key_reset,
|
||||
periodic_stale_reservation_sweep,
|
||||
)
|
||||
from ..auth import periodic_key_reset, periodic_stale_reservation_sweep
|
||||
from ..balance import balance_router, deprecated_wallet_router
|
||||
from ..lightning import lightning_router, periodic_invoice_watcher
|
||||
from ..nostr import (
|
||||
@@ -28,7 +24,6 @@ from ..payment.models import models_router, update_sats_pricing
|
||||
from ..payment.price import update_prices_periodically
|
||||
from ..proxy import initialize_upstreams, proxy_router, refresh_model_maps_periodically
|
||||
from ..upstream.auto_topup import periodic_auto_topup
|
||||
from ..upstream.deepseek_v4_pricing_shim import register_deepseek_v4_pricing
|
||||
from ..upstream.litellm_routing import configure_litellm
|
||||
from ..wallet import periodic_payout, periodic_refund_sweep, periodic_routstr_fee_payout
|
||||
from .admin import admin_router
|
||||
@@ -60,7 +55,6 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
|
||||
model_maps_refresh_task = None
|
||||
key_reset_task = None
|
||||
stale_reservation_task = None
|
||||
dead_key_prune_task = None
|
||||
auto_topup_task = None
|
||||
refund_sweep_task = None
|
||||
routstr_fee_task = None
|
||||
@@ -71,11 +65,6 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
|
||||
# debug logging) before any upstream provider dispatches a request.
|
||||
configure_litellm()
|
||||
|
||||
# TEMPORARY: backfill DeepSeek V4 pricing missing from litellm's cost
|
||||
# map (BerriAI/litellm#30430). Remove this call and
|
||||
# deepseek_v4_pricing_shim.py once litellm ships these models.
|
||||
register_deepseek_v4_pricing()
|
||||
|
||||
# Run database migrations on startup
|
||||
run_migrations()
|
||||
|
||||
@@ -137,7 +126,6 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
|
||||
stale_reservation_task = asyncio.create_task(
|
||||
periodic_stale_reservation_sweep()
|
||||
)
|
||||
dead_key_prune_task = asyncio.create_task(periodic_dead_key_prune())
|
||||
auto_topup_task = asyncio.create_task(periodic_auto_topup())
|
||||
refund_sweep_task = asyncio.create_task(periodic_refund_sweep())
|
||||
routstr_fee_task = asyncio.create_task(periodic_routstr_fee_payout())
|
||||
@@ -177,8 +165,6 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
|
||||
key_reset_task.cancel()
|
||||
if stale_reservation_task is not None:
|
||||
stale_reservation_task.cancel()
|
||||
if dead_key_prune_task is not None:
|
||||
dead_key_prune_task.cancel()
|
||||
if auto_topup_task is not None:
|
||||
auto_topup_task.cancel()
|
||||
if refund_sweep_task is not None:
|
||||
@@ -210,8 +196,6 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
|
||||
tasks_to_wait.append(key_reset_task)
|
||||
if stale_reservation_task is not None:
|
||||
tasks_to_wait.append(stale_reservation_task)
|
||||
if dead_key_prune_task is not None:
|
||||
tasks_to_wait.append(dead_key_prune_task)
|
||||
if auto_topup_task is not None:
|
||||
tasks_to_wait.append(auto_topup_task)
|
||||
if refund_sweep_task is not None:
|
||||
|
||||
@@ -73,14 +73,6 @@ class Settings(BaseSettings):
|
||||
stale_reservation_timeout_seconds: int = Field(
|
||||
default=300, env="STALE_RESERVATION_TIMEOUT_SECONDS"
|
||||
)
|
||||
# Background prune of dead (zero balance, never used) API keys.
|
||||
# Interval 0 disables it; min-age is a grace period (default 1 week).
|
||||
dead_key_prune_interval_seconds: int = Field(
|
||||
default=3600, env="DEAD_KEY_PRUNE_INTERVAL_SECONDS"
|
||||
)
|
||||
dead_key_min_age_seconds: int = Field(
|
||||
default=604_800, env="DEAD_KEY_MIN_AGE_SECONDS"
|
||||
)
|
||||
|
||||
# Network
|
||||
cors_origins: list[str] = Field(default_factory=lambda: ["*"], env="CORS_ORIGINS")
|
||||
|
||||
@@ -418,21 +418,12 @@ def _calculate_from_tokens(
|
||||
},
|
||||
)
|
||||
|
||||
# Fold the cache-read/write cost into the visible ``input_msats`` so a
|
||||
# dashboard that renders I / O / T sees ``input + output == total``
|
||||
# exactly. This mirrors ``_fold_cache_into_input_tokens`` (which rolls the
|
||||
# cache token counts into the visible prompt total). The standalone
|
||||
# ``cache_read_msats`` / ``cache_creation_msats`` fields stay populated for
|
||||
# clients that want the breakdown; nothing sums the components to derive
|
||||
# ``total_msats`` (it is computed independently above), so this is
|
||||
# display-only and does not change what is billed.
|
||||
visible_output_msats = int(calc_output_msats)
|
||||
visible_input_msats = token_based_cost - visible_output_msats
|
||||
visible_input_msats = int(calc_input_msats + calc_cache_read_msats)
|
||||
|
||||
return CostData(
|
||||
base_msats=0,
|
||||
input_msats=visible_input_msats,
|
||||
output_msats=visible_output_msats,
|
||||
output_msats=int(calc_output_msats),
|
||||
total_msats=token_based_cost,
|
||||
total_usd=total_usd,
|
||||
input_tokens=input_tokens,
|
||||
|
||||
+45
-59
@@ -39,6 +39,7 @@ from ..payment.models import (
|
||||
list_models,
|
||||
)
|
||||
from ..payment.price import sats_usd_price
|
||||
from ..payment.usage import normalize_usage
|
||||
from ..wallet import recieve_token, send_token
|
||||
from . import messages_dispatch
|
||||
from .cache_breakpoints import (
|
||||
@@ -157,48 +158,56 @@ class BaseUpstreamProvider:
|
||||
|
||||
@staticmethod
|
||||
def _fold_cache_into_input_tokens(usage: object) -> None:
|
||||
"""Fold cache token counts into ``input_tokens`` / ``prompt_tokens``.
|
||||
"""Fold additive cache token counts into Anthropic ``input_tokens``.
|
||||
|
||||
Cost calculation has already used the per-bucket counts to bill the
|
||||
request correctly; what the client sees in the visible token total
|
||||
should be a single rolled-up prompt count *including* the cache
|
||||
portion. The standalone ``cache_read_input_tokens`` /
|
||||
request correctly; what the client sees in Anthropic-shaped visible
|
||||
token totals should be a single rolled-up input count *including* the
|
||||
cache portion. OpenAI-compatible ``prompt_tokens`` is already inclusive
|
||||
(DeepSeek, OpenAI, OpenRouter, litellm), so adding cache fields there
|
||||
would double-count.
|
||||
|
||||
The standalone ``cache_read_input_tokens`` /
|
||||
``cache_creation_input_tokens`` fields are left in place for clients
|
||||
that want the breakdown.
|
||||
|
||||
For Anthropic-shaped responses (``input_tokens`` present), the cache
|
||||
fields are forced to ``0`` when the upstream omitted them, so the
|
||||
client always sees a consistent shape.
|
||||
"""
|
||||
if not isinstance(usage, dict):
|
||||
return
|
||||
|
||||
# Normalise missing cache fields to 0 on Anthropic-shaped usage so
|
||||
# downstream consumers can rely on them being present.
|
||||
if "input_tokens" in usage:
|
||||
usage.setdefault("cache_read_input_tokens", 0)
|
||||
usage.setdefault("cache_creation_input_tokens", 0)
|
||||
# ``prompt_tokens`` is an inclusive OpenAI-compatible grand total.
|
||||
# Fold only native Anthropic-style usage where ``input_tokens`` excludes
|
||||
# cache reads/writes.
|
||||
if "input_tokens" not in usage or "prompt_tokens" in usage:
|
||||
return
|
||||
|
||||
usage.setdefault("cache_read_input_tokens", 0)
|
||||
usage.setdefault("cache_creation_input_tokens", 0)
|
||||
|
||||
try:
|
||||
cache_read = int(usage.get("cache_read_input_tokens") or 0)
|
||||
cache_creation = int(usage.get("cache_creation_input_tokens") or 0)
|
||||
input_tokens = int(usage.get("input_tokens") or 0)
|
||||
except (TypeError, ValueError):
|
||||
return
|
||||
extra = cache_read + cache_creation
|
||||
if extra <= 0:
|
||||
if extra > 0:
|
||||
usage["input_tokens"] = input_tokens + extra
|
||||
|
||||
@staticmethod
|
||||
def _add_normalized_usage_fields(response_json: object) -> None:
|
||||
"""Preserve raw usage while adding canonical fields for billing/display."""
|
||||
if not isinstance(response_json, dict):
|
||||
return
|
||||
if "input_tokens" in usage:
|
||||
try:
|
||||
usage["input_tokens"] = int(usage.get("input_tokens") or 0) + extra
|
||||
except (TypeError, ValueError):
|
||||
pass
|
||||
if "prompt_tokens" in usage:
|
||||
try:
|
||||
usage["prompt_tokens"] = (
|
||||
int(usage.get("prompt_tokens") or 0) + extra
|
||||
)
|
||||
except (TypeError, ValueError):
|
||||
pass
|
||||
usage = response_json.get("usage")
|
||||
if not isinstance(usage, dict):
|
||||
return
|
||||
normalized = normalize_usage(usage)
|
||||
if normalized is None:
|
||||
return
|
||||
usage.setdefault("input_tokens", normalized.input_tokens)
|
||||
usage.setdefault("output_tokens", normalized.output_tokens)
|
||||
usage.setdefault("cache_read_input_tokens", normalized.cache_read_tokens)
|
||||
usage.setdefault("cache_creation_input_tokens", normalized.cache_write_tokens)
|
||||
|
||||
def _apply_provider_field(self, response_json: object) -> None:
|
||||
"""Stamp the routstr ``provider`` field onto an upstream response payload.
|
||||
@@ -766,9 +775,7 @@ class BaseUpstreamProvider:
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def _process_event(
|
||||
raw_event: bytes, final: bool = False
|
||||
) -> Iterator[bytes]:
|
||||
def _process_event(raw_event: bytes) -> Iterator[bytes]:
|
||||
"""Process one complete SSE event block (lines up to a blank line).
|
||||
|
||||
Handles arbitrary upstream framing across every supported
|
||||
@@ -860,6 +867,7 @@ class BaseUpstreamProvider:
|
||||
k: v for k, v in obj.items() if k != "choices"
|
||||
}
|
||||
usage_chunk_data["choices"] = []
|
||||
self._add_normalized_usage_fields(usage_chunk_data)
|
||||
# Forward the content now, without usage, so token
|
||||
# usage is reported exactly once (in the trailer).
|
||||
forward = {k: v for k, v in obj.items() if k != "usage"}
|
||||
@@ -871,15 +879,10 @@ class BaseUpstreamProvider:
|
||||
)
|
||||
return
|
||||
usage_chunk_data = obj
|
||||
self._add_normalized_usage_fields(usage_chunk_data)
|
||||
return
|
||||
yield prefix + b"data: " + json.dumps(obj).encode() + b"\n\n"
|
||||
else:
|
||||
if final:
|
||||
# Final flush of a truncated tail: the upstream closed
|
||||
# mid-event, so ``data`` is incomplete JSON. Emitting it
|
||||
# as a ``data:`` frame would hand the client invalid
|
||||
# JSON (the "unexpected token" parse error). Drop it.
|
||||
return
|
||||
# Non-JSON data payload (partial fragment already reassembled
|
||||
# by buffering, or a provider control string). Re-prefix each
|
||||
# line so multi-line ``data`` stays valid SSE framing - a bare
|
||||
@@ -898,13 +901,7 @@ class BaseUpstreamProvider:
|
||||
# boundary-independent for every provider.
|
||||
buffer = b""
|
||||
async for chunk in response.aiter_bytes():
|
||||
# Normalize the *joined* buffer, not each chunk in
|
||||
# isolation: a CRLF event delimiter can straddle two
|
||||
# ``aiter_bytes`` chunks (``...\r`` then ``\n...``). A
|
||||
# per-chunk replace would leave a stray ``\r`` and the
|
||||
# ``\n\n`` split would miss the delimiter, merging two
|
||||
# events into one frame and breaking SSE clients.
|
||||
buffer = (buffer + chunk).replace(b"\r\n", b"\n")
|
||||
buffer += chunk.replace(b"\r\n", b"\n")
|
||||
while b"\n\n" in buffer:
|
||||
raw_event, buffer = buffer.split(b"\n\n", 1)
|
||||
for out in _process_event(raw_event):
|
||||
@@ -912,7 +909,7 @@ class BaseUpstreamProvider:
|
||||
|
||||
# Flush any trailing event that lacked a final blank line.
|
||||
if buffer.strip():
|
||||
for out in _process_event(buffer, final=True):
|
||||
for out in _process_event(buffer):
|
||||
yield out
|
||||
|
||||
async with create_session() as session:
|
||||
@@ -920,6 +917,8 @@ class BaseUpstreamProvider:
|
||||
if fresh_key:
|
||||
cost_data: dict
|
||||
try:
|
||||
if usage_chunk_data is not None:
|
||||
self._add_normalized_usage_fields(usage_chunk_data)
|
||||
adjustment_input = (
|
||||
usage_chunk_data
|
||||
if usage_chunk_data is not None
|
||||
@@ -1218,9 +1217,7 @@ class BaseUpstreamProvider:
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def _process_event(
|
||||
raw_event: bytes, final: bool = False
|
||||
) -> Iterator[bytes]:
|
||||
def _process_event(raw_event: bytes) -> Iterator[bytes]:
|
||||
"""Process one complete SSE event block for the Responses API.
|
||||
|
||||
Buffers full events (delimited by a blank line) so parsing is
|
||||
@@ -1290,11 +1287,6 @@ class BaseUpstreamProvider:
|
||||
|
||||
yield prefix + b"data: " + json.dumps(obj).encode() + b"\n\n"
|
||||
else:
|
||||
if final:
|
||||
# Final flush of a truncated tail: upstream closed
|
||||
# mid-event, so ``data`` is incomplete JSON. Dropping it
|
||||
# avoids handing the client an invalid ``data:`` frame.
|
||||
return
|
||||
# Re-prefix each line so multi-line ``data`` stays valid SSE
|
||||
# framing for the client.
|
||||
body = b"".join(
|
||||
@@ -1307,20 +1299,14 @@ class BaseUpstreamProvider:
|
||||
# delimiter so parsing is independent of byte boundaries.
|
||||
buffer = b""
|
||||
async for chunk in response.aiter_bytes():
|
||||
# Normalize the *joined* buffer, not each chunk in
|
||||
# isolation: a CRLF event delimiter can straddle two
|
||||
# ``aiter_bytes`` chunks (``...\r`` then ``\n...``). A
|
||||
# per-chunk replace would leave a stray ``\r`` and the
|
||||
# ``\n\n`` split would miss the delimiter, merging two
|
||||
# events into one frame and breaking SSE clients.
|
||||
buffer = (buffer + chunk).replace(b"\r\n", b"\n")
|
||||
buffer += chunk.replace(b"\r\n", b"\n")
|
||||
while b"\n\n" in buffer:
|
||||
raw_event, buffer = buffer.split(b"\n\n", 1)
|
||||
for out in _process_event(raw_event):
|
||||
yield out
|
||||
|
||||
if buffer.strip():
|
||||
for out in _process_event(buffer, final=True):
|
||||
for out in _process_event(buffer):
|
||||
yield out
|
||||
|
||||
# Always emit a cost-bearing data chunk
|
||||
|
||||
@@ -1,66 +0,0 @@
|
||||
"""TEMPORARY: local DeepSeek V4 pricing shim.
|
||||
|
||||
litellm's bundled cost map does not yet ship ``deepseek-v4-flash`` /
|
||||
``deepseek-v4-pro``. Without an entry, ``backfill_cache_pricing`` cannot find a
|
||||
``cache_read_input_token_cost`` and cache reads fall back to the full input
|
||||
rate — a ~60% overcharge on cache hits (DeepSeek hits are ~0.2x input).
|
||||
|
||||
This module injects the missing entries into ``litellm.model_cost`` at startup
|
||||
so the existing backfill path resolves them. Rates mirror the open upstream PR
|
||||
https://github.com/BerriAI/litellm/pull/26380 (issue
|
||||
https://github.com/BerriAI/litellm/issues/30430).
|
||||
|
||||
=== REMOVAL (once litellm ships these models) ===
|
||||
Delete this file and the single ``register_deepseek_v4_pricing()`` call in
|
||||
``routstr/core/main.py``. Nothing else depends on it. Entries are only added
|
||||
when absent, so a stale shim is harmless after upstream lands — but remove it.
|
||||
"""
|
||||
|
||||
import litellm
|
||||
|
||||
from ..core import get_logger
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
# USD per token. Source: BerriAI/litellm PR #26380.
|
||||
_DEEPSEEK_V4_RATES: dict[str, dict[str, float]] = {
|
||||
"deepseek-v4-flash": {
|
||||
"input_cost_per_token": 1.4e-07,
|
||||
"output_cost_per_token": 2.8e-07,
|
||||
"cache_read_input_token_cost": 2.8e-08,
|
||||
"cache_creation_input_token_cost": 0.0,
|
||||
"input_cost_per_token_cache_hit": 2.8e-08,
|
||||
},
|
||||
"deepseek-v4-pro": {
|
||||
"input_cost_per_token": 1.74e-06,
|
||||
"output_cost_per_token": 3.48e-06,
|
||||
"cache_read_input_token_cost": 1.4e-07,
|
||||
"cache_creation_input_token_cost": 0.0,
|
||||
"input_cost_per_token_cache_hit": 1.4e-07,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def register_deepseek_v4_pricing() -> None:
|
||||
"""Inject DeepSeek V4 pricing into ``litellm.model_cost`` if absent.
|
||||
|
||||
Idempotent and non-destructive: a key already present in the cost map
|
||||
(e.g. once litellm ships it) is left untouched. Registers both the bare
|
||||
(``deepseek-v4-flash``) and prefixed (``deepseek/deepseek-v4-flash``)
|
||||
spellings since ``backfill_cache_pricing`` tries both.
|
||||
"""
|
||||
added = []
|
||||
for bare, rates in _DEEPSEEK_V4_RATES.items():
|
||||
for key in (bare, f"deepseek/{bare}"):
|
||||
if key in litellm.model_cost:
|
||||
continue
|
||||
entry: dict[str, object] = dict(rates)
|
||||
entry["litellm_provider"] = "deepseek"
|
||||
entry["mode"] = "chat"
|
||||
litellm.model_cost[key] = entry
|
||||
added.append(key)
|
||||
if added:
|
||||
logger.info(
|
||||
"Registered temporary DeepSeek V4 pricing shim",
|
||||
extra={"models": added},
|
||||
)
|
||||
+14
-39
@@ -35,24 +35,6 @@ async def get_balance(unit: str) -> int:
|
||||
return wallet.available_balance.amount
|
||||
|
||||
|
||||
async def _redeem_same_mint(
|
||||
wallet: Wallet, token_obj: Token
|
||||
) -> tuple[int, str, str]: # amount, unit, mint_url
|
||||
"""Redeem proofs at their own issuing mint (no cross-mint swap).
|
||||
|
||||
split() re-mints the incoming proofs into fresh ones we own so the sender
|
||||
can't double-spend them. With include_fees=True the mint deducts its NUT-02
|
||||
per-proof input fee, so we end up holding only `amount - input_fees`. Credit
|
||||
that, not the face value, or routstr over-credits the user and its wallet
|
||||
drifts insolvent.
|
||||
"""
|
||||
await wallet.load_mint(keyset_id=token_obj.keysets[0])
|
||||
wallet.verify_proofs_dleq(token_obj.proofs)
|
||||
input_fees = wallet.get_fees_for_proofs(token_obj.proofs)
|
||||
await wallet.split(proofs=token_obj.proofs, amount=0, include_fees=True)
|
||||
return int(token_obj.amount) - input_fees, token_obj.unit, token_obj.mint
|
||||
|
||||
|
||||
async def recieve_token(
|
||||
token: str,
|
||||
) -> tuple[int, str, str]: # amount, unit, mint_url
|
||||
@@ -66,7 +48,12 @@ async def recieve_token(
|
||||
if token_obj.mint not in settings.cashu_mints:
|
||||
return await swap_to_primary_mint(token_obj, wallet)
|
||||
|
||||
return await _redeem_same_mint(wallet, token_obj)
|
||||
await wallet.load_mint(keyset_id=token_obj.keysets[0])
|
||||
|
||||
wallet.verify_proofs_dleq(token_obj.proofs)
|
||||
await wallet.split(proofs=token_obj.proofs, amount=0, include_fees=True)
|
||||
|
||||
return token_obj.amount, token_obj.unit, token_obj.mint
|
||||
|
||||
|
||||
async def send(amount: int, unit: str, mint_url: str | None = None) -> tuple[int, str]:
|
||||
@@ -226,9 +213,10 @@ async def swap_to_primary_mint(
|
||||
amount_msat = token_amount
|
||||
else:
|
||||
raise ValueError("Invalid unit")
|
||||
# If the token is already from the primary mint, we don't need a cross-mint
|
||||
# swap — redeem it same-mint. There's no melt/Lightning fee, but the mint's
|
||||
# NUT-02 input fee still applies; _redeem_same_mint accounts for it.
|
||||
primary_wallet = await get_wallet(settings.primary_mint, settings.primary_mint_unit)
|
||||
|
||||
# If the token is already from the primary mint, we don't need to swap
|
||||
# and we definitely don't want to calculate or pay fees.
|
||||
if token_obj.mint == settings.primary_mint:
|
||||
logger.info(
|
||||
"swap_to_primary_mint: token already on primary mint, skipping swap",
|
||||
@@ -238,9 +226,8 @@ async def swap_to_primary_mint(
|
||||
"unit": token_obj.unit,
|
||||
},
|
||||
)
|
||||
return await _redeem_same_mint(token_wallet, token_obj)
|
||||
|
||||
primary_wallet = await get_wallet(settings.primary_mint, settings.primary_mint_unit)
|
||||
await token_wallet.split(proofs=token_obj.proofs, amount=0, include_fees=True)
|
||||
return token_amount, token_obj.unit, token_obj.mint
|
||||
|
||||
minted_amount = await _calculate_swap_amount(
|
||||
amount_msat,
|
||||
@@ -441,11 +428,7 @@ async def credit_balance(
|
||||
.where(col(db.ApiKey.hashed_key) == key.hashed_key)
|
||||
.values(balance=(db.ApiKey.balance) + amount)
|
||||
)
|
||||
result = await session.exec(stmt) # type: ignore[call-overload]
|
||||
# If pruning removed this key after redemption, do not commit a no-op
|
||||
# balance update and pretend the top-up succeeded.
|
||||
if (getattr(result, "rowcount", 0) or 0) == 0:
|
||||
raise ValueError("API key disappeared before credit could be recorded")
|
||||
await session.exec(stmt) # type: ignore[call-overload]
|
||||
await session.commit()
|
||||
await session.refresh(key)
|
||||
|
||||
@@ -584,19 +567,11 @@ async def fetch_all_balances(
|
||||
}
|
||||
return error_result
|
||||
|
||||
# Build the set of mints to inspect. Received tokens are stored against
|
||||
# ``primary_mint`` (which defaults to a real mint even when ``cashu_mints``
|
||||
# is empty), so include it as a fallback — otherwise a node that accepts
|
||||
# payments would still report empty balances when ``cashu_mints`` is unset.
|
||||
mint_urls: list[str] = list(settings.cashu_mints)
|
||||
if settings.primary_mint and settings.primary_mint not in mint_urls:
|
||||
mint_urls.append(settings.primary_mint)
|
||||
|
||||
# Create tasks for all mint/unit combinations
|
||||
async with db.create_session() as session:
|
||||
tasks = [
|
||||
fetch_balance(session, mint_url, unit)
|
||||
for mint_url in mint_urls
|
||||
for mint_url in settings.cashu_mints
|
||||
for unit in units
|
||||
]
|
||||
|
||||
|
||||
@@ -1,283 +0,0 @@
|
||||
"""
|
||||
Tests for prune_dead_api_keys — the janitor that removes provably-dead 0/0/0
|
||||
API keys (funded keys fully refunded/expired without ever being used, plus bare
|
||||
orphans), while protecting keys that are still meaningful.
|
||||
"""
|
||||
|
||||
import time
|
||||
import uuid
|
||||
from collections.abc import Awaitable, Callable
|
||||
from typing import Any, cast
|
||||
|
||||
import pytest
|
||||
from sqlalchemy.sql.dml import Update
|
||||
from sqlmodel import col, update
|
||||
|
||||
from routstr.core.db import (
|
||||
ApiKey,
|
||||
CashuTransaction,
|
||||
LightningInvoice,
|
||||
create_session,
|
||||
prune_dead_api_keys,
|
||||
)
|
||||
|
||||
OLD = 100 # min_age_seconds used by the tests
|
||||
NOW = int(time.time())
|
||||
LONG_AGO = NOW - 10_000 # well past the grace period
|
||||
|
||||
|
||||
async def _exists(key_hash: str) -> bool:
|
||||
async with create_session() as session:
|
||||
return (await session.get(ApiKey, key_hash)) is not None
|
||||
|
||||
|
||||
def _dead_key(created_at: int | None) -> ApiKey:
|
||||
return ApiKey(
|
||||
hashed_key=f"dead_{uuid.uuid4().hex}",
|
||||
balance=0,
|
||||
reserved_balance=0,
|
||||
total_spent=0,
|
||||
total_requests=0,
|
||||
created_at=created_at,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_prunes_old_refunded_zero_key(patched_db_engine: None) -> None:
|
||||
"""A funded-then-refunded key (0/0/0, NULL parent, old) is pruned."""
|
||||
key = _dead_key(LONG_AGO)
|
||||
async with create_session() as session:
|
||||
session.add(key)
|
||||
await session.commit()
|
||||
|
||||
async with create_session() as session:
|
||||
pruned = await prune_dead_api_keys(session, OLD)
|
||||
|
||||
assert pruned == 1
|
||||
assert not await _exists(key.hashed_key)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_grace_period_protects_fresh_key(patched_db_engine: None) -> None:
|
||||
"""A dead-looking but recently created key is protected by the grace period."""
|
||||
key = _dead_key(int(time.time()))
|
||||
async with create_session() as session:
|
||||
session.add(key)
|
||||
await session.commit()
|
||||
|
||||
async with create_session() as session:
|
||||
pruned = await prune_dead_api_keys(session, OLD)
|
||||
|
||||
assert pruned == 0
|
||||
assert await _exists(key.hashed_key)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_used_key_never_pruned(patched_db_engine: None) -> None:
|
||||
"""Keys with any spend/requests or live balance are never pruned."""
|
||||
spent = _dead_key(LONG_AGO)
|
||||
spent.total_spent = 1
|
||||
requested = _dead_key(LONG_AGO)
|
||||
requested.total_requests = 1
|
||||
funded = _dead_key(LONG_AGO)
|
||||
funded.balance = 1000
|
||||
reserved = _dead_key(LONG_AGO)
|
||||
reserved.reserved_balance = 500
|
||||
|
||||
async with create_session() as session:
|
||||
for k in (spent, requested, funded, reserved):
|
||||
session.add(k)
|
||||
await session.commit()
|
||||
|
||||
async with create_session() as session:
|
||||
pruned = await prune_dead_api_keys(session, OLD)
|
||||
|
||||
assert pruned == 0
|
||||
for k in (spent, requested, funded, reserved):
|
||||
assert await _exists(k.hashed_key)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_parent_and_child_keys_are_not_pruned(
|
||||
patched_db_engine: None,
|
||||
) -> None:
|
||||
"""Pruning must not orphan child keys or delete valid children."""
|
||||
parent = _dead_key(LONG_AGO)
|
||||
child = ApiKey(
|
||||
hashed_key=f"child_{uuid.uuid4().hex}",
|
||||
balance=0,
|
||||
reserved_balance=0,
|
||||
total_spent=0,
|
||||
total_requests=0,
|
||||
created_at=LONG_AGO,
|
||||
parent_key_hash=parent.hashed_key,
|
||||
)
|
||||
async with create_session() as session:
|
||||
session.add(parent)
|
||||
session.add(child)
|
||||
await session.commit()
|
||||
|
||||
async with create_session() as session:
|
||||
pruned = await prune_dead_api_keys(session, OLD)
|
||||
|
||||
assert pruned == 0
|
||||
assert await _exists(parent.hashed_key)
|
||||
assert await _exists(child.hashed_key)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_pending_invoice_protects_key(patched_db_engine: None) -> None:
|
||||
"""A key referenced by a pending topup invoice is never pruned mid-topup."""
|
||||
key = _dead_key(LONG_AGO)
|
||||
invoice = LightningInvoice(
|
||||
id=f"inv_{uuid.uuid4().hex}",
|
||||
bolt11=f"lnbc_{uuid.uuid4().hex}",
|
||||
amount_sats=10,
|
||||
description="topup",
|
||||
payment_hash=uuid.uuid4().hex,
|
||||
status="pending",
|
||||
api_key_hash=key.hashed_key,
|
||||
purpose="topup",
|
||||
expires_at=NOW + 10_000,
|
||||
)
|
||||
async with create_session() as session:
|
||||
session.add(key)
|
||||
session.add(invoice)
|
||||
await session.commit()
|
||||
|
||||
async with create_session() as session:
|
||||
pruned = await prune_dead_api_keys(session, OLD)
|
||||
|
||||
assert pruned == 0
|
||||
assert await _exists(key.hashed_key)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_paid_invoice_does_not_protect_key(patched_db_engine: None) -> None:
|
||||
"""A settled (non-pending) invoice does not keep a dead key alive."""
|
||||
key = _dead_key(LONG_AGO)
|
||||
invoice = LightningInvoice(
|
||||
id=f"inv_{uuid.uuid4().hex}",
|
||||
bolt11=f"lnbc_{uuid.uuid4().hex}",
|
||||
amount_sats=10,
|
||||
description="topup",
|
||||
payment_hash=uuid.uuid4().hex,
|
||||
status="paid",
|
||||
api_key_hash=key.hashed_key,
|
||||
purpose="topup",
|
||||
expires_at=NOW - 1,
|
||||
)
|
||||
async with create_session() as session:
|
||||
session.add(key)
|
||||
session.add(invoice)
|
||||
await session.commit()
|
||||
|
||||
async with create_session() as session:
|
||||
pruned = await prune_dead_api_keys(session, OLD)
|
||||
|
||||
assert pruned == 1
|
||||
assert not await _exists(key.hashed_key)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_key_that_becomes_meaningful_during_prune_survives(
|
||||
patched_db_engine: None, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""Revalidate before unlink/delete so a late top-up cannot be pruned."""
|
||||
key = _dead_key(LONG_AGO)
|
||||
txn = CashuTransaction(
|
||||
id=uuid.uuid4().hex,
|
||||
token="cashuABC",
|
||||
amount=21,
|
||||
unit="sat",
|
||||
type="in",
|
||||
source="apikey",
|
||||
api_key_hashed_key=key.hashed_key,
|
||||
)
|
||||
async with create_session() as session:
|
||||
session.add(key)
|
||||
session.add(txn)
|
||||
await session.commit()
|
||||
|
||||
async with create_session() as session:
|
||||
original_exec = cast(Callable[..., Awaitable[Any]], session.exec)
|
||||
topped_up = False
|
||||
|
||||
async def exec_with_late_topup(
|
||||
statement: Any, *args: Any, **kwargs: Any
|
||||
) -> Any:
|
||||
nonlocal topped_up
|
||||
if (
|
||||
not topped_up
|
||||
and isinstance(statement, Update)
|
||||
and getattr(statement.table, "name", None) == "cashu_transactions"
|
||||
):
|
||||
topped_up = True
|
||||
async with create_session() as topup_session:
|
||||
await topup_session.exec( # type: ignore[call-overload]
|
||||
update(ApiKey)
|
||||
.where(col(ApiKey.hashed_key) == key.hashed_key)
|
||||
.values(balance=42)
|
||||
)
|
||||
await topup_session.commit()
|
||||
return await original_exec(statement, *args, **kwargs)
|
||||
|
||||
monkeypatch.setattr(session, "exec", exec_with_late_topup)
|
||||
pruned = await prune_dead_api_keys(session, OLD)
|
||||
|
||||
assert pruned == 0
|
||||
assert await _exists(key.hashed_key)
|
||||
async with create_session() as session:
|
||||
surviving = await session.get(CashuTransaction, txn.id)
|
||||
assert surviving is not None
|
||||
assert surviving.api_key_hashed_key == key.hashed_key
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_transaction_audit_trail_preserved(patched_db_engine: None) -> None:
|
||||
"""Pruning a refunded key keeps its cashu_transactions, unlinked from the key."""
|
||||
key = _dead_key(LONG_AGO)
|
||||
txn = CashuTransaction(
|
||||
id=uuid.uuid4().hex,
|
||||
token="cashuABC",
|
||||
amount=21,
|
||||
unit="sat",
|
||||
type="in",
|
||||
source="apikey",
|
||||
api_key_hashed_key=key.hashed_key,
|
||||
)
|
||||
async with create_session() as session:
|
||||
session.add(key)
|
||||
session.add(txn)
|
||||
await session.commit()
|
||||
|
||||
async with create_session() as session:
|
||||
pruned = await prune_dead_api_keys(session, OLD)
|
||||
|
||||
assert pruned == 1
|
||||
assert not await _exists(key.hashed_key)
|
||||
|
||||
async with create_session() as session:
|
||||
surviving = await session.get(CashuTransaction, txn.id)
|
||||
assert surviving is not None, "Financial audit row must survive key deletion"
|
||||
assert surviving.api_key_hashed_key is None, "Link must be nulled, not dangling"
|
||||
assert surviving.amount == 21
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_periodic_prune_disabled_returns_immediately(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""Non-positive intervals disable the janitor."""
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
from routstr import auth
|
||||
from routstr.core.settings import settings
|
||||
|
||||
monkeypatch.setattr(settings, "dead_key_prune_interval_seconds", 0)
|
||||
sleep_mock = AsyncMock()
|
||||
monkeypatch.setattr(auth.asyncio, "sleep", sleep_mock)
|
||||
|
||||
await auth.periodic_dead_key_prune()
|
||||
|
||||
sleep_mock.assert_not_called()
|
||||
@@ -1,59 +0,0 @@
|
||||
import hashlib
|
||||
from types import SimpleNamespace
|
||||
from typing import AsyncGenerator
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import pytest
|
||||
from fastapi import HTTPException
|
||||
from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
|
||||
from sqlalchemy.pool import StaticPool
|
||||
from sqlmodel import SQLModel
|
||||
from sqlmodel.ext.asyncio.session import AsyncSession
|
||||
|
||||
from routstr.auth import validate_bearer_key
|
||||
from routstr.core.db import ApiKey
|
||||
|
||||
|
||||
def _make_engine() -> AsyncEngine:
|
||||
return create_async_engine(
|
||||
"sqlite+aiosqlite://",
|
||||
poolclass=StaticPool,
|
||||
connect_args={"check_same_thread": False},
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def session() -> AsyncGenerator[AsyncSession, None]:
|
||||
engine = _make_engine()
|
||||
async with engine.begin() as conn:
|
||||
await conn.run_sync(SQLModel.metadata.create_all)
|
||||
db_session = AsyncSession(engine, expire_on_commit=False)
|
||||
try:
|
||||
yield db_session
|
||||
finally:
|
||||
await db_session.close()
|
||||
await engine.dispose()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_failed_first_cashu_redemption_rolls_back_empty_api_key(
|
||||
session: AsyncSession,
|
||||
) -> None:
|
||||
token = "cashuAfirst_seen_but_redemption_fails"
|
||||
hashed_key = hashlib.sha256(token.encode()).hexdigest()
|
||||
token_obj = SimpleNamespace(mint="http://mint:3338", unit="sat")
|
||||
|
||||
from routstr.core.settings import settings
|
||||
|
||||
with (
|
||||
patch.object(settings, "cashu_mints", ["http://mint:3338"]),
|
||||
patch("routstr.auth.deserialize_token_from_string", return_value=token_obj),
|
||||
patch(
|
||||
"routstr.auth.credit_balance",
|
||||
new=AsyncMock(side_effect=ValueError("token already spent")),
|
||||
),
|
||||
):
|
||||
with pytest.raises(HTTPException):
|
||||
await validate_bearer_key(token, session)
|
||||
|
||||
assert await session.get(ApiKey, hashed_key) is None
|
||||
@@ -182,13 +182,11 @@ async def test_deepseek_cache_hits_billed_at_cache_rate(model_pricing: Mock) ->
|
||||
|
||||
assert isinstance(result, CostData)
|
||||
# 1000 input @ 1 msat + 9000 cache reads @ 0.1 msat + 500 output @ 2 msat.
|
||||
# input_msats folds the cache-read cost in (1000 + 900) so a dashboard
|
||||
# rendering I/O/T sees input + output == total; the cache portion stays
|
||||
# visible in cache_read_msats.
|
||||
# Client-visible input_msats includes cache-read input cost for display,
|
||||
# while cache_read_msats keeps the detailed breakdown.
|
||||
assert result.input_msats == 1900
|
||||
assert result.cache_read_msats == 900
|
||||
assert result.output_msats == 1000
|
||||
assert result.input_msats == 1900
|
||||
assert result.input_msats + result.output_msats == result.total_msats
|
||||
assert result.total_msats == 2900
|
||||
|
||||
|
||||
|
||||
@@ -1,73 +0,0 @@
|
||||
from contextlib import asynccontextmanager
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from routstr.wallet import fetch_all_balances
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def _fake_session(): # type: ignore[no-untyped-def]
|
||||
yield MagicMock()
|
||||
|
||||
|
||||
def _patches(proof_amount: int = 1000): # type: ignore[no-untyped-def]
|
||||
proof = MagicMock(amount=proof_amount)
|
||||
return [
|
||||
patch("routstr.wallet.get_wallet", AsyncMock(return_value=MagicMock())),
|
||||
patch(
|
||||
"routstr.wallet.get_proofs_per_mint_and_unit",
|
||||
MagicMock(return_value=[proof]),
|
||||
),
|
||||
patch(
|
||||
"routstr.wallet.slow_filter_spend_proofs",
|
||||
AsyncMock(side_effect=lambda proofs, wallet: proofs),
|
||||
),
|
||||
patch(
|
||||
"routstr.wallet.db.balances_for_mint_and_unit",
|
||||
AsyncMock(return_value=0),
|
||||
),
|
||||
patch("routstr.wallet.db.create_session", _fake_session),
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_fetch_all_balances_falls_back_to_primary_mint() -> None:
|
||||
"""With empty cashu_mints, balances are still fetched for primary_mint."""
|
||||
from routstr.core.settings import settings
|
||||
|
||||
with patch.object(settings, "cashu_mints", []), patch.object(
|
||||
settings, "primary_mint", "http://primary:3338"
|
||||
):
|
||||
for p in _patches(proof_amount=1000):
|
||||
p.start()
|
||||
try:
|
||||
details, total_wallet, total_user, owner = await fetch_all_balances(
|
||||
units=["sat"]
|
||||
)
|
||||
finally:
|
||||
patch.stopall()
|
||||
|
||||
assert [d["mint_url"] for d in details] == ["http://primary:3338"]
|
||||
assert total_wallet == 1000
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_fetch_all_balances_no_duplicate_primary_mint() -> None:
|
||||
"""primary_mint already in cashu_mints is not inspected twice."""
|
||||
from routstr.core.settings import settings
|
||||
|
||||
with patch.object(
|
||||
settings, "cashu_mints", ["http://primary:3338"]
|
||||
), patch.object(settings, "primary_mint", "http://primary:3338"):
|
||||
for p in _patches(proof_amount=1000):
|
||||
p.start()
|
||||
try:
|
||||
details, total_wallet, _total_user, _owner = await fetch_all_balances(
|
||||
units=["sat"]
|
||||
)
|
||||
finally:
|
||||
patch.stopall()
|
||||
|
||||
assert [d["mint_url"] for d in details] == ["http://primary:3338"]
|
||||
assert total_wallet == 1000
|
||||
@@ -20,6 +20,7 @@ comment ever reaches the client. That invariant is exactly what the buggy
|
||||
|
||||
import json
|
||||
from collections.abc import AsyncGenerator
|
||||
from typing import cast
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
import pytest
|
||||
@@ -107,6 +108,76 @@ def _assert_clean(out: list[bytes]) -> list[dict]:
|
||||
return objs
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_deepseek_usage_chunk_is_normalized_before_billing() -> None:
|
||||
"""DeepSeek stream trailers keep raw fields and add canonical billing fields."""
|
||||
usage = {
|
||||
"prompt_tokens": 10000,
|
||||
"completion_tokens": 500,
|
||||
"total_tokens": 10500,
|
||||
"prompt_cache_hit_tokens": 9000,
|
||||
"prompt_cache_miss_tokens": 1000,
|
||||
}
|
||||
chunks = [
|
||||
b'data: {"id":"ds","model":"deepseek-chat","choices":[{"delta":{"content":"ok"}}]}\n\n',
|
||||
b"data: "
|
||||
+ json.dumps(
|
||||
{
|
||||
"id": "ds",
|
||||
"model": "deepseek-chat",
|
||||
"choices": [],
|
||||
"usage": usage,
|
||||
}
|
||||
).encode()
|
||||
+ b"\n\n",
|
||||
b"data: [DONE]\n\n",
|
||||
]
|
||||
|
||||
await _drive(chunks)
|
||||
|
||||
adjust_mock = cast(AsyncMock, base.adjust_payment_for_tokens)
|
||||
adjustment_input = adjust_mock.call_args.args[1]
|
||||
billed_usage = adjustment_input["usage"]
|
||||
assert billed_usage["prompt_tokens"] == 10000
|
||||
assert billed_usage["prompt_cache_hit_tokens"] == 9000
|
||||
assert billed_usage["prompt_cache_miss_tokens"] == 1000
|
||||
assert billed_usage["input_tokens"] == 1000
|
||||
assert billed_usage["output_tokens"] == 500
|
||||
assert billed_usage["cache_read_input_tokens"] == 9000
|
||||
assert billed_usage["cache_creation_input_tokens"] == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_fold_cache_tokens_does_not_double_count_inclusive_prompt_tokens() -> None:
|
||||
"""Visible usage mutation must not inflate OpenAI-compatible prompt totals."""
|
||||
usage = {
|
||||
"prompt_tokens": 10000,
|
||||
"completion_tokens": 500,
|
||||
"cache_read_input_tokens": 9000,
|
||||
"prompt_cache_hit_tokens": 9000,
|
||||
}
|
||||
|
||||
BaseUpstreamProvider._fold_cache_into_input_tokens(usage)
|
||||
|
||||
assert usage["prompt_tokens"] == 10000
|
||||
assert usage["cache_read_input_tokens"] == 9000
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_fold_cache_tokens_still_rolls_up_anthropic_input_tokens() -> None:
|
||||
"""Anthropic native input_tokens excludes cache and still needs rollup."""
|
||||
usage = {
|
||||
"input_tokens": 1000,
|
||||
"output_tokens": 500,
|
||||
"cache_read_input_tokens": 9000,
|
||||
"cache_creation_input_tokens": 200,
|
||||
}
|
||||
|
||||
BaseUpstreamProvider._fold_cache_into_input_tokens(usage)
|
||||
|
||||
assert usage["input_tokens"] == 10200
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_openai_style_plain_stream() -> None:
|
||||
"""OpenAI / Groq / Fireworks / xAI / Perplexity: plain data + [DONE]."""
|
||||
@@ -337,75 +408,3 @@ async def test_multiline_non_json_data_each_line_prefixed() -> None:
|
||||
continue
|
||||
assert line.startswith(b"data: "), f"bare line leaked to client: {line!r}"
|
||||
assert b"data: line one" in blob and b"data: line two" in blob
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_crlf_delimiter_split_across_chunk_boundary() -> None:
|
||||
"""CRLF event delimiter straddling two TCP reads must not merge events.
|
||||
|
||||
Regression: a per-chunk ``replace(b"\\r\\n", b"\\n")`` left a stray ``\\r``
|
||||
when a ``\\r\\n`` of the ``\\r\\n\\r\\n`` delimiter landed at the very end of
|
||||
one ``aiter_bytes`` chunk and the matching ``\\n`` opened the next. The
|
||||
``\\n\\n`` split then missed the boundary, glued two events into one frame
|
||||
with two ``data:`` lines, and the client's ``JSON.parse`` threw on the
|
||||
concatenated payload (the "unexpected token"/"Extra data" crash).
|
||||
"""
|
||||
e1 = b'data: {"id":"x","choices":[{"delta":{"content":"a"}}]}'
|
||||
e2 = b'data: {"id":"x","choices":[{"delta":{"content":"b"}}]}'
|
||||
chunks = [
|
||||
e1 + b"\r\n\r", # delimiter cut mid-CRLF
|
||||
b"\n" + e2 + b"\r\n\r\n",
|
||||
b"data: [DONE]\r\n\r\n",
|
||||
]
|
||||
out = await _drive(chunks)
|
||||
|
||||
# Client-accurate check: a real SSE client concatenates all ``data:`` lines
|
||||
# *within one event* (events are ``\n\n``-delimited) before parsing. A
|
||||
# merged frame would surface here as two objects glued into one payload,
|
||||
# which ``_assert_clean`` (per-line) would miss.
|
||||
blob = b"".join(out)
|
||||
contents: list[str] = []
|
||||
for event in blob.split(b"\n\n"):
|
||||
datas = [
|
||||
ln[len(b"data: ") :]
|
||||
for ln in event.split(b"\n")
|
||||
if ln.startswith(b"data: ")
|
||||
]
|
||||
if not datas:
|
||||
continue
|
||||
payload = b"".join(datas)
|
||||
if payload.strip() == b"[DONE]":
|
||||
continue
|
||||
obj = json.loads(payload) # raises if two events were merged into one
|
||||
for c in obj.get("choices", []):
|
||||
if "delta" in c:
|
||||
contents.append(c["delta"]["content"])
|
||||
assert contents == ["a", "b"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_truncated_json_tail_on_connection_close() -> None:
|
||||
"""A stream that drops mid-event must not emit the partial JSON downstream.
|
||||
|
||||
Regression: the end-of-stream flush ran ``_process_event`` on the leftover
|
||||
buffer unconditionally. When the upstream connection closed mid-event the
|
||||
leftover was incomplete JSON, which fell through to the raw-forward path and
|
||||
handed the client a ``data: {partial`` frame -> ``Unterminated string`` parse
|
||||
error. The truncated tail must be dropped instead.
|
||||
"""
|
||||
chunks = [
|
||||
b'data: {"id":"x","choices":[{"delta":{"content":"ok"}}]}\n\n',
|
||||
b'data: {"id":"x","choices":[{"delta":{"con', # connection dies here
|
||||
]
|
||||
out = await _drive(chunks)
|
||||
objs = _assert_clean(out) # raises if the partial tail leaked as a data frame
|
||||
contents = [
|
||||
c["delta"]["content"]
|
||||
for o in objs
|
||||
for c in o.get("choices", [])
|
||||
if "delta" in c
|
||||
]
|
||||
# The one complete chunk is delivered; the truncated fragment is dropped
|
||||
# entirely (no second delta), and _assert_clean above guarantees nothing
|
||||
# non-JSON ever reached the client.
|
||||
assert contents == ["ok"]
|
||||
|
||||
+3
-115
@@ -39,8 +39,6 @@ async def test_recieve_token_valid() -> None:
|
||||
|
||||
mock_wallet = Mock()
|
||||
mock_wallet.split = AsyncMock()
|
||||
# Fee-free trusted mint (e.g. Minibits): nothing deducted.
|
||||
mock_wallet.get_fees_for_proofs = Mock(return_value=0)
|
||||
|
||||
from routstr.core.settings import settings
|
||||
|
||||
@@ -63,69 +61,6 @@ async def test_recieve_token_valid() -> None:
|
||||
assert mint == "http://mint:3338"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_recieve_token_trusted_mint_deducts_input_fee() -> None:
|
||||
"""A trusted mint that charges NUT-02 input fees.
|
||||
|
||||
The same-mint receive (`wallet.split(..., include_fees=True)`, a NUT-03 swap
|
||||
at the same mint — not swap_to_primary_mint) pays the mint's per-proof fee,
|
||||
so routstr only ends up with `face - input_fee` in fresh proofs. The credited
|
||||
amount must reflect that, otherwise routstr over-credits the user and its own
|
||||
wallet drifts toward insolvency.
|
||||
"""
|
||||
token_data = {
|
||||
"token": [
|
||||
{
|
||||
"mint": "http://mint:3338",
|
||||
"proofs": [
|
||||
{"amount": 1000, "id": "test", "secret": "secret", "C": "curve"}
|
||||
],
|
||||
}
|
||||
],
|
||||
"unit": "sat",
|
||||
}
|
||||
token_json = json.dumps(token_data)
|
||||
token_b64 = base64.urlsafe_b64encode(token_json.encode()).decode()
|
||||
token_str = f"cashuA{token_b64}"
|
||||
|
||||
mock_wallet = Mock()
|
||||
mock_wallet.split = AsyncMock()
|
||||
# Mock a 3-sat input fee from the Cashu wallet API.
|
||||
mock_wallet.get_fees_for_proofs = Mock(return_value=3)
|
||||
|
||||
from routstr.core.settings import settings
|
||||
|
||||
with patch.object(settings, "cashu_mints", ["http://mint:3338"]):
|
||||
with patch("routstr.wallet.deserialize_token_from_string") as mock_deserialize:
|
||||
mock_token = Mock()
|
||||
mock_token.keysets = ["keyset1"]
|
||||
mock_token.mint = "http://mint:3338"
|
||||
mock_token.unit = "sat"
|
||||
mock_token.amount = 1000
|
||||
mock_token.proofs = [{"amount": 1000}]
|
||||
mock_deserialize.return_value = mock_token
|
||||
|
||||
mock_wallet.load_mint = AsyncMock()
|
||||
mock_wallet.load_proofs = AsyncMock()
|
||||
# Patch get_wallet directly so the module-level `_wallets` cache
|
||||
# (keyed by mint URL) can't hand back a wallet from another test.
|
||||
with patch(
|
||||
"routstr.wallet.get_wallet",
|
||||
AsyncMock(return_value=mock_wallet),
|
||||
):
|
||||
amount, unit, mint = await recieve_token(token_str)
|
||||
assert amount == 997 # 1000 face - 3 sat input fee paid on swap
|
||||
assert unit == "sat"
|
||||
assert mint == "http://mint:3338"
|
||||
mock_wallet.get_fees_for_proofs.assert_called_once_with(
|
||||
mock_token.proofs
|
||||
)
|
||||
# DLEQ is verified before re-minting the incoming proofs.
|
||||
mock_wallet.verify_proofs_dleq.assert_called_once_with(
|
||||
mock_token.proofs
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_token() -> None:
|
||||
mock_wallet = Mock()
|
||||
@@ -150,7 +85,6 @@ async def test_credit_balance() -> None:
|
||||
mock_key.balance = 5000000
|
||||
mock_key.hashed_key = "test_hash"
|
||||
mock_session = AsyncMock()
|
||||
mock_session.exec.return_value.rowcount = 1
|
||||
|
||||
# Mock session.refresh to update the balance (simulates DB reload)
|
||||
async def mock_refresh(key: ApiKey) -> None:
|
||||
@@ -207,38 +141,6 @@ async def test_credit_balance_rejects_zero_amount() -> None:
|
||||
assert not mock_session.commit.called
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_credit_balance_rejects_missing_key() -> None:
|
||||
"""A top-up must fail if the key was pruned after redemption."""
|
||||
token_data = {
|
||||
"token": [{"mint": "http://mint:3338", "proofs": [{"amount": 1000}]}],
|
||||
"unit": "sat",
|
||||
}
|
||||
token_json = json.dumps(token_data)
|
||||
token_b64 = base64.urlsafe_b64encode(token_json.encode()).decode()
|
||||
token_str = f"cashuA{token_b64}"
|
||||
|
||||
mock_key = Mock()
|
||||
mock_key.balance = 0
|
||||
mock_key.hashed_key = "test_hash"
|
||||
mock_session = AsyncMock()
|
||||
mock_session.exec.return_value.rowcount = 0
|
||||
|
||||
from routstr.core.settings import settings
|
||||
|
||||
with patch.object(settings, "cashu_mints", ["http://mint:3338"]):
|
||||
with patch(
|
||||
"routstr.wallet.recieve_token",
|
||||
return_value=(1000, "sat", "http://mint:3338"),
|
||||
):
|
||||
with pytest.raises(ValueError, match="disappeared"):
|
||||
await credit_balance(token_str, mock_key, mock_session)
|
||||
|
||||
# UPDATE matched nothing; committing would hide the failed credit.
|
||||
assert mock_session.exec.called
|
||||
assert not mock_session.commit.called
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_swap_to_primary_mint_insufficient_for_fees() -> None:
|
||||
"""Token amount is less than melt_quote.amount + melt_quote.fee_reserve."""
|
||||
@@ -352,31 +254,19 @@ async def test_recieve_token_untrusted_mint() -> None:
|
||||
assert mint == "http://mint:3338"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.asyncio
|
||||
async def test_swap_to_primary_mint_already_on_primary() -> None:
|
||||
"""Same-mint shortcut: the token is already on the primary mint.
|
||||
|
||||
No cross-mint swap (no melt/mint), but the same-mint split(include_fees=True)
|
||||
still burns the mint's NUT-02 input fee, so the credited amount must be face
|
||||
minus the input fee — not full face value (the over-credit bug). DLEQ is
|
||||
verified too, matching the trusted same-mint receive path.
|
||||
"""
|
||||
from routstr.core.settings import settings
|
||||
from routstr.wallet import swap_to_primary_mint
|
||||
|
||||
mock_token = Mock()
|
||||
mock_token.mint = settings.primary_mint
|
||||
mock_token.keysets = ["keyset1"]
|
||||
mock_token.amount = 1000
|
||||
mock_token.unit = "sat"
|
||||
mock_token.proofs = [{"amount": 1000}]
|
||||
mock_token.proofs = []
|
||||
|
||||
mock_token_wallet = Mock()
|
||||
mock_token_wallet.load_mint = AsyncMock()
|
||||
mock_token_wallet.load_proofs = AsyncMock()
|
||||
mock_token_wallet.verify_proofs_dleq = Mock()
|
||||
# Mock a 3-sat input fee from the Cashu wallet API.
|
||||
mock_token_wallet.get_fees_for_proofs = Mock(return_value=3)
|
||||
mock_token_wallet.split = AsyncMock(return_value=None)
|
||||
mock_token_wallet.request_mint = AsyncMock()
|
||||
mock_token_wallet.melt_quote = AsyncMock()
|
||||
@@ -384,11 +274,9 @@ async def test_swap_to_primary_mint_already_on_primary() -> None:
|
||||
with patch("routstr.wallet.get_wallet", AsyncMock(return_value=mock_token_wallet)):
|
||||
amount, unit, mint = await swap_to_primary_mint(mock_token, mock_token_wallet)
|
||||
|
||||
assert amount == 997 # 1000 face - 3 sat input fee
|
||||
assert amount == 1000
|
||||
assert unit == "sat"
|
||||
assert mint == settings.primary_mint
|
||||
mock_token_wallet.verify_proofs_dleq.assert_called_once_with(mock_token.proofs)
|
||||
mock_token_wallet.get_fees_for_proofs.assert_called_once_with(mock_token.proofs)
|
||||
mock_token_wallet.split.assert_called_once()
|
||||
mock_token_wallet.request_mint.assert_not_called()
|
||||
mock_token_wallet.melt_quote.assert_not_called()
|
||||
|
||||
@@ -297,13 +297,16 @@ export default function ProvidersPage() {
|
||||
};
|
||||
|
||||
const toggleProviderExpansion = (providerId: number) => {
|
||||
if (expandedProviders.has(providerId)) {
|
||||
setExpandedProviders(new Set());
|
||||
const newExpanded = new Set(expandedProviders);
|
||||
if (newExpanded.has(providerId)) {
|
||||
newExpanded.delete(providerId);
|
||||
} else {
|
||||
newExpanded.add(providerId);
|
||||
}
|
||||
setExpandedProviders(newExpanded);
|
||||
if (!newExpanded.has(providerId)) {
|
||||
setViewingModels(null);
|
||||
} else {
|
||||
// Accordion: only one provider open at a time so switching to another
|
||||
// provider's models auto-collapses the previously expanded one.
|
||||
setExpandedProviders(new Set([providerId]));
|
||||
setViewingModels(providerId);
|
||||
}
|
||||
};
|
||||
|
||||
Reference in New Issue
Block a user