clean up

added opencode display usage correctly
Merge pull request #432 from Routstr/fix-usage-msat-pricing
2026-07-31 15:56:14 +00:00 · 2026-04-05 00:30:24 +02:00 · 2026-04-03 23:37:04 +02:00 · 2026-04-03 23:14:17 +02:00 · 2026-04-03 22:26:48 +02:00 · 2026-04-02 13:31:01 +02:00
6 changed files with 840 additions and 39 deletions
@@ -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
@@ -5,6 +5,7 @@ from time import monotonic
 from typing import Annotated, NoReturn

 from fastapi import APIRouter, Depends, Header, HTTPException
+from fastapi.responses import JSONResponse
 from pydantic import BaseModel
 from sqlmodel import select

@@ -204,19 +205,57 @@ async def _refund_cache_set(authorization: str, value: dict[str, str]) -> None:
        _refund_cache[key] = (expiry, value)


-@router.post("/refund")
+@router.post("/refund", response_model=None)
 async def refund_wallet_endpoint(
-    authorization: Annotated[str, Header(...)],
+    authorization: Annotated[str | None, Header()] = None,
+    x_cashu: Annotated[str | None, Header()] = None,
    session: AsyncSession = Depends(get_session),
-) -> dict[str, str]:
-    if not authorization.startswith("Bearer "):
+) -> JSONResponse | dict[str, str]:
+    if x_cashu:
+        # Find the "in" transaction by the original payment token
+        in_tx_result = await session.exec(
+            select(CashuTransaction).where(
+                CashuTransaction.token == x_cashu,
+                CashuTransaction.type == "in",
+            )
+        )
+        in_tx = in_tx_result.first()
+        if in_tx is None:
+            raise HTTPException(status_code=404, detail="Refund not found")
+
+        # Use the request_id to find the associated "out" (refund) transaction
+        if in_tx.request_id is None:
+            raise HTTPException(status_code=404, detail="Refund not found")
+
+        out_tx_result = await session.exec(
+            select(CashuTransaction).where(
+                CashuTransaction.request_id == in_tx.request_id,
+                CashuTransaction.type == "out",
+            )
+        )
+        out_tx = out_tx_result.first()
+        if out_tx is None:
+            raise HTTPException(status_code=404, detail="Refund not found")
+        if out_tx.swept:
+            raise HTTPException(status_code=410, detail="Refund has been swept")
+
+        out_tx.collected = True
+        session.add(out_tx)
+        await session.commit()
+        body: dict[str, str] = {"token": out_tx.token}
+        if out_tx.unit == "sat":
+            body["sats"] = str(out_tx.amount)
+        else:
+            body["msats"] = str(out_tx.amount)
+        return JSONResponse(content=body, headers={"X-Cashu": out_tx.token})
+
+    if authorization is None or not authorization.startswith("Bearer "):
        raise HTTPException(
            status_code=401,
            detail="Invalid authorization. Use 'Bearer <cashu-token>' or 'Bearer <api-key>'",
        )

    bearer_value: str = authorization[7:]
-
    key: ApiKey = await validate_bearer_key(bearer_value, session)

    if key.total_balance <= 0:
@@ -229,6 +268,12 @@ async def refund_wallet_endpoint(
            detail="Cannot refund child key. Please refund the parent key instead.",
        )

+    if key.reserved_balance > 0:
+        raise HTTPException(
+            status_code=400,
+            detail="Cannot refund key. There are ongoing requests for this api key.",
+        )
+
    remaining_balance_msats: int = key.total_balance

    if key.refund_currency == "sat":
