fix(STR-1654): price LLM usage from model config

docs/advanced/configuration.mdx

@@ -19,6 +19,18 @@ Configure Strix using environment variables or a config file.
   Custom API base URL. Also accepts `OPENAI_API_BASE`, `LITELLM_BASE_URL`, or `OLLAMA_API_BASE`.
 </ParamField>
 
+<ParamField path="STRIX_LLM_INPUT_COST_PER_MTOK" type="number">
+  Price per 1M input tokens in USD. When set, Strix computes cost and `--max-budget` from these prices instead of the provider-reported or LiteLLM cost. Use it for gateway or self-hosted model ids that LiteLLM can't price.
+</ParamField>
+
+<ParamField path="STRIX_LLM_OUTPUT_COST_PER_MTOK" type="number">
+  Price per 1M output tokens in USD. When set, Strix computes cost and `--max-budget` from these prices instead of the provider-reported or LiteLLM cost. Use it for gateway or self-hosted model ids that LiteLLM can't price.
+</ParamField>
+
+<ParamField path="STRIX_LLM_CACHED_INPUT_COST_PER_MTOK" type="number">
+  Price per 1M cached input tokens in USD. When set, Strix computes cost and `--max-budget` from these prices instead of the provider-reported or LiteLLM cost. If unset, cached tokens use the input-token price.
+</ParamField>
+
 <ParamField path="STRIX_API_TYPE" type="string">
   Select the OpenAI API path for the model: `responses` or `chat_completions`.
   By default the model decides: Responses when LiteLLM's model catalog lists

strix/report/usage.py

@@ -3,6 +3,8 @@
 from __future__ import annotations
 
 import logging
+import math
+import os
 from typing import Any
 
 from agents.usage import Usage, deserialize_usage, serialize_usage
@@ -13,6 +15,53 @@ from strix.report.pricing import resolve_litellm_model
 
 logger = logging.getLogger(__name__)
 
+_PRICE_ENV = (
+    "STRIX_LLM_INPUT_COST_PER_MTOK",
+    "STRIX_LLM_OUTPUT_COST_PER_MTOK",
+    "STRIX_LLM_CACHED_INPUT_COST_PER_MTOK",
+)
+_warned_invalid_prices: set[str] = set()
+
+
+def _configured_prices() -> tuple[float | None, float | None, float | None]:
+    prices: list[float | None] = []
+    for name in _PRICE_ENV:
+        raw = os.environ.get(name)
+        if raw is None or raw == "":
+            prices.append(None)
+            continue
+        try:
+            price = float(raw)
+        except ValueError:
+            price = math.nan
+        if not math.isfinite(price) or price < 0:
+            if name not in _warned_invalid_prices:
+                logger.warning("Ignoring invalid %s; expected a finite non-negative rate", name)
+                _warned_invalid_prices.add(name)
+            prices.append(None)
+            continue
+        prices.append(price)
+    return prices[0], prices[1], prices[2]
+
+
+def _configured_usage_cost(
+    usage: Usage,
+    prices: tuple[float | None, float | None, float | None],
+) -> float:
+    input_rate, output_rate, cached_rate = prices
+    input_tokens = max(0, _int_or_zero(getattr(usage, "input_tokens", 0)))
+    cached_details = getattr(usage, "input_tokens_details", None)
+    cached_tokens = max(0, _int_or_zero(getattr(cached_details, "cached_tokens", 0)))
+    cached_tokens = min(cached_tokens, input_tokens)
+    output_tokens = max(0, _int_or_zero(getattr(usage, "output_tokens", 0)))
+    billable_input = input_tokens - cached_tokens
+    cost = (
+        billable_input * (input_rate or 0)
+        + cached_tokens * (cached_rate if cached_rate is not None else input_rate or 0)
+        + output_tokens * (output_rate or 0)
+    ) / 1_000_000
+    return _round_cost(cost)
+
 
 class ProviderUsage(BaseModel):
     """Running totals for one upstream provider OpenRouter routed calls to."""
@@ -114,11 +163,23 @@ class LLMUsageLedger:
     def total_cost(self) -> float:
         if self.zero_cost:
             return 0.0
+        prices = _configured_prices()
+        if any(price is not None for price in prices):
+            return _configured_usage_cost(self._total_usage, prices)
         return _round_cost(self._observed_cost if self._has_observed_cost else self._estimated_cost)
 
     def to_record(self) -> dict[str, Any]:
         record = serialize_usage(self._total_usage)
