max cost filtered models

example.py

@@ -0,0 +1,37 @@
+import os
+
+import openai
+
+client = openai.OpenAI(
+    api_key=os.environ["CASHU_TOKEN"],
+    base_url=os.environ.get("ROUTSTR_API_URL", "https://api.routstr.com/v1"),
+    # base_url="http://roustrjfsdgfiueghsklchg.onion/v1",
+    # client=httpx.AsyncClient(
+    #     proxies={"http": "socks5://localhost:9050"},
+    # ),  # to use onion proxy (tor)
+)
+history: list = []
+
+
+def chat() -> None:
+    while True:
+        user_msg = {"role": "user", "content": input("\nYou: ")}
+        history.append(user_msg)
+        ai_msg = {"role": "assistant", "content": ""}
+
+        for chunk in client.chat.completions.create(
+            model=os.environ.get("MODEL", "openai/gpt-4o-mini"),
+            messages=history,
+            stream=True,
+        ):
+            if len(chunk.choices) > 0:
+                content = chunk.choices[0].delta.content
+                if content is not None:
+                    ai_msg["content"] += content
+                    print(content, end="", flush=True)
+        print()
+        history.append(ai_msg)
+
+
+if __name__ == "__main__":
+    chat()

examples/balance/check_balance.py

@@ -1,11 +0,0 @@
-import os
-
-import httpx
-
-# Use your Cashu token or API key as the Bearer token,
-# cashu token is hashed on the server and acts as an Temporary API key
-headers = {"Authorization": f"Bearer {os.environ.get('TOKEN')}"}
-base_url = os.environ.get("API_URL", "https://api.routstr.com/v1")
-
-resp = httpx.get(f"{base_url}/balance/info", headers=headers)
-print(resp.json())

examples/balance/create_balance.py

@@ -1,15 +0,0 @@
-import os
-
-import httpx
-
-# Send a Cashu token to the /create endpoint to get a persistent API key
-token = os.environ.get("TOKEN")
-if not token:
-    print("Please set TOKEN environment variable with a Cashu token")
-    exit(1)
-
-base_url = os.environ.get("API_URL", "https://api.routstr.com/v1")
-
-resp = httpx.get(f"{base_url}/balance/create", params={"initial_balance_token": token})
-
-print(resp.json())

examples/balance/refund_balance.py

@@ -1,12 +0,0 @@
-import os
-
-import httpx
-
-# Use your Cashu token or API key as the Bearer token
-headers = {"Authorization": f"Bearer {os.environ.get('TOKEN')}"}
-base_url = os.environ.get("API_URL", "https://api.routstr.com/v1")
-
-resp = httpx.post(f"{base_url}/balance/refund", headers=headers)
-
-print("Refund successful!")
-print(resp.json())

examples/balance/topup_balance.py

@@ -1,16 +0,0 @@
-import os
-
-import httpx
-
-# Use your Cashu token or API key as the Bearer token
-headers = {"Authorization": f"Bearer {os.environ.get('TOKEN')}"}
-base_url = os.environ.get("API_URL", "https://api.routstr.com/v1")
-
-# The Cashu token to top up with
-cashu_token = input("Enter Cashu token to top up: ")
-
-resp = httpx.post(
-    f"{base_url}/balance/topup", headers=headers, json={"cashu_token": cashu_token}
-)
-
-print(resp.json())

examples/chat_completions.py

@@ -1,15 +0,0 @@
-import os
-
-from openai import OpenAI
-
-client = OpenAI(
-    api_key=os.environ.get("TOKEN"),
-    base_url=os.environ.get("API_URL", "https://api.routstr.com/v1"),
-)
-
-response = client.chat.completions.create(
-    model=os.environ.get("MODEL", "gpt-5-nano"),
-    messages=[{"role": "user", "content": "Hello!"}],
-)
-
-print(response.choices[0].message.content)

examples/list_models.py

@@ -1,19 +0,0 @@
-import os
-
-import httpx
-from openai import OpenAI
-
-client = OpenAI(
-    api_key=os.environ.get("TOKEN", ""),
-    base_url=os.environ.get("API_URL", "https://api.routstr.com/v1"),
-)
-
-for model in client.models.list():
-    print(model.id)
-
-# OR
-
-models = httpx.get(
-    f"{client.base_url}/v1/models",
-    headers={"Authorization": f"Bearer {client.api_key}"},
-).json()

examples/responses/conversation.py

@@ -1,31 +0,0 @@
-import os
-
-from openai import OpenAI
-
-client = OpenAI(
-    api_key=os.environ.get("TOKEN"),
-    base_url=os.environ.get("API_URL", "https://api.routstr.com/v1"),
-)
-
-conversation = []  # type: ignore
-
-# First turn
-response1 = client.responses.create(  # type: ignore
-    model="o4-mini",
-    input="Hi, my name is Alice.",
-    conversation=conversation,
-)
-print("Response 1:", response1.output)
-
-# Note: The 'conversation' parameter might need to be constructed differently
-# depending on exact SDK/API spec. Typically, you pass back the previous turn's data.
-# Assuming the SDK manages or returns a conversation object/ID:
-# conversation.append(response1)
-
-# Second turn - demonstrating intent, actual implementation depends on strict API spec
-# response2 = client.responses.create(
-#     model="openai/gpt-4o-mini",
-#     input="What is my name?",
-#     conversation=conversation,
-# )
-# print("Response 2:", response2.output)

examples/responses/create.py

@@ -1,17 +0,0 @@
-import os
-
-from openai import OpenAI
-
-# The OpenAI SDK handles the 'responses' endpoint if it's updated to the latest version
-# and the base_url points to a compatible proxy like Routstr.
-client = OpenAI(
-    api_key=os.environ.get("TOKEN"),
-    base_url=os.environ.get("API_URL", "https://api.routstr.com/v1"),
-)
-
-response = client.responses.create(
-    model="gpt-5-mini",
-    input="Tell me a three sentence bedtime story about a unicorn.",
-)
-
-print(response.output)

examples/responses/streaming_response.py

@@ -1,20 +0,0 @@
-import os
-
-from openai import OpenAI
-
-client = OpenAI(
-    api_key=os.environ.get("TOKEN"),
-    base_url=os.environ.get("API_URL", "https://api.routstr.com/v1"),
-)
-
-stream = client.responses.create(
-    model="claude-4.5-sonnet",
-    input="Write a short poem about rust.",
-    stream=True,
-)
-
-for event in stream:
-    # Note: Depending on the SDK version and response structure,
-    # you might access event.output_delta or similar fields
-    print(event, end="", flush=True)
-print()

examples/responses/web_search.py

@@ -1,16 +0,0 @@
-import os
-
-from openai import OpenAI
-
-client = OpenAI(
-    api_key=os.environ.get("TOKEN"),
-    base_url=os.environ.get("API_URL", "https://api.routstr.com/v1"),
-)
-
-response = client.responses.create(
-    model="gpt-5-mini",
-    input="What is the latest news about AI?",
-    tools=[{"type": "web_search"}],  # type: ignore
-)
-
-print(response.output)

examples/streaming.py

@@ -1,28 +0,0 @@
-import os
-
-from openai import OpenAI
-
-client = OpenAI(
-    api_key=os.environ.get("TOKEN"),
-    base_url=os.environ.get("API_URL", "https://api.routstr.com/v1"),
-)
-
-messages = []
-while True:
-    messages.append({"role": "user", "content": input("\nYou: ")})
-
-    stream = client.chat.completions.create(
-        model=os.environ.get("MODEL", "gpt-5.1-mini"),
-        messages=messages,  # type: ignore
-        stream=True,
-    )
-
-    print("AI: ", end="")
-    response_content = ""
-    for chunk in stream:
-        if content := chunk.choices[0].delta.content:  # type: ignore
-            print(content, end="", flush=True)
-            response_content += content
-    print()
-
-    messages.append({"role": "assistant", "content": response_content})

examples/tor.py

