From eebcc67c85304d31e71ea7253cca24e42fa3da48 Mon Sep 17 00:00:00 2001 From: 9qeklajc Date: Wed, 15 Apr 2026 17:19:13 +0200 Subject: [PATCH 01/10] add cost usages --- routstr/upstream/base.py | 82 +++++++----- tests/integration/test_usage_injection.py | 145 ++++++++++++++++++++++ 2 files changed, 197 insertions(+), 30 deletions(-) create mode 100644 tests/integration/test_usage_injection.py diff --git a/routstr/upstream/base.py b/routstr/upstream/base.py index b506e388..10232749 100644 --- a/routstr/upstream/base.py +++ b/routstr/upstream/base.py @@ -119,6 +119,39 @@ class BaseUpstreamProvider: "can_show_balance": False, } + def inject_cost_metadata( + self, response_json: dict, cost_data: CostData | MaxCostData, key: ApiKey + ) -> None: + """Unifies the injection of cost and usage metadata across all completion types.""" + sats_cost = cost_data.total_msats // 1000 + + # Inject into top-level usage block (OpenAI/Anthropic style) + if "usage" in response_json: + response_json["usage"]["cost"] = cost_data.total_usd + response_json["usage"]["sats_cost"] = sats_cost + response_json["usage"]["remaining_balance_msats"] = key.balance + + # Inject into Anthropic nested usage block if present + if ( + "message" in response_json + and isinstance(response_json["message"], dict) + and "usage" in response_json["message"] + ): + response_json["message"]["usage"]["sats_cost"] = sats_cost + + # Unified Routstr metadata + response_json["metadata"] = response_json.get("metadata", {}) + response_json["metadata"]["routstr"] = { + "cost": cost_data.dict(), + "sats_cost": sats_cost, + "remaining_balance_msats": key.balance, + } + + # Legacy/Compatibility fields + response_json["cost"] = cost_data.dict() + response_json["cost"]["sats_cost"] = sats_cost + response_json["cost"]["remaining_balance_msats"] = key.balance + def prepare_headers(self, request_headers: dict) -> dict: """Prepare headers for upstream request by removing proxy-specific headers and adding authentication. @@ -568,7 +601,7 @@ class BaseUpstreamProvider: usage_chunk_data["usage"]["cost"] = cost_data.get( "total_usd", 0.0 ) - usage_chunk_data["usage"]["cost_sats"] = ( + usage_chunk_data["usage"]["sats_cost"] = ( cost_data.get("total_msats", 0) // 1000 ) usage_chunk_data["usage"]["remaining_balance_msats"] = ( @@ -683,31 +716,10 @@ class BaseUpstreamProvider: key, response_json, session, deducted_max_cost ) - await session.refresh(key) - remaining_balance_msats = key.balance - - # Merge cost into usage for OpenCode - if "usage" in response_json: - response_json["usage"]["cost"] = cost_data.get("total_usd", 0.0) - response_json["usage"]["cost_sats"] = ( - cost_data.get("total_msats", 0) // 1000 - ) - response_json["usage"]["remaining_balance_msats"] = ( - remaining_balance_msats - ) - - # Keep detailed cost - response_json["metadata"] = response_json.get("metadata", {}) - response_json["metadata"]["routstr"] = {"cost": cost_data} - response_json["metadata"]["routstr"]["cost"]["sats_cost"] = ( - cost_data.get("total_msats", 0) // 1000 - ) - response_json["metadata"]["routstr"]["cost"]["remaining_balance_msats"] = ( - remaining_balance_msats - ) - response_json["cost"] = cost_data - response_json["cost"]["sats_cost"] = cost_data.get("total_msats", 0) // 1000 - response_json["cost"]["remaining_balance_msats"] = remaining_balance_msats + if isinstance(cost_data, (CostData, MaxCostData)): + self.inject_cost_metadata(response_json, cost_data, key) + else: + response_json["cost"] = cost_data logger.info( "Payment adjustment completed for non-streaming", @@ -895,7 +907,7 @@ class BaseUpstreamProvider: cost_data.get("total_usd", 0.0) ) usage_chunk_data["response"]["usage"][ - "cost_sats" + "sats_cost" ] = cost_data.get("total_msats", 0) // 1000 usage_chunk_data["response"]["usage"][ "remaining_balance_msats" @@ -904,7 +916,7 @@ class BaseUpstreamProvider: usage_chunk_data["usage"]["cost"] = cost_data.get( "total_usd", 0.0 ) - usage_chunk_data["usage"]["cost_sats"] = ( + usage_chunk_data["usage"]["sats_cost"] = ( cost_data.get("total_msats", 0) // 1000 ) usage_chunk_data["usage"][ @@ -1020,7 +1032,7 @@ class BaseUpstreamProvider: # Merge cost into usage for OpenCode if "usage" in response_json: response_json["usage"]["cost"] = cost_data.get("total_usd", 0.0) - response_json["usage"]["cost_sats"] = ( + response_json["usage"]["sats_cost"] = ( cost_data.get("total_msats", 0) // 1000 ) response_json["usage"]["remaining_balance_msats"] = ( @@ -1225,6 +1237,12 @@ class BaseUpstreamProvider: new_session, max_cost_for_model, ) + + if isinstance(cost_data, (CostData, MaxCostData)): + self.inject_cost_metadata( + combined_data, cost_data, fresh_key + ) + usage_finalized = True yield f"event: cost\ndata: {json.dumps({'cost': cost_data})}\n\n".encode() except Exception: @@ -1276,7 +1294,11 @@ class BaseUpstreamProvider: cost_data = await adjust_payment_for_tokens( key, response_json, session, deducted_max_cost ) - response_json["cost"] = cost_data + + if isinstance(cost_data, (CostData, MaxCostData)): + self.inject_cost_metadata(response_json, cost_data, key) + else: + response_json["cost"] = cost_data allowed_headers = { "content-type", diff --git a/tests/integration/test_usage_injection.py b/tests/integration/test_usage_injection.py new file mode 100644 index 00000000..59b672fd --- /dev/null +++ b/tests/integration/test_usage_injection.py @@ -0,0 +1,145 @@ +import json +from collections.abc import AsyncGenerator +from unittest.mock import AsyncMock, patch + +import pytest +from httpx import AsyncClient + + +@pytest.mark.integration +@pytest.mark.asyncio +async def test_usage_and_metadata_injection_all_endpoints( + authenticated_client: AsyncClient, +) -> None: + """Test that usage, cost, and metadata are injected into all completion endpoints.""" + + endpoints = [ + ( + "/v1/chat/completions", + { + "model": "gpt-3.5-turbo", + "messages": [{"role": "user", "content": "Hello"}], + }, + { + "id": "chatcmpl-123", + "object": "chat.completion", + "model": "gpt-3.5-turbo", + "choices": [{"message": {"content": "Hi"}, "finish_reason": "stop"}], + "usage": { + "prompt_tokens": 10, + "completion_tokens": 5, + "total_tokens": 15, + }, + }, + ), + ( + "/v1/messages", + { + "model": "claude-3-opus-20240229", + "messages": [{"role": "user", "content": "Hello"}], + "max_tokens": 100, + }, + { + "id": "msg_123", + "type": "message", + "role": "assistant", + "model": "claude-3-opus-20240229", + "content": [{"type": "text", "text": "Hi"}], + "usage": {"input_tokens": 10, "output_tokens": 5}, + }, + ), + ] + + for path, payload, mock_data in endpoints: + with patch("httpx.AsyncClient.send") as mock_send: + + async def mock_iter_bytes( + *args: object, **kwargs: object + ) -> AsyncGenerator[bytes, None]: + yield json.dumps(mock_data).encode() + + mock_response = AsyncMock() + mock_response.status_code = 200 + mock_response.headers = {"content-type": "application/json"} + mock_response.text = json.dumps(mock_data) + mock_response.json = AsyncMock(return_value=mock_data) + mock_response.aiter_bytes = mock_iter_bytes + mock_send.return_value = mock_response + + response = await authenticated_client.post(path, json=payload) + assert response.status_code == 200 + + data = response.json() + if hasattr(data, "__await__"): + data = await data + + # Check unified usage injection + assert "usage" in data + assert "cost" in data["usage"] + assert "sats_cost" in data["usage"] + assert "remaining_balance_msats" in data["usage"] + + # Check metadata injection + assert "metadata" in data + assert "routstr" in data["metadata"] + assert "cost" in data["metadata"]["routstr"] + assert "sats_cost" in data["metadata"]["routstr"] + assert "remaining_balance_msats" in data["metadata"]["routstr"] + + # Check legacy cost injection + assert "cost" in data + assert "sats_cost" in data["cost"] + + +@pytest.mark.integration +@pytest.mark.asyncio +async def test_streaming_usage_injection( + authenticated_client: AsyncClient, +) -> None: + """Test usage injection for streaming messages.""" + + path = "/v1/messages" + payload = { + "model": "claude-3-opus-20240229", + "messages": [{"role": "user", "content": "Hello"}], + "max_tokens": 100, + "stream": True, + } + + # Final usage chunk for Anthropic + streaming_chunks = [ + b'event: message_start\ndata: {"type": "message_start", "message": {"id": "msg_123", "type": "message", "role": "assistant", "model": "claude-3-opus-20240229", "usage": {"input_tokens": 10, "output_tokens": 0}}}\n\n', + b'event: message_delta\ndata: {"type": "message_delta", "delta": {"stop_reason": "end_turn", "stop_sequence": null}, "usage": {"output_tokens": 5}}\n\n', + b'event: message_stop\ndata: {"type": "message_stop"}\n\n', + ] + + with patch("httpx.AsyncClient.send") as mock_send: + + async def mock_iter_bytes( + *args: object, **kwargs: object + ) -> AsyncGenerator[bytes, None]: + for chunk in streaming_chunks: + yield chunk + + mock_response = AsyncMock() + mock_response.status_code = 200 + mock_response.headers = {"content-type": "text/event-stream"} + mock_response.aiter_bytes = mock_iter_bytes + mock_send.return_value = mock_response + + response = await authenticated_client.post(path, json=payload) + assert response.status_code == 200 + + # In streaming, we look for the final 'event: cost' which is added by Routstr + response_text = response.text + if hasattr(response_text, "__await__"): + response_text = await response_text + lines = response_text.split("\n") + + cost_event = next( + (line for line in lines if line.startswith('data: {"cost":')), None + ) + assert cost_event is not None + + cost_data = json.loads(cost_event[6:])["cost"] + assert "total_msats" in cost_data From a1ae6e94e91deecafb6cb1d19290de14db7da4c0 Mon Sep 17 00:00:00 2001 From: 9qeklajc Date: Wed, 15 Apr 2026 19:20:50 +0200 Subject: [PATCH 02/10] match model when versioned --- routstr/proxy.py | 20 ++- tests/integration/test_usage_injection.py | 145 ---------------------- 2 files changed, 19 insertions(+), 146 deletions(-) delete mode 100644 tests/integration/test_usage_injection.py diff --git a/routstr/proxy.py b/routstr/proxy.py index ddab1c31..9d2dbf09 100644 --- a/routstr/proxy.py +++ b/routstr/proxy.py @@ -69,7 +69,25 @@ 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()) + if not model_id: + return None + + model_id_lower = model_id.lower() + # Try exact match first + if model := _model_instances.get(model_id_lower): + return model + + # Try stripping common version suffixes (e.g., -20251222) + # This handles cases where upstream returns a specific version + # but we only track the base model name. + import re + + base_model_id = re.sub(r"-\d{8}$", "", model_id_lower) + if base_model_id != model_id_lower: + if model := _model_instances.get(base_model_id): + return model + + return None def get_provider_for_model(model_id: str) -> list[BaseUpstreamProvider] | None: diff --git a/tests/integration/test_usage_injection.py b/tests/integration/test_usage_injection.py deleted file mode 100644 index 59b672fd..00000000 --- a/tests/integration/test_usage_injection.py +++ /dev/null @@ -1,145 +0,0 @@ -import json -from collections.abc import AsyncGenerator -from unittest.mock import AsyncMock, patch - -import pytest -from httpx import AsyncClient - - -@pytest.mark.integration -@pytest.mark.asyncio -async def test_usage_and_metadata_injection_all_endpoints( - authenticated_client: AsyncClient, -) -> None: - """Test that usage, cost, and metadata are injected into all completion endpoints.""" - - endpoints = [ - ( - "/v1/chat/completions", - { - "model": "gpt-3.5-turbo", - "messages": [{"role": "user", "content": "Hello"}], - }, - { - "id": "chatcmpl-123", - "object": "chat.completion", - "model": "gpt-3.5-turbo", - "choices": [{"message": {"content": "Hi"}, "finish_reason": "stop"}], - "usage": { - "prompt_tokens": 10, - "completion_tokens": 5, - "total_tokens": 15, - }, - }, - ), - ( - "/v1/messages", - { - "model": "claude-3-opus-20240229", - "messages": [{"role": "user", "content": "Hello"}], - "max_tokens": 100, - }, - { - "id": "msg_123", - "type": "message", - "role": "assistant", - "model": "claude-3-opus-20240229", - "content": [{"type": "text", "text": "Hi"}], - "usage": {"input_tokens": 10, "output_tokens": 5}, - }, - ), - ] - - for path, payload, mock_data in endpoints: - with patch("httpx.AsyncClient.send") as mock_send: - - async def mock_iter_bytes( - *args: object, **kwargs: object - ) -> AsyncGenerator[bytes, None]: - yield json.dumps(mock_data).encode() - - mock_response = AsyncMock() - mock_response.status_code = 200 - mock_response.headers = {"content-type": "application/json"} - mock_response.text = json.dumps(mock_data) - mock_response.json = AsyncMock(return_value=mock_data) - mock_response.aiter_bytes = mock_iter_bytes - mock_send.return_value = mock_response - - response = await authenticated_client.post(path, json=payload) - assert response.status_code == 200 - - data = response.json() - if hasattr(data, "__await__"): - data = await data - - # Check unified usage injection - assert "usage" in data - assert "cost" in data["usage"] - assert "sats_cost" in data["usage"] - assert "remaining_balance_msats" in data["usage"] - - # Check metadata injection - assert "metadata" in data - assert "routstr" in data["metadata"] - assert "cost" in data["metadata"]["routstr"] - assert "sats_cost" in data["metadata"]["routstr"] - assert "remaining_balance_msats" in data["metadata"]["routstr"] - - # Check legacy cost injection - assert "cost" in data - assert "sats_cost" in data["cost"] - - -@pytest.mark.integration -@pytest.mark.asyncio -async def test_streaming_usage_injection( - authenticated_client: AsyncClient, -) -> None: - """Test usage injection for streaming messages.""" - - path = "/v1/messages" - payload = { - "model": "claude-3-opus-20240229", - "messages": [{"role": "user", "content": "Hello"}], - "max_tokens": 100, - "stream": True, - } - - # Final usage chunk for Anthropic - streaming_chunks = [ - b'event: message_start\ndata: {"type": "message_start", "message": {"id": "msg_123", "type": "message", "role": "assistant", "model": "claude-3-opus-20240229", "usage": {"input_tokens": 10, "output_tokens": 0}}}\n\n', - b'event: message_delta\ndata: {"type": "message_delta", "delta": {"stop_reason": "end_turn", "stop_sequence": null}, "usage": {"output_tokens": 5}}\n\n', - b'event: message_stop\ndata: {"type": "message_stop"}\n\n', - ] - - with patch("httpx.AsyncClient.send") as mock_send: - - async def mock_iter_bytes( - *args: object, **kwargs: object - ) -> AsyncGenerator[bytes, None]: - for chunk in streaming_chunks: - yield chunk - - mock_response = AsyncMock() - mock_response.status_code = 200 - mock_response.headers = {"content-type": "text/event-stream"} - mock_response.aiter_bytes = mock_iter_bytes - mock_send.return_value = mock_response - - response = await authenticated_client.post(path, json=payload) - assert response.status_code == 200 - - # In streaming, we look for the final 'event: cost' which is added by Routstr - response_text = response.text - if hasattr(response_text, "__await__"): - response_text = await response_text - lines = response_text.split("\n") - - cost_event = next( - (line for line in lines if line.startswith('data: {"cost":')), None - ) - assert cost_event is not None - - cost_data = json.loads(cost_event[6:])["cost"] - assert "total_msats" in cost_data From 45bdfaee58eee4757da70261803f666ec9a1b445 Mon Sep 17 00:00:00 2001 From: 9qeklajc Date: Wed, 15 Apr 2026 20:18:56 +0200 Subject: [PATCH 03/10] best effort to be compatible with legacy code --- routstr/upstream/base.py | 39 ++++++++++++++++++++++----------------- 1 file changed, 22 insertions(+), 17 deletions(-) diff --git a/routstr/upstream/base.py b/routstr/upstream/base.py index 10232749..b1827a57 100644 --- a/routstr/upstream/base.py +++ b/routstr/upstream/base.py @@ -120,14 +120,26 @@ class BaseUpstreamProvider: } def inject_cost_metadata( - self, response_json: dict, cost_data: CostData | MaxCostData, key: ApiKey + self, + response_json: dict, + cost_data: CostData | MaxCostData | dict, + key: ApiKey, ) -> None: """Unifies the injection of cost and usage metadata across all completion types.""" - sats_cost = cost_data.total_msats // 1000 + if isinstance(cost_data, dict): + total_msats = cost_data.get("total_msats", 0) + total_usd = cost_data.get("total_usd", 0.0) + cost_dict = cost_data + else: + total_msats = cost_data.total_msats + total_usd = cost_data.total_usd + cost_dict = cost_data.dict() + + sats_cost = total_msats // 1000 # Inject into top-level usage block (OpenAI/Anthropic style) if "usage" in response_json: - response_json["usage"]["cost"] = cost_data.total_usd + response_json["usage"]["cost"] = total_usd response_json["usage"]["sats_cost"] = sats_cost response_json["usage"]["remaining_balance_msats"] = key.balance @@ -142,13 +154,13 @@ class BaseUpstreamProvider: # Unified Routstr metadata response_json["metadata"] = response_json.get("metadata", {}) response_json["metadata"]["routstr"] = { - "cost": cost_data.dict(), + "cost": cost_dict, "sats_cost": sats_cost, "remaining_balance_msats": key.balance, } # Legacy/Compatibility fields - response_json["cost"] = cost_data.dict() + response_json["cost"] = cost_dict.copy() response_json["cost"]["sats_cost"] = sats_cost response_json["cost"]["remaining_balance_msats"] = key.balance @@ -716,10 +728,7 @@ class BaseUpstreamProvider: key, response_json, session, deducted_max_cost ) - if isinstance(cost_data, (CostData, MaxCostData)): - self.inject_cost_metadata(response_json, cost_data, key) - else: - response_json["cost"] = cost_data + self.inject_cost_metadata(response_json, cost_data, key) logger.info( "Payment adjustment completed for non-streaming", @@ -1238,10 +1247,9 @@ class BaseUpstreamProvider: max_cost_for_model, ) - if isinstance(cost_data, (CostData, MaxCostData)): - self.inject_cost_metadata( - combined_data, cost_data, fresh_key - ) + self.inject_cost_metadata( + combined_data, cost_data, fresh_key + ) usage_finalized = True yield f"event: cost\ndata: {json.dumps({'cost': cost_data})}\n\n".encode() @@ -1295,10 +1303,7 @@ class BaseUpstreamProvider: key, response_json, session, deducted_max_cost ) - if isinstance(cost_data, (CostData, MaxCostData)): - self.inject_cost_metadata(response_json, cost_data, key) - else: - response_json["cost"] = cost_data + self.inject_cost_metadata(response_json, cost_data, key) allowed_headers = { "content-type", From 25f39897d7119dcaaa7d1c3e1089859c74501431 Mon Sep 17 00:00:00 2001 From: 9qeklajc Date: Wed, 15 Apr 2026 20:45:13 +0200 Subject: [PATCH 04/10] mirror logic from chat completion --- routstr/upstream/base.py | 66 ++++++++++++++++++++++++++++++++++------ 1 file changed, 57 insertions(+), 9 deletions(-) diff --git a/routstr/upstream/base.py b/routstr/upstream/base.py index b1827a57..d349ab7f 100644 --- a/routstr/upstream/base.py +++ b/routstr/upstream/base.py @@ -140,7 +140,7 @@ class BaseUpstreamProvider: # Inject into top-level usage block (OpenAI/Anthropic style) if "usage" in response_json: response_json["usage"]["cost"] = total_usd - response_json["usage"]["sats_cost"] = sats_cost + response_json["usage"]["cost_sats"] = sats_cost response_json["usage"]["remaining_balance_msats"] = key.balance # Inject into Anthropic nested usage block if present @@ -613,7 +613,7 @@ class BaseUpstreamProvider: usage_chunk_data["usage"]["cost"] = cost_data.get( "total_usd", 0.0 ) - usage_chunk_data["usage"]["sats_cost"] = ( + usage_chunk_data["usage"]["cost_sats"] = ( cost_data.get("total_msats", 0) // 1000 ) usage_chunk_data["usage"]["remaining_balance_msats"] = ( @@ -1160,7 +1160,11 @@ class BaseUpstreamProvider: ) async def handle_streaming_messages_completion( - self, response: httpx.Response, key: ApiKey, max_cost_for_model: int + self, + response: httpx.Response, + key: ApiKey, + max_cost_for_model: int, + requested_model: str | None = None, ) -> StreamingResponse: async def stream_with_cost( max_cost_for_model: int, @@ -1199,6 +1203,8 @@ class BaseUpstreamProvider: stored_chunks.append(chunk) try: decoded_chunk = chunk.decode("utf-8", errors="ignore") + modified_lines = [] + changed = False for line in decoded_chunk.split("\n"): if line.startswith("data: "): try: @@ -1208,6 +1214,20 @@ class BaseUpstreamProvider: if msg and msg.get("model"): last_model_seen = str(msg.get("model")) + if requested_model: + # Apply requested_model override + model_updated = False + if msg: + msg["model"] = requested_model + model_updated = True + if data.get("model"): + data["model"] = requested_model + model_updated = True + + if model_updated: + line = "data: " + json.dumps(data) + changed = True + if usage := msg.get("usage"): input_tokens += usage.get("input_tokens", 0) output_tokens += usage.get( @@ -1221,10 +1241,14 @@ class BaseUpstreamProvider: ) except json.JSONDecodeError: pass - except Exception: - pass + modified_lines.append(line) - yield chunk + if changed: + yield "\n".join(modified_lines).encode("utf-8") + else: + yield chunk + except Exception: + yield chunk usage_data = { "input_tokens": input_tokens, @@ -1290,11 +1314,22 @@ class BaseUpstreamProvider: session: AsyncSession, deducted_max_cost: int, path: str, + requested_model: str | None = None, ) -> Response: try: content = await response.aread() response_json = json.loads(content) + if requested_model: + if "model" in response_json: + response_json["model"] = requested_model + if ( + "message" in response_json + and isinstance(response_json["message"], dict) + and "model" in response_json["message"] + ): + response_json["message"]["model"] = requested_model + if path.endswith("count_tokens") and "usage" not in response_json: input_tokens = response_json.get("input_tokens", 0) response_json["usage"] = {"input_tokens": input_tokens} @@ -1457,7 +1492,10 @@ class BaseUpstreamProvider: if is_streaming and response.status_code == 200: result = await self.handle_streaming_messages_completion( - response, key, max_cost_for_model + response, + key, + max_cost_for_model, + requested_model=original_model_id, ) background_tasks = BackgroundTasks() background_tasks.add_task(response.aclose) @@ -1468,7 +1506,12 @@ class BaseUpstreamProvider: if response.status_code == 200: try: return await self.handle_non_streaming_messages_completion( - response, key, session, max_cost_for_model, path + response, + key, + session, + max_cost_for_model, + path, + requested_model=original_model_id, ) finally: await response.aclose() @@ -1478,7 +1521,12 @@ class BaseUpstreamProvider: if response.status_code == 200: try: return await self.handle_non_streaming_messages_completion( - response, key, session, max_cost_for_model, path + response, + key, + session, + max_cost_for_model, + path, + requested_model=original_model_id, ) finally: await response.aclose() From 81146710bad26eab48d01e72321502bec75d005c Mon Sep 17 00:00:00 2001 From: 9qeklajc Date: Wed, 15 Apr 2026 21:27:02 +0200 Subject: [PATCH 05/10] revert --- routstr/upstream/base.py | 20 ++++++++++---------- 1 file changed, 10 insertions(+), 10 deletions(-) diff --git a/routstr/upstream/base.py b/routstr/upstream/base.py index d349ab7f..ee8a2f61 100644 --- a/routstr/upstream/base.py +++ b/routstr/upstream/base.py @@ -149,19 +149,19 @@ class BaseUpstreamProvider: and isinstance(response_json["message"], dict) and "usage" in response_json["message"] ): - response_json["message"]["usage"]["sats_cost"] = sats_cost + response_json["message"]["usage"]["cost_sats"] = sats_cost # Unified Routstr metadata response_json["metadata"] = response_json.get("metadata", {}) response_json["metadata"]["routstr"] = { "cost": cost_dict, - "sats_cost": sats_cost, + "cost_sats": sats_cost, "remaining_balance_msats": key.balance, } # Legacy/Compatibility fields response_json["cost"] = cost_dict.copy() - response_json["cost"]["sats_cost"] = sats_cost + response_json["cost"]["cost_sats"] = sats_cost response_json["cost"]["remaining_balance_msats"] = key.balance def prepare_headers(self, request_headers: dict) -> dict: @@ -627,7 +627,7 @@ class BaseUpstreamProvider: "cost": cost_data } usage_chunk_data["metadata"]["routstr"]["cost"][ - "sats_cost" + "cost_sats" ] = cost_data.get("total_msats", 0) // 1000 usage_chunk_data["metadata"]["routstr"]["cost"][ "remaining_balance_msats" @@ -916,7 +916,7 @@ class BaseUpstreamProvider: cost_data.get("total_usd", 0.0) ) usage_chunk_data["response"]["usage"][ - "sats_cost" + "cost_sats" ] = cost_data.get("total_msats", 0) // 1000 usage_chunk_data["response"]["usage"][ "remaining_balance_msats" @@ -925,7 +925,7 @@ class BaseUpstreamProvider: usage_chunk_data["usage"]["cost"] = cost_data.get( "total_usd", 0.0 ) - usage_chunk_data["usage"]["sats_cost"] = ( + usage_chunk_data["usage"]["cost_sats"] = ( cost_data.get("total_msats", 0) // 1000 ) usage_chunk_data["usage"][ @@ -940,7 +940,7 @@ class BaseUpstreamProvider: "cost": cost_data } usage_chunk_data["metadata"]["routstr"]["cost"][ - "sats_cost" + "cost_sats" ] = cost_data.get("total_msats", 0) // 1000 usage_chunk_data["metadata"]["routstr"]["cost"][ "remaining_balance_msats" @@ -1041,7 +1041,7 @@ class BaseUpstreamProvider: # Merge cost into usage for OpenCode if "usage" in response_json: response_json["usage"]["cost"] = cost_data.get("total_usd", 0.0) - response_json["usage"]["sats_cost"] = ( + response_json["usage"]["cost_sats"] = ( cost_data.get("total_msats", 0) // 1000 ) response_json["usage"]["remaining_balance_msats"] = ( @@ -1051,14 +1051,14 @@ class BaseUpstreamProvider: # Keep detailed cost response_json["metadata"] = response_json.get("metadata", {}) response_json["metadata"]["routstr"] = {"cost": cost_data} - response_json["metadata"]["routstr"]["cost"]["sats_cost"] = ( + response_json["metadata"]["routstr"]["cost"]["cost_sats"] = ( cost_data.get("total_msats", 0) // 1000 ) response_json["metadata"]["routstr"]["cost"]["remaining_balance_msats"] = ( remaining_balance_msats ) response_json["cost"] = cost_data - response_json["cost"]["sats_cost"] = cost_data.get("total_msats", 0) // 1000 + response_json["cost"]["cost_sats"] = cost_data.get("total_msats", 0) // 1000 response_json["cost"]["remaining_balance_msats"] = remaining_balance_msats logger.info( From 8b6393f7943595c16a001c3b9ec2bfcff653f042 Mon Sep 17 00:00:00 2001 From: 9qeklajc Date: Wed, 15 Apr 2026 21:50:12 +0200 Subject: [PATCH 06/10] clean up --- routstr/upstream/base.py | 40 ++++++++++++++++++++++++++++++++-------- 1 file changed, 32 insertions(+), 8 deletions(-) diff --git a/routstr/upstream/base.py b/routstr/upstream/base.py index ee8a2f61..8c93d88f 100644 --- a/routstr/upstream/base.py +++ b/routstr/upstream/base.py @@ -149,19 +149,19 @@ class BaseUpstreamProvider: and isinstance(response_json["message"], dict) and "usage" in response_json["message"] ): - response_json["message"]["usage"]["cost_sats"] = sats_cost + response_json["message"]["usage"]["sats_cost"] = sats_cost # Unified Routstr metadata response_json["metadata"] = response_json.get("metadata", {}) response_json["metadata"]["routstr"] = { "cost": cost_dict, - "cost_sats": sats_cost, + "sats_cost": sats_cost, "remaining_balance_msats": key.balance, } # Legacy/Compatibility fields response_json["cost"] = cost_dict.copy() - response_json["cost"]["cost_sats"] = sats_cost + response_json["cost"]["sats_cost"] = sats_cost response_json["cost"]["remaining_balance_msats"] = key.balance def prepare_headers(self, request_headers: dict) -> dict: @@ -627,7 +627,7 @@ class BaseUpstreamProvider: "cost": cost_data } usage_chunk_data["metadata"]["routstr"]["cost"][ - "cost_sats" + "sats_cost" ] = cost_data.get("total_msats", 0) // 1000 usage_chunk_data["metadata"]["routstr"]["cost"][ "remaining_balance_msats" @@ -728,7 +728,31 @@ class BaseUpstreamProvider: key, response_json, session, deducted_max_cost ) - self.inject_cost_metadata(response_json, cost_data, key) + await session.refresh(key) + remaining_balance_msats = key.balance + + # Merge cost into usage for OpenCode + if "usage" in response_json: + response_json["usage"]["cost"] = cost_data.get("total_usd", 0.0) + response_json["usage"]["cost_sats"] = ( + cost_data.get("total_msats", 0) // 1000 + ) + response_json["usage"]["remaining_balance_msats"] = ( + remaining_balance_msats + ) + + # Keep detailed cost + response_json["metadata"] = response_json.get("metadata", {}) + response_json["metadata"]["routstr"] = {"cost": cost_data} + response_json["metadata"]["routstr"]["cost"]["sats_cost"] = ( + cost_data.get("total_msats", 0) // 1000 + ) + response_json["metadata"]["routstr"]["cost"]["remaining_balance_msats"] = ( + remaining_balance_msats + ) + response_json["cost"] = cost_data + response_json["cost"]["sats_cost"] = cost_data.get("total_msats", 0) // 1000 + response_json["cost"]["remaining_balance_msats"] = remaining_balance_msats logger.info( "Payment adjustment completed for non-streaming", @@ -940,7 +964,7 @@ class BaseUpstreamProvider: "cost": cost_data } usage_chunk_data["metadata"]["routstr"]["cost"][ - "cost_sats" + "sats_cost" ] = cost_data.get("total_msats", 0) // 1000 usage_chunk_data["metadata"]["routstr"]["cost"][ "remaining_balance_msats" @@ -1051,14 +1075,14 @@ class BaseUpstreamProvider: # Keep detailed cost response_json["metadata"] = response_json.get("metadata", {}) response_json["metadata"]["routstr"] = {"cost": cost_data} - response_json["metadata"]["routstr"]["cost"]["cost_sats"] = ( + response_json["metadata"]["routstr"]["cost"]["sats_cost"] = ( cost_data.get("total_msats", 0) // 1000 ) response_json["metadata"]["routstr"]["cost"]["remaining_balance_msats"] = ( remaining_balance_msats ) response_json["cost"] = cost_data - response_json["cost"]["cost_sats"] = cost_data.get("total_msats", 0) // 1000 + response_json["cost"]["sats_cost"] = cost_data.get("total_msats", 0) // 1000 response_json["cost"]["remaining_balance_msats"] = remaining_balance_msats logger.info( From 58e16203479fba9c7243dde911e6db366f8e529e Mon Sep 17 00:00:00 2001 From: 9qeklajc Date: Sat, 18 Apr 2026 00:30:19 +0200 Subject: [PATCH 07/10] add logging reason for bad not succeed requests --- routstr/upstream/base.py | 58 ++++++++++++++++++++++++++++------------ 1 file changed, 41 insertions(+), 17 deletions(-) diff --git a/routstr/upstream/base.py b/routstr/upstream/base.py index 8c93d88f..a4769000 100644 --- a/routstr/upstream/base.py +++ b/routstr/upstream/base.py @@ -1467,15 +1467,28 @@ class BaseUpstreamProvider: stream=True, ) - logger.info( - "Received upstream response", - extra={ - "status_code": response.status_code, - "path": path, - "key_hash": key.hashed_key[:8] + "...", - "content_type": response.headers.get("content-type", "unknown"), - }, - ) + if response.status_code != 200: + logger.error( + "Received upstream response", + extra={ + "reason_phrase": response.reason_phrase, + "status_code": response.status_code, + "path": path, + "key_hash": key.hashed_key[:8] + "...", + "content_type": response.headers.get("content-type", "unknown"), + }, + ) + else: + logger.info( + "Received upstream response", + extra={ + "reason_phrase": response.reason_phrase, + "status_code": response.status_code, + "path": path, + "key_hash": key.hashed_key[:8] + "...", + "content_type": response.headers.get("content-type", "unknown"), + }, + ) if response.status_code != 200: if response.status_code >= 500: @@ -2620,14 +2633,25 @@ class BaseUpstreamProvider: stream=True, ) - logger.debug( - "Received upstream response", - extra={ - "status_code": response.status_code, - "path": path, - "response_headers": dict(response.headers), - }, - ) + if response.status_code != 200: + logger.error( + "Received upstream response", + extra={ + "reason_phrase": response.reason_phrase, + "status_code": response.status_code, + "path": path, + "response_headers": dict(response.headers), + }, + ) + else: + logger.debug( + "Received upstream response", + extra={ + "status_code": response.status_code, + "path": path, + "response_headers": dict(response.headers), + }, + ) if response.status_code != 200: logger.warning( From d95c09e00c3f4d9b1a50e97d123a1f002b20a28b Mon Sep 17 00:00:00 2001 From: 9qeklajc Date: Sun, 19 Apr 2026 15:21:23 +0200 Subject: [PATCH 08/10] fix test model connection --- routstr/payment/models.py | 72 ++++++++++++++++++++++++++- ui/components/api-endpoint-tester.tsx | 6 ++- ui/components/model-tester.tsx | 6 ++- 3 files changed, 81 insertions(+), 3 deletions(-) diff --git a/routstr/payment/models.py b/routstr/payment/models.py index 95d7ad89..afed3238 100644 --- a/routstr/payment/models.py +++ b/routstr/payment/models.py @@ -7,7 +7,7 @@ from fastapi import APIRouter, Depends from pydantic.v1 import BaseModel from sqlmodel.ext.asyncio.session import AsyncSession -from ..core.db import ModelRow, get_session +from ..core.db import ModelRow, UpstreamProviderRow, get_session from ..core.logging import get_logger from ..core.settings import settings from .price import sats_usd_price @@ -405,6 +405,76 @@ async def update_sats_pricing() -> None: logger.error(f"Error updating sats pricing: {e}") +class ModelTestRequest(BaseModel): + model_id: str + endpoint_type: str + request_data: dict + + +@models_router.post("/api/models/test") +async def test_model( + payload: ModelTestRequest, + session: AsyncSession = Depends(get_session), +) -> dict: + """Test a model by sending a request through its configured upstream provider.""" + from sqlmodel import select + + result = await session.execute( + select(ModelRow).where(ModelRow.id == payload.model_id) + ) + model_row = result.scalars().first() + + if not model_row: + return { + "success": False, + "error": f"Model '{payload.model_id}' not found in database", + "status_code": 404, + } + + provider = await session.get(UpstreamProviderRow, model_row.upstream_provider_id) + if not provider: + return { + "success": False, + "error": "Upstream provider not found", + "status_code": 404, + } + + base_url = provider.base_url.rstrip("/") + if payload.endpoint_type == "chat-completions": + url = f"{base_url}/chat/completions" + else: + url = f"{base_url}/{payload.endpoint_type}" + + actual_model_id = model_row.forwarded_model_id or model_row.id + request_data = dict(payload.request_data) + request_data["model"] = actual_model_id + + headers = { + "Content-Type": "application/json", + "Authorization": f"Bearer {provider.api_key}", + } + + try: + async with httpx.AsyncClient(timeout=30.0) as client: + response = await client.post(url, json=request_data, headers=headers) + try: + response_data = response.json() + except Exception: + response_data = {"raw": response.text} + + return { + "success": response.status_code < 400, + "data": response_data, + "status_code": response.status_code, + } + except Exception as e: + return { + "success": False, + "error": str(e), + "status_code": 500, + } + + @models_router.get("/v1/models") @models_router.get("/v1/models/", include_in_schema=False) @models_router.get("/models") diff --git a/ui/components/api-endpoint-tester.tsx b/ui/components/api-endpoint-tester.tsx index a77b4248..dbb91266 100644 --- a/ui/components/api-endpoint-tester.tsx +++ b/ui/components/api-endpoint-tester.tsx @@ -471,7 +471,11 @@ export function ApiEndpointTester({ models }: ApiEndpointTesterProps) { testEndpointMutation.mutate(requestData); }; - const enabledModels = models.filter((model) => model.isEnabled); + const enabledModels = Array.from( + new Map( + models.filter((model) => model.isEnabled).map((m) => [m.id, m]) + ).values() + ); const credentials = selectedModel ? getModelCredentials(selectedModel) : null; const endpointUrl = credentials ? buildEndpointUrl( diff --git a/ui/components/model-tester.tsx b/ui/components/model-tester.tsx index dee500a7..3ecd43b3 100644 --- a/ui/components/model-tester.tsx +++ b/ui/components/model-tester.tsx @@ -197,7 +197,11 @@ export function ModelTester({ models }: ModelTesterProps) { testModelMutation.mutate(request); }; - const enabledModels = models.filter((model) => model.isEnabled); + const enabledModels = Array.from( + new Map( + models.filter((model) => model.isEnabled).map((m) => [m.id, m]) + ).values() + ); const credentials = selectedModel ? getModelCredentials(selectedModel) : null; return ( From 3681fc8aabef8789ea40c577f10241f3b953fc76 Mon Sep 17 00:00:00 2001 From: 9qeklajc Date: Sun, 19 Apr 2026 15:22:47 +0200 Subject: [PATCH 09/10] no advanced testing for now --- ui/components/models-page.tsx | 2 ++ 1 file changed, 2 insertions(+) diff --git a/ui/components/models-page.tsx b/ui/components/models-page.tsx index f8505d7f..bff3a379 100644 --- a/ui/components/models-page.tsx +++ b/ui/components/models-page.tsx @@ -124,12 +124,14 @@ export function ModelsPage() { > Basic Testing + {/* API Endpoints + */} From 8c6d1f89dccb46e65c66574ed6997d89b10d097a Mon Sep 17 00:00:00 2001 From: 9qeklajc Date: Mon, 20 Apr 2026 16:36:41 +0200 Subject: [PATCH 10/10] fix forwarding upstream error responses --- routstr/upstream/base.py | 116 +++++++++++++++++++++------------------ 1 file changed, 62 insertions(+), 54 deletions(-) diff --git a/routstr/upstream/base.py b/routstr/upstream/base.py index a4769000..458e5dfd 100644 --- a/routstr/upstream/base.py +++ b/routstr/upstream/base.py @@ -421,75 +421,83 @@ class BaseUpstreamProvider: """ pass - async def map_upstream_error_response( + async def forward_upstream_error_response( self, request: Request, path: str, upstream_response: httpx.Response ) -> Response: - """Map upstream error responses to appropriate proxy error responses. - - Args: - request: Original FastAPI request - path: Request path - upstream_response: Response from upstream service - - Returns: - Mapped error response with appropriate status code and error type - """ + """Log upstream errors and forward the upstream response unchanged.""" status_code = upstream_response.status_code headers = dict(upstream_response.headers) - content_type = headers.get("content-type", "") + content_type = headers.get("content-type") or headers.get("Content-Type", "") + upstream_request_id = ( + headers.get("request-id") + or headers.get("Request-Id") + or headers.get("x-request-id") + or headers.get("X-Request-Id") + or headers.get("anthropic-request-id") + or headers.get("openai-request-id") + ) + + body_read_error = None try: body_bytes = await upstream_response.aread() - except Exception: + except Exception as exc: body_bytes = b"" + body_read_error = f"{type(exc).__name__}: {exc}" message, upstream_code = self._extract_upstream_error_message(body_bytes) - lowered_message = message.lower() - lowered_code = (upstream_code or "").lower() + body_preview = body_bytes.decode("utf-8", errors="ignore").strip()[:500] - error_type = "upstream_error" - mapped_status = 502 - - if status_code in (400, 422): - error_type = "invalid_request_error" - mapped_status = 400 - elif status_code in (401, 403): - error_type = "upstream_auth_error" - mapped_status = 502 - elif status_code == 404: - if path.endswith("chat/completions"): - error_type = "invalid_model" - mapped_status = 400 - if not message or message == "Upstream request failed": - message = "Requested model is not available upstream" - elif "model" in lowered_message or "model" in lowered_code: - error_type = "invalid_model" - mapped_status = 400 - if not message or message == "Upstream request failed": - message = "Requested model is not available upstream" - else: - error_type = "upstream_error" - mapped_status = 502 - elif status_code == 429: - error_type = "rate_limit_exceeded" - mapped_status = 429 - elif status_code >= 500: - error_type = "upstream_error" - mapped_status = 502 - - logger.debug( - "Mapped upstream error", + logger.warning( + "Forwarding upstream error response as-is", extra={ "path": path, + "provider": self.provider_type, "upstream_status": status_code, - "mapped_status": mapped_status, - "error_type": error_type, + "upstream_code": upstream_code, "upstream_content_type": content_type, + "upstream_request_id": upstream_request_id, "message_preview": message[:200], + "body_preview": body_preview, + "body_read_error": body_read_error, + "method": request.method, }, ) - return create_error_response( - error_type, message, mapped_status, request=request + for header_name in ( + "content-length", + "Content-Length", + "transfer-encoding", + "Transfer-Encoding", + "content-encoding", + "Content-Encoding", + "connection", + "Connection", + "keep-alive", + "Keep-Alive", + "proxy-authenticate", + "Proxy-Authenticate", + "proxy-authorization", + "Proxy-Authorization", + "te", + "TE", + "trailer", + "Trailer", + "upgrade", + "Upgrade", + ): + headers.pop(header_name, None) + + if not content_type: + headers.pop("content-type", None) + headers.pop("Content-Type", None) + + media_type = content_type or None + + return Response( + content=body_bytes, + status_code=status_code, + headers=headers, + media_type=media_type, ) async def handle_streaming_chat_completion( @@ -1500,7 +1508,7 @@ class BaseUpstreamProvider: ) try: - mapped_error = await self.map_upstream_error_response( + mapped_error = await self.forward_upstream_error_response( request, path, response ) finally: @@ -1807,7 +1815,7 @@ class BaseUpstreamProvider: ) try: - mapped_error = await self.map_upstream_error_response( + mapped_error = await self.forward_upstream_error_response( request, path, response ) finally: @@ -1979,7 +1987,7 @@ class BaseUpstreamProvider: ) if response.status_code != 200: try: - mapped = await self.map_upstream_error_response( + mapped = await self.forward_upstream_error_response( request, path, response ) finally: