feat(s7b): IR-projeksjonen er valgfri paa alle tre kallsteder, og fraveret sies

DEL A + DEL C av ordre 20260903T204605Z-215167684-from-.claude.

Soem 1: okf.load_optional_ir_projection ved siden av den fail-faste, moensteret
fra load_cost_baseline/load_optional_cost_baseline. Et PAR, ikke et required=-flagg:
PM-tillegg 5 maalte hva en uoevet parameter koster (elleve evidence_for-kallsteder
brukte defaulten til den andre grenen raatnet), og et flagg ville dessuten gjort
usanne de fem docstringene som siterer load_ir_projection som DEN fail-faste
presedensen. Toleransen stopper ved fravaer: en malformed projeksjon reiser fortsatt.

De tre kallstedene fikk HVER SIN stilling:
  * _project_from_bundle - fravaer hopper over en fail-fast som ikke har noe aa
    sjekke mot; en projeksjon som FINNES og navngir et annet prosjekt nekter
    fortsatt. Divergens-vakten er kontrakten multi-base-dispatchen hviler paa.
  * run_mandate_across_bundles - FILA FOERST, basens ERKLAERTE bundle_id som
    fallback (S7a-3). Presedensen baerer i begge retninger: erklaering-foerst ville
    re-adressert hver eksisterende base der project_id != bundle_id.
  * bundle_candidate_features - optional_bundle_candidate_features, og de TO
    konsumentene svarer ULIKT paa fravaeret. Steg-1-folden HOPPER OVER og sier
    hvor mange tidligere dommer som dermed aldri naadde hypotese-prompten;
    seed_store_from_bundle NEKTER ved navn (VerdictKeyUnavailable), fordi aa mynte
    en noekkel for en dom som erklaerer ingen er nettopp defekten S3.2 lukker.

Synligheten: RunResult.unkeyed_verdicts (ANTALL, ikke flagg - koe-(y)-regelen) +
run.unkeyed_verdicts_notice som ENESTE renderer, None ved null (omisjon, aldri tom
rad). Baereren er MAALT: dry-run-kuttet returnerer OVER folden, saa et felt paa
DryRunReport kunne bare rapportert null - ulikt cost_baseline_anchored og
skipped_links, begge opploest over kuttet. Ikke paa ProvenanceStamp: stempelet
beskriver gaten som doemte EN kandidat.

Prosjektnavnet var et ikke-spoersmaal, og det er maalt: SavingsProposal har intet
navnefelt, saa projeksjonen har aldri vaert en navnekilde. Project.name kommer
fortsatt fra type: project-konseptets title med id-en som siste utvei.

DEL C: --mandate lagt i report_forbidden. Den var ELDRE enn partisjonen og hadde
aldri faatt en rad, saa --report --ledger X --mandate Y droppet kommisjonen i
STILLHET - F4-klassen. Testarmen kjoerer mot en argv report-modus ellers ville
AKSEPTERT, med en kontroll som beviser rc 0 uten flagget.

Kontroll: 1290 passed / 5 skipped (fra 1275/5 - supersett, 0 fjernet).
Golden demo-transcript.stdout BYTE-UENDRET, shasum -a 1 (INNHOLD, ikke git-blob)
= ea8c534773acdbe41ae68f2c55724d69aaf8be4f. ruff + mypy rene.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Kjell Tore Guttormsen 2026-09-04 08:09:03 +02:00
commit b75387ca25
4 changed files with 593 additions and 14 deletions

View file

