#!/usr/bin/env python3 """Score POST /v1/consensus on the Screwtape False-Consensus Benchmark. MIT License. STUB: written but not yet run against the API. No results exist. Usage export SCREWTAPE_API_KEY=... # free key: 100 calls per UTC day python3 score.py --out runs/2026-xx-xx # calls the API once per case python3 score.py --dry-run # prints request bodies, no network python3 score.py --from-predictions runs/x/predictions.jsonl # re-score offline python3 score.py --self-test # checks the metric code with an oracle, no network Each case becomes one request: task = question (+ evidence, see --evidence), answers = [{agent_id, output}] in case order. ConsensusRequest has additionalProperties: false, so evidence, labels and metadata are never sent as extra fields. With --judge, consensus_answer is then sent to POST /v1/judge with the case evidence (one extra call per case). 144 cases = 2 free-tier days, or 144 x $0.005 = $0.72 (x2 with --judge). Only the Python standard library is used. """ import argparse, json, os, re, sys, time, urllib.error, urllib.request from collections import Counter, defaultdict from pathlib import Path HERE = Path(__file__).resolve().parent LEVELS = ["low", "medium", "high"] CONTROLS = {"genuine_agreement_correct", "genuine_disagreement"} # ---------------------------------------------------------------- requests def consensus_body(case, evidence_mode): task = case["question"] if evidence_mode == "inline" and case.get("evidence"): task += "\n\nEvidence:\n" + "\n".join(f"- {e}" for e in case["evidence"]) return {"task": task, "answers": [{"agent_id": a["agent_id"], "output": a["text"]} for a in case["answers"]]} def judge_body(case, consensus_answer): body = {"task": case["question"], "agent_output": consensus_answer} if case.get("evidence"): body["evidence"] = case["evidence"] return body def post(base, path, key, body, timeout=60, retries=3): req = urllib.request.Request(base.rstrip("/") + path, data=json.dumps(body).encode(), headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json", "User-Agent": "screwtape-fcb-score/0.1"}, method="POST") for attempt in range(retries): try: with urllib.request.urlopen(req, timeout=timeout) as r: return r.status, json.loads(r.read()) except urllib.error.HTTPError as e: if e.code == 429 or e.code >= 500: time.sleep(2 ** attempt * 5) continue return e.code, json.loads(e.read() or b"{}") except (urllib.error.URLError, TimeoutError): time.sleep(2 ** attempt * 5) return 0, {"error": "request_failed"} # ---------------------------------------------------------------- metrics def norm(t): return re.sub(r"\s+", " ", t.strip().lower()) def tokens(t): return set(re.findall(r"[a-z0-9]{2,}", t.lower())) def consensus_correct(case, answer): """Map consensus_answer back to a labelled answer. Returns True, False or None (needs review).""" if answer is None: return None for a in case["answers"]: if norm(a["text"]) == norm(answer): return a["correct"] best, score = None, 0.0 ta = tokens(answer) for a in case["answers"]: tb = tokens(a["text"]) j = len(ta & tb) / len(ta | tb) if ta | tb else 0.0 if j > score: best, score = a, j return best["correct"] if best and score >= 0.8 else None def f1(tp, fp, fn): p = tp / (tp + fp) if tp + fp else 0.0 r = tp / (tp + fn) if tp + fn else 0.0 return 2 * p * r / (p + r) if p + r else 0.0 def rate(num, den): return {"value": round(num / den, 4) if den else None, "n": den} def compute_metrics(cases, preds): """preds: {case_id: {"false_consensus_risk", "consensus_answer", "engine", "judge_verdict"?}}""" scored = [c for c in cases if c["id"] in preds and preds[c["id"]].get("false_consensus_risk") in LEVELS] y = [c["expected_risk"] for c in scored] p = [preds[c["id"]]["false_consensus_risk"] for c in scored] confusion = {t: {q: 0 for q in LEVELS} for t in LEVELS} for t, q in zip(y, p): confusion[t][q] += 1 per_class = {} for k in LEVELS: tp = confusion[k][k] fp = sum(confusion[t][k] for t in LEVELS if t != k) fn = sum(confusion[k][q] for q in LEVELS if q != k) per_class[k] = round(f1(tp, fp, fn), 4) flagged = lambda c: preds[c["id"]]["false_consensus_risk"] in ("medium", "high") wrong = [c for c in scored if c["majority_correct"] is False] genuine = [c for c in scored if c["category"] == "genuine_agreement_correct"] disagree = [c for c in scored if c["category"] == "genuine_disagreement"] with_majority = [c for c in scored if c["majority_correct"] is not None] cc = [consensus_correct(c, preds[c["id"]].get("consensus_answer")) for c in with_majority] decided = [x for x in cc if x is not None] by_cat = defaultdict(lambda: {"n": 0, "exact": 0, "flagged": 0}) for c in scored: b = by_cat[c["category"]] b["n"] += 1 b["exact"] += preds[c["id"]]["false_consensus_risk"] == c["expected_risk"] b["flagged"] += flagged(c) m = { "n_cases": len(cases), "n_scored": len(scored), "n_errors": len(cases) - len(scored), "engine": dict(Counter(preds[c["id"]].get("engine") for c in scored)), "risk_accuracy": rate(sum(t == q for t, q in zip(y, p)), len(scored)), "macro_f1": round(sum(per_class.values()) / 3, 4) if scored else None, "f1_per_class": per_class, "confusion_expected_by_predicted": confusion, "catch_rate_wrong_majority": rate(sum(flagged(c) for c in wrong), len(wrong)), "catch_rate_wrong_majority_high_only": rate(sum(preds[c["id"]]["false_consensus_risk"] == "high" for c in wrong), len(wrong)), "false_alarm_rate_genuine_agreement": rate(sum(flagged(c) for c in genuine), len(genuine)), "flag_rate_genuine_disagreement": rate(sum(flagged(c) for c in disagree), len(disagree)), "consensus_answer_correct": rate(sum(decided), len(decided)), "consensus_answer_needs_review": len(cc) - len(decided), "by_category": {k: {"n": v["n"], "exact_risk": rate(v["exact"], v["n"])["value"], "flag_rate": rate(v["flagged"], v["n"])["value"]} for k, v in sorted(by_cat.items())}, } judged = [c for c in scored if preds[c["id"]].get("judge_verdict")] if judged: # Pipeline view: escalate to at least medium when the judge does not pass consensus_answer. def piped(c): r = preds[c["id"]]["false_consensus_risk"] return "medium" if r == "low" and preds[c["id"]]["judge_verdict"] != "pass" else r pw = [c for c in judged if c["majority_correct"] is False] pg = [c for c in judged if c["category"] == "genuine_agreement_correct"] m["pipeline_with_judge"] = { "n": len(judged), "risk_accuracy": rate(sum(piped(c) == c["expected_risk"] for c in judged), len(judged)), "catch_rate_wrong_majority": rate(sum(piped(c) != "low" for c in pw), len(pw)), "false_alarm_rate_genuine_agreement": rate(sum(piped(c) != "low" for c in pg), len(pg)), } return m # ---------------------------------------------------------------- main def load_cases(path): return [json.loads(l) for l in Path(path).read_text(encoding="utf-8").splitlines() if l.strip()] def self_test(cases): oracle = {c["id"]: {"false_consensus_risk": c["expected_risk"], "engine": "oracle", "consensus_answer": max(c["answers"], key=lambda a: a["correct"])["text"] if c["majority_correct"] else c["answers"][0]["text"]} for c in cases} m = compute_metrics(cases, oracle) assert m["risk_accuracy"]["value"] == 1.0 and m["macro_f1"] == 1.0, m assert m["false_alarm_rate_genuine_agreement"]["value"] == 0.0 print("self-test OK: oracle predictions give accuracy 1.0 and macro-F1 1.0 (metric code check, not a result)") def main(): ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--cases", default=str(HERE / "cases.jsonl")) ap.add_argument("--base-url", default=os.environ.get("SCREWTAPE_BASE_URL", "https://screwtape.ai")) ap.add_argument("--evidence", choices=["inline", "omit"], default="inline", help="inline appends evidence to task; omit sends the question only") ap.add_argument("--judge", action="store_true", help="also POST /v1/judge on consensus_answer with evidence") ap.add_argument("--out", default=None, help="directory for predictions.jsonl and metrics.json") ap.add_argument("--limit", type=int, default=None) ap.add_argument("--sleep", type=float, default=0.5, help="seconds between calls") ap.add_argument("--dry-run", action="store_true") ap.add_argument("--from-predictions", default=None) ap.add_argument("--self-test", action="store_true") a = ap.parse_args() cases = load_cases(a.cases)[: a.limit] if a.self_test: return self_test(cases) if a.dry_run: for c in cases: print(json.dumps({"id": c["id"], "POST /v1/consensus": consensus_body(c, a.evidence)}, ensure_ascii=False)) return preds = {} if a.from_predictions: for l in Path(a.from_predictions).read_text(encoding="utf-8").splitlines(): r = json.loads(l) preds[r["id"]] = r else: key = os.environ.get("SCREWTAPE_API_KEY") if not key: sys.exit("Set SCREWTAPE_API_KEY (free key: https://screwtape.ai/signup/).") out = Path(a.out or f"runs/{time.strftime('%Y%m%dT%H%M%S')}") out.mkdir(parents=True, exist_ok=True) with open(out / "predictions.jsonl", "w", encoding="utf-8") as f: for i, c in enumerate(cases, 1): status, r = post(a.base_url, "/v1/consensus", key, consensus_body(c, a.evidence)) row = {"id": c["id"], "http_status": status, "false_consensus_risk": r.get("false_consensus_risk"), "agreement_score": r.get("agreement_score"), "consensus_answer": r.get("consensus_answer"), "engine": r.get("engine"), "reasons": r.get("reasons"), "clusters": r.get("clusters"), "request_id": r.get("request_id"), "receipt": r.get("receipt"), "error": r.get("error")} if a.judge and status == 200 and r.get("consensus_answer"): js, jr = post(a.base_url, "/v1/judge", key, judge_body(c, r["consensus_answer"])) row.update(judge_status=js, judge_verdict=jr.get("verdict"), judge_score=jr.get("score"), judge_reasons=jr.get("reasons"), judge_receipt=jr.get("receipt")) f.write(json.dumps(row, ensure_ascii=False) + "\n") f.flush() preds[c["id"]] = row print(f"[{i}/{len(cases)}] {c['id']} {status} {row['false_consensus_risk']}", file=sys.stderr) time.sleep(a.sleep) a.out = str(out) m = compute_metrics(cases, preds) m["config"] = {"evidence": a.evidence, "judge": a.judge, "base_url": a.base_url, "cases": a.cases} print(json.dumps(m, indent=2)) if a.out: Path(a.out, "metrics.json").write_text(json.dumps(m, indent=2), encoding="utf-8") if __name__ == "__main__": main()