1"""Independent lower-bound checker. Imports nothing from the solver modules.23For every record in results/boundary_certificates.jsonl with sat == False:4 1. rebuilds the CNF F(n,p) of Lemma 3.4 from the graph definition with its5 own sequential-counter implementation of "exactly p" and "at most 11",6 reads the stored DIMACS file and checks that the stored clause set is7 *implied-equivalent in the required direction*: every stored clause must be8 one of the rebuilt clauses (so the stored formula is no weaker than the9 intended one in the sense that it is a sub-formula ... see note), and the10 rebuilt core clauses (edge clauses, normalisation, cardinality) are all present;11 in practice we require byte-level equality of the sorted clause multisets12 of the stored file and an independent regeneration with python-sat's13 CardEnc (same encoding, same variable order), plus an independent14 structural check of the non-cardinality clauses.15 2. verifies SHA-256 of CNF and DRAT files against the record,16 3. runs drat-trim on (CNF, DRAT) and requires "s VERIFIED" and exit code 0.17For every record in results/lower_bound_certificates.jsonl (fort cover):18 1. checks every stored fort vertex-by-vertex (nonempty; every outside vertex19 has 0 or >= 2 neighbours in it),20 2. checks that the stored CNF consists of: the at-most-b counter over the21 seed variables, the four first-force selector blocks, and exactly the22 stored fort clauses,23 3. runs drat-trim.24Writes results/lower_bound_verification.json. Exit code 1 on any failure.25"""26import hashlib27import json28import os29import subprocess30import sys3132ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))33os.chdir(ROOT)34sys.path.insert(0, os.path.join(ROOT, ".pydeps"))35from pysat.card import CardEnc, EncType # noqa: E402 (only the cardinality encoder is reused)3637DRAT_TRIM = os.path.join(ROOT, "code", "tools", "drat-trim", "drat-trim")383940def sha256(path):41 h = hashlib.sha256()42 with open(path, "rb") as f:43 for chunk in iter(lambda: f.read(1 << 24), b""):44 h.update(chunk)45 return h.hexdigest()464748def parse_nset(spec):49 """'24,26-40,50' -> set of ints; '' or None -> None (no restriction); 'none' -> empty set."""50 if spec is None or spec == "":51 return None52 if spec.strip().lower() == "none":53 return set()54 out = set()55 for part in spec.split(","):56 part = part.strip()57 if "-" in part:58 a, b = part.split("-")59 out.update(range(int(a), int(b) + 1))60 elif part:61 out.add(int(part))62 return out636465def skip_reason(r, nset, max_bytes):66 """Returns None if the record must be checked, else a string explaining why it is skipped.67 Skipped records are reported with ok = None and never counted as verified."""68 if nset is not None and r["n"] not in nset:69 return "not in the requested n-set"70 if not os.path.exists(r["proof_path"]):71 return "proof file not present on disk (not shipped; regenerate with code/run_boundary_window.py or code/regen_fortcover_proof.py)"72 if max_bytes and os.path.getsize(r["proof_path"]) > max_bytes:73 return "proof larger than VERIFY_MAX_PROOF_BYTES=%d (run with FULL=1 / no limit to check it)" % max_bytes74 return None757677def adjacency(n, k):78 adj = [set() for _ in range(2 * n)]79 for i in range(n):80 for a, b in ((i, (i + 1) % n), (i, n + i), (n + i, n + (i + k) % n)):81 adj[a].add(b)82 adj[b].add(a)83 return adj848586def read_dimacs(path):87 clauses = []88 nv = 089 with open(path) as f:90 for line in f:91 if line.startswith("c"):92 continue93 if line.startswith("p"):94 nv = int(line.split()[2])95 continue96 lits = [int(t) for t in line.split()]97 if lits and lits[-1] == 0:98 lits = lits[:-1]99 if lits:100 clauses.append(tuple(lits))101 return nv, clauses102103104def card_clauses(lits, bound, top, kind):105 enc = (CardEnc.atmost if kind == "atmost" else CardEnc.equals)(lits=lits, bound=bound, top_id=top, encoding=EncType.seqcounter)106 return [tuple(c) for c in enc.clauses], max(top, enc.nv)107108109def rebuild_boundary(n, k, p, bmax):110 adj = adjacency(n, k)111 N = 2 * n112 x = list(range(1, N + 1))113 z = list(range(N + 1, 2 * N + 1))114 core = []115 for w in range(N):116 for t in adj[w]:117 core.append((x[w], -x[t], z[w]))118 core.append((-z[w], -x[w]))119 top = 2 * N120 c1, top = card_clauses(x, p, top, "equals")121 c2, top = card_clauses(z, bmax, top, "atmost")122 norm = [(x[0], x[n])]123 return core, c1 + c2, norm124125126def check_boundary_record(r):127 n, p = r["n"], r["p"]128 out = {"n": n, "p": p, "kind": "boundary"}129 cnf_path, proof_path = r["cnf_path"], r["proof_path"]130 out["cnf_sha_ok"] = sha256(cnf_path) == r["cnf_sha256"]131 out["proof_sha_ok"] = sha256(proof_path) == r["proof_sha256"]132 nv, stored = read_dimacs(cnf_path)133 core, card, norm = rebuild_boundary(n, 5, p, 11)134 stored_set = sorted(stored)135 rebuilt = sorted(core + card + norm)136 out["cnf_matches_rebuild"] = stored_set == rebuilt137 # structural check independent of CardEnc: every core clause and the normalisation clause are present138 sset = set(stored)139 out["core_clauses_present"] = all(c in sset for c in core) and norm[0] in sset140 res = subprocess.run([DRAT_TRIM, cnf_path, proof_path], capture_output=True, text=True)141 out["drat_trim_exit"] = res.returncode142 out["drat_trim_verified"] = "s VERIFIED" in res.stdout143 out["ok"] = all(out[k] for k in ("cnf_sha_ok", "proof_sha_ok", "cnf_matches_rebuild", "core_clauses_present", "drat_trim_verified")) and res.returncode == 0144 return out145146147def is_fort(adj, F, N):148 Fs = set(F)149 if not Fs:150 return False151 for w in range(N):152 if w in Fs:153 continue154 if len(adj[w] & Fs) == 1:155 return False156 return True157158159def check_fort_record(r):160 n, b = r["n"], r["bound"]161 out = {"n": n, "bound": b, "kind": "fortcover"}162 cnf_path, proof_path, forts_path = r["cnf_path"], r["proof_path"], r["forts_path"]163 out["cnf_sha_ok"] = sha256(cnf_path) == r["cnf_sha256"]164 out["proof_sha_ok"] = sha256(proof_path) == r["proof_sha256"]165 out["forts_sha_ok"] = sha256(forts_path) == r["forts_sha256"]166 fd = json.load(open(forts_path))167 adj = adjacency(n, 5)168 N = 2 * n169 svar = fd["svar"]170 out["all_forts_valid"] = all(is_fort(adj, F, N) for F in fd["forts"])171 out["num_forts"] = len(fd["forts"])172 nv, stored = read_dimacs(cnf_path)173 sset = set(stored)174 # expected non-fort clauses175 expected = []176 card, top = card_clauses(svar, b, max(svar), "atmost")177 expected += card178 sel = []179 reps = [(0, n), (0, 1 % n), (n, 0), (n, n + (5 % n))]180 for (src, tgt) in reps:181 top += 1182 c = top183 sel.append(c)184 expected.append((-c, svar[src]))185 expected.append((-c, -svar[tgt]))186 for w2 in sorted(adj[src]):187 if w2 != tgt:188 expected.append((-c, svar[w2]))189 expected.append(tuple(sel))190 fort_clauses = [tuple(svar[w] for w in F) for F in fd["forts"]]191 out["cnf_matches_rebuild"] = sorted(stored) == sorted(expected + fort_clauses)192 res = subprocess.run([DRAT_TRIM, cnf_path, proof_path], capture_output=True, text=True)193 out["drat_trim_exit"] = res.returncode194 out["drat_trim_verified"] = "s VERIFIED" in res.stdout195 out["ok"] = all(out[k] for k in ("cnf_sha_ok", "proof_sha_ok", "forts_sha_ok", "all_forts_valid", "cnf_matches_rebuild", "drat_trim_verified")) and res.returncode == 0196 return out197198199if __name__ == "__main__":200 # Environment (all optional; defaults check every record whose proof is on disk):201 # VERIFY_BOUNDARY_N e.g. "24,26-40,50,60" restrict boundary records to these n ("none": skip all)202 # VERIFY_FORTCOVER_N likewise for fort-cover records203 # VERIFY_MAX_PROOF_BYTES skip proofs larger than this (0 = no limit)204 # VERIFY_OUT output file (default results/lower_bound_verification.json)205 # VERIFY_MERGE=1 keep rows of the existing output file for records not checked in this run206 # Skipped records are listed with ok = null and a "skipped" reason; they are never counted as verified.207 import datetime208 from concurrent.futures import ThreadPoolExecutor209 workers = int(sys.argv[1]) if len(sys.argv) > 1 else 8210 bset = parse_nset(os.environ.get("VERIFY_BOUNDARY_N"))211 fset = parse_nset(os.environ.get("VERIFY_FORTCOVER_N"))212 max_bytes = int(os.environ.get("VERIFY_MAX_PROOF_BYTES", "0") or 0)213 out_path = os.environ.get("VERIFY_OUT") or "results/lower_bound_verification.json"214 merge = os.environ.get("VERIFY_MERGE") == "1" and os.path.exists(out_path)215 old = json.load(open(out_path)) if merge else {"boundary": [], "fortcover": []}216 report = {"utc": datetime.datetime.now(datetime.timezone.utc).isoformat(timespec="seconds"),217 "settings": {"boundary_n": os.environ.get("VERIFY_BOUNDARY_N"), "fortcover_n": os.environ.get("VERIFY_FORTCOVER_N"),218 "max_proof_bytes": max_bytes, "merge": merge, "workers": workers},219 "boundary": [], "fortcover": []}220 ok_all = True221 counts = {}222 for path, fn, key, nset, idf in (("results/boundary_certificates.jsonl", check_boundary_record, "boundary", bset, lambda r: (r["n"], r["p"])),223 ("results/lower_bound_certificates.jsonl", check_fort_record, "fortcover", fset, lambda r: (r["n"], r["bound"]))):224 if not os.path.exists(path):225 continue226 recs = [json.loads(line) for line in open(path)]227 recs = [r for r in recs if r.get("sat") is False and "error" not in r]228 todo, rows = [], []229 for r in recs:230 why = skip_reason(r, nset, max_bytes)231 if why is None:232 todo.append(r)233 else:234 kept = next((o for o in old.get(key, []) if idf(o) == idf(r) and o.get("ok") is not None), None) if merge else None235 if kept is not None:236 kept = dict(kept, reused_from_previous_run=old.get("utc", "earlier run"))237 rows.append(kept)238 else:239 rows.append({"n": r["n"], ("p" if key == "boundary" else "bound"): idf(r)[1], "kind": key, "ok": None, "skipped": why})240 with ThreadPoolExecutor(max_workers=workers) as ex:241 for o in ex.map(fn, todo):242 rows.append(o)243 print(json.dumps(o), flush=True)244 report[key] = rows245 checked = [o for o in rows if o.get("ok") is not None and not o.get("reused_from_previous_run")]246 counts[key] = {"records": len(rows), "checked_now": len(checked), "ok_now": sum(1 for o in checked if o["ok"]),247 "reused": sum(1 for o in rows if o.get("reused_from_previous_run")),248 "skipped": sum(1 for o in rows if o.get("ok") is None)}249 ok_all &= all(o["ok"] for o in rows if o.get("ok") is not None)250 report["counts"] = counts251 report["all_checked_ok"] = ok_all252 report["all_ok"] = ok_all and all(c["skipped"] == 0 for c in counts.values())253 report["drat_trim_sha256"] = sha256(os.path.join(ROOT, "code", "tools", "drat-trim", "drat-trim.c"))254 with open(out_path, "w") as f:255 json.dump(report, f, indent=1)256 print("counts: %s; every checked certificate ok: %s; all records checked: %s" % (json.dumps(counts), ok_all, report["all_ok"]))257 sys.exit(0 if ok_all else 1)258