"""Independent lower-bound checker. Imports nothing from the solver modules. For every record in results/boundary_certificates.jsonl with sat == False: 1. rebuilds the CNF F(n,p) of Lemma 3.4 from the graph definition with its own sequential-counter implementation of "exactly p" and "at most 11", reads the stored DIMACS file and checks that the stored clause set is *implied-equivalent in the required direction*: every stored clause must be one of the rebuilt clauses (so the stored formula is no weaker than the intended one in the sense that it is a sub-formula ... see note), and the rebuilt core clauses (edge clauses, normalisation, cardinality) are all present; in practice we require byte-level equality of the sorted clause multisets of the stored file and an independent regeneration with python-sat's CardEnc (same encoding, same variable order), plus an independent structural check of the non-cardinality clauses. 2. verifies SHA-256 of CNF and DRAT files against the record, 3. runs drat-trim on (CNF, DRAT) and requires "s VERIFIED" and exit code 0. For every record in results/lower_bound_certificates.jsonl (fort cover): 1. checks every stored fort vertex-by-vertex (nonempty; every outside vertex has 0 or >= 2 neighbours in it), 2. checks that the stored CNF consists of: the at-most-b counter over the seed variables, the four first-force selector blocks, and exactly the stored fort clauses, 3. runs drat-trim. Writes results/lower_bound_verification.json. Exit code 1 on any failure. """ import hashlib import json import os import subprocess import sys ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) os.chdir(ROOT) sys.path.insert(0, os.path.join(ROOT, ".pydeps")) from pysat.card import CardEnc, EncType # noqa: E402 (only the cardinality encoder is reused) DRAT_TRIM = os.path.join(ROOT, "code", "tools", "drat-trim", "drat-trim") def sha256(path): h = hashlib.sha256() with open(path, "rb") as f: for chunk in iter(lambda: f.read(1 << 24), b""): h.update(chunk) return h.hexdigest() def parse_nset(spec): """'24,26-40,50' -> set of ints; '' or None -> None (no restriction); 'none' -> empty set.""" if spec is None or spec == "": return None if spec.strip().lower() == "none": return set() out = set() for part in spec.split(","): part = part.strip() if "-" in part: a, b = part.split("-") out.update(range(int(a), int(b) + 1)) elif part: out.add(int(part)) return out def skip_reason(r, nset, max_bytes): """Returns None if the record must be checked, else a string explaining why it is skipped. Skipped records are reported with ok = None and never counted as verified.""" if nset is not None and r["n"] not in nset: return "not in the requested n-set" if not os.path.exists(r["proof_path"]): return "proof file not present on disk (not shipped; regenerate with code/run_boundary_window.py or code/regen_fortcover_proof.py)" if max_bytes and os.path.getsize(r["proof_path"]) > max_bytes: return "proof larger than VERIFY_MAX_PROOF_BYTES=%d (run with FULL=1 / no limit to check it)" % max_bytes return None def adjacency(n, k): adj = [set() for _ in range(2 * n)] for i in range(n): for a, b in ((i, (i + 1) % n), (i, n + i), (n + i, n + (i + k) % n)): adj[a].add(b) adj[b].add(a) return adj def read_dimacs(path): clauses = [] nv = 0 with open(path) as f: for line in f: if line.startswith("c"): continue if line.startswith("p"): nv = int(line.split()[2]) continue lits = [int(t) for t in line.split()] if lits and lits[-1] == 0: lits = lits[:-1] if lits: clauses.append(tuple(lits)) return nv, clauses def card_clauses(lits, bound, top, kind): enc = (CardEnc.atmost if kind == "atmost" else CardEnc.equals)(lits=lits, bound=bound, top_id=top, encoding=EncType.seqcounter) return [tuple(c) for c in enc.clauses], max(top, enc.nv) def rebuild_boundary(n, k, p, bmax): adj = adjacency(n, k) N = 2 * n x = list(range(1, N + 1)) z = list(range(N + 1, 2 * N + 1)) core = [] for w in range(N): for t in adj[w]: core.append((x[w], -x[t], z[w])) core.append((-z[w], -x[w])) top = 2 * N c1, top = card_clauses(x, p, top, "equals") c2, top = card_clauses(z, bmax, top, "atmost") norm = [(x[0], x[n])] return core, c1 + c2, norm def check_boundary_record(r): n, p = r["n"], r["p"] out = {"n": n, "p": p, "kind": "boundary"} cnf_path, proof_path = r["cnf_path"], r["proof_path"] out["cnf_sha_ok"] = sha256(cnf_path) == r["cnf_sha256"] out["proof_sha_ok"] = sha256(proof_path) == r["proof_sha256"] nv, stored = read_dimacs(cnf_path) core, card, norm = rebuild_boundary(n, 5, p, 11) stored_set = sorted(stored) rebuilt = sorted(core + card + norm) out["cnf_matches_rebuild"] = stored_set == rebuilt # structural check independent of CardEnc: every core clause and the normalisation clause are present sset = set(stored) out["core_clauses_present"] = all(c in sset for c in core) and norm[0] in sset res = subprocess.run([DRAT_TRIM, cnf_path, proof_path], capture_output=True, text=True) out["drat_trim_exit"] = res.returncode out["drat_trim_verified"] = "s VERIFIED" in res.stdout 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 == 0 return out def is_fort(adj, F, N): Fs = set(F) if not Fs: return False for w in range(N): if w in Fs: continue if len(adj[w] & Fs) == 1: return False return True def check_fort_record(r): n, b = r["n"], r["bound"] out = {"n": n, "bound": b, "kind": "fortcover"} cnf_path, proof_path, forts_path = r["cnf_path"], r["proof_path"], r["forts_path"] out["cnf_sha_ok"] = sha256(cnf_path) == r["cnf_sha256"] out["proof_sha_ok"] = sha256(proof_path) == r["proof_sha256"] out["forts_sha_ok"] = sha256(forts_path) == r["forts_sha256"] fd = json.load(open(forts_path)) adj = adjacency(n, 5) N = 2 * n svar = fd["svar"] out["all_forts_valid"] = all(is_fort(adj, F, N) for F in fd["forts"]) out["num_forts"] = len(fd["forts"]) nv, stored = read_dimacs(cnf_path) sset = set(stored) # expected non-fort clauses expected = [] card, top = card_clauses(svar, b, max(svar), "atmost") expected += card sel = [] reps = [(0, n), (0, 1 % n), (n, 0), (n, n + (5 % n))] for (src, tgt) in reps: top += 1 c = top sel.append(c) expected.append((-c, svar[src])) expected.append((-c, -svar[tgt])) for w2 in sorted(adj[src]): if w2 != tgt: expected.append((-c, svar[w2])) expected.append(tuple(sel)) fort_clauses = [tuple(svar[w] for w in F) for F in fd["forts"]] out["cnf_matches_rebuild"] = sorted(stored) == sorted(expected + fort_clauses) res = subprocess.run([DRAT_TRIM, cnf_path, proof_path], capture_output=True, text=True) out["drat_trim_exit"] = res.returncode out["drat_trim_verified"] = "s VERIFIED" in res.stdout 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 == 0 return out if __name__ == "__main__": # Environment (all optional; defaults check every record whose proof is on disk): # VERIFY_BOUNDARY_N e.g. "24,26-40,50,60" restrict boundary records to these n ("none": skip all) # VERIFY_FORTCOVER_N likewise for fort-cover records # VERIFY_MAX_PROOF_BYTES skip proofs larger than this (0 = no limit) # VERIFY_OUT output file (default results/lower_bound_verification.json) # VERIFY_MERGE=1 keep rows of the existing output file for records not checked in this run # Skipped records are listed with ok = null and a "skipped" reason; they are never counted as verified. import datetime from concurrent.futures import ThreadPoolExecutor workers = int(sys.argv[1]) if len(sys.argv) > 1 else 8 bset = parse_nset(os.environ.get("VERIFY_BOUNDARY_N")) fset = parse_nset(os.environ.get("VERIFY_FORTCOVER_N")) max_bytes = int(os.environ.get("VERIFY_MAX_PROOF_BYTES", "0") or 0) out_path = os.environ.get("VERIFY_OUT") or "results/lower_bound_verification.json" merge = os.environ.get("VERIFY_MERGE") == "1" and os.path.exists(out_path) old = json.load(open(out_path)) if merge else {"boundary": [], "fortcover": []} report = {"utc": datetime.datetime.now(datetime.timezone.utc).isoformat(timespec="seconds"), "settings": {"boundary_n": os.environ.get("VERIFY_BOUNDARY_N"), "fortcover_n": os.environ.get("VERIFY_FORTCOVER_N"), "max_proof_bytes": max_bytes, "merge": merge, "workers": workers}, "boundary": [], "fortcover": []} ok_all = True counts = {} for path, fn, key, nset, idf in (("results/boundary_certificates.jsonl", check_boundary_record, "boundary", bset, lambda r: (r["n"], r["p"])), ("results/lower_bound_certificates.jsonl", check_fort_record, "fortcover", fset, lambda r: (r["n"], r["bound"]))): if not os.path.exists(path): continue recs = [json.loads(line) for line in open(path)] recs = [r for r in recs if r.get("sat") is False and "error" not in r] todo, rows = [], [] for r in recs: why = skip_reason(r, nset, max_bytes) if why is None: todo.append(r) else: 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 None if kept is not None: kept = dict(kept, reused_from_previous_run=old.get("utc", "earlier run")) rows.append(kept) else: rows.append({"n": r["n"], ("p" if key == "boundary" else "bound"): idf(r)[1], "kind": key, "ok": None, "skipped": why}) with ThreadPoolExecutor(max_workers=workers) as ex: for o in ex.map(fn, todo): rows.append(o) print(json.dumps(o), flush=True) report[key] = rows checked = [o for o in rows if o.get("ok") is not None and not o.get("reused_from_previous_run")] counts[key] = {"records": len(rows), "checked_now": len(checked), "ok_now": sum(1 for o in checked if o["ok"]), "reused": sum(1 for o in rows if o.get("reused_from_previous_run")), "skipped": sum(1 for o in rows if o.get("ok") is None)} ok_all &= all(o["ok"] for o in rows if o.get("ok") is not None) report["counts"] = counts report["all_checked_ok"] = ok_all report["all_ok"] = ok_all and all(c["skipped"] == 0 for c in counts.values()) report["drat_trim_sha256"] = sha256(os.path.join(ROOT, "code", "tools", "drat-trim", "drat-trim.c")) with open(out_path, "w") as f: json.dump(report, f, indent=1) print("counts: %s; every checked certificate ok: %s; all records checked: %s" % (json.dumps(counts), ok_all, report["all_ok"])) sys.exit(0 if ok_all else 1)