oak/benchmarks

"""Quiet-host calibration for the scale lane: evidence verified by recomputation.

A scale row's ``claim_eligible`` / ``evidence_class`` are DERIVED, never a

free flag, and nothing a calibration record *states* is trusted:

1. ``run_scale_suite.py calibrate`` runs an identical-binary A/A campaign

   (git, git_aa, oak_installed, oak_installed_aa) plus host probes (load

   average, small-file creation rate, F_FULLFSYNC latency) and writes its rows

   to a JSONL file. The record names that file's content sha256.

2. Publishing requires that calibration JSONL among publish_gate's inputs,

   matched by content sha256. The gate RECOMPUTES the A/A cell count, per-cell

   floors (oakbench.cellstats, the P1 machinery), identical-binary false

   claims, host load and disk probes from those rows, and checks that the

   calibration rows and the published rows share runner identity, host

   cpu_count and physical memory, the Oak and Git binary sha256, and

   registry-pinned fixtures. The published run must carry its own quiet disk

   probes too.

This detects edited, resealed, copied or invented RECORDS (the sealed digest

is only an integrity check of the record's own content). It does not

authenticate the campaign JSONL: editing its load samples and probe values

(a few dozen numbers) and resealing a record to the new file digest passes, as

would a fabricated campaign. Closing that needs a signed runner, which the

lane does not have. Do not describe it as forgery-proof.

"""

from __future__ import annotations

import hashlib

import json

import os

import statistics

import time

from pathlib import Path

from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple

SCHEMA_VERSION = 2

KIND = "oakbench_quiet_host_calibration"

CALIBRATED_EVIDENCE_CLASS = "calibrated_quiet_host"

DEVELOPMENT_EVIDENCE_CLASS = "development_only_loaded_host"

POLICY_VERSION = "2026-09-29.3"

# Quiet-host policy (provisional; re-derive on the dedicated runner and bump

# POLICY_VERSION). Load/core 0.1 is ~1.4 runnable threads on a 14-core host.

MAX_LOAD_PER_CORE = 0.10

MIN_AA_CELLS = 20

# A calibration needs two independent halves of >= 5 runs: floors come from

# one, A/A claims are counted on the other (a floor computed from the same

# A/A pair it judges can never produce a claim).

MIN_RUNS = 10

MAX_AA_FLOOR_MEDIAN_PCT = 10.0

# Per arm: the median A/A floor AND the share of cells whose floor exceeds

# MAX_AA_FLOOR_MEDIAN_PCT must be small (a broken A/A arm is systematic).

MAX_AA_CELLS_OVER_FLOOR_SHARE = 0.10

# Disk contention (load average does not see it; Oak clone/switch are

# fsync-bound, fb-549). Measured on the loaded dev host: F_FULLFSYNC median

# 3.7 ms, file creation as low as 170 files/s.

# Per flush primitive; a probe must use the platform's full-flush primitive

# (F_FULLFSYNC on macOS: plain fsync there skips the drive-cache flush).

# PROVISIONAL: no idle baseline was obtainable on the shared dev host (internal

# SSD, loaded: F_FULLFSYNC 3.7-3.9 ms); 2.5 ms may be unreachable even idle.

# The dedicated runner must re-derive these (P6) and bump POLICY_VERSION.

MAX_FLUSH_MEDIAN_MS = {"F_FULLFSYNC": 2.5, "fsync": 2.5}

MAX_FULLFSYNC_MEDIAN_MS = MAX_FLUSH_MEDIAN_MS["F_FULLFSYNC"]

MIN_CREATE_FILES_PER_S = 2000.0

# Work volumes whose flushes do not reach durable media cannot certify runs.

ACCEPTED_MEDIA = ("physical",)

PROBE_OPERATIONS = ("host.disk_fullfsync", "host.file_create_rate")

AA_PAIRS = (("git", "git_aa"), ("oak_installed", "oak_installed_aa"))

