Download tools/outside_eval.py from Lawless1987/covenant: direct link, hf CLI and curl.
- Browser
- Download file 25.8 kB
-
https://huggingface.co/Lawless1987/covenant/resolve/main/tools/outside_eval.py
- Command line
-
hf download hf://Lawless1987/covenant/tools/outside_eval.py
-
curl -L -o outside_eval.py https://huggingface.co/Lawless1987/covenant/resolve/main/tools/outside_eval.py
25.8 kB
| #!/usr/bin/env python3 | |
| """tools/outside_eval.py -- the deployed gate, measured on labels nobody in this project wrote. | |
| WHY (A327, 2026-10-10). The promoted student scores 2,751 of 2,885 on a k-fold over the project's own | |
| ledger, while a random 24 of its live forum convictions were all benign (A322). Internal labels cannot | |
| settle which number is right. This runs the node's own gate on published, human-labelled sets and | |
| reports BOTH error rates, because a gate that holds everything also "fails closed". The plan, fixed | |
| before any data was run, is docs/OUTSIDE_EVAL.md. | |
| NOTHING AN OUTSIDE ITEM PRODUCES MAY REACH A TRAINING LEDGER. The deferring judge appends verdicts to | |
| ops/verdicts.jsonl and ops/verdicts_live.jsonl (nightly training reads both) and ops/judged_by_student.jsonl. | |
| During a run those paths are rebound to a temporary directory, and the outside texts found in the real | |
| files are counted before and after: if the counts differ the run is VOID and says so. | |
| python tools/outside_eval.py convert xstest <xstest_prompts.csv> -> eval/outside/xstest.jsonl | |
| python tools/outside_eval.py convert ethics-cm <test.csv> -> eval/outside/ethics_cm_short.jsonl | |
| python tools/outside_eval.py run [SET.jsonl ...] [--out DIR] -> DIR/outside_eval_results.jsonl, DIR/summary.json | |
| """ | |
| import csv | |
| import glob | |
| import hashlib | |
| import json | |
| import math | |
| import os | |
| import sys | |
| import shutil | |
| import tempfile | |
| import time | |
| HERE = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) | |
| # Run as `python tools/outside_eval.py`, sys.path[0] is tools/, and the gate's modules live at the root. | |
| # The first real run died on this import while OE1 passed, because the test put the root on the path itself | |
| # (A327 addendum); OE1.15 now runs the command line as a subprocess. | |
| if HERE not in sys.path: | |
| sys.path.insert(0, HERE) | |
| EVAL_DIR = os.path.join(HERE, "eval", "outside") | |
| OPS = os.path.join(HERE, "ops") | |
| LEDGERS = [os.path.join(OPS, "verdicts.jsonl"), os.path.join(OPS, "verdicts_live.jsonl"), | |
| os.path.join(OPS, "judged_by_student.jsonl")] | |
| SEATS = {"Ora": "fallback_model.json", "Sena": "fallback_model_2.json"} | |
| RESULTS_NAME = "outside_eval_results.jsonl" # classified in verify_bundle.INPUT_LEDGERS (A255): published, never rewritten | |
| TEXT_CAP = 4000 # record_verdict keeps text[:4000]; a leak is matched on the same prefix | |
| SOURCES = { | |
| "xstest": {"name": "XSTest (Rottger et al., NAACL 2024)", "license": "CC-BY-4.0", | |
| "labelled_by": "the paper's authors", "url": "https://github.com/paul-rottger/xstest", | |
| "out": "xstest.jsonl"}, | |
| "ethics-cm": {"name": "ETHICS commonsense, test split, short scenarios (Hendrycks et al., ICLR 2021)", | |
| "license": "MIT", "labelled_by": "crowd workers, per the paper", | |
| "url": "https://huggingface.co/datasets/hendrycks/ethics", "out": "ethics_cm_short.jsonl"}, | |
| } | |
| def sha256_file(path): | |
| h = hashlib.sha256() | |
| with open(path, "rb") as fh: | |
| for block in iter(lambda: fh.read(1 << 16), b""): | |
| h.update(block) | |
| return h.hexdigest() | |
| def _row(kind, sid, text, expect, extra, digest): | |
| s = SOURCES[kind] | |
| return dict({"id": "%s:%s" % (kind, sid), "source": s["name"], "source_id": str(sid), "license": s["license"], | |
| "labelled_by": s["labelled_by"], "source_sha256": digest, "text": text, | |
| "expect_violates": bool(expect)}, **extra) | |
| def convert_xstest(path): | |
| """XSTest: label 'safe' -> expect clean, 'unsafe' -> expect violates; the type column is kept.""" | |
| digest = sha256_file(path) | |
| rows = [] | |
| with open(path, encoding="utf-8", newline="") as fh: | |
| for r in csv.DictReader(fh): | |
| label = (r.get("label") or "").strip().lower() | |
| if label not in ("safe", "unsafe") or not (r.get("prompt") or "").strip(): | |
| raise SystemExit("xstest row %r has label %r -- the mapping in docs/OUTSIDE_EVAL.md does not cover it" | |
| % (r.get("id"), label)) | |
| rows.append(_row("xstest", r.get("id") or len(rows), r["prompt"].strip(), label == "unsafe", | |
| {"outside_label": label, "type": (r.get("type") or "").strip()}, digest)) | |
| return rows | |
| def convert_ethics_cm(path): | |
| """ETHICS commonsense: label 1 ('clearly wrong') -> expect violates, 0 -> expect clean; short items only.""" | |
| digest = sha256_file(path) | |
| rows = [] | |
| with open(path, encoding="utf-8", newline="") as fh: | |
| for i, r in enumerate(csv.DictReader(fh)): | |
| if str(r.get("is_short", "")).strip().lower() not in ("true", "1"): | |
| continue | |
| label = str(r.get("label", "")).strip() | |
| if label not in ("0", "1") or not (r.get("input") or "").strip(): | |
| raise SystemExit("ethics row %d has label %r -- the mapping does not cover it" % (i, label)) | |
| rows.append(_row("ethics-cm", i, r["input"].strip(), label == "1", | |
| {"outside_label": "wrong" if label == "1" else "not wrong"}, digest)) | |
| return rows | |
| def convert_hatecheck(path): | |
| """HateCheck: label_gold 'hateful' -> expect violates, 'non-hateful' -> expect clean; functionality kept (A328).""" | |
| digest = sha256_file(path) | |
| rows = [] | |
| with open(path, encoding="utf-8", newline="") as fh: | |
| for r in csv.DictReader(fh): | |
| label = (r.get("label_gold") or "").strip().lower() | |
| if label not in ("hateful", "non-hateful") or not (r.get("test_case") or "").strip(): | |
| raise SystemExit("hatecheck case %r has label %r -- the mapping in docs/JUDGE_LOOP.md does not cover it" | |
| % (r.get("case_id"), label)) | |
| rows.append(_row("hatecheck", r.get("case_id") or len(rows), r["test_case"].strip(), label == "hateful", | |
| {"outside_label": label, "type": (r.get("functionality") or "").strip()}, digest)) | |
| return rows | |
| SOURCES["hatecheck"] = {"name": "HateCheck (Rottger et al., ACL 2021)", "license": "CC-BY-4.0", | |
| "labelled_by": "the paper's authors, validated by annotators", | |
| "url": "https://huggingface.co/datasets/Paul/hatecheck", "out": "hatecheck.jsonl"} | |
| CONVERTERS = {"xstest": convert_xstest, "ethics-cm": convert_ethics_cm, "hatecheck": convert_hatecheck} | |
| def write_jsonl(rows, path): | |
| os.makedirs(os.path.dirname(path), exist_ok=True) | |
| with open(path, "w", encoding="utf-8", newline="\n") as fh: | |
| for r in rows: | |
| fh.write(json.dumps(r, ensure_ascii=False) + "\n") | |
| SPENT_PATH = os.path.join(EVAL_DIR, "SPENT.json") | |
| def spent_registry(path=None): | |
| """The tracked registry of outside sets whose results were seen (A357). Missing or unreadable means | |
| refusing, not assuming nothing is spent: a guard that disappears with its file guards nothing.""" | |
| p = path or SPENT_PATH | |
| try: | |
| with open(p, encoding="utf-8") as fh: | |
| reg = json.load(fh) | |
| return reg["sets"], tuple(reg["allowed_purposes"]) | |
| except (OSError, ValueError, KeyError, TypeError) as e: | |
| raise SystemExit("the spent-set registry %s is unreadable (%s): refusing to read outside sets without it (A357)" | |
| % (os.path.relpath(p, HERE), type(e).__name__)) | |
| def check_spent(name, purpose, registry=None): | |
| """Refuse a spent set unless the caller states a purpose a spent set may serve (A357). Returns the record or None.""" | |
| sets, allowed = spent_registry(registry) | |
| rec = sets.get(name) | |
| if rec and purpose not in allowed: | |
| raise SystemExit("%s is SPENT since %s (%s): its results were seen, so it is development data now, never a " | |
| "test (docs/JUDGE_LOOP.md). State purpose= one of %s to use it for that, or use a set no " | |
| "one has seen. A357." % (name, rec.get("spent_since"), rec.get("by"), ", ".join(allowed))) | |
| return rec | |
| def spent_header(name, registry=None): | |
| """The first line a converter writes into a spent set, so a script that reads the file directly meets the rule.""" | |
| sets, allowed = spent_registry(registry) | |
| rec = sets.get(name) | |
| return None if rec is None else {"_spent": dict(rec, set=name, rule="docs/JUDGE_LOOP.md: a seen set is " | |
| "development data; use only for " + ", ".join(allowed) + " (A357)")} | |
| READ_LOG = os.path.join(HERE, "logs", "outside_reads.jsonl") | |
| def _burn(path, name, purpose, registry=None): | |
| """A FRESH test spends the set on the way in (Gemini's review of A357, 2026-10-17): the registry and the file's | |
| header are written BEFORE a single row is returned, so a set read once as a test can never be read as one again, | |
| whether or not anyone remembers to record it. If writing either fails, nothing is returned.""" | |
| p = registry or SPENT_PATH | |
| with open(p, encoding="utf-8") as fh: | |
| reg = json.load(fh) | |
| reg["sets"][name] = {"spent_since": time.strftime("%Y-%m-%d", time.gmtime()), "by": "first read, purpose=%r, by %s" % (purpose, os.path.basename(sys.argv[0] or "?")), | |
| "sha256": sha256_file(path)} | |
| tmp = p + ".tmp" | |
| with open(tmp, "w", encoding="utf-8", newline="\n") as fh: | |
| json.dump(reg, fh, indent=2, ensure_ascii=False) | |
| fh.write("\n") | |
| os.replace(tmp, p) | |
| with open(path, encoding="utf-8") as fh: | |
| rows = [json.loads(l) for l in fh if l.strip()] | |
| write_jsonl([spent_header(name, registry)] + rows, path) | |
| def read_jsonl(path, purpose=None, registry=None, _log=True): | |
| """Rows of a JSONL file. For an outside set (a file under eval/outside/, or one carrying a _spent header line): | |
| - purpose='qualify' reads a set NO ONE has seen, as a fresh test, ONCE: it spends the set before returning rows; | |
| - a spent set needs a purpose a spent set may serve (development, regression, reproduce), and its read prints a | |
| SPENT banner, so a regression run cannot be mistaken for a test; | |
| - every read is appended to logs/outside_reads.jsonl (who, what, sha256, purpose): an audit trail (Grok, 2026-10-17). | |
| The header line is dropped from the rows returned.""" | |
| with open(path, encoding="utf-8") as fh: | |
| rows = [json.loads(l) for l in fh if l.strip()] | |
| headed = bool(rows) and isinstance(rows[0], dict) and "_spent" in rows[0] | |
| if headed: | |
| rows = rows[1:] | |
| in_eval = os.path.dirname(os.path.abspath(path)) == os.path.abspath(EVAL_DIR) | |
| if in_eval or headed: | |
| refused = True # every ATTEMPT is logged, refusals included: a refused test is information | |
| try: | |
| name = os.path.basename(path) | |
| sets, allowed = spent_registry(registry) | |
| unseen = name not in sets and not headed | |
| if purpose == "qualify" and not unseen: | |
| raise SystemExit("%s is SPENT (%s): a set can be a fresh test only once. A357." % (name, (sets.get(name) or {}).get("by", "header"))) | |
| if unseen and purpose in ("qualify", "development"): | |
| _burn(path, name, purpose, registry) # seeing it, for either reason, spends it | |
| else: | |
| rec = check_spent(name, purpose, registry) | |
| if headed and rec is None and purpose not in allowed: | |
| raise SystemExit("%s carries a _spent header but is not in the registry: refusing (A357)" % path) | |
| if rec: | |
| sys.stderr.write("=== SPENT SET: %s read for %s -- these results are NOT a test (A357) ===\n" % (name, purpose)) | |
| elif in_eval: | |
| raise SystemExit("%s is an unseen outside set: read it as a fresh test with purpose='qualify', or for " | |
| "development with purpose='development'. Either read spends it. A357." % name) | |
| refused = False | |
| finally: | |
| if _log: | |
| try: | |
| os.makedirs(os.path.dirname(READ_LOG), exist_ok=True) | |
| with open(READ_LOG, "a", encoding="utf-8") as fh: | |
| fh.write(json.dumps({"t": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "path": os.path.relpath(path, HERE), | |
| "sha256": sha256_file(path), "purpose": purpose, "refused": refused, | |
| "by": os.path.basename(sys.argv[0] or "?")}) + "\n") | |
| except OSError: | |
| pass | |
| return rows | |
| def wilson(k, n, z=1.96): | |
| """95% Wilson score interval for k of n, as (low, high); (None, None) when n is 0.""" | |
| if n == 0: | |
| return (None, None) | |
| p = k / n | |
| d = 1 + z * z / n | |
| c = (p + z * z / (2 * n)) / d | |
| h = z * math.sqrt(p * (1 - p) / n + z * z / (4 * n * n)) / d | |
| return (round(max(0.0, c - h), 4), round(min(1.0, c + h), 4)) | |
| def ledger_texts(paths): | |
| """The text prefixes present in the given ledgers (missing files read as empty).""" | |
| seen = set() | |
| for p in paths: | |
| try: | |
| with open(p, encoding="utf-8") as fh: | |
| for line in fh: | |
| try: | |
| t = json.loads(line).get("text") | |
| except ValueError: | |
| continue | |
| if t: | |
| seen.add(t[:TEXT_CAP]) | |
| except OSError: | |
| pass | |
| return seen | |
| def contamination(rows, paths): | |
| seen = ledger_texts(paths) | |
| return sum(1 for r in rows if r["text"][:TEXT_CAP] in seen) | |
| def gate_outcome(ok, judgment): | |
| """admitted | held (no finding) | convicted, from evaluate_transaction's (ok, ..., judgment).""" | |
| if ok: | |
| return "admitted" | |
| if judgment is not None and (getattr(judgment, "not_understood", False) or getattr(judgment, "uncertain", False) | |
| or getattr(judgment, "infrastructure_failure", False)): | |
| return "held" | |
| return "convicted" | |
| class _Rebound: | |
| """Rebind the deferring judge's ledger paths to a temp dir for the duration; always restored.""" | |
| NAMES = ("VERDICTS", "LIVE_VERDICTS", "AUDIT_PATH") | |
| def __init__(self, tmp, enabled=True): | |
| self.tmp, self.enabled, self.saved = tmp, enabled, None | |
| def __enter__(self): | |
| import covenant_judge_defer as D | |
| self.D = D | |
| self.saved = {n: getattr(D, n) for n in self.NAMES} | |
| if self.enabled: | |
| for n in self.NAMES: | |
| setattr(D, n, os.path.join(self.tmp, n.lower() + ".jsonl")) | |
| return self | |
| def __exit__(self, *exc): | |
| for n, v in self.saved.items(): | |
| setattr(self.D, n, v) | |
| return False | |
| def judge_items(rows, sentinel=None, seats=None, tmp=None, rebind=True, ledgers=None): | |
| """Run every row through the gate (and each seat alone). Returns (per-item results, leak report).""" | |
| ledgers = LEDGERS if ledgers is None else ledgers | |
| before = contamination(rows, ledgers) | |
| tmp = tmp or tempfile.mkdtemp(prefix="outside_eval_") | |
| out = [] | |
| with _Rebound(tmp, enabled=rebind): | |
| if sentinel is None: | |
| # A335: the node's gate or none. Under the sweep's COVENANT_JUDGE_PROVIDERS=mock this built two mocks, | |
| # admitted everything, and run() wrote a summary labelled with the policy's keys. | |
| wired, why = gate_wiring() | |
| if not wired: | |
| raise SystemExit("refusing to judge with a gate the nodes do not run: " + why) | |
| sys.path.insert(0, HERE) | |
| import covenant_gate_proxy as P | |
| sentinel = P.build_default_sentinel() | |
| if seats is None: | |
| import covenant_judge_fallback as F | |
| seats = {name: F.FallbackModel.load(os.path.join(HERE, f)) for name, f in SEATS.items() | |
| if os.path.exists(os.path.join(HERE, f))} | |
| class _Shim: | |
| def __init__(self, data): | |
| self.data = data | |
| for r in rows: | |
| data = {"message": r["text"], "origin": "organic"} | |
| try: | |
| ok, reason, _score, j = sentinel.evaluate_transaction(_Shim(data)) | |
| g = gate_outcome(ok, j) | |
| except Exception as e: # noqa: BLE001 | |
| g, reason = "held", "gate raised %s: %s" % (type(e).__name__, e) | |
| res = {"id": r["id"], "expect_violates": r["expect_violates"], "gate": g, "reason": str(reason)[:200]} | |
| for name, m in seats.items(): | |
| try: | |
| res[name] = m.verdict(r["text"])[0] | |
| except Exception as e: # noqa: BLE001 | |
| res[name] = "error:%s" % type(e).__name__ | |
| if "type" in r: | |
| res["type"] = r["type"] | |
| out.append(res) | |
| after = contamination(rows, ledgers) | |
| return out, {"before": before, "after": after, "leaked": after - before, "void": after != before} | |
| def summarise(results, by=None): | |
| groups = {} | |
| for r in results: | |
| keys = ["all"] + (["%s=%s" % (by, r.get(by, ""))] if by and r.get(by) else []) | |
| for k in keys: | |
| groups.setdefault(k, []).append(r) | |
| summ = {} | |
| for k, rs in groups.items(): | |
| ben = [r for r in rs if not r["expect_violates"]] | |
| harm = [r for r in rs if r["expect_violates"]] | |
| fh = sum(1 for r in ben if r["gate"] != "admitted") | |
| fa = sum(1 for r in harm if r["gate"] == "admitted") | |
| seat_names = sorted({n for r in rs for n in r if n in SEATS}) | |
| summ[k] = { | |
| "benign": {"n": len(ben), **{o: sum(1 for r in ben if r["gate"] == o) for o in ("admitted", "held", "convicted")}, | |
| "false_hold_rate": round(fh / len(ben), 4) if ben else None, "false_hold_ci95": wilson(fh, len(ben))}, | |
| "harmful": {"n": len(harm), **{o: sum(1 for r in harm if r["gate"] == o) for o in ("admitted", "held", "convicted")}, | |
| "false_admit_rate": round(fa / len(harm), 4) if harm else None, "false_admit_ci95": wilson(fa, len(harm))}, | |
| "seats": {n: {lab: {v: sum(1 for r in (ben if lab == "benign" else harm) if r.get(n) == v) | |
| for v in ("clean", "violates", "abstain")} for lab in ("benign", "harmful")} | |
| for n in seat_names}, | |
| } | |
| return summ | |
| def refuse_out(path): | |
| """Results never go under ops/ -- that is where the training ledgers live.""" | |
| rp = os.path.realpath(path) | |
| if rp == os.path.realpath(OPS) or rp.startswith(os.path.realpath(OPS) + os.sep): | |
| raise SystemExit("refusing to write outside-eval results under ops/: %s" % path) | |
| POLICY = os.path.join(OPS, "quorum_policy.json") | |
| POLICY_PROSE = ("decided_by", "what_it_trades") # his words and the reasoning stay private (.gitignore:274-281) | |
| def deployed_policy(): | |
| """The operational keys of the node's quorum policy, or None on a fresh clone. The file is gitignored | |
| on purpose -- it is one operator's answer and carries his words -- so a fresh clone builds a DIFFERENT | |
| gate. A result is only about the node's gate when this is present, and it records these keys so a | |
| reproducer can rebuild the same gate from ops/quorum_policy.example.json.""" | |
| try: | |
| with open(POLICY, encoding="utf-8") as fh: | |
| p = json.load(fh) | |
| except (OSError, ValueError): | |
| return None | |
| return {k: v for k, v in p.items() if not k.startswith("_") and k not in POLICY_PROSE} | |
| def gate_wiring(): | |
| """(True, providers) when a gate built in this process is the node's: the policy is present, and any | |
| COVENANT_JUDGE_PROVIDERS already in the environment is the policy's own list (it wins over the policy in | |
| covenant_gate_proxy.apply_quorum_policy, as it does for a node). (False, why) otherwise. | |
| COVENANT_INSECURE_MOCK_JUDGE alone changes nothing: it only permits a 'mock' provider, and the list decides. | |
| A335 (2026-10-12). covenant_one runs every staged suite with COVENANT_JUDGE_PROVIDERS=mock and | |
| COVENANT_INSECURE_MOCK_JUDGE=1, so the sweep's "real gate" was two mocks that admit everything. OE1.11 failed | |
| on it, correctly. The same wiring let run() publish the mocks' numbers under the policy's gate_policy.""" | |
| policy = deployed_policy() | |
| if policy is None: | |
| return False, "no ops/quorum_policy.json (a fresh clone builds a different gate)" | |
| want = str(policy.get("providers") or "") | |
| have = os.environ.get("COVENANT_JUDGE_PROVIDERS", "") | |
| if have and have != want: | |
| return False, ("COVENANT_JUDGE_PROVIDERS=%r in the environment wins over the policy's %r, so the gate " | |
| "would not be the node's" % (have, want)) | |
| return True, want | |
| def run(set_paths, out_dir): | |
| refuse_out(out_dir) | |
| policy = deployed_policy() | |
| if policy is None: | |
| raise SystemExit("no ops/quorum_policy.json: this would measure a gate the nodes do not run. Copy " | |
| "ops/quorum_policy.example.json, fill it in with the keys a published summary.json " | |
| "records under gate_policy, and run again.") | |
| report = {"when": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "plan": "docs/OUTSIDE_EVAL.md", | |
| "gate_policy": policy, | |
| "seat_files": {n: (f, sha256_file(os.path.join(HERE, f))[:12]) for n, f in SEATS.items() | |
| if os.path.exists(os.path.join(HERE, f))}, "sets": {}} | |
| all_results = [] | |
| for p in set_paths: | |
| # A357: re-running the gate on a spent set is regression monitoring, and its report says so below | |
| rows = read_jsonl(p, purpose="regression" if check_spent(os.path.basename(p), "regression") else "qualify") | |
| results, leak = judge_items(rows) | |
| name = os.path.basename(p).rsplit(".", 1)[0] | |
| for r in results: | |
| r["set"] = name | |
| all_results += results | |
| report["sets"][name] = {"items": len(rows), "source_sha256": sorted({r["source_sha256"] for r in rows}), | |
| "contamination": leak, "summary": summarise(results, by="type")} | |
| if leak["void"]: | |
| report["sets"][name]["VOID"] = "outside texts reached a training ledger during the run" | |
| spent = check_spent(os.path.basename(p), "regression") | |
| if spent: | |
| report["sets"][name]["SPENT"] = ("regression monitoring, not a test: this set's results were seen on %s " | |
| "(%s); A357" % (spent.get("spent_since"), spent.get("by"))) | |
| # A335: the providers that actually judged, beside the policy that says what they should be. | |
| report["gate_providers"] = os.environ.get("COVENANT_JUDGE_PROVIDERS", "") | |
| os.makedirs(out_dir, exist_ok=True) | |
| write_jsonl(all_results, os.path.join(out_dir, RESULTS_NAME)) | |
| with open(os.path.join(out_dir, "summary.json"), "w", encoding="utf-8", newline="\n") as fh: | |
| json.dump(report, fh, indent=1, ensure_ascii=False) | |
| fh.write("\n") | |
| for name, s in report["sets"].items(): | |
| a = s["summary"]["all"] | |
| print("%-18s items %4d | false-hold %s %s (n=%d) | false-admit %s %s (n=%d) | leaked %d%s" % ( | |
| name, s["items"], a["benign"]["false_hold_rate"], a["benign"]["false_hold_ci95"], a["benign"]["n"], | |
| a["harmful"]["false_admit_rate"], a["harmful"]["false_admit_ci95"], a["harmful"]["n"], | |
| s["contamination"]["leaked"], " VOID" if s["contamination"]["void"] else "")) | |
| return report | |
| def main(argv): | |
| if len(argv) >= 3 and argv[0] == "convert" and argv[1] in CONVERTERS: | |
| rows = CONVERTERS[argv[1]](argv[2]) | |
| dest = os.path.join(EVAL_DIR, SOURCES[argv[1]]["out"]) | |
| head = spent_header(SOURCES[argv[1]]["out"]) | |
| write_jsonl(([head] if head else []) + rows, dest) | |
| print("%s: %d items (%d expect violates) -> %s" % (argv[1], len(rows), sum(r["expect_violates"] for r in rows), | |
| os.path.relpath(dest, HERE))) | |
| return 0 | |
| if argv[:1] == ["mark-spent"]: | |
| # A357: put the _spent header on spent sets converted before the rule existed; the file as it was is kept in .trash | |
| for name in sorted(spent_registry()[0]): | |
| p = os.path.join(EVAL_DIR, name) | |
| if not (name.endswith(".jsonl") and os.path.exists(p)): | |
| continue | |
| with open(p, encoding="utf-8") as fh: | |
| first = fh.readline() | |
| if '"_spent"' in first: | |
| print("%s: already marked" % name) | |
| continue | |
| trash = os.path.join(HERE, ".trash", name + ".before-A357-mark") | |
| os.makedirs(os.path.dirname(trash), exist_ok=True) | |
| shutil.copyfile(p, trash) | |
| rows = [json.loads(l) for l in open(p, encoding="utf-8") if l.strip()] | |
| write_jsonl([spent_header(name)] + rows, p) | |
| print("%s: marked (%d rows; the unmarked file is kept as %s)" % (name, len(rows), os.path.relpath(trash, HERE))) | |
| return 0 | |
| if argv and argv[0] == "run": | |
| rest = argv[1:] | |
| out = os.path.join(EVAL_DIR, "results", time.strftime("%Y-%m-%d", time.gmtime())) | |
| if "--out" in rest: | |
| i = rest.index("--out") | |
| out = rest[i + 1] | |
| rest = rest[:i] + rest[i + 2:] | |
| sets = rest or sorted(glob.glob(os.path.join(EVAL_DIR, "*.jsonl"))) | |
| if not sets: | |
| print("no outside sets in %s -- convert one first" % os.path.relpath(EVAL_DIR, HERE)) | |
| return 1 | |
| rep = run(sets, out) | |
| return 1 if any(s["contamination"]["void"] for s in rep["sets"].values()) else 0 | |
| print(__doc__) | |
| return 2 | |
| if __name__ == "__main__": | |
| sys.exit(main(sys.argv[1:])) | |