@ -1520,3 +1520,27 @@ def load_ir_projection(bundle_dir: str, name: str = _IR_PROJECTION) -> dict[str,
raise FileNotFoundError(f"IR projection not found in bundle: {name!r}") raise FileNotFoundError(f"IR projection not found in bundle: {name!r}")
data: dict[str, Any] = json.loads(resolved.read_text(encoding="utf-8")) data: dict[str, Any] = json.loads(resolved.read_text(encoding="utf-8"))
return data return data
def load_optional_ir_projection(
bundle_dir: str, name: str = _IR_PROJECTION
) -> dict[str, Any] | None:
"""``load_ir_projection`` where a MISSING file is legitimate: returns ``None`` instead of
raising (S7b søm 1). This is the run path's loader — an INGESTED corpus carries no hand-written
IR projection, and requiring one meant such a base could be navigated, catalogued and judged
through the deterministic mandate door and still not run the eight-step loop at all.
A PAIR beside the fail-fast loader rather than a ``required=`` flag on it, mirroring
``load_cost_baseline`` / ``load_optional_cost_baseline`` exactly. A flag would leave one branch
unexercised wherever callers took the default, which is the rot PM-tillegg 5 measured on
``evidence_for``; and it would falsify the several docstrings across the package that cite
``load_ir_projection`` as *the* fail-fast precedent.
The tolerance stops at absence: a projection that EXISTS but is malformed still raises, and a
name escaping the bundle still raises. Reading a corrupt projection as "no projection" would
hand back an unkeyed run under the appearance of a keyed one the same reasoning as
``load_optional_cost_baseline`` and ``budget.read_spend``."""
try:
return load_ir_projection(bundle_dir, name)
except FileNotFoundError:
return None

View file

@ -121,7 +121,7 @@ from portfolio_optimiser.verdicts import (
ProposalFeatures, ProposalFeatures,
Verdict, Verdict,
VerdictStore, VerdictStore,
bundle_candidate_features, optional_bundle_candidate_features,
capture_verdict, capture_verdict,
load_verdicts_from_dir, load_verdicts_from_dir,
similarity, similarity,
@ -183,6 +183,22 @@ class RunResult:
#: describes the gate that judged ONE candidate. EMPTY on the road path (no bundle is navigated) #: describes the gate that judged ONE candidate. EMPTY on the road path (no bundle is navigated)
#: and on any bundle that was read whole; it defaults for the same reason ``coverage`` does. #: and on any bundle that was read whole; it defaults for the same reason ``coverage`` does.
skipped_links: tuple[okf.SkippedLink, ...] = () skipped_links: tuple[okf.SkippedLink, ...] = ()
#: How many prior expert verdicts were in the store but could NOT be folded into the hypothesis
#: prompt, because the knowledge base declares no IR projection to key retrieval against (S7b
#: søm 1). 0 is the honest POSITIVE statement — every verdict the store held was keyable, which
#: covers both "the fold ran" and "there was nothing to fold" — so it defaults, exactly as
#: ``skipped_links``' empty tuple does.
#:
#: A COUNT, not a flag: "the fold did not happen" and "two judgements never reached the model"
#: are different operative facts, which is ``BudgetExceeded``'s kø-(y) rule one level down.
#:
#: Carried HERE and on neither of the other two carriers, and that placement is MEASURED.
#: ``ProvenanceStamp`` describes the gate that judged ONE candidate, whereas this is settled
#: once per run before any candidate exists (``skipped_links``' own reason). ``DryRunReport``
#: cannot carry it at all: the dry-run cut returns ABOVE the fold, so a field there could only
#: ever report zero — unlike ``cost_baseline_anchored`` and ``skipped_links``, both resolved
#: above that cut.
unkeyed_verdicts: int = 0
@property @property
def verdict_key(self) -> str: def verdict_key(self) -> str:
@ -523,11 +539,22 @@ def _project_from_bundle(
"""Derive a minimal ``Project`` from an OKF bundle (so a bundle the loop runs need NOT be a """Derive a minimal ``Project`` from an OKF bundle (so a bundle the loop runs need NOT be a
road reference-domain project). Only ``id`` + ``name`` reach the generation prompt road reference-domain project). Only ``id`` + ``name`` reach the generation prompt
(``generate._build_messages``), so ``cost_items`` is empty and ``verdict_input`` is left unset (``generate._build_messages``), so ``cost_items`` is empty and ``verdict_input`` is left unset
here (the Layer-2 decision flows via ``run_project``'s own ``verdict_input`` argument). Fail-fast: the bundle's IR here (the Layer-2 decision flows via ``run_project``'s own ``verdict_input`` argument).
``project_id`` must match the requested id. ``bundle`` reuses an already-navigated bundle to
avoid a second navigation.""" **The IR projection is OPTIONAL, and the tolerance stops at absence** (S7b søm 1). A base that
ir = okf.load_ir_projection(bundle_dir) HAS one must still agree with the requested id that divergence guard is the existing contract
if ir["project_id"] != project_id: multi-base dispatch rests on ("den eksisterende fail-fasten blir rutingsnøkkelen"), so loosening
absence must not loosen disagreement. A base WITHOUT one has nothing to check against, and an
ingested corpus is exactly that case.
**The name is unaffected, which is measured rather than assumed.** ``SavingsProposal`` has no
name field, so the projection has never been a name source: ``Project.name`` comes from the
``type: project`` concept's ``title``, with the requested id as the last resort — before and
after this change alike.
``bundle`` reuses an already-navigated bundle to avoid a second navigation."""
ir = okf.load_optional_ir_projection(bundle_dir)
if ir is not None and ir["project_id"] != project_id:
raise ValueError(f"bundle project_id {ir['project_id']!r} != requested {project_id!r}") raise ValueError(f"bundle project_id {ir['project_id']!r} != requested {project_id!r}")
nav = bundle if bundle is not None else okf.navigate_bundle(bundle_dir) nav = bundle if bundle is not None else okf.navigate_bundle(bundle_dir)
project_file = next((f for f in nav.files if f.type == "project"), None) project_file = next((f for f in nav.files if f.type == "project"), None)
@ -671,6 +698,32 @@ def collision_notice(collisions: tuple[VerdictCollision, ...]) -> str | None:
return "\n".join(lines) return "\n".join(lines)
def unkeyed_verdicts_notice(unkeyed: int) -> str | None:
"""Render the prior verdicts that could NOT reach the hypothesis prompt, or ``None`` when every
one of them could (S7b søm 1).
The measured silence this closes: making the IR projection optional lets an ingested corpus run
the whole loop, but it also removes the pre-hypothesis candidate the Step-1 ExpeL fold is keyed
on. Without a line here, a base holding a dozen prior expert judgements would run and simply not
use them indistinguishable, on stdout and in the artefact alike, from a base that had never
been judged at all. That is the same class of silence ``skipped_links_notice`` exists for.
ONE renderer, taking the already-resolved COUNT rather than a store or a bundle path: a renderer
that re-read either would be a second resolution of the run's own fold, free to disagree with it.
``None`` at zero omission, never an empty row (``mandate.announce``'s rule, the one
``cost_baseline_notice`` and ``skipped_links_notice`` both follow). Zero is the honest positive
statement, and a run that folded everything has nothing to report."""
if unkeyed <= 0:
return None
plural = "" if unkeyed == 1 else "s"
return (
f" Knowledge base: {unkeyed} prior expert verdict{plural} NOT folded into the hypothesis "
"prompt — the base declares no IR projection (validator-input.json), so there is no "
"candidate to key retrieval against"
)
def skipped_links_notice(skipped: tuple[okf.SkippedLink, ...]) -> str | None: def skipped_links_notice(skipped: tuple[okf.SkippedLink, ...]) -> str | None:
"""Render what the run could NOT read, or ``None`` when every cross-link was followed. """Render what the run could NOT read, or ``None`` when every cross-link was followed.
@ -945,12 +998,20 @@ async def run_project(
else None else None
) )
# S7b søm 1: the fold needs a pre-hypothesis CANDIDATE to rank prior verdicts against, and that
# candidate comes from the base's IR projection. A base without one (every ingested corpus) can
# now run — but it cannot key retrieval, and the verdicts it holds would otherwise be dropped in
# silence. Counted here and reported; the fold itself is unchanged when the key exists.
unkeyed_verdicts = 0
if bundle_dir is not None and store is not None and store.verdicts: if bundle_dir is not None and store is not None and store.verdicts:
expel_query = bundle_candidate_features(bundle_dir) expel_query = optional_bundle_candidate_features(bundle_dir)
fewshot = ExpeLContextProvider( if expel_query is None:
store, expel_query, k=top_k, retriever=ranker unkeyed_verdicts = len(store.verdicts)
).format_fewshot() else:
gen_context = f"{fewshot}\n\n{gen_context}" fewshot = ExpeLContextProvider(
store, expel_query, k=top_k, retriever=ranker
).format_fewshot()
gen_context = f"{fewshot}\n\n{gen_context}"
# 5. Structured candidate(s) -> blocking validation on the NUMBERS; token bound = the meter. # 5. Structured candidate(s) -> blocking validation on the NUMBERS; token bound = the meter.
# Without a mandate this is the single pre-Trekk-A call, unchanged. With one, every # Without a mandate this is the single pre-Trekk-A call, unchanged. With one, every
@ -1157,6 +1218,7 @@ async def run_project(
coverage=coverage, coverage=coverage,
refinements=tuple(refinements), refinements=tuple(refinements),
skipped_links=skipped_links, skipped_links=skipped_links,
unkeyed_verdicts=unkeyed_verdicts,
) )
@ -1735,7 +1797,16 @@ async def run_mandate_across_bundles(
# ONE reading of the base's own project id, used both to ADDRESS the run and to LABEL it. # ONE reading of the base's own project id, used both to ADDRESS the run and to LABEL it.
# A second lookup for the label would be the kø-(p) duplicate free to drift from the value # A second lookup for the label would be the kø-(p) duplicate free to drift from the value
# the run was actually dispatched with. # the run was actually dispatched with.
project_id = str(okf.load_ir_projection(bundle_dir)["project_id"]) #
# S7b søm 1: the hand-written projection FIRST, the base's DECLARED id as the fallback. The
# precedence is load-bearing in both directions. Declaration-first would re-address every
# existing base whose ``project_id`` differs from its ``bundle_id`` — the file is what those
# bases have always been routed by. File-only was the refusal this seam removes: an ingested
# corpus carries no projection, so it could not be routed at all. ``bundle_id`` is the
# identity every other door already resolves through (S7a-3), so the fallback introduces no
# third notion of what a base is called; ``by_id`` above is that same resolution, reused.
declared_ir = okf.load_optional_ir_projection(bundle_dir)
project_id = str(declared_ir["project_id"]) if declared_ir is not None else bundle_id
# D2: the id of the verdict THIS base minted, taken from ``run_project``'s existing # D2: the id of the verdict THIS base minted, taken from ``run_project``'s existing
# ``notify`` seam rather than off the returned ``RunResult``. ``notify`` fires inside the # ``notify`` seam rather than off the returned ``RunResult``. ``notify`` fires inside the
# capture block, so it is called exactly when a verdict exists (F2: never when nobody # capture block, so it is called exactly when a verdict exists (F2: never when nobody
@ -2219,6 +2290,12 @@ def main(argv: list[str] | None = None) -> int:
# when only one is given, so at most one situation reaches this list: both set. # when only one is given, so at most one situation reaches this list: both set.
"--decision": args.decision is not None, "--decision": args.decision is not None,
"--rationale": args.rationale is not None, "--rationale": args.rationale is not None,
# DEL C, the side-finding økt 82 measured and reported rather than fixed: --mandate is
# OLDER than this partition and never got a row, so ``--report --ledger X --mandate Y``
# dropped the commission in SILENCE — announced nothing, settled nothing, exit 0. The
# F4 class exactly: report mode returns above every dispatch, so an omission here is a
# silent drop and not a refusal.
"--mandate": args.mandate is not None,
} }
if any(report_forbidden.values()): if any(report_forbidden.values()):
print( print(
@ -3106,6 +3183,11 @@ def main(argv: list[str] | None = None) -> int:
id_notice = bundle_id_notice(result.provenance.bundle_id_source) id_notice = bundle_id_notice(result.provenance.bundle_id_source)
if id_notice is not None: if id_notice is not None:
print(id_notice) print(id_notice)
# Full run only, and structurally so: the fold happens BELOW the ``--live-dry-run`` cut, so a
# dry run has nothing to report here (contrast the three notices above, all resolved above it).
fold_notice = unkeyed_verdicts_notice(result.unkeyed_verdicts)
if fold_notice is not None:
print(fold_notice)
# The settlement against the commission (Trekk A4). Empty without a mandate, so an # The settlement against the commission (Trekk A4). Empty without a mandate, so an
# un-commissioned run prints exactly what it printed before. # un-commissioned run prints exactly what it printed before.
settlement = settle(result.coverage) settlement = settle(result.coverage)

View file

@ -503,10 +503,36 @@ def _features_from_ir(ir: dict[str, Any]) -> ProposalFeatures:
def bundle_candidate_features(bundle_dir: str) -> ProposalFeatures: def bundle_candidate_features(bundle_dir: str) -> ProposalFeatures:
"""The pre-hypothesis ExpeL query key: the candidate measure's structural features, read from """The pre-hypothesis ExpeL query key: the candidate measure's structural features, read from
the OKF bundle's IR projection. Available BEFORE any proposal is generated — which is what lets the OKF bundle's IR projection. Available BEFORE any proposal is generated — which is what lets
Step-1 retrieve prior verdicts and fold them into the hypothesis prompt (målbilde §2 step 1).""" Step-1 retrieve prior verdicts and fold them into the hypothesis prompt (målbilde §2 step 1).
Fail-fast. Use ``optional_bundle_candidate_features`` where a base without a projection is
legitimate and note that its two consumers answer that absence DIFFERENTLY, on purpose."""
return _features_from_ir(okf.load_ir_projection(bundle_dir)) return _features_from_ir(okf.load_ir_projection(bundle_dir))
def optional_bundle_candidate_features(bundle_dir: str) -> ProposalFeatures | None:
"""``bundle_candidate_features`` where a base with no IR projection is legitimate: ``None``
instead of raising (S7b søm 1). An ingested corpus has no hand-written projection, so it has no
pre-hypothesis candidate to key retrieval against a fact about the base, not an error.
Tolerance stops at absence, as everywhere: a projection that exists but is malformed still
raises (``okf.load_optional_ir_projection``'s rule)."""
ir = okf.load_optional_ir_projection(bundle_dir)
return None if ir is None else _features_from_ir(ir)
class VerdictKeyUnavailable(ValueError):
"""A ``type: verdict`` concept declares no structural key AND its bundle has no IR projection to
fall back on (S7b søm 1).
Fail-closed, and the ONE place in this seam where absence is not tolerable. The pre-S3.2
fallback keys such a verdict on the bundle's projection candidate; with no projection there is
no candidate, and minting a key anyway would attach the verdict to a candidate it is not about
precisely the defect S3.2 closes. Validation, never repair (``write_concept_file``'s rule): the
three fields are the author's to declare, and ``promote_verdict`` writes them, so a base grown
by the loop itself is never affected."""
# S3.2: a verdict file MAY carry its own structural key. All three fields or none — see # S3.2: a verdict file MAY carry its own structural key. All three fields or none — see
# ``_features_from_verdict_frontmatter``. # ``_features_from_verdict_frontmatter``.
_STRUCTURAL_FRONTMATTER_KEYS = ("affected_codes", "measure_type", "claimed_saving_nok") _STRUCTURAL_FRONTMATTER_KEYS = ("affected_codes", "measure_type", "claimed_saving_nok")
@ -618,7 +644,15 @@ def seed_store_from_bundle(bundle_dir: str) -> VerdictStore:
if features is None: if features is None:
# Read the IR projection lazily: a bundle whose verdicts all carry their own key does # Read the IR projection lazily: a bundle whose verdicts all carry their own key does
# not need one, and this keeps the fallback path's behaviour byte-identical. # not need one, and this keeps the fallback path's behaviour byte-identical.
fallback = fallback if fallback is not None else bundle_candidate_features(bundle_dir) fallback = (
fallback if fallback is not None else optional_bundle_candidate_features(bundle_dir)
)
if fallback is None:
raise VerdictKeyUnavailable(
f"{vf.name}: this verdict declares no structural key "
f"({', '.join(_STRUCTURAL_FRONTMATTER_KEYS)}) and the knowledge base carries no "
"IR projection to fall back on, so there is no candidate to key it to"
)
features = fallback features = fallback
verdicts.append( verdicts.append(
capture_verdict( capture_verdict(

View file

@ -0,0 +1,439 @@
"""The hand-written IR projection is OPTIONAL, and its absence is said rather than raised
(S7b søm 1, ordre ``20260903T204605Z-215167684-from-.claude`` DEL A).
**The seam this closes was measured before it was built** (``docs/2026-09-03-forslag-fra-mandat.md``
§ 1.1): ``okf.load_ir_projection`` is fail-fast required at THREE call sites, and an ingested tender
corpus carries no hand-written ``validator-input.json``, so such a base could be navigated
(``read_bundle``), catalogued (``list_bundles``) and judged through the deterministic mandate door
(``evaluate_mandate_candidates``, which never reads the file) and still could not run the
eight-step loop at all. The refusal fired before the first model call, which is the one mercy it
had.
**The pattern is the repo's own, not a new one.** ``load_cost_baseline`` /
``load_optional_cost_baseline`` already say exactly this: ABSENCE is tolerated and the caller takes
a stance; a file that EXISTS but is malformed, or that disagrees with the run, still refuses. The
S4.0 row states the rule in one line *"toleransen stopper ved fravær"* and reading a corrupt
projection as "no projection" would hand back an unkeyed run wearing a keyed one's clothes.
**A pair, never a ``required=`` flag.** PM-tillegg 5 measured what an unexercised parameter costs:
all eleven ``evidence_for`` call sites used the default, so the other branch rotted until it raised.
A pair also keeps true the five docstrings across ``hitl``/``ledger``/``verdicts``/``contracts`` that
cite ``load_ir_projection`` as *the* fail-fast precedent.
**The project NAME was a non-question, and that is measured rather than assumed.** The archived
self-order flagged the fallback as a real decision. Measured: ``SavingsProposal`` has five fields
(``project_id``, ``measure``, ``affected_items``, ``claimed_saving_nok``, ``assumptions``) and none
of them is a name the projection has never been a name source. ``Project.name`` comes from the
``type: project`` concept's ``title`` with the id as last resort, before and after, so there is no
new implicit ``project_id`` here; there is an old one, unchanged.
**Each of the three sites needed its OWN stance, which is why this is not one loader swap.**
* ``run._project_from_bundle`` absence skips a fail-fast that has nothing to check against;
a projection that is PRESENT and names another project still refuses. That divergence guard is
what multi-base dispatch rests on ("den eksisterende fail-fasten blir rutingsnøkkelen"), so
loosening absence must not loosen disagreement.
* ``run.run_mandate_across_bundles`` the routed run's ``project_id``. FILE FIRST, the base's
DECLARED ``bundle_id`` as the fallback (S7a-3). The precedence is load-bearing in the other
direction too: every existing base whose ``project_id`` differs from its ``bundle_id`` must keep
routing on the file.
* ``verdicts.bundle_candidate_features`` the ExpeL query key, and the ONE place where absence is
not merely tolerable. Two consumers, two different answers: the run's Step-1 fold SKIPS and SAYS
how many prior verdicts did not reach the prompt, while ``seed_store_from_bundle``'s S3.2 fallback
REFUSES BY NAME, because minting a key for a verdict that declares none is precisely the defect
S3.2 closes.
**The visibility carrier is ``RunResult`` alone, and that is a measurement.**
``cost_baseline_anchored`` and ``skipped_links`` are both resolved above the ``live_dry_run`` cut
(``run.py``: the cut returns before generation), so both can be carried on ``DryRunReport``. The
ExpeL fold happens BELOW that cut a dry run stops before it so a ``DryRunReport`` field could
only ever report zero. It is not on ``ProvenanceStamp`` either, for ``skipped_links``' own reason:
this is a RUN-level fact settled once, before any candidate exists, whereas the stamp describes the
gate that judged ONE candidate.
**Predictable vacuity, hunted for deliberately.** The shipped fixtures declare no ``bundle_id``
(S7a-3 measured zero ``^bundle_id`` matches under ``tests/``), so on them the declared id and the
mount basename COINCIDE and a routing arm cannot tell the two apart every routing arm here
therefore runs against a base crafted to declare an id differing from its directory name (M14's
lesson). And an arm that asserts only ``pytest.raises`` cannot discriminate when both
implementations raise (M5's lesson), so the refusal arms assert the NAMED class plus a
distinguishing token.
"""
from __future__ import annotations
import json
import shutil
from pathlib import Path
from typing import Any
import pytest
from portfolio_optimiser import okf, run, verdicts
from portfolio_optimiser.mandate import Approach, Mandate
from portfolio_optimiser.run import RunResult
from portfolio_optimiser.simulation import scripted_factory
from portfolio_optimiser.verdicts import ProposalFeatures, VerdictStore, capture_verdict
_FIXTURES = Path(__file__).parent / "fixtures"
#: MAJOR-4's priced fixture. It ships WITHOUT a ``validator-input.json`` — that absence is the whole
#: subject here, so it is used exactly as it sits on disk unless an arm says otherwise.
_PRICED = str(_FIXTURES / "k2-prisskjema-SYNTETISK")
_PROJECT = "K2"
#: A proposal the deterministic gate validates against the fixture's derived schedule: 21.1 is
#: 1250 x 850 = 1 062 500, and ``energy_efficiency`` caps the claim at 15 % of that (F8's registry).
_PROPOSAL_JSON = json.dumps(
{
"project_id": _PROJECT,
"measure": "energy_efficiency",
"affected_items": [{"code": "21.1", "quantity": 1250.0, "unit_cost": 850.0}],
"claimed_saving_nok": 100_000.0,
"assumptions": {"21.1": [800.0, 900.0]},
}
)
def _copy(tmp_path: Path, name: str, *, declared: str | None = None) -> Path:
"""A working copy of the priced fixture, optionally DECLARING a bundle id that differs from its
mount name the only shape under which a routing arm can distinguish declaration from mount."""
root = tmp_path / name
shutil.copytree(_PRICED, root)
if declared is not None:
for path in root.glob("*.md"):
text = path.read_text(encoding="utf-8")
path.write_text(
text.replace("---\ntype:", f"---\nbundle_id: {declared}\ntype:", 1), "utf-8"
)
return root
def _write_projection(root: Path, project_id: str) -> None:
(root / "validator-input.json").write_text(
json.dumps(
{
"project_id": project_id,
"measure": "energy_efficiency",
"affected_items": [{"code": "21.1", "quantity": 1250.0, "unit_cost": 850.0}],
"claimed_saving_nok": 100_000.0,
}
),
encoding="utf-8",
)
def _factory(sink: list[str] | None = None) -> Any:
return scripted_factory(
{"proposer": _PROPOSAL_JSON, "checker": "VERDICT: APPROVE"},
[] if sink is None else sink,
)
async def _run(bundle_dir: str, *, sink: list[str] | None = None, **kwargs: Any) -> Any:
return await run.run_project(
_PROJECT,
"local",
docs_dir=bundle_dir,
bundle_dir=bundle_dir,
derive_cost_baseline=True,
client_factory=_factory(sink),
max_rounds=2,
**kwargs,
)
# --------------------------------------------------------------------------------------------
# (a) the loader pair itself
# --------------------------------------------------------------------------------------------
def test_the_optional_loader_returns_none_for_a_base_that_has_no_projection() -> None:
assert okf.load_optional_ir_projection(_PRICED) is None
with pytest.raises(FileNotFoundError):
okf.load_ir_projection(_PRICED)
def test_the_optional_loader_returns_the_projection_when_the_base_has_one(tmp_path: Path) -> None:
root = _copy(tmp_path, "with-projection")
_write_projection(root, _PROJECT)
loaded = okf.load_optional_ir_projection(str(root))
assert loaded is not None
assert loaded["project_id"] == _PROJECT
def test_the_tolerance_stops_at_absence_a_malformed_projection_still_raises(
tmp_path: Path,
) -> None:
"""S4.0's rule, transplanted verbatim: reading a corrupt projection as "no projection" would
hand back an unkeyed run under the appearance of a keyed one."""
root = _copy(tmp_path, "malformed")
(root / "validator-input.json").write_text("{not json", encoding="utf-8")
with pytest.raises(json.JSONDecodeError):
okf.load_optional_ir_projection(str(root))
# --------------------------------------------------------------------------------------------
# (b) the run path: a base with no projection runs the WHOLE loop
# --------------------------------------------------------------------------------------------
async def test_a_base_without_a_projection_runs_the_whole_pipeline() -> None:
"""THE HEADLINE. Before this seam the same call raised ``FileNotFoundError`` from
``_project_from_bundle`` before the first model call, which is why an ingested corpus could be
navigated and never run."""
result = await _run(_PRICED)
assert isinstance(result, RunResult)
assert result.provenance.validator_decision in {"validated", "rejected"}
assert result.provenance.cost_baseline_anchored is True
async def test_a_projection_that_names_another_project_still_refuses(tmp_path: Path) -> None:
"""The divergence guard, which multi-base dispatch rests on. Absence became tolerable;
DISAGREEMENT did not, and a loader swap that loosened both would be green on every other arm
here."""
root = _copy(tmp_path, "wrong-project")
_write_projection(root, "SOMEONE-ELSE")
with pytest.raises(ValueError) as excinfo:
await _run(str(root))
message = str(excinfo.value)
assert "SOMEONE-ELSE" in message
assert _PROJECT in message
async def test_a_base_that_carries_the_projection_runs_exactly_as_before(tmp_path: Path) -> None:
"""The control the order asks for: with the file present nothing about the run changes."""
root = _copy(tmp_path, "runnable")
_write_projection(root, _PROJECT)
result = await _run(str(root))
assert isinstance(result, RunResult)
assert result.provenance.validator_decision in {"validated", "rejected"}
# --------------------------------------------------------------------------------------------
# (c) the dispatcher's routing key: file first, declared bundle id as the fallback
# --------------------------------------------------------------------------------------------
async def test_the_dispatcher_routes_a_projectionless_base_on_its_declared_id(
tmp_path: Path,
) -> None:
"""``run_mandate_across_bundles`` takes no ``project_id`` parameter BY DESIGN — it reads each
base's own. Without a projection the base's DECLARED id is what remains, and S7a-3 already made
that the identity every other door uses."""
declared = "k2-trinn1-20260903"
root = _copy(tmp_path, "mounted-elsewhere", declared=declared)
assert Path(root).name != declared # the two are genuinely distinct here
result = await run.run_mandate_across_bundles(
Mandate(
objective="Finn besparelser",
approaches=(Approach(id="a1", label="Tiltak", bundle_id=declared),),
allow_own_proposals=True,
),
(str(root),),
"local",
client_factory=_factory(),
max_rounds=2,
)
assert [r.bundle_id for r in result.runs] == [declared]
assert [r.project_id for r in result.runs] == [declared]
async def test_the_dispatcher_still_prefers_the_projection_when_the_base_has_one(
tmp_path: Path,
) -> None:
"""PRECEDENCE, and it is load-bearing in the direction the headline arm cannot see: every base
whose ``project_id`` differs from its ``bundle_id`` must keep routing on the FILE. The base here
declares both, and they disagree an implementation that read the declared id first would send
the run under the wrong project name while every other arm stayed green."""
declared = "k2-trinn1-20260903"
root = _copy(tmp_path, "both-known", declared=declared)
_write_projection(root, _PROJECT)
assert _PROJECT != declared
result = await run.run_mandate_across_bundles(
Mandate(
objective="Finn besparelser",
approaches=(Approach(id="a1", label="Tiltak", bundle_id=declared),),
allow_own_proposals=True,
),
(str(root),),
"local",
client_factory=_factory(),
max_rounds=2,
)
assert [r.project_id for r in result.runs] == [_PROJECT]
# --------------------------------------------------------------------------------------------
# (d) the ExpeL fold: skipped for want of a key, and SAID
# --------------------------------------------------------------------------------------------
#: A rationale that appears nowhere in the fixture, so its presence in a prompt can only come from
#: the ExpeL fold (``simulate_learning_loop``'s marker rule).
_MARKER = "TIDLIGERE-DOM-MARKOER"
def _store() -> VerdictStore:
return VerdictStore(
verdicts=[
capture_verdict(
ProposalFeatures(
affected_codes=frozenset({"21.1"}),
measure_type="energy_efficiency",
claimed_saving_nok=90_000.0,
),
"approved",
_MARKER,
),
capture_verdict(
ProposalFeatures(
affected_codes=frozenset({"36.1"}),
measure_type="ventilation",
claimed_saving_nok=50_000.0,
),
"rejected",
"tidligere dom nummer to",
),
]
)
async def test_prior_verdicts_that_could_not_be_keyed_are_counted_not_silently_dropped() -> None:
"""Without a projection there is no pre-hypothesis candidate to key the fold against, so the
fold cannot run. The run says how MANY prior verdicts therefore never reached the hypothesis
prompt the ``BudgetExceeded`` -(y) rule applied to a fold: "it did not happen" and "you lost
two judgements" are different operative facts."""
store = _store()
sink: list[str] = []
result = await _run(_PRICED, store=store, sink=sink)
assert isinstance(result, RunResult)
assert result.unkeyed_verdicts == 2
# THE DISCRIMINATOR: a counter can be wired to anything, so the arm reads the PROMPT the fold
# would have written into. The marker is a rationale that exists only in this store.
assert not any(_MARKER in prompt for prompt in sink)
async def test_a_run_whose_fold_was_keyed_counts_none(tmp_path: Path) -> None:
"""The control. A constant-count implementation and a constant-zero one are both green without
this pair and it is what makes the renderer's OMISSION itself gated."""
root = _copy(tmp_path, "keyed")
_write_projection(root, _PROJECT)
sink: list[str] = []
result = await _run(str(root), store=_store(), sink=sink)
assert isinstance(result, RunResult)
assert result.unkeyed_verdicts == 0
# And the fold genuinely RAN — without this the arm above would be green against an
# implementation that never folds at all and merely counts zero here.
assert any(_MARKER in prompt for prompt in sink)
def test_the_renderer_is_silent_when_every_prior_verdict_was_keyed() -> None:
"""Omission, never an empty row (``mandate.announce``'s rule, the one every ``*_notice``
renderer here follows)."""
assert run.unkeyed_verdicts_notice(0) is None
line = run.unkeyed_verdicts_notice(2)
assert line is not None
assert "2" in line
def test_the_cli_prints_the_notice_for_a_projectionless_base_with_prior_verdicts(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
"""The renderer reaching stdout is its own seam (the ``cost_baseline_notice`` precedent: a
renderer nothing calls is a line nobody sees). Driven through ``main()`` with the verdict inbox,
which is the only route by which the CLI can hand ``run_project`` a non-empty store."""
inbox = tmp_path / "inbox"
inbox.mkdir()
verdicts.write_verdict(str(inbox), _store().verdicts[0])
monkeypatch.setattr(run, "_default_factory", lambda _profile: _factory())
rc = run.main(
[
_PROJECT,
"--docs-dir",
_PRICED,
"--bundle-dir",
_PRICED,
"--derive-cost-baseline",
"--verdict-dir",
str(inbox),
]
)
assert rc == 0
assert "1 prior expert verdict" in capsys.readouterr().out
# --------------------------------------------------------------------------------------------
# (e) the seed fallback: REFUSED by name, never keyed on a guess
# --------------------------------------------------------------------------------------------
def _verdict_concept(root: Path, name: str, *, structural: bool) -> None:
key = (
'affected_codes: "21.1"\nmeasure_type: "energy_efficiency"\nclaimed_saving_nok: "90000"\n'
if structural
else ""
)
(root / name).write_text(
f'---\ntype: verdict\ntitle: "Dom"\n{key}decision: approved\n---\n\nEn tidligere dom.\n',
encoding="utf-8",
)
index = root / "index.md"
index.write_text(
index.read_text(encoding="utf-8") + f"\nOgsaa [dom]({name}).\n", encoding="utf-8"
)
def test_a_verdict_without_its_own_key_in_a_projectionless_base_is_refused_by_name(
tmp_path: Path,
) -> None:
"""S3.2's rule at the one point where absence is NOT tolerable. The pre-S3.2 fallback keys such
a verdict on the bundle's projection candidate; with no projection there is no candidate, and
minting one would attach the verdict to a candidate it is not about the exact defect S3.2
closes. Validation, never repair (``write_concept_file``'s rule)."""
root = _copy(tmp_path, "unkeyable")
_verdict_concept(root, "dom.md", structural=False)
with pytest.raises(verdicts.VerdictKeyUnavailable) as excinfo:
verdicts.seed_store_from_bundle(str(root))
message = str(excinfo.value)
assert "dom.md" in message
assert "affected_codes" in message
def test_a_verdict_that_declares_its_own_key_seeds_without_any_projection(tmp_path: Path) -> None:
"""The control, and the reason the refusal above is NARROW: S3.2's fields are what
``promote_verdict`` writes, so a base grown by the loop itself seeds unchanged."""
root = _copy(tmp_path, "self-keyed")
_verdict_concept(root, "dom.md", structural=True)
store = verdicts.seed_store_from_bundle(str(root))
assert len(store.verdicts) == 1
assert store.verdicts[0].proposal_features.affected_codes == frozenset({"21.1"})
# --------------------------------------------------------------------------------------------
# (f) DEL C — the report-mode partition (the gap økt 82 measured and reported)
# --------------------------------------------------------------------------------------------
def test_report_mode_refuses_a_mandate_instead_of_dropping_it(
tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
"""``--mandate`` was the one flag older than ``report_forbidden`` that never got a row, so
``--report --ledger X --mandate Y`` announced nothing and settled nothing a SILENT DROP, the
F4 class. The argv is one report mode would otherwise ACCEPT (a valid ``--ledger``, and the
control proves rc 0 without the flag), so rc 1 is the mutant's opposite outcome rather than the
same refusal arriving by another route."""
ledger = tmp_path / "ledger.json"
ledger.write_text(json.dumps([]), encoding="utf-8")
mandate = tmp_path / "mandate.json"
mandate.write_text(
Mandate(objective="x", approaches=(Approach(id="a1", label="t"),)).model_dump_json(),
encoding="utf-8",
)
assert run.main(["--report", "--ledger", str(ledger)]) == 0
capsys.readouterr()
rc = run.main(["--report", "--ledger", str(ledger), "--mandate", str(mandate)])
assert rc == 1
assert "mode-exclusive" in capsys.readouterr().err