# ---------------------------------------------------------------------------

# Host facts and probes

# ---------------------------------------------------------------------------

def host_facts() -> Dict[str, Any]:

    try:

        mem = int(os.sysconf("SC_PAGE_SIZE")) * int(os.sysconf("SC_PHYS_PAGES"))

    except (ValueError, OSError, AttributeError):

        mem = None

    return {"host_cpu_count": os.cpu_count(), "host_mem_bytes": mem}

def expected_flush_kind(system: Optional[str] = None) -> str:

    import platform

    return "F_FULLFSYNC" if (system or platform.system()) == "Darwin" else "fsync"

def _diskutil(target: str) -> Dict[str, Any]:

    import plistlib

    import subprocess

    try:

        out = subprocess.run(["/usr/sbin/diskutil", "info", "-plist", target], capture_output=True, timeout=20)

    except (OSError, subprocess.TimeoutExpired):

        return {}

    if out.returncode != 0:

        return {}

    try:

        return plistlib.loads(out.stdout)

    except Exception:  # noqa: BLE001 - malformed output means unknown medium

        return {}

def classify_medium(volume: Dict[str, Any], store: Dict[str, Any]) -> str:

    """physical | ram_or_disk_image | tmpfs | unknown (only physical certifies)."""

    fstype = str(volume.get("fstype") or "").lower()

    if fstype in ("tmpfs", "ramfs", "devfs", "overlay", "squashfs"):

        return "tmpfs"

    info = store or volume.get("_diskutil") or {}

    if not info:

        return "unknown"

    if info.get("BusProtocol") == "Disk Image" or info.get("VirtualOrPhysical") == "Virtual" \

            or "Disk Image" in str(info.get("MediaName") or ""):

        return "ram_or_disk_image"

    return "physical" if info.get("BusProtocol") else "unknown"

def volume_identity(path: Path) -> Dict[str, Any]:

    """Filesystem type, mount point and backing device/medium of ``path``'s volume."""

    import platform

    import subprocess

    path = Path(path)

    probe = path if path.exists() else path.parent

    identity: Dict[str, Any] = {"mount_point": None, "device": None, "fstype": None, "volume_uuid": None,

                                "bus_protocol": None, "physical_store": None, "medium": "unknown"}

    try:

        df = subprocess.run(["df", "-P", str(probe)], capture_output=True, text=True, timeout=20).stdout

        fields = df.strip().splitlines()[-1].split()

        identity["device"], identity["mount_point"] = fields[0], fields[-1]

    except (OSError, IndexError, subprocess.TimeoutExpired):

        return identity

    if platform.system() == "Darwin":

        info = _diskutil(identity["mount_point"])

        identity["fstype"] = info.get("FilesystemType")

        identity["volume_uuid"] = info.get("VolumeUUID")

        store_info: Dict[str, Any] = {}

        stores = info.get("APFSPhysicalStores") or []

        if stores:

            store = stores[0].get("APFSPhysicalStore")

            identity["physical_store"] = store

            store_info = _diskutil(store) if store else {}

        else:

            store_info = info

        identity["bus_protocol"] = store_info.get("BusProtocol")

        identity["medium"] = classify_medium(identity, store_info)

    else:

        try:

            for line in Path("/proc/mounts").read_text().splitlines():

                parts = line.split()

                if len(parts) >= 3 and parts[1] == identity["mount_point"]:

                    identity["fstype"] = parts[2]

        except OSError:

            pass

        fstype = str(identity["fstype"] or "").lower()

        if fstype in ("tmpfs", "ramfs", "overlay", "squashfs"):

            identity["medium"] = "tmpfs"

        elif str(identity["device"]).startswith("/dev/") and not str(identity["device"]).startswith("/dev/ram") \

                and not str(identity["device"]).startswith("/dev/zram"):

            identity["medium"] = "physical"

        identity["volume_uuid"] = identity["device"]

    return identity

