diff --git a/app/models/usage_record.py b/app/models/usage_record.py
index 37df57e..30e3dff 100644
--- a/app/models/usage_record.py
+++ b/app/models/usage_record.py
@@ -62,6 +62,14 @@ class UsageRecord(db.Model):
     #: Estimated cost in the configured display currency. ``None`` means the
     #: provider/model has no known price; it is never silently reported as zero.
     cost_estimate = db.Column(db.Float, nullable=True)
+    #: Correlation id shared with the request log (``X-Request-ID``). Nullable
+    #: because rows can be written by background/CLI work with no request.
+    request_id = db.Column(db.String(64), nullable=True, index=True)
+    #: End-to-end latency of the instrumented operation, in milliseconds.
+    latency_ms = db.Column(db.Integer, nullable=True)
+    #: Outcome of the operation: ``success``, ``error`` or ``denied``. Unknown
+    #: values are preserved so readers can group historical rows safely.
+    status = db.Column(db.String(20), nullable=True, index=True)
     created_at = db.Column(
         db.DateTime(timezone=True),
         nullable=False,
@@ -85,6 +93,9 @@ def to_dict(self) -> dict:
             "total_tokens": self.total_tokens,
             "tokens_estimated": bool(self.tokens_estimated),
             "cost_estimate": self.cost_estimate,
+            "request_id": self.request_id,
+            "latency_ms": self.latency_ms,
+            "status": self.status,
             "created_at": self.created_at.isoformat() if self.created_at else None,
         }
 
diff --git a/app/reviews/routes.py b/app/reviews/routes.py
index 9dfabb5..af99e69 100644
--- a/app/reviews/routes.py
+++ b/app/reviews/routes.py
@@ -38,6 +38,7 @@
 )
 from app.models.review_finding import CATEGORIES_BY_KIND, SEVERITIES
 from app.reviews import bp
+from app.services import chat_audit, usage_ledger
 from app.services import metrics as metrics_service
 from app.services import reviews as reviews_service
 from app.services.events import emit_event
@@ -416,6 +417,30 @@ def _save_result(review: Review, result: dict, config: dict) -> None:
     else:
         review.status = STATUS_COMPLETED
         review.error_message = None
+
+    # One ledger row per review run (success or failure), carrying the project
+    # scope, token counts, provider/model, request id, latency and outcome
+    # (issue #430, extended by #428).
+    usage = result.get("usage")
+    if not isinstance(usage, dict):
+        usage = {}
+    project = db.session.get(Project, review.project_id) if review.project_id else None
+    usage_ledger.record_usage(
+        source=usage_ledger.SOURCE_REVIEW,
+        prompt_tokens=usage.get("prompt_tokens", 0),
+        completion_tokens=usage.get("completion_tokens", 0),
+        total_tokens=usage.get("total_tokens"),
+        tokens_estimated=usage.get("tokens_estimated", True),
+        provider=usage.get("provider"),
+        model=usage.get("model"),
+        user_id=review.user_id,
+        project=project,
+        workspace_id=project.workspace_id if project else None,
+        request_id=chat_audit.current_request_id() or f"review:{review.id}",
+        latency_ms=usage.get("latency_ms"),
+        status=usage_ledger.STATUS_ERROR if error else usage_ledger.STATUS_SUCCESS,
+        commit=False,
+    )
     db.session.commit()
 
 
diff --git a/app/services/budget_alerts.py b/app/services/budget_alerts.py
new file mode 100644
index 0000000..ffa4969
--- /dev/null
+++ b/app/services/budget_alerts.py
@@ -0,0 +1,59 @@
+"""Budget threshold alert evaluation (Phase 10, issue #471).
+
+The ledger records usage; this module decides when a workspace's spend crosses
+its budget limit. Alerts are deduplicated per workspace and period so a burst of
+usage produces one notification instead of one per request. The in-process
+deduplicator is intentionally small and resettable (tests and multi-process
+workers each keep their own view); a persistent dedupe store can replace it
+without changing the public :func:`check_budget_alerts` contract.
+"""
+
+from __future__ import annotations
+
+from app.services.token_usage import BudgetAlertDeduplicator
+
+#: Alert kind emitted when spend is above the configured limit.
+ALERT_OVER_BUDGET = "over_budget"
+
+_deduplicator = BudgetAlertDeduplicator()
+
+
+def check_budget_alerts(workspace_id, *, spend, limit, period) -> list[dict]:
+    """Return alerts for ``workspace_id`` for the given budget ``period``.
+
+    ``spend`` and ``limit`` are compared as numbers. A missing, zero or negative
+    limit means no budget is configured, so no alert is emitted. The same alert
+    (workspace + period + kind) fires at most once per period; starting a new
+    period resets deduplication through
+    :class:`~app.services.token_usage.BudgetAlertDeduplicator`.
+    """
+    try:
+        spend_value = float(spend or 0)
+        limit_value = float(limit or 0)
+    except (TypeError, ValueError):
+        return []
+    if limit_value <= 0 or spend_value <= limit_value:
+        return []
+
+    period_key = str(period or "")
+    if not _deduplicator.check_and_mark((workspace_id, period_key), ALERT_OVER_BUDGET):
+        return []
+    return [
+        {
+            "workspace_id": workspace_id,
+            "kind": ALERT_OVER_BUDGET,
+            "period": period_key,
+            "spend": spend_value,
+            "limit": limit_value,
+            "over_by": round(spend_value - limit_value, 6),
+        }
+    ]
+
+
+def reset() -> None:
+    """Forget deduplication state (used between tests/periods)."""
+    global _deduplicator
+    _deduplicator = BudgetAlertDeduplicator()
+
+
+__all__ = ["ALERT_OVER_BUDGET", "check_budget_alerts", "reset"]
diff --git a/app/services/email.py b/app/services/email.py
index eb37cf8..a0724c9 100644
--- a/app/services/email.py
+++ b/app/services/email.py
@@ -1,14 +1,15 @@
 """Transactional email service.
 
-Sends plain-text + simple HTML email over SMTS using environment config. Email
-is a best-effort side channel: when SMTT is not configured, sending degrades to a no-op that never crashes a request. The in-app notification (see
-``app/services/notifications.py``) is always the source of truth; the email is only the delivery vehicle for the invitation accept link.
+Sends plain-text + simple HTML email over SMTP using environment config. Email
+is a best-effort side channel: when SMTP is not configured, sending degrades to
+a no-op that never crashes a request. The in-app notification (see
+``app/services/notifications.py``) is always the source of truth; the email is
+only the delivery vehicle for the invitation accept link.
 """
 
 from __future__ import annotations
 
 import logging
