Kjell Tore Guttormsen
50c9763706
Measured 2026-09-17 17:43: vegnormal-okf rebuilt build/ferdig/r761-2025 while this repository's v1 gate, the stress judge and four corpus tests pointed straight at it. Rows 6-7 went IKKE MAALT and five tests fell, for a change no one here made. The failure mode was never falsehood - the gate says IKKE MAALT and exits non-zero, never green - it was instability: two projects shared a directory neither owns, so what this repository MEASURES could move without a commit here. A copy alone would push that directory one move away, so the copy comes with a pin. frozen_bundles.json (tracked) carries path + sha256 + file count per base; the bundles themselves are NEVER committed here. Three states, separated by construction: match -> resolves; gone -> FrozenBundleMissing (an OSError, so the gate's existing except OSError gives IKKE MAALT + exit 1 unchanged and the corpus tests SKIP, MAJOR-3's ceiling); drift -> FrozenBundleDrift (a ValueError), loud, named, and never a skip. The two classes are deliberately unrelated: a caller that catches "missing" to skip must not swallow "drift". The NAME is hashed alongside the bytes, and the directory name carries the first 12 chars of the digest so a stale copy is visible in ls. Renewal is a decision: new copy + new pin in the SAME commit (README). --bundle-root / PORTFOLIO_VEGNORMAL_ROOT stays as the operator's explicit, UNPINNED live mount. Iron Law: the tests were written and run RED first (collection error, then two arms of my own making). Load-bearing MEASURED, eight mutations all red against the WHOLE suite with a green control of 1984 passed / 5 skipped / 5 xfailed and a strict node-id superset (1977 -> 1994, 0 removed): M1 the pin is never verified (7) - M2 drift collapsed into missing (5) - M3 the name is not hashed (40) - M4 the gate seam reverted to root/name (1) - M5 the corpus helpers skip on drift too (4, one per file) - M6a the slash spelling back in src (1) - M6b the quoted path segment back in a test (1) - M7 the directory name drops the short digest (1, and 45 skipped, which proves absence is a SKIP and not a false green) - M8 the explicit override ignored (3, two of them in test_stress_judge_loadbearing.py, independent witnesses older than this work). M2 FALSIFIED THE TEST FIRST: the four parametrised arms did not go red, they went to SKIP (5 -> 9 skipped) and stayed green - pytest.skip inside a pytest.raises is not a failure. The arm now catches pytest.skip.Exception explicitly and turns it into an AssertionError. grep -rnE 'vegnormal-okf/build|["'"'"']vegnormal-okf["'"'"']' src tests contexts -> 0 (3 + 4 hits before; the three remaining prose mentions document history and are allowed). Gate re-run against the frozen copy: identical to the live mount (rows 0/3 - 0/3 - 3/8 - no report - 3/8 - IKKE MAALT - 1/20, exit 1). Order 20260917T223645Z-1296211942-from-.claude. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
1042 lines
41 KiB
Python
1042 lines
41 KiB
Python
"""The v1 gate: how far portfolio-optimiser is from v1, row by row, with an exit code.
|
||
|
||
One command, deterministic, offline: no model call, no network, no quota. Exit 0 only when every
|
||
FAILING row (1-6) is green; exit 1 otherwise; exit 2 on wrong usage. Row 7 is a diagnosis and never
|
||
moves the exit code.
|
||
|
||
The gate defines the CONTRACT a later capability must deliver into, not the generator. It reads a
|
||
rounds directory of a fixed shape (see ``--help``); nothing in the product writes it yet. Every
|
||
denominator comes from a source outside the thing being measured: the number of rounds is the
|
||
operator's choice (3), the feedback types are the eight named in ``v1_gate.json``, the MAF points
|
||
are a listed, approved-or-not set, and rows 6-7 read the stress round's own artefacts.
|
||
|
||
Rows 3 and 6 run NAMED tests (``v1_gate.json``) in a child pytest with ``--runxfail``: a type
|
||
counts only when every test registered for it passes, so a missing surface is a red test, never a
|
||
missing one.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import ast
|
||
import json
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
import xml.etree.ElementTree as ET
|
||
from collections.abc import Callable, Iterable, Mapping, Sequence
|
||
from dataclasses import asdict, dataclass, field
|
||
from datetime import datetime, timezone
|
||
import difflib
|
||
import re
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from portfolio_optimiser.validator import rejection_stage
|
||
|
||
_DATA = Path(__file__).with_name("v1_gate.json")
|
||
_PACKAGE_SRC = Path(__file__).resolve().parents[1]
|
||
_REPO_ROOT = Path(__file__).resolve().parents[3]
|
||
#: Gitignored by default: the remote is public, and a domain expert's feedback must never reach it.
|
||
DEFAULT_ROUNDS_DIR = "v1-rounds"
|
||
#: A line of an AI-authored document shorter than this is too generic to identify its origin.
|
||
_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"
|
||
RED = "RØD"
|
||
DIAGNOSIS = "DIAGNOSE"
|
||
#: A row whose evidence could not be read. Never green: on a failing row it fails the exit code.
|
||
NOT_MEASURED = "IKKE MÅLT"
|
||
|
||
ROUNDS_CONTRACT = """\
|
||
Rundekatalogen (--rounds-dir) har fast form. Runde n = tilbakemelding på rapport n-1, så kjøring n:
|
||
|
||
<rounds-dir>/0/report.md rapporten fra grunnkjøringen (runde 0)
|
||
<rounds-dir>/0/outcome.json grunnkjøringen som runde 1 måles mot
|
||
<rounds-dir>/<n>/feedback.json fagpersonens tilbakemelding på rapport n-1 (n = 1, 2, 3)
|
||
<rounds-dir>/<n>/outcome.json kjøring n, gjort ETTER den tilbakemeldingen
|
||
<rounds-dir>/<n>/report.md rapporten bygget fra kjøring n
|
||
<rounds-dir>/3/report.kept.md runde 3-rapporten slik fagpersonen BEHOLDT den
|
||
|
||
feedback.json:
|
||
{"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>"}]}
|
||
|
||
outcome.json (hver rad sjekkes mot kjøringens egen <outbox>/<run_id>-coverage.json):
|
||
{"run_id": "<kjøringen>", "outbox": "<utboksen, relativ til denne fila eller absolutt>",
|
||
"approaches": [{"id": "<tilnærming>", "validated": true|false,
|
||
"stage": "<avvisningsstadium som validator.rejection_stage gir, tom når validert>",
|
||
"validated_nok": <tall eller null>,
|
||
"feedback_ids": ["<id-er fra feedback.json som forklarer raden>"]}],
|
||
"removed": [{"id": "<tilnærming fjernet siden forrige runde>", "feedback_ids": [...]}]}
|
||
|
||
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 (endring
|
||
under 1 % er støy) — OG minst én endret rad bærer en feedback-id gitt i DENNE runden, gitt mellom
|
||
de to kjøringene. Hver runde må ha minst ett nytt punkt og egne id-er. Tekst tatt fra et
|
||
AI-forfattet dokument (docs/ekspert-svar.md) teller aldri. Rad 1-2 beviser FORM, ikke forfatterskap.
|
||
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.
|
||
"""
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Row:
|
||
key: str
|
||
title: str
|
||
k: int | None
|
||
n: int | None
|
||
status: str
|
||
reason: str
|
||
failing: bool = True
|
||
exceptions: tuple[str, ...] = ()
|
||
diagnostics: tuple[str, ...] = ()
|
||
|
||
def line(self) -> str:
|
||
k = "–" if self.k is None else str(self.k)
|
||
n = "–" if self.n is None else str(self.n)
|
||
return f"{self.title} | {k} av {n} | {self.status} | {self.reason}"
|
||
|
||
|
||
def load_config(path: Path = _DATA) -> dict[str, Any]:
|
||
data: dict[str, Any] = json.loads(path.read_text(encoding="utf-8"))
|
||
return data
|
||
|
||
|
||
# ---------------------------------------------------------------------------------------------
|
||
# Rows 1 and 2 — rounds
|
||
# ---------------------------------------------------------------------------------------------
|
||
|
||
|
||
def _norm(text: str) -> str:
|
||
"""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:
|
||
"""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
|
||
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] = []
|
||
for rel in docs:
|
||
path = repo_root / rel
|
||
if not path.is_file():
|
||
return None
|
||
texts.append(path.read_text(encoding="utf-8"))
|
||
lines = [
|
||
_norm(line.strip().lstrip(">*-#|` ").strip())
|
||
for text in texts
|
||
for line in text.splitlines()
|
||
]
|
||
return _norm("\n".join(texts)), "\n".join(x for x in lines if len(x) >= _AI_LINE_MIN)
|
||
|
||
|
||
def _is_ai_text(text: str, ai: tuple[str, str]) -> bool:
|
||
whole, lines = ai
|
||
item = _norm(text)
|
||
if len(item) >= _AI_LINE_MIN and item in whole:
|
||
return True
|
||
return any(line in item for line in lines.splitlines() if line)
|
||
|
||
|
||
def _parse_time(value: Any) -> datetime | None:
|
||
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"
|
||
if not path.is_file():
|
||
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 ""
|
||
return None, f"{name}: feedback.json mangler{extra}"
|
||
try:
|
||
data = json.loads(path.read_text(encoding="utf-8"))
|
||
author = str(data["author"]).strip()
|
||
items = list(data["items"])
|
||
given_raw = data["given_at"]
|
||
except (ValueError, KeyError, TypeError) as exc:
|
||
return None, f"{name}: feedback.json uleselig ({exc!r})"
|
||
if not author:
|
||
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:
|
||
return None, f"{name}: AI-vakten kunne ikke lese sine kilder"
|
||
ids: set[str] = set()
|
||
texts: set[str] = set()
|
||
for item in items:
|
||
try:
|
||
item_id, item_type, text = str(item["id"]), int(item["type"]), str(item["text"])
|
||
except (KeyError, TypeError, ValueError):
|
||
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:
|
||
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):
|
||
return None, f"{name}: punkt {item_id!r} er AI-forfattet tekst"
|
||
ids.add(item_id)
|
||
texts.add(_norm(text))
|
||
if not ids:
|
||
return None, f"{name}: feedback.json har ingen punkter"
|
||
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:
|
||
exceptions: list[str] = []
|
||
k = 0
|
||
if not rounds_dir.is_dir():
|
||
exceptions.append(f"{rounds_dir} finnes ikke")
|
||
else:
|
||
for _, (feedback, why) in sorted(read_rounds(rounds_dir, required, ai).items()):
|
||
if feedback is not None:
|
||
k += 1
|
||
else:
|
||
exceptions.append(why)
|
||
status = GREEN if k == required else RED
|
||
reason = (
|
||
"form verifisert i alle runder (forfatterskap: se attestering)"
|
||
if k == required
|
||
else exceptions[0]
|
||
)
|
||
return Row(
|
||
"rounds",
|
||
"1 runder med ekte fagperson",
|
||
k,
|
||
required,
|
||
status,
|
||
reason,
|
||
exceptions=tuple(exceptions),
|
||
)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
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:
|
||
data = json.loads(path.read_text(encoding="utf-8"))
|
||
rows = {str(a["id"]): a for a in data["approaches"]}
|
||
removed = {
|
||
str(r["id"]): set(map(str, r.get("feedback_ids", ()))) for r in data.get("removed", ())
|
||
}
|
||
run_id = str(data["run_id"]).strip()
|
||
outbox = Path(str(data["outbox"])).expanduser()
|
||
except FileNotFoundError:
|
||
return None, f"{path} mangler"
|
||
except (ValueError, KeyError, TypeError) as exc:
|
||
return None, f"{path} uleselig ({exc!r})"
|
||
if not run_id:
|
||
return None, f"{path}: run_id er tom"
|
||
if not outbox.is_absolute():
|
||
outbox = path.parent / outbox
|
||
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]:
|
||
nok = row.get("validated_nok")
|
||
return (
|
||
bool(row.get("validated")),
|
||
str(row.get("stage") or ""),
|
||
None if nok is None else float(nok),
|
||
)
|
||
|
||
|
||
def _nok_changed(before: Any, after: Any) -> bool:
|
||
"""(d) with a noise floor. ``validated_nok`` is the model's own claim, so two runs on the same
|
||
input can differ by rounding; a change smaller than 1 % of the earlier figure (and never less
|
||
than 1 NOK) is not something an expert's feedback asked for, and it does not count."""
|
||
if (before is None) != (after is None):
|
||
return True
|
||
if before is None or after is None:
|
||
return False
|
||
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]] = {}
|
||
for aid, row in cur.rows.items():
|
||
if aid not in prev.rows or _changed(prev.rows[aid], row):
|
||
changed[aid] = set(map(str, row.get("feedback_ids", ())))
|
||
for aid in prev.rows.keys() - cur.rows.keys():
|
||
changed[aid] = cur.removed.get(aid, set())
|
||
if not changed:
|
||
return False, "ingen endring i (a)-(d) over støygrensen"
|
||
traced = sorted(aid for aid, ids in changed.items() if ids & feedback_ids)
|
||
if not traced:
|
||
return False, f"{len(changed)} rad(er) endret, ingen sporet til rundens feedback-id-er"
|
||
return True, f"{len(changed)} rad(er) endret, sporet: {', '.join(traced)}"
|
||
|
||
|
||
def score_changes(rounds_dir: Path, required: int, ai: tuple[str, str] | None) -> Row:
|
||
exceptions: list[str] = []
|
||
k = 0
|
||
base_path = rounds_dir / "0" / "outcome.json"
|
||
base_outcome, base_why = read_outcome(base_path)
|
||
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):
|
||
feedback, why = feedback_by_round.get(n, (None, f"runde {n}: {rounds_dir} finnes ikke"))
|
||
if feedback is None:
|
||
exceptions.append(why)
|
||
continue
|
||
prev, why_prev = read_outcome(rounds_dir / str(n - 1) / "outcome.json")
|
||
cur, why_cur = read_outcome(rounds_dir / str(n) / "outcome.json")
|
||
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:
|
||
k += 1
|
||
else:
|
||
exceptions.append(f"runde {n}: {detail}")
|
||
status = GREEN if k == required else RED
|
||
return Row(
|
||
"changes",
|
||
"2 runder med målbar endring",
|
||
k,
|
||
required,
|
||
status,
|
||
base,
|
||
exceptions=tuple(exceptions),
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------------------------
|
||
# Row 3 and the row 6 probes — named tests run in a child
|
||
# ---------------------------------------------------------------------------------------------
|
||
|
||
ProbeRunner = Callable[[Sequence[str]], Mapping[str, str]]
|
||
|
||
|
||
def run_probes(nodeids: Sequence[str], repo_root: Path = _REPO_ROOT) -> dict[str, str]:
|
||
"""``nodeid -> "passed" | "failed" | "skipped" | "missing"``, from a child pytest run with
|
||
``--runxfail`` so a known gap shows as the failure it is."""
|
||
outcomes = {n: "missing" for n in nodeids}
|
||
if not (repo_root / "tests").is_dir():
|
||
return outcomes
|
||
env = {**os.environ, "PYTHONDONTWRITEBYTECODE": "1"}
|
||
base = [sys.executable, "-m", "pytest", "-p", "no:cacheprovider", "-q"]
|
||
files = sorted(
|
||
{n.split("::", 1)[0] for n in nodeids if (repo_root / n.split("::")[0]).is_file()}
|
||
)
|
||
if not files:
|
||
return outcomes
|
||
collect = subprocess.run(
|
||
[*base, "--collect-only", *files], cwd=repo_root, env=env, capture_output=True, text=True
|
||
)
|
||
collected = {line.strip() for line in collect.stdout.splitlines() if "::" in line}
|
||
present = [n for n in nodeids if n in collected]
|
||
if not present:
|
||
return outcomes
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
junit = Path(tmp) / "junit.xml"
|
||
subprocess.run(
|
||
[*base, "--runxfail", f"--junitxml={junit}", *present],
|
||
cwd=repo_root,
|
||
env=env,
|
||
capture_output=True,
|
||
text=True,
|
||
)
|
||
if not junit.is_file():
|
||
return {n: ("failed" if n in present else "missing") for n in nodeids}
|
||
for case in ET.parse(junit).getroot().iter("testcase"):
|
||
nodeid = case.get("classname", "").replace(".", "/") + ".py::" + case.get("name", "")
|
||
if nodeid not in outcomes:
|
||
continue
|
||
if case.find("failure") is not None or case.find("error") is not None:
|
||
outcomes[nodeid] = "failed"
|
||
elif case.find("skipped") is not None:
|
||
outcomes[nodeid] = "skipped"
|
||
else:
|
||
outcomes[nodeid] = "passed"
|
||
return {n: (o if o != "missing" or n not in present else "failed") for n, o in outcomes.items()}
|
||
|
||
|
||
def green_types(types: Mapping[str, Any], outcomes: Mapping[str, str]) -> dict[int, str]:
|
||
"""``type -> ""`` when green, else the reason. Partial is no: every registered test passes."""
|
||
result: dict[int, str] = {}
|
||
for key, spec in types.items():
|
||
evidence = list(spec.get("evidence", ()))
|
||
if not evidence:
|
||
result[int(key)] = "ingen probe registrert"
|
||
continue
|
||
bad = [
|
||
f"{n.split('::')[-1]}={outcomes.get(n, 'missing')}"
|
||
for n in evidence
|
||
if outcomes.get(n) != "passed"
|
||
]
|
||
result[int(key)] = "; ".join(bad)
|
||
return result
|
||
|
||
|
||
def score_types(types: Mapping[str, Any], outcomes: Mapping[str, str]) -> Row:
|
||
verdicts = green_types(types, outcomes)
|
||
k = sum(1 for why in verdicts.values() if not why)
|
||
n = len(types)
|
||
exceptions = tuple(
|
||
f"type {t} ({types[str(t)]['label']}): {why}" for t, why in sorted(verdicts.items()) if why
|
||
)
|
||
green = ", ".join(str(t) for t, why in sorted(verdicts.items()) if not why) or "ingen"
|
||
status = GREEN if k == n else RED
|
||
return Row(
|
||
"types",
|
||
"3 tilbakemeldingstyper med vei inn OG handling",
|
||
k,
|
||
n,
|
||
status,
|
||
f"grønne: {green}",
|
||
exceptions=exceptions,
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------------------------
|
||
# Row 4 — the round 3 report kept
|
||
# ---------------------------------------------------------------------------------------------
|
||
|
||
|
||
#: 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():
|
||
return Kept(0, 0, 0, False, "ingen rapport")
|
||
raw_report = report.read_bytes()
|
||
raw_kept = kept.read_bytes()
|
||
before = content_lines(raw_report.decode("utf-8"))
|
||
after = content_lines(raw_kept.decode("utf-8"))
|
||
if not before:
|
||
return Kept(0, 0, 0, False, "tom rapport")
|
||
matcher = difflib.SequenceMatcher(None, before, after, autojunk=False)
|
||
same = sum(block.size for block in matcher.get_matching_blocks())
|
||
return Kept(same, len(before), len(after) - same, raw_report == raw_kept)
|
||
|
||
|
||
def score_kept(rounds_dir: Path, threshold: float, ai: tuple[str, str] | None = None) -> Row:
|
||
result = kept_ratio(rounds_dir / "3" / "report.md", rounds_dir / "3" / "report.kept.md")
|
||
title = f"4 runde 3-rapport beholdt (≥ {threshold:.0%} innholdslinjer)"
|
||
if result.why:
|
||
return Row("kept", title, None, result.total or None, RED, result.why)
|
||
added = f"; {result.added} linje(r) lagt til av fagpersonen"
|
||
if result.untouched:
|
||
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(
|
||
"kept",
|
||
title,
|
||
result.kept,
|
||
result.total,
|
||
GREEN if ok else RED,
|
||
f"{result.kept / result.total:.1%} av innholdslinjene uendret og i rekkefølge{added}",
|
||
diagnostics=(f"lagt til: {result.added}",),
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------------------------
|
||
# Row 5 — MAF points, each pointing at a type
|
||
# ---------------------------------------------------------------------------------------------
|
||
|
||
|
||
def _imports(tree: ast.AST, construct: str, package: str) -> set[str]:
|
||
names: set[str] = set()
|
||
for node in ast.walk(tree):
|
||
if isinstance(node, ast.ImportFrom) and node.module and node.module.startswith(package):
|
||
names |= {a.asname or a.name for a in node.names if a.name == construct}
|
||
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:
|
||
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]:
|
||
"""(construct imported from MAF and used somewhere in ``src``, named call site uses it)."""
|
||
present = False
|
||
for path in sorted(src.glob("*.py")):
|
||
tree = ast.parse(path.read_text(encoding="utf-8"))
|
||
names = _imports(tree, point["construct"], point["package"])
|
||
body = [
|
||
n for n in getattr(tree, "body", []) if not isinstance(n, (ast.Import, ast.ImportFrom))
|
||
]
|
||
if names and _referenced(body, names):
|
||
present = True
|
||
break
|
||
site = point.get("callsite")
|
||
if not site:
|
||
return present, False
|
||
path = src / site["module"]
|
||
if not path.is_file():
|
||
return present, False
|
||
tree = ast.parse(path.read_text(encoding="utf-8"))
|
||
names = _imports(tree, point["construct"], point["package"])
|
||
scopes = [
|
||
n
|
||
for n in getattr(tree, "body", [])
|
||
if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef))
|
||
and n.name == site["scope"]
|
||
]
|
||
return present, bool(names) and _referenced(scopes, names)
|
||
|
||
|
||
def score_maf(maf: Mapping[str, Any], type_verdicts: Mapping[int, str], src: Path) -> Row:
|
||
points = list(maf["points"])
|
||
exceptions: list[str] = []
|
||
presence = callsites = pointers = 0
|
||
for point in points:
|
||
present, site = maf_presence(point, src)
|
||
types_green = all(not type_verdicts.get(int(t), "ukjent") for t in point["types"])
|
||
presence += present
|
||
callsites += site
|
||
counts = present and site and types_green
|
||
pointers += counts
|
||
if not counts:
|
||
missing = [
|
||
w
|
||
for w, ok in (
|
||
("presence", present),
|
||
("kallsted", site),
|
||
(f"type {point['types']} grønn", types_green),
|
||
)
|
||
if not ok
|
||
]
|
||
exceptions.append(f"{point['u_id']} {point['construct']}: mangler {', '.join(missing)}")
|
||
n = len(points)
|
||
diagnostics = (
|
||
f"presence {presence} av {n}",
|
||
f"kallsted verifisert {callsites} av {n}",
|
||
f"med grønn typepeker {pointers} av {n}",
|
||
)
|
||
title = "5 MAF-punkter med typepeker"
|
||
if maf.get("approved") is not True:
|
||
return Row(
|
||
"maf",
|
||
title,
|
||
0,
|
||
n,
|
||
RED,
|
||
"M ikke godkjent av operatøren",
|
||
exceptions=tuple(exceptions),
|
||
diagnostics=diagnostics,
|
||
)
|
||
return Row(
|
||
"maf",
|
||
title,
|
||
pointers,
|
||
n,
|
||
GREEN if pointers == n else RED,
|
||
f"presence {presence} av {n}",
|
||
exceptions=tuple(exceptions),
|
||
diagnostics=diagnostics,
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------------------------
|
||
# Rows 6 and 7 — the stress round's artefacts
|
||
# ---------------------------------------------------------------------------------------------
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class StressMeasure:
|
||
validated: int = 0
|
||
undeclared: int = 0
|
||
undeclared_anywhere: int = 0
|
||
named: int = 0
|
||
rows: int = 0
|
||
commissioned: int = 0
|
||
where: str = ""
|
||
missing: str = ""
|
||
#: Declarations with no ``approach_id`` — written before the rule; the row cannot be measured.
|
||
unaddressed: int = 0
|
||
#: Of ``validated``, how many were the runs' own proposals (M-3).
|
||
own_validated: int = 0
|
||
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(
|
||
evidence: Mapping[str, Any], repo_root: Path, stress_root: Path, bundle_root: Path | None
|
||
) -> StressMeasure:
|
||
"""Re-judge every listed outbox with the current judge. Any run that cannot be judged makes the
|
||
whole measurement absent — a partial one would carry the wrong denominator."""
|
||
from portfolio_optimiser import frozen_bundles, stress
|
||
from portfolio_optimiser.mandate import load_mandate
|
||
|
||
verdicts = []
|
||
contexts: set[str] = set()
|
||
for run_spec in evidence["runs"]:
|
||
context = repo_root / run_spec["context"]
|
||
outbox = stress_root / run_spec["outbox"]
|
||
if not outbox.is_dir():
|
||
return StressMeasure(where=str(stress_root), missing=f"{outbox} finnes ikke")
|
||
declared = stress.read_bundle_declarations(context / "bundle.txt")
|
||
wanted = run_spec.get("bundle")
|
||
chosen = [d for d in declared if wanted in (None, d["name"], d["bundle_id"])]
|
||
if len(chosen) != 1:
|
||
return StressMeasure(where=str(stress_root), missing=f"{context}: base ikke entydig")
|
||
try:
|
||
# The frozen copy this repository pins, unless the operator named a live mount.
|
||
# Drift is a ValueError and absence an OSError: both land in ``missing`` below, so a
|
||
# corpus that moved is IKKE MÅLT with the reason said, never a silently wrong number.
|
||
base = frozen_bundles.bundle_dir(chosen[0]["name"], override=bundle_root)
|
||
verdicts.append(
|
||
stress.score_context_set(
|
||
context,
|
||
outbox,
|
||
run_spec["run_id"],
|
||
base,
|
||
bundle_id=chosen[0]["bundle_id"] if len(declared) > 1 else None,
|
||
)
|
||
)
|
||
except (stress.EmptyMeasurement, OSError, ValueError) as exc:
|
||
return StressMeasure(where=str(stress_root), missing=f"{run_spec['run_id']}: {exc}")
|
||
contexts.add(run_spec["context"])
|
||
approaches = [a for v in verdicts for a in v.approaches]
|
||
validated = [a for a in approaches if a.status == "validated"]
|
||
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)
|
||
commissioned = sum(
|
||
len(load_mandate(repo_root / c / "mandate.json").approaches) for c in contexts
|
||
)
|
||
return StressMeasure(
|
||
validated=len(validated) + own_validated,
|
||
undeclared=len(undeclared) + own_undeclared,
|
||
undeclared_anywhere=sum(1 for a in validated if a.requirement_source == "absent"),
|
||
named=sum(1 for a in approaches if a.named),
|
||
rows=len(approaches),
|
||
commissioned=commissioned,
|
||
where=str(stress_root),
|
||
unaddressed=unaddressed,
|
||
undeclared_ids=tuple(a.approach_id for a in undeclared) + own_ids,
|
||
own_validated=own_validated,
|
||
)
|
||
|
||
|
||
def score_undeclared(
|
||
probes: Sequence[str], outcomes: Mapping[str, str], m: StressMeasure, label: str
|
||
) -> Row:
|
||
"""GREEN only when every probe passes AND the artefacts were measured with k = 0. Evidence that
|
||
could not be read is IKKE MÅLT — never green, and it fails the exit code like red does."""
|
||
failing = [
|
||
f"{n.split('::')[-1]}={outcomes.get(n, 'missing')}"
|
||
for n in probes
|
||
if outcomes.get(n) != "passed"
|
||
]
|
||
if not probes:
|
||
failing.append("ingen probe registrert")
|
||
title = "6 validert UTEN erklært krav (tilnærmingens egen)"
|
||
exceptions = [f"probe {x}" for x in failing]
|
||
probe_state = "prober røde" if failing else "prober grønne"
|
||
k: int | None = None
|
||
n: int | None = None
|
||
diagnostics: tuple[str, ...] = ()
|
||
if m.missing:
|
||
reason = f"{probe_state}; {label}: ikke målt, artefakter mangler ({m.missing})"
|
||
elif m.unaddressed:
|
||
reason = (
|
||
f"{probe_state}; {label}: ikke målt: artefaktene er eldre enn regelen "
|
||
f"(approach_id mangler på {m.unaddressed} erklæring(er))"
|
||
)
|
||
diagnostics = (
|
||
f"før regelen: {m.undeclared} av {m.validated} validerte (hvorav {m.own_validated} "
|
||
"egne forslag) uten tilnærmingens egen erklæring — regelen ville gjort dem "
|
||
"unsupported, men modellen fikk aldri spørsmålet",
|
||
)
|
||
else:
|
||
k, n = m.undeclared, m.validated
|
||
reason = (
|
||
f"{probe_state}; {label} ({m.where}): {k} av {n} validerte uten erklæring fra "
|
||
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]
|
||
if failing or k:
|
||
status = RED
|
||
elif k is None:
|
||
status = NOT_MEASURED
|
||
else:
|
||
status = GREEN
|
||
return Row(
|
||
"undeclared",
|
||
title,
|
||
k,
|
||
n,
|
||
status,
|
||
reason,
|
||
exceptions=tuple(exceptions),
|
||
diagnostics=diagnostics,
|
||
)
|
||
|
||
|
||
NAMED_WARNING = (
|
||
"en prompt-endring som ber modellen gjengi `ref` gjør `named` til noe modellen blir BEDT om, "
|
||
"og svekker den som uavhengig måling"
|
||
)
|
||
|
||
|
||
def score_named(m: StressMeasure, label: str) -> Row:
|
||
title = "7 named (diagnose, ingen terskel)"
|
||
if m.missing:
|
||
return Row(
|
||
"named",
|
||
title,
|
||
None,
|
||
None,
|
||
NOT_MEASURED,
|
||
f"{label}: ikke målt, artefakter mangler ({m.missing})",
|
||
failing=False,
|
||
diagnostics=(NAMED_WARNING,),
|
||
)
|
||
extra = "" if m.rows == m.commissioned else f"; {m.rows} rader dømt"
|
||
return Row(
|
||
"named",
|
||
title,
|
||
m.named,
|
||
m.commissioned,
|
||
DIAGNOSIS,
|
||
f"{label}: nevner = bestilte tilnærminger i fasitsettene{extra}",
|
||
failing=False,
|
||
diagnostics=(NAMED_WARNING,),
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------------------------
|
||
# The command
|
||
# ---------------------------------------------------------------------------------------------
|
||
|
||
|
||
def evaluate(
|
||
*,
|
||
rounds_dir: Path,
|
||
config: Mapping[str, Any],
|
||
repo_root: Path = _REPO_ROOT,
|
||
src: Path = _PACKAGE_SRC,
|
||
stress_root: Path | None = None,
|
||
bundle_root: Path | None = None,
|
||
probe_runner: ProbeRunner | None = None,
|
||
stress_measure: StressMeasure | None = None,
|
||
) -> list[Row]:
|
||
ai = ai_authored_lines(repo_root, config["ai_authored"])
|
||
required = int(config["rounds_required"])
|
||
types = config["feedback_types"]
|
||
probes = list(config["row6_evidence"])
|
||
nodeids = [n for spec in types.values() for n in spec.get("evidence", ())] + probes
|
||
outcomes = (probe_runner or (lambda ids: run_probes(ids, repo_root)))(nodeids)
|
||
evidence = config["stress_evidence"]
|
||
if stress_measure is None:
|
||
stress_measure = measure_stress(
|
||
evidence,
|
||
repo_root,
|
||
stress_root or repo_root / evidence["root"],
|
||
bundle_root,
|
||
)
|
||
return [
|
||
score_rounds(rounds_dir, required, ai),
|
||
score_changes(rounds_dir, required, ai),
|
||
score_types(types, outcomes),
|
||
score_kept(rounds_dir, float(config["keep_threshold"]), ai),
|
||
score_maf(config["maf_points"], green_types(types, outcomes), src),
|
||
score_undeclared(probes, outcomes, stress_measure, evidence["label"]),
|
||
score_named(stress_measure, evidence["label"]),
|
||
]
|
||
|
||
|
||
def exit_code(rows: Sequence[Row]) -> int:
|
||
return 0 if all(r.status == GREEN for r in rows if r.failing) else 1
|
||
|
||
|
||
def render(rows: Sequence[Row]) -> str:
|
||
out = ["rad | k av N | status | grunn"]
|
||
out += [r.line() for r in rows]
|
||
out.append("")
|
||
out.append(f"Attestering: {ATTESTATION}.")
|
||
out.append("")
|
||
out.append("Unntak fra 100 %:")
|
||
for r in rows:
|
||
for x in r.exceptions:
|
||
out.append(f" [{r.title.split()[0]}] {x}")
|
||
for d in r.diagnostics:
|
||
out.append(f" [{r.title.split()[0]}] diagnose: {d}")
|
||
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:
|
||
parser = argparse.ArgumentParser(
|
||
prog="python -m portfolio_optimiser.evals.v1_gate",
|
||
description="Hvor langt portfolio-optimiser er fra v1, rad for rad. Exit 0 kun når alle "
|
||
"fellende rader (1-6) er grønne, 1 ellers, 2 ved feil bruk. Ingen modellkall, intet nett.",
|
||
epilog=ROUNDS_CONTRACT,
|
||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||
)
|
||
parser.add_argument(
|
||
"--rounds-dir",
|
||
default=None,
|
||
help=f"rundekatalogen (default {DEFAULT_ROUNDS_DIR}/, gitignored)",
|
||
)
|
||
parser.add_argument(
|
||
"--stress-root", default=None, help="utboks-roten for stressrunden rad 6-7 dømmer"
|
||
)
|
||
parser.add_argument(
|
||
"--bundle-root",
|
||
default=None,
|
||
help="en EKSPLISITT, UPINNET levende montering; uten den svarer den frosne kopien og "
|
||
"sha256-pinnen verifiseres",
|
||
)
|
||
parser.add_argument("--json", action="store_true", help="maskinlesbar output")
|
||
args = parser.parse_args(argv)
|
||
|
||
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")
|
||
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
|
||
rows = evaluate(
|
||
rounds_dir=rounds_dir,
|
||
config=load_config(),
|
||
stress_root=Path(args.stress_root) if args.stress_root else None,
|
||
bundle_root=Path(args.bundle_root).expanduser() if args.bundle_root else None,
|
||
)
|
||
code = exit_code(rows)
|
||
if args.json:
|
||
print(
|
||
json.dumps(
|
||
{"exit": code, "attestation": ATTESTATION, "rows": [asdict(r) for r in rows]},
|
||
ensure_ascii=False,
|
||
indent=2,
|
||
)
|
||
)
|
||
else:
|
||
print(render(rows))
|
||
return code
|
||
|
||
|
||
if __name__ == "__main__": # pragma: no cover - exercised by a subprocess test
|
||
raise SystemExit(main())
|