def volume_problems(volume: Any) -> List[str]:

    if not isinstance(volume, dict):

        return ["work volume unrecorded"]

    if volume.get("medium") not in ACCEPTED_MEDIA:

        return [f"work volume medium {volume.get('medium')!r} ({volume.get('fstype')}, {volume.get('device')}) "

                "cannot certify durable-flush costs"]

    return []

def volume_key(volume: Any) -> Optional[str]:

    if not isinstance(volume, dict):

        return None

    return str(volume.get("volume_uuid") or f"{volume.get('device')}@{volume.get('mount_point')}")

def disk_probe(directory: Path, samples: int = 20, files: int = 500) -> Dict[str, Any]:

    """F_FULLFSYNC (fsync elsewhere) latency of a 4 KiB write, and small-file creation rate."""

    import fcntl

    directory = Path(directory)

    directory.mkdir(parents=True, exist_ok=True)

    full = getattr(fcntl, "F_FULLFSYNC", None)

    latencies = []

    path = directory / ".oakbench-fsync-probe"

    for _ in range(samples):

        fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)

        try:

            os.write(fd, os.urandom(4096))

            start = time.perf_counter()

            if full is not None:

                fcntl.fcntl(fd, full)

            else:

                os.fsync(fd)

            latencies.append((time.perf_counter() - start) * 1000.0)

        finally:

            os.close(fd)

    path.unlink()

    tree = directory / ".oakbench-create-probe"

    tree.mkdir(exist_ok=True)

    start = time.perf_counter()

    for index in range(files):

        with open(tree / f"f{index}", "wb") as handle:

            handle.write(b"x" * 64)

    elapsed = time.perf_counter() - start

    for index in range(files):

        (tree / f"f{index}").unlink()

    tree.rmdir()

    return {"fullfsync_ms": [round(v, 4) for v in latencies],

            "fullfsync_kind": "F_FULLFSYNC" if full is not None else "fsync",

            "create_files": files, "create_elapsed_s": round(elapsed, 6), "work_volume": volume_identity(directory)}

def probe_rows(probe: Dict[str, Any], base: Dict[str, Any], phase: str) -> List[Dict[str, Any]]:

    """Diagnostic rows carrying probe samples (so the gate can recompute them)."""

    common = {**base, "subject": "harness", "subject_kind": "harness", "command": [], "returncode": 0,

              "diagnostic": True, "probe_phase": phase, "measurement_source": "harness_phase_wall_clock",

              "evidence_class": DEVELOPMENT_EVIDENCE_CLASS, "claim_eligible": False,

              "claim_ineligible_reasons": ["host probe (diagnostic)"], **host_facts(),

              "work_volume": probe.get("work_volume")}

    rows = [dict(common, operation="host.disk_fullfsync", run=index, elapsed_ms=value,

                 fsync_kind=probe["fullfsync_kind"]) for index, value in enumerate(probe["fullfsync_ms"])]

    rate = probe["create_files"] / probe["create_elapsed_s"] if probe["create_elapsed_s"] > 0 else 0.0

    rows.append(dict(common, operation="host.file_create_rate", run=0,

                     elapsed_ms=round(probe["create_elapsed_s"] * 1000.0, 3), files=probe["create_files"],

                     files_per_s=round(rate, 1)))

    return rows

