""" SNIN Emission Paper Pilot — v0.4.2 runner. Covers all three boundary blockers from astranaut01 #24948 + 8 falsifiers + 7 adversarial reversals. """ import json import emission as E import ledger_store as LS import classifier as CL V = json.load(open("test_vectors_v0.4.2.json")) fails = [] def check(name, ok, detail=""): print(("PASS " if ok else "FAIL ") + name + (f" [{detail}]" if detail else "")) if not ok: fails.append(name + " " + detail) def strip_fixture(rows): return [{k: v for k, v in r.items() if k != "is_fixture"} for r in rows] def renumber(rows): """Each fixture history is an independent simulation: renumber seq to 1..N in order and remap reverses_event_id references.""" mapping = {} out = [] for i, r in enumerate(rows, start=1): mapping[r["seq"]] = i nr = dict(r) nr["seq"] = i if nr.get("reverses_event_id"): nr["reverses_event_id"] = mapping.get(nr["reverses_event_id"], nr["reverses_event_id"]) out.append(nr) return out # ---- setup: issue assignments + attestations ---------------------------- store = LS.LedgerStore() for a in V["assignments"]: store.issue_assignment(a) for a in V["attestations"] + V.get("attestation_extra", []): store.add_attestation(a) # ---- 1. P1 > reds (formula via store) ----------------------------------- nets = {} for name, fx in V["fixture_histories"].items(): if fx.get("expected") == "storage_reject": s2 = LS.LedgerStore() for a in V["assignments"]: s2.issue_assignment(a) for a in V["attestations"] + V.get("attestation_extra", []): s2.add_attestation(a) try: s2.append(dict(fx["rows"][0])) nets[name] = ("NO_REJECT", None) except LS.LedgerError: nets[name] = ("REJECTED", None) continue rows = renumber(strip_fixture(fx["rows"])) s2 = LS.LedgerStore() for a in V["assignments"]: s2.issue_assignment(a) for a in V["attestations"] + V.get("attestation_extra", []): s2.add_attestation(a) for r in rows: s2.append(r) nets[name] = (E.net_balance(s2.snapshot(), fx["actor"]), s2) check("R1_copycat storage-rejected (excluded class)", nets["R1_copycat_digest"][0] == "REJECTED") reds = [] for k in ("R2_plausible_incorrect", "R3_self_circle"): val = nets[k][0] check(f"{k} net == 0", val == 0, f"got {val}") if isinstance(val, int): reds.append(val) p1 = nets["P1_corrective"][0] check("P1 > max(R1..R3): 100M > 0", isinstance(p1, int) and p1 > max(reds + [0]), f"P1={p1}, reds={reds}") # ---- 2. all 8 falsifiers rejected (blocker 1) --------------------------- for f in V["falsifiers"]: s2 = LS.LedgerStore() for a in V["assignments"]: s2.issue_assignment(a) for a in V["attestations"] + V.get("attestation_extra", []): s2.add_attestation(a) try: s2.append(dict(f["row"])) check(f"falsifier:{f['name']} REJECTED", False, "accepted!") except LS.LedgerError: check(f"falsifier:{f['name']} rejected", True) # ---- 3. all 7 adversarial reversals rejected ---------------------------- # ---- 3. all 7 adversarial reversals rejected ---------------------------- # store: emission (was seq 2001) -> seq 1; first reversal (setup) -> seq 2 SEQ_MAP = {2001: 1, 2003: 2, 2010: 2, 2011: 2} base = renumber(strip_fixture(V["fixture_histories"]["R2_plausible_incorrect"]["rows"][:1])) def fix_rev(r): rr = {k: v for k, v in r.items() if k != "seq"} if rr.get("reverses_event_id") in SEQ_MAP: rr["reverses_event_id"] = SEQ_MAP[rr["reverses_event_id"]] rr.update({"type": "reversal", "role": "system", "ts": 999, "policy_version": "0.4.2", "task_id": "T2001"}) return rr for adv in V["adversarial_reversals"]: s2 = LS.LedgerStore() for a in V["assignments"]: s2.issue_assignment(a) for a in V["attestations"] + V.get("attestation_extra", []): s2.add_attestation(a) s2.append(base[0]) if adv.get("setup_first"): s2.append(fix_rev(adv["setup_first"])) try: s2.append(fix_rev(adv["reversal"])) check(f"adv:{adv['name']} REJECTED", False, "accepted!") except LS.LedgerError: check(f"adv:{adv['name']} rejected", True) # ---- 4. classifier reads attestation, not self-report (blocker 2) ------- for case in V["classifier_cases"]: if case.get("skip"): continue att = store.attestations.get(case["attestation_id"]) if att is None: check(f"classifier {case['name']} attestation missing", False) continue got, reason = CL.classify_attestation(att) check(f"classifier {case['name']} == {case['expect']}", got == case['expect'], f"got {got}: {reason} (self-reported sim={case.get('self_reported',{}).get('similarity','?')})") # C1 specifically: self-reported 0.30 clean, attested 0.97 -> flag c1_att = store.attestations.get("AT_FORGED") if c1_att: got, reason = CL.classify_attestation(c1_att) check("classifier C1 forged self-report (0.30) flagged on attested 0.97", got == "flag", f"got {got}: {reason}") # ---- 5. partial rating_adjustment (blocker 3) --------------------------- # emission 100 rating, reversal 10 rating_adjustment -> net rating 90 s3 = LS.LedgerStore() s3.issue_assignment({"task_id": "T_PARTIAL", "assignee": "A", "emission_class": "verify_accept", "amount_cap_micro": 100, "ts_issued": 1, "review_of": "B"}) s3.add_attestation({"attestation_id": "AT_PARTIAL", "controller": "CTRL", "similarity_measured": 0.5, "text_len_measured": 100, "authorship_self_verified": False, "ts": 1, "kind": "review"}) s3.append({"seq": 1, "type": "emission", "actor": "A", "role": "review", "emission_class": "verify_accept", "amount_micro": 100, "outcome": "accepted", "ts": 1, "policy_version": "0.4.2", "task_id": "T_PARTIAL", "attestation_id": "AT_PARTIAL"}) s3.append({"seq": 2, "type": "reversal", "actor": "A", "role": "system", "reverses_event_id": 1, "effect": "rating_adjustment", "basis": "misclassification", "amount_micro": 10, "ts": 2, "policy_version": "0.4.2", "task_id": "T_PARTIAL"}) rat = E.derived_rating(s3.snapshot()) check("partial rating_adjustment: 100-10=90", rat.get("A") == 90, f"got {rat.get('A')}") # ---- 6. conservation with pool from caps -------------------------------- all_rows = [] for name in ("R2_plausible_incorrect", "R3_self_circle", "P1_corrective"): all_rows += strip_fixture(V["fixture_histories"][name]["rows"]) used_tasks = {"T2001", "T2002", "T3001", "T3002", "T4001", "T4002"} used_assign = [a for a in V["assignments"] if a["task_id"] in used_tasks] ok1, d1 = E.conservation(all_rows, used_assign, edge=0) check("conservation pool=sum(caps) net+carry=pool", ok1, str(d1)) check("conservation carry = 200M (unspent pool remainder)", ok1 and d1["carry"] == 200_000_000, str(d1)) # ---- 7. fixture gate + presence gate ------------------------------------ s4 = LS.LedgerStore() for a in V["assignments"]: s4.issue_assignment(a) for a in V["attestations"] + V.get("attestation_extra", []): s4.add_attestation(a) try: s4.append({"seq": 1, "type": "emission", "actor": "A", "role": "create", "emission_class": "verify_accept", "amount_micro": 500, "outcome": "accepted", "ts": 1, "policy_version": "0.4.2", "task_id": "T2001", "is_fixture": True}) check("fixture gate (is_fixture+amount>0)", False) except LS.LedgerError: check("fixture gate (is_fixture+amount>0) rejected", True) s5 = LS.LedgerStore() s5.issue_assignment({"task_id": "T_PRES", "assignee": "A", "emission_class": "presence", "amount_cap_micro": 100, "ts_issued": 1}) try: s5.append({"seq": 1, "type": "emission", "actor": "A", "role": "create", "emission_class": "presence", "amount_micro": 100, "outcome": "recorded", "ts": 1, "policy_version": "0.4.2", "task_id": "T_PRES"}) check("presence gate (amount>0)", False) except LS.LedgerError: check("presence gate (amount>0) rejected", True) # ---- 8. double rating_adjustment guard on raw input --------------------- try: E.derived_rating([ {"type": "emission", "seq": 1, "actor": "A", "emission_class": "verify_accept", "amount_micro": 100, "outcome": "accepted", "ts": 1, "policy_version": "0.4.2"}, {"type": "reversal", "seq": 2, "actor": "A", "reverses_event_id": 1, "effect": "rating_adjustment", "basis": "fraud", "amount_micro": 50, "ts": 2, "policy_version": "0.4.2"}, {"type": "reversal", "seq": 3, "actor": "A", "reverses_event_id": 1, "effect": "rating_adjustment", "basis": "fraud", "amount_micro": 50, "ts": 3, "policy_version": "0.4.2"}, ]) check("double rating_adjustment guard", False) except ValueError: check("double rating_adjustment guard raises", True) # ---- 9. duplicate seq --------------------------------------------------- s6 = LS.LedgerStore() s6.issue_assignment({"task_id": "T_DUP", "assignee": "A", "emission_class": "verify_accept", "amount_cap_micro": 100, "ts_issued": 1, "review_of": "B"}) s6.add_attestation({"attestation_id": "AT_DUP", "controller": "CTRL", "similarity_measured": 0.5, "text_len_measured": 100, "authorship_self_verified": False, "ts": 1, "kind": "review"}) try: s6.append({"seq": 1, "type": "emission", "actor": "A", "role": "review", "emission_class": "verify_accept", "amount_micro": 100, "outcome": "accepted", "ts": 1, "policy_version": "0.4.2", "task_id": "T_DUP", "attestation_id": "AT_DUP"}) s6.append({"seq": 1, "type": "emission", "actor": "B", "role": "review", "emission_class": "verify_reject", "amount_micro": 100, "outcome": "rejected", "ts": 2, "policy_version": "0.4.2", "task_id": "T_DUP", "attestation_id": "AT_DUP"}) check("duplicate seq rejected", False) except LS.LedgerError: check("duplicate seq rejected", True) print("-" * 60) if fails: print(f"FAILURES: {len(fails)}") for f in fails: print(" FAIL", f) raise SystemExit(1) print("ALL TESTS PASSED")