From e9449ef0fe5bedd75682e960defae442d8aa004d Mon Sep 17 00:00:00 2001 From: Drew T <50529377+Druthulu@users.noreply.github.com> Date: Sat, 29 Aug 2026 18:45:49 -0600 Subject: [PATCH] =?UTF-8?q?feat(tools):=20parallel=5Fgate.py=20+=20twin=5F?= =?UTF-8?q?sweep.py=20=E2=80=94=20worktree-isolated=20concurrent=20gates,?= =?UTF-8?q?=20and=20the=20twin=20lever=20as=20one=20command?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit parallel_gate.py — the per-binary byte gate was serial BY HARNESS, not by nature. Each binary already compiles into its own build//, links its own .ld and checks its own SHA; what serialized it was shared mutable state in the ONE checkout (the splice, assert_write_set's GLOBAL git status, and the deliberately-broad `git add -u src/` that must stay broad). Measured: a 109-binary sweep ran ~1 min/binary on a 32-core box at load 1.4 (~4% utilisation), and an xargs -P 4 attempt over the shared tree CORRUPTED it earlier today. Fix is ISOLATION, not locking: one git worktree per worker (own index, own src/, own build/). Workers gate and NEVER commit; the orchestrator adopts only drafts the gate ACCEPTED, and only where the main tree still matches the pinned baseline (otherwise REFUSED, never clobbered), then ONE commit and ONE R22 clean-fleet sweep verifies the merged whole. Measured on 85 binaries / 234 candidates: 178 banked in 12m19s wall for 127m40s CPU = 10.4x parallelism, ~7x end-to-end vs serial, 99 files merged, 0 refused, check-all 213/213. Five things a fresh worktree does NOT have, each found by measurement and each first appearing as "the draft failed": splat-generated include/*.inc, the EMPTY tools/maspsx submodule, gitignored tools/bin (cc1) + tools/psyq, build/{,assets/} outputs, and extracted/retail. Dirs mixing tracked and untracked content cannot be symlinked wholesale (ln -s nests INSIDE them) — hence link_missing(). The negative control that catches all of it: an UNMODIFIED binary must build BYTE-IDENTICAL in the worktree. Before that control, the first parallel run reported a clean, plausible "0 banked across 4 binaries" that was pure environment artifact. twin_sweep.py — enumerate every open stub that has an ALREADY-BANKED structural twin, remap it mechanically, gate it. Yields measured today: h_exact 139/157 = 88.5%, h_norm 36/45 then 178/234 = 76-80%. h_seq stays refused (Phase-26) and is not used. THE POOL REFILLS: each bank becomes an exemplar for its siblings — 165 fresh candidates existed immediately after banking 178. Run it BEFORE drawing any wave; t5_cards does not build seed_ref, so cards assert "no banked twin" for these and agents redraft answers we already hold (a t5u Opus slot ground ov_SC03_023:func_8017BEBC to closeness 45 while ov_SC02_004 held a byte-identical banked copy). --- tools/parallel_gate.py | 279 +++++++++++++++++++++++++++++++++++++++++ tools/twin_sweep.py | 135 ++++++++++++++++++++ 2 files changed, 414 insertions(+) create mode 100644 tools/parallel_gate.py create mode 100644 tools/twin_sweep.py diff --git a/tools/parallel_gate.py b/tools/parallel_gate.py new file mode 100644 index 000000000..4c394a616 --- /dev/null +++ b/tools/parallel_gate.py @@ -0,0 +1,279 @@ +#!/usr/bin/env python3 +"""parallel_gate.py — gate MANY binaries CONCURRENTLY in isolated git worktrees, merge only the +passing drafts back into the main tree, and commit ONCE (P31 S65). + +WHY THIS EXISTS +--------------- +`gate_stage.py` is single-threaded BY HARNESS, not by nature. The build is already per-binary +(`build//*.o`, its own `.ld`, its own SHA check), so N binaries could compile at once. What +serializes them is SHARED MUTABLE STATE, all of it in the working tree: + + * the splice writes `src//*.c` in the one checkout; + * `assert_write_set` measures a GLOBAL `git status`, so a concurrent run's writes read as this + run's blast-radius violation; + * the commit is a deliberately broad `git add -u src/` (it must be — propagation legitimately + touches many overlays, and a narrower glob once DROPPED four R22-verified banks). + +Measured cost of that serialization on 2026-08-29: a 109-binary h_norm sweep ran ~1 min/binary on a +32-core box at load 1.4 — about 4% utilisation — while an `xargs -P 4` attempt over the SAME tree +corrupted it (one aborted run's stage edits were swept into a concurrent run's commit, 696 broken +lines into ov_MAIN_012, check-all 212/213). + +THE FIX IS ISOLATION, NOT LOCKING. Each worker gets its own `git worktree` (own index, own +`src/`, own `build/`), so the three shared-state problems above simply do not exist. Workers NEVER +commit and never touch the main tree. The orchestrator then applies only the drafts the gate +ACCEPTED, commits once, and runs ONE R22 clean-fleet sweep over the merged result — so the +whole-binary byte gate remains the sole arbiter (G3/P9) and the final state is verified as a whole, +not as N independent claims. + +WHAT A WORKER NEEDS (measured, not guessed) +------------------------------------------- + * tracked sources: come free with the worktree (src/, include/, config/, tools/, Makefile) + * `asm/` -> SYMLINK to the main tree. A gate never writes it (only `make extract` does), and + it is 453 MB — copying it per worker would be the whole cost of the exercise. + * `.venv/` -> SYMLINK (gitignored, so absent in a fresh worktree; gate_stage shells `.venv/bin/python`) + * `build//{.ld,undefined_syms_auto.txt,undefined_funcs_auto.txt}` -> COPY (1.9 MB/binary). + These are splat-extract outputs, untracked, and the link step needs them. Copying + them is what lets a worker skip `make extract` entirely. + +MERGE SAFETY +------------ +Every worktree is created from ONE pinned commit. Before adopting a worker's file the orchestrator +asserts the main tree's copy still matches that pinned version — if anything else changed that +binary meanwhile, the file is REFUSED rather than clobbered (the failure mode this whole tool exists +to prevent). Adopting is per-BINARY and per-FILE, never a blanket add. + + parallel_gate.py --plan plan.json [--workers 8] [--commit] [--r22] [--keep] + plan.json: [{"binary": "ov_SC03_099", "drafts": "/abs/path/to/dir"}, ...] +""" +import argparse, json, os, shutil, subprocess, sys, time +from concurrent.futures import ThreadPoolExecutor, as_completed + +REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +PY = os.path.join(REPO, ".venv/bin/python") +WT_ROOT = os.path.join(REPO, ".run/pgate") +GEN = ("{b}.ld", "undefined_syms_auto.txt", "undefined_funcs_auto.txt") + + +def sh(cmd, cwd=REPO, timeout=None): + return subprocess.run(cmd, capture_output=True, text=True, cwd=cwd, timeout=timeout) + + +def head_commit(): + return sh(["git", "rev-parse", "HEAD"]).stdout.strip() + + +def make_worktree(idx, pin): + """One reusable worktree per worker slot, pinned to `pin` so the merge check has a baseline.""" + wt = os.path.join(WT_ROOT, "wt%d" % idx) + if os.path.exists(wt): + sh(["git", "worktree", "remove", "--force", wt]) + os.makedirs(WT_ROOT, exist_ok=True) + r = sh(["git", "worktree", "add", "--detach", wt, pin]) + if r.returncode: + raise SystemExit("[pgate] worktree add failed: %s" % (r.stderr or r.stdout)[-300:]) + # read-only, huge, gitignored: asm/ (453 MB of .s), the venv gate_stage shells, expected/. + # A gate never writes them — only `make extract` does — so symlinks are correct and copying + # would dominate the cost. + for link in ("asm", ".venv", "expected"): + tgt, dst = os.path.join(REPO, link), os.path.join(wt, link) + if os.path.exists(tgt) and not os.path.exists(dst): + os.symlink(tgt, dst) + # DIRECTORIES THAT MIX TRACKED AND UNTRACKED CONTENT cannot be symlinked wholesale: the worktree + # already has the tracked half, so `ln -s` would nest the link INSIDE it (measured, twice). Link + # the MISSING ENTRIES instead. extracted/retail holds 3 tracked files in git and the multi-GB + # ROM payload only on disk; the OBJCOPY size/pad step stats that payload. + link_missing(os.path.join(REPO, "extracted/retail"), os.path.join(wt, "extracted/retail")) + # UNTRACKED GENERATED HEADERS. splat writes these into include/ and they are gitignored, so a + # fresh worktree has none of them and `as` dies with "can't open macro.inc for reading" on the + # very first .data.o — which surfaces as every draft "failed" and reads exactly like a wave of + # bad drafts (measured: 4 binaries, 10 drafts, 0 banked, before this line existed). Copy, do not + # symlink: they are small and a worker must never be able to write the main tree's copy. + for f in ("macro.inc", "gte_macros.inc", "labels.inc", "include_asm.h"): + srcf = os.path.join(REPO, "include", f) + if os.path.exists(srcf): + shutil.copy2(srcf, os.path.join(wt, "include", f)) + # THE TOOLCHAIN ITSELF. `tools/maspsx` is a git SUBMODULE (a worktree creates the directory and + # leaves it EMPTY) and `tools/bin/**` (cc1) + parts of `tools/psyq` are gitignored binaries. The + # directories therefore EXIST but are short, which is why a naive "is it missing?" check passes + # and the build then dies with "cc1: No such file or directory". Link the CONTENTS, per entry, + # so an existing-but-empty dir is repaired rather than skipped. + for sub in ("bin", "maspsx", "psyq"): + link_missing(os.path.join(REPO, "tools", sub), os.path.join(wt, "tools", sub)) + return wt + + +def link_missing(s_dir, d_dir): + """Symlink every entry of s_dir that d_dir lacks. Never replaces an existing entry, so a + directory holding BOTH tracked files (present in the worktree) and gitignored ones (present only + in the main tree) is repaired entry-by-entry.""" + if not os.path.isdir(s_dir): + return + os.makedirs(d_dir, exist_ok=True) + for e in os.listdir(s_dir): + dst = os.path.join(d_dir, e) + if not os.path.lexists(dst): + os.symlink(os.path.join(s_dir, e), dst) + + +def stage_generated(wt, binary): + """Copy splat's untracked per-binary outputs so the worker can LINK without re-extracting.""" + src, dst = os.path.join(REPO, "build", binary), os.path.join(wt, "build", binary) + os.makedirs(dst, exist_ok=True) + missing = [] + for pat in GEN: + f = pat.format(b=binary) + s = os.path.join(src, f) + if os.path.exists(s): + shutil.copy2(s, os.path.join(dst, f)) + else: + missing.append(f) + # EXTRACTED ASSET OBJECTS. The link line pulls build/assets//*.o (splat's binary-data + # objects, e.g. trailing.o). Without them the compile succeeds and the LINK dies with + # "cannot find build/assets//trailing.o" — a failure that arrives late and, again, reads + # as a bad draft. 8 KB per binary; copy them. + a_src, a_dst = os.path.join(REPO, "build/assets", binary), os.path.join(wt, "build/assets", binary) + if os.path.isdir(a_src): + os.makedirs(a_dst, exist_ok=True) + for f in os.listdir(a_src): + sp = os.path.join(a_src, f) + if os.path.isfile(sp): + shutil.copy2(sp, os.path.join(a_dst, f)) + return missing + + +def stubs_of(wt, binary): + """{fn} still INCLUDE_ASM in THIS tree — the bank oracle (a bank REMOVES a stub).""" + r = sh([PY, "-c", + "import sys;sys.path.insert(0,'tools');import corpus;" + "print('\\n'.join(sorted(s.symbol for s in corpus.stubs(%r).values())))" % binary], + cwd=wt) + if r.returncode: + return None + return set(r.stdout.split()) + + +def gate_one(idx, pin, job): + binary, drafts = job["binary"], job["drafts"] + t0 = time.time() + wt = job.get("_wt") + try: + missing = stage_generated(wt, binary) + before = stubs_of(wt, binary) + if before is None: + return {"binary": binary, "banked": [], "error": "corpus refused in worktree"} + r = sh([PY, "tools/gate_stage.py", "--drafts", drafts, "--binary", binary, + "--no-propagate", "--source-tag", "pgate"], cwd=wt, timeout=3600) + after = stubs_of(wt, binary) + banked = sorted(before - after) if after is not None else [] + files = {} + if banked: # capture the worker's resulting TU text for the merge + for rel in sh(["git", "status", "--porcelain", "--", "src/%s/" % binary], + cwd=wt).stdout.splitlines(): + p = rel[3:].strip() + if p: + files[p] = open(os.path.join(wt, p)).read() + return {"binary": binary, "banked": banked, "files": files, "secs": round(time.time() - t0, 1), + "missing_generated": missing, + "rc": r.returncode, "tail": (r.stdout or r.stderr)[-200:] if not banked else ""} + except Exception as e: + return {"binary": binary, "banked": [], "error": "%s: %s" % (type(e).__name__, str(e)[:160])} + + +def main(): + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("--plan", required=True, help="JSON [{binary, drafts}, ...]") + ap.add_argument("--workers", type=int, default=8) + ap.add_argument("--commit", action="store_true") + ap.add_argument("--r22", action="store_true", help="clean-fleet verify after the merge (do it)") + ap.add_argument("--keep", action="store_true", help="keep worktrees for inspection") + ap.add_argument("--no-merge", action="store_true", + help="gate in the worktrees and REPORT only; never write the main tree. The " + "dirty-tree precondition applies to the MERGE, not to gating — a worker " + "is isolated by construction — so this mode runs safely alongside another " + "gate/sweep and is how you measure the speedup without waiting for a quiet tree.") + a = ap.parse_args() + + if not a.no_merge: + dirty = sh(["git", "status", "--porcelain", "--", "src/", "config/"]).stdout.strip() + if dirty: + sys.exit("REFUSED: src/ or config/ is dirty — commit or inspect first (R42):\n" + dirty[:400]) + + jobs = json.load(open(a.plan)) + jobs = [j for j in jobs if os.path.isdir(j["drafts"]) and + any(f.endswith(".c") for f in os.listdir(j["drafts"]))] + if not jobs: + sys.exit("no jobs with drafts") + pin = head_commit() + nw = max(1, min(a.workers, len(jobs))) + print("[pgate] %d binaries, %d workers, pinned at %s" % (len(jobs), nw, pin[:9]), flush=True) + + wts = [make_worktree(i, pin) for i in range(nw)] + results, t0 = [], time.time() + try: + # one worktree per SLOT, handed round-robin: a worker is reused across jobs, never shared + # concurrently (each future owns its slot for its whole run). + slot_q = list(range(nw)) + def run(job, slot): + job["_wt"] = wts[slot] + return gate_one(slot, pin, job) + with ThreadPoolExecutor(max_workers=nw) as ex: + futs = {} + for i, job in enumerate(jobs): + futs[ex.submit(run, job, i % nw)] = job["binary"] + for f in as_completed(futs): + r = f.result(); results.append(r) + print("[pgate] %-14s banked %-3d %s" % (r["binary"], len(r["banked"]), + r.get("error") or ("%.0fs" % r.get("secs", 0))), flush=True) + finally: + if not a.keep: + for wt in wts: + sh(["git", "worktree", "remove", "--force", wt]) + + total = sum(len(r["banked"]) for r in results) + print("[pgate] %d banked across %d binaries in %.0fs wall (workers=%d)" + % (total, sum(1 for r in results if r["banked"]), time.time() - t0, nw), flush=True) + + # ---- MERGE: adopt only files whose main-tree copy is still the pinned version (never clobber) + adopted, refused = [], [] + if a.no_merge: + print("[pgate] --no-merge: main tree untouched; %d file(s) held in .run/pgate_results.json" + % sum(len(r.get("files") or {}) for r in results), flush=True) + json.dump(results, open(os.path.join(REPO, ".run/pgate_results.json"), "w"), indent=1) + return + for r in results: + for p, text in (r.get("files") or {}).items(): + base = sh(["git", "show", "%s:%s" % (pin, p)]).stdout + cur = open(os.path.join(REPO, p)).read() if os.path.exists(os.path.join(REPO, p)) else None + if cur != base: + refused.append(p); continue + open(os.path.join(REPO, p), "w").write(text) + adopted.append(p) + print("[pgate] merged %d file(s); REFUSED %d (main tree moved under them): %s" + % (len(adopted), len(refused), " ".join(refused[:5])), flush=True) + + if a.r22 and adopted: + print("[pgate] R22 clean-fleet verify …", flush=True) + sh(["make", "clean"], timeout=1800) + sh(["make", "extract-all", "JOBS=32"], timeout=7200) + c = sh(["make", "check-all", "JOBS=32"], timeout=7200) + line = [l for l in c.stdout.splitlines() if l.startswith("check-all:")] + print("[pgate] %s" % (line[-1] if line else "check-all produced no summary"), flush=True) + if not line or "0 failed" not in line[-1]: + sys.exit("[pgate] ABORT — fleet NOT green after merge; files left in tree for inspection (R42)") + + if a.commit and adopted: + sh(["git", "add", "--"] + adopted) + msg = ("feat(decomp): parallel gate — %d fns across %d binaries (%d workers)\n\n%s" + % (total, sum(1 for r in results if r["banked"]), nw, + "\n".join(" %-14s %s" % (r["binary"], " ".join(r["banked"])) + for r in results if r["banked"])[:3000])) + sh(["git", "-c", "user.name=Drew T", "-c", "user.email=50529377+Druthulu@users.noreply.github.com", + "commit", "-q", "-m", msg]) + print("[pgate] committed %s" % head_commit()[:9], flush=True) + + json.dump(results, open(os.path.join(REPO, ".run/pgate_results.json"), "w"), indent=1) + + +if __name__ == "__main__": + main() diff --git a/tools/twin_sweep.py b/tools/twin_sweep.py new file mode 100644 index 000000000..9e915f766 --- /dev/null +++ b/tools/twin_sweep.py @@ -0,0 +1,135 @@ +#!/usr/bin/env python3 +"""twin_sweep.py — bank every open stub that has an ALREADY-BANKED structural twin, for ~0 tokens. + +THE LEVER (P31 S65, byte-measured). For each open `INCLUDE_ASM` stub, ask whether ANY already-banked +function fleet-wide shares its signature hash. If one does, `family_remap` rewrites that exemplar's C +to this overlay's symbols mechanically and the whole-binary gate decides (G3/P9). Measured yields: + + h_exact (identical instruction bytes modulo reloc fields) : 139/157 = 88.5% + h_norm (identical after masking the reloc fields) : 36/45 = 80% (first probe) + 178/234 = 76% (parallel sweep) + h_seq (mnemonic skeleton, immediates may differ) : 0% — Phase-26 refuted it; NOT USED + +**THE POOL REFILLS.** Every bank becomes an exemplar for its own siblings, so a sweep that ends with +0 candidates is NOT the end: re-run after any wave, recovery pass, or sweep. Measured: immediately +after banking 178 functions, 165 fresh twin candidates existed that had none before. + +**RUN THIS BEFORE DRAWING A WAVE.** A drafting slot spent on a function that has a banked twin is +pure waste, and `t5_cards.py` does NOT build `seed_ref`, so cards actively assert "no banked twin — +derive from the .s" for these. Measured cost of not doing it: a t5u Opus slot ground +ov_SC03_023:func_8017BEBC to closeness 45 while ov_SC02_004 held a byte-identical banked copy. + + twin_sweep.py [--tier exact|norm|both] [--workers 10] [--plan-only] [--commit] [--r22] + +Staging is pure computation (family_remap only READS build/ + asm/), so it is safe to run while +something else holds the tree; only the gate needs `tools/parallel_gate.py` and a quiet tree. +""" +import argparse, collections, json, os, shutil, subprocess, sys +from concurrent.futures import ThreadPoolExecutor + +REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +os.chdir(REPO) +sys.path.insert(0, "tools") +PY = ".venv/bin/python" + + +def candidates(tier): + """[{to, to_addr, from, addr, same_addr, nins, tier}] — an open stub + its banked exemplar. + + Prefers a SAME-ADDRESS exemplar (the same engine fn in a sibling overlay) because the remap is + then a pure symbol substitution; a cross-address twin additionally needs --to-addr.""" + import family_sweep as FS, corpus, progress + sigs = FS.load_sigs() + open_by = {} + for b in progress.BINARIES: + try: + open_by[b] = {int(s.symbol[5:], 16) for s in corpus.stubs(b).values()} + except Exception: + pass + banked = {"exact": collections.defaultdict(list), "norm": collections.defaultdict(list)} + for b, addrs in sigs.items(): + op = open_by.get(b, set()) + for addr, (nins, hx, hn) in addrs.items(): + if addr in op: + continue + banked["exact"][hx].append((b, addr, nins)) + banked["norm"][hn].append((b, addr, nins)) + want = ("exact", "norm") if tier == "both" else (tier,) + out, seen = [], set() + for b, addrs in sigs.items(): + op = open_by.get(b, set()) + for addr, (nins, hx, hn) in addrs.items(): + if addr not in op or (b, addr) in seen: + continue + for t in want: # exact first: a stricter twin is a safer remap + pool = banked[t].get(hx if t == "exact" else hn) + if not pool: + continue + fb, fa, _ = sorted(pool, key=lambda x: (x[1] != addr, x[0]))[0] + out.append({"to": b, "to_addr": "0x%08x" % addr, "from": fb, "addr": "0x%08x" % fa, + "same_addr": fa == addr, "nins": nins, "tier": t}) + seen.add((b, addr)) + break + return out + + +def stage(item): + b, rows, root = item + d = os.path.abspath(os.path.join(root, b)) + shutil.rmtree(d, ignore_errors=True) + os.makedirs(d) + made = 0 + for r in rows: + fn = "func_%s" % r["to_addr"][2:].upper() + cmd = [PY, "tools/family_remap.py", "--addr", r["addr"], "--from", r["from"], + "--to", b, "--out", "%s/%s.c" % (d, fn)] + if not r["same_addr"]: + cmd += ["--to-addr", r["to_addr"]] + p = subprocess.run(cmd, capture_output=True, text=True) + if p.returncode == 0 and os.path.exists("%s/%s.c" % (d, fn)): + made += 1 + return b, made, len(rows), d + + +def main(): + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("--tier", default="both", choices=("exact", "norm", "both")) + ap.add_argument("--workers", type=int, default=10) + ap.add_argument("--plan-only", action="store_true") + ap.add_argument("--commit", action="store_true") + ap.add_argument("--r22", action="store_true") + ap.add_argument("--root", default=".run/twin_stage") + a = ap.parse_args() + + cands = candidates(a.tier) + by = collections.defaultdict(list) + for r in cands: + by[r["to"]].append(r) + tc = collections.Counter(r["tier"] for r in cands) + print("[twin] %d candidate(s) across %d binaries (%d ins) — %s" + % (len(cands), len(by), sum(r["nins"] for r in cands), + ", ".join("%s %d" % kv for kv in sorted(tc.items()))), flush=True) + if not cands: + print("[twin] nothing to do — re-run after the next wave/recovery bank (the pool REFILLS)") + return + + with ThreadPoolExecutor(max_workers=12) as ex: + res = list(ex.map(stage, [(b, rows, a.root) for b, rows in by.items()])) + plan = [{"binary": b, "drafts": d} for b, made, _, d in res if made] + json.dump(plan, open(".run/twin_plan.json", "w"), indent=1) + print("[twin] staged %d/%d remap(s); plan .run/twin_plan.json (%d binaries)" + % (sum(m for _, m, _, _ in res), sum(t for _, _, t, _ in res), len(plan)), flush=True) + if a.plan_only: + return + + cmd = [PY, "tools/parallel_gate.py", "--plan", ".run/twin_plan.json", + "--workers", str(a.workers)] + if a.commit: + cmd.append("--commit") + if a.r22: + cmd.append("--r22") + sys.exit(subprocess.call(cmd)) + + +if __name__ == "__main__": + main()