open/portfolio-optimiser
1
0
Fork 0
Code Issues Releases 4 Activity

fix(v1-gate): row 2 binds a round to the run's own artefact family, and says out loud what it still cannot prove

Browse source 
The 18.09 re-measurement took row 2 to 3 of 3 GREEN on a tree this product had never
run in: four handwritten outcome.json, four handwritten <run_id>-coverage.json in an
outbox the forger named in those same files, and os.utime for the ordering. "Round 0
must be a named real run" was implemented as "a file with that name exists" — which
touch satisfies. The attack is committed as a red test in b769537.

Three bindings, chosen because each removes one of the forger's three moves:

1. The outbox is DERIVED, never declared. It is <rounds-dir>/<n>/outbox/, and an
   outcome.json that names one is refused by name. A path a round file chooses is a
   path it can point at a directory the same hand just wrote.

2. "The run exists" now means the run's own artefacts agree WITH EACH OTHER on content
   the gate recomputes (verify_run). Every evaluated approach has the proposal/outcome
   pair write_outbox actually persists; outcome_type IS the coverage status, reason IS
   its detail, a validated row's figure IS the proposal's own claimed_saving_nok, the
   provenance stamp follows the same branch, and verdict_id is RE-MINTED here from the
   proposal's own IR with the product's one minting rule (A5) rather than read. A
   not_evaluated approach wrote neither file, and an artefact naming an approach the
   coverage omits belongs to another run. verdicts._features_from_ir is made public for
   this: a second private copy in the gate could drift from the rule the run stamped
   with, which would turn the binding into a coincidence.

3. mtime decides nothing. The run's time is the round's declared ran_at (ISO-8601 with
   zone, required). An mtime is not evidence — it is a filesystem attribute one utime
   call sets.

What this does NOT do is prove a run happened. Nothing in a directory can: the outbox
writers are byte-deterministic and carry no clock by contract. So row 2 states its own
limit on every run instead of leaving GREEN to imply it (Row.attests / RUN_ATTESTATION):
that a run was actually made, and when, is the operator's to confirm. The cost of a
forgery moves from touch to reproducing the product's own artefact set, minting rule
included.

