#!/usr/bin/env python3 """ run_cert.py — KKLT-CERT W0 orchestrator. Stage 1 (exact, banked): frame solve (+ xi mutation test), tower extension to N_EXT with recurrence regression, w0-extension cross-check vs the banked 900-term series, bank towers_ext.json + frame_solution.json. Stage 2 (per precision, per route): certified transport MUM -> s_vac and flux contraction; checks c1-c4; bank cert_results.json. Usage: python3 run_cert.py stage1 python3 run_cert.py route (R|C) (dps) python3 run_cert.py checks Resources: single thread. """ from fractions import Fraction as Fr import json, os, sys, time HERE = os.path.dirname(os.path.abspath(__file__)) N_EXT = 420 S_STAR = Fr("0.0134681831826400193897483719453") # exact pin (restrict/s_vac.json) TAU_PIN = Fr("6.8554572563187259046361432042141669") # exact pin (pilot dps-90) PILOT_W0 = "2.0371060933111834182083117862342977e-8" PILOT_TAU = "6.8554572563187259046361432042141669" PILOT_U1 = "2.7421769802085209924953942571371399" PILOT_U2 = "2.0566327731047339299455907474511510" NAMES = ["F0", "F1", "F2", "X0", "X1", "X2"] def stage1(): import cert_lib as CL t0 = time.time() coeffs, towers = CL.solve_frame(M_FIT_LO=14, M_FIT_HI=20, L_CHK=240) print(f"[stage1] frame solved + checks f1 (held-out 15..20) and f2 " f"(L-annihilation to s^240) PASS ({time.time()-t0:.1f}s)", flush=True) for i, c in enumerate(coeffs): print(f" {NAMES[i]}: " + ", ".join( f"c[{''.join(map(str,al))};v^{p};z3^{q}]={v}" for (al, p, q), v in sorted(c.items()))) # mutation test: flip xi sign -> the zeta3 sector must need a counterterm CL.XI_COEF = Fr(270) try: cm, _ = CL.solve_frame(M_FIT_LO=14, M_FIT_HI=20, L_CHK=100) z3c = [(i, k, v) for i, c in enumerate(cm) for k, v in c.items() if k[2] == 1] print(f"[stage1] xi-flip mutation: fit still closes but needs zeta3 " f"counterterms {z3c} (nonzero => B4 frame pinned)", flush=True) assert z3c, "xi-flip mutation UNDETECTED -- B4 not pinned by jets" except AssertionError as e: print(f"[stage1] xi-flip mutation: solve FAILS ({e}) -- B4 pinned", flush=True) CL.XI_COEF = Fr(-270) t0 = time.time() ext = [CL.extend_tower(T, N_EXT) for T in towers] print(f"[stage1] towers extended to N={N_EXT} with recurrence regression " f"on 200..240 ({time.time()-t0:.1f}s)", flush=True) # cross-check: X0 tower is v^3 * w0; compare vs banked 900-term series SC = json.load(open(os.path.join(CL.REST, "series_curve.json"))) a900 = [int(x) for x in SC["a"]] TX0 = ext[3][(0, 3, 0)] assert all(TX0[m] == a900[m] for m in range(N_EXT + 1)), "w0 extension FAIL" print(f"[stage1] w0-extension == banked series_curve to m={N_EXT} PASS") nser = sum(len(T) for T in ext) bank = {"N_EXT": N_EXT, "periods": [CL.tower_to_json(T) for T in ext]} CL.bank_json(os.path.join(HERE, "towers_ext.json"), bank) CL.bank_json(os.path.join(HERE, "frame_solution.json"), {"note": "v^3*Pi_i = sum c * v^p z3^q * Phi_alpha", "coeffs": [{f"{a[0]}{a[1]},{p},{q}": str(v) for (a, p, q), v in c.items()} for c in coeffs]}) print(f"[stage1] banked {nser} graded layer series; done.") def load_towers(): import cert_lib as CL bank = json.load(open(os.path.join(HERE, "towers_ext.json"))) assert bank["N_EXT"] == N_EXT return [CL.tower_from_json(d, N_EXT) for d in bank["periods"]] def schedule(s0, hints, s_star): """auto-refine waypoint list: insert midpoints until every leg fits inside 0.55 * local majorant radius.""" import cert_transport as CT path = [s0] + hints + [s_star] for _ in range(25): newpath, changed = [path[0]], False for i in range(len(path) - 1): x, t = path[i], path[i + 1] rho = CT.pick_rho(CT.local_shift(x)) h = CT.cxq_abs_ub(CT.cxq_sub(t, x)) if h > Fr(55, 100) * rho: mid = CT.cxq(t) xx = CT.cxq(x) m = ((xx[0] + mid[0]) / 2, (xx[1] + mid[1]) / 2) m = (Fr(round(m[0] * 40000), 40000), Fr(round(m[1] * 40000), 40000)) newpath.append(m); changed = True newpath.append(t) path = newpath if not changed: break assert not changed, "schedule did not converge" return path[1:-1] def run_one(route, dps): import cert_transport as CT from flint import arb, acb, ctx prec = int(dps * 3.3219) + 64 ctx.prec = prec towers = load_towers() if route == "R": s0, rho0 = Fr(3, 400), Fr(1, 80) hints = [Fr(21, 2000), Fr(1, 80)] else: s0, rho0 = Fr(1, 160), Fr(1, 80) hints = [(Fr(1, 125), Fr(3, 2000)), (Fr(11, 1000), Fr(3, 1000)), (Fr(33, 2500), Fr(3, 2000))] ways = schedule(s0, hints, S_STAR) print(f"[route {route} dps {dps}] prec={prec} bits; s0={s0}; " f"waypoints={[(float(CT.cxq(w)[0]), float(CT.cxq(w)[1])) for w in ways]}", flush=True) dstep = Fr(1, 10**6) targets = [S_STAR, S_STAR - dstep, S_STAR + dstep] t0 = time.time() out, diags = CT.run_route(towers, N_EXT, s0, rho0, ways, targets, prec) wall = time.time() - t0 for dg in diags: print(" ", dg, flush=True) tpin = TAU_PIN res = {t: CT.contract(out[t], tpin) for t in targets} r = res[S_STAR] # tau-pin sensitivity: same Pi, tau_im shifted by +-1e-6 tstep = Fr(1, 10**6) rtm = CT.contract(out[S_STAR], tpin - tstep) rtp = CT.contract(out[S_STAR], tpin + tstep) tslope = (rtp["W0"] - rtm["W0"]) / (2 * CT.fr2arb(tstep)) def show(name, z, digits=None): print(f" {name} = {z.str(digits or dps)}") print(f"[route {route} dps {dps}] wall {wall:.1f}s; at (s*, tau*=i*pin):") for key in ["U1", "U2", "W0", "emKcs_over_X0sq"]: show(key, res[S_STAR][key]) print(f" tau_hat (diagnostic, curve-corrupted) = {r['tau_hat'].str(12)}") print(f" |DtauW/W| at pin (offset scale) = {r['DtauW_rel'].str(6)}") w0m, w0p = res[S_STAR - dstep]["W0"], res[S_STAR + dstep]["W0"] slope = (w0p - w0m) / (2 * CT.fr2arb(dstep)) print(f" |W0|(s*-d),(s*+d) d=1e-6: {w0m.str(20)} {w0p.str(20)}") print(f" certified diff-quotients: d|W0|/ds ~ {slope.str(10)} " f"d|W0|/dImtau ~ {tslope.str(10)}") out_json = { "route": route, "dps": dps, "prec_bits": prec, "wall_s": wall, "tau_pin_im": str(tpin), "W0": r["W0"].str(dps + 10, radius=True), "absW": r["absW"].str(dps + 10, radius=True), "emK": r["emK"].str(dps + 10, radius=True), "emKcs_gauge": r["emKcs_over_X0sq"].str(30, radius=True), "U1_im": r["U1"].imag.str(dps + 10, radius=True), "U2_im": r["U2"].imag.str(dps + 10, radius=True), "tau_hat": r["tau_hat"].str(12), "DtauW_rel": r["DtauW_rel"].str(6), "dW_ds": r["dW_ds"].str(12), "W0_rad_rel": str(r["W0"].rad() / abs(r["W0"].mid())), "slope_s_mid": slope.mid().str(10), "slope_tau_mid": tslope.mid().str(10), "diags": [{k: v for k, v in d.items()} for d in diags], } fn = os.path.join(HERE, f"result_{route}_{dps}.json") with open(fn, "w") as f: json.dump(out_json, f, indent=1, default=str) print(f" banked {os.path.basename(fn)}") def parse_ball(s): import mpmath as mp mp.mp.dps = 200 s = s.strip() if s.startswith("["): body = s[1:-1] if "+/-" in body: m, r = body.split("+/-") return mp.mpf(m.strip()), mp.mpf(r.strip()) return mp.mpf(body), mp.mpf(0) return mp.mpf(s), mp.mpf(0) def matched_digits(a, b): import mpmath as mp if a == b: return 999 return int(mp.floor(-mp.log10(abs(a - b) / abs(b)))) def checks(): import mpmath as mp mp.mp.dps = 200 R = {(r, d): json.load(open(os.path.join(HERE, f"result_{r}_{d}.json"))) for r in "RC" for d in (60, 150)} W = {k: parse_ball(v["W0"]) for k, v in R.items()} print("=== KKLT-CERT W0 checks ===") print("certified |W0|(s*, tau*) balls:") for k, (m, r) in W.items(): print(f" route {k[0]} dps {k[1]:3d}: mid={mp.nstr(m, 40)} rad={mp.nstr(r, 3)}" f" rel_rad={mp.nstr(r / m, 3)}") g = {} d1 = min(matched_digits(W[('R', 60)][0], W[('R', 150)][0]), matched_digits(W[('C', 60)][0], W[('C', 150)][0])) g["c1_two_dps_matched_digits"] = d1 print(f"[c1] dps-60 vs dps-150 matched digits (worst of R,C): {d1}" f" -> {'PASS' if d1 >= 55 else 'FAIL'} (bar: 55, dps-60 radius-limited)") ov = all(abs(W[('R', d)][0] - W[('C', d)][0]) <= W[('R', d)][1] + W[('C', d)][1] for d in (60, 150)) dRC = matched_digits(W[('R', 150)][0], W[('C', 150)][0]) g["c4_routes_overlap"] = bool(ov); g["c4_matched_digits"] = dRC print(f"[c4] route R vs C: balls overlap = {ov}; matched digits (dps150) = {dRC}" f" -> {'PASS' if ov and dRC >= 140 else 'FAIL'}") pilot = mp.mpf(PILOT_W0) cert = W[('R', 150)][0] off = abs(pilot - cert) / cert mline = [l for l in open(os.path.join(HERE, "out_model_at_point.txt")) if l.startswith("model |W0| at (tau*, U(s*)) =")][0] c2 = abs(mp.mpf(mline.split("=")[1].split()[0]) - cert) / cert g["c2_same_point_model_rel"] = (f"{mp.nstr(c2, 4)} (dps-70 mpmath model at (tau*,U(s*)) read from" " out_model_at_point.txt; comparison computed here)") g["c2_vacuum_offset_rel"] = mp.nstr(off, 6) print(f"[c2] same-point model-vs-certified: {mp.nstr(c2, 4)} rel (model value read from out_model_at_point.txt;" f" the same model-truncation offset is recomputed at the vacuum by ../../kklt-evaluate.py [v1-vs-ball]" f" and ../cert_w0_vac/run_vac.py checks [gc4])") print(f"[c2] pilot VACUUM value vs certified curve-point value: rel diff = " f"{mp.nstr(off, 6)} == measured transverse-offset penalty (not an error:" f" the curve point is not the vacuum; see README.md)") g["c3_rel_radius_dps150"] = mp.nstr(W[('R', 150)][1] / cert, 3) print(f"[c3] honest radii quoted above; headline rel radius (R,150): " f"{g['c3_rel_radius_dps150']}") # tau_hat structural finding print(f"[st] tau_hat from curve stationarity = {R[('R',150)]['tau_hat']} " f"(vs vacuum 6.85546i): curve-restricted D_tauW=0 is offset-corrupted; " f"tau is fixed externally (see README.md)") g["headline_W0_ball"] = R[('R', 150)]["W0"] g["U1_im"] = R[('R', 150)]["U1_im"]; g["U2_im"] = R[('R', 150)]["U2_im"] g["emKcs_gauge"] = R[('R', 150)]["emKcs_gauge"] g["slopes"] = {"dW0_ds": R[('R', 150)]["slope_s_mid"], "dW0_dImtau": R[('R', 150)]["slope_tau_mid"]} with open(os.path.join(HERE, "checks_cert.json"), "w") as f: json.dump(g, f, indent=1) print("banked checks_cert.json") if not (d1 >= 55 and ov and dRC >= 140): sys.exit(1) if __name__ == "__main__": cmd = sys.argv[1] if cmd == "stage1": stage1() elif cmd == "route": run_one(sys.argv[2], int(sys.argv[3])) elif cmd == "checks": checks()