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

test(v1-gate): the v1 gate, written RED

Browse source 
One command says how far the repo is from v1, row by row, with an exit
code: rounds with a real domain expert 0/3, traced measurable change 0/3,
feedback types with a way in and an action 3/8 (1, 3, 7), round 3 report
kept - none, MAF points with a green type pointer 0/8 (list not approved),
validated without the approach's own declaration 10/10 in stress round 6,
and `named` 1/20 as a diagnosis that never moves the exit code.

The gate defines the contract (a fixed rounds directory, gitignored by
default), not the generator. Rows 3 and 6 run named tests with
--runxfail; the red probes are xfail(strict=True) so the suite stays green
while the gap is real. No product code changed.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Kjell Tore Guttormsen 2026-09-17 15:31:48 +02:00
commit 83c94e4fb6
7 changed files with 1397 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

3
.gitignore vendored
View file

@ -40,3 +40,6 @@ PLAN.md
# S3.1 embedding-vektorstore: rebyggbar cache, aldri autoritativ — repoet holder null binærartefakter
*.npy
# v1 gate: default rounds directory. Holds a domain expert's feedback and reports - never public.
v1-rounds/

22
CLAUDE.md
View file

@ -3090,6 +3090,28 @@ Python ≥3.10. MAF (`agent-framework-core` 1.16.0, `-orchestrations` 1.1.1 —
modell har lest den nye klausulen (structured-output-grensens klasse), DEL D er maalingen; og
klausulen er hjelpetekst, ikke en gate — den kan ikke gjoere en gjettet sti riktig, bare billigere
aa rette.
- **v1-gaten måler avstanden til v1 med exit-kode, og den definerer KONTRAKTEN, ikke generatoren
(17.09):** `uv run python -m portfolio_optimiser.evals.v1_gate` (exit 0 kun når rad 1–6 er grønne,
1 ellers, 2 ved feil bruk; `--json`; ingen modellkall, intet nett). Den leser en rundekatalog med
fast form (`--help`; default `v1-rounds/`, GITIGNORED fordi fagpersonens tilbakemelding aldri skal
nå den offentlige remoten) — `feedback.json` er fastsatt som JSON, ikke `feedback.*`, fordi hvert
punkt må bære id og type for at sporingskravet i rad 2 kan måles. Rad 3 og 6 kjører NAVNGITTE
tester (`evals/v1_gate.json`) i en barne-pytest med `--runxfail`; de røde probene
(`tests/test_v1_probes.py`) er `xfail(strict=True)`, så suiten er grønn mens gapet er ekte og blir
RØD på XPASS den dagen en kapabilitet lukker det. En probe for en type uten flate feiler også når
et matchende flagg DUKKER OPP — «delvis er nei», og den blir grønn først når den skrives om til å
drive døra og observere handlingen. Rad 5 er RØD så lenge datafilens `approved` er `false`, og et
MAF-punkt teller kun når typen det peker på er grønn i rad 3 (kallstedet verifisert med AST, en
kommentar teller ikke). Rad 6 leser tilnærmingens EGEN erklæring (`requirement_source ==
"approach"`) — en kjøringsnivå-erklæring kan ikke tilskrives én tilnærming — og dømmer
stressrunde 6-utboksene på nytt med dagens dommer; kan én kjøring ikke dømmes, er HELE målingen
«ikke målt», aldri 0. Load-bearing MÅLT (`tests/test_v1_gate.py`), seks mutasjoner alle røde:
sporingskravet borte (2) · AI-vakten borte (1) · `>` i stedet for `≥` 80 % (1) · godkjennings-
vakten borte (1) · rad 6 ignorerer k (1) · rad 7 blir fellende (2). **Ærlighets-grenser, uttalt:**
basene rad 6–7 dømmer mot er et annet repos montering og kan være under ombygging (målt 17.09:
`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
rettes (gatet av en egen arm).
- **STATE.md er local-only** (gitignored). Voyage session-state er efemert; STATE.md er kanonisk kontinuitet.
- Prosess: Voyage-plugin (`/trekbrief → /trekplan → /trekexecute → /trekreview`) per større fase.

1
src/portfolio_optimiser/evals/__init__.py Normal file
View file

@ -0,0 +1 @@
"""Deterministic evaluation gates (no model calls)."""

87
src/portfolio_optimiser/evals/v1_gate.json Normal file
View file

@ -0,0 +1,87 @@
{
"rounds_required": 3,
"keep_threshold": 0.8,
"ai_authored": ["docs/ekspert-svar.md"],
"feedback_types": {
"1": {
"label": "feil - rett en feil i forslaget",
"evidence": [
"tests/test_proposal_review_loop_loadbearing.py::test_t13_the_flag_answers_the_review_from_a_real_argv_and_the_answer_is_used"
]
},
"2": {
"label": "bort - ta en retning bort",
"evidence": ["tests/test_v1_probes.py::test_type_2_remove_a_direction_has_a_typed_door"]
},
"3": {
"label": "vinklinger - nye vinklinger",
"evidence": ["tests/test_mandate_cli.py::test_run_settles_against_the_mandate_afterwards"]
},
"4": {
"label": "lette paa krav",
"evidence": ["tests/test_v1_probes.py::test_type_4_relax_a_requirement_has_a_door"]
},
"5": {
"label": "konseptgrafer",
"evidence": ["tests/test_v1_probes.py::test_type_5_edit_the_concept_graph_has_a_door"]
},
"6": {
"label": "skills per analyse",
"evidence": ["tests/test_v1_probes.py::test_type_6_skills_per_analysis_has_a_door"]
},
"7": {
"label": "MCP - verktoey i debatten",
"evidence": [
"tests/test_mcp_run_loadbearing.py::test_configured_server_becomes_a_tool_the_agents_have",
"tests/test_b4_mcp_call_trace_loadbearing.py::test_a_called_mcp_tool_is_recorded_in_provenance"
]
},
"8": {
"label": "inline kontekst",
"evidence": ["tests/test_v1_probes.py::test_type_8_inline_context_has_a_door"]
}
},
"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"
],
"maf_points": {
"approved": false,
"points": [
{"u_id": "U13", "construct": "MagenticPlanReviewResponse", "package": "agent_framework",
"callsite": {"module": "explore.py", "scope": "_drive"}, "types": [1, 2]},
{"u_id": "U9", "construct": "ContextProvider", "package": "agent_framework",
"callsite": {"module": "verdicts.py", "scope": "ExpeLContextProvider"}, "types": [1, 8]},
{"u_id": "U12", "construct": "FileCheckpointStorage", "package": "agent_framework",
"callsite": {"module": "explore.py", "scope": "checkpoint_storage"}, "types": [1]},
{"u_id": "U4", "construct": "MagenticBuilder", "package": "agent_framework",
"callsite": {"module": "explore.py", "scope": "fresh_exploration_workflow"}, "types": [3]},
{"u_id": "U7", "construct": "tool", "package": "agent_framework",
"callsite": {"module": "explore.py", "scope": "navigator_tools"}, "types": [4]},
{"u_id": "U11", "construct": "TextSpanRegion", "package": "agent_framework",
"callsite": {"module": "provenance.py", "scope": "ProvenanceStamp"}, "types": [5]},
{"u_id": "U5", "construct": "SkillsProvider", "package": "agent_framework",
"callsite": null, "types": [6]},
{"u_id": "U6", "construct": "MCPStdioTool", "package": "agent_framework",
"callsite": {"module": "mcp_tools.py", "scope": "build_mcp_tools"}, "types": [7]}
]
},
"stress_evidence": {
"label": "stressrunde 6",
"root": "scratchpad/p22-stress",
"runs": [
{"context": "contexts/gate-nordvik-2027", "outbox": "gate-nordvik-2027",
"run_id": "gate-nordvik-2027-08"},
{"context": "contexts/tunnel-hauglia-2027", "outbox": "tunnel-hauglia-2027",
"run_id": "tunnel-hauglia-2027-08"},
{"context": "contexts/fv412-dekkefornyelse-2027", "outbox": "fv412-dekkefornyelse-2027",
"run_id": "fv412-dekkefornyelse-2027-08"},
{"context": "contexts/kontrakt-sorasen-2027", "outbox": "kontrakt-sorasen-2027",
"run_id": "kontrakt-sorasen-2027-08"},
{"context": "contexts/dekke-og-kontrakt-lindaas-2027", "outbox": "lindaas",
"run_id": "lindaas-04-vegnormal-n200-2024", "bundle": "vegnormal-n200-2024"},
{"context": "contexts/dekke-og-kontrakt-lindaas-2027", "outbox": "lindaas",
"run_id": "lindaas-04-vegnormal-r761-2025", "bundle": "vegnormal-r761-2025"}
]
}
}

722
src/portfolio_optimiser/evals/v1_gate.py Normal file
View file

@ -0,0 +1,722 @@
"""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 import Counter
from collections.abc import Callable, Iterable, Mapping, Sequence
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Any
_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
GREEN = "GRØNN"
RED = "RØD"
DIAGNOSIS = "DIAGNOSE"
ROUNDS_CONTRACT = """\
Rundekatalogen (--rounds-dir) har fast form:
<rounds-dir>/0/outcome.json grunnkjøringen (runde 0) som runde 1 måles mot
<rounds-dir>/<n>/report.md rapporten slik fagpersonen fikk den (n = 1, 2, 3)
<rounds-dir>/<n>/feedback.json fagpersonens tilbakemelding
<rounds-dir>/<n>/outcome.json utfallet rapporten er bygget fra
<rounds-dir>/3/report.kept.md runde 3-rapporten slik fagpersonen BEHOLDT den
feedback.json:
{"author": "<fagpersonen>",
"items": [{"id": "<unik id>", "type": <1-8>, "text": "<tilbakemeldingen>"}]}
outcome.json:
{"run_id": "<kjøringen>",
"approaches": [{"id": "<tilnærming>", "validated": true|false,
"stage": "<avvisningsstadium, 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 den 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
én endret rad bærer en feedback-id gitt i DENNE runden. Tekst tatt fra et AI-forfattet dokument
(docs/ekspert-svar.md) teller aldri som fagperson-tilbakemelding.
"""
@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:
return " ".join(text.split())
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."""
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 read_feedback(round_dir: Path, ai: tuple[str, str] | None) -> tuple[set[str], str]:
"""The ids of a round's feedback items, and ``""`` — or an empty set and the reason."""
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 set(), f"runde {round_dir.name}: feedback.json mangler{extra}"
try:
data = json.loads(path.read_text(encoding="utf-8"))
author = str(data["author"]).strip()
items = list(data["items"])
except (ValueError, KeyError, TypeError) as exc:
return set(), f"runde {round_dir.name}: feedback.json uleselig ({exc!r})"
if not author:
return set(), f"runde {round_dir.name}: feedback.json navngir ingen fagperson"
if ai is None:
return set(), f"runde {round_dir.name}: AI-vakten kunne ikke lese sine kilder"
ids: 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 set(), f"runde {round_dir.name}: et feedback-punkt mangler id/type/text"
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"
if _is_ai_text(text, ai):
return set(), f"runde {round_dir.name}: punkt {item_id!r} er AI-forfattet tekst"
ids.add(item_id)
if not ids:
return set(), f"runde {round_dir.name}: feedback.json har ingen punkter"
return ids, ""
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 n in range(1, required + 1):
ids, why = read_feedback(rounds_dir / str(n), ai)
if ids:
k += 1
else:
exceptions.append(why)
status = GREEN if k == required else RED
reason = "alle runder har fagperson-tilbakemelding" if k == required else exceptions[0]
return Row(
"rounds",
"1 runder med ekte fagperson",
k,
required,
status,
reason,
exceptions=tuple(exceptions),
)
def _read_outcome(path: Path) -> tuple[dict[str, dict[str, Any]], dict[str, set[str]], str, str]:
"""Rows by approach id, removed ids with their feedback ids, the run id, and ``""`` or why."""
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", ())
}
return rows, removed, str(data.get("run_id", "")), ""
except FileNotFoundError:
return {}, {}, "", f"{path} mangler"
except (ValueError, KeyError, TypeError) as exc:
return {}, {}, "", f"{path} uleselig ({exc!r})"
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")
def round_changed(before: Path, after: Path, feedback_ids: set[str]) -> tuple[bool, str]:
"""Whether round ``after`` changed measurably against ``before`` AND the change is traced to
feedback given in this round. The second half is what keeps model noise out."""
prev, _, _, why = _read_outcome(before)
if why:
return False, why
cur, removed, _, why = _read_outcome(after)
if why:
return False, why
changed: dict[str, set[str]] = {}
for aid, row in cur.items():
if aid not in prev or _row_key(prev[aid]) != _row_key(row):
changed[aid] = set(map(str, row.get("feedback_ids", ())))
for aid in prev.keys() - cur.keys():
changed[aid] = removed.get(aid, set())
if not changed:
return False, "ingen endring i (a)-(d)"
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_run, base_why = _read_outcome(rounds_dir / "0" / "outcome.json")
base = f"runde 0 = {rounds_dir / '0' / 'outcome.json'}"
base += f" ({base_run})" if base_run else (f" — {base_why}" if base_why else "")
for n in range(1, required + 1):
ids, why = read_feedback(rounds_dir / str(n), ai)
if not ids:
exceptions.append(why)
continue
ok, detail = round_changed(
rounds_dir / str(n - 1) / "outcome.json", rounds_dir / str(n) / "outcome.json", 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
# ---------------------------------------------------------------------------------------------
def kept_ratio(report: Path, kept: Path) -> tuple[int, int, str]:
if not report.is_file() or not kept.is_file():
return 0, 0, "ingen rapport"
lines = [x for x in report.read_text(encoding="utf-8").splitlines() if x.strip()]
if not lines:
return 0, 0, "tom rapport"
pool = Counter(x for x in kept.read_text(encoding="utf-8").splitlines() if x.strip())
same = 0
for line in lines:
if pool[line] > 0:
pool[line] -= 1
same += 1
return same, len(lines), ""
def score_kept(rounds_dir: Path, threshold: float) -> Row:
same, total, why = kept_ratio(
rounds_dir / "3" / "report.md", rounds_dir / "3" / "report.kept.md"
)
title = f"4 runde 3-rapport beholdt (≥ {threshold:.0%} linjer)"
if why:
return Row("kept", title, None, total or None, RED, why)
ok = same >= threshold * total
return Row(
"kept",
title,
same,
total,
GREEN if ok else RED,
f"{same / total:.1%} av ikke-tomme linjer uendret",
)
# ---------------------------------------------------------------------------------------------
# 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 _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))
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 = ""
undeclared_ids: tuple[str, ...] = field(default=())
def measure_stress(
evidence: Mapping[str, Any], repo_root: Path, stress_root: Path, bundle_root: Path
) -> 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 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")
base = bundle_root / chosen[0]["name"]
try:
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"]
commissioned = sum(
len(load_mandate(repo_root / c / "mandate.json").approaches) for c in contexts
)
return StressMeasure(
validated=len(validated),
undeclared=len(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),
undeclared_ids=tuple(a.approach_id for a in undeclared),
)
def score_undeclared(
probes: Sequence[str], outcomes: Mapping[str, str], m: StressMeasure, label: str
) -> Row:
failing = [
f"{n.split('::')[-1]}={outcomes.get(n, 'missing')}"
for n in probes
if outcomes.get(n) != "passed"
]
title = "6 validert UTEN erklært krav (tilnærmingens egen)"
exceptions = [f"probe {x}" for x in failing]
if m.missing:
reason = f"{label}: ikke målt, artefakter mangler ({m.missing})"
k: int | None = None
n: int | None = None
else:
k, n = m.undeclared, m.validated
reason = (
f"{label} ({m.where}): {k} av {n} validerte uten erklæring fra tilnærmingen; "
f"{m.undeclared_anywhere} uten noen erklæring i kjøringen"
)
exceptions += [f"validert uten erklæring: {a}" for a in m.undeclared_ids]
red = bool(failing) or bool(k)
if not probes:
red, exceptions = True, [*exceptions, "ingen probe registrert"]
return Row(
"undeclared", title, k, n, RED if red else GREEN, reason, exceptions=tuple(exceptions)
)
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,
DIAGNOSIS,
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
or Path(
os.environ.get("PORTFOLIO_VEGNORMAL_ROOT", "~/repos/vegnormal-okf/build/ferdig")
).expanduser(),
)
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"])),
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("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 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="der stressrundens kunnskapsbaser er montert"
)
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")
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, "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())

402
tests/test_v1_gate.py Normal file
View file

@ -0,0 +1,402 @@
"""The v1 gate's own tests: every row CAN go green and CAN go red.
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
(``probe_runner`` / ``stress_measure``) so the logic is exercised without a child pytest; one
subprocess arm runs the real command end to end.
"""
from __future__ import annotations
import ast
import json
import subprocess
import sys
from pathlib import Path
from typing import Any
import pytest
from portfolio_optimiser.evals import v1_gate as gate
_REPO = Path(__file__).resolve().parents[1]
_CONFIG = gate.load_config()
_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(
_CONFIG["row6_evidence"]
)
def _write(path: Path, payload: Any) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
text = payload if isinstance(payload, str) else json.dumps(payload, ensure_ascii=False)
path.write_text(text, encoding="utf-8")
def _feedback(round_dir: Path, *items: tuple[str, int, str], author: str = "fagperson") -> None:
_write(
round_dir / "feedback.json",
{"author": author, "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:
_write(
round_dir / "outcome.json",
{"run_id": f"r{round_dir.name}", "approaches": rows, "removed": list(removed)},
)
def _row(aid: str, validated: bool, nok: float | None, *ids: str, stage: str = "") -> dict:
return {
"id": aid,
"validated": validated,
"stage": stage,
"validated_nok": nok,
"feedback_ids": list(ids),
}
def _green_rounds(root: Path) -> Path:
"""Three traced rounds and a round 3 report kept at 100 %."""
_outcome(root / "0", [_row("a1", False, None, stage="stage0")])
for n in (1, 2, 3):
_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}")])
_write(root / str(n) / "report.md", f"# Rapport {n}\n\nlinje\n")
_write(root / "3" / "report.kept.md", "# Rapport 3\n\nlinje\n")
return root
def _all_pass(ids: Any) -> dict[str, str]:
return {n: "passed" for n in ids}
_CLEAN = gate.StressMeasure(
validated=10, undeclared=0, named=1, rows=20, commissioned=20, where="x"
)
# ---------------------------------------------------------------------------------------------
# Row 1 — rounds with a real domain expert
# ---------------------------------------------------------------------------------------------
def test_row1_is_red_with_no_rounds_and_green_with_three(tmp_path: Path) -> None:
assert gate.score_rounds(tmp_path / "none", 3, _AI).k == 0
assert gate.score_rounds(tmp_path / "none", 3, _AI).status == gate.RED
row = gate.score_rounds(_green_rounds(tmp_path / "r"), 3, _AI)
assert (row.k, row.status) == (3, gate.GREEN)
def test_row1_an_empty_or_wrongly_shaped_feedback_does_not_count(tmp_path: Path) -> None:
root = _green_rounds(tmp_path)
_write(root / "1" / "feedback.json", {"author": "fagperson", "items": []})
(root / "2" / "feedback.json").unlink()
_write(root / "2" / "feedback.md", "Dette er min tilbakemelding.")
_feedback(root / "3", ("f3", 1, "noe"), author="")
row = gate.score_rounds(root, 3, _AI)
assert row.k == 0
assert any("feedback.md" in x for x in row.exceptions)
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
the expert's own words in the same shape IS counted."""
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)
root = _green_rounds(tmp_path)
_feedback(root / "1", ("f1", 1, lifted.lstrip("> ")))
_feedback(root / "2", ("f2", 1, "Se her: " + lifted.lstrip("> ") + " Takk."))
row = gate.score_rounds(root, 3, _AI)
assert row.k == 1
assert sum("AI-forfattet" in x for x in row.exceptions) == 2
# The guard fails closed when it cannot read its source.
assert gate.score_rounds(root, 3, None).k == 0
# ---------------------------------------------------------------------------------------------
# Row 2 — rounds with a measurable change
# ---------------------------------------------------------------------------------------------
def test_row2_a_traced_change_counts(tmp_path: Path) -> None:
row = gate.score_changes(_green_rounds(tmp_path), 3, _AI)
assert (row.k, row.status) == (3, gate.GREEN)
assert "runde 0 =" in row.reason and "(r0)" in row.reason
def test_row2_a_change_without_a_trace_is_model_noise(tmp_path: Path) -> None:
root = _green_rounds(tmp_path)
_outcome(root / "2", [_row("a1", True, 2000.0)]) # changed nok, no feedback id
_outcome(root / "3", [_row("a1", True, 3000.0, "f1")]) # traced to an EARLIER round's id
row = gate.score_changes(root, 3, _AI)
assert (row.k, row.status) == (1, gate.RED)
assert sum("ingen sporet" in x for x in row.exceptions) == 2
def test_row2_no_change_does_not_count(tmp_path: Path) -> None:
root = _green_rounds(tmp_path)
_outcome(root / "2", [_row("a1", True, 1000.0, "f2")]) # identical to round 1
row = gate.score_changes(root, 3, _AI)
assert row.k == 2
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:
ids = {"f"}
base = [_row("a1", False, None, stage="stage0")]
cases = {
"a-added": [*base, _row("a2", False, None, "f", stage="stage0")],
"b-validated": [_row("a1", True, None, "f", stage="stage0")],
"c-stage": [_row("a1", False, None, "f", stage="stage4")],
"d-nok": [_row("a1", False, 5.0, "f", stage="stage0")],
}
for name, rows in cases.items():
_outcome(tmp_path / name / "0", base)
_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)
# (a) by removal: traced only through the ``removed`` list.
_outcome(tmp_path / "rm" / "0", [*base, _row("a2", False, None)])
_outcome(tmp_path / "rm" / "1", base)
assert not gate.round_changed(
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]
# ---------------------------------------------------------------------------------------------
# Row 3 — feedback types
# ---------------------------------------------------------------------------------------------
def test_row3_all_probes_passing_is_eight_of_eight() -> None:
row = gate.score_types(_CONFIG["feedback_types"], _all_pass(_ALL_NODEIDS))
assert (row.k, row.n, row.status) == (8, 8, gate.GREEN)
def test_row3_partial_is_no_and_a_missing_probe_is_red() -> None:
types = json.loads(json.dumps(_CONFIG["feedback_types"]))
outcomes = _all_pass(_ALL_NODEIDS)
outcomes[types["7"]["evidence"][1]] = "failed" # one of type 7's two tests
types["4"]["evidence"] = []
outcomes[types["1"]["evidence"][0]] = "missing"
row = gate.score_types(types, outcomes)
assert (row.k, row.status) == (5, gate.RED)
assert {x.split()[1] for x in row.exceptions} == {"1", "4", "7"}
def test_row3_every_registered_test_exists() -> None:
"""A renamed test would silently turn a type red; this names the drift instead."""
for nodeid in _ALL_NODEIDS:
path, name = nodeid.split("::")
tree = ast.parse((_REPO / path).read_text(encoding="utf-8"))
names = {n.name for n in tree.body if isinstance(n, ast.FunctionDef | ast.AsyncFunctionDef)}
assert name in names, nodeid
# ---------------------------------------------------------------------------------------------
# Row 4 — round 3 kept
# ---------------------------------------------------------------------------------------------
@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:
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)]
_write(tmp_path / "3" / "report.kept.md", "\n".join(kept) + "\n")
row = gate.score_kept(tmp_path, 0.8)
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:
_write(tmp_path / "3" / "report.md", "a\nb\n")
row = gate.score_kept(tmp_path, 0.8)
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:
_write(tmp_path / "3" / "report.md", "x\nx\ny\n")
_write(tmp_path / "3" / "report.kept.md", "x\nz\n")
assert gate.score_kept(tmp_path, 0.8).k == 1
# ---------------------------------------------------------------------------------------------
# Row 5 — MAF points
# ---------------------------------------------------------------------------------------------
def test_row5_is_red_until_the_operator_approves_the_list() -> None:
assert _CONFIG["maf_points"]["approved"] is False
verdicts = gate.green_types(_CONFIG["feedback_types"], _all_pass(_ALL_NODEIDS))
row = gate.score_maf(_CONFIG["maf_points"], verdicts, gate._PACKAGE_SRC)
assert (row.k, row.status, row.reason) == (0, gate.RED, "M ikke godkjent av operatøren")
assert "presence 7 av 8" in row.diagnostics
def _synthetic_src(tmp: Path, *, comment_only: bool = False) -> Path:
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)
return tmp
def _one_point(types: list[int]) -> dict[str, Any]:
point = {
"u_id": "U5",
"construct": "SkillsProvider",
"package": "agent_framework",
"callsite": {"module": "skills.py", "scope": "build"},
"types": types,
}
return {"approved": True, "points": [point]}
def test_row5_an_approved_point_counts_only_with_a_green_type(tmp_path: Path) -> None:
src = _synthetic_src(tmp_path)
green = gate.score_maf(_one_point([6]), {6: ""}, src)
assert (green.k, green.status) == (1, gate.GREEN)
red = gate.score_maf(_one_point([6]), {6: "failed"}, src)
assert (red.k, red.status) == (0, gate.RED)
def test_row5_a_comment_is_not_a_call_site(tmp_path: Path) -> None:
src = _synthetic_src(tmp_path, comment_only=True)
row = gate.score_maf(_one_point([6]), {6: ""}, src)
assert row.k == 0
assert "presence 0 av 1" in row.diagnostics
def test_row5_real_call_sites_are_found_by_ast() -> None:
found = {
p["u_id"]: gate.maf_presence(p, gate._PACKAGE_SRC) for p in _CONFIG["maf_points"]["points"]
}
assert found.pop("U5") == (False, False)
assert set(found.values()) == {(True, True)}
# ---------------------------------------------------------------------------------------------
# Rows 6 and 7 — the stress artefacts
# ---------------------------------------------------------------------------------------------
_PROBES = list(_CONFIG["row6_evidence"])
def test_row6_green_needs_both_the_probes_and_zero_undeclared() -> None:
row = gate.score_undeclared(_PROBES, _all_pass(_PROBES), _CLEAN, "s")
assert (row.k, row.n, row.status) == (0, 10, gate.GREEN)
dirty = gate.StressMeasure(validated=10, undeclared=3, undeclared_ids=("a4",), where="x")
assert gate.score_undeclared(_PROBES, _all_pass(_PROBES), dirty, "s").status == gate.RED
failing = {**_all_pass(_PROBES), _PROBES[1]: "failed"}
assert gate.score_undeclared(_PROBES, failing, _CLEAN, "s").status == gate.RED
def test_row6_missing_artefacts_are_never_zero() -> None:
row = gate.score_undeclared(
_PROBES, _all_pass(_PROBES), gate.StressMeasure(missing="ut finnes ikke"), "s"
)
assert row.k is None
assert "ikke målt, artefakter mangler" in row.reason
assert "– av –" in row.line()
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."""
evidence = _CONFIG["stress_evidence"]
root = _REPO / evidence["root"]
bundles = Path("~/repos/vegnormal-okf/build/ferdig").expanduser()
if not root.is_dir() or not bundles.is_dir():
m = gate.measure_stress(evidence, _REPO, tmp_path / "absent", bundles)
assert m.missing and m.validated == 0
pytest.skip(f"stress artefacts not mounted ({root}, {bundles})")
m = gate.measure_stress(evidence, _REPO, root, bundles)
if m.missing:
# The mount belongs to another repository and can be mid-rebuild; the gate then says
# "ikke målt", which test_row6_missing_artefacts_are_never_zero already pins.
assert m.validated == 0
pytest.skip(f"stress artefacts not judgeable right now: {m.missing}")
assert (m.validated, m.undeclared, m.named, m.commissioned) == (10, 10, 1, 20)
def test_row7_is_a_diagnosis_and_never_moves_the_exit_code() -> None:
row = gate.score_named(_CLEAN, "s")
assert (row.k, row.n, row.status, row.failing) == (1, 20, gate.DIAGNOSIS, False)
assert gate.NAMED_WARNING in row.diagnostics
assert gate.exit_code([row]) == 0
# ---------------------------------------------------------------------------------------------
# The whole gate
# ---------------------------------------------------------------------------------------------
def test_every_failing_row_green_is_exit_zero_and_one_red_is_exit_one(tmp_path: Path) -> None:
config = json.loads(json.dumps(_CONFIG))
config["maf_points"]["approved"] = True
config["maf_points"]["points"] = _one_point([6])["points"]
rows = gate.evaluate(
rounds_dir=_green_rounds(tmp_path),
config=config,
repo_root=_REPO,
src=_synthetic_src(tmp_path / "src"),
probe_runner=_all_pass,
stress_measure=_CLEAN,
)
assert [r.status for r in rows[:6]] == [gate.GREEN] * 6, gate.render(rows)
assert gate.exit_code(rows) == 0
(tmp_path / "3" / "report.kept.md").unlink()
rows = gate.evaluate(
rounds_dir=tmp_path,
config=config,
repo_root=_REPO,
src=tmp_path / "src",
probe_runner=_all_pass,
stress_measure=_CLEAN,
)
assert gate.exit_code(rows) == 1
def test_the_default_rounds_dir_is_gitignored() -> None:
proc = subprocess.run(
["git", "check-ignore", "-q", gate.DEFAULT_ROUNDS_DIR + "/1/feedback.json"], cwd=_REPO
)
assert proc.returncode == 0
def _cli(*args: str) -> subprocess.CompletedProcess[str]:
return subprocess.run(
[sys.executable, "-m", "portfolio_optimiser.evals.v1_gate", *args],
cwd=_REPO,
capture_output=True,
text=True,
)
def test_wrong_usage_is_exit_two(tmp_path: Path) -> None:
assert _cli("--rounds-dir", str(tmp_path / "missing")).returncode == 2
assert _cli("--no-such-flag").returncode == 2
help_text = _cli("--help").stdout
assert "report.kept.md" in help_text and "feedback_ids" in help_text
def test_the_command_is_red_today_with_every_row_in_its_output(tmp_path: Path) -> None:
proc = _cli("--rounds-dir", str(tmp_path), "--json")
assert proc.returncode == 1, proc.stderr
payload = json.loads(proc.stdout)
assert payload["exit"] == 1
rows = {r["key"]: r for r in payload["rows"]}
assert list(rows) == ["rounds", "changes", "types", "kept", "maf", "undeclared", "named"]
assert (rows["rounds"]["k"], rows["changes"]["k"]) == (0, 0)
assert (rows["types"]["k"], rows["types"]["n"]) == (3, 8)
assert rows["kept"]["status"] == gate.RED
assert rows["maf"]["reason"] == "M ikke godkjent av operatøren"
assert rows["undeclared"]["status"] == gate.RED
assert rows["named"]["failing"] is False

160
tests/test_v1_probes.py Normal file
View file

@ -0,0 +1,160 @@
"""v1 gate probes — the named tests the v1 gate (``python -m portfolio_optimiser.evals.v1_gate``)
runs to decide two of its rows. Every test here that is RED today carries
``xfail(strict=True)``, so the ordinary suite stays green while the gap is real, and the gate runs
the file with ``--runxfail`` so the gap shows as red there. ``strict`` is the other half: the day a
capability makes one of these pass, the suite goes RED on the XPASS until the marker is removed —
a closed gap cannot stay labelled open.
**Row 3 (feedback types with a way in AND an action).** Types 1, 3 and 7 are proven by EXISTING
tests elsewhere in the suite (registered by node id in ``evals/v1_gate.json``). The five types with
no complete surface (2, 4, 5, 6, 8) get a probe here that is red BECAUSE the surface is missing,
never a missing test. Each probe measures the absence (the CLI's own ``--help``); if a matching
option appears it STILL fails, naming the option — "partial is no", and a door with no observed
action is exactly partial. Such a probe goes green only when it is rewritten to drive the new door
and observe what it does.
**Row 6 (a validated proposal whose approach declared no requirement).** Two probes against the
real ``run_project``: no declaration anywhere, and a declaration made by the RUN (the debate) but
not by the approach. The second is the reading the gate measures the stress outboxes with: a
run-level declaration cannot be attributed to one approach (the judge labels it ``run``), so it
does not count as the approach having declared anything.
"""
from __future__ import annotations
import contextlib
import io
import json
import re
from pathlib import Path
from typing import Any
import pytest
from portfolio_optimiser import okf, run
from portfolio_optimiser.mandate import Approach, Mandate
from portfolio_optimiser.run import run_project
from portfolio_optimiser.simulation import scripted_factory
from portfolio_optimiser.verdicts import VerdictStore
_BUNDLE = Path(__file__).resolve().parents[1] / "shared" / "examples" / "bygg-energi-mikro"
_BASE_ID = "bygg-energi-mikro"
_PID = "BYGG-KONTOR-NORD"
_VALID_REPLY = (
'{"measure":"LED-retrofit av kontorbelysning","affected_items":'
'[{"code":"ENERGI-TOTAL-EL","quantity":300000,"unit_cost":1.0}],"claimed_saving_nok":30000}'
)
_CHECKER_REPLY = "Reasoning holds.\nVERDICT: APPROVE"
_NO_SURFACE = "v1 probe: no surface"
_PARTIAL = "v1 probe: surface without an observed action"
def _cli_options() -> set[str]:
"""Every option string the CLI's own ``--help`` prints — the surface, measured."""
buffer = io.StringIO()
with contextlib.redirect_stdout(buffer), pytest.raises(SystemExit):
run.main(["--help"])
return set(re.findall(r"--[a-z][a-z-]*", buffer.getvalue()))
def _surface_or_fail(type_no: int, what: str, keywords: tuple[str, ...]) -> None:
hits = sorted(o for o in _cli_options() if any(k in o for k in keywords))
if not hits:
pytest.fail(f"{_NO_SURFACE}: type {type_no} ({what}) — no CLI option matches {keywords}")
pytest.fail(
f"{_PARTIAL}: type {type_no} ({what}) — {hits} appeared; rewrite this probe to drive it "
"and observe the action"
)
# ---------------------------------------------------------------------------------------------
# Row 3 — the five types without a complete surface
# ---------------------------------------------------------------------------------------------
@pytest.mark.xfail(strict=True, reason="type 2: no typed removal; only revise free text")
def test_type_2_remove_a_direction_has_a_typed_door() -> None:
_surface_or_fail(2, "take a direction away", ("drop", "remove", "exclude", "withdraw"))
@pytest.mark.xfail(strict=True, reason="type 4: no surface relaxes a requirement")
def test_type_4_relax_a_requirement_has_a_door() -> None:
_surface_or_fail(4, "relax a requirement", ("relax", "waive", "loosen"))
@pytest.mark.xfail(strict=True, reason="type 5: concept graph edits have no CLI door")
def test_type_5_edit_the_concept_graph_has_a_door() -> None:
_surface_or_fail(5, "edit the concept graph", ("promote", "concept", "graph"))
@pytest.mark.xfail(strict=True, reason="type 6: no skills flag")
def test_type_6_skills_per_analysis_has_a_door() -> None:
_surface_or_fail(6, "skills per analysis", ("skill",))
@pytest.mark.xfail(strict=True, reason="type 8: no door for inline context such as meeting notes")
def test_type_8_inline_context_has_a_door() -> None:
_surface_or_fail(8, "inline context", ("note", "minutes", "inline", "attach"))
# ---------------------------------------------------------------------------------------------
# Row 6 — a validated proposal must rest on a declaration its approach made
# ---------------------------------------------------------------------------------------------
def _mandate() -> Mandate:
return Mandate(
objective="Kutt energikostnad",
approaches=(Approach(id="a1", label="LED-retrofit", description="expert's reason"),),
)
async def _statuses(script: dict[str, Any], tmp_path: Path) -> dict[str, str]:
result = await run_project(
_PID,
"local",
docs_dir=str(_BUNDLE),
bundle_dir=str(_BUNDLE),
store=VerdictStore(verdicts=[]),
client_factory=scripted_factory(script, []),
mandate=_mandate(),
outbox_dir=str(tmp_path),
run_id="v1-row6",
)
return {row.id: row.status for row in result.coverage}
@pytest.mark.xfail(strict=True, reason="row 6: no stage refuses a validation with no declaration")
@pytest.mark.asyncio
async def test_row6_an_approach_that_declared_nothing_cannot_be_validated(tmp_path: Path) -> None:
statuses = await _statuses({"proposer": _VALID_REPLY, "checker": _CHECKER_REPLY}, tmp_path)
debate = json.loads((tmp_path / "v1-row6-debate.json").read_text(encoding="utf-8"))
assert debate["requirements"] == [] # precondition: nothing was declared anywhere
assert statuses["a1"] != "validated", "validated without any declared requirement"
@pytest.mark.xfail(strict=True, reason="row 6: a run-level declaration still stands in")
@pytest.mark.asyncio
async def test_row6_a_run_level_declaration_does_not_stand_in_for_the_approach(
tmp_path: Path,
) -> None:
concepts = [f.name for f in okf.navigate_bundle(str(_BUNDLE)).context_files][:3]
script = {
"proposer": [
*({"call": "read_file", "args": {"bundle_id": _BASE_ID, "path": n}} for n in concepts),
{
"call": "declare_requirement",
"args": {"bundle_id": _BASE_ID, "path": concepts[0], "ref": "probe"},
},
_VALID_REPLY,
_VALID_REPLY,
_VALID_REPLY,
_VALID_REPLY,
],
"checker": _CHECKER_REPLY,
}
statuses = await _statuses(script, tmp_path)
debate = json.loads((tmp_path / "v1-row6-debate.json").read_text(encoding="utf-8"))
assert [r["path"] for r in debate["requirements"]] == [concepts[0]] # precondition
assert statuses["a1"] != "validated", "validated on a declaration the approach never made"