igneum/tools/ci/p01-vectors.py

#!/usr/bin/env python3
"""P01 part A, the million-vector campaign per backend (the founder's Test and Acceptance Standard, POW-01; 8 October 2026).

The GPU side is the real worker (the kit's CUDA, OpenCL or Metal binary) driven as the miner drives it: `job` lines on stdin
with an all-pass share target, so every nonce comes back as `found <job> <nonce> <hash>`; the CPU side is the reference list
the hash lane writes with `igneum-pow hash-bound --count N` (one `<nonce> <hash>` per line, hex or decimal nonce, 16-hex hash).
Every found hash is compared to the reference's hash for that nonce; the evidence JSON carries the counts, the first ten
disagreements, the pack id, the program id, the device line and the manifest sha; exit 1 on any disagreement or any nonce
the worker never answered.

  tools/ci/p01-vectors.py --worker <bin> [--worker-arg X ...] --pack <dir> --reference <file> --count 1000000
                          [--start 0] [--job-nonces 1048576] [--prehash <64 hex>] [--manifest <sha>] --out <evidence.json>
  tools/ci/p01-vectors.py --self-test
The job line is pool.rs's: `job <seq> <prehash hex> <share_target64 hex16> <start nonce> <nonces> <epoch seed bytes hex> <day bytes hex> class=<c> era=<era hex>`.
"""
import argparse, json, os, subprocess, sys, tempfile, time

def read_reference(path):
    ref = {}
    with open(path) as f:
        for line in f:
            p = line.split()
            if len(p) < 2 or p[0].startswith('#'): continue
            n = int(p[0], 16) if p[0].lower().startswith('0x') else int(p[0]); h = p[1].lower().replace('0x', '')
            ref[n] = h.zfill(16)
    return ref

def pack_fields(pack):
    j = json.load(open(os.path.join(pack, 'program.json')))
    cls = j.get('program_class', '?'); era = j.get('era_seed_bytes', '')
    return j.get('seed_bytes', ''), j['dataset']['day_bytes'], cls, era, j.get('program_id', '?'), j.get('generator', '?')

def run(a):
    ref = read_reference(a.reference)
    seed_bytes, day_bytes, cls, era, program_id, generator = pack_fields(a.pack)
    p = subprocess.Popen([a.worker] + a.worker_arg, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1)
    ready = None; t0 = time.time()
    for line in p.stdout:
        if line.startswith('ready'): ready = line.strip(); break
        if time.time() - t0 > 120: break
    if not ready: p.kill(); return {'error': 'the worker never said ready'}, 2
    got = {}; agree = 0; disagree = []; seq = 0; start = a.start; end = a.start + a.count; pending = set()
    def feed():
        nonlocal seq, start
        while start < end and len(pending) < 4:
            n = min(a.job_nonces, end - start); seq += 1
            p.stdin.write(f"job {seq} {a.prehash} {'f'*16} {start} {n} {seed_bytes} {day_bytes} class={cls} era={era}\n"); p.stdin.flush()
            pending.add(seq); start += n
    feed()
    for line in p.stdout:
        it = line.split()
        if not it: continue
        if it[0] == 'found' and len(it) >= 4:
            n = int(it[2]); h = it[3].lower().replace('0x', '').zfill(16); got[n] = h
            r = ref.get(n)
            if r == h: agree += 1
            else:
                if len(disagree) < 10: disagree.append({'nonce': n, 'gpu': h, 'cpu': r})
                disagree_count[0] += 1
        elif it[0] in ('done', 'error') and len(it) >= 2:
            try: pending.discard(int(it[1]))
            except ValueError: pass
            if it[0] == 'error': errors.append(line.strip()[:200])
            if start >= end and not pending: break
            feed()
    try: p.stdin.close()
    except Exception: pass
    p.wait(timeout=30)
    missing = [n for n in range(a.start, end) if n not in got]
    ev = {'case': 'POW-01', 'profile': 'P01', 'pack': os.path.basename(os.path.abspath(a.pack)), 'program_id': program_id, 'generator': generator, 'class': cls,
          'worker': a.worker, 'worker_args': a.worker_arg, 'device_line': ready, 'manifest_sha': a.manifest, 'prehash': a.prehash,
          'nonces': {'start': a.start, 'count': a.count}, 'answered': len(got), 'agree': agree, 'disagree': disagree_count[0], 'missing': len(missing),
          'first_disagreements': disagree, 'first_missing': missing[:10], 'worker_errors': errors[:10], 'seconds': round(time.time() - t0, 1), 'at': time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())}
    ev['verdict'] = 'PASS' if (agree == a.count and not disagree and not missing) else 'FAIL'
    return ev, (0 if ev['verdict'] == 'PASS' else 1)

