diff --git a/docs/advanced/configuration.mdx b/docs/advanced/configuration.mdx
index dfd28d23f..76d434ebc 100644
--- a/docs/advanced/configuration.mdx
+++ b/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`.
+
+ 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.
+
+
+
+ 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.
+
+
+
+ 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.
+
+
Select the OpenAI API path for the model: `responses` or `chat_completions`.
By default the model decides: Responses when LiteLLM's model catalog lists
diff --git a/strix/report/usage.py b/strix/report/usage.py
index bb20b254d..4e03944b1 100644
--- a/strix/report/usage.py
+++ b/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)
diff --git a/tests/test_hooks.py b/tests/test_hooks.py
index d19c4b250..9fca882e4 100644
--- a/tests/test_hooks.py
+++ b/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,
diff --git a/tests/test_pricing.py b/tests/test_pricing.py
index 8916982d3..f27d8a9a7 100644
--- a/tests/test_pricing.py
+++ b/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