better fallback

fix: treat 'quote not found' as definitive unpaid for invoice expiry

docs/provider/configuration.md

@@ -48,6 +48,68 @@ Connect to your AI provider(s):
 | **Upstream URL** | API endpoint (e.g., `https://api.openai.com/v1`) |
 | **API Key**      | Your provider's API key                          |
 
+### PPQ Auto Top-up
+
+PPQ providers can automatically purchase more credits when their USD balance
+falls below a configured threshold. Configure this per provider in the Admin
+Dashboard by editing a **PPQ.AI** provider and opening **PPQ Auto Top-up**.
+There are no environment variables for this feature.
+
+#### Requirements
+
+Before enabling auto top-up, make sure that:
+
+- the PPQ provider has a valid API key;
+- at least one trusted Cashu mint is configured;
+- the node wallet has enough **node-owned** funds at one mint to pay the
+  Lightning invoice; client balances are never used; and
+- the node has a current BTC/USD price for validating the invoice amount.
+
+| Setting | Description |
+| ------- | ----------- |
+| **Enable Auto Top-up** | Enables automatic PPQ credit purchases for this provider. |
+| **When credits are below (USD)** | Starts a top-up when the reported PPQ balance is below this positive USD value. |
+| **Purchase this amount (USD)** | Amount of PPQ credit to buy per top-up. Must be a whole number from **1 to 500 USD**. |
+
+For example, a threshold of `5` and purchase amount of `20` buys 20 USD of
+credit when the PPQ balance drops below 5 USD.
+
+#### How it works
+
+The worker checks eligible providers approximately once per minute. When the
+balance is below the threshold, it:
+
+1. verifies the node has enough owner funds before creating an invoice;
+2. requests a USD-denominated Lightning top-up invoice from PPQ;
+3. rejects expired, mismatched, or unexpectedly expensive invoices (more than
+   10% above the local BTC/USD estimate);
+4. pays from the configured Cashu mint with sufficient owner funds; and
+5. waits for PPQ to confirm that the credit settled.
+
+Only one attempt can be active for a provider. An attempt that was active at
+the start of a cycle suppresses another top-up for that entire cycle, even if
+PPQ reports it settled immediately. This prevents a temporarily stale PPQ
+balance from causing a duplicate purchase.
+
+Completed PPQ payments appear in the dashboard transaction history with source
+`ppq_auto_topup`. The payment record is separate from the internal claim used
+to prevent concurrent attempts.
+
+#### Payment recovery
+
+If the Cashu mint paid the invoice but PPQ settlement cannot be confirmed, the
+provider card shows **Auto top-up needs review**. A payment still owned by a
+running worker is shown as **Paying invoice** and cannot be released.
+
+Before choosing **Release top-up**, manually verify both PPQ and the Cashu mint.
+Release the claim only when the previous Lightning payment is definitively
+unable to settle. Releasing an ambiguous payment allows the next cycle to try
+again and can therefore cause a duplicate top-up.
+
+Disabling auto top-up prevents new purchases, but the node continues to
+reconcile an already active payment until it reaches a safe terminal state or
+requires operator review.
+
 ### Node Identity
 
 How your node appears to clients:

routstr/core/admin.py

@@ -862,6 +862,33 @@ class UpstreamProviderUpdateBySlug(BaseModel):
     provider_settings: dict | None = None
 
 
