fix: phase 9/10 usage ledger, review instrumentation, cost tests, dependency conflicts (#393, #428, #430, #445)

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,
         }
 

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()
 
 

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"]

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

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,18 +258,23 @@ 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,
         root_id: str,
         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))