Measured, in a scratch copy, never in the work tree — four new mutants in this class,
each one line, each felled by the whole suite:
  M7  if "outbox" in data:          -> if False:   1 failed (declares its own outbox)
  M8  if verdict_id != minted:      -> if False:   1 failed (key is not the IR's)
  M9  if strays:                    -> if False:   1 failed (artefact of another run)
  M10 ran_at                        -> coverage mtime  18 failed (incl. the utime test)
Control, same scratch copy, unmutated: 1993 passed, 10 skipped, 5 xfailed.
Work tree, re-run after git add: uv run pytest -q -> 1998 passed, 5 skipped, 5 xfailed.
Gate: uv run python -m portfolio_optimiser.evals.v1_gate -> exit 1, row 2 RED (0 of 3).

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Kjell Tore Guttormsen 2026-09-18 15:32:14 +02:00
commit a362504108
Signed by: ktg
SSH key fingerprint: SHA256:JakMjO6FTBBzN0Bhfj9saOoEjaFxlSdYuZQQpM/lF9Q
5 changed files with 389 additions and 42 deletions
Download patch file Download diff file Expand all files Collapse all files

217
src/portfolio_optimiser/evals/v1_gate.py
View file

@ -27,7 +27,7 @@ import tempfile
import xml.etree.ElementTree as ET
from collections.abc import Callable, Iterable, Mapping, Sequence
from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
from datetime import datetime
import difflib
import re
from pathlib import Path
@ -46,6 +46,17 @@ _AI_LINE_MIN = 30
_NOK_NOISE = 0.01
#: Printed on every run: the one thing rows 1-2 cannot prove.
ATTESTATION = "rad 1–2 beviser ikke at en fagperson skrev feedbacken; det bekrefter operatøren"
#: Row 2: the run's own outbox, DERIVED from the round directory and never declared inside it.
#: A path a round file names is a path a round file can point anywhere, including at a directory
#: the same hand just wrote (measured 18.09).
RUN_OUTBOX = "outbox"
#: Printed on row 2 every run: what an artefact family cannot show, however consistent it is.
RUN_ATTESTATION = (
"at en kjøring FAKTISK ble gjort, og NÅR, står i ingen artefakt — utboksfilene er "
"byte-deterministiske og bærer med vilje ingen klokke. Rekkefølgen leses av ran_at i "
"rundefila, som er oppgitt; at kjøring n ble gjort etter tilbakemeldingen, bekrefter "
"operatøren"
)
GREEN = "GRØNN"
RED = "RØD"
@ -58,8 +69,10 @@ Rundekatalogen (--rounds-dir) har fast form. Runde n = tilbakemelding på rappor
<rounds-dir>/0/report.md rapporten fra grunnkjøringen (runde 0)
<rounds-dir>/0/outcome.json grunnkjøringen som runde 1 måles mot
<rounds-dir>/0/outbox/ grunnkjøringens EGEN utboks, kopiert hit urørt
<rounds-dir>/<n>/feedback.json fagpersonens tilbakemelding på rapport n-1 (n = 1, 2, 3)
<rounds-dir>/<n>/outcome.json kjøring n, gjort ETTER den tilbakemeldingen
<rounds-dir>/<n>/outbox/ kjøring n sin EGEN utboks, kopiert hit urørt
<rounds-dir>/<n>/report.md rapporten bygget fra kjøring n
<rounds-dir>/3/report.kept.md runde 3-rapporten slik fagpersonen BEHOLDT den
@ -68,8 +81,8 @@ feedback.json:
"report_unchanged": true (valgfri, kun runde 3: kvitterer for en urørt rapport),
"items": [{"id": "<unik id>", "type": <1-8>, "text": "<tilbakemeldingen>"}]}
outcome.json (hver rad sjekkes mot kjøringens egen <outbox>/<run_id>-coverage.json):
{"run_id": "<kjøringen>", "outbox": "<utboksen, relativ til denne fila eller absolutt>",
outcome.json (hver rad sjekkes mot kjøringens egen <n>/outbox/<run_id>-coverage.json):
{"run_id": "<kjøringen>", "ran_at": "<ISO-8601 med tidssone: da kjøringen ble gjort>",
"approaches": [{"id": "<tilnærming>", "validated": true|false,
"stage": "<avvisningsstadium som validator.rejection_stage gir, tom når validert>",
"validated_nok": <tall eller null>,
@ -81,6 +94,13 @@ tilnærmings-id-er, (b) hvilke som er validert, (c) avvisningsstadium, (d) valid
under 1 % er støy) — OG minst én endret rad bærer en feedback-id gitt i DENNE runden, gitt mellom
de to kjøringene. Hver runde må ha minst ett nytt punkt og egne id-er. Tekst tatt fra et
AI-forfattet dokument (docs/ekspert-svar.md) teller aldri. Rad 1-2 beviser FORM, ikke forfatterskap.
Utboksen OPPGIS IKKE i outcome.json — den er <n>/outbox/, og en fil som oppgir den avvises. En
kjøring må stå inne for seg selv: hver EVALUERT tilnærming har <run_id>-<id>-proposal.json og
<run_id>-<id>-outcome.json i utboksen, som stemmer med coverage-raden på type, grunn og beløp, og
hvis verdict_id gaten selv minter PÅ NYTT fra forslagets egen IR (verdicts.verdict_key). En
not_evaluated-rad har ingen av delene, og en artefakt for en tilnærming coverage ikke nevner hører
til en annen kjøring. Dette beviser ikke at kjøringen skjedde — se attesteringen for rad 2.
Rad 4 teller innholdslinjer (ikke blanke, skillelinjer eller tabellrammer) som står uendret og i
samme rekkefølge; fagpersonens tillegg vises som eget tall. --rounds-dir inne i repoet må være
gitignored.
@ -98,6 +118,9 @@ class Row:
failing: bool = True
exceptions: tuple[str, ...] = ()
diagnostics: tuple[str, ...] = ()
#: What this row CANNOT prove, in the operator's words. Printed under the attestation on every
#: run — a row states its own limit, rather than leaving the reader to infer it from GREEN.
attests: tuple[str, ...] = ()
def line(self) -> str:
k = "–" if self.k is None else str(self.k)
@ -299,10 +322,136 @@ def _stage_of(status: str, detail: str) -> str:
return rejection_stage(detail)
def read_outcome(path: Path) -> tuple[Outcome | None, str]:
"""A round's outcome file, VERIFIED against the run it names: ``outbox`` must hold
``<run_id>-coverage.json``, and every row's (a)-(d) must equal that run's own coverage. A
handwritten outcome with no run behind it is refused; the run's time is the coverage file's."""
def _read_json(path: Path) -> tuple[Any, str]:
"""One artefact's payload, or ``(None, why)`` — one reader for the run's own files, so an
unreadable one is refused BY NAME instead of by exception type at four call sites."""
try:
return json.loads(path.read_text(encoding="utf-8")), ""
except FileNotFoundError:
return None, f"{path.name} mangler"
except (ValueError, OSError) as exc:
return None, f"{path.name} uleselig ({exc!r})"
def _approach_of(name: str, run_id: str) -> str | None:
"""The approach id a per-approach artefact of ``run_id`` is about, or ``None`` when the file
is not one (``<run_id>-coverage.json`` and the run's other artefacts are not)."""
for suffix in ("-proposal.json", "-outcome.json"):
if name.startswith(f"{run_id}-") and name.endswith(suffix):
return name[len(run_id) + 1 : -len(suffix)]
return None
def _minted_key(ir: Mapping[str, Any]) -> str | None:
"""The verdict key ``verdicts.verdict_key`` mints for THIS proposal IR, or ``None`` when the
IR cannot be keyed at all.
Imported here rather than restated: one minting rule (A5), and a private second copy in the
gate could drift from the one the run actually stamped with — which would turn this check
from a binding into a coincidence. Imported INSIDE the function because ``verdicts`` pulls in
``agent_framework``, and every other row of this gate is stdlib plus the MAF-free validator."""
from portfolio_optimiser.verdicts import features_from_ir, verdict_key
try:
return verdict_key(features_from_ir(dict(ir)))
except (KeyError, TypeError, ValueError):
return None
def verify_run(outbox: Path, run_id: str, coverage: Sequence[Mapping[str, Any]]) -> str:
"""Whether the run's OWN artefacts stand up to EACH OTHER on content this function recomputes
— ``""`` when they do, else the first thing that does not hold.
``<run_id>-coverage.json`` on its own says nothing: one hand writes it, and a second hand can
write it identically. Measured 18.09 on ``9825b26``: row 2 read 3 of 3 GREEN from four
handwritten outcome files and four handwritten coverage files, because "round 0 must be a
named real run" was implemented as "a file with that name exists" — which ``touch`` satisfies.
A run that EVALUATED an approach also wrote ``<run_id>-<id>-proposal.json`` and
``<run_id>-<id>-outcome.json`` (``outbox.write_outbox``, A5), and those artefacts are not free
of the coverage row or of each other: the outcome's ``outcome_type`` IS the row's status, its
``reason`` IS the row's ``detail``, a validated row's figure IS the proposal's own
``claimed_saving_nok``, the stamp's ``validator_decision`` follows the same branch
(``run_project``: validated/unsupported -> ``"validated"``), and ``verdict_id`` is
``verdicts.verdict_key`` of the proposal's own IR — which this gate RE-MINTS rather than
reads. An approach the run never reached (``not_evaluated``) wrote NEITHER file, and an
artefact naming an approach the coverage does not list belongs to some other run: the family
has to be complete in both directions.
This does not prove a run happened. Nothing in a directory can — the outbox writers are
byte-deterministic and carry no clock by contract — and ``RUN_ATTESTATION`` says so on the row
on every run. What it does is move the cost of a forgery from ``touch`` to reproducing the
product's own artefact set, minting rule included."""
listed: set[str] = set()
for row in coverage:
try:
aid, status = str(row["id"]), str(row["status"])
except (KeyError, TypeError) as exc:
return f"en coverage-rad mangler id/status ({exc!r})"
listed.add(aid)
stem = f"{run_id}-{aid}"
proposal_path, outcome_path = (
outbox / f"{stem}-proposal.json",
outbox / f"{stem}-outcome.json",
)
if status == "not_evaluated":
wrote = [p.name for p in (proposal_path, outcome_path) if p.is_file()]
if wrote:
return f"{aid} ble ikke evaluert, men kjøringen skrev {', '.join(wrote)}"
continue
proposal, why = _read_json(proposal_path)
if proposal is None:
return f"{aid}: {why}"
result, why = _read_json(outcome_path)
if result is None:
return f"{aid}: {why}"
try:
ir = dict(proposal["proposal"])
decision = str(proposal["provenance"]["validator_decision"])
keyed = {(str(a["run_id"]), str(a["approach_id"])) for a in (proposal, result)}
outcome_type, verdict_id = str(result["outcome_type"]), str(result["verdict_id"])
except (KeyError, TypeError) as exc:
return f"{aid}: artefaktene mangler et felt ({exc!r})"
if keyed != {(run_id, aid)}:
return f"{aid}: artefaktene er merket {sorted(keyed)}, ikke ({run_id!r}, {aid!r})"
if outcome_type != status:
return f"{aid}: utfallet er {outcome_type!r}, coverage sier {status!r}"
if status == "validated":
claimed, reported = ir.get("claimed_saving_nok"), row.get("saving_nok")
if claimed is None or reported is None or float(claimed) != float(reported):
return f"{aid}: coverage sier {reported} NOK, forslaget selv sier {claimed}"
elif str(result.get("reason", "")) != str(row.get("detail", "")):
return f"{aid}: utfallets grunn er ikke coverage-radens detail"
expected = "rejected" if status == "rejected" else "validated"
if decision != expected:
return f"{aid}: stempelet sier validator_decision={decision!r} for en {status!r} rad"
minted = _minted_key(ir)
if minted is None:
return f"{aid}: forslagets egen IR kan ikke nøkles ({stem}-proposal.json)"
if verdict_id != minted:
return f"{aid}: verdict_id er ikke nøkkelen forslagets egen IR minter"
strays = sorted(
p.name
for p in outbox.glob(f"{run_id}-*.json")
if (found := _approach_of(p.name, run_id)) is not None and found not in listed
)
if strays:
return f"utboksen har artefakter kjøringens egen coverage ikke nevner: {', '.join(strays)}"
return ""
def read_outcome(round_dir: Path) -> tuple[Outcome | None, str]:
"""A round's outcome file, VERIFIED against the run it names — and the run is the one in the
round's OWN ``outbox/``, never a path the file points at.
Three things must hold, and the third is what ``verify_run`` exists for: the coverage file
must be stamped with the run the round names, every row's (a)-(d) must equal that coverage,
and the run's artefacts must stand up to each other. The run's TIME is the round's declared
``ran_at`` and no longer the coverage file's mtime — an mtime is not evidence, it is a
filesystem attribute one ``os.utime`` call sets, and reading it as proof made row 2 green on a
tree nothing had ever run in (18.09). What ``ran_at`` is instead is stated, not implied:
``RUN_ATTESTATION``."""
path = round_dir / "outcome.json"
try:
data = json.loads(path.read_text(encoding="utf-8"))
rows = {str(a["id"]): a for a in data["approaches"]}
@ -310,36 +459,48 @@ def read_outcome(path: Path) -> tuple[Outcome | None, str]:
str(r["id"]): set(map(str, r.get("feedback_ids", ()))) for r in data.get("removed", ())
}
run_id = str(data["run_id"]).strip()
outbox = Path(str(data["outbox"])).expanduser()
ran_at = _parse_time(data["ran_at"])
except FileNotFoundError:
return None, f"{path} mangler"
except (ValueError, KeyError, TypeError) as exc:
return None, f"{path} uleselig ({exc!r})"
if not run_id:
return None, f"{path}: run_id er tom"
if not outbox.is_absolute():
outbox = path.parent / outbox
if "outbox" in data:
return None, (
f"{path}: outbox kan ikke oppgis — kjøringen leses fra "
f"{round_dir.name}/{RUN_OUTBOX}/, ikke fra en sti denne fila velger"
)
if ran_at is None:
return None, f"{path}: ran_at er ikke et ISO-tidsstempel med tidssone"
outbox = round_dir / RUN_OUTBOX
coverage_path = outbox / f"{run_id}-coverage.json"
if not coverage_path.is_file():
return None, f"{path}: kjøringen {run_id!r} finnes ikke ({coverage_path} mangler)"
payload, why = _read_json(coverage_path)
if payload is None:
return None, f"{path}: kjøringen {run_id!r} finnes ikke ({RUN_OUTBOX}/{why})"
try:
coverage = json.loads(coverage_path.read_text(encoding="utf-8"))["rows"]
except (ValueError, KeyError, TypeError) as exc:
coverage = list(payload["rows"])
stamped = str(payload["run_id"])
truth = {
str(r["id"]): (
r["status"] == "validated",
_stage_of(str(r["status"]), str(r.get("detail", ""))),
r.get("saving_nok") if r["status"] == "validated" else None,
)
for r in coverage
}
except (KeyError, TypeError) as exc:
return None, f"{coverage_path} uleselig ({exc!r})"
if stamped != run_id:
return None, f"{path}: coverage-fila er kjøring {stamped!r}, ikke {run_id!r}"
if not coverage:
return None, f"{path}: kjøringen {run_id!r} evaluerte ingen tilnærming"
truth = {
str(r["id"]): (
r["status"] == "validated",
_stage_of(str(r["status"]), str(r.get("detail", ""))),
r.get("saving_nok") if r["status"] == "validated" else None,
)
for r in coverage
}
claimed = {aid: _row_key(row) for aid, row in rows.items()}
if claimed != truth:
return None, f"{path}: (a)-(d) stemmer ikke med kjøringens egen coverage ({run_id})"
ran_at = datetime.fromtimestamp(coverage_path.stat().st_mtime, tz=timezone.utc)
unheld = verify_run(outbox, run_id, coverage)
if unheld:
return None, f"{path}: kjøringen {run_id!r} står ikke inne for seg selv — {unheld}"
return Outcome(rows, removed, run_id, ran_at), ""
@ -389,7 +550,7 @@ def score_changes(rounds_dir: Path, required: int, ai: tuple[str, str] | None) -
exceptions: list[str] = []
k = 0
base_path = rounds_dir / "0" / "outcome.json"
base_outcome, base_why = read_outcome(base_path)
base_outcome, base_why = read_outcome(rounds_dir / "0")
base = f"runde 0 = {base_path}"
base += f" (kjøring {base_outcome.run_id})" if base_outcome else f" — {base_why}"
feedback_by_round = read_rounds(rounds_dir, required, ai) if rounds_dir.is_dir() else {}
@ -398,8 +559,8 @@ def score_changes(rounds_dir: Path, required: int, ai: tuple[str, str] | None) -
if feedback is None:
exceptions.append(why)
continue
prev, why_prev = read_outcome(rounds_dir / str(n - 1) / "outcome.json")
cur, why_cur = read_outcome(rounds_dir / str(n) / "outcome.json")
prev, why_prev = read_outcome(rounds_dir / str(n - 1))
cur, why_cur = read_outcome(rounds_dir / str(n))
if prev is None or cur is None:
exceptions.append(f"runde {n}: {why_prev or why_cur}")
continue
@ -422,6 +583,7 @@ def score_changes(rounds_dir: Path, required: int, ai: tuple[str, str] | None) -
status,
base,
exceptions=tuple(exceptions),
attests=(RUN_ATTESTATION,),
)
@ -958,6 +1120,9 @@ def render(rows: Sequence[Row]) -> str:
out += [r.line() for r in rows]
out.append("")
out.append(f"Attestering: {ATTESTATION}.")
for r in rows:
for a in r.attests:
out.append(f" [{r.title.split()[0]}] {a}.")
out.append("")
out.append("Unntak fra 100 %:")
for r in rows:

16
src/portfolio_optimiser/verdicts.py
View file

@ -489,9 +489,15 @@ def seed_store() -> VerdictStore:
# --- OKF-bundle seeding (Fase 2a): turn a project's bundle into the ExpeL substrate ---
def _features_from_ir(ir: dict[str, Any]) -> ProposalFeatures:
"""Map a bundle's IR projection (``validator-input.json``) to the structural features the
store ranks on: the affected cost-code set, the measure string, and the claimed magnitude."""
def features_from_ir(ir: dict[str, Any]) -> ProposalFeatures:
"""Map a proposal IR dict (a bundle's ``validator-input.json`` projection, or an outbox
artefact's own ``proposal`` payload) to the structural features the store ranks on: the
affected cost-code set, the measure string, and the claimed magnitude.
PUBLIC for the same reason ``verdict_key`` is (A5, one minting rule): the v1 gate re-mints an
artefact's verdict id from the artefact's OWN IR to check that a run's outbox agrees with
itself, and a second private copy of this mapping in the gate would be a rule that can drift
from the one the run actually used — the defect, not the convenience."""
return ProposalFeatures(
affected_codes=frozenset(item["code"] for item in ir["affected_items"]),
measure_type=ir["measure"],
@ -507,7 +513,7 @@ def bundle_candidate_features(bundle_dir: str) -> ProposalFeatures:
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:
@ -518,7 +524,7 @@ def optional_bundle_candidate_features(bundle_dir: str) -> ProposalFeatures | No
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)
return None if ir is None else features_from_ir(ir)
class VerdictKeyUnavailable(ValueError):

2
tests/test_run_cli.py
View file

@ -462,7 +462,7 @@ def _tied_pair_verdicts():
affected_codes=query.affected_codes | {extra_code},
measure_type=query.measure_type,
# ``description == measure`` is what both live minting paths emit
# (``run._features_of`` / ``verdicts._features_from_ir``).
# (``run._features_of`` / ``verdicts.features_from_ir``).
claimed_saving_nok=query.claimed_saving_nok,
description=query.measure_type,
),

