fix migration

fix forwarding upstream error responses

migrations/versions/02650cd6f028_add_routstr_fees_table.py

@@ -0,0 +1,32 @@
+"""add routstr_fees table
+
+Revision ID: 02650cd6f028
+Revises: c3d4e5f6a7b8
+Create Date: 2026-04-24 00:00:00.000000
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "02650cd6f028"
+down_revision = "c3d4e5f6a7b8"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "routstr_fees",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("accumulated_msats", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("total_paid_msats", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("last_paid_at", sa.Integer(), nullable=True),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    # Seed with a single row
+    op.execute("INSERT INTO routstr_fees (id, accumulated_msats, total_paid_msats) VALUES (1, 0, 0)")
+
+
+def downgrade() -> None:
+    op.drop_table("routstr_fees")

migrations/versions/e8f9a0b1c2d3_merge_heads.py

@@ -0,0 +1,20 @@
+"""merge heads: routstr_fees + api_key_to_cashu_transactions
+
+Revision ID: e8f9a0b1c2d3
+Revises: 02650cd6f028, d4e5f6a7b8c9
+Create Date: 2026-04-24 00:00:00.000000
+"""
+
+# revision identifiers, used by Alembic.
+revision = "e8f9a0b1c2d3"
+down_revision = ("02650cd6f028", "d4e5f6a7b8c9")
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    pass
+
+
+def downgrade() -> None:
+    pass

routstr/auth.py

@@ -12,7 +12,7 @@ from sqlalchemy.exc import IntegrityError
 from sqlmodel import col, select, update
 
 from .core import get_logger
-from .core.db import ApiKey, AsyncSession
+from .core.db import ApiKey, AsyncSession, accumulate_routstr_fee
 from .core.settings import settings
 from .payment.cost_calculation import (
     CostData,
@@ -25,6 +25,12 @@ from .wallet import credit_balance, deserialize_token_from_string
 logger = get_logger(__name__)
 payments_logger = get_logger("routstr.payments")
 
+# Routstr platform fee constants
+ROUTSTR_FEE_PERCENT: float = 2.1
+ROUTSTR_LN_ADDRESS: str = "npub130mznv74rxs032peqym6g3wqavh472623mt3z5w73xq9r6qqdufs7ql29s@npub.cash"
+ROUTSTR_FEE_PAYOUT_INTERVAL_SECONDS: int = 900
+ROUTSTR_FEE_DEFAULT_PAYOUT: int = 200
+
 # TODO: implement prepaid api key (not like it was before)
 # PREPAID_API_KEY = os.environ.get("PREPAID_API_KEY", None)
 # PREPAID_BALANCE = int(os.environ.get("PREPAID_BALANCE", "0")) * 1000  # Convert to msats
@@ -741,6 +747,17 @@ async def adjust_payment_for_tokens(
                 },
             )
 
+    async def _accumulate_fee(total_cost_msats: int) -> None:
+        if total_cost_msats > 0 and ROUTSTR_FEE_PERCENT > 0:
+            fee_msats = math.ceil(total_cost_msats * ROUTSTR_FEE_PERCENT / 100)
+            try:
+                await accumulate_routstr_fee(session, fee_msats)
+            except Exception as e:
+                logger.warning(
+                    "Failed to accumulate Routstr fee",
+                    extra={"error": str(e), "fee_msats": fee_msats},
+                )
+
     match await calculate_cost(response_data, deducted_max_cost, session):
         case MaxCostData() as cost:
             logger.debug(
@@ -832,6 +849,7 @@ async def adjust_payment_for_tokens(
                         "model": model,
                     },
                 )
+                await _accumulate_fee(cost.total_msats)
                 payments_logger.info(
                     "FINALIZE",
                     extra={
@@ -935,6 +953,7 @@ async def adjust_payment_for_tokens(
                 await session.refresh(billing_key)
                 if billing_key.hashed_key != key.hashed_key:
                     await session.refresh(key)
+                await _accumulate_fee(total_cost_msats)
                 payments_logger.info(
                     "FINALIZE",
                     extra={
@@ -1005,6 +1024,7 @@ async def adjust_payment_for_tokens(
                             "model": model,
                         },
                     )
+                    await _accumulate_fee(total_cost_msats)
                     payments_logger.info(
                         "FINALIZE",
                         extra={
@@ -1130,6 +1150,7 @@ async def adjust_payment_for_tokens(
                             "model": model,
                         },
                     )
+                    await _accumulate_fee(total_cost_msats)
                     payments_logger.info(
                         "FINALIZE",
                         extra={

routstr/core/db.py

@@ -12,7 +12,7 @@ from alembic.util.exc import CommandError
 from sqlalchemy import UniqueConstraint
 from sqlalchemy.exc import OperationalError
 from sqlalchemy.ext.asyncio.engine import create_async_engine
-from sqlmodel import Field, Relationship, SQLModel, func, select, update
+from sqlmodel import Field, Relationship, SQLModel, col, func, select, update
 from sqlmodel.ext.asyncio.session import AsyncSession
 
 from .logging import get_logger
@@ -223,6 +223,50 @@ class UpstreamProviderRow(SQLModel, table=True):  # type: ignore
     )
 
 
+class RoutstrFee(SQLModel, table=True):  # type: ignore
+    __tablename__ = "routstr_fees"
+    id: int = Field(default=1, primary_key=True)
+    accumulated_msats: int = Field(default=0)
+    total_paid_msats: int = Field(default=0)
+    last_paid_at: int | None = Field(default=None)
+
+
+async def accumulate_routstr_fee(session: AsyncSession, amount_msats: int) -> None:
+    stmt = (
+        update(RoutstrFee)
+        .where(col(RoutstrFee.id) == 1)
+        .values(accumulated_msats=RoutstrFee.accumulated_msats + amount_msats)
+    )
+    result = await session.exec(stmt)  # type: ignore[call-overload]
+    if result.rowcount == 0:
+        session.add(RoutstrFee(id=1, accumulated_msats=amount_msats))
+    await session.commit()
+
+
+async def get_routstr_fee(session: AsyncSession) -> RoutstrFee:
+    fee = await session.get(RoutstrFee, 1)
+    if fee is None:
+        fee = RoutstrFee(id=1, accumulated_msats=0, total_paid_msats=0)
+        session.add(fee)
+        await session.commit()
+        await session.refresh(fee)
+    return fee
+
+
+async def reset_routstr_fee(session: AsyncSession, paid_msats: int) -> None:
+    stmt = (
+        update(RoutstrFee)
+        .where(col(RoutstrFee.id) == 1)
+        .values(
+            accumulated_msats=RoutstrFee.accumulated_msats - paid_msats,
+            total_paid_msats=RoutstrFee.total_paid_msats + paid_msats,
+            last_paid_at=int(time.time()),
+        )
+    )
+    await session.exec(stmt)  # type: ignore[call-overload]
+    await session.commit()
+
+
 async def balances_for_mint_and_unit(
     db_session: AsyncSession, mint_url: str, unit: str
 ) -> int:

routstr/core/main.py

@@ -22,7 +22,7 @@ from ..payment.models import models_router, update_sats_pricing
 from ..payment.price import update_prices_periodically
 from ..proxy import initialize_upstreams, proxy_router, refresh_model_maps_periodically
 from ..upstream.auto_topup import periodic_auto_topup
-from ..wallet import periodic_payout, periodic_refund_sweep
+from ..wallet import periodic_payout, periodic_refund_sweep, periodic_routstr_fee_payout
 from .admin import admin_router
 from .db import create_session, init_db, run_migrations
 from .exceptions import general_exception_handler, http_exception_handler
@@ -56,6 +56,7 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
     key_reset_task = None
     auto_topup_task = None
     refund_sweep_task = None
+    routstr_fee_task = None
 
     try:
         # Run database migrations on startup
@@ -115,6 +116,7 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
         key_reset_task = asyncio.create_task(periodic_key_reset())
         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())
 
         yield
 
@@ -152,6 +154,8 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
             auto_topup_task.cancel()
         if refund_sweep_task is not None:
             refund_sweep_task.cancel()
+        if routstr_fee_task is not None:
+            routstr_fee_task.cancel()
 
         try:
             tasks_to_wait = []
@@ -177,6 +181,8 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
                 tasks_to_wait.append(auto_topup_task)
             if refund_sweep_task is not None:
                 tasks_to_wait.append(refund_sweep_task)
+            if routstr_fee_task is not None:
+                tasks_to_wait.append(routstr_fee_task)
 
             if tasks_to_wait:
                 await asyncio.gather(*tasks_to_wait, return_exceptions=True)

routstr/payment/models.py

@@ -7,7 +7,7 @@ from fastapi import APIRouter, Depends
 from pydantic.v1 import BaseModel
 from sqlmodel.ext.asyncio.session import AsyncSession
 
-from ..core.db import ModelRow, get_session
+from ..core.db import ModelRow, UpstreamProviderRow, get_session
 from ..core.logging import get_logger
 from ..core.settings import settings
 from .price import sats_usd_price
@@ -405,6 +405,76 @@ async def update_sats_pricing() -> None:
             logger.error(f"Error updating sats pricing: {e}")
 
 
+class ModelTestRequest(BaseModel):
+    model_id: str
+    endpoint_type: str
+    request_data: dict
+
+
+@models_router.post("/api/models/test")
+async def test_model(
+    payload: ModelTestRequest,
+    session: AsyncSession = Depends(get_session),
+) -> dict:
+    """Test a model by sending a request through its configured upstream provider."""
+    from sqlmodel import select
+
+    result = await session.execute(
+        select(ModelRow).where(ModelRow.id == payload.model_id)
+    )
+    model_row = result.scalars().first()
+
+    if not model_row:
+        return {
+            "success": False,
+            "error": f"Model '{payload.model_id}' not found in database",
+            "status_code": 404,
+        }
+
+    provider = await session.get(UpstreamProviderRow, model_row.upstream_provider_id)
+    if not provider:
+        return {
+            "success": False,
+            "error": "Upstream provider not found",
+            "status_code": 404,
+        }
+
+    base_url = provider.base_url.rstrip("/")
+    if payload.endpoint_type == "chat-completions":
+        url = f"{base_url}/chat/completions"
+    else:
+        url = f"{base_url}/{payload.endpoint_type}"
+
+    actual_model_id = model_row.forwarded_model_id or model_row.id
+    request_data = dict(payload.request_data)
+    request_data["model"] = actual_model_id
+
+    headers = {
+        "Content-Type": "application/json",
+        "Authorization": f"Bearer {provider.api_key}",
+    }
+
+    try:
+        async with httpx.AsyncClient(timeout=30.0) as client:
+            response = await client.post(url, json=request_data, headers=headers)
+            try:
+                response_data = response.json()
+            except Exception:
+                response_data = {"raw": response.text}
+
+            return {
+                "success": response.status_code < 400,
+                "data": response_data,
+                "status_code": response.status_code,
+            }
+    except Exception as e:
+        return {
+            "success": False,
+            "error": str(e),
+            "status_code": 500,
+        }
+
+
 @models_router.get("/v1/models")
 @models_router.get("/v1/models/", include_in_schema=False)
 @models_router.get("/models")

routstr/proxy.py

@@ -69,7 +69,25 @@ def get_upstreams() -> list[BaseUpstreamProvider]:
 
 def get_model_instance(model_id: str) -> Model | None:
     """Get Model instance by ID from global cache."""
-    return _model_instances.get(model_id.lower())
+    if not model_id:
+        return None
+
+    model_id_lower = model_id.lower()
+    # Try exact match first
+    if model := _model_instances.get(model_id_lower):
+        return model
+
+    # Try stripping common version suffixes (e.g., -20251222)
+    # This handles cases where upstream returns a specific version
+    # but we only track the base model name.
+    import re
+
+    base_model_id = re.sub(r"-\d{8}$", "", model_id_lower)
+    if base_model_id != model_id_lower:
+        if model := _model_instances.get(base_model_id):
+            return model
+
+    return None
 
 
 def get_provider_for_model(model_id: str) -> list[BaseUpstreamProvider] | None:

routstr/upstream/base.py

@@ -119,6 +119,51 @@ class BaseUpstreamProvider:
             "can_show_balance": False,
         }
 
+    def inject_cost_metadata(
+        self,
+        response_json: dict,
+        cost_data: CostData | MaxCostData | dict,
+        key: ApiKey,
+    ) -> None:
+        """Unifies the injection of cost and usage metadata across all completion types."""
+        if isinstance(cost_data, dict):
+            total_msats = cost_data.get("total_msats", 0)
+            total_usd = cost_data.get("total_usd", 0.0)
+            cost_dict = cost_data
+        else:
+            total_msats = cost_data.total_msats
+            total_usd = cost_data.total_usd
+            cost_dict = cost_data.dict()
+
+        sats_cost = total_msats // 1000
+
+        # Inject into top-level usage block (OpenAI/Anthropic style)
+        if "usage" in response_json:
+            response_json["usage"]["cost"] = total_usd
+            response_json["usage"]["cost_sats"] = sats_cost
+            response_json["usage"]["remaining_balance_msats"] = key.balance
+
+        # Inject into Anthropic nested usage block if present
+        if (
+            "message" in response_json
+            and isinstance(response_json["message"], dict)
+            and "usage" in response_json["message"]
+        ):
+            response_json["message"]["usage"]["sats_cost"] = sats_cost
+
+        # Unified Routstr metadata
+        response_json["metadata"] = response_json.get("metadata", {})
+        response_json["metadata"]["routstr"] = {
+            "cost": cost_dict,
+            "sats_cost": sats_cost,
+            "remaining_balance_msats": key.balance,
+        }
+
+        # Legacy/Compatibility fields
+        response_json["cost"] = cost_dict.copy()
+        response_json["cost"]["sats_cost"] = sats_cost
+        response_json["cost"]["remaining_balance_msats"] = key.balance
+
     def prepare_headers(self, request_headers: dict) -> dict:
         """Prepare headers for upstream request by removing proxy-specific headers and adding authentication.
 
