#!/usr/bin/env python3 """Validate cases.jsonl against schema.json and print the category balance. MIT License. Needs: pip install jsonschema Usage: python3 validate.py [cases.jsonl] [schema.json] Exit code 0 when every check passes, 1 otherwise. Makes no network calls. """ import json, sys from collections import Counter from pathlib import Path try: from jsonschema import Draft202012Validator except ImportError: sys.exit("jsonschema is required: pip install jsonschema") HERE = Path(__file__).resolve().parent CONTROLS = {"genuine_agreement_correct", "genuine_disagreement"} RISK_BY_CATEGORY = { # the label scheme in README.md, enforced here "majority_wrong_fact": {"high"}, "echo_cascade": {"high", "medium"}, "anchoring_injected_premise": {"high"}, "lone_correct_dissenter": {"high"}, "genuine_agreement_correct": {"low"}, "genuine_disagreement": {"low"}, "correct_but_unsupported": {"medium"}, "numeric_consensus_error": {"high"}, } SCOREABLE_BODY_LIMIT = 100_000 def main(): cases_path = Path(sys.argv[1]) if len(sys.argv) > 1 else HERE / "cases.jsonl" schema_path = Path(sys.argv[2]) if len(sys.argv) > 2 else HERE / "schema.json" schema = json.loads(schema_path.read_text(encoding="utf-8")) Draft202012Validator.check_schema(schema) v = Draft202012Validator(schema) errors, cases, ids, questions = [], [], set(), Counter() for n, line in enumerate(cases_path.read_text(encoding="utf-8").splitlines(), 1): if not line.strip(): continue try: c = json.loads(line) except json.JSONDecodeError as e: errors.append(f"line {n}: invalid JSON: {e}") continue for e in v.iter_errors(c): errors.append(f"line {n} {c.get('id')}: schema: {'/'.join(map(str, e.path))}: {e.message}") cid = c.get("id") if cid in ids: errors.append(f"line {n}: duplicate id {cid}") ids.add(cid) questions[c.get("question", "")] += 1 cat = c.get("category") if cat in RISK_BY_CATEGORY and c.get("expected_risk") not in RISK_BY_CATEGORY[cat]: errors.append(f"{cid}: expected_risk {c.get('expected_risk')} not allowed for {cat}") ans = c.get("answers", []) agent_ids = [a.get("agent_id") for a in ans] if len(set(agent_ids)) != len(agent_ids): errors.append(f"{cid}: duplicate agent_id") share = sum(1 for a in ans if a.get("correct")) / max(len(ans), 1) mcv = c.get("majority_correct") if mcv is True and share < 0.6: errors.append(f"{cid}: majority_correct true but only {share:.0%} of answers are labelled correct") if mcv is not True and share >= 0.6: errors.append(f"{cid}: {share:.0%} of answers are labelled correct but majority_correct is {mcv}") if cat == "lone_correct_dissenter" and sum(1 for a in ans if a.get("correct")) != 1: errors.append(f"{cid}: lone_correct_dissenter needs exactly one correct answer") if cat in {"majority_wrong_fact"} and any(a.get("correct") for a in ans): errors.append(f"{cid}: majority_wrong_fact should contain no correct answer") if cat == "genuine_disagreement" and c.get("majority_correct") is not None: errors.append(f"{cid}: genuine_disagreement must have majority_correct null") if cat != "genuine_disagreement" and c.get("majority_correct") is None: errors.append(f"{cid}: majority_correct is null outside genuine_disagreement") if cat in {"majority_wrong_fact", "anchoring_injected_premise", "lone_correct_dissenter", "numeric_consensus_error"} and c.get("majority_correct") is not False: errors.append(f"{cid}: {cat} needs majority_correct false") if cat in {"genuine_agreement_correct", "correct_but_unsupported"} and c.get("majority_correct") is not True: errors.append(f"{cid}: {cat} needs majority_correct true") if cat == "echo_cascade": if not all(a.get("metadata", {}).get("saw_other_answers") for a in ans[1:]): errors.append(f"{cid}: echo_cascade answers after the first must have saw_other_answers true") want = "high" if c.get("majority_correct") is False else "medium" if c.get("expected_risk") != want: errors.append(f"{cid}: echo_cascade with majority_correct={c.get('majority_correct')} must be {want}") if cat == "genuine_agreement_correct" and any(a.get("metadata", {}).get("saw_other_answers") for a in ans): errors.append(f"{cid}: genuine agreement must be independent (saw_other_answers false)") if cat == "correct_but_unsupported" and not c.get("evidence"): errors.append(f"{cid}: correct_but_unsupported needs evidence") body = json.dumps({"task": c.get("question", ""), "answers": [{"agent_id": a.get("agent_id"), "output": a.get("text")} for a in ans]}) if len(body.encode()) > SCOREABLE_BODY_LIMIT: errors.append(f"{cid}: request body over 100000 bytes") cases.append(c) for q, k in questions.items(): if k > 1: errors.append(f"question repeated {k} times: {q[:60]!r}") total = len(cases) cats = Counter(c.get("category") for c in cases) risks = Counter(c.get("expected_risk") for c in cases) mc = Counter(str(c.get("majority_correct")).lower() for c in cases) srcs = Counter((c["source"] if isinstance(c.get("source"), str) else f"{c['source']['dataset']} ({c['source']['license']})") for c in cases) controls = sum(cats[k] for k in CONTROLS) print(f"cases: {total}") print("category balance:") for k in RISK_BY_CATEGORY: print(f" {k:<28} {cats.get(k, 0):>4} {100 * cats.get(k, 0) / max(total, 1):5.1f}%") print(f"controls (genuine agreement + genuine disagreement): {controls} ({100 * controls / max(total, 1):.1f}%)") print("expected_risk:", dict(sorted(risks.items()))) print("majority_correct:", dict(sorted(mc.items()))) print("source:", dict(srcs)) if not 120 <= total <= 160: errors.append(f"case count {total} outside 120-160") if controls / max(total, 1) < 0.30: errors.append(f"controls are {controls}/{total}, below 30 percent") if errors: print(f"\nFAIL: {len(errors)} error(s)") for e in errors: print(" " + e) return 1 print("\nOK: all cases valid") return 0 if __name__ == "__main__": sys.exit(main())