# v.0.2.17 # { "Depends": "py-genlayer:1jb45aa8ynh2a9c9xn3b7qqh8sm5q93hwfp7jqmwsfhh8jpz09h6" } from genlayer import * from datetime import datetime, timezone import json import typing class ReasonLayer(gl.Contract): # ── Primitive config ────────────────────────────────────────────────── owner: str # hex address as string paused: bool total_members: u32 total_validators: u32 total_evaluations: u32 dispute_count: u32 # ── Role registry ───────────────────────────────────────────────────── # value: "admin" | "validator" | "member" roles: TreeMap[Address, str] # ── Profiles (JSON-encoded dicts stored as str) ─────────────────────── members: TreeMap[Address, str] validators: TreeMap[Address, str] # ── Trust scores (JSON-encoded) ─────────────────────────────────────── trust_scores: TreeMap[Address, str] # ── Evaluations ─────────────────────────────────────────────────────── # key: str(ev_id) value: JSON-encoded evaluation record evaluations: TreeMap[str, str] # key: address value: JSON-encoded list of string IDs member_evaluation_ids: TreeMap[Address, str] # ── Endorsements ────────────────────────────────────────────────────── # key: address value: JSON-encoded list of endorser hex strings endorsements: TreeMap[Address, str] # key: "from_hex:to_hex" value: True if endorsement exists endorsement_given: TreeMap[str, bool] # ── Disputes ───────────────────────────────────────────────────────── # key: str(dispute_id) value: JSON-encoded dispute record disputes: TreeMap[str, str] # ── Recommendations ─────────────────────────────────────────────────── # key: address value: JSON-encoded recommendation set recommendations: TreeMap[Address, str] # ------------------------------------------------------------------------- # Constructor # ------------------------------------------------------------------------- def __init__(self) -> None: self.owner = str(gl.message.sender_address) self.paused = False self.total_members = u32(0) self.total_validators = u32(0) self.total_evaluations = u32(0) self.dispute_count = u32(0) # ------------------------------------------------------------------------- # Internal helpers # ------------------------------------------------------------------------- def _now(self) -> int: return int(datetime.now(timezone.utc).timestamp()) def _require_not_paused(self) -> None: if self.paused: raise Exception("Contract is paused") def _require_owner(self) -> None: if str(gl.message.sender_address) != self.owner: raise Exception("Owner only") def _require_admin(self) -> None: role = self.roles.get(gl.message.sender_address, "") if role != "admin" and str(gl.message.sender_address) != self.owner: raise Exception("Admin only") def _require_registered(self) -> None: if gl.message.sender_address not in self.members: raise Exception("Not a registered member") def _score_to_tier(self, score: int) -> str: if score < 20: return "Unverified" if score < 40: return "Bronze" if score < 60: return "Silver" if score < 80: return "Gold" return "Platinum" def _parse_llm_json(self, raw: str) -> dict: cleaned = raw.replace("```json", "").replace("```", "").strip() return json.loads(cleaned) # ------------------------------------------------------------------------- # View Methods # ------------------------------------------------------------------------- @gl.public.view def get_member(self, address: str) -> str: addr = Address(address) if addr not in self.members: raise Exception("Member not found") return self.members[addr] @gl.public.view def get_validator(self, address: str) -> str: addr = Address(address) if addr not in self.validators: raise Exception("Validator not found") return self.validators[addr] @gl.public.view def get_trust_passport(self, address: str) -> str: addr = Address(address) if addr not in self.trust_scores: raise Exception("No trust passport -- request evaluation first") return self.trust_scores[addr] @gl.public.view def get_evaluations(self, address: str, limit: u32, offset: u32) -> str: addr = Address(address) raw = self.member_evaluation_ids.get(addr, "[]") ids = json.loads(raw) start = int(offset) end = min(start + int(limit), len(ids)) result = [] for eid in ids[start:end]: key = str(eid) if key in self.evaluations: result.append(json.loads(self.evaluations[key])) return json.dumps(result) @gl.public.view def get_recommendations(self, address: str) -> str: addr = Address(address) if addr not in self.recommendations: raise Exception("No recommendations -- generate them first") return self.recommendations[addr] @gl.public.view def get_validator_registry(self, limit: u32, offset: u32) -> str: result = [] count = 0 skipped = 0 for addr in self.validators: if skipped < int(offset): skipped += 1 continue if count >= int(limit): break v = json.loads(self.validators[addr]) if v.get("is_active", False): result.append(v) count += 1 return json.dumps(result) @gl.public.view def get_endorsements(self, address: str) -> str: addr = Address(address) return self.endorsements.get(addr, "[]") @gl.public.view def get_dispute(self, dispute_id: u32) -> str: key = str(int(dispute_id)) if key not in self.disputes: raise Exception("Dispute not found") return self.disputes[key] @gl.public.view def is_registered(self, address: str) -> bool: return Address(address) in self.members @gl.public.view def get_stats(self) -> str: return json.dumps({ "total_members": int(self.total_members), "total_validators": int(self.total_validators), "total_evaluations": int(self.total_evaluations), "total_disputes": int(self.dispute_count), }) # ------------------------------------------------------------------------- # Deterministic Write Methods # ------------------------------------------------------------------------- @gl.public.write def register_member(self, display_name: str, bio: str) -> None: self._require_not_paused() sender = gl.message.sender_address if sender in self.members: raise Exception("Already registered") if not display_name.strip(): raise Exception("display_name cannot be empty") self.members[sender] = json.dumps({ "address": str(sender), "display_name": display_name.strip()[:64], "bio": bio.strip()[:512], "registered_at": self._now(), "is_active": True, "endorsement_count": 0, "dispute_count": 0, }) self.roles[sender] = "member" self.total_members = u32(int(self.total_members) + 1) @gl.public.write def register_validator(self, moniker: str, website: str) -> None: self._require_not_paused() sender = gl.message.sender_address self._require_registered() if sender in self.validators: raise Exception("Already registered as validator") if not moniker.strip(): raise Exception("moniker cannot be empty") self.validators[sender] = json.dumps({ "address": str(sender), "moniker": moniker.strip()[:64], "website": website.strip()[:256], "is_active": True, "registered_at": self._now(), "uptime_score": 100, "total_evaluations_given": 0, }) self.roles[sender] = "validator" self.total_validators = u32(int(self.total_validators) + 1) @gl.public.write def update_profile(self, display_name: str, bio: str) -> None: self._require_not_paused() sender = gl.message.sender_address self._require_registered() if not display_name.strip(): raise Exception("display_name cannot be empty") profile = json.loads(self.members[sender]) profile["display_name"] = display_name.strip()[:64] profile["bio"] = bio.strip()[:512] self.members[sender] = json.dumps(profile) @gl.public.write def submit_endorsement(self, target: str) -> None: self._require_not_paused() sender = gl.message.sender_address self._require_registered() target_addr = Address(target) if target_addr not in self.members: raise Exception("Target is not a registered member") if str(sender) == str(target_addr): raise Exception("Cannot endorse yourself") ekey = str(sender) + ":" + str(target_addr) if self.endorsement_given.get(ekey, False): raise Exception("Already endorsed this member") self.endorsement_given[ekey] = True existing = json.loads(self.endorsements.get(target_addr, "[]")) existing.append(str(sender)) self.endorsements[target_addr] = json.dumps(existing) profile = json.loads(self.members[target_addr]) profile["endorsement_count"] = profile.get("endorsement_count", 0) + 1 self.members[target_addr] = json.dumps(profile) @gl.public.write def submit_dispute(self, target: str, reason: str, evidence_url: str) -> None: self._require_not_paused() sender = gl.message.sender_address self._require_registered() target_addr = Address(target) if target_addr not in self.members: raise Exception("Target is not a registered member") if str(sender) == str(target_addr): raise Exception("Cannot dispute yourself") if not reason.strip(): raise Exception("reason cannot be empty") d_id = int(self.dispute_count) self.disputes[str(d_id)] = json.dumps({ "d_id": d_id, "target": str(target_addr), "raised_by": str(sender), "reason": reason.strip()[:1024], "evidence_url": evidence_url.strip()[:512], "resolution": "pending", "ai_verdict": "", "raised_at": self._now(), "resolved_at": 0, }) self.dispute_count = u32(d_id + 1) profile = json.loads(self.members[target_addr]) profile["dispute_count"] = profile.get("dispute_count", 0) + 1 self.members[target_addr] = json.dumps(profile) @gl.public.write def set_validator_uptime(self, address: str, score: u32) -> None: self._require_admin() addr = Address(address) if addr not in self.validators: raise Exception("Validator not found") if int(score) > 100: raise Exception("score must be 0-100") v = json.loads(self.validators[addr]) v["uptime_score"] = int(score) self.validators[addr] = json.dumps(v) @gl.public.write def grant_role(self, address: str, role: str) -> None: self._require_owner() if role not in ("admin", "validator", "member"): raise Exception("Invalid role") self.roles[Address(address)] = role @gl.public.write def pause_contract(self) -> None: self._require_owner() self.paused = True @gl.public.write def unpause_contract(self) -> None: self._require_owner() self.paused = False # ------------------------------------------------------------------------- # AI Leader Functions # ------------------------------------------------------------------------- @gl.public.write def request_trust_evaluation(self, address: str) -> None: # Equivalence: gl.vm.run_nondet_unsafe with partial field match. # tier must agree exactly (derived deterministically from score range). # score tolerance: +/- 5 points (LLM numeric variance). # reasoning and flags are informational only -- not compared. self._require_not_paused() addr = Address(address) if addr not in self.members: raise Exception("Member not found") profile = json.loads(self.members[addr]) endorsement_count = profile.get("endorsement_count", 0) dispute_count = profile.get("dispute_count", 0) bio = profile.get("bio", "") prompt = ( "You are an AI trust evaluator for ReasonLayer, a decentralized validator intelligence network.\n" "Evaluate the trustworthiness of this on-chain member profile.\n\n" "\n" "Address: " + address + "\n" "Display Name: " + profile.get("display_name", "") + "\n" "Bio: " + bio + "\n" "Endorsement Count: " + str(endorsement_count) + "\n" "Dispute Count: " + str(dispute_count) + "\n" "\n\n" "Scoring rules:\n" " - Start at 50 (neutral baseline)\n" " - Each endorsement adds 5 points (maximum bonus: +25)\n" " - Each dispute subtracts 15 points\n" " - Bio longer than 100 characters adds 5 points\n" " - No bio or very short bio subtracts 5 points\n" " - Clamp final score to 0-100\n\n" "Tier mapping:\n" " 0-19: Unverified | 20-39: Bronze | 40-59: Silver | 60-79: Gold | 80-100: Platinum\n\n" "Respond ONLY with a JSON object. No markdown. No explanation outside the JSON.\n" "Required keys:\n" ' "score": integer 0-100\n' ' "tier": one of "Unverified", "Bronze", "Silver", "Gold", "Platinum"\n' ' "reasoning": string, 2-3 sentences\n' ' "flags": array of risk-factor strings (empty array if none)\n' ) def leader_fn(): raw = gl.nondet.exec_prompt(prompt) data = self._parse_llm_json(raw) score = max(0, min(100, int(data.get("score", 50)))) tier = self._score_to_tier(score) return { "score": score, "tier": tier, "reasoning": str(data.get("reasoning", "")), "flags": data.get("flags", []), } def validator_fn(leader_result) -> bool: if not isinstance(leader_result, gl.vm.Return): return False mine = leader_fn() data = leader_result.calldata return ( data["tier"] == mine["tier"] and abs(data["score"] - mine["score"]) <= 5 ) result = gl.vm.run_nondet_unsafe(leader_fn, validator_fn) # All storage writes happen after consensus ev_id = int(self.total_evaluations) ev_record = { "ev_id": ev_id, "subject": address, "requested_by": str(gl.message.sender_address), "score": result["score"], "tier": result["tier"], "reasoning": result["reasoning"], "flags": result["flags"], "evaluated_at": self._now(), } self.evaluations[str(ev_id)] = json.dumps(ev_record) existing_ids = json.loads(self.member_evaluation_ids.get(addr, "[]")) existing_ids.append(ev_id) self.member_evaluation_ids[addr] = json.dumps(existing_ids) self.total_evaluations = u32(ev_id + 1) prev_count = 0 if addr in self.trust_scores: prev = json.loads(self.trust_scores[addr]) prev_count = prev.get("evaluation_count", 0) self.trust_scores[addr] = json.dumps({ "address": address, "score": result["score"], "tier": result["tier"], "reasoning": result["reasoning"], "flags": result["flags"], "last_evaluated_at": self._now(), "evaluation_count": prev_count + 1, }) @gl.public.write def generate_recommendations(self, for_address: str) -> None: # Equivalence: prompt_non_comparative. # Validators judge the leader output against criteria without # re-running generation -- rankings are open-ended and cannot # be reproduced identically across nodes. self._require_not_paused() addr = Address(for_address) if addr not in self.members: raise Exception("Member not found") profile = json.loads(self.members[addr]) trust_info = None if addr in self.trust_scores: ts = json.loads(self.trust_scores[addr]) trust_info = {"score": ts.get("score", 0), "tier": ts.get("tier", "Unverified")} validator_list = [] count = 0 for vaddr in self.validators: v = json.loads(self.validators[vaddr]) if v.get("is_active", False) and v.get("uptime_score", 0) >= 80: validator_list.append({ "address": str(vaddr), "moniker": v.get("moniker", ""), "uptime_score": v.get("uptime_score", 0), }) count += 1 if count >= 20: break if count < 3: raise Exception("Not enough active validators (need at least 3)") member_context = json.dumps({ "address": for_address, "display_name": profile.get("display_name", ""), "bio": profile.get("bio", ""), "endorsement_count": profile.get("endorsement_count", 0), "dispute_count": profile.get("dispute_count", 0), "trust": trust_info, "active_validators": validator_list, }) task = ( "Review this member profile and the active validator registry. " "Recommend the top 3 most compatible validators for this member " "based on their trust tier, profile content, and network activity." ) criteria = ( "Return a JSON object with exactly two keys: " '"addresses" (array of exactly 3 validator wallet address strings ' "from the active_validators list) and " '"reasoning" (2-3 sentence string explaining the selection). ' "Addresses must appear verbatim from active_validators. " "No markdown. JSON only." ) result = gl.eq_principle.prompt_non_comparative( lambda: member_context, task=task, criteria=criteria, ) parsed = self._parse_llm_json(result) if isinstance(result, str) else result self.recommendations[addr] = json.dumps({ "for_address": for_address, "validator_addresses": parsed.get("addresses", []), "reasoning": str(parsed.get("reasoning", "")), "generated_at": self._now(), }) @gl.public.write def resolve_dispute(self, dispute_id: u32) -> None: # Equivalence: prompt_comparative. # Leader and validators independently assess the dispute. # The "verdict" field must match exactly; reasoning may differ. self._require_not_paused() self._require_admin() key = str(int(dispute_id)) if key not in self.disputes: raise Exception("Dispute not found") dispute = json.loads(self.disputes[key]) if dispute.get("resolution") != "pending": raise Exception("Dispute already resolved") target_name = "" target_endorsements = 0 target_disputes_count = 0 try: target_addr = Address(dispute["target"]) if target_addr in self.members: tp = json.loads(self.members[target_addr]) target_name = tp.get("display_name", "") target_endorsements = tp.get("endorsement_count", 0) target_disputes_count = tp.get("dispute_count", 0) except Exception: pass dispute_context = json.dumps({ "target": dispute.get("target", ""), "raised_by": dispute.get("raised_by", ""), "reason": dispute.get("reason", ""), "evidence_url": dispute.get("evidence_url", ""), "target_display_name": target_name, "target_endorsements": target_endorsements, "target_disputes": target_disputes_count, }) def evaluate_dispute(): prompt = ( "You are an AI arbitrator for ReasonLayer.\n" "Assess this dispute and return a verdict.\n\n" "\n" + dispute_context + "\n\n\n" 'Return "upheld" if the reason is serious and evidence is credible.\n' 'Return "rejected" if the reason is vague or appears bad-faith.\n\n' "Respond ONLY with a JSON object. No markdown.\n" "Required keys:\n" ' "verdict": "upheld" or "rejected"\n' ' "reasoning": string, 2-3 sentences\n' ' "confidence": integer 0-100\n' ) raw = gl.nondet.exec_prompt(prompt) data = self._parse_llm_json(raw) verdict = str(data.get("verdict", "")).lower().strip() if verdict not in ("upheld", "rejected"): raise Exception("Invalid verdict: " + verdict) return json.dumps({ "verdict": verdict, "reasoning": str(data.get("reasoning", "")), "confidence": int(data.get("confidence", 50)), }, sort_keys=True) result_str = gl.eq_principle.prompt_comparative( evaluate_dispute, principle=( 'The "verdict" field must be exactly the same ' '("upheld" or "rejected"). Reasoning may differ.' ), ) parsed = self._parse_llm_json(result_str) if isinstance(result_str, str) else result_str dispute["resolution"] = parsed["verdict"] dispute["ai_verdict"] = parsed["reasoning"] dispute["resolved_at"] = self._now() self.disputes[key] = json.dumps(dispute)