"""Independent evaluator for the post-training model-convergence protocol. This module is deliberately data-only: it reads JSON manifests, result records and stored prediction records. It never imports an adapter, optional detection package, dataset, checkpoint, or model. A result is useful only when all of the frozen matrix and sealing invariants can be demonstrated from the saved evidence. """ from __future__ import annotations import argparse import hashlib import json import math import os import random import sys import tempfile from pathlib import Path from typing import Any, Iterable, Mapping, Sequence EVALUATOR_VERSION = "POST-TRAINING-MODEL-CONVERGENCE-EVALUATOR 1.0.0" PROTOCOL_VERSION = "post-training-model-convergence-1.0.0" WORKLOADS = ("cifar10_resnet18", "cifar10_resnet50", "voc_yolo11n") CLASSIFICATION_WORKLOADS = WORKLOADS[:2] DETECTION_WORKLOAD = WORKLOADS[2] BASE_SEEDS = (501, 502, 503) SWARM_SEEDS = (601, 602, 603) SPLIT_SEED = 20260908 PROJECTION_SEED = 20260909 BOOTSTRAP_SEED = 20260910 PARTICLES = 12 PRIMARY_GENERATIONS = 60 ENSEMBLE_GENERATIONS = 20 PRIMARY_QUERIES = PARTICLES * PRIMARY_GENERATIONS ENSEMBLE_QUERIES = PARTICLES * ENSEMBLE_GENERATIONS PRIMARY_RUNS = len(WORKLOADS) * len(BASE_SEEDS) * len(SWARM_SEEDS) ENSEMBLE_RUNS = len(WORKLOADS) * len(SWARM_SEEDS) TOTAL_PSO_QUERIES = PRIMARY_RUNS * PRIMARY_QUERIES + ENSEMBLE_RUNS * ENSEMBLE_QUERIES OBJECTIVE_SAMPLES = {"cifar10_resnet18": 1024, "cifar10_resnet50": 1024, "voc_yolo11n": 512} TOTAL_CANDIDATE_SAMPLES = sum( (len(BASE_SEEDS) * len(SWARM_SEEDS) * PRIMARY_QUERIES + len(SWARM_SEEDS) * ENSEMBLE_QUERIES) * OBJECTIVE_SAMPLES[w] for w in WORKLOADS ) BOOTSTRAP_RESAMPLES = 2000 BOOTSTRAP_ALPHA = 0.05 / 6.0 ISSUE_CATEGORIES = ( "schema", "provenance", "matrix", "accounting", "seal", "selection", "finite", "metrics", "plateau", "overfit", "leakage", "bootstrap", "gates", ) def _issue(issues: dict[str, list[str]], category: str, message: str) -> None: issues.setdefault(category, []).append(message) def _finite(value: Any) -> bool: return isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(float(value)) def _walk_nonfinite(value: Any, path: str = "") -> list[str]: out: list[str] = [] if isinstance(value, float) and not math.isfinite(value): out.append(path or "$") elif isinstance(value, Mapping): for key, item in value.items(): out.extend(_walk_nonfinite(item, f"{path}.{key}" if path else str(key))) elif isinstance(value, (list, tuple)): for i, item in enumerate(value): out.extend(_walk_nonfinite(item, f"{path}[{i}]")) return out def _json(path: Path) -> Any: with path.open("r", encoding="utf-8") as handle: return json.load(handle) def _canonical(value: Any) -> bytes: return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") def _sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def _atomic_json(path: Path, value: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) fd, name = tempfile.mkstemp(prefix=f".{path.name}.", dir=str(path.parent)) try: with os.fdopen(fd, "w", encoding="utf-8") as handle: json.dump(value, handle, sort_keys=True, separators=(",", ":"), ensure_ascii=False) handle.write("\n") handle.flush() os.fsync(handle.fileno()) os.replace(name, path) except BaseException: try: os.unlink(name) except OSError: pass raise def _number(value: Any, *keys: str) -> float | None: if isinstance(value, Mapping): for key in keys: candidate = value.get(key) if _finite(candidate): return float(candidate) for candidate in value.values(): found = _number(candidate, *keys) if found is not None: return found elif isinstance(value, (list, tuple)): for candidate in value: found = _number(candidate, *keys) if found is not None: return found return None def _int(value: Any, *keys: str) -> int | None: number = _number(value, *keys) if number is None or not number.is_integer(): return None return int(number) def _same(a: Any, b: Any, tol: float = 1e-9) -> bool: return _finite(a) and _finite(b) and math.isclose(float(a), float(b), rel_tol=tol, abs_tol=tol) def _resolve_json(root: Path, value: Any) -> Any: """Resolve a saved JSON prediction reference without opening model/data files.""" if isinstance(value, Mapping): for key in ("path", "file", "artifact", "prediction_artifact", "predictions_path"): ref = value.get(key) if isinstance(ref, str) and ref.lower().endswith((".json", ".jsonl", ".pt")): return _resolve_json(root, ref) return value if isinstance(value, str) and value.lower().endswith((".json", ".jsonl", ".pt")): path = (root / value).resolve() if root.resolve() not in path.parents: raise ValueError(f"prediction artifact escapes run root: {value}") if not path.is_file(): raise FileNotFoundError(path) if path.suffix == ".jsonl": return [_json_line for _json_line in (json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()) if _json_line] if path.suffix == ".pt": try: import torch return torch.load(path, map_location="cpu", weights_only=True) except (ImportError, OSError, RuntimeError, TypeError, ValueError) as exc: raise ValueError(f"unable to load weights-only prediction artifact: {path}: {exc}") from exc return _json(path) return value def _hash_manifest(root: Path, manifest: Mapping[str, Any], issues: dict[str, list[str]]) -> bool: good = True if manifest.get("protocol_version") != PROTOCOL_VERSION: _issue(issues, "provenance", f"frozen manifest protocol mismatch: {manifest.get('protocol_version')!r}") good = False if manifest.get("state") != "frozen": _issue(issues, "seal", f"frozen manifest state must be 'frozen', got {manifest.get('state')!r}") good = False declared = manifest.get("manifest_hash") payload = {key: manifest[key] for key in ("protocol_version", "config", "artifacts", "state") if key in manifest} if not isinstance(declared, str) or hashlib.sha256(_canonical(payload)).hexdigest() != declared: _issue(issues, "seal", "frozen_manifest.json self-hash mismatch") good = False artifacts = manifest.get("artifacts") if not isinstance(artifacts, Mapping) or not artifacts: _issue(issues, "schema", "frozen manifest requires a non-empty artifacts map") return False for name, expected in artifacts.items(): if not isinstance(name, str) or not isinstance(expected, str) or len(expected) != 64: _issue(issues, "schema", f"invalid frozen artifact hash declaration: {name!r}") good = False continue path = (root / name).resolve() if root.resolve() not in path.parents: _issue(issues, "seal", f"frozen artifact escapes run root: {name}") good = False elif not path.is_file(): _issue(issues, "seal", f"frozen artifact is missing: {name}") good = False elif _sha256(path) != expected: _issue(issues, "seal", f"frozen artifact hash drift: {name}") good = False return good def _config_checks(config: Any, issues: dict[str, list[str]], workload_id: str | None = None) -> None: if not isinstance(config, Mapping): _issue(issues, "schema", "missing or non-object config") return expected: dict[str, Any] = { "protocol_version": PROTOCOL_VERSION, "split_seed": SPLIT_SEED, "projection_seed": PROJECTION_SEED, "bootstrap_seed": BOOTSTRAP_SEED, "particle_count": PARTICLES, "pso_generations": PRIMARY_GENERATIONS, "residual_dimension": 64, "residual_bound": 1.0, "initial_radius": 0.25, "objective_checkpoints": [0, 10, 20, 30, 40, 50, 60], "base_seeds": list(BASE_SEEDS), "swarm_seeds": list(SWARM_SEEDS), } for key, value in expected.items(): got = config.get(key) if got != value and not (isinstance(got, tuple) and list(got) == value): _issue(issues, "provenance", f"config.{key} must be {value!r}, got {got!r}") ids = config.get("workload_ids") if ids is not None and sorted(ids) != sorted(WORKLOADS): _issue(issues, "matrix", f"config.workload_ids must be {list(WORKLOADS)!r}") if workload_id and config.get("workload_id") not in (None, workload_id): _issue(issues, "matrix", f"config.workload_id disagrees with {workload_id}") def _counter(value: Any, keys: Sequence[str]) -> int | None: if not isinstance(value, Mapping): return None for key in keys: candidate = value.get(key) if isinstance(candidate, int) and not isinstance(candidate, bool): return candidate return None def _check_leakage(result: Mapping[str, Any], issues: dict[str, list[str]], workload: str) -> None: leakage = result.get("leakage_counters") if not isinstance(leakage, Mapping): _issue(issues, "schema", f"{workload}: missing leakage_counters") leakage = {} loaded_before = leakage.get("official_test_data_loaded_before_freeze", leakage.get("test_data_loaded_before_freeze")) if loaded_before is not False: _issue(issues, "leakage", f"{workload}: official test data must be explicitly marked not loaded before freeze") evaluated_before = leakage.get("official_test_evaluations_before_freeze", leakage.get("test_evaluations_before_freeze", leakage.get("official_test_forward_passes_before_freeze"))) if evaluated_before != 0: _issue(issues, "leakage", f"{workload}: official test exposure before freeze must be explicitly zero") for key in ("official_test_data_loaded_before_freeze", "official_test_evaluations_before_freeze"): if key in result and ((key.endswith("freeze") and result[key] not in (False, 0))): _issue(issues, "leakage", f"{workload}: contradictory top-level {key}") construction = _counter(leakage, ("official_test_construction", "official_test_dataset_construction")) if construction != 1: _issue(issues, "leakage", f"{workload}: official test construction must equal one, got {construction}") forwards = _counter(leakage, ("official_test_forward_passes", "official_test_evaluations")) if forwards is None or forwards < 1: _issue(issues, "leakage", f"{workload}: official test forward/evaluation ledger must be positive, got {forwards}") confirmation = result.get("confirmation") if not isinstance(confirmation, Mapping): _issue(issues, "schema", f"{workload}: missing confirmation record") return for key in ("second_confirmation", "confirmation_repeated", "post_test_tuning", "post_test_reruns"): if confirmation.get(key) not in (None, False, 0, []): _issue(issues, "leakage", f"{workload}: forbidden repeated confirmation/tuning flag {key}") for key in ("official_test_data_loaded_before_freeze", "official_test_evaluations_before_freeze"): if key in confirmation and confirmation[key] not in (False, 0): _issue(issues, "leakage", f"{workload}: confirmation contradicts sealed pre-freeze {key}") def _record_queries(record: Mapping[str, Any]) -> tuple[int | None, int | None]: counters = record.get("counters") if isinstance(record.get("counters"), Mapping) else record queries = _int(counters, "objective_queries", "queries", "query_count", "total_queries", "evaluated_queries") samples = _int(counters, "objective_samples", "samples", "sample_evaluations", "candidate_sample_evaluations", "total_sample_evaluations") return queries, samples def _seed_from_key(key: Any) -> int | None: try: text = str(key) if text.isdigit(): return int(text) except Exception: pass return None def _collect_cells(node: Any, base: int | None = None, swarm: int | None = None) -> list[dict[str, Any]]: """Collect cells from either explicit records or base->swarm mappings.""" found: list[dict[str, Any]] = [] if isinstance(node, Mapping): b = _int(node, "base_seed") if _int(node, "base_seed") is not None else base s = _int(node, "swarm_seed") if _int(node, "swarm_seed") is not None else swarm if s is None: s = _int(node, "seed") if b is not None and s is not None and any(k in node for k in ("counters", "objective_queries", "objective_samples", "queries", "samples", "generation", "endpoint", "metrics")): item = dict(node); item.setdefault("base_seed", b); item.setdefault("swarm_seed", s); found.append(item) for key, value in node.items(): key_seed = _seed_from_key(key) if key_seed in BASE_SEEDS: found.extend(_collect_cells(value, key_seed, s)) elif key_seed in SWARM_SEEDS: found.extend(_collect_cells(value, b, key_seed)) elif key not in {"base_seed", "swarm_seed"}: found.extend(_collect_cells(value, b, s)) elif isinstance(node, list): for value in node: found.extend(_collect_cells(value, base, swarm)) return found def _collect_base_cells(node: Any, base: int | None = None) -> list[dict[str, Any]]: """Collect one record per base seed from flattened or base->record schemas.""" found: list[dict[str, Any]] = [] if isinstance(node, Mapping): b = _int(node, "base_seed") if _int(node, "base_seed") is not None else base if b is not None and any(k in node for k in ("updates", "gradient_updates", "counters", "metrics", "checkpoint", "endpoint", "objective")): item = dict(node); item.setdefault("base_seed", b); found.append(item) for key, value in node.items(): key_seed = _seed_from_key(key) if key_seed in BASE_SEEDS: found.extend(_collect_base_cells(value, key_seed)) elif key not in {"base_seed", "swarm_seed"}: found.extend(_collect_base_cells(value, b)) elif isinstance(node, list): for value in node: found.extend(_collect_base_cells(value, base)) return found def _method_base_cells(result: Mapping[str, Any], method: str) -> list[dict[str, Any]]: arms = result.get("arms") return _collect_base_cells(arms.get(method)) if isinstance(arms, Mapping) and method in arms else [] def _method_evidence(value: Any, method: str, family: str, root: Path) -> bool: """Require non-empty metric or prediction evidence below an exact method key.""" if isinstance(value, Mapping): for key, item in value.items(): normalized = str(key).lower().replace("-", "_") if normalized == method: if _prediction_records(item, root) is not None: return True if isinstance(item, Mapping) and (_number(item, "nll", "loss", "accuracy", "map50_95", "map50", "mAP50-95") is not None): return True if _method_evidence(item, method, family, root): return True if _method_evidence(item, method, family, root): return True elif isinstance(value, list): return bool(value) and any(_method_evidence(item, method, family, root) for item in value) return False def _require_confirmation_methods(result: Mapping[str, Any], workload: str, family: str, root: Path, issues: dict[str, list[str]]) -> None: confirmation = result.get("confirmation") required = {"feature_pso", "feature_random", "feature_adam", "head_adam"} required |= ({"uniform", "uniform_temperature", "slsqp_weights", "ensemble_pso"} if family == "classification" else {"uniform_wbf", "ensemble_pso", "ensemble_random"}) for method in sorted(required): if not _method_evidence(confirmation, method, family, root): _issue(issues, "metrics", f"{workload}: confirmation lacks predictions/metrics for required method {method}") def _method_cells(result: Mapping[str, Any], method: str) -> list[dict[str, Any]]: arms = result.get("arms") if not isinstance(arms, Mapping): return [] value = arms.get(method) return _collect_cells(value) if value is not None else [] def _verify_matrix(result: Mapping[str, Any], workload: str, issues: dict[str, list[str]]) -> dict[str, Any]: arms = result.get("arms") if not isinstance(arms, Mapping): _issue(issues, "schema", f"{workload}: arms must be an object") arms = {} required = {"feature_pso", "feature_random", "feature_adam", "head_adam"} missing = required - set(arms) if missing: _issue(issues, "matrix", f"{workload}: missing arms {sorted(missing)}") stats: dict[str, Any] = {"primary_queries": 0, "primary_random_queries": 0, "ensemble_queries": 0, "primary_samples": 0, "primary_random_samples": 0, "ensemble_samples": 0, "cells": {}} for method in ("feature_pso", "feature_random"): cells = _method_cells(result, method) stats["cells"][method] = len(cells) expected = len(BASE_SEEDS) * len(SWARM_SEEDS) if len(cells) != expected: _issue(issues, "matrix", f"{workload}: {method} requires {expected} base/swarm cells, got {len(cells)}") seen: set[tuple[int, int]] = set() for cell in cells: key = (_int(cell, "base_seed") or -1, _int(cell, "swarm_seed") or -1) if key in seen or key[0] not in BASE_SEEDS or key[1] not in SWARM_SEEDS: _issue(issues, "matrix", f"{workload}: invalid or duplicate {method} cell {key}") seen.add(key) queries, samples = _record_queries(cell) if queries != PRIMARY_QUERIES: _issue(issues, "accounting", f"{workload}: {method} {key} queries must be {PRIMARY_QUERIES}, got {queries}") if samples != PRIMARY_QUERIES * OBJECTIVE_SAMPLES[workload]: _issue(issues, "accounting", f"{workload}: {method} {key} samples must be {PRIMARY_QUERIES * OBJECTIVE_SAMPLES[workload]}, got {samples}") if method == "feature_pso": if queries is not None: stats["primary_queries"] += queries if samples is not None: stats["primary_samples"] += samples else: if queries is not None: stats["primary_random_queries"] += queries if samples is not None: stats["primary_random_samples"] += samples for method in ("feature_adam", "head_adam"): cells = _method_base_cells(result, method); stats["cells"][method] = len(cells) if len(cells) != len(BASE_SEEDS): _issue(issues, "matrix", f"{workload}: {method} requires exactly three base cells, got {len(cells)}") seen = set() for cell in cells: seed = _int(cell, "base_seed") if seed in seen or seed not in BASE_SEEDS: _issue(issues, "matrix", f"{workload}: invalid or duplicate {method} base cell {seed}") if seed is not None: seen.add(seed) ensemble = result.get("ensemble") if not isinstance(ensemble, Mapping): _issue(issues, "schema", f"{workload}: ensemble must be an object") ensemble = {} required_ensemble = ("uniform", "uniform_temperature", "slsqp_weights", "ensemble_pso") if workload in CLASSIFICATION_WORKLOADS else ("uniform_wbf", "ensemble_pso", "ensemble_random") for method in required_ensemble: if method not in ensemble or ensemble.get(method) in (None, {}, []): _issue(issues, "matrix", f"{workload}: missing required ensemble method {method}") ens_pso = ensemble.get("ensemble_pso", ensemble.get("pso")) cells = _collect_cells(ens_pso) if ens_pso is not None else [] stats["cells"]["ensemble_pso"] = len(cells) if len(cells) != len(SWARM_SEEDS): _issue(issues, "matrix", f"{workload}: ensemble_pso requires three swarm cells, got {len(cells)}") seen_swarm: set[int] = set() for cell in cells: seed = _int(cell, "swarm_seed", "seed") if seed in seen_swarm or seed not in SWARM_SEEDS: _issue(issues, "matrix", f"{workload}: invalid or duplicate ensemble swarm cell {seed}") if seed is not None: seen_swarm.add(seed) queries, samples = _record_queries(cell) if queries != ENSEMBLE_QUERIES: _issue(issues, "accounting", f"{workload}: ensemble_pso queries must be {ENSEMBLE_QUERIES}, got {queries}") if samples != ENSEMBLE_QUERIES * OBJECTIVE_SAMPLES[workload]: _issue(issues, "accounting", f"{workload}: ensemble_pso samples must be {ENSEMBLE_QUERIES * OBJECTIVE_SAMPLES[workload]}, got {samples}") if queries is not None: stats["ensemble_queries"] += queries if samples is not None: stats["ensemble_samples"] += samples if workload == DETECTION_WORKLOAD: random_cells = _collect_cells(ensemble.get("ensemble_random")) if ensemble.get("ensemble_random") is not None else [] stats["cells"]["ensemble_random"] = len(random_cells) if len(random_cells) != len(SWARM_SEEDS): _issue(issues, "matrix", f"{workload}: ensemble_random requires three swarm cells, got {len(random_cells)}") seen_random = set() for cell in random_cells: seed = _int(cell, "swarm_seed", "seed") if seed in seen_random or seed not in SWARM_SEEDS: _issue(issues, "matrix", f"{workload}: invalid or duplicate ensemble_random seed {seed}") if seed is not None: seen_random.add(seed) return stats def _verify_selection(result: Mapping[str, Any], workload: str, issues: dict[str, list[str]]) -> None: selection = result.get("development_selection") if not isinstance(selection, Mapping): _issue(issues, "selection", f"{workload}: missing development_selection") return selected = selection.get("primary", selection.get("feature_pso", selection.get("selected"))) if not isinstance(selected, Mapping): _issue(issues, "selection", f"{workload}: missing primary selected endpoints") return for base in BASE_SEEDS: entry = selected.get(str(base), selected.get(base)) if not isinstance(entry, Mapping): _issue(issues, "selection", f"{workload}: no selected endpoint for base seed {base}") continue swarm = _int(entry, "swarm_seed", "seed") generation = _int(entry, "generation", "final_generation") if swarm not in SWARM_SEEDS: _issue(issues, "selection", f"{workload}: selected seed {base} has invalid swarm {swarm}") if generation != PRIMARY_GENERATIONS: _issue(issues, "selection", f"{workload}: selected endpoint {base} is not final generation 60") matches = [c for c in _method_cells(result, "feature_pso") if _int(c, "base_seed") == base and _int(c, "swarm_seed") == swarm] if not matches: _issue(issues, "selection", f"{workload}: selected endpoint {base}/{swarm} is not a feature_pso cell") elif entry.get("vector_hash") and matches[0].get("vector_hash") and entry["vector_hash"] != matches[0]["vector_hash"]: _issue(issues, "selection", f"{workload}: selected vector hash drift for base {base}") def _prediction_records(value: Any, root: Path) -> list[dict[str, Any]] | None: try: value = _resolve_json(root, value) except (OSError, ValueError, json.JSONDecodeError): return None if isinstance(value, Mapping): if "probabilities" in value and "targets" in value: probabilities, targets = value["probabilities"], value["targets"] for attr in ("detach", "cpu"): if hasattr(probabilities, attr): probabilities = getattr(probabilities, attr)() if hasattr(targets, attr): targets = getattr(targets, attr)() if hasattr(probabilities, "tolist"): probabilities = probabilities.tolist() if hasattr(targets, "tolist"): targets = targets.tolist() if isinstance(probabilities, (list, tuple)) and isinstance(targets, (list, tuple)) and len(probabilities) == len(targets): return [{"probabilities": list(probability), "target": int(target)} for probability, target in zip(probabilities, targets)] return None for key in ("records", "predictions", "images", "examples", "data"): if key in value: got = _prediction_records(value[key], root) if got is not None: return got return None if isinstance(value, list) and all(isinstance(x, Mapping) for x in value): return [dict(x) for x in value] return None def _find_prediction_sets(value: Any, root: Path, prefix: str = "$") -> dict[str, list[dict[str, Any]]]: found: dict[str, list[dict[str, Any]]] = {} if isinstance(value, Mapping): for key, item in value.items(): name = f"{prefix}.{key}" if "prediction" in str(key).lower() or str(key).lower() in {"base", "pso", "feature_pso", "test"}: records = _prediction_records(item, root) if records is not None: found[name] = records found.update(_find_prediction_sets(item, root, name)) elif isinstance(value, list) and value and isinstance(value[0], Mapping): records = _prediction_records(value, root) if records is not None: found[prefix] = records return found def classification_metrics(records: Sequence[Mapping[str, Any]]) -> dict[str, Any]: """Recompute unrounded NLL and accuracy from per-image probabilities/targets.""" probs: list[list[float]] = []; targets: list[int] = [] for i, record in enumerate(records): p = record.get("probabilities", record.get("probs", record.get("prob"))) target = record.get("target", record.get("label", record.get("class_id"))) if not isinstance(p, (list, tuple)) or not p or not isinstance(target, int) or isinstance(target, bool) or target < 0 or target >= len(p): raise ValueError(f"invalid classification prediction at index {i}") vals = [float(x) for x in p] if not all(math.isfinite(x) and x >= 0.0 for x in vals) or not math.isclose(math.fsum(vals), 1.0, rel_tol=1e-6, abs_tol=1e-6): raise ValueError(f"probabilities must be finite and sum to one at index {i}") probs.append(vals); targets.append(target) nll = math.fsum( -math.log(max(p[t], 1e-300)) for p, t in zip(probs, targets) ) / len(probs) predictions = [ max(range(len(p)), key=p.__getitem__) for p in probs ] accuracy = sum( prediction == target for prediction, target in zip(predictions, targets) ) / len(probs) brier = math.fsum( math.fsum( (probability - float(index == target)) ** 2 for index, probability in enumerate(p) ) for p, target in zip(probs, targets) ) / len(probs) confidences = [p[prediction] for p, prediction in zip(probs, predictions)] ece = 0.0 for bin_index in range(15): lower = bin_index / 15.0 upper = (bin_index + 1) / 15.0 members = [ index for index, confidence in enumerate(confidences) if lower <= confidence <= upper if bin_index == 14 or confidence < upper ] if members: bin_accuracy = math.fsum( predictions[index] == targets[index] for index in members ) / len(members) bin_confidence = math.fsum( confidences[index] for index in members ) / len(members) ece += ( abs(bin_accuracy - bin_confidence) * len(members) / len(probs) ) return { "n": len(probs), "nll": nll, "accuracy": accuracy, "brier": brier, "ece15": ece, "probabilities": probs, "targets": targets, } def _box(record: Mapping[str, Any]) -> tuple[float, float, float, float] | None: value = record.get("box", record.get("bbox", record.get("xyxy"))) if not isinstance(value, (list, tuple)) or len(value) != 4 or not all(_finite(x) for x in value): return None x1, y1, x2, y2 = map(float, value) return (x1, y1, x2, y2) if x2 >= x1 and y2 >= y1 else None def _iou(a: Sequence[float], b: Sequence[float]) -> float: x1, y1 = max(a[0], b[0]), max(a[1], b[1]); x2, y2 = min(a[2], b[2]), min(a[3], b[3]) inter = max(0.0, x2 - x1) * max(0.0, y2 - y1) area_a = max(0.0, a[2] - a[0]) * max(0.0, a[3] - a[1]); area_b = max(0.0, b[2] - b[0]) * max(0.0, b[3] - b[1]) return inter / (area_a + area_b - inter) if area_a + area_b - inter > 0 else 0.0 def _interp_ap_101(recall: Sequence[float], precision: Sequence[float]) -> float: """Ultralytics compute_ap: precision envelope and 101-point trapezoid.""" mrec = [0.0, *map(float, recall), 1.0] mpre = [1.0, *map(float, precision), 0.0] for i in range(len(mpre) - 2, -1, -1): mpre[i] = max(mpre[i], mpre[i + 1]) values: list[float] = [] for k in range(101): x = k / 100.0; j = 0 while j + 1 < len(mrec) and mrec[j + 1] <= x: j += 1 if j + 1 >= len(mrec): values.append(mpre[-1]); continue span = mrec[j + 1] - mrec[j] values.append(mpre[j] if span <= 0 else mpre[j] + (mpre[j + 1] - mpre[j]) * (x - mrec[j]) / span) return sum((values[i] + values[i + 1]) * 0.5 / 100.0 for i in range(100)) def detection_metrics(records: Sequence[Mapping[str, Any]], class_count: int | None = None) -> dict[str, Any]: """Recompute Ultralytics-style whole-dataset AP at IoU .50:.95. Matching is performed independently per image. Candidate matches are sorted by IoU and deduplicated by prediction and ground truth, as in DetectionValidator.process_batch; AP then uses the pinned 101-point interpolated trapezoid. """ thresholds = [0.50 + 0.05 * i for i in range(10)] parsed: list[tuple[list[dict[str, Any]], list[dict[str, Any]]]] = []; max_class = -1 for i, image in enumerate(records): predictions = image.get("predictions", image.get("detections", image.get("pred", []))) truth = image.get("ground_truth", image.get("targets", image.get("gt", image.get("labels", [])))) if not isinstance(predictions, list) or not isinstance(truth, list): raise ValueError(f"invalid detection image record {i}") pp: list[dict[str, Any]] = []; gg: list[dict[str, Any]] = [] for item in predictions: if not isinstance(item, Mapping) or _box(item) is None: raise ValueError(f"invalid detection prediction {i}") cls = item.get("class_id", item.get("class", item.get("cls"))); score = item.get("score", item.get("confidence", item.get("conf"))) if not isinstance(cls, int) or isinstance(cls, bool) or not _finite(score): raise ValueError(f"invalid detection prediction fields {i}") pp.append({"box": _box(item), "class_id": cls, "score": float(score)}); max_class = max(max_class, cls) for item in truth: if not isinstance(item, Mapping) or _box(item) is None: raise ValueError(f"invalid ground truth {i}") cls = item.get("class_id", item.get("class", item.get("cls"))) if not isinstance(cls, int) or isinstance(cls, bool): raise ValueError(f"invalid ground truth class {i}") gg.append({"box": _box(item), "class_id": cls}); max_class = max(max_class, cls) parsed.append((pp, gg)) present = sorted({g["class_id"] for _, gt in parsed for g in gt}) classes = present if class_count is None else [c for c in range(class_count) if c in present] if not classes: classes = list(range(class_count or (max_class + 1))) or [0] aps: dict[str, list[float]] = {}; precision50: list[float] = []; recall50: list[float] = [] for cls in classes: gt_count = sum(sum(x["class_id"] == cls for x in gt) for _, gt in parsed); class_aps: list[float] = [] for threshold in thresholds: true_by_image: list[list[bool]] = [] for preds, gt in parsed: candidates = [] for pi, pred in enumerate(preds): if pred["class_id"] != cls: continue for gi, target in enumerate(gt): if target["class_id"] == cls: overlap = _iou(pred["box"], target["box"]) if overlap >= threshold: candidates.append((overlap, pi, gi)) candidates.sort(key=lambda x: -x[0]); used_pred: set[int] = set(); used_gt: set[int] = set(); matched: set[int] = set() for overlap, pi, gi in candidates: if pi not in used_pred and gi not in used_gt: used_pred.add(pi); used_gt.add(gi); matched.add(pi) true_by_image.append([i in matched for i in range(len(preds))]) ranked = sorted(((pred["score"], hit) for (preds, _), hits in zip(parsed, true_by_image) for pred, hit in zip(preds, hits) if pred["class_id"] == cls), key=lambda x: -x[0]) tp=[]; fp=[]; ctp=cfp=0 for _, hit in ranked: ctp += int(hit); cfp += int(not hit); tp.append(ctp); fp.append(cfp) if gt_count == 0 or not ranked: # Ultralytics ap_per_class skips classes with no predictions; AP is zero. class_aps.append(0.0); continue recalls = [x / gt_count for x in tp]; precisions = [x / max(x + y, 1) for x, y in zip(tp, fp)] class_aps.append(_interp_ap_101(recalls, precisions)) if threshold == 0.5: precision50.append(precisions[-1] if precisions else 0.0); recall50.append(recalls[-1] if recalls else 0.0) aps[str(cls)] = class_aps map50 = math.fsum(v[0] for v in aps.values()) / len(aps); map5095 = math.fsum(x for values in aps.values() for x in values) / (len(aps) * 10) return {"per_class_ap": aps, "n": len(records), "map50": map50, "map50_95": map5095, "precision": math.fsum(precision50) / len(precision50) if precision50 else 0.0, "recall": math.fsum(recall50) / len(recall50) if recall50 else 0.0, "ground_truth": sum(len(gt) for _, gt in parsed), "predictions": sum(len(preds) for preds, _ in parsed)} def _metric_from_record(record: Any, family: str) -> dict[str, float] | None: if not isinstance(record, Mapping): return None keys = ("nll", "loss") if family == "classification" else ("map50_95", "map50-95", "mAP50-95", "map5095") primary = _number(record, *keys); accuracy = _number(record, "accuracy", "acc") map50 = _number(record, "map50", "mAP50") if family == "classification" and primary is not None and accuracy is not None: return {"nll": primary, "accuracy": accuracy} if family == "detection" and primary is not None: return {"map50_95": primary, **({"map50": map50} if map50 is not None else {})} return None def _quantile(values: Sequence[float], q: float) -> float: ordered = sorted(float(x) for x in values) if not ordered: raise ValueError("cannot quantile an empty sequence") position = (len(ordered) - 1) * q; lower = int(math.floor(position)); upper = min(lower + 1, len(ordered) - 1) return ordered[lower] + (ordered[upper] - ordered[lower]) * (position - lower) def _record_identity(record: Mapping[str, Any], index: int) -> Any: for key in ("image_id", "id", "key", "filename", "path", "index"): if key in record: return (key, str(record[key])) return ("position", index) def _record_ground_truth(record: Mapping[str, Any]) -> Any: return record.get("ground_truth", record.get("targets", record.get("gt", record.get("labels", [])))) def _bootstrap_alignment(pairs: Sequence[tuple[Sequence[Mapping[str, Any]], Sequence[Mapping[str, Any]]]], family: str) -> tuple[bool, str]: if not pairs or any(len(a) != len(b) or not a for a, b in pairs): return False, "incomplete or length-mismatched paired records" reference_ids = [_record_identity(x, i) for i, x in enumerate(pairs[0][0])] reference_gt = [_record_ground_truth(x) for x in pairs[0][0]] for pair_index, (base, pso) in enumerate(pairs): if [_record_identity(x, i) for i, x in enumerate(base)] != reference_ids or [_record_identity(x, i) for i, x in enumerate(pso)] != reference_ids: return False, f"pair {pair_index} image IDs/order differ" if family == "classification": base_targets = [x.get("target", x.get("label", x.get("class_id"))) for x in base] pso_targets = [x.get("target", x.get("label", x.get("class_id"))) for x in pso] ref_targets = [x.get("target", x.get("label", x.get("class_id"))) for x in pairs[0][0]] if base_targets != ref_targets or pso_targets != ref_targets: return False, f"pair {pair_index} targets differ" elif [_canonical(_record_ground_truth(x)) for x in base] != [_canonical(x) for x in reference_gt] or [_canonical(_record_ground_truth(x)) for x in pso] != [_canonical(x) for x in reference_gt]: return False, f"pair {pair_index} ground truth differs" return True, "aligned" def _bootstrap_from_records(pairs: Sequence[tuple[Sequence[Mapping[str, Any]], Sequence[Mapping[str, Any]]]], family: str) -> dict[str, Any]: aligned, reason = _bootstrap_alignment(pairs, family) if not aligned: return {"available": False, "seed": BOOTSTRAP_SEED, "resamples": BOOTSTRAP_RESAMPLES, "reason": reason} rng = random.Random(BOOTSTRAP_SEED); stats: list[float] = [] if family == "classification": parsed = [(classification_metrics(a), classification_metrics(b)) for a, b in pairs] by_class: dict[int, list[int]] = {} for i, target in enumerate(parsed[0][0]["targets"]): by_class.setdefault(target, []).append(i) for _ in range(BOOTSTRAP_RESAMPLES): # One class-stratified draw is shared by every paired base/PSO model. indices = [rng.choice(class_indices) for class_indices in by_class.values() for _ in class_indices] deltas = [] for base, pso in parsed: b_nll = math.fsum(-math.log(max(base["probabilities"][i][base["targets"][i]], 1e-300)) for i in indices) / len(indices) p_nll = math.fsum(-math.log(max(pso["probabilities"][i][pso["targets"][i]], 1e-300)) for i in indices) / len(indices) deltas.append((b_nll - p_nll) / b_nll if b_nll else 0.0) stats.append(math.fsum(deltas) / len(deltas)) else: for _ in range(BOOTSTRAP_RESAMPLES): # One whole-image draw is shared by every paired base/PSO model. indices = [rng.randrange(len(pairs[0][0])) for _ in pairs[0][0]] deltas = [] for base, pso in pairs: b = detection_metrics([base[i] for i in indices])["map50_95"] p = detection_metrics([pso[i] for i in indices])["map50_95"] deltas.append(p - b) stats.append(math.fsum(deltas) / len(deltas)) lo = _quantile(stats, BOOTSTRAP_ALPHA); hi = _quantile(stats, 1.0 - BOOTSTRAP_ALPHA) return {"available": True, "seed": BOOTSTRAP_SEED, "resamples": BOOTSTRAP_RESAMPLES, "alpha": BOOTSTRAP_ALPHA, "lower": lo, "upper": hi, "statistic": math.fsum(stats) / len(stats), "excludes_zero": lo > 0 or hi < 0} def _prediction_pairs(result: Mapping[str, Any], root: Path, family: str) -> list[tuple[int, list[dict[str, Any]], list[dict[str, Any]]]]: """Find one test base/selected pair per frozen base seed by explicit names.""" sets = _find_prediction_sets(result.get("confirmation", {}), root) out: list[tuple[int, list[dict[str, Any]], list[dict[str, Any]]]] = [] for seed in BASE_SEEDS: candidates = [(name, records) for name, records in sets.items() if str(seed) in name] base = next((records for name, records in candidates if any(x in name.lower() for x in ("base", "frozen"))), None) pso = next((records for name, records in candidates if any(x in name.lower() for x in ("feature_pso", "selected", "pso")) and "ensemble" not in name.lower()), None) if base is not None and pso is not None: out.append((seed, base, pso)) return out def _audit_series(value: Any) -> list[dict[str, Any]]: found: list[dict[str, Any]] = [] if isinstance(value, Mapping): for item in value.values(): found.extend(_audit_series(item)) elif isinstance(value, list) and value and all(isinstance(item, Mapping) for item in value): if any("epoch" in item or "step" in item for item in value) and any(_number(item, "loss", "audit_loss") is not None for item in value): found.append({"records": value}) else: for item in value: found.extend(_audit_series(item)) return found def _plateau_flags(result: Mapping[str, Any], workload: str, family: str, issues: dict[str, list[str]]) -> dict[str, Any]: baselines = result.get("baselines", {}) if not isinstance(baselines, Mapping): return {"available": False, "passed": False} per_seed: dict[str, bool] = {}; details: dict[str, Any] = {} for seed in BASE_SEEDS: entry = baselines.get(str(seed), baselines.get(seed)) series = _audit_series(entry) records = series[0]["records"] if series else [] records = records[-11:] if len(records) >= 11 else [] losses = [_number(x, "loss", "audit_loss") for x in records] metric_keys = ("accuracy", "primary_metric", "selection_accuracy") if family == "classification" else ("map50_95", "mAP50-95", "primary_metric", "selection_metric") metrics = [_number(x, *metric_keys) for x in records] loss_ok = len(losses) == 11 and all(x is not None for x in losses) metric_ok = len(metrics) == 11 and all(x is not None for x in metrics) if loss_ok: mean = math.fsum(float(x) for x in losses) / len(losses) loss_ok = (max(losses) - min(losses)) / max(abs(mean), 1e-12) <= 0.01 if metric_ok: metric_ok = max(metrics) - min(metrics) <= 0.005 passed = bool(loss_ok and metric_ok); per_seed[str(seed)] = passed details[str(seed)] = {"loss_ok": bool(loss_ok), "metric_ok": bool(metric_ok), "observations": len(records)} declared = _number(entry, "baseline_plateau") if isinstance(entry, Mapping) else None if isinstance(entry, Mapping) and "baseline_plateau" in entry and bool(entry["baseline_plateau"]) != passed: _issue(issues, "plateau", f"{workload}: baseline seed {seed} inflated/incorrect plateau flag") return {"available": bool(per_seed), "per_seed": per_seed, "details": details, "passed": bool(per_seed) and all(per_seed.values())} def _workload_gates(result: Mapping[str, Any], workload: str, family: str, root: Path, issues: dict[str, list[str]]) -> dict[str, Any]: pairs = _prediction_pairs(result, root, family) per_seed: dict[str, Any] = {} pair_records: list[tuple[Sequence[Mapping[str, Any]], Sequence[Mapping[str, Any]]]] = [] for seed, base_records, pso_records in pairs: try: base = classification_metrics(base_records) if family == "classification" else detection_metrics(base_records, 20) pso = classification_metrics(pso_records) if family == "classification" else detection_metrics(pso_records, 20) except (ValueError, ZeroDivisionError) as exc: _issue(issues, "metrics", f"{workload}: test pair {seed} cannot be recomputed: {exc}"); continue pair_records.append((base_records, pso_records)) if family == "classification": per_seed[str(seed)] = {"base": {"nll": base["nll"], "accuracy": base["accuracy"]}, "pso": {"nll": pso["nll"], "accuracy": pso["accuracy"]}, "relative_nll_reduction": (base["nll"] - pso["nll"]) / base["nll"], "accuracy_delta": pso["accuracy"] - base["accuracy"]} else: per_seed[str(seed)] = {"base": {"map50_95": base["map50_95"], "map50": base["map50"]}, "pso": {"map50_95": pso["map50_95"], "map50": pso["map50"]}, "map50_95_delta": pso["map50_95"] - base["map50_95"], "map50_delta": pso["map50"] - base["map50"]} ci = _bootstrap_from_records(pair_records, family) # Confirmation predictions and the paired CI are required integrity evidence. # A numerically negative CI is a valid scientific result; an unavailable CI # means the frozen/confirmed artifact set is incomplete or tampered. if len(pairs) != len(BASE_SEEDS): _issue(issues, "metrics", f"{workload}: required confirmation pairs are incomplete ({len(pairs)}/{len(BASE_SEEDS)})") if not ci.get("available", False): _issue(issues, "metrics", f"{workload}: required paired bootstrap unavailable: {ci.get('reason', 'missing prediction evidence')}") if family == "classification" and per_seed: reductions = [x["relative_nll_reduction"] for x in per_seed.values()]; acc_deltas = [x["accuracy_delta"] for x in per_seed.values()] gate = len(reductions) == 3 and math.fsum(reductions) / 3 >= 0.01 and math.fsum(acc_deltas) / 3 >= -0.002 and min(acc_deltas) >= -0.005 and sum(x > 0 for x in reductions) >= 2 and ci.get("excludes_zero", False) elif family == "detection" and per_seed: deltas = [x["map50_95_delta"] for x in per_seed.values()] gate = len(deltas) == 3 and math.fsum(deltas) / 3 >= 0.005 and min(deltas) >= -0.005 and sum(x > 0 for x in deltas) >= 2 and ci.get("excludes_zero", False) else: gate = False if len(pairs) != len(BASE_SEEDS): _issue(issues, "bootstrap", f"{workload}: missing complete official-test base/feature_pso prediction pairs ({len(pairs)}/3)") return {"available": bool(per_seed), "per_seed": per_seed, "bootstrap": ci, "generalization_pass": bool(gate)} def _verify_hashes(result: Mapping[str, Any], root: Path, issues: dict[str, list[str]], workload: str) -> None: declared = result.get("artifact_hashes") if not isinstance(declared, Mapping): _issue(issues, "schema", f"{workload}: missing artifact_hashes") return for name, expected in declared.items(): if not isinstance(name, str) or not isinstance(expected, str): _issue(issues, "seal", f"{workload}: malformed artifact hash entry") continue path = (root / name).resolve() if root.resolve() not in path.parents or not path.is_file(): _issue(issues, "seal", f"{workload}: missing/escaping artifact {name}") elif _sha256(path) != expected: _issue(issues, "seal", f"{workload}: artifact hash drift {name}") def _compare_development_snapshot(development: Any, current: Mapping[str, Any], workload: str, issues: dict[str, list[str]]) -> None: """Ensure confirmation did not alter any sealed development decision/state.""" if not isinstance(development, Mapping): _issue(issues, "schema", f"{workload}: development_result.json must contain an object") return development_leakage = development.get("leakage_counters") if not isinstance(development_leakage, Mapping): _issue(issues, "leakage", f"{workload}: development snapshot lacks leakage_counters") else: loaded = development_leakage.get("official_test_data_loaded_before_freeze", development_leakage.get("test_data_loaded_before_freeze")) evaluated = development_leakage.get("official_test_evaluations_before_freeze", development_leakage.get("test_evaluations_before_freeze", development_leakage.get("official_test_forward_passes_before_freeze"))) if loaded is not False or evaluated != 0: _issue(issues, "leakage", f"{workload}: development snapshot records pre-freeze official-test exposure") for key in ("official_test_construction", "official_test_dataset_construction", "official_test_forward_passes", "official_test_evaluations"): if key in development_leakage and development_leakage[key] != 0: _issue(issues, "leakage", f"{workload}: development snapshot {key} must be zero") development_confirmation = development.get("confirmation") if development_confirmation not in ({}, None): _issue(issues, "leakage", f"{workload}: development snapshot confirmation must be empty") fields = ("config", "manifests", "provenance", "baselines", "arms", "ensemble", "development_selection", "integrity", "resource_ledger", "artifact_hashes") for field in fields: if field not in development: _issue(issues, "seal", f"{workload}: development_result.json missing sealed field {field}") continue if field not in current: _issue(issues, "seal", f"{workload}: current result missing sealed field {field}") continue if field == "artifact_hashes": # Confirmation may add test prediction hashes. Every development hash # must nevertheless remain present and byte-identical. old_hashes = development[field]; new_hashes = current[field] if not isinstance(old_hashes, Mapping) or not isinstance(new_hashes, Mapping): _issue(issues, "seal", f"{workload}: artifact_hashes changed shape across confirmation") else: for name, value in old_hashes.items(): if new_hashes.get(name) != value: _issue(issues, "seal", f"{workload}: sealed artifact hash drift for {name}") elif field == "integrity": old_integrity = development[field]; new_integrity = current[field] if not isinstance(old_integrity, Mapping) or not isinstance(new_integrity, Mapping): if old_integrity != new_integrity: _issue(issues, "seal", f"{workload}: pre-confirmation integrity changed") else: for name, value in old_integrity.items(): if new_integrity.get(name) != value: _issue(issues, "seal", f"{workload}: pre-confirmation integrity field changed: {name}") elif _canonical(development[field]) != _canonical(current[field]): _issue(issues, "seal", f"{workload}: sealed pre-confirmation field changed: {field}") def _evaluate_workload(result: Any, root: Path, workload: str, issues: dict[str, list[str]], development: Any = None) -> dict[str, Any]: if not isinstance(result, Mapping): _issue(issues, "schema", f"{workload}: result must be an object"); return {"workload_id": workload, "valid": False} if result.get("workload_id") != workload: _issue(issues, "matrix", f"{workload}: workload_id mismatch") if development is not None: _compare_development_snapshot(development, result, workload, issues) family = "detection" if workload == DETECTION_WORKLOAD else "classification" if result.get("family") != family: _issue(issues, "matrix", f"{workload}: family must be {family}") for key in ("manifests", "provenance", "baselines", "arms", "ensemble", "development_selection", "confirmation", "integrity", "leakage_counters", "resource_ledger", "artifact_hashes"): if key not in result: _issue(issues, "schema", f"{workload}: missing top-level {key}") _config_checks(result.get("config"), issues, workload); _check_leakage(result, issues, workload); _verify_hashes(result, root, issues, workload) accounting = _verify_matrix(result, workload, issues); _verify_selection(result, workload, issues); _require_confirmation_methods(result, workload, family, root, issues) if result.get("integrity", {}).get("confirmed") is False if isinstance(result.get("integrity"), Mapping) else False: _issue(issues, "seal", f"{workload}: integrity declares confirmation failure") prediction_sets = _find_prediction_sets(result.get("confirmation", {}), root) recomputed: dict[str, Any] = {} for name, records in prediction_sets.items(): try: recomputed[name] = classification_metrics(records) if family == "classification" else detection_metrics(records, 20) except (ValueError, ZeroDivisionError) as exc: _issue(issues, "metrics", f"{workload}: invalid stored predictions at {name}: {exc}") # Check every explicitly stored metric that has a corresponding recomputation. for name, metric in recomputed.items(): if "pso" in name.lower() and isinstance(metric, Mapping): stored = _metric_from_record(result.get("confirmation"), family) if stored and family == "classification" and not (_same(stored.get("nll"), metric.get("nll"), 1e-7) and _same(stored.get("accuracy"), metric.get("accuracy"), 1e-7)): _issue(issues, "metrics", f"{workload}: stored classification metric disagrees with probabilities at {name}") plateau = _plateau_flags(result, workload, family, issues) gates = _workload_gates(result, workload, family, root, issues) objective = _number(result.get("development_selection"), "objective_improvement", "relative_objective_improvement") selection_metric = _number(result.get("development_selection"), "selection_metric", "selection_nll", "selection_map50_95") overfit = bool(objective is not None and objective >= 0.01 and selection_metric is not None and ((family == "classification" and selection_metric > 0.01) or (family == "detection" and selection_metric < -0.005))) if overfit: _issue(issues, "overfit", f"{workload}: objective improvement conflicts with held-out/selection metric") return {"workload_id": workload, "family": family, "valid": True, "accounting": accounting, "recomputed": recomputed, "plateau": plateau, "gates": gates, "overfit_signal": overfit} def evaluate_run(run_root: str | os.PathLike[str]) -> dict[str, Any]: """Evaluate one frozen run, returning findings even when artifacts are malformed.""" root = Path(run_root); issues = {key: [] for key in ISSUE_CATEGORIES}; workloads: dict[str, Any] = {} manifest: Any = None try: manifest = _json(root / "frozen_manifest.json") except (OSError, ValueError, json.JSONDecodeError) as exc: _issue(issues, "schema", f"cannot load frozen_manifest.json: {exc}") if isinstance(manifest, Mapping): for location in _walk_nonfinite(manifest): _issue(issues, "finite", f"non-finite value in frozen manifest at {location}") _hash_manifest(root, manifest, issues); _config_checks(manifest.get("config"), issues) frozen_config = manifest.get("config") if isinstance(manifest.get("config"), Mapping) else {} if sorted(frozen_config.get("workload_ids", ())) != sorted(WORKLOADS): _issue(issues, "matrix", "frozen manifest does not seal all three workloads") for workload in WORKLOADS: path = root / "workloads" / workload / "result.json" development_path = root / "workloads" / workload / "development_result.json" if isinstance(manifest, Mapping): manifest_artifacts = manifest.get("artifacts", {}) expected_development = f"workloads/{workload}/development_result.json" if not isinstance(manifest_artifacts, Mapping) or expected_development not in manifest_artifacts: _issue(issues, "seal", f"{workload}: frozen manifest must seal {expected_development}") try: result = _json(path) except (OSError, ValueError, json.JSONDecodeError) as exc: _issue(issues, "schema", f"{workload}: cannot load result.json: {exc}"); continue try: development = _json(development_path) except (OSError, ValueError, json.JSONDecodeError) as exc: _issue(issues, "seal", f"{workload}: cannot load development_result.json: {exc}"); development = None for location in _walk_nonfinite(result): _issue(issues, "finite", f"{workload}: non-finite value at {location}") if development is not None: for location in _walk_nonfinite(development): _issue(issues, "finite", f"{workload}: non-finite development value at {location}") workloads[workload] = _evaluate_workload(result, root, workload, issues, development) # Cross-workload exact accounting is intentionally independent of stored totals. totals = {key: sum(int(w.get("accounting", {}).get(key, 0)) for w in workloads.values()) for key in ("primary_queries", "primary_random_queries", "ensemble_queries", "primary_samples", "primary_random_samples", "ensemble_samples")} totals["pso_queries"] = totals["primary_queries"] + totals["ensemble_queries"] totals["candidate_samples"] = totals["primary_samples"] + totals["ensemble_samples"] if totals["pso_queries"] != TOTAL_PSO_QUERIES: _issue(issues, "accounting", f"total PSO queries must be {TOTAL_PSO_QUERIES}, got {totals['pso_queries']}") if totals["candidate_samples"] != TOTAL_CANDIDATE_SAMPLES: _issue(issues, "accounting", f"total candidate-sample evaluations must be {TOTAL_CANDIDATE_SAMPLES}, got {totals['candidate_samples']}") # A success flag is never consumed; it is checked against independently observed integrity. integrity_ok = not any(issues[key] for key in ("schema", "provenance", "matrix", "accounting", "seal", "selection", "finite", "leakage", "metrics")) payload = {"evaluator_version": EVALUATOR_VERSION, "protocol_version": PROTOCOL_VERSION, "run_root": str(root), "pass": integrity_ok, "integrity_pass": integrity_ok, "workloads": workloads, "accounting": {**totals, "expected_pso_queries": TOTAL_PSO_QUERIES, "expected_candidate_samples": TOTAL_CANDIDATE_SAMPLES}, "issues": issues, "issue_counts": {key: len(value) for key, value in issues.items()}} return payload def build_cli_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="Evaluate frozen post-training model-convergence artifacts") parser.add_argument("--run-root", type=Path, required=True) parser.add_argument("--output", type=Path, default=None) return parser def main(argv: Sequence[str] | None = None) -> int: args = build_cli_parser().parse_args(argv); payload = evaluate_run(args.run_root) destination = args.output or args.run_root / "evaluation.json" try: _atomic_json(destination, payload) except OSError as exc: print(f"evaluator output failed: {exc}", file=sys.stderr); return 2 print(json.dumps(payload, sort_keys=True, separators=(",", ":"))) return 0 if payload["pass"] else 1 __all__ = [ "BASE_SEEDS", "BOOTSTRAP_ALPHA", "BOOTSTRAP_RESAMPLES", "BOOTSTRAP_SEED", "CLASSIFICATION_WORKLOADS", "DETECTION_WORKLOAD", "EVALUATOR_VERSION", "ENSEMBLE_QUERIES", "TOTAL_CANDIDATE_SAMPLES", "TOTAL_PSO_QUERIES", "WORKLOADS", "classification_metrics", "detection_metrics", "evaluate_run", "main", "build_cli_parser", ] if __name__ == "__main__": raise SystemExit(main())