""" SNIN Emission Paper Pilot — append-only ledger store v0.4.2 (FROZEN). Boundary blockers #24948 fixed: (1) JSON Schema is ENFORCED on every append (jsonschema library). - required fields enforced (actor/role/ts/outcome/task_id/...) - seq MUST be monotonic: seq == last_seq + 1 - policy_version MUST equal the window version (0.4.2) (2) Capability/receipt boundary against self-attestation: - emission rows exist ONLY against a pre-issued task_assignment - class == assignment class; actor == assignee; amount <= amount_cap - review classes: assignee MUST differ from review_of (no self-award) - review emissions REQUIRE a controller attestation_id; the classifier reads ONLY attested fields (never self-reported) (3) Partial rating_adjustment: reversal amount is capped by the target credit and derived_rating subtracts exactly the reversal amount. Falsifiers from #24948 all rejected here: missing-required-field, seq 10 then seq 2, policy 0.3 in frozen 0.4.2, self-awarded verify_accept, forged classifier features, partial 10/100. """ import json import jsonschema import emission as E WINDOW_POLICY_VERSION = "0.4.2" REVIEW_CLASSES = {"verify_accept", "verify_reject", "recheck_assigned", "recheck_confirm", "recheck_overturn"} with open("schema_v0.4.2.json") as f: SCHEMA = json.load(f) class LedgerError(Exception): pass class LedgerStore: def __init__(self, assignments=None, attestations=None, rows=None, policy_version=WINDOW_POLICY_VERSION): self._rows = [] self._by_seq = {} self.policy_version = policy_version self.assignments = {a["task_id"]: a for a in (assignments or [])} self.attestations = {a["attestation_id"]: a for a in (attestations or [])} if rows: for r in rows: self.append(r) # ---- capability registry ------------------------------------------- def issue_assignment(self, assignment): a = dict(assignment) a["policy_version"] = self.policy_version tid = a["task_id"] if tid in self.assignments: raise LedgerError(f"duplicate task assignment {tid}") jsonschema.validate(a, SCHEMA["$defs"]["task_assignment"]) self.assignments[tid] = a return tid def add_attestation(self, att): a = dict(att) a["policy_version"] = self.policy_version aid = a["attestation_id"] if aid in self.attestations: raise LedgerError(f"duplicate attestation {aid}") jsonschema.validate(a, SCHEMA["$defs"]["attestation"]) self.attestations[aid] = a return aid # ---- validation ---------------------------------------------------- def _check_capability(self, row): """Emission exists only against a pre-issued assignment (blocker 2).""" task = self.assignments.get(row.get("task_id")) if task is None: raise LedgerError(f"no pre-issued task assignment for task_id={row.get('task_id')} " "(capability boundary: emission requires issued capability)") if task["emission_class"] != row["emission_class"]: raise LedgerError(f"task {task['task_id']} class {task['emission_class']} " f"!= row class {row['emission_class']}") if task["assignee"] != row["actor"]: raise LedgerError(f"task assignee {task['assignee']} != row actor {row['actor']}") if int(row["amount_micro"]) > int(task["amount_cap_micro"]): raise LedgerError(f"amount {row['amount_micro']} > task cap {task['amount_cap_micro']} " "(cap comes from policy, not from the row)") if row["emission_class"] in REVIEW_CLASSES: if task.get("review_of") is None: raise LedgerError(f"review task {task['task_id']} missing review_of") if task["assignee"] == task["review_of"]: raise LedgerError("self-award: reviewer == claim author, forbidden") return True def _check_attestation(self, row): """Review emissions must cite a controller attestation (blocker 2).""" if row["emission_class"] not in REVIEW_CLASSES: return True aid = row.get("attestation_id") if not aid: raise LedgerError(f"review emission {row['seq']} missing attestation_id") att = self.attestations.get(aid) if att is None: raise LedgerError(f"attestation {aid} does not exist") if att.get("controller") == row["actor"]: raise LedgerError("attestation controller must not be the row actor (independent measurement)") return True def validate_emission(self, row): if row.get("is_fixture") and int(row.get("amount_micro", 0)) != 0: raise LedgerError("fixture gate: is_fixture=true => amount_micro MUST be 0") if row["emission_class"] in ("presence", "task_receipt") and int(row.get("amount_micro", 0)) != 0: raise LedgerError(f"{row['emission_class']} never emits: amount MUST be 0") self._check_capability(row) self._check_attestation(row) return True def validate_reversal(self, row): if not row.get("reverses_event_id"): raise LedgerError("reversal missing reverses_event_id") target = self._by_seq.get(row["reverses_event_id"]) if target is None: raise LedgerError(f"reversal target seq {row['reverses_event_id']} does not exist") if target.get("type") != "emission": raise LedgerError("reversal target must be an emission row, not another reversal") if target.get("is_fixture"): raise LedgerError("reversal target must not be a fixture row") if row.get("actor") != target.get("actor"): raise LedgerError(f"reversal actor {row.get('actor')} != target actor {target.get('actor')}") basis = row.get("basis") effect = row.get("effect") if effect not in E.REVERSAL_BASIS_EFFECTS.get(basis, set()): raise LedgerError(f"effect {effect} not permitted for basis {basis}") for r in self._rows: if r.get("type") == "reversal" and \ r.get("reverses_event_id") == row["reverses_event_id"] and \ r.get("effect") == effect: raise LedgerError(f"duplicate reversal ({row['reverses_event_id']}, {effect})") q = E.quoted_emission(target) cap = q.get("balance", 0) if effect == "balance_clawback" else q.get("rating", 0) if int(row.get("amount_micro", 0)) > cap: raise LedgerError(f"over-{effect}: {row.get('amount_micro')} > target credit {cap}") return True # ---- append -------------------------------------------------------- def append(self, row): r = dict(row) r["policy_version"] = r.get("policy_version", self.policy_version) # (1) fixed window policy_version if r["policy_version"] != self.policy_version: raise LedgerError(f"policy_version {r['policy_version']} != frozen {self.policy_version}") r["type"] = r.get("type") or ("reversal" if r.get("reverses_event_id") else "emission") # (1) monotonic seq: seq == last_seq + 1 last = max([x["seq"] for x in self._rows] or [0]) if r.get("seq") is None: r["seq"] = last + 1 if r["seq"] != last + 1: raise LedgerError(f"non-monotonic seq: got {r['seq']}, expected {last + 1}") # (1) full JSON Schema validation on append try: jsonschema.validate(r, SCHEMA) except jsonschema.ValidationError as e: raise LedgerError(f"schema violation: {e.message}") if r["type"] == "emission": self.validate_emission(r) else: self.validate_reversal(r) self._rows.append(r) self._by_seq[r["seq"]] = r return r["seq"] def snapshot(self): return json.loads(json.dumps(self._rows))