+async def _active_ppq_claim_in_session(session: AsyncSession, provider_id: int) -> bool:
+    """Check for an active claim inside the caller's transaction.
+
+    Must share the transaction of whatever destructive write it is guarding —
+    a check in its own session leaves a window for a worker to create the
+    claim between the check and the commit.
+    """
+    from ..upstream.auto_topup import _ppq_state_id_for_provider
+
+    claim = await session.get(CashuTransaction, _ppq_state_id_for_provider(provider_id))
+    return claim is not None and not claim.collected and not claim.swept
+
+
+def _require_valid_ppq_auto_topup(
+    provider_type: str, settings: dict | None
+) -> None:
+    """Reject PPQ auto top-up settings the worker would later refuse."""
+    if provider_type != "ppqai":
+        return
+
+    from ..upstream.auto_topup import validate_ppq_auto_topup_settings
+
+    problem = validate_ppq_auto_topup_settings(settings)
+    if problem is not None:
+        raise HTTPException(status_code=400, detail=problem)
+
+
 async def _apply_provider_update(
     session: AsyncSession,
     provider: UpstreamProviderRow,
@@ -873,6 +900,29 @@ async def _apply_provider_update(
         await _ensure_unique_slug(session, validated, exclude_id=provider.id)
         provider.slug = validated
 
+    provider_type_changed = (
+        payload.provider_type is not None
+        and payload.provider_type != provider.provider_type
+    )
+    ppq_type_changed = provider_type_changed and (
+        provider.provider_type == "ppqai" or payload.provider_type == "ppqai"
+    )
+    if (
+        provider_type_changed
+        and provider.provider_type == "ppqai"
+        and provider.id is not None
+        and await _active_ppq_claim_in_session(session, provider.id)
+    ):
+        # Changing the type would orphan the claim: the PPQ endpoints refuse
+        # non-ppqai providers, so nobody could ever inspect or release it.
+        raise HTTPException(
+            status_code=409,
+            detail=(
+                "This provider has an active PPQ auto top-up claim. Release "
+                "it before changing the provider type"
+            ),
+        )
+
     if payload.provider_type is not None:
         provider.provider_type = payload.provider_type
     if payload.base_url is not None:
@@ -885,6 +935,41 @@ async def _apply_provider_update(
         provider.enabled = payload.enabled
     if payload.provider_fee is not None:
         provider.provider_fee = payload.provider_fee
+
+    # Auto-top-up fields have provider-specific units and meaning. Reusing
+    # enabled Routstr settings for PPQ (or vice versa) can silently reinterpret
+    # sats as USD, so a type change must provide settings for the new type.
+    if (
+        ppq_type_changed
+        and payload.provider_settings is None
+        and provider.provider_settings
+    ):
+        try:
+            stored_settings = json.loads(provider.provider_settings)
+        except (json.JSONDecodeError, TypeError):
+            stored_settings = None
+        if isinstance(stored_settings, dict) and stored_settings.get("auto_topup"):
+            raise HTTPException(
+                status_code=400,
+                detail=(
+                    "Changing provider type requires explicit auto-top-up "
+                    "settings because the units are provider-specific"
+                ),
+            )
+
+    # Validate against the effective type and effective settings.
+    effective_settings = payload.provider_settings
+    if effective_settings is None and payload.provider_type is not None:
+        try:
+            effective_settings = (
+                json.loads(provider.provider_settings)
+                if provider.provider_settings
+                else None
+            )
+        except (json.JSONDecodeError, TypeError):
+            effective_settings = None
+    if effective_settings is not None:
+        _require_valid_ppq_auto_topup(provider.provider_type, effective_settings)
     if payload.provider_settings is not None:
         provider.provider_settings = json.dumps(payload.provider_settings)
 
@@ -924,6 +1009,10 @@ async def create_upstream_provider(
         else:
             slug = await allocate_unique_provider_slug(session, payload.provider_type)
 
+        _require_valid_ppq_auto_topup(
+            payload.provider_type, payload.provider_settings
+        )
+
         provider = UpstreamProviderRow(
             slug=slug,
             provider_type=payload.provider_type,
@@ -1015,6 +1104,25 @@ async def delete_upstream_provider(provider_id: str) -> dict[str, object]:
     async with create_session() as session:
         provider = await _get_upstream_provider_by_ref(session, provider_id)
         deleted_id = _provider_pk(provider)
+
+        # Checked inside the delete transaction: the worker's claim creation
+        # re-reads the provider inside its own transaction, so these two
+        # writes serialise — either the claim lands first and this 409s, or
+        # the delete lands first and the worker refuses to claim.
+        if provider.provider_type == "ppqai" and await _active_ppq_claim_in_session(
+            session, deleted_id
+        ):
+            # Deleting now would orphan the claim and any funds it tracks:
+            # the PPQ endpoints 404 without the provider row, so the claim
+            # could never again be inspected or released.
+            raise HTTPException(
+                status_code=409,
+                detail=(
+                    "This provider has an active PPQ auto top-up claim. "
+                    "Resolve and release it before deleting the provider"
+                ),
+            )
+
         await session.delete(provider)
         await session.commit()
     await reinitialize_upstreams()
@@ -1623,6 +1731,78 @@ async def get_log_dates_api(request: Request) -> dict[str, object]:
     return {"dates": dates}
 
 
+_PPQ_RELEASE_ERRORS = {
+    "no_active_claim": "No active PPQ claim to release",
+    "stale_state": ("The claim changed since it was reviewed; reload and check again"),
+    "payment_in_flight": (
+        "A Lightning payment is still in flight for this claim. Wait for it to "
+        "finish or expire before releasing"
+    ),
+    "claim_changed": (
+        "The claim changed while the release was being applied; reload and check again"
+    ),
+}
+
+
+class ReleasePPQAutoTopupRequest(BaseModel):
+    confirmed_safe_to_retry: bool
+    # Echoes the state_token the admin reviewed — the claim's full versioned
+    # state, not just its operation id. Any change since the review (a new
+    # attempt, a phase change, a renewed lease) fails the match, so the
+    # release cannot land on a state the admin never saw.
+    state_token: str | None = None
+
+
+async def _require_ppq_provider(provider_id: int) -> UpstreamProviderRow:
+    async with create_session() as session:
+        provider = await session.get(UpstreamProviderRow, provider_id)
+    if provider is None:
+        raise HTTPException(status_code=404, detail="Provider not found")
+    if provider.provider_type != "ppqai":
+        raise HTTPException(status_code=400, detail="Provider is not PPQ")
+    return provider
+
+
+@admin_router.get(
+    "/api/upstream-providers/{provider_id}/ppq-auto-topup",
+    dependencies=[Depends(require_admin_api)],
+)
+async def get_ppq_auto_topup_api(provider_id: int) -> dict[str, object]:
+    await _require_ppq_provider(provider_id)
+    from ..upstream.auto_topup import get_ppq_auto_topup_state
+
+    return {"ok": True, **await get_ppq_auto_topup_state(provider_id)}
+
+
+@admin_router.post(
+    "/api/upstream-providers/{provider_id}/ppq-auto-topup/release",
+    dependencies=[Depends(require_admin_api)],
+)
+async def release_ppq_auto_topup_api(
+    provider_id: int, payload: ReleasePPQAutoTopupRequest
+) -> dict[str, object]:
+    await _require_ppq_provider(provider_id)
+    if not payload.confirmed_safe_to_retry:
+        raise HTTPException(
+            status_code=400,
+            detail="Confirm the Lightning payment outcome is safe before releasing",
+        )
+
+    from ..upstream.auto_topup import release_ppq_auto_topup_state
+
+    outcome = await release_ppq_auto_topup_state(
+        provider_id, state_token=payload.state_token
+    )
+    if not outcome.released:
+        raise HTTPException(status_code=409, detail=_PPQ_RELEASE_ERRORS[outcome.reason])
+
+    logger.warning(
+        "Admin released PPQ auto top-up claim after manual reconciliation",
+        extra={"provider_id": provider_id, "state_token": payload.state_token},
+    )
+    return {"ok": True, "released": True}
+
+
 @admin_router.get("/api/transactions", dependencies=[Depends(require_admin_api)])
 async def get_transactions_api(
     type: str | None = None,
@@ -1635,7 +1815,11 @@ async def get_transactions_api(
     async with create_session() as session:
         from sqlmodel import col, func
 
-        base = select(CashuTransaction)
+        # Hide only the deterministic PPQ claim-lock rows. Append-only PPQ
+        # payment rows remain visible as the audit trail for irreversible melts.
+        base = select(CashuTransaction).where(
+            ~col(CashuTransaction.id).like("ppq-auto-topup-%")
+        )
         if type:
             base = base.where(CashuTransaction.type == type)
         if source:

routstr/lightning.py

@@ -456,11 +456,20 @@ async def check_invoice_payment(
 
             mint_url = settlement.mint_url or settings.primary_mint
             wallet = await get_wallet(mint_url, "sat")
-            mint_status = await run_mint_operation(
-                lambda: wallet.get_mint_quote(settlement.payment_hash),
-                op_name="get_mint_quote",
-                mint_url=mint_url,
-            )
+            try:
+                mint_status = await run_mint_operation(
+                    lambda: wallet.get_mint_quote(settlement.payment_hash),
+                    op_name="get_mint_quote",
+                    mint_url=mint_url,
+                )
+            except Exception as error:
+                if not _is_quote_not_found(error):
+                    raise
+                logger.info(
+                    "Invoice quote no longer exists at mint, marking expired",
+                    extra={"invoice_id": invoice.id, "error": str(error)},
+                )
+                return True
             if not mint_status.paid:
                 return getattr(mint_status, "state", None) == MintQuoteState.unpaid
             payment_confirmed = True
@@ -575,6 +584,15 @@ async def check_invoice_payment(
             return False
 
 
+def _is_quote_not_found(error: BaseException) -> bool:
+    """Check if the error indicates the mint no longer has this quote."""
+    message = str(error)
+    return bool(
+        re.search(r"\bquote\s+not\s+found\b", message, re.IGNORECASE)
+        and re.search(r"\bcode\s*:?\s*0\b", message, re.IGNORECASE)
+    )
+
+
 def _is_outputs_already_signed(error: BaseException) -> bool:
     message = str(error)
     return bool(

routstr/payment/cost_calculation.py

@@ -183,6 +183,28 @@ async def calculate_cost(
                 cost_details.get("output_cost")
                 or cost_details.get("upstream_inference_completions_cost")
             )
+            cache_pricing_rates: tuple[float, float, float, float] | None = None
+            if cache_read_tokens > 0 or cache_creation_tokens > 0:
+                try:
+                    cache_pricing_rates = _get_pricing_rates(
+                        response_data, model_obj, provider_fee
+                    )
+                except ValueError:
+                    logger.warning(
+                        "Cache pricing unavailable for USD cost breakdown; "
+                        "leaving cache cost components unknown",
+                        extra={"model": response_data.get("model", "unknown")},
+                    )
+                if cache_pricing_rates is None and settings.fixed_pricing:
+                    fixed_input_rate = (
+                        float(settings.fixed_per_1k_input_tokens) * 1000.0
+                    )
+                    cache_pricing_rates = (
+                        fixed_input_rate,
+                        float(settings.fixed_per_1k_output_tokens) * 1000.0,
+                        fixed_input_rate,
+                        fixed_input_rate,
+                    )
             return _calculate_from_usd_cost(
                 usd_cost,
                 input_usd,
@@ -193,6 +215,7 @@ async def calculate_cost(
                 output_tokens,
                 response_data,
                 provider_fee,
+                cache_pricing_rates,
             )
         except Exception as e:
             logger.warning(
@@ -451,6 +474,7 @@ def _calculate_from_usd_cost(
     output_tokens: int,
     response_data: dict,
     provider_fee: float | None,
+    pricing_rates: tuple[float, float, float, float] | None = None,
 ) -> CostData:
     """Calculate cost from USD figures, deriving input/output split from tokens."""
     if provider_fee is None:
@@ -460,15 +484,20 @@ def _calculate_from_usd_cost(
     output_usd = output_usd * provider_fee
     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)
+    raw_cost_msats = cost_in_sats * 1000
+    cost_in_msats = math.ceil(raw_cost_msats)
+    raw_input_msats = 0.0
 
     if input_usd > 0 or output_usd > 0:
         # The total is the authoritative billed amount. Allocating that integer
         # total proportionally avoids losing sub-millisatoshi remainders when
         # input and output components are each truncated independently.
         component_usd = input_usd + output_usd
-        input_msats = math.floor(cost_in_msats * input_usd / component_usd)
-        output_msats = cost_in_msats - input_msats
+        # Match the token-priced path: truncate the visible output component
+        # and assign the authoritative total's rounding remainder to input.
+        output_msats = math.floor(cost_in_msats * output_usd / component_usd)
+        input_msats = cost_in_msats - output_msats
+        raw_input_msats = raw_cost_msats * input_usd / component_usd
     else:
         effective_input_tokens = (
             input_tokens + cache_read_tokens + cache_creation_tokens
@@ -480,6 +509,38 @@ def _calculate_from_usd_cost(
             else 0
         )
         output_msats = cost_in_msats - input_msats
+        raw_input_msats = (
+            raw_cost_msats * effective_input_tokens / total_tokens
+            if total_tokens > 0
+            else 0.0
+        )
+
+    # Preserve the same cache-rate ratios as the token-priced path while the
+    # upstream USD total remains authoritative. Cache values are informational
+    # subcomponents of the inclusive input cost.
+    cache_read_msats = 0
+    cache_creation_msats = 0
+    if pricing_rates is not None:
+        input_rate, _, cache_read_rate, cache_creation_rate = pricing_rates
+        regular_weight = input_tokens * input_rate
+        cache_read_weight = cache_read_tokens * cache_read_rate
+        cache_creation_weight = cache_creation_tokens * cache_creation_rate
+        total_input_weight = (
+            regular_weight + cache_read_weight + cache_creation_weight
+        )
+        if total_input_weight > 0:
+            cache_read_msats = int(
+                round(
+                    raw_input_msats * cache_read_weight / total_input_weight,
+                    3,
+                )
+            )
+            cache_creation_msats = int(
+                round(
+                    raw_input_msats * cache_creation_weight / total_input_weight,
+                    3,
+                )
+            )
 
     logger.info(
         "Using cost from usage data/details",
@@ -487,6 +548,8 @@ def _calculate_from_usd_cost(
             "usd_cost": usd_cost,
             "cost_in_sats": cost_in_sats,
             "cost_in_msats": cost_in_msats,
+            "cache_read_msats": cache_read_msats,
+            "cache_creation_msats": cache_creation_msats,
             "model": response_data.get("model", "unknown"),
         },
     )
@@ -501,8 +564,8 @@ def _calculate_from_usd_cost(
         output_tokens=output_tokens,
         cache_read_input_tokens=cache_read_tokens,
         cache_creation_input_tokens=cache_creation_tokens,
-        cache_read_msats=0,
-        cache_creation_msats=0,
+        cache_read_msats=cache_read_msats,
+        cache_creation_msats=cache_creation_msats,
     )
 
 

routstr/upstream/base.py

@@ -70,6 +70,81 @@ if typing.TYPE_CHECKING:
 logger = get_logger(__name__)
 
 
+CostMetadata = CostData | MaxCostData | dict[str, Any]
+
+
+def _cost_field(
+    cost_data: CostMetadata, field: str, default: int | float = 0
+) -> int | float:
+    if isinstance(cost_data, dict):
+        value = cost_data.get(field, default)
+    else:
+        value = getattr(cost_data, field, default)
+    return value if isinstance(value, (int, float)) else default
+
+
+def _inject_cost_response_headers(
+    headers: dict[str, str], cost_data: CostMetadata
+) -> None:
+    """Inject per-request cost breakdown into response headers.
+
+    The SDK's ``extractUsageFromResponseHeaders`` reads these to populate
+    ``inputMsats``, ``outputMsats``, ``totalMsats`` and ``satsCost`` in the
+    usage tracking entry — without them, x-cashu requests show 0.0 for all
+    sat cost fields.
+    """
+    headers["X-Routstr-Cost-Msats"] = str(
+        int(_cost_field(cost_data, "total_msats"))
+    )
+    headers["X-Routstr-Input-Cost-Msats"] = str(
+        int(_cost_field(cost_data, "input_msats"))
+    )
+    headers["X-Routstr-Output-Cost-Msats"] = str(
+        int(_cost_field(cost_data, "output_msats"))
+    )
+    total_usd = float(_cost_field(cost_data, "total_usd", 0.0))
+    if total_usd:
+        headers["X-Routstr-Cost-Usd"] = str(total_usd)
+
+
+def _inject_cost_into_usage(response_json: dict, cost_data: CostMetadata) -> None:
+    """Inject cost breakdown into the response body's ``usage.cost`` object.
+
+    The SDK's ``extractUsageFromResponseBody`` expects ``usage.cost`` to be
+    an object with ``total_msats``/``input_msats``/``output_msats`` (not a
+    plain USD number). When the upstream returns ``cost`` as a number, the
+    SDK cannot extract the msats breakdown from the body alone.
+    """
+    usage = response_json.get("usage")
+    if not isinstance(usage, dict):
+        return
+    # Direct assignment (not setdefault) so routstr's authoritative cost
+    # data always overwrites any upstream-provided cost values. Using
+    # setdefault would silently keep stale upstream values and drop our
+    # calculated msats breakdown.
+    cost_obj: dict[str, int | float] = {
+        "base_msats": int(_cost_field(cost_data, "base_msats")),
+        "input_msats": int(_cost_field(cost_data, "input_msats")),
+        "output_msats": int(_cost_field(cost_data, "output_msats")),
+        "total_msats": int(_cost_field(cost_data, "total_msats")),
+        "cache_read_input_tokens": int(
+            _cost_field(cost_data, "cache_read_input_tokens")
+        ),
+        "cache_creation_input_tokens": int(
+            _cost_field(cost_data, "cache_creation_input_tokens")
+        ),
+        "cache_read_msats": int(_cost_field(cost_data, "cache_read_msats")),
+        "cache_creation_msats": int(
+            _cost_field(cost_data, "cache_creation_msats")
+        ),
+    }
+    total_usd = float(_cost_field(cost_data, "total_usd", 0.0))
+    if total_usd:
+        cost_obj["total_usd"] = total_usd
+    usage["cost"] = cost_obj
+    usage["cost_sats"] = int(_cost_field(cost_data, "total_msats")) // 1000
+
+
 def _is_json_content_type(content_type: str | None) -> bool:
     """Return True when the upstream response should be parsed as JSON."""
     if not content_type:
@@ -276,30 +351,24 @@ class BaseUpstreamProvider:
         self._apply_provider_field(response_json)
         if isinstance(cost_data, dict):
             total_msats = cost_data.get("total_msats", 0)
-            total_usd = cost_data.get("total_usd", 0.0)
             cost_dict = cost_data
         else:
             total_msats = cost_data.total_msats
-            total_usd = cost_data.total_usd
             cost_dict = cost_data.dict()
 
         sats_cost = total_msats // 1000
 
-        # Inject into top-level usage block (OpenAI/Anthropic style)
-        if "usage" in response_json:
-            response_json["usage"]["cost"] = total_usd
-            response_json["usage"]["cost_sats"] = sats_cost
+        # Inject the shared SDK cost contract into every usage shape.
+        if isinstance(response_json.get("usage"), dict):
+            _inject_cost_into_usage(response_json, cost_data)
             response_json["usage"]["remaining_balance_msats"] = key.balance
             self._fold_cache_into_input_tokens(response_json["usage"])
 
-        # Inject into Anthropic nested usage block if present
-        if (
-            "message" in response_json
-            and isinstance(response_json["message"], dict)
-            and "usage" in response_json["message"]
-        ):
-            response_json["message"]["usage"]["sats_cost"] = sats_cost
-            self._fold_cache_into_input_tokens(response_json["message"]["usage"])
+        message = response_json.get("message")
+        if isinstance(message, dict) and isinstance(message.get("usage"), dict):
+            _inject_cost_into_usage(message, cost_data)
+            message["usage"]["remaining_balance_msats"] = key.balance
+            self._fold_cache_into_input_tokens(message["usage"])
 
         # Unified Routstr metadata
         response_json["metadata"] = response_json.get("metadata", {})
@@ -1230,12 +1299,9 @@ class BaseUpstreamProvider:
             await session.refresh(key)
             remaining_balance_msats = key.balance
 
-            # Merge cost into usage for OpenCode
+            # Merge the shared cost contract into usage for SDKs and OpenCode.
             if "usage" in response_json:
-                response_json["usage"]["cost"] = cost_data.get("total_usd", 0.0)
-                response_json["usage"]["cost_sats"] = (
-                    cost_data.get("total_msats", 0) // 1000
-                )
+                _inject_cost_into_usage(response_json, cost_data)
                 response_json["usage"]["remaining_balance_msats"] = (
                     remaining_balance_msats
                 )
@@ -1282,6 +1348,7 @@ class BaseUpstreamProvider:
                 for k, v in response.headers.items()
                 if k.lower() in allowed_headers
             }
+            _inject_cost_response_headers(response_headers, cost_data)
 
             if requested_model:
                 response_json["model"] = requested_model
@@ -1667,12 +1734,9 @@ class BaseUpstreamProvider:
             await session.refresh(key)
             remaining_balance_msats = key.balance
 
-            # Merge cost into usage for OpenCode
+            # Merge the shared cost contract into usage for SDKs and OpenCode.
             if "usage" in response_json:
-                response_json["usage"]["cost"] = cost_data.get("total_usd", 0.0)
-                response_json["usage"]["cost_sats"] = (
-                    cost_data.get("total_msats", 0) // 1000
-                )
+                _inject_cost_into_usage(response_json, cost_data)
                 response_json["usage"]["remaining_balance_msats"] = (
                     remaining_balance_msats
                 )
@@ -1719,6 +1783,7 @@ class BaseUpstreamProvider:
                 for k, v in response.headers.items()
                 if k.lower() in allowed_headers
             }
+            _inject_cost_response_headers(response_headers, cost_data)
 
             if requested_model:
                 response_json["model"] = requested_model
@@ -2154,6 +2219,9 @@ class BaseUpstreamProvider:
                 if k.lower() in allowed_headers
             }
 
+            # Inject the same cost headers used by every paid response path.
+            _inject_cost_response_headers(response_headers, cost_data)
+
             return Response(
                 content=json.dumps(response_json).encode(),
                 status_code=response.status_code,
@@ -2243,9 +2311,14 @@ class BaseUpstreamProvider:
         )
         self.inject_cost_metadata(response_json, cost_data, key)
 
+        # Inject the same cost headers used by every paid response path.
+        response_headers: dict[str, str] = {}
+        _inject_cost_response_headers(response_headers, cost_data)
+
         return Response(
             content=json.dumps(response_json).encode(),
             status_code=200,
+            headers=response_headers,
             media_type="application/json",
         )
 
@@ -2296,11 +2369,12 @@ class BaseUpstreamProvider:
             and "usage" in response_json
             and isinstance(response_json["usage"], dict)
         ):
-            response_json["usage"]["cost_sats"] = cost_data.total_msats // 1000
+            _inject_cost_into_usage(response_json, cost_data)
             self._fold_cache_into_input_tokens(response_json["usage"])
 
         response_headers: dict[str, str] = {}
         if cost_data:
+            _inject_cost_response_headers(response_headers, cost_data)
             refund_amount = messages_dispatch.compute_refund(
                 amount, unit, cost_data.total_msats
             )
@@ -2546,7 +2620,7 @@ class BaseUpstreamProvider:
         the cost of a wire-format change for clients that read ``X-Cashu``
         from headers today.
         """
-        buffered: list[bytes] = []
+        buffered: list[messages_dispatch.AnnotatedEvent] = []
         last_model_seen: str | None = None
         input_tokens = 0
         output_tokens = 0
@@ -2574,7 +2648,7 @@ class BaseUpstreamProvider:
             total_cost = max(total_cost, annotated.total_cost)
             input_cost = max(input_cost, annotated.input_cost)
             output_cost = max(output_cost, annotated.output_cost)
-            buffered.append(annotated.sse_bytes)
+            buffered.append(annotated)
 
         response_headers: dict[str, str] = {
             "Cache-Control": "no-cache",
@@ -2601,6 +2675,7 @@ class BaseUpstreamProvider:
                 },
             )
 
+        cost_data: CostData | MaxCostData | None = None
         if (
             input_tokens > 0
             or output_tokens > 0
@@ -2656,9 +2731,30 @@ class BaseUpstreamProvider:
                     },
                 )
 
+        if cost_data:
+            _inject_cost_response_headers(response_headers, cost_data)
+            for index, annotated in enumerate(buffered):
+                event = annotated.event
+                changed = False
+                message = event.get("message")
+                if isinstance(message, dict) and isinstance(message.get("usage"), dict):
+                    _inject_cost_into_usage(message, cost_data)
+                    changed = True
+                if isinstance(event.get("usage"), dict):
+                    _inject_cost_into_usage(event, cost_data)
+                    changed = True
+                if changed:
+                    event_type = str(event.get("type") or "")
+                    prefix = f"event: {event_type}\n" if event_type else ""
+                    buffered[index] = annotated._replace(
+                        sse_bytes=(
+                            f"{prefix}data: {json.dumps(event)}\n\n".encode()
+                        )
+                    )
+
         async def replay() -> AsyncGenerator[bytes, None]:
-            for chunk in buffered:
-                yield chunk
+            for annotated in buffered:
+                yield annotated.sse_bytes
 
         return StreamingResponse(
             replay(),
@@ -3701,6 +3797,11 @@ class BaseUpstreamProvider:
                                 "model": model,
                             },
                         )
+
+                    # Inject cost breakdown headers so the SDK's
+                    # extractUsageFromResponseHeaders can populate
+                    # inputMsats/outputMsats/totalMsats for x-cashu requests.
+                    _inject_cost_response_headers(response_headers, cost_data)
             except Exception as e:
                 logger.error(
                     "Error calculating cost for streaming response",
@@ -3723,8 +3824,12 @@ class BaseUpstreamProvider:
                     if "provider" not in data_json:
                         self._apply_provider_field(data_json)
                         changed = True
-                    if cost_data and "usage" in data_json and data_json["usage"]:
-                        data_json["usage"]["cost_sats"] = cost_data.total_msats // 1000
+                    if (
+                        cost_data
+                        and "usage" in data_json
+                        and data_json["usage"]
+                    ):
+                        _inject_cost_into_usage(data_json, cost_data)
                         changed = True
                     if changed:
                         lines[i] = "data: " + json.dumps(data_json)
@@ -3778,7 +3883,10 @@ class BaseUpstreamProvider:
             )
 
             if cost_data and "usage" in response_json:
-                response_json["usage"]["cost_sats"] = cost_data.total_msats // 1000
+                # Inject cost breakdown into both the response body (so the
+                # SDK's body extractor picks up the msats breakdown) and the
+                # response headers (so the SDK's header extractor works too).
+                _inject_cost_into_usage(response_json, cost_data)
 
             if not cost_data:
                 logger.error(
@@ -3809,6 +3917,8 @@ class BaseUpstreamProvider:
             if "content-encoding" in response_headers:
                 del response_headers["content-encoding"]
 
+            _inject_cost_response_headers(response_headers, cost_data)
+
             if unit == "msat":
                 refund_amount = amount - cost_data.total_msats
             elif unit == "sat":
@@ -4682,6 +4792,11 @@ class BaseUpstreamProvider:
                                 "model": model,
                             },
                         )
+
+                    # Inject cost breakdown headers so the SDK's
+                    # extractUsageFromResponseHeaders can populate
+                    # inputMsats/outputMsats/totalMsats for x-cashu requests.
+                    _inject_cost_response_headers(response_headers, cost_data)
             except Exception as e:
                 logger.error(
                     "Error calculating cost for streaming Responses API response",
@@ -4704,8 +4819,12 @@ class BaseUpstreamProvider:
                     if "provider" not in data_json:
                         self._apply_provider_field(data_json)
                         changed = True
-                    if cost_data and "usage" in data_json and data_json["usage"]:
-                        data_json["usage"]["cost_sats"] = cost_data.total_msats // 1000
+                    if (
+                        cost_data
+                        and "usage" in data_json
+                        and data_json["usage"]
+                    ):
+                        _inject_cost_into_usage(data_json, cost_data)
                         changed = True
                     if changed:
                         lines[i] = "data: " + json.dumps(data_json)
@@ -4748,7 +4867,7 @@ class BaseUpstreamProvider:
             )
 
             if cost_data and "usage" in response_json:
-                response_json["usage"]["cost_sats"] = cost_data.total_msats // 1000
+                _inject_cost_into_usage(response_json, cost_data)
 
             if not cost_data:
                 logger.error(
@@ -4779,6 +4898,8 @@ class BaseUpstreamProvider:
             if "content-encoding" in response_headers:
                 del response_headers["content-encoding"]
 
+            _inject_cost_response_headers(response_headers, cost_data)
+
             if unit == "msat":
                 refund_amount = amount - cost_data.total_msats
             elif unit == "sat":

routstr/upstream/ppqai.py

@@ -441,7 +441,7 @@ class PPQAIUpstreamProvider(BaseUpstreamProvider):
         """
         data = await self.check_balance()
         balance = data.get("balance")
-        if isinstance(balance, (int, float)):
+        if isinstance(balance, (int, float)) and not isinstance(balance, bool):
             return float(balance)
         return None
 

routstr/wallet.py

@@ -6,11 +6,12 @@ import time
 import typing
 from contextlib import asynccontextmanager
 from contextvars import ContextVar
+from dataclasses import dataclass
 from pathlib import Path
 from typing import AsyncGenerator, TypedDict
 
 import httpx
-from cashu.core.base import MeltQuoteState, MintQuote, Proof, Token
+from cashu.core.base import MeltQuote, MeltQuoteState, MintQuote, Proof, Token
 from cashu.core.mint_info import MintInfo as _CashuMintInfo
 from cashu.wallet.helpers import deserialize_token_from_string
 from cashu.wallet.wallet import Wallet as _CashuWallet
@@ -105,6 +106,11 @@ def _msats_to_sats(amount: int) -> int:
     return amount // 1000
 
 
+def _msats_to_sats_ceil(amount: int) -> int:
+    """Round liabilities up so fractional sats are never treated as owner funds."""
+    return (amount + 999) // 1000
+
+
 def _mints_to_inspect() -> list[str]:
     """Return configured mints plus the primary mint, without duplicates."""
     mint_urls = list(settings.cashu_mints)
@@ -390,9 +396,7 @@ async def _recieve_token_locked(
         else list(dict.fromkeys([settings.primary_mint, *settings.cashu_mints]))
     )
     output_unit = (
-        token_obj.unit
-        if token_obj.mint in destinations
-        else settings.primary_mint_unit
+        token_obj.unit if token_obj.mint in destinations else settings.primary_mint_unit
     )
     if destination_unit is not None and output_unit != destination_unit:
         raise ValueError(
@@ -493,6 +497,275 @@ async def send_token(amount: int, unit: str, mint_url: str | None = None) -> str
     return token
 
 
+class Bolt11PaymentNotAttempted(Exception):
+    """The invoice was definitively not paid, so the attempt can be retried.
+
+    Raised only where the mint's own answer rules out a settlement: coin
+    selection never reached ``melt``, or ``melt`` returned an explicit unpaid
+    state. Any proofs reserved along the way are released before this is
+    raised.
+    """
+
+
+class Bolt11PaymentAmbiguous(Exception):
+    """The payment may or may not have settled, so it must not be retried.
+
+    Raised when ``melt`` errored, timed out, or came back pending. The selected
+    proofs stay reserved: the mint may still complete the payment with them,
+    and spending them elsewhere would be a double spend.
+    """
+
+
+@dataclass
+class Bolt11PaymentPlan:
+    invoice: str
+    wallet: Wallet
+    proofs: list[Proof]
+    quote: MeltQuote
+    mint_url: str
+    unit: str
+
+    @property
+    def invoice_amount_sats(self) -> int:
+        amount = int(self.quote.amount)
+        return amount if self.unit == "sat" else (amount + 999) // 1000
+
+    @property
+    def maximum_spend_sats(self) -> int:
+        maximum = (
+            int(self.quote.amount)
+            + int(self.quote.fee_reserve)
+            + int(self.wallet.get_fees_for_proofs(self.proofs))
+        )
+        return maximum if self.unit == "sat" else (maximum + 999) // 1000
+
+
+async def _owner_balance_for_mint_and_unit(
+    mint_url: str, unit: str, proofs_balance: int
+) -> int:
+    """Return spendable node-owned funds without crossing user liabilities."""
+    async with db.create_session() as session:
+        # Refund mint is a preference, not funding provenance. Mirror payout's
+        # conservative rule and protect the full liability at every mint.
+        user_liability = await db.total_user_liability(session)
+    # API-key balances are stored in msats. Cashu ``sat`` proofs are not.
+    if unit == "sat":
+        user_liability = _msats_to_sats_ceil(user_liability)
+    return max(0, proofs_balance - user_liability)
+
+
+async def maximum_owner_cashu_balance_sats() -> int:
+    """Return the largest conservatively owner-funded mint/unit balance."""
+    details, _, _, _ = await fetch_all_balances()
+    async with db.create_session() as session:
+        liability_sats = _msats_to_sats_ceil(await db.total_user_liability(session))
+    balances = [
+        (
+            detail["wallet_balance"]
+            if detail["unit"] == "sat"
+            else _msats_to_sats(detail["wallet_balance"])
+        )
+        - liability_sats
+        for detail in details
+        if not detail.get("error")
+    ]
+    return max([0, *balances])
+
+
+async def prepare_bolt11_payment(invoice: str) -> Bolt11PaymentPlan:
+    """Choose the sufficiently funded configured mint with most owner funds.
+
+    Candidate discovery reads balances, user liabilities, and melt quotes. Coin
+    selection, which may split proofs, is deferred until the winner is known.
+
+    Runs under ``wallet_operation_guard``: the plan snapshots live proof state,
+    which another worker process could otherwise mutate mid-read. Callers that
+    go on to execute the plan should hold the guard across both calls so the
+    snapshot stays valid.
+    """
+    async with wallet_operation_guard():
+        return await _prepare_bolt11_payment(invoice)
+
+
+async def _prepare_bolt11_payment(invoice: str) -> Bolt11PaymentPlan:
+    mint_urls = list(dict.fromkeys([*settings.cashu_mints, settings.primary_mint]))
+    candidates: list[tuple[int, Wallet, list[Proof], MeltQuote, str, str]] = []
+    failures: list[dict[str, str]] = []
+    evaluated = 0
+
+    for mint_url in mint_urls:
+        if not mint_url:
+            continue
+        for unit in ("sat", "msat"):
+            try:
+                # force_reload: the guard's flock only serializes access — a
+                # cached wallet can still hold proof state from before another
+                # process's reservation landed on disk.
+                wallet = await get_wallet(mint_url, unit, force_reload=True)
+                proofs = get_proofs_per_mint_and_unit(
+                    wallet, mint_url, unit, not_reserved=True
+                )
+                proofs = await slow_filter_spend_proofs(proofs, wallet)
+                proofs_balance = sum(proof.amount for proof in proofs)
+                if proofs_balance <= 0:
+                    evaluated += 1
+                    continue
+
+                quote = await wallet.melt_quote(invoice=invoice)
+                evaluated += 1
+                # select_to_send runs with include_fees=True, so the input fee
+                # has to be part of sufficiency too. Without it a mint passes
+                # this filter and then fails coin selection.
+                required = (
+                    quote.amount
+                    + quote.fee_reserve
+                    + wallet.get_fees_for_proofs(proofs)
+                )
+                owner_balance = await _owner_balance_for_mint_and_unit(
+                    mint_url, unit, proofs_balance
+                )
+                if owner_balance < required:
+                    continue
+                owner_balance_msats = (
+                    owner_balance * 1000 if unit == "sat" else owner_balance
+                )
+                candidates.append(
+                    (owner_balance_msats, wallet, proofs, quote, mint_url, unit)
+                )
+            except Exception as e:
+                failures.append({"mint_url": mint_url, "unit": unit, "error": str(e)})
+                logger.debug(
+                    "Cashu mint cannot fund BOLT11 invoice",
+                    extra={"mint_url": mint_url, "unit": unit, "error": str(e)},
+                )
+
+    if not candidates:
+        if failures:
+            logger.warning(
+                "No Cashu mint could fund the BOLT11 invoice",
+                extra={"evaluated": evaluated, "failures": failures},
+            )
+        if evaluated == 0 and failures:
+            raise RuntimeError("Every configured Cashu mint refused the payment")
+        raise ValueError(
+            "No configured Cashu mint has enough balance after user liabilities to pay invoice"
+        )
+
+    _, wallet, proofs, quote, mint_url, unit = max(candidates, key=lambda item: item[0])
+    return Bolt11PaymentPlan(invoice, wallet, proofs, quote, mint_url, unit)
+
+
+async def execute_bolt11_payment(plan: Bolt11PaymentPlan) -> tuple[int, str, str]:
+    """Execute a prepared payment, separating retryable from ambiguous failure.
+
+    Raises ``Bolt11PaymentNotAttempted`` when the invoice provably did not
+    settle, and ``Bolt11PaymentAmbiguous`` when the outcome is unknown. Callers
+    may safely retry the first and must never retry the second.
+
+    Runs under ``wallet_operation_guard``: coin selection and reservation must
+    not race another worker process spending the same proofs.
+    """
+    async with wallet_operation_guard():
+        return await _execute_bolt11_payment(plan)
+
+
+async def _execute_bolt11_payment(plan: Bolt11PaymentPlan) -> tuple[int, str, str]:
+    # Select unreserved, mirroring send_token: a selection failure must not
+    # strand proofs that were never handed to the mint.
+    try:
+        selected, _ = await plan.wallet.select_to_send(
+            plan.proofs,
+            plan.quote.amount + plan.quote.fee_reserve,
+            set_reserved=False,
+            include_fees=True,
+        )
+    except Exception as e:
+        raise Bolt11PaymentNotAttempted(f"Coin selection failed: {e}") from e
+
+    await plan.wallet.set_reserved_for_send(selected, reserved=True)
+
+    try:
+        result = await asyncio.wait_for(
+            plan.wallet.melt(
+                proofs=selected,
+                invoice=plan.invoice,
+                fee_reserve_sat=plan.quote.fee_reserve,
+                quote_id=plan.quote.quote,
+            ),
+            timeout=60,
+        )
+    except BaseException as e:
+        # The mint may still be settling with these proofs, so they must stay
+        # reserved — but cashu's melt() un-reserves them itself on a mint
+        # transport error, the exact ambiguous case. Re-reserve with the melt
+        # quote id, not as a send: get_melt_quote() finds the proofs to settle
+        # by melt_id, so a send-style reservation would strand them — paid
+        # proofs never invalidated, unpaid ones never released. BaseException
+        # includes task cancellation after the melt was submitted.
+        try:
+            await asyncio.shield(
+                plan.wallet.set_reserved_for_melt(
+                    selected, reserved=True, quote_id=plan.quote.quote
+                )
+            )
+        except BaseException:
+            logger.critical(
+                "Could not re-reserve proofs after an ambiguous melt",
+                extra={"mint_url": plan.mint_url, "quote_id": plan.quote.quote},
+            )
+        if isinstance(e, asyncio.CancelledError):
+            raise
+        raise Bolt11PaymentAmbiguous(f"Cashu melt did not return: {e}") from e
+
+    raw_state = getattr(result, "state", None)
+    state = str(raw_state).lower().rsplit(".", 1)[-1] if raw_state is not None else ""
+    if state == "paid" or getattr(result, "paid", None) is True:
+        change = getattr(result, "change", None) or []
+        paid = sum(proof.amount for proof in selected) - sum(
+            int(item.amount) for item in change
+        )
+        return paid, plan.mint_url, plan.unit
+
+    if state == "unpaid":
+        # The mint is telling us it did not pay, so the proofs are ours again.
+        await plan.wallet.set_reserved_for_send(selected, reserved=False)
+        raise Bolt11PaymentNotAttempted("Cashu mint reported the melt as unpaid")
+
+    raise Bolt11PaymentAmbiguous(
+        f"Cashu melt did not reach a final state: {state or 'unknown'}"
+    )
+
+
+async def check_bolt11_payment_status(mint_url: str, unit: str, quote_id: str) -> str:
+    """Ask the mint what became of an earlier melt attempt.
+
+    Returns ``"paid"``, ``"unpaid"``, ``"pending"``, or ``"unknown"``. This is
+    the durable reconciliation path for an ambiguous payment: cashu's
+    ``get_melt_quote`` also settles the wallet database — invalidating the
+    proofs on ``paid`` and releasing their reservation on ``unpaid`` — so a
+    caller that sees ``"unpaid"`` may safely retry with the same funds.
+
+    Runs under ``wallet_operation_guard`` because of that side effect: it
+    mutates proof state and must not race other processes' wallet operations.
+    """
+    try:
+        async with wallet_operation_guard():
+            wallet = await get_wallet(mint_url, unit, force_reload=True)
+            quote = await wallet.get_melt_quote(quote_id)
+    except Exception as e:
+        logger.warning(
+            "Could not query the mint for a melt quote's status",
+            extra={"mint_url": mint_url, "quote_id": quote_id, "error": str(e)},
+        )
+        return "unknown"
+    if quote is None:
+        return "unknown"
+    state = str(getattr(quote, "state", "")).lower().rsplit(".", 1)[-1]
+    if state in ("paid", "unpaid", "pending"):
+        return state
+    return "unknown"
+
+
 async def release_token_reservation(token: str) -> None:
     """Release a token that was created locally but never handed off."""
     async with wallet_operation_guard():
@@ -723,6 +996,7 @@ async def _request_mint_with_fallback(
                 lambda: wallet.request_mint(amount),
                 op_name=op_name,
                 mint_url=mint_url,
+                retry_timeouts=False,
                 retry_on_rate_limit=False,
             )
             logger.info(
@@ -1724,9 +1998,7 @@ async def fetch_all_balances(
 
             try:
                 async with mint_check_limit:
-                    wallet = await get_wallet(
-                        mint_url, unit, retry_on_rate_limit=False
-                    )
+                    wallet = await get_wallet(mint_url, unit, retry_on_rate_limit=False)
                     proofs = get_proofs_per_mint_and_unit(
                         wallet, mint_url, unit, not_reserved=True
                     )
@@ -1840,9 +2112,7 @@ async def _payout_mint_and_unit(mint_url: str, unit: str) -> None:
         # snapshot up to 30s stale from another process's reservation, so the
         # cross-process lock is only safe with a fresh reload.
         wallet = await get_wallet(mint_url, unit, force_reload=True)
-        proofs = get_proofs_per_mint_and_unit(
-            wallet, mint_url, unit, not_reserved=True
-        )
+        proofs = get_proofs_per_mint_and_unit(wallet, mint_url, unit, not_reserved=True)
         proofs = await slow_filter_spend_proofs(proofs, wallet)
         await asyncio.sleep(5)
     except Exception as e:
@@ -1948,6 +2218,12 @@ async def _refund_sweep_once(cutoff: int) -> None:
             db.CashuTransaction.type == "out",
             db.CashuTransaction.collected == False,  # noqa: E712
             db.CashuTransaction.swept == False,  # noqa: E712
+            # PPQ rows describe a Lightning spend or claim lock, not a
+            # refundable Cashu token. Preserve legacy rows without a source.
+            col(db.CashuTransaction.source).is_(None)
+            | col(db.CashuTransaction.source).notin_(
+                ["ppq_auto_topup", "ppq_auto_topup_claim"]
+            ),
             db.CashuTransaction.created_at < cutoff,
             claim_available,
         )
@@ -2180,16 +2456,10 @@ async def periodic_routstr_fee_payout() -> None:
 
 async def send_to_lnurl(amount: int, unit: str, mint: str, address: str) -> int:
     async with wallet_operation_guard():
-        mint = await find_trusted_mint_with_funds(
-            amount, unit, mint, force_reload=True
-        )
+        mint = await find_trusted_mint_with_funds(amount, unit, mint, force_reload=True)
         wallet = await get_wallet(mint, unit)
-        available = get_proofs_per_mint_and_unit(
-            wallet, mint, unit, not_reserved=True
-        )
-        proofs, _ = await wallet.select_to_send(
-            available, amount, set_reserved=True
-        )
+        available = get_proofs_per_mint_and_unit(wallet, mint, unit, not_reserved=True)
+        proofs, _ = await wallet.select_to_send(available, amount, set_reserved=True)
         return await raw_send_to_lnurl(wallet, proofs, address, unit)
 
 

tests/integration/test_ppq_auto_topup_claim.py

@@ -0,0 +1,602 @@
+"""Real-database tests for the PPQ auto top-up claim lifecycle.
+
+These exercise the claim against actual SQL rather than mocked sessions,
+because the guarantees under test are all about what the database will and
+will not let two concurrent writers do.
+"""
+
+import time
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from sqlmodel import select
+
+from routstr.core.db import CashuTransaction, create_session
+from routstr.upstream.auto_topup import (
+    PPQ_PHASE_CLAIMED,
+    PPQ_PHASE_IN_FLIGHT,
+    PPQ_PHASE_RECONCILE,
+    _claim_ppq_topup,
+    _ppq_payment_id,
+    _ppq_payment_usd,
+    _ppq_request_id,
+    _ppq_spent_last_24h_usd,
+    _ppq_state_id_for_provider,
+    _record_ppq_invoice,
+    _set_ppq_state_terminal,
+    get_ppq_auto_topup_state,
+    release_ppq_auto_topup_state,
+)
+
+pytestmark = pytest.mark.asyncio
+
+
+def _row(provider_id: int = 1) -> MagicMock:
+    row = MagicMock()
+    row.id = provider_id
+    return row
+
+
+async def _seed_provider(provider_id: int = 1, slug: str = "ppq") -> None:
+    """Claim creation is fenced on the provider row existing; seed it."""
+    from routstr.core.db import UpstreamProviderRow
+
+    async with create_session() as session:
+        session.add(
+            UpstreamProviderRow(
+                id=provider_id,
+                slug=slug,
+                provider_type="ppqai",
+                base_url="https://api.ppq.ai",
+                api_key="secret",
+                enabled=True,
+            )
+        )
+        await session.commit()
+
+
+async def _state_row(provider_id: int = 1) -> CashuTransaction | None:
+    async with create_session() as session:
+        return await session.get(
+            CashuTransaction, _ppq_state_id_for_provider(provider_id)
+        )
+
+
+async def _seed_claim(
+    provider_id: int,
+    phase: str,
+    invoice_id: str,
+    lease_expires_at: int,
+    quote_id: str = "quote-1",
+) -> str:
+    """Seed a claim row and return its state token (the full request_id)."""
+    token = _ppq_request_id(
+        "operation-1", lease_expires_at, phase, invoice_id, quote_id
+    )
+    async with create_session() as session:
+        session.add(
+            CashuTransaction(
+                id=_ppq_state_id_for_provider(provider_id),
+                token="lnbc-invoice",
+                amount=102,
+                unit="sat",
+                type="out",
+                request_id=token,
+                mint_url="https://mint.test",
+                collected=False,
+                source="ppq_auto_topup",
+            )
+        )
+        await session.commit()
+    return token
+
+
+async def test_second_claim_is_refused_while_the_first_is_active(
+    patched_db_engine: Any,
+) -> None:
+    await _seed_provider()
+    assert await _claim_ppq_topup(_row()) is not None
+    # The whole point of the claim: a concurrent cycle must not get one.
+    assert await _claim_ppq_topup(_row()) is None
+
+    async with create_session() as session:
+        rows = (await session.exec(select(CashuTransaction))).all()
+    assert len(rows) == 1
+
+
+async def test_claim_is_reusable_once_the_previous_attempt_finished(
+    patched_db_engine: Any,
+) -> None:
+    await _seed_provider()
+    first = await _claim_ppq_topup(_row())
+    assert first is not None
+    assert await _set_ppq_state_terminal(_row(), first, collected=True, swept=False)
+
+    second = await _claim_ppq_topup(_row())
+    assert second is not None and second != first
+
+
+async def test_recording_the_invoice_moves_the_claim_in_flight(
+    patched_db_engine: Any,
+) -> None:
+    await _seed_provider()
+    operation_id = await _claim_ppq_topup(_row())
+    assert operation_id is not None
+
+    state = await get_ppq_auto_topup_state(1)
+    assert state["phase"] == PPQ_PHASE_CLAIMED
+    assert state["releasable"] is True
+    assert state["invoice_id"] is None
+
+    lease = await _record_ppq_invoice(
+        _row(),
+        operation_id,
+        invoice="lnbc-invoice",
+        invoice_id="invoice-1",
+        quote_id="quote-1",
+        amount=102,
+        amount_usd=10,
+        unit="sat",
+        mint_url="https://mint.test",
+    )
+    assert lease > int(time.time())
+
+    state = await get_ppq_auto_topup_state(1)
+    assert state["phase"] == PPQ_PHASE_IN_FLIGHT
+    assert state["invoice_id"] == "invoice-1"
+    # A payment is committed to a mint, so an admin must not sweep it.
+    assert state["releasable"] is False
+    # The raw BOLT11 invoice must never reach the admin API.
+    assert "token" not in state
+
+
+async def test_release_refuses_an_in_flight_claim(patched_db_engine: Any) -> None:
+    token = await _seed_claim(
+        1, PPQ_PHASE_IN_FLIGHT, "invoice-1", int(time.time()) + 900
+    )
+
+    outcome = await release_ppq_auto_topup_state(1, state_token=token)
+
+    assert outcome.released is False
+    assert outcome.reason == "payment_in_flight"
+    row = await _state_row()
+    assert row is not None and row.swept is False
+
+
+async def test_release_refuses_a_stale_state_token(patched_db_engine: Any) -> None:
+    await _seed_claim(1, PPQ_PHASE_RECONCILE, "invoice-1", int(time.time()) + 900)
+
+    outcome = await release_ppq_auto_topup_state(1, state_token="ppq:stale:token")
+
+    assert outcome.released is False
+    assert outcome.reason == "stale_state"
+    row = await _state_row()
+    assert row is not None and row.swept is False
+
+
+async def test_release_accepts_a_reconcile_claim(patched_db_engine: Any) -> None:
+    token = await _seed_claim(
+        1, PPQ_PHASE_RECONCILE, "invoice-1", int(time.time()) + 900
+    )
+
+    outcome = await release_ppq_auto_topup_state(1, state_token=token)
+
+    assert outcome.released is True
+    row = await _state_row()
+    assert row is not None and row.swept is True
+
+
+async def test_expired_in_flight_claim_becomes_releasable(
+    patched_db_engine: Any,
+) -> None:
+    # A worker that died mid-payment must not lock the provider forever.
+    token = await _seed_claim(1, PPQ_PHASE_IN_FLIGHT, "invoice-1", int(time.time()) - 1)
+
+    assert (await get_ppq_auto_topup_state(1))["releasable"] is True
+    outcome = await release_ppq_auto_topup_state(1, state_token=token)
+    assert outcome.released is True
+
+
+async def test_release_reports_no_active_claim_once_swept(
+    patched_db_engine: Any,
+) -> None:
+    token = await _seed_claim(
+        1, PPQ_PHASE_RECONCILE, "invoice-1", int(time.time()) + 900
+    )
+    assert (await release_ppq_auto_topup_state(1, state_token=token)).released
+
+    outcome = await release_ppq_auto_topup_state(1, state_token=token)
+    assert outcome.released is False
+    assert outcome.reason == "no_active_claim"
+
+
+async def test_terminal_write_fails_after_the_claim_was_released(
+    patched_db_engine: Any,
+) -> None:
+    """The symptom an admin release leaves behind for the owning worker."""
+    token = await _seed_claim(
+        1, PPQ_PHASE_RECONCILE, "invoice-1", int(time.time()) + 900
+    )
+    assert (await release_ppq_auto_topup_state(1, state_token=token)).released
+
+    assert (
+        await _set_ppq_state_terminal(
+            _row(), "operation-1", collected=True, swept=False
+        )
+        is False
+    )
+
+
+async def test_ppq_claim_rows_are_excluded_from_the_admin_transaction_list(
+    patched_db_engine: Any,
+) -> None:
+    from routstr.core.admin import get_transactions_api
+
+    await _seed_provider()
+    await _claim_ppq_topup(_row())
+    async with create_session() as session:
+        session.add(
+            CashuTransaction(
+                id="real-transaction",
+                token="cashuAreal",
+                amount=50,
+                unit="sat",
+                type="out",
+                source="x-cashu",
+            )
+        )
+        await session.commit()
+
+    result = await get_transactions_api()
+
+    ids = {t["id"] for t in result["transactions"]}  # type: ignore[index,union-attr]
+    assert "real-transaction" in ids
+    assert _ppq_state_id_for_provider(1) not in ids
+
+
+async def test_ppq_payment_audit_row_is_visible_and_survives_next_claim(
+    patched_db_engine: Any,
+) -> None:
+    from routstr.core.admin import get_transactions_api
+
+    await _seed_provider()
+    operation_id = await _claim_ppq_topup(_row())
+    assert operation_id is not None
+    await _record_ppq_invoice(
+        _row(),
+        operation_id,
+        invoice="lnbc-secret-invoice",
+        invoice_id="invoice-1",
+        quote_id="quote-1",
+        amount=102,
+        amount_usd=10,
+        unit="sat",
+        mint_url="https://mint.test",
+    )
+    assert await _set_ppq_state_terminal(
+        _row(), operation_id, collected=True, swept=False
+    )
+
+    result = await get_transactions_api(source="ppq_auto_topup")
+    transactions = result["transactions"]
+    assert len(transactions) == 1
+    audit = transactions[0]
+    assert audit["id"] == _ppq_payment_id(operation_id)
+    assert audit["token"] == "ppq-invoice:invoice-1:usd:10"
+    assert audit["collected"] is True
+    assert "lnbc-secret-invoice" not in audit["token"]
+
+    # Reusing the deterministic claim lock must not overwrite history.
+    assert await _claim_ppq_topup(_row()) is not None
+    async with create_session() as session:
+        assert await session.get(CashuTransaction, audit["id"]) is not None
+
+
+async def test_reconcile_settles_a_recorded_invoice(patched_db_engine: Any) -> None:
+    from routstr.upstream.auto_topup import _reconcile_ppq_state
+
+    await _seed_claim(1, PPQ_PHASE_IN_FLIGHT, "invoice-1", int(time.time()) + 900)
+    provider = MagicMock()
+    provider.check_topup_status = AsyncMock(return_value=True)
+
+    # Still suppresses this cycle, but the claim is now finished.
+    assert await _reconcile_ppq_state(_row(), provider) is True
+
+    row = await _state_row()
+    assert row is not None and row.collected is True
+
+
+async def test_stale_token_from_before_a_phase_change_cannot_release(
+    patched_db_engine: Any,
+) -> None:
+    """The blocker scenario: admin reviews `claimed`, payment turns ambiguous.
+
+    The operation id is identical in both states, so an id-based fence would
+    let the stale confirmation land. The full state token must not.
+    """
+    await _seed_provider()
+    operation_id = await _claim_ppq_topup(_row())
+    assert operation_id is not None
+    reviewed = await get_ppq_auto_topup_state(1)
+    assert reviewed["phase"] == PPQ_PHASE_CLAIMED
+
+    # Worker records the invoice: same operation, new phase, proofs committed.
+    await _record_ppq_invoice(
+        _row(),
+        operation_id,
+        invoice="lnbc-invoice",
+        invoice_id="invoice-1",
+        quote_id="quote-1",
+        amount=102,
+        amount_usd=10,
+        unit="sat",
+        mint_url="https://mint.test",
+    )
+
+    outcome = await release_ppq_auto_topup_state(
+        1, state_token=str(reviewed["state_token"])
+    )
+    assert outcome.released is False
+    assert outcome.reason == "stale_state"
+    row = await _state_row()
+    assert row is not None and row.swept is False
+
+
+async def test_concurrent_claims_only_one_wins(patched_db_engine: Any) -> None:
+    import asyncio
+
+    await _seed_provider()
+
+    results = await asyncio.gather(
+        *(_claim_ppq_topup(_row()) for _ in range(5)), return_exceptions=True
+    )
+    winners = [r for r in results if isinstance(r, str)]
+    assert len(winners) == 1
+
+    async with create_session() as session:
+        rows = (await session.exec(select(CashuTransaction))).all()
+    assert len(rows) == 1
+
+
+async def test_reconcile_releases_claim_when_mint_reports_unpaid(
+    patched_db_engine: Any,
+) -> None:
+    from routstr.upstream.auto_topup import _reconcile_ppq_state
+
+    # Lease expired, PPQ never credited: only the mint's own "unpaid" answer
+    # may hand the claim back.
+    await _seed_claim(1, PPQ_PHASE_RECONCILE, "invoice-1", int(time.time()) - 1)
+    provider = MagicMock()
+    provider.check_topup_status = AsyncMock(return_value=False)
+
+    with patch(
+        "routstr.upstream.auto_topup.check_bolt11_payment_status",
+        AsyncMock(return_value="unpaid"),
+    ) as status:
+        suppressed = await _reconcile_ppq_state(_row(), provider)
+
+    status.assert_awaited_once_with("https://mint.test", "sat", "quote-1")
+    assert suppressed is False
+    row = await _state_row()
+    assert row is not None and row.swept is True
+
+
+async def test_reconcile_keeps_claim_when_mint_answer_is_not_final(
+    patched_db_engine: Any,
+) -> None:
+    from routstr.upstream.auto_topup import _reconcile_ppq_state
+
+    await _seed_claim(1, PPQ_PHASE_RECONCILE, "invoice-1", int(time.time()) - 1)
+    provider = MagicMock()
+    provider.check_topup_status = AsyncMock(return_value=False)
+
+    for answer in ("paid", "pending", "unknown"):
+        with patch(
+            "routstr.upstream.auto_topup.check_bolt11_payment_status",
+            AsyncMock(return_value=answer),
+        ):
+            assert await _reconcile_ppq_state(_row(), provider) is True
+        row = await _state_row()
+        assert row is not None and row.swept is False, answer
+
+
+async def test_release_endpoint_maps_refusals_to_409(patched_db_engine: Any) -> None:
+    from fastapi import HTTPException
+
+    from routstr.core.admin import (
+        ReleasePPQAutoTopupRequest,
+        release_ppq_auto_topup_api,
+    )
+
+    provider_row = MagicMock()
+    provider_row.provider_type = "ppqai"
+
+    token = await _seed_claim(
+        1, PPQ_PHASE_IN_FLIGHT, "invoice-1", int(time.time()) + 900
+    )
+
+    with patch(
+        "routstr.core.admin._require_ppq_provider",
+        AsyncMock(return_value=provider_row),
+    ):
+        with pytest.raises(HTTPException) as excinfo:
+            await release_ppq_auto_topup_api(
+                1,
+                ReleasePPQAutoTopupRequest(
+                    confirmed_safe_to_retry=True, state_token=token
+                ),
+            )
+        assert excinfo.value.status_code == 409
+        assert "in flight" in excinfo.value.detail
+
+        with pytest.raises(HTTPException) as excinfo:
+            await release_ppq_auto_topup_api(
+                1,
+                ReleasePPQAutoTopupRequest(
+                    confirmed_safe_to_retry=True, state_token="ppq:wrong"
+                ),
+            )
+        assert excinfo.value.status_code == 409
+        assert "changed since" in excinfo.value.detail
+
+
+async def test_provider_delete_is_blocked_by_an_active_claim(
+    patched_db_engine: Any,
+) -> None:
+    from fastapi import HTTPException
+
+    from routstr.core.admin import delete_upstream_provider
+    from routstr.core.db import UpstreamProviderRow
+
+    async with create_session() as session:
+        session.add(
+            UpstreamProviderRow(
+                id=1,
+                slug="ppq",
+                provider_type="ppqai",
+                base_url="https://api.ppq.ai",
+                api_key="secret",
+                enabled=True,
+            )
+        )
+        await session.commit()
+    await _seed_claim(1, PPQ_PHASE_RECONCILE, "invoice-1", int(time.time()) + 900)
+
+    with pytest.raises(HTTPException) as excinfo:
+        await delete_upstream_provider("1")
+    assert excinfo.value.status_code == 409
+
+    # Provider must still exist.
+    async with create_session() as session:
+        assert await session.get(UpstreamProviderRow, 1) is not None
+
+
+async def test_claim_is_refused_when_the_provider_row_is_gone(
+    patched_db_engine: Any,
+) -> None:
+    """The worker's half of the delete race: no provider row, no claim."""
+    assert await _claim_ppq_topup(_row()) is None
+
+    async with create_session() as session:
+        rows = (await session.exec(select(CashuTransaction))).all()
+    assert rows == []
+
+
+async def test_claim_is_refused_after_a_provider_type_change(
+    patched_db_engine: Any,
+) -> None:
+    from routstr.core.db import UpstreamProviderRow
+
+    await _seed_provider()
+    async with create_session() as session:
+        provider = await session.get(UpstreamProviderRow, 1)
+        assert provider is not None
+        provider.provider_type = "openai"
+        session.add(provider)
+        await session.commit()
+
+    assert await _claim_ppq_topup(_row()) is None
+
+
+async def test_disabled_provider_with_claim_still_reconciles(
+    patched_db_engine: Any,
+) -> None:
+    """A claim tracks committed money; eligibility must not stop reconciling."""
+    from routstr.core.db import UpstreamProviderRow
+    from routstr.upstream.auto_topup import _reconcile_all_ppq_claims
+
+    await _seed_provider()
+    async with create_session() as session:
+        provider = await session.get(UpstreamProviderRow, 1)
+        assert provider is not None
+        provider.enabled = False
+        session.add(provider)
+        await session.commit()
+    await _seed_claim(1, PPQ_PHASE_RECONCILE, "invoice-1", int(time.time()) + 900)
+
+    ppq = MagicMock()
+    ppq.check_topup_status = AsyncMock(return_value=True)
+    with patch(
+        "routstr.upstream.auto_topup.PPQAIUpstreamProvider.from_db_row",
+        return_value=ppq,
+    ):
+        await _reconcile_all_ppq_claims()
+
+    row = await _state_row()
+    assert row is not None and row.collected is True
+
+
+async def test_claim_without_api_key_still_reconciles_via_the_mint(
+    patched_db_engine: Any,
+) -> None:
+    from routstr.core.db import UpstreamProviderRow
+    from routstr.upstream.auto_topup import _reconcile_all_ppq_claims
+
+    await _seed_provider()
+    async with create_session() as session:
+        provider = await session.get(UpstreamProviderRow, 1)
+        assert provider is not None
+        provider.api_key = ""
+        session.add(provider)
+        await session.commit()
+    # Lease expired, so the mint may be consulted.
+    await _seed_claim(1, PPQ_PHASE_RECONCILE, "invoice-1", int(time.time()) - 1)
+
+    with patch(
+        "routstr.upstream.auto_topup.check_bolt11_payment_status",
+        AsyncMock(return_value="unpaid"),
+    ) as status:
+        await _reconcile_all_ppq_claims()
+
+    # No API key: PPQ was never polled, but the mint was, and its definitive
+    # "unpaid" released the claim.
+    status.assert_awaited_once()
+    row = await _state_row()
+    assert row is not None and row.swept is True
+
+
+def test_ppq_payment_usd_prefers_stamped_amount() -> None:
+    # Stamped rows must not move with the BTC price.
+    assert _ppq_payment_usd(102, "sat", "ppq-invoice:a:usd:10", 0.5) == 10.0
+
+
+def test_ppq_payment_usd_falls_back_to_current_price() -> None:
+    # Rows recorded before the stamp existed convert sats at today's price.
+    assert _ppq_payment_usd(2000, "sat", "ppq-invoice:legacy", 0.001) == 2.0
+    assert _ppq_payment_usd(2_000_000, "msat", "ppq-invoice:legacy", 0.001) == 2.0
+
+
+def test_ppq_payment_usd_survives_malformed_stamp() -> None:
+    assert _ppq_payment_usd(3000, "sat", "ppq-invoice:x:usd:oops", 0.001) == 3.0
+
+
+async def test_daily_spend_ignores_provably_unattempted_payments(
+    patched_db_engine: Any,
+) -> None:
+    def _payment(
+        id_: str, token: str, collected: bool, swept: bool
+    ) -> CashuTransaction:
+        return CashuTransaction(
+            id=id_,
+            token=token,
+            amount=1,
+            unit="sat",
+            type="out",
+            source="ppq_auto_topup",
+            collected=collected,
+            swept=swept,
+        )
+
+    async with create_session() as session:
+        # Settled, in-flight, and provably-unattempted payments plus a
+        # pre-stamp row: only the unattempted one must be excluded.
+        session.add(_payment("pay-usd-1", "ppq-invoice:a:usd:100", True, False))
+        session.add(_payment("pay-usd-2", "ppq-invoice:b:usd:50", False, False))
+        session.add(_payment("pay-usd-3", "ppq-invoice:c:usd:25", False, True))
+        legacy = _payment("pay-usd-4", "ppq-invoice:legacy", True, False)
+        legacy.amount = 2000
+        session.add(legacy)
+        await session.commit()
+
+    assert await _ppq_spent_last_24h_usd(0.001) == 152.0

tests/integration/test_wallet_melt_restart.py

@@ -0,0 +1,115 @@
+"""Restart reconciliation for ambiguous melts, against a real cashu wallet DB.
+
+The ambiguous-melt path in ``execute_bolt11_payment`` re-reserves proofs with
+``set_reserved_for_melt(..., quote_id=...)`` after cashu's ``melt()`` clears
+both the reservation and the ``melt_id`` on a transport error. These tests
+prove, on cashu's actual sqlite store rather than mocks, that the recovery
+survives a process restart: a fresh wallet instance on the same database can
+still find the proofs by ``melt_id`` — the lookup ``get_melt_quote()`` uses to
+invalidate them on "paid" or release them on "unpaid".
+"""
+
+from pathlib import Path
+
+import pytest
+from cashu.core.base import Proof
+from cashu.wallet import crud
+from cashu.wallet.wallet import Wallet
+
+pytestmark = pytest.mark.asyncio
+
+QUOTE_ID = "quote-restart-1"
+
+
+def _proof(secret: str, amount: int = 64) -> Proof:
+    return Proof(
+        id="009a1f293253e41e",
+        amount=amount,
+        secret=secret,
+        C="02bc9097997d81afb2cc7346b5e4345a9346bd2a506eb7958598a72f0cf85163ea",
+    )
+
+
+async def _wallet(db_dir: Path) -> Wallet:
+    # with_db builds the instance and runs migrations locally; nothing here
+    # talks to a mint.
+    return await Wallet.with_db("https://mint.test", str(db_dir))
+
+
+async def _seed_ambiguous_melt(wallet: Wallet) -> list[Proof]:
+    """Reproduce the exact sequence of an ambiguous melt failure.
+
+    1. Proofs exist and are selected for a melt.
+    2. cashu's melt() reserves them with the quote id, then hits a transport
+       error and rolls that back — reservation gone, melt_id gone.
+    3. Our recovery in execute_bolt11_payment re-reserves with the quote id.
+    """
+    proofs = [_proof("secret-a"), _proof("secret-b", amount=32)]
+    for proof in proofs:
+        await crud.store_proof(proof, db=wallet.db)
+
+    await wallet.set_reserved_for_melt(proofs, reserved=True, quote_id=QUOTE_ID)
+    # cashu's `except` block in melt():
+    await wallet.set_reserved_for_melt(proofs, reserved=False, quote_id=None)
+    # our recovery:
+    await wallet.set_reserved_for_melt(proofs, reserved=True, quote_id=QUOTE_ID)
+    return proofs
+
+
+async def test_melt_recovery_is_findable_by_quote_after_restart(
+    tmp_path: Path,
+) -> None:
+    wallet = await _wallet(tmp_path)
+    await _seed_ambiguous_melt(wallet)
+
+    # "Restart": a brand-new wallet on the same database file, as after a
+    # process crash between the melt and any reconciliation.
+    restarted = await _wallet(tmp_path)
+    found = await crud.get_proofs(db=restarted.db, melt_id=QUOTE_ID)
+
+    # This is get_melt_quote()'s own lookup. If it comes back empty, a "paid"
+    # answer can never invalidate these proofs and an "unpaid" answer can
+    # never release them — the strand the send-style re-reserve caused.
+    assert sorted(p.secret for p in found) == ["secret-a", "secret-b"]
+    assert all(p.reserved for p in found)
+    assert all(p.melt_id == QUOTE_ID for p in found)
+
+
+async def test_send_style_reservation_would_not_be_reconcilable(
+    tmp_path: Path,
+) -> None:
+    """The defect the fix removed, demonstrated on the real store."""
+    wallet = await _wallet(tmp_path)
+    proofs = [_proof("secret-send")]
+    for proof in proofs:
+        await crud.store_proof(proof, db=wallet.db)
+
+    await wallet.set_reserved_for_melt(proofs, reserved=True, quote_id=QUOTE_ID)
+    await wallet.set_reserved_for_melt(proofs, reserved=False, quote_id=None)
+    # The old recovery: reserve as a send, no quote association.
+    await wallet.set_reserved_for_send(proofs, reserved=True)
+
+    restarted = await _wallet(tmp_path)
+    found = await crud.get_proofs(db=restarted.db, melt_id=QUOTE_ID)
+    assert found == []  # reconciliation would never see these proofs
+
+
+async def test_unpaid_reconciliation_releases_recovered_proofs_after_restart(
+    tmp_path: Path,
+) -> None:
+    """The full recovery arc: crash, restart, mint says unpaid, funds usable."""
+    wallet = await _wallet(tmp_path)
+    await _seed_ambiguous_melt(wallet)
+
+    restarted = await _wallet(tmp_path)
+    found = await crud.get_proofs(db=restarted.db, melt_id=QUOTE_ID)
+    assert len(found) == 2
+
+    # What get_melt_quote() does on an "unpaid" answer.
+    await restarted.set_reserved_for_melt(found, reserved=False, quote_id=None)
+
+    released = await crud.get_proofs(db=restarted.db, melt_id=QUOTE_ID)
+    assert released == []
+    all_proofs = await crud.get_proofs(db=restarted.db)
+    assert len(all_proofs) == 2
+    assert all(not p.reserved for p in all_proofs)  # spendable again

tests/unit/test_auto_topup.py

@@ -4,7 +4,29 @@ from unittest.mock import AsyncMock, MagicMock, patch
 import pytest
 
 from routstr.core.db import CashuTransaction
-from routstr.upstream.auto_topup import _check_and_topup
+from routstr.upstream.auto_topup import (
+    _check_and_topup,
+    _parse_ppq_request_id,
+    _run_auto_topup_cycle,
+    validate_ppq_auto_topup_settings,
+)
+from routstr.upstream.ppqai import PPQAIUpstreamProvider
+from routstr.wallet import Bolt11PaymentAmbiguous, Bolt11PaymentNotAttempted
+
+
+def test_ppq_claim_parser_rejects_invalid_expiry() -> None:
+    assert (
+        _parse_ppq_request_id("ppq:operation:not-a-timestamp:claimed:invoice:none")
+        is None
+    )
+
+
+@pytest.mark.asyncio
+async def test_ppq_balance_rejects_boolean_api_value() -> None:
+    provider = PPQAIUpstreamProvider("secret")
+    provider.check_balance = AsyncMock(return_value={"balance": False})  # type: ignore[method-assign]
+
+    assert await provider.get_balance() is None
 
 
 def _row() -> MagicMock:
@@ -12,6 +34,7 @@ def _row() -> MagicMock:
     row.id = "provider-1"
     row.base_url = "https://provider.test"
     row.api_key = "secret"
+    row.provider_type = "routstr"
     row.provider_settings = json.dumps(
         {
             "auto_topup": True,
@@ -151,3 +174,563 @@ async def test_auto_topup_does_not_send_untracked_token() -> None:
 
     reclaim.assert_awaited_once_with("cashu-token")
     provider.topup.assert_not_awaited()
+
+
+def _ppq_row() -> MagicMock:
+    row = MagicMock()
+    row.id = "ppq-provider-1"
+    row.base_url = "https://api.ppq.ai"
+    row.api_key = "secret"
+    row.provider_type = "ppqai"
+    row.provider_settings = json.dumps(
+        {
+            "auto_topup": True,
+            "topup_threshold": 5.0,
+            "topup_amount_limit": 10,
+        }
+    )
+    return row
+
+
+@pytest.mark.asyncio
+async def test_ppq_auto_topup_pays_invoice_and_confirms_settlement() -> None:
+    provider = MagicMock()
+    provider.get_balance = AsyncMock(return_value=2.5)
+    provider.initiate_topup = AsyncMock(
+        return_value=MagicMock(
+            invoice_id="invoice-1",
+            payment_request="lnbc-invoice",
+            amount=10,
+            currency="USD",
+            expires_at=None,
+        )
+    )
+    provider.check_topup_status = AsyncMock(return_value=True)
+    plan = MagicMock()
+    plan.invoice_amount_sats = 100
+    plan.maximum_spend_sats = 102
+    plan.quote.amount = 100
+    plan.quote.fee_reserve = 2
+    plan.mint_url = "https://mint-rich.test"
+    plan.unit = "sat"
+    row = _ppq_row()
+
+    with (
+        patch(
+            "routstr.upstream.auto_topup.PPQAIUpstreamProvider.from_db_row",
+            return_value=provider,
+        ),
+        patch(
+            "routstr.upstream.auto_topup._reconcile_ppq_state",
+            AsyncMock(return_value=False),
+        ),
+        patch(
+            "routstr.upstream.auto_topup._claim_ppq_topup",
+            AsyncMock(return_value="operation-1"),
+        ),
+        patch(
+            "routstr.upstream.auto_topup.maximum_owner_cashu_balance_sats",
+            AsyncMock(return_value=10_000),
+        ),
+        patch(
+            "routstr.upstream.auto_topup._ppq_spent_last_24h_usd",
+            AsyncMock(return_value=0.0),
+        ),
+        patch(
+            "routstr.upstream.auto_topup.prepare_bolt11_payment",
+            AsyncMock(return_value=plan),
+        ) as prepare,
+        patch(
+            "routstr.upstream.auto_topup.execute_bolt11_payment",
+            AsyncMock(return_value=(101, "https://mint-rich.test", "sat")),
+        ) as execute,
+        patch("routstr.upstream.auto_topup._record_ppq_invoice", AsyncMock()) as record,
+        patch(
+            "routstr.upstream.auto_topup._record_ppq_payment_spent", AsyncMock()
+        ) as record_spent,
+        patch(
+            "routstr.upstream.auto_topup._set_ppq_state_terminal", AsyncMock()
+        ) as terminal,
+        patch("routstr.upstream.auto_topup.sats_usd_price", return_value=0.001),
+    ):
+        await _check_and_topup(row)
+
+    provider.initiate_topup.assert_awaited_once_with(10)
+    prepare.assert_awaited_once_with("lnbc-invoice")
+    execute.assert_awaited_once_with(plan)
+    record.assert_awaited_once()
+    record_spent.assert_awaited_once_with("operation-1", 101)
+    provider.check_topup_status.assert_awaited_once_with("invoice-1")
+    terminal.assert_awaited_once_with(row, "operation-1", collected=True, swept=False)
+
+
+@pytest.mark.asyncio
+async def test_ppq_ambiguous_melt_keeps_claim_and_emits_critical_alert() -> None:
+    provider = MagicMock()
+    provider.get_balance = AsyncMock(return_value=2.5)
+    provider.initiate_topup = AsyncMock(
+        return_value=MagicMock(
+            invoice_id="invoice-1",
+            payment_request="lnbc-invoice",
+            amount=10,
+            currency="USD",
+            expires_at=None,
+        )
+    )
+    plan = MagicMock(maximum_spend_sats=102, mint_url="https://mint.test", unit="sat")
+    plan.quote.amount = 100
+    plan.quote.fee_reserve = 2
+    row = _ppq_row()
+
+    with (
+        patch(
+            "routstr.upstream.auto_topup.PPQAIUpstreamProvider.from_db_row",
+            return_value=provider,
+        ),
+        patch(
+            "routstr.upstream.auto_topup._reconcile_ppq_state",
+            AsyncMock(return_value=False),
+        ),
+        patch(
+            "routstr.upstream.auto_topup._claim_ppq_topup",
+            AsyncMock(return_value="operation-1"),
+        ),
+        patch(
+            "routstr.upstream.auto_topup.maximum_owner_cashu_balance_sats",
+            AsyncMock(return_value=10_000),
+        ),
+        patch(
+            "routstr.upstream.auto_topup._ppq_spent_last_24h_usd",
+            AsyncMock(return_value=0.0),
+        ),
+        patch(
+            "routstr.upstream.auto_topup.prepare_bolt11_payment",
+            AsyncMock(return_value=plan),
+        ),
+        patch(
+            "routstr.upstream.auto_topup.execute_bolt11_payment",
+            AsyncMock(side_effect=Bolt11PaymentAmbiguous("ambiguous melt")),
+        ),
+        patch(
+            "routstr.upstream.auto_topup._record_ppq_invoice",
+            AsyncMock(return_value=2_000_000_000),
+        ),
+        patch(
+            "routstr.upstream.auto_topup._mark_ppq_reconcile", AsyncMock()
+        ) as reconcile_mark,
+        patch(
+            "routstr.upstream.auto_topup._set_ppq_state_terminal", AsyncMock()
+        ) as terminal,
+        patch("routstr.upstream.auto_topup.sats_usd_price", return_value=0.001),
+        patch("routstr.upstream.auto_topup.logger.critical") as critical,
+    ):
+        with pytest.raises(Bolt11PaymentAmbiguous, match="ambiguous melt"):
+            await _check_and_topup(row)
+
+    # The claim is never released — it moves to reconcile for the admin.
+    terminal.assert_not_awaited()
+    reconcile_mark.assert_awaited_once()
+    critical.assert_called_once()
+    assert "admin reconciliation" in critical.call_args.args[0]
+
+
+@pytest.mark.asyncio
+async def test_ppq_payment_not_attempted_releases_claim_for_retry() -> None:
+    provider = MagicMock()
+    provider.get_balance = AsyncMock(return_value=2.5)
+    provider.initiate_topup = AsyncMock(
+        return_value=MagicMock(
+            invoice_id="invoice-1",
+            payment_request="lnbc-invoice",
+            amount=10,
+            currency="USD",
+            expires_at=None,
+        )
+    )
+    plan = MagicMock(maximum_spend_sats=102, mint_url="https://mint.test", unit="sat")
+    plan.quote.amount = 100
+    plan.quote.fee_reserve = 2
+    plan.quote.quote = "quote-1"
+    terminal = AsyncMock(return_value=True)
+    row = _ppq_row()
+
+    with (
+        patch(
+            "routstr.upstream.auto_topup.PPQAIUpstreamProvider.from_db_row",
+            return_value=provider,
+        ),
+        patch(
+            "routstr.upstream.auto_topup._reconcile_ppq_state",
+            AsyncMock(return_value=False),
+        ),
+        patch(
+            "routstr.upstream.auto_topup.maximum_owner_cashu_balance_sats",
+            AsyncMock(return_value=10_000),
+        ),
+        patch(
+            "routstr.upstream.auto_topup._ppq_spent_last_24h_usd",
+            AsyncMock(return_value=0.0),
+        ),
+        patch(
+            "routstr.upstream.auto_topup._claim_ppq_topup",
+            AsyncMock(return_value="operation-1"),
+        ),
+        patch(
+            "routstr.upstream.auto_topup.prepare_bolt11_payment",
+            AsyncMock(return_value=plan),
+        ),
+        patch(
+            "routstr.upstream.auto_topup._record_ppq_invoice",
+            AsyncMock(return_value=2_000_000_000),
+        ),
+        patch(
+            "routstr.upstream.auto_topup.execute_bolt11_payment",
+            AsyncMock(side_effect=Bolt11PaymentNotAttempted("unpaid")),
+        ),
+        patch("routstr.upstream.auto_topup._set_ppq_state_terminal", terminal),
+        patch(
+            "routstr.upstream.auto_topup._mark_ppq_reconcile", AsyncMock()
+        ) as reconcile,
+        patch("routstr.upstream.auto_topup.sats_usd_price", return_value=0.001),
+        pytest.raises(Bolt11PaymentNotAttempted, match="unpaid"),
+    ):
+        await _check_and_topup(row)
+
+    terminal.assert_awaited_once_with(row, "operation-1", collected=False, swept=True)
+    reconcile.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_ppq_status_error_after_payment_marks_reconcile_and_alerts() -> None:
+    provider = MagicMock()
+    provider.get_balance = AsyncMock(return_value=2.5)
+    provider.initiate_topup = AsyncMock(
+        return_value=MagicMock(
+            invoice_id="invoice-1",
+            payment_request="lnbc-invoice",
+            amount=10,
+            currency="USD",
+            expires_at=None,
+        )
+    )
+    provider.check_topup_status = AsyncMock(side_effect=RuntimeError("PPQ 502"))
+    plan = MagicMock(maximum_spend_sats=102, mint_url="https://mint.test", unit="sat")
+    plan.quote.amount = 100
+    plan.quote.fee_reserve = 2
+    plan.quote.quote = "quote-1"
+
+    with (
+        patch(
+            "routstr.upstream.auto_topup.PPQAIUpstreamProvider.from_db_row",
+            return_value=provider,
+        ),
+        patch(
+            "routstr.upstream.auto_topup._reconcile_ppq_state",
+            AsyncMock(return_value=False),
+        ),
+        patch(
+            "routstr.upstream.auto_topup.maximum_owner_cashu_balance_sats",
+            AsyncMock(return_value=10_000),
+        ),
+        patch(
+            "routstr.upstream.auto_topup._ppq_spent_last_24h_usd",
+            AsyncMock(return_value=0.0),
+        ),
+        patch(
+            "routstr.upstream.auto_topup._claim_ppq_topup",
+            AsyncMock(return_value="operation-1"),
+        ),
+        patch(
+            "routstr.upstream.auto_topup.prepare_bolt11_payment",
+            AsyncMock(return_value=plan),
+        ),
+        patch(
+            "routstr.upstream.auto_topup._record_ppq_invoice",
+            AsyncMock(return_value=2_000_000_000),
+        ),
+        patch(
+            "routstr.upstream.auto_topup.execute_bolt11_payment",
+            AsyncMock(return_value=(101, "https://mint.test", "sat")),
+        ),
+        patch(
+            "routstr.upstream.auto_topup._record_ppq_payment_spent", AsyncMock()
+        ) as spent,
+        patch(
+            "routstr.upstream.auto_topup._mark_ppq_reconcile", AsyncMock()
+        ) as reconcile,
+        patch(
+            "routstr.upstream.auto_topup._set_ppq_state_terminal", AsyncMock()
+        ) as terminal,
+        patch("routstr.upstream.auto_topup.sats_usd_price", return_value=0.001),
+        patch("routstr.upstream.auto_topup.logger.critical") as critical,
+    ):
+        await _check_and_topup(_ppq_row())
+
+    spent.assert_awaited_once_with("operation-1", 101)
+    reconcile.assert_awaited_once()
+    terminal.assert_not_awaited()
+    assert "settlement polling failed" in critical.call_args.args[0]
+
+
+@pytest.mark.asyncio
+async def test_ppq_preflight_funding_check_happens_before_invoice_creation() -> None:
+    provider = MagicMock()
+    provider.get_balance = AsyncMock(return_value=2.5)
+    provider.initiate_topup = AsyncMock()
+
+    with (
+        patch(
+            "routstr.upstream.auto_topup.PPQAIUpstreamProvider.from_db_row",
+            return_value=provider,
+        ),
+        patch(
+            "routstr.upstream.auto_topup._reconcile_ppq_state",
+            AsyncMock(return_value=False),
+        ),
+        patch(
+            "routstr.upstream.auto_topup.maximum_owner_cashu_balance_sats",
+            AsyncMock(return_value=1),
+        ),
+        patch("routstr.upstream.auto_topup._claim_ppq_topup", AsyncMock()) as claim,
+        patch("routstr.upstream.auto_topup.sats_usd_price", return_value=0.001),
+    ):
+        await _check_and_topup(_ppq_row())
+
+    provider.initiate_topup.assert_not_awaited()
+    claim.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_active_claim_at_cycle_start_suppresses_topup_for_whole_cycle() -> None:
+    row = _ppq_row()
+    row.id = 1
+    session = AsyncMock()
+    result = MagicMock()
+    result.all.return_value = [row]
+    session.exec.return_value = result
+    context = MagicMock()
+    context.__aenter__ = AsyncMock(return_value=session)
+    context.__aexit__ = AsyncMock(return_value=None)
+
+    with (
+        patch(
+            "routstr.upstream.auto_topup._reconcile_all_ppq_claims",
+            AsyncMock(return_value={1}),
+        ),
+        patch("routstr.upstream.auto_topup.create_session", return_value=context),
+        patch("routstr.upstream.auto_topup._check_and_topup", AsyncMock()) as check,
+    ):
+        await _run_auto_topup_cycle()
+
+    check.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_ppq_auto_topup_skips_when_balance_meets_threshold() -> None:
+    provider = MagicMock()
+    provider.get_balance = AsyncMock(return_value=5.0)
+    provider.initiate_topup = AsyncMock()
+
+    with (
+        patch(
+            "routstr.upstream.auto_topup.PPQAIUpstreamProvider.from_db_row",
+            return_value=provider,
+        ),
+        patch(
+            "routstr.upstream.auto_topup._reconcile_ppq_state",
+            AsyncMock(return_value=False),
+        ),
+    ):
+        await _check_and_topup(_ppq_row())
+
+    provider.initiate_topup.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_ppq_auto_topup_skips_when_daily_spend_cap_reached() -> None:
+    provider = MagicMock()
+    provider.get_balance = AsyncMock(return_value=2.5)
+    provider.initiate_topup = AsyncMock()
+
+    with (
+        patch(
+            "routstr.upstream.auto_topup.PPQAIUpstreamProvider.from_db_row",
+            return_value=provider,
+        ),
+        patch(
+            "routstr.upstream.auto_topup._reconcile_ppq_state",
+            AsyncMock(return_value=False),
+        ),
+        patch(
+            "routstr.upstream.auto_topup.maximum_owner_cashu_balance_sats",
+            AsyncMock(return_value=10_000_000),
+        ),
+        # 1000 USD already spent, exactly the daily cap: the next 10 USD
+        # top-up must be refused.
+        patch(
+            "routstr.upstream.auto_topup._ppq_spent_last_24h_usd",
+            AsyncMock(return_value=1000.0),
+        ),
+        patch(
+            "routstr.upstream.auto_topup._claim_ppq_topup",
+            AsyncMock(),
+        ) as claim,
+        patch("routstr.upstream.auto_topup.sats_usd_price", return_value=0.001),
+    ):
+        await _check_and_topup(_ppq_row())
+
+    claim.assert_not_awaited()
+    provider.initiate_topup.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_ppq_pending_attempt_suppresses_duplicate_topup() -> None:
+    provider = MagicMock()
+    provider.get_balance = AsyncMock()
+
+    with (
+        patch(
+            "routstr.upstream.auto_topup.PPQAIUpstreamProvider.from_db_row",
+            return_value=provider,
+        ),
+        patch(
+            "routstr.upstream.auto_topup._reconcile_ppq_state",
+            AsyncMock(return_value=True),
+        ),
+    ):
+        await _check_and_topup(_ppq_row())
+
+    provider.get_balance.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_ppq_auto_topup_rejects_non_finite_balance() -> None:
+    provider = MagicMock()
+    provider.get_balance = AsyncMock(return_value=float("nan"))
+
+    with (
+        patch(
+            "routstr.upstream.auto_topup.PPQAIUpstreamProvider.from_db_row",
+            return_value=provider,
+        ),
+        patch(
+            "routstr.upstream.auto_topup._reconcile_ppq_state",
+            AsyncMock(return_value=False),
+        ),
+        patch("routstr.upstream.auto_topup._claim_ppq_topup", AsyncMock()) as claim,
+    ):
+        await _check_and_topup(_ppq_row())
+
+    claim.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_settled_topup_alerts_when_its_claim_was_already_released() -> None:
+    provider = MagicMock()
+    provider.get_balance = AsyncMock(return_value=2.5)
+    provider.initiate_topup = AsyncMock(
+        return_value=MagicMock(
+            invoice_id="invoice-1",
+            payment_request="lnbc-invoice",
+            amount=10,
+            currency="USD",
+            expires_at=None,
+        )
+    )
+    provider.check_topup_status = AsyncMock(return_value=True)
+    plan = MagicMock()
+    plan.maximum_spend_sats = 102
+    plan.quote.amount = 100
+    plan.quote.fee_reserve = 2
+    plan.mint_url = "https://mint-rich.test"
+    plan.unit = "sat"
+
+    with (
+        patch(
+            "routstr.upstream.auto_topup.PPQAIUpstreamProvider.from_db_row",
+            return_value=provider,
+        ),
+        patch(
+            "routstr.upstream.auto_topup._reconcile_ppq_state",
+            AsyncMock(return_value=False),
+        ),
+        patch(
+            "routstr.upstream.auto_topup._claim_ppq_topup",
+            AsyncMock(return_value="operation-1"),
+        ),
+        patch(
+            "routstr.upstream.auto_topup.prepare_bolt11_payment",
+            AsyncMock(return_value=plan),
+        ),
+        patch(
+            "routstr.upstream.auto_topup.maximum_owner_cashu_balance_sats",
+            AsyncMock(return_value=10_000),
+        ),
+        patch(
+            "routstr.upstream.auto_topup._ppq_spent_last_24h_usd",
+            AsyncMock(return_value=0.0),
+        ),
+        patch(
+            "routstr.upstream.auto_topup.execute_bolt11_payment",
+            AsyncMock(return_value=(101, "https://mint-rich.test", "sat")),
+        ),
+        patch("routstr.upstream.auto_topup._record_ppq_invoice", AsyncMock()),
+        patch("routstr.upstream.auto_topup._record_ppq_payment_spent", AsyncMock()),
+        patch(
+            "routstr.upstream.auto_topup._set_ppq_state_terminal",
+            AsyncMock(return_value=False),
+        ),
+        patch("routstr.upstream.auto_topup.sats_usd_price", return_value=0.001),
+        patch("routstr.upstream.auto_topup.logger") as log,
+    ):
+        await _check_and_topup(_ppq_row())
+
+    assert any(
+        "claim was already released" in call.args[0]
+        for call in log.critical.call_args_list
+    )
+
+
+@pytest.mark.parametrize(
+    ("settings", "expected"),
+    [
+        ({"auto_topup": False, "topup_threshold": -1}, None),
+        (
+            {"auto_topup": True, "topup_threshold": 5, "topup_amount_limit": 10},
+            None,
+        ),
+        (
+            {"auto_topup": True, "topup_threshold": None, "topup_amount_limit": 10},
+            "threshold",
+        ),
+        (
+            {"auto_topup": True, "topup_threshold": 5, "topup_amount_limit": 0.5},
+            "whole number",
+        ),
+        (
+            {"auto_topup": True, "topup_threshold": 5, "topup_amount_limit": 5000},
+            "between",
+        ),
+        (
+            {"auto_topup": True, "topup_threshold": True, "topup_amount_limit": 10},
+            "threshold",
+        ),
+    ],
+)
+def test_ppq_auto_topup_settings_validation(
+    settings: dict, expected: str | None
+) -> None:
+    problem = validate_ppq_auto_topup_settings(settings)
+    if expected is None:
+        assert problem is None
+    else:
+        assert problem is not None and expected in problem
+
+
+def test_ppq_auto_topup_settings_validation_survives_huge_json_integers() -> None:
+    # json.loads happily produces integers past float range; float() raises
+    # OverflowError there instead of returning inf.
+    problem = validate_ppq_auto_topup_settings(
+        {"auto_topup": True, "topup_threshold": 10**400, "topup_amount_limit": 10}
+    )
+    assert problem is not None and "threshold" in problem

tests/unit/test_cost_calculation_caching.py

@@ -15,6 +15,7 @@ os.environ.setdefault("LIGHTNING_ADDRESS", "test@stm.to")
 
 from routstr.core.settings import settings
 from routstr.payment.cost_calculation import CostData, MaxCostData, calculate_cost
+from routstr.payment.models import Architecture, Model, Pricing
 
 
 @pytest.fixture(autouse=True)
@@ -527,11 +528,136 @@ async def test_openrouter_upstream_inference_cost_components_are_used() -> None:
     result = await calculate_cost(response, max_cost=100000)
 
     assert isinstance(result, CostData)
-    assert result.input_msats == 994
-    assert result.output_msats == 3477
+    assert result.input_msats == 995
+    assert result.output_msats == 3476
+    assert result.cache_read_msats == 758
+    assert result.cache_creation_msats == 0
     assert result.input_msats + result.output_msats == result.total_msats == 4471
 
 
+@pytest.mark.asyncio
+async def test_usd_cache_breakdown_matches_token_priced_path(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Authoritative USD totals must retain model-specific cache-rate ratios."""
+    monkeypatch.setattr(settings, "fixed_pricing", False)
+    model = Model(
+        id="cache-priced-model",
+        name="cache-priced-model",
+        created=0,
+        description="",
+        context_length=8192,
+        architecture=Architecture(
+            modality="text",
+            input_modalities=["text"],
+            output_modalities=["text"],
+            tokenizer="test",
+            instruct_type=None,
+        ),
+        pricing=Pricing(prompt=0.01, completion=0.02),
+        sats_pricing=Pricing(
+            prompt=0.01,
+            completion=0.02,
+            input_cache_read=0.001,
+            input_cache_write=0.01,
+        ),
+        per_request_limits=None,
+        top_provider=None,
+    )
+    usage = {
+        "prompt_tokens": 1000,
+        "completion_tokens": 100,
+        "prompt_tokens_details": {"cached_tokens": 900},
+    }
+
+    token_result = await calculate_cost(
+        {"model": model.id, "usage": usage},
+        max_cost=100_000,
+        model_obj=model,
+    )
+    usd_result = await calculate_cost(
+        {
+            "model": model.id,
+            "usage": {
+                **usage,
+                "cost": 0.000195,
+                "cost_details": {
+                    "input_cost": 0.000095,
+                    "output_cost": 0.0001,
+                },
+            },
+        },
+        max_cost=100_000,
+        model_obj=model,
+        provider_fee=1.0,
+    )
+
+    assert isinstance(token_result, CostData)
+    assert isinstance(usd_result, CostData)
+    assert usd_result.total_msats == token_result.total_msats == 3900
+    assert usd_result.input_msats + usd_result.output_msats == usd_result.total_msats
+    assert usd_result.cache_read_msats == token_result.cache_read_msats == 900
+
+
+@pytest.mark.asyncio
+async def test_usd_cache_breakdown_does_not_absorb_total_rounding_remainder(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Sub-msat cache components truncate like the token-priced path."""
+    monkeypatch.setattr(settings, "fixed_pricing", False)
+    model = Model(
+        id="sub-msat-cache-model",
+        name="sub-msat-cache-model",
+        created=0,
+        description="",
+        context_length=8192,
+        architecture=Architecture(
+            modality="text",
+            input_modalities=["text"],
+            output_modalities=["text"],
+            tokenizer="test",
+            instruct_type=None,
+        ),
+        pricing=Pricing(prompt=0.001, completion=0.001),
+        sats_pricing=Pricing(
+            prompt=0.001,
+            completion=0.001,
+            input_cache_write=0.0006,
+        ),
+        per_request_limits=None,
+        top_provider=None,
+    )
+    usage = {
+        "input_tokens": 0,
+        "output_tokens": 0,
+        "cache_creation_input_tokens": 1,
+    }
+
+    token_result = await calculate_cost(
+        {"model": model.id, "usage": usage},
+        max_cost=100_000,
+        model_obj=model,
+    )
+    usd_result = await calculate_cost(
+        {
+            "model": model.id,
+            "usage": {
+                **usage,
+                "cost": 0.00000003,
+                "cost_details": {"input_cost": 0.00000003},
+            },
+        },
+        max_cost=100_000,
+        model_obj=model,
+        provider_fee=1.0,
+    )
+
+    assert isinstance(token_result, CostData)
+    assert isinstance(usd_result, CostData)
+    assert usd_result.total_msats == token_result.total_msats == 1
+    assert usd_result.cache_creation_msats == token_result.cache_creation_msats == 0
+
+
 # ============================================================================
 # PPQ.AI BYOK: upstream_inference_cost + BYOK fee billing
 #
@@ -568,12 +694,14 @@ async def test_ppq_byok_bills_upstream_inference_cost_plus_fee() -> None:
     # msats), not the fee alone (~0.0023 USD → ~45k msats).  ~20× correction.
     assert result.total_msats == 940274
     assert result.input_msats + result.output_msats == result.total_msats
-    assert result.input_msats == 926546
-    assert result.output_msats == 13728
+    assert result.input_msats == 926547
+    assert result.output_msats == 13727
     assert result.total_usd == pytest.approx(0.047013667305)
     # Token normalisation (OpenAI dialect: cached included in prompt_tokens)
     assert result.input_tokens == 5070  # 164371 - 159301
     assert result.cache_read_input_tokens == 159301
+    assert result.cache_read_msats == 897966
+    assert result.cache_creation_msats == 0
     assert result.output_tokens == 99
 
 

tests/unit/test_cost_response_metadata.py

@@ -0,0 +1,113 @@
+"""Response-contract tests for Routstr cost metadata across paid paths."""
+
+import json
+import os
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import httpx
+import pytest
+
+os.environ.setdefault("UPSTREAM_BASE_URL", "http://test")
+os.environ.setdefault("UPSTREAM_API_KEY", "test")
+
+from routstr.core.db import ApiKey  # noqa: E402
+from routstr.upstream.base import BaseUpstreamProvider  # noqa: E402
+
+COST_DATA = {
+    "base_msats": 0,
+    "input_msats": 1_200,
+    "output_msats": 300,
+    "total_msats": 1_500,
+    "total_usd": 0.0001,
+    "input_tokens": 10,
+    "output_tokens": 3,
+    "cache_read_input_tokens": 8,
+    "cache_creation_input_tokens": 2,
+    "cache_read_msats": 80,
+    "cache_creation_msats": 40,
+}
+
+
+def _provider() -> BaseUpstreamProvider:
+    return BaseUpstreamProvider(base_url="http://test", api_key="upstream-key")
+
+
+def _key() -> ApiKey:
+    return ApiKey(hashed_key="abcdef0123" * 4, balance=1_000_000)
+
+
+def _session() -> Any:
+    session = MagicMock()
+    session.refresh = AsyncMock()
+    return session
+
+
+def _upstream_response(payload: dict) -> httpx.Response:
+    return httpx.Response(
+        200,
+        json=payload,
+        request=httpx.Request("POST", "http://test"),
+    )
+
+
+def _assert_cost_contract(response: Any) -> None:
+    body = json.loads(response.body)
+    assert body["usage"]["cost"] == {
+        "base_msats": 0,
+        "input_msats": 1_200,
+        "output_msats": 300,
+        "total_msats": 1_500,
+        "total_usd": 0.0001,
+        "cache_read_input_tokens": 8,
+        "cache_creation_input_tokens": 2,
+        "cache_read_msats": 80,
+        "cache_creation_msats": 40,
+    }
+    assert response.headers["X-Routstr-Cost-Msats"] == "1500"
+    assert response.headers["X-Routstr-Input-Cost-Msats"] == "1200"
+    assert response.headers["X-Routstr-Output-Cost-Msats"] == "300"
+
+
+@pytest.mark.asyncio
+async def test_balance_chat_completion_uses_shared_cost_contract() -> None:
+    provider = _provider()
+    with patch(
+        "routstr.upstream.base.adjust_payment_for_tokens",
+        new=AsyncMock(return_value=dict(COST_DATA)),
+    ):
+        response = await provider.handle_non_streaming_chat_completion(
+            _upstream_response(
+                {
+                    "model": "test-model",
+                    "usage": {"prompt_tokens": 10, "completion_tokens": 3},
+                }
+            ),
+            _key(),
+            _session(),
+            deducted_max_cost=10_000,
+        )
+
+    _assert_cost_contract(response)
+
+
+@pytest.mark.asyncio
+async def test_balance_responses_completion_uses_shared_cost_contract() -> None:
+    provider = _provider()
+    with patch(
+        "routstr.upstream.base.adjust_payment_for_tokens",
+        new=AsyncMock(return_value=dict(COST_DATA)),
+    ):
+        response = await provider.handle_non_streaming_responses_completion(
+            _upstream_response(
+                {
+                    "model": "test-model",
+                    "usage": {"input_tokens": 10, "output_tokens": 3},
+                }
+            ),
+            _key(),
+            _session(),
+            deducted_max_cost=10_000,
+        )
+
+    _assert_cost_contract(response)

tests/unit/test_lightning_settlement.py

@@ -149,6 +149,72 @@ async def test_invoice_mint_rejects_unrelated_concurrent_balance_growth() -> Non
         await _mint_invoice_quote(wallet, invoice)  # type: ignore[arg-type]
 
 
+@pytest.mark.asyncio
+async def test_quote_not_found_is_definitively_unpaid() -> None:
+    _invoice_settlement_locks.clear()
+    invoice = _invoice(status="pending", expires_at=0)
+    session = AsyncMock()
+    wallet = Mock(
+        get_mint_quote=AsyncMock(
+            side_effect=Exception("Mint Error: quote not found (Code: 0)")
+        )
+    )
+
+    with (
+        patch("routstr.lightning.get_wallet", AsyncMock(return_value=wallet)),
+        patch("routstr.lightning._reload_invoice_view", AsyncMock()),
+    ):
+        result = await check_invoice_payment(invoice, session)  # type: ignore[arg-type]
+
+    assert result is True
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "message",
+    [
+        "Mint Error: quote not found (Code: 10000)",
+        "Mint Error: quote not found (Code: 01)",
+        "Mint Error: quote not found (Code: 0x10)",
+    ],
+)
+async def test_quote_not_found_without_exact_code_0_is_not_definitively_unpaid(
+    message: str,
+) -> None:
+    _invoice_settlement_locks.clear()
+    invoice = _invoice(status="pending", expires_at=0)
+    session = AsyncMock()
+    wallet = Mock(get_mint_quote=AsyncMock(side_effect=Exception(message)))
+
+    with (
+        patch("routstr.lightning.get_wallet", AsyncMock(return_value=wallet)),
+        patch("routstr.lightning._reload_invoice_view", AsyncMock()),
+    ):
+        result = await check_invoice_payment(invoice, session)  # type: ignore[arg-type]
+
+    assert result is False
+
+
+@pytest.mark.asyncio
+async def test_quote_not_found_case_insensitive() -> None:
+    _invoice_settlement_locks.clear()
+    invoice = _invoice(status="pending", expires_at=0)
+    session = AsyncMock()
+    wallet = Mock(
+        get_mint_quote=AsyncMock(
+            side_effect=Exception("MINT ERROR: Quote Not Found (code 0)")
+        )
+    )
+
+    with (
+        patch("routstr.lightning.get_wallet", AsyncMock(return_value=wallet)),
+        patch("routstr.lightning._reload_invoice_view", AsyncMock()),
+    ):
+        result = await check_invoice_payment(invoice, session)  # type: ignore[arg-type]
+
+    assert result is True
+
+
 @pytest.mark.asyncio
 async def test_non_pending_invoice_is_not_minted() -> None:
     _invoice_settlement_locks.clear()
@@ -194,6 +260,36 @@ async def test_ambiguous_invoice_mint_timeout_remains_recoverable() -> None:
     session.commit.assert_awaited_once()
 
 
+@pytest.mark.asyncio
+async def test_quote_not_found_after_payment_confirmation_is_not_unpaid() -> None:
+    _invoice_settlement_locks.clear()
+    invoice = _invoice()
+    session = AsyncMock()
+    wallet = Mock(get_mint_quote=AsyncMock(return_value=Mock(paid=True)))
+    state_session = AsyncMock()
+    state_session.exec.return_value.rowcount = 1
+
+    @asynccontextmanager
+    async def owned_session() -> AsyncIterator[AsyncMock]:
+        yield state_session
+
+    with (
+        patch("routstr.lightning.get_wallet", AsyncMock(return_value=wallet)),
+        patch("routstr.lightning.create_session", owned_session),
+        patch(
+            "routstr.lightning._mint_invoice_quote",
+            AsyncMock(
+                side_effect=Exception("Mint Error: quote not found (Code: 0)")
+            ),
+        ),
+        patch("routstr.lightning._reload_invoice_view", AsyncMock()),
+    ):
+        result = await check_invoice_payment(invoice, session)  # type: ignore[arg-type]
+
+    assert result is False
+    assert invoice.status == "settlement_pending"
+
+
 @pytest.mark.asyncio
 async def test_quote_lookup_timeout_is_not_definitively_unpaid() -> None:
     _invoice_settlement_locks.clear()

tests/unit/test_messages_litellm_dispatch.py

@@ -451,7 +451,8 @@ async def test_non_streaming_dispatches_via_litellm_and_returns_anthropic_respon
     assert payload["model"] == "openai/gpt-4o-mini"  # mapped back to requested
     assert payload["usage"]["input_tokens"] == 5
     assert payload["usage"]["output_tokens"] == 3
-    assert payload["usage"]["cost"] == 0.0001
+    assert payload["usage"]["cost"]["total_msats"] == 1234
+    assert payload["usage"]["cost"]["total_usd"] == 0.0001
     assert payload["usage"]["cost_sats"] == 1
 
 
@@ -854,6 +855,9 @@ async def test_x_cashu_streaming_replays_events_and_sets_refund_header() -> None
 
     assert isinstance(result, StreamingResponse)
     assert result.headers.get("X-Cashu") == "cashuSTREAM"
+    assert result.headers.get("X-Routstr-Cost-Msats") == "1500000"
+    assert result.headers.get("X-Routstr-Input-Cost-Msats") == "1000000"
+    assert result.headers.get("X-Routstr-Output-Cost-Msats") == "500000"
     # 1_500_000 msats → 1500 sats. Refund = 5000 - 1500 = 3500.
     mock_refund.assert_awaited_once()
     refund_call = mock_refund.await_args
@@ -872,6 +876,9 @@ async def test_x_cashu_streaming_replays_events_and_sets_refund_header() -> None
     assert "event: message_start" in joined
     assert "event: message_delta" in joined
     assert "event: message_stop" in joined
+    assert '"total_msats": 1500000' in joined
+    assert '"input_msats": 1000000' in joined
+    assert '"output_msats": 500000' in joined
 
 
 # ---------------------------------------------------------------------------

tests/unit/test_wallet.py

@@ -4,7 +4,7 @@ import json
 import socket
 from collections.abc import AsyncIterator, Generator
 from contextlib import asynccontextmanager
-from unittest.mock import AsyncMock, Mock, patch
+from unittest.mock import AsyncMock, MagicMock, Mock, patch
 
 import httpx
 import pytest
@@ -12,13 +12,17 @@ from cashu.core.base import MeltQuoteState
 
 from routstr.core.db import ApiKey
 from routstr.wallet import (
+    Bolt11PaymentAmbiguous,
+    Bolt11PaymentNotAttempted,
     MintConnectionError,
     TokenConsumedError,
     _is_mint_rate_limited,
     classify_redemption_error,
     credit_balance,
+    execute_bolt11_payment,
     get_balance,
     is_mint_connection_error,
+    prepare_bolt11_payment,
     recieve_token,
     send,
     send_token,
@@ -239,9 +243,7 @@ async def test_recieve_token_uses_only_requested_destination_mint() -> None:
         )
 
     assert result == (99, "sat", destination)
-    swap.assert_awaited_once_with(
-        token, source_wallet, destination_mints=[destination]
-    )
+    swap.assert_awaited_once_with(token, source_wallet, destination_mints=[destination])
 
 
 @pytest.mark.asyncio
@@ -492,9 +494,7 @@ async def test_send_refreshes_reservations_inside_wallet_guard() -> None:
     ):
         assert await send(1000, "sat", mint) == (1000, "token")
 
-    wallet.set_reserved_for_send.assert_awaited_once_with(
-        [proof], reserved=True
-    )
+    wallet.set_reserved_for_send.assert_awaited_once_with([proof], reserved=True)
 
 
 @pytest.mark.asyncio
@@ -888,9 +888,7 @@ def _make_swap_mocks(
             quote=f"melt_quote_{invoice}", amount=invoice, fee_reserve=_next_fee()
         )
     )
-    mock_token_wallet.melt = AsyncMock(
-        return_value=Mock(state=MeltQuoteState.paid)
-    )
+    mock_token_wallet.melt = AsyncMock(return_value=Mock(state=MeltQuoteState.paid))
 
     return mock_token, mock_token_wallet, mock_primary_wallet
 
@@ -1804,6 +1802,220 @@ async def test_swap_melt_transport_error_is_never_reported_reusable() -> None:
     assert mock_token_wallet.melt.call_count == 1
 
 
+@pytest.mark.asyncio
+async def test_execute_bolt11_payment_rejects_unpaid_melt_state() -> None:
+    plan = MagicMock()
+    plan.proofs = [MagicMock(amount=110)]
+    plan.quote.amount = 100
+    plan.quote.fee_reserve = 10
+    plan.quote.quote = "quote-1"
+    plan.invoice = "lnbc-invoice"
+    plan.wallet.select_to_send = AsyncMock(return_value=(plan.proofs, 0))
+    plan.wallet.set_reserved_for_send = AsyncMock()
+    plan.wallet.melt = AsyncMock(return_value=MagicMock(state="UNPAID", change=[]))
+
+    with pytest.raises(Bolt11PaymentNotAttempted):
+        await execute_bolt11_payment(plan)
+
+    # An explicit unpaid answer means the proofs are ours again.
+    plan.wallet.set_reserved_for_send.assert_awaited_with(plan.proofs, reserved=False)
+
+
+@pytest.mark.asyncio
+async def test_execute_bolt11_payment_accepts_legacy_paid_response() -> None:
+    plan = MagicMock()
+    plan.proofs = [MagicMock(amount=110)]
+    plan.quote.amount = 100
+    plan.quote.fee_reserve = 10
+    plan.quote.quote = "quote-1"
+    plan.invoice = "lnbc-invoice"
+    plan.mint_url = "https://mint.test"
+    plan.unit = "sat"
+    plan.wallet.select_to_send = AsyncMock(return_value=(plan.proofs, 0))
+    plan.wallet.set_reserved_for_send = AsyncMock()
+    plan.wallet.melt = AsyncMock(
+        return_value=MagicMock(state=None, paid=True, change=[])
+    )
+
+    assert await execute_bolt11_payment(plan) == (
+        110,
+        "https://mint.test",
+        "sat",
+    )
+
+
+@pytest.mark.asyncio
+async def test_execute_bolt11_payment_keeps_proofs_reserved_when_melt_errors() -> None:
+    plan = MagicMock()
+    plan.proofs = [MagicMock(amount=110)]
+    plan.quote.amount = 100
+    plan.quote.fee_reserve = 10
+    plan.quote.quote = "quote-1"
+    plan.invoice = "lnbc-invoice"
+    plan.wallet.select_to_send = AsyncMock(return_value=(plan.proofs, 0))
+    plan.wallet.set_reserved_for_send = AsyncMock()
+    plan.wallet.set_reserved_for_melt = AsyncMock()
+    plan.wallet.melt = AsyncMock(side_effect=TimeoutError("no answer"))
+
+    with pytest.raises(Bolt11PaymentAmbiguous):
+        await execute_bolt11_payment(plan)
+
+    # The mint may still settle with these proofs. cashu's own melt()
+    # un-reserves them on a mint transport error, so the ambiguous path must
+    # re-reserve — and it must do so with the melt quote id, because
+    # get_melt_quote() finds the proofs to settle by melt_id.
+    plan.wallet.set_reserved_for_melt.assert_awaited_once_with(
+        plan.proofs, reserved=True, quote_id="quote-1"
+    )
+
+
+@pytest.mark.asyncio
+async def test_execute_bolt11_payment_does_not_reserve_when_selection_fails() -> None:
+    plan = MagicMock()
+    plan.proofs = [MagicMock(amount=110)]
+    plan.quote.amount = 100
+    plan.quote.fee_reserve = 10
+    plan.wallet.select_to_send = AsyncMock(side_effect=ValueError("insufficient"))
+    plan.wallet.set_reserved_for_send = AsyncMock()
+    plan.wallet.melt = AsyncMock()
+
+    with pytest.raises(Bolt11PaymentNotAttempted):
+        await execute_bolt11_payment(plan)
+
+    plan.wallet.set_reserved_for_send.assert_not_awaited()
+    plan.wallet.melt.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_prepare_bolt11_payment_counts_input_fees_in_sufficiency() -> None:
+    from routstr.core.settings import settings
+
+    wallet = MagicMock()
+    wallet.proofs = [MagicMock(amount=105)]
+    wallet.melt_quote = AsyncMock(
+        return_value=MagicMock(amount=100, fee_reserve=2, quote="quote-1")
+    )
+    # Balance covers amount + fee_reserve (102) but not the 5 sat input fee.
+    wallet.get_fees_for_proofs = Mock(return_value=5)
+
+    async def get_wallet(mint_url: str, unit: str = "sat", **_: object) -> MagicMock:
+        if unit == "msat":
+            raise ValueError("unit unsupported")
+        return wallet
+
+    with (
+        patch.object(settings, "cashu_mints", ["https://only.test"]),
+        patch.object(settings, "primary_mint", "https://only.test"),
+        patch("routstr.wallet.get_wallet", side_effect=get_wallet),
+        patch(
+            "routstr.wallet.get_proofs_per_mint_and_unit",
+            side_effect=lambda wallet, *args, **kwargs: wallet.proofs,
+        ),
+        patch(
+            "routstr.wallet.slow_filter_spend_proofs",
+            side_effect=lambda proofs, wallet: proofs,
+        ),
+        pytest.raises(ValueError, match="enough balance"),
+    ):
+        await prepare_bolt11_payment("lnbc-invoice")
+
+
+@pytest.mark.asyncio
+async def test_prepare_bolt11_payment_does_not_spend_user_liabilities() -> None:
+    from routstr.core.settings import settings
+
+    wallet = MagicMock()
+    wallet.proofs = [MagicMock(amount=500)]
+    wallet.melt_quote = AsyncMock(
+        return_value=MagicMock(amount=100, fee_reserve=2, quote="quote-1")
+    )
+    wallet.get_fees_for_proofs = Mock(return_value=0)
+
+    async def get_wallet(mint_url: str, unit: str = "sat", **_: object) -> MagicMock:
+        if unit == "msat":
+            raise ValueError("unit unsupported")
+        return wallet
+
+    with (
+        patch.object(settings, "cashu_mints", ["https://only.test"]),
+        patch.object(settings, "primary_mint", "https://only.test"),
+        patch("routstr.wallet.get_wallet", side_effect=get_wallet),
+        patch(
+            "routstr.wallet.get_proofs_per_mint_and_unit",
+            side_effect=lambda wallet, *args, **kwargs: wallet.proofs,
+        ),
+        patch(
+            "routstr.wallet.slow_filter_spend_proofs",
+            side_effect=lambda proofs, wallet: proofs,
+        ),
+        patch(
+            "routstr.wallet._owner_balance_for_mint_and_unit",
+            AsyncMock(return_value=90),
+        ),
+        pytest.raises(ValueError, match="user liabilities"),
+    ):
+        await prepare_bolt11_payment("lnbc-invoice")
+
+
+@pytest.mark.asyncio
+async def test_prepare_bolt11_payment_rounds_user_liability_up_to_whole_sats() -> None:
+    from routstr.core.settings import settings
+
+    wallet = MagicMock()
+    wallet.proofs = [MagicMock(amount=100)]
+    wallet.melt_quote = AsyncMock(
+        return_value=MagicMock(amount=1, fee_reserve=0, quote="quote-1")
+    )
+    wallet.get_fees_for_proofs = Mock(return_value=0)
+
+    async def get_wallet(mint_url: str, unit: str = "sat", **_: object) -> MagicMock:
+        if unit == "msat":
+            raise ValueError("unit unsupported")
+        return wallet
+
+    with (
+        patch.object(settings, "cashu_mints", ["https://only.test"]),
+        patch.object(settings, "primary_mint", "https://only.test"),
+        patch("routstr.wallet.get_wallet", side_effect=get_wallet),
+        patch(
+            "routstr.wallet.get_proofs_per_mint_and_unit",
+            side_effect=lambda wallet, *args, **kwargs: wallet.proofs,
+        ),
+        patch(
+            "routstr.wallet.slow_filter_spend_proofs",
+            side_effect=lambda proofs, wallet: proofs,
+        ),
+        patch(
+            "routstr.wallet.db.total_user_liability",
+            AsyncMock(return_value=99_999),
+        ),
+        pytest.raises(ValueError, match="user liabilities"),
+    ):
+        await prepare_bolt11_payment("lnbc-invoice")
+
+
+@pytest.mark.asyncio
+async def test_execute_bolt11_payment_rereserves_when_cancelled() -> None:
+    plan = MagicMock()
+    plan.proofs = [MagicMock(amount=110)]
+    plan.quote.amount = 100
+    plan.quote.fee_reserve = 10
+    plan.quote.quote = "quote-1"
+    plan.invoice = "lnbc-invoice"
+    plan.mint_url = "https://mint.test"
+    plan.wallet.select_to_send = AsyncMock(return_value=(plan.proofs, 0))
+    plan.wallet.set_reserved_for_send = AsyncMock()
+    plan.wallet.set_reserved_for_melt = AsyncMock()
+    plan.wallet.melt = AsyncMock(side_effect=asyncio.CancelledError())
+
+    with pytest.raises(asyncio.CancelledError):
+        await execute_bolt11_payment(plan)
+
+    plan.wallet.set_reserved_for_melt.assert_awaited_once_with(
+        plan.proofs, reserved=True, quote_id="quote-1"
+    )
+
+
 # ---------------------------------------------------------------------------
 # Per-mint adaptive guard + _mint_operation factory/retry
 # ---------------------------------------------------------------------------
@@ -2014,9 +2226,7 @@ async def test_default_timeout_allows_retry_after_rate_limit_cooldown() -> None:
     response = httpx.Response(429, request=request)
     operation = AsyncMock(
         side_effect=[
-            httpx.HTTPStatusError(
-                "rate limited", request=request, response=response
-            ),
+            httpx.HTTPStatusError("rate limited", request=request, response=response),
             "ok",
         ]
     )
@@ -2426,6 +2636,43 @@ async def test_wallet_fallback_on_429_no_in_place_retry() -> None:
     mock_sleep.assert_not_called()
 
 
+@pytest.mark.asyncio
+async def test_wallet_fallback_on_timeout_no_in_place_retry() -> None:
+    """A timeout from one destination must immediately try the next mint."""
+    from routstr.core.settings import settings
+    from routstr.wallet import _request_mint_with_fallback
+
+    primary = "http://primary:3338"
+    secondary = "http://secondary:3338"
+    primary_wallet = Mock(
+        request_mint=AsyncMock(side_effect=httpx.TimeoutException("timed out"))
+    )
+    quote = Mock(quote="q_secondary", request="lnbc1secondary")
+    secondary_wallet = Mock(request_mint=AsyncMock(return_value=quote))
+    wallets = {primary: primary_wallet, secondary: secondary_wallet}
+
+    with (
+        patch.object(settings, "primary_mint", primary),
+        patch.object(settings, "cashu_mints", [primary, secondary]),
+        patch.object(settings, "mint_retry_max_attempts", 3),
+        patch.object(settings, "mint_max_concurrency", 0),
+        patch.object(settings, "mint_operation_timeout_seconds", 0),
+        patch("routstr.mint.asyncio.sleep", AsyncMock()) as sleep,
+        patch(
+            "routstr.wallet.get_wallet",
+            AsyncMock(side_effect=lambda mint, *args, **kwargs: wallets[mint]),
+        ),
+    ):
+        _, mint_url, _ = await _request_mint_with_fallback(
+            1000, op_name="test_timeout_fallback"
+        )
+
+    assert mint_url == secondary
+    primary_wallet.request_mint.assert_awaited_once_with(1000)
+    secondary_wallet.request_mint.assert_awaited_once_with(1000)
+    sleep.assert_not_awaited()
+
+
 @pytest.mark.asyncio
 async def test_wallet_fallback_skips_mint_during_cooldown() -> None:
     from routstr.core.settings import settings

tests/unit/test_x_cashu_cost_sats.py

@@ -67,8 +67,13 @@ async def test_non_streaming_includes_cost_sats() -> None:
         )
 
     body = json.loads(response.body)
-    assert "cost_sats" in body["usage"]
     assert body["usage"]["cost_sats"] == 5  # 5000 msats // 1000
+    assert body["usage"]["cost"]["total_msats"] == 5000
+    assert body["usage"]["cost"]["input_msats"] == 3000
+    assert body["usage"]["cost"]["output_msats"] == 2000
+    assert response.headers["x-routstr-cost-msats"] == "5000"
+    assert response.headers["x-routstr-input-cost-msats"] == "3000"
+    assert response.headers["x-routstr-output-cost-msats"] == "2000"
 
 
 @pytest.mark.asyncio
@@ -96,7 +101,7 @@ async def test_non_streaming_cost_sats_value_rounds_down() -> None:
 
 
 @pytest.mark.asyncio
-async def test_non_streaming_preserves_existing_usage_fields() -> None:
+async def test_non_streaming_preserves_tokens_and_replaces_upstream_cost() -> None:
     provider = _make_provider()
     cost_data = _make_cost_data(total_msats=3000)
 
@@ -127,7 +132,8 @@ async def test_non_streaming_preserves_existing_usage_fields() -> None:
     assert usage["prompt_tokens"] == 100
     assert usage["completion_tokens"] == 50
     assert usage["total_tokens"] == 150
-    assert usage["cost"] == 0.00015
+    assert usage["cost"]["total_msats"] == 3000
+    assert usage["cost"]["total_usd"] == 0.00025
     assert usage["cost_sats"] == 3
 
 

ui/components/provider-card.tsx

@@ -1,3 +1,5 @@
+import { AdminService } from '@/lib/api/services/admin';
+import type { PPQAutoTopupState } from '@/lib/api/services/admin';
 import type {
   AdminModel,
   ProviderModels,
@@ -20,12 +22,16 @@ import {
   Trash2,
   Key,
   RotateCcw,
+  AlertTriangle,
+  Unlock,
+  Loader2,
 } from 'lucide-react';
 import { ProviderBalance } from '@/components/provider-balance';
 import { ProviderModelsPanel } from '@/components/provider-models-panel';
 import { RoutstrCreateKeySection } from '@/components/providers/RoutstrCreateKeySection';
 import { RoutstrProviderService } from '@/lib/api/services/routstr-provider';
-import { useMutation, useQueryClient } from '@tanstack/react-query';
+import { getErrorStatus } from '@/lib/api/client';
+import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query';
 import { useState } from 'react';
 import { toast } from 'sonner';
 import { cn } from '@/lib/utils';
@@ -36,6 +42,16 @@ import {
   DialogHeader,
   DialogTitle,
 } from '@/components/ui/dialog';
+import {
+  AlertDialog,
+  AlertDialogAction,
+  AlertDialogCancel,
+  AlertDialogContent,
+  AlertDialogDescription,
+  AlertDialogFooter,
+  AlertDialogHeader,
+  AlertDialogTitle,
+} from '@/components/ui/alert-dialog';
 
 interface ProviderCardProps {
   provider: UpstreamProvider;
@@ -77,8 +93,71 @@ export function ProviderCard({
 }: ProviderCardProps) {
   const queryClient = useQueryClient();
   const [isKeyModalOpen, setIsKeyModalOpen] = useState(false);
+  const [isReleaseDialogOpen, setIsReleaseDialogOpen] = useState(false);
+  // The claim as the query cache held it when the admin opened the dialog.
+  // The mutation sends this token rather than re-reading the query at submit
+  // time: a background refetch after the dialog opened must not swap in a
+  // state the admin never saw. The server rejects a stale token with a 409,
+  // which is the authoritative guard.
+  const [reviewedState, setReviewedState] = useState<PPQAutoTopupState | null>(
+    null
+  );
   const hasDetails = Boolean(provider.api_version) || isExpanded;
   const isRoutstr = provider.provider_type === 'routstr';
+  const isPPQ = provider.provider_type === 'ppqai';
+
+  const { data: ppqAutoTopupState, isError: ppqStateFetchFailed } = useQuery({
+    queryKey: ['ppq-auto-topup-state', provider.id],
+    queryFn: () => AdminService.getPPQAutoTopupState(provider.id),
+    enabled: isPPQ,
+    refetchInterval: 30000,
+  });
+
+  // A claim the server will not let us release: a worker is between reserving
+  // proofs and hearing back from the mint, and sweeping it would let the next
+  // cycle pay a second invoice.
+  const isPPQPaymentInFlight =
+    Boolean(ppqAutoTopupState?.active) &&
+    ppqAutoTopupState?.releasable === false;
+
+  const openReleaseDialog = () => {
+    setReviewedState(ppqAutoTopupState ?? null);
+    setIsReleaseDialogOpen(true);
+  };
+
+  const releasePPQMutation = useMutation({
+    mutationFn: () =>
+      AdminService.releasePPQAutoTopup(
+        provider.id,
+        reviewedState?.state_token ?? null
+      ),
+    onSuccess: () => {
+      queryClient.invalidateQueries({
+        queryKey: ['ppq-auto-topup-state', provider.id],
+      });
+      setIsReleaseDialogOpen(false);
+      setReviewedState(null);
+      toast.success('PPQ auto top-up claim released');
+    },
+    onError: (error: Error) => {
+      queryClient.invalidateQueries({
+        queryKey: ['ppq-auto-topup-state', provider.id],
+      });
+      if (getErrorStatus(error) === 409) {
+        // The claim changed since it was reviewed; the stale snapshot is
+        // useless, so force a fresh review.
+        setIsReleaseDialogOpen(false);
+        setReviewedState(null);
+        toast.error(
+          'PPQ claim changed since it was reviewed; reopen to see the new state'
+        );
+        return;
+      }
+      // Transient failure: keep the dialog and the reviewed snapshot so the
+      // admin can retry without re-navigating.
+      toast.error(`Failed to release PPQ claim: ${error.message}`);
+    },
+  });
 
   const refundMutation = useMutation({
     mutationFn: () => RoutstrProviderService.refundBalance(provider.id),
@@ -113,6 +192,35 @@ export function ProviderCard({
               >
                 {provider.enabled ? 'Enabled' : 'Disabled'}
               </Badge>
+              {ppqAutoTopupState?.active && (
+                <Badge
+                  variant='outline'
+                  className={cn(
+                    'w-fit gap-1',
+                    isPPQPaymentInFlight
+                      ? 'border-blue-500 text-blue-700 dark:text-blue-400'
+                      : 'border-amber-500 text-amber-700 dark:text-amber-400'
+                  )}
+                >
+                  {isPPQPaymentInFlight ? (
+                    <Loader2 className='h-3 w-3 animate-spin' />
+                  ) : (
+                    <AlertTriangle className='h-3 w-3' />
+                  )}
+                  {isPPQPaymentInFlight
+                    ? 'Paying invoice'
+                    : 'Auto top-up needs review'}
+                </Badge>
+              )}
+              {isPPQ && ppqStateFetchFailed && (
+                <Badge
+                  variant='outline'
+                  className='border-destructive text-destructive w-fit gap-1'
+                >
+                  <AlertTriangle className='h-3 w-3' />
+                  Top-up status unavailable
+                </Badge>
+              )}
             </div>
             <CardDescription className='break-all'>
               {provider.base_url}
@@ -153,6 +261,19 @@ export function ProviderCard({
               </Button>
             )}
 
+            {isPPQ && ppqAutoTopupState?.active && !isPPQPaymentInFlight && (
+              <Button
+                variant='outline'
+                size='sm'
+                onClick={openReleaseDialog}
+                className='justify-center gap-1.5 border-amber-500 text-amber-700 dark:text-amber-400'
+                title='Release only after manually verifying the Lightning payment outcome'
+              >
+                <Unlock className='h-4 w-4' />
+                <span>Release top-up</span>
+              </Button>
+            )}
+
             {isRoutstr && provider.api_key && (
               <Button
                 variant='outline'
@@ -212,6 +333,42 @@ export function ProviderCard({
         </div>
       </CardHeader>
 
+      <AlertDialog
+        open={isReleaseDialogOpen}
+        onOpenChange={setIsReleaseDialogOpen}
+      >
+        <AlertDialogContent>
+          <AlertDialogHeader>
+            <AlertDialogTitle>Release PPQ auto top-up?</AlertDialogTitle>
+            <AlertDialogDescription>
+              Only continue after checking PPQ and the Cashu mint and confirming
+              the previous Lightning payment cannot later settle. Releasing an
+              ambiguous payment can allow a duplicate top-up.
+              {reviewedState?.invoice_id
+                ? ` Invoice: ${reviewedState.invoice_id}`
+                : ''}
+            </AlertDialogDescription>
+          </AlertDialogHeader>
+          <AlertDialogFooter>
+            <AlertDialogCancel>Cancel</AlertDialogCancel>
+            <AlertDialogAction
+              onClick={(e) => {
+                // Radix closes the dialog on action click by default; the
+                // mutation handlers decide whether it closes (kept open on
+                // transient errors so the admin can retry).
+                e.preventDefault();
+                releasePPQMutation.mutate();
+              }}
+              disabled={releasePPQMutation.isPending}
+            >
+              {releasePPQMutation.isPending
+                ? 'Releasing...'
+                : 'I verified it is safe to retry'}
+            </AlertDialogAction>
+          </AlertDialogFooter>
+        </AlertDialogContent>
+      </AlertDialog>
+
       <Dialog open={isKeyModalOpen} onOpenChange={setIsKeyModalOpen}>
         <DialogContent className='max-h-[90dvh] overflow-y-auto sm:max-w-[500px]'>
           <DialogHeader>

ui/components/provider-form-dialog-content.tsx

@@ -12,6 +12,7 @@ import {
   DialogTitle,
 } from '@/components/ui/dialog';
 import { ProviderFormFields } from '@/components/provider-form-fields';
+import { ppqAutoTopupSettingsInvalid } from '@/components/providers/PPQAutoTopupSettings';
 
 interface ProviderFormDialogContentProps {
   mode: 'create' | 'edit';
@@ -52,6 +53,11 @@ export function ProviderFormDialogContent({
   isSubmitting,
   availableMints,
 }: ProviderFormDialogContentProps) {
+  // The server re-validates these bounds; this only stops submitting a form
+  // whose inline errors are already visible.
+  const hasInvalidSettings =
+    formData.provider_type === 'ppqai' &&
+    ppqAutoTopupSettingsInvalid(formData.provider_settings || {});
   return (
     <DialogContent className='max-h-[90dvh] overflow-y-auto sm:max-w-[500px]'>
       <DialogHeader>
@@ -80,7 +86,7 @@ export function ProviderFormDialogContent({
         </Button>
         <Button
           onClick={onSubmit}
-          disabled={isSubmitting}
+          disabled={isSubmitting || hasInvalidSettings}
           className='w-full sm:w-auto'
         >
           {isSubmitting ? submittingLabel : submitLabel}

ui/components/provider-form-fields.tsx

@@ -17,6 +17,7 @@ import {
   SelectValue,
 } from '@/components/ui/select';
 import { Switch } from '@/components/ui/switch';
+import { PPQAutoTopupSettings } from '@/components/providers/PPQAutoTopupSettings';
 import { RoutstrNodeSettings } from '@/components/providers/RoutstrNodeSettings';
 import { RoutstrCreateKeySection } from '@/components/providers/RoutstrCreateKeySection';
 
@@ -78,6 +79,8 @@ export function ProviderFormFields({
               provider_type: value,
               base_url: getDefaultBaseUrl(value),
               provider_fee: value === 'openrouter' ? 1.06 : 1.01,
+              provider_settings:
+                value === prev.provider_type ? prev.provider_settings : {},
             }));
           }}
         >
@@ -118,6 +121,19 @@ export function ProviderFormFields({
         />
       )}
 
+      {formData.provider_type === 'ppqai' && (
+        <PPQAutoTopupSettings
+          settings={formData.provider_settings || {}}
+          onSettingsChange={(settings) =>
+            setFormData((prev) => ({
+              ...prev,
+              provider_settings: settings,
+            }))
+          }
+          idPrefix={mode === 'edit' ? 'edit' : ''}
+        />
+      )}
+
       <div className='grid gap-2'>
         <Label htmlFor={`${idPrefix}slug`}>
           Slug {mode === 'create' ? '(optional, auto-generated)' : ''}

ui/components/providers/PPQAutoTopupSettings.tsx

@@ -0,0 +1,191 @@
+'use client';
+
+import { Input } from '@/components/ui/input';
+import { Label } from '@/components/ui/label';
+import { Switch } from '@/components/ui/switch';
+
+interface ProviderSettings {
+  auto_topup?: boolean;
+  topup_threshold?: number;
+  topup_amount_limit?: number;
+  [key: string]: unknown;
+}
+
+interface PPQAutoTopupSettingsProps {
+  settings: ProviderSettings;
+  onSettingsChange: (settings: ProviderSettings) => void;
+  idPrefix?: string;
+}
+
+/**
+ * Field-level validation shared with the dialog's submit gating. The server
+ * enforces the same bounds authoritatively; this only keeps a knowingly
+ * invalid form from being submitted.
+ *
+ * Validation only applies while auto top-up is enabled: a disabled toggle
+ * hides the fields, and stale out-of-range values behind it must not block
+ * submission invisibly. When enabled, both fields are required — a blank
+ * field would otherwise submit and fail server-side.
+ */
+export function ppqAutoTopupSettingsErrors(settings: ProviderSettings): {
+  thresholdError?: string;
+  amountError?: string;
+} {
+  if (!settings.auto_topup) {
+    return {};
+  }
+  const threshold = settings.topup_threshold;
+  const amount = settings.topup_amount_limit;
+  return {
+    thresholdError:
+      threshold === undefined
+        ? 'Required when auto top-up is enabled'
+        : threshold <= 0
+          ? 'Must be greater than 0'
+          : undefined,
+    amountError:
+      amount === undefined
+        ? 'Required when auto top-up is enabled'
+        : amount < 1 || amount > 500
+          ? 'Must be between 1 and 500 USD'
+          : undefined,
+  };
+}
+
+export function ppqAutoTopupSettingsInvalid(
+  settings: ProviderSettings
+): boolean {
+  const { thresholdError, amountError } = ppqAutoTopupSettingsErrors(settings);
+  return Boolean(thresholdError || amountError);
+}
+
+export function PPQAutoTopupSettings({
+  settings,
+  onSettingsChange,
+  idPrefix = '',
+}: PPQAutoTopupSettingsProps) {
+  const prefix = idPrefix ? `${idPrefix}_` : '';
+  const update = (patch: Partial<ProviderSettings>) =>
+    onSettingsChange({ ...settings, ...patch });
+
+  /**
+   * Clearing the field yields '' and parse* yields NaN, which JSON.stringify
+   * turns into null. Drop the key instead so the server rejects a missing
+   * value rather than storing a broken one.
+   */
+  const updateNumber = (
+    key: 'topup_threshold' | 'topup_amount_limit',
+    raw: string,
+    parse: (value: string) => number
+  ) => {
+    const next = { ...settings };
+    const parsed = parse(raw);
+    if (raw === '' || Number.isNaN(parsed)) {
+      delete next[key];
+    } else {
+      next[key] = parsed;
+    }
+    onSettingsChange(next);
+  };
+
+  const { thresholdError, amountError } = ppqAutoTopupSettingsErrors(settings);
+
+  return (
+    <div className='bg-muted/30 grid gap-4 rounded-lg border p-4'>
+      <Label className='text-sm font-semibold'>PPQ Auto Top-up</Label>
+
+      <div className='flex items-center justify-between'>
+        <Label htmlFor={`${prefix}ppq_auto_topup`} className='text-sm'>
+          Enable Auto Top-up
+        </Label>
+        <Switch
+          id={`${prefix}ppq_auto_topup`}
+          checked={!!settings.auto_topup}
+          onCheckedChange={(checked) => update({ auto_topup: checked })}
+        />
+      </div>
+
+      {settings.auto_topup && (
+        <div className='border-primary/20 grid gap-4 border-l-2 pt-2 pl-4'>
+          <div className='grid gap-2'>
+            <Label
+              htmlFor={`${prefix}ppq_topup_threshold`}
+              className='text-xs font-medium'
+            >
+              When credits are below (USD)
+            </Label>
+            <Input
+              id={`${prefix}ppq_topup_threshold`}
+              type='number'
+              min='0.01'
+              step='0.01'
+              className='h-9'
+              placeholder='e.g. 5'
+              value={settings.topup_threshold ?? ''}
+              aria-invalid={Boolean(thresholdError)}
+              aria-describedby={
+                thresholdError
+                  ? `${prefix}ppq_topup_threshold_error`
+                  : undefined
+              }
+              onChange={(e) =>
+                updateNumber('topup_threshold', e.target.value, parseFloat)
+              }
+            />
+            {thresholdError && (
+              <p
+                id={`${prefix}ppq_topup_threshold_error`}
+                className='text-destructive text-[10px]'
+              >
+                {thresholdError}
+              </p>
+            )}
+          </div>
+
+          <div className='grid gap-2'>
+            <Label
+              htmlFor={`${prefix}ppq_topup_amount_limit`}
+              className='text-xs font-medium'
+            >
+              Purchase this amount (USD)
+            </Label>
+            <Input
+              id={`${prefix}ppq_topup_amount_limit`}
+              type='number'
+              min='1'
+              max='500'
+              step='1'
+              className='h-9'
+              placeholder='e.g. 10'
+              value={settings.topup_amount_limit ?? ''}
+              aria-invalid={Boolean(amountError)}
+              aria-describedby={
+                amountError
+                  ? `${prefix}ppq_topup_amount_limit_error`
+                  : undefined
+              }
+              onChange={(e) =>
+                updateNumber('topup_amount_limit', e.target.value, (v) =>
+                  parseInt(v, 10)
+                )
+              }
+            />
+            {amountError && (
+              <p
+                id={`${prefix}ppq_topup_amount_limit_error`}
+                className='text-destructive text-[10px]'
+              >
+                {amountError}
+              </p>
+            )}
+          </div>
+
+          <p className='text-muted-foreground text-[10px]'>
+            Pays PPQ&apos;s Lightning invoice from the sufficiently funded Cashu
+            mint with the highest available balance.
+          </p>
+        </div>
+      )}
+    </div>
+  );
+}

ui/lib/api/client.ts

@@ -144,3 +144,18 @@ export class ApiError extends Error {
     this.data = data;
   }
 }
+
+/**
+ * HTTP status of a caught request error, whatever shape it arrived in.
+ * apiClient methods rethrow raw Axios errors, so callers must not rely on
+ * `instanceof ApiError` alone to read a status code.
+ */
+export function getErrorStatus(error: unknown): number | undefined {
+  if (error instanceof ApiError) {
+    return error.status;
+  }
+  if (axios.isAxiosError(error)) {
+    return error.response?.status;
+  }
+  return undefined;
+}

ui/lib/api/services/admin.ts

@@ -1013,6 +1013,32 @@ export class AdminService {
     }>(`/admin/api/upstream-providers/${providerId}/balance`);
   }
 
+  static async getPPQAutoTopupState(
+    providerId: number
+  ): Promise<PPQAutoTopupState> {
+    return await apiClient.get<PPQAutoTopupState>(
+      `/admin/api/upstream-providers/${providerId}/ppq-auto-topup`
+    );
+  }
+
+  /**
+   * `stateToken` must be the `state_token` snapshotted when the admin opened
+   * the confirmation — not re-read at submit time. The server rejects a
+   * release whose claim changed in any way since that snapshot.
+   */
+  static async releasePPQAutoTopup(
+    providerId: number,
+    stateToken: string | null
+  ): Promise<{
+    ok: boolean;
+    released: boolean;
+  }> {
+    return await apiClient.post<{ ok: boolean; released: boolean }>(
+      `/admin/api/upstream-providers/${providerId}/ppq-auto-topup/release`,
+      { confirmed_safe_to_retry: true, state_token: stateToken }
+    );
+  }
+
   // ── CLI Tokens ──
 
   static async listCliTokens(): Promise<CliTokenListItem[]> {
@@ -1244,6 +1270,30 @@ export interface TransactionsResponse {
   total: number;
 }
 
+export interface PPQAutoTopupState {
+  ok: boolean;
+  active: boolean;
+  /**
+   * Opaque version of the claim as reviewed. Echo it back verbatim to
+   * release; the server rejects a release whose claim changed in any way
+   * (new attempt, phase change, renewed lease) since this was read.
+   */
+  state_token?: string | null;
+  /** Identifies the attempt currently holding the claim. Informational. */
+  operation_id?: string | null;
+  /** 'claimed' | 'in_flight' | 'reconcile'. Null when the claim is malformed. */
+  phase?: string | null;
+  /** False while a payment is in flight — the server rejects a release then. */
+  releasable?: boolean;
+  expires_at?: number | null;
+  invoice_id?: string | null;
+  created_at?: number;
+  amount?: number;
+  unit?: string;
+  mint_url?: string | null;
+  malformed?: boolean;
+}
+
 export interface LightningInvoice {
   id: string;
   bolt11: string;