portfolio-optimiser/src/portfolio_optimiser/validator.py

"""Deterministic, blocking hybrid validator (B1) — the obligatory, non-optional gate.

Pure module: **NO** ``agent_framework`` import (D7-portable core). Three stages over the
typed IR (``ir.py``):

1. **Pydantic IR** invariants already ran at construction (``ir.SavingsProposal``).
2. **PuLP solver-in-the-loop** — a real CBC solve bounds the maximum feasible saving (R2).
   CBC ships in PuLP's wheel; if it is genuinely absent the step **escalates**
   (``CbcUnavailable``) — no silent LP-relaxation fallback.
3. **Monte Carlo** — stdlib ``random`` (seeded ``_MC_SEED``) + ``statistics.quantiles`` over
   uncertain unit-costs give P10/P50/P90 of the feasible saving.

The structural block (stage 4) returns a ``Rejection`` that is a *different type* from
``ValidatedProposal`` and carries no percentiles, so it can never be consumed as validated.

Promoted verbatim from ``spikes/c_validator.py``. The one deliberate change vs the spike:
``self_repair`` no longer borrows the harness ``Budget`` (that pulls ``agent_framework`` via
``spikes/_harness``) — it loops directly on ``max_attempts``; the token-budget bound is
layered on by the Step 10 generate loop, keeping THIS module pure.
"""

from __future__ import annotations

import random
import statistics
import warnings
from collections.abc import Callable, Mapping
from contextlib import contextmanager
from dataclasses import dataclass

import pulp

from portfolio_optimiser.ir import AffectedItem, CostBaseline, CostBaselineLine, SavingsProposal
from portfolio_optimiser.reference_domain import Project

MAX_SAVING_FRACTION = 0.30
"""Policy cap: at most 30% of an affected item's cost is realistically recoverable as a
saving. The LP bounds the feasible saving by this fraction."""

_ENERGY_METHOD_MEASURE = "energy_efficiency"
_ENERGY_METHOD_MAX_FRACTION = 0.15
"""Step 9 (SC7-B): the IPMVP Option A method-specific cap. Option A measures only the KEY parameter
and STIPULATES the rest (operating hours), so the defensibly-verifiable saving is more conservative
than the generic policy cap — deliberately STRICTER than ``MAX_SAVING_FRACTION`` so this rule is an
INDEPENDENT gate: it can reject a proposal the generic P90 stage passes (not redundant). The concrete
fraction is calibrated against the reference domain; the CONDITION (a method-scoped stricter cap) is
the encoded rule. Returns the same ``Rejection`` type — a validator stage, not a new gate."""

METHOD_CAPS: dict[str, float] = {_ENERGY_METHOD_MEASURE: _ENERGY_METHOD_MAX_FRACTION}
"""S4.0 (F8): the method-cap REGISTRY — measure type -> method-scoped max saving fraction. The
rule used to be an ``if proposal.measure == "energy_efficiency"`` branch, so encoding a second
assessment method meant editing the validator. It is now data: a caller passes its own registry
(``validate_proposal(..., method_caps=...)``), keyed by the measure type a dimension admits
(``dimension.allowed_measure_types``), and the built-in entry stays the default so the Step-9
behaviour is unchanged. Deliberately NOT a config file yet — the deliverable is the key-by-config
seam (90%-prinsippet), not a settings format."""

BASELINE_TOLERANCE_DEFAULT = 0.05
"""S4.0: the relative deviation a reconciled ``AffectedItem`` may show against its cost-baseline
line (5%). A tolerance is needed at all because a proposer restates magnitudes in prose-derived,
rounded form; it is small because its whole purpose is to leave no room for a FABRICATED magnitude.
Config, not policy: every caller can tighten or loosen it per run (``tolerance=``)."""

_MC_SAMPLES = 512
_MC_SEED = 20260624


class CbcUnavailable(RuntimeError):
    """PuLP's bundled CBC solver is not available — escalate (no silent fallback)."""


@contextmanager
def _quiet_pulp():
    """Silence PuLP 3.x's ``PULP_CBC_CMD`` DeprecationWarning. The bundled CBC is only
    reachable via ``PULP_CBC_CMD``; PuLP 4.0 will require ``pip install pulp[cbc]`` +
    COIN_CMD (a migration note). The warning is cosmetic here."""
    with warnings.catch_warnings():
        warnings.simplefilter("ignore", DeprecationWarning)
        yield


@dataclass(frozen=True)
class ValidatedProposal:
    """A proposal that passed every stage. Carries the Monte Carlo percentiles."""

    proposal: SavingsProposal
    p10: float
    p50: float
    p90: float
    nominal_feasible: float


@dataclass(frozen=True)
class Rejection:
    """A structurally-blocked proposal. Distinct type, no percentiles — it can never be
    consumed as a ``ValidatedProposal``."""

    proposal: SavingsProposal
    reason: str


