test(v1-gate): harden the gate against a handwritten green

An independent review made rows 1, 2 and 4 green from a handwritten
directory in a minute, and 10 of 20 mutants survived the gate's tests.

Rounds now need a new point and their own ids, a timezone-aware given_at
in order, and a report the feedback was given on; every outcome must name
a run whose own coverage confirms (a)-(d), the feedback must fall between
the two runs, and a NOK change under 1 % is noise. Row 4 counts content
lines kept unchanged and in order, shows the expert's additions, and calls
a byte-identical copy untouched unless round 3 acknowledges it. Row 6
counts the runs' own proposals. Types 3 and 7 are proven through the real
flags with the action in the result (still 3 of 8). The contract numbers
and the evidence register are pinned to their source. Every run prints
that rows 1-2 cannot prove who wrote the feedback. A rounds directory
inside the repo that git would commit, and a missing stress or bundle
root, are usage errors.

The review's 20 mutants, re-run: 20 of 20 killed.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Kjell Tore Guttormsen 2026-09-17 17:53:40 +02:00
commit 9825b2677c
5 changed files with 979 additions and 213 deletions

View file

@ -3112,6 +3112,23 @@ Python ≥3.10. MAF (`agent-framework-core` 1.16.0, `-orchestrations` 1.1.1 —
`r761-2025` uten `index.md`) — `--bundle-root` peker da på en utpakket kopi; bevisene for type 1, 3 `r761-2025` uten `index.md`) — `--bundle-root` peker da på en utpakket kopi; bevisene for type 1, 3
og 7 er EKSISTERENDE tester registrert ved node-id, så en omdøping gjør typen rød til registeret og 7 er EKSISTERENDE tester registrert ved node-id, så en omdøping gjør typen rød til registeret
rettes (gatet av en egen arm). rettes (gatet av en egen arm).
- **v1-gaten er HERDET mot forfalskning — og sier det den ikke kan bevise (uavhengig review 17.09):**
reviewen gjorde rad 1, 2 og 4 grønne fra en håndskrevet katalog på ett minutt, og 10 av 20 mutanter
overlevde `tests/test_v1_gate.py`. Nå: (M-1) hver runde må ha minst ett NYTT punkt og egne id-er,
et `given_at` med tidssone i stigende rekkefølge, og være gitt på en rapport (`<n-1>/report.md`);
hver `outcome.json` må navngi en kjøring (`run_id` + `outbox`) hvis EGEN `coverage.json` bekrefter
(a)(d) — en håndskrevet eller tom grunnkjøring nektes — og feedbacken må være gitt MELLOM de to
kjøringene (coverage-filens tid); en NOK-endring under 1 % er støy. AI-vakten er et kjent-tekst-
filter (store/små bokstaver ignoreres), ALDRI en detektor, og gaten skriver på hver kjøring at
rad 12 ikke beviser forfatterskap (`ATTESTATION`). (M-2) rad 4 teller INNHOLDSLINJER (blanke,
skillelinjer og tabellrammer ute; etterstilte mellomrom strippet) som står uendret OG i rekkefølge
(`difflib`), viser tilleggene som eget tall, og en byte-identisk kopi er «ikke rørt» med mindre
runde 3 kvitterer `report_unchanged: true`. (M-3) rad 6 teller kjøringenes egne forslag
(15 validerte i stressrunde 6, ikke 10). (M-4) type 3 og 7 bevises gjennom de EKTE flaggene med
handlingen i resultatet (`tests/test_v1_probes.py`); tallet står på 3 av 8. (M-5) kontraktens
tall og bevisregisteret er pinnet mot kilden i testen. Småfunn: rundekatalog inne i repoet uten
gitignore og manglende `--stress-root`/`--bundle-root` gir exit 2; en typeannotasjon teller ikke
som kallsted. Reviewens 20 mutanter kjørt på nytt: **20 av 20 røde**.
- **Et forslag uten tilnærmingens EGEN erklæring kan ikke bære `validated` (rad 6, 17.09):** målt - **Et forslag uten tilnærmingens EGEN erklæring kan ikke bære `validated` (rad 6, 17.09):** målt
på stressrunde 6 hadde alle 10 validerte tilnærmingene bare kjørings-erklæringer, som ingen kan på stressrunde 6 hadde alle 10 validerte tilnærmingene bare kjørings-erklæringer, som ingen kan
knytte til én tilnærming — og tre falsifiseringsarmer validerte. `declare_requirement` tar derfor knytte til én tilnærming — og tre falsifiseringsarmer validerte. `declare_requirement` tar derfor

View file

