1"""Certified boundary window (L3): for fixed p and each n in [NMIN, NMAX], build2F(n,p) = "exists X, |X| = p, |dX| <= 11" (Lemma 3.4), solve with proof logging3(CaDiCaL 1.9.5 via python-sat), write DIMACS + DRAT to4code/proofs/certificates/boundary/, check the proof with drat-trim, and record5everything in results/boundary_certificates.jsonl. SAT outcomes (obstructions)6are recorded with the explicit X and go to results/counterexamples.jsonl.78usage: python3 code/run_boundary_window.py P NMIN NMAX [workers]9 python3 code/run_boundary_window.py --job P N10"""11import json12import os13import subprocess14import sys15import time16from concurrent.futures import ThreadPoolExecutor1718import common19from graph_constructor import edge_list_sha256, edges_formula, label20from sat_or_ilp_solver import boundary_clauses, boundary_of, solve_with_proof2122CERT_DIR = os.path.join(common.CERTS, "boundary")232425def job(p, n):26 os.makedirs(CERT_DIR, exist_ok=True)27 t0 = time.time()28 cl, x, z, nv = boundary_clauses(n, 5, p, 11)29 cnf_path = os.path.join(CERT_DIR, "P%d_5_p%d_b11.cnf" % (n, p))30 proof_path = os.path.join(CERT_DIR, "P%d_5_p%d_b11.drat" % (n, p))31 r = solve_with_proof(cl, nv, "cadical195", cnf_path, proof_path)32 r.update({"n": n, "k": 5, "p": p, "bound": 11, "graph_sha256": edge_list_sha256(edges_formula(n, 5)),33 "x_vars": "1..%d" % (2 * n), "z_vars": "%d..%d" % (2 * n + 1, 4 * n), "wall_seconds": round(time.time() - t0, 3),34 "utc": common.utcnow(), "peak_rss_mb": round(common.peak_rss_mb(), 1)})35 r.update(common.versions())36 if r["sat"]:37 model = r.pop("model")38 X = [w for w in range(2 * n) if model[x[w] - 1] > 0]39 r["X"] = [label(w, n) for w in X]40 r["boundary"] = [label(w, n) for w in boundary_of(n, 5, X)]41 r["boundary_size"] = len(r["boundary"])42 assert len(X) == p and r["boundary_size"] <= 1143 os.remove(cnf_path)44 return r454647def run_sub(args):48 p, n = args49 out = subprocess.run([sys.executable, os.path.abspath(__file__), "--job", str(p), str(n)], capture_output=True, text=True)50 if out.returncode != 0:51 return {"n": n, "p": p, "error": out.stderr[-800:]}52 return json.loads(out.stdout.strip().splitlines()[-1])535455if __name__ == "__main__":56 if sys.argv[1] == "--job":57 print(json.dumps(job(int(sys.argv[2]), int(sys.argv[3]))))58 sys.exit(0)59 p, nmin, nmax = (int(a) for a in sys.argv[1:4])60 workers = int(sys.argv[4]) if len(sys.argv) > 4 else 861 out = os.path.join(common.RESULTS, "boundary_certificates.jsonl")62 common.log_event("F", "boundary_window_start", p=p, nmin=nmin, nmax=nmax, workers=workers)63 tasks = [(p, n) for n in range(nmin, nmax + 1)]64 with ThreadPoolExecutor(max_workers=workers) as ex:65 for r in ex.map(run_sub, tasks):66 common.append_jsonl(out, r)67 if "error" in r:68 print("n=%d p=%d ERROR %s" % (r["n"], r["p"], r["error"]), flush=True)69 elif r["sat"]:70 common.append_jsonl(os.path.join(common.RESULTS, "counterexamples.jsonl"),71 {"type": "boundary_obstruction", "n": r["n"], "p": r["p"], "X": r["X"], "boundary": r["boundary"], "utc": r["utc"]})72 print("n=%d p=%d SAT (obstruction) |dX|=%d" % (r["n"], r["p"], r["boundary_size"]), flush=True)73 else:74 print("n=%d p=%d UNSAT solve=%.1fs check=%.1fs verified=%s exit=%d" % (75 r["n"], r["p"], r["solve_seconds"], r["checker_seconds"], r["checker_verified"], r["checker_exit"]), flush=True)76 common.log_event("F", "boundary_window_end", p=p, nmin=nmin, nmax=nmax)77