1"""Run a prepared jobs file through the Rust engine (optionally K engine processes in parallel, each job single-threaded)2and record provenance on the executing host. Stdlib only, Python ≥ 3.9.34Usage: python3 run_batch_remote.py --engine ./rho_engine --jobs study.jobs.json --out study_<host>.json [--parallel K] [--params params.json]5"""6import argparse, json, pathlib, subprocess, sys, time, os7sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[2] / "src"))8from wp34.provenance import record91011def main():12 ap = argparse.ArgumentParser()13 ap.add_argument("--engine", required=True); ap.add_argument("--jobs", required=True); ap.add_argument("--out", required=True)14 ap.add_argument("--params", default=None); ap.add_argument("--parallel", type=int, default=1); ap.add_argument("--seed", type=int, default=34)15 a = ap.parse_args(); t0 = time.time()16 jobs = json.loads(pathlib.Path(a.jobs).read_text())17 out = pathlib.Path(a.out); out.parent.mkdir(parents=True, exist_ok=True)18 K = max(1, a.parallel)19 chunks = [jobs[i::K] for i in range(K)]20 procs, files = [], []21 for i, ch in enumerate(chunks):22 if not ch:23 continue24 jf = out.with_suffix(f".chunk{i}.jobs.json"); rf = out.with_suffix(f".chunk{i}.raw.json")25 jf.write_text(json.dumps(ch)); files.append((i, rf))26 procs.append(subprocess.Popen([a.engine, "batch", "--file", str(jf), "--out", str(rf)], stdout=subprocess.DEVNULL, stderr=open(out.with_suffix(f".chunk{i}.err"), "w")))27 for p in procs:28 p.wait()29 # merge in original order30 merged = [None] * len(jobs)31 for i, rf in files:32 res = json.loads(rf.read_text())33 for k, r in enumerate(res):34 merged[i + k * K] = r35 ok = sum(1 for r in merged if r and r.get("ok"))36 params = json.loads(pathlib.Path(a.params).read_text()) if a.params else {}37 params.update({"engine": a.engine, "jobs_file": a.jobs, "parallel_processes": K})38 record("ecdlp_study", params, {"n_jobs": len(jobs), "n_ok": ok, "runs": merged}, seed=a.seed, out=str(out), started=t0)39 for i, rf in files:40 rf.unlink(); out.with_suffix(f".chunk{i}.jobs.json").unlink()41 print(f"done: {ok}/{len(jobs)} verified in {time.time()-t0:.0f}s on {os.uname().nodename} → {out}")424344if __name__ == "__main__":45 main()46