covenant / tools /outside_eval.py
Lawless1987's picture
Snapshot of covenant main (full git history on GitHub) (part 4)
f373219 verified
Raw History Blame Contribute Delete
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:]))