| 1 | """Quiet-host calibration for the scale lane: evidence verified by recomputation.
|
| 2 |
|
| 3 | A scale row's ``claim_eligible`` / ``evidence_class`` are DERIVED, never a
|
| 4 | free flag, and nothing a calibration record *states* is trusted:
|
| 5 |
|
| 6 | 1. ``run_scale_suite.py calibrate`` runs an identical-binary A/A campaign
|
| 7 | (git, git_aa, oak_installed, oak_installed_aa) plus host probes (load
|
| 8 | average, small-file creation rate, F_FULLFSYNC latency) and writes its rows
|
| 9 | to a JSONL file. The record names that file's content sha256.
|
| 10 | 2. Publishing requires that calibration JSONL among publish_gate's inputs,
|
| 11 | matched by content sha256. The gate RECOMPUTES the A/A cell count, per-cell
|
| 12 | floors (oakbench.cellstats, the P1 machinery), identical-binary false
|
| 13 | claims, host load and disk probes from those rows, and checks that the
|
| 14 | calibration rows and the published rows share runner identity, host
|
| 15 | cpu_count and physical memory, the Oak and Git binary sha256, and
|
| 16 | registry-pinned fixtures. The published run must carry its own quiet disk
|
| 17 | probes too.
|
| 18 |
|
| 19 | This detects edited, resealed, copied or invented RECORDS (the sealed digest
|
| 20 | is only an integrity check of the record's own content). It does not
|
| 21 | authenticate the campaign JSONL: editing its load samples and probe values
|
| 22 | (a few dozen numbers) and resealing a record to the new file digest passes, as
|
| 23 | would a fabricated campaign. Closing that needs a signed runner, which the
|
| 24 | lane does not have. Do not describe it as forgery-proof.
|
| 25 | """
|
| 26 |
|
| 27 | from __future__ import annotations
|
| 28 |
|
| 29 | import hashlib
|
| 30 | import json
|
| 31 | import os
|
| 32 | import statistics
|
| 33 | import time
|
| 34 | from pathlib import Path
|
| 35 | from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple
|
| 36 |
|
| 37 | SCHEMA_VERSION = 2
|
| 38 | KIND = "oakbench_quiet_host_calibration"
|
| 39 | CALIBRATED_EVIDENCE_CLASS = "calibrated_quiet_host"
|
| 40 | DEVELOPMENT_EVIDENCE_CLASS = "development_only_loaded_host"
|
| 41 | POLICY_VERSION = "2026-09-29.3"
|
| 42 | # Quiet-host policy (provisional; re-derive on the dedicated runner and bump
|
| 43 | # POLICY_VERSION). Load/core 0.1 is ~1.4 runnable threads on a 14-core host.
|
| 44 | MAX_LOAD_PER_CORE = 0.10
|
| 45 | MIN_AA_CELLS = 20
|
| 46 | # A calibration needs two independent halves of >= 5 runs: floors come from
|
| 47 | # one, A/A claims are counted on the other (a floor computed from the same
|
| 48 | # A/A pair it judges can never produce a claim).
|
| 49 | MIN_RUNS = 10
|
| 50 | MAX_AA_FLOOR_MEDIAN_PCT = 10.0
|
| 51 | # Per arm: the median A/A floor AND the share of cells whose floor exceeds
|
| 52 | # MAX_AA_FLOOR_MEDIAN_PCT must be small (a broken A/A arm is systematic).
|
| 53 | MAX_AA_CELLS_OVER_FLOOR_SHARE = 0.10
|
| 54 | # Disk contention (load average does not see it; Oak clone/switch are
|
| 55 | # fsync-bound, fb-549). Measured on the loaded dev host: F_FULLFSYNC median
|
| 56 | # 3.7 ms, file creation as low as 170 files/s.
|
| 57 | # Per flush primitive; a probe must use the platform's full-flush primitive
|
| 58 | # (F_FULLFSYNC on macOS: plain fsync there skips the drive-cache flush).
|
| 59 | # PROVISIONAL: no idle baseline was obtainable on the shared dev host (internal
|
| 60 | # SSD, loaded: F_FULLFSYNC 3.7-3.9 ms); 2.5 ms may be unreachable even idle.
|
| 61 | # The dedicated runner must re-derive these (P6) and bump POLICY_VERSION.
|
| 62 | MAX_FLUSH_MEDIAN_MS = {"F_FULLFSYNC": 2.5, "fsync": 2.5}
|
| 63 | MAX_FULLFSYNC_MEDIAN_MS = MAX_FLUSH_MEDIAN_MS["F_FULLFSYNC"]
|
| 64 | MIN_CREATE_FILES_PER_S = 2000.0
|
| 65 | # Work volumes whose flushes do not reach durable media cannot certify runs.
|
| 66 | ACCEPTED_MEDIA = ("physical",)
|
| 67 | PROBE_OPERATIONS = ("host.disk_fullfsync", "host.file_create_rate")
|
| 68 | AA_PAIRS = (("git", "git_aa"), ("oak_installed", "oak_installed_aa"))
|
| 69 |
|
| 70 |
|
| 71 | # ---------------------------------------------------------------------------
|
| 72 | # Host facts and probes
|
| 73 | # ---------------------------------------------------------------------------
|
| 74 |
|
| 75 |
|
| 76 | def host_facts() -> Dict[str, Any]:
|
| 77 | try:
|
| 78 | mem = int(os.sysconf("SC_PAGE_SIZE")) * int(os.sysconf("SC_PHYS_PAGES"))
|
| 79 | except (ValueError, OSError, AttributeError):
|
| 80 | mem = None
|
| 81 | return {"host_cpu_count": os.cpu_count(), "host_mem_bytes": mem}
|
| 82 |
|
| 83 |
|
| 84 | def expected_flush_kind(system: Optional[str] = None) -> str:
|
| 85 | import platform
|
| 86 | return "F_FULLFSYNC" if (system or platform.system()) == "Darwin" else "fsync"
|
| 87 |
|
| 88 |
|
| 89 | def _diskutil(target: str) -> Dict[str, Any]:
|
| 90 | import plistlib
|
| 91 | import subprocess
|
| 92 | try:
|
| 93 | out = subprocess.run(["/usr/sbin/diskutil", "info", "-plist", target], capture_output=True, timeout=20)
|
| 94 | except (OSError, subprocess.TimeoutExpired):
|
| 95 | return {}
|
| 96 | if out.returncode != 0:
|
| 97 | return {}
|
| 98 | try:
|
| 99 | return plistlib.loads(out.stdout)
|
| 100 | except Exception: # noqa: BLE001 - malformed output means unknown medium
|
| 101 | return {}
|
| 102 |
|
| 103 |
|
| 104 | def classify_medium(volume: Dict[str, Any], store: Dict[str, Any]) -> str:
|
| 105 | """physical | ram_or_disk_image | tmpfs | unknown (only physical certifies)."""
|
| 106 | fstype = str(volume.get("fstype") or "").lower()
|
| 107 | if fstype in ("tmpfs", "ramfs", "devfs", "overlay", "squashfs"):
|
| 108 | return "tmpfs"
|
| 109 | info = store or volume.get("_diskutil") or {}
|
| 110 | if not info:
|
| 111 | return "unknown"
|
| 112 | if info.get("BusProtocol") == "Disk Image" or info.get("VirtualOrPhysical") == "Virtual" \
|
| 113 | or "Disk Image" in str(info.get("MediaName") or ""):
|
| 114 | return "ram_or_disk_image"
|
| 115 | return "physical" if info.get("BusProtocol") else "unknown"
|
| 116 |
|
| 117 |
|
| 118 | def volume_identity(path: Path) -> Dict[str, Any]:
|
| 119 | """Filesystem type, mount point and backing device/medium of ``path``'s volume."""
|
| 120 | import platform
|
| 121 | import subprocess
|
| 122 | path = Path(path)
|
| 123 | probe = path if path.exists() else path.parent
|
| 124 | identity: Dict[str, Any] = {"mount_point": None, "device": None, "fstype": None, "volume_uuid": None,
|
| 125 | "bus_protocol": None, "physical_store": None, "medium": "unknown"}
|
| 126 | try:
|
| 127 | df = subprocess.run(["df", "-P", str(probe)], capture_output=True, text=True, timeout=20).stdout
|
| 128 | fields = df.strip().splitlines()[-1].split()
|
| 129 | identity["device"], identity["mount_point"] = fields[0], fields[-1]
|
| 130 | except (OSError, IndexError, subprocess.TimeoutExpired):
|
| 131 | return identity
|
| 132 | if platform.system() == "Darwin":
|
| 133 | info = _diskutil(identity["mount_point"])
|
| 134 | identity["fstype"] = info.get("FilesystemType")
|
| 135 | identity["volume_uuid"] = info.get("VolumeUUID")
|
| 136 | store_info: Dict[str, Any] = {}
|
| 137 | stores = info.get("APFSPhysicalStores") or []
|
| 138 | if stores:
|
| 139 | store = stores[0].get("APFSPhysicalStore")
|
| 140 | identity["physical_store"] = store
|
| 141 | store_info = _diskutil(store) if store else {}
|
| 142 | else:
|
| 143 | store_info = info
|
| 144 | identity["bus_protocol"] = store_info.get("BusProtocol")
|
| 145 | identity["medium"] = classify_medium(identity, store_info)
|
| 146 | else:
|
| 147 | try:
|
| 148 | for line in Path("/proc/mounts").read_text().splitlines():
|
| 149 | parts = line.split()
|
| 150 | if len(parts) >= 3 and parts[1] == identity["mount_point"]:
|
| 151 | identity["fstype"] = parts[2]
|
| 152 | except OSError:
|
| 153 | pass
|
| 154 | fstype = str(identity["fstype"] or "").lower()
|
| 155 | if fstype in ("tmpfs", "ramfs", "overlay", "squashfs"):
|
| 156 | identity["medium"] = "tmpfs"
|
| 157 | elif str(identity["device"]).startswith("/dev/") and not str(identity["device"]).startswith("/dev/ram") \
|
| 158 | and not str(identity["device"]).startswith("/dev/zram"):
|
| 159 | identity["medium"] = "physical"
|
| 160 | identity["volume_uuid"] = identity["device"]
|
| 161 | return identity
|
| 162 |
|
| 163 |
|
| 164 | def volume_problems(volume: Any) -> List[str]:
|
| 165 | if not isinstance(volume, dict):
|
| 166 | return ["work volume unrecorded"]
|
| 167 | if volume.get("medium") not in ACCEPTED_MEDIA:
|
| 168 | return [f"work volume medium {volume.get('medium')!r} ({volume.get('fstype')}, {volume.get('device')}) "
|
| 169 | "cannot certify durable-flush costs"]
|
| 170 | return []
|
| 171 |
|
| 172 |
|
| 173 | def volume_key(volume: Any) -> Optional[str]:
|
| 174 | if not isinstance(volume, dict):
|
| 175 | return None
|
| 176 | return str(volume.get("volume_uuid") or f"{volume.get('device')}@{volume.get('mount_point')}")
|
| 177 |
|
| 178 |
|
| 179 | def disk_probe(directory: Path, samples: int = 20, files: int = 500) -> Dict[str, Any]:
|
| 180 | """F_FULLFSYNC (fsync elsewhere) latency of a 4 KiB write, and small-file creation rate."""
|
| 181 | import fcntl
|
| 182 | directory = Path(directory)
|
| 183 | directory.mkdir(parents=True, exist_ok=True)
|
| 184 | full = getattr(fcntl, "F_FULLFSYNC", None)
|
| 185 | latencies = []
|
| 186 | path = directory / ".oakbench-fsync-probe"
|
| 187 | for _ in range(samples):
|
| 188 | fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
|
| 189 | try:
|
| 190 | os.write(fd, os.urandom(4096))
|
| 191 | start = time.perf_counter()
|
| 192 | if full is not None:
|
| 193 | fcntl.fcntl(fd, full)
|
| 194 | else:
|
| 195 | os.fsync(fd)
|
| 196 | latencies.append((time.perf_counter() - start) * 1000.0)
|
| 197 | finally:
|
| 198 | os.close(fd)
|
| 199 | path.unlink()
|
| 200 | tree = directory / ".oakbench-create-probe"
|
| 201 | tree.mkdir(exist_ok=True)
|
| 202 | start = time.perf_counter()
|
| 203 | for index in range(files):
|
| 204 | with open(tree / f"f{index}", "wb") as handle:
|
| 205 | handle.write(b"x" * 64)
|
| 206 | elapsed = time.perf_counter() - start
|
| 207 | for index in range(files):
|
| 208 | (tree / f"f{index}").unlink()
|
| 209 | tree.rmdir()
|
| 210 | return {"fullfsync_ms": [round(v, 4) for v in latencies],
|
| 211 | "fullfsync_kind": "F_FULLFSYNC" if full is not None else "fsync",
|
| 212 | "create_files": files, "create_elapsed_s": round(elapsed, 6), "work_volume": volume_identity(directory)}
|
| 213 |
|
| 214 |
|
| 215 | def probe_rows(probe: Dict[str, Any], base: Dict[str, Any], phase: str) -> List[Dict[str, Any]]:
|
| 216 | """Diagnostic rows carrying probe samples (so the gate can recompute them)."""
|
| 217 | common = {**base, "subject": "harness", "subject_kind": "harness", "command": [], "returncode": 0,
|
| 218 | "diagnostic": True, "probe_phase": phase, "measurement_source": "harness_phase_wall_clock",
|
| 219 | "evidence_class": DEVELOPMENT_EVIDENCE_CLASS, "claim_eligible": False,
|
| 220 | "claim_ineligible_reasons": ["host probe (diagnostic)"], **host_facts(),
|
| 221 | "work_volume": probe.get("work_volume")}
|
| 222 | rows = [dict(common, operation="host.disk_fullfsync", run=index, elapsed_ms=value,
|
| 223 | fsync_kind=probe["fullfsync_kind"]) for index, value in enumerate(probe["fullfsync_ms"])]
|
| 224 | rate = probe["create_files"] / probe["create_elapsed_s"] if probe["create_elapsed_s"] > 0 else 0.0
|
| 225 | rows.append(dict(common, operation="host.file_create_rate", run=0,
|
| 226 | elapsed_ms=round(probe["create_elapsed_s"] * 1000.0, 3), files=probe["create_files"],
|
| 227 | files_per_s=round(rate, 1)))
|
| 228 | return rows
|
| 229 |
|
| 230 |
|
| 231 | def probe_verdict(rows: Iterable[Dict[str, Any]]) -> Tuple[List[str], Dict[str, Any]]:
|
| 232 | rows = list(rows)
|
| 233 | flush_rows = [r for r in rows if r.get("operation") == "host.disk_fullfsync"
|
| 234 | and isinstance(r.get("elapsed_ms"), (int, float))]
|
| 235 | fsyncs = [r["elapsed_ms"] for r in flush_rows]
|
| 236 | rates = [r.get("files_per_s") for r in rows if r.get("operation") == "host.file_create_rate"]
|
| 237 | reasons, facts = [], {}
|
| 238 | if len(fsyncs) < 10 or not rates:
|
| 239 | return ["disk probes missing"], facts
|
| 240 | kinds = {r.get("fsync_kind") for r in flush_rows}
|
| 241 | systems = {(r.get("runner_profile") or {}).get("system") for r in flush_rows}
|
| 242 | expected = {expected_flush_kind(system) for system in systems if system} or {None}
|
| 243 | if len(kinds) != 1 or kinds != expected or next(iter(kinds)) not in MAX_FLUSH_MEDIAN_MS:
|
| 244 | reasons.append(f"flush probe kind {sorted(map(str, kinds))} is not the platform's full-flush "
|
| 245 | f"primitive {sorted(map(str, expected))}")
|
| 246 | return reasons, facts
|
| 247 | kind = next(iter(kinds))
|
| 248 | for row in rows:
|
| 249 | if row.get("operation", "").startswith("host."):
|
| 250 | reasons += volume_problems(row.get("work_volume"))
|
| 251 | volumes = {volume_key(r.get("work_volume")) for r in rows if r.get("operation", "").startswith("host.")}
|
| 252 | if len(volumes) != 1:
|
| 253 | reasons.append(f"probes ran on several volumes {sorted(map(str, volumes))}")
|
| 254 | reasons = list(dict.fromkeys(reasons))
|
| 255 | facts["work_volume_key"] = next(iter(volumes)) if len(volumes) == 1 else None
|
| 256 | facts["fullfsync_median_ms"] = statistics.median(fsyncs)
|
| 257 | facts["create_files_per_s_min"] = min(r for r in rates if isinstance(r, (int, float)))
|
| 258 | limit = MAX_FLUSH_MEDIAN_MS[kind]
|
| 259 | if facts["fullfsync_median_ms"] > limit:
|
| 260 | reasons.append(f"disk contended: {kind} median {facts['fullfsync_median_ms']:.2f} ms > {limit} ms")
|
| 261 | if facts["create_files_per_s_min"] < MIN_CREATE_FILES_PER_S:
|
| 262 | reasons.append(f"disk contended: {facts['create_files_per_s_min']:.0f} files/s < {MIN_CREATE_FILES_PER_S:.0f}")
|
| 263 | return reasons, facts
|
| 264 |
|
| 265 |
|
| 266 | # ---------------------------------------------------------------------------
|
| 267 | # Records
|
| 268 | # ---------------------------------------------------------------------------
|
| 269 |
|
| 270 |
|
| 271 | def file_sha256(path: Path) -> str:
|
| 272 | digest = hashlib.sha256()
|
| 273 | with Path(path).open("rb") as handle:
|
| 274 | for chunk in iter(lambda: handle.read(1 << 20), b""):
|
| 275 | digest.update(chunk)
|
| 276 | return digest.hexdigest()
|
| 277 |
|
| 278 |
|
| 279 | def record_digest(record: Dict[str, Any]) -> str:
|
| 280 | body = {key: value for key, value in record.items() if key != "record_sha256"}
|
| 281 | return hashlib.sha256(json.dumps(body, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
|
| 282 |
|
| 283 |
|
| 284 | def seal(record: Dict[str, Any]) -> Dict[str, Any]:
|
| 285 | sealed = dict(record)
|
| 286 | sealed.update(schema_version=SCHEMA_VERSION, kind=KIND, policy_version=POLICY_VERSION)
|
| 287 | sealed["record_sha256"] = record_digest(sealed)
|
| 288 | return sealed
|
| 289 |
|
| 290 |
|
| 291 | def load_rows(path: Path) -> List[Dict[str, Any]]:
|
| 292 | return [json.loads(line) for line in Path(path).read_text().splitlines() if line.strip()]
|
| 293 |
|
| 294 |
|
| 295 | def _pin_problems(rows: Iterable[Dict[str, Any]], registry: Optional[Dict[str, Any]]) -> List[str]:
|
| 296 | if registry is None:
|
| 297 | return []
|
| 298 | pins = {entry.get("spec_sha256"): entry.get("refs_sha256") for entry in registry.values()
|
| 299 | if isinstance(entry, dict) and entry.get("spec_sha256")}
|
| 300 | bad = set()
|
| 301 | for row in rows:
|
| 302 | if row.get("subject_kind") == "harness" or row.get("diagnostic"):
|
| 303 | continue
|
| 304 | spec, refs = row.get("fixture_spec_sha256"), row.get("fixture_manifest_sha256")
|
| 305 | if not spec or pins.get(spec) is None or pins.get(spec) != refs:
|
| 306 | bad.add(str(row.get("fixture_id")))
|
| 307 | return [f"fixture not registry-pinned or pin mismatch: {sorted(bad)[:3]}"] if bad else []
|
| 308 |
|
| 309 |
|
| 310 | def campaign_identity(rows: Sequence[Dict[str, Any]]) -> Tuple[Dict[str, Any], List[str]]:
|
| 311 | """Runner/host/binary identity shared by every row of a run (or problems)."""
|
| 312 | reasons: List[str] = []
|
| 313 | identity: Dict[str, Any] = {}
|
| 314 | volumes = {volume_key(row.get("work_volume")) for row in rows}
|
| 315 | if len(volumes) != 1 or None in volumes:
|
| 316 | reasons.append(f"rows disagree on or lack work_volume: {sorted(map(str, volumes))[:3]}")
|
| 317 | else:
|
| 318 | identity["work_volume_key"] = volumes.pop()
|
| 319 | for key in ("runner_id", "host_cpu_count", "host_mem_bytes"):
|
| 320 | values = {row.get(key) for row in rows}
|
| 321 | if len(values) != 1 or None in values:
|
| 322 | reasons.append(f"rows disagree on or lack {key}: {sorted(map(str, values))[:3]}")
|
| 323 | else:
|
| 324 | identity[key] = values.pop()
|
| 325 | for kind in ("git", "oak"):
|
| 326 | shas = {row.get("binary_sha256") for row in rows if row.get("subject_kind") == kind
|
| 327 | and not row.get("diagnostic")}
|
| 328 | shas.discard(None)
|
| 329 | if len(shas) > 1:
|
| 330 | reasons.append(f"{kind} rows name several binaries")
|
| 331 | identity[f"{kind}_binary_sha256"] = shas.pop() if len(shas) == 1 else None
|
| 332 | return identity, reasons
|
| 333 |
|
| 334 |
|
| 335 | def recompute(rows: Sequence[Dict[str, Any]], registry: Optional[Dict[str, Any]] = None
|
| 336 | ) -> Tuple[Dict[str, Any], List[str]]:
|
| 337 | """Recompute every calibration number from the campaign's own rows."""
|
| 338 | from oakbench import cellstats
|
| 339 | identity, reasons = campaign_identity(rows)
|
| 340 | measured = [r for r in rows if r.get("profile") == "scale" and r.get("returncode") == 0
|
| 341 | and not r.get("diagnostic") and isinstance(r.get("elapsed_ms"), (int, float))]
|
| 342 | value = lambda r: r.get("elapsed_ms") # noqa: E731
|
| 343 | floors: List[float] = []
|
| 344 | false_claims = 0
|
| 345 | per_arm: Dict[str, Dict[str, Any]] = {}
|
| 346 | for arm, control in AA_PAIRS:
|
| 347 | runs = sorted({r.get("run") for r in measured if r.get("subject") == arm})
|
| 348 | if len(runs) < MIN_RUNS:
|
| 349 | reasons.append(f"{arm}: fewer than {MIN_RUNS} runs (two independent halves of >= 5)")
|
| 350 | continue
|
| 351 | pair = [r for r in measured if r.get("subject") in (arm, control)]
|
| 352 | # Per-arm floors over all runs: the A/A noise this arm pair shows.
|
| 353 | cell = cellstats.cell_floors(pair, arm, control, value)
|
| 354 | floors.extend(cell.values())
|
| 355 | arm_floors = sorted(cell.values())
|
| 356 | over = sum(1 for f in arm_floors if f > MAX_AA_FLOOR_MEDIAN_PCT)
|
| 357 | per_arm[arm] = {"cells": len(arm_floors),
|
| 358 | "floor_median_pct": round(statistics.median(arm_floors), 3) if arm_floors else None,
|
| 359 | "cells_over_threshold": over}
|
| 360 | # Held-out A/A claims: floors from the first half, verdicts on the second.
|
| 361 | first = set(runs[: len(runs) // 2])
|
| 362 | screen = [r for r in pair if r.get("run") in first]
|
| 363 | held = [r for r in pair if r.get("run") not in first]
|
| 364 | held_floors = cellstats.cell_floors(screen, arm, control, value)
|
| 365 | verdicts = cellstats.classify_cells(held, control, arm, held_floors, value_of=value)
|
| 366 | claims = sum(1 for item in verdicts.values() if item.get("verdict") in ("regression", "improvement"))
|
| 367 | per_arm[arm]["held_out_claims"] = claims
|
| 368 | false_claims += claims
|
| 369 | cpus = identity.get("host_cpu_count")
|
| 370 | loads = [r["environment_sample"]["loadavg"][0] for r in rows
|
| 371 | if isinstance((r.get("environment_sample") or {}).get("loadavg"), list)]
|
| 372 | load_per_core = max(loads) / cpus if loads and isinstance(cpus, int) and cpus > 0 else None
|
| 373 | disk_reasons, disk = probe_verdict(rows)
|
| 374 | facts = {**identity, "aa_cells": len(floors), "aa_per_arm": per_arm,
|
| 375 | "aa_floor_median_pct": round(statistics.median(floors), 3) if floors else None,
|
| 376 | "aa_false_claims": false_claims,
|
| 377 | "loadavg_per_core_max": round(load_per_core, 4) if load_per_core is not None else None, **disk}
|
| 378 | if len(floors) < MIN_AA_CELLS:
|
| 379 | reasons.append(f"too few A/A cells ({len(floors)} < {MIN_AA_CELLS})")
|
| 380 | if facts["aa_floor_median_pct"] is None or facts["aa_floor_median_pct"] > MAX_AA_FLOOR_MEDIAN_PCT:
|
| 381 | reasons.append(f"A/A floor median {facts['aa_floor_median_pct']}% > {MAX_AA_FLOOR_MEDIAN_PCT}%")
|
| 382 | if false_claims:
|
| 383 | reasons.append(f"identical-binary A/A produced {false_claims} held-out claim(s)")
|
| 384 | for arm, info in per_arm.items():
|
| 385 | if info["floor_median_pct"] is None or info["floor_median_pct"] > MAX_AA_FLOOR_MEDIAN_PCT:
|
| 386 | reasons.append(f"{arm} A/A floor median {info['floor_median_pct']}% > {MAX_AA_FLOOR_MEDIAN_PCT}%")
|
| 387 | if info["cells"] and info["cells_over_threshold"] / info["cells"] > MAX_AA_CELLS_OVER_FLOOR_SHARE:
|
| 388 | reasons.append(f"{arm}: {info['cells_over_threshold']}/{info['cells']} cells have an A/A floor over "
|
| 389 | f"{MAX_AA_FLOOR_MEDIAN_PCT}% (broken or noisy A/A arm)")
|
| 390 | if load_per_core is None or load_per_core > MAX_LOAD_PER_CORE:
|
| 391 | reasons.append(f"host not quiet: load/core {load_per_core} > {MAX_LOAD_PER_CORE}")
|
| 392 | reasons += disk_reasons
|
| 393 | reasons += _pin_problems(rows, registry)
|
| 394 | return facts, reasons
|
| 395 |
|
| 396 |
|
| 397 | def verify_campaign(record: Any, calibration_rows: Optional[Sequence[Dict[str, Any]]],
|
| 398 | calibration_sha256: Optional[str], registry: Optional[Dict[str, Any]] = None
|
| 399 | ) -> Tuple[Dict[str, Any], List[str]]:
|
| 400 | """Validate a record against its recomputed campaign; returns (facts, reasons)."""
|
| 401 | if not isinstance(record, dict):
|
| 402 | return {}, ["calibration record missing or not an object"]
|
| 403 | reasons = []
|
| 404 | if record.get("schema_version") != SCHEMA_VERSION or record.get("kind") != KIND:
|
| 405 | reasons.append("wrong schema_version/kind")
|
| 406 | if record.get("record_sha256") != record_digest(record):
|
| 407 | reasons.append("record_sha256 does not match content (edited record)")
|
| 408 | if calibration_rows is None:
|
| 409 | return {}, reasons + ["calibration campaign rows not provided (need the JSONL named by the record)"]
|
| 410 | if not calibration_sha256 or calibration_sha256 != record.get("calibration_rows_sha256"):
|
| 411 | return {}, reasons + ["calibration JSONL content sha256 does not match the record"]
|
| 412 | facts, problems = recompute(calibration_rows, registry)
|
| 413 | reasons += problems
|
| 414 | for key in ("runner_id", "host_cpu_count", "host_mem_bytes", "oak_binary_sha256", "git_binary_sha256",
|
| 415 | "work_volume_key", "aa_cells", "aa_false_claims"):
|
| 416 | if record.get(key) != facts.get(key):
|
| 417 | reasons.append(f"record {key} {record.get(key)!r} != recomputed {facts.get(key)!r}")
|
| 418 | return facts, reasons
|
| 419 |
|
| 420 |
|
| 421 | def row_problems(row: Dict[str, Any], facts: Dict[str, Any]) -> List[str]:
|
| 422 | """A published row must come from the calibrated runner, host and binaries, on a quiet host."""
|
| 423 | reasons = []
|
| 424 | for key in ("runner_id", "host_cpu_count", "host_mem_bytes"):
|
| 425 | if row.get(key) != facts.get(key):
|
| 426 | reasons.append(f"row {key} {row.get(key)!r} != calibration {facts.get(key)!r}")
|
| 427 | if volume_key(row.get("work_volume")) != facts.get("work_volume_key"):
|
| 428 | reasons.append("row ran on a different work volume than the calibration")
|
| 429 | reasons += volume_problems(row.get("work_volume"))
|
| 430 | kind = row.get("subject_kind")
|
| 431 | if kind in ("git", "oak") and row.get("binary_sha256") != facts.get(f"{kind}_binary_sha256"):
|
| 432 | reasons.append(f"row {kind} binary differs from the calibrated binary")
|
| 433 | sample = row.get("environment_sample") or {}
|
| 434 | load = sample.get("loadavg") if isinstance(sample, dict) else None
|
| 435 | cpus = facts.get("host_cpu_count")
|
| 436 | if not isinstance(load, list) or not load or not isinstance(cpus, int) or load[0] / cpus > MAX_LOAD_PER_CORE:
|
| 437 | reasons.append("row host load/core not quiet")
|
| 438 | if row.get("fixture_pin_status") != "verified":
|
| 439 | reasons.append("row fixture not verified against its registry pin")
|
| 440 | if row.get("returncode") != 0:
|
| 441 | reasons.append("row is not a successful measurement")
|
| 442 | return reasons
|
| 443 |
|
| 444 |
|
| 445 | def stamp(row: Dict[str, Any], record: Optional[Dict[str, Any]], facts: Optional[Dict[str, Any]],
|
| 446 | campaign_reasons: Sequence[str]) -> Dict[str, Any]:
|
| 447 | """Derive evidence_class / claim_eligible for one row of a run."""
|
| 448 | if record is None:
|
| 449 | reasons = ["no calibration record"]
|
| 450 | else:
|
| 451 | reasons = list(campaign_reasons) or row_problems(row, facts or {})
|
| 452 | if reasons:
|
| 453 | row["evidence_class"] = DEVELOPMENT_EVIDENCE_CLASS
|
| 454 | row["claim_eligible"] = False
|
| 455 | row["claim_ineligible_reasons"] = reasons[:5]
|
| 456 | for key in ("calibration_record", "calibration_record_sha256"):
|
| 457 | row.pop(key, None)
|
| 458 | else:
|
| 459 | row["evidence_class"] = CALIBRATED_EVIDENCE_CLASS
|
| 460 | row["claim_eligible"] = True
|
| 461 | row["claim_ineligible_reasons"] = []
|
| 462 | row["calibration_record"] = record
|
| 463 | row["calibration_record_sha256"] = record["record_sha256"]
|
| 464 | return row
|
| 465 |
|
| 466 |
|
| 467 | def gate(scale_rows: Sequence[Tuple[str, Dict[str, Any]]], inputs: Dict[str, List[Dict[str, Any]]],
|
| 468 | input_sha256: Dict[str, str], registry: Optional[Dict[str, Any]]) -> List[Tuple[str, Any, Any, Any, str]]:
|
| 469 | """publish_gate: every publishable scale row is re-verified by recomputation.
|
| 470 |
|
| 471 | ``inputs``/``input_sha256`` map each input path to its rows and content
|
| 472 | sha256; the calibration JSONL must be one of them. Returns refusals.
|
| 473 | """
|
| 474 | by_sha = {sha: path for path, sha in input_sha256.items()}
|
| 475 | # Inputs that are the calibration campaigns referenced by rows are evidence
|
| 476 | # FOR the gate, not measurements being published.
|
| 477 | campaign_paths = {by_sha.get((row.get("calibration_record") or {}).get("calibration_rows_sha256"))
|
| 478 | for _, row in scale_rows if isinstance(row.get("calibration_record"), dict)}
|
| 479 | campaign_paths.discard(None)
|
| 480 | verified: Dict[str, Tuple[Dict[str, Any], List[str]]] = {}
|
| 481 | run_probe_cache: Dict[Tuple[str, str], List[str]] = {}
|
| 482 | refused = []
|
| 483 | for path, row in scale_rows:
|
| 484 | if path in campaign_paths:
|
| 485 | continue
|
| 486 | if row.get("diagnostic") or row.get("returncode") == 77 or row.get("subject_kind") == "harness":
|
| 487 | continue
|
| 488 | where = (path, row.get("scenario"), row.get("operation"), row.get("subject"))
|
| 489 | if row.get("evidence_class") != CALIBRATED_EVIDENCE_CLASS or row.get("claim_eligible") is not True:
|
| 490 | refused.append((*where, f"evidence_class {row.get('evidence_class')!r} is not calibrated"))
|
| 491 | continue
|
| 492 | record = row.get("calibration_record")
|
| 493 | if not isinstance(record, dict) or row.get("calibration_record_sha256") != record.get("record_sha256"):
|
| 494 | refused.append((*where, "calibration record missing or its digest does not match the row"))
|
| 495 | continue
|
| 496 | key = record.get("record_sha256")
|
| 497 | if key not in verified:
|
| 498 | cal_sha = record.get("calibration_rows_sha256")
|
| 499 | cal_path = by_sha.get(cal_sha)
|
| 500 | verified[key] = verify_campaign(record, inputs.get(cal_path) if cal_path else None,
|
| 501 | cal_sha if cal_path else None, registry)
|
| 502 | facts, reasons = verified[key]
|
| 503 | if reasons:
|
| 504 | refused.append((*where, reasons[0]))
|
| 505 | continue
|
| 506 | problems = row_problems(row, facts)
|
| 507 | probe_key = (path, str(row.get("bench_id")))
|
| 508 | if probe_key not in run_probe_cache:
|
| 509 | run_rows = [r for r in inputs.get(path, []) if r.get("bench_id") == row.get("bench_id")]
|
| 510 | run_probe_cache[probe_key] = probe_verdict(run_rows)[0]
|
| 511 | problems += [f"published run: {reason}" for reason in run_probe_cache[probe_key]]
|
| 512 | if problems:
|
| 513 | refused.append((*where, problems[0]))
|
| 514 | return refused
|