4
tests/test_semretrieval_loadbearing.py
View file

@ -250,7 +250,7 @@ def test_semretrieval_imports_no_network_modules() -> None:
#
# Every feature set carries ``description == measure_type``, which is exactly what both live
# minting paths emit (``run._features_of`` sets ``description=proposal.measure``;
# ``verdicts._features_from_ir`` sets ``description=ir["measure"]``).
# ``verdicts.features_from_ir`` sets ``description=ir["measure"]``).
_MEASURE = "asfalt"
_QUERY_CODES = frozenset({"05.1", "05.2"})
@ -373,7 +373,7 @@ def test_minted_and_authored_verdicts_embed_identically_on_a_structural_tie() ->
affected_codes=_CORRECT_CODES,
measure_type=_MEASURE,
claimed_saving_nok=200_000.0,
description=_MEASURE, # the shape run._features_of / _features_from_ir emit
description=_MEASURE, # the shape run._features_of / features_from_ir emit
)
minted = capture_verdict(features, "approved", "framework-captured ruling")

192
tests/test_v1_gate.py
View file

@ -15,6 +15,7 @@ from __future__ import annotations
import ast
import json
import os
import shutil
import subprocess
import sys
from pathlib import Path
@ -24,7 +25,9 @@ import pytest
from portfolio_optimiser import frozen_bundles
from portfolio_optimiser.evals import v1_gate as gate
from portfolio_optimiser.ir import AffectedItem, SavingsProposal
from portfolio_optimiser.validator import UNSUPPORTED_REASON
from portfolio_optimiser.verdicts import features_from_ir, verdict_key
_REPO = Path(__file__).resolve().parents[1]
_CONFIG = gate.load_config()
@ -98,6 +101,54 @@ def _coverage_row(row: dict[str, Any]) -> dict[str, Any]:
return {"id": row["id"], "status": status, "detail": _DETAIL[row["stage"]], "saving_nok": None}
def _ir(aid: str, nok: float | None) -> dict[str, Any]:
"""The proposal IR a run would have written for this approach — built through the real model
so the fixture cannot drift from the shape ``outbox.write_outbox`` actually persists, which is
the shape the gate re-mints the verdict key from."""
saving = 1.0 if nok is None else float(nok)
return SavingsProposal(
project_id="p",
measure=f"tiltak {aid}",
affected_items=[AffectedItem(code=f"K-{aid}", quantity=1.0, unit_cost=saving)],
claimed_saving_nok=saving,
).model_dump()
def _run_family(outbox: Path, run_id: str, coverage: list[dict[str, Any]]) -> None:
"""One proposal/outcome pair per EVALUATED approach, agreeing with its coverage row — the
artefact family ``gate.verify_run`` re-checks. ``verdict_id`` is minted with the product's own
rule (A5), because that is the field the gate mints again rather than reads."""
for cov in coverage:
if cov["status"] == "not_evaluated":
continue
aid = cov["id"]
ir = _ir(aid, cov.get("saving_nok"))
stem = f"{run_id}-{aid}"
_write(
outbox / f"{stem}-proposal.json",
{
"run_id": run_id,
"approach_id": aid,
"proposal": ir,
"provenance": {
"validator_decision": (
"rejected" if cov["status"] == "rejected" else "validated"
)
},
},
)
_write(
outbox / f"{stem}-outcome.json",
{
"run_id": run_id,
"approach_id": aid,
"outcome_type": cov["status"],
"reason": cov.get("detail", ""),
"verdict_id": verdict_key(features_from_ir(ir)),
},
)
def _outcome(
round_dir: Path,
rows: list[dict[str, Any]],
@ -106,19 +157,24 @@ def _outcome(
at: int | None = None,
coverage: list[dict[str, Any]] | None = None,
) -> None:
"""The round's outcome file AND the run it names: an outbox holding the run's own coverage,
stamped at a fixed time (the run's time is the coverage file's)."""
"""The round's outcome file AND the run it names: the run's OWN outbox inside the round,
holding its coverage and one artefact pair per evaluated approach. The run's time is DECLARED
(``ran_at``) — an mtime is not evidence, and the gate no longer reads one."""
run_id = f"r{round_dir.name}"
outbox = round_dir.parent / "runs" / run_id
outbox = round_dir / "outbox"
if outbox.is_dir():
shutil.rmtree(outbox) # a re-written round brings its own run, not the last one's leftovers
rendered = [_coverage_row(r) for r in rows] if coverage is None else coverage
_write(outbox / f"{run_id}-coverage.json", {"rows": rendered, "stop_reason": ""})
stamp = _T0 + (int(round_dir.name) * 20 if at is None else at)
os.utime(outbox / f"{run_id}-coverage.json", (stamp, stamp))
_write(
outbox / f"{run_id}-coverage.json",
{"run_id": run_id, "rows": rendered, "stop_reason": ""},
)
_run_family(outbox, run_id, rendered)
_write(
round_dir / "outcome.json",
{
"run_id": run_id,
"outbox": f"../runs/{run_id}",
"ran_at": _iso(int(round_dir.name) * 20 if at is None else at),
"approaches": rows,
"removed": list(removed),
},
@ -356,7 +412,7 @@ def test_m1_a_change_below_one_percent_is_noise(tmp_path: Path) -> None:
def test_m1_an_outcome_must_name_a_run_that_exists_and_agree_with_it(tmp_path: Path) -> None:
root = _green_rounds(tmp_path)
run = root / "runs" / "r1" / "r1-coverage.json"
run = root / "1" / "outbox" / "r1-coverage.json"
run.unlink()
assert "finnes ikke" in " ".join(gate.score_changes(root, 3, _AI).exceptions)
_outcome(
@ -451,6 +507,126 @@ def test_m6_a_handwritten_outbox_is_not_a_run(tmp_path: Path) -> None:
assert (row.k, row.status) == (0, gate.RED), row.exceptions
def _edit(path: Path, **changes: Any) -> None:
data = json.loads(path.read_text(encoding="utf-8"))
data.update(changes)
_write(path, data)
def _mut_declares_its_own_outbox(root: Path) -> None:
_edit(root / "1" / "outcome.json", outbox=str(root / "1" / "outbox"))
def _mut_coverage_is_another_run(root: Path) -> None:
_edit(root / "1" / "outbox" / "r1-coverage.json", run_id="r9")
def _mut_the_approach_has_no_artefact(root: Path) -> None:
(root / "1" / "outbox" / "r1-a1-outcome.json").unlink()
def _mut_the_key_is_not_the_irs(root: Path) -> None:
_edit(
root / "1" / "outbox" / "r1-a1-outcome.json",
verdict_id=verdict_key(features_from_ir(_ir("a1", 424242.0))),
)
def _mut_the_stamp_contradicts_the_row(root: Path) -> None:
path = root / "1" / "outbox" / "r1-a1-proposal.json"
data = json.loads(path.read_text(encoding="utf-8"))
data["provenance"]["validator_decision"] = "rejected"
_write(path, data)
def _mut_the_outcome_type_contradicts_the_row(root: Path) -> None:
_edit(root / "1" / "outbox" / "r1-a1-outcome.json", outcome_type="rejected")
def _mut_the_figure_is_not_the_proposals(root: Path) -> None:
"""The IR says one figure and the coverage another — with the key RE-MINTED off the changed
IR, so this arm is felled by the figures and not by the key check standing in front of it."""
path = root / "1" / "outbox" / "r1-a1-proposal.json"
data = json.loads(path.read_text(encoding="utf-8"))
data["proposal"] = _ir("a1", 424242.0)
_write(path, data)
_edit(
root / "1" / "outbox" / "r1-a1-outcome.json",
verdict_id=verdict_key(features_from_ir(data["proposal"])),
)
def _mut_the_reason_is_not_the_rows(root: Path) -> None:
_edit(root / "0" / "outbox" / "r0-a1-outcome.json", reason="en annen grunn")
def _mut_an_artefact_belongs_to_another_run(root: Path) -> None:
_run_family(
root / "1" / "outbox",
"r1",
[{"id": "a9", "status": "validated", "detail": "", "saving_nok": 5.0}],
)
def _mut_an_unevaluated_approach_has_artefacts(root: Path) -> None:
_outcome(
root / "1",
[_row("a1", True, 1000.0, "f1"), _row("a2", False, None, stage="not_evaluated")],
)
_run_family(
root / "1" / "outbox",
"r1",
[{"id": "a2", "status": "rejected", "detail": "x", "saving_nok": None}],
)
def _mut_the_run_time_has_no_zone(root: Path) -> None:
_edit(root / "1" / "outcome.json", ran_at="2026-09-18T10:00:00")
@pytest.mark.parametrize(
("mutate", "marker"),
[
(_mut_declares_its_own_outbox, "outbox kan ikke oppgis"),
(_mut_coverage_is_another_run, "coverage-fila er kjøring"),
(_mut_the_approach_has_no_artefact, "r1-a1-outcome.json mangler"),
(_mut_the_key_is_not_the_irs, "verdict_id er ikke nøkkelen"),
(_mut_the_stamp_contradicts_the_row, "validator_decision"),
(_mut_the_outcome_type_contradicts_the_row, "utfallet er"),
(_mut_the_figure_is_not_the_proposals, "forslaget selv sier"),
(_mut_the_reason_is_not_the_rows, "utfallets grunn"),
(_mut_an_artefact_belongs_to_another_run, "coverage ikke nevner"),
(_mut_an_unevaluated_approach_has_artefacts, "ble ikke evaluert, men kjøringen skrev"),
(_mut_the_run_time_has_no_zone, "ran_at er ikke et ISO"),
],
)
def test_m6_the_run_family_must_stand_up_to_itself(
tmp_path: Path, mutate: Any, marker: str
) -> None:
"""One binding at a time, on a tree that is green until the mutation lands — and each arm is
read back by the reason it was refused for, so an arm cannot pass because some OTHER rule
happened to fire. The rc-0 control is the unmutated tree, asserted first."""
root = _green_rounds(tmp_path)
assert gate.score_changes(root, 3, _AI).status == gate.GREEN
mutate(root)
row = gate.score_changes(root, 3, _AI)
assert row.status == gate.RED
assert any(marker in x for x in row.exceptions), (marker, row.exceptions)
def test_m6_an_mtime_is_not_evidence_and_no_longer_decides_anything(tmp_path: Path) -> None:
"""The forger's third move was ``os.utime``. Every artefact in a green tree is stamped far in
the future, in the wrong order, and row 2 does not move: the run's time is the round's
declared ``ran_at``, and the row says so out loud."""
root = _green_rounds(tmp_path)
for n, path in enumerate(sorted(root.rglob("*.json"))):
os.utime(path, (_T0 - n * 1000, _T0 - n * 1000))
row = gate.score_changes(root, 3, _AI)
assert (row.k, row.status) == (3, gate.GREEN), row.exceptions
assert row.attests == (gate.RUN_ATTESTATION,)
assert "bekrefter operatøren" in gate.RUN_ATTESTATION
def _outcome_obj(rows: list[dict[str, Any]], removed: dict[str, set[str]] | None = None) -> Any:
from datetime import datetime, timezone