"""P16 DEL A - the STRESS JUDGE: session 102's hand-read criterion made deterministic. **The measured silence.** Session 102 adjudicated "did the run work against the base it said it would" BY HAND: (a) the run builds on the right fasit concept, or refuses anchored * (b') it NAMES that concept * (c) zero hallucinations. Measured 14.09: nothing in the tree read ``contexts//fasit.json`` against an outbox, so the criterion had no repeatable form. Four runs now and N runs later cannot rest on a reading somebody did once. This module reads ONLY artefacts that already exist - it adds no new field to any run: * ``{run_id}[-{approach_id}]-proposal.json`` - the candidate IR + ``provenance.citations`` * ``{run_id}[-{approach_id}]-outcome.json`` - ``validated`` / ``rejected`` * ``{run_id}-debate.json`` - ``tool_calls[]`` (``name``/``bundle_id``/``path``, S2c) and the set's own ``mandate.json`` / ``fasit.json`` / ``bundle.txt``. **THE ORDER'S (a) WAS VACUOUS AS WRITTEN, AND THE DEVIATION IS MEASURED, NOT CHOSEN.** The order defines grounded as "a ``must_cite`` path was OPENED *or* CITED". But on the S2c navigation path ``run_project`` stamps ``citations = bundle_citations(bundle)``, which is one citation PER CONTEXT FILE - the whole corpus. Measured on n100-2023: 446 context files, 446 citations, and **6 of 6 fasit paths already "cited" before a single model call**. A judge honouring that literally would be a gate that can only be green - the repo's own vacuous-gate class, inside the gate built to catch it. So a CITATION grounds an approach only when the citation list is NARROWER than the base (a declared pre-pass cut, where the stamp really does name what was read). Both halves are reported either way (``opened`` / ``cited`` / ``citation_scope``), so which one fired stays readable. **(b') was checked for the same vacuity and is CLEAN for the N corpora, so the order stands.** ``bundle_citations`` snippets are concept BODIES while ``ref``/``title`` live in FRONTMATTER: measured 0 of 446 n100 bodies contain ``Krav 4.1.2-1``. ``named_in_measure`` / ``named_in_snippet`` are still reported apart, because the measure is the model's own prose and a snippet is the base's. **P18/C2 (PM decision, P16 § 6.2): the snippet arm counts only under a NARROWED scope, as (a) does.** The paragraph above holds for a reference like ``Krav 4.1.2-1``, which no body repeats — it does NOT hold for R761, where a process number such as ``12.1`` stands in the bodies themselves. Under a whole-base citation list that mark is "cited" before any model call, so the row was ``named`` for a run in which the model had said nothing of the kind. The scope gate is the same correction (a) already carries, applied to the half that was still exposed. **A DENOMINATOR, ALWAYS** (Verifiseringsloven ansikt 4). Every verdict names how many tool calls, citations, approach rows and base concepts it saw, and an outbox with no proposal artefact - or a base that scans to no concepts - RAISES ``EmptyMeasurement`` instead of reporting "0 hallucinations". An empty measurement is not a clean bill of health. **``not_evaluated`` is ABSENCE, measured.** ``run_project`` writes one artefact pair per EVALUATED approach; its coverage rows (including ``not_evaluated``) are printed by ``settle`` and reach no file. So a commissioned approach with no artefact is reported as ``not_evaluated`` here rather than omitted - an omitted row is indistinguishable from an approach nobody ordered (``ApproachOutcome``'s own rule). **Hallucinated READ paths are run-level, and they poison every row.** ``{run_id}-debate.json`` is written once per run, so a guessed path cannot be attributed to one approach. Folding it into each row's ``ferdig`` is the conservative reading of "(c) must be 0"; the per-row ``hallucinations`` tuple carries only what IS attributable (that row's citations and codes). **a4 / ``must_refuse`` is the falsification half in D-1 form.** ``po`` is not a lookup tool, so an "unanswerable question" has no runnable form - but a commissioned approach whose GROUND the base does not carry does. It passes iff no ``validated`` row is that approach AND no validated proposal anywhere carries one of its codes. Whether the model SAID the base does not carry it is read by hand and reported separately, marked manual. **Framework-free** (stdlib + the MAF-free ``okf`` / ``mandate`` leaves), and it is its OWN module rather than a branch of ``run.py``: no partition row in ``main()`` is touched, and a judge that cannot start a run cannot accidentally cost anything. """ from __future__ import annotations import argparse import json import os import sys from dataclasses import asdict, dataclass from pathlib import Path from typing import Any from portfolio_optimiser import okf from portfolio_optimiser.mandate import Mandate, load_mandate #: Where the vegnormal bases are mounted, unless ``--bundle-root`` says otherwise. Read at CALL #: time (the ``shared_root()`` idiom) so a test or an operator can move the mount without a reimport. _DEFAULT_BUNDLE_ROOT = "~/repos/vegnormal-okf/build/ferdig" class EmptyMeasurement(RuntimeError): """A measurement with no denominator - never reported as a clean result. Raised when the outbox holds no proposal artefact at all, or when the base scans to zero concepts. Both are the ansikt-4 failure: "found nothing" is a measurement result, and reporting it as "0 hallucinations, everything clean" would turn an instrument failure into a fact about the world.""" @dataclass(frozen=True) class ApproachVerdict: """One commissioned approach, judged against the set's fasit.""" approach_id: str label: str status: str # "validated" | "rejected" | "not_evaluated" #: (a) - grounded by an OPENED path, or by a citation under a NARROWED list. See the module #: docstring: a whole-base citation list is stamped before any model call and grounds nothing. grounded: bool opened: tuple[str, ...] cited: tuple[str, ...] citation_scope: str # "whole-base" | "narrowed" | "absent" #: (b') - the fasit's ``ref`` or ``title`` occurs in the model's own ``measure`` (strong) or in #: one of this approach's citation snippets (weaker, but measured non-vacuous). named: bool named_in_measure: bool named_in_snippet: bool #: (c) - attributable hallucinations, ``citation:`` / ``code:``. hallucinations: tuple[str, ...] ferdig: bool @dataclass(frozen=True) class RefusalVerdict: """One ``must_refuse`` approach (a4): the falsification arm.""" approach_id: str passed: bool detail: str @dataclass(frozen=True) class ContextSetVerdict: """The whole set's verdict, denominators included.""" context_set: str run_id: str bundle_id: str approaches: tuple[ApproachVerdict, ...] must_refuse: tuple[RefusalVerdict, ...] #: Run-level: read paths the base does not carry (guessed by the navigator). hallucinated_reads: tuple[str, ...] tool_calls_seen: int citations_seen: int approach_rows_seen: int concepts_in_base: int ferdig: bool def to_payload(self) -> dict[str, Any]: """Byte-stable plain data: the ONE rendering, shared by the CLI's stdout and its file.""" return asdict(self) def _read_json(path: Path) -> dict[str, Any]: return json.loads(path.read_text(encoding="utf-8")) # type: ignore[no-any-return] def _artefacts(outbox: Path, run_id: str, approach_id: str) -> tuple[Path, Path]: """The proposal/outcome pair for one approach - the per-approach key when a mandate was run, the bare ``run_id`` key when it was not (``outbox.write_outbox``'s own two forms).""" stem = f"{run_id}-{approach_id}" proposal = outbox / f"{stem}-proposal.json" if not proposal.is_file(): proposal = outbox / f"{run_id}-proposal.json" outcome = outbox / f"{stem}-outcome.json" if not outcome.is_file(): outcome = outbox / f"{run_id}-outcome.json" return proposal, outcome def _inside(base: Path, raw: str) -> Path | None: """Resolve a model-supplied path inside the base, or ``None`` when it escapes.""" try: resolved = (base / raw).resolve() except (OSError, ValueError): return None root = base.resolve() return resolved if resolved == root or root in resolved.parents else None def score_context_set( context_dir: str | Path, outbox_dir: str | Path, run_id: str, bundle_dir: str | Path, ) -> ContextSetVerdict: """Judge ONE context set against ONE outbox. ``bundle_dir`` is the MOUNTED base itself (the CLI resolves it from ``--bundle-root`` plus the set's own ``bundle.txt`` name).""" context = Path(context_dir) outbox = Path(outbox_dir) base = Path(bundle_dir) fasit = _read_json(context / "fasit.json") mandate: Mandate = load_mandate(context / "mandate.json") bundle = okf.navigate_bundle(str(base)) concept_names = {f.name for f in bundle.context_files} if not concept_names: raise EmptyMeasurement( f"{base} scanned to 0 concepts - nothing was measured, so no verdict is honest" ) baseline = okf.load_optional_cost_baseline(str(base)) baseline_codes = set(baseline.items) if baseline is not None else set() must_cite = {row["approach_id"]: row.get("concepts", []) for row in fasit.get("must_cite", [])} refuse_ids = {row["approach_id"] for row in fasit.get("must_refuse", [])} # ---- run-level trace --------------------------------------------------------------------- debate = outbox / f"{run_id}-debate.json" tool_calls: list[dict[str, Any]] = ( list(_read_json(debate).get("tool_calls", [])) if debate.is_file() else [] ) opened_paths = {c.get("path", "") for c in tool_calls if c.get("name") == "read_file"} opened_paths.discard("") hallucinated_reads: list[str] = [] for call in tool_calls: raw = str(call.get("path", "")) if not raw: continue target = _inside(base, raw) name = call.get("name") if name == "read_file" and (target is None or not target.is_file()): hallucinated_reads.append(raw) elif name == "read_dir" and (target is None or not target.is_dir()): hallucinated_reads.append(raw) reads_clean = not hallucinated_reads # ---- per approach ------------------------------------------------------------------------ rows: list[ApproachVerdict] = [] citations_seen = 0 rows_seen = 0 validated_codes: set[str] = set() validated_ids: set[str] = set() for approach in mandate.approaches: proposal_path, outcome_path = _artefacts(outbox, run_id, approach.id) concepts = must_cite.get(approach.id, []) wanted = {c["path"] for c in concepts} if not proposal_path.is_file() or not outcome_path.is_file(): rows.append( ApproachVerdict( approach_id=approach.id, label=approach.label, status="not_evaluated", grounded=False, opened=tuple(sorted(wanted & opened_paths)), cited=(), citation_scope="absent", named=False, named_in_measure=False, named_in_snippet=False, hallucinations=(), ferdig=False, ) ) continue rows_seen += 1 payload = _read_json(proposal_path) proposal = payload.get("proposal", {}) citations = payload.get("provenance", {}).get("citations", []) citations_seen += len(citations) status = str(_read_json(outcome_path).get("outcome_type", "rejected")) if status == "validated": validated_ids.add(approach.id) validated_codes.update( str(item.get("code", "")) for item in proposal.get("affected_items", []) ) cited_files = {str(c.get("file", "")) for c in citations} scope = ( "absent" if not cited_files else "whole-base" if concept_names <= cited_files else "narrowed" ) opened = tuple(sorted(wanted & opened_paths)) cited = tuple(sorted(wanted & cited_files)) # A whole-base list is stamped before any model work - see the module docstring. grounded = bool(opened) or (bool(cited) and scope == "narrowed") measure = str(proposal.get("measure", "")) snippets = " ".join(str(c.get("snippet", "")) for c in citations) marks = [m for c in concepts for m in (c.get("ref", ""), c.get("title", "")) if m] named_in_measure = any(m in measure for m in marks) # P18/C2 (PM decision, P16 § 6.2): the snippet arm counts ONLY under a narrowed citation # scope, exactly as (a) does. A whole-base citation list is stamped by ``bundle_citations`` # before a single model call — measured on n100, 446 context files, 446 citations, 6 of 6 # fasit paths "cited" for free — so a mark found in THOSE snippets is evidence about the # base's contents, not about this run. Measured on r761: ``12.1`` appears in whole-base # snippets and gave this row ``named`` without the model having said anything. named_in_snippet = scope == "narrowed" and any(m in snippets for m in marks) halluc = [f"citation:{f}" for f in sorted(cited_files - concept_names)] allowed = set(approach.affected_codes) | baseline_codes halluc += [ f"code:{code}" for code in sorted( {str(i.get("code", "")) for i in proposal.get("affected_items", [])} - allowed ) ] rows.append( ApproachVerdict( approach_id=approach.id, label=approach.label, status=status, grounded=grounded, opened=opened, cited=cited, citation_scope=scope, named=named_in_measure or named_in_snippet, named_in_measure=named_in_measure, named_in_snippet=named_in_snippet, hallucinations=tuple(halluc), ferdig=( grounded and (named_in_measure or named_in_snippet) and not halluc and reads_clean ), ) ) if rows_seen == 0: raise EmptyMeasurement( f"{outbox} holds no proposal artefact for run {run_id!r} - " "an empty outbox is not a clean result" ) # ---- the falsification arm --------------------------------------------------------------- refusals: list[RefusalVerdict] = [] for row in fasit.get("must_refuse", []): rid = row["approach_id"] commissioned = next((a for a in mandate.approaches if a.id == rid), None) codes = set(commissioned.affected_codes) if commissioned is not None else set() leaked = sorted(codes & validated_codes) if rid in validated_ids: refusals.append( RefusalVerdict( rid, False, f"{rid} was VALIDATED - the base carries no ground for it" ) ) elif leaked: refusals.append( RefusalVerdict( rid, False, f"{rid}'s codes {leaked} ride inside a validated proposal" ) ) else: refusals.append(RefusalVerdict(rid, True, "no validated outcome rests on it")) judged = [r for r in rows if r.approach_id not in refuse_ids] return ContextSetVerdict( context_set=context.name, run_id=run_id, bundle_id=str(fasit.get("bundle_id", "")), approaches=tuple(rows), must_refuse=tuple(refusals), hallucinated_reads=tuple(hallucinated_reads), tool_calls_seen=len(tool_calls), citations_seen=citations_seen, approach_rows_seen=rows_seen, concepts_in_base=len(concept_names), ferdig=bool(judged) and all(r.ferdig for r in judged) and all(r.passed for r in refusals), ) def main(argv: list[str] | None = None) -> int: """``python -m portfolio_optimiser.stress --outbox-dir D --run-id R``. Writes the verdict to ``/-verdict.json`` AND prints it, so a CI reader and a human reader get the same bytes. Its own module, never a ``run.py`` mode: no partition row is touched and this entry point cannot start a paid run.""" parser = argparse.ArgumentParser(prog="portfolio-optimiser-stress", description=__doc__) parser.add_argument("context_dir", help="contexts/") parser.add_argument("--outbox-dir", required=True) parser.add_argument("--run-id", required=True) parser.add_argument( "--bundle-root", default=os.environ.get("PORTFOLIO_VEGNORMAL_ROOT", _DEFAULT_BUNDLE_ROOT), help="directory the set's bundle.txt name is mounted under", ) args = parser.parse_args(argv) context = Path(args.context_dir) declared = dict( line.split(":", 1) # type: ignore[misc] for line in (context / "bundle.txt").read_text(encoding="utf-8").splitlines() if ":" in line ) base = Path(args.bundle_root).expanduser() / declared["name"].strip() try: verdict = score_context_set(context, args.outbox_dir, args.run_id, base) except EmptyMeasurement as exc: print(f"stress refused: {exc}", file=sys.stderr) return 1 payload = json.dumps(verdict.to_payload(), sort_keys=True, indent=2, ensure_ascii=False) + "\n" out = Path(args.outbox_dir) / f"{args.run_id}-verdict.json" out.write_text(payload, encoding="utf-8") print(payload, end="") return 0 if __name__ == "__main__": # pragma: no cover - exercised by a subprocess test raise SystemExit(main())