undo 43e9732

routstr/payment/cost_calculation.py

@@ -70,7 +70,6 @@ async def calculate_cost(  # todo: can be sync
     MSATS_PER_1K_OUTPUT_TOKENS: float = (
         float(settings.fixed_per_1k_output_tokens) * 1000.0
     )
-    MSATS_PER_1K_IMAGE_COMPLETION_TOKENS: float = 0.0
 
     if not settings.fixed_pricing:
         response_model = response_data.get("model", "")
@@ -105,13 +104,11 @@ async def calculate_cost(  # todo: can be sync
         try:
             mspp = float(model_obj.sats_pricing.prompt)
             mspc = float(model_obj.sats_pricing.completion)
-            mspci = float(getattr(model_obj.sats_pricing, "completion_image", 0.0))
         except Exception:
             return CostDataError(message="Invalid pricing data", code="pricing_invalid")
 
         MSATS_PER_1K_INPUT_TOKENS = mspp * 1_000_000.0
         MSATS_PER_1K_OUTPUT_TOKENS = mspc * 1_000_000.0
-        MSATS_PER_1K_IMAGE_COMPLETION_TOKENS = mspci * 1_000_000.0
 
         logger.info(
             "Applied model-specific pricing",
@@ -119,7 +116,6 @@ async def calculate_cost(  # todo: can be sync
                 "model": response_model,
                 "input_price_msats_per_1k": MSATS_PER_1K_INPUT_TOKENS,
                 "output_price_msats_per_1k": MSATS_PER_1K_OUTPUT_TOKENS,
-                "image_completion_price_msats_per_1k": MSATS_PER_1K_IMAGE_COMPLETION_TOKENS,
             },
         )
 
@@ -132,44 +128,13 @@ async def calculate_cost(  # todo: can be sync
             },
         )
         return cost_data
-    usage_data = response_data["usage"]
-    input_tokens = usage_data.get("prompt_tokens", 0)
-    output_tokens = usage_data.get("completion_tokens", 0)
 
-    # added for response api
-    input_tokens = (
-        input_tokens if input_tokens != 0 else usage_data.get("input_tokens", 0)
-    )
-    output_tokens = (
-        output_tokens if output_tokens != 0 else usage_data.get("output_tokens", 0)
-    )
-
-    # Calculate image completion cost
-    image_completion_msats = 0.0
-    if MSATS_PER_1K_IMAGE_COMPLETION_TOKENS > 0:
-        completion_details = usage_data.get("completion_tokens_details", {})
-        image_tokens = completion_details.get("image_tokens", 0)
-
-        if image_tokens > 0:
-            if output_tokens >= image_tokens:
-                output_tokens -= image_tokens
-
-            image_completion_msats = round(
-                image_tokens / 1000 * MSATS_PER_1K_IMAGE_COMPLETION_TOKENS, 3
-            )
-
-            logger.info(
-                "Calculated image completion cost",
-                extra={
-                    "image_tokens": image_tokens,
-                    "image_completion_msats": image_completion_msats,
-                },
-            )
+    input_tokens = response_data.get("usage", {}).get("prompt_tokens", 0)
+    output_tokens = response_data.get("usage", {}).get("completion_tokens", 0)
 
     input_msats = round(input_tokens / 1000 * MSATS_PER_1K_INPUT_TOKENS, 3)
-
     output_msats = round(output_tokens / 1000 * MSATS_PER_1K_OUTPUT_TOKENS, 3)
-    token_based_cost = math.ceil(input_msats + output_msats + image_completion_msats)
+    token_based_cost = math.ceil(input_msats + output_msats)
 
     logger.info(
         "Calculated token-based cost",
@@ -178,7 +143,6 @@ async def calculate_cost(  # todo: can be sync
             "output_tokens": output_tokens,
             "input_cost_msats": input_msats,
             "output_cost_msats": output_msats,
-            "image_completion_msats": image_completion_msats,
             "total_cost_msats": token_based_cost,
             "model": response_data.get("model", "unknown"),
         },

routstr/payment/models.py

@@ -31,7 +31,6 @@ class Pricing(BaseModel):
     completion: float
     request: float = 0.0
     image: float = 0.0
-    completion_image: float = 0.0
     web_search: float = 0.0
     internal_reasoning: float = 0.0
     input_cache_read: float = 0.0
@@ -41,13 +40,6 @@ class Pricing(BaseModel):
     max_cost: float = 0.0  # in sats not msats
 
 
-PRICING_OVERRIDES = {
-    "gemini-3-pro-image-preview": {"completion_image": 0.00012},
-    "gemini-2.5-flash-image": {"completion_image": 0.00003},
-    "gemini-2.0-flash": {"completion_image": 0.00003},
-}
-
-
 class TopProvider(BaseModel):
     context_length: int | None = None
     max_completion_tokens: int | None = None
@@ -101,12 +93,32 @@ async def async_fetch_openrouter_models(source_filter: str | None = None) -> lis
 
     try:
         async with httpx.AsyncClient() as client:
-            response = await client.get(f"{base_url}/models", timeout=30)
-            response.raise_for_status()
-            data = response.json()
+            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 []
 
             models_data: list[dict] = []
-            for model in data.get("data", []):
+            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:
                 model_id = model.get("id", "")
 
                 if source_filter:
@@ -124,19 +136,9 @@ async def async_fetch_openrouter_models(source_filter: str | None = None) -> lis
                 if not _has_valid_pricing(model):
                     continue
 
-                # Apply manual pricing overrides
-                if model_id in PRICING_OVERRIDES:
-                    pricing = model.get("pricing", {})
-                    if pricing:
-                        for k, v in PRICING_OVERRIDES[model_id].items():
-                            pricing[k] = str(
-                                v
-                            )  # OpenRouter API returns strings for pricing
-                        model["pricing"] = pricing
+                filtered_models.append(model)
 
-                models_data.append(model)
-
-            return models_data
+            return filtered_models
     except Exception as e:
         logger.error(f"Error (async) fetching models from OpenRouter API: {e}")
         return []
@@ -166,12 +168,6 @@ def _row_to_model(
     if isinstance(pricing, dict) and float(pricing.get("request", 0.0)) <= 0.0:
         pricing["request"] = max(pricing.get("request", 0.0), 0.0)
 
-    # Apply defaults for missing fields from manual overrides
-    if row.id in PRICING_OVERRIDES and isinstance(pricing, dict):
-        for k, v in PRICING_OVERRIDES[row.id].items():
-            if k not in pricing:
-                pricing[k] = v
-
     parsed_pricing = Pricing.parse_obj(pricing)
     model = Model(
         id=row.id,
@@ -531,7 +527,6 @@ def _pricing_matches(
         "completion",
         "request",
         "image",
-        "completion_image",
         "web_search",
         "internal_reasoning",
     ]

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)
+    return _model_instances.get(model_id.lower())
 
 
 def get_provider_for_model(model_id: str) -> BaseUpstreamProvider | None:

routstr/upstream/base.py

@@ -734,51 +734,53 @@ class BaseUpstreamProvider:
                     await client.aclose()
                 return mapped_error
 
-            if path.endswith("chat/completions"):
-                client_wants_streaming = False
-                if request_body:
-                    try:
-                        request_data = json.loads(request_body)
-                        client_wants_streaming = request_data.get("stream", False)
-                        logger.debug(
-                            "Chat completion request analysis",
-                            extra={
-                                "client_wants_streaming": client_wants_streaming,
-                                "model": request_data.get("model", "unknown"),
-                                "key_hash": key.hashed_key[:8] + "...",
-                            },
-                        )
-                    except json.JSONDecodeError:
-                        logger.warning(
-                            "Failed to parse request body JSON for streaming detection"
-                        )
+            if path.endswith("chat/completions") or path.endswith("embeddings"):
+                if path.endswith("chat/completions"):
+                    client_wants_streaming = False
+                    if request_body:
+                        try:
+                            request_data = json.loads(request_body)
+                            client_wants_streaming = request_data.get("stream", False)
+                            logger.debug(
+                                "Chat completion request analysis",
+                                extra={
+                                    "client_wants_streaming": client_wants_streaming,
+                                    "model": request_data.get("model", "unknown"),
+                                    "key_hash": key.hashed_key[:8] + "...",
+                                },
+                            )
+                        except json.JSONDecodeError:
+                            logger.warning(
+                                "Failed to parse request body JSON for streaming detection"
+                            )
 
-                content_type = response.headers.get("content-type", "")
-                upstream_is_streaming = "text/event-stream" in content_type
-                is_streaming = client_wants_streaming and upstream_is_streaming
+                    content_type = response.headers.get("content-type", "")
+                    upstream_is_streaming = "text/event-stream" in content_type
+                    is_streaming = client_wants_streaming and upstream_is_streaming
 
-                logger.debug(
-                    "Response type analysis",
-                    extra={
-                        "is_streaming": is_streaming,
-                        "client_wants_streaming": client_wants_streaming,
-                        "upstream_is_streaming": upstream_is_streaming,
-                        "content_type": content_type,
-                        "key_hash": key.hashed_key[:8] + "...",
-                    },
-                )
-
-                if is_streaming and response.status_code == 200:
-                    result = await self.handle_streaming_chat_completion(
-                        response, key, max_cost_for_model
+                    logger.debug(
+                        "Response type analysis",
+                        extra={
+                            "is_streaming": is_streaming,
+                            "client_wants_streaming": client_wants_streaming,
+                            "upstream_is_streaming": upstream_is_streaming,
+                            "content_type": content_type,
+                            "key_hash": key.hashed_key[:8] + "...",
+                        },
                     )
-                    background_tasks = BackgroundTasks()
-                    background_tasks.add_task(response.aclose)
-                    background_tasks.add_task(client.aclose)
-                    result.background = background_tasks
-                    return result
 
-                elif response.status_code == 200:
+                    if is_streaming and response.status_code == 200:
+                        result = await self.handle_streaming_chat_completion(
+                            response, key, max_cost_for_model
+                        )
+                        background_tasks = BackgroundTasks()
+                        background_tasks.add_task(response.aclose)
+                        background_tasks.add_task(client.aclose)
+                        result.background = background_tasks
+                        return result
+
+                # Handle both non-streaming chat completions and embeddings
+                if response.status_code == 200:
                     try:
                         return await self.handle_non_streaming_chat_completion(
                             response, key, session, max_cost_for_model
@@ -1519,9 +1521,9 @@ class BaseUpstreamProvider:
                     error_response.headers["X-Cashu"] = refund_token
                     return error_response
 
-                if path.endswith("chat/completions"):
+                if path.endswith("chat/completions") or path.endswith("embeddings"):
                     logger.debug(
-                        "Processing chat completion response",
+                        "Processing completion/embeddings response",
                         extra={"path": path, "amount": amount, "unit": unit},
                     )
 
@@ -1770,15 +1772,32 @@ class BaseUpstreamProvider:
     async def _fetch_openrouter_models(self) -> list[dict]:
         """Fetch models from OpenRouter API."""
         url = "https://openrouter.ai/api/v1/models"
+        embeddings_url = "https://openrouter.ai/api/v1/embeddings/models"
+
         async with httpx.AsyncClient(timeout=30.0) as client:
-            response = await client.get(url)
-            response.raise_for_status()
-            models = response.json()
-            return [
-                model
-                for model in models.get("data", [])
-                if ":free" not in model.get("id", "").lower()
-            ]
+            models_response, embeddings_response = await asyncio.gather(
+                client.get(url), client.get(embeddings_url), return_exceptions=True
+            )
+
+            all_models = []
+
+            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 []
+
+            all_models.extend(process_models_response(models_response))
+            all_models.extend(process_models_response(embeddings_response))
+
+            return all_models
 
     async def _fetch_provider_models(self) -> dict:
         """Fetch models from provider's API."""

routstr/upstream/openrouter.py

@@ -50,7 +50,13 @@ class OpenRouterUpstreamProvider(BaseUpstreamProvider):
     async def fetch_models(self) -> list[Model]:
         """Fetch all OpenRouter models."""
         models_data = await async_fetch_openrouter_models()
-        return [Model(**model) for model in models_data]  # type: ignore
+        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
 
     async def get_balance(self) -> float | None:
         """Get the current account balance from OpenRouter.

tests/integration/test_embeddings.py

@@ -0,0 +1,97 @@
+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]