@@ -1,20 +0,0 @@
-import os
-
-import httpx
-from openai import OpenAI
-
-# Requires `pip install "httpx[socks]"` and a running Tor proxy on port 9050
-client = OpenAI(
-    api_key=os.environ.get("TOKEN"),
-    base_url=os.environ.get("ONION_URL", "http://roustrjfsdgfiueghsklchg.onion/v1"),
-    http_client=httpx.Client(proxies="socks5://localhost:9050"),
-)
-
-print(
-    client.chat.completions.create(
-        model="openai/gpt-4o-mini",
-        messages=[{"role": "user", "content": "Hello from Tor!"}],
-    )
-    .choices[0]
-    .message.content
-)

pyproject.toml

@@ -73,7 +73,6 @@ packages = ["routstr"]
 [tool.ruff.lint]
 select = ["E", "F", "I"]
 ignore = ["E501"]
-exclude = ["examples"]
 
 [tool.mypy]
 python_version = "3.11"

routstr/payment/cost_calculation.py

@@ -5,7 +5,6 @@ from pydantic.v1 import BaseModel
 from ..core import get_logger
 from ..core.db import AsyncSession
 from ..core.settings import settings
-from .price import sats_usd_price
 
 logger = get_logger(__name__)
 
@@ -65,56 +64,6 @@ async def calculate_cost(  # todo: can be sync
         )
         return cost_data
 
-    usage_data = response_data["usage"]
-
-    usd_cost = 0.0
-
-    # Prioritize cost_details.upstream_inference_cost
-    if "cost_details" in usage_data:
-        usd_cost = float(
-            usage_data["cost_details"].get("upstream_inference_cost", 0) or 0
-        )
-
-    # Fallback to cost field if upstream_inference_cost is 0
-    if usd_cost == 0 and "cost" in usage_data:
-        try:
-            usd_cost = float(usage_data.get("cost", 0) or 0)
-        except Exception:
-            pass
-
-    if usd_cost > 0:
-        try:
-            sats_per_usd = 1.0 / sats_usd_price()
-            cost_in_sats = usd_cost * sats_per_usd
-            cost_in_msats = math.ceil(cost_in_sats * 1000)
-
-            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,
-                    "model": response_data.get("model", "unknown"),
-                },
-            )
-
-            return CostData(
-                base_msats=-1,
-                input_msats=-1,  # Cost field doesn't break down by token type
-                output_msats=-1,
-                total_msats=cost_in_msats,
-            )
-        except Exception as e:
-            logger.warning(
-                "Error calculating cost from usage data",
-                extra={
-                    "error": str(e),
-                    "usd_cost": usd_cost,
-                    "model": response_data.get("model", "unknown"),
-                },
-            )
-            # Fall through to token-based calculation
-
     MSATS_PER_1K_INPUT_TOKENS: float = (
         float(settings.fixed_per_1k_input_tokens) * 1000.0
     )
@@ -180,15 +129,19 @@ async def calculate_cost(  # todo: can be sync
         )
         return cost_data
 
-    input_tokens = usage_data.get("prompt_tokens", 0)
-    output_tokens = usage_data.get("completion_tokens", 0)
+    input_tokens = response_data.get("usage", {}).get("prompt_tokens", 0)
+    output_tokens = response_data.get("usage", {}).get("completion_tokens", 0)
 
     # added for response api
     input_tokens = (
-        input_tokens if input_tokens != 0 else usage_data.get("input_tokens", 0)
+        input_tokens
+        if input_tokens != 0
+        else response_data.get("usage", {}).get("input_tokens", 0)
     )
     output_tokens = (
-        output_tokens if output_tokens != 0 else usage_data.get("output_tokens", 0)
+        output_tokens
+        if output_tokens != 0
+        else response_data.get("usage", {}).get("output_tokens", 0)
     )
 
     input_msats = round(input_tokens / 1000 * MSATS_PER_1K_INPUT_TOKENS, 3)

routstr/payment/models.py

@@ -3,14 +3,15 @@ import json
 import random
 
 import httpx
-from fastapi import APIRouter, Depends
+from fastapi import APIRouter, Depends, Request
 from pydantic.v1 import BaseModel
 from sqlmodel import select
 from sqlmodel.ext.asyncio.session import AsyncSession
 
-from ..core.db import ModelRow, create_session, get_session
+from ..core.db import ApiKey, ModelRow, create_session, get_session
 from ..core.logging import get_logger
 from ..core.settings import settings
+from ..wallet import deserialize_token_from_string
 from .price import sats_usd_price
 
 logger = get_logger(__name__)
@@ -93,32 +94,12 @@ async def async_fetch_openrouter_models(source_filter: str | None = None) -> lis
 
     try:
         async with httpx.AsyncClient() as client:
-            models_response, embeddings_response = await asyncio.gather(
-                client.get(f"{base_url}/models", timeout=30),
-                client.get(f"{base_url}/embeddings/models", timeout=30),
-                return_exceptions=True,
-            )
-
-            def process_models_response(
-                response: httpx.Response | BaseException,
-            ) -> list[dict]:
-                if not isinstance(response, BaseException):
-                    response.raise_for_status()
-                    data = response.json()
-                    return [
-                        model
-                        for model in data.get("data", [])
-                        if ":free" not in model.get("id", "").lower()
-                    ]
-                return []
+            response = await client.get(f"{base_url}/models", timeout=30)
+            response.raise_for_status()
+            data = response.json()
 
             models_data: list[dict] = []
-            models_data.extend(process_models_response(models_response))
-            models_data.extend(process_models_response(embeddings_response))
-
-            # Apply source filter and exclusions
-            filtered_models = []
-            for model in models_data:
+            for model in data.get("data", []):
                 model_id = model.get("id", "")
 
                 if source_filter:
@@ -136,9 +117,9 @@ async def async_fetch_openrouter_models(source_filter: str | None = None) -> lis
                 if not _has_valid_pricing(model):
                     continue
 
-                filtered_models.append(model)
+                models_data.append(model)
 
-            return filtered_models
+            return models_data
     except Exception as e:
         logger.error(f"Error (async) fetching models from OpenRouter API: {e}")
         return []
