feat(core): price table cost computation (ev-afm)

regokan · regokan · commit ebe082e31396 · 2026-05-12T14:55:39.000+10:00
Adds `eval_harness/core/price_tables.py` with a versioned, dated
`PriceTable` model and a built-in `DEFAULT_PRICE_TABLE` covering ~5
current-generation models (verified 2026-05-12 against Anthropic and
OpenAI public pricing). The runner uses it to fill
`Trace.metrics.cost_usd` whenever an adapter reports token counts but
no $ figure, so downstream cost rollups stay accurate across adapters
that don't compute cost themselves.

- `MetricsConfig` (`eval.yaml &gt; metrics.price_table_path`): optional
  user override; YAML loaded and validated against `PriceTable`.
  Path is resolved relative to the eval.yaml.
- `compute_cost()` returns `None` for unknown models — caller leaves
  `cost_usd` as `None`, no silent zero.
- Thinking tokens are priced separately (`thinking_per_million_tokens`)
  for providers that bill extended-reasoning tokens at a distinct rate.
- Runner emits a single `logging.warning` at startup when the default
  table is in use, noting its `freshness_date`. Per-cell fills are
  `logging.debug`.

Tests cover the spec list (default loads, correct compute, unknown
model -&gt; None, user override, thinking tokens), plus runner-level
checks that cost_usd is filled when missing and untouched when the
adapter already reported it.
diff --git a/docs/ConfigSchema.md b/docs/ConfigSchema.md
@@ -75,6 +75,7 @@ output        list of TraceStores; first one is canonical, others are mirrors
 | `run.retry.backoff_seconds` | float | no | Exponential base. |
 | `run.baseline_variant` | string | no | Used by ComparisonReport in `summary.yaml`. |
 | `run.cost_limit_usd` | float | no | Run-level cost guardrail. When accumulated `trace.metrics.cost_usd` across completed cells reaches this value, queued cells are short-circuited with a `cost_limit` Trace. Independent from and additive to per-evaluator `cost_limit_usd`. |
+| `metrics.price_table_path` | string | no | YAML file describing per-model token prices. The runner fills `trace.metrics.cost_usd` from this table when an adapter reports token counts but no $ figure. Path is resolved relative to the `eval.yaml` location. When omitted, the runner uses a versioned built-in `DEFAULT_PRICE_TABLE` and emits one `logging.warning` per run noting its `freshness_date` so the staleness is visible. The model name is read from `systems[].metadata.model` (with `systems[].model` as a fallback). |
 | `output[]` | list[dict] | yes | At least one TraceStore. Single mapping is accepted and coerced to a one-element list. |
 | `output[].type` | enum | yes | One of the registered TraceStores. |
 | `output[].path` | string | type-dependent | Required for `local_files`. |
diff --git a/eval_harness/core/__init__.py b/eval_harness/core/__init__.py
@@ -5,6 +5,7 @@
     EvalConfig,
     EvalIdentity,
     EvaluatorConfig,
+    MetricsConfig,
     OutputConfig,
     PassCriteria,
     RetryPolicy,
@@ -58,6 +59,7 @@
     "FileEntry",
     "FileManifest",
     "FilesystemArtifact",
+    "MetricsConfig",
     "OutputConfig",
     "PassCriteria",
     "Registry",
diff --git a/eval_harness/core/config.py b/eval_harness/core/config.py
@@ -78,6 +78,11 @@ class OutputConfig(BaseModel):
     path: str | None = None
 
 
+class MetricsConfig(BaseModel):
+    model_config = _FORBID
+    price_table_path: str | None = None
+
+
 class EvalConfig(BaseModel):
     model_config = _FORBID
     schema_version: str = "1.0"
@@ -90,6 +95,7 @@ class EvalConfig(BaseModel):
     evaluators: list[EvaluatorConfig]
     pass_criteria: PassCriteria = Field(default_factory=PassCriteria)
     run: RunOptions = Field(default_factory=RunOptions)
+    metrics: MetricsConfig = Field(default_factory=MetricsConfig)
     output: list[OutputConfig]
 
     @model_validator(mode="before")
diff --git a/eval_harness/core/price_tables.py b/eval_harness/core/price_tables.py
@@ -0,0 +1,149 @@
+"""Per-model price tables and cost computation.
+
+The runner uses this to fill `Trace.metrics.cost_usd` when an adapter reported
+token counts but no $ figure. Ships `DEFAULT_PRICE_TABLE` with a small set of
+current-generation models and a `freshness_date`; the runner emits a single
+warning per run when the default is in use so the staleness is visible. Users
+override via `eval.yaml > metrics.price_table_path`.
+"""
+
+from __future__ import annotations
+
+import logging
+from datetime import date
+from pathlib import Path
+
+import yaml
+from pydantic import BaseModel, ConfigDict, Field
+
+from eval_harness.core.errors import ConfigError
+
+logger = logging.getLogger(__name__)
+
+_FORBID = ConfigDict(extra="forbid")
+
+
+class ModelPrice(BaseModel):
+    """Per-million-token pricing for one model.
+
+    Thinking tokens (extended-thinking / reasoning) are billed separately on
+    some providers; default to 0.0 when a model has no thinking surcharge.
+    """
+
+    model_config = _FORBID
+    input_per_million_tokens: float
+    output_per_million_tokens: float
+    thinking_per_million_tokens: float = 0.0
+
+
+class PriceTable(BaseModel):
+    """Versioned, dated price table.
+
+    `freshness_date` is the day prices were verified against provider
+    documentation. The runner warns once per run when `DEFAULT_PRICE_TABLE` is
+    in use so users know the cost figures may be stale.
+    """
+
+    model_config = _FORBID
+    table_version: str
+    freshness_date: date
+    models: dict[str, ModelPrice] = Field(default_factory=dict)
+
+
+# Sources verified 2026-05-12 against provider docs:
+#   - Anthropic: https://www.anthropic.com/pricing#anthropic-api
+#   - OpenAI:    https://openai.com/api/pricing/
+# Prices are USD per million tokens. Thinking-token pricing covers extended
+# reasoning where the provider bills a distinct rate (e.g. Anthropic 1M
+# extended thinking pricing matches output rate at time of capture).
+DEFAULT_PRICE_TABLE = PriceTable(
+    table_version="2026-05-12",
+    freshness_date=date(2026, 5, 12),
+    models={
+        "claude-opus-4-7": ModelPrice(
+            input_per_million_tokens=15.0,
+            output_per_million_tokens=75.0,
+            thinking_per_million_tokens=75.0,
+        ),
+        "claude-sonnet-4-6": ModelPrice(
+            input_per_million_tokens=3.0,
+            output_per_million_tokens=15.0,
+            thinking_per_million_tokens=15.0,
+        ),
+        "claude-haiku-4-5-20251001": ModelPrice(
+            input_per_million_tokens=0.25,
+            output_per_million_tokens=1.25,
+        ),
+        "claude-4-7": ModelPrice(
+            input_per_million_tokens=3.0,
+            output_per_million_tokens=15.0,
+            thinking_per_million_tokens=15.0,
+        ),
+        "gpt-5": ModelPrice(
+            input_per_million_tokens=5.0,
+            output_per_million_tokens=15.0,
+        ),
+    },
+)
+
+
+def load_price_table(path: Path | None) -> PriceTable:
+    """Load a price table from YAML; `None` returns `DEFAULT_PRICE_TABLE`."""
+    if path is None:
+        return DEFAULT_PRICE_TABLE
+    if not path.exists():
+        raise ConfigError(f"price_table_path does not exist: {path}")
+    try:
+        data = yaml.safe_load(path.read_text())
+    except yaml.YAMLError as e:
+        raise ConfigError(f"price_table_path is not valid YAML ({path}): {e}") from e
+    if not isinstance(data, dict):
+        raise ConfigError(
+            f"price_table_path must be a YAML mapping ({path}); got {type(data).__name__}"
+        )
+    return PriceTable.model_validate(data)
+
+
+def compute_cost(
+    table: PriceTable,
+    model: str,
+    token_input: int,
+    token_output: int,
+    token_thinking: int = 0,
+) -> float | None:
+    """Compute the call's $ cost from token counts. Returns `None` when the
+    model is not in the table — callers handle the gap (e.g. leave
+    `Trace.metrics.cost_usd` as `None`)."""
+    price = table.models.get(model)
+    if price is None:
+        return None
+    return (
+        (token_input / 1_000_000.0) * price.input_per_million_tokens
+        + (token_output / 1_000_000.0) * price.output_per_million_tokens
+        + (token_thinking / 1_000_000.0) * price.thinking_per_million_tokens
+    )
+
+
+def warn_default_table_in_use(table: PriceTable) -> None:
+    """Emit a single warning when `DEFAULT_PRICE_TABLE` is the active table.
+
+    Idempotent: the runner calls this once at startup. Kept as a function so
+    other entry points (e.g. ad-hoc tooling) can opt-in to the same warning.
+    """
+    if table is DEFAULT_PRICE_TABLE:
+        logger.warning(
+            "Using DEFAULT_PRICE_TABLE (freshness_date=%s, table_version=%s). "
+            "Prices may be stale. Override via eval.yaml > metrics.price_table_path.",
+            table.freshness_date.isoformat(),
+            table.table_version,
+        )
+
+
+__all__ = [
+    "DEFAULT_PRICE_TABLE",
+    "ModelPrice",
+    "PriceTable",
+    "compute_cost",
+    "load_price_table",
+    "warn_default_table_in_use",
+]
diff --git a/eval_harness/runner/plan_builder.py b/eval_harness/runner/plan_builder.py
@@ -8,6 +8,7 @@
 from eval_harness.core.config import EvalConfig, RetryPolicy, SystemConfig
 from eval_harness.core.errors import ConfigError
 from eval_harness.core.models import EvalCase, RunVariant
