#!/usr/bin/env python3 r"""lbl3vp-points-evaluate.py -- LBL3VP (row 33 of the paper's table of integrals): the dispersive evaluator of I_VP(eps) at FOUR FURTHER Euclidean points, gated against an independent AMFlow grid at each point. Third script of the page, beside lbl3vp-evaluate.py (the fixed-eps value and the pole layers at the reference point (s, t, m^2) = (-1, -1/3, 1)) and lbl3vp-eps0-evaluate.py (the eps^0 layer there). This one answers the function-level question: does the SAME construction -- the DE-transported box kernel times the symbolic density, seeded by exact Feynman-parameter integrals -- reproduce the integral at other kinematics, where nothing was tuned? THE OBJECT I_VP(eps; s, t) = (1/pi) Int_1^inf rho_VP(w'; eps) K_P(w'; eps; s, t) dw' (m^2 = 1), the vacuum-polarisation-dressed light-by-light box (three loops). rho_VP is kinematics-free; the kernel K_P carries (s, t) through its one-loop box w'-DE and its threshold moments. THE FOUR POINTS (vendor_row33_points/points/points.json carries the rule and the numbers) P1: (s, t) = (-1/2, -1), u = -s-t = 3/2 P2: (s, t) = (-3/2, -1/4), u = 7/4 P3: (s, t) = (-3/4, -1/2), u = 5/4 P4: (s, t) = (-1/4, -1/2), u = 3/4 All four Euclidean (s < 0, t < 0, u < 4 m^2), with the kernel-analyticity margin 1 - u/4 >= 1/2, at distances >= 9/4 from the u = 4 m^2 threshold and >= 17/4 from s = 4 m^2, with rational heights comparable to the reference and (|s|, |t|, |u|) multisets different from the reference's and from each other's. P1 and P4 lie on the line t = 2s (u = 3/2 and 3/4), where an odd-order Gauss-Legendre rule in the closed one-loop box would put a node on a removable zero of its integrand; the shipped lbl3disp uses an even order, which has no node there. For each point two things were made once and are shipped beside this script: its one-loop box w'-DE (an exact-rational 8x8 system reconstructed from IBP samples at that point) and an independent AMFlow grid of the full three-loop integral at eps = 2^-k, k = 4..18 (15 samples, Arb balls; two AMFlow legs at x_order 200 and 300 agree to 54-63 digits per sample). Any other (s, t) needs both and is refused by name (exit 2): the script computes nothing it cannot gate. THE ROUTE (the research evaluator, vendored whole: vendor_row33_points/eval_row33.py, build B1) * kernel seed at w' = 5: the eight exact fixed-eps Feynman-parameter masters (k33lib, (s, u) as exact rationals); threshold Taylor block K_n(eps) at w' = 1: the exact moment two-folds; * kernel transport: Taylor-step transport of the point's exact-rational w'-DE (RatDE), the step-size singularity list derived from that DE's denominators at fixed eps; * density: closed two-body form on (1, 9), Frobenius series at the three-body threshold w' = 9, real-axis transport of the exact self-energy system above (rho33lib); * tanh-sinh panel quadrature at two levels + one Richardson step. No AMFlow, no fit, no stored value enters the computation; the only file inputs are the two exact-rational systems and the point's w'-DE. WHAT THE RUN CHECKS Fixed-eps mode (default; --eps 2^-k, or --all-eps for k = 6..13): I_VP(2^-k) at the point against the shipped grid's sample at that eps, as -log10 of the relative difference. Bar: 30 digits. The recorded run (points/P/record.json: the same evaluator at the same point, dps 110) reached 37.58-38.87 digits over the eight eps at every one of the four points (P1 37.60-38.86; P2 37.58-38.83; P3 37.60-38.85; P4 37.61-38.87); its figure for the requested eps is printed on the line after the live one. A live figure below the bar is GATE FAILED (exit 1). --pole-layers: the eps^-2 and eps^-1 Laurent layers of I_VP at the point (from the closed dilogarithmic kernel's threshold Taylor fit + the exact parametric threshold two-fold + three quadratures of the eps^-1 density layer -- eval_row33.pole_layers) against the Neville peel of the shipped grid (c_-2, c_-1, c_0 with leave-one-out bars, computed here from the grid file at 400 digits by the recorded run's own arithmetic). The agreement is capped by the peel (about 45 / 40 digits); bar 30 digits on both layers and on the in-run fit-vs-two-fold check of K_2^(0). Every comparison value is read from the shipped files and never enters the computation. FILES (beside this script, in vendor_row33_points/; every one sha256-pinned below and REFUSED on any mismatch before anything is imported or computed) eval_row33.py the evaluator (build B1: --kin / --box-de / --oracle-json generalisation of the reference-point script; RatDE transport, I_disp_at_eps, pole_layers, the grid loader) k33lib.py exact fixed-eps Feynman-parameter masters and threshold moments, (s, u) parameters rho33lib.py the symbolic spectral density lbl3disp.py the closed dilogarithmic one-loop box (the eps^0 kernel for the pole layers) row33_data.json the exact-rational self-energy system (4 masters) + the reference-point data points/points.json the rule and the four points points/P/box1eq_de.json the point's one-loop box w'-DE (8 masters, exact rational) points/P/amflow_grid.json the independent AMFlow grid at the point (15 samples, Arb balls) points/P/record.json the recorded run at the point: per-eps values and agreements, the peel, the pole layers, walls, provenance in plain words The vendored modules are the research code with only docstrings, comments and message strings edited: their numeric path is identical statement for statement to the build they come from. EXIT CODES: 0 every gate passes; 1 a gate fails (GATE FAILED by name -- what --mutate must produce); 2 usage (a point outside the Euclidean rule, a point with no shipped DE and grid, an eps off the grid, --dps below 60); 3 a pinned file's sha256 does not match (refused before any computation); 4 a required file, or mpmath / sympy, is missing. USAGE python3 lbl3vp-points-evaluate.py # P1, eps = 2^-10, dps 110 python3 lbl3vp-points-evaluate.py --point P2 --eps 2^-7 # another point (P1, P2, P3, P4) / eps (k = 4..18) python3 lbl3vp-points-evaluate.py --point P3 --all-eps --jobs 8 # P3 = (-3/4,-1/2), the eight eps python3 lbl3vp-points-evaluate.py --point=-1/2,-1 # the same as --point P1 (s,t[,m2] form) python3 lbl3vp-points-evaluate.py --all-eps --jobs 8 # k = 6..13, eight processes python3 lbl3vp-points-evaluate.py --pole-layers # eps^-2 / eps^-1 at the point vs the peel python3 lbl3vp-points-evaluate.py --mutate # control: K_2(eps) shifted by 1e-30 after it # is computed; the value moves at the 1e-29 level and the gate must FAIL (exit 1) python3 lbl3vp-points-evaluate.py --dps 60 # the cheapest accepted precision --json OUT writes the run (values as full-precision strings, agreements, diagnostics, walls). Not offered: a --check two-precision rerun (run two --dps values instead) and the reference-point modes (those are lbl3vp-evaluate.py's). TIMING (measured; one process each, on a shared 96-core server at the loadavg stamped below, so an upper bound for an idle machine of the same class) default run (P1, eps = 2^-10, dps 110): 295 s --point P2 --eps 2^-7 --dps 110: 290 s --pole-layers at P1, dps 110: 161 s --mutate at P1, eps = 2^-10, dps 110 (the control; exits 1): 290 s --all-eps --jobs 8 --dps 60 at P1 (eight eps in eight processes): 168 s loadavg at launch: 110.44 --point P3 --all-eps --jobs 8 --dps 110 (eight eps in eight processes): 288 s --point P4 --all-eps --jobs 8 --dps 110 (eight eps in eight processes): 284 s --point P3 --pole-layers, dps 110: 145 s --point P4 --pole-layers, dps 110: 144 s --point P3 --mutate, eps = 2^-10, dps 110 (the control; exits 1): 273 s --point P3 --dps 60 (one eps, the cheap first run): 149 s --point P4 --dps 60 (one eps): 146 s loadavg at launch: 114.32 (P3 / P4 at dps 60), 106.36 (P3 all-eps), 115.82 (P4 all-eps), 101.12 (pole layers and the control) (the P3 / P4 lines above; the P1 / P2 lines were measured at the loadavg stated before them) The cost is the kernel transport (about 5800 target points) and the density transport; it does not depend on the point. --dps 60 (the quadrature limiter then sits near 33 digits) is the cheap first run. Dependencies: python3 >= 3.8, mpmath (pip install mpmath), sympy (pip install sympy). Output is line-buffered; the first line prints at once. """ import argparse import datetime import hashlib import json import os import sys import time EXIT_PASS, EXIT_FAIL, EXIT_USAGE, EXIT_PIN, EXIT_MISSING = 0, 1, 2, 3, 4 HERE = os.path.dirname(os.path.abspath(__file__)) VENDOR = os.path.join(HERE, "vendor_row33_points") PINS = { "eval_row33.py": "8028aad292c42dc85ba914cc0177b2abef42cb97be305075be524de6770661b2", "k33lib.py": "520923388506dd8cf2ff1622718b2f916d764ec2b6884b4694c4d010d038220e", "rho33lib.py": "44767fab87e66115957d8f175793185eb7d50ce38245c2ee40675a70d17a9d02", "lbl3disp.py": "fbee9161004f1a7577417b59fb6264ffb16012e518dc992c402a84ca3703d5ff", "row33_data.json": "4735e2721a903b3124e2be2ec0c68f1b68346c330c5570dd22b0cdc9cd758dd9", "points/points.json": "00fb8e7b126d166e460e60d4b43511cdd2c366e4959c2f6c1013223bd2e5d906", "points/P1/box1eq_de.json": "271cdf23343154dc113479e1d7986c813871f3ade689993751ca29105814e154", "points/P1/amflow_grid.json": "1d99d0cabb15d2d61c51078e8347ce5e09001248d25dca2932f8a4c610f86fbe", "points/P1/record.json": "7151dab6a8872971a1ffc2b4604d153df87a4cc49c0d34fb8a22dbafdf7ef668", "points/P2/box1eq_de.json": "8112bb4ce60df93dbc28c2c0cf4315604295df662cf7dd6db76aaff99b5ca650", "points/P2/amflow_grid.json": "e0f60910d3bf051048cba18851a57e3241f6f4ad550978ec07b130a25521e846", "points/P2/record.json": "b9960411bca7c40d0235d91223b93e2c5bd2e9b1803c475a7b3fff74ad9a4f4d", "points/P3/box1eq_de.json": "38a1d1b0c62ed58d8656172d36c83d661ecfd22aac41047e7b9bca001157577e", "points/P3/amflow_grid.json": "984768ace5f8286d4f8078f49a9611c4fd6e0e41ba34e7af126b1799842614a4", "points/P3/record.json": "120c92482f2eef8c82a1bd841c7055cced2326651203117d9933c82079c3349f", "points/P4/box1eq_de.json": "f928e6437577f26932e4118a6b4ed868d1f293e8cdbf1516eed264617d712631", "points/P4/amflow_grid.json": "0814488d53fa85e2c781a266bc6f03ff9892001be3f32f882916158d84ebe5d2", "points/P4/record.json": "d3c28db61ab5862dcbe5e880e8758bdf5bb04d8240d5832722ab9441b3228aa5", } MIN_DPS = 60 BAR_D = 30.0 KS_ALL = list(range(6, 14)) KS_GRID = list(range(4, 19)) def sha256_file(path): with open(path, "rb") as fh: return hashlib.sha256(fh.read()).hexdigest() def check_pins(): """Refuse every vendored file whose sha256 does not match its pin -- before importing any of them.""" for rel, pin in PINS.items(): path = os.path.join(VENDOR, rel) if not os.path.exists(path): print(f"MISSING: vendor_row33_points/{rel} must sit beside this script (download the " f"vendor_row33_points folder from the same page); nothing computed") sys.exit(EXIT_MISSING) sha = sha256_file(path) if sha != pin: print(f"REFUSED: vendor_row33_points/{rel} sha256 {sha} does not match the pin {pin} -- the file " f"was altered or is not the released version; nothing computed") sys.exit(EXIT_PIN) print(f"[pins] {len(PINS)} vendored files match their sha256 pins " f"(evaluator {PINS['eval_row33.py'][:16]}..., data {PINS['row33_data.json'][:16]}...)") def utc_stamp(): return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def resolve_point(txt, points, sp): """--point |s,t[,m2] -> the shipped point dict, or a usage exit by name. The tags come from points/points.json; a tag listed there is accepted only if its DE, grid and record are pinned above.""" t = txt.strip() shipped = [p for p in points["points"] if all(f"points/{p['tag']}/{f}" in PINS for f in ("box1eq_de.json", "amflow_grid.json", "record.json"))] by_tag = {p["tag"]: p for p in shipped} tags = ", ".join(by_tag) pairs = [f"{p['tag']} = ({p['s']},{p['t']})" for p in shipped] listing = (", ".join(pairs[:-1]) + " and " + pairs[-1]) if len(pairs) > 1 else pairs[0] unpinned = [p["tag"] for p in points["points"] if p["tag"] not in by_tag] if t.upper() in unpinned: print(f"REFUSED (exit {EXIT_USAGE}): points/points.json lists {t.upper()} but no pinned one-loop box w'-DE, AMFlow grid " f"and record ship for it; only {tags} ship all three. Nothing is computed ungated.") sys.exit(EXIT_USAGE) if t.upper() in by_tag: p = by_tag[t.upper()] return p, sp.Rational(p["s"]), sp.Rational(p["t"]), sp.Rational(p["msq"]) toks = t.replace(",", " ").split() if len(toks) not in (2, 3): print(f"USAGE (exit {EXIT_USAGE}): --point takes {tags} or s,t[,m2] (rationals, e.g. --point=-1/2,-1); got {txt!r}") sys.exit(EXIT_USAGE) try: s, tt = sp.Rational(toks[0]), sp.Rational(toks[1]) m2 = sp.Rational(toks[2]) if len(toks) == 3 else sp.Integer(1) except Exception as ex: print(f"USAGE (exit {EXIT_USAGE}): --point values must be rationals; {ex}") sys.exit(EXIT_USAGE) if not (s < 0 and tt < 0 and m2 > 0 and -(s + tt) < 4 * m2): print(f"REFUSED (exit {EXIT_USAGE}): (s,t,m^2) = ({s},{tt},{m2}) is outside the Euclidean rule of " f"points/points.json: s < 0, t < 0, m^2 > 0 and u = -s-t < 4 m^2 are required") sys.exit(EXIT_USAGE) sh, th = s / m2, tt / m2 for p in points["points"]: if sp.Rational(p["s"]) / sp.Rational(p["msq"]) == sh and sp.Rational(p["t"]) / sp.Rational(p["msq"]) == th: if p["tag"] not in by_tag: print(f"REFUSED (exit {EXIT_USAGE}): (s,t)/m^2 = ({sh},{th}) is {p['tag']} in points/points.json, but no pinned " f"one-loop box w'-DE, AMFlow grid and record ship for it; only {tags} ship all three. Nothing is computed ungated.") sys.exit(EXIT_USAGE) return p, s, tt, m2 print(f"REFUSED (exit {EXIT_USAGE}): no shipped data at (s,t)/m^2 = ({sh},{th}). A point needs its own " f"one-loop box w'-DE reconstruction and an independent AMFlow grid to gate against; only {listing} " f"ship both. Nothing is computed ungated.") sys.exit(EXIT_USAGE) def parse_k(txt, sp): """--eps '2^-k' (or the exact rational 1/2^k) with k on the shipped grid, else a usage exit by name.""" t = txt.strip() try: if t.startswith("2^-"): k = int(t[3:]) e = sp.Rational(1, 2 ** k) else: e = sp.Rational(t) k = e.q.bit_length() - 1 if not (e.p == 1 and (e.q & (e.q - 1)) == 0): raise ValueError("not a power of two") except Exception: print(f"USAGE (exit {EXIT_USAGE}): --eps must be 2^-k with k = 4..18 (the shipped grid's samples); got {txt!r}") sys.exit(EXIT_USAGE) if k not in KS_GRID: print(f"USAGE (exit {EXIT_USAGE}): --eps 2^-{k} is off the shipped grid (k = 4..18); every served value is gated, " f"so an eps without a grid sample is refused") sys.exit(EXIT_USAGE) return k def install_mutation(K, mp): """--mutate: wrap k33lib.kn_moments so that the threshold moment K_2(eps) comes back shifted by 1e-30 once it is computed. K_2 carries the eps^-2 pole of I_VP (through the threshold-panel subtraction), so the value moves by about 2e-29 relative, the internal two-precision checks (which see the same shifted moment) still pass, and the gate against the grid must FAIL. The vendored numeric code is untouched; the evaluator looks the function up by name in the k33lib module at call time.""" orig = K.kn_moments def kn_moments_mutated(eps, nmax, n, work, s=K.S_REF, u=K.U_REF): out = orig(eps, nmax, n, work, s, u) out[2] = out[2] + mp.mpf("1e-30") return out K.kn_moments = kn_moments_mutated print(" !! MUTATE: K_2(eps) (the second threshold Taylor moment of the box kernel) shifted by 1e-30 after " "it is computed -- the value moves at the 1e-29 level and the gate must FAIL", flush=True) # ---- the recorded run's peel arithmetic, ported as a unit ---- def peel_parse_ball(x, mp, re): """Arb ball string -> mid. '[mid +/- rad]' -> mid; '[+/- rad]' (no significant digit) -> None.""" x = x.strip() if x.startswith('[+/-'): return None m = re.search(r'\[([-+0-9.eE]+)\s', x) return mp.mpf(m.group(1)) if m else mp.mpf(x) def peel_nev(xs, ys): T = list(ys); n = len(xs) for m in range(1, n): for i in range(n - m): T[i] = (xs[i+m]*T[i] - xs[i]*T[i+1])/(xs[i+m]-xs[i]) return T[0] def peel_agree(a, b, mp): return float('inf') if a == b else float(-mp.log10(abs(a - b)/abs(b))) def peel_grid(grid_path, mp): """The Neville peel c_-2, c_-1, c_0 of the grid with leave-one-out bars at kmin = 4, 6, 8, at 400 digits: the recorded run's arithmetic on the same bytes (its GATE producer), so the strings must equal record.json's.""" import re G = json.load(open(grid_path)); res = G['result'][0] assert res['integral']['indices'][:8] == [1, 1, 1, 2, 1, 2, 1, 1], res['integral'] grid = {} with mp.workdps(400): for s in res['samples']: e = mp.mpf(s['eps']['re']); assert mp.mpf(s['eps']['im']) == 0 k = int(mp.nint(-mp.log(e, 2))); assert mp.mpf(2)**(-k) == e, e re_ = peel_parse_ball(s['value']['re'], mp, re) grid[k] = {'re': re_, 'certified': re_ is not None} ks = sorted(grid) assert all(grid[k]['certified'] for k in ks), "uncertified grid sample" peel = {} def ext(fn, kmin): xs = [mp.mpf(1)/2**k for k in ks if k >= kmin]; kk = [k for k in ks if k >= kmin] ys = [fn(x, grid[k]['re']) for x, k in zip(xs, kk)] v = peel_nev(xs, ys); loo = [peel_nev(xs[:j]+xs[j+1:], ys[:j]+ys[j+1:]) for j in range(len(xs))] return v, max(abs(l - v) for l in loo) for kmin in (4, 6, 8): if len([k for k in ks if k >= kmin]) < 6: continue cm2, s2 = ext(lambda e, I: I*e**2, kmin) cm1, s1 = ext(lambda e, I: (I*e**2 - cm2)/e, kmin) c0, s0 = ext(lambda e, I: (I*e**2 - cm2 - cm1*e)/e**2, kmin) peel[str(kmin)] = {'c-2': mp.nstr(cm2, 60), 'c-2_loo_bar_d': float(-mp.log10(s2/abs(cm2))), 'c-1': mp.nstr(cm1, 60), 'c-1_loo_bar_d': float(-mp.log10(s1/abs(cm1))), 'c0': mp.nstr(c0, 60), 'c0_loo_bar_d': float(-mp.log10(s0/abs(c0)))} return peel def run_eps(E, mp, sp, DPS, k, verbose): """I_VP(2^-k) at the module's point, gated live against the shipped grid inside the evaluator's own run_gate (oracle_for / agree_d). Returns the run as plain data.""" e = sp.Rational(1, 2 ** k) t0 = time.time() out = E.one_pass(DPS, [e], None, verbose=verbose) lab = E.eps_label(e) info = out["_info"][lab] val = out[f"I_disp({lab})"] with mp.workdps(DPS + 20): v_full = mp.nstr(val, DPS + 10) v40 = mp.nstr(val, 40) o40 = mp.nstr(mp.mpf(info["oracle"]), 40) if info["oracle"] is not None else None diag = {kk: (float(v) if isinstance(v, (int, float, mp.mpf)) else v) for kk, v in info["diag"].items()} return {"k": k, "eps": lab, "I_disp": v_full, "I_disp_40d": v40, "I_AMF_40d": o40, "agreement_d": info["agreement_d"], "diag": diag, "wall_s": round(time.time() - t0, 2)} def main(): try: sys.stdout.reconfigure(line_buffering=True) except Exception: pass ap = argparse.ArgumentParser( description="Row 33 (LBL3VP): the dispersive evaluator at four further Euclidean points, P1 = (-1/2,-1), P2 = (-3/2,-1/4), P3 = (-3/4,-1/2) and P4 = (-1/4,-1/2), " "each gated against its own independent AMFlow grid (bar 30 digits; the recorded runs reached " "37.6-38.9). Measured walls, one process per run on a loaded 96-core server (upper bounds): " "--all-eps --jobs 8 --dps 110 at P3 288 s, at P4 284 s; --pole-layers at P3 145 s, at P4 144 s; one eps at --dps 60 at P3 149 s, at P4 146 s. See the module docstring for every measured cost.") ap.add_argument("--point", default="P1", metavar="P", help="P1, P2, P3, P4, or s,t[,m2] as rationals (use the = form for a leading minus: --point=-1/2,-1); " "a point without shipped data is refused (exit 2). Default P1.") ap.add_argument("--eps", default="2^-10", metavar="2^-k", help="eps = 2^-k, k = 4..18 (the shipped grid's samples). Default 2^-10.") ap.add_argument("--dps", type=int, default=110, help="transport working precision in decimal digits (default 110, the recorded run's; minimum 60)") ap.add_argument("--all-eps", action="store_true", help="evaluate k = 6..13 (the recorded run's set) instead of one eps") ap.add_argument("--jobs", type=int, default=1, help="with --all-eps: number of parallel processes (default 1)") ap.add_argument("--pole-layers", action="store_true", help="eps^-2 / eps^-1 Laurent layers at the point vs the Neville peel of the shipped grid") ap.add_argument("--mutate", action="store_true", help="control: shift K_2(eps) by 1e-30 after it is computed; the gate must FAIL (exit 1)") ap.add_argument("--json", default=None, metavar="OUT", help="write the run (values, agreements, diagnostics) as json") args = ap.parse_args() if args.dps < MIN_DPS: print(f"USAGE (exit {EXIT_USAGE}): --dps must be >= {MIN_DPS} (below it the quadrature limiter sits under the 30-digit bar)") return EXIT_USAGE if args.jobs < 1: print(f"USAGE (exit {EXIT_USAGE}): --jobs must be >= 1") return EXIT_USAGE T0 = time.time() mode = "pole-layers" if args.pole_layers else ("all-eps" if args.all_eps else "fixed-eps") print(f"lbl3vp-points-evaluate.py: row 33 (LBL3VP) dispersive evaluator at a further Euclidean point, {mode} mode, " f"dps={args.dps}{' [MUTATE control]' if args.mutate else ''}") try: import mpmath as mp except ImportError: print("MISSING: python3 module mpmath (pip install mpmath); nothing computed") return EXIT_MISSING try: import sympy as sp except ImportError: print("MISSING: python3 module sympy (pip install sympy; it parses the exact-rational systems); nothing computed") return EXIT_MISSING check_pins() # exits 3 / 4 before any vendored code is imported sys.dont_write_bytecode = True # leave no __pycache__ beside the downloaded files sys.path.insert(0, VENDOR) import eval_row33 as E # noqa: E402 import k33lib as K # noqa: E402 points = json.load(open(os.path.join(VENDOR, "points", "points.json"))) pt, s_, t_, m2_ = resolve_point(args.point, points, sp) tag = pt["tag"] de_path = os.path.join(VENDOR, "points", tag, "box1eq_de.json") grid_path = os.path.join(VENDOR, "points", tag, "amflow_grid.json") rec = json.load(open(os.path.join(VENDOR, "points", tag, "record.json"))) if rec["point"]["tag"] != tag or rec["grid"]["sha256"] != PINS[f"points/{tag}/amflow_grid.json"]: print(f"REFUSED (exit {EXIT_PIN}): points/{tag}/record.json names the grid sha256 {rec['grid']['sha256'][:16]}... but the " f"pinned grid is {PINS[f'points/{tag}/amflow_grid.json'][:16]}... -- the shipped files disagree; nothing computed") return EXIT_PIN # the point + its DE + its grid into the evaluator's module state (the --kin / --box-de / --oracle-json path) E.set_kin(s_, t_, m2_) de = json.load(open(de_path)) if len(de["A"]) != 8 or any(len(r) != 8 for r in de["A"]) or len(de["masters"]) != 8 \ or de.get("var", "wp") != "wp" or de.get("dim_var", "d") != "d": print(f"REFUSED: points/{tag}/box1eq_de.json is not an 8x8 system over 8 masters in (wp, d)") return EXIT_PIN E.BOX_DE = de E.BOX_DE_PATH = f"vendor_row33_points/points/{tag}/box1eq_de.json" E.NO_ORACLE = False E.ORACLE_PATH = grid_path E.ORACLE_GRID = E.load_amflow_grid(grid_path) missing = [k for k in KS_GRID if k not in E.ORACLE_GRID] if missing: print(f"REFUSED: points/{tag}/amflow_grid.json carries no certified sample at k = {missing}") return EXIT_PIN sh, th, uh = E.kin_hat() print(f"point {tag}: (s, t, m^2) = ({s_}, {t_}, {m2_}), u = {-(s_ + t_)}" + (f"; evaluated as Ihat(s/m^2, t/m^2) = Ihat({sh}, {th})" if m2_ != 1 else "") + f"; one-loop box w'-DE points/{tag}/box1eq_de.json; grid points/{tag}/amflow_grid.json " f"({len(E.ORACLE_GRID)} samples, sha256 {rec['grid']['sha256'][:16]}...)") if args.mutate: install_mutation(K, mp) result = {"script": os.path.basename(__file__), "script_sha256": sha256_file(os.path.abspath(__file__)), "mode": mode, "point": {"tag": tag, "s": str(s_), "t": str(t_), "msq": str(m2_), "u": str(-(s_ + t_))}, "dps": args.dps, "bar_d": BAR_D, "mutate": bool(args.mutate), "argv": sys.argv[1:], "started": utc_stamp()} verdict_ok = True if args.pole_layers: DPS = args.dps qdps = max(55, DPS - 15); kdps = max(90, DPS + 20); nfit = E.nfit_for(DPS) print(f"Laurent pole layers at (s,t) = ({sh}, {th}) from the closed-form kernel (quad dps {qdps}, kernel dps " f"{kdps}, fit degree {nfit}); comparison = the Neville peel of the shipped grid (computed now, 400 digits)") t0 = time.time() mp.mp.dps = kdps + 60 shat = mp.mpf(sh.p) / sh.q; that = mp.mpf(th.p) / th.q res = E.pole_layers(shat, that, False, qdps, kdps, nfit=nfit) wall_pole = time.time() - t0 with mp.workdps(qdps + 20): em2_s = mp.nstr(res["I_em2"], qdps); em1_s = mp.nstr(res["I_em1"], qdps); bsm1_s = mp.nstr(res["Bsm1"], qdps) fit_chk = float(res["fit_chk"]) peel = peel_grid(grid_path, mp) P = peel.get("4") or peel[min(peel)] with mp.workdps(400): d2 = peel_agree(mp.mpf(em2_s), mp.mpf(P["c-2"]), mp) d1 = peel_agree(mp.mpf(em1_s), mp.mpf(P["c-1"]), mp) rp = rec["pole_layers"] peel_same = all(peel[kmin] == rec["peel"][kmin] for kmin in rec["peel"]) and set(peel) == set(rec["peel"]) print(f" peel of the grid (kmin 4): c_-2 = {P['c-2'][:42]}... (leave-one-out bar {P['c-2_loo_bar_d']:.1f} d)") print(f" c_-1 = {P['c-1'][:42]}... (bar {P['c-1_loo_bar_d']:.1f} d)") print(f" c_0 = {P['c0'][:42]}... (bar {P['c0_loo_bar_d']:.1f} d)") print(f" peel strings equal the recorded run's (kmin 4/6/8): {'yes' if peel_same else 'NO'}") print(f" [check] live closed-dilog K_2^(0) fit vs exact parametric two-fold: {fit_chk:.2f} d (recorded run: {rp['fit_chk_d']:.2f} d)") print(f" eps^-2: -K_2^(0) = {em2_s[:38]} (computed)") print(f" agreement vs peel = {d2:.2f} d (recorded run: {rp['eps-2_d']:.2f} d; capped by the peel's {P['c-2_loo_bar_d']:.1f} d bar)") print(f" eps^-1: -K_2^(1)+B[s_-1]^(0) = {em1_s[:38]} (computed)") print(f" agreement vs peel = {d1:.2f} d (recorded run: {rp['eps-1_d']:.2f} d)") result["pole_layers"] = {"I_em2": em2_s, "I_em1": em1_s, "Bsm1": bsm1_s, "fit_chk_d": fit_chk, "eps-2_d": d2, "eps-1_d": d1, "settings": {"qdps": qdps, "kdps": kdps, "nfit": nfit}, "wall_s": round(wall_pole, 2), "recorded": {"fit_chk_d": rp["fit_chk_d"], "eps-2_d": rp["eps-2_d"], "eps-1_d": rp["eps-1_d"]}} result["peel"] = peel result["peel_equals_recorded"] = peel_same worst = min(d2, d1, fit_chk) if not peel_same: print(f"\n[gate] FAIL (exit {EXIT_FAIL}): GATE FAILED -- the live peel of the shipped grid differs from the recorded run's strings") verdict_ok = False elif worst < BAR_D: print(f"\n[gate] FAIL (exit {EXIT_FAIL}): GATE FAILED -- pole-layer agreement {worst:.2f} d is below the bar {BAR_D:.0f} d " f"(eps^-2 {d2:.2f}, eps^-1 {d1:.2f}, fit-vs-two-fold {fit_chk:.2f})") verdict_ok = False else: print(f"\n[gate] PASS: eps^-2 {d2:.2f} d and eps^-1 {d1:.2f} d vs the shipped grid's peel, fit-vs-two-fold {fit_chk:.2f} d, " f"all above the bar {BAR_D:.0f} d (computed at dps {DPS}, {wall_pole:.0f} s)") else: DPS = args.dps ks = KS_ALL if args.all_eps else [parse_k(args.eps, sp)] runs = {} if args.all_eps and args.jobs > 1 and len(ks) > 1: import multiprocessing ctx = multiprocessing.get_context("fork") print(f" {len(ks)} eps in {min(args.jobs, len(ks))} processes (k = {ks[0]}..{ks[-1]}); one line per eps as it lands") def worker(k, conn): try: conn.send((k, run_eps(E, mp, sp, DPS, k, verbose=False), None)) except Exception as ex: conn.send((k, None, repr(ex))) conn.close() pending = list(ks); active = []; pipes = {}; procs = {} while pending or active: while pending and len(active) < args.jobs: k = pending.pop(0) pr, pw = ctx.Pipe(False) p = ctx.Process(target=worker, args=(k, pw)); p.start() pipes[k], procs[k] = pr, p; active.append(k) k = active.pop(0) kk, r, err = pipes[k].recv(); procs[k].join() if r is None: print(f" eps = 2^-{kk}: FAILED: {err}") return EXIT_FAIL runs[kk] = r rk = rec["fixed_eps"]["per_eps"].get(str(kk)) print(f" eps = 2^-{kk}: I_disp = {r['I_disp_40d']} I_AMF = {r['I_AMF_40d']} agreement {r['agreement_d']:.2f} d" + (f" (recorded run: {rk['agreement_d']:.2f} d)" if rk else "") + f" [{r['wall_s']:.0f} s]") else: for k in ks: r = run_eps(E, mp, sp, DPS, k, verbose=True) runs[k] = r rk = rec["fixed_eps"]["per_eps"].get(str(k)) if rk: print(f" recorded run: {rk['agreement_d']:.2f} d (the same point and eps at dps 110; points/{tag}/record.json)") if rk["I_AMF_40d"] != r["I_AMF_40d"]: print(f" NOTE: the grid sample printed above differs from the recorded run's I_AMF_40d {rk['I_AMF_40d']}") else: print(f" (no recorded figure at this eps; the recorded run covers k = 6..13)") result["runs"] = {str(k): runs[k] for k in sorted(runs)} ds = [runs[k]["agreement_d"] for k in runs] worst = min(ds) if worst < BAR_D: bad = [f"2^-{k}: {runs[k]['agreement_d']:.2f} d" for k in sorted(runs) if runs[k]["agreement_d"] < BAR_D] print(f"\n[gate] FAIL (exit {EXIT_FAIL}): GATE FAILED -- agreement with the shipped AMFlow grid below the bar {BAR_D:.0f} d at " f"{', '.join(bad)}" + (" (the --mutate control: this failure is the expected result)" if args.mutate else "")) verdict_ok = False else: span = f"{worst:.2f} d" if len(ds) == 1 else f"{worst:.2f}-{max(ds):.2f} d over k = {ks[0]}..{ks[-1]}" rspan = rec["fixed_eps"] print(f"\n[gate] PASS: I_VP at {tag} agrees with the shipped AMFlow grid to {span} (bar {BAR_D:.0f} d; recorded run " f"{rspan['min_agreement_d']:.2f}-{rspan['max_agreement_d']:.2f} d over k = 6..13; computed at dps {DPS}, {time.time() - T0:.0f} s)") result["verdict"] = "PASS" if verdict_ok else "GATE FAILED" result["wall_s"] = round(time.time() - T0, 2) result["finished"] = utc_stamp() if args.json: with open(args.json, "w") as fh: json.dump(result, fh, indent=1) print(f" wrote {args.json}") return EXIT_PASS if verdict_ok else EXIT_FAIL if __name__ == "__main__": sys.exit(main())