@@ -376,75 +421,83 @@ class BaseUpstreamProvider:
         """
         pass
 
-    async def map_upstream_error_response(
+    async def forward_upstream_error_response(
         self, request: Request, path: str, upstream_response: httpx.Response
     ) -> Response:
-        """Map upstream error responses to appropriate proxy error responses.
-
-        Args:
-            request: Original FastAPI request
-            path: Request path
-            upstream_response: Response from upstream service
-
-        Returns:
-            Mapped error response with appropriate status code and error type
-        """
+        """Log upstream errors and forward the upstream response unchanged."""
         status_code = upstream_response.status_code
         headers = dict(upstream_response.headers)
-        content_type = headers.get("content-type", "")
+        content_type = headers.get("content-type") or headers.get("Content-Type", "")
+        upstream_request_id = (
+            headers.get("request-id")
+            or headers.get("Request-Id")
+            or headers.get("x-request-id")
+            or headers.get("X-Request-Id")
+            or headers.get("anthropic-request-id")
+            or headers.get("openai-request-id")
+        )
+
+        body_read_error = None
         try:
             body_bytes = await upstream_response.aread()
-        except Exception:
+        except Exception as exc:
             body_bytes = b""
+            body_read_error = f"{type(exc).__name__}: {exc}"
 
         message, upstream_code = self._extract_upstream_error_message(body_bytes)
-        lowered_message = message.lower()
-        lowered_code = (upstream_code or "").lower()
+        body_preview = body_bytes.decode("utf-8", errors="ignore").strip()[:500]
 
-        error_type = "upstream_error"
-        mapped_status = 502
-
-        if status_code in (400, 422):
-            error_type = "invalid_request_error"
-            mapped_status = 400
-        elif status_code in (401, 403):
-            error_type = "upstream_auth_error"
-            mapped_status = 502
-        elif status_code == 404:
-            if path.endswith("chat/completions"):
-                error_type = "invalid_model"
-                mapped_status = 400
-                if not message or message == "Upstream request failed":
-                    message = "Requested model is not available upstream"
-            elif "model" in lowered_message or "model" in lowered_code:
-                error_type = "invalid_model"
-                mapped_status = 400
-                if not message or message == "Upstream request failed":
-                    message = "Requested model is not available upstream"
-            else:
-                error_type = "upstream_error"
-                mapped_status = 502
-        elif status_code == 429:
-            error_type = "rate_limit_exceeded"
-            mapped_status = 429
-        elif status_code >= 500:
-            error_type = "upstream_error"
-            mapped_status = 502
-
-        logger.debug(
-            "Mapped upstream error",
+        logger.warning(
+            "Forwarding upstream error response as-is",
             extra={
                 "path": path,
+                "provider": self.provider_type,
                 "upstream_status": status_code,
-                "mapped_status": mapped_status,
-                "error_type": error_type,
+                "upstream_code": upstream_code,
                 "upstream_content_type": content_type,
+                "upstream_request_id": upstream_request_id,
                 "message_preview": message[:200],
+                "body_preview": body_preview,
+                "body_read_error": body_read_error,
+                "method": request.method,
             },
         )
 
-        return create_error_response(
-            error_type, message, mapped_status, request=request
+        for header_name in (
+            "content-length",
+            "Content-Length",
+            "transfer-encoding",
+            "Transfer-Encoding",
+            "content-encoding",
+            "Content-Encoding",
+            "connection",
+            "Connection",
+            "keep-alive",
+            "Keep-Alive",
+            "proxy-authenticate",
+            "Proxy-Authenticate",
+            "proxy-authorization",
+            "Proxy-Authorization",
+            "te",
+            "TE",
+            "trailer",
+            "Trailer",
+            "upgrade",
+            "Upgrade",
+        ):
+            headers.pop(header_name, None)
+
+        if not content_type:
+            headers.pop("content-type", None)
+            headers.pop("Content-Type", None)
+
+        media_type = content_type or None
+
+        return Response(
+            content=body_bytes,
+            status_code=status_code,
+            headers=headers,
+            media_type=media_type,
         )
 
     async def handle_streaming_chat_completion(
@@ -518,24 +571,32 @@ class BaseUpstreamProvider:
                             continue
 
                         try:
-                            obj = json.loads(part)
-                            if isinstance(obj, dict):
-                                if obj.get("model"):
-                                    last_model_seen = str(obj.get("model"))
-
-                                if requested_model:
-                                    obj["model"] = requested_model
-
-                                if "id" not in obj or not isinstance(obj["id"], str):
-                                    obj["id"] = f"chatcmpl-{uuid.uuid4()}"
-
-                                if isinstance(obj.get("usage"), dict):
-                                    # Hold this chunk back to merge cost later
-                                    usage_chunk_data = obj
+                            # Only parse if it looks like a JSON object to avoid SSE control messages or partials
+                            if part.strip().startswith(b"{") and part.strip().endswith(
+                                b"}"
+                            ):
+                                obj = json.loads(part)
+                                if isinstance(obj, dict):
+                                    if obj.get("model"):
+                                        last_model_seen = str(obj.get("model"))
+                                    if requested_model:
+                                        obj["model"] = requested_model
+                                    if (
+                                        "id" not in obj
+                                        or not isinstance(obj["id"], str)
+                                        or obj["id"] == "existing-id"
+                                    ):
+                                        if not hasattr(self, "_current_stream_id"):
+                                            self._current_stream_id = (
+                                                f"chatcmpl-{uuid.uuid4()}"
+                                            )
+                                        obj["id"] = self._current_stream_id
+                                    if isinstance(obj.get("usage"), dict):
+                                        usage_chunk_data = obj
+                                        continue
+                                    yield b"data: " + json.dumps(obj).encode() + b"\n\n"
                                     continue
-                                yield b"data: " + json.dumps(obj).encode() + b"\n\n"
-                                continue
-                        except json.JSONDecodeError:
+                        except Exception:
                             pass
 
                         prefix = (
@@ -1131,7 +1192,11 @@ class BaseUpstreamProvider:
                 )
 
     async def handle_streaming_messages_completion(
-        self, response: httpx.Response, key: ApiKey, max_cost_for_model: int
+        self,
+        response: httpx.Response,
+        key: ApiKey,
+        max_cost_for_model: int,
+        requested_model: str | None = None,
     ) -> StreamingResponse:
         async def stream_with_cost(
             max_cost_for_model: int,
@@ -1170,6 +1235,8 @@ class BaseUpstreamProvider:
                     stored_chunks.append(chunk)
                     try:
                         decoded_chunk = chunk.decode("utf-8", errors="ignore")
+                        modified_lines = []
+                        changed = False
                         for line in decoded_chunk.split("\n"):
                             if line.startswith("data: "):
                                 try:
@@ -1179,6 +1246,20 @@ class BaseUpstreamProvider:
                                         if msg and msg.get("model"):
                                             last_model_seen = str(msg.get("model"))
 
+                                        if requested_model:
+                                            # Apply requested_model override
+                                            model_updated = False
+                                            if msg:
+                                                msg["model"] = requested_model
+                                                model_updated = True
+                                            if data.get("model"):
+                                                data["model"] = requested_model
+                                                model_updated = True
+
+                                            if model_updated:
+                                                line = "data: " + json.dumps(data)
+                                                changed = True
+
                                         if usage := msg.get("usage"):
                                             input_tokens += usage.get("input_tokens", 0)
                                             output_tokens += usage.get(
@@ -1192,10 +1273,14 @@ class BaseUpstreamProvider:
                                             )
                                 except json.JSONDecodeError:
                                     pass
-                    except Exception:
-                        pass
+                            modified_lines.append(line)
 
-                    yield chunk
+                        if changed:
+                            yield "\n".join(modified_lines).encode("utf-8")
+                        else:
+                            yield chunk
+                    except Exception:
+                        yield chunk
 
                 usage_data = {
                     "input_tokens": input_tokens,
@@ -1217,6 +1302,11 @@ class BaseUpstreamProvider:
                                     new_session,
                                     max_cost_for_model,
                                 )
+
+                                self.inject_cost_metadata(
+                                    combined_data, cost_data, fresh_key
+                                )
+
                                 usage_finalized = True
                                 yield f"event: cost\ndata: {json.dumps({'cost': cost_data})}\n\n".encode()
                             except Exception:
@@ -1256,11 +1346,22 @@ class BaseUpstreamProvider:
         session: AsyncSession,
         deducted_max_cost: int,
         path: str,
+        requested_model: str | None = None,
     ) -> Response:
         try:
             content = await response.aread()
             response_json = json.loads(content)
 
+            if requested_model:
+                if "model" in response_json:
+                    response_json["model"] = requested_model
+                if (
+                    "message" in response_json
+                    and isinstance(response_json["message"], dict)
+                    and "model" in response_json["message"]
+                ):
+                    response_json["message"]["model"] = requested_model
+
             if path.endswith("count_tokens") and "usage" not in response_json:
                 input_tokens = response_json.get("input_tokens", 0)
                 response_json["usage"] = {"input_tokens": input_tokens}
@@ -1268,7 +1369,8 @@ class BaseUpstreamProvider:
             cost_data = await adjust_payment_for_tokens(
                 key, response_json, session, deducted_max_cost
             )
-            response_json["cost"] = cost_data
+
+            self.inject_cost_metadata(response_json, cost_data, key)
 
             allowed_headers = {
                 "content-type",
@@ -1373,15 +1475,28 @@ class BaseUpstreamProvider:
                     stream=True,
                 )
 
-            logger.info(
-                "Received upstream response",
-                extra={
-                    "status_code": response.status_code,
-                    "path": path,
-                    "key_hash": key.hashed_key[:8] + "...",
-                    "content_type": response.headers.get("content-type", "unknown"),
-                },
-            )
+            if response.status_code != 200:
+                logger.error(
+                    "Received upstream response",
+                    extra={
+                        "reason_phrase": response.reason_phrase,
+                        "status_code": response.status_code,
+                        "path": path,
+                        "key_hash": key.hashed_key[:8] + "...",
+                        "content_type": response.headers.get("content-type", "unknown"),
+                    },
+                )
+            else:
+                logger.info(
+                    "Received upstream response",
+                    extra={
+                        "reason_phrase": response.reason_phrase,
+                        "status_code": response.status_code,
+                        "path": path,
+                        "key_hash": key.hashed_key[:8] + "...",
+                        "content_type": response.headers.get("content-type", "unknown"),
+                    },
+                )
 
             if response.status_code != 200:
                 if response.status_code >= 500:
@@ -1393,7 +1508,7 @@ class BaseUpstreamProvider:
                     )
 
                 try:
-                    mapped_error = await self.map_upstream_error_response(
+                    mapped_error = await self.forward_upstream_error_response(
                         request, path, response
                     )
                 finally:
@@ -1422,7 +1537,10 @@ class BaseUpstreamProvider:
 
                     if is_streaming and response.status_code == 200:
                         result = await self.handle_streaming_messages_completion(
-                            response, key, max_cost_for_model
+                            response,
+                            key,
+                            max_cost_for_model,
+                            requested_model=original_model_id,
                         )
                         background_tasks = BackgroundTasks()
                         background_tasks.add_task(response.aclose)
@@ -1433,7 +1551,12 @@ class BaseUpstreamProvider:
                     if response.status_code == 200:
                         try:
                             return await self.handle_non_streaming_messages_completion(
-                                response, key, session, max_cost_for_model, path
+                                response,
+                                key,
+                                session,
+                                max_cost_for_model,
+                                path,
+                                requested_model=original_model_id,
                             )
                         finally:
                             await response.aclose()
@@ -1443,7 +1566,12 @@ class BaseUpstreamProvider:
                     if response.status_code == 200:
                         try:
                             return await self.handle_non_streaming_messages_completion(
-                                response, key, session, max_cost_for_model, path
+                                response,
+                                key,
+                                session,
+                                max_cost_for_model,
+                                path,
+                                requested_model=original_model_id,
                             )
                         finally:
                             await response.aclose()
@@ -1687,7 +1815,7 @@ class BaseUpstreamProvider:
                     )
 
                 try:
-                    mapped_error = await self.map_upstream_error_response(
+                    mapped_error = await self.forward_upstream_error_response(
                         request, path, response
                     )
                 finally:
@@ -1859,7 +1987,7 @@ class BaseUpstreamProvider:
                 )
                 if response.status_code != 200:
                     try:
-                        mapped = await self.map_upstream_error_response(
+                        mapped = await self.forward_upstream_error_response(
                             request, path, response
                         )
                     finally:
@@ -2513,14 +2641,25 @@ class BaseUpstreamProvider:
                     stream=True,
                 )
 
-                logger.debug(
-                    "Received upstream response",
-                    extra={
-                        "status_code": response.status_code,
-                        "path": path,
-                        "response_headers": dict(response.headers),
-                    },
-                )
+                if response.status_code != 200:
+                    logger.error(
+                        "Received upstream response",
+                        extra={
+                            "reason_phrase": response.reason_phrase,
+                            "status_code": response.status_code,
+                            "path": path,
+                            "response_headers": dict(response.headers),
+                        },
+                    )
+                else:
+                    logger.debug(
+                        "Received upstream response",
+                        extra={
+                            "status_code": response.status_code,
+                            "path": path,
+                            "response_headers": dict(response.headers),
+                        },
+                    )
 
                 if response.status_code != 200:
                     logger.warning(

routstr/upstream/ppqai.py

@@ -1,9 +1,9 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Optional
 
 import httpx
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 
 from ..core.logging import get_logger
 from ..payment.models import Architecture, Model, Pricing, async_fetch_openrouter_models
@@ -16,18 +16,20 @@ logger = get_logger(__name__)
 
 
 class PPQAIModelPricing(BaseModel):
-    ui: dict[str, float]
-    api: dict[str, float]
+    ui: Optional[dict[str, float]] = None
+    api: Optional[dict[str, float]] = None
+    input_per_1M_tokens: Optional[float] = Field(None, alias="input_per_1M_tokens")
+    output_per_1M_tokens: Optional[float] = Field(None, alias="output_per_1M_tokens")
 
 
 class PPQAIModel(BaseModel):
     id: str
-    provider: str
+    provider: Optional[str] = None
     name: str
     created_at: int
     context_length: int
     pricing: PPQAIModelPricing
-    popular: bool
+    popular: bool = False
 
 
 class PPQAIUpstreamProvider(BaseUpstreamProvider):
@@ -134,31 +136,54 @@ class PPQAIUpstreamProvider(BaseUpstreamProvider):
                         )
 
                         if or_model:
-                            if input_price := ppqai_model.pricing.api.get(
-                                "input_per_1M"
-                            ):
+                            input_price = None
+                            if ppqai_model.pricing.api:
+                                input_price = ppqai_model.pricing.api.get(
+                                    "input_per_1M"
+                                )
+                            elif ppqai_model.pricing.input_per_1M_tokens:
+                                input_price = ppqai_model.pricing.input_per_1M_tokens
+
+                            if input_price is not None:
                                 or_model.pricing.prompt = input_price / 1_000_000
-                            if output_price := ppqai_model.pricing.api.get(
-                                "output_per_1M"
-                            ):
+
+                            output_price = None
+                            if ppqai_model.pricing.api:
+                                output_price = ppqai_model.pricing.api.get(
+                                    "output_per_1M"
+                                )
+                            elif ppqai_model.pricing.output_per_1M_tokens:
+                                output_price = ppqai_model.pricing.output_per_1M_tokens
+
+                            if output_price is not None:
                                 or_model.pricing.completion = output_price / 1_000_000
+
                             if cl := ppqai_model.context_length:
                                 or_model.context_length = cl
                             models.append(or_model)
                         else:
-                            input_price = ppqai_model.pricing.api.get(
-                                "input_per_1M", 0.0
-                            )
-                            output_price = ppqai_model.pricing.api.get(
-                                "output_per_1M", 0.0
-                            )
+                            input_price = 0.0
+                            if ppqai_model.pricing.api:
+                                input_price = ppqai_model.pricing.api.get(
+                                    "input_per_1M", 0.0
+                                )
+                            elif ppqai_model.pricing.input_per_1M_tokens:
+                                input_price = ppqai_model.pricing.input_per_1M_tokens
+
+                            output_price = 0.0
+                            if ppqai_model.pricing.api:
+                                output_price = ppqai_model.pricing.api.get(
+                                    "output_per_1M", 0.0
+                                )
+                            elif ppqai_model.pricing.output_per_1M_tokens:
+                                output_price = ppqai_model.pricing.output_per_1M_tokens
 
                             models.append(
                                 Model(
                                     id=ppqai_model.id,
                                     name=ppqai_model.name,
                                     created=ppqai_model.created_at // 1000,
-                                    description=f"{ppqai_model.provider} model",
+                                    description=f"{ppqai_model.provider or 'PPQ.AI'} model",
                                     context_length=ppqai_model.context_length,
                                     architecture=Architecture(
                                         modality="text->text",

routstr/wallet.py

@@ -155,6 +155,20 @@ async def swap_to_primary_mint(
         raise ValueError("Invalid unit")
     primary_wallet = await get_wallet(settings.primary_mint, settings.primary_mint_unit)
 
+    # If the token is already from the primary mint, we don't need to swap
+    # and we definitely don't want to calculate or pay fees.
+    if token_obj.mint == settings.primary_mint:
+        logger.info(
+            "swap_to_primary_mint: token already on primary mint, skipping swap",
+            extra={
+                "mint": token_obj.mint,
+                "amount": token_amount,
+                "unit": token_obj.unit,
+            },
+        )
+        await token_wallet.split(proofs=token_obj.proofs, amount=0, include_fees=True)
+        return token_amount, token_obj.unit, token_obj.mint
+
     minted_amount = await _calculate_swap_amount(
         amount_msat,
         token_obj.unit,
@@ -559,6 +573,46 @@ async def periodic_refund_sweep() -> None:
             )
 
 
+async def periodic_routstr_fee_payout() -> None:
+    from .auth import (
+        ROUTSTR_FEE_DEFAULT_PAYOUT,
+        ROUTSTR_FEE_PAYOUT_INTERVAL_SECONDS,
+        ROUTSTR_LN_ADDRESS,
+    )
+
+    if not ROUTSTR_LN_ADDRESS:
+        logger.info("ROUTSTR_LN_ADDRESS not set, skipping fee payout")
+        return
+    while True:
+        await asyncio.sleep(ROUTSTR_FEE_PAYOUT_INTERVAL_SECONDS)
+        try:
+            async with db.create_session() as session:
+                fee = await db.get_routstr_fee(session)
+                accumulated_sats = fee.accumulated_msats // 1000
+                if accumulated_sats >= ROUTSTR_FEE_DEFAULT_PAYOUT:
+                    wallet = await get_wallet(settings.primary_mint, "sat")
+                    proofs = get_proofs_per_mint_and_unit(
+                        wallet, settings.primary_mint, "sat", not_reserved=True
+                    )
+                    amount_received = await raw_send_to_lnurl(
+                        wallet, proofs, ROUTSTR_LN_ADDRESS, "sat", amount=accumulated_sats
+                    )
+                    paid_msats = accumulated_sats * 1000
+                    await db.reset_routstr_fee(session, paid_msats)
+                    logger.info(
+                        "Routstr fee payout sent",
+                        extra={
+                            "accumulated_sats": accumulated_sats,
+                            "amount_received": amount_received,
+                        },
+                    )
+        except Exception as e:
+            logger.error(
+                f"Error in Routstr fee payout: {type(e).__name__}",
+                extra={"error": str(e)},
+            )
+
+
 async def send_to_lnurl(amount: int, unit: str, mint: str, address: str) -> int:
     wallet = await get_wallet(mint, unit)
     proofs = wallet._get_proofs_per_keyset(wallet.proofs)[wallet.keyset_id]

tests/unit/test_stream_id_injection.py

@@ -0,0 +1,109 @@
+import json
+from collections.abc import AsyncGenerator
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from routstr.core.db import ApiKey
+from routstr.upstream.base import BaseUpstreamProvider
+
+
+@pytest.mark.asyncio
+async def test_stream_with_id_injection() -> None:
+    """Test that stream_with_cost correctly injects IDs into complete JSON chunks but skips partials."""
+    provider = BaseUpstreamProvider(
+        base_url="https://api.example.com", api_key="test_key"
+    )
+
+    # Mock response with mixed chunks:
+    # 1. Complete JSON without ID
+    # 2. Partial JSON (should be passed through)
+    # 3. Complete JSON with ID (should be preserved or updated if requested_model is set)
+    # 4. [DONE] message
+    chunks = [
+        b'data: {"choices": [{"delta": {"content": "Hello"}}]}\n\n',
+        b'data: {"choices": [{"delta": {"content": "',  # Partial
+        b'world"}}]}\n\n',
+        b'data: {"id": "existing-id", "choices": [{"delta": {"content": "!"}}]}\n\n',
+        b"data: [DONE]\n\n",
+    ]
+
+    async def aiter_bytes() -> AsyncGenerator[bytes, None]:
+        for chunk in chunks:
+            yield chunk
+
+    mock_response = MagicMock()
+    mock_response.status_code = 200
+    mock_response.headers = {"content-type": "text/event-stream"}
+    mock_response.aiter_bytes = aiter_bytes
+
+    key = MagicMock(spec=ApiKey)
+    key.hashed_key = "test_hash"
+    key.balance = 1000
+
+    background_tasks = MagicMock()
+
+    # We need to mock adjust_payment_for_tokens since it's called at the end
+    with MagicMock():
+        from routstr.upstream import base
+
+        # Mocking the module-level function used in the generator
+        base.adjust_payment_for_tokens = AsyncMock(
+            return_value={"total_usd": 0.1, "total_msats": 100}
+        )
+        base.create_session = MagicMock()
+
+        streaming_response = await provider.handle_streaming_chat_completion(
+            response=mock_response,
+            key=key,
+            max_cost_for_model=100,
+            background_tasks=background_tasks,
+            requested_model="test-model",
+        )
+
+        results = []
+        async for chunk in streaming_response.body_iterator:
+            results.append(chunk)
+
+        # Parse results
+        parsed_results = []
+        for r in results:
+            if isinstance(r, bytes) and r.startswith(b"data: "):
+                data = r[6:].decode().strip()
+                if data == "[DONE]":
+                    parsed_results.append(data)
+                else:
+                    try:
+                        parsed_results.append(json.loads(data))
+                    except (json.JSONDecodeError, UnicodeDecodeError):
+                        parsed_results.append(
+                            data
+                        )  # Keep as string if it failed to parse
+
+        # Verifications
+        # 1. First chunk should have an injected ID and the requested model
+        assert isinstance(parsed_results[0], dict)
+        assert "id" in parsed_results[0]
+        assert parsed_results[0]["id"].startswith("chatcmpl-")
+        assert parsed_results[0]["model"] == "test-model"
+
+        # 2. Second chunk was partial, should be passed as-is
+        # In current implementation, re.split(b"data: ", b'data: {...') gives ['', '{...']
+        # The first empty part is skipped. The second part is processed.
+
+        # Check that we have results
+        assert len(parsed_results) >= 4
+
+        # Find the chunk that was "existing-id"
+        id_chunk = next(
+            r
+            for r in parsed_results
+            if isinstance(r, dict)
+            and "choices" in r
+            and r["choices"][0]["delta"].get("content") == "!"
+        )
+        assert id_chunk["id"] == parsed_results[0]["id"]
+        assert id_chunk["model"] == "test-model"
+
+        # 4. [DONE] should be there
+        assert "[DONE]" in parsed_results

tests/unit/test_wallet.py

@@ -220,6 +220,33 @@ async def test_recieve_token_untrusted_mint() -> None:
 
 
 @pytest.mark.asyncio
+@pytest.mark.asyncio
+async def test_swap_to_primary_mint_already_on_primary() -> None:
+    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.amount = 1000
+    mock_token.unit = "sat"
+    mock_token.proofs = []
+
+    mock_token_wallet = Mock()
+    mock_token_wallet.split = AsyncMock(return_value=None)
+    mock_token_wallet.request_mint = AsyncMock()
+    mock_token_wallet.melt_quote = AsyncMock()
+
+    with patch("routstr.wallet.get_wallet", AsyncMock(return_value=mock_token_wallet)):
+        amount, unit, mint = await swap_to_primary_mint(mock_token, mock_token_wallet)
+
+    assert amount == 1000
+    assert unit == "sat"
+    assert mint == settings.primary_mint
+    mock_token_wallet.split.assert_called_once()
+    mock_token_wallet.request_mint.assert_not_called()
+    mock_token_wallet.melt_quote.assert_not_called()
+
+
 async def test_swap_to_primary_mint_success() -> None:
     """Test successful swap with dynamic fee calculation."""
     from routstr.wallet import swap_to_primary_mint

ui/components/api-endpoint-tester.tsx

@@ -471,7 +471,11 @@ export function ApiEndpointTester({ models }: ApiEndpointTesterProps) {
     testEndpointMutation.mutate(requestData);
   };
 
-  const enabledModels = models.filter((model) => model.isEnabled);
+  const enabledModels = Array.from(
+    new Map(
+      models.filter((model) => model.isEnabled).map((m) => [m.id, m])
+    ).values()
+  );
   const credentials = selectedModel ? getModelCredentials(selectedModel) : null;
   const endpointUrl = credentials
     ? buildEndpointUrl(

ui/components/model-tester.tsx

@@ -197,7 +197,11 @@ export function ModelTester({ models }: ModelTesterProps) {
     testModelMutation.mutate(request);
   };
 
-  const enabledModels = models.filter((model) => model.isEnabled);
+  const enabledModels = Array.from(
+    new Map(
+      models.filter((model) => model.isEnabled).map((m) => [m.id, m])
+    ).values()
+  );
   const credentials = selectedModel ? getModelCredentials(selectedModel) : null;
 
   return (

ui/components/models-page.tsx

@@ -124,12 +124,14 @@ export function ModelsPage() {
             >
               Basic Testing
             </TabsTrigger>
+            {/*
             <TabsTrigger
               value='test-api'
               className='h-9 snap-start px-2 text-[13px] sm:h-10 sm:px-2.5 sm:text-sm'
             >
               API Endpoints
             </TabsTrigger>
+            */}
           </TabsList>
 
           <TabsContent value='manage' className='mt-0'>