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:
parent
938a1ca30e
commit
9825b2677c
5 changed files with 979 additions and 213 deletions
17
CLAUDE.md
17
CLAUDE.md
|
|
@ -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 1–2 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
|
||||||
|
|
|
||||||
|
|
@ -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"
|
||||||
]
|
]
|
||||||
},
|
},
|
||||||
|
|
|
||||||
|
|
@ -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 1–2 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 på rapport n-1, så 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 på 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 på minst én av (a) settet av
|
En runde har målbar endring når kjøringen skiller seg fra forrige på 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 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.
|
||||||
|
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.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -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*)*\|?| )\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:
|
||||||
|
|
|
||||||
|
|
@ -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 \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)
|
||||||
|
|
|
||||||
|
|
@ -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
|
||||||
# ---------------------------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------------------------
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue