feat: implement Epic 07 Phase 5: versioned cost accounting on usage_events, OTel metric instruments, and span-helper wiring

backend-python/.env.example

@@ -241,3 +241,9 @@ OBSERVABILITY_ENABLED=false
 #   staging/CI:        0.25 (25%)
 #   production:        0.05 (5%)
 # OTEL_TRACES_SAMPLE_RATIO=1.0
+
+# Cost accounting (honoured only when OBSERVABILITY_ENABLED=true).
+# Rates live in git-tracked config/model_pricing.yaml — not env-overridable.
+# OBSERVABILITY_COST_PRICING_FILE=config/model_pricing.yaml
+# Must match pricing_version in the pricing file at startup.
+# OBSERVABILITY_COST_PRICING_VERSION=2026-08

backend-python/alembic/versions/0008_observability_usage_cost.py

@@ -0,0 +1,43 @@
+"""0008 observability usage cost
+
+Revision ID: 0008_observability_usage_cost
+Revises: 0007_workflow_tables
+Create Date: 2026-08-08
+
+Epic 07 Phase 5: additive ``cost_usd`` / ``pricing_version`` on ``usage_events``.
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+revision: str = "0008_observability_usage_cost"
+down_revision: Union[str, None] = "0007_workflow_tables"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "usage_events",
+        sa.Column("cost_usd", sa.Numeric(precision=12, scale=6), nullable=True),
+    )
+    op.add_column(
+        "usage_events",
+        sa.Column("pricing_version", sa.Text(), nullable=True),
+    )
+    op.create_index(
+        "ix_usage_events_provider_model_created",
+        "usage_events",
+        ["provider", "model", "created_at"],
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(
+        "ix_usage_events_provider_model_created",
+        table_name="usage_events",
+    )
+    op.drop_column("usage_events", "pricing_version")
+    op.drop_column("usage_events", "cost_usd")

backend-python/app/ai/observability/__init__.py

@@ -3,6 +3,8 @@
 See ``docs/plans/post-mvp-v2-epic-07-observability-and-evaluation.md`` Part I § Public APIs.
 """
 
