""" SNIN Emission Paper Pilot — v0.4.1 runner. Usage: python3 run_tests.py Covers all four blockers from astranaut01 #24832: 1. canonical model (amount_micro only, derived deltas) 2. ledger_store enforces reversal integrity (adversarial vectors) 3. executable N-classifier with a real flagging surface 4. conservation with clawbacks and computed carry Exit 0 only if all pass. """ import json import emission as E import ledger_store as LS import classifier as CL V = json.load(open("test_vectors_v0.4.1.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): """Fixture rows model REAL events when testing the formula (astranaut01 #24180: P1 check calls the pure calculation, not the payment channel).""" return [{k: v for k, v in r.items() if k != "is_fixture"} for r in rows] # ---- 1. positive vs red (formula) ------------------------------------ nets = {} for name, fx in V["fixture_histories"].items(): rows = strip_fixture(fx.get("rows", [])) if fx.get("expected") == "storage_reject": store = LS.LedgerStore() try: store.append(rows[0]) nets[name] = ("NO_REJECT", None) except LS.LedgerError: nets[name] = ("REJECTED", None) continue actor = fx["actor"] try: store = LS.LedgerStore(rows) nets[name] = (E.net_balance(store.snapshot(), actor), store) except LS.LedgerError as e: nets[name] = (f"ERR:{e}", None) check("R1_copycat_digest 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. adversarial reversals (blocker #2) ---------------------------- base = strip_fixture(V["fixture_histories"]["R2_plausible_incorrect"]["rows"][:1]) # only 2001 for adv in V["adversarial_reversals"]: store = LS.LedgerStore(list(base)) # setup extra target rows where the test needs them for extra in adv.get("setup", []): store.append(extra) rev = dict(adv["reversal"]) rev.update({"type": "reversal", "role": "system", "ts": 999, "policy_version": "0.4.1"}) try: store.append(rev) check(f"adv:{adv['name']} REJECTED", False, "accepted — integrity hole!") except LS.LedgerError: check(f"adv:{adv['name']} rejected", True) # ---- fixture gate & class gates ---------------------------------------- g = LS.LedgerStore() try: g.append({"seq": 1, "type": "emission", "actor": "A", "role": "create", "emission_class": "verify_accept", "amount_micro": 500, "outcome": "accepted", "ts": 1, "policy_version": "0.4.1", "is_fixture": True}) check("fixture gate (is_fixture + amount>0)", False, "accepted") except LS.LedgerError: check("fixture gate (is_fixture + amount>0) rejected", True) g2 = LS.LedgerStore() try: g2.append({"seq": 1, "type": "emission", "actor": "A", "role": "create", "emission_class": "presence", "amount_micro": 100, "outcome": "recorded", "ts": 1, "policy_version": "0.4.1"}) check("presence gate (amount>0)", False, "accepted") except LS.LedgerError: check("presence gate (amount>0) rejected", True) # ---- 3. classifier: N clean, C flagged (blocker #3) -------------------- claims = V["claims"] exp = V["classifier_expectations"] for cid, want in exp.items(): claim = next(c for c in claims.values() if c["claim_id"] == cid or c.get("claim_id") == cid or list(claims.values()).index(c) == list(claims.keys()).index(cid)) for name, want in exp.items(): claim = claims[name] got, reason = CL.classify_claim(claim) check(f"classifier {name} == {want}", got == want, f"got {got}: {reason}") # prove the classifier CAN flag (meaningful negative control) flag_ok = any(CL.classify_claim(c)[0] == "flag" for c in claims.values()) check("classifier provably capable of flagging", flag_ok) # ---- 4. conservation with clawbacks (blocker #4) ------------------------ # full P1+R2+R3 histories: gross 600M, clawbacks 200M, net 400M all_rows = [] for name in ("R2_plausible_incorrect", "R3_self_circle", "P1_corrective"): all_rows += strip_fixture(V["fixture_histories"][name]["rows"]) ok1, d1 = E.conservation(all_rows, budget_micro=400_000_000, edge=0) check("conservation net(400M)+carry(0)==budget(400M)", ok1, str(d1)) ok2, d2 = E.conservation(all_rows, budget_micro=500_000_000, edge=0) check("conservation net(400M)+carry(100M)==budget(500M)", ok2 and d2["carry"] == 100_000_000, str(d2)) # duplicate seq rejected dup = LS.LedgerStore() try: dup.append({"seq": 1, "type": "emission", "actor": "A", "role": "create", "emission_class": "verify_accept", "amount_micro": 100, "outcome": "accepted", "ts": 1, "policy_version": "0.4.1"}) dup.append({"seq": 1, "type": "emission", "actor": "B", "role": "create", "emission_class": "verify_reject", "amount_micro": 100, "outcome": "rejected", "ts": 2, "policy_version": "0.4.1"}) check("duplicate seq rejected", False) except LS.LedgerError: check("duplicate seq rejected", True) # 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.1"}, {"type": "reversal", "seq": 2, "actor": "A", "reverses_event_id": 1, "effect": "rating_adjustment", "basis": "fraud", "amount_micro": 100, "ts": 2, "policy_version": "0.4.1"}, {"type": "reversal", "seq": 3, "actor": "A", "reverses_event_id": 1, "effect": "rating_adjustment", "basis": "fraud", "amount_micro": 100, "ts": 3, "policy_version": "0.4.1"}, ]) check("double rating_adjustment guard", False) except ValueError: check("double rating_adjustment guard raises", True) print("-" * 60) if fails: print(f"FAILURES: {len(fails)}") for f in fails: print(" FAIL", f) raise SystemExit(1) print("ALL TESTS PASSED")