-
 import smtplib
 from email.mime.multipart import MIMEMultipart
 from email.mime.text import MIMEText
diff --git a/app/services/plugin_deps.py b/app/services/plugin_deps.py
index 0ea060a..3be9c3f 100644
--- a/app/services/plugin_deps.py
+++ b/app/services/plugin_deps.py
@@ -28,12 +28,15 @@
 
 from __future__ import annotations
 
+import contextlib
 import logging
 from collections.abc import Callable
 from dataclasses import dataclass, field
 from importlib import metadata
 
 from packaging.requirements import InvalidRequirement, Requirement
+from packaging.utils import canonicalize_name
+from packaging.version import InvalidVersion, Version
 
 from app.services.plugin_versions import InvalidPluginVersionError, parse_version
 
@@ -69,6 +72,10 @@ class PluginDependencyUnparsableError(PluginDependencyError):
     """Raised when a declared dependency is not a valid PEP 508 requirement."""
 
 
+class PluginDependencyConflictError(PluginDependencyError):
+    """Raised when no single version can satisfy the resolved graph's ranges."""
+
+
 @dataclass(frozen=True)
 class PluginRef:
     """Normalized plugin record consumed by the resolver.
@@ -94,6 +101,56 @@ def __str__(self) -> str:
 PackageVersion = Callable[[str], str]
 
 
+def _candidate_versions(specifier_sets) -> set[Version]:
+    """Return boundary versions worth probing for a satisfiable intersection.
+
+    PEP 440 does not expose set intersection directly, so conflicts are detected
+    by probing the versions named in the constraints (plus the versions either
+    side of every boundary). For ordinary ranges this is exact: if none of the
+    probes satisfies every specifier, no version can.
+    """
+    candidates = {Version("0"), Version("1")}
+    for specifier_set in specifier_sets:
+        for specifier in specifier_set:
+            raw = specifier.version
+            base = raw[:-2] if raw.endswith(".*") else raw
+            try:
+                version = Version(base)
+            except InvalidVersion:
+                continue
+            candidates.add(version)
+            for suffix in (".post0.dev0", ".post0", ".post1", ".dev0"):
+                with contextlib.suppress(InvalidVersion):
+                    candidates.add(Version(f"{base}{suffix}"))
+            release = version.release
+            for index in range(len(release)):
+                higher = (
+                    *release[:index],
+                    release[index] + 1,
+                    *(0 for _ in release[index + 1 :]),
+                )
+                candidates.add(Version(".".join(str(part) for part in higher)))
+                if release[index] > 0:
+                    lower = (
+                        *release[:index],
+                        release[index] - 1,
+                        *(0 for _ in release[index + 1 :]),
+                    )
+                    candidates.add(Version(".".join(str(part) for part in lower)))
+    return candidates
+
+
+def _specifiers_conflict(specifier_sets) -> bool:
+    """Return ``True`` when the specifier sets share no satisfying version."""
+    constrained = [specifier_set for specifier_set in specifier_sets if specifier_set]
+    if len(constrained) < 2:
+        return False
+    return not any(
+        all(specifier_set.contains(candidate, prereleases=True) for specifier_set in constrained)
+        for candidate in _candidate_versions(constrained)
+    )
+
+
 class DependencyResolver:
     """Resolve and validate a plugin's declared dependencies.
 
@@ -116,19 +173,70 @@ def resolve(self, plugin: PluginRef) -> None:
         """Verify ``plugin`` and all of its transitive dependencies.
 
         Raises a :class:`PluginDependencyError` subclass when any required
-        dependency is missing, disabled, version-incompatible, or cyclic. The
-        plugin's state is never modified here; callers decide how to react.
+        dependency is missing, disabled, version-incompatible, or cyclic. When
+        two paths in the graph constrain the same dependency to disjoint
+        ranges, :class:`PluginDependencyConflictError` names both paths so the
+        install can be refused before anything changes. The plugin's state is
+        never modified here; callers decide how to react.
         """
-        self._check_plugin(plugin, ())
+        constraints: dict[str, list[tuple[Requirement, tuple[str, ...]]]] = {}
+        failures: list[tuple[str, Requirement, str]] = []
+        self._check_plugin(plugin, (), constraints, failures)
+        if failures:
+            root_id, requirement, installed = failures[0]
+            raise PluginDependencyVersionError(
+                root_id,
+                f"dependency '{requirement}' is not satisfied by "
+                f"installed version {installed}.",
+            )
         logger.debug("Plugin dependencies resolved for %s", plugin.id)
 
     # ---------------- internals ----------------
 
-    def _check_plugin(self, plugin: PluginRef, visiting: tuple[str, ...]) -> None:
+    def _register_constraint(
+        self,
+        constraints: dict[str, list[tuple[Requirement, tuple[str, ...]]]],
+        requirement: Requirement,
+        path: tuple[str, ...],
+    ) -> None:
+        """Record a version constraint and reject a disjoint intersection.
+
+        ``path`` is the chain of plugin ids from the requested root to the
+        plugin that declared this requirement, so a conflict can report exactly
+        which paths disagree.
+        """
+        if not requirement.specifier:
+            return
+        key = canonicalize_name(requirement.name)
+        entries = constraints.setdefault(key, [])
+        entries.append((requirement, path))
+        if len(entries) < 2:
+            return
+        if not _specifiers_conflict([entry[0].specifier for entry in entries]):
+            return
+        details = "; ".join(
+            f"{entry_path[-1]} requires '{entry_requirement}' " f"(path: {' -> '.join(entry_path)})"
+            for entry_requirement, entry_path in entries
+        )
+        raise PluginDependencyConflictError(
+            path[0],
+            f"version conflict for '{requirement.name}': {details}",
+        )
+
+    def _check_plugin(
+        self,
+        plugin: PluginRef,
+        visiting: tuple[str, ...],
+        constraints: dict[str, list[tuple[Requirement, tuple[str, ...]]]],
+        failures: list[tuple[str, Requirement, str]],
+    ) -> None:
         """Recursively validate ``plugin``'s dependencies.
 
         ``visiting`` holds the plugin ids on the current path from the root
         (used for cycle detection); a tuple keeps it immutable and cheap.
+        Version mismatches are collected and reported after the whole graph has
+        been walked, so a conflict between two branches is reported as a
+        conflict rather than as whichever branch happened to fail first.
         """
         if plugin.id in visiting:
             cycle = (*visiting, plugin.id)
