# { "Depends": "py-genlayer:1jb45aa8ynh2a9c9xn3b7qqh8sm5q93hwfp7jqmwsfhh8jpz09h6" } """A reusable, consensus-backed admission gate for public evidence bundles.""" from genlayer import * import json import typing class EvidenceQuorum(gl.Contract): """Persist policy-bound evidence decisions instead of a mutable latest result.""" decision_count: u64 latest_decision_id: str decision_ids: DynArray[str] decisions: TreeMap[str, str] def __init__(self): self.decision_count = u64(0) self.latest_decision_id = "" @gl.public.view def get_decision_count(self) -> u64: return self.decision_count @gl.public.view def get_latest_decision_id(self) -> str: return self.latest_decision_id @gl.public.view def get_decision(self, decision_id: str) -> str: return self.decisions.get(decision_id, "") @gl.public.view def list_decision_ids(self) -> str: return json.dumps([item for item in self.decision_ids], separators=(",", ":")) @gl.public.write def assess_bundle(self, purpose: str, policy: str, source_urls: DynArray[str]) -> str: if len(purpose.strip()) < 12: raise Exception("purpose is too short") if len(policy.strip()) < 24: raise Exception("policy must describe a real admission rule") urls = _validate_urls([url for url in source_urls]) manifest = _manifest(urls) if len({item["host"] for item in manifest}) < 2: raise Exception("at least two independent source hosts are required") def leader_fn(): return _assess(purpose, policy, urls, manifest) def validator_fn(leader_result) -> bool: if not isinstance(leader_result, gl.vm.Return): return False try: proposed = _parse(leader_result.calldata) independent = _parse(_assess(purpose, policy, urls, manifest)) except Exception: return False return ( proposed["decision"] == independent["decision"] and proposed["policy_hash"] == independent["policy_hash"] and proposed["evidence_bundle_hash"] == independent["evidence_bundle_hash"] and proposed["independent_hosts"] == independent["independent_hosts"] and abs(proposed["confidence"] - independent["confidence"]) <= 18 and abs(proposed["satisfying_sources"] - independent["satisfying_sources"]) <= 1 ) result = json.loads(gl.vm.run_nondet_unsafe(leader_fn, validator_fn)) self.decision_count = u64(int(self.decision_count) + 1) decision_id = str(int(self.decision_count)) record = { "decision_id": decision_id, "purpose": purpose, "policy": policy, "policy_hash": result["policy_hash"], "source_manifest": manifest, "submitted_by": str(gl.message.sender_address), "result": result, "accepted_write": { "decision_id": decision_id, "registry_sequence": int(self.decision_count), "evidence_bundle_hash": result["evidence_bundle_hash"], }, } self.decisions[decision_id] = json.dumps(record, sort_keys=True, separators=(",", ":")) self.decision_ids.append(decision_id) self.latest_decision_id = decision_id return decision_id def _assess(purpose: str, policy: str, urls: typing.Sequence[str], manifest: typing.Sequence[dict]) -> str: snapshots, hashes = [], [] for index, url in enumerate(urls): text = gl.nondet.web.render(url, mode="text")[:6000] snapshot_hash = _stable_hash(text) hashes.append(snapshot_hash) snapshots.append({"index": index + 1, "host": manifest[index]["host"], "snapshot_hash": snapshot_hash, "text": text}) policy_hash = _stable_hash(policy.strip()) bundle_hash = _stable_hash(json.dumps(hashes, separators=(",", ":"))) prompt = f""" Assess whether this public evidence bundle satisfies the stated admission policy. Purpose: {purpose} Policy: {policy} Provenance: {json.dumps(manifest)} Evidence: {json.dumps(snapshots)} Return only minified JSON with exactly: decision ("admit", "reject", or "needs_review"), confidence (0-100), independent_hosts, satisfying_sources, policy_hash (exactly "{policy_hash}"), evidence_bundle_hash (exactly "{bundle_hash}"), and summary (under 450 chars). Admit only when the policy is directly supported by diverse sources. Reject when material evidence contradicts it. Use needs_review for incomplete, inaccessible, duplicated, or ambiguous evidence. """ raw = json.loads(gl.nondet.exec_prompt(prompt)) result = { "decision": str(raw["decision"]).lower(), "confidence": max(0, min(100, int(raw["confidence"]))), "independent_hosts": int(raw["independent_hosts"]), "satisfying_sources": int(raw["satisfying_sources"]), "policy_hash": str(raw["policy_hash"]), "evidence_bundle_hash": str(raw["evidence_bundle_hash"]), "summary": str(raw["summary"])[:450], } return json.dumps(result, sort_keys=True, separators=(",", ":")) def _parse(raw: str) -> dict: item = json.loads(raw) if item["decision"] not in ("admit", "reject", "needs_review"): raise Exception("invalid decision") if not 0 <= int(item["confidence"]) <= 100: raise Exception("invalid confidence") if not 2 <= int(item["independent_hosts"]) <= 5: raise Exception("invalid host count") if not 0 <= int(item["satisfying_sources"]) <= 5: raise Exception("invalid source count") if not str(item["policy_hash"]).startswith("h") or not str(item["evidence_bundle_hash"]).startswith("h"): raise Exception("invalid commitment") if not 0 < len(str(item["summary"])) <= 450: raise Exception("invalid summary") item["confidence"] = int(item["confidence"]) item["independent_hosts"] = int(item["independent_hosts"]) item["satisfying_sources"] = int(item["satisfying_sources"]) return item def _validate_urls(raw_urls: typing.Sequence[str]) -> typing.Sequence[str]: if len(raw_urls) < 2 or len(raw_urls) > 5: raise Exception("use two to five sources") urls, seen = [], set() for raw in raw_urls: url = str(raw).strip() key = url.rstrip("/").lower() if not url.startswith("https://") or len(url) > 500: raise Exception("sources must be short https URLs") if key in seen: raise Exception("duplicate sources are not allowed") seen.add(key) urls.append(url) return urls def _manifest(urls: typing.Sequence[str]) -> typing.Sequence[dict]: return [{"source_index": index + 1, "url": url, "host": _host(url), "url_hash": _stable_hash(url)} for index, url in enumerate(urls)] def _host(url: str) -> str: host = url.split("//", 1)[1].split("/", 1)[0].lower() if not host or "." not in host: raise Exception("invalid source host") return host def _stable_hash(value: str) -> str: total = 0 for char in value: total = (total * 131 + ord(char)) % 1000000007 return "h" + str(total)