core/validation_layer.py

"""
core/validation_layer.py — final accept / retry / reject decision.

v4.7 additions
==============
* ``VerificationScore`` — the "Verification & Trust" scoring engine.
  It combines four independent signals into a single 0-100 confidence
  score and a colour band (red / yellow / green) used by the UI badge.

  Components (equal-weighted by default, tunable via env / constructor):
    1. **Test Pass Rate**       — did the repo-native tests pass? (0 or 100)
    2. **Lint Pass Rate**       — ruff + eslint static-analysis clean rate.
    3. **Patch Memory Relevance** — inverse of how similar this patch is to
       previously-failed patches (lower similarity = higher trust).
    4. **Semantic Analysis**    — reviewer verdict + risk score signal.

  The class is *stateless* and safe to instantiate on demand. It never
  raises — missing signals simply score 0 and the reason is recorded.
"""

from __future__ import annotations

import os
from dataclasses import dataclass, field, asdict
from typing import Any, Dict, List, Optional


# ---------------------------------------------------------------------------
# VerificationScore — the "Confidence Badge" scoring engine.
# ---------------------------------------------------------------------------

# Colour thresholds (0-100). Anything at/above GREEN is green, at/above
# YELLOW but below GREEN is yellow, everything else is red.
GREEN_THRESHOLD = 80
YELLOW_THRESHOLD = 55


@dataclass
class VerificationScore:
    """
    Four-component confidence score for a single AI patch.

    Usage::

        vs = VerificationScore()
        result = vs.calculate(state)
        # -> {"score": 82, "band": "green", "components": {...}, ...}
    """

    # Weights are normalised inside ``calculate`` so callers can pass any
    # positive numbers.
    test_weight: float = 1.0
    lint_weight: float = 1.0
    memory_weight: float = 1.0
    semantic_weight: float = 1.0

    # Score band thresholds (0-100). Overridable via env for A/B testing.
    green_threshold: int = GREEN_THRESHOLD
    yellow_threshold: int = YELLOW_THRESHOLD

    @classmethod
    def from_env(cls) -> "VerificationScore":
        """Build a scorer honouring optional ``DEVAI_VS_*`` env vars."""
        def _f(key: str, default: float) -> float:
            try:
                return float(os.environ.get(key, default))
            except (TypeError, ValueError):
                return default

        return cls(
            test_weight=_f("DEVAI_VS_TEST_WEIGHT", 1.0),
            lint_weight=_f("DEVAI_VS_LINT_WEIGHT", 1.0),
            memory_weight=_f("DEVAI_VS_MEMORY_WEIGHT", 1.0),
            semantic_weight=_f("DEVAI_VS_SEMANTIC_WEIGHT", 1.0),
            green_threshold=int(_f("DEVAI_VS_GREEN", GREEN_THRESHOLD)),
            yellow_threshold=int(_f("DEVAI_VS_YELLOW", YELLOW_THRESHOLD)),
        )

    # ---- Individual component scorers -------------------------------------

    @staticmethod
    def _test_pass_score(state: Dict[str, Any]) -> Dict[str, Any]:
        """0 or 100 based on test_passed. Skipped tests = neutral 60."""
        test_passed = state.get("test_passed")
        # Some pipelines mark the test stage as skipped (no test files).
        test_logs = str(state.get("test_logs") or "")
        skipped = (
            test_passed is None
            or "no tests" in test_logs.lower()
            or "skipped" in test_logs.lower()[:400]
        )
        if skipped and test_passed is not True:
            return {
                "score": 60.0,
                "reason": "no repo-native tests detected — neutral score",
                "raw": {"test_passed": test_passed, "skipped": True},
            }
        score = 100.0 if bool(test_passed) else 0.0
        return {
            "score": score,
            "reason": "tests passed" if score else "tests failed",
            "raw": {
                "test_passed": bool(test_passed),
                "auto_debug_attempts": int(state.get("auto_debug_attempts", 0) or 0),
            },
        }

    @staticmethod
    def _lint_pass_score(state: Dict[str, Any]) -> Dict[str, Any]:
        """Score from static-analysis report (0-100)."""
        report = state.get("static_analysis_report") or {}
        # v4.6 stashes the report either as a dict (to_dict) or nested.
        if not report:
            return {
                "score": 70.0,
                "reason": "no lint report available — neutral score",
                "raw": {},
            }
        if report.get("skipped"):
            return {
                "score": 70.0,
                "reason": "static analysis skipped (linters unavailable)",
                "raw": {"skipped": True},
            }
        passed = bool(report.get("passed"))
        total_issues = int(report.get("total_issues", 0) or 0)
        if passed and total_issues == 0:
            return {
                "score": 100.0,
                "reason": "lint clean",
                "raw": {"total_issues": 0},
            }
        # Decay: 5 issues ≈ 75, 10 ≈ 55, 20 ≈ 15, capped at 0.
        decay = max(0.0, 100.0 - total_issues * 5.0)
        return {
            "score": decay,
            "reason": f"{total_issues} lint issue(s)",
            "raw": {"total_issues": total_issues, "passed": passed},
        }

    @staticmethod
    def _patch_memory_score(state: Dict[str, Any]) -> Dict[str, Any]:
        """Higher score = this patch does NOT resemble past failed patches."""
        matches: List[Dict[str, Any]] = state.get("patch_memory_matches") or []
        if not matches:
            return {
                "score": 90.0,
                "reason": "no similar past failures — high trust",
                "raw": {"matches": 0},
            }
        top_sim = max(float(m.get("similarity", 0.0) or 0.0) for m in matches)
        # Invert similarity: 0.0 sim ⇒ 100, 1.0 sim ⇒ 0.
        score = round((1.0 - min(top_sim, 1.0)) * 100.0, 2)
        return {
            "score": score,
            "reason": f"{len(matches)} similar past failure(s), top sim={top_sim:.2f}",
            "raw": {
                "match_count": len(matches),
                "top_similarity": top_sim,
            },
        }

    @staticmethod
    def _semantic_score(state: Dict[str, Any]) -> Dict[str, Any]:
        """From reviewer verdict + risk score (0-10)."""
        review = state.get("review_result") or {}
        verdict = str(review.get("verdict") or "").lower()
        risk = int(review.get("risk_score", 0) or 0)

        if verdict == "rejected":
            return {
                "score": 0.0,
                "reason": f"reviewer rejected (risk={risk})",
                "raw": {"verdict": verdict, "risk_score": risk},
            }
        if verdict == "approved":
            base = 100.0
        elif verdict == "approved_with_warnings":
            base = 75.0
        else:
            base = 60.0  # unknown / missing verdict
        # Risk penalty: risk 0 ⇒ no penalty, risk 10 ⇒ -100.
        risk_penalty = min(risk, 10) * 10.0
        score = max(0.0, base - risk_penalty)
        return {
            "score": score,
            "reason": f"verdict={verdict or 'n/a'}, risk={risk}/10",
            "raw": {"verdict": verdict, "risk_score": risk},
        }

    # ---- Public API -------------------------------------------------------

    def calculate(self, state: Dict[str, Any]) -> Dict[str, Any]:
        """Return the combined score + per-component breakdown."""
        components = {
            "test": self._test_pass_score(state),
            "lint": self._lint_pass_score(state),
            "memory": self._patch_memory_score(state),
            "semantic": self._semantic_score(state),
        }
        weights = {
            "test": max(0.0, self.test_weight),
            "lint": max(0.0, self.lint_weight),
            "memory": max(0.0, self.memory_weight),
            "semantic": max(0.0, self.semantic_weight),
        }
        total_weight = sum(weights.values()) or 1.0
        weighted_sum = sum(components[k]["score"] * weights[k] for k in components)
        score = round(weighted_sum / total_weight, 1)

        # Colour band
        if score >= self.green_threshold:
            band, emoji, label = "green", "🟢", "High Confidence"
        elif score >= self.yellow_threshold:
            band, emoji, label = "yellow", "🟡", "Medium Confidence"
        else:
            band, emoji, label = "red", "🔴", "Low Confidence"

        result = {
            "score": score,
            "band": band,
            "emoji": emoji,
            "label": label,
            "components": components,
            "weights": weights,
            "thresholds": {
                "green": self.green_threshold,
                "yellow": self.yellow_threshold,
            },
        }
        state["verification_score"] = result
        return result


