fix(pricing): price Venice models from live metadata

agent/model_metadata.py

@@ -953,6 +953,37 @@ def _extract_pricing(payload: Dict[str, Any]) -> Dict[str, Any]:
             pricing["completion"] = str(float(novita_output) / 10_000 / 1_000_000)
         return pricing
 
+    # Venice.ai ships pricing under ``model_spec.pricing.{input,output,...}.usd``
+    # (sometimes top-level ``pricing``) with $/MTok values — nested dicts, not
+    # bare numbers. Convert to per-token strings before the generic alias scan
+    # (which crashes on unhashable dict values).
+    # Example: deepseek-v4-flash → input.usd=0.138, output.usd=0.275, cache_input.usd=0.028
+    model_spec = payload.get("model_spec") if isinstance(payload.get("model_spec"), dict) else {}
+    venice_pricing = None
+    for candidate in (payload.get("pricing"), model_spec.get("pricing") if model_spec else None):
+        if isinstance(candidate, dict) and any(
+            isinstance(candidate.get(k), dict) and candidate.get(k, {}).get("usd") is not None
+            for k in ("input", "output", "cache_input", "cache_read", "cache_write")
+        ):
+            venice_pricing = candidate
+            break
+    if isinstance(venice_pricing, dict):
+        vin = venice_pricing.get("input")
+        vout = venice_pricing.get("output")
+        vcache = venice_pricing.get("cache_input") or venice_pricing.get("cache_read")
+        vwrite = venice_pricing.get("cache_write")
+        result: Dict[str, Any] = {}
+        if isinstance(vin, dict) and vin.get("usd") is not None:
+            result["prompt"] = str(float(vin["usd"]) / 1_000_000)
+        if isinstance(vout, dict) and vout.get("usd") is not None:
+            result["completion"] = str(float(vout["usd"]) / 1_000_000)
+        if isinstance(vcache, dict) and vcache.get("usd") is not None:
+            result["cache_read"] = str(float(vcache["usd"]) / 1_000_000)
+        if isinstance(vwrite, dict) and vwrite.get("usd") is not None:
+            result["cache_write"] = str(float(vwrite["usd"]) / 1_000_000)
+        if result:
+            return result
+
     # DeepInfra ships pricing under ``metadata.pricing`` with $/MTok values:
     # ``input_tokens``, ``output_tokens``, ``cache_read_tokens``. Convert to
     # per-token strings so the generic cost machinery (usage_pricing.py)
@@ -985,8 +1016,15 @@ def _extract_pricing(payload: Dict[str, Any]) -> Dict[str, Any]:
         pricing: Dict[str, Any] = {}
         for target, aliases in alias_map.items():
             for alias in aliases:
-                if alias in normalized and normalized[alias] not in {None, ""}:
-                    pricing[target] = normalized[alias]
+                if alias not in normalized:
+                    continue
+                val = normalized[alias]
+                # Skip nested dicts (e.g. Venice-style pricing objects) — they
+                # are not scalar costs and break set membership tests.
+                if isinstance(val, (dict, list)):
+                    continue
+                if val not in {None, ""}:
+                    pricing[target] = val
                     break
         if pricing:
             return pricing

agent/usage_pricing.py

@@ -1007,6 +1007,17 @@ def resolve_billing_route(
         return BillingRoute(provider="openrouter", model=model, base_url=base_url or "", billing_mode="official_models_api")
     if provider_name == "nous" or base_url_host_matches(base_url or "", "inference-api.nousresearch.com"):
         return BillingRoute(provider="nous", model=model, base_url=base_url or _NOUS_DEFAULT_BASE_URL, billing_mode="official_models_api")
+    # Venice must be detected BEFORE the custom/local short-circuit: Hermes
+    # default profile uses provider=custom with base_url=api.venice.ai, and
+    # Venice publishes live $/MTok rates on GET /models (needs ADMIN or
+    # inference key for the call — pricing is public on the model object).
+    if provider_name == "venice" or base_url_host_matches(base_url or "", "api.venice.ai"):
+        return BillingRoute(
+            provider="venice",
+            model=model.split("/")[-1] if model else "",
+            base_url=base_url or "https://api.venice.ai/api/v1",
+            billing_mode="official_models_api",
+        )
     if provider_name == "anthropic":
         return BillingRoute(provider="anthropic", model=model.split("/")[-1], base_url=base_url or "", billing_mode="official_docs_snapshot")
     # "openai-api" is the picker/registry slug for direct api.openai.com; it

plugins/observability/langfuse/__init__.py

@@ -538,6 +538,25 @@ def _serialize_assistant_message(message: Any) -> dict[str, Any]:
     }
 
 