-        record["cost"] = self.total_cost
+        prices = _configured_prices()
+        configured = any(price is not None for price in prices)
+        total_cost = (
+            0.0
+            if self.zero_cost
+            else _configured_usage_cost(self._total_usage, prices)
+            if configured
+            else self.total_cost
+        )
+        record["cost"] = total_cost
         record["providers"] = {name: tally.model_dump() for name, tally in self._providers.items()}
         record["agents"] = []
 
@@ -128,7 +189,11 @@ class LLMUsageLedger:
             usage = self._agent_usage[agent_id]
             metadata = self._agent_metadata.get(agent_id, {})
             agent_cost = (
-                self.total_cost * (agent_tokens[agent_id] / total_tokens) if total_tokens else 0.0
+                _configured_usage_cost(usage, prices)
+                if configured and not self.zero_cost
+                else total_cost * (agent_tokens[agent_id] / total_tokens)
+                if total_tokens
+                else 0.0
             )
 
             agent_record = serialize_usage(usage)

tests/test_hooks.py

@@ -2,10 +2,12 @@
 
 from __future__ import annotations
 
+from types import SimpleNamespace
 from typing import Any
 from unittest.mock import MagicMock, patch
 
 import pytest
+from agents.usage import Usage
 
 from strix.core.agents import AgentCoordinator
 from strix.core.hooks import (
@@ -15,6 +17,7 @@ from strix.core.hooks import (
     SubagentBudgetReservedError,
     recomputed_budget_flags,
 )
+from strix.report.usage import LLMUsageLedger
 
 
 def _make_hooks(max_budget: float | None) -> ReportUsageHooks:
@@ -85,6 +88,35 @@ async def test_over_budget_raises() -> None:
         await hooks.on_llm_end(_make_context(), MagicMock(), MagicMock())
 
 
+@pytest.mark.asyncio
+async def test_manual_rates_make_unpriced_model_trip_budget(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("STRIX_LLM_INPUT_COST_PER_MTOK", "10")
+    monkeypatch.delenv("STRIX_LLM_OUTPUT_COST_PER_MTOK", raising=False)
+    monkeypatch.delenv("STRIX_LLM_CACHED_INPUT_COST_PER_MTOK", raising=False)
+    usage = Usage()
+    usage.requests = 1
+    usage.input_tokens = 1000
+    usage.total_tokens = 1000
+    ledger = LLMUsageLedger()
+
+    class ReportState:
+        def record_sdk_usage(self, **kwargs: Any) -> None:
+            ledger.record(**kwargs)
+
+        def get_total_llm_cost(self) -> float:
+            return ledger.total_cost
+
+    hooks = _make_hooks(0.005)
+    response: Any = SimpleNamespace(usage=usage)
+    with (
+        patch("strix.core.hooks.get_global_report_state", return_value=ReportState()),
+        pytest.raises(BudgetExceededError),
+    ):
+        await hooks.on_llm_end(_make_context(), MagicMock(), response)
+
+
 @pytest.mark.asyncio
 async def test_budget_check_uses_live_cost_accessor() -> None:
     # The check must read the live ledger, not the persisted run-record snapshot,

tests/test_pricing.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
 from unittest.mock import patch
 
 import litellm
@@ -9,6 +10,10 @@ from strix.report.pricing import resolve_litellm_model
 from strix.report.usage import LLMUsageLedger
 
 
+if TYPE_CHECKING:
+    import pytest
+
+
 def test_resolves_common_bare_model_names() -> None:
     resolve_litellm_model.cache_clear()
     assert resolve_litellm_model("deepseek-v4-flash") == "deepseek/deepseek-v4-flash"
@@ -60,6 +65,117 @@ def test_ledger_prefers_observed_cost_over_estimate() -> None:
     assert ledger.total_cost == 0.17
 
 
+def test_unpriced_model_without_manual_rates_costs_zero(monkeypatch: pytest.MonkeyPatch) -> None:
+    for name in (
+        "STRIX_LLM_INPUT_COST_PER_MTOK",
+        "STRIX_LLM_OUTPUT_COST_PER_MTOK",
+        "STRIX_LLM_CACHED_INPUT_COST_PER_MTOK",
+    ):
+        monkeypatch.delenv(name, raising=False)
+    usage = Usage()
+    usage.requests = 1
+    usage.input_tokens = 1000
+    usage.output_tokens = 200
+    usage.total_tokens = 1200
+    ledger = LLMUsageLedger()
+
+    ledger.record(agent_id="a", usage=usage, model="gateway/unpriced-model")
+
+    assert ledger.total_cost == 0.0
+
+
+def test_manual_rates_price_usage_and_override_observed_cost(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("STRIX_LLM_INPUT_COST_PER_MTOK", "2")
+    monkeypatch.setenv("STRIX_LLM_OUTPUT_COST_PER_MTOK", "8")
+    monkeypatch.delenv("STRIX_LLM_CACHED_INPUT_COST_PER_MTOK", raising=False)
+    usage = Usage()
+    usage.requests = 1
+    usage.input_tokens = 1000
+    usage.output_tokens = 200
+    usage.total_tokens = 1200
+    ledger = LLMUsageLedger()
+
+    ledger.record(agent_id="a", usage=usage, model="gateway/unpriced-model")
+    ledger.record_observed_cost(0.17)
+
+    assert ledger.total_cost == 0.0036
+    assert ledger.to_record()["agents"][0]["cost"] == 0.0036
+
+
+def test_cached_tokens_fall_back_to_input_rate(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("STRIX_LLM_INPUT_COST_PER_MTOK", "5")
+    monkeypatch.setenv("STRIX_LLM_OUTPUT_COST_PER_MTOK", "7")
+    monkeypatch.delenv("STRIX_LLM_CACHED_INPUT_COST_PER_MTOK", raising=False)
+    usage = Usage()
+    usage.requests = 1
+    usage.input_tokens = 10
+    usage.output_tokens = 1
+    usage.total_tokens = 11
+    usage.input_tokens_details.cached_tokens = 2
+    ledger = LLMUsageLedger()
+
+    ledger.record(agent_id="a", usage=usage, model="gateway/unpriced-model")
+
+    assert ledger.total_cost == 0.000057
+
+
+def test_hydrated_usage_uses_configured_rates_not_persisted_cost(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("STRIX_LLM_INPUT_COST_PER_MTOK", "2")
+    monkeypatch.setenv("STRIX_LLM_OUTPUT_COST_PER_MTOK", "8")
+    usage = Usage()
+    usage.requests = 1
+    usage.input_tokens = 1000
+    usage.output_tokens = 200
+    usage.total_tokens = 1200
+    original = LLMUsageLedger()
+    original.record(agent_id="a", usage=usage, model="gateway/unpriced-model")
+    persisted = original.to_record()
+    persisted["cost"] = 42.0
+    hydrated = LLMUsageLedger()
+
+    hydrated.hydrate(persisted)
+
+    assert hydrated.total_cost == 0.0036
+
+
+def test_invalid_manual_rate_is_ignored(
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    monkeypatch.setenv("STRIX_LLM_INPUT_COST_PER_MTOK", "invalid")
+    monkeypatch.setenv("STRIX_LLM_OUTPUT_COST_PER_MTOK", "10")
+    monkeypatch.delenv("STRIX_LLM_CACHED_INPUT_COST_PER_MTOK", raising=False)
+    usage = Usage()
+    usage.requests = 1
+    usage.input_tokens = 1000
+    usage.output_tokens = 200
+    usage.total_tokens = 1200
+    ledger = LLMUsageLedger()
+
+    ledger.record(agent_id="a", usage=usage, model="gateway/unpriced-model")
+
+    assert ledger.total_cost == 0.002
+    assert "Ignoring invalid STRIX_LLM_INPUT_COST_PER_MTOK" in caplog.text
+
+
+def test_zero_cost_overrides_manual_rates(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("STRIX_LLM_INPUT_COST_PER_MTOK", "10")
+    usage = Usage()
+    usage.requests = 1
+    usage.input_tokens = 1000
+    usage.total_tokens = 1000
+    ledger = LLMUsageLedger()
+    ledger.zero_cost = True
+
+    ledger.record(agent_id="a", usage=usage, model="gateway/unpriced-model")
+
+    assert ledger.total_cost == 0.0
+
+
 def test_hydrated_estimate_continues_accumulating_new_estimates() -> None:
     usage = Usage()
     usage.requests = 1