fix: refund users when model returns zero tokens with non-zero cost

When a model produces empty output (zero tokens reported) but the cost
calculation returns a non-zero USD cost, the previous behavior was to
charge the full amount as output_msats. This is incorrect — if no tokens
were produced, the user should not be charged.

Now the else branch:
- Logs a warning with the usd_cost and model for debugging
- Sets input_msats, output_msats, and cost_in_msats to 0 (full refund)

.github/workflows/container.yml

@@ -14,6 +14,14 @@ jobs:
     steps:
       - name: Checkout code
         uses: actions/checkout@v3
+        with:
+          fetch-depth: 0
+
+      - name: Resolve git metadata
+        id: gitmeta
+        run: |
+          echo "sha=$(git rev-parse --short=7 HEAD)" >> "$GITHUB_OUTPUT"
+          echo "tag=$(git describe --tags --exact-match HEAD 2>/dev/null || true)" >> "$GITHUB_OUTPUT"
 
       - name: Set up Docker Buildx
         uses: docker/setup-buildx-action@v2
@@ -29,7 +37,11 @@ jobs:
         uses: docker/build-push-action@v4
         with:
           context: .
+          file: Dockerfile.full
           push: true
+          build-args: |
+            GIT_COMMIT=${{ steps.gitmeta.outputs.sha }}
+            GIT_TAG=${{ steps.gitmeta.outputs.tag }}
           tags: |
             ghcr.io/routstr/proxy:latest
             ghcr.io/routstr/core:latest

Dockerfile

@@ -21,6 +21,10 @@ WORKDIR /app
 
 COPY . .
 
+ARG GIT_COMMIT=""
+ARG GIT_TAG=""
+ENV GIT_COMMIT=${GIT_COMMIT}
+ENV GIT_TAG=${GIT_TAG}
 ENV PORT=8000
 ENV PYTHONUNBUFFERED=1
 

Dockerfile.full

@@ -41,6 +41,10 @@ RUN uv sync --no-dev
 # Copy the built UI from the ui-builder stage
 COPY --from=ui-builder /app/ui/out ./ui_out
 
+ARG GIT_COMMIT=""
+ARG GIT_TAG=""
+ENV GIT_COMMIT=${GIT_COMMIT}
+ENV GIT_TAG=${GIT_TAG}
 ENV PORT=8000
 ENV PYTHONUNBUFFERED=1
 

docs/provider/deployment.md

@@ -2,6 +2,57 @@
 
 Production deployment guide for Routstr Provider nodes.
 
+## All-in-One Docker Image (Preferred)
+
+The easiest way to deploy Routstr is using the all-in-one Docker image from Docker Hub, which includes both the FastAPI backend and the Next.js admin dashboard in a single container.
+
+### Quick Start
+
+```bash
+docker run -d \
+  --name routstr \
+  -p 8000:8000 \
+  -v routstr-data:/app/data \
+  -e DATABASE_URL="sqlite:////app/data/routstr.db" \
+  9qeklajc/routstr:latest
+```
+
+Access your node:
+- **API & Admin Dashboard**: http://localhost:8000
+
+### Docker Compose Setup
+
+Create `docker-compose.yml`:
+
+```yaml
+version: '3.8'
+
+services:
+  routstr:
+    image: 9qeklajc/routstr:latest
+    container_name: routstr
+    restart: unless-stopped
+    ports:
+      - "8000:8000"
+    volumes:
+      - routstr-data:/app/data
+    environment:
+      DATABASE_URL: "sqlite:////app/data/routstr.db"
+      ADMIN_KEY: "your-secure-admin-key"
+      LOG_LEVEL: "info"
+
+volumes:
+  routstr-data:
+```
+
+Start it:
+
+```bash
+docker compose up -d
+```
+
+---
+
 ## Docker Compose (Recommended)
 
 For production, use Docker Compose with persistent storage and optional Tor support.

plans/other-mints-balance-tracking-plan.md

@@ -0,0 +1,355 @@
+# Plan: Track unsupported incoming mints in `other_mints` and include them in balances
+
+## Goal
+
+When a Cashu token arrives from a mint that is **not** in `settings.cashu_mints`, we currently create/use a wallet for that mint and swap value into the primary mint. Any change/surplus left behind after `melt()` remains in the foreign `token_wallet`, but that balance is not surfaced by `fetch_all_balances()` because it only iterates over configured mints.
+
+This plan adds persistent tracking for those foreign mints in a new database table called `other_mints`, and updates balance reporting to include them.
+
+---
+
+## Current behavior
+
+### Incoming unsupported mint flow
+
+In `routstr/wallet.py`:
+
+- `recieve_token()` deserializes the token
+- if `token_obj.mint not in settings.cashu_mints`, it calls `swap_to_primary_mint(token_obj, wallet)`
+- `swap_to_primary_mint()` calls `token_wallet.melt(...)` to pay the primary mint invoice
+
+### Important detail: change is retained, not discarded
+
+The underlying Cashu wallet library keeps any melt change:
+
+- `Wallet.melt()` constructs blank outputs for change
+- when the melt succeeds, returned change is reconstructed into proofs
+- those proofs are appended to `self.proofs` and stored in the wallet DB
+
+So surplus from unsupported mints is **not discarded**, but it may become invisible operationally.
+
+### Visibility problem
+
+`fetch_all_balances()` currently only loops over:
+
+- `settings.cashu_mints`
+- units `sat` and `msat`
+
+This means balances left on unsupported mints are not shown in admin balance reporting.
+
+---
+
+## Proposed design
+
+## 1. Add a new DB table: `other_mints`
+
+Add a small table in `routstr/core/db.py` to persist unsupported mints we have seen in incoming tokens.
+
+Suggested schema:
+
+- `mint_url: str` primary key
+- `created_at: int`
+- `last_seen_at: int`
+
+Minimal model:
+
+```python
+class OtherMint(SQLModel, table=True):
+    __tablename__ = "other_mints"
+
+    mint_url: str = Field(primary_key=True)
+    created_at: int = Field(default_factory=lambda: int(time.time()))
+    last_seen_at: int = Field(default_factory=lambda: int(time.time()))
+```
+
+Why minimal:
+
+- the only required function is mint discovery/tracking
+- unit handling can remain dynamic via existing balance queries over `sat` and `msat`
+
+---
+
+## 2. Add DB helpers for `other_mints`
+
+In `routstr/core/db.py`, add helper functions:
+
+### `register_other_mint(mint_url: str) -> None`
+
+Behavior:
+
+- if the mint is not present, insert it
+- if it already exists, update `last_seen_at`
+
+### `list_other_mints(session) -> list[str]`
+
+Behavior:
+
+- return all tracked unsupported mint URLs
+
+Optional later:
+
+- `delete_other_mint(...)`
+- admin cleanup helpers
+
+---
+
+## 3. Register unsupported mints during token receipt
+
+Update `recieve_token()` in `routstr/wallet.py`.
+
+Current logic:
+
+```python
+if token_obj.mint not in settings.cashu_mints:
+    return await swap_to_primary_mint(token_obj, wallet)
+```
+
+Planned logic:
+
+```python
+if token_obj.mint not in settings.cashu_mints:
+    await db.register_other_mint(token_obj.mint)
+    return await swap_to_primary_mint(token_obj, wallet)
+```
+
+Why here:
+
+- this is the earliest reliable point where we know the mint came in via an actual token
+- this is exactly the path that can leave foreign-mint change behind
+- it avoids needing to infer unsupported mints later from wallet internals
+
+---
+
+## 4. Update `fetch_all_balances()` to include `other_mints`
+
+Current behavior only includes configured mints.
+
+Planned behavior:
+
+- load tracked unsupported mints from DB
+- combine them with `settings.cashu_mints`
+- dedupe while preserving order
+- fetch balances for all tracked mints across requested units
+
+Conceptual flow:
+
+```python
+tracked_mints = dedupe(settings.cashu_mints + other_mints_from_db)
+```
+
+Then existing per-mint/per-unit balance logic can remain mostly unchanged.
+
+This ensures that retained change on unsupported mints becomes visible in admin balance reporting.
+
+---
+
+## 5. Add a balance source marker
+
+Extend `BalanceDetail` in `routstr/wallet.py` to identify whether a balance row comes from a configured mint or an `other_mints` entry.
+
+Suggested field:
+
+- `source: str` with values:
+  - `"configured"`
+  - `"other"`
+
+Updated shape:
+
+```python
+class BalanceDetail(TypedDict, total=False):
+    mint_url: str
+    unit: str
+    source: str
+    wallet_balance: int
+    user_balance: int
+    owner_balance: int
+    error: str
+```
+
+Why this helps:
+
+- admin can distinguish normal configured wallet balances from foreign/unsupported balances
+- avoids confusion if unexpected mint URLs show up in the balances API/UI
+
+---
+
+## 6. Admin/API impact
+
+Backend impact is minimal because `/admin/api/balances` already returns `fetch_all_balances()` output.
+
+Effects:
+
+- supported mints continue to show as before
+- tracked unsupported mints will also appear
+- UI can optionally display the new `source` field
+
+No API contract break is expected if the frontend ignores unknown fields.
+
+---
+
+## 7. Payout behavior: do not change in phase 1
+
+`periodic_payout()` currently only iterates over `settings.cashu_mints`.
+
+Recommendation for this change:
+
+- **do not** expand `periodic_payout()` to include `other_mints` yet
+- only improve visibility through balance reporting
+
+Reason:
+
+- automatic payout from unsupported/foreign mints may be operationally undesirable
+- visibility should come first, automation second
+
+Possible future phase:
+
+- add optional sweeping/payout support for `other_mints`
+- or provide an admin-triggered withdrawal/sweep flow
+
+---
+
+## 8. Logging improvements (optional)
+
+Optional follow-up improvement in `swap_to_primary_mint()`:
+
+- capture the return value from `token_wallet.melt(...)`
+- if feasible, log any reported change amount
+- otherwise, rely on wallet balance reporting to surface residual amounts
+
+This is useful but not required for the first implementation.
+
+---
+
+## Files to change
+
+### `routstr/core/db.py`
+
+Add:
+
+- `OtherMint` SQLModel
+- `register_other_mint()`
+- `list_other_mints()`
+
+### `migrations/versions/<new_revision>_add_other_mints_table.py`
+
+Create migration to add the `other_mints` table.
+
+### `routstr/wallet.py`
+
+Update:
+
+- `recieve_token()` to register unsupported mints
+- `BalanceDetail` to include `source`
+- `fetch_all_balances()` to include both configured and tracked unsupported mints
+
+### `routstr/core/admin.py`
+
+Likely no backend changes required unless a dedicated `other_mints` API is desired.
+
+---
+
+## Behavior rules
+
+### Register a mint when
+
+- an incoming token is processed
+- the token mint is not in `settings.cashu_mints`
+
+### Do not remove automatically when
+
+- balance reaches zero
+
+Reason:
+
+- historical visibility is useful
+- avoids flapping entries in the admin balance list
+- mint may receive additional unsupported tokens later
+
+Potential future enhancement:
+
+- admin endpoint to prune zero-balance `other_mints`
+
+---
+
+## Edge cases
+
+### A mint later becomes configured
+
+If a mint in `other_mints` is later added to `settings.cashu_mints`:
+
+- deduplication prevents duplicate balance rows
+- `source` should resolve to `configured`
+
+### Unsupported mint with zero balance
+
+A tracked unsupported mint may show zero balances.
+
+Initial recommendation:
+
+- allow it to appear
+- consider later filtering zero-balance `other` rows if the UI becomes noisy
+
+### Units
+
+Balance fetching can continue to query both `sat` and `msat` for each tracked mint.
+
+If a mint has no proofs in one unit, current error/zero handling can continue to apply.
+
+---
+
+## Test plan
+
+### DB tests
+
+- registering a new unsupported mint inserts a row
+- registering the same mint again updates `last_seen_at` without duplication
+- listing other mints returns expected mint URLs
+
+### Wallet tests
+
+#### `recieve_token()`
+
+- when mint is unsupported, `db.register_other_mint()` is called before swap
+- when mint is configured, `db.register_other_mint()` is not called
+
+#### `fetch_all_balances()`
+
+- includes configured mints
+- includes `other_mints` from DB
+- dedupes if a mint exists in both configured and other lists
+- sets `source` correctly
+
+### Regression tests
+
+- existing trusted mint balance reporting remains unchanged
+- `/admin/api/balances` continues to work
+
+---
+
+## Recommended implementation order
+
+1. Add `OtherMint` model to `routstr/core/db.py`
+2. Add Alembic migration for `other_mints`
+3. Add `register_other_mint()` and `list_other_mints()` helpers
+4. Update `recieve_token()` to register unsupported mints
+5. Update `fetch_all_balances()` to union configured + tracked other mints
+6. Add `source` to `BalanceDetail`
+7. Add/adjust tests
+
+---
+
+## Summary
+
+This change solves an operational visibility problem:
+
+- unsupported incoming mints can leave retained change in foreign wallets
+- those funds are currently preserved but not surfaced in balance reporting
+- introducing `other_mints` makes those mints discoverable and auditable
+- expanding `fetch_all_balances()` ensures their balances are visible in admin tooling
+
+Recommended scope for the first pass:
+
+- track unsupported mints in DB
+- include them in balance reporting
+- mark them as `source="other"`
+- do not yet change payout/sweeping behavior