+from app.ai.observability.cost.calculator import CostCalculator
+from app.ai.observability.cost.pricing import ModelPricingTable
 from app.ai.observability.exceptions import (
     ObservabilityConfigError,
     ObservabilityDisabledError,
@@ -28,7 +30,9 @@
 
 __all__ = [
     "OBSERVABILITY_ENABLED",
+    "CostCalculator",
     "MeterRegistry",
+    "ModelPricingTable",
     "ObservabilityConfigError",
     "ObservabilityDisabledError",
     "ObservabilityError",

backend-python/app/ai/observability/cost/__init__.py

@@ -0,0 +1,11 @@
+"""Cost accounting public API."""
+
+from app.ai.observability.cost.calculator import CostCalculator, CostRegistry
+from app.ai.observability.cost.pricing import ModelPricingEntry, ModelPricingTable
+
+__all__ = [
+    "CostCalculator",
+    "CostRegistry",
+    "ModelPricingEntry",
+    "ModelPricingTable",
+]

backend-python/app/ai/observability/cost/calculator.py

@@ -1,5 +1,87 @@
-"""Approximate token cost calculator — implemented in Phase 5."""
+"""Approximate token cost calculator — invoked only from ``SqlUsageStore.record()``."""
 
 from __future__ import annotations
 
-# TODO(epic-07): Phase 5 — CostCalculator.price(provider, model, usage).
+from app.ai.observability.cost.pricing import ModelPricingTable
+from app.ai.observability.exceptions import ObservabilityConfigError
+from app.core.config import Settings
+from app.core.logging import get_logger
+from app.providers.base import ProviderUsage
+
+logger = get_logger(__name__)
+
+
+class CostCalculator:
+    """Convert ``ProviderUsage`` into approximate USD cost using a pricing table."""
+
+    def __init__(self, table: ModelPricingTable) -> None:
+        self._table = table
+
+    @property
+    def pricing_version(self) -> str:
+        return self._table.pricing_version
+
+    @property
+    def pricing_table(self) -> ModelPricingTable:
+        return self._table
+
+    def price(
+        self,
+        provider: str,
+        model: str,
+        usage: ProviderUsage | None,
+    ) -> tuple[float | None, str | None]:
+        if usage is None:
+            return None, None
+        if usage.prompt_tokens is None or usage.completion_tokens is None:
+            return None, None
+
+        entry = self._table.lookup(provider, model)
+        if entry is None:
+            return None, None
+
+        cost = (usage.prompt_tokens / 1000.0) * entry.input_usd_per_1k + (
+            usage.completion_tokens / 1000.0
+        ) * entry.output_usd_per_1k
+        return cost, self._table.pricing_version
+
+
+class CostRegistry:
+    """Process-wide ``CostCalculator`` accessor (real or unset when disabled)."""
+
+    _calculator: CostCalculator | None = None
+    _initialized = False
+
+    @classmethod
+    def initialize(cls, settings: Settings) -> None:
+        if cls._initialized:
+            return
+
+        cls._initialized = True
+        if not settings.observability_enabled:
+            cls._calculator = None
+            return
+
+        try:
+            table = ModelPricingTable.load(settings)
+        except ObservabilityConfigError:
+            raise
+        except Exception as exc:
+            raise ObservabilityConfigError(
+                f"Failed to load model pricing table: {exc}"
+            ) from exc
+
+        cls._calculator = CostCalculator(table)
+
+    @classmethod
+    def get_calculator(cls) -> CostCalculator | None:
+        return cls._calculator
+
+    @classmethod
+    def is_enabled(cls) -> bool:
+        return cls._calculator is not None
+
+    @classmethod
+    def reset_for_tests(cls) -> None:
+        cls._initialized = False
+        cls._calculator = None

backend-python/app/ai/observability/cost/pricing.py

@@ -1,5 +1,138 @@
-"""Model pricing table loader — implemented in Phase 5."""
+"""Model pricing table loader — git-tracked ``config/model_pricing.yaml``."""
 
 from __future__ import annotations
 
-# TODO(epic-07): Phase 5 — ModelPricingTable loads config/model_pricing.yaml.
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+import yaml
+
+from app.ai.observability.exceptions import ObservabilityConfigError
+from app.core.config import Settings
+
+_BACKEND_ROOT = Path(__file__).resolve().parents[4]
+
+
+@dataclass(frozen=True)
+class ModelPricingEntry:
+    provider: str
+    model: str
+    input_usd_per_1k: float
+    output_usd_per_1k: float
+
+
+class ModelPricingTable:
+    """Version-locked per-(provider, model) token rates loaded at startup."""
+
+    def __init__(
+        self,
+        *,
+        pricing_version: str,
+        entries: dict[tuple[str, str], ModelPricingEntry],
+    ) -> None:
+        self._pricing_version = pricing_version
+        self._entries = entries
+
+    @property
+    def pricing_version(self) -> str:
+        return self._pricing_version
+
+    @property
+    def model_registry(self) -> frozenset[str]:
+        return frozenset(entry.model for entry in self._entries.values())
+
+    def lookup(self, provider: str, model: str) -> ModelPricingEntry | None:
+        return self._entries.get((provider, model))
+
+    @classmethod
+    def load(
+        cls, settings: Settings, *, pricing_file: Path | None = None
+    ) -> ModelPricingTable:
+        path = pricing_file or (
+            _BACKEND_ROOT / settings.observability_cost_pricing_file
+        )
+        if not path.is_file():
+            raise ObservabilityConfigError(f"Model pricing file not found: {path}")
+
+        raw = yaml.safe_load(path.read_text(encoding="utf-8"))
+        if not isinstance(raw, dict):
+            raise ObservabilityConfigError(
+                "Model pricing file must contain a YAML mapping at the top level."
+            )
+
+        file_version = raw.get("pricing_version")
+        if not isinstance(file_version, str) or not file_version.strip():
+            raise ObservabilityConfigError(
+                "Model pricing file requires a non-empty pricing_version."
+            )
+        if file_version != settings.observability_cost_pricing_version:
+            raise ObservabilityConfigError(
+                "Model pricing version mismatch: "
+                f"file has {file_version!r}, "
+                f"settings.observability_cost_pricing_version is "
+                f"{settings.observability_cost_pricing_version!r}."
+            )
+
+        models_raw = raw.get("models")
+        if not isinstance(models_raw, list) or not models_raw:
+            raise ObservabilityConfigError(
+                "Model pricing file requires a non-empty models list."
+            )
+
+        entries: dict[tuple[str, str], ModelPricingEntry] = {}
+        for index, item in enumerate(models_raw):
+            entry = cls._parse_entry(item, index=index)
+            key = (entry.provider, entry.model)
+            if key in entries:
+                raise ObservabilityConfigError(
+                    f"Duplicate pricing entry for provider={entry.provider!r}, "
+                    f"model={entry.model!r}."
+                )
+            entries[key] = entry
+
+        return cls(pricing_version=file_version, entries=entries)
+
+    @staticmethod
+    def _parse_entry(raw: Any, *, index: int) -> ModelPricingEntry:
+        if not isinstance(raw, dict):
+            raise ObservabilityConfigError(f"models[{index}] must be a mapping.")
+
+        provider = raw.get("provider")
+        model = raw.get("model")
+        input_rate = raw.get("input_usd_per_1k")
+        output_rate = raw.get("output_usd_per_1k")
+
+        if not isinstance(provider, str) or not provider.strip():
+            raise ObservabilityConfigError(
+                f"models[{index}].provider must be a non-empty string."
+            )
+        if not isinstance(model, str) or not model.strip():
+            raise ObservabilityConfigError(
+                f"models[{index}].model must be a non-empty string."
+            )
+
+        validated_input_rate = input_rate
+        validated_output_rate = output_rate
+        for field_name, value in (
+            ("input_usd_per_1k", validated_input_rate),
+            ("output_usd_per_1k", validated_output_rate),
+        ):
+            if not isinstance(value, (int, float)) or isinstance(value, bool):
+                raise ObservabilityConfigError(
+                    f"models[{index}].{field_name} must be a finite number >= 0."
+                )
+            if value < 0:
+                raise ObservabilityConfigError(
+                    f"models[{index}].{field_name} must be >= 0."
+                )
+
+        assert isinstance(validated_input_rate, (int, float))
+        assert isinstance(validated_output_rate, (int, float))
+
+        return ModelPricingEntry(
+            provider=provider.strip(),
+            model=model.strip(),
+            input_usd_per_1k=float(validated_input_rate),
+            output_usd_per_1k=float(validated_output_rate),
+        )