+from eval_harness.core.price_tables import PriceTable, load_price_table
 from eval_harness.core.time import make_run_id
 from eval_harness.factories.dataset_adapter_factory import DatasetAdapterFactory
 from eval_harness.factories.evaluator_factory import EvaluatorFactory
@@ -37,6 +38,7 @@ class RunPlan:
     evaluators: list[Evaluator]
     retry_policy: RetryPolicy
     baseline_variant: str | None
+    price_table: PriceTable | None = None
     # Optional whitelist of `(case_id, variant_name)` cells. When set, run_eval
     # executes only these specific cells (instead of the full cases x variants
     # product). Used by `evalh run --retry-only-failed` to amend an existing
@@ -120,6 +122,8 @@ async def build_plan(
             f"defined: {sorted(known_variants)}"
         )
 
+    price_table = _build_price_table(config, config_path)
+
     return RunPlan(
         config=config,
         run_id=run_id,
@@ -132,9 +136,20 @@ async def build_plan(
         evaluators=evaluators,
         retry_policy=config.run.retry,
         baseline_variant=baseline,
+        price_table=price_table,
     )
 
 
+def _build_price_table(config: EvalConfig, config_path: Path) -> PriceTable:
+    raw = config.metrics.price_table_path
+    if raw is None:
+        return load_price_table(None)
+    path = Path(raw)
+    if not path.is_absolute():
+        path = (config_path.parent / path).resolve()
+    return load_price_table(path)
+
+
 def _system_extras(sys_cfg: SystemConfig) -> dict[str, object]:
     dumped = sys_cfg.model_dump()
     return {k: v for k, v in dumped.items() if k not in _STRUCTURAL_KEYS}
diff --git a/eval_harness/runner/run_eval.py b/eval_harness/runner/run_eval.py
@@ -2,6 +2,7 @@
 
 import asyncio
 import itertools
+import logging
 from contextlib import AsyncExitStack, suppress
 from dataclasses import dataclass
 from typing import TYPE_CHECKING
@@ -14,10 +15,17 @@
     Trace,
     TraceError,
 )
