diff --git a/docs/advanced/configuration.mdx b/docs/advanced/configuration.mdx
index 2048eefe65..84eb612646 100644
--- a/docs/advanced/configuration.mdx
+++ b/docs/advanced/configuration.mdx
@@ -19,6 +19,22 @@ 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 for `STRIX_LLM`. Use it for gateway or self-hosted model ids that LiteLLM can't price.
+
+
+
+ Price per 1M output tokens in USD for `STRIX_LLM`.
+
+
+
+ Price per 1M cached input tokens in USD for `STRIX_LLM`. If unset, cached tokens use the input price.
+
+
+
+ Setting any of these prices replaces the provider-reported and LiteLLM cost for `STRIX_LLM` in reports and `--max-budget`. An unset input or output price counts as `0`, so set both. A separate `STRIX_DEDUPE_MODEL` keeps its LiteLLM 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/config/settings.py b/strix/config/settings.py
index eb61a793d4..4217469026 100644
--- a/strix/config/settings.py
+++ b/strix/config/settings.py
@@ -61,6 +61,15 @@ class LlmSettings(BaseSettings):
repr=False,
)
reasoning_effort: ReasoningEffort = Field(default="high", alias="STRIX_REASONING_EFFORT")
+ input_cost_per_mtok: float | None = Field(
+ default=None, ge=0, allow_inf_nan=False, alias="STRIX_LLM_INPUT_COST_PER_MTOK"
+ )
+ output_cost_per_mtok: float | None = Field(
+ default=None, ge=0, allow_inf_nan=False, alias="STRIX_LLM_OUTPUT_COST_PER_MTOK"
+ )
+ cached_input_cost_per_mtok: float | None = Field(
+ default=None, ge=0, allow_inf_nan=False, alias="STRIX_LLM_CACHED_INPUT_COST_PER_MTOK"
+ )
force_required_tool_choice: bool = Field(
default=False,
alias="STRIX_FORCE_REQUIRED_TOOL_CHOICE",
diff --git a/strix/report/usage.py b/strix/report/usage.py
index bb20b254d4..3f3f10d016 100644
--- a/strix/report/usage.py
+++ b/strix/report/usage.py
@@ -8,11 +8,38 @@
from agents.usage import Usage, deserialize_usage, serialize_usage
from pydantic import BaseModel, TypeAdapter, ValidationError
+from strix.config.loader import load_settings
from strix.report.pricing import resolve_litellm_model
logger = logging.getLogger(__name__)
+Prices = tuple[float | None, float | None, float | None]
+
+
+def _configured_prices() -> tuple[str | None, Prices | None]:
+ llm = load_settings().llm
+ prices = (llm.input_cost_per_mtok, llm.output_cost_per_mtok, llm.cached_input_cost_per_mtok)
+ if all(price is None for price in prices):
+ return None, None
+ return (llm.model or "").strip() or None, prices
+
+
+def _configured_usage_cost(usage: Usage, prices: Prices) -> 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."""
@@ -37,6 +64,7 @@ def __init__(self) -> None:
self._total_usage = Usage()
self._agent_usage: dict[str, Usage] = {}
self._agent_metadata: dict[str, dict[str, str]] = {}
+ self._agent_estimated_cost: dict[str, float] = {}
self._observed_cost = 0.0
self._estimated_cost = 0.0
self._has_observed_cost = False
@@ -73,6 +101,9 @@ def record(
estimated = _estimate_litellm_cost(usage, model)
if estimated:
self._estimated_cost += estimated
+ self._agent_estimated_cost[normalized_agent_id] = (
+ self._agent_estimated_cost.get(normalized_agent_id, 0.0) + estimated
+ )
return True
@@ -114,11 +145,23 @@ def record_provider(
def total_cost(self) -> float:
if self.zero_cost:
return 0.0
+ model, prices = _configured_prices()
+ if prices is not None:
+ return _round_cost(
+ sum(self._configured_agent_cost(aid, model, prices) for aid in self._agent_usage)
+ )
return _round_cost(self._observed_cost if self._has_observed_cost else self._estimated_cost)
+ def _configured_agent_cost(self, agent_id: str, model: str | None, prices: Prices) -> float:
+ if self._agent_metadata.get(agent_id, {}).get("model") == model:
+ return _configured_usage_cost(self._agent_usage[agent_id], prices)
+ return self._agent_estimated_cost.get(agent_id, 0.0)
+
def to_record(self) -> dict[str, Any]:
record = serialize_usage(self._total_usage)
- record["cost"] = self.total_cost
+ model, prices = _configured_prices()
+ total_cost = self.total_cost
+ record["cost"] = total_cost
record["providers"] = {name: tally.model_dump() for name, tally in self._providers.items()}
record["agents"] = []
@@ -128,7 +171,11 @@ def to_record(self) -> dict[str, Any]:
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
+ self._configured_agent_cost(agent_id, model, prices)
+ if prices is not None 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)
@@ -148,6 +195,7 @@ def hydrate(self, raw_usage: Any) -> None:
self._total_usage = Usage()
self._agent_usage.clear()
self._agent_metadata.clear()
+ self._agent_estimated_cost.clear()
self._observed_cost = 0.0
self._estimated_cost = 0.0
self._has_observed_cost = False
@@ -183,6 +231,7 @@ def hydrate(self, raw_usage: Any) -> None:
logger.exception("Failed to hydrate llm_usage for agent %s", agent_id)
self._agent_usage[agent_id] = Usage()
+ self._agent_estimated_cost[agent_id] = _float_or_zero(raw_agent.get("cost"))
metadata: dict[str, str] = {}
agent_name = raw_agent.get("agent_name")
model = raw_agent.get("model")
diff --git a/tests/test_hooks.py b/tests/test_hooks.py
index d19c4b250b..b13b2f310b 100644
--- a/tests/test_hooks.py
+++ b/tests/test_hooks.py
@@ -2,11 +2,14 @@
from __future__ import annotations
-from typing import Any
+from types import SimpleNamespace
+from typing import TYPE_CHECKING, Any
from unittest.mock import MagicMock, patch
import pytest
+from agents.usage import Usage
+from strix.config import loader
from strix.core.agents import AgentCoordinator
from strix.core.hooks import (
BudgetExceededError,
@@ -15,6 +18,11 @@
SubagentBudgetReservedError,
recomputed_budget_flags,
)
+from strix.report.usage import LLMUsageLedger
+
+
+if TYPE_CHECKING:
+ from pathlib import Path
def _make_hooks(max_budget: float | None) -> ReportUsageHooks:
@@ -85,6 +93,38 @@ 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, tmp_path: Path
+) -> None:
+ monkeypatch.setenv("STRIX_LLM", "test-model")
+ 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)
+ monkeypatch.setattr(loader, "_override", tmp_path / "missing.json")
+ monkeypatch.setattr(loader, "_cached", None)
+ 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 8916982d30..ee1130e664 100644
--- a/tests/test_pricing.py
+++ b/tests/test_pricing.py
@@ -1,14 +1,50 @@
from __future__ import annotations
+import json
+from typing import TYPE_CHECKING
from unittest.mock import patch
import litellm
+import pytest
from agents.usage import Usage
+from pydantic import ValidationError
+from strix.config import loader
+from strix.config.settings import LlmSettings
from strix.report.pricing import resolve_litellm_model
from strix.report.usage import LLMUsageLedger
+if TYPE_CHECKING:
+ from pathlib import Path
+
+
+PRICE_ENV = (
+ "STRIX_LLM_INPUT_COST_PER_MTOK",
+ "STRIX_LLM_OUTPUT_COST_PER_MTOK",
+ "STRIX_LLM_CACHED_INPUT_COST_PER_MTOK",
+)
+
+
+@pytest.fixture(autouse=True)
+def _isolated_settings(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+ for name in PRICE_ENV:
+ monkeypatch.delenv(name, raising=False)
+ monkeypatch.setenv("STRIX_LLM", "gateway/unpriced-model")
+ monkeypatch.setattr(loader, "_override", tmp_path / "missing.json")
+ monkeypatch.setattr(loader, "_cached", None)
+
+
+def _usage(input_tokens: int, output_tokens: int = 0, cached_tokens: int = 0) -> Usage:
+ usage = Usage()
+ usage.requests = 1
+ usage.input_tokens = input_tokens
+ usage.output_tokens = output_tokens
+ usage.total_tokens = input_tokens + output_tokens
+ usage.input_tokens_details.cached_tokens = cached_tokens
+ return usage
+
+
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 +96,112 @@ def test_ledger_prefers_observed_cost_over_estimate() -> None:
assert ledger.total_cost == 0.17
+def test_unpriced_model_without_manual_rates_costs_zero() -> None:
+ ledger = LLMUsageLedger()
+
+ ledger.record(agent_id="a", usage=_usage(1000, 200), 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")
+ ledger = LLMUsageLedger()
+
+ ledger.record(agent_id="a", usage=_usage(1000, 200), 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_manual_rates_from_config_file(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+ path = tmp_path / "config.json"
+ path.write_text(
+ json.dumps(
+ {
+ "env": {
+ "STRIX_LLM_INPUT_COST_PER_MTOK": "2",
+ "STRIX_LLM_OUTPUT_COST_PER_MTOK": "8",
+ }
+ }
+ ),
+ encoding="utf-8",
+ )
+ monkeypatch.setattr(loader, "_override", path)
+ ledger = LLMUsageLedger()
+
+ ledger.record(agent_id="a", usage=_usage(1000, 200), model="gateway/unpriced-model")
+
+ assert ledger.total_cost == 0.0036
+
+
+def test_manual_rates_only_price_the_configured_model(monkeypatch: pytest.MonkeyPatch) -> None:
+ monkeypatch.setenv("STRIX_LLM_INPUT_COST_PER_MTOK", "2")
+ monkeypatch.setenv("STRIX_LLM_OUTPUT_COST_PER_MTOK", "8")
+ ledger = LLMUsageLedger()
+
+ ledger.record(agent_id="a", usage=_usage(1000, 200), model="gateway/unpriced-model")
+ with patch("litellm.completion_cost", return_value=0.42):
+ ledger.record(agent_id="dedupe", usage=_usage(500, 50), model="deepseek/deepseek-v4-flash")
+ record = ledger.to_record()
+
+ assert ledger.total_cost == 0.4236
+ assert {agent["agent_id"]: agent["cost"] for agent in record["agents"]} == {
+ "a": 0.0036,
+ "dedupe": 0.42,
+ }
+
+
+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")
+ ledger = LLMUsageLedger()
+
+ ledger.record(
+ agent_id="a", usage=_usage(10, 1, cached_tokens=2), 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")
+ original = LLMUsageLedger()
+ original.record(agent_id="a", usage=_usage(1000, 200), model="gateway/unpriced-model")
+ persisted = original.to_record()
+ persisted["cost"] = 42.0
+ persisted["agents"][0]["cost"] = 42.0
+ hydrated = LLMUsageLedger()
+
+ hydrated.hydrate(persisted)
+
+ assert hydrated.total_cost == 0.0036
+
+
+@pytest.mark.parametrize("value", ["invalid", "-1", "nan", "inf"])
+def test_invalid_manual_rate_is_rejected(value: str, monkeypatch: pytest.MonkeyPatch) -> None:
+ monkeypatch.setenv("STRIX_LLM_INPUT_COST_PER_MTOK", value)
+ with pytest.raises(ValidationError):
+ LlmSettings()
+
+
+def test_zero_cost_overrides_manual_rates(monkeypatch: pytest.MonkeyPatch) -> None:
+ monkeypatch.setenv("STRIX_LLM_INPUT_COST_PER_MTOK", "10")
+ ledger = LLMUsageLedger()
+ ledger.zero_cost = True
+
+ ledger.record(agent_id="a", usage=_usage(1000), model="gateway/unpriced-model")
+
+ assert ledger.total_cost == 0.0
+
+
def test_hydrated_estimate_continues_accumulating_new_estimates() -> None:
usage = Usage()
usage.requests = 1