Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 44 additions & 1 deletion agent/usage_pricing.py
Original file line number Diff line number Diff line change
Expand Up @@ -561,6 +561,37 @@ def _to_int(value: Any) -> int:
return 0


# Model-name prefixes whose published pricing is maintained in
# _OFFICIAL_DOCS_PRICING. When a proxied model name starts with one of
# these prefixes we can infer the upstream billing provider and resolve
# to official-docs pricing instead of returning "unknown".
_MODEL_PREFIX_TO_PROVIDER: tuple[tuple[str, str], ...] = (
("claude-", "anthropic"),
("gpt-", "openai"),
("o1-", "openai"),
("o3-", "openai"),
("o4-", "openai"),
("gemini-", "google"),
("deepseek-", "deepseek"),
)


def _infer_upstream_provider(model: str) -> Optional[str]:
"""Infer the upstream billing provider from model name patterns.

When a model is served through a local proxy (LiteLLM, vertex-proxy,
custom gateway) the Hermes provider name is a user-defined alias that
doesn't match any known billing route. This helper checks the model
name against well-known vendor prefixes so the pricing lookup can fall
through to official-docs snapshot entries.
"""
lower = model.lower()
for prefix, provider in _MODEL_PREFIX_TO_PROVIDER:
if lower.startswith(prefix):
return provider
return None


def resolve_billing_route(
model_name: str,
provider: Optional[str] = None,
Expand All @@ -587,8 +618,20 @@ def resolve_billing_route(
return BillingRoute(provider="openai", model=model.split("/")[-1], base_url=base_url or "", billing_mode="official_docs_snapshot")
if provider_name in {"minimax", "minimax-cn"}:
return BillingRoute(provider=provider_name, model=model.split("/")[-1], base_url=base_url or "", billing_mode="official_docs_snapshot")
if provider_name in {"custom", "local"} or (base and "localhost" in base):
if provider_name in {"custom", "local"} or (base and ("localhost" in base or "127.0.0.1" in base)):
# Infer upstream billing provider from model name patterns so
# proxied models (e.g. vertex-proxy, LiteLLM, custom gateways)
# resolve to official-docs pricing instead of "unknown".
inferred = _infer_upstream_provider(model)
if inferred:
return BillingRoute(provider=inferred, model=model, base_url=base_url or "", billing_mode="official_docs_snapshot")
return BillingRoute(provider=provider_name or "custom", model=model, base_url=base_url or "", billing_mode="unknown")
# Final fallback: unrecognized provider -- still attempt model-name
# inference before giving up (covers user-defined provider names like
# "vertex-opus46" that proxy a known upstream model).
inferred = _infer_upstream_provider(model)
if inferred:
return BillingRoute(provider=inferred, model=model, base_url=base_url or "", billing_mode="official_docs_snapshot")
return BillingRoute(provider=provider_name or "unknown", model=model.split("/")[-1] if model else "", base_url=base_url or "", billing_mode="unknown")


Expand Down
117 changes: 117 additions & 0 deletions tests/agent/test_usage_pricing.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,9 +2,11 @@

from agent.usage_pricing import (
CanonicalUsage,
_infer_upstream_provider,
estimate_usage_cost,
get_pricing_entry,
normalize_usage,
resolve_billing_route,
)


Expand Down Expand Up @@ -322,3 +324,118 @@ def test_bedrock_claude_cached_session_estimates_cost_not_unknown():
)
assert result.status == "estimated"
assert result.amount_usd is not None


# ── Proxy model-name inference ───────────────────────────────────────────


class TestInferUpstreamProvider:
"""_infer_upstream_provider maps well-known model prefixes to billing providers."""

def test_claude_models_infer_anthropic(self):
assert _infer_upstream_provider("claude-opus-4.6") == "anthropic"
assert _infer_upstream_provider("claude-sonnet-4") == "anthropic"
assert _infer_upstream_provider("claude-haiku") == "anthropic"
assert _infer_upstream_provider("Claude-Opus-4.6") == "anthropic"

def test_gpt_models_infer_openai(self):
assert _infer_upstream_provider("gpt-4o") == "openai"
assert _infer_upstream_provider("gpt-5.3") == "openai"

def test_o_series_models_infer_openai(self):
assert _infer_upstream_provider("o1-preview") == "openai"
assert _infer_upstream_provider("o3-pro") == "openai"
assert _infer_upstream_provider("o4-mini") == "openai"

def test_gemini_models_infer_google(self):
assert _infer_upstream_provider("gemini-pro") == "google"
assert _infer_upstream_provider("gemini-flash") == "google"

def test_deepseek_models_infer_deepseek(self):
assert _infer_upstream_provider("deepseek-v4-pro") == "deepseek"

def test_unknown_model_returns_none(self):
assert _infer_upstream_provider("my-custom-model") is None
assert _infer_upstream_provider("llama-3.3-70b") is None
assert _infer_upstream_provider("") is None


class TestProxyBillingRoute:
"""Proxied models on localhost resolve to official-docs pricing routes."""

def test_vertex_proxy_claude_resolves_to_anthropic(self):
route = resolve_billing_route(
"claude-opus-4.6",
provider="vertex-opus46",
base_url="http://127.0.0.1:8788/anthropic",
)
assert route.provider == "anthropic"
assert route.model == "claude-opus-4.6"
assert route.billing_mode == "official_docs_snapshot"

def test_litellm_proxy_gpt_resolves_to_openai(self):
route = resolve_billing_route(
"gpt-4o",
provider="litellm",
base_url="http://localhost:4000/v1",
)
assert route.provider == "openai"
assert route.model == "gpt-4o"
assert route.billing_mode == "official_docs_snapshot"

def test_localhost_proxy_unknown_model_stays_unknown(self):
route = resolve_billing_route(
"my-custom-finetune",
provider="custom",
base_url="http://localhost:11434/v1",
)
assert route.provider == "custom"
assert route.billing_mode == "unknown"

def test_canonical_providers_bypass_inference(self):
"""Named providers like 'anthropic' and 'openai' must still use their
direct routing paths, not the inference fallback."""
route = resolve_billing_route(
"claude-sonnet-4", provider="anthropic", base_url=""
)
assert route.provider == "anthropic"
assert route.billing_mode == "official_docs_snapshot"


def test_proxy_claude_pricing_end_to_end():
"""Full pricing path: a proxied Claude model on localhost must resolve
to Anthropic official-docs pricing with real dollar amounts.

This is the user-visible regression: vertex-proxy / LiteLLM users saw
'unknown' cost for every session because the custom provider name
didn't match any billing route.
"""
entry = get_pricing_entry(
"claude-opus-4-6",
provider="vertex-opus46",
base_url="http://127.0.0.1:8788/anthropic",
)
assert entry is not None
assert entry.source == "official_docs_snapshot"
assert float(entry.input_cost_per_million) > 0
assert float(entry.output_cost_per_million) > 0


def test_proxy_claude_cost_estimation_end_to_end():
"""estimate_usage_cost through a proxy must return a dollar amount,
not status='unknown'."""
usage = CanonicalUsage(
input_tokens=1000,
output_tokens=500,
cache_read_tokens=5000,
cache_write_tokens=200,
)
result = estimate_usage_cost(
"claude-sonnet-4-6",
usage,
provider="vertex-sonnet46",
base_url="http://127.0.0.1:8788/anthropic",
)
assert result.status == "estimated"
assert result.amount_usd is not None
assert float(result.amount_usd) > 0