def probe_verdict(rows: Iterable[Dict[str, Any]]) -> Tuple[List[str], Dict[str, Any]]:

    rows = list(rows)

    flush_rows = [r for r in rows if r.get("operation") == "host.disk_fullfsync"

                  and isinstance(r.get("elapsed_ms"), (int, float))]

    fsyncs = [r["elapsed_ms"] for r in flush_rows]

    rates = [r.get("files_per_s") for r in rows if r.get("operation") == "host.file_create_rate"]

    reasons, facts = [], {}

    if len(fsyncs) < 10 or not rates:

        return ["disk probes missing"], facts

    kinds = {r.get("fsync_kind") for r in flush_rows}

    systems = {(r.get("runner_profile") or {}).get("system") for r in flush_rows}

    expected = {expected_flush_kind(system) for system in systems if system} or {None}

    if len(kinds) != 1 or kinds != expected or next(iter(kinds)) not in MAX_FLUSH_MEDIAN_MS:

        reasons.append(f"flush probe kind {sorted(map(str, kinds))} is not the platform's full-flush "

                       f"primitive {sorted(map(str, expected))}")

        return reasons, facts

    kind = next(iter(kinds))

    for row in rows:

        if row.get("operation", "").startswith("host."):

            reasons += volume_problems(row.get("work_volume"))

    volumes = {volume_key(r.get("work_volume")) for r in rows if r.get("operation", "").startswith("host.")}

    if len(volumes) != 1:

        reasons.append(f"probes ran on several volumes {sorted(map(str, volumes))}")

    reasons = list(dict.fromkeys(reasons))

    facts["work_volume_key"] = next(iter(volumes)) if len(volumes) == 1 else None

    facts["fullfsync_median_ms"] = statistics.median(fsyncs)

    facts["create_files_per_s_min"] = min(r for r in rates if isinstance(r, (int, float)))

    limit = MAX_FLUSH_MEDIAN_MS[kind]

    if facts["fullfsync_median_ms"] > limit:

        reasons.append(f"disk contended: {kind} median {facts['fullfsync_median_ms']:.2f} ms > {limit} ms")

    if facts["create_files_per_s_min"] < MIN_CREATE_FILES_PER_S:

        reasons.append(f"disk contended: {facts['create_files_per_s_min']:.0f} files/s < {MIN_CREATE_FILES_PER_S:.0f}")

    return reasons, facts

# ---------------------------------------------------------------------------

# Records

# ---------------------------------------------------------------------------

def file_sha256(path: Path) -> str:

    digest = hashlib.sha256()

    with Path(path).open("rb") as handle:

        for chunk in iter(lambda: handle.read(1 << 20), b""):

            digest.update(chunk)

    return digest.hexdigest()

def record_digest(record: Dict[str, Any]) -> str:

    body = {key: value for key, value in record.items() if key != "record_sha256"}

    return hashlib.sha256(json.dumps(body, sort_keys=True, separators=(",", ":")).encode()).hexdigest()

def seal(record: Dict[str, Any]) -> Dict[str, Any]:

    sealed = dict(record)

    sealed.update(schema_version=SCHEMA_VERSION, kind=KIND, policy_version=POLICY_VERSION)

    sealed["record_sha256"] = record_digest(sealed)

    return sealed

def load_rows(path: Path) -> List[Dict[str, Any]]:

    return [json.loads(line) for line in Path(path).read_text().splitlines() if line.strip()]

def _pin_problems(rows: Iterable[Dict[str, Any]], registry: Optional[Dict[str, Any]]) -> List[str]:

    if registry is None:

        return []

    pins = {entry.get("spec_sha256"): entry.get("refs_sha256") for entry in registry.values()

            if isinstance(entry, dict) and entry.get("spec_sha256")}

    bad = set()

    for row in rows:

        if row.get("subject_kind") == "harness" or row.get("diagnostic"):

            continue

        spec, refs = row.get("fixture_spec_sha256"), row.get("fixture_manifest_sha256")

        if not spec or pins.get(spec) is None or pins.get(spec) != refs:

            bad.add(str(row.get("fixture_id")))

    return [f"fixture not registry-pinned or pin mismatch: {sorted(bad)[:3]}"] if bad else []