+def _pricing_api_key(*, provider: str, base_url: str) -> str:
+    """Return a credential only for the provider endpoint it belongs to."""
+    try:
+        from agent.model_metadata import base_url_host_matches
+
+        is_venice = (provider or "").strip().lower() == "venice" or (
+            base_url_host_matches(base_url or "", "api.venice.ai")
+        )
+    except Exception:
+        is_venice = False
+    if not is_venice:
+        return ""
+    return (
+        os.environ.get("VENICE_ADMIN_KEY")
+        or os.environ.get("VENICE_API_KEY")
+        or ""
+    ).strip()
+
+
 def _usage_and_cost(response: Any, *, provider: str, api_mode: str, model: str, base_url: str) -> tuple[dict[str, int], dict[str, float]]:
     usage_details: Dict[str, int] = {}
     cost_details: Dict[str, float] = {}
@@ -566,12 +585,15 @@ def _usage_and_cost(response: Any, *, provider: str, api_mode: str, model: str,
             usage_details["cache_creation_input_tokens"] = canonical.cache_write_tokens
         if canonical.reasoning_tokens:
             usage_details["reasoning_tokens"] = canonical.reasoning_tokens
+        # Venice requires a Bearer key for GET /models. Keep credential
+        # selection host-scoped so a custom endpoint never receives it.
+        pricing_api_key = _pricing_api_key(provider=provider, base_url=base_url)
         cost = estimate_usage_cost(
             model,
             canonical,
             provider=provider,
             base_url=base_url,
-            api_key="",
+            api_key=pricing_api_key,
         )
         if cost.amount_usd is not None:
             # Langfuse cost_details keys must match usage_details keys.
@@ -580,7 +602,12 @@ def _usage_and_cost(response: Any, *, provider: str, api_mode: str, model: str,
                 from agent.usage_pricing import get_pricing_entry
                 from decimal import Decimal
                 _ONE_M = Decimal("1000000")
-                entry = get_pricing_entry(model, provider=provider, base_url=base_url)
+                entry = get_pricing_entry(
+                    model,
+                    provider=provider,
+                    base_url=base_url,
+                    api_key=pricing_api_key,
+                )
                 if entry:
                     if entry.input_cost_per_million is not None and canonical.input_tokens:
                         cost_details["input"] = float(Decimal(canonical.input_tokens) * entry.input_cost_per_million / _ONE_M)
@@ -1000,7 +1027,13 @@ def on_post_llm_call(*, task_id: str = "", session_id: str = "", provider: str =
                 cache_write_tokens=_cache_write,
                 reasoning_tokens=_reasoning,
             )
-            entry = get_pricing_entry(model, provider=provider, base_url=base_url)
+            _pricing_key = _pricing_api_key(provider=provider, base_url=base_url)
+            entry = get_pricing_entry(
+                model,
+                provider=provider,
+                base_url=base_url,
+                api_key=_pricing_key,
+            )
             if entry:
                 if entry.input_cost_per_million is not None and _input:
                     cost_details["input"] = float(Decimal(_input) * entry.input_cost_per_million / _ONE_M)
@@ -1011,7 +1044,13 @@ def on_post_llm_call(*, task_id: str = "", session_id: str = "", provider: str =
                 if entry.cache_write_cost_per_million is not None and _cache_write:
                     cost_details["cache_creation_input_tokens"] = float(Decimal(_cache_write) * entry.cache_write_cost_per_million / _ONE_M)
             else:
-                _cost = estimate_usage_cost(model, _cu, provider=provider, base_url=base_url, api_key="")
+                _cost = estimate_usage_cost(
+                    model,
+                    _cu,
+                    provider=provider,
+                    base_url=base_url,
+                    api_key=_pricing_key,
+                )
                 if _cost.amount_usd is not None:
                     cost_details["total"] = float(_cost.amount_usd)
         except Exception:

tests/agent/test_usage_pricing.py

@@ -193,6 +193,57 @@ def test_openrouter_models_api_pricing_is_converted_from_per_token_to_per_millio
     assert float(entry.cache_write_cost_per_million) == 6.25
 
 
+def test_venice_custom_endpoint_routes_to_live_models_api():
+    route = resolve_billing_route(
+        "venice/deepseek-v4-flash",
+        provider="custom",
+        base_url="https://api.venice.ai/api/v1",
+    )
+
+    assert route.provider == "venice"
+    assert route.model == "deepseek-v4-flash"
+    assert route.base_url == "https://api.venice.ai/api/v1"
+    assert route.billing_mode == "official_models_api"
+
+
+def test_venice_live_pricing_uses_bearer_key_and_nested_usd_rates(monkeypatch):
+    seen = {}
+
+    def fake_fetch(base_url, *, api_key=""):
+        seen["base_url"] = base_url
+        seen["api_key"] = api_key
+        return {
+            "deepseek-v4-flash": {
+                "pricing": {
+                    "prompt": "0.000000138",
+                    "completion": "0.000000275",
+                    "cache_read": "0.000000028",
+                }
+            }
+        }
+
+    monkeypatch.setattr(
+        "agent.usage_pricing.fetch_endpoint_model_metadata",
+        fake_fetch,
+    )
+
+    entry = get_pricing_entry(
+        "deepseek-v4-flash",
+        provider="custom",
+        base_url="https://api.venice.ai/api/v1",
+        api_key="venice-test-key",
+    )
+
+    assert seen == {
+        "base_url": "https://api.venice.ai/api/v1",
+        "api_key": "venice-test-key",
+    }
+    assert entry is not None
+    assert float(entry.input_cost_per_million) == 0.138
+    assert float(entry.output_cost_per_million) == 0.275
+    assert float(entry.cache_read_cost_per_million) == 0.028
+
+
 def test_estimate_usage_cost_marks_subscription_routes_included():
     result = estimate_usage_cost(
         "gpt-5.3-codex",

tests/hermes_cli/test_api_key_providers.py

@@ -1212,6 +1212,36 @@ def test_novita_pricing_unit_conversion(self):
         assert float(result["prompt"]) == 2690 / 10_000 / 1_000_000
         assert float(result["completion"]) == 4000 / 10_000 / 1_000_000
 
+    def test_venice_nested_pricing_unit_conversion(self):
+        """Venice returns nested USD-per-Mtok objects from its models API."""
+        from agent.model_metadata import _extract_pricing
+
+        result = _extract_pricing({
+            "id": "deepseek-v4-flash",
+            "model_spec": {
+                "pricing": {
+                    "input": {"usd": 0.138},
+                    "output": {"usd": 0.275},
+                    "cache_input": {"usd": 0.028},
+                }
+            },
+        })
+
+        assert float(result["prompt"]) == 0.138 / 1_000_000
+        assert float(result["completion"]) == 0.275 / 1_000_000
+        assert float(result["cache_read"]) == 0.028 / 1_000_000
+
+    def test_nested_non_pricing_values_are_ignored(self):
+        """Generic pricing aliases must not crash on nested provider data."""
+        from agent.model_metadata import _extract_pricing
+
+        assert _extract_pricing({
+            "pricing": {
+                "prompt": {"currency": "USD"},
+                "completion": {"currency": "USD"},
+            }
+        }) == {}
+
     def test_novita_pricing_cache(self, monkeypatch):
         """_fetch_novita_pricing should cache results in _pricing_cache."""
         from hermes_cli import models as models_mod

tests/plugins/test_langfuse_plugin.py

@@ -1021,3 +1021,27 @@ class _Resp:
 
         assert seen["resp"] is resp
         assert captured["usage_details"] == {"input": 7, "output": 3}
+
+
+class TestPricingApiKey:
+    def test_venice_host_prefers_admin_key(self, monkeypatch):
+        sys.modules.pop("plugins.observability.langfuse", None)
+        mod = importlib.import_module("plugins.observability.langfuse")
+        monkeypatch.setenv("VENICE_ADMIN_KEY", "admin-test-key")
+        monkeypatch.setenv("VENICE_API_KEY", "inference-test-key")
+
+        assert mod._pricing_api_key(
+            provider="custom",
+            base_url="https://api.venice.ai/api/v1",
+        ) == "admin-test-key"
+
+    def test_custom_non_venice_host_receives_no_provider_key(self, monkeypatch):
+        sys.modules.pop("plugins.observability.langfuse", None)
+        mod = importlib.import_module("plugins.observability.langfuse")
+        monkeypatch.setenv("VENICE_ADMIN_KEY", "must-not-leak")
+        monkeypatch.setenv("OPENAI_API_KEY", "also-must-not-leak")
+
+        assert mod._pricing_api_key(
+            provider="custom",
+            base_url="https://models.example.test/v1",
+        ) == ""