fix: use direct assignment instead of setdefault in _inject_cost_into_usage

Per Claude Code review (PR #30 comment #3039): setdefault would silently
keep stale upstream-provided cost values and drop routstr's authoritative
msats breakdown. Switch to direct assignment so routstr's calculated cost
data always takes precedence.

routstr/payment/cost_calculation.py

@@ -481,12 +481,37 @@ def _calculate_from_usd_cost(
         )
         output_msats = cost_in_msats - input_msats
 
+    # Allocate cache read/creation msats proportionally from the input portion.
+    # Without this, cache_read_msats and cache_creation_msats are always 0 in
+    # the USD-cost path, and the SDK cannot report cache cost breakdowns.
+    cache_read_msats = 0
+    cache_creation_msats = 0
+    if cache_read_tokens > 0 or cache_creation_tokens > 0:
+        cache_tokens = cache_read_tokens + cache_creation_tokens
+        regular_input_tokens = input_tokens
+        total_input_tokens = regular_input_tokens + cache_tokens
+        if total_input_tokens > 0:
+            # Split input_msats into regular input, cache_read, cache_creation
+            cache_read_msats = (
+                int(input_msats * cache_read_tokens / total_input_tokens)
+                if cache_read_tokens > 0
+                else 0
+            )
+            cache_creation_msats = (
+                int(input_msats * cache_creation_tokens / total_input_tokens)
+                if cache_creation_tokens > 0
+                else 0
+            )
+            input_msats = input_msats - cache_read_msats - cache_creation_msats
+
     logger.info(
         "Using cost from usage data/details",
         extra={
             "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 +526,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

@@ -64,6 +64,56 @@ if typing.TYPE_CHECKING:
 logger = get_logger(__name__)
 
 
+def _inject_cost_response_headers(
+    headers: dict[str, str], cost_data: CostData | MaxCostData
+) -> 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(cost_data.total_msats)
+    headers["X-Routstr-Input-Cost-Msats"] = str(cost_data.input_msats)
+    headers["X-Routstr-Output-Cost-Msats"] = str(cost_data.output_msats)
+    if cost_data.total_usd:
+        headers["X-Routstr-Cost-Usd"] = str(cost_data.total_usd)
+
+
+def _inject_cost_into_usage(
+    response_json: dict, cost_data: CostData | MaxCostData
+) -> 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 = {
+        "base_msats": cost_data.base_msats,
+        "input_msats": cost_data.input_msats,
+        "output_msats": cost_data.output_msats,
+        "total_msats": cost_data.total_msats,
+        "cache_read_input_tokens": cost_data.cache_read_input_tokens,
+        "cache_creation_input_tokens": cost_data.cache_creation_input_tokens,
+        "cache_read_msats": cost_data.cache_read_msats,
+        "cache_creation_msats": cost_data.cache_creation_msats,
+    }
+    if cost_data.total_usd:
+        cost_obj["total_usd"] = cost_data.total_usd
+    usage["cost"] = cost_obj
+    usage["cost_sats"] = 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.
     """
@@ -282,9 +332,27 @@ class BaseUpstreamProvider:
 
         sats_cost = total_msats // 1000
 
+        # Build the cost object that the SDK's extractUsageFromResponseBody
+        # and extractUsageFromSSEJson expect: an object with total_msats,
+        # input_msats, output_msats, cache_read_msats, cache_creation_msats,
+        # etc.  Setting usage.cost to a plain float (total_usd) means the SDK
+        # cannot extract the msats breakdown — cache_read_msats and
+        # cache_creation_msats in particular are lost.
+        cost_obj = {
+            "base_msats": cost_dict.get("base_msats", 0),
+            "input_msats": cost_dict.get("input_msats", 0),
+            "output_msats": cost_dict.get("output_msats", 0),
+            "total_msats": total_msats,
+            "total_usd": total_usd,
+            "cache_read_input_tokens": cost_dict.get("cache_read_input_tokens", 0),
+            "cache_creation_input_tokens": cost_dict.get("cache_creation_input_tokens", 0),
+            "cache_read_msats": cost_dict.get("cache_read_msats", 0),
+            "cache_creation_msats": cost_dict.get("cache_creation_msats", 0),
+        }
+
         # Inject into top-level usage block (OpenAI/Anthropic style)
         if "usage" in response_json:
-            response_json["usage"]["cost"] = total_usd
+            response_json["usage"]["cost"] = cost_obj
             response_json["usage"]["cost_sats"] = sats_cost
             response_json["usage"]["remaining_balance_msats"] = key.balance
             self._fold_cache_into_input_tokens(response_json["usage"])
@@ -2060,6 +2128,21 @@ class BaseUpstreamProvider:
                 if k.lower() in allowed_headers
             }
 
+            # Inject cost breakdown headers so the SDK's
+            # extractUsageFromResponseHeaders can populate
+            # inputMsats/outputMsats/totalMsats for balance-mode requests.
+            if isinstance(cost_data, dict):
+                _cost_data_obj = CostData(
+                    base_msats=cost_data.get("base_msats", 0),
+                    input_msats=cost_data.get("input_msats", 0),
+                    output_msats=cost_data.get("output_msats", 0),
+                    total_msats=cost_data.get("total_msats", 0),
+                    total_usd=cost_data.get("total_usd", 0.0),
+                )
+            else:
+                _cost_data_obj = cost_data
+            _inject_cost_response_headers(response_headers, _cost_data_obj)
+
             return Response(
                 content=json.dumps(response_json).encode(),
                 status_code=response.status_code,
@@ -2148,9 +2231,24 @@ class BaseUpstreamProvider:
         )
         self.inject_cost_metadata(response_json, cost_data, key)
 
+        # Inject cost breakdown headers for balance-mode requests.
+        if isinstance(cost_data, dict):
+            _cost_data_obj = CostData(
+                base_msats=cost_data.get("base_msats", 0),
+                input_msats=cost_data.get("input_msats", 0),
+                output_msats=cost_data.get("output_msats", 0),
+                total_msats=cost_data.get("total_msats", 0),
+                total_usd=cost_data.get("total_usd", 0.0),
+            )
+        else:
+            _cost_data_obj = cost_data
+        response_headers: dict[str, str] = {}
+        _inject_cost_response_headers(response_headers, _cost_data_obj)
+
         return Response(
             content=json.dumps(response_json).encode(),
             status_code=200,
+            headers=response_headers,
             media_type="application/json",
         )
 
@@ -2199,11 +2297,12 @@ class BaseUpstreamProvider:
         if cost_data 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
             )
@@ -3551,6 +3650,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",
@@ -3578,9 +3682,7 @@ class BaseUpstreamProvider:
                         and "usage" in data_json
                         and data_json["usage"]
                     ):
-                        data_json["usage"]["cost_sats"] = (
-                            cost_data.total_msats // 1000
-                        )
+                        _inject_cost_into_usage(data_json, cost_data)
                         changed = True
                     if changed:
                         lines[i] = "data: " + json.dumps(data_json)
@@ -3634,7 +3736,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(
@@ -3665,6 +3770,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":
@@ -4538,6 +4645,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",
@@ -4565,9 +4677,7 @@ class BaseUpstreamProvider:
                         and "usage" in data_json
                         and data_json["usage"]
                     ):
-                        data_json["usage"]["cost_sats"] = (
-                            cost_data.total_msats // 1000
-                        )
+                        _inject_cost_into_usage(data_json, cost_data)
                         changed = True
                     if changed:
                         lines[i] = "data: " + json.dumps(data_json)
@@ -4610,7 +4720,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(
@@ -4641,6 +4751,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":

tests/unit/test_coverage_admin.py

@@ -1,7 +1,7 @@
 """Coverage tests for admin.py (currently 35%).
 
 Tests admin endpoints that are testable without full app setup:
-withdraw validation, authentication guards, and slug validation.
+withdraw validation, password update, CLI token lifecycle.
 """
 
 from unittest.mock import Mock, patch
@@ -64,6 +64,43 @@ async def test_withdraw_rejects_insufficient_balance() -> None:
         assert "Insufficient" in str(exc_info.value.detail)
 
 
+# ===========================================================================
+# update_password — validation
+# ===========================================================================
+
+@pytest.mark.asyncio
+async def test_update_password_rejects_empty_new() -> None:
+    """update_password rejects empty new password."""
+    from routstr.core.admin import PasswordUpdate, update_password
+
+    request = Request(scope={"type": "http", "method": "POST"})
+
+    with pytest.raises(HTTPException) as exc_info:
+        await update_password(
+            request,
+            PasswordUpdate(current_password="old", new_password=""),
+        )
+
+    # Returns 500 (no admin password configured) or 400 (validation)
+    assert exc_info.value.status_code in (400, 500, 422)
+
+
+@pytest.mark.asyncio
+async def test_update_password_rejects_short_new() -> None:
+    """update_password rejects short passwords."""
+    from routstr.core.admin import PasswordUpdate, update_password
+
+    request = Request(scope={"type": "http", "method": "POST"})
+
+    with pytest.raises(HTTPException) as exc_info:
+        await update_password(
+            request,
+            PasswordUpdate(current_password="old", new_password="ab"),
+        )
+
+    assert exc_info.value.status_code in (400, 500, 422)
+
+
 # ===========================================================================
 # require_admin_api guard
 # ===========================================================================

tests/unit/test_coverage_base2.py

@@ -75,14 +75,16 @@ def test_extract_error_simple_error_string_not_parsed() -> None:
 async def test_on_upstream_error_redirect_noop() -> None:
     """Default implementation is a no-op for non-redirect statuses."""
     p = BaseUpstreamProvider("https://api.test.com", "sk-test")
-    await p.on_upstream_error_redirect(402, "Insufficient balance")
+    result = await p.on_upstream_error_redirect(402, "Insufficient balance")
+    assert result is None
 
 
 @pytest.mark.asyncio
 async def test_on_upstream_error_redirect_429() -> None:
     """429 rate limit passes through (subclasses may override)."""
     p = BaseUpstreamProvider("https://api.test.com", "sk-test")
-    await p.on_upstream_error_redirect(429, "Rate limited")
+    result = await p.on_upstream_error_redirect(429, "Rate limited")
+    assert result is None
 
 
 # ===========================================================================

tests/unit/test_coverage_payment_helpers.py

@@ -37,7 +37,7 @@ async def test_check_token_balance_no_x_cashu_raises() -> None:
 
     from routstr.payment.helpers import check_token_balance
 
-    headers: dict[str, str] = {}
+    headers = {}
     body = {"model": "gpt-4"}
 
     with pytest.raises(HTTPException) as exc_info:

tests/unit/test_db_and_payout_resilience.py

@@ -0,0 +1,89 @@
+"""Tests asserting CORRECT behavior for DB persistence and payout safety.
+
+RED tests — FAIL against current main until bugs are fixed.
+"""
+
+import inspect
+
+# ===========================================================================
+# RED TESTS: Fee payout crash safety
+# ===========================================================================
+
+def test_fee_payout_pre_reset_or_lock_exists() -> None:
+    """FIX REQUIRED: pay-then-reset must become lock-then-pay-then-unlock.
+
+    wallet.py:1076-1080 currently: raw_send_to_lnurl() THEN reset_routstr_fee().
+    A crash between these lines causes double payment.
+
+    Fix: set a lock flag BEFORE paying, clear it AFTER resetting.
+    On startup, reconcile any locked-but-not-reset payouts.
+    """
+    from routstr import wallet
+
+    source = inspect.getsource(wallet.periodic_routstr_fee_payout)
+
+    pay_pos = source.find("raw_send_to_lnurl")
+    reset_pos = source.find("reset_routstr_fee")
+
+    assert pay_pos > 0 and reset_pos > 0, "Pay and reset both exist"
+
+    # After fix: lock/safeguard must exist BEFORE the pay call
+    pre_pay_section = source[:pay_pos]
+    has_pre_guard = any(
+        kw in pre_pay_section.lower()
+        for kw in ["lock", "payout_state", "is_paying", "in_progress",
+                    "pre_reset", "reconcile", "checkpoint"]
+    )
+
+    assert has_pre_guard, (
+        "FIX REQUIRED: Fee payout pays before resetting with no crash guard. "
+        "A crash between pay and reset causes double payment. "
+        "Fix: add a DB lock/payout_state flag before paying."
+    )
+
+
+# ===========================================================================
+# RED TESTS: DB store resilience
+# ===========================================================================
+
+def test_retry_wrapper_exists() -> None:
+    """FIX REQUIRED: A retry wrapper for critical DB writes must exist."""
+    from routstr.core import db
+
+    assert hasattr(db, "store_cashu_transaction_with_retry"), (
+        "FIX REQUIRED: No retry wrapper exists for critical money-path "
+        "DB writes. Was merged (#600) then reverted (#604). Must be "
+        "reinstated with CRITICAL logging on final failure."
+    )
+
+
+# ===========================================================================
+# Wallet caching mechanism (informational — not a bug on main)
+# ===========================================================================
+
+def test_wallet_cache_uses_global_dict() -> None:
+    """get_wallet uses a global _wallets dict — verify mechanism."""
+    from routstr import wallet
+
+    source = inspect.getsource(wallet.get_wallet)
+    assert "_wallets" in source
+    assert "load_mint" in source
+    assert "load_proofs" in source
+
+
+# ===========================================================================
+# Mint rate limiter setting (informational)
+# ===========================================================================
+
+def test_mint_concurrency_setting_exists_or_documents_gap() -> None:
+    """If mint_max_concurrency exists, it must NOT be 0.
+
+    0 disables 429 cooldown tracking on the PR #597 branch.
+    """
+    from routstr.core.settings import settings
+
+    concurrency = getattr(settings, "mint_max_concurrency", None)
+    if concurrency is not None:
+        assert concurrency > 0, (
+            f"mint_max_concurrency = {concurrency}. 0 disables 429 cooldown."
+        )

tests/unit/test_emergency_refund_integrity.py

@@ -0,0 +1,215 @@
+"""Tests asserting CORRECT behavior for emergency refund and DB persistence.
+
+These tests FAIL against current main because the code is buggy.
+They serve as the "RED" phase of TDD — once the bugs are fixed, they go green.
+
+Correct behavior required:
+1. store_cashu_transaction should raise on failure (not silently return False)
+2. Emergency refund paths must NOT use try/except/pass for DB stores
+3. A retry wrapper must exist for critical money-path DB writes
+"""
+
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+# ===========================================================================
+# RED TESTS: store_cashu_transaction should RAISE on failure
+# ===========================================================================
+
+@pytest.mark.asyncio
+async def test_store_cashu_raises_on_db_failure_not_returns_false() -> None:
+    """FIX REQUIRED: store_cashu_transaction must raise on DB failure.
+
+    Currently returns False silently — callers never detect the failure.
+    Correct behavior: raise an exception so callers can recover.
+    """
+    from routstr.core.db import store_cashu_transaction
+
+    with patch("routstr.core.db.create_session") as mock_create:
+        mock_session = AsyncMock()
+        mock_session.commit = AsyncMock(side_effect=OSError("disk full"))
+        mock_session.__aenter__ = AsyncMock(return_value=mock_session)
+        mock_session.__aexit__ = AsyncMock(return_value=None)
+        mock_create.return_value = mock_session
+
+        with pytest.raises(Exception) as exc_info:
+            await store_cashu_transaction(
+                token="cashuAtest_refund_token",
+                amount=1000,
+                unit="sat",
+                mint_url="http://mint:3338",
+                typ="out",
+                request_id="req-123",
+            )
+
+        # Must raise a meaningful exception, not silently return False
+        # OSError or a custom DB error is acceptable
+        assert "disk full" in str(exc_info.value) or isinstance(
+            exc_info.value, (OSError, RuntimeError)
+        ), (
+            f"Expected store to propagate the failure, got {type(exc_info.value).__name__}: "
+            f"{exc_info.value}"
+        )
+
+
+@pytest.mark.asyncio
+async def test_store_cashu_raises_on_any_error() -> None:
+    """FIX REQUIRED: All DB errors must propagate, not just OSError."""
+    from routstr.core.db import store_cashu_transaction
+
+    errors = [
+        OSError("disk full"),
+        RuntimeError("connection lost"),
+        ConnectionRefusedError("db down"),
+    ]
+
+    for error in errors:
+        with patch("routstr.core.db.create_session") as mock_create:
+            mock_session = AsyncMock()
+            mock_session.commit = AsyncMock(side_effect=error)
+            mock_session.__aenter__ = AsyncMock(return_value=mock_session)
+            mock_session.__aexit__ = AsyncMock(return_value=None)
+            mock_create.return_value = mock_session
+
+            with pytest.raises(Exception):
+                await store_cashu_transaction(
+                    token="cashuAtest",
+                    amount=1000,
+                    unit="sat",
+                    typ="out",
+                )
+
+
+# ===========================================================================
+# RED TESTS: Retry wrapper must exist
+# ===========================================================================
+
+def test_retry_wrapper_exists_for_critical_writes() -> None:
+    """FIX REQUIRED: store_cashu_transaction_with_retry must exist.
+
+    Currently reverted (#600 → #604). All critical money-path DB writes
+    (after minting a token) need retry with backoff + CRITICAL logging.
+    """
+    from routstr.core import db
+
+    assert hasattr(db, "store_cashu_transaction_with_retry"), (
+        "FIX REQUIRED: store_cashu_transaction_with_retry does not exist. "
+        "Was merged in PR #600, reverted in PR #604. "
+        "All post-mint DB writes need retry + backoff + CRITICAL logging."
+    )
+
+
+@pytest.mark.asyncio
+async def test_retry_wrapper_retries_on_transient_failure() -> None:
+    """FIX REQUIRED: retry wrapper must retry, not fail on first attempt."""
+    from routstr.core import db
+
+    # Skip if the retry wrapper doesn't exist yet
+    if not hasattr(db, "store_cashu_transaction_with_retry"):
+        pytest.skip("store_cashu_transaction_with_retry does not exist yet")
+
+    with patch("routstr.core.db.store_cashu_transaction") as mock_store:
+        mock_store = AsyncMock()
+        mock_store.side_effect = [OSError("transient"), None]  # 1st fails, 2nd succeeds
+        # We'd test that the wrapper retries, but it doesn't exist yet
+        # This test documents the expected behavior
+
+
+# ===========================================================================
+# RED TESTS: Emergency refund must not silently lose tokens
+# ===========================================================================
+
+def test_emergency_refund_no_try_except_pass() -> None:
+    """FIX REQUIRED: Emergency refund paths must NOT use try/except/pass.
+
+    base.py:3643-3653 (chat) and base.py:4607-4617 (responses) both use
+    try/except/pass around store_cashu_transaction after minting a refund
+    token. If DB write fails, the token is permanently lost.
+
+    The fix: remove try/except/pass. Let the exception propagate so
+    the caller can detect failure and at minimum log the token.
+    """
+    import inspect
+
+    from routstr.upstream.base import BaseUpstreamProvider
+
+    # Check chat emergency refund handler
+    chat_src = inspect.getsource(
+        BaseUpstreamProvider.handle_x_cashu_non_streaming_response
+    )
+
+    # Find the emergency refund section
+    emergency_start = chat_src.find("emergency_refund = amount")
+    assert emergency_start > 0, "Emergency refund path exists"
+
+    emergency_section = chat_src[emergency_start : emergency_start + 500]
+
+    # The try/except/pass around store_cashu_transaction must NOT exist
+    has_except_pass = "except Exception:" in emergency_section and "pass" in emergency_section
+
+    assert not has_except_pass, (
+        "FIX REQUIRED: Emergency refund (chat) uses try/except/pass around "
+        "store_cashu_transaction. A failed DB write silently loses the minted "
+        "token. Fix: let the exception propagate or log at CRITICAL with the "
+        "full token for manual recovery."
+    )
+
+
+def test_emergency_refund_responses_api_no_silent_failure() -> None:
+    """FIX REQUIRED: Responses API emergency refund same fix as chat."""
+    import inspect
+
+    from routstr.upstream.base import BaseUpstreamProvider
+
+    responses_src = inspect.getsource(
+        BaseUpstreamProvider.handle_x_cashu_non_streaming_responses_response
+    )
+
+    has_emergency = "emergency_refund = amount" in responses_src
+    if has_emergency:
+        emergency_start = responses_src.find("emergency_refund = amount")
+        emergency_section = responses_src[emergency_start : emergency_start + 500]
+        has_except_pass = (
+            "except Exception:" in emergency_section and "pass" in emergency_section
+        )
+        assert not has_except_pass, (
+            "FIX REQUIRED: Responses API emergency refund also uses "
+            "try/except/pass. Same fund-loss vulnerability as chat path."
+        )
+
+
+# ===========================================================================
+# RED TESTS: Fee payout crash safety
+# ===========================================================================
+
+def test_fee_payout_has_crash_guard() -> None:
+    """FIX REQUIRED: Fee payout must have guard against double-pay on crash.
+
+    wallet.py:1076-1080 pays LNURL THEN resets the fee counter.
+    A crash between these steps causes double payment on restart.
+
+    Fix options:
+    1. Pre-reset the counter before paying (if pay fails, restore it)
+    2. Add a "payout_lock" DB flag that's set before pay and cleared after
+    3. Record payout in DB and reconcile on startup
+    """
+    import inspect
+
+    from routstr import wallet
+
+    source = inspect.getsource(wallet.periodic_routstr_fee_payout)
+
+    # After the fix, the pay-then-reset pattern should be replaced
+    # with a safe sequence. Verify the guard exists.
+    has_guard = any(
+        kw in source.lower()
+        for kw in ["payout_lock", "is_paying", "payout_in_progress",
+                    "pre_reset", "reset_before", "reconcile"]
+    )
+
+    assert has_guard, (
+        "FIX REQUIRED: Fee payout has no crash guard. Pay-then-reset "
+        "pattern in periodic_routstr_fee_payout can double-pay on "
+        "process restart."
+    )

tests/unit/test_wallet_money_paths.py

@@ -113,10 +113,7 @@ async def test_get_balance_returns_integer() -> None:
     mock_wallet.load_mint = AsyncMock()
     mock_wallet.load_proofs = AsyncMock()
 
-    with (
-        patch("routstr.wallet._wallets", {}),
-        patch("routstr.wallet.Wallet.with_db", return_value=mock_wallet),
-    ):
+    with patch("routstr.wallet.Wallet.with_db", return_value=mock_wallet):
         balance = await get_balance("sat")
         assert isinstance(balance, int)
         assert balance == 50000

tests/unit/test_zero_cost_fallback.py

@@ -0,0 +1,171 @@
+"""Tests asserting CORRECT behavior for the streaming billing fallback.
+
+These tests FAIL against current main because the zero-cost fallback at
+base.py:1012-1030 gives users free service on billing errors.
+
+Correct behavior required:
+1. Billing errors must NOT hardcode total_msats=0 — free service is theft
+2. Reserved balance must be released when billing fails
+3. Error must be logged at CRITICAL level, not just logger.exception
+4. The except clause must be narrow, not catch-all Exception
+"""
+
+import inspect
+
+# ===========================================================================
+# RED TESTS: No hardcoded zero-cost on billing error
+# ===========================================================================
+
+def test_billing_error_must_not_hardcode_zero_cost() -> None:
+    """FIX REQUIRED: billing errors must not result in zero-cost billing.
+
+    base.py:1012-1030 substitutes total_msats=0, total_usd=0.0 when
+    adjust_payment_for_tokens raises ANY exception. This means:
+    - User gets free inference
+    - Reserved balance is never released
+    - Operator has no idea money was lost
+    """
+    from routstr.upstream.base import BaseUpstreamProvider
+
+    source = inspect.getsource(
+        BaseUpstreamProvider.handle_streaming_chat_completion
+    )
+
+    fallback_start = source.find("Error during usage finalization")
+    assert fallback_start > 0, (
+        "Fallback block exists — it must be removed or fixed"
+    )
+
+    fallback_section = source[fallback_start : fallback_start + 600]
+
+    has_zero_msats = '"total_msats": 0' in fallback_section
+    has_zero_usd = '"total_usd": 0.0' in fallback_section
+
+    assert not has_zero_msats, (
+        "FIX REQUIRED: total_msats is hardcoded to 0 on billing error. "
+        "User gets free service. Fix: propagate the error as a 500 response "
+        "with the token refunded to the user."
+    )
+    assert not has_zero_usd, (
+        "FIX REQUIRED: total_usd is hardcoded to 0.0. No billing occurs. "
+        "Fix: propagate the error."
+    )
+
+
+def test_billing_error_must_release_reserved_balance() -> None:
+    """FIX REQUIRED: billing errors must release the reserved balance.
+
+    When adjust_payment_for_tokens fails, the reserved balance on the
+    API key must be released. Currently it's stuck forever.
+    """
+    from routstr.upstream.base import BaseUpstreamProvider
+
+    source = inspect.getsource(
+        BaseUpstreamProvider.handle_streaming_chat_completion
+    )
+
+    fallback_start = source.find("Error during usage finalization")
+    assert fallback_start > 0, "Fallback exists"
+
+    fallback_section = source[fallback_start : fallback_start + 600]
+
+    has_release = any(
+        kw in fallback_section
+        for kw in ["reserved_balance", "release_reservation", "adjust_reserved",
+                    "reset_reserved", "clear_reserved"]
+    )
+
+    assert has_release, (
+        "FIX REQUIRED: Zero-cost fallback does NOT release the reserved "
+        "balance. Funds are permanently stuck. Fix: add reserved_balance "
+        "release in the error path."
+    )
+
+
+def test_billing_error_catch_is_too_broad() -> None:
+    """FIX REQUIRED: except clause must not catch all Exception types.
+
+    `except Exception as e:` catches transient DB errors, logic bugs,
+    and serialization failures — all resulting in free service.
+    The catch should be specific (e.g., TemporaryDBError) or the error
+    should propagate as a 500.
+    """
+    from routstr.upstream.base import BaseUpstreamProvider
+
+    source = inspect.getsource(
+        BaseUpstreamProvider.handle_streaming_chat_completion
+    )
+
+    fallback_start = source.find("Error during usage finalization")
+    # Look at the except clause above the fallback
+    pre_fallback = source[max(0, fallback_start - 250) : fallback_start]
+
+    assert "except Exception" not in pre_fallback, (
+        "FIX REQUIRED: The except clause catches all Exception types. "
+        "A transient DB hiccup results in free inference. "
+        "Fix: narrow the exception type or propagate the error."
+    )
+
+
+def test_billing_error_must_log_critical() -> None:
+    """FIX REQUIRED: billing failure must log at CRITICAL level.
+
+    Currently uses logger.exception() which is ERROR level.
+    A billing failure means the operator is losing money — this must
+    be CRITICAL so monitoring/monitoring systems catch it.
+    """
+    from routstr.upstream.base import BaseUpstreamProvider
+
+    source = inspect.getsource(
+        BaseUpstreamProvider.handle_streaming_chat_completion
+    )
+
+    fallback_start = source.find("Error during usage finalization")
+    fallback_section = source[fallback_start : fallback_start + 600]
+
+    has_critical = "CRITICAL" in fallback_section or "critical" in fallback_section
+
+    assert has_critical, (
+        "FIX REQUIRED: Billing error is logged at ERROR level. "
+        "Money is being lost — this must be CRITICAL so operators "
+        "get alerted."
+    )
+
+
+# ===========================================================================
+# RED TESTS: Messages streaming billing
+# ===========================================================================
+
+def test_messages_streaming_no_silent_billing_failure() -> None:
+    """FIX REQUIRED: messages streaming must not silently swallow billing errors.
+
+    handle_streaming_messages_completion uses `except Exception: pass`
+    for the finalize path, silently dropping the billing attachment.
+    After the fix, this catch block must either:
+    - Log at CRITICAL level with the error details
+    - Propagate the error to surface an HTTP 500
+    - Release reserved balance and refund the token
+    """
+    from routstr.upstream.base import BaseUpstreamProvider
+
+    source = inspect.getsource(
+        BaseUpstreamProvider.handle_streaming_messages_completion
+    )
+
+    # After fix: the silent pass in finalize_without_usage must be replaced
+    # The fix must include at least one of: CRITICAL logging, error propagation,
+    # or balance release in the error path.
+
+    # The silent pass must NOT exist around billing finalization
+    silent_pass_exists = False
+    for segment in source.split("except Exception:"):
+        if "adjust_payment_for_tokens" in segment:
+            if "pass" in segment[:150]:
+                silent_pass_exists = True
+                break
+
+    assert not silent_pass_exists, (
+        "FIX REQUIRED: finalize_without_usage in messages streaming "
+        "silently swallows billing errors with `except Exception: pass`. "
+        "User gets unbilled inference with no log record."
+    )