def _solve_max_feasible(items: list[AffectedItem], fraction: float) -> float:
    """Real CBC solve: maximize total saving subject to a per-item upper bound and a
    global fraction cap. Raises ``CbcUnavailable`` if CBC is genuinely missing."""
    with _quiet_pulp():
        solver = pulp.PULP_CBC_CMD(msg=False)
        if not solver.available():
            raise CbcUnavailable("PuLP's bundled CBC solver is not available on this platform")
        prob = pulp.LpProblem("max_feasible_saving", pulp.LpMaximize)
        xs = [pulp.LpVariable(f"x_{i}", lowBound=0, upBound=it.total) for i, it in enumerate(items)]
        prob += pulp.lpSum(xs)
        prob += pulp.lpSum(xs) <= fraction * sum(it.total for it in items)
        status = prob.solve(solver)
    if pulp.LpStatus[status] != "Optimal":
        raise CbcUnavailable(
            f"CBC did not reach an optimal solution (status={pulp.LpStatus[status]})"
        )
    return float(pulp.value(prob.objective))


def _monte_carlo(
    proposal: SavingsProposal, *, fraction: float = MAX_SAVING_FRACTION
) -> tuple[float, float, float]:
    """Vary uncertain unit-costs (seeded) and return (P10, P50, P90) of the feasible
    saving. Uses the LP's closed-form optimum (= fraction x sum of sampled totals), which
    is exact here, so we do NOT spawn a CBC subprocess per sample (D6)."""
    rng = random.Random(_MC_SEED)
    feasibles: list[float] = []
    for _ in range(_MC_SAMPLES):
        total = 0.0
        for item in proposal.affected_items:
            rng_range = proposal.assumptions.get(item.code)
            unit_cost = rng.uniform(*rng_range) if rng_range else item.unit_cost
            total += item.quantity * unit_cost
        feasibles.append(fraction * total)
    deciles = statistics.quantiles(feasibles, n=10, method="inclusive")
    return deciles[0], deciles[4], deciles[8]  # P10, P50, P90


def baseline_from_project(project: Project) -> CostBaseline:
    """Project a road reference-domain ``Project``'s ``cost_items`` into the ``CostBaseline``
    contract — the road-path counterpart of ``okf.load_cost_baseline`` (S4.0). The road path always
    HAS its baseline (the estimate is the project), so this projection is total: no optional
    variant, and a run on this path is always anchored."""
    return CostBaseline(
        project_id=project.id,
        items={
            ci.code: CostBaselineLine(quantity=ci.quantity, unit_cost=ci.unit_cost)
            for ci in project.cost_items
        },
    )


def _reconcile_against_baseline(
    proposal: SavingsProposal, baseline: CostBaseline, tolerance: float
) -> Rejection | None:
    """S4.0 (F3): every affected item must correspond to a REAL line of the project's cost baseline.

    Two independent failures, both fail-closed:

    * the cost code is absent from the baseline — a fabricated line;
    * the code is real but its ``quantity``/``unit_cost`` deviates from the baseline line by more
      than ``tolerance`` (relative to the BASELINE value, which is the ground truth) — a real code
      carrying a fabricated magnitude.

    Returns the first ``Rejection`` (validator's own type — never a new gate), or ``None`` when the
    proposal reconciles. Items are checked in their stated order so the reason is deterministic.
    A validation, never a repair: the proposal is rejected, not silently corrected to the baseline."""
    for item in proposal.affected_items:
        line = baseline.items.get(item.code)
        if line is None:
            return Rejection(
                proposal=proposal,
                reason=(
                    f"unknown cost code {item.code!r}: not in project {baseline.project_id}'s "
                    f"cost baseline ({len(baseline.items)} known codes)"
                ),
            )
        for field, claimed, actual in (
            ("quantity", item.quantity, line.quantity),
            ("unit_cost", item.unit_cost, line.unit_cost),
        ):
            if abs(claimed - actual) > tolerance * actual:
                return Rejection(
                    proposal=proposal,
                    reason=(
                        f"{field} {claimed:g} for cost code {item.code!r} is outside the "
                        f"{tolerance:.1%} tolerance around the baseline {field} {actual:g}"
                    ),
                )
    return None