__all__ = ["ValidationLayer", "VerificationScore", "GREEN_THRESHOLD", "YELLOW_THRESHOLD"]


# ---------------------------------------------------------------------------
# Existing ValidationLayer — extended to also compute the VerificationScore.
# ---------------------------------------------------------------------------


class ValidationLayer:

    def __init__(self, scorer: Optional[VerificationScore] = None) -> None:
        # Lazy default — env-configurable.
        self._scorer = scorer or VerificationScore.from_env()

    def decide(self, state: Dict[str, Any]) -> Dict[str, Any]:
        review = state.get("review_result", {}) or {}
        verdict = (review.get("verdict") or "").lower()
        risk = int(review.get("risk_score", 0) or 0)
        test_passed = bool(state.get("test_passed"))
        attempts = int(state.get("auto_debug_attempts", 0))

        if verdict == "rejected" or risk >= 8:
            decision = "rejected"
            reason = f"reviewer rejected (risk={risk})"
        elif not test_passed and attempts >= 3:
            decision = "rejected"
            reason = "tests failed after auto-debug attempts"
        elif not test_passed:
            decision = "retry_needed"
            reason = "tests still failing"
        elif verdict == "approved_with_warnings" or risk >= 5:
            decision = "accepted_with_warning"
            reason = f"approved but risk={risk}"
        else:
            decision = "accepted"
            reason = "all checks passed"

        result = {
            "decision": decision,
            "reason": reason,
            "risk_score": risk,
            "test_passed": test_passed,
            "auto_debug_attempts": attempts,
        }
        state["validation_result"] = result
        state["final_status"] = decision

        # v4.7 — always compute the Verification & Trust score.
        try:
            self._scorer.calculate(state)
        except Exception as exc:  # never let scoring break the pipeline
            state["verification_score"] = {
                "score": 0.0,
                "band": "red",
                "emoji": "⚠️",
                "label": "Score unavailable",
                "error": str(exc),
                "components": {},
            }
        return result