@@ -541,11 +522,89 @@ def _pricing_matches(
     return True
 
 
+
+async def _get_request_balance(request: Request, session: AsyncSession) -> int | None:
+    """Get the balance from the request headers if authentication is provided."""
+    headers = request.headers
+    token: str | None = None
+
+    if x_cashu := headers.get("x-cashu"):
+        token = x_cashu
+    elif auth := headers.get("authorization"):
+        parts = auth.split(" ")
+        if len(parts) > 1:
+            token = parts[1]
+
+    if not token:
+        return None
+
+    # Handle API keys (sk-*)
+    if token.startswith("sk-"):
+        try:
+            # sk- keys use the part after "sk-" as the ID
+            key_id = token[3:]
+            key = await session.get(ApiKey, key_id)
+            if key:
+                return key.balance - key.reserved_balance
+        except Exception as e:
+            logger.warning(f"Error checking API key balance: {e}")
+        return None
+
+    # Handle Cashu tokens
+    try:
+        token_obj = deserialize_token_from_string(token)
+        amount_msat = (
+            token_obj.amount if token_obj.unit == "msat" else token_obj.amount * 1000
+        )
+        return amount_msat
+    except Exception as e:
+        logger.debug(f"Failed to deserialize cashu token for balance check: {e}")
+        return None
+
+
 @models_router.get("/v1/models")
 @models_router.get("/models", include_in_schema=False)
-async def models(session: AsyncSession = Depends(get_session)) -> dict:
+async def models(
+    request: Request, session: AsyncSession = Depends(get_session)
+) -> dict:
     """Get all available models from all providers with database overrides applied."""
     from ..proxy import get_unique_models
 
     items = get_unique_models()
+    
+    # Optional: Filter by user balance if authenticated
+    user_balance = await _get_request_balance(request, session)
+    if user_balance is not None:
+        filtered_items = []
+        
+        # Calculate tolerance factor once
+        tol_factor = 1.0
+        if not settings.fixed_pricing and settings.tolerance_percentage > 0:
+            tol_factor = 1.0 - (settings.tolerance_percentage / 100.0)
+            
+        fixed_cost = settings.fixed_cost_per_request * 1000
+        min_cost = settings.min_request_msat
+
+        for model in items:
+            model_cost_msats = 0
+            
+            if settings.fixed_pricing:
+                model_cost_msats = max(min_cost, fixed_cost)
+            elif model.sats_pricing:
+                # Use model specific pricing
+                max_cost = (
+                    model.sats_pricing.max_cost
+                    * 1000
+                    * tol_factor
+                )
+                model_cost_msats = max(min_cost, int(max_cost))
+            else:
+                # Fallback if no pricing found
+                model_cost_msats = max(min_cost, fixed_cost)
+            
+            if model_cost_msats <= user_balance:
+                filtered_items.append(model)
+        
+        items = filtered_items
+
     return {"data": items}

routstr/proxy.py

@@ -65,7 +65,7 @@ def get_upstreams() -> list[BaseUpstreamProvider]:
 
 def get_model_instance(model_id: str) -> Model | None:
     """Get Model instance by ID from global cache."""
-    return _model_instances.get(model_id.lower())
+    return _model_instances.get(model_id)
 
 
 def get_provider_for_model(model_id: str) -> BaseUpstreamProvider | None:
@@ -141,20 +141,14 @@ async def proxy(
             "unauthorized", "Unauthorized", 401, request=request
         )
 
-    logger.info(  # TODO: move to middleware, async
-        "Received proxy request",
-        extra={
-            "method": request.method,
-            "path": path,
-            "client_host": request.client.host if request.client else "unknown",
-            "user_agent": request.headers.get("user-agent", "unknown")[:100],
-        },
-    )
-
+    is_responses_api = path.startswith("v1/responses") or path.startswith("responses")
     request_body = await request.body()
     request_body_dict = parse_request_body_json(request_body, path)
 
-    model_id = request_body_dict.get("model", "unknown")
+    if is_responses_api:
+        model_id = extract_model_from_responses_api_request(request_body_dict)
+    else:
+        model_id = request_body_dict.get("model", "unknown")
 
     model_obj = get_model_instance(model_id)
     if not model_obj:
@@ -180,9 +174,14 @@ async def proxy(
     check_token_balance(headers, request_body_dict, max_cost_for_model)
 
     if x_cashu := headers.get("x-cashu", None):
-        return await upstream.handle_x_cashu(
-            request, x_cashu, path, max_cost_for_model, model_obj
-        )
+        if is_responses_api:
+            return await upstream.handle_x_cashu_responses_api(
+                request, x_cashu, path, max_cost_for_model, model_obj
+            )
+        else:
+            return await upstream.handle_x_cashu(
+                request, x_cashu, path, max_cost_for_model, model_obj
+            )
 
     elif auth := headers.get("authorization", None):
         key = await get_bearer_token_key(headers, path, session, auth)
@@ -197,28 +196,36 @@ async def proxy(
             )
 
         logger.debug("Processing unauthenticated GET request", extra={"path": path})
-        # TODO: why is this needed? can we remove it?
         headers = upstream.prepare_headers(dict(request.headers))
         return await upstream.forward_get_request(request, path, headers)
 
-    # Only pay for request if we have request body data (for completions endpoints)
     if request_body_dict:
         await pay_for_request(key, max_cost_for_model, session)
 
-    # Prepare headers for upstream
     headers = upstream.prepare_headers(dict(request.headers))
 
-    # Forward to upstream and handle response
-    response = await upstream.forward_request(
-        request,
-        path,
-        headers,
-        request_body,
-        key,
-        max_cost_for_model,
-        session,
-        model_obj,
-    )
+    if is_responses_api:
+        response = await upstream.forward_responses_api_request(
+            request,
+            path,
+            headers,
+            request_body,
+            key,
+            max_cost_for_model,
+            session,
+            model_obj,
+        )
+    else:
+        response = await upstream.forward_request(
+            request,
+            path,
+            headers,
+            request_body,
+            key,
+            max_cost_for_model,
+            session,
+            model_obj,
+        )
 
     if response.status_code != 200:
         await revert_pay_for_request(key, session, max_cost_for_model)
@@ -321,6 +328,24 @@ async def get_bearer_token_key(
         raise
 
 
+def extract_model_from_responses_api_request(request_body_dict: dict[str, Any]) -> str:
+    if model := request_body_dict.get("model"):
+        return model
+
+    if input_data := request_body_dict.get("input"):
+        if isinstance(input_data, dict) and (model := input_data.get("model")):
+            return model
+
+    if request_body_dict.get("messages"):
+        return "unknown"
+
+    logger.warning(
+        "No model found in Responses API request",
+        extra={"body_keys": list(request_body_dict.keys())},
+    )
+    return "unknown"
+
+
 def parse_request_body_json(request_body: bytes, path: str) -> dict[str, Any]:
     request_body_dict = {}
     if request_body:

routstr/upstream/handlers/cashu_payment_handler.py

@@ -0,0 +1,583 @@
+"""X-Cashu payment request handlers for different API types."""
+
+from __future__ import annotations
+
+import json
+from typing import TYPE_CHECKING, Callable
+
+import httpx
+from fastapi import BackgroundTasks, Request
+from fastapi.responses import Response, StreamingResponse
+
+from ...core import get_logger
+from ...wallet import send_token
+from ..payment.cashu_handler import (
+    CashuTokenError,
+    create_cashu_error_response,
+    validate_and_redeem_token,
+)
+from ..processing.http_client import HttpForwarder
+from ..processing.response_handler import ChatCompletionProcessor, ResponsesApiProcessor
+
+if TYPE_CHECKING:
+    from ...payment.models import Model
+
+logger = get_logger(__name__)
+
+
+class BaseCashuHandler:
+    """Base handler for X-Cashu payment requests."""
+
+    def __init__(self, base_url: str):
+        self.base_url = base_url
+        self.http_forwarder = HttpForwarder(base_url)
+
+    async def send_refund(self, amount: int, unit: str, mint: str | None = None) -> str:
+        """Create and send a refund token to the user."""
+        logger.debug(
+            "Creating refund token",
+            extra={"amount": amount, "unit": unit, "mint": mint},
+        )
+
+        max_retries = 3
+        last_exception = None
+
+        for attempt in range(max_retries):
+            try:
+                refund_token = await send_token(amount, unit=unit, mint_url=mint)
+
+                logger.info(
+                    "Refund token created successfully",
+                    extra={
+                        "amount": amount,
+                        "unit": unit,
+                        "mint": mint,
+                        "attempt": attempt + 1,
+                        "token_preview": refund_token[:20] + "..."
+                        if len(refund_token) > 20
+                        else refund_token,
+                    },
+                )
+
+                return refund_token
+            except Exception as e:
+                last_exception = e
+                if attempt < max_retries - 1:
+                    logger.warning(
+                        "Refund token creation failed, retrying",
+                        extra={
+                            "error": str(e),
+                            "error_type": type(e).__name__,
+                            "attempt": attempt + 1,
+                            "max_retries": max_retries,
+                            "amount": amount,
+                            "unit": unit,
+                            "mint": mint,
+                        },
+                    )
+                else:
+                    logger.error(
+                        "Failed to create refund token after all retries",
+                        extra={
+                            "error": str(e),
+                            "error_type": type(e).__name__,
+                            "attempt": attempt + 1,
+                            "max_retries": max_retries,
+                            "amount": amount,
+                            "unit": unit,
+                            "mint": mint,
+                        },
+                    )
+
+        raise Exception(
+            f"failed to create refund after {max_retries} attempts: {str(last_exception)}"
+        )
+
+    def _calculate_refund_amount(self, amount: int, unit: str, cost_msats: int) -> int:
+        """Calculate refund amount based on unit and cost."""
+        if unit == "msat":
+            return amount - cost_msats
+        elif unit == "sat":
+            return amount - (cost_msats + 999) // 1000
+        else:
+            raise ValueError(f"Invalid unit: {unit}")
+
+    async def _handle_upstream_error(
+        self,
+        response: httpx.Response,
+        amount: int,
+        unit: str,
+        mint: str | None,
+        api_type: str = "API",
+    ) -> Response:
+        """Handle upstream service errors with refund."""
+        logger.warning(
+            f"Upstream {api_type} request failed, processing refund",
+            extra={
+                "status_code": response.status_code,
+                "amount": amount,
+                "unit": unit,
+            },
+        )
+
+        refund_token = await self.send_refund(amount - 60, unit, mint)
+
+        logger.info(
+            f"Refund processed for failed upstream {api_type} request",
+            extra={
+                "status_code": response.status_code,
+                "refund_amount": amount,
+                "unit": unit,
+                "refund_token_preview": refund_token[:20] + "..."
+                if len(refund_token) > 20
+                else refund_token,
+            },
+        )
+
+        error_response = Response(
+            content=json.dumps(
+                {
+                    "error": {
+                        "message": f"Error forwarding {api_type} request to upstream",
+                        "type": "upstream_error",
+                        "code": response.status_code,
+                        "refund_token": refund_token,
+                    }
+                }
+            ),
+            status_code=response.status_code,
+            media_type="application/json",
+        )
+        error_response.headers["X-Cashu"] = refund_token
+        return error_response
+
+
+class ChatCompletionCashuHandler(BaseCashuHandler):
+    """Handler for X-Cashu paid chat completion requests."""
+
+    async def handle_request(
+        self,
+        request: Request,
+        x_cashu_token: str,
+        path: str,
+        headers: dict,
+        max_cost_for_model: int,
+        model_obj: Model,
+        prepare_request_body_func: Callable,
+        get_x_cashu_cost_func: Callable,
+        query_params_func: Callable,
+    ) -> Response | StreamingResponse:
+        """Handle chat completion request with X-Cashu payment."""
+        try:
+            amount, unit, mint = await validate_and_redeem_token(x_cashu_token, request)
+
+            # Forward request to upstream
+            request_body = await request.body()
+            response = await self.http_forwarder.forward_request(
+                request=request,
+                path=path,
+                headers=headers,
+                request_body=request_body,
+                query_params=query_params_func(path, request.query_params),
+                transform_body_func=prepare_request_body_func,
+                model_obj=model_obj,
+            )
+
+            # Handle upstream errors
+            if response.status_code != 200:
+                return await self._handle_upstream_error(
+                    response, amount, unit, mint, "chat completion"
+                )
+
+            # Process chat completion response
+            if path.endswith("chat/completions"):
+                result = await self._handle_chat_completion_response(
+                    response,
+                    amount,
+                    unit,
+                    max_cost_for_model,
+                    mint,
+                    get_x_cashu_cost_func,
+                )
+                background_tasks = BackgroundTasks()
+                background_tasks.add_task(response.aclose)
+                result.background = background_tasks
+                return result
+
+            # Default streaming response for other endpoints
+            return self._create_default_streaming_response(response)
+
+        except CashuTokenError as e:
+            return create_cashu_error_response(e, request, x_cashu_token)
+
+    async def _handle_chat_completion_response(
+        self,
+        response: httpx.Response,
+        amount: int,
+        unit: str,
+        max_cost_for_model: int,
+        mint: str | None,
+        get_x_cashu_cost_func: Callable,
+    ) -> Response | StreamingResponse:
+        """Handle chat completion response processing."""
+        content = await response.aread()
+        content_str = content.decode("utf-8") if isinstance(content, bytes) else content
+
+        is_streaming = ChatCompletionProcessor.is_streaming_response(content_str)
+
+        if is_streaming:
+            return await self._handle_streaming_response(
+                content_str,
+                response,
+                amount,
+                unit,
+                max_cost_for_model,
+                mint,
+                get_x_cashu_cost_func,
+            )
+        else:
+            return await self._handle_non_streaming_response(
+                content_str,
+                response,
+                amount,
+                unit,
+                max_cost_for_model,
+                mint,
+                get_x_cashu_cost_func,
+            )
+
+    async def _handle_streaming_response(
+        self,
+        content_str: str,
+        response: httpx.Response,
+        amount: int,
+        unit: str,
+        max_cost_for_model: int,
+        mint: str | None,
+        get_x_cashu_cost_func: Callable,
+    ) -> StreamingResponse:
+        """Handle streaming chat completion response with refund calculation."""
+        response_headers = ChatCompletionProcessor.clean_response_headers(
+            dict(response.headers)
+        )
+
+        usage_data, model = ChatCompletionProcessor.extract_usage_from_streaming(
+            content_str
+        )
+
+        if usage_data and model:
+            try:
+                response_data = {"usage": usage_data, "model": model}
+                cost_data = await get_x_cashu_cost_func(
+                    response_data, max_cost_for_model
+                )
+
+                if cost_data:
+                    refund_amount = self._calculate_refund_amount(
+                        amount, unit, cost_data.total_msats
+                    )
+
+                    if refund_amount > 0:
+                        refund_token = await self.send_refund(refund_amount, unit, mint)
+                        response_headers["X-Cashu"] = refund_token
+
+            except Exception as e:
+                logger.error(
+                    "Error calculating cost for streaming response",
+                    extra={"error": str(e), "error_type": type(e).__name__},
+                )
+
+        lines = content_str.strip().split("\n")
+        return StreamingResponse(
+            ChatCompletionProcessor.create_streaming_generator(lines),
+            status_code=response.status_code,
+            headers=response_headers,
+            media_type="text/plain",
+        )
+
+    async def _handle_non_streaming_response(
+        self,
+        content_str: str,
+        response: httpx.Response,
+        amount: int,
+        unit: str,
+        max_cost_for_model: int,
+        mint: str | None,
+        get_x_cashu_cost_func: Callable,
+    ) -> Response:
+        """Handle non-streaming chat completion response with refund calculation."""
+        response_headers = ChatCompletionProcessor.clean_response_headers(
+            dict(response.headers)
+        )
+
+        try:
+            response_json = json.loads(content_str)
+            cost_data = await get_x_cashu_cost_func(response_json, max_cost_for_model)
+
+            if cost_data:
+                refund_amount = self._calculate_refund_amount(
+                    amount, unit, cost_data.total_msats
+                )
+
+                if refund_amount > 0:
+                    refund_token = await self.send_refund(refund_amount, unit, mint)
+                    response_headers["X-Cashu"] = refund_token
+
+            return Response(
+                content=content_str,
+                status_code=response.status_code,
+                headers=response_headers,
+                media_type="application/json",
+            )
+
+        except json.JSONDecodeError:
+            # Emergency refund on parse error
+            emergency_refund = amount
+            refund_token = await send_token(emergency_refund, unit=unit, mint_url=mint)
+            response_headers["X-Cashu"] = refund_token
+
+            logger.warning(
+                "Emergency refund issued due to JSON parse error",
+                extra={
+                    "original_amount": amount,
+                    "refund_amount": emergency_refund,
+                    "deduction": 60,
+                },
+            )
+
+            return Response(
+                content=content_str,
+                status_code=response.status_code,
+                headers=response_headers,
+                media_type="application/json",
+            )
+
+    def _create_default_streaming_response(
+        self, response: httpx.Response
+    ) -> StreamingResponse:
+        """Create default streaming response for non-chat endpoints."""
+        background_tasks = BackgroundTasks()
+        background_tasks.add_task(response.aclose)
+
+        return StreamingResponse(
+            response.aiter_bytes(),
+            status_code=response.status_code,
+            headers=dict(response.headers),
+            background=background_tasks,
+        )
+
+
+class ResponsesApiCashuHandler(BaseCashuHandler):
+    """Handler for X-Cashu paid Responses API requests."""
+
+    async def handle_request(
+        self,
+        request: Request,
+        x_cashu_token: str,
+        path: str,
+        headers: dict,
+        max_cost_for_model: int,
+        model_obj: Model,
+        prepare_responses_api_request_body_func: Callable,
+        get_x_cashu_cost_func: Callable,
+        query_params_func: Callable,
+    ) -> Response | StreamingResponse:
+        """Handle Responses API request with X-Cashu payment."""
+        try:
+            amount, unit, mint = await validate_and_redeem_token(x_cashu_token, request)
+
+            # Forward request to upstream
+            request_body = await request.body()
+            response = await self.http_forwarder.forward_request(
+                request=request,
+                path=path,
+                headers=headers,
+                request_body=request_body,
+                query_params=query_params_func(path, request.query_params),
+                transform_body_func=prepare_responses_api_request_body_func,
+                model_obj=model_obj,
+            )
+
+            # Handle upstream errors
+            if response.status_code != 200:
+                return await self._handle_upstream_error(
+                    response, amount, unit, mint, "Responses API"
+                )
+
+            # Process Responses API response
+            if path.startswith("responses"):
+                result = await self._handle_responses_api_completion(
+                    response,
+                    amount,
+                    unit,
+                    max_cost_for_model,
+                    mint,
+                    get_x_cashu_cost_func,
+                )
+                background_tasks = BackgroundTasks()
+                background_tasks.add_task(response.aclose)
+                result.background = background_tasks
+                return result
+
+            # Default streaming response for other endpoints
+            return self._create_default_streaming_response(response)
+
+        except CashuTokenError as e:
+            return create_cashu_error_response(e, request, x_cashu_token)
+
+    async def _handle_responses_api_completion(
+        self,
+        response: httpx.Response,
+        amount: int,
+        unit: str,
+        max_cost_for_model: int,
+        mint: str | None,
+        get_x_cashu_cost_func: Callable,
+    ) -> Response | StreamingResponse:
+        """Handle Responses API completion response processing."""
+        content = await response.aread()
+        content_str = content.decode("utf-8") if isinstance(content, bytes) else content
+
+        is_streaming = ResponsesApiProcessor.is_streaming_response(content_str)
+
+        if is_streaming:
+            return await self._handle_streaming_responses_api_response(
+                content_str,
+                response,
+                amount,
+                unit,
+                max_cost_for_model,
+                mint,
+                get_x_cashu_cost_func,
+            )
+        else:
+            return await self._handle_non_streaming_responses_api_response(
+                content_str,
+                response,
+                amount,
+                unit,
+                max_cost_for_model,
+                mint,
+                get_x_cashu_cost_func,
+            )
+
+    async def _handle_streaming_responses_api_response(
+        self,
+        content_str: str,
+        response: httpx.Response,
+        amount: int,
+        unit: str,
+        max_cost_for_model: int,
+        mint: str | None,
+        get_x_cashu_cost_func: Callable,
+    ) -> StreamingResponse:
+        """Handle streaming Responses API response with refund calculation."""
+        response_headers = ResponsesApiProcessor.clean_response_headers(
+            dict(response.headers)
+        )
+
+        # Extract usage data (ignoring reasoning tokens as per requirement)
+        usage_data, model = ResponsesApiProcessor.extract_usage_with_reasoning_tokens(
+            content_str
+        )
+
+        if usage_data and model:
+            try:
+                response_data = {"usage": usage_data, "model": model}
+                cost_data = await get_x_cashu_cost_func(
+                    response_data, max_cost_for_model
+                )
+
+                if cost_data:
+                    refund_amount = self._calculate_refund_amount(
+                        amount, unit, cost_data.total_msats
+                    )
+
+                    if refund_amount > 0:
+                        refund_token = await self.send_refund(refund_amount, unit, mint)
+                        response_headers["X-Cashu"] = refund_token
+
+            except Exception as e:
+                logger.error(
+                    "Error calculating cost for streaming Responses API response",
+                    extra={"error": str(e), "error_type": type(e).__name__},
+                )
+
+        lines = content_str.strip().split("\n")
+        return StreamingResponse(
+            ResponsesApiProcessor.create_streaming_generator(lines),
+            status_code=response.status_code,
+            headers=response_headers,
+            media_type="text/plain",
+        )
+
+    async def _handle_non_streaming_responses_api_response(
+        self,
+        content_str: str,
+        response: httpx.Response,
+        amount: int,
+        unit: str,
+        max_cost_for_model: int,
+        mint: str | None,
+        get_x_cashu_cost_func: Callable,
+    ) -> Response:
+        """Handle non-streaming Responses API response with refund calculation."""
+        response_headers = ResponsesApiProcessor.clean_response_headers(
+            dict(response.headers)
+        )
+
+        try:
+            response_json = json.loads(content_str)
+            cost_data = await get_x_cashu_cost_func(response_json, max_cost_for_model)
+
+            if cost_data:
+                refund_amount = self._calculate_refund_amount(
+                    amount, unit, cost_data.total_msats
+                )
+
+                if refund_amount > 0:
+                    refund_token = await self.send_refund(refund_amount, unit, mint)
+                    response_headers["X-Cashu"] = refund_token
+
+            return Response(
+                content=content_str,
+                status_code=response.status_code,
+                headers=response_headers,
+                media_type="application/json",
+            )
+
+        except json.JSONDecodeError:
+            # Emergency refund on parse error
+            emergency_refund = amount
+            refund_token = await send_token(emergency_refund, unit=unit, mint_url=mint)
+            response_headers["X-Cashu"] = refund_token
+
+            logger.warning(
+                "Emergency refund issued for Responses API due to JSON parse error",
+                extra={
+                    "original_amount": amount,
+                    "refund_amount": emergency_refund,
+                    "deduction": 60,
+                },
+            )
+
+            return Response(
+                content=content_str,
+                status_code=response.status_code,
+                headers=response_headers,
+                media_type="application/json",
+            )
+
+    def _create_default_streaming_response(
+        self, response: httpx.Response
+    ) -> StreamingResponse:
+        """Create default streaming response for non-responses endpoints."""
+        background_tasks = BackgroundTasks()
+        background_tasks.add_task(response.aclose)
+
+        return StreamingResponse(
+            response.aiter_bytes(),
+            status_code=response.status_code,
+            headers=dict(response.headers),
+            background=background_tasks,
+        )

routstr/upstream/openrouter.py

@@ -50,13 +50,7 @@ class OpenRouterUpstreamProvider(BaseUpstreamProvider):
     async def fetch_models(self) -> list[Model]:
         """Fetch all OpenRouter models."""
         models_data = await async_fetch_openrouter_models()
-        models = [Model(**model) for model in models_data]  # type: ignore
-        # manual alias for openai/text-embedding-ada-002 due to openrouter api bug
-        for model in models:
-            if model.id == "openai/text-embedding-ada-002":
-                model.alias_ids = ["text-embedding-ada-002-v2"]
-                break
-        return models
+        return [Model(**model) for model in models_data]  # type: ignore
 
     async def get_balance(self) -> float | None:
         """Get the current account balance from OpenRouter.

routstr/upstream/payment/cashu_handler.py

@@ -0,0 +1,111 @@
+"""X-Cashu payment handling utilities."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from fastapi import Request
+from fastapi.responses import Response
+
+from ...core import get_logger
+from ...payment.helpers import create_error_response
+from ...wallet import recieve_token
+
+if TYPE_CHECKING:
+    pass
+
+logger = get_logger(__name__)
+
+
+class CashuTokenError(Exception):
+    """Exception raised for X-Cashu token processing errors."""
+
+    def __init__(self, error_type: str, message: str, status_code: int = 400):
+        self.error_type = error_type
+        self.message = message
+        self.status_code = status_code
+        super().__init__(message)
+
+
+async def validate_and_redeem_token(
+    x_cashu_token: str, request: Request
+) -> tuple[int, str, str | None]:
+    """Validate and redeem X-Cashu token.
+
+    Args:
+        x_cashu_token: The X-Cashu token to validate and redeem
+        request: Original FastAPI request (for error responses)
+
+    Returns:
+        Tuple of (amount, unit, mint_url)
+
+    Raises:
+        CashuTokenError: If token validation or redemption fails
+    """
+    logger.info(
+        "Processing X-Cashu token redemption",
+        extra={
+            "token_preview": x_cashu_token[:20] + "..."
+            if len(x_cashu_token) > 20
+            else x_cashu_token,
+        },
+    )
+
+    try:
+        amount, unit, mint = await recieve_token(x_cashu_token)
+        logger.info(
+            "X-Cashu token redeemed successfully",
+            extra={"amount": amount, "unit": unit, "mint": mint},
+        )
+        return amount, unit, mint
+    except Exception as e:
+        error_message = str(e)
+        logger.error(
+            "X-Cashu token redemption failed",
+            extra={
+                "error": error_message,
+                "error_type": type(e).__name__,
+            },
+        )
+
+        # Determine specific error type
+        if "already spent" in error_message.lower():
+            raise CashuTokenError(
+                "token_already_spent",
+                "The provided CASHU token has already been spent",
+                400,
+            )
+        elif "invalid token" in error_message.lower():
+            raise CashuTokenError(
+                "invalid_token", "The provided CASHU token is invalid", 400
+            )
+        elif "mint error" in error_message.lower():
+            raise CashuTokenError(
+                "mint_error", f"CASHU mint error: {error_message}", 422
+            )
+        else:
+            raise CashuTokenError(
+                "cashu_error", f"CASHU token processing failed: {error_message}", 400
+            )
+
+
+def create_cashu_error_response(
+    error: CashuTokenError, request: Request, x_cashu_token: str
+) -> Response:
+    """Create error response for X-Cashu token errors.
+
+    Args:
+        error: The CashuTokenError that occurred
+        request: Original FastAPI request
+        x_cashu_token: The token that caused the error
+
+    Returns:
+        Error response with appropriate status code and message
+    """
+    return create_error_response(
+        error.error_type,
+        error.message,
+        error.status_code,
+        request=request,
+        token=x_cashu_token,
+    )

routstr/upstream/processing/http_client.py

@@ -0,0 +1,200 @@
+"""HTTP client utilities for upstream requests."""
+
+from __future__ import annotations
+
+import traceback
+from typing import TYPE_CHECKING, Callable, Mapping
+
+import httpx
+from fastapi import BackgroundTasks, Request
+from fastapi.responses import StreamingResponse
+
+from ...core import get_logger
+
+if TYPE_CHECKING:
+    from ...payment.models import Model
+
+logger = get_logger(__name__)
+
+
+class HttpForwarder:
+    """Handles HTTP request forwarding to upstream services."""
+
+    def __init__(self, base_url: str):
+        self.base_url = base_url
+
+    def _prepare_path(self, path: str) -> str:
+        """Prepare path by removing v1/ prefix if present."""
+        if path.startswith("v1/"):
+            path = path.replace("v1/", "")
+        return path
+
+    async def forward_request(
+        self,
+        request: Request,
+        path: str,
+        headers: dict,
+        request_body: bytes | None,
+        query_params: Mapping[str, str] | None,
+        transform_body_func: Callable | None = None,
+        model_obj: Model | None = None,
+    ) -> httpx.Response:
+        """Forward HTTP request to upstream service.
+
+        Args:
+            request: Original FastAPI request
+            path: Request path
+            headers: Prepared headers for upstream
+            request_body: Request body bytes, if any
+            query_params: Query parameters for the request
+            transform_body_func: Optional function to transform request body
+            model_obj: Model object for request body transformation
+
+        Returns:
+            Response from upstream service
+
+        Raises:
+            httpx.RequestError: If request fails
+            Exception: For other unexpected errors
+        """
+        path = self._prepare_path(path)
+        url = f"{self.base_url}/{path}"
+
+        # Transform body if function provided
+        transformed_body = request_body
+        if transform_body_func and request_body and model_obj:
+            transformed_body = transform_body_func(request_body, model_obj)
+
+        logger.info(
+            "Forwarding request to upstream",
+            extra={
+                "url": url,
+                "method": request.method,
+                "path": path,
+                "has_request_body": request_body is not None,
+            },
+        )
+
+        client = httpx.AsyncClient(
+            transport=httpx.AsyncHTTPTransport(retries=1),
+            timeout=None,
+        )
+
+        try:
+            if transformed_body is not None:
+                response = await client.send(
+                    client.build_request(
+                        request.method,
+                        url,
+                        headers=headers,
+                        content=transformed_body,
+                        params=query_params,
+                    ),
+                    stream=True,
+                )
+            else:
+                response = await client.send(
+                    client.build_request(
+                        request.method,
+                        url,
+                        headers=headers,
+                        content=request.stream(),
+                        params=query_params,
+                    ),
+                    stream=True,
+                )
+
+            logger.info(
+                "Received upstream response",
+                extra={
+                    "status_code": response.status_code,
+                    "path": path,
+                    "content_type": response.headers.get("content-type", "unknown"),
+                },
+            )
+
+            return response
+
+        except httpx.RequestError as exc:
+            await client.aclose()
+            error_type = type(exc).__name__
+            error_details = str(exc)
+
+            logger.error(
+                "HTTP request error to upstream",
+                extra={
+                    "error_type": error_type,
+                    "error_details": error_details,
+                    "method": request.method,
+                    "url": url,
+                    "path": path,
+                    "query_params": dict(request.query_params),
+                },
+            )
+
+            if isinstance(exc, httpx.ConnectError):
+                error_message = "Unable to connect to upstream service"
+            elif isinstance(exc, httpx.TimeoutException):
+                error_message = "Upstream service request timed out"
+            elif isinstance(exc, httpx.NetworkError):
+                error_message = "Network error while connecting to upstream service"
+            else:
+                error_message = f"Error connecting to upstream service: {error_type}"
+
+            raise httpx.RequestError(error_message) from exc
+
+        except Exception as exc:
+            await client.aclose()
+            tb = traceback.format_exc()
+
+            logger.error(
+                "Unexpected error in upstream forwarding",
+                extra={
+                    "error": str(exc),
+                    "error_type": type(exc).__name__,
+                    "method": request.method,
+                    "url": url,
+                    "path": path,
+                    "query_params": dict(request.query_params),
+                    "traceback": tb,
+                },
+            )
+
+            raise
+
+
+class StreamingResponseWrapper:
+    """Wrapper for creating streaming responses with proper cleanup."""
+
+    @staticmethod
+    def create_streaming_response(
+        response: httpx.Response,
+        client: httpx.AsyncClient,
+    ) -> StreamingResponse:
+        """Create a streaming response with background cleanup tasks.
+
+        Args:
+            response: httpx response to stream
+            client: httpx client to clean up
+
+        Returns:
+            StreamingResponse with background cleanup
+        """
+        background_tasks = BackgroundTasks()
+        background_tasks.add_task(response.aclose)
+        background_tasks.add_task(client.aclose)
+
+        logger.debug(
+            "Creating streaming response",
+            extra={
+                "status_code": response.status_code,
+                "content_type": response.headers.get("content-type", "unknown"),
+            },
+        )
+
+        return StreamingResponse(
+            response.aiter_bytes(),
+            status_code=response.status_code,
+            headers=dict(response.headers),
+            background=background_tasks,
+        )

routstr/upstream/processing/response_handler.py

@@ -0,0 +1,170 @@
+"""Response processing utilities for different API types."""
+
+from __future__ import annotations
+
+import json
+import re
+from collections.abc import AsyncGenerator
+from typing import TYPE_CHECKING
+
+from ...core import get_logger
+
+if TYPE_CHECKING:
+    pass
+
+logger = get_logger(__name__)
+
+
+class ResponseProcessor:
+    """Processes responses from upstream services."""
+
+    @staticmethod
+    def is_streaming_response(content_str: str) -> bool:
+        """Determine if response content indicates streaming.
+
+        Args:
+            content_str: Response content as string
+
+        Returns:
+            True if response is streaming, False otherwise
+        """
+        return content_str.startswith("data:") or "data:" in content_str
+
+    @staticmethod
+    def extract_usage_from_streaming(
+        content_str: str,
+    ) -> tuple[dict | None, str | None]:
+        """Extract usage data and model from streaming response content.
+
+        Args:
+            content_str: Streaming response content as string
+
+        Returns:
+            Tuple of (usage_data, model) or (None, None) if not found
+        """
+        usage_data = None
+        model = None
+
+        lines = content_str.strip().split("\n")
+        for line in lines:
+            if line.startswith("data: "):
+                try:
+                    data_json = json.loads(line[6:])
+                    if "usage" in data_json:
+                        usage_data = data_json["usage"]
+                        model = data_json.get("model")
+                    elif "model" in data_json and not model:
+                        model = data_json["model"]
+                except json.JSONDecodeError:
+                    continue
+
+        return usage_data, model
+
+    @staticmethod
+    def clean_response_headers(headers: dict) -> dict:
+        """Clean response headers by removing encoding-related headers.
+
+        Args:
+            headers: Original response headers
+
+        Returns:
+            Cleaned headers dict
+        """
+        cleaned_headers = dict(headers)
+        headers_to_remove = ["transfer-encoding", "content-encoding", "content-length"]
+
+        for header in headers_to_remove:
+            cleaned_headers.pop(header, None)
+
+        return cleaned_headers
+
+    @staticmethod
+    async def create_streaming_generator(
+        lines: list[str],
+    ) -> AsyncGenerator[bytes, None]:
+        """Create async generator for streaming response lines.
+
+        Args:
+            lines: List of response lines to stream
+
+        Yields:
+            Encoded response lines
+        """
+        for line in lines:
+            yield (line + "\n").encode("utf-8")
+
+
+class ChatCompletionProcessor(ResponseProcessor):
+    """Processor specifically for chat completion responses."""
+
+    @staticmethod
+    def extract_model_from_chunks(stored_chunks: list[bytes]) -> str | None:
+        """Extract model name from stored response chunks.
+
+        Args:
+            stored_chunks: List of response chunks from streaming
+
+        Returns:
+            Model name if found, None otherwise
+        """
+        last_model_seen = None
+
+        for i in range(len(stored_chunks) - 1, -1, -1):
+            chunk = stored_chunks[i]
+            if not chunk:
+                continue
+            try:
+                events = re.split(b"data: ", chunk)
+                for event_data in events:
+                    if not event_data or event_data.strip() in (b"[DONE]", b""):
+                        continue
+                    try:
+                        data = json.loads(event_data)
+                        if isinstance(data, dict) and data.get("model"):
+                            return str(data.get("model"))
+                    except json.JSONDecodeError:
+                        continue
+            except Exception as e:
+                logger.debug(
+                    "Error processing chunk for model extraction",
+                    extra={"error": str(e), "error_type": type(e).__name__},
+                )
+
+        return last_model_seen
+
+
+class ResponsesApiProcessor(ResponseProcessor):
+    """Processor specifically for Responses API responses."""
+
+    @staticmethod
+    def extract_usage_with_reasoning_tokens(
+        content_str: str,
+    ) -> tuple[dict | None, str | None]:
+        """Extract usage data including reasoning tokens from Responses API content.
+
+        Args:
+            content_str: Streaming response content as string
+
+        Returns:
+            Tuple of (usage_data, model) with reasoning tokens preserved
+        """
+        usage_data = None
+        model = None
+
+        lines = content_str.strip().split("\n")
+        for line in lines:
+            if line.startswith("data: "):
+                try:
+                    data_json = json.loads(line[6:])
+                    if "usage" in data_json:
+                        usage_data = data_json["usage"]
+                        model = data_json.get("model")
+                        # Note: We now only track input_tokens and output_tokens
+                        # reasoning_tokens are ignored per the requirement
+                        break
+                    elif "model" in data_json and not model:
+                        model = data_json["model"]
+                except json.JSONDecodeError:
+                    continue
+
+        return usage_data, model

tests/integration/test_embeddings.py

@@ -1,97 +0,0 @@
-import json
-from typing import Any
-from unittest.mock import AsyncMock, MagicMock, patch
-
-import pytest
-from httpx import AsyncClient
-
-
-@pytest.mark.integration
-@pytest.mark.asyncio
-async def test_proxy_embeddings_endpoint(authenticated_client: AsyncClient) -> None:
-    """Test the embeddings endpoint proxy functionality"""
-
-    test_payload = {
-        "model": "text-embedding-ada-002",
-        "input": "The quick brown fox",
-    }
-
-    mock_response_data = {
-        "object": "list",
-        "data": [
-            {"object": "embedding", "embedding": [0.0023, -0.0012, 0.0045], "index": 0}
-        ],
-        "model": "text-embedding-ada-002",
-        "usage": {"prompt_tokens": 5, "total_tokens": 5},
-    }
-
-    with patch("httpx.AsyncClient.send") as mock_send:
-        # Create a proper async generator for iter_bytes
-        async def mock_iter_bytes(*args: Any, **kwargs: Any) -> Any:
-            yield json.dumps(mock_response_data).encode()
-
-        mock_response = AsyncMock()
-        mock_response.status_code = 200
-        mock_response.headers = {"content-type": "application/json"}
-        mock_response.text = json.dumps(mock_response_data)
-        # Use MagicMock for synchronous .json() method
-        mock_response.json = MagicMock(return_value=mock_response_data)
-        mock_response.iter_bytes = mock_iter_bytes
-        mock_response.aiter_bytes = mock_iter_bytes
-        mock_send.return_value = mock_response
-
-        # Make POST request to embeddings endpoint
-        response = await authenticated_client.post("/v1/embeddings", json=test_payload)
-
-        assert response.status_code == 200
-        response_data = response.json()
-        assert response_data["object"] == "list"
-        assert len(response_data["data"]) == 1
-        assert response_data["data"][0]["object"] == "embedding"
-
-        # Verify request was forwarded
-        mock_send.assert_called_once()
-        forwarded_request = mock_send.call_args[0][0]
-        # Verify the path ends with embeddings
-        # Note: forwarded path might be full URL
-        assert str(forwarded_request.url).endswith("embeddings")
-
-
-@pytest.mark.integration
-@pytest.mark.asyncio
-async def test_model_case_insensitivity(authenticated_client: AsyncClient) -> None:
-    """Test that model lookups are case insensitive"""
-
-    # We'll use a mixed-case model ID that should match the lowercase one in the system
-    # We assume 'gpt-3.5-turbo' is available in the mock env/database
-
-    test_payload = {
-        "model": "GPT-3.5-TURBO",
-        "messages": [{"role": "user", "content": "Hello"}],
-    }
-
-    with patch("httpx.AsyncClient.send") as mock_send:
-        mock_response_data = {
-            "id": "chatcmpl-123",
-            "object": "chat.completion",
-            "choices": [{"message": {"content": "Hi"}}],
-            "usage": {"total_tokens": 10},
-        }
-
-        async def mock_iter_bytes(*args: Any, **kwargs: Any) -> Any:
-            yield json.dumps(mock_response_data).encode()
-
-        mock_response = AsyncMock()
-        mock_response.status_code = 200
-        mock_response.headers = {"content-type": "application/json"}
-        mock_response.text = json.dumps(mock_response_data)
-        mock_response.json = MagicMock(return_value=mock_response_data)
-        mock_response.iter_bytes = mock_iter_bytes
-        mock_response.aiter_bytes = mock_iter_bytes
-        mock_send.return_value = mock_response
-
-        response = await authenticated_client.post(
-            "/v1/chat/completions", json=test_payload
-        )
-
-        assert response.status_code == 200

tests/integration/test_wallet_authentication.py

@@ -3,6 +3,7 @@ Integration tests for wallet authentication system including API key generation
 Tests POST /v1/wallet/topup endpoint and authorization header validation.
 """
 
+import hashlib
 from datetime import datetime, timedelta
 from typing import Any
 
@@ -14,6 +15,7 @@ from routstr.core.db import ApiKey
 
 from .utils import (
     CashuTokenGenerator,
+    ConcurrencyTester,
     ResponseValidator,
 )
 
@@ -387,6 +389,69 @@ async def test_api_key_with_expiry_time(
     # The expiry time and refund address functionality is tested elsewhere
 
 
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_concurrent_token_submissions(
+    integration_client: AsyncClient, testmint_wallet: Any, integration_session: Any
+) -> None:
+    """Test concurrent submissions of different tokens"""
+
+    # Generate multiple unique tokens with known amounts
+    num_tokens = 10
+    tokens = []
+    expected_balances = {}
+
+    for i in range(num_tokens):
+        amount = 100 + i * 10
+        token = await testmint_wallet.mint_tokens(amount)
+        tokens.append(token)
+        # Store expected balance by token hash
+        hashed_key = hashlib.sha256(token.encode()).hexdigest()
+        expected_balances[hashed_key] = amount * 1000  # msats
+
+    # Create concurrent requests
+    requests = [
+        {
+            "method": "GET",
+            "url": "/v1/wallet/info",
+            "headers": {"Authorization": f"Bearer {token}"},
+        }
+        for token in tokens
+    ]
+
+    # Execute concurrently
+    tester = ConcurrencyTester()
+    responses = await tester.run_concurrent_requests(
+        integration_client, requests, max_concurrent=5
+    )
+
+    # All should succeed
+    assert len(responses) == num_tokens
+    api_keys = set()
+
+    for response in responses:
+        assert response.status_code == 200
+        data = response.json()
+        api_key = data["api_key"]
+        api_keys.add(api_key)
+
+        # Verify balance matches the expected amount
+        hashed_key = api_key[3:]  # Remove "sk-" prefix
+        assert data["balance"] == expected_balances[hashed_key]
+
+    # Should have created unique API keys
+    assert len(api_keys) == num_tokens
+
+    # Verify all keys exist in database
+    for api_key in api_keys:
+        hashed_key = api_key[3:]  # Remove "sk-" prefix
+        result = await integration_session.execute(
+            select(ApiKey).where(ApiKey.hashed_key == hashed_key)  # type: ignore[arg-type]
+        )
+        db_key = result.scalar_one()
+        assert db_key.balance == expected_balances[hashed_key]
+
+
 @pytest.mark.integration
 @pytest.mark.asyncio
 async def test_authorization_with_cashu_token_directly(
@@ -439,6 +504,48 @@ async def test_x_cashu_header_support(
     assert response.status_code == 200
 
 
+@pytest.mark.integration
+@pytest.mark.asyncio
+@pytest.mark.slow
+async def test_api_key_consistency_under_load(
+    integration_client: AsyncClient, testmint_wallet: Any, integration_session: Any
+) -> None:
+    """Test API key generation consistency under concurrent load"""
+
+    # Generate a single token
+    token = await testmint_wallet.mint_tokens(1000)
+
+    # First request to create the API key
+    integration_client.headers["Authorization"] = f"Bearer {token}"
+    initial_response = await integration_client.get("/v1/wallet/info")
+    assert initial_response.status_code == 200
+    expected_api_key = initial_response.json()["api_key"]
+    expected_balance = initial_response.json()["balance"]
+
+    # Try to use the same token concurrently multiple times
+    # All should return the same API key since it's already created
+    requests = [
+        {
+            "method": "GET",
+            "url": "/v1/wallet/info",
+            "headers": {"Authorization": f"Bearer {token}"},
+        }
+        for _ in range(20)  # 20 concurrent attempts
+    ]
+
+    tester = ConcurrencyTester()
+    responses = await tester.run_concurrent_requests(
+        integration_client, requests, max_concurrent=10
+    )
+
+    # All should succeed and return the same API key
+    for response in responses:
+        assert response.status_code == 200
+        data = response.json()
+        assert data["api_key"] == expected_api_key
+        assert data["balance"] == expected_balance
+
+
 @pytest.mark.integration
 @pytest.mark.asyncio
 async def test_database_timestamp_accuracy(

tests/integration/test_wallet_information.py

@@ -13,7 +13,7 @@ from sqlmodel import select, update
 
 from routstr.core.db import ApiKey
 
-from .utils import ResponseValidator
+from .utils import ConcurrencyTester, ResponseValidator
 
 
 @pytest.mark.integration
@@ -204,6 +204,45 @@ async def test_expired_api_key_behavior(
     assert db_key.refund_address == "test@lightning.address"
 
 
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_concurrent_access_same_api_key(
+    integration_client: AsyncClient, authenticated_client: AsyncClient
+) -> None:
+    """Test concurrent access with the same API key"""
+
+    # Get the API key from authenticated client
+    response = await authenticated_client.get("/v1/wallet/")
+    api_key = response.json()["api_key"]
+    initial_balance = response.json()["balance"]
+
+    # Create multiple concurrent requests
+    requests = []
+    for i in range(20):
+        # Alternate between both endpoints
+        endpoint = "/v1/wallet/" if i % 2 == 0 else "/v1/wallet/info"
+        requests.append(
+            {
+                "method": "GET",
+                "url": endpoint,
+                "headers": {"Authorization": f"Bearer {api_key}"},
+            }
+        )
+
+    # Execute concurrently
+    tester = ConcurrencyTester()
+    responses = await tester.run_concurrent_requests(
+        integration_client, requests, max_concurrent=10
+    )
+
+    # All should succeed with consistent data
+    for response in responses:
+        assert response.status_code == 200
+        data = response.json()
+        assert data["api_key"] == api_key
+        assert data["balance"] == initial_balance
+
+
 @pytest.mark.integration
 @pytest.mark.asyncio
 async def test_wallet_info_data_consistency(

tests/integration/test_wallet_topup.py

@@ -15,6 +15,7 @@ from routstr.core.db import ApiKey
 
 from .utils import (
     CashuTokenGenerator,
+    ConcurrencyTester,
     ResponseValidator,
 )
 
@@ -283,6 +284,60 @@ async def test_transaction_history_tracking(  # type: ignore[no-untyped-def]
     assert response.status_code == 400
 
 
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_concurrent_topups_same_api_key(  # type: ignore[no-untyped-def]
+    integration_client: AsyncClient,
+    authenticated_client: AsyncClient,
+    testmint_wallet: Any,
+) -> None:
+    """Test concurrent top-ups to the same API key"""
+
+    # Get API key
+    response = await authenticated_client.get("/v1/wallet/")
+    api_key = response.json()["api_key"]
+    initial_balance = response.json()["balance"]
+
+    # Generate multiple unique tokens
+    num_tokens = 10
+    tokens = []
+    total_amount = 0
+
+    for i in range(num_tokens):
+        amount = 100 + i * 10  # Different amounts
+        token = await testmint_wallet.mint_tokens(amount)
+        tokens.append(token)
+        total_amount += amount
+
+    # Create concurrent top-up requests
+    requests = [
+        {
+            "method": "POST",
+            "url": "/v1/wallet/topup",
+            "params": {"cashu_token": token},
+            "headers": {"Authorization": f"Bearer {api_key}"},
+        }
+        for token in tokens
+    ]
+
+    # Execute concurrently
+    tester = ConcurrencyTester()
+    responses = await tester.run_concurrent_requests(
+        integration_client, requests, max_concurrent=5
+    )
+
+    # All should succeed
+    for response in responses:
+        assert response.status_code == 200
+        assert "msats" in response.json()
+
+    # Verify final balance is correct
+    final_response = await authenticated_client.get("/v1/wallet/")
+    final_balance = final_response.json()["balance"]
+    expected_balance = initial_balance + (total_amount * 1000)
+    assert final_balance == expected_balance
+
+
 @pytest.mark.integration
 @pytest.mark.asyncio
 async def test_topup_during_active_proxy_request(  # type: ignore[no-untyped-def]