disagree_count = [0]; errors = []

def self_test():
    d = tempfile.mkdtemp(); fails = 0
    pack = os.path.join(d, 'pack'); os.makedirs(pack)
    json.dump({'seed_bytes': 'aa' * 32, 'program_class': 'v5', 'era_seed_bytes': 'bb' * 32, 'program_id': '0x1', 'generator': 5, 'dataset': {'day_bytes': 'cc' * 19, 'log2_words': 20}}, open(os.path.join(pack, 'program.json'), 'w'))
    # a fake worker: hashes nonce n as (n * 0x9e3779b97f4a7c15) mod 2^64; nonce 7 wrong when WRONG=1; nonce 9 never answered when DROP=1
    w = os.path.join(d, 'worker.py')
    open(w, 'w').write('''import sys, os
print("ready fake-gpu 0 prepare 1", flush=True)
for line in sys.stdin:
    p = line.split()
    if p[0] != "job": continue
    seq, start, n = int(p[1]), int(p[4]), int(p[5])
    for k in range(start, start + n):
        if os.environ.get("DROP") == "1" and k == 9: continue
        h = (k * 0x9e3779b97f4a7c15) % (1 << 64)
        if os.environ.get("WRONG") == "1" and k == 7: h ^= 1
        print(f"found {seq} {k} {h:016x}", flush=True)
    print(f"done {seq} {n} 1.0", flush=True)
''')
    ref = os.path.join(d, 'ref.txt'); open(ref, 'w').write(''.join(f"{k} {(k * 0x9e3779b97f4a7c15) % (1 << 64):016x}\n" for k in range(0, 20)))
    base = ['--worker', sys.executable, '--worker-arg', w, '--pack', pack, '--reference', ref, '--count', '20', '--job-nonces', '8', '--manifest', 'deadbeef']
    def go(env, tag):
        out = os.path.join(d, f'{tag}.json'); e = dict(os.environ); e.update(env)
        r = subprocess.run([sys.executable, __file__] + base + ['--out', out], env=e, capture_output=True, text=True); return r.returncode, json.load(open(out)) if os.path.exists(out) else {}
    rc, ev = go({}, 'ok')
    if not (rc == 0 and ev.get('verdict') == 'PASS' and ev['agree'] == 20 and ev['disagree'] == 0 and ev['missing'] == 0): print(f"self-test failed: a clean run was not PASS: rc={rc} {ev}"); fails = 1
    rc, ev = go({'WRONG': '1'}, 'wrong')
    if not (rc == 1 and ev.get('verdict') == 'FAIL' and ev['disagree'] == 1 and ev['first_disagreements'][0]['nonce'] == 7): print(f"self-test failed: one wrong hash was not a FAIL naming nonce 7: rc={rc} {ev.get('first_disagreements')}"); fails = 1
    rc, ev = go({'DROP': '1'}, 'drop')
    if not (rc == 1 and ev['missing'] == 1 and ev['first_missing'] == [9]): print(f"self-test failed: an unanswered nonce was not a FAIL naming nonce 9: rc={rc} {ev.get('first_missing')}"); fails = 1
    if not fails: print('self-test passed: a clean million-shape run is PASS with the counts; one wrong hash is FAIL naming the nonce, the gpu and cpu hashes; an unanswered nonce is FAIL naming it; the job lines carry the pack fields and the all-pass target')
    return fails

if __name__ == '__main__':
    if '--self-test' in sys.argv: sys.exit(self_test())
    ap = argparse.ArgumentParser(); ap.add_argument('--worker', required=True); ap.add_argument('--worker-arg', action='append', default=[]); ap.add_argument('--pack', required=True)
    ap.add_argument('--reference', required=True); ap.add_argument('--count', type=int, default=1_000_000); ap.add_argument('--start', type=int, default=0); ap.add_argument('--job-nonces', type=int, default=1 << 20)
    ap.add_argument('--prehash', default='00' * 31 + '01'); ap.add_argument('--manifest', default=''); ap.add_argument('--out', required=True)
    a = ap.parse_args(); ev, rc = run(a)
    os.makedirs(os.path.dirname(os.path.abspath(a.out)), exist_ok=True); json.dump(ev, open(a.out, 'w'), indent=2)
    print(f"p01-vectors: {ev.get('verdict', 'ERROR')}: answered {ev.get('answered')} agree {ev.get('agree')} disagree {ev.get('disagree')} missing {ev.get('missing')} in {ev.get('seconds')} s -> {a.out}")
    sys.exit(rc)