@ -15,7 +15,10 @@
}, },
"3": { "3": {
"label": "vinklinger - nye vinklinger", "label": "vinklinger - nye vinklinger",
"evidence": ["tests/test_mandate_cli.py::test_run_settles_against_the_mandate_afterwards"] "evidence": [
"tests/test_v1_probes.py::test_type_3_a_commissioned_angle_is_evaluated_through_the_cli",
"tests/test_v1_probes.py::test_type_3_a_new_angle_changes_the_outcome"
]
}, },
"4": { "4": {
"label": "lette paa krav", "label": "lette paa krav",
@ -32,7 +35,7 @@
"7": { "7": {
"label": "MCP - verktoey i debatten", "label": "MCP - verktoey i debatten",
"evidence": [ "evidence": [
"tests/test_mcp_run_loadbearing.py::test_configured_server_becomes_a_tool_the_agents_have", "tests/test_v1_probes.py::test_type_7_the_mcp_flag_puts_a_service_the_run_calls_into_the_result",
"tests/test_b4_mcp_call_trace_loadbearing.py::test_a_called_mcp_tool_is_recorded_in_provenance" "tests/test_b4_mcp_call_trace_loadbearing.py::test_a_called_mcp_tool_is_recorded_in_provenance"
] ]
}, },

View file

@ -25,12 +25,16 @@ import subprocess
import sys import sys
import tempfile import tempfile
import xml.etree.ElementTree as ET import xml.etree.ElementTree as ET
from collections import Counter
from collections.abc import Callable, Iterable, Mapping, Sequence from collections.abc import Callable, Iterable, Mapping, Sequence
from dataclasses import asdict, dataclass, field from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
import difflib
import re
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
from portfolio_optimiser.validator import rejection_stage
_DATA = Path(__file__).with_name("v1_gate.json") _DATA = Path(__file__).with_name("v1_gate.json")
_PACKAGE_SRC = Path(__file__).resolve().parents[1] _PACKAGE_SRC = Path(__file__).resolve().parents[1]
_REPO_ROOT = Path(__file__).resolve().parents[3] _REPO_ROOT = Path(__file__).resolve().parents[3]
@ -38,6 +42,10 @@ _REPO_ROOT = Path(__file__).resolve().parents[3]
DEFAULT_ROUNDS_DIR = "v1-rounds" DEFAULT_ROUNDS_DIR = "v1-rounds"
#: A line of an AI-authored document shorter than this is too generic to identify its origin. #: A line of an AI-authored document shorter than this is too generic to identify its origin.
_AI_LINE_MIN = 30 _AI_LINE_MIN = 30
#: Row 2 (d): a change in validated NOK below this share of the earlier figure is noise.
_NOK_NOISE = 0.01
#: Printed on every run: the one thing rows 1-2 cannot prove.
ATTESTATION = "rad 12 beviser ikke at en fagperson skrev feedbacken; det bekrefter operatøren"
GREEN = "GRØNN" GREEN = "GRØNN"
RED = "RØD" RED = "RØD"
@ -46,30 +54,36 @@ DIAGNOSIS = "DIAGNOSE"
NOT_MEASURED = "IKKE MÅLT" NOT_MEASURED = "IKKE MÅLT"
ROUNDS_CONTRACT = """\ ROUNDS_CONTRACT = """\
Rundekatalogen (--rounds-dir) har fast form: Rundekatalogen (--rounds-dir) har fast form. Runde n = tilbakemelding rapport n-1, kjøring n:
<rounds-dir>/0/outcome.json grunnkjøringen (runde 0) som runde 1 måles mot <rounds-dir>/0/report.md rapporten fra grunnkjøringen (runde 0)
<rounds-dir>/<n>/report.md rapporten slik fagpersonen fikk den (n = 1, 2, 3) <rounds-dir>/0/outcome.json grunnkjøringen som runde 1 måles mot
<rounds-dir>/<n>/feedback.json fagpersonens tilbakemelding <rounds-dir>/<n>/feedback.json fagpersonens tilbakemelding rapport n-1 (n = 1, 2, 3)
<rounds-dir>/<n>/outcome.json utfallet rapporten er bygget fra <rounds-dir>/<n>/outcome.json kjøring n, gjort ETTER den tilbakemeldingen
<rounds-dir>/<n>/report.md rapporten bygget fra kjøring n
<rounds-dir>/3/report.kept.md runde 3-rapporten slik fagpersonen BEHOLDT den <rounds-dir>/3/report.kept.md runde 3-rapporten slik fagpersonen BEHOLDT den
feedback.json: feedback.json:
{"author": "<fagpersonen>", {"author": "<fagpersonen>", "given_at": "<ISO-8601 med tidssone>",
"report_unchanged": true (valgfri, kun runde 3: kvitterer for en urørt rapport),
"items": [{"id": "<unik id>", "type": <1-8>, "text": "<tilbakemeldingen>"}]} "items": [{"id": "<unik id>", "type": <1-8>, "text": "<tilbakemeldingen>"}]}
outcome.json: outcome.json (hver rad sjekkes mot kjøringens egen <outbox>/<run_id>-coverage.json):
{"run_id": "<kjøringen>", {"run_id": "<kjøringen>", "outbox": "<utboksen, relativ til denne fila eller absolutt>",
"approaches": [{"id": "<tilnærming>", "validated": true|false, "approaches": [{"id": "<tilnærming>", "validated": true|false,
"stage": "<avvisningsstadium, tom når validert>", "stage": "<avvisningsstadium som validator.rejection_stage gir, tom når validert>",
"validated_nok": <tall eller null>, "validated_nok": <tall eller null>,
"feedback_ids": ["<id-er fra feedback.json som forklarer raden>"]}], "feedback_ids": ["<id-er fra feedback.json som forklarer raden>"]}],
"removed": [{"id": "<tilnærming fjernet siden forrige runde>", "feedback_ids": [...]}]} "removed": [{"id": "<tilnærming fjernet siden forrige runde>", "feedback_ids": [...]}]}
En runde har målbar endring når den skiller seg fra forrige minst én av (a) settet av En runde har målbar endring når kjøringen skiller seg fra forrige minst én av (a) settet av
tilnærmings-id-er, (b) hvilke som er validert, (c) avvisningsstadium, (d) validert NOK OG minst tilnærmings-id-er, (b) hvilke som er validert, (c) avvisningsstadium, (d) validert NOK (endring
én endret rad bærer en feedback-id gitt i DENNE runden. Tekst tatt fra et AI-forfattet dokument under 1 % er støy) OG minst én endret rad bærer en feedback-id gitt i DENNE runden, gitt mellom
(docs/ekspert-svar.md) teller aldri som fagperson-tilbakemelding. de to kjøringene. Hver runde 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.
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 være
gitignored.
""" """
@ -102,13 +116,18 @@ def load_config(path: Path = _DATA) -> dict[str, Any]:
def _norm(text: str) -> str: def _norm(text: str) -> str:
return " ".join(text.split()) """Whitespace collapsed and case folded — the AI guard's one normalisation."""
return " ".join(text.split()).casefold()
def ai_authored_lines(repo_root: Path, docs: Sequence[str]) -> tuple[str, str] | None: def ai_authored_lines(repo_root: Path, docs: Sequence[str]) -> tuple[str, str] | None:
"""The normalised full text and the joined long lines of every AI-authored document, or """The normalised full text and the joined long lines of every AI-authored document, or
``None`` when one of them cannot be read the guard then cannot run, and a round it cannot ``None`` when one of them cannot be read the guard then cannot run, and a round it cannot
check is never counted.""" check is never counted.
A KNOWN-TEXT filter, not an authorship detector: it refuses text lifted from the listed
documents (case and whitespace ignored), and nothing else. Authorship itself is not verifiable
here, and the gate's output says so on every run (``ATTESTATION``)."""
texts: list[str] = [] texts: list[str] = []
for rel in docs: for rel in docs:
path = repo_root / rel path = repo_root / rel
@ -131,37 +150,109 @@ def _is_ai_text(text: str, ai: tuple[str, str]) -> bool:
return any(line in item for line in lines.splitlines() if line) return any(line in item for line in lines.splitlines() if line)
def read_feedback(round_dir: Path, ai: tuple[str, str] | None) -> tuple[set[str], str]: def _parse_time(value: Any) -> datetime | None:
"""The ids of a round's feedback items, and ``""`` — or an empty set and the reason.""" try:
stamp = datetime.fromisoformat(str(value))
except ValueError:
return None
return stamp if stamp.tzinfo is not None else None
@dataclass(frozen=True)
class Feedback:
ids: frozenset[str]
texts: frozenset[str]
given_at: datetime
report_unchanged: bool
def _read_feedback_file(round_dir: Path, ai: tuple[str, str] | None) -> tuple[Feedback | None, str]:
"""One round's feedback file, checked on its own; ``(None, why)`` when it does not hold."""
name = f"runde {round_dir.name}"
path = round_dir / "feedback.json" path = round_dir / "feedback.json"
if not path.is_file(): if not path.is_file():
others = sorted(p.name for p in round_dir.glob("feedback.*")) if round_dir.is_dir() else [] others = sorted(p.name for p in round_dir.glob("feedback.*")) if round_dir.is_dir() else []
extra = f" (fant {', '.join(others)}; kontrakten er feedback.json)" if others else "" extra = f" (fant {', '.join(others)}; kontrakten er feedback.json)" if others else ""
return set(), f"runde {round_dir.name}: feedback.json mangler{extra}" return None, f"{name}: feedback.json mangler{extra}"
try: try:
data = json.loads(path.read_text(encoding="utf-8")) data = json.loads(path.read_text(encoding="utf-8"))
author = str(data["author"]).strip() author = str(data["author"]).strip()
items = list(data["items"]) items = list(data["items"])
given_raw = data["given_at"]
except (ValueError, KeyError, TypeError) as exc: except (ValueError, KeyError, TypeError) as exc:
return set(), f"runde {round_dir.name}: feedback.json uleselig ({exc!r})" return None, f"{name}: feedback.json uleselig ({exc!r})"
if not author: if not author:
return set(), f"runde {round_dir.name}: feedback.json navngir ingen fagperson" return None, f"{name}: feedback.json navngir ingen fagperson"
given_at = _parse_time(given_raw)
if given_at is None:
return None, f"{name}: given_at er ikke et ISO-tidsstempel med tidssone"
if ai is None: if ai is None:
return set(), f"runde {round_dir.name}: AI-vakten kunne ikke lese sine kilder" return None, f"{name}: AI-vakten kunne ikke lese sine kilder"
ids: set[str] = set() ids: set[str] = set()
texts: set[str] = set()
for item in items: for item in items:
try: try:
item_id, item_type, text = str(item["id"]), int(item["type"]), str(item["text"]) item_id, item_type, text = str(item["id"]), int(item["type"]), str(item["text"])
except (KeyError, TypeError, ValueError): except (KeyError, TypeError, ValueError):
return set(), f"runde {round_dir.name}: et feedback-punkt mangler id/type/text" return None, f"{name}: et feedback-punkt mangler id/type/text"
if not item_id or not text.strip() or not 1 <= item_type <= 8: if not item_id or not text.strip() or not 1 <= item_type <= 8:
return set(), f"runde {round_dir.name}: punkt {item_id!r} er tomt eller har ukjent type" return None, f"{name}: punkt {item_id!r} er tomt eller har ukjent type"
if item_id in ids:
return None, f"{name}: punkt-id {item_id!r} er brukt to ganger"
if _is_ai_text(text, ai): if _is_ai_text(text, ai):
return set(), f"runde {round_dir.name}: punkt {item_id!r} er AI-forfattet tekst" return None, f"{name}: punkt {item_id!r} er AI-forfattet tekst"
ids.add(item_id) ids.add(item_id)
texts.add(_norm(text))
if not ids: if not ids:
return set(), f"runde {round_dir.name}: feedback.json har ingen punkter" return None, f"{name}: feedback.json har ingen punkter"
return ids, "" unchanged = data.get("report_unchanged") is True
return Feedback(frozenset(ids), frozenset(texts), given_at, unchanged), ""
def read_rounds(
rounds_dir: Path, required: int, ai: tuple[str, str] | None
) -> dict[int, tuple[Feedback | None, str]]:
"""Every round's feedback, with the cross-round rules applied in round order: a round must
bring at least one point no earlier round gave, may not reuse an earlier round's ids (tracing is
per round), must come after the previous round's feedback, and must have been given on a
report (``<n-1>/report.md``)."""
result: dict[int, tuple[Feedback | None, str]] = {}
seen_ids: set[str] = set()
seen_texts: set[str] = set()
last: datetime | None = None
for n in range(1, required + 1):
feedback, why = _read_feedback_file(rounds_dir / str(n), ai)
if feedback is not None:
if not (rounds_dir / str(n - 1) / "report.md").is_file():
feedback, why = (
None,
f"runde {n}: gitt på en rapport som mangler ({n - 1}/report.md)",
)
elif feedback.ids & seen_ids:
reused = ", ".join(sorted(feedback.ids & seen_ids))
feedback, why = (
None,
f"runde {n}: id-er fra en tidligere runde gjenbrukt ({reused})",
)
elif feedback.texts <= seen_texts:
feedback, why = (
None,
f"runde {n}: ingen punkt som ikke alt er gitt i en tidligere runde",
)
elif last is not None and feedback.given_at <= last:
feedback, why = None, f"runde {n}: given_at er ikke etter forrige rundes"
if feedback is not None:
seen_ids |= feedback.ids
seen_texts |= feedback.texts
last = feedback.given_at
result[n] = (feedback, why)
return result
def read_feedback(round_dir: Path, ai: tuple[str, str] | None) -> tuple[set[str], str]:
"""The ids of ONE round's feedback items checked on its own, and ``""`` — or why not."""
feedback, why = _read_feedback_file(round_dir, ai)
return (set(feedback.ids), "") if feedback is not None else (set(), why)
def score_rounds(rounds_dir: Path, required: int, ai: tuple[str, str] | None) -> Row: def score_rounds(rounds_dir: Path, required: int, ai: tuple[str, str] | None) -> Row:
@ -170,14 +261,17 @@ def score_rounds(rounds_dir: Path, required: int, ai: tuple[str, str] | None) ->
if not rounds_dir.is_dir(): if not rounds_dir.is_dir():
exceptions.append(f"{rounds_dir} finnes ikke") exceptions.append(f"{rounds_dir} finnes ikke")
else: else:
for n in range(1, required + 1): for _, (feedback, why) in sorted(read_rounds(rounds_dir, required, ai).items()):
ids, why = read_feedback(rounds_dir / str(n), ai) if feedback is not None:
if ids:
k += 1 k += 1
else: else:
exceptions.append(why) exceptions.append(why)
status = GREEN if k == required else RED status = GREEN if k == required else RED
reason = "alle runder har fagperson-tilbakemelding" if k == required else exceptions[0] reason = (
"form verifisert i alle runder (forfatterskap: se attestering)"
if k == required
else exceptions[0]
)
return Row( return Row(
"rounds", "rounds",
"1 runder med ekte fagperson", "1 runder med ekte fagperson",
@ -189,42 +283,102 @@ def score_rounds(rounds_dir: Path, required: int, ai: tuple[str, str] | None) ->
) )
def _read_outcome(path: Path) -> tuple[dict[str, dict[str, Any]], dict[str, set[str]], str, str]: @dataclass(frozen=True)
"""Rows by approach id, removed ids with their feedback ids, the run id, and ``""`` or why.""" class Outcome:
rows: dict[str, dict[str, Any]]
removed: dict[str, set[str]]
run_id: str
ran_at: datetime
def _stage_of(status: str, detail: str) -> str:
if status == "validated":
return ""
if status == "not_evaluated":
return "not_evaluated"
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."""
try: try:
data = json.loads(path.read_text(encoding="utf-8")) data = json.loads(path.read_text(encoding="utf-8"))
rows = {str(a["id"]): a for a in data["approaches"]} rows = {str(a["id"]): a for a in data["approaches"]}
removed = { removed = {
str(r["id"]): set(map(str, r.get("feedback_ids", ()))) for r in data.get("removed", ()) str(r["id"]): set(map(str, r.get("feedback_ids", ()))) for r in data.get("removed", ())
} }
return rows, removed, str(data.get("run_id", "")), "" run_id = str(data["run_id"]).strip()
outbox = Path(str(data["outbox"])).expanduser()
except FileNotFoundError: except FileNotFoundError:
return {}, {}, "", f"{path} mangler" return None, f"{path} mangler"
except (ValueError, KeyError, TypeError) as exc: except (ValueError, KeyError, TypeError) as exc:
return {}, {}, "", f"{path} uleselig ({exc!r})" 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
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)"
try:
coverage = json.loads(coverage_path.read_text(encoding="utf-8"))["rows"]
except (ValueError, KeyError, TypeError) as exc:
return None, f"{coverage_path} uleselig ({exc!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)
return Outcome(rows, removed, run_id, ran_at), ""
def _row_key(row: Mapping[str, Any]) -> tuple[bool, str, Any]: def _row_key(row: Mapping[str, Any]) -> tuple[bool, str, Any]:
return bool(row.get("validated")), str(row.get("stage") or ""), row.get("validated_nok") nok = row.get("validated_nok")
return (
bool(row.get("validated")),
str(row.get("stage") or ""),
None if nok is None else float(nok),
)
def round_changed(before: Path, after: Path, feedback_ids: set[str]) -> tuple[bool, str]: def _nok_changed(before: Any, after: Any) -> bool:
"""Whether round ``after`` changed measurably against ``before`` AND the change is traced to """(d) with a noise floor. ``validated_nok`` is the model's own claim, so two runs on the same
feedback given in this round. The second half is what keeps model noise out.""" input can differ by rounding; a change smaller than 1 % of the earlier figure (and never less
prev, _, _, why = _read_outcome(before) than 1 NOK) is not something an expert's feedback asked for, and it does not count."""
if why: if (before is None) != (after is None):
return False, why return True
cur, removed, _, why = _read_outcome(after) if before is None or after is None:
if why: return False
return False, why return abs(float(after) - float(before)) >= max(1.0, _NOK_NOISE * abs(float(before)))
def _changed(before: Mapping[str, Any], after: Mapping[str, Any]) -> bool:
b, a = _row_key(before), _row_key(after)
return b[:2] != a[:2] or _nok_changed(b[2], a[2])
def outcomes_changed(prev: Outcome, cur: Outcome, feedback_ids: set[str]) -> tuple[bool, str]:
"""Whether ``cur`` changed measurably against ``prev`` AND the change is traced to feedback
given before ``cur`` ran. The second half is what keeps model noise out."""
changed: dict[str, set[str]] = {} changed: dict[str, set[str]] = {}
for aid, row in cur.items(): for aid, row in cur.rows.items():
if aid not in prev or _row_key(prev[aid]) != _row_key(row): if aid not in prev.rows or _changed(prev.rows[aid], row):
changed[aid] = set(map(str, row.get("feedback_ids", ()))) changed[aid] = set(map(str, row.get("feedback_ids", ())))
for aid in prev.keys() - cur.keys(): for aid in prev.rows.keys() - cur.rows.keys():
changed[aid] = removed.get(aid, set()) changed[aid] = cur.removed.get(aid, set())
if not changed: if not changed:
return False, "ingen endring i (a)-(d)" return False, "ingen endring i (a)-(d) over støygrensen"
traced = sorted(aid for aid, ids in changed.items() if ids & feedback_ids) traced = sorted(aid for aid, ids in changed.items() if ids & feedback_ids)
if not traced: if not traced:
return False, f"{len(changed)} rad(er) endret, ingen sporet til rundens feedback-id-er" return False, f"{len(changed)} rad(er) endret, ingen sporet til rundens feedback-id-er"
@ -234,17 +388,27 @@ def round_changed(before: Path, after: Path, feedback_ids: set[str]) -> tuple[bo
def score_changes(rounds_dir: Path, required: int, ai: tuple[str, str] | None) -> Row: def score_changes(rounds_dir: Path, required: int, ai: tuple[str, str] | None) -> Row:
exceptions: list[str] = [] exceptions: list[str] = []
k = 0 k = 0
_, _, base_run, base_why = _read_outcome(rounds_dir / "0" / "outcome.json") base_path = rounds_dir / "0" / "outcome.json"
base = f"runde 0 = {rounds_dir / '0' / 'outcome.json'}" base_outcome, base_why = read_outcome(base_path)
base += f" ({base_run})" if base_run else (f"{base_why}" if base_why else "") 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 {}
for n in range(1, required + 1): for n in range(1, required + 1):
ids, why = read_feedback(rounds_dir / str(n), ai) feedback, why = feedback_by_round.get(n, (None, f"runde {n}: {rounds_dir} finnes ikke"))
if not ids: if feedback is None:
exceptions.append(why) exceptions.append(why)
continue continue
ok, detail = round_changed( prev, why_prev = read_outcome(rounds_dir / str(n - 1) / "outcome.json")
rounds_dir / str(n - 1) / "outcome.json", rounds_dir / str(n) / "outcome.json", ids cur, why_cur = read_outcome(rounds_dir / str(n) / "outcome.json")
) if prev is None or cur is None:
exceptions.append(f"runde {n}: {why_prev or why_cur}")
continue
if not prev.ran_at <= feedback.given_at <= cur.ran_at:
exceptions.append(
f"runde {n}: feedbacken er ikke gitt mellom kjøring {prev.run_id} og {cur.run_id}"
)
continue
ok, detail = outcomes_changed(prev, cur, set(feedback.ids))
if ok: if ok:
k += 1 k += 1
else: else:
@ -354,36 +518,74 @@ def score_types(types: Mapping[str, Any], outcomes: Mapping[str, str]) -> Row:
# --------------------------------------------------------------------------------------------- # ---------------------------------------------------------------------------------------------
def kept_ratio(report: Path, kept: Path) -> tuple[int, int, str]: #: A line that carries no reading: horizontal rules, table rules, and non-breaking-space fillers.
_MARKUP_ONLY = re.compile(
r"^\s*(?:(?:[-*_=]\s*){3,}|\|?\s*:?-+:?\s*(?:\|\s*:?-+:?\s*)*\|?|&nbsp;)\s*$"
)
def content_lines(text: str) -> list[str]:
"""The lines a reader keeps or rewrites: trailing whitespace stripped (an editor's doing, never
the expert's), blank lines and markup-only lines dropped, everything else — headings included —
kept in order. Nothing else is normalised."""
lines = [line.rstrip() for line in text.splitlines()]
return [x for x in lines if re.search(r"\w", x) and not _MARKUP_ONLY.match(x)]
@dataclass(frozen=True)
class Kept:
kept: int
total: int
added: int
untouched: bool
why: str = ""
def kept_ratio(report: Path, kept: Path) -> Kept:
"""How much of the report the expert kept: content lines of ``report`` that survive in
``kept`` IN ORDER (each line matched at most once a multiset, and a reshuffle is a change),
plus the expert's additions as their own number. A byte-identical copy is flagged as untouched:
nobody can tell it from a report nobody read."""
if not report.is_file() or not kept.is_file(): if not report.is_file() or not kept.is_file():
return 0, 0, "ingen rapport" return Kept(0, 0, 0, False, "ingen rapport")
lines = [x for x in report.read_text(encoding="utf-8").splitlines() if x.strip()] raw_report = report.read_bytes()
if not lines: raw_kept = kept.read_bytes()
return 0, 0, "tom rapport" before = content_lines(raw_report.decode("utf-8"))
pool = Counter(x for x in kept.read_text(encoding="utf-8").splitlines() if x.strip()) after = content_lines(raw_kept.decode("utf-8"))
same = 0 if not before:
for line in lines: return Kept(0, 0, 0, False, "tom rapport")
if pool[line] > 0: matcher = difflib.SequenceMatcher(None, before, after, autojunk=False)
pool[line] -= 1 same = sum(block.size for block in matcher.get_matching_blocks())
same += 1 return Kept(same, len(before), len(after) - same, raw_report == raw_kept)
return same, len(lines), ""
def score_kept(rounds_dir: Path, threshold: float) -> Row: def score_kept(rounds_dir: Path, threshold: float, ai: tuple[str, str] | None = None) -> Row:
same, total, why = kept_ratio( result = kept_ratio(rounds_dir / "3" / "report.md", rounds_dir / "3" / "report.kept.md")
rounds_dir / "3" / "report.md", rounds_dir / "3" / "report.kept.md" title = f"4 runde 3-rapport beholdt (≥ {threshold:.0%} innholdslinjer)"
) if result.why:
title = f"4 runde 3-rapport beholdt (≥ {threshold:.0%} linjer)" return Row("kept", title, None, result.total or None, RED, result.why)
if why: added = f"; {result.added} linje(r) lagt til av fagpersonen"
return Row("kept", title, None, total or None, RED, why) if result.untouched:
ok = same >= threshold * total feedback, _ = _read_feedback_file(rounds_dir / "3", ai)
if feedback is None or not feedback.report_unchanged:
return Row(
"kept",
title,
None,
result.total,
RED,
"ikke rørt: report.kept.md er byte-identisk med report.md; kvitter med "
'"report_unchanged": true i 3/feedback.json',
)
ok = result.kept >= threshold * result.total
return Row( return Row(
"kept", "kept",
title, title,
same, result.kept,
total, result.total,
GREEN if ok else RED, GREEN if ok else RED,
f"{same / total:.1%} av ikke-tomme linjer uendret", f"{result.kept / result.total:.1%} av innholdslinjene uendret og i rekkefølge{added}",
diagnostics=(f"lagt til: {result.added}",),
) )
@ -400,8 +602,31 @@ def _imports(tree: ast.AST, construct: str, package: str) -> set[str]:
return names return names
def _annotations(node: ast.AST) -> set[int]:
"""ids of every node inside a type annotation — a name used only as a type is not a use."""
found: set[int] = set()
for x in ast.walk(node):
parts: list[ast.AST | None] = []
if isinstance(x, (ast.FunctionDef, ast.AsyncFunctionDef)):
parts.append(x.returns)
elif isinstance(x, ast.arg):
parts.append(x.annotation)
elif isinstance(x, ast.AnnAssign):
parts.append(x.annotation)
for part in parts:
if part is not None:
found |= {id(y) for y in ast.walk(part)}
return found
def _referenced(nodes: Iterable[ast.AST], names: set[str]) -> bool: def _referenced(nodes: Iterable[ast.AST], names: set[str]) -> bool:
return any(isinstance(x, ast.Name) and x.id in names for node in nodes for x in ast.walk(node)) for node in nodes:
typed = _annotations(node)
if any(
isinstance(x, ast.Name) and x.id in names and id(x) not in typed for x in ast.walk(node)
):
return True
return False
def maf_presence(point: Mapping[str, Any], src: Path) -> tuple[bool, bool]: def maf_presence(point: Mapping[str, Any], src: Path) -> tuple[bool, bool]:
@ -502,9 +727,38 @@ class StressMeasure:
missing: str = "" missing: str = ""
#: Declarations with no ``approach_id`` — written before the rule; the row cannot be measured. #: Declarations with no ``approach_id`` — written before the rule; the row cannot be measured.
unaddressed: int = 0 unaddressed: int = 0
#: Of ``validated``, how many were the runs' own proposals (M-3).
own_validated: int = 0
undeclared_ids: tuple[str, ...] = field(default=()) undeclared_ids: tuple[str, ...] = field(default=())
def _own_proposals(
evidence: Mapping[str, Any], stress_root: Path
) -> tuple[int, int, tuple[str, ...]]:
"""(validated own proposals, of those without an ``own-proposal`` declaration, their labels)."""
validated = undeclared = 0
ids: list[str] = []
for run_spec in evidence["runs"]:
outbox = stress_root / run_spec["outbox"]
run_id = run_spec["run_id"]
outcome = outbox / f"{run_id}-own-proposal-outcome.json"
if not outcome.is_file():
continue
if json.loads(outcome.read_text(encoding="utf-8")).get("outcome_type") != "validated":
continue
validated += 1
debate = outbox / f"{run_id}-debate.json"
records = (
json.loads(debate.read_text(encoding="utf-8")).get("requirements", [])
if debate.is_file()
else []
)
if not any(r.get("approach_id") == "own-proposal" for r in records):
undeclared += 1
ids.append(f"own-proposal ({run_id})")
return validated, undeclared, tuple(ids)
def measure_stress( def measure_stress(
evidence: Mapping[str, Any], repo_root: Path, stress_root: Path, bundle_root: Path evidence: Mapping[str, Any], repo_root: Path, stress_root: Path, bundle_root: Path
) -> StressMeasure: ) -> StressMeasure:
@ -542,20 +796,24 @@ def measure_stress(
approaches = [a for v in verdicts for a in v.approaches] approaches = [a for v in verdicts for a in v.approaches]
validated = [a for a in approaches if a.status == "validated"] validated = [a for a in approaches if a.status == "validated"]
undeclared = [a for a in validated if a.requirement_source != "approach"] undeclared = [a for a in validated if a.requirement_source != "approach"]
# M-3: the run's OWN proposal is gated by the same rule, but the judge scores only the
# commissioned approaches (the fasit has rows for nothing else). Read straight off each outbox.
own_validated, own_undeclared, own_ids = _own_proposals(evidence, stress_root)
unaddressed = sum(v.unaddressed_declarations for v in verdicts) unaddressed = sum(v.unaddressed_declarations for v in verdicts)
commissioned = sum( commissioned = sum(
len(load_mandate(repo_root / c / "mandate.json").approaches) for c in contexts len(load_mandate(repo_root / c / "mandate.json").approaches) for c in contexts
) )
return StressMeasure( return StressMeasure(
validated=len(validated), validated=len(validated) + own_validated,
undeclared=len(undeclared), undeclared=len(undeclared) + own_undeclared,
undeclared_anywhere=sum(1 for a in validated if a.requirement_source == "absent"), undeclared_anywhere=sum(1 for a in validated if a.requirement_source == "absent"),
named=sum(1 for a in approaches if a.named), named=sum(1 for a in approaches if a.named),
rows=len(approaches), rows=len(approaches),
commissioned=commissioned, commissioned=commissioned,
where=str(stress_root), where=str(stress_root),
unaddressed=unaddressed, unaddressed=unaddressed,
undeclared_ids=tuple(a.approach_id for a in undeclared), undeclared_ids=tuple(a.approach_id for a in undeclared) + own_ids,
own_validated=own_validated,
) )
@ -585,14 +843,16 @@ def score_undeclared(
f"(approach_id mangler på {m.unaddressed} erklæring(er))" f"(approach_id mangler på {m.unaddressed} erklæring(er))"
) )
diagnostics = ( diagnostics = (
f"før regelen: {m.undeclared} av {m.validated} validerte uten tilnærmingens egen " f"før regelen: {m.undeclared} av {m.validated} validerte (hvorav {m.own_validated} "
"erklæring — regelen ville gjort dem unsupported, men modellen fikk aldri spørsmålet", "egne forslag) uten tilnærmingens egen erklæring — regelen ville gjort dem "
"unsupported, men modellen fikk aldri spørsmålet",
) )
else: else:
k, n = m.undeclared, m.validated k, n = m.undeclared, m.validated
reason = ( reason = (
f"{probe_state}; {label} ({m.where}): {k} av {n} validerte uten erklæring fra " f"{probe_state}; {label} ({m.where}): {k} av {n} validerte uten erklæring fra "
f"tilnærmingen; {m.undeclared_anywhere} uten noen erklæring i kjøringen" f"tilnærmingen (hvorav {m.own_validated} egne forslag i nevneren); "
f"{m.undeclared_anywhere} uten noen erklæring i kjøringen"
) )
exceptions += [f"validert uten erklæring: {a}" for a in m.undeclared_ids] exceptions += [f"validert uten erklæring: {a}" for a in m.undeclared_ids]
if failing or k: if failing or k:
@ -682,7 +942,7 @@ def evaluate(
score_rounds(rounds_dir, required, ai), score_rounds(rounds_dir, required, ai),
score_changes(rounds_dir, required, ai), score_changes(rounds_dir, required, ai),
score_types(types, outcomes), score_types(types, outcomes),
score_kept(rounds_dir, float(config["keep_threshold"])), score_kept(rounds_dir, float(config["keep_threshold"]), ai),
score_maf(config["maf_points"], green_types(types, outcomes), src), score_maf(config["maf_points"], green_types(types, outcomes), src),
score_undeclared(probes, outcomes, stress_measure, evidence["label"]), score_undeclared(probes, outcomes, stress_measure, evidence["label"]),
score_named(stress_measure, evidence["label"]), score_named(stress_measure, evidence["label"]),
@ -697,6 +957,8 @@ def render(rows: Sequence[Row]) -> str:
out = ["rad | k av N | status | grunn"] out = ["rad | k av N | status | grunn"]
out += [r.line() for r in rows] out += [r.line() for r in rows]
out.append("") out.append("")
out.append(f"Attestering: {ATTESTATION}.")
out.append("")
out.append("Unntak fra 100 %:") out.append("Unntak fra 100 %:")
for r in rows: for r in rows:
for x in r.exceptions: for x in r.exceptions:
@ -706,6 +968,20 @@ def render(rows: Sequence[Row]) -> str:
return "\n".join(out) return "\n".join(out)
def _safe_rounds_dir(path: Path) -> bool:
"""Outside the repository, or inside it and ignored by git."""
resolved = path.resolve()
try:
resolved.relative_to(_REPO_ROOT)
except ValueError:
return True
probe = resolved / "1" / "feedback.json"
proc = subprocess.run(
["git", "check-ignore", "-q", str(probe)], cwd=_REPO_ROOT, capture_output=True
)
return proc.returncode == 0
def main(argv: Sequence[str] | None = None) -> int: def main(argv: Sequence[str] | None = None) -> int:
parser = argparse.ArgumentParser( parser = argparse.ArgumentParser(
prog="python -m portfolio_optimiser.evals.v1_gate", prog="python -m portfolio_optimiser.evals.v1_gate",
@ -730,6 +1006,14 @@ def main(argv: Sequence[str] | None = None) -> int:
if args.rounds_dir is not None and not Path(args.rounds_dir).is_dir(): if args.rounds_dir is not None and not Path(args.rounds_dir).is_dir():
parser.error(f"--rounds-dir {args.rounds_dir} finnes ikke") parser.error(f"--rounds-dir {args.rounds_dir} finnes ikke")
if args.rounds_dir is not None and not _safe_rounds_dir(Path(args.rounds_dir)):
parser.error(
f"--rounds-dir {args.rounds_dir} ligger i repoet uten å være gitignored — "
"fagpersonens tilbakemelding kunne da bli committet til den offentlige remoten"
)
for flag, value in (("--stress-root", args.stress_root), ("--bundle-root", args.bundle_root)):
if value is not None and not Path(value).expanduser().is_dir():
parser.error(f"{flag} {value} finnes ikke")
rounds_dir = Path(args.rounds_dir) if args.rounds_dir else _REPO_ROOT / DEFAULT_ROUNDS_DIR rounds_dir = Path(args.rounds_dir) if args.rounds_dir else _REPO_ROOT / DEFAULT_ROUNDS_DIR
rows = evaluate( rows = evaluate(
rounds_dir=rounds_dir, rounds_dir=rounds_dir,
@ -741,7 +1025,9 @@ def main(argv: Sequence[str] | None = None) -> int:
if args.json: if args.json:
print( print(
json.dumps( json.dumps(
{"exit": code, "rows": [asdict(r) for r in rows]}, ensure_ascii=False, indent=2 {"exit": code, "attestation": ATTESTATION, "rows": [asdict(r) for r in rows]},
ensure_ascii=False,
indent=2,
) )
) )
else: else:

View file

@ -1,15 +1,20 @@
"""The v1 gate's own tests: every row CAN go green and CAN go red. """The v1 gate's own tests: every row CAN go green and CAN go red — and cannot be FAKED green.
A gate that can only be red is as worthless as one that can only be green, so each row is driven A gate that can only be red is as worthless as one that can only be green, so each row is driven
from fixtures on both sides of its line. The probes and the stress measurement are injected here from fixtures on both sides of its line. An independent review (17.09) then showed three rows
(``probe_runner`` / ``stress_measure``) so the logic is exercised without a child pytest; one could be made green from a handwritten directory in a minute, and ten of twenty mutants survived
subprocess arm runs the real command end to end. this file; the arms marked M-1 M-5 and m-1 are the answer, each named for the finding it pins.
The probes and the stress measurement are injected here (``probe_runner`` / ``stress_measure``) so
the logic is exercised without a child pytest; ``run_probes`` gets its own arm against a throwaway
test file, and one subprocess arm runs the real command end to end.
""" """
from __future__ import annotations from __future__ import annotations
import ast import ast
import json import json
import os
import subprocess import subprocess
import sys import sys
from pathlib import Path from pathlib import Path
@ -18,6 +23,7 @@ from typing import Any
import pytest import pytest
from portfolio_optimiser.evals import v1_gate as gate from portfolio_optimiser.evals import v1_gate as gate
from portfolio_optimiser.validator import UNSUPPORTED_REASON
_REPO = Path(__file__).resolve().parents[1] _REPO = Path(__file__).resolve().parents[1]
_CONFIG = gate.load_config() _CONFIG = gate.load_config()
@ -25,6 +31,15 @@ _AI = gate.ai_authored_lines(_REPO, _CONFIG["ai_authored"])
_ALL_NODEIDS = [n for spec in _CONFIG["feedback_types"].values() for n in spec["evidence"]] + list( _ALL_NODEIDS = [n for spec in _CONFIG["feedback_types"].values() for n in spec["evidence"]] + list(
_CONFIG["row6_evidence"] _CONFIG["row6_evidence"]
) )
_T0 = 1_780_000_000 # a fixed epoch: every fixture run and feedback is ordered against it
#: The reason text a rejection at each stage carries, so a fixture run's coverage produces the
#: stage ``validator.rejection_stage`` would read off a real run.
_DETAIL = {
"stage0-baseline": "unknown cost code 'X': not in project P's cost baseline (1 known codes)",
"stage4-p90": "claimed 9 exceeds P90 feasible 1",
"unsupported": UNSUPPORTED_REASON,
}
def _write(path: Path, payload: Any) -> None: def _write(path: Path, payload: Any) -> None:
@ -33,17 +48,28 @@ def _write(path: Path, payload: Any) -> None:
path.write_text(text, encoding="utf-8") path.write_text(text, encoding="utf-8")
def _feedback(round_dir: Path, *items: tuple[str, int, str], author: str = "fagperson") -> None: def _iso(offset: int) -> str:
from datetime import datetime, timezone
return datetime.fromtimestamp(_T0 + offset, tz=timezone.utc).isoformat()
def _feedback(
round_dir: Path,
*items: tuple[str, int, str],
author: str = "fagperson",
at: int | None = None,
**extra: Any,
) -> None:
offset = (int(round_dir.name) * 20 - 10) if at is None else at
_write( _write(
round_dir / "feedback.json", round_dir / "feedback.json",
{"author": author, "items": [{"id": i, "type": t, "text": x} for i, t, x in items]}, {
) "author": author,
"given_at": _iso(offset),
"items": [{"id": i, "type": t, "text": x} for i, t, x in items],
def _outcome(round_dir: Path, rows: list[dict[str, Any]], removed: Any = ()) -> None: **extra,
_write( },
round_dir / "outcome.json",
{"run_id": f"r{round_dir.name}", "approaches": rows, "removed": list(removed)},
) )
@ -57,14 +83,57 @@ def _row(aid: str, validated: bool, nok: float | None, *ids: str, stage: str = "
} }
def _coverage_row(row: dict[str, Any]) -> dict[str, Any]:
if row["validated"]:
return {
"id": row["id"],
"status": "validated",
"detail": "",
"saving_nok": row["validated_nok"],
}
if row["stage"] == "not_evaluated":
return {"id": row["id"], "status": "not_evaluated", "detail": "budget", "saving_nok": None}
status = "unsupported" if row["stage"] == "unsupported" else "rejected"
return {"id": row["id"], "status": status, "detail": _DETAIL[row["stage"]], "saving_nok": None}
def _outcome(
round_dir: Path,
rows: list[dict[str, Any]],
removed: Any = (),
*,
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)."""
run_id = f"r{round_dir.name}"
outbox = round_dir.parent / "runs" / run_id
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(
round_dir / "outcome.json",
{
"run_id": run_id,
"outbox": f"../runs/{run_id}",
"approaches": rows,
"removed": list(removed),
},
)
def _green_rounds(root: Path) -> Path: def _green_rounds(root: Path) -> Path:
"""Three traced rounds and a round 3 report kept at 100 %.""" """Three traced rounds, each run after its feedback, and a round 3 report the expert kept
_outcome(root / "0", [_row("a1", False, None, stage="stage0")]) whole while adding a line of their own."""
_outcome(root / "0", [_row("a1", False, None, stage="stage0-baseline")])
_write(root / "0" / "report.md", "# Rapport 0\n\nlinje\n")
for n in (1, 2, 3): for n in (1, 2, 3):
_feedback(root / str(n), (f"f{n}", 1, f"Tallet for linje {n} er feil, bruk kontrakten.")) _feedback(root / str(n), (f"f{n}", 1, f"Tallet for linje {n} er feil, bruk kontrakten."))
_outcome(root / str(n), [_row("a1", True, 1000.0 * n, f"f{n}")]) _outcome(root / str(n), [_row("a1", True, 1000.0 * n, f"f{n}")])
_write(root / str(n) / "report.md", f"# Rapport {n}\n\nlinje\n") _write(root / str(n) / "report.md", f"# Rapport {n}\n\nlinje\n")
_write(root / "3" / "report.kept.md", "# Rapport 3\n\nlinje\n") _write(root / "3" / "report.kept.md", "# Rapport 3\n\nlinje\nmin egen merknad\n")
return root return root
@ -77,6 +146,87 @@ _CLEAN = gate.StressMeasure(
) )
# ---------------------------------------------------------------------------------------------
# M-5 — the contract is pinned to its SOURCE, not to whatever the data file says today
# ---------------------------------------------------------------------------------------------
#: Operator's choice of the v1 destination, 16.09: three feedback rounds with a real expert.
_ROUNDS_REQUIRED = 3
#: Operator's choice, 16.09: the expert keeps at least 80 % of the round 3 report.
_KEEP_THRESHOLD = 0.8
#: The eight feedback types of the v1 outcome basis (item 9 is a note, not a type).
_TYPES = {"1", "2", "3", "4", "5", "6", "7", "8"}
#: Operator-approved 17.09: M = 8 U-IDs and the types each points at.
_M_LIST = [
("U13", [1, 2]),
("U9", [1, 8]),
("U12", [1]),
("U4", [3]),
("U7", [4]),
("U11", [5]),
("U5", [6]),
("U6", [7]),
]
#: The evidence each type is judged on. Changing it moves a type's verdict, so it is a contract
#: change: rewriting a probe goes through the PM checkpoint, and this list is where that shows.
_EVIDENCE = {
"1": [
"tests/test_proposal_review_loop_loadbearing.py::"
"test_t13_the_flag_answers_the_review_from_a_real_argv_and_the_answer_is_used"
],
"2": ["tests/test_v1_probes.py::test_type_2_remove_a_direction_has_a_typed_door"],
"3": [
"tests/test_v1_probes.py::test_type_3_a_commissioned_angle_is_evaluated_through_the_cli",
"tests/test_v1_probes.py::test_type_3_a_new_angle_changes_the_outcome",
],
"4": ["tests/test_v1_probes.py::test_type_4_relax_a_requirement_has_a_door"],
"5": ["tests/test_v1_probes.py::test_type_5_edit_the_concept_graph_has_a_door"],
"6": ["tests/test_v1_probes.py::test_type_6_skills_per_analysis_has_a_door"],
"7": [
"tests/test_v1_probes.py::"
"test_type_7_the_mcp_flag_puts_a_service_the_run_calls_into_the_result",
"tests/test_b4_mcp_call_trace_loadbearing.py::test_a_called_mcp_tool_is_recorded_in_provenance",
],
"8": ["tests/test_v1_probes.py::test_type_8_inline_context_has_a_door"],
}
def test_m5_the_contract_numbers_match_their_source() -> None:
assert _CONFIG["rounds_required"] == _ROUNDS_REQUIRED
assert _CONFIG["keep_threshold"] == _KEEP_THRESHOLD
assert set(_CONFIG["feedback_types"]) == _TYPES
assert _CONFIG["ai_authored"] == ["docs/ekspert-svar.md"]
maf = _CONFIG["maf_points"]
assert [(p["u_id"], p["types"]) for p in maf["points"]] == _M_LIST
assert (maf["approved"], maf["approved_on"], maf["approved_by"]) == (
True,
"2026-09-17",
"operatørgodkjent",
)
def test_m5_the_evidence_register_is_pinned() -> None:
assert {k: v["evidence"] for k, v in _CONFIG["feedback_types"].items()} == _EVIDENCE
assert _CONFIG["row6_evidence"] == [
"tests/test_v1_probes.py::test_row6_an_approach_that_declared_nothing_cannot_be_validated",
"tests/test_v1_probes.py::test_row6_a_run_level_declaration_does_not_stand_in_for_the_approach",
]
def test_m5_evaluate_uses_the_configured_numbers(tmp_path: Path) -> None:
"""The pins above would be decoration if ``evaluate`` hard-coded its own numbers."""
config = json.loads(json.dumps(_CONFIG))
config["rounds_required"] = 1
rows = gate.evaluate(
rounds_dir=tmp_path,
config=config,
repo_root=_REPO,
probe_runner=_all_pass,
stress_measure=_CLEAN,
)
assert (rows[0].n, rows[1].n) == (1, 1)
# --------------------------------------------------------------------------------------------- # ---------------------------------------------------------------------------------------------
# Row 1 — rounds with a real domain expert # Row 1 — rounds with a real domain expert
# --------------------------------------------------------------------------------------------- # ---------------------------------------------------------------------------------------------
@ -89,25 +239,35 @@ def test_row1_is_red_with_no_rounds_and_green_with_three(tmp_path: Path) -> None
assert (row.k, row.status) == (3, gate.GREEN) assert (row.k, row.status) == (3, gate.GREEN)
def test_row1_one_round_is_not_three(tmp_path: Path) -> None:
root = _green_rounds(tmp_path)
(root / "2" / "feedback.json").unlink()
(root / "3" / "feedback.json").unlink()
row = gate.score_rounds(root, 3, _AI)
assert (row.k, row.status) == (1, gate.RED)
def test_row1_an_empty_or_wrongly_shaped_feedback_does_not_count(tmp_path: Path) -> None: def test_row1_an_empty_or_wrongly_shaped_feedback_does_not_count(tmp_path: Path) -> None:
root = _green_rounds(tmp_path) root = _green_rounds(tmp_path)
_write(root / "1" / "feedback.json", {"author": "fagperson", "items": []}) _write(root / "1" / "feedback.json", {"author": "x", "given_at": _iso(10), "items": []})
(root / "2" / "feedback.json").unlink() (root / "2" / "feedback.json").unlink()
_write(root / "2" / "feedback.md", "Dette er min tilbakemelding.") _write(root / "2" / "feedback.md", "Dette er min tilbakemelding.")
_feedback(root / "3", ("f3", 1, "noe"), author="") _feedback(root / "3", ("f3", 1, "noe"), author="")
row = gate.score_rounds(root, 3, _AI) row = gate.score_rounds(root, 3, _AI)
assert row.k == 0 assert row.k == 0
assert any("feedback.md" in x for x in row.exceptions) assert any("feedback.md" in x for x in row.exceptions)
assert any("navngir ingen fagperson" in x for x in row.exceptions)
def test_row1_the_ai_authored_answer_sheet_can_never_be_counted_in(tmp_path: Path) -> None: def test_row1_the_ai_authored_answer_sheet_can_never_be_counted_in(tmp_path: Path) -> None:
"""``docs/ekspert-svar.md`` is AI-authored: text lifted from it is refused, and a control with """``docs/ekspert-svar.md`` is AI-authored: text lifted from it is refused — also with its
the expert's own words in the same shape IS counted.""" case changed (M-1: one letter used to get it through) and the expert's own words count."""
doc = (_REPO / "docs" / "ekspert-svar.md").read_text(encoding="utf-8") doc = (_REPO / "docs" / "ekspert-svar.md").read_text(encoding="utf-8")
lifted = next(line for line in doc.splitlines() if "Skal en dom telle som fagdom" in line) lifted = next(line for line in doc.splitlines() if "Skal en dom telle som fagdom" in line)
lifted = lifted.lstrip("> ")
root = _green_rounds(tmp_path) root = _green_rounds(tmp_path)
_feedback(root / "1", ("f1", 1, lifted.lstrip("> "))) _feedback(root / "1", ("f1", 1, lifted))
_feedback(root / "2", ("f2", 1, "Se her: " + lifted.lstrip("> ") + " Takk.")) _feedback(root / "2", ("f2", 1, "Se her: " + lifted.swapcase() + " Takk."))
row = gate.score_rounds(root, 3, _AI) row = gate.score_rounds(root, 3, _AI)
assert row.k == 1 assert row.k == 1
assert sum("AI-forfattet" in x for x in row.exceptions) == 2 assert sum("AI-forfattet" in x for x in row.exceptions) == 2
@ -115,6 +275,46 @@ def test_row1_the_ai_authored_answer_sheet_can_never_be_counted_in(tmp_path: Pat
assert gate.score_rounds(root, 3, None).k == 0 assert gate.score_rounds(root, 3, None).k == 0
def test_m1_the_same_feedback_copied_into_every_round_counts_once(tmp_path: Path) -> None:
root = _green_rounds(tmp_path)
for n in (1, 2, 3):
_feedback(root / str(n), (f"f{n}", 1, "ok"))
row = gate.score_rounds(root, 3, _AI)
assert row.k == 1
assert sum("ingen punkt som ikke alt er gitt" in x for x in row.exceptions) == 2
def test_m1_ids_are_per_round(tmp_path: Path) -> None:
root = _green_rounds(tmp_path)
_feedback(root / "2", ("f1", 1, "Et helt nytt punkt i runde to."))
assert "gjenbrukt" in " ".join(gate.score_rounds(root, 3, _AI).exceptions)
def test_m1_feedback_is_timestamped_and_ordered(tmp_path: Path) -> None:
root = _green_rounds(tmp_path)
_write(
root / "1" / "feedback.json",
{
"author": "x",
"given_at": "2026-09-17T10:00:00",
"items": [{"id": "f1", "type": 1, "text": "t"}],
},
)
_feedback(root / "3", ("f3", 1, "Et nytt punkt, men gitt for tidlig."), at=5)
row = gate.score_rounds(root, 3, _AI)
assert row.k == 1
assert any("tidssone" in x for x in row.exceptions)
assert any("ikke etter forrige" in x for x in row.exceptions)
def test_m1_feedback_is_given_on_a_report(tmp_path: Path) -> None:
root = _green_rounds(tmp_path)
(root / "0" / "report.md").unlink()
row = gate.score_rounds(root, 3, _AI)
assert row.k == 2
assert any("0/report.md" in x for x in row.exceptions)
# --------------------------------------------------------------------------------------------- # ---------------------------------------------------------------------------------------------
# Row 2 — rounds with a measurable change # Row 2 — rounds with a measurable change
# --------------------------------------------------------------------------------------------- # ---------------------------------------------------------------------------------------------
@ -122,53 +322,90 @@ def test_row1_the_ai_authored_answer_sheet_can_never_be_counted_in(tmp_path: Pat
def test_row2_a_traced_change_counts(tmp_path: Path) -> None: def test_row2_a_traced_change_counts(tmp_path: Path) -> None:
row = gate.score_changes(_green_rounds(tmp_path), 3, _AI) row = gate.score_changes(_green_rounds(tmp_path), 3, _AI)
assert (row.k, row.status) == (3, gate.GREEN) assert (row.k, row.status) == (3, gate.GREEN), row.exceptions
assert "runde 0 =" in row.reason and "(r0)" in row.reason assert "runde 0 =" in row.reason and "(kjøring r0)" in row.reason
def test_row2_a_change_without_a_trace_is_model_noise(tmp_path: Path) -> None: def test_row2_a_change_without_a_trace_is_model_noise(tmp_path: Path) -> None:
root = _green_rounds(tmp_path) root = _green_rounds(tmp_path)
_outcome(root / "2", [_row("a1", True, 2000.0)]) # changed nok, no feedback id _outcome(root / "2", [_row("a1", True, 5000.0)]) # changed nok, no feedback id
_outcome(root / "3", [_row("a1", True, 3000.0, "f1")]) # traced to an EARLIER round's id _outcome(root / "3", [_row("a1", True, 9000.0, "f1")]) # traced to an EARLIER round's id
row = gate.score_changes(root, 3, _AI) row = gate.score_changes(root, 3, _AI)
assert (row.k, row.status) == (1, gate.RED) assert (row.k, row.status) == (1, gate.RED)
assert sum("ingen sporet" in x for x in row.exceptions) == 2 assert sum("ingen sporet" in x for x in row.exceptions) == 2
def test_row2_no_change_does_not_count(tmp_path: Path) -> None: def test_row2_no_change_does_not_count_and_is_measured_against_the_previous_round(
tmp_path: Path,
) -> None:
root = _green_rounds(tmp_path) root = _green_rounds(tmp_path)
_outcome(root / "2", [_row("a1", True, 1000.0, "f2")]) # identical to round 1 _outcome(root / "2", [_row("a1", True, 1000.0, "f2")]) # identical to round 1, not to round 0
row = gate.score_changes(root, 3, _AI) row = gate.score_changes(root, 3, _AI)
assert row.k == 2 assert row.k == 2
assert any("ingen endring" in x for x in row.exceptions) assert any("ingen endring" in x for x in row.exceptions)
def test_row2_each_of_a_to_d_is_a_change(tmp_path: Path) -> None: def test_m1_a_change_below_one_percent_is_noise(tmp_path: Path) -> None:
root = _green_rounds(tmp_path)
_outcome(root / "2", [_row("a1", True, 1009.99, "f2")])
assert gate.score_changes(root, 3, _AI).k == 2
_outcome(root / "2", [_row("a1", True, 1010.0, "f2")])
assert gate.score_changes(root, 3, _AI).k == 3
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.unlink()
assert "finnes ikke" in " ".join(gate.score_changes(root, 3, _AI).exceptions)
_outcome(
root / "1",
[_row("a1", True, 1000.0, "f1")],
coverage=[{"id": "a1", "status": "rejected", "detail": _DETAIL["stage4-p90"]}],
)
assert "stemmer ikke" in " ".join(gate.score_changes(root, 3, _AI).exceptions)
def test_m1_an_empty_baseline_run_is_refused(tmp_path: Path) -> None:
root = _green_rounds(tmp_path)
_outcome(root / "0", [], coverage=[])
row = gate.score_changes(root, 3, _AI)
assert "evaluerte ingen" in row.reason
assert row.k == 2
def test_m1_the_feedback_must_come_between_the_two_runs(tmp_path: Path) -> None:
root = _green_rounds(tmp_path)
_outcome(root / "1", [_row("a1", True, 1000.0, "f1")], at=5) # ran before its feedback (10)
row = gate.score_changes(root, 3, _AI)
assert row.k < 3
assert any("ikke gitt mellom kjøring" in x for x in row.exceptions)
def _outcome_obj(rows: list[dict[str, Any]], removed: dict[str, set[str]] | None = None) -> Any:
from datetime import datetime, timezone
return gate.Outcome(
{r["id"]: r for r in rows}, removed or {}, "r", datetime.now(tz=timezone.utc)
)
def test_row2_each_of_a_to_d_is_a_change() -> None:
ids = {"f"} ids = {"f"}
base = [_row("a1", False, None, stage="stage0")] base = [_row("a1", False, None, stage="stage0-baseline")]
cases = { cases = {
"a-added": [*base, _row("a2", False, None, "f", stage="stage0")], "a-added": [*base, _row("a2", False, None, "f", stage="stage0-baseline")],
"b-validated": [_row("a1", True, None, "f", stage="stage0")], "b-validated": [_row("a1", True, 7.0, "f")],
"c-stage": [_row("a1", False, None, "f", stage="stage4")], "c-stage": [_row("a1", False, None, "f", stage="stage4-p90")],
"d-nok": [_row("a1", False, 5.0, "f", stage="stage0")], "d-nok": [_row("a1", False, 5.0, "f", stage="stage0-baseline")],
} }
for name, rows in cases.items(): for name, rows in cases.items():
_outcome(tmp_path / name / "0", base) ok, why = gate.outcomes_changed(_outcome_obj(base), _outcome_obj(rows), ids)
_outcome(tmp_path / name / "1", rows)
ok, why = gate.round_changed(
tmp_path / name / "0" / "outcome.json", tmp_path / name / "1" / "outcome.json", ids
)
assert ok, (name, why) assert ok, (name, why)
# (a) by removal: traced only through the ``removed`` list. before = _outcome_obj([*base, _row("a2", False, None)])
_outcome(tmp_path / "rm" / "0", [*base, _row("a2", False, None)]) assert not gate.outcomes_changed(before, _outcome_obj(base), ids)[0]
_outcome(tmp_path / "rm" / "1", base) removed = _outcome_obj(base, {"a2": {"f"}})
assert not gate.round_changed( assert gate.outcomes_changed(before, removed, ids)[0]
tmp_path / "rm" / "0" / "outcome.json", tmp_path / "rm" / "1" / "outcome.json", ids
)[0]
_outcome(tmp_path / "rm" / "1", base, removed=[{"id": "a2", "feedback_ids": ["f"]}])
assert gate.round_changed(
tmp_path / "rm" / "0" / "outcome.json", tmp_path / "rm" / "1" / "outcome.json", ids
)[0]
# --------------------------------------------------------------------------------------------- # ---------------------------------------------------------------------------------------------
@ -201,31 +438,106 @@ def test_row3_every_registered_test_exists() -> None:
assert name in names, nodeid assert name in names, nodeid
def test_m1_run_probes_reads_every_outcome_honestly(tmp_path: Path) -> None:
"""m-1 (M06-M08, M16): an erroring or skipped test is not a pass, and a known gap marked
``xfail(strict=True)`` is run as the failure it is."""
_write(
tmp_path / "tests" / "test_probe.py",
"import pytest\n\n"
"@pytest.fixture\n"
"def broken():\n raise RuntimeError('fixture')\n\n"
"def test_pass():\n pass\n\n"
"def test_fail():\n assert False\n\n"
"def test_error(broken):\n pass\n\n"
"def test_skip():\n pytest.skip('no')\n\n"
"@pytest.mark.xfail(strict=True)\n"
"def test_gap():\n assert False\n",
)
names = ["pass", "fail", "error", "skip", "gap", "gone"]
ids = [f"tests/test_probe.py::test_{n}" for n in names]
outcomes = gate.run_probes(ids, repo_root=tmp_path)
assert [outcomes[i] for i in ids] == [
"passed",
"failed",
"failed",
"skipped",
"failed",
"missing",
]
# --------------------------------------------------------------------------------------------- # ---------------------------------------------------------------------------------------------
# Row 4 — round 3 kept # Row 4 — round 3 kept
# --------------------------------------------------------------------------------------------- # ---------------------------------------------------------------------------------------------
def _report(root: Path, report: str, kept: str | None) -> None:
_write(root / "3" / "report.md", report)
if kept is not None:
_write(root / "3" / "report.kept.md", kept)
def _kept_row(root: Path, report: str, kept: str) -> gate.Row:
_report(root, report, kept)
return gate.score_kept(root, 0.8, _AI)
@pytest.mark.parametrize(("kept_lines", "status"), [(79, gate.RED), (80, gate.GREEN)]) @pytest.mark.parametrize(("kept_lines", "status"), [(79, gate.RED), (80, gate.GREEN)])
def test_row4_the_line_is_eighty_percent(tmp_path: Path, kept_lines: int, status: str) -> None: def test_row4_the_line_is_eighty_percent(tmp_path: Path, kept_lines: int, status: str) -> None:
lines = [f"linje {i}" for i in range(100)] lines = [f"linje {i}" for i in range(100)]
_write(tmp_path / "3" / "report.md", "\n\n".join(lines) + "\n")
kept = lines[:kept_lines] + [f"endret {i}" for i in range(100 - kept_lines)] kept = lines[:kept_lines] + [f"endret {i}" for i in range(100 - kept_lines)]
_write(tmp_path / "3" / "report.kept.md", "\n".join(kept) + "\n") _report(tmp_path, "\n\n".join(lines) + "\n", "\n".join(kept) + "\n")
row = gate.score_kept(tmp_path, 0.8) row = gate.score_kept(tmp_path, 0.8, _AI)
assert (row.k, row.n, row.status) == (kept_lines, 100, status) assert (row.k, row.n, row.status) == (kept_lines, 100, status)
def test_row4_a_missing_kept_report_is_red_never_full(tmp_path: Path) -> None: def test_row4_a_missing_kept_report_is_red_never_full(tmp_path: Path) -> None:
_write(tmp_path / "3" / "report.md", "a\nb\n") _report(tmp_path, "a\nb\n", None)
row = gate.score_kept(tmp_path, 0.8) row = gate.score_kept(tmp_path, 0.8, _AI)
assert (row.k, row.status, row.reason) == (None, gate.RED, "ingen rapport") assert (row.k, row.status, row.reason) == (None, gate.RED, "ingen rapport")
def test_row4_a_line_kept_once_counts_once(tmp_path: Path) -> None: def test_row4_a_line_kept_once_counts_once(tmp_path: Path) -> None:
_write(tmp_path / "3" / "report.md", "x\nx\ny\n") _report(tmp_path, "x\nx\ny\n", "x\nz\n")
_write(tmp_path / "3" / "report.kept.md", "x\nz\n") assert gate.score_kept(tmp_path, 0.8, _AI).k == 1
assert gate.score_kept(tmp_path, 0.8).k == 1
def test_m2_separators_do_not_make_a_rewritten_report_kept(tmp_path: Path) -> None:
rules = "---\n| --- | --- |\n&nbsp;\n" * 34
report = rules + "".join(f"innhold {i}\n" for i in range(5))
kept = rules + "".join(f"omskrevet {i}\n" for i in range(5))
row = _kept_row(tmp_path, report, kept)
assert (row.k, row.n, row.status) == (0, 5, gate.RED)
def test_m2_trailing_whitespace_is_an_editor_not_an_edit(tmp_path: Path) -> None:
report = "".join(f"linje {i} \n" for i in range(10))
row = _kept_row(tmp_path, report, report.replace(" \n", "\n"))
assert (row.k, row.n, row.status) == (10, 10, gate.GREEN)
def test_m2_a_reshuffle_is_a_change(tmp_path: Path) -> None:
lines = [f"linje {i}" for i in range(10)]
row = _kept_row(tmp_path, "\n".join(lines), "\n".join(reversed(lines)))
assert row.k == 1 and row.status == gate.RED
def test_m2_additions_are_their_own_number(tmp_path: Path) -> None:
report = "".join(f"linje {i}\n" for i in range(10))
kept = report + "".join(f"innvending {i}\n" for i in range(200))
row = _kept_row(tmp_path, report, kept)
assert (row.k, row.status) == (10, gate.GREEN)
assert "200 linje(r) lagt til" in row.reason
def test_m2_an_untouched_copy_needs_a_receipt(tmp_path: Path) -> None:
report = "".join(f"linje {i}\n" for i in range(10))
row = _kept_row(tmp_path, report, report)
assert (row.k, row.status) == (None, gate.RED)
assert "ikke rørt" in row.reason
_feedback(tmp_path / "3", ("f3", 1, "Rapporten kan stå som den er."), report_unchanged=True)
row = gate.score_kept(tmp_path, 0.8, _AI)
assert (row.k, row.status) == (10, gate.GREEN)
# --------------------------------------------------------------------------------------------- # ---------------------------------------------------------------------------------------------
@ -242,24 +554,7 @@ def test_row5_is_red_until_the_operator_approves_the_list() -> None:
def test_row5_the_approved_list_counts_only_points_whose_types_are_green() -> None: def test_row5_the_approved_list_counts_only_points_whose_types_are_green() -> None:
"""The tracked list is operator-approved (17.09): eight U-IDs, and a point counts only when
every type it points at is green. With today's green types (1, 3, 7) that is U12, U4, U6."""
maf = _CONFIG["maf_points"] maf = _CONFIG["maf_points"]
assert (maf["approved"], maf["approved_on"], maf["approved_by"]) == (
True,
"2026-09-17",
"operatørgodkjent",
)
assert [(p["u_id"], p["types"]) for p in maf["points"]] == [
("U13", [1, 2]),
("U9", [1, 8]),
("U12", [1]),
("U4", [3]),
("U7", [4]),
("U11", [5]),
("U5", [6]),
("U6", [7]),
]
today = { today = {
n: ("passed" if int(t) in (1, 3, 7) else "failed") n: ("passed" if int(t) in (1, 3, 7) else "failed")
for t, spec in _CONFIG["feedback_types"].items() for t, spec in _CONFIG["feedback_types"].items()
@ -270,40 +565,54 @@ def test_row5_the_approved_list_counts_only_points_whose_types_are_green() -> No
assert all(p not in " ".join(row.exceptions) for p in ("U12 ", "U4 ", "U6 ")) assert all(p not in " ".join(row.exceptions) for p in ("U12 ", "U4 ", "U6 "))
def _synthetic_src(tmp: Path, *, comment_only: bool = False) -> Path: def _synthetic_src(tmp: Path, body: str | None = None) -> Path:
body = ( body = "def build():\n return SkillsProvider()\n" if body is None else body
"# uses SkillsProvider\n" if comment_only else "def build():\n return SkillsProvider()\n"
)
_write(tmp / "skills.py", "from agent_framework import SkillsProvider\n\n" + body) _write(tmp / "skills.py", "from agent_framework import SkillsProvider\n\n" + body)
return tmp return tmp
def _one_point(types: list[int]) -> dict[str, Any]: def _one_point(types: list[int], scope: str = "build") -> dict[str, Any]:
point = { point = {
"u_id": "U5", "u_id": "U5",
"construct": "SkillsProvider", "construct": "SkillsProvider",
"package": "agent_framework", "package": "agent_framework",
"callsite": {"module": "skills.py", "scope": "build"}, "callsite": {"module": "skills.py", "scope": scope},
"types": types, "types": types,
} }
return {"approved": True, "points": [point]} return {"approved": True, "points": [point]}
def test_row5_an_approved_point_counts_only_with_a_green_type(tmp_path: Path) -> None: def test_row5_an_approved_point_counts_only_when_every_type_it_points_at_is_green(
tmp_path: Path,
) -> None:
src = _synthetic_src(tmp_path) src = _synthetic_src(tmp_path)
green = gate.score_maf(_one_point([6]), {6: ""}, src) green = gate.score_maf(_one_point([6]), {6: ""}, src)
assert (green.k, green.status) == (1, gate.GREEN) assert (green.k, green.status) == (1, gate.GREEN)
red = gate.score_maf(_one_point([6]), {6: "failed"}, src) assert gate.score_maf(_one_point([6]), {6: "failed"}, src).k == 0
assert (red.k, red.status) == (0, gate.RED) assert gate.score_maf(_one_point([6, 7]), {6: "", 7: "failed"}, src).k == 0
def test_row5_a_comment_is_not_a_call_site(tmp_path: Path) -> None: def test_row5_a_comment_is_not_a_call_site(tmp_path: Path) -> None:
src = _synthetic_src(tmp_path, comment_only=True) src = _synthetic_src(tmp_path, "# uses SkillsProvider\n")
row = gate.score_maf(_one_point([6]), {6: ""}, src) row = gate.score_maf(_one_point([6]), {6: ""}, src)
assert row.k == 0 assert row.k == 0
assert "presence 0 av 1" in row.diagnostics assert "presence 0 av 1" in row.diagnostics
def test_row5_the_named_scope_is_required(tmp_path: Path) -> None:
src = _synthetic_src(tmp_path, "def other():\n return SkillsProvider()\n")
row = gate.score_maf(_one_point([6]), {6: ""}, src)
assert row.k == 0
assert "kallsted verifisert 0 av 1" in row.diagnostics
def test_row5_a_type_annotation_is_not_a_use(tmp_path: Path) -> None:
src = _synthetic_src(
tmp_path, "def build(p: SkillsProvider) -> SkillsProvider:\n return p\n"
)
assert gate.maf_presence(_one_point([6])["points"][0], src) == (False, False)
def test_row5_real_call_sites_are_found_by_ast() -> None: def test_row5_real_call_sites_are_found_by_ast() -> None:
found = { found = {
p["u_id"]: gate.maf_presence(p, gate._PACKAGE_SRC) for p in _CONFIG["maf_points"]["points"] p["u_id"]: gate.maf_presence(p, gate._PACKAGE_SRC) for p in _CONFIG["maf_points"]["points"]
@ -326,6 +635,7 @@ def test_row6_green_needs_both_the_probes_and_zero_undeclared() -> None:
assert gate.score_undeclared(_PROBES, _all_pass(_PROBES), dirty, "s").status == gate.RED assert gate.score_undeclared(_PROBES, _all_pass(_PROBES), dirty, "s").status == gate.RED
failing = {**_all_pass(_PROBES), _PROBES[1]: "failed"} failing = {**_all_pass(_PROBES), _PROBES[1]: "failed"}
assert gate.score_undeclared(_PROBES, failing, _CLEAN, "s").status == gate.RED assert gate.score_undeclared(_PROBES, failing, _CLEAN, "s").status == gate.RED
assert gate.score_undeclared([], {}, _CLEAN, "s").status == gate.RED
def test_row6_missing_artefacts_are_never_zero_and_never_green() -> None: def test_row6_missing_artefacts_are_never_zero_and_never_green() -> None:
@ -348,29 +658,22 @@ def test_row6_artefacts_older_than_the_rule_are_not_measured() -> None:
assert "eldre enn regelen" in row.reason and "approach_id mangler" in row.reason assert "eldre enn regelen" in row.reason and "approach_id mangler" in row.reason
def test_m3_own_proposals_are_in_the_denominator(tmp_path: Path) -> None:
runs = {"runs": [{"outbox": "o", "run_id": "r1"}, {"outbox": "o", "run_id": "r2"}]}
_write(tmp_path / "o" / "r1-own-proposal-outcome.json", {"outcome_type": "validated"})
_write(tmp_path / "o" / "r2-own-proposal-outcome.json", {"outcome_type": "validated"})
_write(
tmp_path / "o" / "r2-debate.json",
{"requirements": [{"path": "p", "approach_id": "own-proposal"}]},
)
assert gate._own_proposals(runs, tmp_path) == (2, 1, ("own-proposal (r1)",))
def test_row7_not_measured_is_not_green_either() -> None: def test_row7_not_measured_is_not_green_either() -> None:
row = gate.score_named(gate.StressMeasure(missing="borte"), "s") row = gate.score_named(gate.StressMeasure(missing="borte"), "s")
assert (row.k, row.status, row.failing) == (None, gate.NOT_MEASURED, False) assert (row.k, row.status, row.failing) == (None, gate.NOT_MEASURED, False)
@pytest.mark.parametrize("missing", ["all", "outcome0", "kept"])
def test_rows_1_2_4_with_missing_files_are_red(tmp_path: Path, missing: str) -> None:
root = _green_rounds(tmp_path / "r")
if missing == "all":
root = tmp_path / "absent"
elif missing == "outcome0":
(root / "0" / "outcome.json").unlink()
else:
(root / "3" / "report.kept.md").unlink()
rows = [
gate.score_rounds(root, 3, _AI),
gate.score_changes(root, 3, _AI),
gate.score_kept(root, 0.8),
]
assert gate.exit_code(rows) == 1
assert gate.GREEN not in {r.status for r in rows} or missing != "all"
def test_row6_measures_the_stress_outboxes_when_they_exist(tmp_path: Path) -> None: def test_row6_measures_the_stress_outboxes_when_they_exist(tmp_path: Path) -> None:
"""Against the real artefacts when this machine has them; otherwise the absence is named.""" """Against the real artefacts when this machine has them; otherwise the absence is named."""
evidence = _CONFIG["stress_evidence"] evidence = _CONFIG["stress_evidence"]
@ -386,7 +689,14 @@ def test_row6_measures_the_stress_outboxes_when_they_exist(tmp_path: Path) -> No
# "ikke målt", which test_row6_missing_artefacts_are_never_zero already pins. # "ikke målt", which test_row6_missing_artefacts_are_never_zero already pins.
assert m.validated == 0 assert m.validated == 0
pytest.skip(f"stress artefacts not judgeable right now: {m.missing}") pytest.skip(f"stress artefacts not judgeable right now: {m.missing}")
assert (m.validated, m.undeclared, m.named, m.commissioned) == (10, 10, 1, 20) # 10 commissioned + 5 of the runs' own proposals (M-3).
assert (m.validated, m.undeclared, m.own_validated, m.named, m.commissioned) == (
15,
15,
5,
1,
20,
)
assert m.unaddressed > 0 # stress round 6 predates approach-addressed declarations assert m.unaddressed > 0 # stress round 6 predates approach-addressed declarations
row = gate.score_undeclared(_PROBES, _all_pass(_PROBES), m, "s") row = gate.score_undeclared(_PROBES, _all_pass(_PROBES), m, "s")
assert (row.k, row.status) == (None, gate.NOT_MEASURED) assert (row.k, row.status) == (None, gate.NOT_MEASURED)
@ -404,30 +714,37 @@ def test_row7_is_a_diagnosis_and_never_moves_the_exit_code() -> None:
# --------------------------------------------------------------------------------------------- # ---------------------------------------------------------------------------------------------
def test_every_failing_row_green_is_exit_zero_and_one_red_is_exit_one(tmp_path: Path) -> None: def _all_green(tmp_path: Path) -> tuple[list[gate.Row], dict[str, Any]]:
config = json.loads(json.dumps(_CONFIG)) config = json.loads(json.dumps(_CONFIG))
config["maf_points"]["approved"] = True
config["maf_points"]["points"] = _one_point([6])["points"] config["maf_points"]["points"] = _one_point([6])["points"]
rows = gate.evaluate( rows = gate.evaluate(
rounds_dir=_green_rounds(tmp_path), rounds_dir=_green_rounds(tmp_path / "rounds"),
config=config, config=config,
repo_root=_REPO, repo_root=_REPO,
src=_synthetic_src(tmp_path / "src"), src=_synthetic_src(tmp_path / "src"),
probe_runner=_all_pass, probe_runner=_all_pass,
stress_measure=_CLEAN, stress_measure=_CLEAN,
) )
return rows, config
def test_every_failing_row_green_is_exit_zero(tmp_path: Path) -> None:
rows, _ = _all_green(tmp_path)
assert [r.status for r in rows[:6]] == [gate.GREEN] * 6, gate.render(rows) assert [r.status for r in rows[:6]] == [gate.GREEN] * 6, gate.render(rows)
assert gate.exit_code(rows) == 0 assert gate.exit_code(rows) == 0
(tmp_path / "3" / "report.kept.md").unlink()
rows = gate.evaluate(
rounds_dir=tmp_path, @pytest.mark.parametrize("index", range(6))
config=config, def test_each_failing_row_alone_moves_the_exit_code(tmp_path: Path, index: int) -> None:
repo_root=_REPO, rows, _ = _all_green(tmp_path)
src=tmp_path / "src", red = list(rows)
probe_runner=_all_pass, red[index] = gate.Row(rows[index].key, rows[index].title, 0, 1, gate.RED, "r")
stress_measure=_CLEAN, assert gate.exit_code(red) == 1
)
assert gate.exit_code(rows) == 1
def test_the_attestation_is_printed(tmp_path: Path) -> None:
rows, _ = _all_green(tmp_path)
assert gate.ATTESTATION in gate.render(rows)
def test_the_default_rounds_dir_is_gitignored() -> None: def test_the_default_rounds_dir_is_gitignored() -> None:
@ -449,8 +766,13 @@ def _cli(*args: str) -> subprocess.CompletedProcess[str]:
def test_wrong_usage_is_exit_two(tmp_path: Path) -> None: def test_wrong_usage_is_exit_two(tmp_path: Path) -> None:
assert _cli("--rounds-dir", str(tmp_path / "missing")).returncode == 2 assert _cli("--rounds-dir", str(tmp_path / "missing")).returncode == 2
assert _cli("--no-such-flag").returncode == 2 assert _cli("--no-such-flag").returncode == 2
assert _cli("--stress-root", str(tmp_path / "missing")).returncode == 2
assert _cli("--bundle-root", str(tmp_path / "missing")).returncode == 2
# m-2: a rounds directory inside the repository that git would commit.
tracked = _cli("--rounds-dir", str(_REPO / "contexts"))
assert tracked.returncode == 2 and "gitignored" in tracked.stderr
help_text = _cli("--help").stdout help_text = _cli("--help").stdout
assert "report.kept.md" in help_text and "feedback_ids" in help_text assert "report.kept.md" in help_text and "given_at" in help_text and "outbox" in help_text
def test_the_command_is_red_today_with_every_row_in_its_output(tmp_path: Path) -> None: def test_the_command_is_red_today_with_every_row_in_its_output(tmp_path: Path) -> None:
@ -458,6 +780,7 @@ def test_the_command_is_red_today_with_every_row_in_its_output(tmp_path: Path) -
assert proc.returncode == 1, proc.stderr assert proc.returncode == 1, proc.stderr
payload = json.loads(proc.stdout) payload = json.loads(proc.stdout)
assert payload["exit"] == 1 assert payload["exit"] == 1
assert payload["attestation"] == gate.ATTESTATION
rows = {r["key"]: r for r in payload["rows"]} rows = {r["key"]: r for r in payload["rows"]}
assert list(rows) == ["rounds", "changes", "types", "kept", "maf", "undeclared", "named"] assert list(rows) == ["rounds", "changes", "types", "kept", "maf", "undeclared", "named"]
assert (rows["rounds"]["k"], rows["changes"]["k"]) == (0, 0) assert (rows["rounds"]["k"], rows["changes"]["k"]) == (0, 0)

View file

@ -31,7 +31,7 @@ from typing import Any
import pytest import pytest
from portfolio_optimiser import okf, run from portfolio_optimiser import okf, run
from portfolio_optimiser.mandate import Approach, Mandate from portfolio_optimiser.mandate import Approach, BindingRequirement, Mandate
from portfolio_optimiser.run import run_project from portfolio_optimiser.run import run_project
from portfolio_optimiser.simulation import scripted_factory from portfolio_optimiser.simulation import scripted_factory
from portfolio_optimiser.verdicts import VerdictStore from portfolio_optimiser.verdicts import VerdictStore
@ -67,6 +67,143 @@ def _surface_or_fail(type_no: int, what: str, keywords: tuple[str, ...]) -> None
) )
# ---------------------------------------------------------------------------------------------
# Row 3 — types 3 and 7: the way in is the REAL flag, and the action shows in the RESULT
# ---------------------------------------------------------------------------------------------
_REQUIREMENT = {"path": "tiltak-led-retrofit.md", "ref": "Krav 1"}
def _cli_run(tmp_path: Path, replies: dict[str, Any], *extra: str) -> tuple[int, Path, str]:
"""One in-process CLI run on the micro base with scripted replies, writing an outbox."""
replies_file = tmp_path / "replies.json"
replies_file.write_text(json.dumps(replies), encoding="utf-8")
out = tmp_path / "out"
buffer = io.StringIO()
with contextlib.redirect_stdout(buffer):
rc = run.main(
[
_PID,
"--bundle-dir",
str(_BUNDLE),
"--scripted-replies",
str(replies_file),
"--outbox-dir",
str(out),
"--run-id",
"probe",
*extra,
]
)
return rc, out, buffer.getvalue()
def test_type_3_a_commissioned_angle_is_evaluated_through_the_cli(tmp_path: Path) -> None:
"""``--mandate`` is the way in; the action is a VALIDATED outcome for that angle in the
outbox. A settlement line alone is not enough a ``NOT EVALUATED`` row prints the id too."""
mandate = tmp_path / "mandate.json"
mandate.write_text(
json.dumps(
{
"objective": "Kutt energikostnad",
"approaches": [
{"id": "ny-vinkling", "label": "LED-retrofit", "requirement": _REQUIREMENT}
],
"allow_own_proposals": False,
}
),
encoding="utf-8",
)
rc, out, stdout = _cli_run(
tmp_path, {"proposer": _VALID_REPLY, "checker": _CHECKER_REPLY}, "--mandate", str(mandate)
)
assert rc == 0, stdout
coverage = json.loads((out / "probe-coverage.json").read_text(encoding="utf-8"))
assert [(r["id"], r["status"]) for r in coverage["rows"]] == [("ny-vinkling", "validated")]
outcome = json.loads((out / "probe-ny-vinkling-outcome.json").read_text(encoding="utf-8"))
assert outcome["outcome_type"] == "validated"
@pytest.mark.asyncio
async def test_type_3_a_new_angle_changes_the_outcome() -> None:
"""Adding an angle in a later round changes what the run carries: the new angle's own reply
(reachable only if its label reached the prompt) becomes the selected, validated outcome."""
labels = {"Behovsstyrt belysning": 30_000, "Nattsenking av temperatur": 60_000}
def select(prompt: str, _role: str) -> str:
claimed = next((v for k, v in labels.items() if k in prompt), 10_000)
return _VALID_REPLY.replace("30000}", f"{claimed}}}")
requirement = BindingRequirement(**_REQUIREMENT)
first = Approach(id="a1", label="Behovsstyrt belysning", requirement=requirement)
second = Approach(id="a2", label="Nattsenking av temperatur", requirement=requirement)
async def outcome(*approaches: Approach) -> Any:
return await run_project(
_PID,
"local",
docs_dir=str(_BUNDLE),
bundle_dir=str(_BUNDLE),
store=VerdictStore(verdicts=[]),
client_factory=scripted_factory({"proposer": select, "checker": _CHECKER_REPLY}, []),
mandate=Mandate(objective="o", approaches=approaches, allow_own_proposals=False),
)
before = await outcome(first)
after = await outcome(first, second)
assert before.outcome.proposal.claimed_saving_nok == 30_000
assert {r.id: r.status for r in after.coverage} == {"a1": "validated", "a2": "validated"}
assert after.outcome.proposal.claimed_saving_nok == 60_000
def test_type_7_the_mcp_flag_puts_a_service_the_run_calls_into_the_result(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""``--mcp-config`` is the way in; the action is the RESULT recording that the debate called
the configured service. Without the flag the same script leaves no external call."""
from tests.test_b4_mcp_call_trace_loadbearing import _as_context_manager, _lookup_unit_price
monkeypatch.setattr(
run, "build_mcp_tools", lambda _c: [_as_context_manager(_lookup_unit_price)]
)
config = tmp_path / "mcp.json"
config.write_text(
json.dumps(
{
"servers": [
{
"name": "prisregister",
"transport": "http",
"url": "https://intern.example/mcp",
"allowed_tools": ["lookup_unit_price"],
"timeout_seconds": 15,
}
]
}
),
encoding="utf-8",
)
replies = {
"proposer": [
{"call": "lookup_unit_price", "args": {"code": "ENERGI-TOTAL-EL"}},
*([_VALID_REPLY] * 4),
],
"checker": _CHECKER_REPLY,
}
def calls(sub: str, *extra: str) -> list[Any]:
(tmp_path / sub).mkdir()
rc, out, stdout = _cli_run(tmp_path / sub, replies, *extra)
assert rc == 0, stdout
proposal = json.loads((out / "probe-proposal.json").read_text(encoding="utf-8"))
return list(proposal["provenance"]["external_calls"])
assert calls("with", "--mcp-config", str(config)) == [
{"server": "prisregister", "tool": "lookup_unit_price"}
]
assert calls("without") == []
# --------------------------------------------------------------------------------------------- # ---------------------------------------------------------------------------------------------
# Row 3 — the five types without a complete surface # Row 3 — the five types without a complete surface
# --------------------------------------------------------------------------------------------- # ---------------------------------------------------------------------------------------------