"""Run every adapter in adapters.REGISTRY against MALFORMED-300, once. Produces leaderboard.json. Reports two numbers per parser, because these libraries do not all share a design goal: exact_300 exact match over all 300 cases, refusals graded by the spec (an unrecoverable case passes ONLY by refusing). exact_275 exact match over the 275 RECOVERABLE cases only. This removes the refusal-policy question entirely, so a library that is designed to always return something is not penalised for its design -- it is just measured on what it recovers. invented how many of the 25 unrecoverable cases got a value back instead of a refusal. Not a bug: for some of these libraries it is the documented behaviour. It is reported because it is the thing that silently corrupts an agent's state. Nothing here is tuned. Each parser is run once, on the frozen corpus (sha256 recorded below), and the numbers are written down whatever they say. CC0-1.0. """ import hashlib import importlib import json import os import subprocess import sys import time HERE = os.path.dirname(os.path.abspath(__file__)) CORPUS = os.path.join(HERE, "..", "malformed300", "malformed300.jsonl") SCORER = os.path.join(HERE, "..", "malformed300", "score.py") sys.path.insert(0, HERE) sys.path.insert(0, os.path.join(HERE, "..", "jsonshim")) import adapters # noqa: E402 def corpus_sha256(path): h = hashlib.sha256() with open(path, "rb") as f: h.update(f.read()) return h.hexdigest() def load_cases(path): with open(path) as f: return [json.loads(l) for l in f if l.strip()] def canon(v): return json.dumps(v, sort_keys=True, separators=(",", ":"), ensure_ascii=False) def version_of(modname): if not modname: return None try: m = importlib.import_module(modname) except Exception: return None for attr in ("__version__", "VERSION", "version"): v = getattr(m, attr, None) if isinstance(v, str): return v if isinstance(v, tuple): return ".".join(str(x) for x in v) try: from importlib.metadata import version as _v return _v(modname.replace("_", "-")) except Exception: return None def preflight(registry): """PREFLIGHT: import every declared module before a single case is scored. Added after a run in which five libraries were not importable, every adapter silently refused, and all five scored exactly what the stdlib control scores. The output looked like a result. Nothing about it was. """ missing = [] for name, (_fn, modname) in registry.items(): if not modname: continue try: importlib.import_module(modname) except Exception as e: missing.append("%s (module %s): %s" % (name, modname, e)) if missing: sys.stderr.write("PREFLIGHT FAILED - not scoring anything:\n " + "\n ".join(missing) + "\n") raise SystemExit(3) print("preflight: %d declared modules import" % sum(1 for _n, (_f, m) in registry.items() if m)) def score(fn, cases): by_cat = {} exact = exact_rec = invented = false_refusal = 0 n_rec = 0 rows = [] t0 = time.time() for c in cases: refused = False got = None err = None try: got = fn(c["input"]) if got is None: refused = True except ImportError: # A missing library is a broken environment, not a refusal. Grading it # as one manufactures a plausible score (it lands on exactly the control's # number) out of a machine that never ran the library at all. Abort. raise except Exception as e: refused = True err = "%s: %s" % (type(e).__name__, str(e)[:120]) unrec = c["expected_kind"] == "unrecoverable" if unrec: ok = refused if not refused: invented += 1 else: n_rec += 1 ok = (not refused) and canon(got) == canon(c["expected"]) if refused: false_refusal += 1 if ok: exact_rec += 1 if ok: exact += 1 cat = by_cat.setdefault(c["category"], {"n": 0, "ok": 0}) cat["n"] += 1 cat["ok"] += 1 if ok else 0 rows.append({"id": c["id"], "category": c["category"], "ok": ok, "refused": refused, "error": err}) for v in by_cat.values(): v["rate"] = round(v["ok"] / v["n"], 4) return { "n": len(cases), "exact_300": exact, "rate_300": round(exact / len(cases), 4), "n_recoverable": n_rec, "exact_275": exact_rec, "rate_275": round(exact_rec / n_rec, 4) if n_rec else None, "unrecoverable_n": len(cases) - n_rec, "correctly_refused": (len(cases) - n_rec) - invented, "invented_values": invented, "false_refusals": false_refusal, "seconds": round(time.time() - t0, 2), "by_category": dict(sorted(by_cat.items())), }, rows def main(): cases = load_cases(CORPUS) sha = corpus_sha256(CORPUS) out = { "corpus": "MALFORMED-300", "corpus_sha256": sha, "cases": len(cases), "generated_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "python": sys.version.split()[0], "platform": sys.platform, "grading": "exact_300 = spec grading (unrecoverable passes only by refusing); " "exact_275 = recoverable cases only, refusal policy ignored", "results": {}, } preflight(adapters.REGISTRY) for label, (fnname, modname) in adapters.REGISTRY.items(): fn = getattr(adapters, fnname) try: res, rows = score(fn, cases) except ImportError: raise except Exception as e: out["results"][label] = {"error": "%s: %s" % (type(e).__name__, e)} print("%-28s ERROR %s" % (label, e)) continue res["version"] = version_of(modname) out["results"][label] = res print("%-28s %3d/300 (%5.1f%%) rec %3d/275 (%5.1f%%) invented %2d " "false-refusals %3d %.1fs" % (label, res["exact_300"], res["rate_300"] * 100, res["exact_275"], res["rate_275"] * 100, res["invented_values"], res["false_refusals"], res["seconds"])) with open(os.path.join(HERE, "rows_%s.jsonl" % fnname), "w") as f: for r in rows: f.write(json.dumps(r) + "\n") with open(os.path.join(HERE, "leaderboard.json"), "w") as f: json.dump(out, f, indent=2, sort_keys=False) print("\ncorpus sha256 %s" % sha) print("wrote leaderboard.json") if __name__ == "__main__": main()