def campaign_identity(rows: Sequence[Dict[str, Any]]) -> Tuple[Dict[str, Any], List[str]]:

    """Runner/host/binary identity shared by every row of a run (or problems)."""

    reasons: List[str] = []

    identity: Dict[str, Any] = {}

    volumes = {volume_key(row.get("work_volume")) for row in rows}

    if len(volumes) != 1 or None in volumes:

        reasons.append(f"rows disagree on or lack work_volume: {sorted(map(str, volumes))[:3]}")

    else:

        identity["work_volume_key"] = volumes.pop()

    for key in ("runner_id", "host_cpu_count", "host_mem_bytes"):

        values = {row.get(key) for row in rows}

        if len(values) != 1 or None in values:

            reasons.append(f"rows disagree on or lack {key}: {sorted(map(str, values))[:3]}")

        else:

            identity[key] = values.pop()

    for kind in ("git", "oak"):

        shas = {row.get("binary_sha256") for row in rows if row.get("subject_kind") == kind

                and not row.get("diagnostic")}

        shas.discard(None)

        if len(shas) > 1:

            reasons.append(f"{kind} rows name several binaries")

        identity[f"{kind}_binary_sha256"] = shas.pop() if len(shas) == 1 else None

    return identity, reasons

def recompute(rows: Sequence[Dict[str, Any]], registry: Optional[Dict[str, Any]] = None

              ) -> Tuple[Dict[str, Any], List[str]]:

    """Recompute every calibration number from the campaign's own rows."""

    from oakbench import cellstats

    identity, reasons = campaign_identity(rows)

    measured = [r for r in rows if r.get("profile") == "scale" and r.get("returncode") == 0

                and not r.get("diagnostic") and isinstance(r.get("elapsed_ms"), (int, float))]

    value = lambda r: r.get("elapsed_ms")  # noqa: E731

    floors: List[float] = []

    false_claims = 0

    per_arm: Dict[str, Dict[str, Any]] = {}

    for arm, control in AA_PAIRS:

        runs = sorted({r.get("run") for r in measured if r.get("subject") == arm})

        if len(runs) < MIN_RUNS:

            reasons.append(f"{arm}: fewer than {MIN_RUNS} runs (two independent halves of >= 5)")

            continue

        pair = [r for r in measured if r.get("subject") in (arm, control)]

        # Per-arm floors over all runs: the A/A noise this arm pair shows.

        cell = cellstats.cell_floors(pair, arm, control, value)

        floors.extend(cell.values())

        arm_floors = sorted(cell.values())

        over = sum(1 for f in arm_floors if f > MAX_AA_FLOOR_MEDIAN_PCT)

        per_arm[arm] = {"cells": len(arm_floors),

                        "floor_median_pct": round(statistics.median(arm_floors), 3) if arm_floors else None,

                        "cells_over_threshold": over}

        # Held-out A/A claims: floors from the first half, verdicts on the second.

        first = set(runs[: len(runs) // 2])

        screen = [r for r in pair if r.get("run") in first]

        held = [r for r in pair if r.get("run") not in first]

        held_floors = cellstats.cell_floors(screen, arm, control, value)

        verdicts = cellstats.classify_cells(held, control, arm, held_floors, value_of=value)

        claims = sum(1 for item in verdicts.values() if item.get("verdict") in ("regression", "improvement"))

        per_arm[arm]["held_out_claims"] = claims

        false_claims += claims

    cpus = identity.get("host_cpu_count")

    loads = [r["environment_sample"]["loadavg"][0] for r in rows

             if isinstance((r.get("environment_sample") or {}).get("loadavg"), list)]

    load_per_core = max(loads) / cpus if loads and isinstance(cpus, int) and cpus > 0 else None

    disk_reasons, disk = probe_verdict(rows)

    facts = {**identity, "aa_cells": len(floors), "aa_per_arm": per_arm,

             "aa_floor_median_pct": round(statistics.median(floors), 3) if floors else None,

             "aa_false_claims": false_claims,

             "loadavg_per_core_max": round(load_per_core, 4) if load_per_core is not None else None, **disk}

    if len(floors) < MIN_AA_CELLS:

        reasons.append(f"too few A/A cells ({len(floors)} < {MIN_AA_CELLS})")

    if facts["aa_floor_median_pct"] is None or facts["aa_floor_median_pct"] > MAX_AA_FLOOR_MEDIAN_PCT:

        reasons.append(f"A/A floor median {facts['aa_floor_median_pct']}% > {MAX_AA_FLOOR_MEDIAN_PCT}%")

    if false_claims:

        reasons.append(f"identical-binary A/A produced {false_claims} held-out claim(s)")

    for arm, info in per_arm.items():

        if info["floor_median_pct"] is None or info["floor_median_pct"] > MAX_AA_FLOOR_MEDIAN_PCT:

            reasons.append(f"{arm} A/A floor median {info['floor_median_pct']}% > {MAX_AA_FLOOR_MEDIAN_PCT}%")

        if info["cells"] and info["cells_over_threshold"] / info["cells"] > MAX_AA_CELLS_OVER_FLOOR_SHARE:

            reasons.append(f"{arm}: {info['cells_over_threshold']}/{info['cells']} cells have an A/A floor over "

                           f"{MAX_AA_FLOOR_MEDIAN_PCT}% (broken or noisy A/A arm)")

    if load_per_core is None or load_per_core > MAX_LOAD_PER_CORE:

        reasons.append(f"host not quiet: load/core {load_per_core} > {MAX_LOAD_PER_CORE}")

    reasons += disk_reasons

    reasons += _pin_problems(rows, registry)

    return facts, reasons

def verify_campaign(record: Any, calibration_rows: Optional[Sequence[Dict[str, Any]]],

                    calibration_sha256: Optional[str], registry: Optional[Dict[str, Any]] = None

                    ) -> Tuple[Dict[str, Any], List[str]]:

    """Validate a record against its recomputed campaign; returns (facts, reasons)."""

    if not isinstance(record, dict):

        return {}, ["calibration record missing or not an object"]

    reasons = []

    if record.get("schema_version") != SCHEMA_VERSION or record.get("kind") != KIND:

        reasons.append("wrong schema_version/kind")

    if record.get("record_sha256") != record_digest(record):

        reasons.append("record_sha256 does not match content (edited record)")

    if calibration_rows is None:

        return {}, reasons + ["calibration campaign rows not provided (need the JSONL named by the record)"]

    if not calibration_sha256 or calibration_sha256 != record.get("calibration_rows_sha256"):

        return {}, reasons + ["calibration JSONL content sha256 does not match the record"]

    facts, problems = recompute(calibration_rows, registry)

    reasons += problems

    for key in ("runner_id", "host_cpu_count", "host_mem_bytes", "oak_binary_sha256", "git_binary_sha256",

                "work_volume_key", "aa_cells", "aa_false_claims"):

        if record.get(key) != facts.get(key):

            reasons.append(f"record {key} {record.get(key)!r} != recomputed {facts.get(key)!r}")

    return facts, reasons

def row_problems(row: Dict[str, Any], facts: Dict[str, Any]) -> List[str]:

    """A published row must come from the calibrated runner, host and binaries, on a quiet host."""

    reasons = []

    for key in ("runner_id", "host_cpu_count", "host_mem_bytes"):

        if row.get(key) != facts.get(key):

            reasons.append(f"row {key} {row.get(key)!r} != calibration {facts.get(key)!r}")

    if volume_key(row.get("work_volume")) != facts.get("work_volume_key"):

        reasons.append("row ran on a different work volume than the calibration")

    reasons += volume_problems(row.get("work_volume"))

    kind = row.get("subject_kind")

    if kind in ("git", "oak") and row.get("binary_sha256") != facts.get(f"{kind}_binary_sha256"):

        reasons.append(f"row {kind} binary differs from the calibrated binary")

    sample = row.get("environment_sample") or {}

    load = sample.get("loadavg") if isinstance(sample, dict) else None

    cpus = facts.get("host_cpu_count")

    if not isinstance(load, list) or not load or not isinstance(cpus, int) or load[0] / cpus > MAX_LOAD_PER_CORE:

        reasons.append("row host load/core not quiet")

    if row.get("fixture_pin_status") != "verified":

        reasons.append("row fixture not verified against its registry pin")

    if row.get("returncode") != 0:

        reasons.append("row is not a successful measurement")

    return reasons

def stamp(row: Dict[str, Any], record: Optional[Dict[str, Any]], facts: Optional[Dict[str, Any]],

          campaign_reasons: Sequence[str]) -> Dict[str, Any]:

    """Derive evidence_class / claim_eligible for one row of a run."""

    if record is None:

        reasons = ["no calibration record"]

    else:

        reasons = list(campaign_reasons) or row_problems(row, facts or {})

    if reasons:

        row["evidence_class"] = DEVELOPMENT_EVIDENCE_CLASS

        row["claim_eligible"] = False

        row["claim_ineligible_reasons"] = reasons[:5]

        for key in ("calibration_record", "calibration_record_sha256"):

            row.pop(key, None)

    else:

        row["evidence_class"] = CALIBRATED_EVIDENCE_CLASS

        row["claim_eligible"] = True

        row["claim_ineligible_reasons"] = []

        row["calibration_record"] = record

        row["calibration_record_sha256"] = record["record_sha256"]

    return row

def gate(scale_rows: Sequence[Tuple[str, Dict[str, Any]]], inputs: Dict[str, List[Dict[str, Any]]],

         input_sha256: Dict[str, str], registry: Optional[Dict[str, Any]]) -> List[Tuple[str, Any, Any, Any, str]]:

    """publish_gate: every publishable scale row is re-verified by recomputation.

    ``inputs``/``input_sha256`` map each input path to its rows and content

    sha256; the calibration JSONL must be one of them. Returns refusals.

    """

    by_sha = {sha: path for path, sha in input_sha256.items()}

    # Inputs that are the calibration campaigns referenced by rows are evidence

    # FOR the gate, not measurements being published.

    campaign_paths = {by_sha.get((row.get("calibration_record") or {}).get("calibration_rows_sha256"))

                      for _, row in scale_rows if isinstance(row.get("calibration_record"), dict)}

    campaign_paths.discard(None)

    verified: Dict[str, Tuple[Dict[str, Any], List[str]]] = {}

    run_probe_cache: Dict[Tuple[str, str], List[str]] = {}

    refused = []

    for path, row in scale_rows:

        if path in campaign_paths:

            continue

        if row.get("diagnostic") or row.get("returncode") == 77 or row.get("subject_kind") == "harness":

            continue

        where = (path, row.get("scenario"), row.get("operation"), row.get("subject"))

        if row.get("evidence_class") != CALIBRATED_EVIDENCE_CLASS or row.get("claim_eligible") is not True:

            refused.append((*where, f"evidence_class {row.get('evidence_class')!r} is not calibrated"))

            continue

        record = row.get("calibration_record")

        if not isinstance(record, dict) or row.get("calibration_record_sha256") != record.get("record_sha256"):

            refused.append((*where, "calibration record missing or its digest does not match the row"))

            continue

        key = record.get("record_sha256")

        if key not in verified:

            cal_sha = record.get("calibration_rows_sha256")

            cal_path = by_sha.get(cal_sha)

            verified[key] = verify_campaign(record, inputs.get(cal_path) if cal_path else None,

                                            cal_sha if cal_path else None, registry)

        facts, reasons = verified[key]

        if reasons:

            refused.append((*where, reasons[0]))

            continue

        problems = row_problems(row, facts)

        probe_key = (path, str(row.get("bench_id")))

        if probe_key not in run_probe_cache:

            run_rows = [r for r in inputs.get(path, []) if r.get("bench_id") == row.get("bench_id")]

            run_probe_cache[probe_key] = probe_verdict(run_rows)[0]

        problems += [f"published run: {reason}" for reason in run_probe_cache[probe_key]]

        if problems:

            refused.append((*where, problems[0]))

    return refused