feat: fallback to remote dashboard if not selfhosted

- Moves UI serving logic to `routstr/core/ui.py`.
- Implements a fallback mechanism: if the local UI build is missing (specifically index.html), it acts as a reverse proxy to a configured `fallback_ui_url` (default: https://api.routstr.com).
- Adds `FALLBACK_UI_URL` environment variable support.
- Ensures the application remains usable even without a local UI build.

routstr/core/main.py

@@ -1,13 +1,11 @@
 import asyncio
 import os
 from contextlib import asynccontextmanager
-from pathlib import Path
 from typing import AsyncGenerator
 
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
-from fastapi.responses import FileResponse, RedirectResponse
-from fastapi.staticfiles import StaticFiles
+from fastapi.responses import RedirectResponse
 from starlette.exceptions import HTTPException
 
 from ..balance import balance_router, deprecated_wallet_router
@@ -27,6 +25,7 @@ from .logging import get_logger, setup_logging
 from .middleware import LoggingMiddleware
 from .settings import SettingsService
 from .settings import settings as global_settings
+from .ui import setup_ui
 
 # Initialize logging first
 setup_logging()
@@ -190,143 +189,8 @@ async def providers() -> RedirectResponse:
     return RedirectResponse("/v1/providers/")
 
 
-UI_DIST_PATH = Path(__file__).parent.parent.parent / "ui_out"
-
-if UI_DIST_PATH.exists() and UI_DIST_PATH.is_dir():
-    logger.info(f"Serving static UI from {UI_DIST_PATH}")
-
-    app.mount(
-        "/_next",
-        StaticFiles(directory=UI_DIST_PATH / "_next", check_dir=True),
-        name="next-static",
-    )
-
-    @app.get("/", include_in_schema=False)
-    async def serve_root_ui() -> FileResponse:
-        return FileResponse(UI_DIST_PATH / "index.html")
-
-    # Add explicit route for /index.txt to redirect to /
-    @app.get("/index.txt", include_in_schema=False)
-    async def redirect_index_txt() -> RedirectResponse:
-        return RedirectResponse("/")
-
-    @app.get("/admin")
-    async def admin_redirect() -> FileResponse:
-        return FileResponse(UI_DIST_PATH / "index.html")
-
-    @app.get("/dashboard", include_in_schema=False)
-    async def serve_dashboard_ui() -> FileResponse:
-        return FileResponse(UI_DIST_PATH / "index.html")
-
-    @app.get("/login", include_in_schema=False)
-    async def serve_login_ui() -> FileResponse:
-        return FileResponse(UI_DIST_PATH / "login" / "index.html")
-
-    # Add explicit route for /login/index.txt to redirect to /login
-    @app.get("/login/index.txt", include_in_schema=False)
-    async def redirect_login_index_txt() -> RedirectResponse:
-        return RedirectResponse("/login")
-
-    @app.get("/model", include_in_schema=False)
-    async def serve_models_ui() -> FileResponse:
-        return FileResponse(UI_DIST_PATH / "model" / "index.html")
-
-    # Add explicit route for /model/index.txt to redirect to /model
-    @app.get("/model/index.txt", include_in_schema=False)
-    async def redirect_model_index_txt() -> RedirectResponse:
-        return RedirectResponse("/model")
-
-    @app.get("/providers", include_in_schema=False)
-    async def serve_providers_ui() -> FileResponse:
-        return FileResponse(UI_DIST_PATH / "providers" / "index.html")
-
-    # Add explicit route for /providers/index.txt to redirect to /providers
-    @app.get("/providers/index.txt", include_in_schema=False)
-    async def redirect_providers_index_txt() -> RedirectResponse:
-        return RedirectResponse("/providers")
-
-    @app.get("/settings", include_in_schema=False)
-    async def serve_settings_ui() -> FileResponse:
-        return FileResponse(UI_DIST_PATH / "settings" / "index.html")
-
-    # Add explicit route for /settings/index.txt to redirect to /settings
-    @app.get("/settings/index.txt", include_in_schema=False)
-    async def redirect_settings_index_txt() -> RedirectResponse:
-        return RedirectResponse("/settings")
-
-    @app.get("/transactions", include_in_schema=False)
-    async def serve_transactions_ui() -> FileResponse:
-        return FileResponse(UI_DIST_PATH / "transactions" / "index.html")
-
-    # Add explicit route for /transactions/index.txt to redirect to /transactions
-    @app.get("/transactions/index.txt", include_in_schema=False)
-    async def redirect_transactions_index_txt() -> RedirectResponse:
-        return RedirectResponse("/transactions")
-
-    @app.get("/balances", include_in_schema=False)
-    async def serve_balances_ui() -> FileResponse:
-        return FileResponse(UI_DIST_PATH / "balances" / "index.html")
-
-    # Add explicit route for /balances/index.txt to redirect to /balances
-    @app.get("/balances/index.txt", include_in_schema=False)
-    async def redirect_balances_index_txt() -> RedirectResponse:
-        return RedirectResponse("/balances")
-
-    @app.get("/logs", include_in_schema=False)
-    async def serve_logs_ui() -> FileResponse:
-        return FileResponse(UI_DIST_PATH / "logs" / "index.html")
-
-    # Add explicit route for /logs/index.txt to redirect to /logs
-    @app.get("/logs/index.txt", include_in_schema=False)
-    async def redirect_logs_index_txt() -> RedirectResponse:
-        return RedirectResponse("/logs")
-
-    @app.get("/usage", include_in_schema=False)
-    async def serve_usage_ui() -> FileResponse:
-        return FileResponse(UI_DIST_PATH / "usage" / "index.html")
-
-    # Add explicit route for /usage/index.txt to redirect to /usage
-    @app.get("/usage/index.txt", include_in_schema=False)
-    async def redirect_usage_index_txt() -> RedirectResponse:
-        return RedirectResponse("/usage")
-
-    @app.get("/unauthorized", include_in_schema=False)
-    async def serve_unauthorized_ui() -> FileResponse:
-        return FileResponse(UI_DIST_PATH / "unauthorized" / "index.html")
-
-    # Add explicit route for /unauthorized/index.txt to redirect to /unauthorized
-    @app.get("/unauthorized/index.txt", include_in_schema=False)
-    async def redirect_unauthorized_index_txt() -> RedirectResponse:
-        return RedirectResponse("/unauthorized")
-
-    @app.get("/favicon.ico", include_in_schema=False)
-    async def serve_favicon() -> FileResponse:
-        icon_path = UI_DIST_PATH / "icon.ico"
-        if icon_path.exists():
-            return FileResponse(icon_path)
-        return FileResponse(UI_DIST_PATH / "favicon.ico")
-
-    @app.get("/icon.ico", include_in_schema=False)
-    async def serve_icon() -> FileResponse:
-        return FileResponse(UI_DIST_PATH / "icon.ico")
-
-    app.mount(
-        "/static", StaticFiles(directory=UI_DIST_PATH, check_dir=True), name="ui-static"
-    )
-else:
-    logger.warning(
-        f"UI dist directory not found at {UI_DIST_PATH}, skipping static file serving"
-    )
-
-    @app.get("/", include_in_schema=False)
-    async def root_fallback() -> dict:
-        return {
-            "name": global_settings.name,
-            "description": global_settings.description,
-            "version": __version__,
-            "status": "running",
-            "ui": "not available",
-        }
+# Setup UI routes (local serving or proxy fallback)
+setup_ui(app, __version__)
 
 
 app.include_router(models_router)

routstr/core/settings.py

@@ -6,7 +6,7 @@ import os
 from datetime import datetime, timezone
 from typing import Any
 
-from pydantic.v1 import BaseModel, BaseSettings, Field, validator
+from pydantic.v1 import BaseModel, BaseSettings, Field
 from sqlmodel.ext.asyncio.session import AsyncSession
 
 
@@ -37,13 +37,6 @@ class Settings(BaseSettings):
 
     # Cashu
     cashu_mints: list[str] = Field(default_factory=list, env="CASHU_MINTS")
-
-    @validator("cashu_mints", pre=True, each_item=True)
-    def normalize_mint_url(cls, v: str) -> str:
-        if isinstance(v, str):
-            return v.rstrip("/")
-        return v
-
     receive_ln_address: str = Field(default="", env="RECEIVE_LN_ADDRESS")
     primary_mint: str = Field(default="", env="PRIMARY_MINT_URL")
     primary_mint_unit: str = Field(default="sat", env="PRIMARY_MINT_UNIT")
@@ -96,6 +89,9 @@ class Settings(BaseSettings):
     # Discovery
     relays: list[str] = Field(default_factory=list, env="RELAYS")
 
+    # Cloud UI
+    fallback_ui_url: str = Field(default="https://api.routstr.com", env="FALLBACK_UI_URL")
+
 
 def _compute_primary_mint(cashu_mints: list[str]) -> str:
     return cashu_mints[0] if cashu_mints else "https://mint.minibits.cash/Bitcoin"

routstr/core/ui.py

@@ -0,0 +1,252 @@
+from pathlib import Path
+from typing import Any
+
+import httpx
+from fastapi import FastAPI, Request
+from fastapi.responses import FileResponse, RedirectResponse, StreamingResponse
+from fastapi.staticfiles import StaticFiles
+
+from .logging import get_logger
+from .settings import settings as global_settings
+
+logger = get_logger(__name__)
+
+
+def setup_ui(app: FastAPI, version: str) -> None:
+    UI_DIST_PATH = Path(__file__).parent.parent.parent / "ui_out"
+
+    # Check if we have a valid local UI build
+    # We require at least the index.html to be present because the directory might exist
+    # but be empty (e.g. Docker volume mount before build completes)
+    has_local_ui = (
+        UI_DIST_PATH.exists()
+        and UI_DIST_PATH.is_dir()
+        and (UI_DIST_PATH / "index.html").exists()
+    )
+
+    if has_local_ui:
+        logger.info(f"Serving static UI from {UI_DIST_PATH}")
+
+        app.mount(
+            "/_next",
+            StaticFiles(directory=UI_DIST_PATH / "_next", check_dir=True),
+            name="next-static",
+        )
+
+        @app.get("/", include_in_schema=False)
+        async def serve_root_ui() -> FileResponse:
+            return FileResponse(UI_DIST_PATH / "index.html")
+
+        # Add explicit route for /index.txt to redirect to /
+        @app.get("/index.txt", include_in_schema=False)
+        async def redirect_index_txt() -> RedirectResponse:
+            return RedirectResponse("/")
+
+        @app.get("/admin", include_in_schema=False)
+        async def admin_redirect() -> FileResponse:
+            return FileResponse(UI_DIST_PATH / "index.html")
+
+        @app.get("/dashboard", include_in_schema=False)
+        async def serve_dashboard_ui() -> FileResponse:
+            return FileResponse(UI_DIST_PATH / "index.html")
+
+        @app.get("/login", include_in_schema=False)
+        async def serve_login_ui() -> FileResponse:
+            return FileResponse(UI_DIST_PATH / "login" / "index.html")
+
+        # Add explicit route for /login/index.txt to redirect to /login
+        @app.get("/login/index.txt", include_in_schema=False)
+        async def redirect_login_index_txt() -> RedirectResponse:
+            return RedirectResponse("/login")
+
+        @app.get("/model", include_in_schema=False)
+        async def serve_models_ui() -> FileResponse:
+            return FileResponse(UI_DIST_PATH / "model" / "index.html")
+
+        # Add explicit route for /model/index.txt to redirect to /model
+        @app.get("/model/index.txt", include_in_schema=False)
+        async def redirect_model_index_txt() -> RedirectResponse:
+            return RedirectResponse("/model")
+
+        @app.get("/providers", include_in_schema=False)
+        async def serve_providers_ui() -> FileResponse:
+            return FileResponse(UI_DIST_PATH / "providers" / "index.html")
+
+        # Add explicit route for /providers/index.txt to redirect to /providers
+        @app.get("/providers/index.txt", include_in_schema=False)
+        async def redirect_providers_index_txt() -> RedirectResponse:
+            return RedirectResponse("/providers")
+
+        @app.get("/settings", include_in_schema=False)
+        async def serve_settings_ui() -> FileResponse:
+            return FileResponse(UI_DIST_PATH / "settings" / "index.html")
+
+        # Add explicit route for /settings/index.txt to redirect to /settings
+        @app.get("/settings/index.txt", include_in_schema=False)
+        async def redirect_settings_index_txt() -> RedirectResponse:
+            return RedirectResponse("/settings")
+
+        @app.get("/transactions", include_in_schema=False)
+        async def serve_transactions_ui() -> FileResponse:
+            return FileResponse(UI_DIST_PATH / "transactions" / "index.html")
+
+        # Add explicit route for /transactions/index.txt to redirect to /transactions
+        @app.get("/transactions/index.txt", include_in_schema=False)
+        async def redirect_transactions_index_txt() -> RedirectResponse:
+            return RedirectResponse("/transactions")
+
+        @app.get("/balances", include_in_schema=False)
+        async def serve_balances_ui() -> FileResponse:
+            return FileResponse(UI_DIST_PATH / "balances" / "index.html")
+
+        # Add explicit route for /balances/index.txt to redirect to /balances
+        @app.get("/balances/index.txt", include_in_schema=False)
+        async def redirect_balances_index_txt() -> RedirectResponse:
+            return RedirectResponse("/balances")
+
+        @app.get("/logs", include_in_schema=False)
+        async def serve_logs_ui() -> FileResponse:
+            return FileResponse(UI_DIST_PATH / "logs" / "index.html")
+
+        # Add explicit route for /logs/index.txt to redirect to /logs
+        @app.get("/logs/index.txt", include_in_schema=False)
+        async def redirect_logs_index_txt() -> RedirectResponse:
+            return RedirectResponse("/logs")
+
+        @app.get("/usage", include_in_schema=False)
+        async def serve_usage_ui() -> FileResponse:
+            return FileResponse(UI_DIST_PATH / "usage" / "index.html")
+
+        # Add explicit route for /usage/index.txt to redirect to /usage
+        @app.get("/usage/index.txt", include_in_schema=False)
+        async def redirect_usage_index_txt() -> RedirectResponse:
+            return RedirectResponse("/usage")
+
+        @app.get("/unauthorized", include_in_schema=False)
+        async def serve_unauthorized_ui() -> FileResponse:
+            return FileResponse(UI_DIST_PATH / "unauthorized" / "index.html")
+
+        # Add explicit route for /unauthorized/index.txt to redirect to /unauthorized
+        @app.get("/unauthorized/index.txt", include_in_schema=False)
+        async def redirect_unauthorized_index_txt() -> RedirectResponse:
+            return RedirectResponse("/unauthorized")
+
+        @app.get("/favicon.ico", include_in_schema=False)
+        async def serve_favicon() -> FileResponse:
+            icon_path = UI_DIST_PATH / "icon.ico"
+            if icon_path.exists():
+                return FileResponse(icon_path)
+            return FileResponse(UI_DIST_PATH / "favicon.ico")
+
+        @app.get("/icon.ico", include_in_schema=False)
+        async def serve_icon() -> FileResponse:
+            return FileResponse(UI_DIST_PATH / "icon.ico")
+
+        app.mount(
+            "/static",
+            StaticFiles(directory=UI_DIST_PATH, check_dir=True),
+            name="ui-static",
+        )
+    else:
+        logger.info(
+            f"Local UI not found at {UI_DIST_PATH} (checked for index.html), falling back to cloud/proxy"
+        )
+        if global_settings.fallback_ui_url:
+            logger.info(f"Proxying UI to {global_settings.fallback_ui_url}")
+
+            async def _proxy_ui_request(request: Request, path: str = "") -> Any:
+                # Clean up the target URL
+                base_url = global_settings.fallback_ui_url.rstrip("/")
+                target_url = f"{base_url}/{path.lstrip('/')}"
+
+                headers = dict(request.headers)
+                headers.pop("host", None)
+                headers.pop("content-length", None)
+
+                client = httpx.AsyncClient()
+                try:
+                    rp_req = client.build_request("GET", target_url, headers=headers)
+                    rp_resp = await client.send(rp_req, stream=True)
+
+                    async def stream_response() -> Any:
+                        try:
+                            async for chunk in rp_resp.aiter_raw():
+                                yield chunk
+                        finally:
+                            await rp_resp.aclose()
+                            await client.aclose()
+
+                    response_headers = dict(rp_resp.headers)
+                    # Filter out hop-by-hop headers and others that shouldn't be proxied directly
+                    for key in [
+                        "transfer-encoding",
+                        "connection",
+                        "keep-alive",
+                        "host",
+                    ]:
+                        response_headers.pop(key, None)
+
+                    return StreamingResponse(
+                        stream_response(),
+                        status_code=rp_resp.status_code,
+                        headers=response_headers,
+                    )
+                except Exception as e:
+                    await client.aclose()
+                    logger.error(f"Failed to proxy UI request to {target_url}: {e}")
+                    return {"error": "UI Proxy Error", "details": str(e)}
+
+            @app.get("/", include_in_schema=False)
+            async def proxy_root_ui(request: Request) -> Any:
+                return await _proxy_ui_request(request, "")
+
+            @app.get("/_next/{path:path}", include_in_schema=False)
+            async def proxy_next_assets(request: Request, path: str) -> Any:
+                return await _proxy_ui_request(request, f"_next/{path}")
+
+            @app.get("/static/{path:path}", include_in_schema=False)
+            async def proxy_static_assets(request: Request, path: str) -> Any:
+                return await _proxy_ui_request(request, f"static/{path}")
+
+            # Proxy common assets that might be at root
+            @app.get("/favicon.ico", include_in_schema=False)
+            async def proxy_favicon(request: Request) -> Any:
+                return await _proxy_ui_request(request, "favicon.ico")
+
+            @app.get("/icon.ico", include_in_schema=False)
+            async def proxy_icon(request: Request) -> Any:
+                return await _proxy_ui_request(request, "icon.ico")
+
+            # SPA routes
+            for route in [
+                "/admin",
+                "/dashboard",
+                "/login",
+                "/model",
+                "/providers",
+                "/settings",
+                "/transactions",
+                "/balances",
+                "/logs",
+                "/usage",
+                "/unauthorized",
+            ]:
+
+                @app.get(route, include_in_schema=False)
+                async def proxy_spa_route(request: Request) -> Any:
+                    return await _proxy_ui_request(request, "")
+
+        else:
+            logger.warning(
+                f"UI dist directory not found at {UI_DIST_PATH} and no fallback_ui_url configured"
+            )
+
+            @app.get("/", include_in_schema=False)
+            async def root_fallback() -> dict[str, Any]:
+                return {
+                    "name": global_settings.name,
+                    "description": global_settings.description,
+                    "version": version,
+                    "status": "running",
+                    "ui": "not available",
+                }

routstr/proxy.py

@@ -16,7 +16,6 @@ from .core.db import (
     create_session,
     get_session,
 )
-from .core.settings import settings
 from .payment.helpers import (
     calculate_discounted_max_cost,
     check_token_balance,
@@ -26,7 +25,6 @@ from .payment.helpers import (
 from .payment.models import Model
 from .upstream import BaseUpstreamProvider
 from .upstream.helpers import init_upstreams
-from .wallet import deserialize_token_from_string
 
 logger = get_logger(__name__)
 proxy_router = APIRouter()
@@ -148,24 +146,6 @@ async def proxy(
     else:
         model_id = request_body_dict.get("model", "unknown")
 
-    if "https://testnut.cashu.space" in settings.cashu_mints:
-        try:
-            token_str = None
-            if x_cashu_header := headers.get("x-cashu"):
-                token_str = x_cashu_header
-            elif auth_header := headers.get("authorization"):
-                parts = auth_header.split(" ")
-                if len(parts) > 1 and not parts[1].startswith("sk-"):
-                    token_str = parts[1]
-
-            if token_str:
-                token_obj = deserialize_token_from_string(token_str)
-                if token_obj.mint == "https://testnut.cashu.space":
-                    model_id = "mock/gpt-420-mock"
-                    request_body_dict["model"] = model_id
-        except Exception:
-            pass
-
     model_obj = get_model_instance(model_id)
     if not model_obj:
         return create_error_response(

routstr/upstream/base.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import asyncio
 import json
+import re
 import traceback
 from collections.abc import AsyncGenerator
 from typing import TYPE_CHECKING, Mapping
@@ -168,27 +169,6 @@ class BaseUpstreamProvider:
 
         return headers
 
-    def _extract_usage_from_tail(
-        self, tail_content: str
-    ) -> tuple[dict | None, str | None]:
-        """Extract usage and model from tail content of a stream."""
-        usage_data = None
-        model = None
-
-        lines = tail_content.strip().split("\n")
-        for line in lines:
-            if line.startswith("data: "):
-                try:
-                    data = json.loads(line[6:])
-                    if isinstance(data, dict):
-                        if "usage" in data:
-                            usage_data = data["usage"]
-                        if "model" in data:
-                            model = data["model"]
-                except json.JSONDecodeError:
-                    continue
-        return usage_data, model
-
     def prepare_params(
         self, path: str, query_params: Mapping[str, str] | None
     ) -> Mapping[str, str]:
@@ -254,11 +234,7 @@ class BaseUpstreamProvider:
                     )
 
                 # Handle model in input field (alternative format)
-                if (
-                    "input" in data
-                    and isinstance(data["input"], dict)
-                    and "model" in data["input"]
-                ):
+                if "input" in data and isinstance(data["input"], dict) and "model" in data["input"]:
                     original_model = model_obj.id
                     transformed_model = self.transform_model_name(original_model)
                     data["input"]["model"] = transformed_model
@@ -456,64 +432,155 @@ class BaseUpstreamProvider:
         async def stream_with_cost(
             max_cost_for_model: int,
         ) -> AsyncGenerator[bytes, None]:
-            tail_buffer: list[bytes] = []
-            MAX_TAIL_CHUNKS = 5
+            stored_chunks: list[bytes] = []
+            usage_finalized: bool = False
+            last_model_seen: str | None = None
+
+            async def finalize_without_usage() -> bytes | None:
+                nonlocal usage_finalized
+                if usage_finalized:
+                    return None
+                async with create_session() as new_session:
+                    fresh_key = await new_session.get(key.__class__, key.hashed_key)
+                    if not fresh_key:
+                        return None
+                    try:
+                        fallback: dict = {
+                            "model": last_model_seen or "unknown",
+                            "usage": None,
+                        }
+                        cost_data = await adjust_payment_for_tokens(
+                            fresh_key, fallback, new_session, max_cost_for_model
+                        )
+                        usage_finalized = True
+                        logger.info(
+                            "Finalized streaming payment without explicit usage",
+                            extra={
+                                "key_hash": key.hashed_key[:8] + "...",
+                                "cost_data": cost_data,
+                                "balance_after_adjustment": fresh_key.balance,
+                            },
+                        )
+                        return f"data: {json.dumps({'cost': cost_data})}\n\n".encode()
+                    except Exception as cost_error:
+                        logger.error(
+                            "Error finalizing payment without usage",
+                            extra={
+                                "error": str(cost_error),
+                                "error_type": type(cost_error).__name__,
+                                "key_hash": key.hashed_key[:8] + "...",
+                            },
+                        )
+                        return None
 
             try:
                 async for chunk in response.aiter_bytes():
-                    yield chunk
-                    tail_buffer.append(chunk)
-                    if len(tail_buffer) > MAX_TAIL_CHUNKS:
-                        tail_buffer.pop(0)
-
-                # Post-stream processing
-                tail_content = b"".join(tail_buffer).decode("utf-8", errors="ignore")
-                usage_data, model = self._extract_usage_from_tail(tail_content)
-
-                if usage_data:
-                    # Calculate final cost using usage data extracted from stream tail
-                    async with create_session() as new_session:
-                        fresh_key = await new_session.get(key.__class__, key.hashed_key)
-                        if fresh_key:
+                    stored_chunks.append(chunk)
+                    try:
+                        for part in re.split(b"data: ", chunk):
+                            if not part or part.strip() in (b"[DONE]", b""):
+                                continue
                             try:
-                                # We need to reconstruct a response object for adjust_payment_for_tokens
-                                # or call it with usage directly if supported.
-                                # adjust_payment_for_tokens expects a dict with "usage" and "model" keys usually.
+                                obj = json.loads(part)
+                                if isinstance(obj, dict) and obj.get("model"):
+                                    last_model_seen = str(obj.get("model"))
+                            except json.JSONDecodeError:
+                                pass
+                    except Exception:
+                        pass
 
-                                data = {
-                                    "usage": usage_data,
-                                    "model": model or "unknown",
-                                }
+                    yield chunk
 
-                                cost_data = await adjust_payment_for_tokens(
-                                    fresh_key,
-                                    data,
-                                    new_session,
-                                    max_cost_for_model,
-                                )
+                logger.debug(
+                    "Streaming completed, analyzing usage data",
+                    extra={
+                        "key_hash": key.hashed_key[:8] + "...",
+                        "chunks_count": len(stored_chunks),
+                    },
+                )
 
-                                logger.info(
-                                    "Payment adjustment completed for streaming",
-                                    extra={
-                                        "key_hash": key.hashed_key[:8] + "...",
-                                        "cost_data": cost_data,
-                                        "model": model,
-                                        "balance_after_adjustment": fresh_key.balance,
-                                    },
-                                )
-                                # We yield the cost data event at the end
-                                yield f"data: {json.dumps({'cost': cost_data})}\n\n".encode()
-                            except Exception as cost_error:
-                                logger.error(
-                                    "Error adjusting payment for streaming tokens",
-                                    extra={"error": str(cost_error)},
-                                )
+                for i in range(len(stored_chunks) - 1, -1, -1):
+                    chunk = stored_chunks[i]
+                    if not chunk:
+                        continue
+                    try:
+                        events = re.split(b"data: ", chunk)
+                        for event_data in events:
+                            if not event_data or event_data.strip() in (b"[DONE]", b""):
+                                continue
+                            try:
+                                data = json.loads(event_data)
+                                if isinstance(data, dict) and data.get("model"):
+                                    last_model_seen = str(data.get("model"))
+                                if isinstance(data, dict) and isinstance(
+                                    data.get("usage"), dict
+                                ):
+                                    async with create_session() as new_session:
+                                        fresh_key = await new_session.get(
+                                            key.__class__, key.hashed_key
+                                        )
+                                        if fresh_key:
+                                            try:
+                                                cost_data = (
+                                                    await adjust_payment_for_tokens(
+                                                        fresh_key,
+                                                        data,
+                                                        new_session,
+                                                        max_cost_for_model,
+                                                    )
+                                                )
+                                                usage_finalized = True
+                                                logger.info(
+                                                    "Payment adjustment completed for streaming",
+                                                    extra={
+                                                        "key_hash": key.hashed_key[:8]
+                                                        + "...",
+                                                        "cost_data": cost_data,
+                                                        "model": last_model_seen,
+                                                        "balance_after_adjustment": fresh_key.balance,
+                                                    },
+                                                )
+                                                yield f"data: {json.dumps({'cost': cost_data})}\n\n".encode()
+                                            except Exception as cost_error:
+                                                logger.error(
+                                                    "Error adjusting payment for streaming tokens",
+                                                    extra={
+                                                        "error": str(cost_error),
+                                                        "error_type": type(
+                                                            cost_error
+                                                        ).__name__,
+                                                        "key_hash": key.hashed_key[:8]
+                                                        + "...",
+                                                    },
+                                                )
+                                    break
+                            except json.JSONDecodeError:
+                                continue
+                    except Exception as e:
+                        logger.error(
+                            "Error processing streaming response chunk",
+                            extra={
+                                "error": str(e),
+                                "error_type": type(e).__name__,
+                                "key_hash": key.hashed_key[:8] + "...",
+                            },
+                        )
+
+                if not usage_finalized:
+                    maybe_cost_event = await finalize_without_usage()
+                    if maybe_cost_event is not None:
+                        yield maybe_cost_event
 
             except Exception as stream_error:
                 logger.warning(
-                    "Streaming interrupted",
-                    extra={"error": str(stream_error)},
+                    "Streaming interrupted; finalizing without usage",
+                    extra={
+                        "error": str(stream_error),
+                        "error_type": type(stream_error).__name__,
+                        "key_hash": key.hashed_key[:8] + "...",
+                    },
                 )
+                await finalize_without_usage()
                 raise
 
         # Remove inaccurate encoding headers from upstream response
@@ -655,69 +722,166 @@ class BaseUpstreamProvider:
         async def stream_with_responses_cost(
             max_cost_for_model: int,
         ) -> AsyncGenerator[bytes, None]:
-            tail_buffer: list[bytes] = []
-            MAX_TAIL_CHUNKS = 5
+            stored_chunks: list[bytes] = []
+            usage_finalized: bool = False
+            last_model_seen: str | None = None
+            reasoning_tokens: int = 0
+
+            async def finalize_without_usage() -> bytes | None:
+                nonlocal usage_finalized
+                if usage_finalized:
+                    return None
+                async with create_session() as new_session:
+                    fresh_key = await new_session.get(key.__class__, key.hashed_key)
+                    if not fresh_key:
+                        return None
+                    try:
+                        fallback: dict = {
+                            "model": last_model_seen or "unknown",
+                            "usage": None,
+                        }
+                        cost_data = await adjust_payment_for_tokens(
+                            fresh_key, fallback, new_session, max_cost_for_model
+                        )
+                        usage_finalized = True
+                        logger.info(
+                            "Finalized Responses API streaming payment without explicit usage",
+                            extra={
+                                "key_hash": key.hashed_key[:8] + "...",
+                                "cost_data": cost_data,
+                                "balance_after_adjustment": fresh_key.balance,
+                            },
+                        )
+                        return f"data: {json.dumps({'cost': cost_data})}\\n\\n".encode()
+                    except Exception as cost_error:
+                        logger.error(
+                            "Error finalizing Responses API payment without usage",
+                            extra={
+                                "error": str(cost_error),
+                                "error_type": type(cost_error).__name__,
+                                "key_hash": key.hashed_key[:8] + "...",
+                            },
+                        )
+                        return None
 
             try:
                 async for chunk in response.aiter_bytes():
-                    yield chunk
-                    tail_buffer.append(chunk)
-                    if len(tail_buffer) > MAX_TAIL_CHUNKS:
-                        tail_buffer.pop(0)
-
-                # Post-stream processing
-                tail_content = b"".join(tail_buffer).decode("utf-8", errors="ignore")
-                usage_data, model = self._extract_usage_from_tail(tail_content)
-
-                if usage_data:
-                    async with create_session() as new_session:
-                        fresh_key = await new_session.get(key.__class__, key.hashed_key)
-                        if fresh_key:
+                    stored_chunks.append(chunk)
+                    try:
+                        for part in re.split(b"data: ", chunk):
+                            if not part or part.strip() in (b"[DONE]", b""):
+                                continue
                             try:
-                                # We need to reconstruct a response object for adjust_payment_for_tokens
-                                # or call it with usage directly if supported.
-                                # adjust_payment_for_tokens expects a dict with "usage" and "model" keys usually.
+                                obj = json.loads(part)
+                                if isinstance(obj, dict):
+                                    if obj.get("model"):
+                                        last_model_seen = str(obj.get("model"))
 
-                                data = {
-                                    "usage": usage_data,
-                                    "model": model or "unknown",
-                                }
+                                    # Track reasoning tokens for Responses API
+                                    if usage := obj.get("usage", {}):
+                                        if isinstance(usage, dict) and "reasoning_tokens" in usage:
+                                            reasoning_tokens += usage.get("reasoning_tokens", 0)
+                            except json.JSONDecodeError:
+                                pass
+                    except Exception:
+                        pass
 
-                                cost_data = await adjust_payment_for_tokens(
-                                    fresh_key,
-                                    data,
-                                    new_session,
-                                    max_cost_for_model,
-                                )
+                    yield chunk
 
-                                logger.info(
-                                    "Payment adjustment completed for Responses API streaming",
-                                    extra={
-                                        "key_hash": key.hashed_key[:8] + "...",
-                                        "cost_data": cost_data,
-                                        "model": model,
-                                        "balance_after_adjustment": fresh_key.balance,
-                                    },
-                                )
-                                # We yield the cost data event at the end
-                                yield f"data: {json.dumps({'cost': cost_data})}\\n\\n".encode()
-                            except Exception as cost_error:
-                                logger.error(
-                                    "Error adjusting payment for Responses API streaming tokens",
-                                    extra={
-                                        "error": str(cost_error),
-                                        "key_hash": key.hashed_key[:8] + "...",
-                                    },
-                                )
+                logger.debug(
+                    "Responses API streaming completed, analyzing usage data",
+                    extra={
+                        "key_hash": key.hashed_key[:8] + "...",
+                        "chunks_count": len(stored_chunks),
+                        "reasoning_tokens": reasoning_tokens,
+                    },
+                )
+
+                # Process final usage data
+                for i in range(len(stored_chunks) - 1, -1, -1):
+                    chunk = stored_chunks[i]
+                    if not chunk:
+                        continue
+                    try:
+                        events = re.split(b"data: ", chunk)
+                        for event_data in events:
+                            if not event_data or event_data.strip() in (b"[DONE]", b""):
+                                continue
+                            try:
+                                data = json.loads(event_data)
+                                if isinstance(data, dict) and data.get("model"):
+                                    last_model_seen = str(data.get("model"))
+                                if isinstance(data, dict) and isinstance(
+                                    data.get("usage"), dict
+                                ):
+                                    # Include reasoning tokens in usage calculation
+                                    async with create_session() as new_session:
+                                        fresh_key = await new_session.get(
+                                            key.__class__, key.hashed_key
+                                        )
+                                        if fresh_key:
+                                            try:
+                                                cost_data = (
+                                                    await adjust_payment_for_tokens(
+                                                        fresh_key,
+                                                        data,
+                                                        new_session,
+                                                        max_cost_for_model,
+                                                    )
+                                                )
+                                                usage_finalized = True
+                                                logger.info(
+                                                    "Payment adjustment completed for Responses API streaming",
+                                                    extra={
+                                                        "key_hash": key.hashed_key[:8]
+                                                        + "...",
+                                                        "cost_data": cost_data,
+                                                        "model": last_model_seen,
+                                                        "reasoning_tokens": reasoning_tokens,
+                                                        "balance_after_adjustment": fresh_key.balance,
+                                                    },
+                                                )
+                                                yield f"data: {json.dumps({'cost': cost_data})}\\n\\n".encode()
+                                            except Exception as cost_error:
+                                                logger.error(
+                                                    "Error adjusting payment for Responses API streaming tokens",
+                                                    extra={
+                                                        "error": str(cost_error),
+                                                        "error_type": type(
+                                                            cost_error
+                                                        ).__name__,
+                                                        "key_hash": key.hashed_key[:8]
+                                                        + "...",
+                                                    },
+                                                )
+                                    break
+                            except json.JSONDecodeError:
+                                continue
+                    except Exception as e:
+                        logger.error(
+                            "Error processing Responses API streaming response chunk",
+                            extra={
+                                "error": str(e),
+                                "error_type": type(e).__name__,
+                                "key_hash": key.hashed_key[:8] + "...",
+                            },
+                        )
+
+                if not usage_finalized:
+                    maybe_cost_event = await finalize_without_usage()
+                    if maybe_cost_event is not None:
+                        yield maybe_cost_event
 
             except Exception as stream_error:
                 logger.warning(
-                    "Responses API streaming interrupted",
+                    "Responses API streaming interrupted; finalizing without usage",
                     extra={
                         "error": str(stream_error),
+                        "error_type": type(stream_error).__name__,
                         "key_hash": key.hashed_key[:8] + "...",
                     },
                 )
+                await finalize_without_usage()
                 raise
 
         # Remove inaccurate encoding headers from upstream response
@@ -769,8 +933,8 @@ class BaseUpstreamProvider:
                     "model": response_json.get("model", "unknown"),
                     "has_usage": "usage" in response_json,
                     "has_reasoning_tokens": "usage" in response_json
-                    and isinstance(response_json.get("usage"), dict)
-                    and "reasoning_tokens" in response_json["usage"],
+                        and isinstance(response_json.get("usage"), dict)
+                        and "reasoning_tokens" in response_json["usage"],
                 },
             )
 
@@ -1515,8 +1679,21 @@ class BaseUpstreamProvider:
         if "content-encoding" in response_headers:
             del response_headers["content-encoding"]
 
-        usage_data, model = self._extract_usage_from_tail(content_str)
+        usage_data = None
+        model = None
+
         lines = content_str.strip().split("\n")
+        for line in lines:
+            if line.startswith("data: "):
+                try:
+                    data_json = json.loads(line[6:])
+                    if "usage" in data_json:
+                        usage_data = data_json["usage"]
+                        model = data_json.get("model")
+                    elif "model" in data_json and not model:
+                        model = data_json["model"]
+                except json.JSONDecodeError:
+                    continue
 
         if usage_data and model:
             logger.debug(
@@ -2303,7 +2480,7 @@ class BaseUpstreamProvider:
             extra={
                 "amount": amount,
                 "unit": unit,
-                "content_lines": len(content_str.strip().split("\n")),
+                "content_lines": len(content_str.strip().split("\\n")),
             },
         )
 
@@ -2313,14 +2490,25 @@ class BaseUpstreamProvider:
         if "content-encoding" in response_headers:
             del response_headers["content-encoding"]
 
-        usage_data, model = self._extract_usage_from_tail(content_str)
+        usage_data = None
+        model = None
         reasoning_tokens = 0
 
-        # Responses API specific: check for reasoning tokens
-        if usage_data and "reasoning_tokens" in usage_data:
-            reasoning_tokens = usage_data.get("reasoning_tokens", 0)
-
-        lines = content_str.strip().split("\n")
+        lines = content_str.strip().split("\\n")
+        for line in lines:
+            if line.startswith("data: "):
+                try:
+                    data_json = json.loads(line[6:])
+                    if "usage" in data_json:
+                        usage_data = data_json["usage"]
+                        model = data_json.get("model")
+                        # Track reasoning tokens for Responses API
+                        if isinstance(usage_data, dict) and "reasoning_tokens" in usage_data:
+                            reasoning_tokens = usage_data.get("reasoning_tokens", 0)
+                    elif "model" in data_json and not model:
+                        model = data_json["model"]
+                except json.JSONDecodeError:
+                    continue
 
         if usage_data and model:
             logger.debug(
@@ -2396,7 +2584,7 @@ class BaseUpstreamProvider:
 
         async def generate() -> AsyncGenerator[bytes, None]:
             for line in lines:
-                yield (line + "\n").encode("utf-8")
+                yield (line + "\\n").encode("utf-8")
 
         return StreamingResponse(
             generate(),

routstr/upstream/fake.py

@@ -1,265 +0,0 @@
-import asyncio
-import json
-import random
-from typing import AsyncIterator
-
-from fastapi import Request
-from fastapi.responses import Response, StreamingResponse
-
-from ..core.db import ApiKey, AsyncSession
-from ..payment.models import Architecture, Model, Pricing
-from .base import BaseUpstreamProvider
-
-
-class MockUpstreamProvider(BaseUpstreamProvider):
-    """Fack Mock Upstream provider specifically for Testing."""
-
-    provider_type = "mock"
-
-    async def forward_request(
-        self,
-        request: Request,
-        path: str,
-        headers: dict,
-        request_body: bytes | None,
-        key: ApiKey,
-        max_cost_for_model: int,
-        session: AsyncSession,
-        model_obj: Model,
-    ) -> Response | StreamingResponse:
-        if path.endswith("chat/completions"):
-            is_streaming = False
-            if request_body:
-                request_data = json.loads(request_body)
-                is_streaming = request_data.get("stream", False)
-
-            if is_streaming:
-
-                async def fake_streaming_response(
-                    chunk_size: int | None = None,
-                ) -> AsyncIterator[bytes]:
-                    suffix = random.randint(1000, 9999)
-                    req_id = f"gen-mock-stream-{suffix}"
-                    created = 1766138895
-                    model = "mock/gpt-420-mock"
-
-                    def make_chunk(
-                        delta: dict,
-                        finish_reason: str | None = None,
-                        usage: dict | None = None,
-                    ) -> bytes:
-                        chunk = {
-                            "id": req_id,
-                            "provider": "MockProvider",
-                            "model": model,
-                            "object": "chat.completion.chunk",
-                            "created": created,
-                            "choices": [
-                                {
-                                    "index": 0,
-                                    "delta": delta,
-                                    "finish_reason": finish_reason,
-                                    "native_finish_reason": "completed"
-                                    if finish_reason
-                                    else None,
-                                    "logprobs": None,
-                                }
-                            ],
-                        }
-                        if usage:
-                            chunk["usage"] = usage
-                        return f"data: {json.dumps(chunk)}\n\n".encode()
-
-                    # 1. Initial chunk
-                    yield make_chunk({"role": "assistant", "content": ""})
-                    await asyncio.sleep(0.02)
-
-                    # 2. Reasoning chunks
-                    reasoning_tokens = ["Mock", " reason", "ing", "..."]
-                    for token in reasoning_tokens:
-                        delta = {
-                            "role": "assistant",
-                            "content": "",
-                            "reasoning": token,
-                            "reasoning_details": [
-                                {
-                                    "type": "reasoning.summary",
-                                    "summary": token,
-                                    "format": "openai-responses-v1",
-                                    "index": 0,
-                                }
-                            ],
-                        }
-                        yield make_chunk(delta)
-                        await asyncio.sleep(0.03)
-
-                    # 3. Content chunks
-                    content_tokens = ["This", " is", " a", " mock", " stream", "."]
-                    for token in content_tokens:
-                        yield make_chunk({"role": "assistant", "content": token})
-                        await asyncio.sleep(0.03)
-
-                    # 4. Finish chunk
-                    yield make_chunk(
-                        {"role": "assistant", "content": ""}, finish_reason="stop"
-                    )
-
-                    # 5. Usage chunk
-                    usage_data = {
-                        "prompt_tokens": 10,
-                        "completion_tokens": 20,
-                        "total_tokens": 30,
-                        "cost": 0.001,
-                        "is_byok": False,
-                        "prompt_tokens_details": {
-                            "cached_tokens": 0,
-                            "audio_tokens": 0,
-                            "video_tokens": 0,
-                        },
-                        "cost_details": {
-                            "upstream_inference_cost": None,
-                            "upstream_inference_prompt_cost": 0,
-                            "upstream_inference_completions_cost": 0.001,
-                        },
-                        "completion_tokens_details": {
-                            "reasoning_tokens": 10,
-                            "image_tokens": 0,
-                        },
-                    }
-
-                    usage_chunk = {
-                        "id": req_id,
-                        "provider": "MockProvider",
-                        "model": model,
-                        "object": "chat.completion.chunk",
-                        "created": created,
-                        "choices": [
-                            {
-                                "index": 0,
-                                "delta": {"role": "assistant", "content": ""},
-                                "finish_reason": None,
-                                "native_finish_reason": None,
-                                "logprobs": None,
-                            }
-                        ],
-                        "usage": usage_data,
-                    }
-                    yield f"data: {json.dumps(usage_chunk)}\n\n".encode()
-
-                    # 6. DONE
-                    yield b"data: [DONE]\n\n"
-
-                    # 7. Cost
-                    cost_chunk = {
-                        "cost": {
-                            "base_msats": 0,
-                            "input_msats": 2,
-                            "output_msats": 10,
-                            "total_msats": 12,
-                        }
-                    }
-                    yield f"data: {json.dumps(cost_chunk)}\n\n".encode()
-
-                return StreamingResponse(
-                    fake_streaming_response(),
-                    200,
-                )
-
-            else:
-                suffix = random.randint(1000, 9999)
-                content_dict = {
-                    "id": f"gen-mock-{suffix}",
-                    "provider": "MockProvider",
-                    "model": "mock/gpt-5-mini",
-                    "object": "chat.completion",
-                    "created": 1766138655,
-                    "choices": [
-                        {
-                            "logprobs": None,
-                            "finish_reason": "length",
-                            "native_finish_reason": "max_output_tokens",
-                            "index": 0,
-                            "message": {
-                                "role": "assistant",
-                                "content": f"Mock Content {suffix}",
-                                "refusal": None,
-                                "reasoning": f"Mock Reasoning {suffix}",
-                                "reasoning_details": [
-                                    {
-                                        "format": "openai-responses-v1",
-                                        "index": 0,
-                                        "type": "reasoning.summary",
-                                        "summary": f"Mock Summary {suffix}",
-                                    },
-                                    {
-                                        "id": f"rs_mock_{suffix}",
-                                        "format": "openai-responses-v1",
-                                        "index": 0,
-                                        "type": "reasoning.encrypted",
-                                        "data": "mock_encrypted_data",
-                                    },
-                                ],
-                            },
-                        }
-                    ],
-                    "usage": {
-                        "prompt_tokens": 10,
-                        "completion_tokens": 10,
-                        "total_tokens": 20,
-                        "cost": 0,
-                        "is_byok": False,
-                        "prompt_tokens_details": {
-                            "cached_tokens": 0,
-                            "audio_tokens": 0,
-                            "video_tokens": 0,
-                        },
-                        "cost_details": {
-                            "upstream_inference_cost": None,
-                            "upstream_inference_prompt_cost": 0,
-                            "upstream_inference_completions_cost": 0,
-                        },
-                        "completion_tokens_details": {
-                            "reasoning_tokens": 5,
-                            "image_tokens": 0,
-                        },
-                    },
-                    "cost": {
-                        "base_msats": 0,
-                        "input_msats": 0,
-                        "output_msats": 0,
-                        "total_msats": 0,
-                    },
-                }
-                return Response(json.dumps(content_dict).encode(), 200)
-
-        elif path.endswith("embeddings"):
-            raise NotImplementedError
-        elif path.endswith("responses"):
-            raise NotImplementedError
-        else:
-            raise NotImplementedError
-
-    async def fetch_models(self) -> list[Model]:
-        return [
-            Model(
-                id="mock/gpt-420-mock",
-                name="mock/gpt-420-mock",
-                created=0,
-                description="mock model for testing",
-                context_length=8192,
-                architecture=Architecture(
-                    modality="text",
-                    input_modalities=["text"],
-                    output_modalities=["text"],
-                    tokenizer="",
-                    instruct_type=None,
-                ),
-                pricing=Pricing(prompt=0.01, completion=0.01),
-            ),
-        ]
-
-    def transform_model_name(self, model_id: str) -> str:
-        return "fake-model"
-
-    async def get_balance(self) -> float | None:
-        return 420.69

routstr/upstream/helpers.py

@@ -218,14 +218,6 @@ async def init_upstreams() -> list[BaseUpstreamProvider]:
         results = await asyncio.gather(*tasks)
         upstreams = [p for p in results if p is not None]
 
-        if "https://testnut.cashu.space" in settings.cashu_mints:
-            from .fake import MockUpstreamProvider
-
-            mock_provider = MockUpstreamProvider("mock", "mock")
-            await mock_provider.refresh_models_cache()
-            upstreams.append(mock_provider)
-            logger.info("Initialized MockUpstreamProvider for testnut mint")
-
         return upstreams
 
 

routstr/wallet.py

@@ -313,8 +313,6 @@ async def periodic_payout() -> None:
         try:
             async with db.create_session() as session:
                 for mint_url in settings.cashu_mints:
-                    if mint_url == "https://testnut.cashu.space":
-                        continue
                     for unit in ["sat", "msat"]:
                         wallet = await get_wallet(mint_url, unit)
                         proofs = get_proofs_per_mint_and_unit(