feat(fase2): promote blocking validator + IR to src

src/portfolio_optimiser/validator.py

@@ -0,0 +1,173 @@
+"""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
+from contextlib import contextmanager
+from dataclasses import dataclass
+
+import pulp
+
+from portfolio_optimiser.ir import AffectedItem, 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."""
+
+_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 validate_proposal(proposal: SavingsProposal) -> ValidatedProposal | Rejection:
+    """Deterministic blocking validation. Returns a ``ValidatedProposal`` only when the
+    claim is feasible; otherwise a ``Rejection`` that cannot be consumed as validated."""
+    # 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}",
+        )
+    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 {},
+    )