| """tools/corpus_factory.py — Kenga Corpus Factory (Phase II: Data & Semantics). |
| |
| Generates synthetic, compiler-verified Kenga programs. Every program is |
| executed via kenga-lite (compile -> run -> stdout); only rc==0 programs are |
| kept. Semantic-equivalent body variants are kept only when they reproduce |
| the exact same stdout. Token-level mutations of verified programs produce |
| (broken, fixed) repair pairs labelled by failure mode. |
| |
| Categories: arith (expression functions), loop (accumulators), rec (single |
| self-recursion), chain (call chains). |
| |
| Output: JSONL manifest, one record per program: |
| {"id", "category", "src", "out", |
| "variants": [{"src", "out"}], |
| "mutants": [{"src", "mode"}]} mode: "run" | "value" |
| """ |
| import argparse |
| import json |
| import os |
| import random |
| import re |
| import sys |
| import time |
|
|
| sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) |
| import kenchat |
|
|
|
|
| |
|
|
| def gen_arith(rng): |
| """Expression functions with semantic-equivalent body variants.""" |
| npar = rng.choice([1, 1, 2, 2, 3]) |
| params = sorted(rng.sample(['a', 'b', 'c', 'd', 'e'], npar)) |
|
|
| def expr(d): |
| if d <= 0 or rng.random() < 0.35: |
| if params and rng.random() < 0.65: |
| return rng.choice(params) |
| return str(rng.randint(0, 9)) |
| op = rng.choice(['+', '-', '*', '*', '+']) |
| s = f'{expr(d - 1)} {op} {expr(d - 1)}' |
| return '(' + s + ')' if d >= 2 else s |
|
|
| sig = ', '.join(f'{p}: i64' for p in params) |
| callargs = ', '.join(str(rng.randint(0, 12)) for _ in params) |
| e = expr(rng.randint(1, 3)) |
|
|
| bodies = [f'return {e};', |
| f'let c = {e};\n return c;'] |
| |
| |
| flat = e.strip() |
| if len(flat) > 4 and flat[0] == '(' and flat[-1] == ')': |
| inner = flat[1:-1] |
| depth = 0 |
| for i, ch in enumerate(inner): |
| if ch == '(': |
| depth += 1 |
| elif ch == ')': |
| depth -= 1 |
| elif depth == 0 and ch in '+-*' and i > 0 and inner[i - 1] == ' ' \ |
| and i + 1 < len(inner) and inner[i + 1] == ' ': |
| l, r = inner[:i].strip(), inner[i + 1:].strip() |
| if l.count('(') == l.count(')') and r.count('(') == r.count(')'): |
| op = ch |
| bodies.append(f'let t = {l};\n t = t {op} {r};\n return t;') |
| break |
| rng.shuffle(bodies) |
|
|
| def mk(body): |
| return (f'fn calc({sig}) -> i64 {{\n {body}\n}}\n' |
| f'fn main() -> i64 {{\n' |
| f' let r = calc({callargs});\n println(r);\n return 0;\n}}\n') |
|
|
| return mk(bodies[0]), [mk(v) for v in bodies[1:3]] |
|
|
|
|
| |
|
|
| def gen_loop(rng): |
| kind = rng.choice(['sum_range', 'count_range', 'prod_range']) |
| if kind == 'sum_range': |
| acc = rng.choice(['i', 'i', '2 * i', 'b - i']) |
| step = rng.choice([1, 1, 2]) |
| body = (f'let s = 0;\n let i = a;\n' |
| f' while i <= b {{\n s = s + {acc};\n i = i + {step};\n }}\n' |
| f' return s;') |
| var = (f'let s = 0;\n let i = a;\n while i <= b {{\n' |
| f' let t = {acc};\n s = s + t;\n i = i + {step};\n }}\n return s;') |
| a, b = rng.randint(0, 14), rng.randint(0, 14) |
| elif kind == 'count_range': |
| acc = rng.choice(['i', 'i', 'a + b - i']) |
| step = rng.choice([1, 1, 2]) |
| body = (f'let s = 0;\n let i = b;\n' |
| f' while i >= a {{\n s = s + {acc};\n i = i - {step};\n }}\n' |
| f' return s;') |
| var = (f'let s = 0;\n let i = b;\n while i >= a {{\n' |
| f' let t = {acc};\n s = s + t;\n i = i - {step};\n }}\n return s;') |
| a, b = rng.randint(0, 14), rng.randint(0, 14) |
| else: |
| a = rng.randint(0, 6) |
| b = a + rng.randint(0, 6) |
| body = (f'let p = 1;\n let i = a;\n' |
| f' while i <= b {{\n p = p * i;\n i = i + 1;\n }}\n' |
| f' return p;') |
| var = None |
|
|
| def mk(fnbody): |
| return (f'fn run(a: i64, b: i64) -> i64 {{\n {fnbody}\n}}\n' |
| f'fn main() -> i64 {{\n' |
| f' let r = run({a}, {b});\n println(r);\n return 0;\n}}\n') |
|
|
| variants = [mk(var)] if var else [] |
| return mk(body), variants |
|
|
|
|
| |
|
|
| def gen_rec(rng): |
| shape = rng.choice(['fact', 'sumrec', 'halve', 'fib', 'sumrec1', 'halve1']) |
| if shape == 'fact': |
| base_cond, base_ret, op, dec = 1, 1, '*', 1 |
| n = rng.randint(0, 8) |
| it = (f'let p = 1;\n let i = 2;\n' |
| f' while i <= n {{\n p = p * i;\n i = i + 1;\n }}\n' |
| f' return p;') |
| elif shape == 'sumrec': |
| base_cond, base_ret, op, dec = 1, 0, '+', 1 |
| n = rng.randint(0, 20) |
| it = (f'let s = 0;\n let i = 1;\n' |
| f' while i <= n {{\n s = s + i;\n i = i + 1;\n }}\n' |
| f' return s;') |
| elif shape == 'sumrec1': |
| base_cond, base_ret, op, dec = 1, 1, '+', 1 |
| n = rng.randint(0, 20) |
| it = (f'let s = 0;\n let i = 2;\n' |
| f' while i <= n {{\n s = s + i;\n i = i + 1;\n }}\n' |
| f' return s + 1;') |
| elif shape == 'halve': |
| base_cond, base_ret, op, dec = 2, 'n', '+', 2 |
| n = rng.randint(0, 16) |
| it = (f'let s = 0;\n let i = n;\n' |
| f' while i > 0 {{\n s = s + i;\n i = i - 2;\n }}\n' |
| f' return s;') |
| elif shape == 'halve1': |
| base_cond, base_ret, op, dec = 2, 1, '+', 2 |
| n = rng.randint(0, 16) |
| it = (f'let s = 0;\n let i = n;\n' |
| f' while i > 1 {{\n s = s + i;\n i = i - 2;\n }}\n' |
| f' return s + 1;') |
| else: |
| |
| n = rng.randint(0, 16) |
| rec_fn = (f'fn deep(n: i64) -> i64 {{\n' |
| f' if n < 2 {{ return n; }}\n' |
| f' return deep(n - 1) + deep(n - 2);\n}}') |
| it_fn = (f'fn run(n: i64) -> i64 {{\n' |
| f' let a = 0;\n let b = 1;\n let i = 0;\n' |
| f' while i < n {{\n let t = a + b;\n' |
| f' a = b;\n b = t;\n i = i + 1;\n }}\n' |
| f' return a;\n}}') |
| return (rec_fn + '\n' + it_fn + '\n' |
| f'fn main() -> i64 {{\n let r = deep({n});\n' |
| f' println(r);\n return 0;\n}}\n', |
| [rec_fn + '\n' + it_fn + '\n' |
| f'fn main() -> i64 {{\n let r = run({n});\n' |
| f' println(r);\n return 0;\n}}\n']) |
|
|
| rec_fn = (f'fn deep(n: i64) -> i64 {{\n' |
| f' if n < {base_cond} {{ return {base_ret}; }}\n' |
| f' return n {op} deep(n - {dec});\n}}') |
| it_fn = f'fn run(n: i64) -> i64 {{\n {it}\n}}' |
|
|
| def mk(caller_body): |
| return (rec_fn + '\n' + it_fn + '\n' |
| f'fn main() -> i64 {{\n' |
| f' {caller_body}\n return 0;\n}}\n') |
|
|
| primary = mk(f'let r = deep({n});\n println(r);') |
| variant = mk(f'let r = run({n});\n println(r);') |
| return primary, [variant] |
|
|
|
|
| |
|
|
| def gen_chain(rng): |
| depth = rng.randint(2, 4) |
| consts = [rng.randint(0, 9) for _ in range(depth)] |
| lines = [] |
| prev = 'x' |
| for i in range(depth): |
| fn = f'f{i}' |
| if i == 0: |
| body = f'return x + {consts[0]};' |
| else: |
| op = rng.choice(['+', '*']) |
| rhs = str(consts[i]) if op == '+' else rng.choice(['2', '3']) |
| body = f'return {prev} {op} {rhs};' |
| lines.append(f'fn {fn}(x: i64) -> i64 {{\n {body}\n}}') |
| prev = fn |
| arg = rng.randint(0, 9) |
| lines.append(f'fn main() -> i64 {{\n let r = {prev}({arg});\n' |
| f' println(r);\n return 0;\n}}') |
| return '\n'.join(lines) + '\n', [] |
|
|
|
|
| |
|
|
| OPFLIP = {' + ': ' - ', ' - ': ' + ', ' * ': ' + '} |
|
|
|
|
| def mutate(src, rng): |
| """Return a broken copy of src, or None if nothing applicable.""" |
| kinds = [] |
| if any(o in src for o in OPFLIP): |
| kinds.append('opflip') |
| if re.search(r'\b\d+\b', src): |
| kinds.append('bump') |
| if ')' in src: |
| kinds.append('droppar') |
| m = re.search(r'fn (\w+)\(([^)]*)\)', src) |
| if m and ',' in m.group(2): |
| kinds.append('swapparams') |
| if not kinds: |
| return None |
| k = rng.choice(kinds) |
| if k == 'opflip': |
| opts = [o for o in OPFLIP if o in src] |
| o = rng.choice(opts) |
| return src.replace(o, OPFLIP[o], 1) |
| if k == 'bump': |
| nums = list(re.finditer(r'\b\d+\b', src)) |
| mm = rng.choice(nums) |
| v = int(mm.group(0)) |
| nv = v + 1 if rng.random() < 0.5 else max(0, v - 1) |
| return src[:mm.start()] + str(nv) + src[mm.end():] |
| if k == 'droppar': |
| idxs = [i for i, ch in enumerate(src) if ch == ')'] |
| i = rng.choice(idxs) |
| return src[:i] + src[i + 1:] |
| mm = re.search(r'fn (\w+)\(([^)]*)\)', src) |
| parts = [p.strip() for p in mm.group(2).split(',')] |
| i, j = rng.sample(range(len(parts)), 2) |
| ni, nj = parts[i].split(':')[0], parts[j].split(':')[0] |
| parts[i] = parts[i].replace(ni, nj) |
| parts[j] = parts[j].replace(nj, ni) |
| return src[:mm.start(2)] + ', '.join(parts) + src[mm.end(2):] |
|
|
|
|
| |
|
|
| GENS = {'arith': gen_arith, 'loop': gen_loop, 'rec': gen_rec, 'chain': gen_chain} |
|
|
|
|
| def main(): |
| ap = argparse.ArgumentParser() |
| ap.add_argument('--arith', type=int, default=150) |
| ap.add_argument('--loop', type=int, default=100) |
| ap.add_argument('--rec', type=int, default=80) |
| ap.add_argument('--chain', type=int, default=80) |
| ap.add_argument('--max-mutants', type=int, default=1) |
| ap.add_argument('--seed', type=int, default=13) |
| ap.add_argument('--out', default='minds/corpus_factory/manifest.jsonl') |
| args = ap.parse_args() |
|
|
| rng = random.Random(args.seed) |
| os.makedirs(os.path.dirname(args.out), exist_ok=True) |
|
|
| counts = {'arith': args.arith, 'loop': args.loop, |
| 'rec': args.rec, 'chain': args.chain} |
| seen = set() |
| stats = {'kept': 0, 'dropped_primary': 0, 'variants_kept': 0, |
| 'variants_dropped': 0, 'mut_run': 0, 'mut_value': 0, |
| 'mut_neutral': 0} |
| run_times = [] |
|
|
| t_start = time.time() |
| n_lines = 0 |
| with open(args.out, 'w', encoding='utf-8') as f: |
| for cat, want in counts.items(): |
| made = 0 |
| attempts = 0 |
| while made < want and attempts < want * 6: |
| attempts += 1 |
| primary, variants = GENS[cat](rng) |
| if primary in seen: |
| continue |
|
|
| t0 = time.time() |
| rc, out, err = kenchat.run_via_kenga_lite(primary, timeout=10) |
| run_times.append(time.time() - t0) |
| if rc != 0 or not out.strip(): |
| stats['dropped_primary'] += 1 |
| continue |
| seen.add(primary) |
| rec = {'id': f'{cat}_{made:05d}', 'category': cat, |
| 'src': primary, 'out': out.strip(), |
| 'variants': [], 'mutants': []} |
|
|
| for vsrc in variants: |
| t0 = time.time() |
| vrc, vout, _ = kenchat.run_via_kenga_lite(vsrc, timeout=10) |
| run_times.append(time.time() - t0) |
| if vrc == 0 and vout.strip() == rec['out']: |
| rec['variants'].append({'src': vsrc, 'out': vout.strip()}) |
| stats['variants_kept'] += 1 |
| else: |
| stats['variants_dropped'] += 1 |
|
|
| for _ in range(args.max_mutants): |
| msrc = mutate(primary, rng) |
| if msrc is None: |
| continue |
| t0 = time.time() |
| |
| mrc, mout, _ = kenchat.run_via_kenga_lite(msrc, timeout=4) |
| run_times.append(time.time() - t0) |
| if mrc != 0: |
| rec['mutants'].append({'src': msrc, 'mode': 'run'}) |
| stats['mut_run'] += 1 |
| elif mout.strip() != rec['out']: |
| rec['mutants'].append({'src': msrc, 'mode': 'value'}) |
| stats['mut_value'] += 1 |
| else: |
| stats['mut_neutral'] += 1 |
|
|
| f.write(json.dumps(rec) + '\n') |
| n_lines += 1 |
| made += 1 |
| stats['kept'] += 1 |
| if made % 50 == 0: |
| print(f' [{cat}] {made}/{want} kept, ' |
| f'{len(run_times)} runs, {time.time()-t_start:.0f}s', |
| flush=True) |
|
|
| wall = time.time() - t_start |
| avg_ms = 1000 * sum(run_times) / max(1, len(run_times)) |
| print(f'manifest: {args.out}') |
| print(f'programs kept: {stats["kept"]} (dropped primary: {stats["dropped_primary"]})') |
| print(f'per category (made/requested): ' + |
| ', '.join(f'{c}={counts[c]}' for c in counts)) |
| print(f'variants: kept {stats["variants_kept"]}, dropped {stats["variants_dropped"]}') |
| print(f'mutants: run-fail {stats["mut_run"]}, wrong-value {stats["mut_value"]}, ' |
| f'neutral-discarded {stats["mut_neutral"]}') |
| print(f'kenga-lite runs: {len(run_times)}, avg {avg_ms:.0f} ms, wall {wall:.0f}s') |
| print(f'estimated 100k programs: {avg_ms * 400000 / 1000 / 3600:.1f} h ' |
| f'(4 runs per program incl variants+mutants)') |
| return 0 |
|
|
|
|
| if __name__ == '__main__': |
| sys.exit(main()) |
|
|
