src/redstack/engines/validation.py

from __future__ import annotations

import math
import re
from collections import Counter
from typing import Final, final

from pydantic import BaseModel, ConfigDict

from redstack.domain.enums import ReasoningPolarity, Severity, ValidationCode
from redstack.domain.ranking import RankedCandidate, Ranking
from redstack.domain.validation import ValidationFinding, ValidationReport

_ID_PATTERN: Final[re.Pattern[str]] = re.compile(r"^CAND_[0-9]{7}$")
# Normalization for the templated-reasoning skeleton: collapse ids and numbers.
_NUM_TOKEN: Final[re.Pattern[str]] = re.compile(r"\d+(?:\.\d+)?")
_ID_TOKEN: Final[re.Pattern[str]] = re.compile(r"CAND_[0-9]{7}")
_TEMPLATED_FRACTION: Final[float] = 0.70

_ALL_CODES: Final[frozenset[ValidationCode]] = frozenset(ValidationCode)


@final
class ValidationEngine(BaseModel):
    """Stateless, pure validator over a (reasoned) ``Ranking``."""

    model_config = ConfigDict(
        frozen=True, extra="forbid", arbitrary_types_allowed=False
    )

    expected_size: int = 100

    # ------------------------------------------------------------------ public
    def validate_ranking(self, ranking: Ranking) -> ValidationReport:
        """Run all structural + Stage-4 checks; assemble the ``ValidationReport``.

        Named ``validate_ranking`` rather than ``validate`` so this doesn't
        shadow ``pydantic.BaseModel.validate`` (a deprecated classmethod with
        an unrelated signature).
        """
        findings: list[ValidationFinding] = []
        ordered = ranking.ordered

        findings.extend(self._row_count(ordered))
        findings.extend(self._ranks(ordered))
        findings.extend(self._ids(ordered))
        findings.extend(self._scores(ordered))
        findings.extend(self._reasoning(ordered))

        is_valid = not any(f.severity is Severity.HARD for f in findings)
        return ValidationReport(
            findings=tuple(findings),
            is_valid=is_valid,
            checks_run=_ALL_CODES,
        )

    # ----------------------------------------------------------- structural
    def _row_count(
        self, ordered: tuple[RankedCandidate, ...]
    ) -> tuple[ValidationFinding, ...]:
        if len(ordered) == self.expected_size:
            return ()
        return (
            ValidationFinding(
                code=ValidationCode.WRONG_ROW_COUNT,
                severity=Severity.HARD,
                message=f"expected {self.expected_size} rows, got {len(ordered)}",
                location=None,
            ),
        )

    def _ranks(
        self, ordered: tuple[RankedCandidate, ...]
    ) -> tuple[ValidationFinding, ...]:
        out: list[ValidationFinding] = []
        ranks = [c.rank for c in ordered]
        for candidate in ordered:
            if candidate.rank < 1 or candidate.rank > self.expected_size:
                out.append(
                    ValidationFinding(
                        code=ValidationCode.RANK_OUT_OF_RANGE,
                        severity=Severity.HARD,
                        message=f"rank {candidate.rank} out of [1, {self.expected_size}]",
                        location=candidate.candidate_id,
                    )
                )
        counts = Counter(ranks)
        for rank, freq in sorted(counts.items()):
            if freq > 1:
                out.append(
                    ValidationFinding(
                        code=ValidationCode.DUPLICATE_RANK,
                        severity=Severity.HARD,
                        message=f"rank {rank} appears {freq} times",
                        location=None,
                    )
                )
        expected = set(range(1, self.expected_size + 1))
        missing = expected - set(ranks)
        if missing:
            out.append(
                ValidationFinding(
                    code=ValidationCode.MISSING_RANK,
                    severity=Severity.HARD,
                    message=f"missing ranks: {sorted(missing)[:10]}",
                    location=None,
                )
            )
        return tuple(out)

    def _ids(
        self, ordered: tuple[RankedCandidate, ...]
    ) -> tuple[ValidationFinding, ...]:
        out: list[ValidationFinding] = []
        seen: Counter[str] = Counter(c.candidate_id for c in ordered)
        for candidate in ordered:
            if _ID_PATTERN.match(candidate.candidate_id) is None:
                out.append(
                    ValidationFinding(
                        code=ValidationCode.BAD_ID_FORMAT,
                        severity=Severity.HARD,
                        message=f"id {candidate.candidate_id!r} fails ^CAND_[0-9]{{7}}$",
                        location=candidate.candidate_id,
                    )
                )
        for cid, freq in sorted(seen.items()):
            if freq > 1:
                out.append(
                    ValidationFinding(
                        code=ValidationCode.DUPLICATE_ID,
                        severity=Severity.HARD,
                        message=f"id {cid} appears {freq} times",
                        location=cid,
                    )
                )
        return tuple(out)

    def _scores(
        self, ordered: tuple[RankedCandidate, ...]
    ) -> tuple[ValidationFinding, ...]:
        out: list[ValidationFinding] = []
        for candidate in ordered:
            value = candidate.score
            if not isinstance(value, float) or not math.isfinite(value):
                out.append(
                    ValidationFinding(
                        code=ValidationCode.SCORE_NOT_FLOAT,
                        severity=Severity.HARD,
                        message=f"score {value!r} is not a finite float",
                        location=candidate.candidate_id,
                    )
                )
        for earlier, later in zip(ordered, ordered[1:], strict=False):
            if later.score > earlier.score:
                out.append(
                    ValidationFinding(
                        code=ValidationCode.SCORE_INCREASING,
                        severity=Severity.HARD,
                        message=(
                            f"score rises at rank {later.rank} "
                            f"({earlier.score} -> {later.score})"
                        ),
                        location=later.candidate_id,
                    )
                )
            elif (
                later.score == earlier.score
                and earlier.candidate_id >= later.candidate_id
            ):
                out.append(
                    ValidationFinding(
                        code=ValidationCode.TIEBREAK_VIOLATION,
                        severity=Severity.HARD,
                        message=(
                            f"tie at rank {later.rank}: "
                            f"{earlier.candidate_id} !< {later.candidate_id}"
                        ),
                        location=later.candidate_id,
                    )
                )
        return tuple(out)

    # ------------------------------------------------------------- Stage-4
    def _reasoning(
        self, ordered: tuple[RankedCandidate, ...]
    ) -> tuple[ValidationFinding, ...]:
        out: list[ValidationFinding] = []
        rendered: list[str] = []
        skeletons: list[str] = []

        for candidate in ordered:
            reasoning = candidate.reasoning
            if reasoning is None or not reasoning.rendered.strip():
                out.append(
                    ValidationFinding(
                        code=ValidationCode.EMPTY_REASONING,
                        severity=Severity.HARD,
                        message="reasoning missing or blank",
                        location=candidate.candidate_id,
                    )
                )
                continue

            if any(len(clause.evidence) == 0 for clause in reasoning.clauses):
                out.append(
                    ValidationFinding(
                        code=ValidationCode.HALLUCINATION,
                        severity=Severity.HARD,
                        message="a reasoning clause carries no evidence",
                        location=candidate.candidate_id,
                    )
                )

            out.extend(self._rank_band_consistency(candidate))

            rendered.append(reasoning.rendered)
            skeletons.append(self._skeleton(reasoning.rendered))

        out.extend(self._identical(ordered, rendered))
        out.extend(self._templated(skeletons))
        return tuple(out)

    def _rank_band_consistency(
        self, candidate: RankedCandidate
    ) -> tuple[ValidationFinding, ...]:
        reasoning = candidate.reasoning
        if reasoning is None:
            return ()
        expected = self._expected_band(candidate.rank)
        out: list[ValidationFinding] = []
        if reasoning.rank_band != expected:
            out.append(
                ValidationFinding(
                    code=ValidationCode.RANK_REASONING_MISMATCH,
                    severity=Severity.SOFT,
                    message=(
                        f"rank {candidate.rank} band {reasoning.rank_band!r} "
                        f"!= expected {expected!r}"
                    ),
                    location=candidate.candidate_id,
                )
            )
        has_strength = any(
            c.polarity is ReasoningPolarity.STRENGTH for c in reasoning.clauses
        )
        if expected in ("top", "mid") and not has_strength:
            out.append(
                ValidationFinding(
                    code=ValidationCode.RANK_REASONING_MISMATCH,
                    severity=Severity.SOFT,
                    message=f"{expected}-band candidate has no strength clause",
                    location=candidate.candidate_id,
                )
            )
        return tuple(out)

    def _expected_band(self, rank: int) -> str:
        top_cut = max(1, round(self.expected_size * 0.10))
        mid_cut = max(top_cut + 1, round(self.expected_size * 0.50))
        if rank <= top_cut:
            return "top"
        if rank <= mid_cut:
            return "mid"
        return "tail"

    @staticmethod
    def _identical(
        ordered: tuple[RankedCandidate, ...], rendered: list[str]
    ) -> tuple[ValidationFinding, ...]:
        if len(rendered) < 2:
            return ()
        counts = Counter(rendered)
        out: list[ValidationFinding] = []
        for text, freq in sorted(counts.items()):
            if freq > 1:
                out.append(
                    ValidationFinding(
                        code=ValidationCode.IDENTICAL_REASONING,
                        severity=Severity.HARD,
                        message=f"{freq} candidates share identical reasoning text",
                        location=None,
                    )
                )
        return tuple(out)

    @staticmethod
    def _templated(skeletons: list[str]) -> tuple[ValidationFinding, ...]:
        if len(skeletons) < 2:
            return ()
        counts = Counter(skeletons)
        _, top_freq = counts.most_common(1)[0]
        fraction = top_freq / len(skeletons)
        if fraction >= _TEMPLATED_FRACTION:
            return (
                ValidationFinding(
                    code=ValidationCode.TEMPLATED_REASONING,
                    severity=Severity.HARD,
                    message=(
                        f"{top_freq}/{len(skeletons)} renderings collapse to one "
                        f"templated skeleton ({fraction:.0%})"
                    ),
                    location=None,
                ),
            )
        return ()

    @staticmethod
    def _skeleton(text: str) -> str:
        """Normalize ids and numbers out so genuine variation survives but a fixed
        template (varying only by number/id) collapses to a single skeleton."""
        without_ids = _ID_TOKEN.sub("<id>", text)
        without_nums = _NUM_TOKEN.sub("<n>", without_ids)
        return " ".join(without_nums.lower().split())


__all__: tuple[str, ...] = ("ValidationEngine",)