@@ -137,7 +245,7 @@ def _check_plugin(self, plugin: PluginRef, visiting: tuple[str, ...]) -> None:
                 "dependency cycle detected: " + " -> ".join(cycle),
             )
 
-        visiting = (*visiting, plugin.id)
+        path = (*visiting, plugin.id)
         for raw in plugin.dependencies:
             try:
                 requirement = Requirement(raw.strip())
@@ -150,11 +258,14 @@ def _check_plugin(self, plugin: PluginRef, visiting: tuple[str, ...]) -> None:
                 # Environment markers (e.g. ``sys_platform == "win32"``) that do
                 # not apply to this environment are ignored.
                 continue
+            self._register_constraint(constraints, requirement, path)
             dependency = self._plugin_provider(requirement.name)
             if dependency is not None:
-                self._check_plugin_dependency(plugin.id, requirement, dependency, visiting)
+                self._check_plugin_dependency(
+                    path[0], requirement, dependency, path, constraints, failures
+                )
             else:
-                self._check_package_dependency(plugin.id, requirement)
+                self._check_package_dependency(path[0], requirement, failures)
 
     def _check_plugin_dependency(
         self,
@@ -162,6 +273,8 @@ def _check_plugin_dependency(
         requirement: Requirement,
         dependency: PluginRef,
         visiting: tuple[str, ...],
+        constraints: dict[str, list[tuple[Requirement, tuple[str, ...]]]],
+        failures: list[tuple[str, Requirement, str]],
     ) -> None:
         """Validate a dependency that refers to a registered plugin."""
         if dependency.id in visiting:
@@ -182,17 +295,16 @@ def _check_plugin_dependency(
         if requirement.specifier and not requirement.specifier.contains(
             dependency.version, prereleases=True
         ):
-            raise PluginDependencyVersionError(
-                root_id,
-                f"dependency '{requirement}' is not satisfied by "
-                f"installed version {dependency.version}.",
-            )
-        self._check_plugin(dependency, visiting)
+            failures.append((root_id, requirement, dependency.version))
+        # Keep walking even on a version mismatch so constraints declared below
+        # this dependency can still participate in conflict detection.
+        self._check_plugin(dependency, visiting, constraints, failures)
 
     def _check_package_dependency(
         self,
         root_id: str,
         requirement: Requirement,
+        failures: list[tuple[str, Requirement, str]],
     ) -> None:
         """Validate a dependency that refers to an installed Python package."""
         try:
@@ -214,8 +326,4 @@ def _check_package_dependency(
 
         specifier = requirement.specifier
         if specifier and not specifier.contains(version, prereleases=True):
-            raise PluginDependencyVersionError(
-                root_id,
-                f"dependency '{requirement}' is not satisfied by "
-                f"installed version {installed}.",
-            )
+            failures.append((root_id, requirement, installed))
diff --git a/app/services/pricing.py b/app/services/pricing.py
index 1233966..3afbc3f 100644
--- a/app/services/pricing.py
+++ b/app/services/pricing.py
@@ -10,8 +10,14 @@
 never calls a vendor and therefore always costs zero.
 
 Lookups are provider/model exact matches after normalizing case and
-surrounding whitespace. An unknown provider or model returns ``None`` rather
-than guessing, so callers can distinguish "free" from "we do not know".
+surrounding whitespace. A ``*`` wildcard may be used for the provider and/or
+model so an operator table can define a fallback rate; an exact entry always
+takes precedence over a wildcard. An unknown provider or model returns ``None``
+rather than guessing, so callers can distinguish "free" from "we do not know".
+
+Token counts passed to :func:`estimate_cost_usd` are clamped to
+:data:`MAX_TOKEN_COUNT` so an implausibly large (or malicious) count cannot
+produce an infinite estimate. Negative counts are still rejected.
 """
 
 from __future__ import annotations
@@ -19,6 +25,14 @@
 from dataclasses import asdict, dataclass
 from typing import Any
 
+#: Wildcard key understood by :func:`get_model_price`.
+WILDCARD = "*"
+
+#: Upper bound applied to each token count before costing. A completion with
+#: more tokens than this is almost certainly a bug or hostile input; clamping
+#: keeps the estimate finite without silently accepting an unbounded value.
+MAX_TOKEN_COUNT = 1_000_000_000
+
 #: Shown next to the table in docs/UI wherever costs are displayed.
 ESTIMATE_DISCLAIMER = (
     "Default prices are estimates in USD per 1,000 tokens. "
@@ -93,34 +107,64 @@ def _normalize(value: str | None) -> str:
     return (value or "").strip().lower()
 
 
-def get_model_price(provider: str | None, model: str | None) -> ModelPrice | None:
-    """Return the default price entry for ``provider``/``model``, or ``None``.
-
-    Matching is case-insensitive and ignores surrounding whitespace. Unknown
-    combinations return ``None`` (cost unknown) rather than assuming zero.
-    """
-    return DEFAULT_PRICE_TABLE.get((_normalize(provider), _normalize(model)))
+def _normalized_table(table=None) -> dict[tuple[str, str], ModelPrice]:
+    """Return ``table`` (default: the shipped table) with normalized keys."""
+    source = DEFAULT_PRICE_TABLE if table is None else table
+    return {
+        (_normalize(provider), _normalize(model)): price
+        for (provider, model), price in source.items()
+    }
 
 
-def list_model_prices(provider: str | None = None) -> list[ModelPrice]:
-    """Return the default price entries, optionally filtered by ``provider``.
+def get_model_price(
+    provider: str | None,
+    model: str | None,
+    *,
+    table=None,
+) -> ModelPrice | None:
+    """Return the price entry for ``provider``/``model``, or ``None``.
+
+    Matching is case-insensitive and ignores surrounding whitespace. An exact
+    provider/model entry always wins over a fallback, then the following
+    wildcard order is tried: ``(provider, "*")``, ``("*", model)`` and finally
+    ``("*", "*")``. Unknown combinations return ``None`` (cost unknown) rather
+    than assuming zero.
+    """
+    normalized = _normalized_table(table)
+    provider_key = _normalize(provider)
+    model_key = _normalize(model)
+    for key in (
+        (provider_key, model_key),
+        (provider_key, WILDCARD),
+        (WILDCARD, model_key),
+        (WILDCARD, WILDCARD),
+    ):
+        price = normalized.get(key)
+        if price is not None:
+            return price
+    return None
+
+
+def list_model_prices(provider: str | None = None, *, table=None) -> list[ModelPrice]:
+    """Return the price entries, optionally filtered by ``provider``.
 
     Results are sorted by provider then model so the table (and any API built
     on it) is stable across runs.
     """
     wanted = _normalize(provider)
+    normalized = _normalized_table(table)
     entries = [
         price
-        for (entry_provider, _), price in DEFAULT_PRICE_TABLE.items()
+        for (entry_provider, _), price in normalized.items()
         if not wanted or entry_provider == wanted
     ]
     return sorted(entries, key=lambda price: (price.provider, price.model))
 
 
-def price_table() -> dict[str, list[dict[str, Any]]]:
-    """Return the whole default table grouped by provider (JSON-friendly)."""
+def price_table(*, table=None) -> dict[str, list[dict[str, Any]]]:
+    """Return the whole table grouped by provider (JSON-friendly)."""
     grouped: dict[str, list[dict[str, Any]]] = {}
-    for price in list_model_prices():
+    for price in list_model_prices(table=table):
         grouped.setdefault(price.provider, []).append(price.to_dict())
     return grouped
 
@@ -131,18 +175,23 @@ def estimate_cost_usd(
     *,
     prompt_tokens: int = 0,
     completion_tokens: int = 0,
+    table=None,
 ) -> float | None:
     """Estimate USD cost for a completion, or ``None`` when the price is unknown.
 
     Token counts must be non-negative integers; anything else raises
     :class:`ValueError` so callers cannot silently report a negative cost.
+    Each count is clamped to :data:`MAX_TOKEN_COUNT` before the arithmetic so
+    an implausible value cannot overflow the estimate.
     """
     prompt = int(prompt_tokens)
     completion = int(completion_tokens)
     if prompt < 0 or completion < 0:
         raise ValueError("Token counts must be non-negative.")
+    prompt = min(prompt, MAX_TOKEN_COUNT)
+    completion = min(completion, MAX_TOKEN_COUNT)
 
-    price = get_model_price(provider, model)
+    price = get_model_price(provider, model, table=table)
     if price is None:
         return None
     return prompt / 1000 * price.input_per_1k_usd + completion / 1000 * price.output_per_1k_usd
@@ -151,6 +200,8 @@ def estimate_cost_usd(
 __all__ = [
     "DEFAULT_PRICE_TABLE",
     "ESTIMATE_DISCLAIMER",
+    "MAX_TOKEN_COUNT",
+    "WILDCARD",
     "ModelPrice",
     "estimate_cost_usd",
     "get_model_price",
diff --git a/app/services/reviews.py b/app/services/reviews.py
index 2a58da8..78669be 100644
--- a/app/services/reviews.py
+++ b/app/services/reviews.py
@@ -22,12 +22,14 @@
 
 import json
 import re
+import time
 
 from app.models.review_finding import (
     CATEGORIES_BY_KIND,
     CONFIDENCES,
     SEVERITIES,
 )
+from app.services import token_usage
 from app.services.llm import LLMProviderError, get_provider
 
 _SEVERITY_RANK = {name: index for index, name in enumerate(SEVERITIES)}
@@ -583,6 +585,44 @@ def validate_review_result(result, *, kind: str, known_files=None, max_findings=
 # --------------------------------------------------------------------------
 
 
+def _usage_from_response(response, provider, messages) -> dict:
+    """Return ledger-ready usage plus provider/model metadata for a completion."""
+    usage = token_usage.usage_from_response(response, messages)
+    name = getattr(provider, "name", None)
+    model = getattr(response, "model", None) or getattr(provider, "model", None)
+    if not model:
+        models = getattr(provider, "models", ()) or ()
+        model = models[0] if models else None
+    # The offline provider derives counts from text and never reports vendor
+    # usage; treat it as estimated. Real providers may report only one side.
+    estimated = (
+        name == "mock"
+        or getattr(response, "prompt_tokens", None) is None
+        or getattr(response, "completion_tokens", None) is None
+    )
+    latency = getattr(response, "latency_seconds", None)
+    return {
+        **usage,
+        "provider": name,
+        "model": model,
+        "tokens_estimated": bool(estimated),
+        "latency_ms": (max(0, round(float(latency) * 1000)) if latency is not None else None),
+    }
+
+
+def _fallback_usage(provider, messages, text, latency_ms) -> dict:
+    """Estimate usage for providers/test doubles that only expose ``complete``."""
+    usage = token_usage.usage_from_text(token_usage.messages_text(messages), text)
+    models = getattr(provider, "models", ()) or ()
+    return {
+        **usage,
+        "provider": getattr(provider, "name", None),
+        "model": models[0] if models else None,
+        "tokens_estimated": True,
+        "latency_ms": latency_ms,
+    }
+
+
 def _run_json(
     prompt: str,
     *,
@@ -596,27 +636,53 @@ def _run_json(
     ``known_files`` (when supplied) is the allow-list of real file paths from
     the bounded review context; findings or ``files_affected`` entries outside
     it are rejected so the model cannot invent files.
+
+    The returned dict also carries a ``usage`` entry so the route layer can
+    append one ledger row per review, including the provider-reported (or
+    estimated) token counts, provider/model, latency and outcome.
     """
+    messages = [
+        {"role": "system", "content": _REVIEW_SYSTEM},
+        {"role": "user", "content": prompt},
+    ]
+    started = time.perf_counter()
+    provider = None
     try:
         provider = get_provider()
-        text = provider.complete(
-            [
-                {"role": "system", "content": _REVIEW_SYSTEM},
-                {"role": "user", "content": prompt},
-            ]
-        )
+        # ``chat`` preserves the provider's usage/latency metadata. Test
+        # doubles and older callers may only implement ``complete``.
+        chat = getattr(provider, "chat", None)
+        if callable(chat):
+            response = chat(messages)
+            text = response.content
+            usage = _usage_from_response(response, provider, messages)
+        else:
+            text = provider.complete(messages)
+            elapsed_ms = max(0, round((time.perf_counter() - started) * 1000))
+            usage = _fallback_usage(provider, messages, text, elapsed_ms)
     except LLMProviderError as exc:
+        elapsed_ms = max(0, round((time.perf_counter() - started) * 1000))
         return {
             "summary": {"overall_assessment": f"[review unavailable: {exc}]"},
             "findings": [],
             "raw": "",
             "error": str(exc),
             "known_files": list(known_files) if known_files is not None else None,
+            "usage": {
+                "prompt_tokens": 0,
+                "completion_tokens": 0,
+                "total_tokens": 0,
+                "provider": getattr(provider, "name", None),
+                "model": (getattr(provider, "models", ()) or (None,))[0],
+                "tokens_estimated": True,
+                "latency_ms": elapsed_ms,
+            },
         }
     parsed = parse_review_response(
         text, kind=kind, threshold=threshold, known_files=known_files, max_findings=max_findings
     )
     parsed["error"] = None
+    parsed["usage"] = usage
     return parsed
 
 
diff --git a/app/services/usage_ledger.py b/app/services/usage_ledger.py
index f7c1d0e..de9834f 100644
--- a/app/services/usage_ledger.py
+++ b/app/services/usage_ledger.py
@@ -32,6 +32,11 @@
 )
 UNKNOWN_SOURCE = "unknown"
 
+#: Operation outcomes understood by the ledger (see docs/observability.md).
+STATUS_SUCCESS = "success"
+STATUS_ERROR = "error"
+STATUS_DENIED = "denied"
+
 
 def normalize_source(source: Any) -> str | None:
     """Return the lowercase source label, or ``None`` when it is missing."""
@@ -60,6 +65,29 @@ def _non_negative_int(value, name: str) -> int:
     return count
 
 
+def _optional_non_negative_int(value, name: str) -> int | None:
+    """Coerce an optional measurement to a non-negative integer, or ``None``."""
+    if value in (None, ""):
+        return None
+    try:
+        number = int(value)
+    except (TypeError, ValueError) as exc:
+        raise ValueError(f"{name} must be an integer.") from exc
+    if number < 0:
+        raise ValueError(f"{name} must be non-negative.")
+    return number
+
+
+def _optional_text(value, name: str, limit: int) -> str | None:
+    """Normalize an optional short string field, or return ``None``."""
+    if value is None:
+        return None
+    text = str(value).strip()
+    if not text:
+        return None
+    return text[:limit]
+
+
 def normalize_usage(prompt_tokens=0, completion_tokens=0, total_tokens=None) -> dict[str, int]:
     """Return validated prompt/completion/total token counts.
 
@@ -137,6 +165,9 @@ def record_usage(
     model: str | None = None,
     tokens_estimated: bool = True,
     cost_estimate=None,
+    request_id: str | None = None,
+    latency_ms=None,
+    status: str | None = None,
     commit: bool = True,
 ) -> UsageRecord:
     """Append one usage record and return it.
@@ -144,10 +175,20 @@ def record_usage(
     Scope objects take precedence over their explicit id counterparts. Cost is
     estimated from the configured price table when the caller does not supply a
     value; an unknown provider/model leaves it as ``None`` rather than zero.
+
+    ``request_id``, ``latency_ms`` and ``status`` are optional correlation and
+    performance fields. They default to ``None`` (the operation did not report
+    them) and are validated so a malformed provider payload cannot write a
+    negative latency or an over-long id into the ledger.
     """
     counts = normalize_usage(prompt_tokens, completion_tokens, total_tokens)
     provider = (str(provider).strip()[:50] or None) if provider else None
     model = (str(model).strip()[:100] or None) if model else None
+    request_id = _optional_text(request_id, "request_id", 64)
+    latency_ms = _optional_non_negative_int(latency_ms, "latency_ms")
+    status = _optional_text(status, "status", 20)
+    if status:
+        status = status.lower()
 
     if cost_estimate is None:
         cost_estimate = estimate_cost(
@@ -174,6 +215,9 @@ def record_usage(
         total_tokens=counts["total_tokens"],
         tokens_estimated=bool(tokens_estimated),
         cost_estimate=cost_estimate,
+        request_id=request_id,
+        latency_ms=latency_ms,
+        status=status,
     )
     db.session.add(record)
     if commit:
@@ -284,6 +328,9 @@ def breakdown_payload(
     "SOURCE_PROJECT_CHAT",
     "SOURCE_REVIEW",
     "SOURCE_TOOLS",
+    "STATUS_DENIED",
+    "STATUS_ERROR",
+    "STATUS_SUCCESS",
     "UNKNOWN_SOURCE",
     "breakdown_by_source",
     "breakdown_payload",
diff --git a/migrations/versions/f9e8d7c6b5a4_add_usage_request_status.py b/migrations/versions/f9e8d7c6b5a4_add_usage_request_status.py
new file mode 100644
index 0000000..e8b2e99
--- /dev/null
+++ b/migrations/versions/f9e8d7c6b5a4_add_usage_request_status.py
@@ -0,0 +1,36 @@
+"""add request correlation and performance fields to usage records
+
+Revision ID: f9e8d7c6b5a4
+Revises: c4d5e6f7a8b9
+Create Date: 2026-10-17 00:00:00.000000
+
+Phase 10: enrich usage rows with correlation and performance metadata
+(issue #428). All three columns are nullable so existing rows stay valid and
+operations that do not report the values can leave them empty.
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = "f9e8d7c6b5a4"
+down_revision = "c4d5e6f7a8b9"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.add_column("usage_records", sa.Column("request_id", sa.String(length=64), nullable=True))
+    op.add_column("usage_records", sa.Column("latency_ms", sa.Integer(), nullable=True))
+    op.add_column("usage_records", sa.Column("status", sa.String(length=20), nullable=True))
+    op.create_index("ix_usage_records_request_id", "usage_records", ["request_id"])
+    op.create_index("ix_usage_records_status", "usage_records", ["status"])
+
+
+def downgrade():
+    op.drop_index("ix_usage_records_status", table_name="usage_records")
+    op.drop_index("ix_usage_records_request_id", table_name="usage_records")
+    op.drop_column("usage_records", "status")
+    op.drop_column("usage_records", "latency_ms")
+    op.drop_column("usage_records", "request_id")
diff --git a/tests/conftest.py b/tests/conftest.py
index 05ce0a2..d789a29 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -132,12 +132,20 @@ def _reset_alert_state():
     same alert in another. The reset is best-effort: modules that do not expose
     a reset hook are simply skipped.
     """
-    from app.services import alerts as alerts_service
-
-    reset = getattr(alerts_service, "reset", None)
-    if callable(reset):
+    import importlib
+
+    resetters = []
+    for module_name in ("app.services.alerts", "app.services.budget_alerts"):
+        try:
+            module = importlib.import_module(module_name)
+        except ImportError:  # optional module: nothing to reset
+            continue
+        reset = getattr(module, "reset", None)
+        if callable(reset):
+            resetters.append(reset)
+
+    for reset in resetters:
         reset()
     yield
-    if callable(reset):
+    for reset in resetters:
         reset()
-
diff --git a/tests/test_alerts.py b/tests/test_alerts.py
index 0febc46..6e0048a 100644
--- a/tests/test_alerts.py
+++ b/tests/test_alerts.py
@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 import importlib
-import typing
+import inspect
 
 import pytest
 
@@ -14,7 +14,7 @@ def _load_module():
             return importlib.import_module(name)
         except ImportError:
             continue
-    pytest.skip("alert service module not found")
+    pytest.skip("alert service module not found", allow_module_level=True)
 
 
 alerts = _load_module()
@@ -30,13 +30,10 @@ def send(self, recipient: str, body: str) -> None:
 
 def _call_alert(func, kwargs):
     sig = inspect.signature(func)
-    accepted = {k: v for k: v in kwargs.items() if k in sig.parameters}
+    accepted = {k: v for k, v in kwargs.items() if k in sig.parameters}
     return func(**accepted)
 
 
-import inspect
-
-
 def test_threshold_fires_once_per_window():
     sender = _Recorder()
     service = alerts.AlertService(sender=sender.send)
diff --git a/tests/test_migrations.py b/tests/test_migrations.py
index b3d30e2..3ce26f7 100644
--- a/tests/test_migrations.py
+++ b/tests/test_migrations.py
@@ -126,7 +126,7 @@ class TestMigrationHead:
     def test_head_is_latest_revision(self):
         result = _run_flask(["db", "heads"], {"DATABASE_URL": "sqlite:///:memory:"})
         assert result.returncode == 0, result.stderr
-        assert "c4d5e6f7a8b9" in (result.stdout + result.stderr)
+        assert "f9e8d7c6b5a4" in (result.stdout + result.stderr)
 
     def test_usage_records_table_upgraded(self):
         expected = {
@@ -143,6 +143,9 @@ def test_usage_records_table_upgraded(self):
             "total_tokens",
             "tokens_estimated",
             "cost_estimate",
+            "request_id",
+            "latency_ms",
+            "status",
             "created_at",
         }
         with _migration_db() as db_url, _inspect(db_url) as insp:
@@ -157,8 +160,22 @@ def test_usage_records_table_upgraded(self):
                 "ix_usage_records_project_id",
                 "ix_usage_records_source",
                 "ix_usage_records_created_at",
+                "ix_usage_records_request_id",
+                "ix_usage_records_status",
             } <= indexes
 
+    def test_usage_correlation_columns_downgrade_removed(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            db_url = f"sqlite:///{os.path.join(tmp, 'mig_usage_correlation.db')}"
+            up = _run_flask(["db", "upgrade"], {"DATABASE_URL": db_url})
+            assert up.returncode == 0, up.stderr
+            down = _run_flask(["db", "downgrade", "c4d5e6f7a8b9"], {"DATABASE_URL": db_url})
+            assert down.returncode == 0, down.stderr
+            with _inspect(db_url) as insp:
+                columns = {col["name"] for col in insp.get_columns("usage_records")}
+                assert {"request_id", "latency_ms", "status"}.isdisjoint(columns)
+                assert "usage_records" in set(insp.get_table_names())
+
     def test_usage_records_downgrade_removed(self):
         with tempfile.TemporaryDirectory() as tmp:
             db_url = f"sqlite:///{os.path.join(tmp, 'mig_ledger.db')}"
diff --git a/tests/test_plugin_deps.py b/tests/test_plugin_deps.py
index 385ffa5..115eed7 100644
--- a/tests/test_plugin_deps.py
+++ b/tests/test_plugin_deps.py
@@ -14,6 +14,7 @@
 
 from app.services.plugin_deps import (
     DependencyResolver,
+    PluginDependencyConflictError,
     PluginDependencyCycleError,
     PluginDependencyDisabledError,
     PluginDependencyNotFoundError,
@@ -190,6 +191,70 @@ def test_recursive_satisfied_dependencies(self):
         _resolver(plugins).resolve(_ref("plugin-a", dependencies=["plugin-b"]))
 
 
+# ---------------------------------------------------------------------------
+# Resolver-level: unsatisfiable version ranges across the graph
+# ---------------------------------------------------------------------------
+
+
+class TestVersionConflicts:
+    def test_direct_conflict_between_two_requirements(self):
+        plugins = {"shared": _ref("shared", version="1.5.0")}
+        with pytest.raises(PluginDependencyConflictError) as excinfo:
+            _resolver(plugins).resolve(_ref("plugin-a", dependencies=["shared>=2", "shared<2"]))
+
+        assert excinfo.value.plugin_id == "plugin-a"
+        assert "version conflict" in excinfo.value.message
+        assert "shared>=2" in excinfo.value.message
+        assert "shared<2" in excinfo.value.message
+        assert "path: plugin-a" in excinfo.value.message
+
+    def test_transitive_conflict_reports_both_paths(self):
+        plugins = {
+            "shared": _ref("shared", version="1.5.0"),
+            "plugin-b": _ref("plugin-b", dependencies=["shared>=2"]),
+            "plugin-c": _ref("plugin-c", dependencies=["shared<2"]),
+        }
+        with pytest.raises(PluginDependencyConflictError) as excinfo:
+            _resolver(plugins).resolve(_ref("plugin-a", dependencies=["plugin-b", "plugin-c"]))
+
+        message = excinfo.value.message
+        assert "version conflict for 'shared'" in message
+        assert "path: plugin-a -> plugin-b" in message
+        assert "path: plugin-a -> plugin-c" in message
+
+    def test_transitive_package_conflict_is_detected(self):
+        plugins = {
+            "plugin-b": _ref("plugin-b", dependencies=["shared-pkg>=2"]),
+            "plugin-c": _ref("plugin-c", dependencies=["shared-pkg<2"]),
+        }
+        resolver = _resolver(plugins, installed={"shared-pkg": "1.5.0"})
+        with pytest.raises(PluginDependencyConflictError) as excinfo:
+            resolver.resolve(_ref("plugin-a", dependencies=["plugin-b", "plugin-c"]))
+
+        assert "version conflict for 'shared-pkg'" in excinfo.value.message
+
+    def test_wildcard_pins_that_do_not_overlap_conflict(self):
+        plugins = {"shared": _ref("shared", version="1.5.0")}
+        with pytest.raises(PluginDependencyConflictError):
+            _resolver(plugins).resolve(
+                _ref("plugin-a", dependencies=["shared==1.*", "shared==2.*"])
+            )
+
+    def test_overlapping_ranges_do_not_conflict(self):
+        plugins = {"shared": _ref("shared", version="1.8.0")}
+        _resolver(plugins).resolve(
+            _ref("plugin-a", dependencies=["shared>=1,<3", "shared>=1.5,<2"])
+        )
+
+    def test_compatible_requirements_on_the_same_dependency_resolve(self):
+        plugins = {
+            "shared": _ref("shared", version="1.8.0"),
+            "plugin-b": _ref("plugin-b", dependencies=["shared>=1"]),
+            "plugin-c": _ref("plugin-c", dependencies=["shared<2"]),
+        }
+        _resolver(plugins).resolve(_ref("plugin-a", dependencies=["plugin-b", "plugin-c"]))
+
+
 # ---------------------------------------------------------------------------
 # Resolver-level: cycles
 # ---------------------------------------------------------------------------
@@ -304,3 +369,20 @@ def test_already_enabled_plugin_is_idempotent(self):
         registry.register(plugin)
         registry.enable("plugin-a")  # must not re-resolve or raise
         assert plugin.enabled is True
+
+    def test_enable_refused_for_version_conflict(self):
+        registry = PluginRegistry()
+        registry.register(_registry_plugin("shared", version="1.5.0"))
+        registry.register(_registry_plugin("plugin-b", dependencies=["shared>=2"]))
+        registry.register(_registry_plugin("plugin-c", dependencies=["shared<2"]))
+        plugin_a = _registry_plugin(
+            "plugin-a", dependencies=["plugin-b", "plugin-c"], enabled=False
+        )
+        registry.register(plugin_a)
+
+        with pytest.raises(PluginDependencyResolutionError) as excinfo:
+            registry.enable("plugin-a")
+
+        assert "version conflict" in str(excinfo.value)
+        assert "plugin-a -> plugin-b" in str(excinfo.value)
+        assert plugin_a.enabled is False
diff --git a/tests/test_pricing.py b/tests/test_pricing.py
index 280fc40..657bb47 100644
--- a/tests/test_pricing.py
+++ b/tests/test_pricing.py
@@ -5,6 +5,9 @@
 from app.services.pricing import (
     DEFAULT_PRICE_TABLE,
     ESTIMATE_DISCLAIMER,
+    MAX_TOKEN_COUNT,
+    WILDCARD,
+    ModelPrice,
     estimate_cost_usd,
     get_model_price,
     list_model_prices,
@@ -92,3 +95,113 @@ def test_negative_tokens_are_rejected(self):
             estimate_cost_usd("openai", "gpt-4o", prompt_tokens=-1)
         with pytest.raises(ValueError):
             estimate_cost_usd("openai", "gpt-4o", completion_tokens=-1)
+
+
+def _entry(provider, model, input_rate=1.0, output_rate=2.0):
+    return ModelPrice(
+        provider=provider,
+        model=model,
+        input_per_1k_usd=input_rate,
+        output_per_1k_usd=output_rate,
+    )
+
+
+class TestLookupPrecedence:
+    """Exact entries beat wildcards; provider fallbacks beat global ones."""
+
+    @pytest.fixture()
+    def table(self):
+        return {
+            ("openai", "gpt-4o"): _entry("openai", "gpt-4o", 1.0, 2.0),
+            ("openai", WILDCARD): _entry("openai", WILDCARD, 3.0, 4.0),
+            (WILDCARD, "gpt-4o"): _entry(WILDCARD, "gpt-4o", 5.0, 6.0),
+            (WILDCARD, WILDCARD): _entry(WILDCARD, WILDCARD, 7.0, 8.0),
+        }
+
+    def test_exact_match_wins_over_every_wildcard(self, table):
+        price = get_model_price("openai", "gpt-4o", table=table)
+        assert price.input_per_1k_usd == 1.0
+        assert price.output_per_1k_usd == 2.0
+
+    def test_provider_wildcard_beats_model_and_global_wildcards(self, table):
+        price = get_model_price("openai", "a-new-model", table=table)
+        assert price.input_per_1k_usd == 3.0
+
+    def test_model_wildcard_used_for_unknown_provider(self, table):
+        price = get_model_price("other-provider", "gpt-4o", table=table)
+        assert price.input_per_1k_usd == 5.0
+
+    def test_global_wildcard_is_the_last_resort(self, table):
+        price = get_model_price("other-provider", "other-model", table=table)
+        assert price.input_per_1k_usd == 7.0
+
+    def test_lookup_still_normalizes_case_and_whitespace(self, table):
+        assert get_model_price("  OpenAI ", " GPT-4O ", table=table) == get_model_price(
+            "openai", "gpt-4o", table=table
+        )
+
+    def test_unknown_model_without_a_wildcard_is_none(self):
+        assert get_model_price("does-not-exist", "gpt-4o", table={}) is None
+
+    def test_custom_table_listing_and_grouping(self, table):
+        assert len(list_model_prices("openai", table=table)) == 2
+        grouped = price_table(table=table)
+        assert "openai" in grouped
+        assert {entry["model"] for entry in grouped["openai"]} == {"gpt-4o", WILDCARD}
+
+
+class TestCostArithmeticAndClamping:
+    def test_arithmetic_is_pinned_to_the_price_rates(self):
+        table = {("acme", "model"): _entry("acme", "model", 0.5, 1.5)}
+        cost = estimate_cost_usd(
+            "acme",
+            "model",
+            prompt_tokens=1_234,
+            completion_tokens=5_678,
+            table=table,
+        )
+        expected = 1_234 / 1000 * 0.5 + 5_678 / 1000 * 1.5
+        assert cost == pytest.approx(expected, abs=1e-12)
+
+    def test_wildcard_estimate_uses_the_fallback_rate(self):
+        table = {
+            ("openai", WILDCARD): _entry("openai", WILDCARD, 0.25, 0.75),
+        }
+        cost = estimate_cost_usd(
+            "openai",
+            "brand-new-model",
+            prompt_tokens=2_000,
+            completion_tokens=1_000,
+            table=table,
+        )
+        assert cost == pytest.approx(2 * 0.25 + 1 * 0.75)
+
+    def test_unknown_model_estimate_is_none(self):
+        assert estimate_cost_usd("acme", "missing", prompt_tokens=10, table={}) is None
+
+    def test_token_counts_are_clamped_to_the_maximum(self):
+        price = get_model_price("openai", "gpt-4o")
+        at_limit = estimate_cost_usd(
+            "openai",
+            "gpt-4o",
+            prompt_tokens=MAX_TOKEN_COUNT,
+            completion_tokens=0,
+        )
+        above_limit = estimate_cost_usd(
+            "openai",
+            "gpt-4o",
+            prompt_tokens=MAX_TOKEN_COUNT + 1_000_000,
+            completion_tokens=0,
+        )
+        assert above_limit == pytest.approx(at_limit)
+        assert above_limit == pytest.approx(MAX_TOKEN_COUNT / 1000 * price.input_per_1k_usd)
+
+    def test_completion_count_is_clamped_independently(self):
+        price = get_model_price("openai", "gpt-4o")
+        cost = estimate_cost_usd(
+            "openai",
+            "gpt-4o",
+            prompt_tokens=0,
+            completion_tokens=MAX_TOKEN_COUNT * 5,
+        )
+        assert cost == pytest.approx(MAX_TOKEN_COUNT / 1000 * price.output_per_1k_usd)
diff --git a/tests/test_usage_ledger.py b/tests/test_usage_ledger.py
index 3c18600..fb19d19 100644
--- a/tests/test_usage_ledger.py
+++ b/tests/test_usage_ledger.py
@@ -103,6 +103,44 @@ def test_negative_counts_are_rejected(self, db):
         with pytest.raises(ValueError):
             usage_ledger.record_usage(source="chat", cost_estimate=-0.1)
 
+    def test_records_request_latency_and_status(self, db):
+        record = usage_ledger.record_usage(
+            source="review",
+            provider="openai",
+            model="gpt-4o",
+            request_id="  req-123  ",
+            latency_ms=1250,
+            status="SUCCESS",
+        )
+
+        assert record.request_id == "req-123"
+        assert record.latency_ms == 1250
+        assert record.status == "success"
+        payload = record.to_dict()
+        assert payload["request_id"] == "req-123"
+        assert payload["latency_ms"] == 1250
+        assert payload["status"] == "success"
+
+    def test_correlation_fields_default_to_none(self, db):
+        record = usage_ledger.record_usage(source="tools", provider="mock", model="mock-1")
+
+        assert record.request_id is None
+        assert record.latency_ms is None
+        assert record.status is None
+        assert record.to_dict()["request_id"] is None
+
+    def test_request_id_is_bounded(self, db):
+        record = usage_ledger.record_usage(source="chat", request_id="x" * 200)
+
+        assert record.request_id is not None
+        assert len(record.request_id) == 64
+
+    def test_invalid_latency_is_rejected(self, db):
+        with pytest.raises(ValueError):
+            usage_ledger.record_usage(source="chat", latency_ms=-1)
+        with pytest.raises(ValueError):
+            usage_ledger.record_usage(source="chat", latency_ms="not-a-number")
+
 
 class TestBreakdown:
     def test_groups_known_and_unknown_sources(self, db, make_user):
@@ -289,3 +327,67 @@ def test_each_project_analysis_records_a_row(self, client, db, make_user, login)
             assert row.tokens_estimated is True
             assert row.provider == "mock"
             assert row.model == "mock-1"
+
+
+class TestReviewInstrumentation:
+    def test_each_review_records_a_scoped_review_row(self, client, db, make_user, login):
+        user, workspace, project = _project(
+            make_user,
+            login,
+            username="reviewuser",
+            email="reviewuser@example.com",
+        )
+
+        for kind in ("quality", "security"):
+            response = client.post(
+                "/reviews/api/reviews",
+                json={"source": "project", "project_id": project.id, "kind": kind},
+            )
+            assert response.status_code == 201
+
+        rows = UsageRecord.query.filter_by(source="review").order_by(UsageRecord.id).all()
+        assert len(rows) == 2
+        for row in rows:
+            assert row.user_id == user.id
+            assert row.workspace_id == workspace.id
+            assert row.project_id == project.id
+            assert row.total_tokens > 0
+            assert row.tokens_estimated is True
+            assert row.provider == "mock"
+            assert row.model == "mock-1"
+            assert row.status == usage_ledger.STATUS_SUCCESS
+            assert row.request_id
+            assert row.latency_ms is not None
+            assert row.latency_ms >= 0
+
+    def test_failed_review_records_an_error_row(self, client, db, make_user, login, monkeypatch):
+        from app.services.llm import LLMProviderError
+
+        _, workspace, project = _project(
+            make_user,
+            login,
+            username="failedreview",
+            email="failedreview@example.com",
+        )
+
+        def boom():
+            raise LLMProviderError("provider unavailable")
+
+        monkeypatch.setattr("app.services.reviews.get_provider", boom)
+        response = client.post(
+            "/reviews/api/reviews",
+            json={"source": "project", "project_id": project.id, "kind": "quality"},
+        )
+
+        assert response.status_code == 201
+        assert response.get_json()["status"] == "failed"
+        rows = UsageRecord.query.filter_by(source="review").all()
+        assert len(rows) == 1
+        row = rows[0]
+        assert row.workspace_id == workspace.id
+        assert row.project_id == project.id
+        assert row.total_tokens == 0
+        assert row.tokens_estimated is True
+        assert row.status == usage_ledger.STATUS_ERROR
+        assert row.request_id
+        assert row.latency_ms is not None