@@ -283,11 +328,20 @@ async def refund_wallet_endpoint(

    await _refund_cache_set(bearer_value, result)

+    previous_reserved_balance = key.reserved_balance
    key.balance = 0
    key.reserved_balance = 0
    session.add(key)
    await session.commit()

+    logger.info(
+        "refund_wallet_endpoint: refund successful",
+        extra={
+            "refunded_msats": remaining_balance_msats,
+            "previous_reserved_balance": previous_reserved_balance,
+        },
+    )
+
    return result


@@ -109,12 +109,20 @@ async def calculate_cost(  # todo: can be sync
    )

    usd_cost = 0.0
+    input_usd = 0.0
+    output_usd = 0.0

-    # Prioritize cost_details.upstream_inference_cost
    if "cost_details" in usage_data:
        usd_cost = float(
            usage_data["cost_details"].get("upstream_inference_cost", 0) or 0
        )
+        input_usd = float(
+            usage_data["cost_details"].get("upstream_inference_prompt_cost", 0) or 0
+        )
+        output_usd = float(
+            usage_data["cost_details"].get("upstream_inference_completions_cost", 0)
+            or 0
+        )

    # Fallback to cost field if upstream_inference_cost is 0
    if usd_cost == 0 and "cost" in usage_data:
@@ -123,12 +131,34 @@ async def calculate_cost(  # todo: can be sync
        except Exception:
            pass

+    MSATS_PER_1K_INPUT_TOKENS: float = (
+        float(settings.fixed_per_1k_input_tokens) * 1000.0
+    )
+    MSATS_PER_1K_OUTPUT_TOKENS: float = (
+        float(settings.fixed_per_1k_output_tokens) * 1000.0
+    )
+
    if usd_cost > 0:
        try:
            sats_per_usd = 1.0 / sats_usd_price()
            cost_in_sats = usd_cost * sats_per_usd
            cost_in_msats = math.ceil(cost_in_sats * 1000)

+            input_msats = 0
+            output_msats = 0
+
+            if input_usd > 0 or output_usd > 0:
+                input_msats = int((input_usd * sats_per_usd) * 1000)
+                output_msats = int((output_usd * sats_per_usd) * 1000)
+            else:
+                total_tokens = input_tokens + output_tokens
+                if total_tokens > 0:
+                    input_ratio = input_tokens / total_tokens
+                    input_msats = int(cost_in_msats * input_ratio)
+                    output_msats = cost_in_msats - input_msats
+                else:
+                    output_msats = cost_in_msats
+
            logger.info(
                "Using cost from usage data/details",
                extra={
@@ -140,9 +170,9 @@ async def calculate_cost(  # todo: can be sync
            )

            return CostData(
-                base_msats=-1,
-                input_msats=-1,  # Cost field doesn't break down by token type
-                output_msats=-1,
+                base_msats=0,
+                input_msats=input_msats,
+                output_msats=output_msats,
                total_msats=cost_in_msats,
                total_usd=usd_cost,
                input_tokens=input_tokens,
@@ -159,13 +189,6 @@ async def calculate_cost(  # todo: can be sync
            )
            # Fall through to token-based calculation

-    MSATS_PER_1K_INPUT_TOKENS: float = (
-        float(settings.fixed_per_1k_input_tokens) * 1000.0
-    )
-    MSATS_PER_1K_OUTPUT_TOKENS: float = (
-        float(settings.fixed_per_1k_output_tokens) * 1000.0
-    )
-
    if not settings.fixed_pricing:
        response_model = response_data.get("model", "")
        logger.debug(
@@ -231,10 +254,10 @@ async def calculate_cost(  # todo: can be sync
            output_tokens=output_tokens,
        )

-    input_msats = round(input_tokens / 1000 * MSATS_PER_1K_INPUT_TOKENS, 3)
+    calc_input_msats = round(input_tokens / 1000 * MSATS_PER_1K_INPUT_TOKENS, 3)

-    output_msats = round(output_tokens / 1000 * MSATS_PER_1K_OUTPUT_TOKENS, 3)
-    token_based_cost = math.ceil(input_msats + output_msats)
+    calc_output_msats = round(output_tokens / 1000 * MSATS_PER_1K_OUTPUT_TOKENS, 3)
+    token_based_cost = math.ceil(calc_input_msats + calc_output_msats)
    total_usd = (token_based_cost / 1000.0) * sats_usd_price()

    logger.info(
@@ -242,8 +265,8 @@ async def calculate_cost(  # todo: can be sync
        extra={
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
-            "input_cost_msats": input_msats,
-            "output_cost_msats": output_msats,
+            "input_cost_msats": calc_input_msats,
+            "output_cost_msats": calc_output_msats,
            "total_cost_msats": token_based_cost,
            "total_usd": total_usd,
            "model": response_data.get("model", "unknown"),
@@ -252,8 +275,8 @@ async def calculate_cost(  # todo: can be sync

    return CostData(
        base_msats=0,
-        input_msats=int(input_msats),
-        output_msats=int(output_msats),
+        input_msats=int(calc_input_msats),
+        output_msats=int(calc_output_msats),
        total_msats=token_based_cost,
        total_usd=total_usd,
        input_tokens=input_tokens,
@@ -403,7 +403,9 @@ async def update_sats_pricing() -> None:


@models_router.get("/v1/models")
-@models_router.get("/models", include_in_schema=False)
+@models_router.get("/v1/models/", include_in_schema=False)
+@models_router.get("/models")
+@models_router.get("/models/", include_in_schema=False)
 async def models(session: AsyncSession = Depends(get_session)) -> dict:
    """Get all available models from all providers with database overrides applied."""
    from ..proxy import get_unique_models
@@ -452,6 +452,7 @@ class BaseUpstreamProvider:
        key: ApiKey,
        max_cost_for_model: int,
        background_tasks: BackgroundTasks,
+        requested_model: str | None = None,
    ) -> StreamingResponse:
        """Handle streaming chat completion responses with token usage tracking and cost adjustment.

@@ -520,11 +521,15 @@ class BaseUpstreamProvider:
                            if isinstance(obj, dict):
                                if obj.get("model"):
                                    last_model_seen = str(obj.get("model"))
+                                if requested_model:
+                                    obj["model"] = requested_model

                                if isinstance(obj.get("usage"), dict):
                                    # Hold this chunk back to merge cost later
                                    usage_chunk_data = obj
                                    continue
+                                yield b"data: " + json.dumps(obj).encode() + b"\n\n"
+                                continue
                        except json.JSONDecodeError:
                            pass

@@ -620,6 +625,7 @@ class BaseUpstreamProvider:
        key: ApiKey,
        session: AsyncSession,
        deducted_max_cost: int,
+        requested_model: str | None = None,
    ) -> Response:
        """Handle non-streaming chat completion responses with token usage tracking and cost adjustment.

@@ -668,9 +674,7 @@ class BaseUpstreamProvider:
                response_json["usage"]["cost_sats"] = (
                    cost_data.get("total_msats", 0) // 1000
                )
-                response_json["usage"]["remaining_balance_msats"] = (
-                    remaining_balance_msats
-                )
+                response_json["usage"]["remaining_balance_msats"] = remaining_balance_msats

            # Keep detailed cost
            response_json["metadata"] = response_json.get("metadata", {})
@@ -714,6 +718,8 @@ class BaseUpstreamProvider:
                if k.lower() in allowed_headers
            }

+            if requested_model:
+                response_json["model"] = requested_model
            return Response(
                content=json.dumps(response_json).encode(),
                status_code=response.status_code,
@@ -744,7 +750,7 @@ class BaseUpstreamProvider:
            raise

    async def handle_streaming_responses_completion(
-        self, response: httpx.Response, key: ApiKey, max_cost_for_model: int
+        self, response: httpx.Response, key: ApiKey, max_cost_for_model: int, requested_model: str | None = None
    ) -> StreamingResponse:
        """Handle streaming Responses API responses with token usage tracking and cost adjustment.

@@ -814,6 +820,8 @@ class BaseUpstreamProvider:
                            if isinstance(obj, dict):
                                if obj.get("model"):
                                    last_model_seen = str(obj.get("model"))
+                                if requested_model:
+                                    obj["model"] = requested_model

                                # Track reasoning tokens for Responses API
                                if usage := obj.get("usage", {}):
@@ -934,6 +942,7 @@ class BaseUpstreamProvider:
        key: ApiKey,
        session: AsyncSession,
        deducted_max_cost: int,
+        requested_model: str | None = None,
    ) -> Response:
        """Handle non-streaming Responses API responses with token usage tracking and cost adjustment.

@@ -985,9 +994,7 @@ class BaseUpstreamProvider:
                response_json["usage"]["cost_sats"] = (
                    cost_data.get("total_msats", 0) // 1000
                )
-                response_json["usage"]["remaining_balance_msats"] = (
-                    remaining_balance_msats
-                )
+                response_json["usage"]["remaining_balance_msats"] = remaining_balance_msats

            # Keep detailed cost
            response_json["metadata"] = response_json.get("metadata", {})
@@ -1031,6 +1038,8 @@ class BaseUpstreamProvider:
                if k.lower() in allowed_headers
            }

+            if requested_model:
+                response_json["model"] = requested_model
            return Response(
                content=json.dumps(response_json).encode(),
                status_code=response.status_code,
@@ -1098,6 +1107,174 @@ class BaseUpstreamProvider:
                    },
                )

+    async def handle_streaming_messages_completion(
+        self, response: httpx.Response, key: ApiKey, max_cost_for_model: int
+    ) -> StreamingResponse:
+        async def stream_with_cost(
+            max_cost_for_model: int,
+        ) -> AsyncGenerator[bytes, None]:
+            stored_chunks: list[bytes] = []
+            usage_finalized: bool = False
+            last_model_seen: str | None = None
+            input_tokens: int = 0
+            output_tokens: int = 0
+
+            async def finalize_without_usage() -> bytes | None:
+                nonlocal usage_finalized
+                if usage_finalized:
+                    return None
+                async with create_session() as new_session:
+                    fresh_key = await new_session.get(key.__class__, key.hashed_key)
+                    if not fresh_key:
+                        usage_finalized = True
+                        return None
+                    try:
+                        fallback: dict = {
+                            "model": last_model_seen or "unknown",
+                            "usage": None,
+                        }
+                        cost_data = await adjust_payment_for_tokens(
+                            fresh_key, fallback, new_session, max_cost_for_model
+                        )
+                        usage_finalized = True
+                        return f"event: cost\ndata: {json.dumps({'cost': cost_data})}\n\n".encode()
+                    except Exception:
+                        usage_finalized = True
+                        return None
+
+            try:
+                async for chunk in response.aiter_bytes():
+                    stored_chunks.append(chunk)
+                    try:
+                        decoded_chunk = chunk.decode("utf-8", errors="ignore")
+                        for line in decoded_chunk.split("\n"):
+                            if line.startswith("data: "):
+                                try:
+                                    data = json.loads(line[6:])
+                                    if isinstance(data, dict):
+                                        msg = data.get("message", {})
+                                        if msg and msg.get("model"):
+                                            last_model_seen = str(msg.get("model"))
+
+                                        if usage := msg.get("usage"):
+                                            input_tokens += usage.get("input_tokens", 0)
+                                            output_tokens += usage.get(
+                                                "output_tokens", 0
+                                            )
+
+                                        if usage := data.get("usage"):
+                                            input_tokens += usage.get("input_tokens", 0)
+                                            output_tokens += usage.get(
+                                                "output_tokens", 0
+                                            )
+                                except json.JSONDecodeError:
+                                    pass
+                    except Exception:
+                        pass
+
+                    yield chunk
+
+                usage_data = {
+                    "input_tokens": input_tokens,
+                    "output_tokens": output_tokens,
+                }
+
+                if input_tokens > 0 or output_tokens > 0:
+                    async with create_session() as new_session:
+                        fresh_key = await new_session.get(key.__class__, key.hashed_key)
+                        if fresh_key:
+                            try:
+                                combined_data = {
+                                    "model": last_model_seen or "unknown",
+                                    "usage": usage_data,
+                                }
+                                cost_data = await adjust_payment_for_tokens(
+                                    fresh_key,
+                                    combined_data,
+                                    new_session,
+                                    max_cost_for_model,
+                                )
+                                usage_finalized = True
+                                yield f"event: cost\ndata: {json.dumps({'cost': cost_data})}\n\n".encode()
+                            except Exception:
+                                pass
+
+                if not usage_finalized:
+                    maybe_cost_event = await finalize_without_usage()
+                    if maybe_cost_event is not None:
+                        yield maybe_cost_event
+
+            except httpx.ReadError:
+                if not usage_finalized:
+                    await finalize_without_usage()
+                # Upstream dropped the connection mid-stream; response already started, swallow silently
+            except Exception:
+                if not usage_finalized:
+                    await finalize_without_usage()
+                raise
+            finally:
+                if not usage_finalized:
+                    await finalize_without_usage()
+
+        response_headers = dict(response.headers)
+        response_headers.pop("content-encoding", None)
+        response_headers.pop("content-length", None)
+
+        return StreamingResponse(
+            stream_with_cost(max_cost_for_model),
+            status_code=response.status_code,
+            headers=response_headers,
+        )
+
+    async def handle_non_streaming_messages_completion(
+        self,
+        response: httpx.Response,
+        key: ApiKey,
+        session: AsyncSession,
+        deducted_max_cost: int,
+        path: str,
+    ) -> Response:
+        try:
+            content = await response.aread()
+            response_json = json.loads(content)
+
+            if path.endswith("count_tokens") and "usage" not in response_json:
+                input_tokens = response_json.get("input_tokens", 0)
+                response_json["usage"] = {"input_tokens": input_tokens}
+
+            cost_data = await adjust_payment_for_tokens(
+                key, response_json, session, deducted_max_cost
+            )
+            response_json["cost"] = cost_data
+
+            allowed_headers = {
+                "content-type",
+                "cache-control",
+                "date",
+                "vary",
+                "access-control-allow-origin",
+                "access-control-allow-methods",
+                "access-control-allow-headers",
+                "access-control-allow-credentials",
+                "access-control-expose-headers",
+                "access-control-max-age",
+            }
+
+            response_headers = {
+                k: v
+                for k, v in response.headers.items()
+                if k.lower() in allowed_headers
+            }
+
+            return Response(
+                content=json.dumps(response_json).encode(),
+                status_code=response.status_code,
+                headers=response_headers,
+                media_type="application/json",
+            )
+        except Exception:
+            raise
+
    async def forward_request(
        self,
        request: Request,
@@ -1126,6 +1303,8 @@ class BaseUpstreamProvider:
        path = self.normalize_request_path(path, model_obj)
        url = self.build_request_url(path, model_obj)

+        original_model_id = model_obj.id if model_obj else None
+
        transformed_body = self.prepare_request_body(request_body, model_obj)

        logger.info(
@@ -1197,7 +1376,54 @@ class BaseUpstreamProvider:
                    await client.aclose()
                return mapped_error

-            if path.endswith("chat/completions") or path.endswith("embeddings"):
+            if (
+                path.endswith("chat/completions")
+                or path.endswith("embeddings")
+                or path.endswith("messages")
+                or path.endswith("messages/count_tokens")
+            ):
+                if path.endswith("messages"):
+                    client_wants_streaming = False
+                    if request_body:
+                        try:
+                            request_data = json.loads(request_body)
+                            client_wants_streaming = request_data.get("stream", False)
+                        except json.JSONDecodeError:
+                            pass
+
+                    content_type = response.headers.get("content-type", "")
+                    upstream_is_streaming = "text/event-stream" in content_type
+                    is_streaming = client_wants_streaming and upstream_is_streaming
+
+                    if is_streaming and response.status_code == 200:
+                        result = await self.handle_streaming_messages_completion(
+                            response, key, max_cost_for_model
+                        )
+                        background_tasks = BackgroundTasks()
+                        background_tasks.add_task(response.aclose)
+                        background_tasks.add_task(client.aclose)
+                        result.background = background_tasks
+                        return result
+
+                    if response.status_code == 200:
+                        try:
+                            return await self.handle_non_streaming_messages_completion(
+                                response, key, session, max_cost_for_model, path
+                            )
+                        finally:
+                            await response.aclose()
+                            await client.aclose()
+
+                if path.endswith("messages/count_tokens"):
+                    if response.status_code == 200:
+                        try:
+                            return await self.handle_non_streaming_messages_completion(
+                                response, key, session, max_cost_for_model, path
+                            )
+                        finally:
+                            await response.aclose()
+                            await client.aclose()
+
                if path.endswith("chat/completions"):
                    client_wants_streaming = False
                    if request_body:
@@ -1237,7 +1463,8 @@ class BaseUpstreamProvider:
                        background_tasks.add_task(response.aclose)
                        background_tasks.add_task(client.aclose)
                        result = await self.handle_streaming_chat_completion(
-                            response, key, max_cost_for_model, background_tasks
+                            response, key, max_cost_for_model, background_tasks,
+                            requested_model=original_model_id,
                        )
                        result.background = background_tasks
                        return result
@@ -1246,7 +1473,8 @@ class BaseUpstreamProvider:
                if response.status_code == 200:
                    try:
                        return await self.handle_non_streaming_chat_completion(
-                            response, key, session, max_cost_for_model
+                            response, key, session, max_cost_for_model,
+                            requested_model=original_model_id,
                        )
                    finally:
                        await response.aclose()
@@ -1361,6 +1589,8 @@ class BaseUpstreamProvider:
        path = self.normalize_request_path(path, model_obj)
        url = self.build_request_url(path, model_obj)

+        original_model_id = model_obj.id if model_obj else None
+
        transformed_body = self.prepare_responses_request_body(request_body, model_obj)

        logger.info(
@@ -1447,7 +1677,8 @@ class BaseUpstreamProvider:

                if is_streaming and response.status_code == 200:
                    result = await self.handle_streaming_responses_completion(
-                        response, key, max_cost_for_model
+                        response, key, max_cost_for_model,
+                        requested_model=original_model_id,
                    )
                    background_tasks = BackgroundTasks()
                    background_tasks.add_task(response.aclose)
@@ -1458,7 +1689,8 @@ class BaseUpstreamProvider:
                if response.status_code == 200:
                    try:
                        return await self.handle_non_streaming_responses_completion(
-                            response, key, session, max_cost_for_model
+                            response, key, session, max_cost_for_model,
+                            requested_model=original_model_id,
                        )
                    finally:
                        await response.aclose()
@@ -1825,11 +2057,25 @@ class BaseUpstreamProvider:
            if line.startswith("data: "):
                try:
                    data_json = json.loads(line[6:])
+                    # OpenAI format: usage and model at top level
                    if "usage" in data_json:
                        usage_data = data_json["usage"]
-                        model = data_json.get("model")
+                        model = data_json.get("model") or model
                    elif "model" in data_json and not model:
                        model = data_json["model"]
+                    # Anthropic format: model and input usage inside "message" key
+                    if "message" in data_json:
+                        msg = data_json["message"]
+                        if not model and msg.get("model"):
+                            model = msg["model"]
+                        if msg.get("usage") and not usage_data:
+                            usage_data = msg["usage"]
+                        elif msg.get("usage") and usage_data:
+                            # Merge: message_start has input_tokens, message_delta has output_tokens
+                            merged = dict(usage_data)
+                            for k, v in msg["usage"].items():
+                                merged[k] = merged.get(k, 0) + v
+                            usage_data = merged
                except json.JSONDecodeError:
                    continue

@@ -2262,9 +2508,14 @@ class BaseUpstreamProvider:
                    error_response.headers["X-Cashu"] = refund_token
                    return error_response

-                if path.endswith("chat/completions") or path.endswith("embeddings"):
+                if (
+                    path.endswith("chat/completions")
+                    or path.endswith("embeddings")
+                    or path.endswith("messages")
+                    or path.endswith("messages/count_tokens")
+                ):
                    logger.debug(
-                        "Processing completion/embeddings response",
+                        "Processing completion/embeddings/messages response",
                        extra={"path": path, "amount": amount, "unit": unit},
                    )

@@ -0,0 +1,116 @@
+import json
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from fastapi.responses import JSONResponse
+
+from routstr.balance import refund_wallet_endpoint
+from routstr.core.db import CashuTransaction
+
+
+def _make_cashu_tx(
+    token: str,
+    amount: int,
+    unit: str,
+    type: str = "out",
+    request_id: str | None = "req-abc",
+    swept: bool = False,
+    collected: bool = False,
+) -> CashuTransaction:
+    tx = CashuTransaction(token=token, amount=amount, unit=unit, type=type, request_id=request_id)
+    tx.swept = swept
+    tx.collected = collected
+    return tx
+
+
+def _exec_result(tx: CashuTransaction | None) -> MagicMock:
+    result = MagicMock()
+    result.first.return_value = tx
+    return result
+
+
+@pytest.mark.asyncio
+async def test_refund_x_cashu_returns_token() -> None:
+    x_cashu_token = "cashuAtest_token_value"
+    in_tx = _make_cashu_tx(token=x_cashu_token, amount=0, unit="msat", type="in", request_id="req-abc")
+    out_tx = _make_cashu_tx(token="cashuArefund_token", amount=1000, unit="msat", type="out", request_id="req-abc")
+
+    session = MagicMock()
+    session.exec = AsyncMock(side_effect=[_exec_result(in_tx), _exec_result(out_tx)])
+    session.add = MagicMock()
+    session.commit = AsyncMock()
+
+    result = await refund_wallet_endpoint(
+        authorization="Bearer sk-somekey",
+        x_cashu=x_cashu_token,
+        session=session,
+    )
+
+    assert isinstance(result, JSONResponse)
+    body = json.loads(result.body)
+    assert body["token"] == "cashuArefund_token"
+    assert body["msats"] == "1000"
+    assert result.headers["X-Cashu"] == "cashuArefund_token"
+    assert out_tx.collected is True
+
+
+@pytest.mark.asyncio
+async def test_refund_x_cashu_sat_unit() -> None:
+    x_cashu_token = "cashuAsat_token"
+    in_tx = _make_cashu_tx(token=x_cashu_token, amount=0, unit="sat", type="in", request_id="req-sat")
+    out_tx = _make_cashu_tx(token="cashuArefund_sat", amount=500, unit="sat", type="out", request_id="req-sat")
+
+    session = MagicMock()
+    session.exec = AsyncMock(side_effect=[_exec_result(in_tx), _exec_result(out_tx)])
+    session.add = MagicMock()
+    session.commit = AsyncMock()
+
+    result = await refund_wallet_endpoint(
+        authorization="Bearer sk-somekey",
+        x_cashu=x_cashu_token,
+        session=session,
+    )
+
+    assert isinstance(result, JSONResponse)
+    body = json.loads(result.body)
+    assert body["token"] == "cashuArefund_sat"
+    assert body["sats"] == "500"
+    assert "msats" not in body
+    assert result.headers["X-Cashu"] == "cashuArefund_sat"
+
+
+@pytest.mark.asyncio
+async def test_refund_x_cashu_not_found_raises_404() -> None:
+    from fastapi import HTTPException
+
+    session = MagicMock()
+    session.exec = AsyncMock(return_value=_exec_result(None))
+
+    with pytest.raises(HTTPException) as exc_info:
+        await refund_wallet_endpoint(
+            authorization="Bearer sk-somekey",
+            x_cashu="cashuAmissing_token",
+            session=session,
+        )
+
+    assert exc_info.value.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_refund_x_cashu_swept_raises_410() -> None:
+    from fastapi import HTTPException
+
+    in_tx = _make_cashu_tx(token="cashuAswept_token", amount=0, unit="msat", type="in", request_id="req-swept")
+    out_tx = _make_cashu_tx(token="cashuAswept", amount=100, unit="msat", type="out", request_id="req-swept", swept=True)
+
+    session = MagicMock()
+    session.exec = AsyncMock(side_effect=[_exec_result(in_tx), _exec_result(out_tx)])
+
+    with pytest.raises(HTTPException) as exc_info:
+        await refund_wallet_endpoint(
+            authorization="Bearer sk-somekey",
+            x_cashu="cashuAswept_token",
+            session=session,
+        )
+
+    assert exc_info.value.status_code == 410
Author	SHA1	Message	Date
9qeklajc	31a9730cd8	clean up	2026-04-05 00:30:24 +02:00
9qeklajc	22ede7636a	added opencode display usage correctly	2026-04-03 23:37:04 +02:00
9qeklajcandGitHub	c9a5af0ffa	Merge pull request #432 from Routstr/fix-usage-msat-pricing set in/out msat prices correctly	2026-04-03 23:14:17 +02:00
9qeklajc	072079b33d	set in/out msat prices correctly	2026-04-03 22:26:48 +02:00
9qeklajcandGitHub	2abc28d295	Merge pull request #431 from Routstr/add-refund-success-log added refund success log	2026-04-02 13:31:01 +02:00
9qeklajc	7879b68cad	added refund success log	2026-04-02 13:18:55 +02:00
9qeklajcandGitHub	9eff340c57	Merge pull request #430 from Routstr/add-missing-messages-endpoint add missing messages endpoint	2026-04-01 23:29:35 +02:00
9qeklajc	933ba105d8	add missing messages endpoint	2026-04-01 23:27:15 +02:00
9qeklajcandGitHub	205dc6c3b3	Merge pull request #429 from Routstr/fix-reset-refund-while-reserved-balance-not-zero refund only when reserved balance is zero	2026-04-01 17:21:57 +02:00
9qeklajc	c1ada39278	refund only when reserved balance is zero	2026-04-01 17:17:36 +02:00
9qeklajcandGitHub	200b32f1ff	Merge pull request #428 from Routstr/simplify-x-cashu-refund Simplify x cashu refund	2026-04-01 15:23:59 +02:00
9qeklajcandGitHub	2b623a86c3	Merge pull request #423 from Routstr/fix-display-api-key-infos Fix display api key infos	2026-04-01 15:18:35 +02:00
9qeklajc	905ae67015	update tests	2026-04-01 15:11:43 +02:00
9qeklajc	6fde846be2	refund x cashu	2026-04-01 15:04:38 +02:00
9qeklajc	a833cf429e	add refund x-cashu to default refund endpoint	2026-03-29 21:22:09 +02:00
9qeklajcandGitHub	a7455a48d2	Merge pull request #421 from Routstr/v0.4.1 fix fee calculation	2026-03-25 10:20:14 +01:00
redshiftandGitHub	a090632804	Merge pull request #424 from Routstr/fix/models-trailing-slash fix: add trailing slash routes for /models endpoint	2026-03-24 22:31:14 +00:00
redshiftandGitHub	d6b755ff81	Merge pull request #425 from Routstr/main syncign before merge	2026-03-24 22:28:37 +00:00
redshift	dc8ebe66e0	docs: add other mints balance tracking plan	2026-03-23 21:28:45 +00:00
redshift	a6c6d3e034	fix: add trailing slash routes for /models endpoint The /models/ endpoint with trailing slash was being caught by the proxy catch-all route instead of the models router, causing 'Model unknown not found' errors. Add explicit routes for both /models and /v1/models with and without trailing slashes.	2026-03-23 13:15:59 +00:00