portfolio-optimiser/src/portfolio_optimiser/validator.py
Kjell Tore Guttormsen 938a1ca30e feat(row6): a proposal whose approach declared no requirement is unsupported
Stress round 6 validated three falsification arms, and every validated
approach rested only on run-level declarations nobody can attribute to one
approach. declare_requirement now takes a required approach_id (a mandate
id or own-proposal; an unknown id is refused naming the valid ones), and a
ValidatedProposal whose approach has neither a mandate requirement nor a
declaration under its own id becomes validator.Unsupported - a Rejection
subclass carrying the validator's own ruling, reported as `unsupported` in
coverage, the outcome artefact, the settlement and the judge, and never
counted or summed. The rule is active whenever the debate held the
declaration tool, the micro base included; the road and pre-pass paths are
untouched. Declaration quality is not judged, so the rule can be satisfied
by declaring any document the run read.

The v1 gate's row 6 probes pass; its artefact half reads IKKE MÅLT because
stress round 6 predates approach-addressed declarations, and IKKE MÅLT is
never green - it fails the exit code.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-09-17 16:40:54 +02:00

819 lines
44 KiB
Python

"""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 re
import statistics
import warnings
from collections.abc import Callable, Mapping, Sequence
from contextlib import contextmanager
from functools import cached_property
from dataclasses import dataclass
from typing import Final
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
#: The ONE sentence an unsupported outcome carries. ``rejection_stage`` keys on it, so the judge can
#: tell this falsifier from the numeric ones without a second copy of the wording.
UNSUPPORTED_REASON: Final = (
"no declared requirement for this approach: the numbers held, but no requirement of the "
"knowledge base was declared as binding it"
)
@dataclass(frozen=True)
class Unsupported(Rejection):
"""A proposal whose NUMBERS held but whose approach declared no binding requirement (row 6).
Neither ``validated`` (nothing in the knowledge base was said to support the direction) nor an
ordinary rejection (every numeric stage passed). It subclasses ``Rejection`` on purpose: every
consumer that asks "is this validated?" with ``isinstance(..., ValidatedProposal)`` answers no
without being touched, so it is never counted, summed or carried as a success. The consumers
that NAME the status — coverage, the outcome artefact, the settlement, the judge — check for
this class first. ``validated`` keeps the validator's own ruling, which is what
``provenance.validator_decision`` mirrors: the validator said yes, and the record says so.
"""
validated: ValidatedProposal
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
},
)
#: P22 DEL A - how many of the project's OWN cost codes the stage-0 refusal names.
#:
#: A FIXED WINDOW, never a share of the schedule: a share scales with the corpus again, only with
#: a smaller constant, which is the exact regression the catalogue excerpt's
#: ``_CATALOGUE_EXCERPT_CHARS`` exists for. It counts CODES rather than characters for a measured
#: reason - a character cut can sever a code mid-name and hand the proposer an identifier that
#: exists nowhere, the ``_index_excerpt`` rule inverted (a path that never existed is worse than
#: no path). A count window can only ever emit whole codes.
#:
#: MEASURED with denominators (okt 126): every cost baseline anywhere in this repo or its measured
#: corpora is at most SIX codes - the five context sets carry 5/5/5/5/6, the two shipped
#: ``shared/examples`` baselines carry 1 each, and MAJOR-4's derivation of the synthetic K2 price
#: schedule yields 3 - while the largest REAL delivered price schedule measured is K2's
#: ``prissammenstilling-sheet-1.md``, 14 priced rows of 118 lines. Nothing measured reaches this
#: window. It exists for the unmeasured R761-style mengdebeskrivelse, where a Norwegian road
#: contract is priced BY prosesskode and the corpus declares 2 727 of them.
_KNOWN_CODE_WINDOW: Final = 20
def _known_codes_clause(baseline: CostBaseline) -> str:
"""The parenthetical of stage 0's unknown-code refusal: how many codes the project buys, and
which ones (P22 DEL A).
MEASURED (P21 round 5): all 26 rejections across six paid runs named an invented code and
answered with a COUNT - ``(5 known codes)`` - so the sentence Step 5 feeds verbatim into the
next attempt carried nothing to correct towards, and 0 of 20 approaches validated. The
magnitude half of this same stage NAMES the baseline value, and that is the half that let the
loop converge in okt 94. This is that half's property, given to the other one.
Bound by ``_KNOWN_CODE_WINDOW`` - a fixed count of WHOLE codes, see the constant. The cut is
ANNOUNCED (``first N:``) rather than left for the reader to subtract from the denominator,
which stays in either branch; a schedule that FITS is not marked truncated and gets its whole
list. Omission, never a lie in either direction (``index_truncated``'s rule).
The order is the SCHEDULE's own. A sort would invent a ranking the project never stated, and
the window would then be an alphabetical accident rather than the head of the document the
operator wrote."""
codes = list(baseline.items)
listed = ", ".join(repr(c) for c in codes[:_KNOWN_CODE_WINDOW])
if len(codes) <= _KNOWN_CODE_WINDOW:
return f"{len(codes)} known codes: {listed}"
return f"{len(codes)} known codes, first {_KNOWN_CODE_WINDOW}: {listed}"
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 ONE ``Rejection`` (validator's own type — never a new gate) naming EVERY violation the
attempt carries, or ``None`` when the proposal reconciles. A validation, never a repair: the
proposal is rejected, not silently corrected to the baseline.
**Completeness is load-bearing (K2 finding (b), measured live in økt 94).** Step 5 feeds this
reason VERBATIM into the next attempt's prompt, so a message naming only the FIRST violation
reads as an instruction to fix that one field. Measured on a line broken in both fields
(``docs/2026-09-06-major2-levende-k2.md`` § 4, run 3): the proposer fixed the quantity and
rebroke the unit cost, then fixed the unit cost and rebroke the quantity, and ran out of
attempts — it found each correct value and never both at once. Reporting them together is what
lets the loop converge under the EXISTING cap (``max_attempts`` is not raised, and stays
unexposed). The verdict itself (D6) is unchanged: any violation still rejects, in this stage,
before the solver.
Each violation keeps its sentence VERBATIM, joined with ``"; "``, so exactly one violation
renders byte-identically to before. The joiner is a separator no single-line renderer can
break — chosen over a newline because ``Rejection.reason`` also lands in outbox JSON, the
hosted payloads and terminal notices. Order is the PROPOSAL's own — items in stated order,
``quantity`` before ``unit_cost`` within an item — so two identical attempts get two identical
prompts. An unknown code contributes its one sentence and NO magnitude sentences: there is no
baseline line for its figures to deviate from, and a comparison against nothing is exactly the
fabrication this stage exists to catch."""
violations: list[str] = []
for item in proposal.affected_items:
line = baseline.items.get(item.code)
if line is None:
violations.append(
f"unknown cost code {item.code!r}: not in project {baseline.project_id}'s "
f"cost baseline ({_known_codes_clause(baseline)})"
)
continue
for field, claimed, actual in (
("quantity", item.quantity, line.quantity),
("unit_cost", item.unit_cost, line.unit_cost),
):
if abs(claimed - actual) > tolerance * actual:
violations.append(
f"{field} {claimed:g} for cost code {item.code!r} is outside the "
f"{tolerance:.1%} tolerance around the baseline {field} {actual:g}"
)
if not violations:
return None
return Rejection(proposal=proposal, reason="; ".join(violations))
#: P18/B1 — the shortest identifier the gate will let ground anything.
#:
#: MEASURED (14.09) over every ``must_cite`` reference and every mandate ``affected_code`` in the
#: four context sets: the shortest real identifier is FOUR characters (R761's ``12.1`` / ``52.1``).
#: Set one BELOW that, so the rule cannot refuse anything that has been measured, while a one- or
#: two-character token — which matches by coincidence in any prose — grounds nothing. The honest
#: failure direction for a gate that speaks about a model's invention.
#:
#: Length is NOT what makes the measured defect inert: P16's fabricated ``R761`` is four characters
#: long. The share below is what does. This covers the coincidence class the measurement did not
#: happen to contain.
_GROUNDING_MIN_LENGTH: Final = 3
#: The share of the grounding's DOCUMENTS above which a token identifies nothing.
#:
#: MEASURED over the four delivered corpora, counting document frequency for every code-shaped
#: token (``generate._IDENTIFIER_FORMS``): 1 692 distinct tokens, and NOT ONE reaches 5 % of its
#: base's documents. The highest anywhere is 6 of 446 (1.35 %); the highest that a fasit or mandate
#: actually names is 3 of 446 (0.67 %). P16's fabricated ``R761`` is 2 756 of 2 756 — 100 %.
#: 5 % therefore sits 3.7x above the highest real token measured and 20x below the defect.
_GROUNDING_MAX_DOCUMENT_SHARE: Final = 0.05
#: …and a share is not a measurement without a denominator big enough to take one (ansikt 4).
#: One document of three is 33 % and says nothing at all, so the share only fires once a token is
#: in at least this many documents. MEASURED: the highest ABSOLUTE document count any real
#: identifier reaches in the four corpora is 6, and every test fixture in this repo is far below
#: 10 — which is why every pre-P18 gate is untouched by this rule rather than exempted from it.
_GROUNDING_MIN_INERT_DOCUMENTS: Final = 10
#: The identifier forms the DELIVERED corpora actually carry, TRANSCRIBED from the measurements
#: (``docs/2026-09-09-p8-forankringstilbudet.md`` § 2 for the first two, P19 DEL B for the third
#: and fourth) rather than chosen. They live HERE, in the validator, because P19/B3 made them a
#: GATE as well as P8's report, and ``generate`` imports them from here: two copies of "what an
#: identifier looks like" would let the report and the gate disagree about one run (kø-(p)).
#:
#: Bare numbers are deliberately EXCLUDED, with the number: K2 carries 46 394 occurrences over
#: 2 117 distinct values (P7 § 2), so counting them would make every report positive and the
#: measurement inert — the repo's cardinal class, a gate that can only come out green.
#: The four forms, NAMED rather than reached by index: P20/B needs two of them by themselves
#: (a requirement/process number), and ``IDENTIFIER_FORMS[1:3]`` in a second module would be a
#: positional dependency on a tuple literal — the kø-(p) shape with no compiler to catch it.
_FORM_SEPARATED_UPPER: Final = re.compile(r"\b[A-ZÆØÅ][A-ZÆØÅ0-9]*(?:[-_][A-ZÆØÅ0-9]+)+\b")
_FORM_REQUIREMENT_NUMBER: Final = re.compile(r"Krav\s+\d+(?:\.\d+)*\s*[\u2014-]\s*\d+(?:_\d+)?")
_FORM_PROCESS_NUMBER: Final = re.compile(r"(?<![\d.])[1-9]\d{0,2}(?:\.\d{1,3}){1,4}\b(?!\.\d)")
_FORM_PROCESS_HEADING: Final = re.compile(r"(?<!\d )(?<![\d.])[1-9]\d{0,2} [A-ZÆØÅ]{5,}\b")
IDENTIFIER_FORMS: Final = (
# ``SHA-01``, ``RIM-02``, ``B-20-00-00``, ``FOR-2011-12-06-1357`` (K2's 50) — and, since P19/B2
# made the same forms decide ``prose`` vs ``identifier``, an UPPERCASE separated token with no
# digits at all. MEASURED: this repo's own ``ENERGI-TOTAL-EL`` matched neither of the pre-P19
# forms, so the classifier called a real cost code prose; a gate is only allowed to be wrong in
# the direction that admits too much.
_FORM_SEPARATED_UPPER,
# ``Krav 3.3.1—13`` (the N corpora's dominant form). EM-DASH U+2014 AND the hyphen, because the
# binding known positive is the em-dash spelling and only the em-dash spelling scores 6 of 6.
# The trailing ``(?:_\d+)?`` is MEASURED, not defensive: one of the 26 fasit references is
# ``Krav 3.3.2—1_1``, and without it the classifier called that real reference prose.
_FORM_REQUIREMENT_NUMBER,
# R761's process numbers, ``12.1`` / ``52.11`` (P19 B1). MEASURED: all six ``ref`` values in
# ``contexts/kontrakt-sorasen-2027/fasit.json`` are of this shape and NEITHER of the first two
# forms matches one of them, so r761's whole offer was 3 identifiers over 6.5 MB. The trailing
# ``(?!\.\d)`` is what keeps a Norwegian date out: ``15.09.2026`` would otherwise contribute
# its ``15.09`` prefix, and a date is not a requirement.
_FORM_PROCESS_NUMBER,
# ``65 ASFALTDEKKER`` — a process number and its heading, the form a price schedule's section
# rows carry (P18 § 2 measured it at 29 of 2 756 documents).
_FORM_PROCESS_HEADING,
)
#: The two forms a REQUIREMENT or PROCESS number takes (P20/B). MEASURED over the four delivered
#: bases: these are the shapes that live in ``req_number`` ("Krav 4.1.2—1"), ``prosessnr``
#: ("'11.11'") and ``seksjon`` ("'10.4.3'") — the fields a base uses to number its own clauses.
#: The other two forms are NOT here: ``SHA-01``-style tokens are what a price schedule's cost lines
#: look like, and ``65 ASFALTDEKKER`` IS a schedule section row.
REQUIREMENT_FORMS: Final = (_FORM_REQUIREMENT_NUMBER, _FORM_PROCESS_NUMBER)
def has_requirement_form(code: str) -> bool:
"""Whether ``code`` is shaped like a requirement or process number. FULL-MATCH, never a search,
for ``has_identifier_form``'s reason: ``impulsventilator 12.1`` is not a clause number."""
return any(form.fullmatch(code) for form in REQUIREMENT_FORMS)
def identifier_tokens(text: str) -> set[str]:
"""Every DISTINCT token of any ``IDENTIFIER_FORMS`` shape in ``text``. One reader, two callers.
A form it does not know is a token it fails to count, so it errs toward saying the input offers
LESS than it does. That direction is deliberate and is what makes the same function admissible
on both sides of P19/B3: an under-count keeps the gate OFF, never turns it on wrongly.
"""
found: set[str] = set()
for form in IDENTIFIER_FORMS:
found |= set(form.findall(text))
return found
def has_identifier_form(code: str) -> bool:
"""Whether ``code`` is shaped like an identifier at all — the ``prose`` / ``identifier`` split.
FULL-MATCH, never a search: ``impulsventilator 12.1`` containing a process number does not make
the word a cost code, and a substring rule here would let any prose code carry one along.
"""
return any(form.fullmatch(code) for form in IDENTIFIER_FORMS)
def classify_codes(codes: Sequence[str], grounding: Grounding | None = None) -> dict[str, str]:
"""``{code: "identifier" | "prose" | "requirement"}`` — P19/B2's report, widened by P20/B, in
ONE place for both consumers.
``requirement`` is the third value: a code shaped like a clause number (``REQUIREMENT_FORMS``)
AND declared as one by the input's own documents. It is a REPORT about what the run made of a
code, not the gate — the gate is ``_reference_refusal`` below and fires on the COMPLEMENT, a
requirement-shaped code the base's vocabulary does NOT contain.
``grounding=None`` is the pre-P20 answer exactly: without the input there is no vocabulary to
check against, so no code can be called a requirement. ``stress.py`` re-derives with ``None``
for runs that predate the field, and says so.
"""
vocabulary = frozenset() if grounding is None else grounding.reference_vocabulary
out: dict[str, str] = {}
for code in codes:
if has_requirement_form(code) and code in vocabulary:
out[code] = "requirement"
else:
out[code] = "identifier" if has_identifier_form(code) else "prose"
return out
@dataclass(frozen=True)
class Grounding:
"""The run's non-model-authored input, carried as the DOCUMENTS it is made of.
P7 carried it as ONE string, and P16 measured what that costs: ``R761`` — the base's own NAME,
which every one of its 2 756 concept documents carries — satisfied ``code in grounding`` and
carried a fabricated 250 000 NOK line through the whole gate to ``validated``. Containment in a
concatenation cannot tell "this project has such a line" from "this word is in the letterhead".
A structure rather than a second argument beside the text: the boundaries and the text are ONE
fact, and two carriers for one fact are free to disagree about it (kø-(p)). ``text`` is derived
here, so the gate and P8's report measure the very same characters.
A caller with no boundaries to declare — the road path, a test — builds ONE document from its
text and is unchanged by construction: one document can never reach
``_GROUNDING_MIN_INERT_DOCUMENTS``, so the share cannot fire on it.
"""
documents: tuple[str, ...]
#: Every reference number the input's documents DECLARE in their own top-level frontmatter
#: (``okf.declared_reference_numbers``), one entry per declaration — the base's own vocabulary
#: of requirement, process and section numbers. P20/B checks a requirement-shaped code against
#: it.
#:
#: DEFAULTED, the ``skipped_links`` half rather than ``cost_baseline_anchored``'s: an empty
#: vocabulary is an honest POSITIVE statement ("this input declares no clause numbers"), and it
#: is what keeps every caller written before today — the road path, every fixture, ``of`` —
#: unchanged by construction, since the gate cannot fire without one.
declared_references: tuple[str, ...] = ()
@classmethod
def of(cls, text: str) -> Grounding:
"""One document. The honest reading of a caller that declared no boundaries."""
return cls(documents=(text,))
@property
def text(self) -> str:
"""The ONE composition. Byte-identical to P7's ``"\n".join`` of the same parts."""
return "\n".join(self.documents)
def document_frequency(self, token: str) -> int:
"""How many of the documents contain ``token`` — the numerator, in the unit of the rule."""
return sum(1 for document in self.documents if token in document)
@cached_property
def reference_vocabulary(self) -> frozenset[str]:
"""The DISTINCT reference numbers this input declares. The denominator P20/B names."""
return frozenset(self.declared_references)
@cached_property
def identifiers(self) -> frozenset[str]:
"""Every distinct identifier-shaped token this input OFFERS (P8's count, P19/B3's guard).
Cached because the gate asks for it once per attempt and the composed text is corpus-sized
(K2: 2 005 561 chars). ``cached_property`` writes through the instance ``__dict__``, which a
frozen dataclass still has, so the object stays immutable to every caller.
"""
return frozenset(identifier_tokens(self.text))
def _inert_in(grounding: Grounding, code: str) -> str | None:
"""Why ``code`` identifies nothing in this input, or ``None`` when it identifies something.
The message NAMES THE DENOMINATOR, because "it is everywhere" and "it is not here" are
different findings and Step 5 feeds this reason verbatim into the next attempt's prompt: a
proposer told only "ungrounded" will re-answer with another token of the same kind.
"""
if len(code) < _GROUNDING_MIN_LENGTH:
return (
f"is {len(code)} characters long, too short to identify a cost line — any prose "
"contains it by coincidence"
)
hits = grounding.document_frequency(code)
total = len(grounding.documents)
floor = max(_GROUNDING_MIN_INERT_DOCUMENTS, total * _GROUNDING_MAX_DOCUMENT_SHARE)
if hits >= floor:
return (
f"appears in {hits} of the {total} documents this run was given — a token that is in "
"every document identifies none of them; name a cost line, not the corpus"
)
return None
def _form_refusal(grounding: Grounding, code: str, anchored_codes: frozenset[str]) -> str | None:
"""Why ``code`` cannot be a cost code of THIS input, or ``None`` (P19/B3).
**The guard is what makes this a rule and not a preference.** MEASURED over P18's round 2: two
ordinary words from a standard's prose — ``impulsventilator`` (4 of 270 N500 documents) and
``bituminøst bærelag`` (4 of 1 133 N200 documents) — passed the whole gate to ``validated`` as
``affected_item`` codes. Both are GROUNDED: they appear verbatim in the input, which is all P7
asks. What they are not is an identifier of a cost line.
The rule fires ONLY when the input demonstrably offers identifier forms. A base that carries
none cannot be answered in a form it does not use, and refusing there would be a rule about
shapes rather than about the corpus — the generality guard, stated rather than assumed.
A code the BASELINE carries is exempt, and that is not leniency: stage 0 has already ruled it a
real line of this project, and the weaker stage must never overrule the stronger falsifier
(the same sentence ``_grounding_text`` carries about its third source). A derived schedule whose
codes are bare section numbers would otherwise be refused wholesale by the gate meant to protect
it.
"""
if code in anchored_codes or has_identifier_form(code):
return None
offered = grounding.identifiers
if not offered:
return None
sample = ", ".join(sorted(offered)[:3])
return (
f"has no identifier form; the input this run was given offers {len(offered)} identifiers "
f"of its own (for example {sample}) — name a cost line by its identifier, not by a word "
"from the prose"
)
def _reference_refusal(grounding: Grounding, code: str) -> str | None:
"""Why a requirement-shaped ``code`` cannot be a cost line of an UNANCHORED input (P20/B).
**The order's own rule was FELLED BY MEASUREMENT before anything was built on it.** It reads:
a code is a requirement when it matches form 2 or 3 AND "står som ``req_number``/``prosessnr``
i toppnivå-frontmatter" — refuse that. Measured 15.09 against the two known positives the same
order names:
* ``10.4`` (n500, tunnel-04, ``validated``) is declared NOWHERE in n500's frontmatter. The base
declares ``seksjon: 10.4.1`` … ``10.4.4`` and ``req_number: Krav 10.4.3—2``; the bare ``10.4``
is a section PREFIX that occurs in 12 of 274 documents and is no document's own number;
* ``1.10.4`` (r761, lindaas a4, ``validated``) is not one of r761's 2 727 ``prosessnr`` nor one
of its 2 753 ``seksjon`` values. It occurs in ONE of 2 756 documents, as prose: "iht.
vegnormal N200 Vegbygging kap. 1.10.4".
So the ordered rule fires on NEITHER of its own known positives. The COMPLEMENT does, and it is
the better-grounded rule besides: ``_ground_against_input``'s docstring already admits that this
stage "fails OPEN … on a coincidental match", and for one shape — a clause number — the base
hands us the vocabulary needed to close exactly that hole. A form-3 token that is NOT one of the
numbers this base declares was matched in prose by accident.
MEASURED over every code of round 3 and P17b (24 codes, 10 runs): exactly two are
requirement-shaped, they are the two known positives, and neither is in its base's vocabulary.
All five of ``contexts/kontrakt-sorasen-2027``'s REAL process codes (``12.1``, ``12.12``,
``22.1``, ``52.11``, ``51.1``) ARE declared ``prosessnr`` and pass — which is what keeps the
R761 risk the order names (a process number is both a clause and a settlement post) from
turning into a wholesale refusal of the one context set built on real codes.
**The generality guard, ``_form_refusal``'s pattern:** an input that declares no reference
numbers at all cannot be answered in a vocabulary it does not have, so the rule cannot fire
there. That is what leaves every pre-P20 fixture untouched rather than exempted.
The message NAMES THE DENOMINATOR (ansikt 4, and Step 5 feeds it verbatim into the next
attempt): "not one of the 2 765 it declares" is actionable where "ungrounded" is not.
"""
if not has_requirement_form(code):
return None
vocabulary = grounding.reference_vocabulary
if not vocabulary or code in vocabulary:
return None
sample = ", ".join(sorted(vocabulary)[:3])
return (
f"is shaped like a requirement or process number, but it is not one of the "
f"{len(vocabulary)} this knowledge base declares (for example {sample}) — it was matched "
"in prose by coincidence, and an unanchored base carries no price for a clause number"
)
def _ground_against_input(
proposal: SavingsProposal,
grounding: Grounding,
anchored_codes: frozenset[str] = frozenset(),
*,
anchored: bool = True,
) -> Rejection | None:
"""P7: every identifier the proposal builds on must appear VERBATIM in the input it was built
from, or the verdict falls.
**Why this is a SEPARATE stage and not a widening of stage 0.** ``_reconcile_against_baseline``
already carries the sentence "the cost code is absent from the baseline — a fabricated line",
and it is right. But it is reached only through ``if baseline is not None``, so the falsifier
is tied to whether a cost baseline happens to exist — and MEASURED (økt 108, verdict
``5fd6272e3725fe68``), an unanchored K2 run ended in ``ValidatedProposal`` on two cost codes
(``M-04-01`` / ``M-04-03``) that appear in NO prompt of that run. **The input always exists;
the baseline does not.** This stage therefore runs on its own, whatever ``baseline`` is.
**Exact substring, no pattern.** The check is ``code in grounding``: nothing here needs to know
what an identifier LOOKS like, so nothing here can be wrong about a form the corpus carries.
That is deliberate — measured over the delivered corpora (K2: 1 108 concept files / 2 005 561
chars; the three N payloads: 8 delivered excerpts each), the identifier forms are heterogeneous
(499 ``UPPER-num`` occurrences / 23 unique and 25 single-letter ``B-20-00-00``-style codes in
K2; requirement numbers such as ``Krav 3.3.1—13`` that live in ``req_number``/``title`` and
never in an excerpt body; 71 UUIDs in one payload), and a pattern chosen to cover them would
be a rule about shapes rather than about grounding.
It fails OPEN, never closed, on a coincidental match: measured, K2 carries 46 394 bare-number
occurrences over 2 117 distinct values, so a numeric-only code is almost always "grounded" by
accident. That is a known weakness of this stage, never a false rejection — the honest failure
direction for a gate that speaks about a model's invention.
Returns ONE ``Rejection`` (the validator's own type — never a new gate) naming EVERY ungrounded
identifier, ``"; "``-joined, in the PROPOSAL's own order, exactly as ``_reconcile_against_
baseline`` does and for the same measured reason (økt 94): Step 5 feeds this reason verbatim
into the next attempt's prompt, and a message naming only the first violation reads as an
instruction to fix that one and leave the rest.
Only ``affected_items`` codes are checked. An ``assumptions`` key naming no affected item is
deliberately out of scope: the Monte Carlo never samples such a band (``SavingsProposal.
_assumption_bands_enclose_unit_cost`` says so in the same words), so it cannot move the verdict,
and a check on it would be a branch no recording exercises."""
violations = []
for item in proposal.affected_items:
if item.code not in grounding.text:
violations.append(
f"ungrounded identifier {item.code!r}: it appears nowhere in the input this "
f"proposal was built from ({len(grounding.text)} chars)"
)
continue
# P18/B1: present is not the same as identifying. An identifier that stands everywhere
# identifies nothing, and one too short to be an identifier is matched by coincidence.
inert = _inert_in(grounding, item.code)
if inert is not None:
violations.append(f"ungrounded identifier {item.code!r}: it {inert}")
continue
# P19/B3: present, and not everywhere — but still not an identifier at all.
shapeless = _form_refusal(grounding, item.code, anchored_codes)
if shapeless is not None:
violations.append(f"ungrounded identifier {item.code!r}: it {shapeless}")
continue
# P20/B: shaped, grounded, not inert — and still a clause number the base never declared.
# UNANCHORED only: with a baseline, stage 0 has already ruled every code that reaches here
# a real line of this project, and the weaker stage must not overrule the stronger one (the
# sentence ``_form_refusal`` and ``_grounding_text`` both carry).
if not anchored:
coincidental = _reference_refusal(grounding, item.code)
if coincidental is not None:
violations.append(f"ungrounded identifier {item.code!r}: it {coincidental}")
if not violations:
return None
return Rejection(proposal=proposal, reason="; ".join(violations))
def validate_proposal(
proposal: SavingsProposal,
*,
baseline: CostBaseline | None = None,
grounding: Grounding | 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).
``grounding`` (P7) is the input text the proposal must be grounded in — the rendered prompt the
model actually received. It is OPTIONAL for the same reason ``baseline`` is (``None`` = the
pre-P7 gate, so every caller and every golden is unchanged), but it closes a DIFFERENT hole:
stage 0 only fires when a baseline exists, while the input exists always."""
# 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 0b (P7): every identifier the proposal builds on must appear verbatim in the input. It
# sits OUTSIDE the baseline branch above -- that is the whole point: stage 0 is the stronger
# check but only an anchored run has it, while an unanchored run had NO falsifier for a
# fabricated code at all. Placed AFTER stage 0 so an anchored run's message is byte-identical
# to before: where both would fire, the baseline's sentence is the more actionable one (it
# names the project and how many codes it knows), and Step 5 feeds that sentence back.
if grounding is not None:
adrift = _ground_against_input(
proposal,
grounding,
frozenset() if baseline is None else frozenset(baseline.items),
# P20/B: an EXPLICIT flag, never ``not anchored_codes``. A baseline with no items and
# no baseline at all are different facts, and conflating them is the very shape this
# repo refuses elsewhere (``cost_baseline_anchored`` is required without a default for
# the same reason).
anchored=baseline is not None,
)
if adrift is not None:
return adrift
# 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)
#: P21 B3 — which falsifier a ``Rejection`` came from, keyed on the SENTENCES this module writes.
#:
#: It lives HERE, next to the wordings, and never in the judge: a second copy of "what a stage 0
#: refusal looks like" would be free to drift from the sentence the validator actually emits, and
#: the reader most likely to be misled is the one re-judging a paid run months later (kø-(p)).
#:
#: In PIPELINE order, which is also the only order that can be right: ``validate_proposal`` returns
#: at the FIRST failing stage, so one reason carries violations from exactly one of them.
_REJECTION_STAGES: Final = (
("stage0-baseline", ("cost baseline (", "tolerance around the baseline ")),
("stage0b-grounding", ("ungrounded identifier ",)),
("stage4-p90", ("exceeds P90 feasible",)),
("stage4b-nominal", ("exceeds the nominal feasible",)),
("stage5-method-cap", ("method cap",)),
("unsupported", ("no declared requirement for this approach",)),
)
def rejection_stage(reason: str) -> str:
"""Which stage of the deterministic gate wrote ``reason`` — ``"other"`` when none did.
A REPORT, never a gate: nothing branches on the answer, so an unrecognised sentence costs a
label and not a verdict. That is why ``"other"`` is an honest answer here and would not be one
inside the pipeline.
The measured reason it exists (P21 B3). Before a project price schedule, the ``must_refuse``
arm of every context set fell — when it fell at all — on stage 0b, P7's grounding check, which
can only say "this identifier is not in the delivered text". Stage 0 is the one stage that
knows what the PROJECT buys, and it was skipped in every paid run measured, because no road
normal ships a ``cost-baseline.json``. Saying which stage caught the falsification arm is how a
reader can tell an anchored refusal from an un-anchored one that happened to land."""
for stage, markers in _REJECTION_STAGES:
if any(marker in reason for marker in markers):
return stage
return "other"
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 {},
)