gen.py

evidence/breadth-13/gen.py
#!/usr/bin/env python3
"""PBT generator for docs/spec-fabric-module.md:147-169 (fabric.module body whitelist).

Property under test (source obligation, doc lines 147-169):
  A fabric.module body may contain ONLY fabric.pe, fabric.switch, fabric.mem,
  fabric.fifo, nested named fabric.module, fabric.instantiate, fabric.boundary,
  and the fabric.yield terminator.  Anything else -- explicitly including
  builtin.unrealized_conversion_cast -- must be rejected.

Generation strategy: build a syntactically valid top-level `module { fabric.module ... }`
whose body is a random sequence of whitelisted body ops (positive case) or the same
plus one randomly placed non-whitelisted op (negative case).

Sampling bounds below (body op counts, bit widths, fifo depths, number of seeds)
are test conventions, NOT source requirements.
"""
import json
import os
import random
import sys

# --- whitelisted body-op templates, each verified to parse+verify standalone ---
# Each builder returns (body_lines, produced_ssa_values, extra_top_level_lines)


def mk_fifo(rng, idx, args):
    src = take(rng, args, 1)
    if not src:
        return None
    src = src[0]
    depth = rng.choice([1, 2, 4, 8, 16])
    byp = rng.choice(["true", "false"])
    v = f"%f{idx}"
    line = (f"    {v} = fabric.fifo {src[0]} [max_depth = {depth}, "
            f"bypassable = {byp}] : {src[1]}")
    return [line], [(v, src[1])], []


def mk_switch(rng, idx, args):
    if not args:
        return None
    ty = rng.choice(args)[1]
    same = [a for a in args if a[1] == ty]  # one uniform port type per switch
    n = min(len(same), rng.choice([1, 2, 2, 3]))
    ins = same[:n]
    for a in ins:
        args.remove(a)
    nout = rng.choice([1, 2])
    table = ["1" * n for _ in range(nout)]
    v = f"%s{idx}"
    names = ", ".join(a[0] for a in ins)
    intys = ", ".join(a[1] for a in ins)
    outtys = ", ".join([ty] * nout)
    res = f"{v}:{nout}" if nout > 1 else v
    line = (f"    {res} = fabric.switch [{rng.choice(['spatial'])}] {names} "
            f"[{{connectivity_table = {json.dumps(table)}}}] "
            f": ({intys}) -> ({outtys})")
    vals = [(f"{v}#{i}", ty) for i in range(nout)] if nout > 1 else [(v, ty)]
    return [line], vals, []


def mk_boundary(rng, idx, args):
    picked = take(rng, args, 2)
    if not picked:
        return None
    data, tag = picked
    w = int(data[1].split("<")[1].split(">")[0])
    tw = int(tag[1].split("<")[1].split(">")[0])
    v = f"%b{idx}"
    line = (f"    {v} = fabric.boundary [s2t] {data[0]}, {tag[0]} "
            f": ({data[1]}, {tag[1]}) -> (!fabric.bits_tag<{w}, {tw}>)")
    return [line], [(v, f"!fabric.bits_tag<{w}, {tw}>")], []


def mk_nested_module(rng, idx, args):
    name = f"@nested_{idx}"
    lines = [f"    fabric.module {name}() {{", "    }"]
    return lines, [], []


def take(rng, args, k):
    """Consume k distinct unused entry-block arguments (single-consumer rule)."""
    if len(args) < k:
        return None
    picked = rng.sample(args, k)
    for a in picked:
        args.remove(a)
    return picked


WHITELIST_BUILDERS = [
    ("fabric.fifo", mk_fifo),
    ("fabric.switch", mk_switch),
    ("fabric.boundary", mk_boundary),
    ("fabric.module(nested)", mk_nested_module),
]

# --- non-whitelisted ops (must be rejected).  All from dialects registered in the
# pinned binary so that a parse error cannot masquerade as the whitelist verdict. ---
FORBIDDEN = [
    ('builtin.unrealized_conversion_cast',
     '    "builtin.unrealized_conversion_cast"() : () -> ()'),
    ('arith.constant', '    %bad{i} = arith.constant 7 : i32'),
    ('arith.addi', '    %bad{i} = arith.addi %badc{i}, %badc{i} : i32'),
    ('memref.alloc', '    %bad{i} = memref.alloc() : memref<8xi32>'),
    ('func.call', '    func.call @absent_callee() : () -> ()'),
    ('arith.muli', '    %bad{i} = arith.muli %badc{i}, %badc{i} : i32'),
    ('func.func', '    func.func @nested_f{i}() {{\n      return\n    }}'),
    ('fabric.system', '    fabric.system @sys{i} {{\n    }}'),
]


def gen_case(seed):
    rng = random.Random(seed)
    kind = "negative" if rng.random() < 0.5 else "positive"
    nargs = rng.randint(1, 4)
    args = []
    for i in range(nargs):
        w = rng.choice([1, 4, 8, 16, 32, 64])
        args.append((f"%a{i}", f"!fabric.bits<{w}>"))
    sig = ", ".join(f"{n} : {t}" for n, t in args)
    nbody = rng.randint(0, 3)
    body, produced, whitelisted_used = [], [], []
    avail = list(args)
    for i in range(nbody):
        name, builder = rng.choice(WHITELIST_BUILDERS)
        got = builder(rng, i, avail)
        if got is None:
            continue
        lines, vals, _ = got
        body.extend(lines)
        produced.extend(vals)
        whitelisted_used.append(name)
    forbidden_name = None
    if kind == "negative":
        forbidden_name, tmpl = rng.choice(FORBIDDEN)
        snippet = tmpl.format(i=seed)
        pos = rng.randint(0, len(body))
        extra = []
        if forbidden_name in ("arith.addi", "arith.muli"):
            extra = [f"    %badc{seed} = arith.constant 3 : i32"]
        body = body[:pos] + extra + [snippet] + body[pos:]
    # yield: forward a produced value if any, else empty
    if produced and rng.random() < 0.6:
        v, t = produced[-1]
        outs = f" -> ({t})"
        body.append(f"    fabric.yield {v} : {t}")
    else:
        outs = ""
    text = ("module {\n"
            f"  fabric.module @m{seed}({sig}){outs} {{\n"
            + "\n".join(body) + ("\n" if body else "")
            + "  }\n}\n")
    return {
        "seed": seed,
        "kind": kind,
        "expect": "reject" if kind == "negative" else "accept",
        "forbidden_op": forbidden_name,
        "whitelisted_ops": whitelisted_used,
        "text": text,
    }


def main():
    n = int(sys.argv[1]) if len(sys.argv) > 1 else 40
    outdir = sys.argv[2] if len(sys.argv) > 2 else "cases"
    os.makedirs(outdir, exist_ok=True)
    manifest = []
    for seed in range(1, n + 1):
        c = gen_case(seed)
        p = os.path.join(outdir, f"seed_{seed:04d}.mlir")
        with open(p, "w") as fh:
            fh.write(c["text"])
        rec = {k: v for k, v in c.items() if k != "text"}
        rec["path"] = p
        manifest.append(rec)
    with open(os.path.join(outdir, "manifest.json"), "w") as fh:
        json.dump(manifest, fh, indent=1)
    print(f"wrote {len(manifest)} cases to {outdir}")


if __name__ == "__main__":
    main()