diff --git a/routstr/payment/cost_calculation.py b/routstr/payment/cost_calculation.py index 44330620..292b6d5f 100644 --- a/routstr/payment/cost_calculation.py +++ b/routstr/payment/cost_calculation.py @@ -53,10 +53,17 @@ async def calculate_cost( # todo: can be sync if "usage" not in response_data or response_data["usage"] is None: logger.warning( - "No usage data in response, using base cost only", + "No usage data in response — billing at MaxCostData with zero " + "tokens. Dashboard will show this request as `(0+0)`. Most " + "common cause: upstream stream did not include a final usage " + "chunk (OpenAI-compat backends require " + "`stream_options.include_usage=true`).", extra={ "max_cost_msats": max_cost, "model": response_data.get("model", "unknown"), + "response_keys": sorted(response_data.keys()) + if isinstance(response_data, dict) + else None, }, ) return MaxCostData( @@ -139,6 +146,22 @@ async def calculate_cost( # todo: can be sync ) if usd_cost > 0: + if input_tokens == 0 and output_tokens == 0: + logger.warning( + "Upstream reported a USD cost but no token counts — " + "billing the USD-derived cost while the dashboard will " + "show this request as `(0+0)` tokens. Check that the " + "upstream actually emits `usage.input_tokens` and " + "`usage.output_tokens` (OpenAI-compat streams require " + "`stream_options.include_usage=true`).", + extra={ + "model": response_data.get("model", "unknown"), + "usd_cost": usd_cost, + "usage_keys": sorted(usage_data.keys()) + if isinstance(usage_data, dict) + else None, + }, + ) try: sats_per_usd = 1.0 / sats_usd_price() cost_in_sats = usd_cost * sats_per_usd @@ -239,10 +262,18 @@ async def calculate_cost( # todo: can be sync if not (MSATS_PER_1K_OUTPUT_TOKENS and MSATS_PER_1K_INPUT_TOKENS): logger.warning( - "No token pricing configured, using base cost", + "No token pricing configured — billing at flat MaxCostData. " + "Token counts %s in the upstream response but cannot be " + "priced; the request will appear in dashboards with the " + "raw counts and a fixed max-cost charge.", + "are present" + if (input_tokens > 0 or output_tokens > 0) + else "are zero", extra={ "base_cost_msats": max_cost, "model": response_data.get("model", "unknown"), + "input_tokens": input_tokens, + "output_tokens": output_tokens, }, ) return MaxCostData( diff --git a/routstr/upstream/base.py b/routstr/upstream/base.py index 3aca949e..0640efe0 100644 --- a/routstr/upstream/base.py +++ b/routstr/upstream/base.py @@ -351,7 +351,10 @@ class BaseUpstreamProvider: ) -> bytes | None: """Transform request body for provider-specific requirements. - Automatically transforms model names in the request body. + Automatically transforms model names and, for streaming chat + completions, opts the upstream into emitting per-chunk ``usage`` + so cost tracking can read real token counts instead of falling + back to ``MaxCostData``. Args: body: Original request body bytes @@ -364,9 +367,25 @@ class BaseUpstreamProvider: try: data = json.loads(body) - if isinstance(data, dict) and "model" in data: - original_model = model_obj.id - transformed_model = self.transform_model_name(original_model) + except Exception as e: + logger.debug( + "Could not parse request body for transformation", + extra={ + "error": str(e), + "provider": self.provider_type or self.base_url, + }, + ) + return body + + if not isinstance(data, dict): + return body + + changed = False + + if "model" in data: + original_model = model_obj.id + transformed_model = self.transform_model_name(original_model) + if data["model"] != transformed_model: data["model"] = transformed_model logger.debug( "Transformed model name in request", @@ -376,16 +395,28 @@ class BaseUpstreamProvider: "provider": self.provider_type or self.base_url, }, ) - return json.dumps(data).encode() - except Exception as e: - logger.debug( - "Could not transform request body", - extra={ - "error": str(e), - "provider": self.provider_type or self.base_url, - }, - ) + changed = True + # OpenAI-compatible streaming responses omit ``usage`` unless the + # request sets ``stream_options.include_usage = true``. Without it + # we can't reconcile token counts at end of stream and the + # request gets billed at max-cost with zero tokens. Discriminate + # chat-completions-shaped requests by the ``messages`` field so we + # don't poke unrelated endpoints. + if ( + data.get("stream") is True + and "messages" in data + and isinstance(data.get("messages"), list) + ): + existing = data.get("stream_options") + merged = dict(existing) if isinstance(existing, dict) else {} + if merged.get("include_usage") is not True: + merged["include_usage"] = True + data["stream_options"] = merged + changed = True + + if changed: + return json.dumps(data).encode() return body def _extract_upstream_error_message( @@ -1667,6 +1698,20 @@ class BaseUpstreamProvider: nonlocal usage_finalized if usage_finalized: return None + logger.warning( + "Finalizing /v1/messages stream with no usage data — " + "client will be billed at max-cost with zero tokens. " + "Likely cause: upstream omitted `usage` from the SSE " + "stream (check that the request includes " + "`stream_options.include_usage=true` and that the " + "upstream actually emits a final usage chunk).", + extra={ + "key_hash": key.hashed_key[:8] + "...", + "model": last_model_seen or "unknown", + "provider": self.provider_type or self.base_url, + "max_cost_msats": max_cost_for_model, + }, + ) async with create_session() as new_session: fresh_key = await new_session.get( key.__class__, key.hashed_key @@ -1700,8 +1745,12 @@ class BaseUpstreamProvider: ): if annotated.model: last_model_seen = annotated.model - input_tokens += annotated.input_tokens - output_tokens += annotated.output_tokens + # Anthropic SSE reports usage cumulatively across + # message_start + message_delta — take the max snapshot + # rather than summing, otherwise input tokens + # double-count. + input_tokens = max(input_tokens, annotated.input_tokens) + output_tokens = max(output_tokens, annotated.output_tokens) yield annotated.sse_bytes if input_tokens > 0 or output_tokens > 0: @@ -1785,8 +1834,9 @@ class BaseUpstreamProvider: ): if annotated.model: last_model_seen = annotated.model - input_tokens += annotated.input_tokens - output_tokens += annotated.output_tokens + # See _stream_litellm_messages for why this is max() not +=. + input_tokens = max(input_tokens, annotated.input_tokens) + output_tokens = max(output_tokens, annotated.output_tokens) buffered.append(annotated.sse_bytes) response_headers: dict[str, str] = { @@ -1794,6 +1844,20 @@ class BaseUpstreamProvider: "Connection": "keep-alive", } + if input_tokens == 0 and output_tokens == 0: + logger.warning( + "x-cashu /v1/messages stream finished with no usage data " + "— refund cannot be computed and the client effectively " + "pays the full cashu amount. Likely cause: upstream " + "omitted `usage` from the SSE stream.", + extra={ + "model": last_model_seen or "unknown", + "provider": self.provider_type or self.base_url, + "amount": amount, + "unit": unit, + }, + ) + if input_tokens > 0 or output_tokens > 0: response_data: dict = { "model": last_model_seen or "unknown",