def validate_proposal(
    proposal: SavingsProposal,
    *,
    baseline: CostBaseline | None = None,
    tolerance: float = BASELINE_TOLERANCE_DEFAULT,
    method_caps: Mapping[str, float] | None = None,
) -> ValidatedProposal | Rejection:
    """Deterministic blocking validation. Returns a ``ValidatedProposal`` only when the
    claim is feasible; otherwise a ``Rejection`` that cannot be consumed as validated.

    ``baseline`` (S4.0, F3) anchors the gate to the project's ACTUAL cost lines: without it every
    stage reasons only about numbers the proposal supplied itself, so an internally-consistent
    hallucination clears the gate. It is OPTIONAL — ``None`` is exactly the pre-S4.0 behaviour, so a
    caller with no baseline (a bundle authored before the amendment) is unchanged — but both run
    paths SET it. ``tolerance`` is the reconciliation's config knob; ``method_caps`` overrides the
    built-in method-cap registry (F8)."""
    # Stage 0 (S4.0): reconcile against the cost baseline BEFORE the solver. It is the cheapest
    # stage and the only one that can tell a fabricated line from a real one — spending a CBC solve
    # on numbers that do not belong to the project is work on a claim that cannot be validated.
    if baseline is not None:
        blocked = _reconcile_against_baseline(proposal, baseline, tolerance)
        if blocked is not None:
            return blocked
    # Stage 1 (Pydantic) already ran at construction. Stage 2: real CBC solve.
    nominal = _solve_max_feasible(proposal.affected_items, MAX_SAVING_FRACTION)
    # Stage 3: Monte Carlo percentiles of the feasible saving.
    p10, p50, p90 = _monte_carlo(proposal)
    # Stage 4: structural block — a claim above the optimistic feasible (P90) is out of range.
    if proposal.claimed_saving_nok > p90:
        return Rejection(
            proposal=proposal,
            reason=f"claimed saving {proposal.claimed_saving_nok:.0f} exceeds P90 feasible {p90:.0f}",
        )
    # Stage 4b (S2.7): the validator enforces its OWN stage-2 boundary. The CBC solve already
    # established the nominal feasible saving at the items' stated unit-costs; a claim above it
    # is out of range no matter how the uncertainty bands fall. This is an INDEPENDENT gate, not
    # a restatement of the P90 stage: an upward-skewed band lifts P90 ABOVE nominal (so P90 alone
    # would pass a claim the deterministic bound rejects), while a downward-skewed one pushes P90
    # below it. Neither stage dominates, so both are kept.
    if proposal.claimed_saving_nok > nominal:
        return Rejection(
            proposal=proposal,
            reason=(
                f"claimed saving {proposal.claimed_saving_nok:.0f} exceeds the nominal feasible "
                f"{nominal:.0f} at the items' stated unit-costs"
            ),
        )
    # Stage 5 (Step 9, SC7-B): a method-specific rule STRICTER than the generic cap. A proposal in
    # the energy method (IPMVP Option A) must clear a lower, method-scoped feasible — an INDEPENDENT
    # gate that can reject a proposal the P90 stage passed. Same ``Rejection`` type, not a new gate.
    # F8 (S4.0): the cap is looked up in a REGISTRY keyed by measure type (config), not compared
    # against the ``energy_efficiency`` literal — a second assessment method is now data, not an
    # edit to this function. The built-in registry keeps the Step-9 behaviour identical.
    caps = METHOD_CAPS if method_caps is None else method_caps
    method_fraction = caps.get(proposal.measure)
    if method_fraction is not None:
        method_feasible = method_fraction * sum(it.total for it in proposal.affected_items)
        if proposal.claimed_saving_nok > method_feasible:
            return Rejection(
                proposal=proposal,
                reason=(
                    f"claimed {proposal.claimed_saving_nok:.0f} exceeds the {proposal.measure} "
                    f"method cap {method_feasible:.0f} (stricter than the generic P90)"
                ),
            )
    return ValidatedProposal(proposal=proposal, p10=p10, p50=p50, p90=p90, nominal_feasible=nominal)


def self_repair(
    generate: Callable[[int], SavingsProposal],
    *,
    max_attempts: int = 3,
) -> ValidatedProposal | Rejection:
    """Call ``generate(attempt)`` and validate; retry on rejection up to ``max_attempts``,
    then hard-stop and return the last rejection. Attempts-bounded — never loops forever
    (B4). The token-budget bound is layered on by the Step 10 generate loop, not here (this
    module stays pure: no ``agent_framework`` import)."""
    if max_attempts <= 0:
        raise ValueError(f"max_attempts must be positive, got {max_attempts}")
    last: Rejection | None = None
    for attempt in range(1, max_attempts + 1):
        result = validate_proposal(generate(attempt))
        if isinstance(result, ValidatedProposal):
            return result
        last = result
    assert last is not None
    return last


def proposal_for(
    project: Project,
    codes: list[str],
    *,
    claimed_saving_nok: float,
    measure: str = "Reduce scope on selected cost codes",
    assumptions: dict[str, tuple[float, float]] | None = None,
) -> SavingsProposal:
    """Build a ``SavingsProposal`` from a real reference project's cost items (helper)."""
    items = [
        AffectedItem(code=ci.code, quantity=ci.quantity, unit_cost=ci.unit_cost)
        for ci in project.cost_items
        if ci.code in codes
    ]
    return SavingsProposal(
        project_id=project.id,
        measure=measure,
        affected_items=items,
        claimed_saving_nok=claimed_saving_nok,
        assumptions=assumptions or {},
    )