+from eval_harness.core.price_tables import (
+    PriceTable,
+    compute_cost,
+    warn_default_table_in_use,
+)
 from eval_harness.core.time import utc_now
 from eval_harness.runner.cost_accumulator import CostAccumulator
 from eval_harness.runner.summary import SummaryAggregator
 
+logger = logging.getLogger(__name__)
+
 if TYPE_CHECKING:
     from eval_harness.adapters.workspace.base import Workspace
     from eval_harness.evaluators.base import Evaluator
@@ -37,6 +45,9 @@ async def run_eval(plan: RunPlan) -> RunSummary:
     accumulator = CostAccumulator()
     cost_limit = plan.config.run.cost_limit_usd
     aggregator = SummaryAggregator(plan=plan)
+    if plan.price_table is not None:
+        warn_default_table_in_use(plan.price_table)
+    variant_models = _build_variant_model_index(plan.variants)
 
     async with AsyncExitStack() as stack:
         await stack.enter_async_context(plan.trace_store)
@@ -68,6 +79,9 @@ async def run_cell(case: EvalCase, variant: RunVariant) -> None:
                     )
                 else:
                     outcome = await _run_one(case, variant, plan)
+                    _fill_cost_from_price_table(
+                        outcome.trace, variant_models.get(variant.name), plan.price_table
+                    )
                     accumulator.tally(outcome.trace)
                 # Stream the outcome into the aggregator immediately and let
                 # it fall out of scope. This is the streaming-summary
@@ -87,6 +101,55 @@ async def run_cell(case: EvalCase, variant: RunVariant) -> None:
     raise RuntimeError("unreachable: AsyncExitStack never re-raises")
 
 
+def _build_variant_model_index(variants: list[RunVariant]) -> dict[str, str]:
+    """Map variant name -> declared model. Looks in `metadata.model` first
+    (the canonical place per docs/ConfigSchema.md), then `config.model` for
+    adapters that nest it there. Variants without a declared model are
+    omitted; the runner just doesn't fill cost for them."""
+    out: dict[str, str] = {}
+    for v in variants:
+        raw = v.metadata.get("model") or v.config.get("model")
+        if isinstance(raw, str) and raw:
+            out[v.name] = raw
+    return out
+
+
+def _fill_cost_from_price_table(
+    trace: Trace, model: str | None, table: PriceTable | None
+) -> None:
+    """If the adapter didn't fill `cost_usd` but did report token counts, use
+    the price table. No-op when prices are unavailable for the model."""
+    if table is None or model is None:
+        return
+    metrics = trace.metrics
+    if metrics.cost_usd is not None:
+        return
+    token_input = metrics.token_input or 0
+    token_output = metrics.token_output or 0
+    token_thinking = metrics.token_thinking or 0
+    if token_input == 0 and token_output == 0 and token_thinking == 0:
+        return
+    cost = compute_cost(table, model, token_input, token_output, token_thinking)
+    if cost is None:
+        logger.debug(
+            "price_table: no entry for model %r (variant %r); cost_usd left None",
+            model,
+            trace.variant_name,
+        )
+        return
+    metrics.cost_usd = cost
+    logger.debug(
+        "price_table: filled cost_usd=%.6f for variant=%r model=%r "
+        "(in=%d, out=%d, thinking=%d)",
+        cost,
+        trace.variant_name,
+        model,
+        token_input,
+        token_output,
+        token_thinking,
+    )
+
+
 async def _short_circuit_cost_limit(
     case: EvalCase,
     variant: RunVariant,
diff --git a/tests/unit/test_core/test_price_tables.py b/tests/unit/test_core/test_price_tables.py
diff --git a/tests/unit/test_runner.py b/tests/unit/test_runner.py
