CI: parallelize the recycled-bit frame-pair checks (~2x on the critical path)

research/recycled-bit-integration/certificate.json

@@ -335,7 +335,7 @@
     "research/recycled-bit-integration/plan.json": "ab8887e5d852fdce16a883a1ffc3173e109708a484b986cb2574e1ee634bf784",
     "research/recycled-bit-integration/row.json": "075c8ff2f9f1b25f6b81dc0c6ec31adc766944cb59badc4bc6e42668993ca073",
     "research/recycled-bit-integration/selection146.json": "407fac7fc3f3497ddf347fc7a820a6b863c2147d18f1be325b57d37998475a3e",
-    "research/recycled-bit-integration/verify.py": "25a31677a0abd492404f4674f61eb70a8817cbb7cf96a87141ee3339a690e218",
+    "research/recycled-bit-integration/verify.py": "a5508f429d0ff6d441d9d07f6d1f9a83ce2335eb59b2041abbc65327b6a4a0b9",
     "scripts/paired_cube_bit.py": "677ff674f48860336ca96c4ab6d7fa9053bf7137d20ba368445d20e236765dbe",
     "scripts/paired_cube_network.py": "293662b7f8772dd398399a1c19966faf7d155f024543d0c06601eabb800f35cf",
     "scripts/structured_bulk_assembly.py": "e99467e14211fba699878ed036e1e2362eb8e395dc22f9ecdcecc7fbf2cc736d",

research/recycled-bit-integration/verify.py

@@ -5,6 +5,7 @@
 Finite conditional verification; retained analytic and uniform interfaces are assumptions.
 Default: reconstruct, check exact new frame pairs and complete symbolic scalar maps.
 --all: additionally check every distinct inherited frame transition over Q.
+Both frame-pair loops run on forked workers (VERIFY_JOBS, default the CPU count); each check is unchanged.
 --write: write the deterministic certificate after all checks pass.
 """
 import sys
@@ -17,6 +18,8 @@
 from fractions import Fraction as Q
 from hashlib import sha256
 import json
+import multiprocessing
+import os
 from pathlib import Path
 import random
 import time
@@ -53,6 +56,32 @@ def sources():
               ROOT / 'certificates/paired-cube-complex-input.json']
     return {str(p.relative_to(ROOT)): sha256(p.read_bytes()).hexdigest() for p in sorted(set(paths))}
 
+_EXACT = None
+_NONDEG = False
+
+def _failures(chunk):
+    E = _EXACT
+    return [(reg, a, b) for reg, a, b in chunk
+            if not (E.inside(a, b) and (not _NONDEG or (E.nondegenerate(a) and E.nondegenerate(b))))]
+
+def check_moves(exact, moves, nondegenerate=False):
+    """The moves (reg, a, b) whose frame a is not inside b (or, with nondegenerate, either frame is degenerate).
+
+    Each check is a pure function of the shared exact frames, exactly as in the serial loop, so the moves are
+    split across forked worker processes (VERIFY_JOBS, default the CPU count). Moves are grouped by target key
+    so each worker builds each annihilator basis once. Without fork, or with one job, it runs serially.
+    """
+    global _EXACT, _NONDEG
+    _EXACT, _NONDEG = exact, nondegenerate
+    moves = sorted(moves, key=lambda m: (str(m[2]), str(m[1]), str(m[0])))
+    jobs = int(os.environ.get('VERIFY_JOBS', os.cpu_count() or 1))
+    if jobs <= 1 or len(moves) < 2 * jobs or 'fork' not in multiprocessing.get_all_start_methods():
+        return _failures(moves)
+    size = -(-len(moves) // (4 * jobs))
+    chunks = [moves[i:i + size] for i in range(0, len(moves), size)]
+    with multiprocessing.get_context('fork').Pool(jobs) as pool:
+        return [m for part in pool.map(_failures, chunks) for m in part]
+
 def main():
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument('--all', action='store_true')
@@ -95,11 +124,11 @@ def main():
     assert set(fresh) == set(old.new_moves(old_led)), 'Composition changed the new-frame obligations'
     log('PASS exact composed plan, chronological ledger and unchanged set of new frame pairs')
     exact = Exact(S, W, folder)
-    for reg, a, b in fresh:
-        assert exact.inside(a, b) and exact.nondegenerate(a) and exact.nondegenerate(b), (reg, a, b)
+    failed = check_moves(exact, fresh, nondegenerate=True)
+    assert not failed, failed[:5]
     if args.all:
-        for reg, a, b in led['moves']:
-            assert exact.inside(a, b), (reg, a, b)
+        failed = check_moves(exact, led['moves'])
+        assert not failed, failed[:5]
     log('PASS exact rational nesting (%d new pairs%s)' % (len(fresh), '; all inherited pairs checked' if args.all else ''))
     outputs = {(c, tuple(T)): n for c, T, n in W['outputs']}
     assert word.complete(2, outputs) and word.complete(0, outputs)