routstr/auth.py

@@ -317,13 +317,13 @@ async def validate_bearer_key(
                 extra={"key_hash": hashed_key[:8] + "..."},
             )
 
-            logger.info(
+            logger.debug(
                 "AUTH: About to call credit_balance",
                 extra={"token_preview": bearer_key[:50]},
             )
             try:
                 msats = await credit_balance(bearer_key, new_key, session)
-                logger.info(
+                logger.debug(
                     "AUTH: credit_balance returned successfully", extra={"msats": msats}
                 )
             except Exception as credit_error:

routstr/core/db.py

@@ -79,11 +79,10 @@ class ApiKey(SQLModel, table=True):  # type: ignore
 
 
 async def reset_all_reserved_balances(session: AsyncSession) -> None:
-    logger.info("Resetting all reserved balances to 0")
     stmt = update(ApiKey).values(reserved_balance=0)
     await session.exec(stmt)  # type: ignore[call-overload]
     await session.commit()
-    logger.info("Reserved balances reset successfully")
+    logger.info("Reset reserved balances on startup")
 
 
 class ModelRow(SQLModel, table=True):  # type: ignore

routstr/core/exceptions.py

@@ -22,16 +22,20 @@ async def http_exception_handler(request: Request, exc: Exception) -> JSONRespon
     # Get status code and detail - works for both FastAPI and Starlette HTTPException
     status_code = getattr(exc, "status_code", 500)
     detail = getattr(exc, "detail", str(exc))
+    path = request.url.path
 
-    logger.warning(
-        "HTTP exception",
-        extra={
-            "request_id": request_id,
-            "status_code": status_code,
-            "detail": detail,
-            "path": request.url.path,
-        },
-    )
+    # 4xx is client behaviour; the uvicorn access log already records it.
+    # Only 5xx warrants a server-side warning/error log here.
+    if status_code >= 500:
+        logger.error(
+            f"HTTP {status_code} on {path}: {detail}",
+            extra={
+                "request_id": request_id,
+                "status_code": status_code,
+                "detail": detail,
+                "path": path,
+            },
+        )
 
     return JSONResponse(
         status_code=status_code,

routstr/core/logging.py

@@ -41,14 +41,23 @@ import logging.config
 import logging.handlers
 import os
 import re
+import sys
 import tomllib
 from datetime import datetime
 from pathlib import Path
 from typing import Any
 
 from pythonjsonlogger import jsonlogger
+from rich.console import Console
 from rich.logging import RichHandler
 
+# Only use RichHandler when stdout is a real TTY. In non-TTY contexts
+# (docker logs, pipes, CI) Rich pads every line to width and wraps long
+# records, producing visually-empty trailing whitespace and split records.
+# A plain StreamHandler avoids both problems.
+_stdout_is_tty = sys.stdout.isatty()
+_console = Console(soft_wrap=True) if _stdout_is_tty else None
+
 # Define custom TRACE level
 TRACE_LEVEL = 5
 logging.addLevelName(TRACE_LEVEL, "TRACE")
@@ -261,6 +270,26 @@ def setup_logging() -> None:
     if console_enabled:
         handlers.append("console")
 
+    if _stdout_is_tty:
+        console_handler: dict[str, Any] = {
+            "()": RichHandler,
+            "level": log_level,
+            "show_time": False,
+            "show_path": False,
+            "rich_tracebacks": True,
+            "markup": True,
+            "console": _console,
+            "filters": ["request_id_filter", "security_filter"],
+        }
+    else:
+        console_handler = {
+            "class": "logging.StreamHandler",
+            "level": log_level,
+            "formatter": "plain",
+            "stream": "ext://sys.stdout",
+            "filters": ["request_id_filter", "security_filter"],
+        }
+
     LOGGING_CONFIG = {
         "version": 1,
         "disable_existing_loggers": False,
@@ -270,6 +299,10 @@ def setup_logging() -> None:
                 "format": "%(asctime)s %(name)s %(levelname)s %(message)s %(pathname)s %(lineno)d %(version)s %(request_id)s",
                 "datefmt": "%Y-%m-%d %H:%M:%S",
             },
+            "plain": {
+                "format": "%(asctime)s %(levelname)-7s %(name)s %(message)s",
+                "datefmt": "%Y-%m-%d %H:%M:%S",
+            },
         },
         "filters": {
             "version_filter": {"()": VersionFilter},
@@ -277,15 +310,7 @@ def setup_logging() -> None:
             "security_filter": {"()": SecurityFilter},
         },
         "handlers": {
-            "console": {
-                "()": RichHandler,
-                "level": log_level,
-                "show_time": False,
-                "show_path": False,
-                "rich_tracebacks": True,
-                "markup": True,
-                "filters": ["request_id_filter", "security_filter"],
-            },
+            "console": console_handler,
             "file": {
                 "()": DailyRotatingFileHandler,
                 "level": log_level,

routstr/core/main.py

@@ -1,5 +1,4 @@
 import asyncio
-import os
 from contextlib import asynccontextmanager
 from pathlib import Path
 from typing import AsyncGenerator
@@ -9,6 +8,8 @@ from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import FileResponse, RedirectResponse
 from fastapi.staticfiles import StaticFiles
 from starlette.exceptions import HTTPException
+from starlette.responses import Response as StarletteResponse
+from starlette.types import Scope
 
 from ..auth import periodic_key_reset
 from ..balance import balance_router, deprecated_wallet_router
@@ -22,6 +23,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 ..upstream.litellm_routing import configure_litellm
 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
@@ -30,16 +32,12 @@ from .logging import get_logger, setup_logging
 from .middleware import LoggingMiddleware
 from .settings import SettingsService
 from .settings import settings as global_settings
+from .version import __version__
 
 # Initialize logging first
 setup_logging()
 logger = get_logger(__name__)
 
-if os.getenv("VERSION_SUFFIX") is not None:
-    __version__ = f"0.4.3-{os.getenv('VERSION_SUFFIX')}"
-else:
-    __version__ = "0.4.3"
-
 
 @asynccontextmanager
 async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
@@ -59,6 +57,10 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
     routstr_fee_task = None
 
     try:
+        # Apply litellm-wide settings (drop_params, chat-completions URL,
+        # debug logging) before any upstream provider dispatches a request.
+        configure_litellm()
+
         # Run database migrations on startup
         run_migrations()
 
@@ -197,6 +199,23 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
             )
 
 
+class _ImmutableStaticFiles(StaticFiles):
+    """Static files with long Cache-Control for content-hashed Next.js assets.
+
+    Files under `/_next/static/` are emitted with content hashes in their
+    filenames and never mutate, so we serve them with a one-year immutable
+    cache header so browsers and CDNs stop revalidating on every reload.
+    """
+
+    async def get_response(self, path: str, scope: Scope) -> StarletteResponse:
+        response = await super().get_response(path, scope)
+        if response.status_code == 200:
+            response.headers["Cache-Control"] = (
+                "public, max-age=31536000, immutable"
+            )
+        return response
+
+
 app = FastAPI(version=__version__, lifespan=lifespan)
 
 
@@ -243,7 +262,7 @@ if UI_DIST_PATH.exists() and UI_DIST_PATH.is_dir():
 
     app.mount(
         "/_next",
-        StaticFiles(directory=UI_DIST_PATH / "_next", check_dir=True),
+        _ImmutableStaticFiles(directory=UI_DIST_PATH / "_next", check_dir=True),
         name="next-static",
     )
 
@@ -251,100 +270,70 @@ if UI_DIST_PATH.exists() and UI_DIST_PATH.is_dir():
     async def serve_root_ui() -> FileResponse:
         return FileResponse(UI_DIST_PATH / "index.html")
 
-    # Add explicit route for /index.txt to redirect to /
+    # Serve the App Router RSC payload for the home page.
     @app.get("/index.txt", include_in_schema=False)
-    async def redirect_index_txt() -> RedirectResponse:
-        return RedirectResponse("/")
+    async def serve_root_rsc() -> FileResponse:
+        return FileResponse(
+            UI_DIST_PATH / "index.txt", media_type="text/x-component"
+        )
+
+    # Next.js is built with `trailingSlash: true`, so all UI page URLs end
+    # with a slash (e.g. `/login/`). The proxy router catches `/{path:path}`
+    # before FastAPI's `redirect_slashes` logic can normalize the URL, so we
+    # must register both the with-slash and without-slash variants here.
+    UI_PAGES = (
+        "dashboard",
+        "login",
+        "model",
+        "providers",
+        "settings",
+        "transactions",
+        "balances",
+        "logs",
+        "usage",
+        "unauthorized",
+    )
+
+    def _register_ui_page(name: str) -> None:
+        page_dir = UI_DIST_PATH / name
+        index_html = page_dir / "index.html"
+        index_txt = page_dir / "index.txt"
+
+        async def serve_page() -> FileResponse:
+            return FileResponse(index_html)
+
+        async def serve_page_rsc() -> FileResponse:
+            return FileResponse(index_txt, media_type="text/x-component")
+
+        app.add_api_route(
+            f"/{name}",
+            serve_page,
+            methods=["GET"],
+            include_in_schema=False,
+            name=f"serve_{name}_ui",
+        )
+        app.add_api_route(
+            f"/{name}/",
+            serve_page,
+            methods=["GET"],
+            include_in_schema=False,
+            name=f"serve_{name}_ui_slash",
+        )
+        app.add_api_route(
+            f"/{name}/index.txt",
+            serve_page_rsc,
+            methods=["GET"],
+            include_in_schema=False,
+            name=f"serve_{name}_rsc",
+        )
+
+    for _page in UI_PAGES:
+        _register_ui_page(_page)
 
     @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"
@@ -356,9 +345,6 @@ if UI_DIST_PATH.exists() and UI_DIST_PATH.is_dir():
     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"

routstr/core/middleware.py

@@ -14,8 +14,54 @@ logger = get_logger(__name__)
 request_id_context: ContextVar[str | None] = ContextVar("request_id")
 
 
+# Methods that are never logged: HEAD requests are health probes from
+# monitoring/load balancers, OPTIONS are CORS preflights — both are framework
+# chatter, not user-meaningful events.
+_SKIP_LOG_METHODS: frozenset[str] = frozenset({"HEAD", "OPTIONS"})
+
+# Path prefixes to skip. Includes Next.js static chunks and the admin
+# dashboard's internal polling API (/admin/api/*) which the UI hits on a timer
+# to refresh balances, logs, providers, etc. — high volume, low diagnostic
+# value. Mutating admin actions are recorded separately in the audit log.
+_SKIP_LOG_PREFIXES: tuple[str, ...] = (
+    "/_next/",
+    "/admin/api/",
+)
+
+# Exact paths to skip. RSC payload prefetches (`*/index.txt`) fire automatically
+# as the user hovers near `<Link>`s, and `/v1/wallet/info` is polled by the UI.
+_SKIP_LOG_EXACT: frozenset[str] = frozenset(
+    {
+        "/favicon.ico",
+        "/icon.ico",
+        "/v1/wallet/info",
+        "/index.txt",
+        "/login/index.txt",
+        "/model/index.txt",
+        "/providers/index.txt",
+        "/settings/index.txt",
+        "/transactions/index.txt",
+        "/balances/index.txt",
+        "/logs/index.txt",
+        "/usage/index.txt",
+        "/unauthorized/index.txt",
+    }
+)
+
+
+def _should_log(method: str, path: str) -> bool:
+    if method in _SKIP_LOG_METHODS:
+        return False
+    if path in _SKIP_LOG_EXACT:
+        return False
+    return not any(path.startswith(prefix) for prefix in _SKIP_LOG_PREFIXES)
+
+
 class LoggingMiddleware(BaseHTTPMiddleware):
-    """Middleware to log detailed request and response information."""
+    """Middleware to log proxy interactions and page navigation.
+
+    Skips logging for static assets and Next.js chunks to avoid noise.
+    """
 
     async def dispatch(self, request: Request, call_next: Callable) -> Response:
         # Generate request ID
@@ -25,56 +71,20 @@ class LoggingMiddleware(BaseHTTPMiddleware):
         # Set request ID in context for logging
         token = request_id_context.set(request_id)
 
+        path = request.url.path
+        should_log = _should_log(request.method, path)
+
         # Start timing
         start_time = time.time()
 
-        # Log request details
-        request_body = None
-        if request.method in ["POST", "PUT", "PATCH"]:
-            try:
-                # Only read body for non-streaming requests
-                if hasattr(request, "_body"):
-                    request_body = await request.body()
-            except Exception:
-                pass
-
-        # Log incoming request
-        logger.info(
-            "Incoming request",
-            extra={
-                "request_id": request_id,
-                "method": request.method,
-                "path": request.url.path,
-                "query_params": dict(request.query_params),
-                "headers": {
-                    k: v
-                    for k, v in request.headers.items()
-                    if k.lower()
-                    not in [
-                        "authorization",
-                        "x-cashu",
-                        "cookie",
-                        "cf-connecting-ip",
-                        "cf-ipcountry",
-                        "x-forwarded-for",
-                        "x-real-ip",
-                    ]
-                },
-                "body_size": len(request_body) if request_body else 0,
-            },
-        )
-
-        # Log at TRACE level for full body (security filter will redact sensitive data)
-        if request_body and hasattr(logger, "exception"):
-            logger.exception(
-                "Request body",
+        if should_log:
+            logger.info(
+                "Incoming request",
                 extra={
                     "request_id": request_id,
                     "method": request.method,
-                    "path": request.url.path,
-                    "body": request_body.decode("utf-8", errors="ignore")[
-                        :1000
-                    ],  # Limit size
+                    "path": path,
+                    "query_params": dict(request.query_params),
                 },
             )
 
@@ -82,36 +92,32 @@ class LoggingMiddleware(BaseHTTPMiddleware):
         try:
             response = await call_next(request)
 
-            # Calculate duration
-            duration = time.time() - start_time
-
-            # Log response
-            logger.info(
-                "Request completed",
-                extra={
-                    "request_id": request_id,
-                    "method": request.method,
-                    "path": request.url.path,
-                    "status_code": response.status_code,
-                    "duration_ms": round(duration * 1000, 2),
-                },
-            )
+            if should_log:
+                duration = time.time() - start_time
+                logger.info(
+                    "Request completed",
+                    extra={
+                        "request_id": request_id,
+                        "method": request.method,
+                        "path": path,
+                        "status_code": response.status_code,
+                        "duration_ms": round(duration * 1000, 2),
+                    },
+                )
             if hasattr(response, "headers"):
                 response.headers["x-routstr-request-id"] = request_id
 
             return response
 
         except Exception as e:
-            # Calculate duration
+            # Always log failures, even for skipped paths, so we don't lose errors.
             duration = time.time() - start_time
-
-            # Log error
             logger.error(
                 "Request failed",
                 extra={
                     "request_id": request_id,
                     "method": request.method,
-                    "path": request.url.path,
+                    "path": path,
                     "duration_ms": round(duration * 1000, 2),
                     "error": str(e),
                     "error_type": type(e).__name__,

routstr/core/version.py

@@ -0,0 +1,81 @@
+"""Application version resolution.
+
+Priority order:
+
+1. ``VERSION_SUFFIX`` env var (manual override; preserves prior behaviour).
+2. Bare base version when HEAD is on the matching release tag (detected via
+   ``GIT_TAG`` env or ``git describe --tags --exact-match HEAD``).
+3. ``GIT_COMMIT`` env var (build-time injection) -> ``<base>+g<sha>``.
+4. Local ``.git`` lookup (source checkouts) -> ``<base>+g<sha>``.
+5. Fallback: bare base version.
+
+The ``+g<sha>`` form is PEP 440 local-version syntax so the result remains a
+valid package version.
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+from functools import lru_cache
+from pathlib import Path
+
+BASE_VERSION = "0.4.3"
+
+_REPO_ROOT = Path(__file__).resolve().parents[2]
+_GIT_TIMEOUT_SECONDS = 2.0
+
+
+def _run_git(*args: str) -> str | None:
+    try:
+        result = subprocess.run(  # noqa: S603 - fixed argv, no shell
+            ["git", *args],
+            cwd=_REPO_ROOT,
+            check=False,
+            capture_output=True,
+            text=True,
+            timeout=_GIT_TIMEOUT_SECONDS,
+        )
+    except (FileNotFoundError, subprocess.SubprocessError, OSError):
+        return None
+    if result.returncode != 0:
+        return None
+    return result.stdout.strip() or None
+
+
+def _git_short_sha() -> str | None:
+    sha = os.getenv("GIT_COMMIT", "").strip()
+    if sha:
+        return sha[:7]
+    return _run_git("rev-parse", "--short=7", "HEAD")
+
+
+def _on_tagged_release() -> bool:
+    tag = os.getenv("GIT_TAG", "").strip()
+    if tag:
+        return tag.lstrip("v") == BASE_VERSION
+    described = _run_git("describe", "--tags", "--exact-match", "HEAD")
+    if not described:
+        return False
+    return described.lstrip("v") == BASE_VERSION
+
+
+@lru_cache(maxsize=1)
+def get_version() -> str:
+    suffix = os.getenv("VERSION_SUFFIX")
+    if suffix is not None:
+        return f"{BASE_VERSION}-{suffix}"
+
+    if _on_tagged_release():
+        return BASE_VERSION
+
+    sha = _git_short_sha()
+    if not sha:
+        return BASE_VERSION
+
+    return f"{BASE_VERSION}+g{sha}"
+
+
+__version__ = get_version()
+
+__all__ = ["BASE_VERSION", "__version__", "get_version"]

routstr/nostr/listing.py

@@ -26,7 +26,7 @@ logger = get_logger(__name__)
 
 def get_app_version() -> str | None:
     try:
-        from ..core.main import __version__ as imported_version
+        from ..core.version import __version__ as imported_version
 
         return imported_version
     except Exception:

routstr/payment/cost_calculation.py

@@ -53,10 +53,17 @@ async def calculate_cost(  # todo: can be sync
 
     if "usage" not in response_data or response_data["usage"] is None:
         logger.warning(
-            "No usage data in response, using base cost only",
+            "No usage data in response — billing at MaxCostData with zero "
+            "tokens. Dashboard will show this request as `(0+0)`. Most "
+            "common cause: upstream stream did not include a final usage "
+            "chunk (OpenAI-compat backends require "
+            "`stream_options.include_usage=true`).",
             extra={
                 "max_cost_msats": max_cost,
                 "model": response_data.get("model", "unknown"),
+                "response_keys": sorted(response_data.keys())
+                if isinstance(response_data, dict)
+                else None,
             },
         )
         return MaxCostData(
@@ -139,6 +146,22 @@ async def calculate_cost(  # todo: can be sync
     )
 
     if usd_cost > 0:
+        if input_tokens == 0 and output_tokens == 0:
+            logger.warning(
+                "Upstream reported a USD cost but no token counts — "
+                "billing the USD-derived cost while the dashboard will "
+                "show this request as `(0+0)` tokens. Check that the "
+                "upstream actually emits `usage.input_tokens` and "
+                "`usage.output_tokens` (OpenAI-compat streams require "
+                "`stream_options.include_usage=true`).",
+                extra={
+                    "model": response_data.get("model", "unknown"),
+                    "usd_cost": usd_cost,
+                    "usage_keys": sorted(usage_data.keys())
+                    if isinstance(usage_data, dict)
+                    else None,
+                },
+            )
         try:
             sats_per_usd = 1.0 / sats_usd_price()
             cost_in_sats = usd_cost * sats_per_usd
@@ -157,7 +180,17 @@ async def calculate_cost(  # todo: can be sync
                     input_msats = int(cost_in_msats * input_ratio)
                     output_msats = cost_in_msats - input_msats
                 else:
-                    output_msats = cost_in_msats
+                    # No tokens reported — model produced empty output, issue full refund
+                    logger.warning(
+                        "Zero tokens with non-zero USD cost — issuing full refund",
+                        extra={
+                            "usd_cost": usd_cost,
+                            "model": response_data.get("model", "unknown"),
+                        },
+                    )
+                    input_msats = 0
+                    output_msats = 0
+                    cost_in_msats = 0
 
             logger.info(
                 "Using cost from usage data/details",
@@ -239,10 +272,18 @@ async def calculate_cost(  # todo: can be sync
 
     if not (MSATS_PER_1K_OUTPUT_TOKENS and MSATS_PER_1K_INPUT_TOKENS):
         logger.warning(
-            "No token pricing configured, using base cost",
+            "No token pricing configured — billing at flat MaxCostData. "
+            "Token counts %s in the upstream response but cannot be "
+            "priced; the request will appear in dashboards with the "
+            "raw counts and a fixed max-cost charge.",
+            "are present"
+            if (input_tokens > 0 or output_tokens > 0)
+            else "are zero",
             extra={
                 "base_cost_msats": max_cost,
                 "model": response_data.get("model", "unknown"),
+                "input_tokens": input_tokens,
+                "output_tokens": output_tokens,
             },
         )
         return MaxCostData(

routstr/payment/models.py

@@ -351,6 +351,9 @@ async def _update_sats_pricing_once() -> None:
     from ..proxy import get_upstreams, refresh_model_maps
 
     upstreams = get_upstreams()
+    if not upstreams:
+        return
+
     sats_to_usd = sats_usd_price()
 
     updated_count = 0
@@ -364,7 +367,10 @@ async def _update_sats_pricing_once() -> None:
         updated_count += len(updated_models)
 
     if updated_count > 0:
-        logger.info("Updated sats pricing", extra={"models_updated": updated_count})
+        logger.info(
+            f"Updated sats pricing for {updated_count} models",
+            extra={"models_updated": updated_count},
+        )
         await refresh_model_maps()
 
 

routstr/upstream/gemini.py

@@ -1,17 +1,13 @@
 from __future__ import annotations
 
-import json
-from collections.abc import AsyncGenerator
 from typing import TYPE_CHECKING, Any
 
-from fastapi import Request
-from fastapi.responses import Response, StreamingResponse
-
+from . import gemini_messages
 from .base import BaseUpstreamProvider
 from .clients.gemini import GeminiClient
 
 if TYPE_CHECKING:
-    from ..core.db import ApiKey, AsyncSession, UpstreamProviderRow
+    from ..core.db import UpstreamProviderRow
     from ..payment.models import Model
 
 from ..core.logging import get_logger
@@ -20,6 +16,15 @@ logger = get_logger(__name__)
 
 
 class GeminiUpstreamProvider(BaseUpstreamProvider):
+    """Gemini provider — proxies through Gemini's OpenAI-compat surface.
+
+    The chat-completions, embeddings, and models paths all flow through
+    :meth:`BaseUpstreamProvider.forward_request`; we only override
+    ``get_request_base_url`` to redirect to ``{base}/openai/...`` and
+    ``_dispatch_anthropic_messages`` to inject thought-signatures on the
+    /v1/messages path (see :mod:`gemini_messages` for that rationale).
+    """
+
     provider_type = "gemini"
     default_base_url = "https://generativelanguage.googleapis.com/v1beta"
     platform_url = "https://aistudio.google.com/app/apikey"
@@ -40,7 +45,7 @@ class GeminiUpstreamProvider(BaseUpstreamProvider):
 
     @property
     def client(self) -> GeminiClient:
-        """Get or create the Gemini API client."""
+        """Get or create the Gemini API client (used for the models listing)."""
         if self._client is None:
             self._client = GeminiClient(api_key=self.api_key)
         return self._client
@@ -66,248 +71,66 @@ class GeminiUpstreamProvider(BaseUpstreamProvider):
         }
 
     def transform_model_name(self, model_id: str) -> str:
-        return model_id.removeprefix("gemini/")
+        """Reduce a routstr model id to the bare upstream Gemini name.
 
-    async def forward_request(
+        Gemini's OpenAI-compat surface expects the literal model id
+        (e.g. ``gemini-3.1-flash-lite-preview``) — no ``gemini/`` provider
+        prefix and no ``google/`` vendor sub-prefix. Take the last path
+        segment so we tolerate any of:
+
+            ``gemini-2.0-flash``
+            ``gemini/gemini-2.0-flash``
+            ``gemini/google/gemini-3.1-flash-lite-preview``
+        """
+        return model_id.rsplit("/", 1)[-1]
+
+    @property
+    def compat_base_url(self) -> str:
+        """Gemini's OpenAI-compat surface, regardless of what's stored.
+
+        Stored ``base_url`` may be ``.../v1beta`` (the native Gemini API
+        root) or ``.../v1beta/openai`` (already pointed at the compat
+        surface). Normalize to the latter.
+        """
+        return self.base_url.rstrip("/").removesuffix("/openai") + "/openai"
+
+    def get_request_base_url(
+        self, path: str, model_obj: "Model | None" = None
+    ) -> str:
+        """Route every proxied request to the OpenAI-compat surface.
+
+        Required because the stored ``base_url`` typically points at the
+        native Gemini API (``/v1beta``), but :meth:`forward_request`
+        forwards OpenAI-shaped paths (``/chat/completions``,
+        ``/embeddings``, ``/models``) which only exist under the
+        ``/openai`` subtree.
+        """
+        return self.compat_base_url
+
+    async def _dispatch_anthropic_messages(
         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:
-        # Remove provider prefix from model ID for Gemini API
-        if "/" in model_obj.id:
-            model_obj.id = model_obj.id.split("/", 1)[1]
+        model_obj: "Model",
+        *,
+        log_extra: dict[str, Any] | None = None,
+    ) -> tuple[bool, Any, str | None]:
+        """Dispatch /v1/messages via the gemini-specific httpx path.
 
-        if not path.startswith("chat/completions"):
-            return await super().forward_request(
-                request,
-                path,
-                headers,
-                request_body,
-                key,
-                max_cost_for_model,
-                session,
-                model_obj,
-            )
-
-        if not request_body:
-            return await super().forward_request(
-                request,
-                path,
-                headers,
-                request_body,
-                key,
-                max_cost_for_model,
-                session,
-                model_obj,
-            )
-
-        try:
-            openai_data = json.loads(request_body)
-            messages = openai_data.get("messages", [])
-            temperature = openai_data.get("temperature")
-            max_tokens = openai_data.get("max_tokens")
-            top_p = openai_data.get("top_p")
-            is_streaming = openai_data.get("stream", False)
-
-            logger.info(
-                "Processing Gemini request with client abstraction",
-                extra={
-                    "model": model_obj.id,
-                    "is_streaming": is_streaming,
-                    "message_count": len(messages),
-                    "key_hash": key.hashed_key[:8] + "...",
-                },
-            )
-
-            if is_streaming:
-                final_usage_data: dict | None = None
-
-                def usage_callback(usage_data: dict[str, Any]) -> None:
-                    """Callback to capture usage data during streaming"""
-                    nonlocal final_usage_data
-                    final_usage_data = usage_data
-
-                async def completion_callback(
-                    model: str, usage_data: dict[str, Any] | None
-                ) -> None:
-                    """Callback to handle payment when streaming completes"""
-                    nonlocal final_usage_data
-                    if usage_data:
-                        final_usage_data = usage_data
-
-                    payment_data = {
-                        "model": model,
-                        "usage": final_usage_data,
-                    }
-
-                    from ..auth import adjust_payment_for_tokens
-                    from ..core.db import create_session
-
-                    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,
-                                    payment_data,
-                                    new_session,
-                                    max_cost_for_model,
-                                )
-
-                                logger.info(
-                                    "Gemini streaming payment finalized",
-                                    extra={
-                                        "cost_data": cost_data,
-                                        "usage_data": final_usage_data,
-                                        "key_hash": key.hashed_key[:8] + "...",
-                                    },
-                                )
-                            except Exception as cost_error:
-                                logger.error(
-                                    "Error finalizing Gemini streaming payment",
-                                    extra={
-                                        "error": str(cost_error),
-                                        "key_hash": key.hashed_key[:8] + "...",
-                                    },
-                                )
-
-                response_generator = self.client.generate_content_stream(
-                    model=model_obj.id,
-                    messages=messages,
-                    temperature=temperature,
-                    max_tokens=max_tokens,
-                    top_p=top_p,
-                    usage_callback=usage_callback,
-                    completion_callback=completion_callback,
-                )
-
-                async def stream_with_cost() -> AsyncGenerator[bytes, None]:
-                    payment_finalized = False
-
-                    async def finalize_payment() -> None:
-                        nonlocal payment_finalized
-                        if payment_finalized:
-                            return
-                        from ..auth import adjust_payment_for_tokens
-                        from ..core.db import create_session
-
-                        async with create_session() as new_session:
-                            fresh_key = await new_session.get(
-                                key.__class__, key.hashed_key
-                            )
-                            if fresh_key:
-                                try:
-                                    await adjust_payment_for_tokens(
-                                        fresh_key,
-                                        {
-                                            "model": model_obj.id,
-                                            "usage": final_usage_data,
-                                        },
-                                        new_session,
-                                        max_cost_for_model,
-                                    )
-                                    payment_finalized = True
-                                except Exception as cost_error:
-                                    logger.error(
-                                        "Error finalizing Gemini streaming payment in fallback",
-                                        extra={
-                                            "error": str(cost_error),
-                                            "key_hash": key.hashed_key[:8] + "...",
-                                        },
-                                    )
-
-                    try:
-                        async for chunk in response_generator:
-                            sse_data = f"data: {json.dumps(chunk)}\n\n"
-                            yield sse_data.encode()
-                    except Exception as e:
-                        logger.error(
-                            "Error in Gemini streaming response",
-                            extra={
-                                "error": str(e),
-                                "error_type": type(e).__name__,
-                                "key_hash": key.hashed_key[:8] + "...",
-                            },
-                        )
-                        raise
-                    finally:
-                        if not payment_finalized:
-                            await finalize_payment()
-
-                return StreamingResponse(
-                    stream_with_cost(),
-                    media_type="text/event-stream",
-                    headers={"Cache-Control": "no-cache", "Connection": "keep-alive"},
-                )
-
-            else:
-                openai_format_response = await self.client.generate_content(
-                    model=model_obj.id,
-                    messages=messages,
-                    temperature=temperature,
-                    max_tokens=max_tokens,
-                    top_p=top_p,
-                )
-
-                from ..auth import adjust_payment_for_tokens
-
-                cost_data = await adjust_payment_for_tokens(
-                    key, openai_format_response, session, max_cost_for_model
-                )
-                await session.refresh(key)
-                remaining_balance_msats = key.balance
-                openai_format_response["cost"] = cost_data
-                openai_format_response["cost"]["sats_cost"] = (
-                    cost_data.get("total_msats", 0) // 1000
-                )
-                openai_format_response["cost"]["remaining_balance_msats"] = (
-                    remaining_balance_msats
-                )
-
-                logger.info(
-                    "Gemini non-streaming payment completed",
-                    extra={
-                        "cost_data": cost_data,
-                        "model": model_obj.id,
-                        "key_hash": key.hashed_key[:8] + "...",
-                    },
-                )
-
-                return Response(
-                    content=json.dumps(openai_format_response),
-                    media_type="application/json",
-                    headers={"Cache-Control": "no-cache"},
-                )
-
-        except Exception as e:
-            logger.error(
-                "Error in Gemini forward_request",
-                extra={
-                    "error": str(e),
-                    "error_type": type(e).__name__,
-                    "path": path,
-                    "key_hash": key.hashed_key[:8] + "...",
-                },
-            )
-            return await super().forward_request(
-                request,
-                path,
-                headers,
-                request_body,
-                key,
-                max_cost_for_model,
-                session,
-                model_obj,
-            )
+        See :mod:`routstr.upstream.gemini_messages` for the full rationale
+        (thought-signature injection, why litellm + the openai SDK can't
+        carry the required ``extra_content`` field).
+        """
+        return await gemini_messages.dispatch_gemini_messages(
+            request_body=request_body,
+            model_obj=model_obj,
+            base_url=self.compat_base_url,
+            api_key=self.api_key,
+            transform_model_name=self.transform_model_name,
+            log_extra=log_extra,
+        )
 
     async def _fetch_provider_models(self) -> dict:
-        """Fetch models from Gemini API."""
+        """Fetch models from Gemini API via the OpenAI-compat client."""
         try:
             models_data = await self.client.list_models()
 

routstr/upstream/gemini_messages.py

@@ -0,0 +1,480 @@
+"""Custom /v1/messages dispatcher for Gemini's OpenAI-compat endpoint.
+
+Why this exists
+---------------
+
+Gemini 2.5 / 3 thinking models reject inbound ``functionCall`` parts that
+lack a ``thought_signature`` field once any prior turn in the conversation
+contains a function call. Anthropic-Messages clients (Claude Code etc.)
+have no concept of thought signatures, so multi-turn tool conversations
+fail with::
+
+    Function call is missing a thought_signature in functionCall parts.
+
+Google's published escape hatch (https://ai.google.dev/gemini-api/docs/
+thought-signatures, FAQ #1) is the dummy signature
+``"skip_thought_signature_validator"`` placed at
+``tool_calls[i].extra_content.google.thought_signature`` for every tool
+call in the request. The hatch is documented specifically for
+"transferring a trace from a different model that does not include thought
+signatures" — exactly our case.
+
+Why we can't reach the wire via litellm
+---------------------------------------
+
+``litellm.anthropic.messages.acreate`` flows through the openai SDK, whose
+pydantic ``ChatCompletionMessageToolCall`` model silently drops unknown
+fields like ``extra_content``. Litellm has no openai-compat translator
+that emits ``extra_content.google.thought_signature``. So we bypass both
+litellm and the openai SDK at the transport layer.
+
+Pipeline
+--------
+
+1. Translate Anthropic body → OpenAI body via litellm's
+   ``AnthropicAdapter`` (the same translator
+   ``litellm.anthropic.messages.acreate`` uses internally).
+2. Inject ``extra_content.google.thought_signature`` on every
+   ``tool_calls[]`` entry.
+3. Set ``reasoning_effort="none"`` to disable Gemini's thinking pass.
+4. POST directly to ``{base_url}/chat/completions`` with ``stream=true``
+   via ``httpx`` (preserves arbitrary fields verbatim).
+5. Translate OpenAI streaming chunks → Anthropic SSE events.
+"""
+
+from __future__ import annotations
+
+import json
+import uuid
+from collections.abc import AsyncGenerator, AsyncIterator
+from typing import Any, Callable
+
+import httpx
+
+from ..core import get_logger
+from ..core.exceptions import UpstreamError
+from ..payment.models import Model
+from .messages_dispatch import (
+    ANTHROPIC_ONLY_FIELDS,
+    aggregate_anthropic_events_to_message,
+)
+
+logger = get_logger(__name__)
+
+DUMMY_THOUGHT_SIGNATURE = "skip_thought_signature_validator"
+
+# Mapping: OpenAI finish_reason → Anthropic stop_reason
+_FINISH_TO_STOP = {
+    "stop": "end_turn",
+    "length": "max_tokens",
+    "tool_calls": "tool_use",
+    "function_call": "tool_use",
+    "content_filter": "refusal",
+}
+
+
+def inject_thought_signatures(messages: list[dict]) -> None:
+    """Add ``extra_content.google.thought_signature`` to every tool_call.
+
+    Mutates ``messages`` in place. Idempotent: existing signatures are not
+    overwritten.
+    """
+    for msg in messages:
+        tool_calls = msg.get("tool_calls")
+        if not isinstance(tool_calls, list):
+            continue
+        for tc in tool_calls:
+            if not isinstance(tc, dict):
+                continue
+            extra = tc.get("extra_content")
+            if not isinstance(extra, dict):
+                extra = tc["extra_content"] = {}
+            google_cfg = extra.get("google")
+            if not isinstance(google_cfg, dict):
+                google_cfg = extra["google"] = {}
+            google_cfg.setdefault("thought_signature", DUMMY_THOUGHT_SIGNATURE)
+
+
+def _translate_anthropic_to_openai(body: dict, model: str) -> dict:
+    """Use litellm's translator to convert an Anthropic /messages body to
+    OpenAI /chat/completions kwargs.
+
+    Imported lazily because the litellm internal path is heavy and not
+    needed for any other code path in routstr.
+    """
+    from litellm.llms.anthropic.experimental_pass_through.adapters.transformation import (  # noqa: E501
+        AnthropicAdapter,
+    )
+
+    kwargs = {"model": model, **body}
+    translated = AnthropicAdapter().translate_completion_input_params(kwargs)
+    if translated is None:
+        raise UpstreamError(
+            "Failed to translate Anthropic body to OpenAI format",
+            status_code=500,
+        )
+    return dict(translated)
+
+
+def _sse_event(event_type: str, payload: dict) -> bytes:
+    return f"event: {event_type}\ndata: {json.dumps(payload)}\n\n".encode()
+
+
+async def _openai_chunks_to_anthropic_events(
+    line_iter: AsyncIterator[str], requested_model: str | None
+) -> AsyncGenerator[bytes, None]:
+    """Translate an OpenAI chat-completions SSE byte stream into the
+    Anthropic-Messages SSE event sequence.
+
+    Maintains per-chunk state across:
+
+    * one optional text content block (lazy-opened on first text delta)
+    * any number of tool_use blocks indexed by openai's ``delta.tool_calls[].index``
+    * final ``stop_reason`` / ``usage`` carried out via ``message_delta`` /
+      ``message_stop``
+    """
+    msg_id = f"msg_{uuid.uuid4().hex[:24]}"
+    started = False
+    text_block_idx: int | None = None
+    tool_block_indices: dict[int, int] = {}
+    next_block_idx = 0
+    final_finish_reason: str | None = None
+    final_usage: dict[str, int] = {"input_tokens": 0, "output_tokens": 0}
+
+    def open_text_block() -> bytes:
+        nonlocal text_block_idx, next_block_idx
+        text_block_idx = next_block_idx
+        next_block_idx += 1
+        return _sse_event(
+            "content_block_start",
+            {
+                "type": "content_block_start",
+                "index": text_block_idx,
+                "content_block": {"type": "text", "text": ""},
+            },
+        )
+
+    def open_tool_block(delta_idx: int, tc: dict) -> bytes:
+        nonlocal next_block_idx
+        block_idx = next_block_idx
+        next_block_idx += 1
+        tool_block_indices[delta_idx] = block_idx
+        fn = tc.get("function") or {}
+        return _sse_event(
+            "content_block_start",
+            {
+                "type": "content_block_start",
+                "index": block_idx,
+                "content_block": {
+                    "type": "tool_use",
+                    "id": tc.get("id") or f"toolu_{uuid.uuid4().hex[:24]}",
+                    "name": fn.get("name") or "",
+                    "input": {},
+                },
+            },
+        )
+
+    def close_block(idx: int) -> bytes:
+        return _sse_event(
+            "content_block_stop",
+            {"type": "content_block_stop", "index": idx},
+        )
+
+    async for raw_line in line_iter:
+        line = raw_line.strip()
+        if not line:
+            continue
+        if not line.startswith("data:"):
+            continue
+        payload = line[5:].lstrip()
+        if not payload or payload == "[DONE]":
+            continue
+        try:
+            chunk = json.loads(payload)
+        except json.JSONDecodeError:
+            continue
+        if not isinstance(chunk, dict):
+            continue
+
+        if not started:
+            started = True
+            yield _sse_event(
+                "message_start",
+                {
+                    "type": "message_start",
+                    "message": {
+                        "id": chunk.get("id") or msg_id,
+                        "type": "message",
+                        "role": "assistant",
+                        "model": requested_model or chunk.get("model") or "",
+                        "content": [],
+                        "stop_reason": None,
+                        "stop_sequence": None,
+                        "usage": {
+                            "input_tokens": 0,
+                            "output_tokens": 0,
+                        },
+                    },
+                },
+            )
+
+        usage = chunk.get("usage")
+        if isinstance(usage, dict):
+            in_tok = usage.get("prompt_tokens") or usage.get("input_tokens") or 0
+            out_tok = usage.get("completion_tokens") or usage.get("output_tokens") or 0
+            if in_tok:
+                final_usage["input_tokens"] = int(in_tok)
+            if out_tok:
+                final_usage["output_tokens"] = int(out_tok)
+
+        choices = chunk.get("choices") or []
+        if not choices:
+            continue
+        choice = choices[0] if isinstance(choices[0], dict) else {}
+        delta = choice.get("delta") or {}
+        if not isinstance(delta, dict):
+            delta = {}
+
+        # Text delta
+        text = delta.get("content")
+        if isinstance(text, str) and text:
+            if text_block_idx is None:
+                yield open_text_block()
+            yield _sse_event(
+                "content_block_delta",
+                {
+                    "type": "content_block_delta",
+                    "index": text_block_idx,
+                    "delta": {"type": "text_delta", "text": text},
+                },
+            )
+
+        # Tool call deltas
+        tool_calls_delta = delta.get("tool_calls")
+        if isinstance(tool_calls_delta, list):
+            for tc in tool_calls_delta:
+                if not isinstance(tc, dict):
+                    continue
+                d_idx = int(tc.get("index") or 0)
+                if d_idx not in tool_block_indices:
+                    yield open_tool_block(d_idx, tc)
+                block_idx = tool_block_indices[d_idx]
+                fn = tc.get("function") or {}
+                args = fn.get("arguments")
+                if isinstance(args, str) and args:
+                    yield _sse_event(
+                        "content_block_delta",
+                        {
+                            "type": "content_block_delta",
+                            "index": block_idx,
+                            "delta": {
+                                "type": "input_json_delta",
+                                "partial_json": args,
+                            },
+                        },
+                    )
+
+        finish = choice.get("finish_reason")
+        if finish:
+            final_finish_reason = finish
+
+    # Close any open content blocks
+    if text_block_idx is not None:
+        yield close_block(text_block_idx)
+    for block_idx in tool_block_indices.values():
+        yield close_block(block_idx)
+
+    # message_delta with stop_reason and usage
+    stop_reason = _FINISH_TO_STOP.get(final_finish_reason or "", "end_turn")
+    yield _sse_event(
+        "message_delta",
+        {
+            "type": "message_delta",
+            "delta": {"stop_reason": stop_reason, "stop_sequence": None},
+            "usage": final_usage,
+        },
+    )
+    yield _sse_event("message_stop", {"type": "message_stop"})
+
+
+async def _post_and_stream(
+    base_url: str,
+    api_key: str,
+    payload: dict,
+    log_extra: dict[str, Any] | None,
+) -> tuple[httpx.AsyncClient, httpx.Response]:
+    """POST to upstream chat-completions and return (client, response) for
+    streaming. Caller is responsible for closing both."""
+    url = f"{base_url.rstrip('/')}/chat/completions"
+    client = httpx.AsyncClient(timeout=httpx.Timeout(120.0, read=120.0))
+    try:
+        request = client.build_request(
+            "POST",
+            url,
+            json=payload,
+            headers={
+                "Authorization": f"Bearer {api_key}",
+                "Content-Type": "application/json",
+                "Accept": "text/event-stream",
+            },
+        )
+        response = await client.send(request, stream=True)
+    except Exception as exc:
+        await client.aclose()
+        logger.error(
+            "Gemini messages dispatch HTTP error",
+            extra={"error": str(exc), "url": url, **(log_extra or {})},
+        )
+        raise UpstreamError(
+            f"Failed to reach Gemini upstream: {exc}", status_code=502
+        ) from exc
+
+    if response.status_code >= 400:
+        try:
+            body_bytes = await response.aread()
+        finally:
+            await response.aclose()
+            await client.aclose()
+        body_text = body_bytes.decode("utf-8", errors="replace")
+        logger.error(
+            "Gemini messages dispatch upstream error",
+            extra={
+                "status_code": response.status_code,
+                "body": body_text[:1000],
+                "url": url,
+                **(log_extra or {}),
+            },
+        )
+        raise UpstreamError(
+            f"Upstream error via gemini compat: {body_text}",
+            status_code=response.status_code,
+        )
+
+    return client, response
+
+
+async def dispatch_gemini_messages(
+    *,
+    request_body: bytes | None,
+    model_obj: Model,
+    base_url: str,
+    api_key: str,
+    transform_model_name: Callable[[str], str],
+    log_extra: dict[str, Any] | None = None,
+) -> tuple[bool, Any, str | None]:
+    """Dispatch a /v1/messages request to Gemini's OpenAI-compat endpoint
+    with thought-signature injection.
+
+    Returns ``(client_stream, result, requested_model)`` where ``result``
+    is either an ``AsyncIterator[bytes]`` of Anthropic-format SSE events
+    (for streaming clients) or an Anthropic Message dict (after the caller
+    aggregates).
+    """
+    if not request_body:
+        raise UpstreamError(
+            "Missing request body for /v1/messages", status_code=400
+        )
+
+    try:
+        body: dict = json.loads(request_body)
+    except json.JSONDecodeError as exc:
+        raise UpstreamError(
+            f"Invalid JSON in /v1/messages body: {exc}", status_code=400
+        ) from exc
+
+    body.pop("model", None)
+    client_stream = bool(body.pop("stream", False))
+
+    # Anthropic-Messages-only fields that don't translate to OpenAI
+    # Chat Completions. litellm's translator passes through unknown
+    # top-level fields verbatim and Gemini's compat surface 400s on
+    # unknown names like ``context_management`` / ``output_config``.
+    dropped: dict[str, Any] = {}
+    for field in ANTHROPIC_ONLY_FIELDS:
+        if field in body:
+            dropped[field] = body.pop(field)
+    if dropped:
+        logger.debug(
+            "Dropped anthropic-only fields before gemini compat dispatch",
+            extra={"dropped_keys": sorted(dropped.keys())},
+        )
+
+    requested_model = (
+        (model_obj.forwarded_model_id or model_obj.id) if model_obj else None
+    )
+    upstream_model = transform_model_name(model_obj.id)
+
+    openai_kwargs = _translate_anthropic_to_openai(body, upstream_model)
+
+    messages = openai_kwargs.get("messages") or []
+    if isinstance(messages, list):
+        inject_thought_signatures(messages)
+
+    # Disable Gemini's thinking pass; the dummy signature already lifts
+    # validation, but skipping thinking entirely avoids degraded model
+    # output and keeps tool-calling deterministic.
+    openai_kwargs.setdefault("reasoning_effort", "none")
+    openai_kwargs["stream"] = True
+    openai_kwargs["model"] = upstream_model
+    # OpenAI-compat backends (including Gemini's) only emit a final
+    # ``usage`` chunk when the request opts in via this flag. Without it
+    # the cost-calculation pipeline can't read real token counts and
+    # falls back to MaxCostData billing.
+    existing_stream_options = openai_kwargs.get("stream_options")
+    merged_stream_options = (
+        dict(existing_stream_options)
+        if isinstance(existing_stream_options, dict)
+        else {}
+    )
+    merged_stream_options.setdefault("include_usage", True)
+    openai_kwargs["stream_options"] = merged_stream_options
+
+    logger.info(
+        "Dispatching /v1/messages via gemini compat (httpx)",
+        extra={
+            "model": upstream_model,
+            "client_stream": client_stream,
+            "messages_with_tool_calls": sum(
+                1 for m in messages if isinstance(m, dict) and m.get("tool_calls")
+            ),
+            **(log_extra or {}),
+        },
+    )
+
+    http_client, response = await _post_and_stream(
+        base_url, api_key, openai_kwargs, log_extra
+    )
+
+    async def line_iter() -> AsyncGenerator[str, None]:
+        try:
+            async for line in response.aiter_lines():
+                yield line
+        finally:
+            await response.aclose()
+            await http_client.aclose()
+
+    anthropic_event_iter = _openai_chunks_to_anthropic_events(
+        line_iter(), requested_model
+    )
+
+    if not client_stream:
+        # Aggregate the Anthropic SSE byte stream into a single Message dict
+        # so the rest of the pipeline (cost calc, metadata injection,
+        # response building) can treat it identically to a non-streaming
+        # litellm response.
+        try:
+            aggregated = await aggregate_anthropic_events_to_message(
+                anthropic_event_iter
+            )
+        except Exception as exc:
+            logger.error(
+                "Failed to aggregate Gemini compat events into message",
+                extra={"error": str(exc), **(log_extra or {})},
+            )
+            raise UpstreamError(
+                f"Failed to aggregate upstream stream: {exc}",
+                status_code=502,
+            ) from exc
+        return client_stream, aggregated, requested_model
+
+    return client_stream, anthropic_event_iter, requested_model

routstr/upstream/helpers.py

@@ -197,13 +197,14 @@ async def init_upstreams() -> list[BaseUpstreamProvider]:
         existing_providers = result.all()
 
         if not existing_providers:
-            logger.info(
-                "No upstream providers found in database, seeding from settings"
-            )
             await _seed_providers_from_settings(session, settings)
             await session.commit()
             result = await session.exec(select(UpstreamProviderRow))
             existing_providers = result.all()
+            if existing_providers:
+                logger.info(
+                    f"Seeded {len(existing_providers)} upstream providers from settings"
+                )
 
         async def _init_single_provider(
             provider_row: UpstreamProviderRow,

routstr/upstream/litellm_routing.py

@@ -0,0 +1,162 @@
+"""Map an upstream `base_url` to the correct litellm provider prefix.
+
+Used by `BaseUpstreamProvider.get_litellm_provider_prefix` so that custom /
+generic provider rows (which inherit the base class default) get routed to
+the right litellm backend instead of falling back to `openai/`.
+
+The table is compiled from litellm 1.74's
+`litellm/litellm_core_utils/get_llm_provider_logic.py` (the
+`openai_compatible_endpoints` table) plus the providers documented at
+https://docs.litellm.ai/docs/providers. Substring match is used so that
+URLs with paths, ports, regional subdomains, etc. all resolve correctly.
+
+Order matters: more specific needles must appear before more generic ones
+(e.g. ``openai.azure.com`` before ``api.openai.com``).
+"""
+
+from __future__ import annotations
+
+import os
+from urllib.parse import urlsplit
+
+import litellm
+
+DEFAULT_PREFIX = "openai/"
+
+LITELLM_HOST_PREFIX_MAP: tuple[tuple[str, str], ...] = (
+    # Azure must win over api.openai.com because the host ends with
+    # `openai.azure.com` and we don't want it picked up as plain OpenAI.
+    ("openai.azure.com", "azure/"),
+    # Google
+    ("generativelanguage.googleapis.com", "gemini/"),
+    ("aiplatform.googleapis.com", "vertex_ai/"),
+    # First-class providers with native litellm prefixes
+    ("api.openai.com", "openai/"),
+    ("api.anthropic.com", "anthropic/"),
+    ("api.groq.com", "groq/"),
+    ("api.fireworks.ai", "fireworks_ai/"),
+    ("api.x.ai", "xai/"),
+    ("api.perplexity.ai", "perplexity/"),
+    ("openrouter.ai", "openrouter/"),
+    ("api.deepseek.com", "deepseek/"),
+    ("api.together.xyz", "together_ai/"),
+    ("codestral.mistral.ai", "codestral/"),
+    ("api.mistral.ai", "mistral/"),
+    ("api.cohere.com", "cohere_chat/"),
+    ("api.cohere.ai", "cohere_chat/"),
+    ("api.deepinfra.com", "deepinfra/"),
+    ("api.endpoints.anyscale.com", "anyscale/"),
+    ("api.cerebras.ai", "cerebras/"),
+    ("inference.baseten.co", "baseten/"),
+    ("api.sambanova.ai", "sambanova/"),
+    ("api.ai21.com", "ai21_chat/"),
+    ("api.friendli.ai", "friendliai/"),
+    ("api.galadriel.com", "galadriel/"),
+    ("api.llama.com", "meta_llama/"),
+    ("api.featherless.ai", "featherless_ai/"),
+    ("inference.api.nscale.com", "nscale/"),
+    ("dashscope-intl.aliyuncs.com", "dashscope/"),
+    ("api.moonshot.ai", "moonshot/"),
+    ("api.moonshot.cn", "moonshot/"),
+    ("api.minimax.io", "minimax/"),
+    ("api.minimaxi.com", "minimax/"),
+    ("platform.publicai.co", "publicai/"),
+    ("api.synthetic.new", "synthetic/"),
+    ("api.stima.tech", "apertis/"),
+    ("nano-gpt.com", "nano-gpt/"),
+    ("api.poe.com", "poe/"),
+    ("llm.chutes.ai", "chutes/"),
+    ("api.v0.dev", "v0/"),
+    ("api.lambda.ai", "lambda_ai/"),
+    ("api.hyperbolic.xyz", "hyperbolic/"),
+    ("ai-gateway.vercel.sh", "vercel_ai_gateway/"),
+    ("api.inference.wandb.ai", "wandb/"),
+    ("integrate.api.nvidia.com", "nvidia_nim/"),
+    ("api.studio.nebius.com", "nebius/"),
+    ("api.novita.ai", "novita/"),
+    ("ark.cn-beijing.volces.com", "volcengine/"),
+    ("api.voyageai.com", "voyage/"),
+    ("api.jina.ai", "jina_ai/"),
+    ("api.aimlapi.com", "aiml/"),
+    ("api.snowflakecomputing.com", "snowflake/"),
+    ("databricks.com", "databricks/"),
+    ("huggingface.co", "huggingface/"),
+)
+
+# Substrings that indicate an Ollama deployment regardless of port/scheme.
+OLLAMA_HOST_HINTS: tuple[str, ...] = (
+    "localhost:11434",
+    "127.0.0.1:11434",
+    "ollama",
+)
+
+
+def detect_litellm_prefix(
+    base_url: str | None, default: str = DEFAULT_PREFIX
+) -> str:
+    """Return the litellm provider prefix (`"<provider>/"`) for `base_url`.
+
+    Falls back to `default` when the host doesn't match any known provider.
+    The default is `openai/` because every unmatched OpenAI-compatible
+    server is, by definition, an OpenAI-compatible server.
+    """
+    if not base_url:
+        return default
+
+    parsed = urlsplit(base_url)
+    host = parsed.netloc.lower() or base_url.lower()
+
+    for needle, prefix in LITELLM_HOST_PREFIX_MAP:
+        if needle in host:
+            return prefix
+
+    if any(hint in host for hint in OLLAMA_HOST_HINTS):
+        return "ollama_chat/"
+
+    return default
+
+
+_configured = False
+
+
+def configure_litellm() -> None:
+    """Apply litellm global settings used by the messages-dispatch path.
+
+    Idempotent: safe to call from both app startup and module-level
+    initializers without side effects on the second invocation.
+
+    Settings applied:
+
+    * ``LITELLM_DEBUG=1`` enables litellm's verbose debug logger.
+    * Forces the Anthropic-messages adapter to call OpenAI Chat Completions
+      (POST ``/chat/completions``) instead of the Responses API (POST
+      ``/responses``) for ``openai/``-prefixed providers. OpenAI-compatible
+      upstreams like Google's generativelanguage compat endpoint expose
+      ``/chat/completions`` but not ``/responses``, which would 404. Set
+      ``LITELLM_USE_RESPONSES_API_FOR_ANTHROPIC_MESSAGES=1`` to opt out.
+    * Silently drops Anthropic-Messages-only parameters (``thinking``,
+      ``cache_control``, ``context_management``, ...) when translating to
+      providers that don't accept them, instead of raising
+      ``UnsupportedParamsError``. Set ``LITELLM_STRICT_PARAMS=1`` to opt
+      out.
+    """
+    global _configured
+    if _configured:
+        return
+
+    if os.getenv("LITELLM_DEBUG") == "1":
+        try:
+            litellm._turn_on_debug()  # type: ignore[no-untyped-call]
+        except Exception:
+            pass
+
+    if os.getenv("LITELLM_USE_RESPONSES_API_FOR_ANTHROPIC_MESSAGES") != "1":
+        try:
+            litellm.use_chat_completions_url_for_anthropic_messages = True
+        except Exception:
+            pass
+
+    if os.getenv("LITELLM_STRICT_PARAMS") != "1":
+        litellm.drop_params = True
+
+    _configured = True

routstr/upstream/messages_dispatch.py

@@ -0,0 +1,458 @@
+"""Pure helpers for translating ``/v1/messages`` to upstream chat completions
+via litellm.
+
+This module owns the litellm/Anthropic-Messages translation layer:
+
+* SSE parsing (``parse_sse_blocks``, ``events_from_chunk``)
+* Payload coercion (``coerce_litellm_payload``)
+* Stream aggregation (``aggregate_anthropic_events_to_message``) — drains
+  a streamed Anthropic event sequence into a single Message dict
+* Per-event annotation for streaming (``annotate_event``,
+  ``stream_annotated_events``) — handles the model-rewrite + token-tally
+  bookkeeping shared by the bearer-key and x-cashu streaming paths
+* The dispatch entry point (``dispatch_anthropic_messages``)
+* Refund math (``compute_refund``)
+
+Nothing in here touches ``BaseUpstreamProvider``; the thin instance methods
+on the provider class forward to these functions and only retain logic that
+genuinely needs ``self`` (cost adjustment, metadata injection, refund
+sending).
+"""
+
+from __future__ import annotations
+
+import json
+from collections.abc import AsyncGenerator, AsyncIterator
+from typing import Any, Callable, NamedTuple, cast
+
+import litellm
+
+from ..core import get_logger
+from ..core.exceptions import UpstreamError
+from ..payment.models import Model
+
+logger = get_logger(__name__)
+
+# Anthropic-Messages-only fields that don't translate to OpenAI
+# Chat Completions. ``litellm.drop_params`` only filters *known*
+# unsupported params; these newer/extension fields get passed through
+# verbatim and the upstream rejects them with a 400. Pop them here so the
+# request reaches the upstream cleanly.
+ANTHROPIC_ONLY_FIELDS: tuple[str, ...] = (
+    "thinking",
+    "cache_control",
+    "context_management",
+    "output_config",
+    "mcp_servers",
+    "service_tier",
+    "anthropic_version",
+    "anthropic_beta",
+)
+
+
+def coerce_litellm_payload(payload: object) -> dict:
+    """Convert a litellm event into a plain dict.
+
+    Non-streaming responses come back as Anthropic-shaped pydantic models
+    or dicts. Streaming may yield raw bytes/str (SSE-encoded); those go
+    through ``events_from_chunk`` instead, not here.
+    """
+    if isinstance(payload, dict):
+        return dict(payload)
+    if hasattr(payload, "model_dump"):
+        return cast(dict, payload.model_dump())
+    raise TypeError(f"Cannot coerce {type(payload).__name__} to dict")
+
+
+def parse_sse_blocks(buffer: bytes) -> tuple[list[dict], bytes]:
+    """Parse complete SSE event blocks out of a byte buffer.
+
+    Returns (events, remaining_buffer). Events are JSON objects parsed from
+    one or more ``data:`` lines per block. Comments, blank lines, and
+    ``[DONE]`` sentinels are ignored. A trailing partial block is preserved
+    in remaining_buffer.
+    """
+    events: list[dict] = []
+    while True:
+        sep = buffer.find(b"\n\n")
+        if sep < 0:
+            sep_rn = buffer.find(b"\r\n\r\n")
+            if sep_rn < 0:
+                break
+            block = buffer[:sep_rn]
+            buffer = buffer[sep_rn + 4 :]
+        else:
+            block = buffer[:sep]
+            buffer = buffer[sep + 2 :]
+
+        data_lines: list[str] = []
+        for raw_line in block.replace(b"\r\n", b"\n").split(b"\n"):
+            line = raw_line.decode("utf-8", errors="replace")
+            if line.startswith(":"):
+                continue
+            if line.startswith("data:"):
+                data_lines.append(line[5:].lstrip())
+        if not data_lines:
+            continue
+        payload = "\n".join(data_lines).strip()
+        if not payload or payload == "[DONE]":
+            continue
+        try:
+            obj = json.loads(payload)
+        except json.JSONDecodeError:
+            continue
+        if isinstance(obj, dict):
+            events.append(obj)
+    return events, buffer
+
+
+def events_from_chunk(
+    chunk: object, sse_buffer: bytes
+) -> tuple[list[dict], bytes]:
+    """Normalize a stream chunk into one or more event dicts.
+
+    ``litellm.anthropic.messages.acreate(stream=True)`` yields raw SSE
+    bytes in practice; some adapters yield strings or typed events. Handle
+    all three.
+    """
+    if isinstance(chunk, (bytes, bytearray)):
+        sse_buffer += bytes(chunk)
+        events, sse_buffer = parse_sse_blocks(sse_buffer)
+        return events, sse_buffer
+    if isinstance(chunk, str):
+        sse_buffer += chunk.encode("utf-8")
+        events, sse_buffer = parse_sse_blocks(sse_buffer)
+        return events, sse_buffer
+    return [coerce_litellm_payload(chunk)], sse_buffer
+
+
+async def aggregate_anthropic_events_to_message(
+    iterator: AsyncIterator[Any],
+) -> dict:
+    """Drain an Anthropic-Messages event iterator into a single Message dict.
+
+    Produces the shape ``litellm.anthropic.messages.acreate(stream=False)``
+    would have returned. Used to transparently stream from upstream while
+    still returning a non-streaming response to the client. Lets us
+    sidestep upstream quirks (e.g. Fireworks rejects ``max_tokens > 4096``
+    unless ``stream=true``) without leaking that into client-visible
+    behavior.
+    """
+    sse_buffer = b""
+    message: dict = {}
+    blocks: list[dict] = []
+    partial_json: dict[int, str] = {}
+    final_stop_reason: str | None = None
+    final_stop_sequence: str | None = None
+    final_usage: dict[str, Any] = {}
+    final_model: str | None = None
+
+    async for chunk in iterator:
+        events, sse_buffer = events_from_chunk(chunk, sse_buffer)
+        for event in events:
+            etype = event.get("type")
+            if etype == "message_start":
+                raw = event.get("message") or {}
+                if isinstance(raw, dict):
+                    message = dict(raw)
+                    existing = message.get("content")
+                    blocks = list(existing) if isinstance(existing, list) else []
+                    usage = message.get("usage")
+                    if isinstance(usage, dict):
+                        final_usage = dict(usage)
+                    if isinstance(message.get("model"), str):
+                        final_model = message["model"]
+            elif etype == "content_block_start":
+                idx = int(event.get("index") or 0)
+                cb = event.get("content_block") or {}
+                cb_dict = dict(cb) if isinstance(cb, dict) else {}
+                while len(blocks) <= idx:
+                    blocks.append({})
+                blocks[idx] = cb_dict
+            elif etype == "content_block_delta":
+                idx = int(event.get("index") or 0)
+                if idx >= len(blocks):
+                    continue
+                delta = event.get("delta") or {}
+                if not isinstance(delta, dict):
+                    continue
+                dtype = delta.get("type")
+                block = blocks[idx]
+                if dtype == "text_delta":
+                    block["text"] = (block.get("text") or "") + (
+                        delta.get("text") or ""
+                    )
+                elif dtype == "input_json_delta":
+                    partial_json[idx] = partial_json.get(idx, "") + (
+                        delta.get("partial_json") or ""
+                    )
+                elif dtype == "thinking_delta":
+                    block["thinking"] = (block.get("thinking") or "") + (
+                        delta.get("thinking") or ""
+                    )
+                elif dtype == "signature_delta":
+                    block["signature"] = (block.get("signature") or "") + (
+                        delta.get("signature") or ""
+                    )
+            elif etype == "content_block_stop":
+                idx = int(event.get("index") or 0)
+                raw_json = partial_json.pop(idx, None)
+                if raw_json is not None and idx < len(blocks):
+                    try:
+                        blocks[idx]["input"] = (
+                            json.loads(raw_json) if raw_json else {}
+                        )
+                    except json.JSONDecodeError:
+                        blocks[idx]["input"] = raw_json
+            elif etype == "message_delta":
+                delta = event.get("delta") or {}
+                if isinstance(delta, dict):
+                    if "stop_reason" in delta:
+                        final_stop_reason = delta.get("stop_reason")
+                    if "stop_sequence" in delta:
+                        final_stop_sequence = delta.get("stop_sequence")
+                usage = event.get("usage")
+                if isinstance(usage, dict):
+                    final_usage.update(usage)
+            # message_stop: nothing to merge
+
+    if not message:
+        # Upstream returned no message_start; expose what we can so the
+        # client at least sees the assembled content.
+        message = {
+            "id": "",
+            "type": "message",
+            "role": "assistant",
+            "content": [],
+        }
+
+    message["content"] = blocks
+    if final_model and not message.get("model"):
+        message["model"] = final_model
+    if final_stop_reason is not None:
+        message["stop_reason"] = final_stop_reason
+    if final_stop_sequence is not None:
+        message["stop_sequence"] = final_stop_sequence
+    if final_usage:
+        existing_usage = message.get("usage")
+        merged = dict(existing_usage) if isinstance(existing_usage, dict) else {}
+        merged.update(final_usage)
+        message["usage"] = merged
+    return message
+
+
+class AnnotatedEvent(NamedTuple):
+    """One Anthropic SSE event after model-rewrite + token-tally bookkeeping.
+
+    ``sse_bytes`` is the wire-ready ``event:`` / ``data:`` block; the two
+    streaming paths in ``BaseUpstreamProvider`` consume ``sse_bytes`` plus
+    the tallies and only differ in whether they stream live or buffer
+    first.
+    """
+
+    event: dict
+    sse_bytes: bytes
+    input_tokens: int
+    output_tokens: int
+    model: str | None
+
+
+def annotate_event(event: dict, requested_model: str | None) -> AnnotatedEvent:
+    """Rewrite ``model`` fields and extract per-event token / model info.
+
+    Mutates ``event`` in place when ``requested_model`` is set so the
+    upstream's true model name doesn't leak to the client.
+    """
+    if requested_model:
+        msg = event.get("message")
+        if isinstance(msg, dict) and "model" in msg:
+            msg["model"] = requested_model
+        if "model" in event:
+            event["model"] = requested_model
+
+    in_tokens = 0
+    out_tokens = 0
+    model: str | None = None
+
+    msg_for_meta = event.get("message")
+    if isinstance(msg_for_meta, dict):
+        if msg_for_meta.get("model"):
+            model = str(msg_for_meta["model"])
+        usage = msg_for_meta.get("usage")
+        if isinstance(usage, dict):
+            in_tokens += int(usage.get("input_tokens") or 0)
+            out_tokens += int(usage.get("output_tokens") or 0)
+
+    if isinstance(event.get("usage"), dict):
+        usage = event["usage"]
+        in_tokens += int(usage.get("input_tokens") or 0)
+        out_tokens += int(usage.get("output_tokens") or 0)
+
+    event_type = str(event.get("type") or "")
+    payload = json.dumps(event)
+    if event_type:
+        sse_bytes = f"event: {event_type}\ndata: {payload}\n\n".encode()
+    else:
+        sse_bytes = f"data: {payload}\n\n".encode()
+
+    return AnnotatedEvent(event, sse_bytes, in_tokens, out_tokens, model)
+
+
+async def stream_annotated_events(
+    iterator: AsyncIterator[Any],
+    requested_model: str | None,
+) -> AsyncGenerator[AnnotatedEvent, None]:
+    """Yield annotated, SSE-serialized events from a litellm stream.
+
+    Both streaming paths in ``BaseUpstreamProvider`` consume this; the only
+    divergence between them — yield-as-you-go vs buffer-then-replay — stays
+    in the caller.
+    """
+    sse_buffer = b""
+    async for chunk in iterator:
+        events, sse_buffer = events_from_chunk(chunk, sse_buffer)
+        for event in events:
+            yield annotate_event(event, requested_model)
+
+
+def compute_refund(amount: int, unit: str, cost_msats: int) -> int:
+    if unit == "msat":
+        return amount - cost_msats
+    if unit == "sat":
+        return amount - (cost_msats + 999) // 1000
+    raise ValueError(f"Invalid unit: {unit}")
+
+
+async def dispatch_anthropic_messages(
+    *,
+    request_body: bytes | None,
+    model_obj: Model,
+    base_url: str,
+    api_key: str,
+    provider_prefix: str,
+    transform_model_name: Callable[[str], str],
+    log_extra: dict[str, Any] | None = None,
+) -> tuple[bool, Any, str | None]:
+    """Call ``litellm.anthropic.messages.acreate`` and return
+    ``(client_stream, result, requested_model)``.
+
+    Shared by the bearer-key and x-cashu paths. Raises :class:`UpstreamError`
+    on bad input or upstream failure.
+    """
+    if not request_body:
+        raise UpstreamError(
+            "Missing request body for /v1/messages", status_code=400
+        )
+
+    try:
+        body: dict = json.loads(request_body)
+    except json.JSONDecodeError as exc:
+        raise UpstreamError(
+            f"Invalid JSON in /v1/messages body: {exc}", status_code=400
+        ) from exc
+
+    body.pop("model", None)
+    # `stream` here is what the **client** asked for. Upstream is always
+    # streamed (see `upstream_stream` below); when the client asked for a
+    # non-streaming response we drain and aggregate the events into a
+    # single Anthropic Message dict before returning. This sidesteps
+    # provider-specific non-streaming caps (e.g. Fireworks rejects
+    # `max_tokens > 4096` unless `stream=true`).
+    client_stream = bool(body.pop("stream", False))
+    upstream_stream = True
+
+    dropped: dict[str, Any] = {}
+    for field in ANTHROPIC_ONLY_FIELDS:
+        if field in body:
+            dropped[field] = body.pop(field)
+    if dropped:
+        logger.debug(
+            "Dropped anthropic-only fields before litellm dispatch",
+            extra={"dropped_keys": sorted(dropped.keys())},
+        )
+
+    # Convention: `model.id` is the canonical upstream model name;
+    # `forwarded_model_id` is the public alias the internal API exposes
+    # and echoes back to the client.
+    requested_model = (
+        (model_obj.forwarded_model_id or model_obj.id) if model_obj else None
+    )
+    upstream_model = transform_model_name(model_obj.id)
+    litellm_model = f"{provider_prefix}{upstream_model}"
+
+    kwargs: dict = {
+        "model": litellm_model,
+        "api_base": base_url,
+        "api_key": api_key,
+        "stream": upstream_stream,
+        **body,
+    }
+
+    logger.info(
+        "Dispatching /v1/messages via litellm",
+        extra={
+            "model": litellm_model,
+            "resolved_provider": provider_prefix.rstrip("/"),
+            "client_stream": client_stream,
+            "upstream_stream": upstream_stream,
+            **(log_extra or {}),
+        },
+    )
+
+    try:
+        result = await litellm.anthropic.messages.acreate(**kwargs)
+    except Exception as exc:
+        exc_message = getattr(exc, "message", None) or str(exc) or repr(exc)
+        exc_status = getattr(exc, "status_code", None)
+        exc_response = getattr(exc, "response", None)
+        response_text = None
+        if exc_response is not None:
+            try:
+                response_text = getattr(exc_response, "text", str(exc_response))
+            except Exception:
+                response_text = "<unreadable>"
+        logger.error(
+            "litellm dispatch failed",
+            extra={
+                "error": exc_message,
+                "error_type": type(exc).__name__,
+                "status_code": exc_status,
+                "llm_provider": getattr(exc, "llm_provider", None),
+                "body": getattr(exc, "body", None),
+                "response_text": response_text,
+                "model": litellm_model,
+                "api_base": base_url,
+            },
+        )
+        raise UpstreamError(
+            f"Upstream error via litellm: {exc_message}",
+            status_code=exc_status if isinstance(exc_status, int) else 502,
+        ) from exc
+
+    if not client_stream and hasattr(result, "__aiter__"):
+        # Client asked for a non-streaming response but we always stream
+        # from upstream — drain the events into a single Anthropic Message
+        # dict so the rest of the pipeline can treat it as if upstream had
+        # returned non-streaming. Some litellm adapters return a
+        # non-streaming dict even when ``stream=True``; in that case,
+        # leave the result as-is.
+        try:
+            aggregated: Any = await aggregate_anthropic_events_to_message(
+                cast(AsyncIterator[Any], result)
+            )
+        except Exception as exc:
+            logger.error(
+                "Failed to aggregate streamed events into message",
+                extra={
+                    "error": str(exc),
+                    "error_type": type(exc).__name__,
+                    "model": litellm_model,
+                },
+            )
+            raise UpstreamError(
+                f"Failed to aggregate upstream stream: {exc}",
+                status_code=502,
+            ) from exc
+        return client_stream, aggregated, requested_model
+
+    return client_stream, result, requested_model

routstr/wallet.py

@@ -341,7 +341,7 @@ async def credit_balance(
         except Exception:
             pass
 
-        logger.info(
+        logger.debug(
             "Cashu token successfully redeemed and stored",
             extra={"amount": amount, "unit": unit, "mint_url": mint_url},
         )
@@ -503,7 +503,7 @@ async def fetch_all_balances(
 
 async def periodic_payout() -> None:
     if not settings.receive_ln_address:
-        logger.error("RECEIVE_LN_ADDRESS is not set, skipping payout")
+        logger.warning("RECEIVE_LN_ADDRESS is not set, periodic payout disabled")
         return
     while True:
         await asyncio.sleep(60 * 15)

tests/unit/test_litellm_routing.py

@@ -0,0 +1,93 @@
+"""Tests for `routstr.upstream.litellm_routing.detect_litellm_prefix`."""
+
+from __future__ import annotations
+
+import pytest
+
+from routstr.upstream.litellm_routing import detect_litellm_prefix
+
+
+@pytest.mark.parametrize(
+    "base_url,expected",
+    [
+        # The bug case: custom row pointing at Fireworks must NOT route to openai/.
+        ("https://api.fireworks.ai/inference/v1", "fireworks_ai/"),
+        # Other OpenAI-compatible providers commonly plugged into the custom slot.
+        ("https://api.groq.com/openai/v1", "groq/"),
+        ("https://api.x.ai/v1", "xai/"),
+        ("https://api.deepseek.com/v1", "deepseek/"),
+        ("https://api.together.xyz/v1", "together_ai/"),
+        ("https://api.perplexity.ai", "perplexity/"),
+        ("https://openrouter.ai/api/v1", "openrouter/"),
+        ("https://api.mistral.ai/v1", "mistral/"),
+        ("https://codestral.mistral.ai/v1", "codestral/"),
+        ("https://api.cohere.com/v1", "cohere_chat/"),
+        ("https://api.cohere.ai/v1", "cohere_chat/"),
+        ("https://api.deepinfra.com/v1/openai", "deepinfra/"),
+        ("https://api.cerebras.ai/v1", "cerebras/"),
+        ("https://api.sambanova.ai/v1", "sambanova/"),
+        ("https://api.moonshot.cn/v1", "moonshot/"),
+        ("https://api.moonshot.ai/v1", "moonshot/"),
+        ("https://api.studio.nebius.com/v1", "nebius/"),
+        ("https://api.novita.ai/v3/openai", "novita/"),
+        ("https://api.lambda.ai/v1", "lambda_ai/"),
+        ("https://api.aimlapi.com/v1", "aiml/"),
+        ("https://api.featherless.ai/v1", "featherless_ai/"),
+        ("https://integrate.api.nvidia.com/v1", "nvidia_nim/"),
+        ("https://inference.baseten.co/v1", "baseten/"),
+        ("https://ai-gateway.vercel.sh/v1", "vercel_ai_gateway/"),
+        ("https://api.inference.wandb.ai/v1", "wandb/"),
+        ("https://api.poe.com/v1", "poe/"),
+        ("https://llm.chutes.ai/v1/", "chutes/"),
+        ("https://api.v0.dev/v1", "v0/"),
+        ("https://api.hyperbolic.xyz/v1", "hyperbolic/"),
+        ("https://api.synthetic.new/openai/v1", "synthetic/"),
+        ("https://api.stima.tech/v1", "apertis/"),
+        ("https://nano-gpt.com/api/v1", "nano-gpt/"),
+        ("https://api.friendli.ai/serverless/v1", "friendliai/"),
+        ("https://api.galadriel.com/v1", "galadriel/"),
+        ("https://api.llama.com/compat/v1", "meta_llama/"),
+        ("https://api.minimax.io/v1", "minimax/"),
+        ("https://api.minimaxi.com/v1", "minimax/"),
+        ("https://platform.publicai.co/v1", "publicai/"),
+        ("https://inference.api.nscale.com/v1", "nscale/"),
+        ("https://dashscope-intl.aliyuncs.com/compatible-mode/v1", "dashscope/"),
+        ("https://api.endpoints.anyscale.com/v1", "anyscale/"),
+        ("https://api.ai21.com/studio/v1", "ai21_chat/"),
+        ("https://ark.cn-beijing.volces.com/api/v3", "volcengine/"),
+        ("https://api.voyageai.com/v1", "voyage/"),
+        ("https://api.jina.ai/v1", "jina_ai/"),
+        ("https://api.snowflakecomputing.com", "snowflake/"),
+        ("https://my-workspace.databricks.com/serving-endpoints", "databricks/"),
+        ("https://huggingface.co/api/inference", "huggingface/"),
+        # First-class providers — URL detection still produces the right prefix
+        # so subclasses without an explicit override stay correct.
+        ("https://api.openai.com/v1", "openai/"),
+        ("https://api.anthropic.com/v1", "anthropic/"),
+        ("https://generativelanguage.googleapis.com/v1beta/openai", "gemini/"),
+        ("https://us-central1-aiplatform.googleapis.com/v1", "vertex_ai/"),
+        # Azure ordering: must beat api.openai.com.
+        ("https://my-resource.openai.azure.com/openai/deployments/foo", "azure/"),
+        # Ollama hints.
+        ("http://localhost:11434/v1", "ollama_chat/"),
+        ("http://127.0.0.1:11434/v1", "ollama_chat/"),
+        ("http://my-ollama-host:11434/v1", "ollama_chat/"),
+        # Casing and trailing slash normalisation.
+        ("HTTPS://API.FIREWORKS.AI/INFERENCE/V1/", "fireworks_ai/"),
+        # Unknown host falls back to openai/ (still OpenAI-compatible by convention).
+        ("https://example.com/v1", "openai/"),
+        ("", "openai/"),
+    ],
+)
+def test_detect_litellm_prefix(base_url: str, expected: str) -> None:
+    assert detect_litellm_prefix(base_url) == expected
+
+
+def test_detect_litellm_prefix_none_uses_default() -> None:
+    assert detect_litellm_prefix(None) == "openai/"
+
+
+def test_detect_litellm_prefix_custom_default() -> None:
+    assert detect_litellm_prefix("https://example.com", default="anthropic/") == (
+        "anthropic/"
+    )

tests/unit/test_stream_id_injection.py

@@ -51,7 +51,15 @@ async def test_stream_with_id_injection() -> None:
         base.adjust_payment_for_tokens = AsyncMock(
             return_value={"total_usd": 0.1, "total_msats": 100}
         )
-        base.create_session = MagicMock()
+        # create_session() is used as an async context manager whose entered
+        # value exposes an awaitable .get(). Build a mock that behaves that
+        # way so the post-stream cost-chunk emission can run.
+        mock_session = MagicMock()
+        mock_session.get = AsyncMock(return_value=key)
+        mock_ctx = MagicMock()
+        mock_ctx.__aenter__ = AsyncMock(return_value=mock_session)
+        mock_ctx.__aexit__ = AsyncMock(return_value=None)
+        base.create_session = MagicMock(return_value=mock_ctx)
 
         streaming_response = await provider.handle_streaming_chat_completion(
             response=mock_response,

tests/unit/test_upstream_gemini.py

@@ -0,0 +1,313 @@
+"""Unit tests for the Gemini /v1/messages dispatch path.
+
+The Gemini upstream needs special handling because its OpenAI-compat
+surface rejects inbound ``functionCall`` parts that lack a
+``thought_signature``. We bypass litellm + openai SDK at the wire layer
+(see ``routstr/upstream/gemini_messages.py``) so we can inject Google's
+documented dummy signature (``"skip_thought_signature_validator"``).
+
+These tests cover the two pure helpers that drive the dispatcher:
+
+* ``inject_thought_signatures`` — request-side injection
+* ``_openai_chunks_to_anthropic_events`` — response-side translator
+"""
+
+from __future__ import annotations
+
+import json
+from collections.abc import AsyncGenerator
+from typing import Any
+
+import pytest
+
+from routstr.upstream.gemini_messages import (
+    DUMMY_THOUGHT_SIGNATURE,
+    _openai_chunks_to_anthropic_events,
+    inject_thought_signatures,
+)
+
+# ---------------------------------------------------------------------------
+# inject_thought_signatures
+# ---------------------------------------------------------------------------
+
+
+def test_inject_thought_signatures_adds_dummy_to_each_tool_call() -> None:
+    messages: list[dict[str, Any]] = [
+        {"role": "user", "content": "do thing"},
+        {
+            "role": "assistant",
+            "tool_calls": [
+                {
+                    "id": "toolu_1",
+                    "type": "function",
+                    "function": {"name": "Bash", "arguments": "{}"},
+                },
+                {
+                    "id": "toolu_2",
+                    "type": "function",
+                    "function": {"name": "Read", "arguments": "{}"},
+                },
+            ],
+        },
+    ]
+
+    inject_thought_signatures(messages)
+
+    for tc in messages[1]["tool_calls"]:
+        assert (
+            tc["extra_content"]["google"]["thought_signature"]
+            == DUMMY_THOUGHT_SIGNATURE
+        )
+
+
+def test_inject_thought_signatures_preserves_existing_signature() -> None:
+    """Don't clobber a real signature that came back from a prior turn."""
+    messages: list[dict[str, Any]] = [
+        {
+            "role": "assistant",
+            "tool_calls": [
+                {
+                    "id": "tc1",
+                    "type": "function",
+                    "function": {"name": "fn", "arguments": "{}"},
+                    "extra_content": {
+                        "google": {"thought_signature": "real-signature"}
+                    },
+                }
+            ],
+        }
+    ]
+
+    inject_thought_signatures(messages)
+
+    assert (
+        messages[0]["tool_calls"][0]["extra_content"]["google"][
+            "thought_signature"
+        ]
+        == "real-signature"
+    )
+
+
+def test_inject_thought_signatures_skips_messages_without_tool_calls() -> None:
+    messages: list[dict[str, Any]] = [
+        {"role": "user", "content": "hi"},
+        {"role": "assistant", "content": "hello"},
+    ]
+
+    inject_thought_signatures(messages)
+
+    for m in messages:
+        assert "extra_content" not in m
+
+
+def test_inject_thought_signatures_handles_malformed_extra_content() -> None:
+    """If a caller already set ``extra_content`` to a non-dict (defensive),
+    we replace it instead of crashing."""
+    messages: list[dict[str, Any]] = [
+        {
+            "role": "assistant",
+            "tool_calls": [
+                {
+                    "id": "tc",
+                    "type": "function",
+                    "function": {"name": "fn", "arguments": "{}"},
+                    "extra_content": "garbage",
+                }
+            ],
+        }
+    ]
+
+    inject_thought_signatures(messages)
+
+    extra = messages[0]["tool_calls"][0]["extra_content"]
+    assert isinstance(extra, dict)
+    assert extra["google"]["thought_signature"] == DUMMY_THOUGHT_SIGNATURE
+
+
+# ---------------------------------------------------------------------------
+# _openai_chunks_to_anthropic_events
+# ---------------------------------------------------------------------------
+
+
+async def _lines(*chunks: dict | str) -> AsyncGenerator[str, None]:
+    """Helper to wrap chunk dicts as SSE-style ``data:`` lines."""
+    for c in chunks:
+        if isinstance(c, dict):
+            yield f"data: {json.dumps(c)}"
+        else:
+            yield c
+
+
+def _parse_anthropic_sse(blocks: list[bytes]) -> list[dict]:
+    """Flatten a list of Anthropic SSE byte chunks into event dicts."""
+    events: list[dict] = []
+    for blob in blocks:
+        text = blob.decode()
+        for entry in text.split("\n\n"):
+            for line in entry.splitlines():
+                if line.startswith("data:"):
+                    events.append(json.loads(line[5:].lstrip()))
+    return events
+
+
+@pytest.mark.asyncio
+async def test_translator_emits_text_only_response() -> None:
+    """Plain text response: message_start → content_block_* (text) →
+    message_delta(end_turn) → message_stop."""
+    chunks: list[dict] = [
+        {
+            "id": "chatcmpl-1",
+            "model": "gemini-2.5-flash",
+            "choices": [{"index": 0, "delta": {"role": "assistant"}}],
+        },
+        {"choices": [{"delta": {"content": "Hello"}}]},
+        {"choices": [{"delta": {"content": ", world"}}]},
+        {
+            "choices": [{"delta": {}, "finish_reason": "stop"}],
+            "usage": {"prompt_tokens": 5, "completion_tokens": 7},
+        },
+    ]
+
+    out = []
+    async for event_bytes in _openai_chunks_to_anthropic_events(
+        _lines(*chunks), requested_model="gemini-2.5-flash"
+    ):
+        out.append(event_bytes)
+
+    events = _parse_anthropic_sse(out)
+    types = [e["type"] for e in events]
+    assert types == [
+        "message_start",
+        "content_block_start",
+        "content_block_delta",
+        "content_block_delta",
+        "content_block_stop",
+        "message_delta",
+        "message_stop",
+    ]
+    # Text deltas concatenate to "Hello, world".
+    text_deltas = [
+        e["delta"]["text"]
+        for e in events
+        if e["type"] == "content_block_delta"
+    ]
+    assert "".join(text_deltas) == "Hello, world"
+    # Stop reason was mapped from openai's "stop".
+    msg_delta = next(e for e in events if e["type"] == "message_delta")
+    assert msg_delta["delta"]["stop_reason"] == "end_turn"
+    assert msg_delta["usage"]["input_tokens"] == 5
+    assert msg_delta["usage"]["output_tokens"] == 7
+
+
+@pytest.mark.asyncio
+async def test_translator_emits_tool_use_block() -> None:
+    """tool_calls split across deltas → tool_use content block with
+    accumulated input_json_delta and stop_reason='tool_use'."""
+    chunks: list[dict] = [
+        {
+            "id": "chatcmpl-2",
+            "model": "gemini-2.5-flash",
+            "choices": [{"delta": {"role": "assistant"}}],
+        },
+        {
+            "choices": [
+                {
+                    "delta": {
+                        "tool_calls": [
+                            {
+                                "index": 0,
+                                "id": "call-abc",
+                                "type": "function",
+                                "function": {
+                                    "name": "Bash",
+                                    "arguments": '{"cmd":',
+                                },
+                            }
+                        ]
+                    }
+                }
+            ]
+        },
+        {
+            "choices": [
+                {
+                    "delta": {
+                        "tool_calls": [
+                            {
+                                "index": 0,
+                                "function": {"arguments": ' "ls"}'},
+                            }
+                        ]
+                    }
+                }
+            ]
+        },
+        {"choices": [{"delta": {}, "finish_reason": "tool_calls"}]},
+    ]
+
+    out = []
+    async for event_bytes in _openai_chunks_to_anthropic_events(
+        _lines(*chunks), requested_model="gemini-2.5-flash"
+    ):
+        out.append(event_bytes)
+
+    events = _parse_anthropic_sse(out)
+    types = [e["type"] for e in events]
+    assert types == [
+        "message_start",
+        "content_block_start",
+        "content_block_delta",
+        "content_block_delta",
+        "content_block_stop",
+        "message_delta",
+        "message_stop",
+    ]
+    # Tool use block was opened with the right name.
+    cb_start = next(e for e in events if e["type"] == "content_block_start")
+    assert cb_start["content_block"]["type"] == "tool_use"
+    assert cb_start["content_block"]["name"] == "Bash"
+    assert cb_start["content_block"]["id"] == "call-abc"
+    # Argument deltas were forwarded as input_json_delta partials.
+    deltas = [e for e in events if e["type"] == "content_block_delta"]
+    assert all(d["delta"]["type"] == "input_json_delta" for d in deltas)
+    assert "".join(d["delta"]["partial_json"] for d in deltas) == (
+        '{"cmd": "ls"}'
+    )
+    # tool_calls finish_reason → tool_use stop_reason.
+    msg_delta = next(e for e in events if e["type"] == "message_delta")
+    assert msg_delta["delta"]["stop_reason"] == "tool_use"
+
+
+@pytest.mark.asyncio
+async def test_translator_handles_done_sentinel_and_blank_lines() -> None:
+    """Spec edge cases from openai SSE: ``data: [DONE]``, blank lines,
+    invalid JSON. Translator should skip them gracefully."""
+    chunks: list[dict | str] = [
+        {
+            "id": "x",
+            "model": "m",
+            "choices": [{"delta": {"role": "assistant"}}],
+        },
+        {"choices": [{"delta": {"content": "ok"}}]},
+        "",
+        ": comment",
+        "data: not-json",
+        "data: [DONE]",
+        {"choices": [{"delta": {}, "finish_reason": "stop"}]},
+    ]
+
+    out = []
+    async for event_bytes in _openai_chunks_to_anthropic_events(
+        _lines(*chunks), requested_model=None
+    ):
+        out.append(event_bytes)
+
+    events = _parse_anthropic_sse(out)
+    assert events[0]["type"] == "message_start"
+    assert events[-1]["type"] == "message_stop"
+    text = "".join(
+        e["delta"]["text"]
+        for e in events
+        if e["type"] == "content_block_delta"
+    )
+    assert text == "ok"